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

@@ -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");

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;

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

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<u32> {
+        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 => {}