release: billing portal 404, per-window pause/resume (#451)

Promotes dev to prod: #450 billing portal 404 when a team has no Polar
customer; #448, #449 per-window pause and resume (neshub). Merge commit,
not squash.

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);
+	});
+});

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<AtomicBool>,
+    /// 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<AtomicBool>,
     shutdown: Arc<AtomicBool>,
     pub capture_fps: Arc<AtomicU32>,
     pub encode_avg_ms: Arc<AtomicU32>,
@@ -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,50 @@ pub struct EncodeSettingsChange {
     pub pyro: Option<PyroSettings>,
 }
 
+/// 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) {
+    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");
+        }
+    } 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");
+        }
+    }
+}
+
+#[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<CapturedFrame>,

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.

apps/neshub/src/capture_cmd.rs

@@ -44,6 +44,17 @@ pub fn keyframe_request(pid: Option<u32>) -> Vec<u8> {
     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<u32>) -> Vec<u8> {
+    to_encoder(pid, &[nesprotocol::MSG_WINDOW_PAUSE])
+}
+
+/// Start `pid`'s encoder again; it answers with a keyframe.
+pub fn window_resume(pid: Option<u32>) -> Vec<u8> {
+    to_encoder(pid, &[nesprotocol::MSG_WINDOW_RESUME])
+}
+
 /// Split a queued command into its target and the command itself.
 pub fn target_of(cmd: Vec<u8>) -> (Option<u32>, Vec<u8>) {
     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!(

apps/neshub/src/e2e_tests.rs

@@ -577,4 +577,61 @@ mod windows {
         assert_eq!(got.len(), 3);
         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;
+        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");
+    }
 }

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;
@@ -304,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/pauses.rs

@@ -0,0 +1,239 @@
+//! 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<u32, u32>,
+    /// 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<u32>,
+}
+
+impl PauseBook {
+    fn users(&self, pid: u32) -> BTreeSet<u32> {
+        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<Vec<u8>> {
+        let mut out = Vec::new();
+        if self.closed.contains(&session) {
+            return out;
+        }
+        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<Vec<u8>> {
+        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()
+        }
+    }
+
+    /// The session is gone: release what it paused and refuse any more.
+    fn close(&mut self, session: u32) -> Vec<Vec<u8>> {
+        self.closed.insert(session);
+        self.release(session)
+    }
+
+    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<Mutex<PauseBook>>,
+    session: u32,
+    cmd_tx: UnboundedSender<Vec<u8>>,
+}
+
+impl PauseHandle {
+    pub fn new(book: Arc<Mutex<PauseBook>>, session: u32, cmd_tx: UnboundedSender<Vec<u8>>) -> Self {
+        Self { book, session, cmd_tx }
+    }
+
+    fn send(&self, cmds: Vec<Vec<u8>>) {
+        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();
+    }
+
+    /// 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) {
+        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<Mutex<PauseBook>>, UnboundedSender<Vec<u8>>, tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>) {
+        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<u8>>) -> Vec<Vec<u8>> {
+        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 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();
+        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))]
+        );
+    }
+}

apps/neshub/src/session.rs

@@ -14,7 +14,7 @@ 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};
@@ -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<AtomicU32>,
+    /// 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<AtomicU32>,
     /// 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.close();
     }
 }
 
@@ -169,6 +174,7 @@ impl ClientSession {
         pyrowave_kbps: u32,
         client: iroh::EndpointId,
         negotiator: Arc<Negotiator>,
+        pauses: Arc<std::sync::Mutex<crate::pauses::PauseBook>>,
     ) -> Self {
         let (gamepad_feedback_tx, gamepad_feedback_rx) = tokio::sync::mpsc::unbounded_channel();
         let (video_tx, video_rx) = tokio::sync::mpsc::unbounded_channel::<MediaItem>();
@@ -176,7 +182,9 @@ impl ClientSession {
         let (cursor_tx, cursor_rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
         let (stats_tx, stats_rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
         let (windows_tx, windows_rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
+        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<AtomicU32>,
     ),
+    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) {
@@ -843,6 +857,36 @@ 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 => {
+                    // 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 = %subscribe.0.fmt_short(), "pause: client pauses encoder {pid}");
+                            pause.pause(pid);
+                        }
+                        None => info!(
+                            client = %subscribe.0.fmt_short(),
+                            "pause: no encoder known for this client yet; ignored"
+                        ),
+                    }
+                }
+                MSG_WINDOW_RESUME => {
+                    // 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);
+                    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)));
+                    }
+                }
                 MSG_CLIENT_CAPS => {
                     // Passed through untouched, and in particular *without*
                     // taking the controller off automatic the way encode
@@ -1208,6 +1252,8 @@ pub struct SessionManager {
         std::sync::Mutex<Option<tokio::sync::mpsc::UnboundedReceiver<(iroh::EndpointId, WindowId)>>>,
     /// The encoder of the primary picture (0: untagged or none yet).
     primary_pid: Arc<AtomicU32>,
+    /// Which sessions have paused which encoders, ref(d-0087).
+    pauses: Arc<std::sync::Mutex<crate::pauses::PauseBook>>,
     /// Where commands for the encoders go; set by `serve_subscriptions`.
     cmd_tx: std::sync::OnceLock<tokio::sync::mpsc::UnboundedSender<Vec<u8>>>,
 }
@@ -1239,9 +1285,33 @@ 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<crate::pauses::PauseHandle> {
+        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)
+    }
+
     /// Take on one of a client's connections, creating the session if this is
     /// the first to arrive.
     ///
@@ -1273,6 +1343,7 @@ impl SessionManager {
                 self.pyrowave_kbps,
                 id,
                 self.negotiator.clone(),
+                self.pauses.clone(),
             )
         });
         session.attach(carrier, conn);
@@ -1439,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 => {}

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),

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. */