fix(neshub): an encoder that missed the codec choice is told again (#389)

A client could connect and state its codecs before the game's encoder
was listening; the command was dropped, the encoder started on AV1, and
a client without AV1 got sound and no picture. Commands now wait for the
encoder's socket, and a frame in a codec no client decodes sends the
agreed set again.

apps/neshub/src/e2e_tests.rs

@@ -80,7 +80,15 @@ async fn rig_with(client_buffer: Option<usize>) -> Rig {
     tokio::spawn({
         let mgr = mgr.clone();
         let path = socket_path.clone();
-        async move { crate::ipc_listener::run_video_listener(path, mgr).await }
+        let controller = Arc::new(tokio::sync::Mutex::new(crate::control::Controller::new(
+            crate::control::Limits::new(8000),
+        )));
+        // Commands go nowhere here: no client in these tests states codecs.
+        let (cmd_tx, _cmd_rx) = tokio::sync::mpsc::unbounded_channel();
+        async move {
+            let _cmd_rx = _cmd_rx;
+            crate::ipc_listener::run_video_listener(path, mgr, controller, cmd_tx).await
+        }
     });
 
     let hub = iroh::Endpoint::builder(presets::N0)

apps/neshub/src/ipc_listener.rs

@@ -22,7 +22,12 @@ fn set_recv_buffer(socket: &UnixDatagram, size: libc::c_int) {
     }
 }
 
-pub async fn run_video_listener(socket_path: PathBuf, session_manager: Arc<SessionManager>) {
+pub async fn run_video_listener(
+    socket_path: PathBuf,
+    session_manager: Arc<SessionManager>,
+    controller: Arc<tokio::sync::Mutex<crate::control::Controller>>,
+    cmd_tx: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
+) {
     if socket_path.exists() {
         let _ = std::fs::remove_file(&socket_path);
     }
@@ -77,6 +82,10 @@ pub async fn run_video_listener(socket_path: PathBuf, session_manager: Arc<Sessi
                         );
                         continue;
                     }
+                    // An encoder on a codec nobody here decodes is told again.
+                    session_manager
+                        .reassert_codec(decoded.codec, &controller, &cmd_tx)
+                        .await;
                     let mut frame_data = Vec::with_capacity(1 + decoded.data.len());
                     frame_data.push(decoded.codec);
                     frame_data.push(decoded.flags);

apps/neshub/src/main.rs

@@ -235,10 +235,26 @@ async fn main() -> Result<()> {
             let cmd_path = std::path::PathBuf::from("/tmp/nescapture-cmd.sock");
             while let Some(mut bytes) = cmd_rx.recv().await {
                 fill_pyrowave_packet_size(&mut bytes, mgr.pyro_packet_size().await);
-                if let Ok(sock) = std::os::unix::net::UnixDatagram::unbound() {
-                    if sock.send_to(&bytes, &cmd_path).is_err() {
-                        tracing::warn!("nescapture cmd send failed at {}", cmd_path.display());
+                let Ok(sock) = std::os::unix::net::UnixDatagram::unbound() else {
+                    continue;
+                };
+                // The encoder's socket is not there until the game draws its
+                // first frame, and a client can connect and say what it decodes
+                // before that. A command sent then used to be lost -- the codec
+                // one with it, leaving the encoder on a default the client may
+                // not decode -- so a command waits for the socket, up to a
+                // minute, and the ones behind it wait their turn.
+                let mut waited = 0u32;
+                while sock.send_to(&bytes, &cmd_path).is_err() {
+                    if waited == 0 {
+                        tracing::info!("nescapture is not listening at {} yet; waiting", cmd_path.display());
                     }
+                    waited += 1;
+                    if waited > 600 {
+                        tracing::warn!("nescapture cmd send failed at {} for a minute; dropped", cmd_path.display());
+                        break;
+                    }
+                    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
                 }
             }
         });
@@ -417,8 +433,8 @@ async fn main() -> Result<()> {
     let stats_ipc = args.stats_ipc.clone();
     let stats_tx_clone = nescope_stats_tx.clone();
     tokio::spawn({
-        let mgr = session_manager.clone();
-        async move { ipc_listener::run_video_listener(video_ipc, mgr).await }
+        let (mgr, controller, cmd_tx) = (session_manager.clone(), controller.clone(), cmd_tx.clone());
+        async move { ipc_listener::run_video_listener(video_ipc, mgr, controller, cmd_tx).await }
     });
     tokio::spawn({
         let mgr = session_manager.clone();

apps/neshub/src/negotiation.rs

@@ -94,12 +94,15 @@ pub fn effective_caps(caps: ClientCaps, pyrowave_format: Option<u8>) -> ClientCa
 /// on it.
 pub struct Negotiator {
     agreement: std::sync::Mutex<Agreement<iroh::EndpointId>>,
+    /// When the encoder was last told the set again; see [`Self::reassert`].
+    reasserted: std::sync::Mutex<Option<std::time::Instant>>,
 }
 
 impl Negotiator {
     pub fn new() -> Self {
         Self {
             agreement: std::sync::Mutex::new(Agreement::default()),
+            reasserted: std::sync::Mutex::new(None),
         }
     }
 
@@ -162,6 +165,46 @@ impl Negotiator {
         nesprotocol::encode_client_caps(&mut cmd, joint);
         let _ = cmd_tx.send(cmd);
     }
+
+    /// A frame arrived in `codec`. When the clients agreed on a set that
+    /// cannot decode it, the encoder never got the set: it was not listening
+    /// yet when the set was sent, or it is a new one -- a launcher handing over
+    /// to the game -- that started on its own default. Either way every client
+    /// is dropping every frame, so the encoder is told again, at most once a
+    /// second.
+    pub async fn reassert(
+        &self,
+        codec: u8,
+        controller: &Mutex<Controller>,
+        cmd_tx: &tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
+    ) {
+        let joint = {
+            let agreement = self.agreement.lock().unwrap();
+            match agreement.joint() {
+                Some(joint) if !joint.is_empty() && !joint.supports_codec(codec) => joint,
+                _ => return,
+            }
+        };
+        {
+            let mut last = self.reasserted.lock().unwrap();
+            if last.is_some_and(|t| t.elapsed() < std::time::Duration::from_secs(1)) {
+                return;
+            }
+            *last = Some(std::time::Instant::now());
+        }
+        if controller.lock().await.mode() == ControlMode::Manual {
+            return;
+        }
+        warn!(
+            codec,
+            joint = format_args!("{:#010b}", joint.bits()),
+            "the encoder sends a codec no client decodes; telling it again what they can"
+        );
+        self.agreement.lock().unwrap().mark_sent(joint);
+        let mut cmd = vec![nesprotocol::MSG_CLIENT_CAPS];
+        nesprotocol::encode_client_caps(&mut cmd, joint);
+        let _ = cmd_tx.send(cmd);
+    }
 }
 
 #[cfg(test)]
@@ -266,4 +309,30 @@ mod tests {
             caps
         );
     }
+
+    /// The encoder came up on AV1 after a client that decodes only H.264 had
+    /// already said so: the set is sent again, once, however many frames say
+    /// it.
+    #[tokio::test]
+    async fn an_encoder_on_a_codec_nobody_decodes_is_told_again() {
+        let negotiator = Negotiator::new();
+        let controller = Mutex::new(Controller::new(crate::control::Limits::new(8000)));
+        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+        let client = iroh::SecretKey::from_bytes(&[7; 32]).public();
+        negotiator.state(client, h264_only());
+        negotiator.settle(&controller, &tx).await;
+        assert!(rx.try_recv().is_ok(), "the first set is sent");
+
+        // Frames in a codec the client decodes say nothing.
+        negotiator.reassert(CODEC_H264, &controller, &tx).await;
+        assert!(rx.try_recv().is_err());
+
+        // AV1 frames: told again, and only once for a burst of them.
+        for _ in 0..60 {
+            negotiator.reassert(CODEC_AV1, &controller, &tx).await;
+        }
+        let cmd = rx.try_recv().expect("the set is sent again");
+        assert_eq!(cmd[0], nesprotocol::MSG_CLIENT_CAPS);
+        assert!(rx.try_recv().is_err(), "once a second, not once a frame");
+    }
 }

apps/neshub/src/session.rs

@@ -975,6 +975,16 @@ impl SessionManager {
         }
     }
 
+    /// A video frame arrived in `codec`; see [`Negotiator::reassert`].
+    pub async fn reassert_codec(
+        &self,
+        codec: u8,
+        controller: &Mutex<Controller>,
+        cmd_tx: &tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
+    ) {
+        self.negotiator.reassert(codec, controller, cmd_tx).await;
+    }
+
     /// Drop a client, and let the encoder move to whatever the clients still
     /// here can all decode -- a better codec, when the one that left was the
     /// one holding the rest back.