feat(host/wire): mid-session shard-payload renegotiation, driven by the MTU verdict
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).
This commit is contained in:
@@ -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.
|
||||
|
||||
@@ -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<usize> {
|
||||
if let Ok(v) = std::env::var("PUNKTFUNK_WIRE_MTU") {
|
||||
if let Ok(mtu) = v.trim().parse::<usize>() {
|
||||
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]
|
||||
|
||||
@@ -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<u8> {
|
||||
// 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<ShardPayloadChanged> {
|
||||
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<u8> {
|
||||
// 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<ShardPayloadAck> {
|
||||
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<u8> {
|
||||
// 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 {
|
||||
|
||||
@@ -59,7 +59,25 @@ fn stream_transport_idle(idle: std::time::Duration) -> Arc<quinn::TransportConfi
|
||||
// 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);
|
||||
// Jumbo opt-in (design/shard-payload-reneg.md Phase 2): with `PUNKTFUNK_JUMBO=1` /
|
||||
// `PUNKTFUNK_WIRE_MTU` > 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)
|
||||
}
|
||||
|
||||
@@ -1091,6 +1091,30 @@ async fn serve_session(
|
||||
// just never fires then.
|
||||
let (cursor_shape_tx, cursor_shape_rx) =
|
||||
tokio::sync::mpsc::unbounded_channel::<punktfunk_core::quic::CursorShape>();
|
||||
// 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::<u16>();
|
||||
let (shard_ack_tx, shard_ack_rx) = tokio::sync::mpsc::unbounded_channel::<u16>();
|
||||
let (shard_apply_tx, shard_apply_rx) = std::sync::mpsc::channel::<usize>();
|
||||
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,
|
||||
|
||||
@@ -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<u32>,
|
||||
// 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<u16>,
|
||||
shard_ack_tx: tokio::sync::mpsc::UnboundedSender<u16>,
|
||||
mut cursor_shape_rx: tokio::sync::mpsc::UnboundedReceiver<punktfunk_core::quic::CursorShape>,
|
||||
cursor_client_draws: Arc<AtomicBool>,
|
||||
clip_enabled: Arc<AtomicBool>,
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -762,6 +762,9 @@ fn send_loop(
|
||||
slice_wire: bool,
|
||||
burst_cap: Option<usize>,
|
||||
fec_target: Arc<AtomicU8>,
|
||||
// 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<usize>,
|
||||
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<u32>,
|
||||
/// 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<usize>,
|
||||
/// 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<PreparedDispl
|
||||
keyframe,
|
||||
rfi,
|
||||
bitrate_rx,
|
||||
shard_rx,
|
||||
compositor,
|
||||
gamescope_route,
|
||||
mut bitrate_kbps,
|
||||
@@ -1771,6 +1799,7 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
|
||||
slice_wire,
|
||||
burst_cap,
|
||||
fec_target,
|
||||
shard_rx,
|
||||
send_stats,
|
||||
timing_conn,
|
||||
phase_send,
|
||||
|
||||
@@ -30,10 +30,30 @@ 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,
|
||||
jumbo_shard_payload_for, jumbo_wire_mtu, mtu1500_shard_payload_for, sealed_datagram_bytes,
|
||||
shard_payload_for_udp_budget, shard_payload_for_wire_mtu, video_datagram_udp_ceiling,
|
||||
};
|
||||
|
||||
/// Everything the MID-SESSION renegotiation driver needs (design/shard-payload-reneg.md
|
||||
/// Phase 2) — `None` at [`spawn_watch`] makes the watcher observe-and-learn only (leg-1
|
||||
/// behavior). Constructed ONLY when the client's `Hello::max_shard_payload` advertised
|
||||
/// per-frame geometry AND the session's wire is not chunk-aligned: a PyroWave client parses
|
||||
/// chunk-aligned AUs in windows of the `Welcome` value pinned at session start (Apple
|
||||
/// `Stage2Pipeline` / `pf-client-core` video.rs read it once over the C ABI), so re-keying
|
||||
/// such a session mid-stream would corrupt its parse — those sessions keep the leg-1
|
||||
/// next-session clamp instead.
|
||||
pub(super) struct ShardReneg {
|
||||
/// The client's advertised receive ceiling (bytes of shard; > 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<u16>,
|
||||
/// ← control task: the client's `ShardPayloadAck`s (the grow gate).
|
||||
pub ack_rx: tokio::sync::mpsc::UnboundedReceiver<u16>,
|
||||
/// → 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<usize>,
|
||||
}
|
||||
|
||||
/// 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<usize>, learned_udp_budget: Option<u16>, 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<ShardReneg>,
|
||||
) {
|
||||
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 <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."
|
||||
);
|
||||
// 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 <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."
|
||||
);
|
||||
// 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;
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user