From 9269ee5d51c34cf00ec6cdb0fb7814efc7d96fbf Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Sat, 10 Oct 2026 15:34:23 +0300 Subject: [PATCH] fix(neshub): an encoder that missed the codec choice is told again The codec the clients agree on goes to the encoder over a socket that does not exist until the game draws, and a client can connect first. The command was dropped, the encoder started on AV1, and a client without AV1 sat on a splash with sound and no picture. Commands now wait for the socket, and a frame in a codec no client decodes sends the agreed set again, which also covers a launcher handing over to a game with an encoder of its own. --- apps/neshub/src/e2e_tests.rs | 10 ++++- apps/neshub/src/ipc_listener.rs | 11 +++++- apps/neshub/src/main.rs | 26 ++++++++++--- apps/neshub/src/negotiation.rs | 69 +++++++++++++++++++++++++++++++++ apps/neshub/src/session.rs | 10 +++++ 5 files changed, 119 insertions(+), 7 deletions(-) diff --git a/apps/neshub/src/e2e_tests.rs b/apps/neshub/src/e2e_tests.rs index d2294406..d474c0a4 100644 --- a/apps/neshub/src/e2e_tests.rs +++ b/apps/neshub/src/e2e_tests.rs @@ -80,7 +80,15 @@ async fn rig_with(client_buffer: Option) -> 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) diff --git a/apps/neshub/src/ipc_listener.rs b/apps/neshub/src/ipc_listener.rs index 4fa38e04..c9b83731 100644 --- a/apps/neshub/src/ipc_listener.rs +++ b/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) { +pub async fn run_video_listener( + socket_path: PathBuf, + session_manager: Arc, + controller: Arc>, + cmd_tx: tokio::sync::mpsc::UnboundedSender>, +) { 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 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(); diff --git a/apps/neshub/src/negotiation.rs b/apps/neshub/src/negotiation.rs index 7adbbbef..ca650d47 100644 --- a/apps/neshub/src/negotiation.rs +++ b/apps/neshub/src/negotiation.rs @@ -94,12 +94,15 @@ pub fn effective_caps(caps: ClientCaps, pyrowave_format: Option) -> ClientCa /// on it. pub struct Negotiator { agreement: std::sync::Mutex>, + /// When the encoder was last told the set again; see [`Self::reassert`]. + reasserted: std::sync::Mutex>, } 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, + cmd_tx: &tokio::sync::mpsc::UnboundedSender>, + ) { + 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"); + } } diff --git a/apps/neshub/src/session.rs b/apps/neshub/src/session.rs index 5ec15e5f..9e5742bf 100644 --- a/apps/neshub/src/session.rs +++ b/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, + cmd_tx: &tokio::sync::mpsc::UnboundedSender>, + ) { + 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.