From 809c37b3553753f387f7d830b99f9e3a01cc1fb6 Mon Sep 17 00:00:00 2001 From: DatCaptainHorse Date: Fri, 16 Oct 2026 20:33:33 +0300 Subject: [PATCH] feat(neshub): decide the video rate per frame from the path, not on thresholds Hub side of nescore's 2026-10-16 video-rate spec. nesprotocol: a per-frame feedback datagram (kind 3, client to hub on the video connection) carrying bytes, fragments, send time, first-fragment arrival, arrival spread and first-fragment queueing delay. The hub stats breakdown gains measured capacity and delivered share. neshub: - rate.rs, a per-client path model. Capacity is measured from each frame's train dispersion. The bottleneck counts as busy when the next train lands no later than its first fragment's own serialisation after the last one, and as overloaded when, during a busy run, the first fragments' queue grows and more is sent than delivered. Overload drops the rate at once to drain at the rate the queue built, then settles at the measured share. Steps up go to the measured capacity, or the highest rate the path has proven quiet at for longer than its own longest gap between overloads at or below it. Silence holds at the last share. Tested against a modelled FIFO bottleneck: above the path, below it, CBR keyframes, a capacity drop and recovery, recurring cross traffic. - control.rs: the slowest client's path decides, per frame as feedback arrives. Gone: the loss thresholds, back-off factor, climb fraction, deadband and queue thresholds. Ceiling, floor, manual and constant quality behave as before. - congestion.rs: an endpoint congestion controller that holds nothing back (window above u32::MAX, where noq stops pacing), so frames leave as trains for the client to measure. media-transport.md now says the hub turns transport pacing off, and why. The client does not send feedback yet (nescore step 2). Until it does, the rate stays where the session started. Co-Authored-By: Claude Opus 5.5 --- apps/neshub/src/congestion.rs | 67 +++ apps/neshub/src/control.rs | 870 +++++++---------------------- apps/neshub/src/dgram.rs | 4 +- apps/neshub/src/main.rs | 32 +- apps/neshub/src/rate.rs | 453 +++++++++++++++ apps/neshub/src/session.rs | 51 ++ crates/nesprotocol/src/datagram.rs | 93 +++ crates/nesprotocol/src/stats.rs | 16 +- docs/media-transport.md | 12 + 9 files changed, 910 insertions(+), 688 deletions(-) create mode 100644 apps/neshub/src/congestion.rs create mode 100644 apps/neshub/src/rate.rs diff --git a/apps/neshub/src/congestion.rs b/apps/neshub/src/congestion.rs new file mode 100644 index 00000000..14a5eaa4 --- /dev/null +++ b/apps/neshub/src/congestion.rs @@ -0,0 +1,67 @@ +//! The transport's congestion control, handed to the media's own. +//! +//! The video rate is decided per frame from what the path measurably did +//! (`rate.rs`), and that decision needs each frame to leave as one train: +//! its fragments back to back, so the bottleneck spaces them by its own rate +//! and the client can measure it. A transport controller in between would +//! window and pace them by its own estimate, smearing the very spacing being +//! measured, and second-guess a rate already chosen from better evidence. +//! +//! So this controller holds nothing back. Its window is larger than noq paces +//! at all (it stops pacing above `u32::MAX`), and it ignores loss, which the +//! media's own controller sees sooner, as queueing. +//! +//! Endpoint-wide, because transport settings are per endpoint, not per +//! connection: audio, input and control get it too. They are small flows the +//! video rate leaves room for; nothing they send can build a queue. + +use std::any::Any; +use std::sync::Arc; + +use iroh::endpoint::{Controller, ControllerFactory}; + +/// Past this noq does not pace: see `noq_proto::connection::pacing`. +const UNPACED_WINDOW: u64 = u32::MAX as u64 * 2; + +#[derive(Debug, Clone, Copy, Default)] +pub struct MediaRate; + +impl Controller for MediaRate { + fn on_congestion_event( + &mut self, + _now: std::time::Instant, + _sent: std::time::Instant, + _is_persistent_congestion: bool, + _is_ecn: bool, + _lost_bytes: u64, + _largest_lost_pn: u64, + ) { + } + + fn on_mtu_update(&mut self, _new_mtu: u16) {} + + fn window(&self) -> u64 { + UNPACED_WINDOW + } + + fn clone_box(&self) -> Box { + Box::new(*self) + } + + fn initial_window(&self) -> u64 { + UNPACED_WINDOW + } + + fn into_any(self: Box) -> Box { + self + } +} + +#[derive(Debug, Default)] +pub struct MediaRateFactory; + +impl ControllerFactory for MediaRateFactory { + fn build(self: Arc, _now: std::time::Instant, _current_mtu: u16) -> Box { + Box::new(MediaRate) + } +} diff --git a/apps/neshub/src/control.rs b/apps/neshub/src/control.rs index ab5fab3c..29436589 100644 --- a/apps/neshub/src/control.rs +++ b/apps/neshub/src/control.rs @@ -5,87 +5,46 @@ //! Nothing used to decide. The encoder took a bitrate from an environment //! variable nobody set, so every session offered the same 10 Mbps whatever the //! path between the two ends could carry. On a path sustaining under 3 Mbps that -//! does not degrade the picture, it removes it: most fragments are dropped, -//! every frame made of several fragments arrives incomplete, and the keyframe -- -//! the largest frame there is -- never completes either, so nothing ever -//! recovers. +//! does not degrade the picture, it removes it. //! -//! # Why the receiver decides, and not this end +//! # From the path, frame by frame //! -//! The obvious inputs are here: the hub holds the QUIC connection and can read -//! its round-trip time, its congestion window and its remaining datagram buffer. -//! They were measured, over an entire multi-session run, saying the path was -//! healthy while the client was receiving almost nothing -- 182 ms round trip, -//! *zero* QUIC packet loss, and arrivals falling from 1776 to 239 datagrams a -//! second. +//! Each client's path is modelled from its per-frame feedback (`rate.rs`): the +//! capacity its frames measure, whether the bottleneck is still busy with one +//! frame when the next arrives, and the share it delivers when it is. Down at +//! once when the queue grows, to that measured share; up only to a rate the path +//! has proven, held on the path's own recurrence. See nescore's +//! `docs/superpowers/specs/2026-10-16-video-rate-from-the-path-design.md`. //! -//! Part of that is structural rather than bad luck. `send_datagram` evicts the -//! oldest queued datagrams and returns `Ok`, so a sender that is overrunning the -//! path is told nothing at all; the "frames dropped" counter it feeds cannot be -//! non-zero however badly things are going. So the primary evidence is the -//! receiver's own report of what arrived, and this end's view is the fallback -//! for when no report has come in. +//! This replaced a controller built from thresholds: loss above 10% to back +//! off, below 2% to climb, back off to 0.85 of goodput, climb a twentieth of the +//! ceiling a second, a 10% deadband, queue limits of 250 and 80 ms. It could +//! only find the path's limit by hitting it, so it sawtoothed, and it heard about +//! the queue it built once a second, after the queue was tens of milliseconds +//! deep. //! -//! # Shape +//! # One encoder, the slowest client //! -//! Slow and dull on purpose. This is not a congestion controller in the QUIC -//! sense and does not try to be: it finds the right order of magnitude for a -//! video bitrate and stays there, at one decision a second. The transport -//! underneath is still doing real congestion control on its own packets. +//! One encoder serves every client, so it can answer only one question: the +//! rate the slowest path allows. Averaging would leave that client never +//! recovering while the numbers look acceptable. -use nesprotocol::{ControlMode, ReceiverReport}; +use std::collections::HashMap; -/// Loss above this means the path is being overrun and the target comes down. -const LOSS_DECREASE: f32 = 0.10; -/// Loss below this means there is room to climb. -const LOSS_INCREASE: f32 = 0.02; +use iroh::EndpointId; +use nesprotocol::ControlMode; +use nesprotocol::datagram::FrameFeedback; -/// What to keep of the measured goodput when backing off. -/// -/// Backing off to *below* what actually got through, rather than to a fraction -/// of what we were asking for, is what makes recovery quick. At 10 Mbps into a -/// 3 Mbps path the next target is about 2.5 Mbps rather than 8.5, so one step -/// does what twenty multiplicative decreases would. -const BACKOFF: f32 = 0.85; - -/// How much of the ceiling to add per calm second. Twenty steps from floor to -/// ceiling: slow enough that a brief quiet spell does not undo a back-off. -const CLIMB_FRACTION: u32 = 20; - -/// Don't actuate for less than this fraction of the current target. -/// -/// Retuning is cheap now -- it no longer rebuilds the encoder or forces a -/// keyframe -- but a line in the log every second saying the bitrate moved by -/// 1% is noise that hides the changes that matter. -const DEADBAND: f32 = 0.10; +use crate::rate::{self, PathModel}; /// Never go below this, whatever the ceiling is. /// -/// Under this a 1080p stream is not worth sending and the right answer is fewer -/// pixels rather than fewer bits, which is a decision about the tier and not one -/// this can make. It also has to leave room for audio, which shares the -/// connection and is not counted against the video ceiling. +/// A policy bound like the ceiling, not a guess about the network: under this a +/// 1080p stream is not worth sending, and the right answer is fewer pixels +/// rather than fewer bits, which is a decision about the tier and not one this +/// can make. const ABSOLUTE_FLOOR_KBPS: u32 = 800; -/// Seconds without a receiver report before this end's own view is used instead. -const SILENT_TICKS_BEFORE_FALLBACK: u32 = 3; - -/// Backlog, in milliseconds of video, that means the path is being overrun. -/// -/// Not a tuned number: it is a latency budget. At sixty frames a second a frame -/// is under 17 ms, so a quarter second of backlog is fifteen frames already -/// handed over and not yet gone -- a player is looking at a picture from before -/// their last four keypresses. There is no bitrate worth that, so past this the -/// target comes down whatever the loss says. -const QUEUE_DECREASE_MS: u32 = 250; - -/// Backlog, in milliseconds, under which the path is considered clear. -/// -/// Climbing needs a stronger warrant than holding does, because climbing is -/// what digs the queue. Five frames or so of backlog is the most that can be -/// outstanding and still be called live. -const QUEUE_CLIMB_MS: u32 = 80; - /// The band a target must stay inside. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct Limits { @@ -112,11 +71,8 @@ impl Limits { } } -/// What this end can see about the path, for when the far end is not talking. -/// -/// Every field is optional because every one of them can be unavailable: a -/// connection with no established path has no round-trip time, and a congestion -/// window is only meaningful once something has been sent. +/// What this end can see about a client's connection: for the stats, not for +/// the decision. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct PathView { /// Congestion window in bytes, as the transport currently believes it. @@ -124,70 +80,54 @@ pub struct PathView { /// Round-trip time in milliseconds. pub rtt_ms: Option, /// Datagram bytes handed to the transport that have not yet left. - /// - /// The backlog is the one thing this end can see that says the path is - /// being overrun *while it is still only late*. Loss says so too, but only - /// afterwards, and on a path that queues rather than drops, "afterwards" - /// can be several seconds. pub backlog_bytes: Option, } impl PathView { /// How long the backlog takes to drain at `drain_kbps`, in milliseconds. - /// - /// Bits divided by kilobits-per-second is milliseconds. The rate to divide - /// by is the one the queue actually drains at -- what is getting through -- - /// and not what we are asking for, which is the number that is too high - /// whenever this matters. pub fn backlog_ms(&self, drain_kbps: u32) -> Option { let backlog = self.backlog_bytes?; let drain = u64::from(drain_kbps.max(1)); Some((backlog.saturating_mul(8) / drain).min(u64::from(u32::MAX)) as u32) } - - /// Delivery rate the window and round trip imply, in kbps. - /// - /// A window is a quantity of bytes in flight for one round trip, so the two - /// together are a rate. Treated as an estimate of last resort: it describes - /// the transport's own opinion of the path, and that opinion is exactly what - /// was observed to be wrong when this mattered most. - pub fn estimate_kbps(&self) -> Option { - let (cwnd, rtt) = (self.cwnd_bytes?, self.rtt_ms?.max(1)); - let bits_per_second = cwnd.saturating_mul(8) * 1000 / u64::from(rtt); - Some((bits_per_second / 1000).min(u64::from(u32::MAX)) as u32) - } } /// Why the target is what it is, for the overlay and the log. #[derive(Debug, Clone, Copy, PartialEq, Eq)] #[repr(u8)] pub enum Reason { - /// Loss was high; backed off to under what was actually getting through. - Congested, - /// Loss was in the middle band; left alone. - Holding, - /// Loss was low; climbing towards the ceiling. - Climbing, - /// No receiver report recently; using this end's view of the path. - Fallback, - /// No report and no usable view either; decaying towards the floor. - Blind, + /// Where the session started; no path has said otherwise yet. + Start, + /// A path's queue was growing; sending below its share to drain it. + Draining, + /// Settled at the share the slowest path delivered while queueing. + Measured, + /// Raised to what the slowest path has proven it carries. + Stepped, + /// No frame came back from the slowest path; held at its last share. + Silent, /// Somebody set the bitrate by hand. Manual, /// The encoder is not under a bitrate at all, so there is nothing to decide. ConstantQuality, /// Nobody is connected, so there is no path to have an opinion about. NoClients, - /// This second's numbers describe frames the hub chose not to send, so they - /// say nothing about the path. - SelfInflicted, - /// Nothing was lost, but the send queue is standing deep enough that the - /// picture is arriving late. Backed off on the queue rather than on loss. - Backlogged, +} + +impl From for Reason { + fn from(r: rate::Reason) -> Self { + match r { + rate::Reason::Start => Self::Start, + rate::Reason::Draining => Self::Draining, + rate::Reason::Measured => Self::Measured, + rate::Reason::Stepped => Self::Stepped, + rate::Reason::Silent => Self::Silent, + } + } } /// The controller's whole state. -#[derive(Debug, Clone, Copy)] +#[derive(Debug)] pub struct Controller { /// The ceiling the box was given, which a client may lower but never raise. /// @@ -196,13 +136,6 @@ pub struct Controller { /// it again, because the only number left to compare against is the one it /// just lowered. box_ceiling_kbps: u32, - /// What the controller currently believes the bitrate should be. - /// - /// Kept apart from `sent_kbps` deliberately. Folding the two together looks - /// simpler and stalls the climb: the step up is a fraction of the *ceiling*, - /// so once the target passes half of it each step is under the deadband, and - /// a target that only moved when it actuated could never accumulate past - /// that point. Belief moves every second; only saying so is rationed. target_kbps: u32, /// What the encoder was last told, or `None` before anything was sent. sent_kbps: Option, @@ -210,19 +143,18 @@ pub struct Controller { mode: ControlMode, /// Set when the encoder is in constant-QP; there is no bitrate to control. constant_quality: bool, - silent_ticks: u32, reason: Reason, - /// Last measured send-queue depth, kept for reporting rather than control. + /// Last measured send-queue depth, for reporting. backlog_ms: u32, + /// Each streaming client's path. + paths: HashMap, } impl Controller { /// Start at the ceiling. /// /// Optimistic on purpose: the common case is a path that can carry the tier - /// that was sold, and a session that opened at its floor and climbed would - /// take twenty seconds to look like what was paid for. The first bad report - /// undoes it in one step. + /// that was sold, and a path that cannot says so within frames. pub fn new(limits: Limits) -> Self { Self { box_ceiling_kbps: limits.ceiling_kbps, @@ -231,9 +163,9 @@ impl Controller { limits, mode: ControlMode::Auto, constant_quality: false, - silent_ticks: 0, - reason: Reason::Holding, + reason: Reason::NoClients, backlog_ms: 0, + paths: HashMap::new(), } } @@ -250,21 +182,31 @@ impl Controller { self.reason } - /// How far behind the send queue was at the last decision, in milliseconds. + /// How far behind the send queue was at the last tick, in milliseconds. pub fn backlog_ms(&self) -> u32 { self.backlog_ms } + /// The slowest path's measured capacity, in kbps; 0 before one is measured. + pub fn capacity_kbps(&self) -> u32 { + self.slowest() + .and_then(PathModel::capacity_kbps) + .unwrap_or(0) + } + + /// The slowest path's last delivered share, in kbps; 0 before it queued. + pub fn delivery_kbps(&self) -> u32 { + self.slowest() + .and_then(PathModel::delivery_kbps) + .unwrap_or(0) + } + /// The ceiling the box was given, whatever a client has since asked for. pub fn box_ceiling_kbps(&self) -> u32 { self.box_ceiling_kbps } /// Adopt a ceiling a client asked for, never above the one the box was given. - /// - /// A client may lower its own ceiling -- to test a path, or because it knows - /// something about its link that this end does not -- but it may not raise - /// the one the tier bought. pub fn set_ceiling(&mut self, ceiling_kbps: u32) { self.limits = Limits::new(ceiling_kbps.min(self.box_ceiling_kbps).max(1)); self.target_kbps = self.limits.clamp(self.target_kbps); @@ -293,39 +235,45 @@ impl Controller { self.constant_quality = constant_quality; } - /// One second's decision. - /// - /// `report` is the worst report across the clients attached, or `None` when - /// none of them said anything. Returns the new target when it is worth - /// sending, and `None` when nothing should be sent -- which is most seconds. - /// One second's decision. - /// - /// `self_inflicted` says this second contained frames the hub deliberately - /// withheld -- a client resynchronising, whose deltas were undecodable and - /// were dropped rather than sent. Those seconds cannot be read as evidence - /// about the path: fewer frames were sent, so fewer arrived, so the measured - /// goodput is low and the loss is high, and a controller anchoring a backoff - /// on that would cut the bitrate in response to its own decision. Every - /// stutter would then also cost bandwidth, which is the opposite of what a - /// resynchronisation needs. - pub fn tick( + /// One frame's feedback from one client. Returns a new target when the + /// encoder should hear one. + pub fn on_feedback( &mut self, - clients: usize, - report: Option, - path: PathView, - self_inflicted: bool, + client: EndpointId, + frame: &FrameFeedback, + now_us: u64, ) -> Option { - if clients == 0 { + let start = self.target_kbps; + self.paths + .entry(client) + .or_insert_with(|| PathModel::new(start, now_us)) + .observe(frame, now_us); + self.decide() + } + + /// Once a second, with the clients that are streaming and this end's view + /// of their connections. Returns a new target when the encoder should hear + /// one. + pub fn tick(&mut self, streaming: &[EndpointId], path: PathView) -> Option { + self.paths.retain(|id, _| streaming.contains(id)); + for model in self.paths.values_mut() { + model.tick(); + } + self.backlog_ms = path.backlog_ms(self.target_kbps).unwrap_or(0); + if streaming.is_empty() { // Nothing is connected, so nothing is being carried and there is no - // path to form an opinion about. Deciding here means deciding on the - // absence of evidence: the controller used to read it as silence and - // decay to the floor, so a box waiting for its first client spent - // that time winding itself down and then jumped back up the moment - // somebody arrived. + // path to form an opinion about. self.reason = Reason::NoClients; - self.silent_ticks = 0; return None; } + self.decide() + } + + fn slowest(&self) -> Option<&PathModel> { + self.paths.values().min_by_key(|m| m.allowed_kbps()) + } + + fn decide(&mut self) -> Option { if self.constant_quality { self.reason = Reason::ConstantQuality; return None; @@ -334,124 +282,44 @@ impl Controller { self.reason = Reason::Manual; return None; } - if self_inflicted { - // Held, not decayed. The path may be fine; this second simply - // cannot say, and silence about the path is not evidence against it. - self.reason = Reason::SelfInflicted; - self.silent_ticks = 0; + let (allowed, reason) = match self.slowest() { + Some(m) => (m.allowed_kbps(), m.reason().into()), + None if self.sent_kbps.is_none() => (self.target_kbps, Reason::Start), + None => return None, + }; + self.reason = reason; + self.target_kbps = self.limits.clamp(allowed); + // Every change is said: they are events -- a queue found, a share + // measured, a step proven -- not a level wandering by a percent. The + // first decision is always said, because the encoder started at + // whatever its environment gave it. + if self.sent_kbps == Some(self.target_kbps) { return None; } - - // What the queue drains at is what is getting through. With no report - // to say, the target is the best guess available -- and a target that - // is too high only makes the backlog look shorter than it is, so this - // errs towards patience rather than towards cutting the rate. - let drain_kbps = match report.as_ref().map(|r| (r.goodput_bps / 1000) as u32) { - Some(measured) if measured > 0 => measured, - _ => self.target_kbps, - }; - self.backlog_ms = path.backlog_ms(drain_kbps).unwrap_or(0); - - let next = match report.as_ref().and_then(|r| r.loss().map(|l| (r, l))) { - Some((report, loss)) => { - self.silent_ticks = 0; - self.decide_from_report(report, loss, path) - } - None => { - self.silent_ticks = self.silent_ticks.saturating_add(1); - if self.silent_ticks < SILENT_TICKS_BEFORE_FALLBACK { - self.reason = Reason::Holding; - return None; - } - self.decide_from_path(path) - } - }; - - self.target_kbps = self.limits.clamp(next); - // Every second moves the belief; only a change worth acting on is said. - // The first decision is always said -- the encoder started at whatever - // its environment gave it, and this end has no way to know it matches. - match self.sent_kbps { - Some(sent) if !worth_sending(sent, self.target_kbps, self.limits) => None, - _ => { - self.sent_kbps = Some(self.target_kbps); - Some(self.target_kbps) - } - } - } - - fn decide_from_report(&mut self, report: &ReceiverReport, loss: f32, path: PathView) -> u32 { - let measured = (report.goodput_bps / 1000) as u32; - // `backlog_ms` is zero both when the queue is empty and when the - // transport cannot say, which are the same thing for this purpose: a - // queue nobody can see is not evidence of one. - let known = path.backlog_bytes.is_some(); - let queued = known && self.backlog_ms > QUEUE_DECREASE_MS; - - if loss > LOSS_DECREASE || queued { - self.reason = if queued { - Reason::Backlogged - } else { - Reason::Congested - }; - // Anchored on what actually arrived, not on what we were asking for. - // `goodput` of zero means nothing completed at all, in which case - // there is no measurement to anchor to and the target is simply - // halved -- the alternative, backing off to zero, would take the - // stream below the floor on a single bad second. - let anchor = if measured == 0 { - self.target_kbps / 2 - } else { - measured.min(self.target_kbps) - }; - return (anchor as f32 * BACKOFF) as u32; - } - // Climbing is what digs the queue, so it needs the queue to be empty as - // well as the loss to be low. Without this the controller climbs all - // the way to the ceiling against a path it is already overrunning, - // because a path that queues instead of dropping reports no loss at all - // until the buffer finally overflows -- and by then the picture is - // seconds behind. - if loss < LOSS_INCREASE && (!known || self.backlog_ms < QUEUE_CLIMB_MS) { - self.reason = Reason::Climbing; - return self - .target_kbps - .saturating_add(self.limits.ceiling_kbps / CLIMB_FRACTION); - } - self.reason = Reason::Holding; - self.target_kbps - } - - fn decide_from_path(&mut self, path: PathView) -> u32 { - match path.estimate_kbps() { - Some(estimate) => { - self.reason = Reason::Fallback; - estimate - } - None => { - // Nothing said and nothing visible. Holding a high target on no - // evidence is how the original failure sustained itself, so this - // decays rather than holds. - self.reason = Reason::Blind; - (self.target_kbps as f32 * BACKOFF) as u32 - } - } + self.sent_kbps = Some(self.target_kbps); + Some(self.target_kbps) } } -/// Whether a change is big enough to be worth actuating. -/// -/// Reaching either end of the band always is: a target pinned to the floor or -/// the ceiling is a fact worth stating even when the step to it was small. -pub fn worth_sending(current: u32, next: u32, limits: Limits) -> bool { - if current == next { - return false; +/// Tell the encoder a target the controller has just decided. +pub fn actuate( + cmd_tx: &tokio::sync::mpsc::UnboundedSender>, + controller: &Controller, + kbps: u32, +) { + let mut cmd = vec![nesprotocol::MSG_ENCODE_SETTINGS]; + nesprotocol::encode_bitrate_only(&mut cmd, kbps); + if cmd_tx.send(cmd).is_err() { + tracing::warn!("encoder command channel closed"); + return; } - if next == limits.floor_kbps() || next == limits.ceiling_kbps { - return true; - } - let change = current.abs_diff(next) as f32; - change >= current as f32 * DEADBAND + // Debug, not info: each is an event, but a path that keeps changing can + // make several a second, which at info buries the lines that matter. + tracing::debug!( + "video target {kbps}/{} kbps: {:?}", + controller.limits().ceiling_kbps, + controller.reason(), + ); } #[cfg(test)] @@ -460,456 +328,130 @@ mod tests { const CEILING: u32 = 10_000; - fn limits() -> Limits { - Limits::new(CEILING) - } - fn controller() -> Controller { - Controller::new(limits()) + Controller::new(Limits::new(CEILING)) } - /// A second in which `released` of `total` frames made it, carrying - /// `goodput_kbps` of completed video. - fn report(released: u32, incomplete: u32, goodput_kbps: u64) -> ReceiverReport { - ReceiverReport { - goodput_bps: goodput_kbps * 1000, - released, - incomplete, - never_arrived: 0, - rtt_ms: 180, - ..Default::default() + fn client(n: u8) -> EndpointId { + iroh::SecretKey::from_bytes(&[n; 32]).public() + } + + /// A frame of `bytes` sent at `t_us`, arriving over a link of `kbps` + /// that `busy` says was still carrying the previous frame. + fn frame(seq: u16, t_us: u64, bytes: u32, kbps: u32, queue_us: u32) -> FrameFeedback { + let spread = (u64::from(bytes - 1_200) * 8_000 / u64::from(kbps)) as u32; + FrameFeedback { + seq, + bytes, + fragments: bytes.div_ceil(1_200) as u16, + first_bytes: 1_200, + sent_us: t_us, + first_arrival_us: t_us + 40_000 + u64::from(queue_us), + spread_us: spread, + first_q_us: queue_us, } } - fn healthy() -> ReceiverReport { - report(60, 0, 9_800) - } - - /// A path with `backlog` bytes handed over and not yet gone. - fn backlogged(backlog: u64) -> PathView { - PathView { - cwnd_bytes: Some(13_000), - rtt_ms: Some(180), - backlog_bytes: Some(backlog), - } - } - - /// The second reported failure, as measured over a 1000-mile link. - /// - /// Every frame arrived and every frame completed -- zero loss, zero QUIC - /// congestion events, a flat 180 ms round trip for the whole session -- and - /// the picture was still several seconds behind, because the offer was - /// several times what the path carried and the difference was sitting in - /// the hub's own send buffer. A controller reading loss alone sees a - /// perfect path here and climbs to the ceiling against it. - #[test] - fn a_path_that_queues_instead_of_dropping_must_not_read_as_healthy() { - let mut c = controller(); - // Two megabits get through; a megabyte is already queued behind them. - let arriving = report(60, 0, 2_000); - let path = backlogged(1024 * 1024); - - for _ in 0..30 { - c.tick(1, Some(arriving), path, false); - } - - assert_ne!( - c.reason(), - Reason::Climbing, - "climbing against a path with four seconds of backlog" - ); - assert!( - c.target_kbps() < 2_000, - "target {} kbps is at or above what is getting through, so the \ - queue can only grow", - c.target_kbps() - ); - } - - /// The sawtooth: floor, climb to ceiling, collapse, repeat every 20 s. - /// - /// Loss alone cannot break this cycle, because on a queueing path loss only - /// appears once the buffer finally overflows -- long after the latency has - /// made the session unplayable, and by then the queue is deep enough that - /// backing off to the floor is the only way out. - #[test] - fn the_target_settles_instead_of_sawtoothing() { - let mut c = controller(); - // A steady 2 Mbps path. The backlog is what the last second of - // over-sending left behind, drained at what actually gets through. - let mut backlog: i64 = 0; - let mut seen = Vec::new(); - - for _ in 0..60 { - let target = c.target_kbps(); - // Whatever was asked for above 2 Mbps piles up; the rest drains. - backlog = (backlog + (i64::from(target) - 2_000) * 1000 / 8).clamp(0, 4 * 1024 * 1024); - c.tick( - 1, - Some(report(60, 0, 2_000)), - backlogged(backlog as u64), - false, - ); - seen.push(c.target_kbps()); - } - - let settled = &seen[30..]; - let (lo, hi) = ( - *settled.iter().min().unwrap(), - *settled.iter().max().unwrap(), - ); - assert!( - hi - lo <= 1_000, - "target still swinging between {lo} and {hi} kbps after 30 seconds" - ); - assert!( - hi <= 2_400, - "settled at {hi} kbps against a path carrying 2000" - ); - } - - /// The reported depth is what the next session will be judged on, so it - /// has to be right: bytes over the rate they leave at, in milliseconds. - #[test] - fn the_queue_depth_is_reported_as_measured() { - let mut c = controller(); - // 250 kB queued behind a 2 Mbps drain is exactly one second. - c.tick(1, Some(report(60, 0, 2_000)), backlogged(250_000), false); - assert_eq!(c.backlog_ms(), 1_000); - - // A transport that cannot say must report no queue, not a wrong one. - c.tick(1, Some(report(60, 0, 2_000)), PathView::default(), false); - assert_eq!(c.backlog_ms(), 0); - } - - /// The backlog must not become a reason never to climb again. - #[test] - fn a_clear_queue_still_climbs() { - let mut c = controller(); - c.set_ceiling(4_000); - let clear = backlogged(0); - for _ in 0..5 { - c.tick(1, Some(report(60, 0, 3_900)), clear, false); - } - assert_eq!(c.reason(), Reason::Climbing); - } - - /// The reported failure, as measured: a 10 Mbps offer into a path carrying - /// under three, where *no* frame completed for minutes. - #[test] - fn the_reported_failure_is_corrected_in_seconds() { - let mut c = controller(); - assert_eq!(c.target_kbps(), CEILING); - - let mut ticks = 0; - while c.target_kbps() > 2_800 && ticks < 10 { - // 46 frames a second starting to arrive, none of them completing. - c.tick(1, Some(report(0, 46, 0)), PathView::default(), false); - ticks += 1; - } - assert!( - c.target_kbps() <= 2_800, - "still at {} kbps after {ticks} seconds", - c.target_kbps(), - ); - assert!( - ticks <= 5, - "took {ticks} seconds to stop overrunning the path" - ); - } - - /// With a goodput measurement to anchor on it should take one step, not - /// several -- that is the whole reason for backing off to what arrived - /// rather than to a fraction of what was asked for. - #[test] - fn a_measured_goodput_is_corrected_in_one_step() { - let mut c = controller(); - let sent = c.tick(1, Some(report(12, 48, 2_850)), PathView::default(), false); - assert_eq!(sent, Some(c.target_kbps())); - assert!( - c.target_kbps() < 2_850, - "backed off to {} kbps, which is above what actually arrived", - c.target_kbps(), - ); - } - - /// The negative that matters most: a path that is fine must be left alone. - #[test] - fn a_healthy_path_is_not_wandered_away_from() { - let mut c = controller(); - // The first decision is always stated. The encoder started at whatever - // its environment gave it and this end cannot know that matches, so the - // target is asserted once rather than assumed. - assert_eq!( - c.tick(1, Some(healthy()), PathView::default(), false), - Some(CEILING) - ); - for _ in 0..60 { - assert_eq!( - c.tick(1, Some(healthy()), PathView::default(), false), - None, - "a healthy path produced a bitrate change", - ); - } - assert_eq!(c.target_kbps(), CEILING); - } - #[test] fn the_first_decision_is_always_stated() { - // Otherwise a session whose path happens to match its ceiling never - // tells the encoder anything, and the encoder keeps whatever its - // environment gave it -- which is the failure this replaces. let mut c = controller(); - assert_eq!( - c.tick(1, Some(healthy()), PathView::default(), false), - Some(CEILING), - "the opening target was never stated", - ); - // And not repeated, now that the encoder has been told. - assert_eq!(c.tick(1, Some(healthy()), PathView::default(), false), None); + assert_eq!(c.tick(&[client(1)], PathView::default()), Some(CEILING)); + assert_eq!(c.tick(&[client(1)], PathView::default()), None); } #[test] - fn a_small_change_is_not_worth_saying() { - // Below the deadband nothing is sent, so the log and the encoder are not - // touched sixty times a minute for changes nobody could see. - let l = limits(); - assert!(!worth_sending(5_000, 5_200, l)); - assert!(worth_sending(5_000, 5_600, l)); - } - - #[test] - fn reaching_either_end_of_the_band_is_always_worth_saying() { - let l = limits(); - assert!(worth_sending(l.floor_kbps() + 1, l.floor_kbps(), l)); - assert!(worth_sending(CEILING - 1, CEILING, l)); - } - - #[test] - fn the_ceiling_is_never_exceeded() { + fn the_slowest_client_decides() { let mut c = controller(); - for _ in 0..200 { - c.tick(1, Some(healthy()), PathView::default(), false); - assert!(c.target_kbps() <= CEILING, "{} kbps", c.target_kbps()); - } - } - - #[test] - fn the_floor_is_never_gone_below() { - let mut c = controller(); - for _ in 0..200 { - c.tick(1, Some(report(0, 60, 0)), PathView::default(), false); - assert!( - c.target_kbps() >= c.limits().floor_kbps(), - "{} kbps is below the floor", - c.target_kbps(), - ); - } - } - - #[test] - fn it_climbs_back_after_a_bad_patch() { - let mut c = controller(); - for _ in 0..5 { - c.tick(1, Some(report(0, 46, 0)), PathView::default(), false); - } - let bottom = c.target_kbps(); - for _ in 0..40 { - c.tick(1, Some(healthy()), PathView::default(), false); + let (fast, slow) = (client(1), client(2)); + let mut t = 0; + let mut q = 0; + for seq in 0..120u16 { + // The fast client's frames drain at once; the slow one's queue + // grows by 6 ms a frame, a train arriving straight behind the last. + c.on_feedback(fast, &frame(seq, t, 20_000, 40_000, 0), t); + let mut f = frame(seq, t, 20_000, 6_000, q); + if seq > 0 { + f.first_arrival_us = t + 40_000 + u64::from(q); + } + c.on_feedback(slow, &f, t); + q += 6_000; + t += 16_667; } assert!( - c.target_kbps() > bottom, - "never recovered from {bottom} kbps" + c.target_kbps() < CEILING, + "the slow client did not bring the rate down" ); - assert_eq!(c.target_kbps(), CEILING, "did not climb all the way back"); } #[test] - fn a_second_the_hub_starved_is_not_read_as_a_bad_path() { - // While a client resynchronises its deltas are withheld, so fewer frames - // are sent, fewer arrive, goodput reads low and loss reads high. Backing - // off on that would cut the bitrate in response to the hub's own - // decision -- and make every stutter cost bandwidth as well. + fn the_ceiling_is_never_exceeded_and_the_floor_never_gone_below() { let mut c = controller(); - assert_eq!( - c.tick(1, Some(healthy()), PathView::default(), false), - Some(CEILING) - ); - - let starved = report(0, 46, 0); - for _ in 0..10 { - assert_eq!(c.tick(1, Some(starved), PathView::default(), true), None); - } - assert_eq!(c.target_kbps(), CEILING, "the hub cut its own bitrate"); - assert_eq!(c.reason(), Reason::SelfInflicted); - } - - #[test] - fn a_genuinely_bad_second_still_acts_once_the_hub_stops_starving_it() { - // The flag holds, it does not blind. The moment a second is the path's - // own, the same evidence is acted on. - let mut c = controller(); - c.tick(1, Some(healthy()), PathView::default(), false); - c.tick(1, Some(report(0, 46, 0)), PathView::default(), true); - assert_eq!(c.target_kbps(), CEILING); - - c.tick(1, Some(report(12, 48, 2_850)), PathView::default(), false); - assert!(c.target_kbps() < 2_850, "a real bad second was ignored too"); - } - - #[test] - fn a_starved_second_does_not_count_towards_silence() { - // It is not a missing report -- one arrived, it just cannot be read. - // Counting it as silence would slide towards the path-based fallback and - // then to decaying blind, which is a different wrong answer. - let mut c = controller(); - c.tick(1, Some(healthy()), PathView::default(), false); - for _ in 0..10 { - c.tick(1, Some(report(0, 60, 0)), PathView::default(), true); - } - assert_eq!(c.reason(), Reason::SelfInflicted); - assert_eq!(c.target_kbps(), CEILING); + c.tick(&[client(1)], PathView::default()); + assert!(c.target_kbps() <= CEILING); + c.set_ceiling(500); + assert!(c.target_kbps() <= 500); + assert!(c.target_kbps() >= c.limits().floor_kbps()); } #[test] fn a_hand_set_bitrate_stands_the_controller_down() { - // How this class of bug gets diagnosed. A controller that overrode a - // person's setting a second later would take the tool away. let mut c = controller(); - c.note_manual_target(1_000); - for _ in 0..30 { - assert_eq!( - c.tick(1, Some(report(0, 60, 0)), PathView::default(), false), - None - ); - } - assert_eq!(c.target_kbps(), 1_000); + c.note_manual_target(3_000); + assert_eq!(c.tick(&[client(1)], PathView::default()), None); assert_eq!(c.reason(), Reason::Manual); + assert_eq!(c.target_kbps(), 3_000); } #[test] fn constant_quality_has_no_bitrate_to_decide() { let mut c = controller(); c.set_constant_quality(true); - assert_eq!( - c.tick(1, Some(report(0, 60, 0)), PathView::default(), false), - None - ); + assert_eq!(c.tick(&[client(1)], PathView::default()), None); assert_eq!(c.reason(), Reason::ConstantQuality); } - #[test] - fn silence_is_tolerated_briefly_and_then_acted_on() { - let mut c = controller(); - // A missed report or two says nothing; the path was fine a second ago. - for _ in 0..(SILENT_TICKS_BEFORE_FALLBACK - 1) { - assert_eq!(c.tick(1, None, PathView::default(), false), None); - assert_eq!(c.target_kbps(), CEILING); - } - // Past that, with nothing visible from this end either, it decays rather - // than holding a high target on no evidence at all -- holding is how the - // original failure sustained itself. - c.tick(1, None, PathView::default(), false); - assert_eq!(c.reason(), Reason::Blind); - assert!(c.target_kbps() < CEILING); - } - - #[test] - fn this_ends_view_is_used_only_when_the_far_end_is_silent() { - let mut c = controller(); - let path = PathView { - cwnd_bytes: Some(120_000), - rtt_ms: Some(200), - backlog_bytes: None, - }; - // A report present means the path view is ignored, however tempting. - c.tick(1, Some(healthy()), path, false); - assert_eq!(c.reason(), Reason::Climbing); - - for _ in 0..SILENT_TICKS_BEFORE_FALLBACK { - c.tick(1, None, path, false); - } - assert_eq!(c.reason(), Reason::Fallback); - // 120 KB in flight per 200 ms is 4.8 Mbps. - assert_eq!(c.target_kbps(), 4_800); - } - #[test] fn nothing_is_decided_while_nobody_is_connected() { - // A box waiting for its first client used to read the silence as a dead - // path and wind itself down to the floor, then jump back up the moment - // somebody arrived. There is no path to have an opinion about yet. let mut c = controller(); - for _ in 0..30 { - assert_eq!(c.tick(0, None, PathView::default(), false), None); - } - assert_eq!(c.target_kbps(), CEILING); + assert_eq!(c.tick(&[], PathView::default()), None); assert_eq!(c.reason(), Reason::NoClients); } #[test] - fn a_lowered_ceiling_can_be_raised_again() { - // The trap in folding the box's ceiling together with the one in force: - // after lowering, the only number left to compare against is the lowered - // one, so the client can never get back up. + fn a_client_that_leaves_takes_its_path_with_it() { let mut c = controller(); - c.set_ceiling(2_000); - assert_eq!(c.limits().ceiling_kbps, 2_000); - c.set_ceiling(5_000); - assert_eq!(c.limits().ceiling_kbps, 5_000); - assert_eq!(c.box_ceiling_kbps(), CEILING); + c.on_feedback(client(1), &frame(0, 0, 20_000, 6_000, 0), 0); + c.tick(&[], PathView::default()); + assert!(c.paths.is_empty()); } #[test] - fn a_ceiling_change_is_restated_to_the_encoder() { - // The encoder is the thing that has to act on it, and a target that - // happens to land on the same number is still a different instruction - // when the band around it moved. + fn a_lowered_ceiling_can_be_raised_again() { let mut c = controller(); - assert!( - c.tick(1, Some(healthy()), PathView::default(), false) - .is_some() - ); - c.set_ceiling(3_000); - assert_eq!( - c.tick(1, Some(healthy()), PathView::default(), false), - Some(3_000) - ); + c.set_ceiling(4_000); + assert_eq!(c.limits().ceiling_kbps, 4_000); + c.set_ceiling(9_000); + assert_eq!(c.limits().ceiling_kbps, 9_000); } #[test] fn a_client_may_lower_its_ceiling_but_not_raise_it() { let mut c = controller(); - c.set_ceiling(4_000); - assert_eq!(c.limits().ceiling_kbps, 4_000); - assert!(c.target_kbps() <= 4_000, "the target outlived its ceiling"); + c.set_ceiling(CEILING * 2); + assert_eq!(c.limits().ceiling_kbps, CEILING); + } - c.set_ceiling(50_000); - assert_eq!( - c.limits().ceiling_kbps, - CEILING, - "a client raised the ceiling the tier bought", - ); + #[test] + fn a_ceiling_change_is_restated_to_the_encoder() { + let mut c = controller(); + c.tick(&[client(1)], PathView::default()); + c.set_ceiling(CEILING); + assert_eq!(c.tick(&[client(1)], PathView::default()), Some(CEILING)); } #[test] fn a_ceiling_under_the_floor_does_not_invert_the_band() { - // A misconfigured tier must not produce a floor above its own ceiling, - // which would put every target above the limit it exists to enforce. - let l = Limits::new(500); - assert!(l.floor_kbps() <= l.ceiling_kbps); - assert_eq!(l.clamp(10_000), 500); - } - - #[test] - fn a_silent_second_is_not_read_as_a_healthy_one() { - // No frames at all accounted for: nothing sent, or nothing arrived. - // Reading it as no loss would climb straight into a path that may be - // carrying nothing. - let mut c = controller(); - let empty = ReceiverReport::default(); - assert_eq!(empty.loss(), None); - c.tick(1, Some(empty), PathView::default(), false); - assert_eq!(c.target_kbps(), CEILING, "climbed on an empty report"); + let limits = Limits::new(500); + assert!(limits.floor_kbps() <= limits.ceiling_kbps); } } diff --git a/apps/neshub/src/dgram.rs b/apps/neshub/src/dgram.rs index ac94d0a6..be6f2015 100644 --- a/apps/neshub/src/dgram.rs +++ b/apps/neshub/src/dgram.rs @@ -38,11 +38,13 @@ pub enum MediaItem { /// Transport settings for an endpoint carrying media datagrams. /// -/// See [`DGRAM_BUFFER_BYTES`] for why the defaults are not enough. +/// See [`DGRAM_BUFFER_BYTES`] for why the defaults are not enough, and +/// `congestion.rs` for why the transport does not decide the rate. pub fn media_transport_config() -> QuicTransportConfig { QuicTransportConfig::builder() .datagram_send_buffer_size(DGRAM_BUFFER_BYTES) .datagram_receive_buffer_size(Some(DGRAM_BUFFER_BYTES)) + .congestion_controller_factory(std::sync::Arc::new(crate::congestion::MediaRateFactory)) .build() } diff --git a/apps/neshub/src/main.rs b/apps/neshub/src/main.rs index ab5b9c0a..b4716bf1 100644 --- a/apps/neshub/src/main.rs +++ b/apps/neshub/src/main.rs @@ -1,3 +1,4 @@ +mod congestion; mod control; mod dgram; #[cfg(test)] @@ -6,6 +7,7 @@ mod ipc_listener; mod keyframe; mod negotiation; mod pyro; +mod rate; mod screenshot; mod session; mod ticket; @@ -362,29 +364,15 @@ async fn main() -> Result<()> { loop { interval.tick().await; - // One decision a second, on the same tick as the stats, because - // a report describes the second that just passed and there is - // nothing to gain from deciding more often than they arrive. + // The rate is decided per frame, as feedback arrives (see + // `session.rs`). Once a second is for what feedback cannot say: + // a path that went silent, and clients that left. { - let clients = mgr.client_count().await; - let (report, path, self_inflicted) = mgr.worst_report().await; + let streaming = mgr.streaming_clients().await; + let (_, path, _) = mgr.worst_report().await; let mut controller = controller.lock().await; - if let Some(kbps) = controller.tick(clients, report, path, self_inflicted) { - let mut cmd = vec![nesprotocol::MSG_ENCODE_SETTINGS]; - nesprotocol::encode_bitrate_only(&mut cmd, kbps); - if cmd_tx.send(cmd).is_err() { - tracing::warn!("encoder command channel closed"); - } else { - // Trace, not info: under a path that keeps moving - // this fires every second, and a per-second line at - // info buries everything worth reading. - tracing::trace!( - "video ceiling {}/{} kbps: {:?}", - kbps, - controller.limits().ceiling_kbps, - controller.reason(), - ); - } + if let Some(kbps) = controller.tick(&streaming, path) { + crate::control::actuate(&cmd_tx, &controller, kbps); } } @@ -435,6 +423,8 @@ async fn main() -> Result<()> { pipeline_p95_ms, pipeline_max_ms, backlog_ms: controller.backlog_ms().min(u32::from(u16::MAX)) as u16, + capacity_kbps: controller.capacity_kbps(), + delivery_kbps: controller.delivery_kbps(), }, ); } diff --git a/apps/neshub/src/rate.rs b/apps/neshub/src/rate.rs new file mode 100644 index 00000000..d135a7a1 --- /dev/null +++ b/apps/neshub/src/rate.rs @@ -0,0 +1,453 @@ +//! What rate one client's path can carry, from how its frames arrived. +//! +//! See nescore's `docs/superpowers/specs/2026-10-16-video-rate-from-the-path-design.md`. +//! +//! # Every frame is a measurement +//! +//! A frame's fragments leave this end back to back (`congestion.rs` sees that +//! nothing holds them), so they reach the client spaced by the slowest link +//! they crossed. Two things follow, both read from the client's per-frame +//! feedback rather than guessed: +//! +//! - **Capacity.** A frame's bytes after its first fragment, over the time its +//! fragments took to arrive, is the bottleneck's rate. Every multi-fragment +//! frame measures it without sending above it for longer than one frame. +//! - **Whether we are over it.** Below capacity a frame's train has drained +//! before the next frame reaches the bottleneck, so consecutive trains arrive +//! with a gap between them. When the next train starts arriving no later than +//! the previous one's next fragment would have, the bottleneck never went idle +//! between them: it is busy with our traffic. Busy is not yet too much -- a +//! keyframe keeps it busy for a few frames and then it drains -- so the test +//! is whether the queue the first fragments find is *growing* across the busy +//! period, and whether more is being sent into it than comes out. +//! +//! # Down at once, up on evidence +//! +//! When the queue grows, what the path delivered over the busy period is its +//! measured share, and the queue already built is drained exactly: sending as +//! far below that share as the rate went above it, for as long, until the +//! trains have a gap between them again. Then the rate sits at the share. +//! +//! Going up uses the capacity the frames measured, and holds each step on the +//! same evidence the audio playout uses: a rate may be used again once the path +//! has gone quiet for longer than the longest it has ever gone between two +//! overloads at or below it, playout's start counting as the first. + +use nesprotocol::datagram::FrameFeedback; + +/// Why a client's rate is what it is. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(u8)] +pub enum Reason { + /// Where the session started; nothing has said otherwise yet. + Start, + /// The queue was growing; sending below the path's share to drain it. + Draining, + /// Settled at the share the path delivered while it was queueing. + Measured, + /// Raised to what the path has proven it carries. + Stepped, + /// No frame has come back; held at the last rate the path proved. + Silent, +} + +/// One time the path was overloaded. +#[derive(Debug, Clone, Copy)] +struct Overload { + at_us: u64, + /// What was being sent into it. + sent_kbps: u32, +} + +/// A run of frames during which the bottleneck never went idle. +#[derive(Debug, Clone, Copy)] +struct Busy { + start_sent_us: u64, + start_arrival_us: u64, + /// Bytes of the frames in the run before the latest one. + bytes: u64, + /// The queue the second frame of the run found: the first frame's own + /// burst included, so a run that only drains never reads as growing. + first_q_us: u32, + /// Whether this run has already been counted as an overload. + counted: bool, +} + +#[derive(Debug)] +pub struct PathModel { + allowed_kbps: u32, + reason: Reason, + started_us: u64, + prev: Option, + busy: Option, + draining: bool, + /// The share the path last delivered while queueing. + delivery_kbps: Option, + /// The highest capacity a frame has measured since the last overload. + capacity_kbps: Option, + overloads: Vec, + /// Whether any feedback arrived since the last [`tick`](Self::tick). + heard: bool, +} + +/// Bits per millisecond, which is kbps, from bytes over microseconds. +fn kbps(bytes: u64, us: u64) -> u32 { + (bytes.saturating_mul(8_000) / us.max(1)).min(u64::from(u32::MAX)) as u32 +} + +impl PathModel { + pub fn new(start_kbps: u32, now_us: u64) -> Self { + Self { + allowed_kbps: start_kbps, + reason: Reason::Start, + started_us: now_us, + prev: None, + busy: None, + draining: false, + delivery_kbps: None, + capacity_kbps: None, + overloads: Vec::new(), + heard: false, + } + } + + pub fn allowed_kbps(&self) -> u32 { + self.allowed_kbps + } + pub fn reason(&self) -> Reason { + self.reason + } + pub fn capacity_kbps(&self) -> Option { + self.capacity_kbps + } + pub fn delivery_kbps(&self) -> Option { + self.delivery_kbps + } + + /// Take one frame's feedback. + pub fn observe(&mut self, frame: &FrameFeedback, now_us: u64) { + self.heard = true; + if frame.fragments >= 2 && frame.spread_us > 0 { + let measured = kbps( + u64::from(frame.bytes.saturating_sub(u32::from(frame.first_bytes))), + u64::from(frame.spread_us), + ); + self.capacity_kbps = Some(self.capacity_kbps.map_or(measured, |c| c.max(measured))); + } + + if let Some(prev) = self.prev.replace(*frame) { + if busy_between(&prev, frame) { + self.continue_busy(&prev, frame, now_us); + } else { + self.busy = None; + if self.draining { + self.draining = false; + if let Some(share) = self.delivery_kbps { + self.allowed_kbps = share; + self.reason = Reason::Measured; + } + } else { + self.step_up(now_us); + } + } + } + } + + fn continue_busy(&mut self, prev: &FrameFeedback, frame: &FrameFeedback, now_us: u64) { + let busy = self.busy.get_or_insert(Busy { + start_sent_us: prev.sent_us, + start_arrival_us: prev.first_arrival_us, + bytes: 0, + first_q_us: frame.first_q_us, + counted: false, + }); + busy.bytes += u64::from(prev.bytes); + let delivered = kbps( + busy.bytes, + frame.first_arrival_us.saturating_sub(busy.start_arrival_us), + ); + let sent = kbps(busy.bytes, frame.sent_us.saturating_sub(busy.start_sent_us)); + let growing = frame.first_q_us > busy.first_q_us; + if !(growing && sent > delivered) { + return; + } + // Drained at the rate it was built: as far below the share as the + // sender went above it, for as long as the run has lasted. + self.allowed_kbps = (2 * u64::from(delivered)) + .saturating_sub(u64::from(sent)) + .max(1) as u32; + self.delivery_kbps = Some(delivered); + self.draining = true; + self.reason = Reason::Draining; + if !busy.counted { + busy.counted = true; + self.overloads.push(Overload { + at_us: now_us, + sent_kbps: sent, + }); + // What the frames measured above this share was not true of the + // path as it is now; it is measured again from here. + self.capacity_kbps = Some(delivered); + } + } + + /// Raise the rate to the highest the path has proven it carries. + fn step_up(&mut self, now_us: u64) { + let Some(capacity) = self.capacity_kbps else { + return; + }; + if capacity <= self.allowed_kbps { + return; + } + // The capacity frames measured, or the highest rate below it that the + // path has gone quiet at for long enough to be trusted. + let mut candidates: Vec = self + .overloads + .iter() + .map(|o| o.sent_kbps) + .filter(|&r| r > self.allowed_kbps && r < capacity) + .chain(std::iter::once(capacity)) + .collect(); + candidates.sort_unstable_by(|a, b| b.cmp(a)); + let Some(&to) = candidates.iter().find(|&&r| self.proven(r, now_us)) else { + return; + }; + if to > self.allowed_kbps { + self.allowed_kbps = to; + self.reason = Reason::Stepped; + } + } + + /// Whether `rate` has been quiet long enough: since the last overload at or + /// below it, longer than the longest gap between two such overloads. + fn proven(&self, rate: u32, now_us: u64) -> bool { + let mut last = self.started_us; + let mut longest = 0; + let mut any = false; + for o in self.overloads.iter().filter(|o| o.sent_kbps <= rate) { + longest = longest.max(o.at_us.saturating_sub(last)); + last = o.at_us; + any = true; + } + !any || now_us.saturating_sub(last) > longest + } + + /// Once a second. A second in which no frame came back is itself a + /// measurement: whatever was sent did not arrive, so the rate goes to the + /// last share the path proved, as for a queue. + pub fn tick(&mut self) { + if !std::mem::take(&mut self.heard) && self.prev.is_some() { + if let Some(share) = self.delivery_kbps { + self.allowed_kbps = self.allowed_kbps.min(share); + } + self.reason = Reason::Silent; + } + } +} + +/// Whether the bottleneck went from `prev`'s train straight into `frame`'s. +/// +/// On a link that never idled, `frame`'s first fragment lands exactly its own +/// serialisation after `prev`'s last, at the rate a train measured. So the +/// test is whether it landed no later than that. With no train long enough to +/// show a rate there is nothing to compare with, and two single-fragment frames +/// cannot tell a busy link from an idle one. +fn busy_between(prev: &FrameFeedback, frame: &FrameFeedback) -> bool { + let per_byte = [prev, frame] + .into_iter() + .find(|f| f.fragments >= 2 && f.bytes > u32::from(f.first_bytes)) + .map(|f| { + ( + u64::from(f.spread_us), + u64::from(f.bytes - u32::from(f.first_bytes)), + ) + }); + let Some((spread, bytes)) = per_byte else { + return false; + }; + let first = (u64::from(frame.first_bytes) * spread).div_ceil(bytes); + let prev_end = prev.first_arrival_us + u64::from(prev.spread_us); + frame.first_arrival_us <= prev_end + first +} + +#[cfg(test)] +mod tests { + use super::*; + + const FRAGMENT: u64 = 1_200; + const INTERVAL_US: u64 = 16_667; + const PROPAGATION_US: u64 = 40_000; + + /// A path with a FIFO bottleneck, fed one frame at a time. + struct Bottleneck { + /// What the bottleneck carries for us, in kbps, by time. + capacity: Box u32>, + /// Extra bytes other traffic puts into it, by frame. + cross: Box u64>, + /// When the bottleneck is next free. + free_at: u64, + seq: u16, + } + + impl Bottleneck { + fn new(capacity: impl Fn(u64) -> u32 + 'static) -> Self { + Self { + capacity: Box::new(capacity), + cross: Box::new(|_| 0), + free_at: 0, + seq: 0, + } + } + + /// Send a frame of `bytes` at `sent_us`; return what the client says. + fn send(&mut self, n: u64, sent_us: u64, bytes: u64) -> FrameFeedback { + let c = u64::from((self.capacity)(sent_us)).max(1); + let at = sent_us + PROPAGATION_US; + // Cross traffic lands just ahead of the frame. + self.free_at = self.free_at.max(at) + (self.cross)(n) * 8_000 / c; + let start = self.free_at.max(at); + let first_q = start - at; + let fragments = bytes.div_ceil(FRAGMENT).max(1); + let serialise = |b: u64| b * 8_000 / c; + let first_done = start + serialise(FRAGMENT.min(bytes)); + let last_done = start + serialise(bytes); + self.free_at = last_done; + self.seq = self.seq.wrapping_add(1); + FrameFeedback { + seq: self.seq, + bytes: bytes as u32, + fragments: fragments as u16, + first_bytes: FRAGMENT.min(bytes) as u16, + sent_us, + first_arrival_us: first_done, + spread_us: (last_done - first_done) as u32, + first_q_us: first_q as u32, + } + } + } + + /// Run `frames` frames at whatever the model allows, the encoder holding + /// its target as an average, as a CBR encoder does. `keyframe_every` + /// frames, one is 8x a delta, and the deltas around it are smaller by what + /// it took. + fn run( + path: &mut Bottleneck, + model: &mut PathModel, + frames: u64, + keyframe_every: Option, + ) -> Vec<(u32, u32)> { + let mut out = Vec::new(); + for n in 0..frames { + let sent = n * INTERVAL_US; + let budget = u64::from(model.allowed_kbps()) * INTERVAL_US / 8_000; + let size = match keyframe_every { + Some(gop) => { + let delta = budget * gop / (gop + 7); + if n % gop == 0 { delta * 8 } else { delta } + } + None => budget, + }; + let fb = path.send(n, sent, size.max(1)); + model.observe(&fb, fb.first_arrival_us); + out.push((model.allowed_kbps(), fb.first_q_us)); + } + out + } + + #[test] + fn a_rate_above_the_path_comes_down_to_what_it_delivers() { + let mut path = Bottleneck::new(|_| 6_000); + let mut model = PathModel::new(10_000, 0); + let trace = run(&mut path, &mut model, 600, None); + let (settled, q) = *trace.last().unwrap(); + assert!( + (5_400..=6_000).contains(&settled), + "settled at {settled} kbps on a 6000 kbps path" + ); + assert!(q < 2_000, "the queue never drained: {q} us"); + let first_low = trace.iter().position(|(r, _)| *r <= 6_000).unwrap(); + assert!(first_low < 20, "took {first_low} frames to come down"); + } + + #[test] + fn a_rate_below_the_path_steps_up_to_what_frames_measured() { + let mut path = Bottleneck::new(|_| 12_000); + let mut model = PathModel::new(3_000, 0); + let trace = run(&mut path, &mut model, 120, None); + let (settled, q) = *trace.last().unwrap(); + assert!( + (11_000..=12_050).contains(&settled), + "settled at {settled} kbps on a 12000 kbps path" + ); + assert!(q < 2_000, "stepping up built a queue: {q} us"); + } + + #[test] + fn a_keyframe_is_not_mistaken_for_overload() { + let mut path = Bottleneck::new(|_| 12_000); + let mut model = PathModel::new(6_000, 0); + run(&mut path, &mut model, 600, Some(120)); + assert!( + model.overloads.is_empty(), + "{} overloads", + model.overloads.len() + ); + } + + #[test] + fn a_capacity_drop_is_followed_down_and_back_up_without_a_sawtooth() { + let mut path = Bottleneck::new(|t| { + if (10_000_000..20_000_000).contains(&t) { + 4_000 + } else { + 9_000 + } + }); + let mut model = PathModel::new(9_000, 0); + let trace = run(&mut path, &mut model, 60 * 40, None); + let during = trace[60 * 15].0; + assert!( + (3_600..=4_000).contains(&during), + "{during} kbps during the drop" + ); + let after = trace.last().unwrap().0; + assert!( + (8_000..=9_000).contains(&after), + "{after} kbps after the drop" + ); + assert!( + model.overloads.len() <= 4, + "{} overloads: a sawtooth", + model.overloads.len() + ); + } + + #[test] + fn recurring_cross_traffic_is_learned_and_not_stepped_into_forever() { + let mut path = Bottleneck::new(|_| 10_000); + // Another flow sends 60 KB every 5 s. + path.cross = Box::new(|n| if n % 300 == 150 { 60_000 } else { 0 }); + let mut model = PathModel::new(9_000, 0); + run(&mut path, &mut model, 60 * 120, None); + let second_minute = model + .overloads + .iter() + .filter(|o| o.at_us > 60_000_000) + .count(); + assert!( + second_minute <= 2, + "{second_minute} overloads in the second minute" + ); + } + + #[test] + fn silence_holds_the_rate_at_the_last_proven_share() { + let mut path = Bottleneck::new(|_| 6_000); + let mut model = PathModel::new(10_000, 0); + run(&mut path, &mut model, 300, None); + model.tick(); + model.tick(); + assert_eq!(model.reason(), Reason::Silent); + assert!(model.allowed_kbps() <= model.delivery_kbps().unwrap()); + } +} diff --git a/apps/neshub/src/session.rs b/apps/neshub/src/session.rs index 8e377516..2ae70787 100644 --- a/apps/neshub/src/session.rs +++ b/apps/neshub/src/session.rs @@ -188,6 +188,14 @@ impl ClientSession { // it is sent, as for any client waiting on one. self.awaiting_keyframe.store(true, Ordering::Relaxed); let _ = self.idr_cmd_tx.send(vec![MSG_IDR_REQUEST]); + // The client says how each frame arrived, back on this same + // connection; the rate is decided on each one as it comes. + self.tasks.push(tokio::spawn(run_feedback_reader( + conn.clone(), + self.client, + self.controller.clone(), + self.idr_cmd_tx.clone(), + ))); let relay_ms = self.relay_ms.clone(); let awaiting = self.awaiting_keyframe.clone(); let withheld = self.withheld.clone(); @@ -890,6 +898,37 @@ async fn run_stats_sender(conn: Connection, mut rx: tokio::sync::mpsc::Unbounded debug!("stats sender exiting"); } +/// Read a client's per-frame feedback and decide the rate on each frame. +/// +/// Acted on here, as it arrives, rather than collected for a once-a-second +/// tick: a queue that starts building is answered a frame interval later, not +/// up to a second later, which at 60 frames a second is the difference +/// between one frame's queue and sixty. +async fn run_feedback_reader( + conn: Connection, + client: iroh::EndpointId, + controller: Arc>, + cmd_tx: tokio::sync::mpsc::UnboundedSender>, +) { + loop { + let datagram = match conn.read_datagram().await { + Ok(d) => d, + Err(e) => { + debug!("feedback reader ended: {e}"); + return; + } + }; + let Some(frame) = nesprotocol::datagram::decode_feedback(&datagram) else { + continue; + }; + let now_us = nesprotocol::clock::monotonic_us(); + let mut controller = controller.lock().await; + if let Some(kbps) = controller.on_feedback(client, &frame, now_us) { + crate::control::actuate(&cmd_tx, &controller, kbps); + } + } +} + /// Whether a broadcast video payload is a keyframe. /// /// The payload is `[1B codec][1B flags][4B ts][2B w][2B h][data]`, the same @@ -1266,6 +1305,18 @@ impl SessionManager { .count() } + /// Which clients have somewhere to send a picture: the paths the rate + /// controller has to answer to. + pub async fn streaming_clients(&self) -> Vec { + self.sessions + .lock() + .await + .iter() + .filter(|(_, s)| s.is_streaming()) + .map(|(id, _)| *id) + .collect() + } + /// Opus actually received from neswire since the last call, in kbps. /// /// Measured, not configured. The reported figure used to be diff --git a/crates/nesprotocol/src/datagram.rs b/crates/nesprotocol/src/datagram.rs index 8bb64ef1..bac7523f 100644 --- a/crates/nesprotocol/src/datagram.rs +++ b/crates/nesprotocol/src/datagram.rs @@ -49,6 +49,75 @@ pub const DGRAM_AUDIO: u8 = 1; /// [`crate::pyrowave`]. pub const DGRAM_PYROWAVE: u8 = crate::pyrowave::DGRAM_PYROWAVE; +/// One completed video frame as the client saw it arrive, client to hub on the +/// video connection. +/// +/// Sent once per frame, so the hub hears about every frame a round trip after +/// it left rather than once a second. What the hub's rate decision rests on: +/// whether the bottleneck was still busy with the last frame when this one +/// reached it, and how fast it delivered this one's fragments. See nescore's +/// `docs/superpowers/specs/2026-10-16-video-rate-from-the-path-design.md`. +pub const DGRAM_FEEDBACK: u8 = 3; + +/// `[1B kind][2B seq][4B bytes][2B fragments][2B first_bytes][8B sent_us] +/// [8B first_arrival_us][4B spread_us][4B first_q_us]`, little-endian. +pub const FEEDBACK_LEN: usize = 35; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct FrameFeedback { + /// The frame's sequence number. + pub seq: u16, + /// Datagram payload bytes that arrived for it, header excluded. + pub bytes: u32, + /// Fragments that arrived. + pub fragments: u16, + /// Payload bytes of the first fragment to arrive, which the arrival spread + /// does not cover: the spread starts when it has landed. + pub first_bytes: u16, + /// When the hub sent the frame, from its datagrams' `sent_us`. + pub sent_us: u64, + /// When the first fragment arrived, on the client's own clock. Only ever + /// compared with another frame's, never with the hub's clock. + pub first_arrival_us: u64, + /// The last fragment's arrival minus the first's. + pub spread_us: u32, + /// Queueing delay of the first fragment on the path (`delay`). + pub first_q_us: u32, +} + +pub fn encode_feedback(f: &FrameFeedback) -> [u8; FEEDBACK_LEN] { + let mut b = [0u8; FEEDBACK_LEN]; + b[0] = DGRAM_FEEDBACK; + b[1..3].copy_from_slice(&f.seq.to_le_bytes()); + b[3..7].copy_from_slice(&f.bytes.to_le_bytes()); + b[7..9].copy_from_slice(&f.fragments.to_le_bytes()); + b[9..11].copy_from_slice(&f.first_bytes.to_le_bytes()); + b[11..19].copy_from_slice(&f.sent_us.to_le_bytes()); + b[19..27].copy_from_slice(&f.first_arrival_us.to_le_bytes()); + b[27..31].copy_from_slice(&f.spread_us.to_le_bytes()); + b[31..35].copy_from_slice(&f.first_q_us.to_le_bytes()); + b +} + +/// `None` for anything that is not a whole feedback datagram. +pub fn decode_feedback(b: &[u8]) -> Option { + if b.len() < FEEDBACK_LEN || b[0] != DGRAM_FEEDBACK { + return None; + } + let u16_at = |o: usize| u16::from_le_bytes([b[o], b[o + 1]]); + let u32_at = |o: usize| u32::from_le_bytes(b[o..o + 4].try_into().unwrap()); + let u64_at = |o: usize| u64::from_le_bytes(b[o..o + 8].try_into().unwrap()); + Some(FrameFeedback { + seq: u16_at(1), + bytes: u32_at(3), + fragments: u16_at(7), + first_bytes: u16_at(9), + sent_us: u64_at(11), + first_arrival_us: u64_at(19), + spread_us: u32_at(27), + first_q_us: u32_at(31), + }) +} /// Bounds reassembly work. A 4K keyframe runs to a few hundred fragments at a /// typical MTU; anything past this did not come from a sender of ours. pub const MAX_FRAGMENTS_PER_FRAME: usize = 4096; @@ -214,6 +283,30 @@ mod tests { assert!(decode_datagram(&buf).is_none()); } + #[test] + fn feedback_round_trips() { + let f = FrameFeedback { + seq: 0xABCD, + bytes: 17_000, + fragments: 15, + first_bytes: 1_180, + sent_us: 0x0102_0304_0506_0708, + first_arrival_us: 0x1112_1314_1516_1718, + spread_us: 9_800, + first_q_us: 450, + }; + let wire = encode_feedback(&f); + assert_eq!(decode_feedback(&wire), Some(f)); + assert_eq!(decode_feedback(&wire[..FEEDBACK_LEN - 1]), None); + let mut other = wire; + other[0] = DGRAM_VIDEO; + assert_eq!( + decode_feedback(&other), + None, + "a video datagram read as feedback" + ); + } + #[test] fn empty_body_still_takes_one_fragment() { assert_eq!(fragment_count(0, 1000), 1); diff --git a/crates/nesprotocol/src/stats.rs b/crates/nesprotocol/src/stats.rs index 5879319e..46dab1e9 100644 --- a/crates/nesprotocol/src/stats.rs +++ b/crates/nesprotocol/src/stats.rs @@ -120,12 +120,18 @@ pub struct VideoBreakdown { /// eight seconds old: loss cannot show a queue, only its overflow. Anything /// but near-zero here means latency is being spent on backlog. pub backlog_ms: u16, + /// The bottleneck capacity the controller has measured from frame + /// dispersion, in kbps; 0 before any frame has measured it. + pub capacity_kbps: u32, + /// The rate the path last delivered while it was queueing, in kbps: the + /// share it proved it could carry. 0 before it has queued at all. + pub delivery_kbps: u32, } /// `[4B key_bps][4B delta_bps][1B keyframes][4B target][4B ceiling][1B reason] /// [1B manual][4B box_ceiling][2B pipeline_p50][2B pipeline_p95][2B pipeline_max] -/// [2B backlog_ms]` -pub const VIDEO_BREAKDOWN_LEN: usize = 31; +/// [2B backlog_ms][4B capacity_kbps][4B delivery_kbps]` +pub const VIDEO_BREAKDOWN_LEN: usize = 39; pub fn encode_video_breakdown(buf: &mut Vec, b: &VideoBreakdown) { buf.reserve(VIDEO_BREAKDOWN_LEN); @@ -141,6 +147,8 @@ pub fn encode_video_breakdown(buf: &mut Vec, b: &VideoBreakdown) { buf.extend_from_slice(&b.pipeline_p95_ms.to_le_bytes()); buf.extend_from_slice(&b.pipeline_max_ms.to_le_bytes()); buf.extend_from_slice(&b.backlog_ms.to_le_bytes()); + buf.extend_from_slice(&b.capacity_kbps.to_le_bytes()); + buf.extend_from_slice(&b.delivery_kbps.to_le_bytes()); } /// Decoded stats from any source. @@ -211,6 +219,8 @@ pub fn decode_stats(msg_type: u8, data: &[u8], stats: &mut PipelineStats) { pipeline_p95_ms: u16::from_le_bytes([d[25], d[26]]), pipeline_max_ms: u16::from_le_bytes([d[27], d[28]]), backlog_ms: u16::from_le_bytes([d[29], d[30]]), + capacity_kbps: u32_at(31), + delivery_kbps: u32_at(35), }); } } @@ -236,6 +246,8 @@ mod breakdown_tests { pipeline_p95_ms: 24, pipeline_max_ms: 61, backlog_ms: 40, + capacity_kbps: 23_500, + delivery_kbps: 7_800, } } diff --git a/docs/media-transport.md b/docs/media-transport.md index 058caab8..e741ff89 100644 --- a/docs/media-transport.md +++ b/docs/media-transport.md @@ -124,6 +124,18 @@ rate, while the pacer underneath is using the measured window and RTT. If the application is producing faster than the pacer drains, the answer is to produce less — which is the bitrate controller's job — not to add a second pacer. +**Since 2026-10-16 the hub turns this pacer off** (`neshub/src/congestion.rs`). +Its controller reports a window above `u32::MAX`, past which noq does not pace, +and ignores loss. The video rate is decided per frame from client feedback +(`neshub/src/rate.rs`), and that needs each frame to leave as a back-to-back +train, so the bottleneck spaces the fragments by its own rate and the client can +measure it. The transport pacing by its window estimate would smear exactly that +spacing. The rate is still produced to fit the path; it is the media's +controller that decides it, on the path's queueing delay, which leads loss and +RTT by tens of milliseconds. Transport settings are per endpoint, so the hub's +audio, input and control connections are unpaced too: they are small flows the +video rate leaves room for. + ## Loss is a lagging indicator; the queue is the leading one A path that is overrun does not necessarily drop anything. It can simply be