Compare commits

...
Author SHA1 Message Date
enricobuehler 66a28d5abb ci(android): run the kit's unit tests
ci / web (pull_request) Successful in 1m3s
ci / rust-arm64 (pull_request) Successful in 2m24s
ci / docs-site (pull_request) Successful in 1m42s
android / android (pull_request) Successful in 3m17s
ci / rust (pull_request) Failing after 13m58s
They were running nowhere. This workflow only assembled, and the screenshot
workflow runs the :app module's tests, so nothing enforced :kit's — the pure
parsers, migrations and feedback policies could go red without anyone
noticing. A couple of seconds against a module the build already produces.
2026-08-04 18:14:13 +02:00
enricobuehler e2faecfd42 fix(client/android): rumble survives a vibrator fault, and an unplug stops leaking
Four faults in the Android feedback path, all of them silent.

Rumble stopped for the rest of the session if one vibrator call threw. The
poll thread called cancel() unguarded while every call around it was already
wrapped, so an unchecked throw — DeadSystemRuntimeException, or the
RuntimeException a dying service wraps a RemoteException in — unwound the
thread. `running` stayed true, so nothing noticed it was gone and nothing
restarted it. Guarding the two bare cancels is not enough on its own: the
binder calls that bind a vibrator can throw just the same, so the loop itself
now survives a failed render, and the same guard covers the hidout thread.

A rumble stop that was never written was treated as one that landed. The
DualSense capture disarmed its backstop timer *before* the write, on a queue
that discarded failed submits without saying so, so a dropped stop left the
motors running with nothing scheduled to try again — and a USB pad holds its
last level until told zero. Writes now report whether they were accepted, the
backstop is disarmed only once the stop is actually on its way, and the
backstop re-arms rather than giving up if its own write is refused.

A full write queue dropped lightbar colours, player-LED masks and trigger
effects. Its overflow rule was "drop the oldest", which is right for rumble —
re-sent continuously, so a lost frame returns milliseconds later — and wrong
for everything else, which the host sends once on change and never repeats.
Eviction is now driven by an explicit key from the caller rather than by
inspecting the bytes: rumble supersedes the pending rumble in place, and a
one-shot is discarded only if the queue holds nothing but one-shots. The key
cannot be recovered from the report itself, which is why this is not keyed by
report id — every DualSense output report carries the *same* id and differs
only in its valid_flag bytes, so an id-keyed rule would let a rumble
supersede a lightbar, which is this bug again by another route.

