From 34ad3cc611810142fa28f52a0d7b63a5dacd29ee Mon Sep 17 00:00:00 2001 From: enricobuehler Date: Tue, 4 Aug 2026 19:42:36 +0200 Subject: [PATCH] feat(host/wire): mid-session shard-payload renegotiation, driven by the MTU verdict MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phases 1-2 of design/shard-payload-reneg.md, on top of the Phase 0 per-frame geometry. The leg-1 watcher stops merely diagnosing the constrained path and heals the CURRENT session; the same machinery, inverted, takes a proven jumbo LAN up to ~8.9 KB shards. - Messages: MSG_SHARD_PAYLOAD_CHANGED (0x08, host→client, {shard_payload u16}) and MSG_SHARD_PAYLOAD_ACK (0x09, the echo). Asymmetric by design: a shrink re-keys the packetizer at the next AU immediately after sending (per-frame pinning makes ordering irrelevant; the ack is telemetry), a grow emits nothing above the old size until the ack — the ack is the gate even though client buffers are statically sized. - Client: one dispatch arm in the shared pump control task (all client families) — validate against the advertised receive bounds, ack; out-of-bounds requests get SILENCE, not an ack, so a buggy host can never read a granted grow out of garbage. - Host driver: the wire_mtu watcher grows a ShardReneg arm — on a below-ceiling verdict it still records the learned budget (session 2 starts right) and now also shrinks session 1 at the ~3-10 s verdict mark; with the jumbo opt-in (PUNKTFUNK_JUMBO=1, or PUNKTFUNK_WIRE_MTU > 1500 — one knob, derived) it sends the ack-gated grow after a settled-at-sealed-jumbo proof and then stays alive as the revert guard: quinn's blackhole detection lowering current_mtu shrinks the wire back through the same path. The QUIC MTUD probe ceiling rises from 1472 to the sealed jumbo size with the opt-in (per-ENDPOINT: a few extra failed probes toward non-jumbo peers, zero cost otherwise). - Apply point: Session::set_shard_payload drained in the send loop next to the adaptive-FEC target, gated on no open streamed AU (a streamed frame's shard-aligned tiling derives from the size it began with). - Renegotiation is gated OFF for PyroWave sessions: their clients parse chunk-aligned AUs in windows of the Welcome value pinned at session start (read once over the C ABI), so a mid-stream re-key would corrupt the parse — those sessions keep the leg-1 next-session clamp. This also settles the plan's open question on the two wire_chunk consumers: both are PyroWave-only, so the gate covers them entirely. - Legacy peers are inert both ways: no Hello advertisement → the host never constructs the driver; an old host never sends the message. core: 296/296 --features quic + clippy -D warnings (macOS), fmt; the regenerated header carries the new message ids (drift gate). --- .../src/client/pump/control_task.rs | 32 +++ crates/punktfunk-core/src/config.rs | 73 ++++++- crates/punktfunk-core/src/quic/control.rs | 101 ++++++++++ crates/punktfunk-core/src/quic/endpoint.rs | 20 +- crates/punktfunk-host/src/native.rs | 27 +++ crates/punktfunk-host/src/native/control.rs | 31 +++ crates/punktfunk-host/src/native/handshake.rs | 7 +- crates/punktfunk-host/src/native/stream.rs | 29 +++ crates/punktfunk-host/src/native/wire_mtu.rs | 182 ++++++++++++++---- include/punktfunk_core.h | 10 + 10 files changed, 468 insertions(+), 44 deletions(-) diff --git a/crates/punktfunk-core/src/client/pump/control_task.rs b/crates/punktfunk-core/src/client/pump/control_task.rs index 291f47a1..2af7b82e 100644 --- a/crates/punktfunk-core/src/client/pump/control_task.rs +++ b/crates/punktfunk-core/src/client/pump/control_task.rs @@ -243,6 +243,38 @@ impl ControlTask { seq: offer.seq, kinds: offer.kinds, }); + } else if let Ok(chg) = crate::quic::ShardPayloadChanged::decode(&msg) { + // Mid-session shard renegotiation (design/shard-payload-reneg.md): the + // host re-keys the sealed video geometry. Per-frame pinning means there + // is nothing to re-key on the receive path — the reassembler follows + // each frame's own header and every buffer is statically sized for the + // ceiling — so the dispatch is validate + ack. The ack is telemetry for + // a shrink and the GATE for a grow (the host emits nothing above the + // old size until it lands). Validate against our own receive bounds — + // the same ceiling we advertised in `Hello::max_shard_payload` — and + // answer an out-of-bounds request with SILENCE, not an ack: a buggy + // host must never read a granted grow out of garbage. + let n = chg.shard_payload as usize; + if (crate::config::MIN_SHARD_PAYLOAD..=crate::config::max_shard_payload()) + .contains(&n) + && n % 2 == 0 + { + tracing::info!( + shard_payload = n, + "host re-keyed the wire shard payload — acking" + ); + let ack = crate::quic::ShardPayloadAck { + shard_payload: chg.shard_payload, + }; + if io::write_msg(&mut ctrl_send, &ack.encode()).await.is_err() { + break; + } + } else { + tracing::warn!( + shard_payload = n, + "out-of-bounds shard-payload change — ignoring (no ack)" + ); + } } else if let Ok(shape) = crate::quic::CursorShape::decode(&msg) { // Pointer bitmap changed (cursor channel, only when negotiated). try_send: // an overflowing ring drops the newest shape — the next change resends. diff --git a/crates/punktfunk-core/src/config.rs b/crates/punktfunk-core/src/config.rs index 1c0d2af7..8ccd2dff 100644 --- a/crates/punktfunk-core/src/config.rs +++ b/crates/punktfunk-core/src/config.rs @@ -373,16 +373,54 @@ pub fn shard_payload_for_udp_budget(udp_budget: usize, peer: core::net::IpAddr) p.clamp(MIN_SHARD_PAYLOAD, mtu1500_shard_payload_for(peer)) } +/// The family's IP+UDP header bytes between an on-wire IP MTU and its UDP payload budget — +/// 28 for IPv4 (and IPv4-mapped), 48 for IPv6. +fn ip_udp_overhead(peer: core::net::IpAddr) -> usize { + 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`] 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) + shard_payload_for_udp_budget(wire_mtu.saturating_sub(ip_udp_overhead(peer)), peer) +} + +/// The operator's jumbo-frames opt-in (design/shard-payload-reneg.md Phase 2): the target +/// on-wire IP MTU, or `None` = no opt-in (nothing above the 1500-default wire is ever probed +/// or grown to). One knob, one code path: a `PUNKTFUNK_WIRE_MTU` above the standard 1500 +/// derives the target from the operator's number; `PUNKTFUNK_JUMBO=1` is the fixed 9000 +/// profile for operators who don't want to think in MTUs. Raising the wire above 1500 is +/// only ever an ACK-GATED mid-session grow toward a client that advertised +/// [`max_shard_payload`] headroom — sessions still START at the family default. +pub fn jumbo_wire_mtu() -> Option { + if let Ok(v) = std::env::var("PUNKTFUNK_WIRE_MTU") { + if let Ok(mtu) = v.trim().parse::() { + if mtu > 1500 { + return Some(mtu); + } + } + } + match std::env::var("PUNKTFUNK_JUMBO") { + Ok(v) if v.trim() == "1" => Some(9000), + _ => None, + } +} + +/// The jumbo sibling of [`shard_payload_for_wire_mtu`]: the largest even shard payload whose +/// sealed datagram fits `wire_mtu`, clamped to the RECEIVE ceiling ([`max_shard_payload`]) +/// instead of the family 1500-default — the up-leg's grow target. Still floored at +/// [`MIN_SHARD_PAYLOAD`]. +pub fn jumbo_shard_payload_for(wire_mtu: usize, peer: core::net::IpAddr) -> usize { + let p = wire_mtu + .saturating_sub(ip_udp_overhead(peer)) + .saturating_sub(HEADER_LEN + CRYPTO_OVERHEAD); + let p = p - p % 2; // FEC requires even shards + p.clamp(MIN_SHARD_PAYLOAD, max_shard_payload()) } /// Everything needed to construct a [`Session`](crate::session::Session). @@ -626,6 +664,29 @@ mod tests { assert_eq!(shard_payload_for_wire_mtu(1280, v6), 1168); } + /// Jumbo grow-target sizing (the up-leg, design/shard-payload-reneg.md): even, sealed + /// fits the wire, clamped to the RECEIVE ceiling instead of the family 1500-default — + /// and the standard 9000 profile lands on the exact documented value. + #[test] + fn jumbo_shard_payload_math() { + use core::net::IpAddr; + let v4: IpAddr = "192.168.1.50".parse().unwrap(); + let v6: IpAddr = "fd00::50".parse().unwrap(); + // 9000 − 28 (IPv4+UDP) − 64 (header+crypto) = 8908 even; sealed 8972 ≤ the 9216 + // datagram ceiling. The v6 sibling: 9000 − 48 − 64 = 8888. + assert_eq!(jumbo_shard_payload_for(9000, v4), 8908); + assert_eq!(sealed_datagram_bytes(8908), 8972); + assert!(sealed_datagram_bytes(8908) <= MAX_DATAGRAM_BYTES); + assert_eq!(jumbo_shard_payload_for(9000, v6), 8888); + // An operator MTU larger than the receive path clamps to the ceiling, smaller ones + // track the wire, and degenerate ones floor at MIN_SHARD_PAYLOAD. + assert_eq!(jumbo_shard_payload_for(64_000, v4), max_shard_payload()); + let p = jumbo_shard_payload_for(4000, v4); + assert_eq!(p % 2, 0); + assert!(sealed_datagram_bytes(p) <= 4000 - 28); + assert_eq!(jumbo_shard_payload_for(100, v4), MIN_SHARD_PAYLOAD); + } + /// 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] diff --git a/crates/punktfunk-core/src/quic/control.rs b/crates/punktfunk-core/src/quic/control.rs index 54155cc4..cce568e4 100644 --- a/crates/punktfunk-core/src/quic/control.rs +++ b/crates/punktfunk-core/src/quic/control.rs @@ -55,6 +55,36 @@ pub struct RfiRequest { pub last_frame: u32, } +/// `host → client`, any time after [`Start`]: the video data plane's sealed shard payload +/// changes mid-session (design/shard-payload-reneg.md Phase 1). Sent ONLY to a client whose +/// [`Hello::max_shard_payload`] advertised per-frame geometry (0/absent = legacy — the host +/// must never send this), and never above that advertised ceiling. Asymmetric semantics: +/// +/// - **Shrink** (the mid-session MTU heal): the host may re-key its packetizer at the next +/// AU boundary immediately after sending — per-frame pinning on the client makes the +/// control-vs-datagram reorder race irrelevant and a smaller shard always fits existing +/// buffers. The [`ShardPayloadAck`] is telemetry. +/// - **Grow** (jumbo): the host must not emit a single sealed datagram above the OLD size +/// until the ack arrives — the ack IS the gate, even when the client's buffers would +/// happen to fit (the rule must not erode if the buffer strategy changes later). +/// +/// No `effective_frame_index`: per-frame pinning makes it redundant — every video packet +/// carries its own `shard_bytes` and the receiver follows each frame's pin. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ShardPayloadChanged { + /// The new sealed shard payload in bytes (even, within the client's advertised bounds). + pub shard_payload: u16, +} + +/// `client → host`: answer to [`ShardPayloadChanged`] — echoes the value the client applied. +/// Only sent for an in-bounds request; an out-of-bounds one is dropped WITHOUT an ack (a +/// buggy host must not read silence-then-garbage as a granted grow). The host treats the +/// echoed value as the grant for a pending grow. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ShardPayloadAck { + pub shard_payload: u16, +} + /// `client → host`, periodic: the client's observed data-plane loss, so the host can size FEC to /// the link instead of a flat percentage (adaptive FEC). `loss_ppm` is parts-per-million of shards /// that arrived missing-but-recovered (plus a bump when frames went unrecoverable) over the report @@ -200,6 +230,10 @@ pub const MSG_SET_BITRATE: u8 = 0x05; pub const MSG_BITRATE_CHANGED: u8 = 0x06; /// Type byte of [`RfiRequest`]. pub const MSG_RFI_REQUEST: u8 = 0x07; +/// Type byte of [`ShardPayloadChanged`]. +pub const MSG_SHARD_PAYLOAD_CHANGED: u8 = 0x08; +/// Type byte of [`ShardPayloadAck`]. +pub const MSG_SHARD_PAYLOAD_ACK: u8 = 0x09; /// Type byte of [`ProbeRequest`]. pub const MSG_PROBE_REQUEST: u8 = 0x20; /// Type byte of [`ProbeResult`]. @@ -306,6 +340,46 @@ impl RfiRequest { } } +impl ShardPayloadChanged { + pub fn encode(&self) -> Vec { + // magic[0..4] type[4] shard_payload[5..7] + let mut b = Vec::with_capacity(7); + b.extend_from_slice(CTL_MAGIC); + b.push(MSG_SHARD_PAYLOAD_CHANGED); + b.extend_from_slice(&self.shard_payload.to_le_bytes()); + b + } + + pub fn decode(b: &[u8]) -> Result { + if b.len() != 7 || &b[0..4] != CTL_MAGIC || b[4] != MSG_SHARD_PAYLOAD_CHANGED { + return Err(PunktfunkError::InvalidArg("bad ShardPayloadChanged")); + } + Ok(ShardPayloadChanged { + shard_payload: u16::from_le_bytes(b[5..7].try_into().unwrap()), + }) + } +} + +impl ShardPayloadAck { + pub fn encode(&self) -> Vec { + // magic[0..4] type[4] shard_payload[5..7] + let mut b = Vec::with_capacity(7); + b.extend_from_slice(CTL_MAGIC); + b.push(MSG_SHARD_PAYLOAD_ACK); + b.extend_from_slice(&self.shard_payload.to_le_bytes()); + b + } + + pub fn decode(b: &[u8]) -> Result { + if b.len() != 7 || &b[0..4] != CTL_MAGIC || b[4] != MSG_SHARD_PAYLOAD_ACK { + return Err(PunktfunkError::InvalidArg("bad ShardPayloadAck")); + } + Ok(ShardPayloadAck { + shard_payload: u16::from_le_bytes(b[5..7].try_into().unwrap()), + }) + } +} + impl LossReport { pub fn encode(&self) -> Vec { // magic[0..4] type[4] loss_ppm[5..9] @@ -1146,6 +1220,33 @@ mod tests { assert!(SetBitrate::decode(&LossReport { loss_ppm: 7 }.encode()).is_err()); } + #[test] + fn shard_payload_messages_roundtrip() { + for shard_payload in [512u16, 1216, 1408, 8908] { + let chg = ShardPayloadChanged { shard_payload }; + assert_eq!(ShardPayloadChanged::decode(&chg.encode()).unwrap(), chg); + let ack = ShardPayloadAck { shard_payload }; + assert_eq!(ShardPayloadAck::decode(&ack.encode()).unwrap(), ack); + // Identical payload shape — the type byte alone must keep the pair disjoint (a + // change echoed back must never re-decode as a change). + assert!(ShardPayloadChanged::decode(&ack.encode()).is_err()); + assert!(ShardPayloadAck::decode(&chg.encode()).is_err()); + } + // Exact length — no trailing bytes, no truncation. + let bytes = ShardPayloadChanged { shard_payload: 512 }.encode(); + assert!(ShardPayloadChanged::decode(&[bytes.as_slice(), &[0]].concat()).is_err()); + assert!(ShardPayloadChanged::decode(&bytes[..bytes.len() - 1]).is_err()); + // Disjoint from the neighboring ids either side (0x07 RfiRequest / 0x20 ProbeRequest). + assert!(ShardPayloadChanged::decode( + &RfiRequest { + first_frame: 1, + last_frame: 2 + } + .encode() + ) + .is_err()); + } + #[test] fn probe_messages_roundtrip() { let req = ProbeRequest { diff --git a/crates/punktfunk-core/src/quic/endpoint.rs b/crates/punktfunk-core/src/quic/endpoint.rs index 8ed25d6b..43a786ff 100644 --- a/crates/punktfunk-core/src/quic/endpoint.rs +++ b/crates/punktfunk-core/src/quic/endpoint.rs @@ -59,7 +59,25 @@ fn stream_transport_idle(idle: std::time::Duration) -> Arc 1500 set, discovery probes up to the sealed JUMBO datagram + // size so a settled connection can PROVE a jumbo path — the actual grow stays + // client-ack-gated (`native/wire_mtu.rs`). The ceiling is per-ENDPOINT, not + // per-connection: with the opt-in set, connections to non-jumbo peers spend a few extra + // failed probes (one PTO each) settling lower; zero cost for anyone who doesn't opt in. + // Derived with the IPv4 overhead — a v6 peer's sealed jumbo target is smaller, so the + // ceiling covers it and discovery settles at the v6 path's own budget. + let probe_ceiling = match crate::config::jumbo_wire_mtu() { + Some(mtu) => { + let shard = crate::config::jumbo_shard_payload_for( + mtu, + std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED), + ); + crate::config::sealed_datagram_bytes(shard) as u16 + } + None => crate::config::video_datagram_udp_ceiling() as u16, + }; + mtud.upper_bound(probe_ceiling); t.mtu_discovery_config(Some(mtud)); Arc::new(t) } diff --git a/crates/punktfunk-host/src/native.rs b/crates/punktfunk-host/src/native.rs index ba41bb18..ac38be13 100644 --- a/crates/punktfunk-host/src/native.rs +++ b/crates/punktfunk-host/src/native.rs @@ -1091,6 +1091,30 @@ async fn serve_session( // just never fires then. let (cursor_shape_tx, cursor_shape_rx) = tokio::sync::mpsc::unbounded_channel::(); + // Mid-session shard renegotiation (design/shard-payload-reneg.md Phase 2): the wire-MTU + // watcher decides (constrained-path shrink / ack-gated jumbo grow), the control task + // writes the `ShardPayloadChanged` and routes the acks back, and the data-plane loop + // applies `Session::set_shard_payload` between AUs (drained next to `bitrate_rx`). + // Channels are wired unconditionally (they just never fire); the DRIVER exists only for + // a client that advertised `Hello::max_shard_payload` on a non-chunk-aligned session — + // PyroWave clients parse chunk-aligned AUs in windows of the `Welcome` value pinned at + // session start (read once over the C ABI), so those sessions keep the leg-1 + // next-session clamp instead of a mid-stream re-key. + let (shard_change_tx, shard_change_rx) = tokio::sync::mpsc::unbounded_channel::(); + let (shard_ack_tx, shard_ack_rx) = tokio::sync::mpsc::unbounded_channel::(); + let (shard_apply_tx, shard_apply_rx) = std::sync::mpsc::channel::(); + let shard_reneg = (hello.max_shard_payload > 0 && codec != crate::encode::Codec::PyroWave) + .then_some(wire_mtu::ShardReneg { + client_ceiling: hello.max_shard_payload, + change_tx: shard_change_tx, + ack_rx: shard_ack_rx, + apply_tx: shard_apply_tx, + }); + // 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) — and, with the driver above, heal or grow THIS + // session mid-stream. Bounded ~10 s task unless a jumbo grow leaves it as revert guard. + wire_mtu::spawn_watch(conn.clone(), welcome.shard_payload as usize, shard_reneg); // Negotiated cursor forwarding: the HOST_CAP_CURSOR bit the Welcome advertised, read back // rather than recomputed (`handshake::cursor_forward` computed it once, with the encoder // blend-capability gate — re-running it here could drift, and would re-probe). @@ -1146,6 +1170,8 @@ async fn serve_session( probe_result_rx, reconfig_result_rx, retarget_rx, + shard_change_rx, + shard_ack_tx, cursor_shape_rx, cursor_client_draws, clip_enabled, @@ -1579,6 +1605,7 @@ async fn serve_session( keyframe: keyframe_rx, rfi: rfi_rx, bitrate_rx, + shard_rx: shard_apply_rx, compositor, gamescope_route, bitrate_kbps, diff --git a/crates/punktfunk-host/src/native/control.rs b/crates/punktfunk-host/src/native/control.rs index e14934c6..a6bdc9ca 100644 --- a/crates/punktfunk-host/src/native/control.rs +++ b/crates/punktfunk-host/src/native/control.rs @@ -43,6 +43,11 @@ pub(super) async fn run( // Host-initiated bitrate re-target (a rebuild re-resolved an Automatic rate): forwarded to // the client as a `BitrateChanged` so its controller's climb base tracks the real encoder. mut retarget_rx: tokio::sync::mpsc::UnboundedReceiver, + // Mid-session shard renegotiation (design/shard-payload-reneg.md): the wire-MTU watcher + // asks for a `ShardPayloadChanged` here (this task is the control stream's sole writer), + // and the client's `ShardPayloadAck`s flow back on `shard_ack_tx` — the grow gate. + mut shard_change_rx: tokio::sync::mpsc::UnboundedReceiver, + shard_ack_tx: tokio::sync::mpsc::UnboundedSender, mut cursor_shape_rx: tokio::sync::mpsc::UnboundedReceiver, cursor_client_draws: Arc, clip_enabled: Arc, @@ -56,6 +61,9 @@ pub(super) async fn run( // Set once `clip_offer_rx` closes (coordinator gone / inert handle) so its `select!` branch // stops firing on a perpetually-ready `None`. let mut clip_offer_closed = false; + // Same discipline for the wire-MTU watcher's channel — its bounded lifetime ends mid-session + // on every healthy path. + let mut shard_change_closed = false; let mut active = initial_mode; // Host-side switch rate limit (a backstop against a hostile/broken client spamming // Reconfigure into pipeline-rebuild churn — the drain-to-newest in the data plane already @@ -214,6 +222,16 @@ pub(super) async fn run( if bitrate_tx.send(resolved).is_err() { break; // data plane gone } + } else if let Ok(ack) = punktfunk_core::quic::ShardPayloadAck::decode(&msg) { + // Mid-session shard renegotiation: the client applied (or granted) a + // geometry change. Forward to the wire-MTU watcher — for a grow this IS + // the gate that lets the packetizer go above the old size. A dropped + // send just means the watcher already ended (shrink acks are telemetry). + tracing::info!( + shard_payload = ack.shard_payload, + "client acked shard-payload change" + ); + let _ = shard_ack_tx.send(ack.shard_payload); } else if let Ok(req) = ProbeRequest::decode(&msg) { tracing::info!( target_kbps = req.target_kbps, @@ -317,6 +335,19 @@ pub(super) async fn run( break; } } + n = shard_change_rx.recv(), if !shard_change_closed => { + // Mid-session shard renegotiation: the wire-MTU watcher decided (shrink on a + // constrained-path verdict / ack-gated jumbo grow). Only ever fires toward a + // client that advertised `Hello::max_shard_payload` — the watcher owns that + // gate. `None` = the watcher's bounded lifetime ended (normal, NOT a session + // end): disable this branch, exactly the `clip_offer_closed` pattern — a + // closed mpsc yields `None` perpetually and would busy-spin the select. + let Some(n) = n else { shard_change_closed = true; continue }; + let msg = punktfunk_core::quic::ShardPayloadChanged { shard_payload: n }; + if io::write_msg(&mut ctrl_send, &msg.encode()).await.is_err() { + break; + } + } shape = cursor_shape_rx.recv() => { // Cursor-forward bridge (M2): the encode loop diffed a new pointer bitmap. // Rare (shape changes are human-paced); ≤ ~58 KiB fits the u16 frame by diff --git a/crates/punktfunk-host/src/native/handshake.rs b/crates/punktfunk-host/src/native/handshake.rs index 29254549..d62e60df 100644 --- a/crates/punktfunk-host/src/native/handshake.rs +++ b/crates/punktfunk-host/src/native/handshake.rs @@ -663,10 +663,9 @@ 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); + // The wire-MTU watch (`wire_mtu::spawn_watch`) is spawned by `serve_session` after the + // control-task channels exist — it now also DRIVES the mid-session shard renegotiation + // (design/shard-payload-reneg.md), which needs the control stream's writer. Ok::<_, anyhow::Error>(( hello, welcome, diff --git a/crates/punktfunk-host/src/native/stream.rs b/crates/punktfunk-host/src/native/stream.rs index 6bba243e..6047d088 100644 --- a/crates/punktfunk-host/src/native/stream.rs +++ b/crates/punktfunk-host/src/native/stream.rs @@ -762,6 +762,9 @@ fn send_loop( slice_wire: bool, burst_cap: Option, fec_target: Arc, + // Mid-session shard-payload re-keys from the wire-MTU watcher (validated + ack-gated + // there) — applied between AUs only (design/shard-payload-reneg.md Phase 1). + shard_rx: std::sync::mpsc::Receiver, stats: SendStats, // `Some` = the client advertised VIDEO_CAP_HOST_TIMING: emit one 0xCF datagram per AU right // after its last packet left the socket (capture→sent, the whole host pipeline incl. pacing). @@ -818,6 +821,25 @@ fn send_loop( } // Adaptive FEC: pick up any new recovery target the control task set from client LossReports. apply_fec_target(&mut session, &fec_target); + // Mid-session shard renegotiation: apply a re-key from the wire-MTU watcher — between + // AUs only, NEVER with a streamed AU open (its shard-aligned tiling derives from the + // size it began with; same gate as the probe burst above). Drain to the newest; the + // protocol side (client advertisement, ack-gated grow) was enforced by the watcher. + if streamed.is_none() { + let mut want_shard = None; + while let Ok(s) = shard_rx.try_recv() { + want_shard = Some(s); + } + if let Some(s) = want_shard { + match session.set_shard_payload(s) { + Ok(()) => tracing::info!(shard_payload = s, "wire shard payload re-keyed"), + // Can't fire for a watcher-driven value (it validates the same bounds) — + // belt-and-suspenders for a future driver. + Err(e) => tracing::warn!(shard_payload = s, error = ?e, + "shard re-key refused by session validation"), + } + } + } // Short timeout so we keep re-checking `stop` + probes when no frames are flowing. match frame_rx.recv_timeout(std::time::Duration::from_millis(50)) { Ok(send_msg) => { @@ -1171,6 +1193,11 @@ pub(super) struct SessionContext { /// Accepted mid-stream bitrate changes (adaptive bitrate, already clamped) — the encoder /// alone is rebuilt in place at the new rate; capture + virtual output are untouched. pub(super) bitrate_rx: std::sync::mpsc::Receiver, + /// Mid-session shard-payload changes from the wire-MTU watcher (already validated + + /// protocol-gated there; a grow arrives only after the client's ack). Applied between + /// AUs via [`Session::set_shard_payload`] — the packetizer re-keys, capture/encoder/ + /// virtual output are untouched (design/shard-payload-reneg.md Phase 1). + pub(super) shard_rx: std::sync::mpsc::Receiver, /// The resolved compositor backend (moot on Windows — `vdisplay::open` ignores it there). pub(super) compositor: crate::vdisplay::Compositor, /// This session's resolved gamescope sub-mode, or `None` for every other backend. Carried here @@ -1385,6 +1412,7 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option 0 by construction). + pub client_ceiling: u16, + /// → control task (the control stream's sole writer): send `ShardPayloadChanged{n}`. + pub change_tx: tokio::sync::mpsc::UnboundedSender, + /// ← control task: the client's `ShardPayloadAck`s (the grow gate). + pub ack_rx: tokio::sync::mpsc::UnboundedReceiver, + /// → data plane: apply [`Session::set_shard_payload`] between AUs + /// (drained next to `bitrate_rx` in the encode loop). + pub apply_tx: std::sync::mpsc::Sender, +} + /// 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 @@ -96,15 +116,26 @@ fn resolve(env_wire_mtu: Option, learned_udp_budget: Option, peer: I } /// 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) { +/// into a verdict — and, with a [`ShardReneg`] driver, act on it MID-SESSION +/// (design/shard-payload-reneg.md Phase 2): a below-ceiling verdict shrinks the live wire at +/// the ~3–10 s mark (session 1 heals instead of staying black), and a settled-at-jumbo +/// verdict grows it, ack-gated, when the operator opted in. Spawned once per negotiated +/// session; without a grow the task ends after the final sample (bounded ~10 s lifetime, +/// holding only a cheap `Connection` handle) — after a grow it stays as the revert guard +/// until the connection closes. +pub(super) fn spawn_watch( + conn: quinn::Connection, + session_shard_payload: usize, + reneg: Option, +) { 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`. + // covers that with margin. Max, because discovery only ever raises `current_mtu` + // (the post-grow revert guard below re-reads it live, where blackhole detection CAN + // lower it again). let mut settled = 0u16; for wait_s in [3u64, 7] { tokio::time::sleep(std::time::Duration::from_secs(wait_s)).await; @@ -113,6 +144,9 @@ pub(super) fn spawn_watch(conn: quinn::Connection, session_shard_payload: usize) break; } } + // The wire this session is CURRENTLY sealed at — moves on a mid-session shrink/grow. + let mut current = session_shard_payload; + let mut reneg = reneg; if settled >= ceiling { // The path carries full-size video datagrams — erase any stale learned clamp so // the next session returns to the default wire. @@ -120,34 +154,116 @@ pub(super) fn spawn_watch(conn: quinn::Connection, session_shard_payload: usize) 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 -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." - ); + // 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(current) <= 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 -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." + ); + // Phase 2 down-leg: heal THIS session at the verdict mark. Shrink is sent + // and applied immediately — per-frame pinning on the client makes ordering + // irrelevant and smaller always fits; the ack is telemetry. The learned + // record above still makes session 2 START right. + if let Some(r) = reneg.as_ref() { + let target = shard_payload_for_udp_budget(settled as usize, peer); + if target < current + && r.change_tx.send(target as u16).is_ok() + && r.apply_tx.send(target).is_ok() + { + tracing::info!( + peer = %peer, + shard_payload = target, + was = current, + "wire MTU: video re-keyed mid-session to fit the constrained path \ + — the stream heals now instead of on the next connect" + ); + current = target; + } + } + } + } + // Phase 2 up-leg: jumbo grow — operator opt-in (PUNKTFUNK_JUMBO / PUNKTFUNK_WIRE_MTU + // > 1500, which also raised the endpoint's probe ceiling so `settled` can even reach + // here), client-advertised headroom, and a settled-at-jumbo proof. The grow is + // ACK-GATED: not one sealed datagram above the old size leaves before the client's + // ack, even though its buffers are statically sized — the rule must not erode. + let (Some(mtu), Some(r)) = (jumbo_wire_mtu(), reneg.as_mut()) else { + return; + }; + let target = jumbo_shard_payload_for(mtu, peer).min(r.client_ceiling as usize); + let target = target - target % 2; + if target <= current || (settled as usize) < sealed_datagram_bytes(target) { + return; + } + if r.change_tx.send(target as u16).is_err() { + return; + } + let acked = tokio::time::timeout(std::time::Duration::from_secs(5), async { + while let Some(v) = r.ack_rx.recv().await { + if v as usize == target { + return true; + } + } + false + }) + .await + .unwrap_or(false); + if !acked { + tracing::warn!(peer = %peer, shard_payload = target, + "wire MTU: jumbo grow not acked — staying at the current wire"); + return; + } + if r.apply_tx.send(target).is_err() { + return; + } + tracing::info!( + peer = %peer, + shard_payload = target, + was = current, + wire_mtu = mtu, + "wire MTU: jumbo grow acked and applied — packets-per-frame cut ~6×" + ); + current = target; + // Revert guard: a mis-proven jumbo hop must self-correct instead of blackholing. + // quinn's PMTU blackhole detection lowers `current_mtu` when the big packets start + // vanishing; sample it and shrink back through the same path the down-leg uses. + loop { + tokio::time::sleep(std::time::Duration::from_secs(5)).await; + if conn.close_reason().is_some() { + return; + } + let mtu_now = conn.stats().path.current_mtu; + if (mtu_now as usize) < sealed_datagram_bytes(current) { + let back = shard_payload_for_udp_budget(mtu_now as usize, peer); + tracing::warn!(peer = %peer, discovered_udp_mtu = mtu_now, + shard_payload = back, was = current, + "wire MTU: jumbo path stopped fitting — reverting the wire to match"); + if r.change_tx.send(back as u16).is_err() || r.apply_tx.send(back).is_err() { + return; + } + current = back; + } } }); } diff --git a/include/punktfunk_core.h b/include/punktfunk_core.h index 154239ae..868dd3f3 100644 --- a/include/punktfunk_core.h +++ b/include/punktfunk_core.h @@ -792,6 +792,16 @@ #define MSG_RFI_REQUEST 7 #endif +#if defined(PUNKTFUNK_FEATURE_QUIC) +// Type byte of [`ShardPayloadChanged`]. +#define MSG_SHARD_PAYLOAD_CHANGED 8 +#endif + +#if defined(PUNKTFUNK_FEATURE_QUIC) +// Type byte of [`ShardPayloadAck`]. +#define MSG_SHARD_PAYLOAD_ACK 9 +#endif + #if defined(PUNKTFUNK_FEATURE_QUIC) // Type byte of [`ProbeRequest`]. #define MSG_PROBE_REQUEST 32