Compare commits

..
Author SHA1 Message Date
enricobuehler 4bc7eecf05 feat(host/wire): MTU resilience for the video data plane
ci / docs-site (pull_request) Successful in 1m5s
ci / web (pull_request) Successful in 1m47s
apple / swift (pull_request) Successful in 1m22s
apple / screenshots (pull_request) Skipped
ci / rust-arm64 (pull_request) Successful in 2m46s
android / android (pull_request) Successful in 3m25s
windows / build (x86_64-pc-windows-msvc) (pull_request) Successful in 2m13s
windows / build (aarch64-pc-windows-msvc) (pull_request) Successful in 3m16s
ci / rust (pull_request) Successful in 22m23s
Video datagrams are sealed at a shard payload sized for a clean 1500-byte
MTU (1472-byte UDP payloads). A host whose route to the client crosses a
smaller-MTU hop (a VPN/overlay adapter claiming the LAN route, a lowered
NIC MTU) delivers every small flow — QUIC control, hole punch, input,
audio — while 100% of video datagrams die: the client sits on a black
screen reporting zero loss and the host streams into the void with every
gauge green. Field-reported as 'connects fine, black screen forever'.

Three legs, none of which changes a session on a healthy path:

- PUNKTFUNK_WIRE_MTU operator override: shard payload derived from a
  given on-wire IP MTU. Wire-compatible — Welcome::shard_payload is
  already negotiated per session (the v4/v6 split ships two values
  today) and every client follows the negotiated value.
- Detection: the QUIC MTU-discovery probe ceiling moves from quinn's
  stock 1452 to exactly the sealed video-datagram size (1472), so a
  control connection's settled MTU becomes a verdict on the path:
  settled at the ceiling proves it carries video, settled below proves
  it cannot. A per-session watcher samples after the search has settled
  (live-connection guard against mid-search false learns) and logs an
  actionable WARN naming the failure shape and the diagnosis commands.
- Healing: the measured budget is recorded per peer IP; the next
  handshake clamps shard_payload to fit, so a reconnect self-heals. A
  later session that reaches the ceiling erases the record.

