feat(protocol): a box is told what it may spend on video, and the receiver can say what it got

Three additions, all of them plumbing for a bitrate that something
actually decides.

`BootDescriptor` gains `VideoLimits`. It rides the boot document rather
than a kernel command line because `nesinit` handles `Boot` by mounting
and *then* bringing the stack up, so the value is in hand before `neshub`
is spawned -- no parsing, no window where the service is running without
its configuration. It is on the descriptor rather than a launch because
its consumer is a service that comes up with the box; geometry went the
other way for the same reason, its consumer being started per launch.

`bitrate_kbps` is an `Option` and the distinction is load-bearing.
"Nobody said" is not zero and is not unlimited, and a reader that
conflates the first with the last reproduces the bug exactly: every
session offered 10 Mbps because no number had ever been chosen and the
encoder's own default stood in for one. `neshub` now says which it got,
and falls back to something modest rather than to whatever it finds.

Note what `deny_unknown_fields` means here, since it is deliberate: a
host that sends `video` to a guest too old to know the field is refused
rather than quietly served. That is the right direction to fail -- the
alternative is a box that boots, streams, and ignores its ceiling -- and
it means the guest image is rebuilt before a host starts sending one.

`MSG_RECEIVER_REPORT` carries what a second looked like from the far
end: goodput actually released to the decoder, frames released,
incomplete and never-arrived, and the receiver's own RTT. The hub cannot
work any of this out for itself. Its own view was measured saying the
path was healthy while almost nothing was arriving, and one reason is
structural -- `send_datagram` evicts the oldest queued datagrams and
returns `Ok`, so the send side has no backpressure signal at all.

`MSG_CONTROL_MODE` says who is choosing the bitrate. Manual exists
because it is how this class of bug gets found: the original report said
the bitrate had already been lowered, and the only way anyone
established otherwise was by setting one by hand and watching the
picture come back.

Both decoders refuse what they cannot read rather than guessing.
`loss()` returns `None` for a second that accounted for no frames at
all, because a second with nothing sent and a second with nothing
arriving are indistinguishable from there, and answering either 0% or
100% would tell a controller something nobody knows.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

apps/neshub/src/main.rs

