diff --git a/crates/punktfunk-host/src/native.rs b/crates/punktfunk-host/src/native.rs index 7dad3025..f2cafd6a 100644 --- a/crates/punktfunk-host/src/native.rs +++ b/crates/punktfunk-host/src/native.rs @@ -1073,6 +1073,17 @@ async fn serve_session( // accepted ack as "the active mode is now X" and fixes itself; old clients just log it. let (reconfig_result_tx, reconfig_result_rx) = tokio::sync::mpsc::unbounded_channel::(); + // Unsolicited bitrate re-target, data plane → control task (the `reconfig_result_tx` pattern + // again, for the same reason). A pipeline rebuild can RE-RESOLVE an Automatic rate — most + // visibly when the source delivers a different size than the session negotiated, e.g. a + // client that asked for 1080p mirroring a 4K panel — and that number is what everything + // downstream reasons about: the send pacer, the console, and the base a `SetBitrate` ack is + // measured against. The client's copy only ever moved on an ack, so it stayed on the + // negotiated rate while the host encoded at another one, and the ABR's first climb computed + // from that stale base asked for LESS than the host was already sending — a re-target + // downward, with the rebuild it costs. Tell the client instead; `BitrateChanged` already + // means exactly this and old clients already handle one arriving unprompted. + let (retarget_tx, retarget_rx) = tokio::sync::mpsc::unbounded_channel::(); // Cursor-forward bridge (M2): the encode loop diffs each frame's cursor serial and hands // changed SHAPES here; the control task (the control stream's sole writer) sends them. // Same shape as `probe_result_tx`. Wired even when the channel wasn't negotiated — it @@ -1133,6 +1144,7 @@ async fn serve_session( probe_tx, probe_result_rx, reconfig_result_rx, + retarget_rx, cursor_shape_rx, cursor_client_draws, clip_enabled, @@ -1579,6 +1591,7 @@ async fn serve_session( probe_rx, probe_result_tx, reconfig_result_tx, + retarget_tx, fec_target: fec_target_dp, phase: phase_ctl, conn: conn_stream, diff --git a/crates/punktfunk-host/src/native/control.rs b/crates/punktfunk-host/src/native/control.rs index e8b89675..e14934c6 100644 --- a/crates/punktfunk-host/src/native/control.rs +++ b/crates/punktfunk-host/src/native/control.rs @@ -40,6 +40,9 @@ pub(super) async fn run( probe_tx: std::sync::mpsc::Sender, mut probe_result_rx: tokio::sync::mpsc::UnboundedReceiver, mut reconfig_result_rx: tokio::sync::mpsc::UnboundedReceiver, + // 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, mut cursor_shape_rx: tokio::sync::mpsc::UnboundedReceiver, cursor_client_draws: Arc, clip_enabled: Arc, @@ -338,6 +341,27 @@ pub(super) async fn run( None => clip_offer_closed = true, } } + retarget = retarget_rx.recv() => { + // A pipeline rebuild re-resolved the Automatic rate (see `retarget_tx`). Same + // message the `SetBitrate` path answers with — the client's controller treats + // any `BitrateChanged` as authoritative for what the encoder now targets, which + // is exactly right here: it IS what the encoder now targets, we just weren't + // asked. PyroWave reaches this too, and should: its rate is pinned against + // mid-stream RETARGETS, but a mode switch legitimately re-resolves the pin + // (~1.6 bpp for the new pixel rate) and the client's live-rate display is + // otherwise stuck on the old one. Its controller is off, so nothing acts on it. + let Some(kbps) = retarget else { break }; // data plane gone + tracing::info!( + kbps, + "encoder re-targeted by a pipeline rebuild — telling the client" + ); + if io::write_msg(&mut ctrl_send, &BitrateChanged { bitrate_kbps: kbps }.encode()) + .await + .is_err() + { + break; + } + } correction = reconfig_result_rx.recv() => { // H2 rollback/correction ack: the data plane reports the mode ACTUALLY live // after a rebuild that failed (stayed at the old mode) or that the backend diff --git a/crates/punktfunk-host/src/native/stream.rs b/crates/punktfunk-host/src/native/stream.rs index 0c263638..6bba243e 100644 --- a/crates/punktfunk-host/src/native/stream.rs +++ b/crates/punktfunk-host/src/native/stream.rs @@ -1214,6 +1214,9 @@ pub(super) struct SessionContext { /// `Reconfigured { accepted: true, mode: }` when a rebuild failed (stayed at /// the old mode) or the backend honored a different refresh than requested. pub(super) reconfig_result_tx: tokio::sync::mpsc::UnboundedSender, + /// Host-initiated bitrate re-target → control task → the client's `BitrateChanged`. Fired + /// by [`adopt_built_bitrate`] when a rebuild lands on a rate the client wasn't told about. + pub(super) retarget_tx: tokio::sync::mpsc::UnboundedSender, /// Adaptive-FEC target the control task updates from the client's loss reports. pub(super) fec_target: Arc, /// The QUIC control connection (carries host→client 0xCE source-HDR metadata mid-stream). @@ -1397,6 +1400,7 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option 1 || pipelined_active || deescalating; // Export "encode can't hold cadence" for the control task's climb refusal. - // An escalated session stays flagged even with the bucket drained: its climb - // headroom is spent, and letting climbs resume would saw against the + // An escalated session is held to a stricter standard — ANY net behind-frame + // keeps it flagged, where an unescalated one is given the full bucket — because + // its climb headroom really is partly spent and a climb would saw against the // escalation and starve the de-escalation clean run below. + // + // But being escalated cannot flag it BY ITSELF, which is what this used to do. + // The client can't tell a transient refusal from an encoder's real ceiling: two + // identical short acks latch a cap, so a session that escalated once — the + // bucket needs ~20 net misses, which a startup hitch supplies while the ABR is + // still in slow start at the 20 Mbps default — got pinned there, and stayed + // pinned long after the escalation had bought back the headroom it was for. + // Escalating exists precisely so cadence CAN be held; once it is (bucket + // drained, every frame on time), refusing climbs is refusing the thing that + // worked. cadence_degraded.store( - escalated || behind_score >= DEPTH_DEGRADE, + encode_behind_cadence(escalated, behind_score, DEPTH_DEGRADE), Ordering::Relaxed, ); if deescalating { @@ -4068,13 +4106,42 @@ impl PaceBudget { } } +/// Does the encoder currently fail to hold the frame cadence? Exported to the control task, which +/// refuses bitrate CLIMBS while it is true (descents always pass — they are the cure). +/// +/// `escalated` = the session has already spent an adaptive-depth / pipelined-retrieve step to buy +/// headroom; `behind_score` is the leaky bucket of frames whose work overran the cadence deadline. +/// An escalated session is judged strictly — ANY net behind-frame keeps it flagged — but being +/// escalated does not flag it on its own. That distinction is the whole point: the client cannot +/// tell a transient refusal from an encoder's hard ceiling (two identical short acks latch a cap), +/// so "escalated ⇒ degraded, permanently" pinned Automatic sessions at whatever rate they happened +/// to hold when a startup hitch escalated them — routinely the 20 Mbps default, while slow start +/// had barely begun. Escalation exists so cadence CAN be held; once it is, refusing climbs refuses +/// the thing that worked. +fn encode_behind_cadence(escalated: bool, behind_score: u32, degrade_at: u32) -> bool { + behind_score >= degrade_at || (escalated && behind_score > 0) +} + /// Adopt the rate a freshly built pipeline's encoder was actually opened at. /// /// The session's own `bitrate_kbps` is the number every later decision reads — the ABR controller's /// climb base, the console's sample, what a `SetBitrate` ack is measured against — so letting it /// disagree with the live encoder means each of those reasons about a stream that doesn't exist. /// Silent when nothing changed, which is the overwhelmingly common case. -fn adopt_built_bitrate(current: &mut u32, built: u32, live: &Arc) { +/// +/// The client keeps its OWN copy of that number, and it used to move only on an ack — so a +/// rebuild that re-resolved an Automatic rate (`build_pipeline` does, whenever the source +/// delivers a size the session did not negotiate) left the two disagreeing for the rest of the +/// session. The ABR's next climb then computed from the stale base and asked for a rate BELOW +/// what the host was already sending: a re-target downward, paying an encoder rebuild to get +/// there. So tell the client too — `BitrateChanged` is the same message the `SetBitrate` path +/// answers with, and means the same thing arriving unprompted. +fn adopt_built_bitrate( + current: &mut u32, + built: u32, + live: &Arc, + retarget: &tokio::sync::mpsc::UnboundedSender, +) { if built == *current { return; } @@ -4085,6 +4152,7 @@ fn adopt_built_bitrate(current: &mut u32, built: u32, live: &Arc) { ); *current = built; live.store(built, Ordering::Relaxed); + let _ = retarget.send(built); // control task gone ⇒ the session is ending anyway } /// Encode-stall recovery: rebuild the encoder in place (keeping capture + the session up) and @@ -4329,6 +4397,38 @@ fn build_pipeline( mod tests { use super::*; + #[test] + fn an_escalated_but_caught_up_encoder_stops_refusing_climbs() { + const DEGRADE: u32 = 10; + // Not escalated: the full bucket is allowed before climbs are refused. + assert!(!encode_behind_cadence(false, 0, DEGRADE)); + assert!(!encode_behind_cadence(false, 9, DEGRADE)); + assert!(encode_behind_cadence(false, 10, DEGRADE)); + // Escalated and still missing deadlines: strict — one net behind-frame is enough. + assert!(encode_behind_cadence(true, 1, DEGRADE)); + // Escalated, bucket fully drained: cadence is being HELD, which is what escalating was + // for. This is the case that used to stay latched for the rest of the session and pin an + // Automatic client at its slow-start rate. + assert!(!encode_behind_cadence(true, 0, DEGRADE)); + } + + #[test] + fn adopting_a_rebuilt_rate_tells_the_client() { + let live = Arc::new(AtomicU32::new(20_000)); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::(); + let mut current = 20_000; + // The overwhelmingly common case: the rebuild landed on the same rate — silent. + adopt_built_bitrate(&mut current, 20_000, &live, &tx); + assert_eq!(rx.try_recv().ok(), None); + // A re-resolve (the client asked 1080p, the source delivers a mirrored 4K panel): the + // host's rate moves, so the client has to hear about it — its controller's climb base is + // its own copy of this number, and a stale one makes the next "climb" a cut. + adopt_built_bitrate(&mut current, 60_000, &live, &tx); + assert_eq!(current, 60_000); + assert_eq!(live.load(Ordering::Relaxed), 60_000); + assert_eq!(rx.try_recv().ok(), Some(60_000)); + } + #[test] fn pacing_never_exceeds_the_session_rate_or_the_display() { // Backend honored the request exactly (the multiplier off): pace at it.