From 849eb70bb992a2281079c56bebd07800d4621c8c Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Sun, 18 Oct 2026 23:33:01 +0300 Subject: [PATCH 1/6] feat(neshub): per-window pause and resume of capture and encode MSG_WINDOW_PAUSE and MSG_WINDOW_RESUME on the control stream. The hub sends them to the encoder of the window the client is shown; nescapture stops capturing in the present hook while paused, and resume requests a keyframe. --- apps/nescapture/src/encode.rs | 58 +++++++++++++++++++++++++++++++++- apps/nescapture/src/present.rs | 11 +++++++ apps/neshub/src/capture_cmd.rs | 30 ++++++++++++++++++ apps/neshub/src/session.rs | 26 +++++++++++++-- crates/nesprotocol/src/lib.rs | 12 +++++++ 5 files changed, 134 insertions(+), 3 deletions(-) diff --git a/apps/nescapture/src/encode.rs b/apps/nescapture/src/encode.rs index 3b41e512..0c3913ab 100644 --- a/apps/nescapture/src/encode.rs +++ b/apps/nescapture/src/encode.rs @@ -55,7 +55,8 @@ use std::time::Instant; use nesprotocol::{ CODEC_AV1, CODEC_H264, CODEC_H265, CODEC_KEEP, CODEC_PYROWAVE, ClientCaps, FLAG_KEYFRAME, - FLAG_RECONFIG, MSG_CLIENT_CAPS, MSG_ENCODE_SETTINGS, MSG_IDR_REQUEST, STREAM_VIDEO, + FLAG_RECONFIG, MSG_CLIENT_CAPS, MSG_ENCODE_SETTINGS, MSG_IDR_REQUEST, MSG_WINDOW_PAUSE, + MSG_WINDOW_RESUME, STREAM_VIDEO, decode_client_caps, decode_encode_settings_ext, encode_ipc_frame_from, }; use pixelforge::{ @@ -998,6 +999,10 @@ pub struct PipelineHandle { stall_probe: StallProbe, progress_epoch: Instant, idr_requested: Arc, + /// Set by a pause from the hub (ref d-0087): the present hook captures + /// nothing while it is, so nothing is blitted, encoded or sent. The game + /// is not touched. + paused: Arc, shutdown: Arc, pub capture_fps: Arc, pub encode_avg_ms: Arc, @@ -1089,6 +1094,8 @@ impl PipelineHandle { shutdown.clone(), ); let idr_requested = Arc::new(AtomicBool::new(false)); + let paused = Arc::new(AtomicBool::new(false)); + let paused_listener = paused.clone(); let capture_fps = Arc::new(AtomicU32::new(0)); let encode_avg_ms = Arc::new(AtomicU32::new(0)); let capture_ms = Arc::new(AtomicU32::new(0)); @@ -1229,6 +1236,9 @@ impl PipelineHandle { let mut buf = [0u8; 128]; loop { match sock.recv(&mut buf) { + Ok(1) if buf[0] == MSG_WINDOW_PAUSE || buf[0] == MSG_WINDOW_RESUME => { + apply_pause_message(buf[0], &paused_listener, &idr_thread); + } Ok(1) if buf[0] == MSG_IDR_REQUEST => { log::info!("IDR requested by client"); idr_thread.store(true, Ordering::Relaxed); @@ -1394,6 +1404,7 @@ impl PipelineHandle { stall_probe, progress_epoch, idr_requested, + paused, shutdown, capture_fps, encode_avg_ms, @@ -1468,6 +1479,17 @@ impl PipelineHandle { true } + /// Whether the window is paused: the hook captures nothing. + pub fn is_paused(&self) -> bool { + self.paused.load(Ordering::Relaxed) + } + + /// Say the capture thread is alive while paused, so the stall watchdog + /// does not report a stopped capture that was stopped on purpose. + pub fn note_paused(&self) { + self.capture_progress.note(self.progress_epoch, 7); + } + pub fn request_idr(&self) { self.idr_requested.store(true, Ordering::Relaxed); } @@ -1524,6 +1546,24 @@ pub struct EncodeSettingsChange { pub pyro: Option, } +/// Apply a pause or resume from the hub. Resume clears the flag and asks for a +/// 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) { + if msg == MSG_WINDOW_PAUSE { + if !paused.swap(true, Ordering::Relaxed) { + log::info!("paused by the client: capture and encode stopped"); + } + } else if msg == MSG_WINDOW_RESUME { + // The keyframe first: the flag lets the next present through, and the + // encoder must already know that frame is to be an IDR. + idr_requested.store(true, Ordering::Relaxed); + if paused.swap(false, Ordering::Relaxed) { + log::info!("resumed by the client: keyframe requested"); + } + } +} + fn encoder_thread( mut cfg: EncoderConfig, frame_rx: mpsc::Receiver, @@ -3848,6 +3888,22 @@ mod encoder_identity_tests { assert!(encoder_still_serves(HD, HD)); } + #[test] + fn pause_stops_and_resume_starts_with_a_keyframe() { + let paused = AtomicBool::new(false); + let idr = AtomicBool::new(false); + apply_pause_message(MSG_WINDOW_PAUSE, &paused, &idr); + assert!(paused.load(Ordering::Relaxed)); + assert!(!idr.load(Ordering::Relaxed), "a pause asks for nothing"); + apply_pause_message(MSG_WINDOW_RESUME, &paused, &idr); + assert!(!paused.load(Ordering::Relaxed)); + assert!(idr.load(Ordering::Relaxed)); + // A resume that was never paused is still a keyframe request, harmless. + idr.store(false, Ordering::Relaxed); + apply_pause_message(MSG_WINDOW_RESUME, &paused, &idr); + assert!(idr.load(Ordering::Relaxed)); + } + #[test] fn a_resolution_change_rebuilds_in_either_direction() { // The regression. Shrinking left the encoder sending the old geometry diff --git a/apps/nescapture/src/present.rs b/apps/nescapture/src/present.rs index c5650223..34b0ba5e 100644 --- a/apps/nescapture/src/present.rs +++ b/apps/nescapture/src/present.rs @@ -213,6 +213,17 @@ unsafe fn try_capture( (images[image_index], fmt, ext, images.len()) }; + // A paused window captures nothing (ref d-0087): no blit, no slot, no + // encode. Before the gate, so the gate's own bookkeeping does not run + // either, and the frame the client resumes on is a fresh one. + if let Ok(enc) = ds.encoder.lock() + && let Some(ref h) = *enc + && h.is_paused() + { + h.note_paused(); + return None; + } + // Gate before any GPU work is queued. A game presenting faster than the // target would otherwise pay a full blit for frames the // encoder throws away moments later. diff --git a/apps/neshub/src/capture_cmd.rs b/apps/neshub/src/capture_cmd.rs index 32cf9d55..66492112 100644 --- a/apps/neshub/src/capture_cmd.rs +++ b/apps/neshub/src/capture_cmd.rs @@ -44,6 +44,17 @@ pub fn keyframe_request(pid: Option) -> Vec { to_encoder(pid, &[MSG_IDR_REQUEST]) } +/// Stop `pid`'s encoder capturing and encoding; it keeps its game running. +/// Never replayed to a later encoder: it is about one window, not a setting. +pub fn window_pause(pid: Option) -> Vec { + to_encoder(pid, &[nesprotocol::MSG_WINDOW_PAUSE]) +} + +/// Start `pid`'s encoder again; it answers with a keyframe. +pub fn window_resume(pid: Option) -> Vec { + to_encoder(pid, &[nesprotocol::MSG_WINDOW_RESUME]) +} + /// Split a queued command into its target and the command itself. pub fn target_of(cmd: Vec) -> (Option, Vec) { if cmd.len() >= 5 && cmd[0] == TO_ENCODER { @@ -221,6 +232,25 @@ mod tests { assert_eq!(f.encoder(11), Some(vec![vec![MSG_IDR_REQUEST]])); } + #[test] + fn pause_and_resume_go_to_one_encoder_and_are_not_replayed() { + let mut f = Fanout::default(); + f.encoder(10); + f.encoder(20); + let (target, cmd) = target_of(window_pause(Some(20))); + assert_eq!(cmd, vec![nesprotocol::MSG_WINDOW_PAUSE]); + assert_eq!( + f.destinations_for(target), + vec![PathBuf::from("/tmp/nescapture-cmd-20.sock")] + ); + let (target, cmd) = target_of(window_resume(Some(20))); + assert_eq!(cmd, vec![nesprotocol::MSG_WINDOW_RESUME]); + assert_eq!(target, Some(20)); + // Targeted commands are not recorded, so a later encoder is not + // born paused. + assert_eq!(f.encoder(30).map(|r| r.len()), Some(1)); + } + #[test] fn pid_from_path() { assert_eq!( diff --git a/apps/neshub/src/session.rs b/apps/neshub/src/session.rs index c615548d..692b95cc 100644 --- a/apps/neshub/src/session.rs +++ b/apps/neshub/src/session.rs @@ -14,13 +14,13 @@ use nesprotocol::{ControlMode, ReceiverReport, decode_control_mode, decode_recei use nesprotocol::{FRAME_HDR_LEN, STREAM_VERSION, encode_frame}; use nesprotocol::{ MSG_CLIENT_CAPS, MSG_CONTROL_MODE, MSG_ENCODE_SETTINGS, MSG_GAMEPAD, MSG_GAMEPAD_FEEDBACK, - MSG_IDR_REQUEST, MSG_INPUT_BATCH, MSG_RECEIVER_REPORT, + MSG_IDR_REQUEST, MSG_INPUT_BATCH, MSG_RECEIVER_REPORT, MSG_WINDOW_PAUSE, MSG_WINDOW_RESUME, }; use crate::control::{Controller, PathView}; use crate::negotiation::{Negotiator, effective_caps}; -use crate::capture_cmd::keyframe_request; +use crate::capture_cmd::{keyframe_request, window_pause, window_resume}; use crate::dgram::{MediaItem, run_datagram_writer}; use crate::windows::{Source, Subscription, WindowTable, route, wants_cursor}; use nesprotocol::window::{MSG_WINDOW_SUBSCRIBE, STREAM_WINDOWS, WINDOW_DEFAULT, WindowEvent, WindowId}; @@ -843,6 +843,28 @@ async fn run_control_reader( // ref(d-0085): the other windows are fine. let _ = idr_cmd_tx.send(keyframe_request(pid_or_all(&subscribe.2))); } + MSG_WINDOW_PAUSE => { + // To the encoder of the window this client is shown, and + // to nobody else. Until the client has been sent a frame + // the hub does not know that encoder, and "every encoder" + // is the wrong answer for a pause: it would stop windows + // nobody asked about. + match pid_or_all(&subscribe.2) { + Some(pid) => { + info!("client pauses encoder {pid}"); + let _ = idr_cmd_tx.send(window_pause(Some(pid))); + } + None => debug!("pause from a client with no encoder yet; ignored"), + } + } + MSG_WINDOW_RESUME => { + // The same gate as a keyframe request: what arrives before + // the keyframe cannot be decoded and starves it. + awaiting_keyframe.store(true, Ordering::Relaxed); + let pid = pid_or_all(&subscribe.2); + info!("client resumes encoder {pid:?}"); + let _ = idr_cmd_tx.send(window_resume(pid)); + } MSG_CLIENT_CAPS => { // Passed through untouched, and in particular *without* // taking the controller off automatic the way encode diff --git a/crates/nesprotocol/src/lib.rs b/crates/nesprotocol/src/lib.rs index 15180c9a..4650a570 100644 --- a/crates/nesprotocol/src/lib.rs +++ b/crates/nesprotocol/src/lib.rs @@ -103,6 +103,15 @@ pub const MSG_IDR_REQUEST: u8 = 0x10; // request a keyframe (desktop → hub → pub const MSG_ENCODE_SETTINGS: u8 = 0x12; // change encoder settings (desktop → hub → nescapture) pub const MSG_CLIENT_CAPS: u8 = 0x15; // what the client can decode (desktop → hub → nescapture) pub const MSG_SURFACE_COLOR: u8 = 0x16; // what the compositor was told a surface is (nescope → nescapture) +/// Stop capturing and encoding the window this client is shown (desktop → hub +/// → that window's nescapture), ref(d-0087). The game keeps running; nothing is +/// sent for the window until [`MSG_WINDOW_RESUME`]. No payload: the hub already +/// knows which window the client is subscribed to, as it does for +/// [`MSG_IDR_REQUEST`]. An old hub logs it as an unknown control message. +pub const MSG_WINDOW_PAUSE: u8 = 0x18; +/// Start again, with a keyframe within a second (desktop → hub → nescapture). +/// Resume is also what forces the keyframe, so the client need not ask for one. +pub const MSG_WINDOW_RESUME: u8 = 0x19; /// What the receiver actually got, once a second (desktop → hub). /// /// The hub cannot see this. Its own view of the path -- RTT, congestion window, @@ -1009,6 +1018,9 @@ mod media_control_tests { ("MSG_ENCODE_SETTINGS", MSG_ENCODE_SETTINGS), ("MSG_CLIENT_CAPS", MSG_CLIENT_CAPS), ("MSG_SURFACE_COLOR", MSG_SURFACE_COLOR), + ("MSG_WINDOW_PAUSE", MSG_WINDOW_PAUSE), + ("MSG_WINDOW_RESUME", MSG_WINDOW_RESUME), + ("MSG_WINDOW_SUBSCRIBE", window::MSG_WINDOW_SUBSCRIBE), ("MSG_RECEIVER_REPORT", MSG_RECEIVER_REPORT), ("MSG_CONTROL_MODE", MSG_CONTROL_MODE), ("MSG_INPUT_BATCH", MSG_INPUT_BATCH), From a26a13bce45ba86c960986677c44a3d3bcf20e30 Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Sun, 18 Oct 2026 23:36:14 +0300 Subject: [PATCH 2/6] test(nescapture): move the pause test into its own module --- apps/nescapture/src/encode.rs | 37 ++++++++++++++++++++--------------- 1 file changed, 21 insertions(+), 16 deletions(-) diff --git a/apps/nescapture/src/encode.rs b/apps/nescapture/src/encode.rs index 0c3913ab..ec967fcd 100644 --- a/apps/nescapture/src/encode.rs +++ b/apps/nescapture/src/encode.rs @@ -1564,6 +1564,27 @@ fn apply_pause_message(msg: u8, paused: &AtomicBool, idr_requested: &AtomicBool) } } +#[cfg(test)] +mod pause_tests { + use super::*; + + #[test] + fn pause_stops_and_resume_starts_with_a_keyframe() { + let paused = AtomicBool::new(false); + let idr = AtomicBool::new(false); + apply_pause_message(MSG_WINDOW_PAUSE, &paused, &idr); + assert!(paused.load(Ordering::Relaxed)); + assert!(!idr.load(Ordering::Relaxed), "a pause asks for nothing"); + apply_pause_message(MSG_WINDOW_RESUME, &paused, &idr); + assert!(!paused.load(Ordering::Relaxed)); + assert!(idr.load(Ordering::Relaxed)); + // A resume that was never paused is still a keyframe request, harmless. + idr.store(false, Ordering::Relaxed); + apply_pause_message(MSG_WINDOW_RESUME, &paused, &idr); + assert!(idr.load(Ordering::Relaxed)); + } +} + fn encoder_thread( mut cfg: EncoderConfig, frame_rx: mpsc::Receiver, @@ -3888,22 +3909,6 @@ mod encoder_identity_tests { assert!(encoder_still_serves(HD, HD)); } - #[test] - fn pause_stops_and_resume_starts_with_a_keyframe() { - let paused = AtomicBool::new(false); - let idr = AtomicBool::new(false); - apply_pause_message(MSG_WINDOW_PAUSE, &paused, &idr); - assert!(paused.load(Ordering::Relaxed)); - assert!(!idr.load(Ordering::Relaxed), "a pause asks for nothing"); - apply_pause_message(MSG_WINDOW_RESUME, &paused, &idr); - assert!(!paused.load(Ordering::Relaxed)); - assert!(idr.load(Ordering::Relaxed)); - // A resume that was never paused is still a keyframe request, harmless. - idr.store(false, Ordering::Relaxed); - apply_pause_message(MSG_WINDOW_RESUME, &paused, &idr); - assert!(idr.load(Ordering::Relaxed)); - } - #[test] fn a_resolution_change_rebuilds_in_either_direction() { // The regression. Shrinking left the encoder sending the old geometry From 78a8bc879e5c18e1bd8b0833d6939df9d969c128 Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Sun, 18 Oct 2026 23:47:12 +0300 Subject: [PATCH 3/6] fix(neshub): a pause belongs to its session, not to the encoder The hub keeps the set of sessions pausing each encoder. An encoder is paused while the set is not empty; a resume goes only to the encoder that session paused; a session that drops, resubscribes or sees the primary picture move releases its pause. --- apps/neshub/src/e2e_tests.rs | 35 +++++++ apps/neshub/src/main.rs | 1 + apps/neshub/src/pauses.rs | 193 +++++++++++++++++++++++++++++++++++ apps/neshub/src/session.rs | 58 ++++++++--- 4 files changed, 274 insertions(+), 13 deletions(-) create mode 100644 apps/neshub/src/pauses.rs diff --git a/apps/neshub/src/e2e_tests.rs b/apps/neshub/src/e2e_tests.rs index 3feb6d43..8220cae7 100644 --- a/apps/neshub/src/e2e_tests.rs +++ b/apps/neshub/src/e2e_tests.rs @@ -577,4 +577,39 @@ mod windows { assert_eq!(got.len(), 3); assert!(got.iter().all(|g| (g.0, g.1) == (0, 0))); } + + #[tokio::test(flavor = "multi_thread")] + async fn a_session_that_goes_while_paused_releases_its_pause() { + 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; + draw(&rig, GAME_A, 0, true).await; + let _ = received(&rig.client).await; + let me = rig._client_ep.addr().id; + let handle = rig.mgr.pause_handle_of(&me).await.expect("a session"); + handle.pause(GAME_A); + assert!(rig.mgr.encoder_paused(GAME_A)); + let controller = tokio::sync::Mutex::new(Controller::new(Limits::new(4_000))); + let (cmd, _rx) = tokio::sync::mpsc::unbounded_channel(); + rig.mgr.remove_session(&me, &controller, &cmd).await; + assert!(!rig.mgr.encoder_paused(GAME_A), "the pause went with the session"); + } + + #[tokio::test(flavor = "multi_thread")] + async fn a_resubscribe_while_paused_releases_the_old_encoder() { + let rig = rig().await; + tell(&rig, |m| encode_event_message(m, &opened(0, "launch_a"))).await; + tell(&rig, |m| encode_event_message(m, &opened(1, "launch_b"))).await; + tell(&rig, |m| encode_window_pid(m, 0, GAME_A)).await; + tell(&rig, |m| encode_window_pid(m, 1, GAME_B)).await; + draw(&rig, GAME_A, 0, true).await; + let _ = received(&rig.client).await; + let me = rig._client_ep.addr().id; + let handle = rig.mgr.pause_handle_of(&me).await.expect("a session"); + handle.pause(GAME_A); + assert!(rig.mgr.encoder_paused(GAME_A)); + let (idr, _rx) = tokio::sync::mpsc::unbounded_channel(); + rig.mgr.subscribe(me, 1, &idr).await; + assert!(!rig.mgr.encoder_paused(GAME_A), "no longer shown, no longer held"); + } } diff --git a/apps/neshub/src/main.rs b/apps/neshub/src/main.rs index 17baae06..99eee293 100644 --- a/apps/neshub/src/main.rs +++ b/apps/neshub/src/main.rs @@ -13,6 +13,7 @@ mod session; mod ticket; mod windows; mod capture_cmd; +mod pauses; use std::net::{Ipv4Addr, SocketAddrV4}; use std::path::PathBuf; diff --git a/apps/neshub/src/pauses.rs b/apps/neshub/src/pauses.rs new file mode 100644 index 00000000..cd802ce0 --- /dev/null +++ b/apps/neshub/src/pauses.rs @@ -0,0 +1,193 @@ +//! Who has paused which encoder, ref(d-0087). +//! +//! A pause is a client's, not an encoder's: two clients can be shown the same +//! window, and one pausing it must not freeze the other's picture. So the hub +//! keeps, for each encoder, the set of sessions pausing it, and the encoder is +//! stopped only while that set is not empty. A session leaving (its +//! connection ending, a resubscribe, the primary picture moving) releases its +//! pause, so a pause can never outlive the client that asked for it. +//! +//! Keyed by the session's gamepad number, which is never reused within a hub's +//! lifetime, so a client that reconnects is a new session and cannot release +//! the pause of the one it replaced, or the other way round. + +use std::collections::{BTreeMap, BTreeSet}; +use std::sync::{Arc, Mutex}; + +use tokio::sync::mpsc::UnboundedSender; + +use crate::capture_cmd::{window_pause, window_resume}; + +#[derive(Debug, Default)] +pub struct PauseBook { + /// The encoder each pausing session paused. One at a time per session. + paused_by: BTreeMap, +} + +impl PauseBook { + fn users(&self, pid: u32) -> BTreeSet { + self.paused_by + .iter() + .filter(|(_, p)| **p == pid) + .map(|(s, _)| *s) + .collect() + } + + /// `session` pauses `pid`. The commands to send, in order. + fn pause(&mut self, session: u32, pid: u32) -> Vec> { + let mut out = Vec::new(); + if self.paused_by.get(&session) == Some(&pid) { + return out; + } + // Moving from one encoder to another releases the first. + out.extend(self.release(session)); + let first = self.users(pid).is_empty(); + self.paused_by.insert(session, pid); + if first { + out.push(window_pause(Some(pid))); + } + out + } + + /// `session` stops pausing. Resumes the encoder it paused, and only that + /// one, and only when no other session still pauses it. + fn release(&mut self, session: u32) -> Vec> { + let Some(pid) = self.paused_by.remove(&session) else { + return Vec::new(); + }; + if self.users(pid).is_empty() { + vec![window_resume(Some(pid))] + } else { + Vec::new() + } + } + + pub fn paused(&self, pid: u32) -> bool { + !self.users(pid).is_empty() + } +} + +/// One session's way of pausing and resuming, shared with its control reader. +#[derive(Clone)] +pub struct PauseHandle { + book: Arc>, + session: u32, + cmd_tx: UnboundedSender>, +} + +impl PauseHandle { + pub fn new(book: Arc>, session: u32, cmd_tx: UnboundedSender>) -> Self { + Self { book, session, cmd_tx } + } + + fn send(&self, cmds: Vec>) { + for c in cmds { + let _ = self.cmd_tx.send(c); + } + } + + /// Pause `pid` for this session. `pid` is the encoder it is shown, which + /// is never 0 here: the caller does not pause what it cannot name. + pub fn pause(&self, pid: u32) { + let cmds = self.book.lock().unwrap_or_else(|e| e.into_inner()).pause(self.session, pid); + self.send(cmds); + } + + /// Resume what this session paused. Never anything else, and nothing at + /// all when it paused nothing. + pub fn resume(&self) { + self.release(); + } + + /// Let go of whatever this session paused: it is leaving, or looking at + /// another picture. + pub fn release(&self) { + let cmds = self.book.lock().unwrap_or_else(|e| e.into_inner()).release(self.session); + self.send(cmds); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn rig() -> (Arc>, UnboundedSender>, tokio::sync::mpsc::UnboundedReceiver>) { + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + (Arc::new(Mutex::new(PauseBook::default())), tx, rx) + } + + fn sent(rx: &mut tokio::sync::mpsc::UnboundedReceiver>) -> Vec> { + std::iter::from_fn(|| rx.try_recv().ok()).collect() + } + + #[test] + fn one_of_two_sessions_pausing_a_shared_encoder_pauses_it_until_both_let_go() { + let (book, tx, mut rx) = rig(); + let a = PauseHandle::new(book.clone(), 1, tx.clone()); + let b = PauseHandle::new(book.clone(), 2, tx); + a.pause(10); + assert_eq!(sent(&mut rx), vec![window_pause(Some(10))]); + // B pauses too: the encoder is already stopped, nothing to send. + b.pause(10); + assert!(sent(&mut rx).is_empty()); + // A resumes, B still pauses: the encoder stays stopped. + a.resume(); + assert!(sent(&mut rx).is_empty()); + assert!(book.lock().unwrap().paused(10)); + b.resume(); + assert_eq!(sent(&mut rx), vec![window_resume(Some(10))]); + assert!(!book.lock().unwrap().paused(10)); + } + + #[test] + fn a_session_that_never_paused_resumes_nothing() { + let (book, tx, mut rx) = rig(); + let a = PauseHandle::new(book.clone(), 1, tx.clone()); + let b = PauseHandle::new(book, 2, tx); + a.pause(10); + sent(&mut rx); + // B, shown the same window, did not pause it: its resume must not + // start the encoder A stopped. + b.resume(); + b.release(); + assert!(sent(&mut rx).is_empty()); + } + + #[test] + fn a_session_that_goes_while_paused_resumes_the_encoder() { + let (book, tx, mut rx) = rig(); + let a = PauseHandle::new(book.clone(), 1, tx); + a.pause(10); + sent(&mut rx); + a.release(); // what its drop does + assert_eq!(sent(&mut rx), vec![window_resume(Some(10))]); + // Releasing twice is not resuming twice. + a.release(); + assert!(sent(&mut rx).is_empty()); + } + + #[test] + fn a_resubscribe_while_paused_releases_the_old_encoder_only() { + let (book, tx, mut rx) = rig(); + let a = PauseHandle::new(book.clone(), 1, tx.clone()); + let other = PauseHandle::new(book.clone(), 2, tx); + a.pause(10); + other.pause(20); + sent(&mut rx); + a.release(); // restart_picture on moving to another window + assert_eq!(sent(&mut rx), vec![window_resume(Some(10))]); + assert!(book.lock().unwrap().paused(20), "someone else's pause is untouched"); + } + + #[test] + fn pausing_another_encoder_moves_the_pause() { + let (book, tx, mut rx) = rig(); + let a = PauseHandle::new(book, 1, tx); + a.pause(10); + a.pause(20); + assert_eq!( + sent(&mut rx), + vec![window_pause(Some(10)), window_resume(Some(10)), window_pause(Some(20))] + ); + } +} diff --git a/apps/neshub/src/session.rs b/apps/neshub/src/session.rs index 692b95cc..5b4f3826 100644 --- a/apps/neshub/src/session.rs +++ b/apps/neshub/src/session.rs @@ -20,7 +20,7 @@ use nesprotocol::{ use crate::control::{Controller, PathView}; use crate::negotiation::{Negotiator, effective_caps}; -use crate::capture_cmd::{keyframe_request, window_pause, window_resume}; +use crate::capture_cmd::keyframe_request; use crate::dgram::{MediaItem, run_datagram_writer}; use crate::windows::{Source, Subscription, WindowTable, route, wants_cursor}; use nesprotocol::window::{MSG_WINDOW_SUBSCRIBE, STREAM_WINDOWS, WINDOW_DEFAULT, WindowEvent, WindowId}; @@ -106,6 +106,9 @@ pub struct ClientSession { /// The encoder whose frames this client was last sent (0: none yet), so /// its keyframe requests go to that encoder alone. encoder_pid: Arc, + /// This session's pause of an encoder, if any (d-0087); released when it + /// goes or looks at another picture. + pause: crate::pauses::PauseHandle, primary_window: Arc, /// Where a subscribe request goes: the manager, which holds the windows. subscribe_tx: tokio::sync::mpsc::UnboundedSender<(iroh::EndpointId, WindowId)>, @@ -150,6 +153,8 @@ impl Drop for ClientSession { for task in &self.tasks { task.abort(); } + // A pause is this client's, and does not outlive it. + self.pause.release(); } } @@ -169,6 +174,7 @@ impl ClientSession { pyrowave_kbps: u32, client: iroh::EndpointId, negotiator: Arc, + pauses: Arc>, ) -> Self { let (gamepad_feedback_tx, gamepad_feedback_rx) = tokio::sync::mpsc::unbounded_channel(); let (video_tx, video_rx) = tokio::sync::mpsc::unbounded_channel::(); @@ -176,7 +182,9 @@ impl ClientSession { let (cursor_tx, cursor_rx) = tokio::sync::mpsc::unbounded_channel::>(); let (stats_tx, stats_rx) = tokio::sync::mpsc::unbounded_channel::>(); let (windows_tx, windows_rx) = tokio::sync::mpsc::unbounded_channel::>(); + let pause = crate::pauses::PauseHandle::new(pauses, gamepad_session, idr_cmd_tx.clone()); Self { + pause, subscription: Arc::new(std::sync::Mutex::new(Subscription::Legacy)), encoder_pid: Arc::new(AtomicU32::new(encoder_pid)), primary_window, @@ -349,6 +357,7 @@ impl ClientSession { client: self.client, negotiator: self.negotiator.clone(), }; + let pause = self.pause.clone(); let subscribe = ( self.client, self.subscribe_tx.clone(), @@ -356,7 +365,7 @@ impl ClientSession { ); self.tasks.push(tokio::spawn(async move { run_control_reader( - conn, idr, reports, controller, awaiting, pyrowave, caps, subscribe, + conn, idr, reports, controller, awaiting, pyrowave, caps, subscribe, pause, ) .await })); @@ -513,6 +522,9 @@ impl ClientSession { /// zero, hold deltas until a keyframe, and forget what it reported about /// the old one. The caller asks for the keyframe. pub fn restart_picture(&self) { + // Another picture: whatever this client paused is no longer the one + // it is shown, so it stops holding it. + self.pause.release(); self.awaiting_keyframe.store(true, Ordering::Relaxed); if self.pending_video.is_none() { let _ = self.send_video.send(MediaItem::Resubscribe); @@ -812,6 +824,7 @@ async fn run_control_reader( tokio::sync::mpsc::UnboundedSender<(iroh::EndpointId, WindowId)>, Arc, ), + pause: crate::pauses::PauseHandle, ) { run_framed_reader(conn, BIDI_CONTROL, "control", None, |msg_type, payload| { let idr_cmd_tx = idr_cmd_tx.clone(); @@ -821,6 +834,7 @@ async fn run_control_reader( let pyrowave = pyrowave.clone(); let caps_sink = caps_sink.clone(); let subscribe = subscribe.clone(); + let pause = pause.clone(); async move { match msg_type { MSG_WINDOW_SUBSCRIBE => match nesprotocol::window::decode_subscribe(&payload) { @@ -844,26 +858,26 @@ async fn run_control_reader( let _ = idr_cmd_tx.send(keyframe_request(pid_or_all(&subscribe.2))); } MSG_WINDOW_PAUSE => { - // To the encoder of the window this client is shown, and - // to nobody else. Until the client has been sent a frame - // the hub does not know that encoder, and "every encoder" - // is the wrong answer for a pause: it would stop windows - // nobody asked about. + // This session's pause of the encoder it is shown, and of + // nothing else. Until it has been sent a frame the hub + // does not know that encoder, and there is nothing to pause. match pid_or_all(&subscribe.2) { Some(pid) => { info!("client pauses encoder {pid}"); - let _ = idr_cmd_tx.send(window_pause(Some(pid))); + pause.pause(pid); } None => debug!("pause from a client with no encoder yet; ignored"), } } MSG_WINDOW_RESUME => { - // The same gate as a keyframe request: what arrives before - // the keyframe cannot be decoded and starves it. + // Only what this session paused. A resume from one that + // paused nothing starts nothing; the keyframe is asked for + // either way, since the client restarted its picture. awaiting_keyframe.store(true, Ordering::Relaxed); - let pid = pid_or_all(&subscribe.2); - info!("client resumes encoder {pid:?}"); - let _ = idr_cmd_tx.send(window_resume(pid)); + pause.resume(); + if let Some(pid) = pid_or_all(&subscribe.2) { + let _ = idr_cmd_tx.send(keyframe_request(Some(pid))); + } } MSG_CLIENT_CAPS => { // Passed through untouched, and in particular *without* @@ -1230,6 +1244,8 @@ pub struct SessionManager { std::sync::Mutex>>, /// The encoder of the primary picture (0: untagged or none yet). primary_pid: Arc, + /// Which sessions have paused which encoders, ref(d-0087). + pauses: Arc>, /// Where commands for the encoders go; set by `serve_subscriptions`. cmd_tx: std::sync::OnceLock>>, } @@ -1261,9 +1277,24 @@ impl SessionManager { relay_ms: Arc::new(AtomicU32::new(0)), pyrowave_kbps, negotiator: Arc::new(Negotiator::new()), + pauses: Default::default(), } } + /// A session's pause handle, for tests. + #[cfg(test)] + pub(crate) async fn pause_handle_of( + &self, + id: &iroh::EndpointId, + ) -> Option { + self.sessions.lock().await.get(id).map(|s| s.pause.clone()) + } + + #[cfg(test)] + pub(crate) fn encoder_paused(&self, pid: u32) -> bool { + self.pauses.lock().unwrap().paused(pid) + } + /// Take on one of a client's connections, creating the session if this is /// the first to arrive. /// @@ -1295,6 +1326,7 @@ impl SessionManager { self.pyrowave_kbps, id, self.negotiator.clone(), + self.pauses.clone(), ) }); session.attach(carrier, conn); From cf9c4136f6b65c9a0701d457afea7f30a859f932 Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Sun, 18 Oct 2026 23:51:27 +0300 Subject: [PATCH 4/6] fix(neshub): a pause made after its session closed is a no-op Dropping a session aborts its tasks, but a control reader mid-poll on another thread could still pause after the release and leave the encoder paused forever. The closed flag is set under the book's lock; pause checks it under the same lock. --- apps/neshub/src/pauses.rs | 46 ++++++++++++++++++++++++++++++++++++++ apps/neshub/src/session.rs | 2 +- 2 files changed, 47 insertions(+), 1 deletion(-) diff --git a/apps/neshub/src/pauses.rs b/apps/neshub/src/pauses.rs index cd802ce0..f3e1e67d 100644 --- a/apps/neshub/src/pauses.rs +++ b/apps/neshub/src/pauses.rs @@ -22,6 +22,11 @@ use crate::capture_cmd::{window_pause, window_resume}; pub struct PauseBook { /// The encoder each pausing session paused. One at a time per session. paused_by: BTreeMap, + /// Sessions that have gone. Set under the same lock that records pauses, + /// so a pause that a still-running reader makes after its session was + /// dropped finds it shut and does nothing, rather than holding an encoder + /// that nobody is left to resume. + closed: BTreeSet, } impl PauseBook { @@ -36,6 +41,9 @@ impl PauseBook { /// `session` pauses `pid`. The commands to send, in order. fn pause(&mut self, session: u32, pid: u32) -> Vec> { let mut out = Vec::new(); + if self.closed.contains(&session) { + return out; + } if self.paused_by.get(&session) == Some(&pid) { return out; } @@ -62,6 +70,12 @@ impl PauseBook { } } + /// The session is gone: release what it paused and refuse any more. + fn close(&mut self, session: u32) -> Vec> { + self.closed.insert(session); + self.release(session) + } + pub fn paused(&self, pid: u32) -> bool { !self.users(pid).is_empty() } @@ -99,6 +113,13 @@ impl PauseHandle { self.release(); } + /// The session is being dropped: release its pause and refuse any pause a + /// reader that is still running makes after this. + pub fn close(&self) { + let cmds = self.book.lock().unwrap_or_else(|e| e.into_inner()).close(self.session); + self.send(cmds); + } + /// Let go of whatever this session paused: it is leaving, or looking at /// another picture. pub fn release(&self) { @@ -179,6 +200,31 @@ mod tests { assert!(book.lock().unwrap().paused(20), "someone else's pause is untouched"); } + #[test] + fn a_pause_after_the_session_is_closed_is_a_no_op() { + let (book, tx, mut rx) = rig(); + let a = PauseHandle::new(book.clone(), 1, tx); + a.pause(10); + sent(&mut rx); + a.close(); + assert_eq!(sent(&mut rx), vec![window_resume(Some(10))]); + // The control reader, still running on another thread, pauses again. + a.pause(10); + assert!(sent(&mut rx).is_empty()); + assert!(!book.lock().unwrap().paused(10)); + } + + #[test] + fn releasing_without_closing_still_allows_a_later_pause() { + let (book, tx, mut rx) = rig(); + let a = PauseHandle::new(book, 1, tx); + a.pause(10); + a.release(); + sent(&mut rx); + a.pause(20); + assert_eq!(sent(&mut rx), vec![window_pause(Some(20))]); + } + #[test] fn pausing_another_encoder_moves_the_pause() { let (book, tx, mut rx) = rig(); diff --git a/apps/neshub/src/session.rs b/apps/neshub/src/session.rs index 5b4f3826..4bcfd0ab 100644 --- a/apps/neshub/src/session.rs +++ b/apps/neshub/src/session.rs @@ -154,7 +154,7 @@ impl Drop for ClientSession { task.abort(); } // A pause is this client's, and does not outlive it. - self.pause.release(); + self.pause.close(); } } From bec943faa6153ef14dde6d6eec38c8f55dc100f5 Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Mon, 19 Oct 2026 00:33:11 +0300 Subject: [PATCH 5/6] 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 => {} From a8f941603a43b8246e8d8eaae582f40011836e81 Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Mon, 19 Oct 2026 11:40:20 +0300 Subject: [PATCH 6/6] fix(api): billing portal answers 404 when a team has no Polar customer --- apps/api/test/billing-portal.test.ts | 81 ++++++++++++++++++++++++++++ packages/core/src/billing/polar.ts | 23 ++++++-- 2 files changed, 100 insertions(+), 4 deletions(-) create mode 100644 apps/api/test/billing-portal.test.ts diff --git a/apps/api/test/billing-portal.test.ts b/apps/api/test/billing-portal.test.ts new file mode 100644 index 00000000..69128a88 --- /dev/null +++ b/apps/api/test/billing-portal.test.ts @@ -0,0 +1,81 @@ +import { afterAll, afterEach, describe, expect, test } from 'bun:test'; + +import { AccessToken } from '@nestri/core/access-token/index'; +import { Polar } from '@nestri/core/billing/polar'; +import { Fixtures } from '@nestri/core/db/fixtures'; +import { testDb } from '@nestri/core/db/test'; +import { Env } from '@nestri/core/env'; +import { Identifier } from '@nestri/core/id'; + +import { app } from '../app/index'; +import './setup'; + +const sql = testDb(); +const createdUserIds: string[] = []; +const realFetch = globalThis.fetch; + +async function person(label: string) { + const owner = await Fixtures.owner(label); + createdUserIds.push(owner.userId); + const pat = await AccessToken.create({ + id: Identifier.ascending('accessToken'), + ownerUserId: owner.userId, + teamId: null, + name: label + }); + return { authorization: `Bearer ${pat.token}` }; +} + +function configure() { + Env.init({ + NODE_ENV: 'test', + FRONTEND_URL: 'http://localhost:5173', + POLAR_ACCESS_TOKEN: 'polar_test', + POLAR_PRODUCT_ID: 'prod_test' + }); + Polar.reset(); +} + +/** Stand in for Polar only; every other request goes where it was going. */ +function polarAnswers(status: number, body: unknown) { + globalThis.fetch = (async (input: Request | string | URL, init?: RequestInit) => { + const url = input instanceof Request ? input.url : String(input); + if (url.includes('polar.sh')) { + return new Response(JSON.stringify(body), { + status, + headers: { 'content-type': 'application/json' } + }); + } + return realFetch(input as never, init); + }) as typeof fetch; +} + +afterEach(() => { + globalThis.fetch = realFetch; + Polar.reset(); +}); + +afterAll(async () => { + if (createdUserIds.length > 0) { + await sql`delete from "user" where id in ${sql(createdUserIds)}`; + } +}); + +describe('GET /billing/portal', () => { + test('a team Polar has never heard of gets a 404, not a 500', async () => { + const headers = await person('portal-none'); + configure(); + polarAnswers(404, { error: 'ResourceNotFound', detail: 'Customer not found' }); + const res = await app.request('/billing/portal', { headers }); + expect(res.status).toBe(404); + expect((await res.json()).message).toBe('This team has no billing yet'); + }); + + test('an unknown external id answered with a 422 is the same 404', async () => { + const headers = await person('portal-422'); + configure(); + polarAnswers(422, { detail: [{ loc: ['body'], msg: 'Customer does not exist', type: 'x' }] }); + const res = await app.request('/billing/portal', { headers }); + expect(res.status).toBe(404); + }); +}); diff --git a/packages/core/src/billing/polar.ts b/packages/core/src/billing/polar.ts index 705c9300..0a65cd49 100644 --- a/packages/core/src/billing/polar.ts +++ b/packages/core/src/billing/polar.ts @@ -207,10 +207,25 @@ export namespace Polar { * them, which is the one thing this integration exists to avoid. */ export const portal = fn(z.object({ teamId: z.string() }), async (input) => { - const session = await client().customerSessions.create({ - externalCustomerId: input.teamId - }); - return { url: session.customerPortalUrl }; + try { + const session = await client().customerSessions.create({ + externalCustomerId: input.teamId + }); + return { url: session.customerPortalUrl }; + } catch (err) { + // A team that never checked out has no customer on their side, and + // they refuse the session (404, or 422 for an unknown external id). + // That is not our failure, so it must not surface as a 500. + const status = (err as { statusCode?: number } | null)?.statusCode; + if (status === 404 || status === 422) { + throw new VisibleError( + 'not_found', + ErrorCodes.NotFound.RESOURCE_NOT_FOUND, + 'This team has no billing yet' + ); + } + throw err; + } }); /** The plan and status a team is on, as far as the provider is concerned. */