fix(host/windows): two virtual pads stop tearing each other's reports

Three faults on the Windows pad path, two of them races that only bite when a
game drives a pad hard enough for two callbacks to overlap.

pf-gamepad's output ring could hand the host a torn report. Publishing is a
read-modify-write — read the cursor, write the slot it names, advance it — and
the framework dispatches output callbacks in parallel, so two could be inside
it at once: both read the same head, both wrote the SAME slot, and both stored
head+1, so the cursor moved once for two reports and the host read a single
entry with two reports mixed into it. An atomic fetch_add does not fix this. It
hands each writer its own slot but advances the cursor before the bytes exist,
so the host is then invited to read a slot still being filled. Serializing the
publish is what makes the cursor bump mean "the slot below is complete". The
ring exists to stop a rumble STOP being coalesced away, and a torn slot can eat
that STOP with no idle watchdog behind it.

Both drivers also promised the host an ordering they never established. The
host loads out_seq and rumble_seq with Acquire and says so in its own comments
— "Acquire pairs with the driver's publish-then-bump store order" — but the
drivers bumped both with plain writes, and an Acquire load pairs with a Release
store and nothing else. On a weakly-ordered core the host could see a fresh seq
against stale bytes. pf-xusb's rumble seq was racy in the same way as the ring:
two SET_STATE calls could both read one value and both write back value+1, so
the host saw one bump for two writes and skipped a level. A skipped stop is the
one that hurts — the pad buzzes until the ~2.5 s idle force-off notices the
game went quiet, which is what bounds the damage.

Diagnosing an unattached driver stalled the session. The pad service thread —
the one feeding input and rumble — waited up to two seconds for a pnputil
enumeration, per unattached pad, at exactly the moment a session was already
going wrong. The diagnosis now runs on its own thread. Off the hot path the
wait no longer has to be a compromise, so it is generous enough to report what
it actually found instead of giving up with "still enumerating" — which, given
pnputil routinely takes longer than the old budget, is what it usually did.

.gitea/workflows/android.yml

@@ -160,14 +160,6 @@ jobs:
           key: gradle-${{ hashFiles('clients/android/**/*.gradle.kts', 'clients/android/gradle/wrapper/gradle-wrapper.properties') }}
           restore-keys: gradle-
 
-      # The kit's JVM unit tests — the pure parsers, migrations and feedback policies. They were
-      # running nowhere: this workflow only assembled, and android-screenshots.yml runs the :app
-      # module's tests, so nothing enforced :kit's. Cheap (a couple of seconds against an already
-      # built module) and it is the only automated cover those behaviours have.
-      - name: kit unit tests
-        working-directory: clients/android
-        run: ./gradlew :kit:testDebugUnitTest --stacktrace
-
       - name: assembleDebug (cargo-ndk → jniLibs → APK)
         working-directory: clients/android
         env:

clients/android/kit/src/main/kotlin/io/unom/punktfunk/kit/DsCapture.kt

@@ -116,9 +116,7 @@ class DsCapture(
             // The interfaces are about to release with the kernel driver still detached — a
             // mid-rumble teardown would leave the motors running with nobody to stop them.
             // EP0-direct (the reader thread is stopping; the queue would never drain).
-            // Nothing can retry after this point, so a failure is worth saying out loud: it is
-            // the difference between a quiet pad and one that buzzes until it is unplugged.
-            if (!usb.writeControl(stopReport(m))) Log.w(TAG, "teardown rumble stop was not written")
+            usb.writeControl(stopReport(m))
         }
         disarmBackstop()
         usb.stop()
@@ -147,9 +145,6 @@ class DsCapture(
         val wasActive = model != null
         model = null
         releaseSlot()
-        // Release the transport too: the link only *signals* the drop, so without this an unplug
-        // left its connection open, its interfaces claimed and its detach receiver registered.
-        usb.stop()
         if (wasActive) onActiveChanged?.invoke(false)
     }
 