@@ -73,6 +73,16 @@ struct Args {
     #[arg(long, env = "NESTRI_AUDIO_BITRATE", default_value_t = 64)]
     audio_bitrate_per_channel: u32,
 
+    /// Ceiling on the video bitrate, in kbps.
+    ///
+    /// Set by `nesinit` from the boot descriptor's video limits, which come
+    /// from the tier the box was sized for. Absent means nobody said -- which is
+    /// not a licence to send whatever the encoder defaults to, since that is
+    /// precisely how every session came to offer 10 Mbps regardless of what the
+    /// path could carry. Unset is reported, and a conservative ceiling is used.
+    #[arg(long, env = "NESTRI_MAX_BITRATE")]
+    max_bitrate_kbps: Option<u32>,
+
     /// Socket nescope sends screenshots on. neshub listens; nescope dials out.
     #[arg(
         long,
@@ -82,6 +92,14 @@ struct Args {
     screenshot_ipc: PathBuf,
 }
 
+/// What to assume when nobody said.
+///
+/// Deliberately modest. A ceiling that was never set should not behave like an
+/// unlimited one: the whole failure this exists to fix was a session offering
+/// 10 Mbps into a path carrying under three, because no number had ever been
+/// chosen and the encoder's own default stood in for one.
+const DEFAULT_MAX_BITRATE_KBPS: u32 = 4_000;
+
 #[tokio::main]
 async fn main() -> Result<()> {
     tracing_subscriber::fmt()
@@ -117,6 +135,15 @@ async fn main() -> Result<()> {
         }
     }
 
+    match args.max_bitrate_kbps {
+        Some(kbps) => tracing::info!("video ceiling: {kbps} kbps, from the boot descriptor"),
+        None => tracing::warn!(
+            "no video ceiling on the boot descriptor; using {DEFAULT_MAX_BITRATE_KBPS} kbps. \
+             A box sized by a tier is told its ceiling -- if this is one, the descriptor did \
+             not carry it."
+        ),
+    }
+
     let endpoint = builder.bind().await?;
     let endpoint_addr = endpoint.addr();
     let ep_id = endpoint_addr.id;

apps/nesinit/src/services.rs

@@ -41,6 +41,7 @@ use tokio::sync::mpsc::{Receiver, Sender};
 
 use crate::reap::{Waiters, Watched};
 use crate::workload::Failure;
+use nesprotocol::lifecycle::VideoLimits;
 
 /// A service that died, and how.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -60,7 +61,13 @@ pub trait Services {
     /// Called once, after the shares are mounted and before anything may be
     /// launched. An empty stack is legitimate: a box with no services still
     /// boots, and a caller can still launch something that needs none.
-    fn bring_up(&mut self) -> Result<Vec<String>, Failure>;
+    ///
+    /// `video` comes from the descriptor and reaches the services that read it.
+    /// It has to arrive here rather than later because a service configured
+    /// after it is already running has a window in which it is not configured,
+    /// and for a bitrate ceiling that window is a session streaming at whatever
+    /// default it started with.
+    fn bring_up(&mut self, video: VideoLimits) -> Result<Vec<String>, Failure>;
 
     /// Deaths, as they happen.
     ///
@@ -308,6 +315,13 @@ pub struct Stack {
     running: Vec<(&'static str, Watched)>,
     deaths: Receiver<Died>,
     reported: Sender<Died>,
+    /// What the host said this box may spend on video, from the descriptor.
+    ///
+    /// Held here because `spawn` is where it reaches a service, and `spawn`
+    /// takes a `&'static Service` whose `env` is a fixed table -- a value that
+    /// arrives at runtime has no route through it otherwise. The same problem
+    /// `RUST_LOG` has, solved the same way.
+    video: VideoLimits,
 }
 
 impl Stack {
@@ -328,6 +342,7 @@ impl Stack {
             running: Vec::new(),
             deaths,
             reported,
+            video: VideoLimits::default(),
         }
     }
 
@@ -390,6 +405,15 @@ impl Stack {
         if let Ok(filter) = std::env::var("RUST_LOG") {
             command.env("RUST_LOG", filter);
         }
+        // The descriptor's video limits, for the services that read them. Same
+        // shape of problem as `RUST_LOG` above -- `env_clear` drops everything
+        // and the service table is a fixed list of literals, so a value that
+        // only exists at runtime has no other route in. `neshub` reads this
+        // through the clap `env =` attribute it already uses for every other
+        // setting.
+        if let Some(kbps) = self.video.bitrate_kbps {
+            command.env("NESTRI_MAX_BITRATE", kbps.to_string());
+        }
         // The service's own entry last, so a service that states one of these
         // for itself wins over the defaults above.
         command.envs(service.env.iter().copied());
@@ -451,7 +475,8 @@ impl Stack {
 }
 
 impl Services for Stack {
-    fn bring_up(&mut self) -> Result<Vec<String>, Failure> {
+    fn bring_up(&mut self, video: VideoLimits) -> Result<Vec<String>, Failure> {
+        self.video = video;
         let mut up = Vec::new();
         // Lifted out so the loop does not hold a borrow of `self` across the
         // start it is asking for.
@@ -604,6 +629,9 @@ pub mod double {
     /// only thing under test.
     pub struct Double {
         pub brought_up: usize,
+        /// What the last `bring_up` was told, so a test can assert the limits
+        /// reached the stack rather than assuming they did.
+        pub video: VideoLimits,
         pub failure: Option<Failure>,
         pub names: Vec<String>,
         deaths: Receiver<Died>,
@@ -625,6 +653,7 @@ pub mod double {
                 names: vec!["dbus-system".into(), "neshub".into()],
                 deaths,
                 report,
+                video: VideoLimits::default(),
             }
         }
 
@@ -637,8 +666,9 @@ pub mod double {
     }
 
     impl Services for Double {
-        fn bring_up(&mut self) -> Result<Vec<String>, Failure> {
+        fn bring_up(&mut self, video: VideoLimits) -> Result<Vec<String>, Failure> {
             self.brought_up += 1;
+            self.video = video;
             match &self.failure {
                 Some(failure) => Err(failure.clone()),
                 None => Ok(self.names.clone()),

apps/nesinit/src/session.rs

@@ -311,7 +311,7 @@ where
                 // A box whose own services will not come up cannot be launched
                 // into, so this is refused rather than reported and carried on
                 // from — unlike a launch, which is the caller's to correct.
-                match services.bring_up() {
+                match services.bring_up(descriptor.video) {
                     Ok(up) => {
                         tracing::info!(services = up.len(), "the box is ready to be launched into");
                         send(&mut writer, &GuestToHost::Initialized { services: up }).await?
@@ -517,6 +517,7 @@ mod tests {
                 ro: false,
             }],
             drives: Vec::new(),
+            video: Default::default(),
         }
     }
 
@@ -715,6 +716,53 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn the_descriptors_video_limits_reach_the_services() {
+        // The ceiling is useless if it stops at the descriptor. `neshub` is the
+        // only thing that can enforce it and it is a service, so the number has
+        // to survive the whole way from the boot document to the spawn.
+        let (guest, host) = tokio::io::duplex(4096);
+        let mut caller = Caller::new(host);
+        let session = spawn(guest, Given::new(Double::exits_when_stopped(Exit::code(0))));
+
+        let mut given = descriptor();
+        given.video.bitrate_kbps = Some(8_000);
+
+        caller.expect_ready().await;
+        caller
+            .say(&HostToGuest::Boot {
+                descriptor: Box::new(given),
+            })
+            .await;
+        caller.expect_booted().await;
+        caller.say(&HostToGuest::Shutdown).await;
+
+        let (_, _, services) = session.await.unwrap();
+        assert_eq!(services.video.bitrate_kbps, Some(8_000));
+    }
+
+    #[tokio::test]
+    async fn a_box_told_nothing_about_video_says_so_rather_than_inventing_a_limit() {
+        // "Unsaid" must not arrive as a number. A stack that cannot tell the two
+        // apart cannot log that it was never told, and a ceiling nobody set is
+        // exactly how every session came to offer 10 Mbps.
+        let (guest, host) = tokio::io::duplex(4096);
+        let mut caller = Caller::new(host);
+        let session = spawn(guest, Given::new(Double::exits_when_stopped(Exit::code(0))));
+
+        caller.expect_ready().await;
+        caller
+            .say(&HostToGuest::Boot {
+                descriptor: Box::new(descriptor()),
+            })
+            .await;
+        caller.expect_booted().await;
+        caller.say(&HostToGuest::Shutdown).await;
+
+        let (_, _, services) = session.await.unwrap();
+        assert_eq!(services.video.bitrate_kbps, None);
+    }
+
     #[tokio::test]
     async fn a_launch_runs_what_it_names_and_is_reported_by_its_id() {
         let (guest, host) = tokio::io::duplex(4096);

apps/nesinit/tests/services_stop.rs

@@ -50,7 +50,7 @@ fn alive(pid: i32) -> bool {
 async fn a_stack_that_goes_away_takes_its_services_with_it() {
     let waiters = Waiters::new();
     let mut stack = Stack::from_table(waiters, SLEEPERS);
-    let up = stack.bring_up().expect("two sleeps did not start");
+    let up = stack.bring_up(Default::default()).expect("two sleeps did not start");
     assert_eq!(up.len(), 2);
 
     let pids = stack.pids();