diff --git a/clients/android/native/src/decode/mod.rs b/clients/android/native/src/decode/mod.rs index a96e4375..982f560b 100644 --- a/clients/android/native/src/decode/mod.rs +++ b/clients/android/native/src/decode/mod.rs @@ -91,7 +91,14 @@ const NO_VIDEO_PATIENCE: std::time::Duration = std::time::Duration::from_millis( /// Re-ask cadence once [`NO_VIDEO_PATIENCE`] has elapsed with still nothing received. Slow, because /// this state is either self-healing on the first ask or not ours to heal — and each pass logs. -const NO_VIDEO_RETRY: std::time::Duration = std::time::Duration::from_millis(2000); +/// +/// ⚠ Taken from core, NOT a local number. `FLUSH_COOLDOWN` (the jump-to-live rate limit) is 2000 ms, +/// and the host classifies a keyframe-recovery cadence by matching a cooldown's period ±10 % to +/// decide WHICH client failure it is looking at. The two are opposites — "I have received nothing" +/// versus "I am drowning in frames I cannot drain" — so while this was also 2000 ms the host +/// confidently reported the wrong one, and a black-screen field case was diagnosed as a slow decoder +/// for days (2026-08-20). Keeping the value in core is what stops the two drifting back together. +const NO_VIDEO_RETRY: std::time::Duration = punktfunk_core::client::NO_VIDEO_RETRY; /// Whether low-latency mode uses the event-driven async decode loop (default) or the synchronous /// poll loop. Flip to `false` to A/B the two on the HUD (`design/…`); the async loop presents a diff --git a/clients/probe/src/main.rs b/clients/probe/src/main.rs index 2e2496a7..ab3d0d07 100644 --- a/clients/probe/src/main.rs +++ b/clients/probe/src/main.rs @@ -53,9 +53,9 @@ use punktfunk_core::config::Role; use punktfunk_core::input::{InputEvent, InputKind}; use punktfunk_core::packet::FLAG_PROBE; use punktfunk_core::quic::{ - endpoint, io, window_loss_ppm, BitrateChanged, CursorRenderMode, Hello, LossReport, - ProbeRequest, ProbeResult, Reconfigure, Reconfigured, RequestKeyframe, SetBitrate, Start, - Welcome, + endpoint, io, window_loss_ppm, BitrateChanged, CursorRenderMode, DeliveryReport, Hello, + LossReport, ProbeRequest, ProbeResult, Reconfigure, Reconfigured, RequestKeyframe, SetBitrate, + Start, Welcome, }; use punktfunk_core::transport::UdpTransport; use punktfunk_core::{CompositorPref, Mode, PunktfunkError, Session}; @@ -987,10 +987,18 @@ async fn session(args: Args) -> Result<()> { let mut ls = send; let lp = loss_ppm.clone(); let df = dropped_frames.clone(); + // Delivery truth for the host's dead-data-plane check: report what actually landed on the + // wire, so the probe reproduces a real client's answer rather than the "cannot answer" + // sentinel — which is exactly what makes it usable for testing that path. + let rxp = rx_wire_packets.clone(); tokio::spawn(async move { use std::sync::atomic::Ordering::Relaxed; let mut last_report = std::time::Instant::now(); let mut last_dropped = 0u64; + // Mirrors the real clients' rule (see `pump/data.rs`): report the delivery count every + // window while it is zero, once when the first packets land, then stop — so a host that + // predates the message is not flooded with "unknown control message" on a good session. + let mut delivery_confirmed = false; loop { tokio::time::sleep(std::time::Duration::from_millis(100)).await; let d = df.load(Relaxed); @@ -1007,6 +1015,25 @@ async fn session(args: Args) -> Result<()> { if last_report.elapsed() >= std::time::Duration::from_millis(750) { last_report = std::time::Instant::now(); let v = lp.swap(u32::MAX, Relaxed); + // Independent of whether there is a fresh loss sample: "no fresh sample" is + // exactly the shape a dead data plane has, so gating it on one would silence + // it in the state it exists to report. + let received = rxp.load(Relaxed); + if received == 0 || !delivery_confirmed { + delivery_confirmed = received > 0; + if io::write_msg( + &mut ls, + &DeliveryReport { + packets_received: received, + } + .encode(), + ) + .await + .is_err() + { + break; // control stream gone + } + } if v != u32::MAX && io::write_msg(&mut ls, &LossReport { loss_ppm: v }.encode()) .await diff --git a/crates/punktfunk-core/cbindgen.toml b/crates/punktfunk-core/cbindgen.toml index 2a181f83..e1828bd2 100644 --- a/crates/punktfunk-core/cbindgen.toml +++ b/crates/punktfunk-core/cbindgen.toml @@ -215,6 +215,7 @@ include = ["PunktfunkEndReason"] "MSG_CLOCK_PROBE" = "PUNKTFUNK_MSG_CLOCK_PROBE" "MSG_CURSOR_RENDER" = "PUNKTFUNK_MSG_CURSOR_RENDER" "MSG_CURSOR_SHAPE" = "PUNKTFUNK_MSG_CURSOR_SHAPE" +"MSG_DELIVERY_REPORT" = "PUNKTFUNK_MSG_DELIVERY_REPORT" "MSG_LOSS_REPORT" = "PUNKTFUNK_MSG_LOSS_REPORT" "MSG_PAIR_CHALLENGE" = "PUNKTFUNK_MSG_PAIR_CHALLENGE" "MSG_PAIR_PROOF" = "PUNKTFUNK_MSG_PAIR_PROOF" diff --git a/crates/punktfunk-core/src/client/control.rs b/crates/punktfunk-core/src/client/control.rs index 140f9196..b0758641 100644 --- a/crates/punktfunk-core/src/client/control.rs +++ b/crates/punktfunk-core/src/client/control.rs @@ -1,7 +1,9 @@ //! `CtrlRequest` (the embedder's control-stream requests) and `Negotiated` (the handshake result). use crate::config::{CompositorPref, GamepadPref, Mode}; -use crate::quic::{ClipControl, ClipOffer, ColorInfo, LossReport, ProbeRequest, RfiRequest}; +use crate::quic::{ + ClipControl, ClipOffer, ColorInfo, DeliveryReport, LossReport, ProbeRequest, RfiRequest, +}; /// A control-stream request the embedder makes on the open handshake stream: a mode switch or a /// speed test. One outbound channel carries both so the worker's `select!` has a single writer @@ -15,6 +17,10 @@ pub(crate) enum CtrlRequest { /// forcing a full IDR. See [`RfiRequest`]. Rfi(RfiRequest), Loss(LossReport), + /// How many data-plane packets have reached us all session — sent straight after every + /// [`CtrlRequest::Loss`], because `loss_ppm` is ambiguous at zero (no loss and no packets look + /// identical) and only this separates them. See [`DeliveryReport`]. + Delivery(DeliveryReport), /// Adaptive bitrate: ask the host to re-target its encoder (kbps). Sent by the pump's /// [`BitrateController`] when the user's bitrate setting is Automatic. SetBitrate(u32), diff --git a/crates/punktfunk-core/src/client/frame_channel.rs b/crates/punktfunk-core/src/client/frame_channel.rs index 47d7db24..f8331538 100644 --- a/crates/punktfunk-core/src/client/frame_channel.rs +++ b/crates/punktfunk-core/src/client/frame_channel.rs @@ -57,6 +57,21 @@ pub(crate) const FLUSH_AFTER: Duration = Duration::from_millis(250); /// the number, so the two can never drift apart. pub const FLUSH_COOLDOWN: Duration = Duration::from_secs(2); +/// Spacing of a client's keyframe re-asks while it has received **no video at all** — the other +/// reason a client asks on a perfectly fixed cadence, and the OPPOSITE fault to [`FLUSH_COOLDOWN`]'s +/// (nothing arriving, versus more arriving than it can drain). +/// +/// **Public, and deliberately a different value, for the same reason [`FLUSH_COOLDOWN`] is public.** +/// While both were 2000 ms the host's recovery-cadence detector could not tell which failure it was +/// looking at, and reported the confident wrong one: a 2026-08-20 field case where not one byte of +/// video ever reached the client was diagnosed for days as a client too slow to keep up. Embedders +/// own the no-video timer (it lives in each decode loop), so this is the value they must use — a +/// local copy is exactly the drift that made the two indistinguishable in the first place. +/// +/// The delivery count on [`crate::quic::LossReport`] settles it outright for clients new enough to +/// send one; this keeps the period itself informative for those that are not. +pub const NO_VIDEO_RETRY: Duration = Duration::from_millis(2600); + /// A clock-triggered jump-to-live that discarded fewer datagrams than this (and no queued AUs) /// found NO local backlog: the frames read as late, but nothing here was actually behind. Two /// causes, and flushing helps neither: a **wall-clock step** (NTP mid-session on either end) diff --git a/crates/punktfunk-core/src/client/mod.rs b/crates/punktfunk-core/src/client/mod.rs index fe58f819..f7a11517 100644 --- a/crates/punktfunk-core/src/client/mod.rs +++ b/crates/punktfunk-core/src/client/mod.rs @@ -42,7 +42,7 @@ mod recovery; mod rumble; mod worker; -pub use self::frame_channel::FLUSH_COOLDOWN; +pub use self::frame_channel::{FLUSH_COOLDOWN, NO_VIDEO_RETRY}; pub use self::planes::AudioPacket; pub use self::probe::ProbeOutcome; pub use self::rumble::{ActuatorQuirks, RumbleCommand}; diff --git a/crates/punktfunk-core/src/client/pump.rs b/crates/punktfunk-core/src/client/pump.rs index 4718eaf6..922183be 100644 --- a/crates/punktfunk-core/src/client/pump.rs +++ b/crates/punktfunk-core/src/client/pump.rs @@ -11,9 +11,9 @@ use crate::abr::BitrateController; use crate::config::Role; use crate::packet::FLAG_PROBE; use crate::quic::{ - io, wall_clock_ns, window_loss_ppm, BitrateChanged, ClipState, ClockEcho, ClockResync, Hello, - LossReport, ProbeResult, Reconfigure, Reconfigured, RequestKeyframe, ResyncAdmit, ResyncGuard, - ResyncStep, SetBitrate, Start, Welcome, + io, wall_clock_ns, window_loss_ppm, BitrateChanged, ClipState, ClockEcho, ClockResync, + DeliveryReport, Hello, LossReport, ProbeResult, Reconfigure, Reconfigured, RequestKeyframe, + ResyncAdmit, ResyncGuard, ResyncStep, SetBitrate, Start, Welcome, }; use crate::session::Session; use crate::transport::UdpTransport; diff --git a/crates/punktfunk-core/src/client/pump/control_task.rs b/crates/punktfunk-core/src/client/pump/control_task.rs index b4c7a0b8..6b29f26c 100644 --- a/crates/punktfunk-core/src/client/pump/control_task.rs +++ b/crates/punktfunk-core/src/client/pump/control_task.rs @@ -107,6 +107,7 @@ impl ControlTask { } CtrlRequest::Rfi(r) => r.encode(), CtrlRequest::Loss(r) => r.encode(), + CtrlRequest::Delivery(r) => r.encode(), CtrlRequest::SetBitrate(k) => SetBitrate { bitrate_kbps: k }.encode(), CtrlRequest::ClockResync => { if clock_rtt_ns.is_none() { diff --git a/crates/punktfunk-core/src/client/pump/data.rs b/crates/punktfunk-core/src/client/pump/data.rs index a844b8d8..3181de57 100644 --- a/crates/punktfunk-core/src/client/pump/data.rs +++ b/crates/punktfunk-core/src/client/pump/data.rs @@ -77,6 +77,12 @@ impl DataPump { // size FEC to the link. Suppressed during a speed test (its FLAG_PROBE filler would skew it). const ADAPT_REPORT_INTERVAL: Duration = Duration::from_millis(750); let mut last_report = Instant::now(); + // Has the host been told, once, that data-plane packets are reaching us? See the send site: + // the delivery count is reported every window while it is ZERO (the state the host acts on) + // and once more when the first packets land, then never again. A host that predates the + // message logs "unknown control message" for each one, so a healthy session must not stream + // them — one line per session is a fair price on an old host, eighty a minute is not. + let mut delivery_confirmed = false; let ( mut last_recovered, mut last_late, @@ -415,6 +421,27 @@ impl DataPump { ); } else { let _ = ctrl_tx.try_send(CtrlRequest::Loss(LossReport { loss_ppm })); + // Rides with the loss report — it is what makes `loss_ppm = 0` readable at the + // host, which cannot otherwise tell a flawless link from one delivering + // nothing. The session TOTAL, not this window's, so one message stands on its + // own. Deliberately inside the same arm: a discarded window is discarded + // because the host was rebuilding or a probe distorted it, and staying silent + // there keeps that contract exact. Nothing is lost — the state this reports + // (no packets at all) produces no discards, so its windows always send. + // + // Sent every window while the count is ZERO, then ONCE when the first packets + // land (so the host stops guessing and can name the other failure confidently), + // then never again: a healthy session must not stream a message that older + // hosts log as unknown on every arrival. + // ponytail: only start-of-session death is covered. A path that dies MID-stream + // leaves the count frozen above zero and silent, which the host still reads as + // healthy — detecting that needs a stalled-counter check with its own timing, + // worth adding if a mid-session case is ever reported. + if should_report_delivery(st.packets_received, &mut delivery_confirmed) { + let _ = ctrl_tx.try_send(CtrlRequest::Delivery(DeliveryReport { + packets_received: st.packets_received, + })); + } } // Standing-latency bleed: close the detector's window with this report's loss // verdict and run its escalation ladder — re-sync first (free; a stale offset @@ -757,10 +784,58 @@ fn take_pipeline_gap(slot: &AtomicU32) -> Option { } } +/// Does this report window owe the host a [`DeliveryReport`], and record that it has been told? +/// +/// Every window while `packets_received` is ZERO — that is the state the host escalates on, and it +/// must keep hearing it — then exactly ONCE more when the first packets land, so the host learns +/// delivery works and can stop hedging its stall diagnosis. Silent after that: a host that predates +/// the message logs every unknown control message, and a healthy hours-long session must not fill +/// its log with them. +fn should_report_delivery(packets_received: u64, confirmed: &mut bool) -> bool { + let owed = packets_received == 0 || !*confirmed; + *confirmed = packets_received > 0; + owed +} + #[cfg(test)] mod tests { use super::*; + /// The host must keep hearing "zero" for as long as it is true (that is the black-screen + /// signal), get exactly one confirmation when video starts, and then silence — the noise budget + /// on an older host, which warns per unknown message, is what pays for the first two. + #[test] + fn the_delivery_count_is_reported_while_zero_then_once_more_and_never_again() { + let mut confirmed = false; + // Nothing arriving: reported every window, for as long as it stays true. + for _ in 0..5 { + assert!( + should_report_delivery(0, &mut confirmed), + "a dead data plane must be re-reported every window" + ); + } + // First packets land: one confirmation, so the host can name the other failure confidently. + assert!(should_report_delivery(500, &mut confirmed)); + // Healthy from here: silent. + for n in [900, 1_200, 90_000] { + assert!( + !should_report_delivery(n, &mut confirmed), + "a healthy session must not stream delivery reports" + ); + } + } + + /// A session that never receives anything must never look confirmed, no matter how long it runs + /// — the whole point is that the host keeps being told. + #[test] + fn a_session_that_receives_nothing_never_reports_itself_healthy() { + let mut confirmed = false; + for _ in 0..100 { + assert!(should_report_delivery(0, &mut confirmed)); + assert!(!confirmed); + } + } + #[test] fn a_pipeline_gap_is_taken_exactly_once() { let slot = AtomicU32::new(0); @@ -935,8 +1010,8 @@ mod tests { .expect("the window after the gap reports on schedule"); assert!( matches!(reported, Some(CtrlRequest::Loss(_))), - "the window after the gap must produce a loss report — an idle session's only \ - outbound request" + "the window after the gap must produce a loss report — the first of the two requests \ + an idle session makes (the delivery count follows it)" ); assert!( started.elapsed() >= Duration::from_millis(1_400), diff --git a/crates/punktfunk-core/src/quic/control.rs b/crates/punktfunk-core/src/quic/control.rs index 9c978d72..8941af14 100644 --- a/crates/punktfunk-core/src/quic/control.rs +++ b/crates/punktfunk-core/src/quic/control.rs @@ -97,6 +97,33 @@ pub struct LossReport { pub loss_ppm: u32, } +/// `client → host`, sent immediately after each [`LossReport`]: data-plane packets this client has +/// received all session, cumulative. +/// +/// ⚠ Exists because `loss_ppm` alone is **ambiguous at zero**: a client receiving a flawless stream +/// and a client receiving *nothing at all* both report `loss_ppm = 0` — loss is a ratio over a +/// window whose denominator is the packets that arrived, so no-packets is indistinguishable from +/// no-loss. That ambiguity let a host decay adaptive FEC to its floor while the client sat behind a +/// black screen having received zero bytes, and the host's own stall diagnosis blamed the client for +/// "not sustaining the stream" it had never been sent (field 2026-08-20: a Windows host whose +/// per-session data port was closed inbound, so the client's hole-punch never opened the return +/// path). `0` while the host has sent frames is the one unambiguous statement of "the video data +/// plane is not reaching me" — the control plane carrying this report is, by construction, healthy. +/// +/// ⚠ A SEPARATE MESSAGE rather than a field appended to [`LossReport`], and that is load-bearing: +/// `LossReport::decode` length-checks EXACTLY, so a longer report is rejected outright by every host +/// already shipped — a new client would silently lose adaptive FEC against them. Mixed versions are +/// normal here (the field case that motivated this ran a current host against a months-old client), +/// so the compatible shape is a new type byte an older host simply ignores, exactly as it already +/// ignores every other control message it predates. +/// +/// Cumulative, not per-window, so a single message is self-contained; `u64` to match the counter it +/// mirrors, with no saturation to reason about. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct DeliveryReport { + pub packets_received: u64, +} + /// `client → host`, any time after [`Start`]: reconfigure the encoder to a new target bitrate /// without reconnecting — the mid-stream lever of adaptive bitrate. The host clamps the request /// exactly like [`Hello::bitrate_kbps`] (its `[MIN, MAX]` band; `0` → host default), answers with @@ -270,6 +297,8 @@ pub const MSG_SHARD_PAYLOAD_ACK: u8 = 0x09; /// and [`BitrateChanged`] already feed. Deliberately NOT in the 0x30 clock block — it carries a /// duration precisely so that no clock domain is involved. pub const MSG_PIPELINE_GAP: u8 = 0x0A; +/// Type byte of [`DeliveryReport`]. +pub const MSG_DELIVERY_REPORT: u8 = 0x0B; /// Type byte of [`ProbeRequest`]. pub const MSG_PROBE_REQUEST: u8 = 0x20; /// Type byte of [`ProbeResult`]. @@ -436,6 +465,26 @@ impl LossReport { } } +impl DeliveryReport { + pub fn encode(&self) -> Vec { + // magic[0..4] type[4] packets_received[5..13] + let mut b = Vec::with_capacity(13); + b.extend_from_slice(CTL_MAGIC); + b.push(MSG_DELIVERY_REPORT); + b.extend_from_slice(&self.packets_received.to_le_bytes()); + b + } + + pub fn decode(b: &[u8]) -> Result { + if b.len() != 13 || &b[0..4] != CTL_MAGIC || b[4] != MSG_DELIVERY_REPORT { + return Err(PunktfunkError::InvalidArg("bad DeliveryReport")); + } + Ok(DeliveryReport { + packets_received: u64::from_le_bytes(b[5..13].try_into().unwrap()), + }) + } +} + impl SetBitrate { pub fn encode(&self) -> Vec { // magic[0..4] type[4] bitrate_kbps[5..9] @@ -1291,6 +1340,41 @@ mod tests { .is_err()); } + #[test] + fn delivery_report_roundtrip() { + for packets_received in [0u64, 1, 9_999, u32::MAX as u64 + 1, u64::MAX] { + let r = DeliveryReport { packets_received }; + assert_eq!(DeliveryReport::decode(&r.encode()).unwrap(), r); + } + assert!(DeliveryReport::decode(&RequestKeyframe.encode()).is_err()); + assert!(DeliveryReport::decode(&LossReport { loss_ppm: 0 }.encode()).is_err()); + } + + /// The delivery count MUST NOT ride on [`LossReport`]: that message is length-checked EXACTLY, + /// so lengthening it would make every already-shipped host reject the loss reports its adaptive + /// FEC runs on — a silent regression for a new client against an old host, which is the normal + /// mixed-version case here (the field report that motivated this ran a current host against a + /// months-old client). Its own type byte keeps `LossReport` byte-identical while an older host + /// simply ignores the message it does not know. + #[test] + fn the_delivery_count_does_not_disturb_the_loss_report_wire_form() { + let loss = LossReport { loss_ppm: 42 }.encode(); + assert_eq!(loss.len(), 9, "LossReport must stay the 9-byte wire form"); + assert_eq!(loss[4], MSG_LOSS_REPORT); + + let delivery = DeliveryReport { + packets_received: 0, + } + .encode(); + assert_ne!( + delivery[4], MSG_LOSS_REPORT, + "a distinct type byte is what makes an old host ignore it instead of failing" + ); + // Neither can be silently mis-parsed as the other. + assert!(LossReport::decode(&delivery).is_err()); + assert!(DeliveryReport::decode(&loss).is_err()); + } + #[test] fn window_loss_ppm_estimates_and_caps() { // No traffic → 0. A clean window (nothing recovered) → 0. diff --git a/crates/punktfunk-host/src/native.rs b/crates/punktfunk-host/src/native.rs index 90a202f8..62442bbd 100644 --- a/crates/punktfunk-host/src/native.rs +++ b/crates/punktfunk-host/src/native.rs @@ -1404,6 +1404,12 @@ async fn serve_session( // evidence (a refusal without the score left a 23-minute floor-pinned field session with no // trace of why). let cadence_behind_score = Arc::new(AtomicU32::new(0)); + // Delivery truth, control task → data plane: the packet count the client reports having + // received all session (`u32::MAX` until a client new enough to answer sends one). The data + // plane needs it to tell a clean link from a dead one — `loss_ppm = 0` means both — before it + // blames the client for a stream that never reached it. + let client_packets_received = Arc::new(AtomicU32::new(u32::MAX)); + let client_packets_received_ctl = client_packets_received.clone(); let (probe_tx, probe_rx) = std::sync::mpsc::channel::(); let (probe_result_tx, probe_result_rx) = tokio::sync::mpsc::unbounded_channel::(); // Mode-switch outcome, data plane → control task (same pattern as `probe_result_tx`): the accept @@ -1535,6 +1541,7 @@ async fn serve_session( encoder_ceiling_kbps.clone(), cadence_degraded.clone(), cadence_behind_score.clone(), + client_packets_received_ctl, fec_target_ctl, phase_ctl_control, reconfig_tx, @@ -2093,6 +2100,26 @@ async fn serve_session( address with no hole-punch; else punched=true → the client's observed source, \ false → no punch seen, the reported address)" ); + // A punch that never arrives is not a routine fallback — it is the fingerprint of a + // data port the client cannot reach INBOUND, and every client punches (5/s for the + // first three seconds, then every two). Video then goes to an address the client only + // CLAIMED, unverified, and if anything on the path needed the flow opened client-first + // it silently goes nowhere: black picture, healthy control plane, no error anywhere. + // On Windows the usual cause is a firewall rule that opens fixed ports only, while + // this port is ephemeral and different every session (fixed by the program-scoped rule + // `service install` now adds — an install predating it still has the old rules). + // `direct` skips the punch by operator choice, so it is not a failure there. + if !direct && !punched { + tracing::warn!( + %client_udp, + udp_port, + "no hole-punch reached this host's data port — inbound UDP to it looks \ + BLOCKED, so video is being sent to the address the client reported without \ + any confirmed return path. If the picture stays black while the session is \ + otherwise healthy, this line is the reason: allow inbound UDP for the host \ + executable (any port), or pin --data-port and open that one" + ); + } let mut session = Session::new(cfg, Box::new(transport)) .map_err(|e| anyhow!("host session: {e:?}"))?; match source { @@ -2127,6 +2154,7 @@ async fn serve_session( encoder_ceiling_kbps, cadence_degraded, cadence_behind_score, + client_packets_received, bitrate_auto, bit_depth, chroma, diff --git a/crates/punktfunk-host/src/native/control.rs b/crates/punktfunk-host/src/native/control.rs index 055558fb..5041323c 100644 --- a/crates/punktfunk-host/src/native/control.rs +++ b/crates/punktfunk-host/src/native/control.rs @@ -30,6 +30,10 @@ pub(super) async fn run( encoder_ceiling_kbps: Arc, cadence_degraded: Arc, cadence_behind_score: Arc, + // Delivery truth, published from every `DeliveryReport` for the data plane's stall diagnosis: + // the packets the client says it has received all session (`u32::MAX` = a client too old to + // send one, the pre-seeded value). + client_packets_received: Arc, fec_target_ctl: Arc, // Phase-locked capture bridge: client PhaseReports land here latest-wins; the encode loop's // controller drains at its own ~1 Hz cadence (design/phase-locked-capture.md). @@ -162,6 +166,16 @@ pub(super) async fn run( if rfi_tx.send((req.first_frame, req.last_frame)).is_err() { break; // data plane gone } + } else if let Ok(rep) = punktfunk_core::quic::DeliveryReport::decode(&msg) { + // What the client has actually RECEIVED — published unconditionally, because it + // is what lets the data plane read `loss_ppm = 0` correctly and must survive + // both the `adaptive_fec` opt-out and a pinned FEC percentage (a host with + // PUNKTFUNK_FEC_PCT set is exactly as blind to a dead data plane otherwise). + // Saturated into the u32 bridge; the value only ever matters near zero. + client_packets_received.store( + rep.packets_received.min(u32::MAX as u64 - 1) as u32, + Ordering::Relaxed, + ); } else if let Ok(rep) = LossReport::decode(&msg) { // Adaptive FEC: size recovery to the loss the client is seeing. The data-plane // send loop reads `fec_target_ctl` and applies it per frame. Ignored when FEC diff --git a/crates/punktfunk-host/src/native/stream.rs b/crates/punktfunk-host/src/native/stream.rs index 293b409d..a89b0e08 100644 --- a/crates/punktfunk-host/src/native/stream.rs +++ b/crates/punktfunk-host/src/native/stream.rs @@ -1319,6 +1319,14 @@ pub(super) struct SessionContext { /// of what held it there — the score is the missing discriminator between "the detector's /// budget is wrong" and "this encoder genuinely can't hold cadence"). pub(super) cadence_behind_score: Arc, + /// Data-plane packets the CLIENT says it has received all session, from the latest + /// [`punktfunk_core::quic::DeliveryReport`] ([`u32::MAX`] = a client too old to send one). + /// + /// The one signal that distinguishes "the link is clean" from "nothing is arriving": both look + /// like `loss_ppm = 0`, because loss is a ratio over the packets that DID arrive. Read by the + /// keyframe-cadence diagnosis below, which without it accuses the client of being too slow for + /// a stream it has never received a byte of. + pub(super) client_packets_received: Arc, /// The client asked for "Automatic" (`Hello::bitrate_kbps == 0`), so `bitrate_kbps` came from /// the host's codec-aware default. For PyroWave that default is the ~1.6 bpp operating point of /// the NEGOTIATED MODE (`resolve_bitrate_kbps_for`) — a mid-stream mode switch re-resolves it @@ -1598,6 +1606,7 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option bool { period.abs_diff(flush) < flush / 10 } +/// The client's OTHER re-ask cooldown: it has received no video whatsoever and is asking for a +/// keyframe on its no-video timer. Kept separate from [`matches_client_flush_cadence`] because the +/// two describe opposite faults — drowning in frames versus receiving none — and only the client's +/// reported delivery count can say which. Both are host-side-irrelevant either way: a fixed +/// software cooldown is never the periodic *disturbance* the metronomic branch reports. +/// +/// Compared against the SHARED constant, never a copy of the number — the same discipline +/// [`matches_client_flush_cadence`] follows, and the one that was missing when the two cooldowns +/// were both 2000 ms and the host could not even tell that it was guessing. +fn matches_client_no_video_cadence(period: std::time::Duration) -> bool { + let no_video = punktfunk_core::client::NO_VIDEO_RETRY; + period.abs_diff(no_video) < no_video / 10 +} + +/// Either client cooldown — the band in which a period tells us about the CLIENT's software, not +/// about anything physical on this host. +fn matches_client_recovery_cooldown(period: std::time::Duration) -> bool { + matches_client_flush_cadence(period) || matches_client_no_video_cadence(period) +} + /// One mode's capture/encode pipeline: (capturer, encoder, first frame, frame interval). /// Dropping the capturer tears down the PipeWire stream and the virtual output with it. type Pipeline = ( @@ -5068,6 +5146,29 @@ mod tests { assert!(!matches_client_flush_cadence(std::time::Duration::ZERO)); } + /// The two client cooldowns must stay TELLABLE APART by period, and both must stay out of the + /// display-disturbance branch. While they were both 2000 ms a black-screen field case (nothing + /// ever reached the client) was reported as "the client cannot sustain the stream" — the exact + /// opposite fault — because the periods were identical and the host guessed. + #[test] + fn the_two_client_cooldowns_are_distinguishable_and_both_excluded_from_display_blame() { + let flush = punktfunk_core::client::FLUSH_COOLDOWN; + let no_video = punktfunk_core::client::NO_VIDEO_RETRY; + assert_ne!( + flush, no_video, + "identical cooldowns make the host's verdict a coin flip" + ); + // Neither may fall inside the other's ±10% band, or the period stops discriminating. + assert!(!matches_client_flush_cadence(no_video)); + assert!(!matches_client_no_video_cadence(flush)); + // Both are client software cooldowns: never the metronomic display-disturbance branch. + assert!(matches_client_recovery_cooldown(flush)); + assert!(matches_client_recovery_cooldown(no_video)); + // A real periodic disturbance still reaches that branch. + assert!(!matches_client_recovery_cooldown(flush * 3)); + assert!(!matches_client_recovery_cooldown(std::time::Duration::ZERO)); + } + #[test] fn an_escalated_but_caught_up_encoder_stops_refusing_climbs() { const DEGRADE: u32 = 10; diff --git a/crates/punktfunk-host/src/windows/service.rs b/crates/punktfunk-host/src/windows/service.rs index f5353415..67656ef5 100644 --- a/crates/punktfunk-host/src/windows/service.rs +++ b/crates/punktfunk-host/src/windows/service.rs @@ -1587,6 +1587,7 @@ fn add_firewall_rules(allow_public: bool) { eprintln!("warning: could not add firewall rule '{name}' (add it manually if needed)"); } } + add_data_plane_firewall_rule(profile); if !allow_public { println!( "Note: streaming ports are open on Private/Domain networks only. On a network Windows \ @@ -1596,7 +1597,75 @@ fn add_firewall_rules(allow_public: bool) { } } +/// Rule name for the program-scoped data-plane rule (see [`add_data_plane_firewall_rule`]). +const FW_DATA_PLANE_RULE: &str = "Punktfunk UDP (data plane)"; + +/// Inbound UDP for the host executable itself, at **any** local port. +/// +/// The media data plane binds an EPHEMERAL port per session (`0.0.0.0:0`, reported to the client in +/// the Welcome), so no `localport=` rule can cover it — the port-scoped rules above open the fixed +/// control/GameStream/mDNS ports and nothing else. Without this, Windows Firewall drops the client's +/// hole-punch (`PUNCH_MAGIC` → the host's data port) on EVERY session: that is what `punched=false` +/// on the host's "data plane bound" line means. The punch then never opens the return path, video +/// falls back to blind-sending at the address the client merely *reported*, and the moment anything +/// on the path needs the flow opened client-first the stream goes black while the control plane +/// stays healthy — no reconnect, no error, just a session that never shows a picture. +/// +/// Program-scoped rather than a pinned port: it covers whatever port the session picks, needs no +/// second rule when the range moves, and cannot collide with another host (a pinned data port in +/// 47998-48010 would land on Sunshine/Apollo's GameStream range). The port rules above are kept as +/// they are — an install whose recorded exe path later moves still has its fixed ports open. +fn add_data_plane_firewall_rule(profile: &str) { + let exe = match std::env::current_exe() { + Ok(p) => p, + Err(e) => { + eprintln!( + "warning: could not resolve the host executable path ({e}) — skipping the \ + data-plane firewall rule; streams may show a black picture behind a healthy \ + connection on networks that need the client's hole-punch to open the path" + ); + return; + } + }; + let ok = run_quiet( + "netsh", + &[ + "advfirewall", + "firewall", + "add", + "rule", + &format!("name={FW_DATA_PLANE_RULE}"), + "dir=in", + "action=allow", + "protocol=UDP", + &format!("program={}", exe.to_string_lossy()), + profile, + ], + ); + if ok { + println!( + "Firewall rule added: {FW_DATA_PLANE_RULE} (any UDP port for {}) [{profile}]", + exe.display() + ); + } else { + eprintln!( + "warning: could not add firewall rule '{FW_DATA_PLANE_RULE}' — the per-session video \ + data port stays closed to inbound, so the client's hole-punch cannot reach it" + ); + } +} + fn remove_firewall_rules() { + let _ = run_quiet( + "netsh", + &[ + "advfirewall", + "firewall", + "delete", + "rule", + &format!("name={FW_DATA_PLANE_RULE}"), + ], + ); for suffix in ["TCP", "UDP"] { // Capital P is the brand; netsh matches a rule name case-INSENSITIVELY, so this still // reaps the lowercase rules every release up to 0.22.1 created — no orphans on upgrade. diff --git a/include/punktfunk_core.h b/include/punktfunk_core.h index fda50dec..19d4a8c1 100644 --- a/include/punktfunk_core.h +++ b/include/punktfunk_core.h @@ -1224,6 +1224,11 @@ #define PUNKTFUNK_MSG_PIPELINE_GAP 10 #endif +#if defined(PUNKTFUNK_FEATURE_QUIC) +// Type byte of [`DeliveryReport`]. +#define PUNKTFUNK_MSG_DELIVERY_REPORT 11 +#endif + #if defined(PUNKTFUNK_FEATURE_QUIC) // Type byte of [`ProbeRequest`]. #define PUNKTFUNK_MSG_PROBE_REQUEST 32 diff --git a/scripts/ci/docs-undocumented-env-baseline.txt b/scripts/ci/docs-undocumented-env-baseline.txt index aeb35ce2..909c8941 100644 --- a/scripts/ci/docs-undocumented-env-baseline.txt +++ b/scripts/ci/docs-undocumented-env-baseline.txt @@ -180,6 +180,7 @@ PUNKTFUNK_MSG_CLOCK_ECHO PUNKTFUNK_MSG_CLOCK_PROBE PUNKTFUNK_MSG_CURSOR_RENDER PUNKTFUNK_MSG_CURSOR_SHAPE +PUNKTFUNK_MSG_DELIVERY_REPORT PUNKTFUNK_MSG_LOSS_REPORT PUNKTFUNK_MSG_PAIR_CHALLENGE PUNKTFUNK_MSG_PAIR_PROOF