@@ -221,20 +216,17 @@ class DsCapture(
 
     override fun rumble(pad: Int, low: Int, high: Int, backstopMs: Long) {
         val m = model ?: return
-        val stop = low == 0 && high == 0
-        if (!stop) armBackstop(backstopMs)
-        val sent = if (m == DsDevice.Model.DUALSHOCK4) {
+        if (low == 0 && high == 0) {
+            disarmBackstop()
+        } else {
+            armBackstop(backstopMs)
+        }
+        if (m == DsDevice.Model.DUALSHOCK4) {
             ds4Low = low
             ds4High = high
             writeDs4()
         } else {
-            usb.writeRaw(0, DsDevice.ds5RumbleReport(m, low, high), OutReportQueue.KEY_RUMBLE)
-        }
-        if (stop) {
-            // Disarm only once the stop is actually on its way. Dropping the net *before* the
-            // write — as this used to — meant a discarded stop left the motors running with
-            // nothing scheduled to try again; a USB pad holds its last level until told zero.
-            if (sent) disarmBackstop() else armBackstop(STOP_RETRY_MS)
+            usb.writeRaw(0, DsDevice.ds5RumbleReport(m, low, high))
         }
     }
 
@@ -260,9 +252,6 @@ class DsCapture(
         usb.writeRaw(0, DsDevice.ds5TriggerReport(m, which, effect))
     }
 
-    // Coalescable: the DS4's write is full-state (motors AND lightbar, rebuilt from the current
-    // fields on every call), so a newer one supersedes an older one wholesale — nothing is lost by
-    // collapsing a backlog of them down to the last.
     private fun writeDs4() = usb.writeRaw(
         0,
         DsDevice.ds4Report(
@@ -272,7 +261,6 @@ class DsCapture(
             (ds4Rgb shr 8) and 0xFF,
             ds4Rgb and 0xFF,
         ),
-        OutReportQueue.KEY_RUMBLE,
     )
 
     /** The report that stops the motors. The DS4's is a full-state write, so it zeroes the
@@ -296,12 +284,7 @@ class DsCapture(
         backstop?.let { mainHandler.removeCallbacks(it) }
         val r = Runnable {
             backstop = null
-            val m = model ?: return@Runnable
-            // The net itself can be refused (a full queue, a connection going away). Re-arm rather
-            // than give up: this is the last thing between a stalled poll thread and a pad that
-            // buzzes until it is unplugged. It stops re-arming as soon as the link closes, which
-            // clears `model` and disarms.
-            if (!usb.writeRaw(0, stopReport(m), OutReportQueue.KEY_RUMBLE)) armBackstop(STOP_RETRY_MS)
+            model?.let { usb.writeRaw(0, stopReport(it)) }
         }
         backstop = r
         mainHandler.postDelayed(r, ms.coerceAtLeast(1))
@@ -314,9 +297,5 @@ class DsCapture(
 
     private companion object {
         const val TAG = "DsCapture"
-
-        /** How soon to retry a rumble stop whose write was rejected. Short: the motors are running
-         *  and the host has already moved on, so nothing else is coming to silence them. */
-        const val STOP_RETRY_MS = 100L
     }
 }

clients/android/kit/src/main/kotlin/io/unom/punktfunk/kit/GamepadFeedback.kt

@@ -88,9 +88,6 @@ class GamepadFeedback(
         const val TAG_PLAYER_LEDS: Byte = 0x02
         const val TAG_TRIGGER: Byte = 0x03
         const val TAG_HID_RAW: Byte = 0x05
-
-        /** Sparse-log cadence for swallowed render failures — see [noteRenderFailure]. */
-        const val LOG_EVERY = 128L
     }
 
     /** One controller's rumble binding — VibratorManager (API 31+) OR the legacy single Vibrator (API 28–30). */
@@ -128,7 +125,6 @@ class GamepadFeedback(
     fun start() {
         running = true
         rumbleThread = Thread({
-            var failures = 0L
             while (running) {
                 val ev = NativeBridge.nativeNextRumble(handle)
                 if (ev < 0L) continue // timeout / closed
@@ -140,50 +136,26 @@ class GamepadFeedback(
                 // the backstop (the hardware net under a stalled poll thread).
                 val pad = ((ev ushr 49) and 0xFL).toInt()
                 val backstopMs = ((ev ushr 32) and 0xFFFF)
-                // Rendering is binder calls into the vibrator service, and every one of them can
-                // throw unchecked — DeadSystemRuntimeException when system_server goes down, and
-                // the ordinary RuntimeException a dying service wraps its RemoteException in.
-                // Unguarded, ONE of those killed this thread outright: `running` stayed true, so
-                // nothing noticed and nothing restarted it, and rumble was gone for the rest of
-                // the session. Losing a single command is recoverable; losing the loop is not.
-                runCatching {
-                    renderRumble(
-                        pad,
-                        ((ev ushr 16) and 0xFFFF).toInt(),
-                        (ev and 0xFFFF).toInt(),
-                        backstopMs,
-                    )
-                }.onFailure { failures = noteRenderFailure("rumble", it, failures) }
+                renderRumble(
+                    pad,
+                    ((ev ushr 16) and 0xFFFF).toInt(),
+                    (ev and 0xFFFF).toInt(),
+                    backstopMs,
+                )
             }
         }, "pf-rumble").apply { isDaemon = true; start() }
 
         hidoutThread = Thread({
             // 128: the raw as-is passthrough events are [pad][kind tag][report kind][≤64 bytes].
             val buf = ByteBuffer.allocateDirect(128)
-            var failures = 0L
             while (running) {
                 val n = NativeBridge.nativeNextHidout(handle, buf)
                 if (n < 0) continue // timeout / closed
-                // Same hazard as the rumble loop above: lights/trigger rendering is binder and USB
-                // calls, and an unchecked throw here would silently end the rich-feedback plane.
-                runCatching { dispatchHidout(buf, n) }
-                    .onFailure { failures = noteRenderFailure("hidout", it, failures) }
+                dispatchHidout(buf, n)
             }
         }, "pf-hidout").apply { isDaemon = true; start() }
     }
 
-    /**
-     * Record a render failure the poll loop swallowed, and return the updated count. Logged on the
-     * first occurrence and sparsely after: a genuinely dead vibrator service fails on *every*
-     * command, which at a rumble plane's rate would bury the log.
-     */
-    private fun noteRenderFailure(plane: String, t: Throwable, seen: Long): Long {
-        if (seen == 0L || seen % LOG_EVERY == 0L) {
-            Log.w(TAG, "$plane render failed (#${seen + 1}) — command dropped, poll loop alive", t)
-        }
-        return seen + 1
-    }
-
     /** Idempotent. Stops + joins the poll threads (must complete before the router is released / handle freed). */
     fun stop() {
         running = false
@@ -297,7 +269,7 @@ class GamepadFeedback(
         val m = bind.vm
         if (m != null) {
             if (lo == 0 && hi == 0) {
-                runCatching { m.cancel() } // (0,0) = stop
+                m.cancel() // (0,0) = stop
                 return
             }
             val combo = CombinedVibration.startParallel()
@@ -322,7 +294,7 @@ class GamepadFeedback(
         // API 28–30 legacy single-motor path: blend both motors into one effect.
         val lv = bind.legacy ?: return
         if (lo == 0 && hi == 0) {
-            runCatching { lv.cancel() } // (0,0) = stop
+            lv.cancel() // (0,0) = stop
             return
         }
         val a = (lo * 0.8 + hi * 0.33).toInt().coerceIn(1, 255)

clients/android/kit/src/main/kotlin/io/unom/punktfunk/kit/HidUsbLink.kt

@@ -14,8 +14,8 @@ import android.hardware.usb.UsbRequest
 import android.os.Build
 import android.util.Log
 import java.nio.ByteBuffer
+import java.util.concurrent.ConcurrentLinkedQueue
 import java.util.concurrent.TimeoutException
-import java.util.concurrent.atomic.AtomicBoolean
 
 /**
  * Generic USB transport for a client-captured HID controller — the device-agnostic half of what
@@ -81,20 +81,14 @@ class HidUsbLink(
 
     /** Pending OUT reports, submitted by the reader thread — only one thread may drive a
      *  connection's [UsbRequest]s ([UsbDeviceConnection.requestWait] returns ANY completed
-     *  request; a second waiter would steal the reader's completions). See [OutReportQueue] for
-     *  what gets discarded when it fills, and why that is not simply "the oldest". */
-    private val outQueue = OutReportQueue()
+     *  request; a second waiter would steal the reader's completions). */
+    private val outQueue = ConcurrentLinkedQueue<ByteArray>()
 
     private var reader: Thread? = null
     private var detachReceiver: BroadcastReceiver? = null
 
     @Volatile private var running = false
 
-    /** Latches on the first "this link is down" signal so [onClosed] fires exactly once, however
-     *  many of the racing detectors (detach broadcast, reader error streak, failed re-queue) see
-     *  it. Reset by [start]. */
-    private val down = AtomicBoolean(false)
-
     /** First attached matching device, or null. Does not need USB permission to enumerate. */
     fun findDevice(): UsbDevice? = usb.deviceList.values.firstOrNull(config.deviceMatch)
 
@@ -120,7 +114,6 @@ class HidUsbLink(
         connection = conn
         device = dev
         claims = claimed
-        down.set(false)
         running = true
         Log.i(
             config.tag,
@@ -141,7 +134,10 @@ class HidUsbLink(
                 val gone: UsbDevice? = intent.getParcelableExtra(UsbManager.EXTRA_DEVICE)
                 if (gone?.deviceName == dev.deviceName) {
                     Log.i(config.tag, "USB detached (${dev.deviceName})")
-                    linkDown()
+                    if (running) {
+                        running = false
+                        onClosed()
+                    }
                 }
             }
         }
@@ -225,9 +221,6 @@ class HidUsbLink(
         if (live.isEmpty()) {
             Log.e(config.tag, "no IN request could be queued")
             finishReader(claims)
-            // `start` already returned true, so without this the owner would sit waiting on a
-            // capture that never streams and never reports itself dead.
-            linkDown()
             return
         }
         val scratch = ByteArray(64)
@@ -302,23 +295,10 @@ class HidUsbLink(
         } finally {
             finishReader(claims)
         }
-        linkDown()
-    }
-
-    /**
-     * Report the link down, exactly once, from whichever detector noticed first — the detach
-     * broadcast (main thread) or the reader thread on its way out.
-     *
-     * This only *signals*; releasing the connection and the interfaces stays the owner's job, via
-     * the [stop] its `onClosed` handler calls. Previously nothing released them on this path: the
-     * detach receiver flipped a flag and fired the callback, so an unplug left the connection open,
-     * the interfaces claimed (the pad could not return to Android's own input stack) and the
-     * receiver still registered — and a re-plug overwrote the field holding it, leaking a receiver
-     * that stayed live for the process's lifetime.
-     */
-    private fun linkDown() {
-        running = false
-        if (down.compareAndSet(false, true)) onClosed()
+        if (running) {
+            running = false
+            onClosed()
+        }
     }
 
     private fun finishReader(claims: List<Claim>) {
@@ -334,35 +314,28 @@ class HidUsbLink(
      * Write one raw report to the device: kind 0 = output report (the active interface's
      * interrupt-OUT, else a `SET_REPORT(Output)` control transfer), kind 1 = feature report
      * (`SET_REPORT(Feature)`). [data] is the full report, id byte first, hidapi framing.
-     *
-     * [coalesce] tells the pending-OUT queue whether a newer report of the same kind may replace
-     * this one — [OutReportQueue.KEY_RUMBLE] for motor levels, the default [OutReportQueue.NO_COALESCE]
-     * for one-shots (lightbar, player LEDs, trigger effects) the sender will not repeat.
-     *
-     * Returns whether the report reached the device or is queued for it. A caller that is writing
-     * a **stop** needs this: a discarded stop has nothing behind it, so it must not be mistaken
-     * for one that landed.
      */
-    fun writeRaw(kind: Int, data: ByteArray, coalesce: Int = OutReportQueue.NO_COALESCE): Boolean {
-        if (data.isEmpty()) return false
-        return when (kind) {
+    fun writeRaw(kind: Int, data: ByteArray) {
+        if (data.isEmpty()) return
+        when (kind) {
             0 -> {
                 if ((activeClaim ?: claims.firstOrNull())?.outReq != null) {
-                    // Interrupt-OUT rides UsbRequests submitted by the reader thread.
-                    outQueue.offer(data, coalesce)
+                    // Interrupt-OUT rides UsbRequests submitted by the reader thread. Bounded,
+                    // newest-wins: these are level-styled commands the sender re-sends anyway.
+                    while (outQueue.size >= 32) outQueue.poll()
+                    outQueue.offer(data)
                 } else {
                     setReport(REPORT_TYPE_OUTPUT, data)
                 }
             }
             1 -> setReport(REPORT_TYPE_FEATURE, data)
-            else -> false
         }
     }
 
-    private fun setReport(type: Int, data: ByteArray): Boolean {
-        val conn = connection ?: return false
-        val ifId = (activeClaim ?: claims.firstOrNull())?.iface?.id ?: return false
-        return sendReport(conn, ifId, type, data)
+    private fun setReport(type: Int, data: ByteArray) {
+        val conn = connection ?: return
+        val ifId = (activeClaim ?: claims.firstOrNull())?.iface?.id ?: return
+        sendReport(conn, ifId, type, data)
     }
 
     /**
@@ -371,8 +344,9 @@ class HidUsbLink(
      * queue would never drain (e.g. a rumble stop before the interfaces release). Safe from any
      * thread: EP0 control transfers are independent of the reader's `requestWait`.
      */
-    fun writeControl(data: ByteArray): Boolean =
-        data.isNotEmpty() && setReport(REPORT_TYPE_OUTPUT, data)
+    fun writeControl(data: ByteArray) {
+        if (data.isNotEmpty()) setReport(REPORT_TYPE_OUTPUT, data)
+    }
 
     private fun sendKeepAlive(conn: UsbDeviceConnection, ifaceId: Int) {
         for (f in config.keepAliveFeatures) sendReport(conn, ifaceId, REPORT_TYPE_FEATURE, f)
@@ -384,48 +358,27 @@ class HidUsbLink(
      * "unnumbered" (id 0 in wValue, id byte stripped from the payload). EP0 is independent of
      * the interrupt endpoints, so this is safe alongside the reader thread's requestWait.
      */
-    private fun sendReport(
-        conn: UsbDeviceConnection,
-        ifaceId: Int,
-        type: Int,
-        data: ByteArray,
-    ): Boolean {
+    private fun sendReport(conn: UsbDeviceConnection, ifaceId: Int, type: Int, data: ByteArray) {
         val id = data[0].toInt() and 0xFF
         val payload = if (id == 0) data.copyOfRange(1, data.size) else data
-        // controlTransfer returns the byte count, or a negative value on failure — a failed write
-        // must be reported as such, not swallowed (a dropped rumble stop has nothing behind it).
-        val n = runCatching {
-            conn.controlTransfer(
-                0x21, // host→device, class, interface
-                0x09, // SET_REPORT
-                (type shl 8) or id,
-                ifaceId,
-                payload,
-                payload.size,
-                WRITE_TIMEOUT_MS,
-            )
-        }.getOrDefault(-1)
-        return n >= 0
+        conn.controlTransfer(
+            0x21, // host→device, class, interface
+            0x09, // SET_REPORT
+            (type shl 8) or id,
+            ifaceId,
+            payload,
+            payload.size,
+            WRITE_TIMEOUT_MS,
+        )
     }
 
-    /**
-     * Stop the read loop and release the interfaces. Idempotent; does not fire [onClosed].
-     *
-     * Safe to call from the `onClosed` handler itself — that is how an unplug now gets cleaned up,
-     * and it arrives on the reader thread, which must not try to join itself.
-     */
+    /** Stop the read loop and release the interfaces. Idempotent; does not fire [onClosed]. */
     fun stop() {
         running = false
-        // Claim the down-latch so the reader's own exit does not report a close the owner asked for.
-        down.set(true)
         detachReceiver?.let { runCatching { context.unregisterReceiver(it) } }
         detachReceiver = null
-        if (reader !== Thread.currentThread()) {
-            runCatching { reader?.join(1000) }
-            // Only forget the thread once it is actually gone: clearing it while it still runs
-            // would let a later stop() skip the join and free the connection under it.
-            reader = null
-        }
+        runCatching { reader?.join(1000) }
+        reader = null
         outQueue.clear()
         activeClaim = null
         for (c in claims) runCatching { connection?.releaseInterface(c.iface) }

clients/android/kit/src/main/kotlin/io/unom/punktfunk/kit/OutReportQueue.kt

@@ -1,89 +0,0 @@
-package io.unom.punktfunk.kit
-
-/**
- * The pending interrupt-OUT reports for a captured controller: a bounded FIFO whose overflow
- * policy knows which reports may be thrown away and which may not.
- *
- * The queue exists because only one thread may drive a connection's `UsbRequest`s, so writes from
- * the feedback threads are handed to the reader thread rather than submitted directly. It has to
- * be bounded — a stalled or unplugged device would otherwise grow it without limit — and the
- * question is what to discard when it fills.
- *
- * The old policy was "newest wins": drop from the head until there is room. That is right for
- * rumble, which is *level-styled* — the host re-sends it continuously, so a dropped frame is
- * replaced milliseconds later and nothing is permanently lost. It is wrong for everything else.
- * A lightbar colour, a player-LED mask and an adaptive-trigger effect are **one-shots**: the host
- * sends them on change and never repeats them. Dropping one leaves the pad wrong until the next
- * time that value happens to change, which may be never.
- *
- * So eviction is driven by an explicit [key] supplied by the caller, not by inspecting the bytes.
- * That distinction cannot be recovered from the report itself: every DualSense output report
- * carries the *same* report id and differs only in its `valid_flag` bytes, so an id-keyed policy
- * would happily let a rumble supersede a lightbar — the very bug this replaces, relocated.
- *
- * Two rules:
- *  - A report offered with a coalescing key **replaces** the pending report with that key, in
- *    place. A burst of rumble collapses to its latest value and never displaces anything else.
- *  - Only when the queue is full does anything get dropped, and then the oldest *coalescable*
- *    report goes first. A one-shot is discarded only if the queue is full of nothing but
- *    one-shots — which needs [cap] distinct one-shots outstanding, far beyond what a real pad
- *    produces.
- *
- * Thread-safe: offered by the feedback threads, drained by the reader thread.
- */
-internal class OutReportQueue(private val cap: Int = CAP) {
-    private class Entry(val key: Int, val data: ByteArray)
-
-    private val items = ArrayDeque<Entry>()
-
-    /**
-     * Queue [data] for submission. [key] is [NO_COALESCE] for a one-shot, or a caller-chosen
-     * constant identifying a level-styled stream whose newer values supersede older ones.
-     *
-     * Returns false only if the report had to be dropped outright — the caller can then treat the
-     * write as failed rather than assuming it is on its way.
-     */
-    fun offer(data: ByteArray, key: Int = NO_COALESCE): Boolean = synchronized(items) {
-        if (key != NO_COALESCE) {
-            val at = items.indexOfFirst { it.key == key }
-            if (at >= 0) {
-                // Supersede in place: keeping the queue position stops a fast rumble stream from
-                // repeatedly jumping the one-shots queued ahead of it.
-                items[at] = Entry(key, data)
-                return true
-            }
-        }
-        if (items.size >= cap) {
-            val victim = items.indexOfFirst { it.key != NO_COALESCE }
-            if (victim >= 0) {
-                items.removeAt(victim)
-            } else if (key != NO_COALESCE) {
-                // Nothing coalescable to sacrifice and this report is itself replaceable — drop it
-                // rather than a one-shot that will never come again.
-                return false
-            } else {
-                items.removeFirst()
-            }
-        }
-        items.addLast(Entry(key, data))
-        return true
-    }
-
-    /** The next report to submit, or null when nothing is pending. */
-    fun poll(): ByteArray? = synchronized(items) { items.removeFirstOrNull()?.data }
-
-    fun clear() = synchronized(items) { items.clear() }
-
-    val size: Int get() = synchronized(items) { items.size }
-
-    companion object {
-        /** This report is a one-shot: never superseded, evicted only as a last resort. */
-        const val NO_COALESCE = 0
-
-        /** Motor levels — re-sent continuously, so only the newest is worth keeping. */
-        const val KEY_RUMBLE = 1
-
-        /** Deep enough to absorb a burst, small enough that a stalled device cannot bloat us. */
-        const val CAP = 32
-    }
-}

clients/android/kit/src/main/kotlin/io/unom/punktfunk/kit/Sc2Capture.kt

@@ -273,20 +273,10 @@ class Sc2Capture(
 
     private fun onLinkClosed() {
         Log.i(TAG, "SC2 link closed (unplug / power-off)")
-        // Both transports share this callback, so read which one was live BEFORE clearing it —
-        // releasing the other would tear down a link that never dropped.
-        val dropped = activeLink
         activeLink = LINK_NONE
         dongleLink = false
         releaseSlot()
         releaseUiKeys()
-        // Release the transport too — see the note in DsCapture.onLinkClosed. The Puck makes this
-        // worse than a single leak: it is the pad that gets power-cycled, so the same process can
-        // round-trip a link many times in one session.
-        when (dropped) {
-            LINK_USB -> usb.stop()
-            LINK_BLE -> ble.stop()
-        }
         onActiveChanged?.invoke(false)
     }
 

clients/android/kit/src/test/kotlin/io/unom/punktfunk/kit/OutReportQueueTest.kt

@@ -1,102 +0,0 @@
-package io.unom.punktfunk.kit
-
-import org.junit.Assert.assertArrayEquals
-import org.junit.Assert.assertEquals
-import org.junit.Assert.assertFalse
-import org.junit.Assert.assertNull
-import org.junit.Assert.assertTrue
-import org.junit.Test
-
-/**
- * The pending-OUT queue's overflow policy. What is being pinned here is the distinction the old
- * "drop from the head until there is room" policy did not make: rumble is re-sent continuously and
- * may be thrown away, while a lightbar/player-LED/trigger report is sent once and never repeated.
- */
-class OutReportQueueTest {
-    /** A report carrying a 0..255 marker so a test can tell which one came back out. */
-    private fun report(marker: Int) = byteArrayOf(0x02, marker.toByte())
-
-    // Masked: the marker rides in a Byte, and Byte.toInt() sign-extends.
-    private fun drain(q: OutReportQueue): List<Int> =
-        generateSequence { q.poll() }.map { it[1].toInt() and 0xFF }.toList()
-
-    @Test
-    fun `rumble supersedes the pending rumble instead of queueing another`() {
-        val q = OutReportQueue()
-        assertTrue(q.offer(report(1), OutReportQueue.KEY_RUMBLE))
-        assertTrue(q.offer(report(2), OutReportQueue.KEY_RUMBLE))
-        assertTrue(q.offer(report(3), OutReportQueue.KEY_RUMBLE))
-        assertEquals("a rumble burst must collapse to one entry", 1, q.size)
-        assertArrayEquals(report(3), q.poll())
-        assertNull(q.poll())
-    }
-
-    @Test
-    fun `superseding keeps the queue position so a rumble stream cannot jump one-shots`() {
-        val q = OutReportQueue()
-        q.offer(report(1), OutReportQueue.KEY_RUMBLE)
-        q.offer(report(10)) // a one-shot queued behind it
-        q.offer(report(2), OutReportQueue.KEY_RUMBLE)
-        // The newer rumble takes the OLD rumble's slot, so the one-shot does not get starved
-        // behind an endlessly-renewed entry.
-        assertEquals(listOf(2, 10), drain(q))
-    }
-
-    @Test
-    fun `a full queue sacrifices rumble, never a one-shot`() {
-        val q = OutReportQueue(cap = 4)
-        q.offer(report(1), OutReportQueue.KEY_RUMBLE)
-        q.offer(report(10))
-        q.offer(report(11))
-        q.offer(report(12))
-        assertEquals(4, q.size)
-        // Full. The old policy dropped the head — here that is a rumble, but only by luck of
-        // ordering; what matters is that the one-shots all survive.
-        assertTrue(q.offer(report(13)))
-        assertEquals(listOf(10, 11, 12, 13), drain(q))
-    }
-
-    @Test
-    fun `the one-shot the host never repeats survives a rumble storm`() {
-        val q = OutReportQueue(cap = 4)
-        // The exact regression: a lightbar colour queued once, then a flood of rumble. Under the
-        // old newest-wins eviction the colour was dropped from the head and never came back,
-        // leaving the pad lit wrong until the value next happened to change.
-        q.offer(report(200)) // lightbar
-        repeat(50) { q.offer(report(it), OutReportQueue.KEY_RUMBLE) }
-        val out = drain(q)
-        assertTrue("the lightbar report must still be queued, got $out", out.contains(200))
-        assertEquals("rumble must not have accumulated", listOf(200, 49), out)
-    }
-
-    @Test
-    fun `a queue full of one-shots refuses a rumble rather than dropping one`() {
-        val q = OutReportQueue(cap = 2)
-        q.offer(report(10))
-        q.offer(report(11))
-        assertFalse(
-            "with nothing coalescable to sacrifice, the replaceable report yields",
-            q.offer(report(1), OutReportQueue.KEY_RUMBLE),
-        )
-        assertEquals(listOf(10, 11), drain(q))
-    }
-
-    @Test
-    fun `only a queue of nothing but one-shots drops one, and it is the oldest`() {
-        val q = OutReportQueue(cap = 2)
-        q.offer(report(10))
-        q.offer(report(11))
-        assertTrue(q.offer(report(12)))
-        assertEquals(listOf(11, 12), drain(q))
-    }
-
-    @Test
-    fun `clear empties the queue`() {
-        val q = OutReportQueue()
-        q.offer(report(1), OutReportQueue.KEY_RUMBLE)
-        q.offer(report(10))
-        q.clear()
-        assertEquals(0, q.size)
-        assertNull(q.poll())
-    }
-}

crates/pf-inject/src/inject/windows/gamepad_raii.rs

@@ -819,46 +819,77 @@ impl DriverAttach {
 
     /// One-shot WARN with everything the host can find out about WHY the driver isn't attached:
     /// driver-store presence, the devnode's PnP status/problem code, and where to look next.
+    ///
+    /// Runs on its own thread and returns immediately. The caller is the session's pad service
+    /// thread — the one feeding input and rumble — and everything below is slow: the driver-store
+    /// check waits up to [`INVENTORY_WAIT`] for a `pnputil` enumeration that can take tens of
+    /// seconds, and the devnode lookup is a synchronous PnP call. Blocking there stalled input for
+    /// up to two seconds *per unattached pad* (the wait is a deadline, not a one-off: while the
+    /// enumeration is still outstanding every pad pays it again), at exactly the moment a session
+    /// is already going wrong. Diagnostics must never be able to hurt the thing they diagnose.
+    ///
+    /// Off the hot path the wait also stops being a compromise — it can afford to be patient and
+    /// report what it actually found rather than "still enumerating".
     fn diagnose(&self) {
-        let store = match driver_store_has(self.inf) {
-            Some(true) => "driver package present in the driver store",
-            Some(false) => {
-                "driver package NOT in the driver store — run: punktfunk-host.exe driver install --gamepad"
-            }
-            None => "driver store could not be queried (pnputil failed or still enumerating)",
-        };
-        let devnode = match &self.instance_id {
-            Some(id) => devnode_status_line(id),
-            None => {
-                "no per-session devnode (SwDeviceCreate failed earlier — see the warning above)"
-                    .to_string()
-            }
-        };
-        tracing::warn!(
-            driver = self.driver,
-            shm = %self.shm_name,
-            grace_secs = ATTACH_GRACE.as_secs(),
-            store,
-            devnode = %devnode,
-            driver_log = self.driver_log,
-            "gamepad driver has not attached to the shared section — the virtual pad exists but no \
-             driver is serving it (games will not see it); an old (pre-sealed-channel) driver also \
-             reads as not-attached: update with punktfunk-host.exe driver install --gamepad \
-             (driver_log is only written by debug driver builds, or with the PFXUSB_DEBUG_LOG / \
-             PFGAMEPAD_DEBUG_LOG / PFMOUSE_DEBUG_LOG system env var set + the device restarted)"
-        );
+        let (driver, inf, driver_log) = (self.driver, self.inf, self.driver_log);
+        let shm_name = self.shm_name.clone();
+        let instance_id = self.instance_id.clone();
+        std::thread::Builder::new()
+            .name("pf-driver-diagnose".into())
+            .spawn(move || diagnose_blocking(driver, inf, driver_log, &shm_name, instance_id))
+            .ok();
     }
 }
 
-/// How long [`driver_store_inventory`] lets the caller wait for the background pnputil query
-/// before reporting without it — [`observe`] runs on the pad service thread, which must keep
-/// draining pad slots even when the driver store is wedged.
-const INVENTORY_WAIT: Duration = Duration::from_secs(2);
+/// The body of [`DriverAttach::diagnose`], on its own thread. Split out rather than inlined into
+/// the closure so the blocking calls stay visible as blocking.
+fn diagnose_blocking(
+    driver: &'static str,
+    inf: &'static str,
+    driver_log: &'static str,
+    shm_name: &str,
+    instance_id: Option<String>,
+) {
+    let store = match driver_store_has(inf) {
+        Some(true) => "driver package present in the driver store",
+        Some(false) => {
+            "driver package NOT in the driver store — run: punktfunk-host.exe driver install --gamepad"
+        }
+        None => "driver store could not be queried (pnputil failed or still enumerating)",
+    };
+    let devnode = match &instance_id {
+        Some(id) => devnode_status_line(id),
+        None => "no per-session devnode (SwDeviceCreate failed earlier — see the warning above)"
+            .to_string(),
+    };
+    tracing::warn!(
+        driver,
+        shm = %shm_name,
+        grace_secs = ATTACH_GRACE.as_secs(),
+        store,
+        devnode = %devnode,
+        driver_log,
+        "gamepad driver has not attached to the shared section — the virtual pad exists but no \
+         driver is serving it (games will not see it); an old (pre-sealed-channel) driver also \
+         reads as not-attached: update with punktfunk-host.exe driver install --gamepad \
+         (driver_log is only written by debug driver builds, or with the PFXUSB_DEBUG_LOG / \
+         PFGAMEPAD_DEBUG_LOG / PFMOUSE_DEBUG_LOG system env var set + the device restarted)"
+    );
+}
+
+/// How long [`driver_store_inventory`] waits for the background pnputil query before reporting
+/// without it. Only [`diagnose_blocking`] waits, and that has a thread to itself, so this is
+/// generous: pnputil routinely takes longer than a couple of seconds on a busy driver store, and
+/// the old two-second budget — chosen to limit the damage while this ran on the pad service thread
+/// — meant the diagnosis usually gave up and printed "still enumerating", which is the one answer
+/// that helps nobody. Nothing waits on this thread, so patience costs only a late log line.
+const INVENTORY_WAIT: Duration = Duration::from_secs(30);
 
 /// Driver-store inventory (`pnputil /enum-drivers`), lower-cased, fetched once per process — only
 /// consulted on the failure path, so the subprocess cost never hits a healthy session. The query
 /// runs on its OWN thread: pnputil can block for tens of seconds on a busy/wedged driver store,
-/// and the caller is the pad service thread. `None` = not available yet (query still running) or
+/// and this keeps one wedged query from being re-run per pad. `None` = not available yet (query
+/// still running past [`INVENTORY_WAIT`]) or
 /// failed; a query that outlives [`INVENTORY_WAIT`] still lands in the cache for later reports.
 fn driver_store_inventory() -> Option<&'static str> {
     static INV: OnceLock<String> = OnceLock::new();

packaging/windows/drivers/pf-gamepad/src/lib.rs

@@ -360,9 +360,32 @@ fn ring_len(view: &pf_umdf_util::section::MappedView) -> u32 {
 /// from being coalesced away by a following LED/trigger report inside one host poll window (the
 /// confirmed stuck-rumble path).
 fn publish_output(view: &pf_umdf_util::section::MappedView, bytes: &[u8]) {
+    // Serialized: the whole publish is a read-modify-write (read the cursor, write the slot it
+    // names, then advance it) and the framework dispatches output callbacks in PARALLEL, so two
+    // can be inside this at once. Unsynchronized, both read the same `ring_head`, both write the
+    // SAME slot — tearing one report's bytes across the other's — and both store head+1, so the
+    // cursor advances once for two reports and the host sees a single torn entry.
+    //
+    // An atomic `fetch_add` on the head does not fix it. That hands each writer a distinct slot,
+    // but it advances the cursor BEFORE the slot bytes exist, so the host can read a slot that is
+    // still being filled — trading a torn slot for a torn slot the host is invited to read. Making
+    // the head-advance mean "the slot below is complete" is exactly what the lock buys.
+    //
+    // Poison-tolerant on purpose. Poison is sticky, so the repo's usual `if let Ok(g) = lock()`
+    // would skip the publish for the REST OF THE PROCESS after a single panic elsewhere — silently
+    // ending game output. Recovering the guard is safe here: the protected state is bytes in a
+    // shared section, not an invariant a panic could have broken.
+    let _publish = RING_PUBLISH
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
     view.write_bytes(OFF_OUTPUT, bytes);
     let seq = view.read_u32(OFF_OUT_SEQ).wrapping_add(1);
-    view.write_u32(OFF_OUT_SEQ, seq);
+    // Release, not a plain write: the host loads `out_seq` with Acquire specifically to order its
+    // copy of the report bytes after it (`dualsense_windows.rs`, "Acquire pairs with the driver's
+    // publish-then-bump store order"). An Acquire load pairs with a Release store and nothing
+    // else, so as a plain write this promised the host an ordering it never actually established —
+    // on a weakly-ordered core (ARM64) the fresh seq could arrive ahead of the bytes it announces.
+    view.store_u32(OFF_OUT_SEQ, seq, Ordering::Release);
     let len = ring_len(view);
     if len != 0 {
         let head = view.read_u32(OFF_RING_HEAD);
@@ -375,6 +398,11 @@ fn publish_output(view: &pf_umdf_util::section::MappedView, bytes: &[u8]) {
     }
 }
 
+/// Serializes [`publish_output`] against itself — see the note there for why an atomic cursor is
+/// not enough. Uncontended in the common case: one output report at a time is the norm, and the
+/// critical section is a few dozen bytes of memcpy into an already-mapped view.
+static RING_PUBLISH: std::sync::Mutex<()> = std::sync::Mutex::new(());
+
 /// The sealed-channel client (per-pad: `ProcessSharingDisabled` gives each pad its own WUDFHost, so
 /// this static is per-pad). The handshake/adoption/validation state machine lives in `pf_umdf_util`.
 static CHANNEL: ChannelClient = ChannelClient::new();

packaging/windows/drivers/pf-xusb/src/lib.rs

@@ -358,20 +358,48 @@ fn read_state(data: Option<&MappedView>) -> (u32, u16, u8, u8, i16, i16, i16, i1
 /// host can tell "driver bound and alive" apart from "driver package missing/failed to bind" and see
 /// the game-visible polling path advance.
 fn touch_driver_marks(data: &MappedView) {
+    let _marks = SECTION_PUBLISH
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
     data.write_u32(OFF_DRIVER_PROTO, GAMEPAD_PROTO_VERSION);
     let hb = data.read_u32(OFF_DRIVER_HEARTBEAT).wrapping_add(1);
     data.write_u32(OFF_DRIVER_HEARTBEAT, hb);
 }
 
 /// Publish a game's rumble (from SET_STATE) into the DATA section for the host to forward.
+///
+/// Serialized and Release-published, because IOCTLs arrive concurrently and neither property held
+/// before. `seq` was a read-modify-write across the two motor bytes: two `SET_STATE` calls could
+/// both read the same value and both write back `seq + 1`, so the host — which treats an unchanged
+/// seq as "nothing new" — saw one bump for two writes and skipped a level entirely. A skipped
+/// **stop** is the one that hurts: the pad keeps buzzing until the host's ~2.5 s idle force-off
+/// notices the game went quiet, which is where the bound on this bug comes from.
+///
+/// The seq store is Release for the same reason as `pf-gamepad`'s `out_seq`: the host loads it with
+/// Acquire and documents that as ordering its read of the motor bytes ("the driver bumps
+/// `rumble_seq` AFTER writing the rumble bytes", `gamepad_windows.rs`). A plain write gives that
+/// Acquire nothing to pair with, so the guarantee the host's comment claims did not exist in either
+/// direction — the host could read a fresh seq against stale motor levels on a weakly-ordered core.
 fn publish_rumble(data: Option<&MappedView>, large: u8, small: u8) {
     let Some(v) = data else { return };
+    let _publish = SECTION_PUBLISH
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
     v.write_u8(OFF_RUMBLE_LARGE, large);
     v.write_u8(OFF_RUMBLE_SMALL, small);
     let seq = v.read_u32(OFF_RUMBLE_SEQ).wrapping_add(1);
-    v.write_u32(OFF_RUMBLE_SEQ, seq);
+    v.store_u32(OFF_RUMBLE_SEQ, seq, Ordering::Release);
 }
 
+/// Serializes the section's read-modify-write publishes ([`publish_rumble`], [`touch_driver_marks`])
+/// against each other. One lock rather than one per field: they are all short byte writes into the
+/// same mapped view, and the contention is nil compared to the IOCTL round trip that reaches them.
+///
+/// Poison-tolerant deliberately — poison is sticky, so bailing out on it would silently stop
+/// forwarding rumble for the rest of the process. The protected state is bytes in a shared section,
+/// not an invariant a panic elsewhere could have violated.
+static SECTION_PUBLISH: std::sync::Mutex<()> = std::sync::Mutex::new(());
+
 // Build the 29-byte GET_STATE buffer (the layout xinput1_4 parses).
 fn build_get_state(data: Option<&MappedView>) -> [u8; 29] {
     let (packet, buttons, lt, rt, lx, ly, rx, ry) = read_state(data);