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/apps/nescapture/src/encode.rs b/apps/nescapture/src/encode.rs index 3b41e512..332abd9c 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,50 @@ 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) { + 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, 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/e2e_tests.rs b/apps/neshub/src/e2e_tests.rs index 3feb6d43..82b5659c 100644 --- a/apps/neshub/src/e2e_tests.rs +++ b/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"); + } } diff --git a/apps/neshub/src/main.rs b/apps/neshub/src/main.rs index 17baae06..e72fe0e8 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; @@ -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()); diff --git a/apps/neshub/src/pauses.rs b/apps/neshub/src/pauses.rs new file mode 100644 index 00000000..f3e1e67d --- /dev/null +++ b/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, + /// 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 { + 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.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> { + 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> { + 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>, + 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(); + } + + /// 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>, 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 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))] + ); + } +} diff --git a/apps/neshub/src/session.rs b/apps/neshub/src/session.rs index c615548d..e3108251 100644 --- a/apps/neshub/src/session.rs +++ b/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, + /// 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.close(); } } @@ -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) { @@ -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>>, /// 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>>, } @@ -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 { + 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) + } + /// 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 => {} 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), 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. */