An unplug leaked the USB connection and the detach receiver. The link only
signalled the drop; neither capture released anything, so the interfaces
stayed claimed (the pad could not return to Android's own input stack) and a
re-plug overwrote the field holding the receiver, stranding one live for the
rest of the process. The captures now release the transport, stop() is safe
to call from the callback it arrives on — the reader thread must not join
itself — and a close is reported exactly once however many detectors see it.
A reader that could not queue a single request now reports itself down too,
instead of leaving the owner waiting on a capture that never streams.
2026-08-04 18:14:07 +02:00
7 changed files with 361 additions and 56 deletions
+8
View File
@@ -160,6 +160,14 @@ jobs:
key: gradle-${{ hashFiles('clients/android/**/*.gradle.kts', 'clients/android/gradle/wrapper/gradle-wrapper.properties') }} key: gradle-${{ hashFiles('clients/android/**/*.gradle.kts', 'clients/android/gradle/wrapper/gradle-wrapper.properties') }}
restore-keys: gradle- 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) - name: assembleDebug (cargo-ndk → jniLibs → APK)
working-directory: clients/android working-directory: clients/android
env: env:
@@ -116,7 +116,9 @@ class DsCapture(
// The interfaces are about to release with the kernel driver still detached — a // 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. // 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). // EP0-direct (the reader thread is stopping; the queue would never drain).
usb.writeControl(stopReport(m)) // 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")
} }
disarmBackstop() disarmBackstop()
usb.stop() usb.stop()
@@ -145,6 +147,9 @@ class DsCapture(
val wasActive = model != null val wasActive = model != null
model = null model = null
releaseSlot() 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) if (wasActive) onActiveChanged?.invoke(false)
} }
@@ -216,17 +221,20 @@ class DsCapture(
override fun rumble(pad: Int, low: Int, high: Int, backstopMs: Long) { override fun rumble(pad: Int, low: Int, high: Int, backstopMs: Long) {
val m = model ?: return val m = model ?: return
if (low == 0 && high == 0) { val stop = low == 0 && high == 0
disarmBackstop() if (!stop) armBackstop(backstopMs)
} else { val sent = if (m == DsDevice.Model.DUALSHOCK4) {
armBackstop(backstopMs)
}
if (m == DsDevice.Model.DUALSHOCK4) {
ds4Low = low ds4Low = low
ds4High = high ds4High = high
writeDs4() writeDs4()
} else { } else {
usb.writeRaw(0, DsDevice.ds5RumbleReport(m, low, high)) 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)
} }
} }
@@ -252,6 +260,9 @@ class DsCapture(
usb.writeRaw(0, DsDevice.ds5TriggerReport(m, which, effect)) 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( private fun writeDs4() = usb.writeRaw(
0, 0,
DsDevice.ds4Report( DsDevice.ds4Report(
@@ -261,6 +272,7 @@ class DsCapture(
(ds4Rgb shr 8) and 0xFF, (ds4Rgb shr 8) and 0xFF,
ds4Rgb 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 /** The report that stops the motors. The DS4's is a full-state write, so it zeroes the
@@ -284,7 +296,12 @@ class DsCapture(
backstop?.let { mainHandler.removeCallbacks(it) } backstop?.let { mainHandler.removeCallbacks(it) }
val r = Runnable { val r = Runnable {
backstop = null backstop = null
model?.let { usb.writeRaw(0, stopReport(it)) } 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)
} }
backstop = r backstop = r
mainHandler.postDelayed(r, ms.coerceAtLeast(1)) mainHandler.postDelayed(r, ms.coerceAtLeast(1))
@@ -297,5 +314,9 @@ class DsCapture(
private companion object { private companion object {
const val TAG = "DsCapture" 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
} }
} }
@@ -88,6 +88,9 @@ class GamepadFeedback(
const val TAG_PLAYER_LEDS: Byte = 0x02 const val TAG_PLAYER_LEDS: Byte = 0x02
const val TAG_TRIGGER: Byte = 0x03 const val TAG_TRIGGER: Byte = 0x03
const val TAG_HID_RAW: Byte = 0x05 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 2830). */ /** One controller's rumble binding — VibratorManager (API 31+) OR the legacy single Vibrator (API 2830). */
@@ -125,6 +128,7 @@ class GamepadFeedback(
fun start() { fun start() {
running = true running = true
rumbleThread = Thread({ rumbleThread = Thread({
var failures = 0L
while (running) { while (running) {
val ev = NativeBridge.nativeNextRumble(handle) val ev = NativeBridge.nativeNextRumble(handle)
if (ev < 0L) continue // timeout / closed if (ev < 0L) continue // timeout / closed
@@ -136,26 +140,50 @@ class GamepadFeedback(
// the backstop (the hardware net under a stalled poll thread). // the backstop (the hardware net under a stalled poll thread).
val pad = ((ev ushr 49) and 0xFL).toInt() val pad = ((ev ushr 49) and 0xFL).toInt()
val backstopMs = ((ev ushr 32) and 0xFFFF) val backstopMs = ((ev ushr 32) and 0xFFFF)
renderRumble( // Rendering is binder calls into the vibrator service, and every one of them can
pad, // throw unchecked — DeadSystemRuntimeException when system_server goes down, and
((ev ushr 16) and 0xFFFF).toInt(), // the ordinary RuntimeException a dying service wraps its RemoteException in.
(ev and 0xFFFF).toInt(), // Unguarded, ONE of those killed this thread outright: `running` stayed true, so
backstopMs, // 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) }
} }
}, "pf-rumble").apply { isDaemon = true; start() } }, "pf-rumble").apply { isDaemon = true; start() }
hidoutThread = Thread({ hidoutThread = Thread({
// 128: the raw as-is passthrough events are [pad][kind tag][report kind][≤64 bytes]. // 128: the raw as-is passthrough events are [pad][kind tag][report kind][≤64 bytes].
val buf = ByteBuffer.allocateDirect(128) val buf = ByteBuffer.allocateDirect(128)
var failures = 0L
while (running) { while (running) {
val n = NativeBridge.nativeNextHidout(handle, buf) val n = NativeBridge.nativeNextHidout(handle, buf)
if (n < 0) continue // timeout / closed if (n < 0) continue // timeout / closed
dispatchHidout(buf, n) // 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) }
} }
}, "pf-hidout").apply { isDaemon = true; start() } }, "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). */ /** Idempotent. Stops + joins the poll threads (must complete before the router is released / handle freed). */
fun stop() { fun stop() {
running = false running = false
@@ -269,7 +297,7 @@ class GamepadFeedback(
val m = bind.vm val m = bind.vm
if (m != null) { if (m != null) {
if (lo == 0 && hi == 0) { if (lo == 0 && hi == 0) {
m.cancel() // (0,0) = stop runCatching { m.cancel() } // (0,0) = stop
return return
} }
val combo = CombinedVibration.startParallel() val combo = CombinedVibration.startParallel()
@@ -294,7 +322,7 @@ class GamepadFeedback(
// API 2830 legacy single-motor path: blend both motors into one effect. // API 2830 legacy single-motor path: blend both motors into one effect.
val lv = bind.legacy ?: return val lv = bind.legacy ?: return
if (lo == 0 && hi == 0) { if (lo == 0 && hi == 0) {
lv.cancel() // (0,0) = stop runCatching { lv.cancel() } // (0,0) = stop
return return
} }
val a = (lo * 0.8 + hi * 0.33).toInt().coerceIn(1, 255) val a = (lo * 0.8 + hi * 0.33).toInt().coerceIn(1, 255)
@@ -14,8 +14,8 @@ import android.hardware.usb.UsbRequest
import android.os.Build import android.os.Build
import android.util.Log import android.util.Log
import java.nio.ByteBuffer import java.nio.ByteBuffer
import java.util.concurrent.ConcurrentLinkedQueue
import java.util.concurrent.TimeoutException 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 * Generic USB transport for a client-captured HID controller — the device-agnostic half of what
@@ -81,14 +81,20 @@ class HidUsbLink(
/** Pending OUT reports, submitted by the reader thread — only one thread may drive a /** Pending OUT reports, submitted by the reader thread — only one thread may drive a
* connection's [UsbRequest]s ([UsbDeviceConnection.requestWait] returns ANY completed * connection's [UsbRequest]s ([UsbDeviceConnection.requestWait] returns ANY completed
* request; a second waiter would steal the reader's completions). */ * request; a second waiter would steal the reader's completions). See [OutReportQueue] for
private val outQueue = ConcurrentLinkedQueue<ByteArray>() * what gets discarded when it fills, and why that is not simply "the oldest". */
private val outQueue = OutReportQueue()
private var reader: Thread? = null private var reader: Thread? = null
private var detachReceiver: BroadcastReceiver? = null private var detachReceiver: BroadcastReceiver? = null
@Volatile private var running = false @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. */ /** First attached matching device, or null. Does not need USB permission to enumerate. */
fun findDevice(): UsbDevice? = usb.deviceList.values.firstOrNull(config.deviceMatch) fun findDevice(): UsbDevice? = usb.deviceList.values.firstOrNull(config.deviceMatch)
@@ -114,6 +120,7 @@ class HidUsbLink(
connection = conn connection = conn
device = dev device = dev
claims = claimed claims = claimed
down.set(false)
running = true running = true
Log.i( Log.i(
config.tag, config.tag,
@@ -134,10 +141,7 @@ class HidUsbLink(
val gone: UsbDevice? = intent.getParcelableExtra(UsbManager.EXTRA_DEVICE) val gone: UsbDevice? = intent.getParcelableExtra(UsbManager.EXTRA_DEVICE)
if (gone?.deviceName == dev.deviceName) { if (gone?.deviceName == dev.deviceName) {
Log.i(config.tag, "USB detached (${dev.deviceName})") Log.i(config.tag, "USB detached (${dev.deviceName})")
if (running) { linkDown()
running = false
onClosed()
}
} }
} }
} }
@@ -221,6 +225,9 @@ class HidUsbLink(
if (live.isEmpty()) { if (live.isEmpty()) {
Log.e(config.tag, "no IN request could be queued") Log.e(config.tag, "no IN request could be queued")
finishReader(claims) 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 return
} }
val scratch = ByteArray(64) val scratch = ByteArray(64)
@@ -295,10 +302,23 @@ class HidUsbLink(
} finally { } finally {
finishReader(claims) finishReader(claims)
} }
if (running) { linkDown()
running = false }
onClosed()
} /**
* 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()
} }
private fun finishReader(claims: List<Claim>) { private fun finishReader(claims: List<Claim>) {
@@ -314,28 +334,35 @@ class HidUsbLink(
* Write one raw report to the device: kind 0 = output report (the active interface's * 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 * 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. * (`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) { fun writeRaw(kind: Int, data: ByteArray, coalesce: Int = OutReportQueue.NO_COALESCE): Boolean {
if (data.isEmpty()) return if (data.isEmpty()) return false
when (kind) { return when (kind) {
0 -> { 0 -> {
if ((activeClaim ?: claims.firstOrNull())?.outReq != null) { if ((activeClaim ?: claims.firstOrNull())?.outReq != null) {
// Interrupt-OUT rides UsbRequests submitted by the reader thread. Bounded, // Interrupt-OUT rides UsbRequests submitted by the reader thread.
// newest-wins: these are level-styled commands the sender re-sends anyway. outQueue.offer(data, coalesce)
while (outQueue.size >= 32) outQueue.poll()
outQueue.offer(data)
} else { } else {
setReport(REPORT_TYPE_OUTPUT, data) setReport(REPORT_TYPE_OUTPUT, data)
} }
} }
1 -> setReport(REPORT_TYPE_FEATURE, data) 1 -> setReport(REPORT_TYPE_FEATURE, data)
else -> false
} }
} }
private fun setReport(type: Int, data: ByteArray) { private fun setReport(type: Int, data: ByteArray): Boolean {
val conn = connection ?: return val conn = connection ?: return false
val ifId = (activeClaim ?: claims.firstOrNull())?.iface?.id ?: return val ifId = (activeClaim ?: claims.firstOrNull())?.iface?.id ?: return false
sendReport(conn, ifId, type, data) return sendReport(conn, ifId, type, data)
} }
/** /**
@@ -344,9 +371,8 @@ class HidUsbLink(
* queue would never drain (e.g. a rumble stop before the interfaces release). Safe from any * 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`. * thread: EP0 control transfers are independent of the reader's `requestWait`.
*/ */
fun writeControl(data: ByteArray) { fun writeControl(data: ByteArray): Boolean =
if (data.isNotEmpty()) setReport(REPORT_TYPE_OUTPUT, data) data.isNotEmpty() && setReport(REPORT_TYPE_OUTPUT, data)
}
private fun sendKeepAlive(conn: UsbDeviceConnection, ifaceId: Int) { private fun sendKeepAlive(conn: UsbDeviceConnection, ifaceId: Int) {
for (f in config.keepAliveFeatures) sendReport(conn, ifaceId, REPORT_TYPE_FEATURE, f) for (f in config.keepAliveFeatures) sendReport(conn, ifaceId, REPORT_TYPE_FEATURE, f)
@@ -358,27 +384,48 @@ class HidUsbLink(
* "unnumbered" (id 0 in wValue, id byte stripped from the payload). EP0 is independent of * "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. * the interrupt endpoints, so this is safe alongside the reader thread's requestWait.
*/ */
private fun sendReport(conn: UsbDeviceConnection, ifaceId: Int, type: Int, data: ByteArray) { private fun sendReport(
conn: UsbDeviceConnection,
ifaceId: Int,
type: Int,
data: ByteArray,
): Boolean {
val id = data[0].toInt() and 0xFF val id = data[0].toInt() and 0xFF
val payload = if (id == 0) data.copyOfRange(1, data.size) else data val payload = if (id == 0) data.copyOfRange(1, data.size) else data
conn.controlTransfer( // controlTransfer returns the byte count, or a negative value on failure — a failed write
0x21, // host→device, class, interface // must be reported as such, not swallowed (a dropped rumble stop has nothing behind it).
0x09, // SET_REPORT val n = runCatching {
(type shl 8) or id, conn.controlTransfer(
ifaceId, 0x21, // host→device, class, interface
payload, 0x09, // SET_REPORT
payload.size, (type shl 8) or id,
WRITE_TIMEOUT_MS, ifaceId,
) payload,
payload.size,
WRITE_TIMEOUT_MS,
)
}.getOrDefault(-1)
return n >= 0
} }
/** Stop the read loop and release the interfaces. Idempotent; does not fire [onClosed]. */ /**
* 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.
*/
fun stop() { fun stop() {
running = false 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?.let { runCatching { context.unregisterReceiver(it) } }
detachReceiver = null detachReceiver = null
runCatching { reader?.join(1000) } if (reader !== Thread.currentThread()) {
reader = null 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
}
outQueue.clear() outQueue.clear()
activeClaim = null activeClaim = null
for (c in claims) runCatching { connection?.releaseInterface(c.iface) } for (c in claims) runCatching { connection?.releaseInterface(c.iface) }
@@ -0,0 +1,89 @@
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
}
}
@@ -273,10 +273,20 @@ class Sc2Capture(
private fun onLinkClosed() { private fun onLinkClosed() {
Log.i(TAG, "SC2 link closed (unplug / power-off)") 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 activeLink = LINK_NONE
dongleLink = false dongleLink = false
releaseSlot() releaseSlot()
releaseUiKeys() 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) onActiveChanged?.invoke(false)
} }
@@ -0,0 +1,102 @@
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())
}
}