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 <noreply@anthropic.com>

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<dyn Controller> {
+        Box::new(*self)
+    }
+
+    fn initial_window(&self) -> u64 {
+        UNPACED_WINDOW
+    }
+
+    fn into_any(self: Box<Self>) -> Box<dyn Any> {
+        self
+    }
+}
+
+#[derive(Debug, Default)]
+pub struct MediaRateFactory;
+
+impl ControllerFactory for MediaRateFactory {
+    fn build(self: Arc<Self>, _now: std::time::Instant, _current_mtu: u16) -> Box<dyn Controller> {
+        Box::new(MediaRate)
+    }
+}

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()
 }
 

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(),
                         },
                     );
                 }

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<FrameFeedback>,
+    busy: Option<Busy>,
+    draining: bool,
+    /// The share the path last delivered while queueing.
+    delivery_kbps: Option<u32>,
+    /// The highest capacity a frame has measured since the last overload.
+    capacity_kbps: Option<u32>,
+    overloads: Vec<Overload>,
+    /// 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<u32> {
+        self.capacity_kbps
+    }
+    pub fn delivery_kbps(&self) -> Option<u32> {
+        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<u32> = 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<dyn Fn(u64) -> u32>,
+        /// Extra bytes other traffic puts into it, by frame.
+        cross: Box<dyn Fn(u64) -> 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<u64>,
+    ) -> 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());
+    }
+}

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<Mutex<Controller>>,
+    cmd_tx: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
+) {
+    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<iroh::EndpointId> {
+        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

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<FrameFeedback> {
+    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);

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<u8>, b: &VideoBreakdown) {
     buf.reserve(VIDEO_BREAKDOWN_LEN);
@@ -141,6 +147,8 @@ pub fn encode_video_breakdown(buf: &mut Vec<u8>, 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,
         }
     }
 

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