From bec943faa6153ef14dde6d6eec38c8f55dc100f5 Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Mon, 19 Oct 2026 00:33:11 +0300 Subject: [PATCH] fix(neshub): a client learns its encoder from the frames it is sent The session's encoder pid was set only at subscribe time (from the encoder that had drawn the window) and on a primary switch for clients that never subscribed. A client that subscribed before the game drew its first frame, which is the real order, kept pid 0, so its pause was ignored. Also logs pause and resume at info in the hub and in nescapture. --- apps/nescapture/src/encode.rs | 5 +++++ apps/neshub/src/e2e_tests.rs | 22 +++++++++++++++++++++ apps/neshub/src/main.rs | 13 ++++++++++++- apps/neshub/src/session.rs | 36 ++++++++++++++++++++++++++++++++--- 4 files changed, 72 insertions(+), 4 deletions(-) diff --git a/apps/nescapture/src/encode.rs b/apps/nescapture/src/encode.rs index ec967fcd..332abd9c 100644 --- a/apps/nescapture/src/encode.rs +++ b/apps/nescapture/src/encode.rs @@ -1550,6 +1550,11 @@ pub struct EncodeSettingsChange { /// keyframe in the same step, so the first frame after it is one the client /// can decode: everything it saw before the pause is stale. fn apply_pause_message(msg: u8, paused: &AtomicBool, idr_requested: &AtomicBool) { + log::info!( + "window {} received (was {})", + if msg == MSG_WINDOW_PAUSE { "pause" } else { "resume" }, + if paused.load(Ordering::Relaxed) { "paused" } else { "running" } + ); if msg == MSG_WINDOW_PAUSE { if !paused.swap(true, Ordering::Relaxed) { log::info!("paused by the client: capture and encode stopped"); diff --git a/apps/neshub/src/e2e_tests.rs b/apps/neshub/src/e2e_tests.rs index 8220cae7..82b5659c 100644 --- a/apps/neshub/src/e2e_tests.rs +++ b/apps/neshub/src/e2e_tests.rs @@ -578,6 +578,28 @@ mod windows { assert!(got.iter().all(|g| (g.0, g.1) == (0, 0))); } + /// The order a real box has: the client subscribes to the window when it + /// opens, and the game's encoder draws its first frame afterwards. The + /// client's encoder must follow the frames, or a pause has nowhere to go + /// (found on a real box, ref(d-0087)). + #[tokio::test(flavor = "multi_thread")] + async fn a_client_that_subscribed_before_the_first_frame_learns_its_encoder() { + let rig = rig().await; + tell(&rig, |m| encode_event_message(m, &opened(0, "launch_a"))).await; + tell(&rig, |m| encode_window_pid(m, 0, GAME_A)).await; + let me = rig._client_ep.addr().id; + let (idr, _rx) = tokio::sync::mpsc::unbounded_channel(); + rig.mgr.subscribe(me, 0, &idr).await; + assert_eq!(rig.mgr.encoder_pid_of(&me).await, Some(0), "nothing has drawn yet"); + draw(&rig, GAME_A, 0, true).await; + draw(&rig, GAME_A, 1, false).await; + let _ = received(&rig.client).await; + assert_eq!(rig.mgr.encoder_pid_of(&me).await, Some(GAME_A)); + let handle = rig.mgr.pause_handle_of(&me).await.expect("a session"); + handle.pause(GAME_A); + assert!(rig.mgr.encoder_paused(GAME_A)); + } + #[tokio::test(flavor = "multi_thread")] async fn a_session_that_goes_while_paused_releases_its_pause() { let rig = rig().await; diff --git a/apps/neshub/src/main.rs b/apps/neshub/src/main.rs index 99eee293..e72fe0e8 100644 --- a/apps/neshub/src/main.rs +++ b/apps/neshub/src/main.rs @@ -305,7 +305,18 @@ async fn main() -> Result<()> { if destinations.first() != Some(&cmd_path) { for path in destinations { match sock.send_to(&bytes, &path) { - Ok(_) => {} + Ok(_) => { + if matches!( + bytes.first(), + Some(&nesprotocol::MSG_WINDOW_PAUSE | &nesprotocol::MSG_WINDOW_RESUME) + ) { + tracing::info!( + "sent {} to {}", + if bytes[0] == nesprotocol::MSG_WINDOW_PAUSE { "pause" } else { "resume" }, + path.display() + ); + } + } // Refused or absent: that process has gone. Err(e) if capture_cmd::encoder_is_gone(&e) => { tracing::info!("encoder at {} is gone: {e}", path.display()); diff --git a/apps/neshub/src/session.rs b/apps/neshub/src/session.rs index 4bcfd0ab..e3108251 100644 --- a/apps/neshub/src/session.rs +++ b/apps/neshub/src/session.rs @@ -863,10 +863,13 @@ async fn run_control_reader( // does not know that encoder, and there is nothing to pause. match pid_or_all(&subscribe.2) { Some(pid) => { - info!("client pauses encoder {pid}"); + info!(client = %subscribe.0.fmt_short(), "pause: client pauses encoder {pid}"); pause.pause(pid); } - None => debug!("pause from a client with no encoder yet; ignored"), + None => info!( + client = %subscribe.0.fmt_short(), + "pause: no encoder known for this client yet; ignored" + ), } } MSG_WINDOW_RESUME => { @@ -874,6 +877,11 @@ async fn run_control_reader( // paused nothing starts nothing; the keyframe is asked for // either way, since the client restarted its picture. awaiting_keyframe.store(true, Ordering::Relaxed); + info!( + client = %subscribe.0.fmt_short(), + "resume: encoder {:?}", + pid_or_all(&subscribe.2) + ); pause.resume(); if let Some(pid) = pid_or_all(&subscribe.2) { let _ = idr_cmd_tx.send(keyframe_request(Some(pid))); @@ -1290,6 +1298,15 @@ impl SessionManager { self.sessions.lock().await.get(id).map(|s| s.pause.clone()) } + #[cfg(test)] + pub(crate) async fn encoder_pid_of(&self, id: &iroh::EndpointId) -> Option { + self.sessions + .lock() + .await + .get(id) + .map(|s| s.encoder_pid.load(Ordering::Relaxed)) + } + #[cfg(test)] pub(crate) fn encoder_paused(&self, pid: u32) -> bool { self.pauses.lock().unwrap().paused(pid) @@ -1493,7 +1510,20 @@ impl SessionManager { } } for session in sessions.values() { - match route(session.subscription(), window, note.primary) { + let routed = route(session.subscription(), window, note.primary); + // A client shown this frame is shown this encoder, whenever it + // subscribed. The pid used to be fixed at subscribe time, and a + // client that subscribed before the game drew its first frame (the + // usual order: the window opens, the client subscribes, the + // encoder draws) kept pid 0 for good, so a pause had no encoder + // to go to and was dropped, ref(d-0087). + if routed.is_some() + && let Some(pid) = source + && session.encoder_pid.load(Ordering::Relaxed) != pid + { + session.encoder_pid.store(pid, Ordering::Relaxed); + } + match routed { Some(WINDOW_DEFAULT) => session.send_video_frame(MediaItem::Coded(data.clone())), Some(w) => session.send_video_frame(MediaItem::Windowed(w, data.clone())), None => {}