Verified: core 286/286 --features quic + clippy -D warnings (macOS);
host clippy -D warnings + native:: tests 44/44 (pf-lxcheck container).
The regenerated C header picks up the new MIN_SHARD_PAYLOAD constant.
2026-08-04 18:30:33 +02:00
13 changed files with 394 additions and 365 deletions
-8
View File
@@ -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:
@@ -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
}
}
@@ -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 2830). */
@@ -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 2830 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)
@@ -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) }
@@ -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
}
}
@@ -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)
}
@@ -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())
}
}
+112
View File
@@ -341,6 +341,50 @@ pub fn mtu1500_shard_payload_for(peer: core::net::IpAddr) -> usize {
}
}
/// Floor for a negotiated `shard_payload` (even, well under every real path). A path whose UDP
/// budget lands below this can't carry the QUIC control plane either (QUIC's own minimum is a
/// 1200-byte UDP payload), so shrinking video shards further buys nothing — the clamp helpers
/// bottom out here instead of producing degenerate confetti-sized shards.
pub const MIN_SHARD_PAYLOAD: usize = 512;
/// The sealed wire size of a video datagram carrying `shard_payload` bytes of shard — what
/// actually leaves the socket as UDP payload (punktfunk header + shard + crypto overhead).
pub const fn sealed_datagram_bytes(shard_payload: usize) -> usize {
HEADER_LEN + shard_payload + CRYPTO_OVERHEAD
}
/// The UDP-payload size a path must carry for full-size IPv4 video datagrams: the sealed size
/// of the [`mtu1500_shard_payload`] default (= 1472, the exact 1500-MTU IPv4 ceiling). Doubles
/// as the QUIC MTU-discovery probe ceiling (`quic/endpoint.rs`): with the ceiling set to
/// exactly this value, a control connection whose discovery settles AT the ceiling has proven
/// the path carries full-size video datagrams, and one that settles BELOW it has proven the
/// path cannot — a discrimination quinn's stock 1452 ceiling can't make in either direction.
pub const fn video_datagram_udp_ceiling() -> usize {
sealed_datagram_bytes(mtu1500_shard_payload())
}
/// Largest even shard payload whose sealed datagram fits in `udp_budget` bytes of UDP payload
/// (the quantity QUIC MTU discovery measures — [`video_datagram_udp_ceiling`] is its probe
/// ceiling). Clamped to the peer's family default ([`mtu1500_shard_payload_for`]) so a generous
/// budget never grows packets past today's wire, and floored at [`MIN_SHARD_PAYLOAD`].
pub fn shard_payload_for_udp_budget(udp_budget: usize, peer: core::net::IpAddr) -> usize {
let p = udp_budget.saturating_sub(HEADER_LEN + CRYPTO_OVERHEAD);
let p = p - p % 2; // FEC requires even shards
p.clamp(MIN_SHARD_PAYLOAD, mtu1500_shard_payload_for(peer))
}
/// [`shard_payload_for_udp_budget`] for an operator-supplied ON-WIRE IP MTU (the number
/// `netsh interface ipv4 show subinterfaces` / `ip link` shows): subtracts the family's IP+UDP
/// headers first — 28 for IPv4 (and IPv4-mapped), 48 for IPv6.
pub fn shard_payload_for_wire_mtu(wire_mtu: usize, peer: core::net::IpAddr) -> usize {
let ip_udp = match peer {
core::net::IpAddr::V4(_) => 28,
core::net::IpAddr::V6(v6) if v6.to_ipv4_mapped().is_some() => 28,
core::net::IpAddr::V6(_) => 48,
};
shard_payload_for_udp_budget(wire_mtu.saturating_sub(ip_udp), peer)
}
/// Everything needed to construct a [`Session`](crate::session::Session).
///
/// `Debug` is implemented by hand to redact `key`/`salt`, and `key`/`salt` are zeroized
@@ -514,6 +558,74 @@ mod tests {
assert!(HEADER_LEN + (p + 2) + CRYPTO_OVERHEAD > 1452, "not maximal");
}
/// The video-datagram ceiling IS the exact v4 sealed size — the QUIC MTU-discovery probe
/// ceiling (endpoint.rs) relies on this equality for its settled-at-vs-below verdict.
#[test]
fn video_datagram_ceiling_is_the_sealed_default() {
assert_eq!(
video_datagram_udp_ceiling(),
HEADER_LEN + mtu1500_shard_payload() + CRYPTO_OVERHEAD
);
assert_eq!(video_datagram_udp_ceiling(), 1472);
}
/// Budget-derived sizing: even, sealed-fits-the-budget, clamped to the family default
/// above and [`MIN_SHARD_PAYLOAD`] below.
#[test]
fn shard_payload_for_udp_budget_math() {
use core::net::IpAddr;
let v4: IpAddr = "192.168.1.50".parse().unwrap();
let v6: IpAddr = "fd00::50".parse().unwrap();
// The full ceiling reproduces the default exactly.
assert_eq!(
shard_payload_for_udp_budget(video_datagram_udp_ceiling(), v4),
mtu1500_shard_payload()
);
// A WARP/Tailscale-shaped 1280 budget: sealed result must fit the budget, stay even.
let p = shard_payload_for_udp_budget(1280, v4);
assert_eq!(p % 2, 0);
assert!(sealed_datagram_bytes(p) <= 1280);
assert!(sealed_datagram_bytes(p + 2) > 1280, "not maximal");
// Odd budgets round down to even shards.
assert_eq!(shard_payload_for_udp_budget(1281, v4) % 2, 0);
// A generous budget never grows past the family default (either family).
assert_eq!(
shard_payload_for_udp_budget(9000, v4),
mtu1500_shard_payload()
);
assert_eq!(
shard_payload_for_udp_budget(9000, v6),
mtu1500_shard_payload_v6()
);
// Degenerate budgets bottom out at the floor instead of confetti.
assert_eq!(shard_payload_for_udp_budget(100, v4), MIN_SHARD_PAYLOAD);
}
/// Operator-facing wire-MTU sizing subtracts the right IP+UDP header per family, and 1500
/// reproduces today's defaults exactly.
#[test]
fn shard_payload_for_wire_mtu_math() {
use core::net::IpAddr;
let v4: IpAddr = "192.168.1.50".parse().unwrap();
let v6: IpAddr = "fd00::50".parse().unwrap();
let mapped: IpAddr = "::ffff:192.168.1.50".parse().unwrap();
assert_eq!(
shard_payload_for_wire_mtu(1500, v4),
mtu1500_shard_payload()
);
assert_eq!(
shard_payload_for_wire_mtu(1500, mapped),
mtu1500_shard_payload()
);
assert_eq!(
shard_payload_for_wire_mtu(1500, v6),
mtu1500_shard_payload_v6()
);
// 1280 wire 28 64 = 1188 (v4); 48 64 = 1168 (v6).
assert_eq!(shard_payload_for_wire_mtu(1280, v4), 1188);
assert_eq!(shard_payload_for_wire_mtu(1280, v6), 1168);
}
/// Family selection: genuine v6 remotes get the v6 size; v4 — including the IPv4-mapped v6
/// form a dual-stack `[::]` socket reports for a v4 client — keeps the v4 size.
#[test]
@@ -47,6 +47,20 @@ fn stream_transport_idle(idle: std::time::Duration) -> Arc<quinn::TransportConfi
// plane latest-wins at the source — ~200 ms of stereo Opus (proportionally less at
// surround bitrates), so sustained congestion costs concealable drops, never lag.
t.datagram_send_buffer_size(4 * 1024);
// MTU discovery probes up to EXACTLY the sealed size of a full IPv4 video datagram (1472)
// instead of quinn's stock 1452. Two reasons: (a) on a clean 1500-MTU path QUIC gets the
// last 20 bytes per packet; (b) the ceiling turns discovery into a video-path verdict the
// host's wire-MTU watcher reads (`punktfunk-host` `native/wire_mtu.rs`) — settled == ceiling
// proves the path carries full-size video datagrams, settled BELOW it proves it cannot (a
// VPN/overlay adapter at MTU ~1280 blackholes every video packet while all the small flows
// pass: the "connects fine, black screen forever" field shape). With the stock 1452 ceiling
// a healthy path and a constrained one are indistinguishable at the top. This is the ONLY
// behavioral change on healthy paths, and it's confined to discovery: probes are padded
// PINGs quinn already expects to lose above a constrained hop — a lost probe settles the
// search lower, exactly as it did before.
let mut mtud = quinn::MtuDiscoveryConfig::default();
mtud.upper_bound(crate::config::video_datagram_udp_ceiling() as u16);
t.mtu_discovery_config(Some(mtud));
Arc::new(t)
}
+4 -3
View File
@@ -26,9 +26,7 @@
#![deny(clippy::undocumented_unsafe_blocks)]
use anyhow::{anyhow, Context, Result};
use punktfunk_core::config::{
mtu1500_shard_payload_for, CompositorPref, FecConfig, FecScheme, GamepadPref, Role,
};
use punktfunk_core::config::{CompositorPref, FecConfig, FecScheme, GamepadPref, Role};
use punktfunk_core::input::{InputEvent, InputKind};
use punktfunk_core::packet::{FLAG_PIC, FLAG_PROBE, FLAG_SOF};
use punktfunk_core::quic::{
@@ -72,6 +70,9 @@ use input::{input_thread, ClientInput};
/// The Hello→Welcome→Start negotiation (plan §W1); `serve_session` calls `handshake::negotiate`
/// after the pairing gate.
mod handshake;
/// MTU resilience for the video data plane: `PUNKTFUNK_WIRE_MTU` override, the per-session
/// path-MTU watch on the control connection, and the per-peer learned shard-payload clamp.
mod wire_mtu;
/// The mid-stream control task (plan §W1); `serve_session` spawns `control::run` after the
/// handshake to multiplex renegotiation / speed-test control messages onto the data-plane channels.
+10 -1
View File
@@ -491,7 +491,12 @@ pub(super) async fn negotiate(
// per-datagram loss on Wi-Fi — the "100 Mbps badly fails on the phone" root cause.
// Negotiated, so the client follows. Jumbo (≈8900) is a future negotiated bump (needs
// MAX_DATAGRAM_BYTES raised + end-to-end 9000 MTU).
shard_payload: mtu1500_shard_payload_for(peer.ip()) as u16,
// Resolution order (wire_mtu.rs): `PUNKTFUNK_WIRE_MTU` operator override, then a path
// budget learned from a prior session whose QUIC MTU discovery settled below the
// video-datagram ceiling (the "VPN on the host blackholes every video packet" field
// shape — small flows pass, the stream is an endless black screen), then this family
// default. Healthy paths take the default branch and are byte-identical to before.
shard_payload: wire_mtu::negotiated_shard_payload(peer.ip()) as u16,
encrypt: true,
key,
salt,
@@ -658,6 +663,10 @@ pub(super) async fn negotiate(
let start =
Start::decode(&io::read_msg(recv).await?).map_err(|e| anyhow!("Start decode: {e:?}"))?;
bringup.mark("start");
// The session is real: watch this connection's MTU discovery settle and turn it into a
// path verdict (WARN + learned clamp for the next session on a constrained path; clears a
// stale clamp on a healthy one). Bounded ~10 s task, ends by itself.
wire_mtu::spawn_watch(conn.clone(), welcome.shard_payload as usize);
Ok::<_, anyhow::Error>((
hello,
welcome,
@@ -0,0 +1,192 @@
//! MTU resilience for the video data plane (the "connects fine, black screen forever" field
//! shape).
//!
//! Video datagrams are sealed at a per-session `shard_payload` sized for a clean 1500-byte MTU
//! (1472-byte UDP payloads). A host whose route to the client runs through a smaller-MTU hop —
//! a VPN/overlay adapter (Tailscale/WARP/ZeroTier default to 1280) claiming the LAN route, or a
//! lowered NIC MTU — delivers every SMALL flow (QUIC control, hole punch, input, audio) while
//! 100 % of video datagrams die by fragmentation or local `WSAEMSGSIZE`: the client sits on a
//! black screen reporting `loss_ppm=0` (it can't see gaps in packets it never saw any of) and
//! the host streams into the void with every gauge green. Neither side observes the failure
//! directly — but the control connection CAN: its MTU discovery probes up to exactly the sealed
//! video-datagram size ([`video_datagram_udp_ceiling`], set in `quic/endpoint.rs`), so its
//! settled MTU is a verdict on the path.
//!
//! Three legs, none of which changes a session on a healthy path:
//! - **`PUNKTFUNK_WIRE_MTU=<bytes>`** — operator override; the shard payload is derived from
//! the given on-wire IP MTU. Wire-compatible with every deployed client:
//! `Welcome::shard_payload` is already negotiated per session (the v4/v6 split ships two
//! values today) and clients follow the negotiated value.
//! - **Watch** — a per-session task samples the control connection's discovered MTU once the
//! search has had time to finish. A connection still alive that settled BELOW the ceiling is
//! proof the path can't carry full-size video: log an actionable WARN and record the measured
//! budget for the peer.
//! - **Heal** — the next handshake from that peer clamps `shard_payload` to the recorded
//! budget, so a reconnect fixes the stream. A later session that reaches the ceiling erases
//! the record (the learn/heal loop is self-correcting in both directions).
use std::collections::HashMap;
use std::net::IpAddr;
use std::sync::{Mutex, OnceLock};
use punktfunk_core::config::{
mtu1500_shard_payload_for, sealed_datagram_bytes, shard_payload_for_udp_budget,
shard_payload_for_wire_mtu, video_datagram_udp_ceiling,
};
/// Measured UDP-payload budget per peer IP, learned from live control connections whose MTU
/// discovery settled below the video-datagram ceiling. In-memory only: a host restart
/// re-learns in one session, and entries self-correct (a later ceiling-hit erases, a lower
/// re-measure overwrites).
fn learned() -> &'static Mutex<HashMap<IpAddr, u16>> {
static LEARNED: OnceLock<Mutex<HashMap<IpAddr, u16>>> = OnceLock::new();
LEARNED.get_or_init(|| Mutex::new(HashMap::new()))
}
/// The shard payload for a new session to `peer`: `PUNKTFUNK_WIRE_MTU` override, else the
/// peer's learned path budget, else the family default (today's exact behavior). Logs whenever
/// the result differs from the default.
pub(super) fn negotiated_shard_payload(peer: IpAddr) -> usize {
let env = match std::env::var("PUNKTFUNK_WIRE_MTU") {
Ok(v) => match v.trim().parse::<usize>() {
Ok(mtu) => Some(mtu),
Err(_) => {
tracing::warn!(value = %v, "PUNKTFUNK_WIRE_MTU is not a number — ignoring it");
None
}
},
Err(_) => None,
};
let learned_budget = learned().lock().unwrap().get(&peer).copied();
resolve(env, learned_budget, peer)
}
/// Pure resolution (env override > learned budget > family default) — the tested core of
/// [`negotiated_shard_payload`].
fn resolve(env_wire_mtu: Option<usize>, learned_udp_budget: Option<u16>, peer: IpAddr) -> usize {
let default = mtu1500_shard_payload_for(peer);
if let Some(mtu) = env_wire_mtu {
let p = shard_payload_for_wire_mtu(mtu, peer);
if p != default {
tracing::info!(
wire_mtu = mtu,
shard_payload = p,
default,
"wire MTU: shard payload set from PUNKTFUNK_WIRE_MTU"
);
}
return p;
}
if let Some(budget) = learned_udp_budget {
let p = shard_payload_for_udp_budget(budget as usize, peer);
if p != default {
tracing::info!(
peer = %peer,
udp_budget = budget,
shard_payload = p,
default,
"wire MTU: shard payload clamped to this peer's measured path MTU (learned \
from a prior session's QUIC MTU discovery) — video datagrams now fit the \
constrained hop"
);
return p;
}
}
default
}
/// Sample the control connection's discovered MTU after the search has settled and turn it
/// into a verdict. Spawned once per negotiated session; the task ends by itself after the
/// final sample (bounded ~10 s lifetime, holding only a cheap `Connection` handle).
pub(super) fn spawn_watch(conn: quinn::Connection, session_shard_payload: usize) {
tokio::spawn(async move {
let peer = conn.remote_address().ip();
let ceiling = video_datagram_udp_ceiling() as u16;
// Discovery finishes in a handful of RTTs on a LAN (well under the first sample) but
// needs a loss timeout per failed probe on a constrained path — the second sample
// covers that with margin. Max, because discovery only ever raises `current_mtu`.
let mut settled = 0u16;
for wait_s in [3u64, 7] {
tokio::time::sleep(std::time::Duration::from_secs(wait_s)).await;
settled = settled.max(conn.stats().path.current_mtu);
if settled >= ceiling {
break;
}
}
if settled >= ceiling {
// The path carries full-size video datagrams — erase any stale learned clamp so
// the next session returns to the default wire.
if learned().lock().unwrap().remove(&peer).is_some() {
tracing::info!(peer = %peer,
"wire MTU: path re-measured at full size — learned clamp cleared");
}
return;
}
// A closed connection stops discovering, so a session that ended before the final
// sample proves nothing (a healthy high-RTT path could still be mid-search): learn
// only from a connection that stayed alive through the whole window.
if conn.close_reason().is_some() {
return;
}
learned().lock().unwrap().insert(peer, settled);
if sealed_datagram_bytes(session_shard_payload) <= settled as usize {
// This session was already clamped small enough — the path is still constrained
// (keep the record fresh) but video fits, so no alarm.
tracing::info!(peer = %peer, discovered_udp_mtu = settled,
"wire MTU: constrained path re-measured; this session's video is sized to fit");
} else {
tracing::warn!(
peer = %peer,
discovered_udp_mtu = settled,
needed_udp_mtu = ceiling,
"wire MTU: this path CANNOT carry full-size video datagrams — the control \
plane works but every video packet is oversized for a hop, which streams as \
an endless black screen with zero reported loss. Typical cause: a VPN/overlay \
adapter (Tailscale / Cloudflare WARP / ZeroTier) claiming the LAN route, or a \
lowered NIC MTU — compare `ping <client> -f -l 1450` vs `-l 1200` and check \
`netsh interface ipv4 show subinterfaces` (Windows) / `ip link` (Linux). The \
measured budget is recorded: the NEXT session from this client sizes video to \
fit automatically. To pin it for all sessions set PUNKTFUNK_WIRE_MTU."
);
}
});
}
#[cfg(test)]
mod tests {
use super::*;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
const V4: IpAddr = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 2));
const V6: IpAddr = IpAddr::V6(Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1));
#[test]
fn default_when_nothing_known() {
assert_eq!(resolve(None, None, V4), mtu1500_shard_payload_for(V4));
assert_eq!(resolve(None, None, V6), mtu1500_shard_payload_for(V6));
}
#[test]
fn env_override_beats_learned() {
// 1280 wire 28 IP/UDP 64 header/crypto = 1188.
assert_eq!(resolve(Some(1280), Some(1472), V4), 1188);
}
#[test]
fn learned_budget_clamps() {
// A WARP-shaped path: 1280-byte UDP budget → 1280 64 = 1216.
assert_eq!(resolve(None, Some(1280), V4), 1216);
}
#[test]
fn learned_at_or_above_ceiling_is_the_default_wire() {
assert_eq!(resolve(None, Some(1472), V4), mtu1500_shard_payload_for(V4));
assert_eq!(resolve(None, Some(2000), V4), mtu1500_shard_payload_for(V4));
}
#[test]
fn env_full_mtu_is_the_default_wire_both_families() {
assert_eq!(resolve(Some(1500), None, V4), mtu1500_shard_payload_for(V4));
assert_eq!(resolve(Some(1500), None, V6), mtu1500_shard_payload_for(V6));
}
}
+6
View File
@@ -333,6 +333,12 @@
#define INBOUND_REQ_FLAG 2147483648
#endif
// Floor for a negotiated `shard_payload` (even, well under every real path). A path whose UDP
// budget lands below this can't carry the QUIC control plane either (QUIC's own minimum is a
// 1200-byte UDP payload), so shrinking video shards further buys nothing — the clamp helpers
// bottom out here instead of producing degenerate confetti-sized shards.
#define MIN_SHARD_PAYLOAD 512
// 16-byte AEAD authentication tag appended by either session cipher.
#define TAG_LEN 16