From be718bb75633a6fc20930bef876111f24612a7e7 Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Sat, 17 Oct 2026 21:16:55 +0300 Subject: [PATCH 1/7] fix(nesinit): hold a loop device open until its mount is made, or autoclear hands it to the next attach --- apps/nesinit/src/workload.rs | 22 ++++++++++++++-------- 1 file changed, 14 insertions(+), 8 deletions(-) diff --git a/apps/nesinit/src/workload.rs b/apps/nesinit/src/workload.rs index cd9d7c4c..a3ad4d48 100644 --- a/apps/nesinit/src/workload.rs +++ b/apps/nesinit/src/workload.rs @@ -498,9 +498,9 @@ fn mount_overlay(overlay: &Overlay) -> Result<(), Failure> { // Everything that can be refused without touching the filesystem is // refused first, so a descriptor this cannot act on leaves nothing behind. let plan = OverlayPlan::new(overlay)?; - let lower = device_for(&plan.lower, true) + let (lower, _lower_held) = device_for(&plan.lower, true) .map_err(|error| plan.failed("the build image", &plan.lower, error))?; - let upper = device_for(&plan.upper, false) + let (upper, _upper_held) = device_for(&plan.upper, false) .map_err(|error| plan.failed("the writable layer", &plan.upper, error))?; mount_one( @@ -587,7 +587,7 @@ impl ImagePlan { /// Mount one read-only image where the descriptor says. fn mount_image(image: &Image) -> Result<(), Failure> { let plan = ImagePlan::new(image)?; - let device = device_for(&plan.device, true).map_err(|error| plan.failed(error))?; + let (device, _held) = device_for(&plan.device, true).map_err(|error| plan.failed(error))?; mount_one(&device, &plan.at, c"erofs", plan.flags, None).map_err(|error| plan.failed(error)) } @@ -642,12 +642,18 @@ fn unmount_one(target: &std::ffi::CStr) -> Result<(), Failure> { /// file in a share is the host's to name, and a link would let whatever wrote /// it name another. Anything that is neither a block device nor a regular file /// is refused too. -fn device_for(source: &std::ffi::CStr, read_only: bool) -> io::Result { +/// +/// A loop device comes back with its descriptor, which the caller holds until +/// the mount is made: it is attached with autoclear, so the moment nothing +/// holds it open the kernel detaches it, and the next attach is handed the +/// same device. Two attaches in a row (a build image, then its writable layer) +/// then both land on one device and the first mount reads the second file. +fn device_for(source: &std::ffi::CStr, read_only: bool) -> io::Result<(CString, Option)> { use std::os::unix::fs::FileTypeExt; let probe = open_no_symlinks(source, libc::O_RDONLY)?; let kind = probe.metadata()?.file_type(); if kind.is_block_device() { - return Ok(source.to_owned()); + return Ok((source.to_owned(), None)); } if !kind.is_file() { return Err(io::Error::new( @@ -656,7 +662,7 @@ fn device_for(source: &std::ffi::CStr, read_only: bool) -> io::Result { )); } drop(probe); - loop_attach(source, read_only) + loop_attach(source, read_only).map(|(path, device)| (path, Some(device))) } /// `struct open_how` from ``. @@ -731,7 +737,7 @@ const LO_FLAGS_AUTOCLEAR: u32 = 4; /// /// Retried a few times on `EBUSY`: a device handed out as free can be taken /// by somebody else before it is configured. -fn loop_attach(file: &std::ffi::CStr, read_only: bool) -> io::Result { +fn loop_attach(file: &std::ffi::CStr, read_only: bool) -> io::Result<(CString, std::fs::File)> { use std::os::fd::AsRawFd; // Opened again, with the same refusal, because the file may have been @@ -775,7 +781,7 @@ fn loop_attach(file: &std::ffi::CStr, read_only: bool) -> io::Result { ) }; if done == 0 { - return CString::new(path).map_err(io::Error::other); + return Ok((CString::new(path).map_err(io::Error::other)?, device)); } last = io::Error::last_os_error(); if last.raw_os_error() != Some(libc::EBUSY) { From aaa8bfa9949c77ae746fbf5ef537cdbbcead5364 Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Sat, 17 Oct 2026 21:34:35 +0300 Subject: [PATCH 2/7] fix(nesinit): a box takes a new launch after one of its games has closed --- apps/nesinit/src/session.rs | 35 +++++++++++------------------------ 1 file changed, 11 insertions(+), 24 deletions(-) diff --git a/apps/nesinit/src/session.rs b/apps/nesinit/src/session.rs index 9cc6dddc..9c44e28d 100644 --- a/apps/nesinit/src/session.rs +++ b/apps/nesinit/src/session.rs @@ -143,8 +143,9 @@ where let mut running: BTreeMap = BTreeMap::new(); let mut exits: Exits = JoinSet::new(); // A terminal launch's exit, once it has ended while others were still - // running. The box stays up for them, takes no new launches, and ends with - // this exit when the last of them has gone. + // running. The box stays up for them, still takes new launches (closing + // one window is not the person being done, ref(d-0085)), and ends with + // this exit when the last launch in it has gone. let mut ending: Option = None; // Each launch's content mounted after boot, by launch, so it can be taken // down with the launch. Recorded when asked for, before it has finished. @@ -433,19 +434,6 @@ where refuse(&mut writer, id, &reason).await?; continue; } - if ending.is_some() { - // The box was ended by a terminal launch and is waiting - // only for the launches already in it. A new one would - // keep alive a box its person has finished with. - refuse( - &mut writer, - id, - "the box is ending: a launch that ends it has exited", - ) - .await?; - continue; - } - let started = start( &mut writer, workload, @@ -1168,10 +1156,11 @@ mod tests { assert_eq!(outcome, Outcome::WorkloadExited(Exit::code(7))); } - /// Once a terminal launch has ended, the box waits only for what is - /// already in it: a new launch is refused. + /// A terminal launch ending beside another is one window closed, not the + /// person done: the box still takes a new launch, and ends only once the + /// last launch in it has gone. #[tokio::test] - async fn an_ending_box_takes_no_new_launch() { + async fn a_box_takes_a_new_launch_after_one_of_its_launches_ended() { let (mut caller, session) = booted(Given::new(Double::exits_when_stopped(Exit::code(0)))).await; @@ -1183,17 +1172,15 @@ mod tests { assert_eq!(caller.expect().await, exited("l-1")); caller.say(&launch_and_stay("l-3")).await; - let GuestToHost::StartFailed { id, reason } = caller.expect().await else { - panic!("an ending box took a new launch") - }; - assert_eq!(id, LaunchId::new("l-3")); - assert!(reason.contains("ending"), "{reason}"); + assert_eq!(caller.expect().await, started("l-3")); caller.say(&HostToGuest::Stop { id: LaunchId::new("l-2") }).await; assert_eq!(caller.expect().await, exited("l-2")); + caller.say(&HostToGuest::Stop { id: LaunchId::new("l-3") }).await; + assert_eq!(caller.expect().await, exited("l-3")); let (outcome, workload, _) = session.await.unwrap(); assert_eq!(outcome, Outcome::WorkloadExited(Exit::code(0))); - assert_eq!(workload.started.len(), 2); + assert_eq!(workload.started.len(), 3); } /// The last launch of an ending box failing to restart leaves nothing to From 4d97424cc92920dec3656cf902ff8555655cda10 Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Sun, 18 Oct 2026 06:44:03 +0300 Subject: [PATCH 3/7] fix(nesinit): a launch is a session of its own and the box owns the compositor Stopping a launch signalled only its top process, so the compositor wrapper, wine and the game survived and the window stayed open. Each launch now runs under setsid; a stop signals the whole session (TERM, then KILL after 5s) and a launch counts as ended only once its session is empty. The compositor moves out of the first launch into the box: the boot descriptor carries an optional Compositor (geometry, rate, uid, env) and nesinit starts it before the box reports ready, so every launch joins it. --- apps/nesinit/src/services.rs | 4 +- apps/nesinit/src/session.rs | 78 +++++- apps/nesinit/src/workload.rs | 381 +++++++++++++++++++++++++++- crates/nesprotocol/src/lifecycle.rs | 57 +++++ 4 files changed, 508 insertions(+), 12 deletions(-) diff --git a/apps/nesinit/src/services.rs b/apps/nesinit/src/services.rs index 4c396edd..dc38ee1c 100644 --- a/apps/nesinit/src/services.rs +++ b/apps/nesinit/src/services.rs @@ -752,7 +752,7 @@ impl Drop for Stack { /// and bind afresh, which is a new inode every time. type Identity = (u64, u64); -fn identity_of(path: &str) -> Option { +pub(crate) fn identity_of(path: &str) -> Option { use std::os::unix::fs::MetadataExt; std::fs::metadata(path).ok().map(|at| (at.dev(), at.ino())) } @@ -769,7 +769,7 @@ fn is_alive(pid: i32) -> bool { /// The waiting itself, with the deadline passed in so a test can assert the /// giving-up without waiting out a real one. -fn await_path( +pub(crate) fn await_path( name: &str, path: &str, before: Option, diff --git a/apps/nesinit/src/session.rs b/apps/nesinit/src/session.rs index 9c44e28d..8ae86a45 100644 --- a/apps/nesinit/src/session.rs +++ b/apps/nesinit/src/session.rs @@ -401,7 +401,15 @@ where // A box whose own services will not come up cannot be launched // into, so this is refused rather than reported and carried on // from — unlike a launch, which is the caller's to correct. - match services.bring_up(descriptor.video) { + match services.bring_up(descriptor.video).and_then(|mut up| { + // After the services and before the box is ready: the + // first launch joins it like every other. ref(d-0085) + if let Some(compositor) = &descriptor.compositor { + workload.start_compositor(compositor)?; + up.push("nescope".to_string()); + } + Ok(up) + }) { Ok(up) => { tracing::info!(services = up.len(), "the box is ready to be launched into"); send(&mut writer, &GuestToHost::Initialized { services: up }).await? @@ -697,6 +705,7 @@ mod tests { overlays: Vec::new(), images: Vec::new(), video: Default::default(), + compositor: None, } } @@ -895,6 +904,73 @@ mod tests { ); } + /// The box's compositor is the box's, started once with the descriptor and + /// before anything is launched. ref(d-0085) + #[tokio::test] + async fn the_compositor_is_started_by_the_box_and_not_by_a_launch() { + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + let session = spawn(guest, Given::new(Double::exits_when_stopped(Exit::code(0)))); + + let mut given = descriptor(); + given.compositor = Some(nesprotocol::lifecycle::Compositor { + width: 1280, + height: 720, + fps: 30, + uid: 1001, + gid: 1001, + env: Default::default(), + }); + caller.expect_ready().await; + caller + .say(&HostToGuest::Boot { + descriptor: Box::new(given.clone()), + }) + .await; + caller.expect_booted().await; + caller.say(&HostToGuest::Shutdown).await; + + let (_, workload, _) = session.await.unwrap(); + assert_eq!(workload.compositors, vec![given.compositor.unwrap()]); + assert!(workload.started.is_empty()); + } + + /// A box whose compositor will not come up cannot be drawn on, so it is + /// refused like a box whose services will not. + #[tokio::test] + async fn a_compositor_that_will_not_start_refuses_the_box() { + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + let mut double = Double::exits_when_stopped(Exit::code(0)); + double.compositor_failure = Some(Failure::new("no gpu")); + let session = spawn(guest, Given::new(double)); + + let mut given = descriptor(); + given.compositor = Some(nesprotocol::lifecycle::Compositor { + width: 1280, + height: 720, + fps: 30, + uid: 1001, + gid: 1001, + env: Default::default(), + }); + caller.expect_ready().await; + caller + .say(&HostToGuest::Boot { + descriptor: Box::new(given), + }) + .await; + assert_eq!(caller.expect().await, GuestToHost::Mounted); + assert_eq!( + caller.expect().await, + GuestToHost::InitFailed { + reason: "no gpu".into() + } + ); + let (outcome, _, _) = session.await.unwrap(); + assert!(matches!(outcome, Outcome::Refused(_)), "{outcome:?}"); + } + #[tokio::test] async fn the_descriptors_video_limits_reach_the_services() { // The ceiling is useless if it stops at the descriptor. `neshub` is the diff --git a/apps/nesinit/src/workload.rs b/apps/nesinit/src/workload.rs index a3ad4d48..17970443 100644 --- a/apps/nesinit/src/workload.rs +++ b/apps/nesinit/src/workload.rs @@ -12,7 +12,7 @@ use std::io; use std::pin::Pin; use nesprotocol::lifecycle::{ - AttachId, Exec, Exit, Image, LAUNCH_ID_ENV, LaunchId, Mount, Overlay, Winsize, + AttachId, Compositor, Exec, Exit, Image, LAUNCH_ID_ENV, LaunchId, Mount, Overlay, Winsize, }; use std::os::unix::process::CommandExt; @@ -84,9 +84,18 @@ pub trait Workload { /// with any of their games in it, not one game's. ref(d-0085) fn start(&mut self, id: &LaunchId, exec: &Exec) -> Result; - /// Ask one launch to stop. Idempotent, and never ends the session by - /// itself. A launch that is not running is not signalled: its pid may - /// already belong to something else. + /// Start the box's one compositor, and return once it is serving. + /// + /// Owned by the box and not by any launch: every launch draws on it, so no + /// launch's stop may reach it. ref(d-0085) + fn start_compositor(&mut self, compositor: &Compositor) -> Result<(), Failure>; + + /// Ask one launch to stop: its whole process group, not its top process. + /// Idempotent, and never ends the session by itself. A launch that is not + /// running is not signalled: its pid may already belong to something else. + /// + /// The launch's exit is delivered when its group is gone, so it reports + /// the game's window closed rather than only the top process having left. fn stop_launch(&mut self, id: &LaunchId); /// Ask every launch to stop. Idempotent, and never ends the session by @@ -101,6 +110,131 @@ pub struct Process { /// delivered and `Watched::running` is what tells the two apart; it is /// replaced when the same id is started again (a restart). running: std::collections::BTreeMap, + /// The box's compositor, if it has one. Not a launch: nothing in + /// `running` is it, and a launch's stop cannot reach it. + compositor: Option, +} + +/// How long a launch's group has to leave after being asked before it is +/// killed. +pub(crate) const STOP_GRACE: std::time::Duration = std::time::Duration::from_secs(5); + +/// The compositor's socket, in its user's runtime directory. nescope's own +/// default, and what a launch is told to join. +pub(crate) const COMPOSITOR_SOCKET: &str = "nescope-0"; + +/// The compositor's binary in the guest image. +const COMPOSITOR_BIN: &str = "/usr/bin/nescope"; + +/// How long the compositor has to publish its socket. +const COMPOSITOR_READY: std::time::Duration = std::time::Duration::from_secs(15); + +/// Every live process in session `sid`. +/// +/// A launch is started in a session of its own (`setsid`), so its session id is +/// its top pid and everything it starts inherits it. Read from `/proc` rather +/// than asked of the kernel with `kill(-pgid)` alone because a Wine or a +/// launcher may move its own children to groups of their own, and a group that +/// has been left is still in the session. Zombies are not members: the reaper +/// clears them, and waiting on one would wait on nobody. +pub(crate) fn session_members(sid: i32) -> Vec { + let Ok(dir) = std::fs::read_dir("/proc") else { + return Vec::new(); + }; + dir.flatten() + .filter_map(|entry| entry.file_name().to_str()?.parse::().ok()) + .filter(|pid| in_session(*pid, sid)) + .collect() +} + +/// Whether `pid` is a live (not zombie) member of session `sid`. +fn in_session(pid: i32, sid: i32) -> bool { + std::fs::read_to_string(format!("/proc/{pid}/stat")) + .ok() + .and_then(|stat| live_session_of(&stat)) + == Some(sid) +} + +/// The session a `/proc//stat` line belongs to, or `None` for a process +/// that is a zombie or whose line cannot be read. +/// +/// `pid (comm) state ppid pgrp session ...`, with a comm that may hold spaces +/// and parentheses, so it is the last `)` that ends it. +fn live_session_of(stat: &str) -> Option { + let (_, after) = stat.rsplit_once(')')?; + let mut fields = after.split_whitespace(); + let state = fields.next()?; + let session = fields.nth(2)?.parse::().ok()?; + (state != "Z" && state != "X").then_some(session) +} + +/// Send `signal` to a launch's whole group: the process group in one call, then +/// every member of the session by pid, which reaches the ones that left the +/// group. +pub(crate) fn signal_session(sid: i32, signal: libc::c_int) { + let members = session_members(sid); + if members.is_empty() { + // Nothing is left to signal, and `sid` is only a number again: a + // group call now could reach whatever the kernel gave it to. + return; + } + // SAFETY: integers only; a group that has gone fails with ESRCH. + unsafe { libc::kill(-sid, signal) }; + for pid in members { + // SAFETY: as above. + unsafe { libc::kill(pid, signal) }; + } +} + +/// Wait for a launch's group to be gone, ending it if it will not go alone. +/// +/// Reached once the top process has been reaped. A launch counts as ended only +/// when nothing of it is left: a window owned by a process still running is a +/// launch still running, whatever became of the process that started it. +async fn group_gone(sid: i32) { + if session_members(sid).is_empty() { + return; + } + tracing::info!(sid, "a launch's top process has gone and its group has not: ending the group"); + signal_session(sid, libc::SIGTERM); + let deadline = std::time::Instant::now() + STOP_GRACE; + let mut killed = false; + // Past the grace plus as long again: a process in uninterruptible sleep + // ignores SIGKILL, and a launch that never ends is worse than one that + // ended with something stuck in it. + let give_up = deadline + STOP_GRACE * 2; + loop { + if session_members(sid).is_empty() { + return; + } + let now = std::time::Instant::now(); + if now >= give_up { + tracing::warn!(sid, "a launch's group would not die; reporting the launch ended anyway"); + return; + } + if now >= deadline && !killed { + killed = true; + signal_session(sid, libc::SIGKILL); + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } +} + +/// Stop a group: TERM now, KILL after the grace if it is still there. On a +/// thread of its own, because the caller cannot wait and the runtime may be +/// busy with the exits it is meant to deliver. +fn stop_session(sid: i32) { + signal_session(sid, libc::SIGTERM); + std::thread::spawn(move || { + let deadline = std::time::Instant::now() + STOP_GRACE; + while std::time::Instant::now() < deadline { + if session_members(sid).is_empty() { + return; + } + std::thread::sleep(std::time::Duration::from_millis(100)); + } + signal_session(sid, libc::SIGKILL); + }); } impl Process { @@ -110,15 +244,22 @@ impl Process { Self { waiters, running: std::collections::BTreeMap::new(), + compositor: None, } } /// Send a signal to the workload and nothing else. /// - /// Nothing here ever signals the process group or every process: the order - /// a shutdown promises is only true if the workload can be stopped alone. + /// The launches' top processes only, never every process: the order a + /// shutdown promises is only true if the workload can be stopped alone. + /// The one exception is SIGKILL, which is the end of the order and takes + /// each launch's whole group; a launch's own stop does too + /// ([`Workload::stop_launch`]). pub fn signal(&self, signal: libc::c_int) { for watched in self.running.values() { + if signal == libc::SIGKILL { + signal_session(watched.pid, signal); + } Self::signal_one(watched, signal); } } @@ -310,6 +451,13 @@ impl Workload for Process { // allocates nothing. unsafe { command.pre_exec(move || { + // A session of its own: the launch is a group that can be + // stopped alone, which is what makes one launch's stop leave + // another's windows open. Before the drops below, which do + // not care. + if libc::setsid() == -1 { + return Err(io::Error::last_os_error()); + } // gid first: dropping the uid first would lose the privilege // needed to set the gid at all. if libc::setgid(gid) != 0 { @@ -346,15 +494,96 @@ impl Workload for Process { self.running.retain(|_, watched| watched.running()); self.running.insert(id.clone(), watched); + // The session id is the top pid (`setsid` above), and stays that + // number for as long as anything is in it. + let sid = self.running.get(id).map(|w| w.pid).expect("just inserted"); Ok(Box::pin(async move { - exit.await - .map_err(|_| io::Error::other("the workload's exit was not delivered")) + let exit = exit + .await + .map_err(|_| io::Error::other("the workload's exit was not delivered"))?; + // The launch ended when its group has, not when its top process + // did. See `group_gone`. + group_gone(sid).await; + Ok(exit) })) } + fn start_compositor(&mut self, compositor: &Compositor) -> Result<(), Failure> { + let (uid, gid) = (compositor.uid, compositor.gid); + let runtime = crate::system::runtime_dir(uid, gid).map_err(|error| { + Failure::new(format!("no runtime directory for the compositor (uid {uid}): {error}")) + })?; + let exec = Exec { + argv: vec![ + COMPOSITOR_BIN.to_string(), + "--width".into(), + compositor.width.to_string(), + "--height".into(), + compositor.height.to_string(), + "--fps".into(), + compositor.fps.to_string(), + ], + env: compositor.env.clone(), + cwd: None, + uid, + gid, + }; + let mut command = std::process::Command::new(&exec.argv[0]); + command.args(&exec.argv[1..]); + command.env_clear(); + command.envs(environment(&exec, Some(&runtime))); + // SAFETY: async-signal-safe calls only, nothing allocated. + unsafe { + command.pre_exec(move || { + if libc::setsid() == -1 { + return Err(io::Error::last_os_error()); + } + if libc::setgid(gid) != 0 { + return Err(io::Error::last_os_error()); + } + if libc::setuid(uid) != 0 { + return Err(io::Error::last_os_error()); + } + Ok(()) + }); + } + let socket = format!("{runtime}/{COMPOSITOR_SOCKET}"); + let before = crate::services::identity_of(&socket); + let mut watched = self + .waiters + .watch(|| Ok(command.spawn()?.id() as i32)) + .map_err(|error| { + Failure::new(format!("{} as {uid}:{gid}: {error}", exec.argv[0])) + })?; + let exit = watched.take_exit().expect("a new watch has its exit"); + // Said when it goes: with it gone no launch has anywhere to draw, and + // nothing else in the box would say why. + if let Ok(handle) = tokio::runtime::Handle::try_current() { + handle.spawn(async move { + if let Ok(exit) = exit.await { + tracing::error!(?exit, "the compositor has exited: no window in this box can draw"); + } + }); + } + let pid = watched.pid; + let flag = &watched; + crate::services::await_path( + "the compositor", + &socket, + before, + &|| flag.running(), + COMPOSITOR_READY, + )?; + tracing::info!(pid, %socket, "the compositor is serving"); + self.compositor = Some(watched); + Ok(()) + } + fn stop_launch(&mut self, id: &LaunchId) { if let Some(watched) = self.running.get(id) { - Self::signal_one(watched, libc::SIGTERM); + // Whether or not the top process is still there: its children may + // be, and they are what holds the window. + stop_session(watched.pid); } } @@ -1059,6 +1288,126 @@ mod loop_tests { mod tests { use super::*; + #[test] + fn a_stat_line_names_its_session_whatever_the_command_is_called() { + assert_eq!(live_session_of("42 (launcher) S 1 42 42 0 -1"), Some(42)); + assert_eq!(live_session_of("43 (a) b (c) R 42 42 42 0"), Some(42)); + assert_eq!(live_session_of("44 (x) Z 42 42 42 0"), None); + assert_eq!(live_session_of("garbage"), None); + } + + /// Reaps one pid, as the guest's reaper does for every child. + fn reap_for(waiters: crate::reap::Waiters, pid: i32) { + tokio::spawn(async move { + loop { + let mut status = 0; + // SAFETY: writes only into `status`. + let seen = unsafe { libc::waitpid(pid, &mut status, libc::WNOHANG) }; + if seen == pid { + waiters.deliver(pid, crate::reap::exit_of(status)); + return; + } + if seen == -1 { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + }); + } + + fn me(script: &str) -> Exec { + // SAFETY: no arguments. + let (uid, gid) = unsafe { (libc::getuid(), libc::getgid()) }; + Exec { + argv: vec!["/bin/sh".into(), "-c".into(), script.into()], + env: Default::default(), + cwd: None, + uid, + gid, + } + } + + async fn started(script: &str) -> (Process, LaunchId, i32, Exited) { + let waiters = crate::reap::Waiters::new(); + let mut process = Process::new(waiters.clone()); + let id = LaunchId::new("g-1"); + let exited = process.start(&id, &me(script)).expect("it starts"); + let top = process.running.get(&id).unwrap().pid; + reap_for(waiters, top); + // Long enough for the shell to have started its children. + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + (process, id, top, exited) + } + + /// The bug this exists for: a stop that reached the top process only left + /// everything it had started running, and with it the game's window. + #[tokio::test] + async fn stopping_a_launch_ends_its_children_and_it_ends_when_they_have() { + let (mut process, id, top, exited) = started("sleep 300 & sleep 300 & wait").await; + assert!(session_members(top).len() >= 3, "{:?}", session_members(top)); + process.stop_launch(&id); + tokio::time::timeout(std::time::Duration::from_secs(10), exited) + .await + .expect("the launch did not end") + .unwrap(); + assert!(session_members(top).is_empty(), "something of the launch is left"); + } + + /// A child that ignores SIGTERM is killed after the grace, and the launch + /// is not reported ended before it has gone. + #[tokio::test] + async fn a_child_that_ignores_the_stop_is_killed_after_the_grace() { + let (mut process, id, top, exited) = + started("trap '' TERM; (trap '' TERM; sleep 300) & wait").await; + process.stop_launch(&id); + let began = std::time::Instant::now(); + tokio::time::timeout(std::time::Duration::from_secs(15), exited) + .await + .expect("the launch did not end") + .unwrap(); + assert!(began.elapsed() >= STOP_GRACE.saturating_sub(std::time::Duration::from_millis(500))); + assert!(session_members(top).is_empty()); + } + + /// The top process leaving by itself does not end the launch while + /// something of it runs; the leftovers are ended and then it has ended. + #[tokio::test] + async fn a_launch_whose_top_process_left_is_not_ended_until_its_group_is() { + let (_process, _id, top, exited) = started("sleep 300 & exit 3").await; + let exit = tokio::time::timeout(std::time::Duration::from_secs(10), exited) + .await + .expect("the launch did not end") + .unwrap(); + assert_eq!(exit, Exit::code(3)); + assert!(session_members(top).is_empty()); + } + + /// Two launches are two groups: stopping one leaves the other running. + #[tokio::test] + async fn stopping_one_launch_leaves_another_running() { + let waiters = crate::reap::Waiters::new(); + let mut process = Process::new(waiters.clone()); + let (a, b) = (LaunchId::new("a"), LaunchId::new("b")); + let exited_a = process.start(&a, &me("sleep 300 & wait")).unwrap(); + let _exited_b = process.start(&b, &me("sleep 300 & wait")).unwrap(); + let (top_a, top_b) = ( + process.running[&a].pid, + process.running[&b].pid, + ); + reap_for(waiters.clone(), top_a); + reap_for(waiters, top_b); + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + process.stop_launch(&a); + tokio::time::timeout(std::time::Duration::from_secs(10), exited_a) + .await + .unwrap() + .unwrap(); + assert!(session_members(top_a).is_empty()); + assert!(session_members(top_b).len() >= 2, "the other launch was touched"); + signal_session(top_b, libc::SIGKILL); + } + + /// The driver override has to reach the workload, because nothing else /// carries it: the image's profile script never runs for an exec'd /// process. Without it a game's OpenGL finds no driver at all. @@ -1370,6 +1719,9 @@ pub mod double { /// Every stop asked for, one launch or all (`None`). pub stopped: Vec>, pub stops: usize, + /// Every compositor asked for. + pub compositors: Vec, + pub compositor_failure: Option, pub mount_failure: Option, pub image_failure: Option, /// Like `image_failure`, but only for a call that names an image, so a @@ -1407,6 +1759,8 @@ pub mod double { started_ids: Vec::new(), stopped: Vec::new(), stops: 0, + compositors: Vec::new(), + compositor_failure: None, mount_failure: None, image_failure: None, named_image_failure: None, @@ -1496,6 +1850,15 @@ pub mod double { })) } + fn start_compositor(&mut self, compositor: &Compositor) -> Result<(), Failure> { + self.order.push("compositor"); + self.compositors.push(compositor.clone()); + match &self.compositor_failure { + Some(failure) => Err(failure.clone()), + None => Ok(()), + } + } + fn stop_launch(&mut self, id: &LaunchId) { self.stops += 1; self.stopped.push(Some(id.clone())); diff --git a/crates/nesprotocol/src/lifecycle.rs b/crates/nesprotocol/src/lifecycle.rs index e88a2aa2..a98edae6 100644 --- a/crates/nesprotocol/src/lifecycle.rs +++ b/crates/nesprotocol/src/lifecycle.rs @@ -284,6 +284,35 @@ pub struct BootDescriptor { /// A number travels to where the thing that reads it is started. #[serde(default)] pub video: VideoLimits, + /// The box's one compositor, started by the init when the box is ready to + /// be launched into. ref(d-0085) + /// + /// Every launch draws on it, the first as much as the tenth: a launch is a + /// process group that can be stopped alone, and a compositor owned by one + /// launch could not be, because stopping that launch would take every other + /// window with it. `None` is a box with no compositor of its own, which is + /// what a workload that draws nothing wants. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub compositor: Option, +} + +/// What the compositor is started with. +/// +/// Geometry and rate are the box's, set once: every window in the box is a +/// stream of its own, but they share one output. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Compositor { + pub width: u32, + pub height: u32, + pub fps: u32, + /// Who it runs as: the same user the box's launches run as, because the + /// socket it publishes is in that user's runtime directory. + pub uid: u32, + pub gid: u32, + /// Environment on top of the guest's own, sorted like [`Exec::env`]. + #[serde(default)] + pub env: BTreeMap, } /// Ceilings on what a box's video may cost. @@ -577,6 +606,7 @@ mod tests { overlays: Vec::new(), images: Vec::new(), video: VideoLimits::default(), + compositor: None, } } @@ -882,6 +912,7 @@ mod video_limits_tests { overlays: Vec::new(), images: Vec::new(), video: VideoLimits::default(), + compositor: None, }; let json = serde_json::to_string(&d).unwrap(); assert!(!json.contains("bitrate"), "{json}"); @@ -896,6 +927,7 @@ mod video_limits_tests { video: VideoLimits { bitrate_kbps: Some(8_000), }, + compositor: None, }; let back: BootDescriptor = serde_json::from_str(&serde_json::to_string(&d).unwrap()).unwrap(); @@ -946,6 +978,7 @@ mod image_tests { at: "/nestri/compat".into(), }], video: VideoLimits::default(), + compositor: None, }; let json = serde_json::to_string(&d).unwrap(); assert!(json.contains(r#""images":[{"device":"/dev/vdd","at":"/nestri/compat"}]"#)); @@ -1008,4 +1041,28 @@ mod attach_tests { fn the_data_port_is_its_own() { assert_ne!(ATTACH_PORT, CONTROL_PORT); } + + /// A box with no compositor says nothing about one, so a descriptor from + /// before the field existed reads the same, and the compositor survives a + /// round trip with its environment. + #[test] + fn a_compositor_is_optional_and_round_trips() { + let none: BootDescriptor = serde_json::from_str("{}").unwrap(); + assert!(none.compositor.is_none()); + assert!(!serde_json::to_string(&none).unwrap().contains("compositor")); + let d = BootDescriptor { + compositor: Some(Compositor { + width: 1920, + height: 1080, + fps: 60, + uid: 1001, + gid: 1001, + env: BTreeMap::from([("A".to_string(), "b".to_string())]), + }), + ..none + }; + let back: BootDescriptor = + serde_json::from_str(&serde_json::to_string(&d).unwrap()).unwrap(); + assert_eq!(back, d); + } } From d5dbefe9d0e1ce78e902befbd793fa14d919b49e Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Sun, 18 Oct 2026 07:17:06 +0300 Subject: [PATCH 4/7] fix(nesinit): a launch owns a cgroup, so a stop reaches Wine's own sessions Wine setsid's its processes, so stopping a launch by session left wineserver, explorer.exe and the game running with their window open. Each launch is now put in cgroup /sys/fs/cgroup/nestri-launch- before exec; stop is SIGTERM to its members, then cgroup.kill after the grace, and the launch ends when the cgroup is empty. Without cgroup v2 the members are the session plus every process whose environment carries NESTRI_LAUNCH_ID. cgroup2 is mounted at /sys/fs/cgroup early in boot. --- apps/nesinit/src/filesystems.rs | 13 ++ apps/nesinit/src/workload.rs | 317 +++++++++++++++++++++++++++----- 2 files changed, 283 insertions(+), 47 deletions(-) diff --git a/apps/nesinit/src/filesystems.rs b/apps/nesinit/src/filesystems.rs index 005c40b3..e6254ff0 100644 --- a/apps/nesinit/src/filesystems.rs +++ b/apps/nesinit/src/filesystems.rs @@ -87,6 +87,19 @@ const EARLY: &[Early] = &[ data: "", cost: "a workload that looks up a device finds nothing", }, + // One cgroup per launch, so that a stop reaches what a launch started + // however many sessions it moved them to (Wine does). `cgroup.kill` needs + // no controller. A kernel without cgroup v2 fails this and launches are + // found by their environment instead. + Early { + source: "cgroup2", + target: "/sys/fs/cgroup", + fstype: "cgroup2", + flags: NOSUID_NODEV | libc::MS_NOEXEC, + data: "", + cost: "a stop cannot reach the processes a launch moved to sessions of \ + their own, and is left to find them by environment", + }, // Writable, and the reason any of this is here. Both sockets in this // component live on a tmpfs because the root they would otherwise sit on // is read-only. diff --git a/apps/nesinit/src/workload.rs b/apps/nesinit/src/workload.rs index 17970443..dc19299d 100644 --- a/apps/nesinit/src/workload.rs +++ b/apps/nesinit/src/workload.rs @@ -9,6 +9,7 @@ use std::ffi::CString; use std::future::Future; use std::io; +use std::os::unix::ffi::OsStringExt as _; use std::pin::Pin; use nesprotocol::lifecycle::{ @@ -110,6 +111,10 @@ pub struct Process { /// delivered and `Watched::running` is what tells the two apart; it is /// replaced when the same id is started again (a restart). running: std::collections::BTreeMap, + /// The group of each launch in `running`, which outlives its top process. + groups: std::collections::BTreeMap>, + /// Where launch cgroups are made: [`CGROUP_ROOT`], except in a test. + cgroup_root: String, /// The box's compositor, if it has one. Not a launch: nothing in /// `running` is it, and a launch's stop cannot reach it. compositor: Option, @@ -168,21 +173,135 @@ fn live_session_of(stat: &str) -> Option { (state != "Z" && state != "X").then_some(session) } -/// Send `signal` to a launch's whole group: the process group in one call, then -/// every member of the session by pid, which reaches the ones that left the -/// group. -pub(crate) fn signal_session(sid: i32, signal: libc::c_int) { - let members = session_members(sid); - if members.is_empty() { - // Nothing is left to signal, and `sid` is only a number again: a - // group call now could reach whatever the kernel gave it to. - return; +/// Where launch cgroups live: the cgroup v2 root, mounted by `filesystems`. +const CGROUP_ROOT: &str = "/sys/fs/cgroup"; + +/// Everything that belongs to one launch. +/// +/// A session is not enough: Wine `setsid`s its own processes (wineserver, +/// services.exe, explorer.exe and the game itself each have a session of their +/// own), so a stop that signalled the launch's session left the game running +/// with its window open. Measured 2026-10-18 on the test box. +/// +/// What a child cannot leave is a cgroup, so a launch is put in one before it +/// execs and the kernel's own list of its members is what is signalled. Where +/// no cgroup could be made (v2 not mounted) the members are the launch's +/// session plus every process whose environment names the launch, which Wine +/// cannot drop either: it is inherited. +#[derive(Debug)] +pub(crate) struct Group { + id: String, + /// The top process, which is also the session id. + sid: i32, + cgroup: Option, +} + +/// An id as a directory name: only what is safe in a path. +fn sanitise(id: &str) -> String { + id.chars() + .map(|c| if c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '.' { c } else { '_' }) + .collect() +} + +/// Make the cgroup for launch `id` under `base`, or `None` when it cannot be +/// (no cgroup v2, no permission, or a populated one of that name already). +fn make_cgroup(base: &str, id: &str) -> Option { + let path = std::path::Path::new(base).join(format!("nestri-launch-{}", sanitise(id))); + match std::fs::create_dir(&path) { + Ok(()) => Some(path), + Err(error) if error.kind() == io::ErrorKind::AlreadyExists => { + // A leftover from a launch with this id is reused only when empty: + // sharing one would make a stop of one launch the stop of another. + let procs = std::fs::read_to_string(path.join("cgroup.procs")).ok()?; + procs.trim().is_empty().then_some(path) + } + Err(error) => { + tracing::warn!(%id, base, "no cgroup for this launch ({error}); its members are found by environment"); + None + } } - // SAFETY: integers only; a group that has gone fails with ESRCH. - unsafe { libc::kill(-sid, signal) }; - for pid in members { - // SAFETY: as above. - unsafe { libc::kill(pid, signal) }; +} + +/// Whether `pid` is not a zombie. +fn is_live(pid: i32) -> bool { + std::fs::read_to_string(format!("/proc/{pid}/stat")) + .ok() + .and_then(|stat| live_session_of(&stat)) + .is_some() +} + +/// Whether the NUL-separated environment `environ` carries `NAME=value`. +fn environ_names(environ: &[u8], id: &str) -> bool { + let want = format!("{LAUNCH_ID_ENV}={id}"); + environ.split(|b| *b == 0).any(|entry| entry == want.as_bytes()) +} + +impl Group { + /// The group of a launch whose top process is `sid`. + fn new(id: &LaunchId, sid: i32, cgroup: Option) -> Self { + Self { id: id.as_str().to_string(), sid, cgroup } + } + + /// Every live process of the launch. + pub(crate) fn members(&self) -> Vec { + if let Some(path) = &self.cgroup { + if let Ok(procs) = std::fs::read_to_string(path.join("cgroup.procs")) { + return procs + .lines() + .filter_map(|line| line.trim().parse::().ok()) + .filter(|pid| is_live(*pid)) + .collect(); + } + } + let mut members = session_members(self.sid); + if let Ok(dir) = std::fs::read_dir("/proc") { + for pid in dir.flatten().filter_map(|e| e.file_name().to_str()?.parse::().ok()) { + if members.contains(&pid) || !is_live(pid) { + continue; + } + let named = std::fs::read(format!("/proc/{pid}/environ")) + .map(|environ| environ_names(&environ, &self.id)) + .unwrap_or(false); + if named { + members.push(pid); + } + } + } + members + } + + /// Signal everything of the launch. + pub(crate) fn signal(&self, signal: libc::c_int) { + let members = self.members(); + if members.is_empty() { + // Only a number again: a group call now could reach whatever the + // kernel gave it to. + return; + } + // SAFETY: integers only; a group that has gone fails with ESRCH. + unsafe { libc::kill(-self.sid, signal) }; + for pid in members { + // SAFETY: as above. + unsafe { libc::kill(pid, signal) }; + } + } + + /// End everything of the launch now, with no chance to refuse. + pub(crate) fn kill(&self) { + if let Some(path) = &self.cgroup { + if std::fs::write(path.join("cgroup.kill"), "1").is_ok() { + return; + } + // Before Linux 5.14 there is no `cgroup.kill`: kill by pid. + } + self.signal(libc::SIGKILL); + } + + /// Remove the launch's cgroup, once it is empty. + fn remove(&self) { + if let Some(path) = &self.cgroup { + let _ = std::fs::remove_dir(path); + } } } @@ -191,12 +310,13 @@ pub(crate) fn signal_session(sid: i32, signal: libc::c_int) { /// Reached once the top process has been reaped. A launch counts as ended only /// when nothing of it is left: a window owned by a process still running is a /// launch still running, whatever became of the process that started it. -async fn group_gone(sid: i32) { - if session_members(sid).is_empty() { +async fn group_gone(group: std::sync::Arc) { + if group.members().is_empty() { + group.remove(); return; } - tracing::info!(sid, "a launch's top process has gone and its group has not: ending the group"); - signal_session(sid, libc::SIGTERM); + tracing::info!(sid = group.sid, "a launch's top process has gone and its group has not: ending the group"); + group.signal(libc::SIGTERM); let deadline = std::time::Instant::now() + STOP_GRACE; let mut killed = false; // Past the grace plus as long again: a process in uninterruptible sleep @@ -204,36 +324,37 @@ async fn group_gone(sid: i32) { // ended with something stuck in it. let give_up = deadline + STOP_GRACE * 2; loop { - if session_members(sid).is_empty() { + if group.members().is_empty() { + group.remove(); return; } let now = std::time::Instant::now(); if now >= give_up { - tracing::warn!(sid, "a launch's group would not die; reporting the launch ended anyway"); + tracing::warn!(sid = group.sid, "a launch's group would not die; reporting the launch ended anyway"); return; } if now >= deadline && !killed { killed = true; - signal_session(sid, libc::SIGKILL); + group.kill(); } tokio::time::sleep(std::time::Duration::from_millis(100)).await; } } -/// Stop a group: TERM now, KILL after the grace if it is still there. On a -/// thread of its own, because the caller cannot wait and the runtime may be -/// busy with the exits it is meant to deliver. -fn stop_session(sid: i32) { - signal_session(sid, libc::SIGTERM); +/// Stop a group: TERM now, then `cgroup.kill` (or KILL) after the grace if it +/// is still there. On a thread of its own, because the caller cannot wait and +/// the runtime may be busy with the exits it is meant to deliver. +fn stop_group(group: std::sync::Arc) { + group.signal(libc::SIGTERM); std::thread::spawn(move || { let deadline = std::time::Instant::now() + STOP_GRACE; while std::time::Instant::now() < deadline { - if session_members(sid).is_empty() { + if group.members().is_empty() { return; } std::thread::sleep(std::time::Duration::from_millis(100)); } - signal_session(sid, libc::SIGKILL); + group.kill(); }); } @@ -244,6 +365,8 @@ impl Process { Self { waiters, running: std::collections::BTreeMap::new(), + groups: std::collections::BTreeMap::new(), + cgroup_root: CGROUP_ROOT.to_string(), compositor: None, } } @@ -257,11 +380,13 @@ impl Process { /// ([`Workload::stop_launch`]). pub fn signal(&self, signal: libc::c_int) { for watched in self.running.values() { - if signal == libc::SIGKILL { - signal_session(watched.pid, signal); - } Self::signal_one(watched, signal); } + if signal == libc::SIGKILL { + for group in self.groups.values() { + group.kill(); + } + } } fn signal_one(watched: &Watched, signal: libc::c_int) { @@ -446,8 +571,16 @@ impl Workload for Process { }; command.envs(launch_environment(exec, runtime.as_deref(), id)); + // The launch's cgroup, made here and joined before exec: what a child + // that makes its own session still cannot leave. The path is built + // now because the closure below may not allocate. + let cgroup = make_cgroup(&self.cgroup_root, id.as_str()); + let procs_file = cgroup + .as_ref() + .and_then(|path| std::ffi::CString::new(path.join("cgroup.procs").into_os_string().into_vec()).ok()); + // SAFETY: the closure runs between fork and exec in the child, where - // only async-signal-safe calls are allowed. These two are, and it + // only async-signal-safe calls are allowed. These are, and it // allocates nothing. unsafe { command.pre_exec(move || { @@ -458,6 +591,16 @@ impl Workload for Process { if libc::setsid() == -1 { return Err(io::Error::last_os_error()); } + // Into the launch's cgroup, as root, before the uid drops. + // A failure is not one: the launch is still found by its + // environment. + if let Some(file) = &procs_file { + let fd = libc::open(file.as_ptr(), libc::O_WRONLY | libc::O_CLOEXEC); + if fd >= 0 { + libc::write(fd, b"0".as_ptr().cast(), 1); + libc::close(fd); + } + } // gid first: dropping the uid first would lose the privilege // needed to set the gid at all. if libc::setgid(gid) != 0 { @@ -474,6 +617,9 @@ impl Workload for Process { .waiters .watch(|| Ok(command.spawn()?.id() as i32)) .map_err(|error| { + if let Some(path) = &cgroup { + let _ = std::fs::remove_dir(path); + } // The program, the user, and what the system said. // // `Permission denied` on its own is the least useful true @@ -497,13 +643,17 @@ impl Workload for Process { // The session id is the top pid (`setsid` above), and stays that // number for as long as anything is in it. let sid = self.running.get(id).map(|w| w.pid).expect("just inserted"); + // Groups with nothing left in them are dropped here, as `running` is. + self.groups.retain(|_, group| !group.members().is_empty()); + let group = std::sync::Arc::new(Group::new(id, sid, cgroup)); + self.groups.insert(id.clone(), group.clone()); Ok(Box::pin(async move { let exit = exit .await .map_err(|_| io::Error::other("the workload's exit was not delivered"))?; // The launch ended when its group has, not when its top process // did. See `group_gone`. - group_gone(sid).await; + group_gone(group).await; Ok(exit) })) } @@ -580,10 +730,10 @@ impl Workload for Process { } fn stop_launch(&mut self, id: &LaunchId) { - if let Some(watched) = self.running.get(id) { + if let Some(group) = self.groups.get(id) { // Whether or not the top process is still there: its children may // be, and they are what holds the window. - stop_session(watched.pid); + stop_group(group.clone()); } } @@ -1330,7 +1480,13 @@ mod tests { async fn started(script: &str) -> (Process, LaunchId, i32, Exited) { let waiters = crate::reap::Waiters::new(); let mut process = Process::new(waiters.clone()); - let id = LaunchId::new("g-1"); + static NEXT: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0); + // Unique per test: members are found by the id in their environment. + let id = LaunchId::new(format!( + "g-{}-{}", + std::process::id(), + NEXT.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + )); let exited = process.start(&id, &me(script)).expect("it starts"); let top = process.running.get(&id).unwrap().pid; reap_for(waiters, top); @@ -1343,22 +1499,24 @@ mod tests { /// everything it had started running, and with it the game's window. #[tokio::test] async fn stopping_a_launch_ends_its_children_and_it_ends_when_they_have() { - let (mut process, id, top, exited) = started("sleep 300 & sleep 300 & wait").await; - assert!(session_members(top).len() >= 3, "{:?}", session_members(top)); + let (mut process, id, _top, exited) = started("sleep 300 & sleep 300 & wait").await; + let group = process.groups[&id].clone(); + assert!(group.members().len() >= 3, "{:?}", group.members()); process.stop_launch(&id); tokio::time::timeout(std::time::Duration::from_secs(10), exited) .await .expect("the launch did not end") .unwrap(); - assert!(session_members(top).is_empty(), "something of the launch is left"); + assert!(group.members().is_empty(), "something of the launch is left"); } /// A child that ignores SIGTERM is killed after the grace, and the launch /// is not reported ended before it has gone. #[tokio::test] async fn a_child_that_ignores_the_stop_is_killed_after_the_grace() { - let (mut process, id, top, exited) = + let (mut process, id, _top, exited) = started("trap '' TERM; (trap '' TERM; sleep 300) & wait").await; + let group = process.groups[&id].clone(); process.stop_launch(&id); let began = std::time::Instant::now(); tokio::time::timeout(std::time::Duration::from_secs(15), exited) @@ -1366,20 +1524,21 @@ mod tests { .expect("the launch did not end") .unwrap(); assert!(began.elapsed() >= STOP_GRACE.saturating_sub(std::time::Duration::from_millis(500))); - assert!(session_members(top).is_empty()); + assert!(group.members().is_empty()); } /// The top process leaving by itself does not end the launch while /// something of it runs; the leftovers are ended and then it has ended. #[tokio::test] async fn a_launch_whose_top_process_left_is_not_ended_until_its_group_is() { - let (_process, _id, top, exited) = started("sleep 300 & exit 3").await; + let (process, id, _top, exited) = started("sleep 300 & exit 3").await; + let group = process.groups[&id].clone(); let exit = tokio::time::timeout(std::time::Duration::from_secs(10), exited) .await .expect("the launch did not end") .unwrap(); assert_eq!(exit, Exit::code(3)); - assert!(session_members(top).is_empty()); + assert!(group.members().is_empty()); } /// Two launches are two groups: stopping one leaves the other running. @@ -1387,7 +1546,7 @@ mod tests { async fn stopping_one_launch_leaves_another_running() { let waiters = crate::reap::Waiters::new(); let mut process = Process::new(waiters.clone()); - let (a, b) = (LaunchId::new("a"), LaunchId::new("b")); + let (a, b) = (LaunchId::new("iso-a"), LaunchId::new("iso-b")); let exited_a = process.start(&a, &me("sleep 300 & wait")).unwrap(); let _exited_b = process.start(&b, &me("sleep 300 & wait")).unwrap(); let (top_a, top_b) = ( @@ -1402,9 +1561,73 @@ mod tests { .await .unwrap() .unwrap(); - assert!(session_members(top_a).is_empty()); - assert!(session_members(top_b).len() >= 2, "the other launch was touched"); - signal_session(top_b, libc::SIGKILL); + assert!(process.groups[&a].members().is_empty()); + assert!(process.groups[&b].members().len() >= 2, "the other launch was touched"); + process.groups[&b].kill(); + } + + /// The case that broke #433: Wine `setsid`s its own processes, so they are + /// in no session of the launch's. The launch still owns them, by its + /// environment (no cgroup here: the test is not root), and a stop ends + /// them. + #[tokio::test] + async fn a_child_that_makes_its_own_session_is_still_stopped() { + let (mut process, id, top, exited) = + started("setsid sleep 300 & setsid sh -c 'sleep 300 & wait' & wait").await; + let group = process.groups[&id].clone(); + assert!( + session_members(top).len() < group.members().len(), + "the setsid children should be outside the launch's session: {:?}", + group.members() + ); + process.stop_launch(&id); + tokio::time::timeout(std::time::Duration::from_secs(10), exited) + .await + .expect("the launch did not end") + .unwrap(); + assert!(group.members().is_empty(), "something of the launch is left"); + } + + /// The same through a real cgroup. Needs a cgroup v2 directory the test + /// may create in, named by `NESTRI_TEST_CGROUP_BASE` (a systemd user + /// delegation, or `/sys/fs/cgroup` as root). + #[tokio::test] + #[ignore = "needs a writable cgroup v2 directory in NESTRI_TEST_CGROUP_BASE"] + async fn a_cgroup_holds_a_child_that_makes_its_own_session() { + let base = std::env::var("NESTRI_TEST_CGROUP_BASE").expect("NESTRI_TEST_CGROUP_BASE"); + let waiters = crate::reap::Waiters::new(); + let mut process = Process::new(waiters.clone()); + process.cgroup_root = base; + let id = LaunchId::new(format!("cg-{}", std::process::id())); + let exited = process + .start(&id, &me("setsid sleep 300 & setsid sh -c 'sleep 300 & wait' & wait")) + .unwrap(); + let top = process.running[&id].pid; + reap_for(waiters, top); + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + let group = process.groups[&id].clone(); + let path = group.cgroup.clone().expect("the cgroup was made"); + assert!(group.members().len() >= 4, "{:?}", group.members()); + process.stop_launch(&id); + tokio::time::timeout(std::time::Duration::from_secs(10), exited) + .await + .expect("the launch did not end") + .unwrap(); + assert!(group.members().is_empty()); + assert!(!path.exists(), "the cgroup was not removed"); + } + + #[test] + fn the_launch_id_is_found_whole_in_an_environment() { + let env = b"A=1\0NESTRI_LAUNCH_ID=g-10\0B=2\0"; + assert!(environ_names(env, "g-10")); + assert!(!environ_names(env, "g-1")); + assert!(!environ_names(b"X_NESTRI_LAUNCH_ID=g-10\0", "g-10")); + } + + #[test] + fn an_id_becomes_a_safe_directory_name() { + assert_eq!(sanitise("a/../b c"), "a_.._b_c"); } From ceebee7183bb8372cd6bc68b9f39bad6c7c249a6 Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Sun, 18 Oct 2026 07:28:56 +0300 Subject: [PATCH 5/7] fix(nesinit): a launch's group survives a failed join, a re-add and a busy rmdir; a dead compositor ends the box --- apps/nesinit/src/session.rs | 51 ++++++++ apps/nesinit/src/workload.rs | 225 ++++++++++++++++++++++++++++------- 2 files changed, 236 insertions(+), 40 deletions(-) diff --git a/apps/nesinit/src/session.rs b/apps/nesinit/src/session.rs index 8ae86a45..4edff5a2 100644 --- a/apps/nesinit/src/session.rs +++ b/apps/nesinit/src/session.rs @@ -158,6 +158,8 @@ where // resolves immediately and forever, so an unguarded branch on one turns // this loop into a spin that still looks like it is waiting. let mut services_open = true; + // The box's compositor exit, once there is a compositor. + let mut compositor_exit: Option = None; loop { // The exits are guarded like the channels below: an empty set resolves @@ -177,6 +179,9 @@ where up = payload.from_workload.recv(), if relay_open => Event::FromWorkload(up), found = addresses.recv(), if carrier_open => Event::Address(found), gone = services.deaths().recv(), if services_open => Event::ServiceDied(gone), + gone = async { compositor_exit.as_mut().expect("guarded").await }, if compositor_exit.is_some() => { + Event::CompositorDied(gone) + }, }; let line = match event { @@ -281,6 +286,18 @@ where send(&mut writer, &GuestToHost::ServiceDied { name, exit }).await?; continue; } + Event::CompositorDied(gone) => { + // Every launch draws on it, so with it gone nothing in this box + // can show a window: said as a service death, then the box + // ends rather than carrying on looking healthy. + let exit = gone.unwrap_or_else(|_| Exit::code(1)); + tracing::error!(?exit, "the compositor has exited: no window in this box can draw"); + send(&mut writer, &GuestToHost::ServiceDied { name: "nescope".into(), exit }).await?; + workload.signal_stop(); + return Ok(Outcome::Refused(Failure::new(format!( + "the compositor exited ({exit:?}), so the box can no longer draw" + )))); + } Event::ServiceDied(None) => { // The supervisor is gone, which on the way down is ordinary. // Nothing to report and nothing to end: a box with no service @@ -411,6 +428,7 @@ where Ok(up) }) { Ok(up) => { + compositor_exit = workload.compositor_exit(); tracing::info!(services = up.len(), "the box is ready to be launched into"); send(&mut writer, &GuestToHost::Initialized { services: up }).await? } @@ -651,6 +669,7 @@ enum Event { FromWorkload(Option), Address(Option), ServiceDied(Option), + CompositorDied(std::io::Result), } /// Hand an envelope to the relay, and treat a relay that is not there as the @@ -971,6 +990,38 @@ mod tests { assert!(matches!(outcome, Outcome::Refused(_)), "{outcome:?}"); } + /// A compositor that exits ends the box, and the host is told which + /// service it was. + #[tokio::test] + async fn a_compositor_that_exits_is_reported_and_ends_the_box() { + let (guest, host) = tokio::io::duplex(4096); + let mut caller = Caller::new(host); + let (tx, rx) = tokio::sync::oneshot::channel(); + let mut double = Double::exits_when_stopped(Exit::code(0)); + double.compositor_exit = Some(rx); + let session = spawn(guest, Given::new(double)); + + let mut given = descriptor(); + given.compositor = Some(nesprotocol::lifecycle::Compositor { + width: 1280, + height: 720, + fps: 30, + uid: 1001, + gid: 1001, + env: Default::default(), + }); + caller.expect_ready().await; + caller.say(&HostToGuest::Boot { descriptor: Box::new(given) }).await; + caller.expect_booted().await; + tx.send(Exit::signal(11)).unwrap(); + assert_eq!( + caller.expect().await, + GuestToHost::ServiceDied { name: "nescope".into(), exit: Exit::signal(11) } + ); + let (outcome, _, _) = session.await.unwrap(); + assert!(matches!(outcome, Outcome::Refused(_)), "{outcome:?}"); + } + #[tokio::test] async fn the_descriptors_video_limits_reach_the_services() { // The ceiling is useless if it stops at the descriptor. `neshub` is the diff --git a/apps/nesinit/src/workload.rs b/apps/nesinit/src/workload.rs index dc19299d..6684bc7d 100644 --- a/apps/nesinit/src/workload.rs +++ b/apps/nesinit/src/workload.rs @@ -91,6 +91,14 @@ pub trait Workload { /// launch's stop may reach it. ref(d-0085) fn start_compositor(&mut self, compositor: &Compositor) -> Result<(), Failure>; + /// The compositor's exit, once, after [`Workload::start_compositor`] has + /// returned. The session reports it and ends the box: a box that cannot + /// draw is no use to anyone in it. `None` when there is no compositor or it + /// was already taken. + fn compositor_exit(&mut self) -> Option { + None + } + /// Ask one launch to stop: its whole process group, not its top process. /// Idempotent, and never ends the session by itself. A launch that is not /// running is not signalled: its pid may already belong to something else. @@ -118,6 +126,9 @@ pub struct Process { /// The box's compositor, if it has one. Not a launch: nothing in /// `running` is it, and a launch's stop cannot reach it. compositor: Option, + /// The compositor's exit, until the session takes it. See + /// [`Workload::compositor_exit`]. + compositor_exit: Option, } /// How long a launch's group has to leave after being asked before it is @@ -194,8 +205,14 @@ pub(crate) struct Group { /// The top process, which is also the session id. sid: i32, cgroup: Option, + /// Set when the same id is started again: this group is no longer the + /// launch's, and nothing waiting on it may act for the launch. + superseded: std::sync::atomic::AtomicBool, } +/// Numbers each group, so two launches of one id never share a cgroup. +static GENERATION: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + /// An id as a directory name: only what is safe in a path. fn sanitise(id: &str) -> String { id.chars() @@ -203,18 +220,14 @@ fn sanitise(id: &str) -> String { .collect() } -/// Make the cgroup for launch `id` under `base`, or `None` when it cannot be -/// (no cgroup v2, no permission, or a populated one of that name already). -fn make_cgroup(base: &str, id: &str) -> Option { - let path = std::path::Path::new(base).join(format!("nestri-launch-{}", sanitise(id))); +/// Make the cgroup for launch `id`, generation `generation`, under `base`, or +/// `None` when it cannot be (no cgroup v2, no permission, or one of that name +/// already). The generation is in the name so a restart never shares a +/// directory with what is left of the launch before it. +fn make_cgroup(base: &str, id: &str, generation: u64) -> Option { + let path = std::path::Path::new(base).join(format!("nestri-launch-{}-{generation}", sanitise(id))); match std::fs::create_dir(&path) { Ok(()) => Some(path), - Err(error) if error.kind() == io::ErrorKind::AlreadyExists => { - // A leftover from a launch with this id is reused only when empty: - // sharing one would make a stop of one launch the stop of another. - let procs = std::fs::read_to_string(path.join("cgroup.procs")).ok()?; - procs.trim().is_empty().then_some(path) - } Err(error) => { tracing::warn!(%id, base, "no cgroup for this launch ({error}); its members are found by environment"); None @@ -222,6 +235,14 @@ fn make_cgroup(base: &str, id: &str) -> Option { } } +/// Whether `pid` is listed in the cgroup at `path`: the proof the launch's +/// child joined it, which its own unchecked write could not give. +fn joined(path: &std::path::Path, pid: i32) -> bool { + std::fs::read_to_string(path.join("cgroup.procs")) + .map(|procs| procs.lines().any(|line| line.trim().parse::().ok() == Some(pid))) + .unwrap_or(false) +} + /// Whether `pid` is not a zombie. fn is_live(pid: i32) -> bool { std::fs::read_to_string(format!("/proc/{pid}/stat")) @@ -239,18 +260,28 @@ fn environ_names(environ: &[u8], id: &str) -> bool { impl Group { /// The group of a launch whose top process is `sid`. fn new(id: &LaunchId, sid: i32, cgroup: Option) -> Self { - Self { id: id.as_str().to_string(), sid, cgroup } + Self { id: id.as_str().to_string(), sid, cgroup, superseded: false.into() } + } + + /// Whether the launch has been started again since this group was made. + fn is_superseded(&self) -> bool { + self.superseded.load(std::sync::atomic::Ordering::SeqCst) } /// Every live process of the launch. pub(crate) fn members(&self) -> Vec { if let Some(path) = &self.cgroup { if let Ok(procs) = std::fs::read_to_string(path.join("cgroup.procs")) { - return procs + let found: Vec = procs .lines() .filter_map(|line| line.trim().parse::().ok()) .filter(|pid| is_live(*pid)) .collect(); + // An empty cgroup with the top process alive is a cgroup the + // launch is not in: the scan below is what finds it. + if !found.is_empty() || !in_session(self.sid, self.sid) { + return found; + } } } let mut members = session_members(self.sid); @@ -297,22 +328,35 @@ impl Group { self.signal(libc::SIGKILL); } - /// Remove the launch's cgroup, once it is empty. - fn remove(&self) { - if let Some(path) = &self.cgroup { - let _ = std::fs::remove_dir(path); + /// Remove the launch's cgroup, once it is empty. A zombie still counts as + /// a member to the kernel until it is reaped, so `EBUSY` is retried for a + /// moment before it is given up on. + async fn remove(&self) { + let Some(path) = &self.cgroup else { return }; + let mut error = None; + for _ in 0..40 { + match std::fs::remove_dir(path) { + Ok(()) => return, + Err(e) if e.kind() == io::ErrorKind::NotFound => return, + Err(e) => error = Some(e), + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + if let Some(error) = error { + tracing::warn!(path = %path.display(), "a launch's cgroup was not removed: {error}"); } } } /// Wait for a launch's group to be gone, ending it if it will not go alone. /// -/// Reached once the top process has been reaped. A launch counts as ended only -/// when nothing of it is left: a window owned by a process still running is a -/// launch still running, whatever became of the process that started it. +/// Reached once the top process has been reaped. When it leaves, whatever it +/// started is ended with it (TERM, then KILL after the grace): a launch is not +/// reported ended while something of it is still running, and what it left +/// behind is not kept alive. async fn group_gone(group: std::sync::Arc) { if group.members().is_empty() { - group.remove(); + group.remove().await; return; } tracing::info!(sid = group.sid, "a launch's top process has gone and its group has not: ending the group"); @@ -325,12 +369,13 @@ async fn group_gone(group: std::sync::Arc) { let give_up = deadline + STOP_GRACE * 2; loop { if group.members().is_empty() { - group.remove(); + group.remove().await; return; } let now = std::time::Instant::now(); if now >= give_up { tracing::warn!(sid = group.sid, "a launch's group would not die; reporting the launch ended anyway"); + group.remove().await; return; } if now >= deadline && !killed { @@ -349,12 +394,17 @@ fn stop_group(group: std::sync::Arc) { std::thread::spawn(move || { let deadline = std::time::Instant::now() + STOP_GRACE; while std::time::Instant::now() < deadline { - if group.members().is_empty() { + // The launch was started again under this id: what is in the + // environment-named fallback is now the new launch's, and a kill + // here would be its death. + if group.is_superseded() || group.members().is_empty() { return; } std::thread::sleep(std::time::Duration::from_millis(100)); } - group.kill(); + if !group.is_superseded() { + group.kill(); + } }); } @@ -368,6 +418,7 @@ impl Process { groups: std::collections::BTreeMap::new(), cgroup_root: CGROUP_ROOT.to_string(), compositor: None, + compositor_exit: None, } } @@ -574,7 +625,8 @@ impl Workload for Process { // The launch's cgroup, made here and joined before exec: what a child // that makes its own session still cannot leave. The path is built // now because the closure below may not allocate. - let cgroup = make_cgroup(&self.cgroup_root, id.as_str()); + let generation = GENERATION.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + let cgroup = make_cgroup(&self.cgroup_root, id.as_str(), generation); let procs_file = cgroup .as_ref() .and_then(|path| std::ffi::CString::new(path.join("cgroup.procs").into_os_string().into_vec()).ok()); @@ -592,8 +644,9 @@ impl Workload for Process { return Err(io::Error::last_os_error()); } // Into the launch's cgroup, as root, before the uid drops. - // A failure is not one: the launch is still found by its - // environment. + // A failure is not one here, where nothing may be said: the + // parent checks the child is in the cgroup after the spawn and + // falls back to the environment when it is not. if let Some(file) = &procs_file { let fd = libc::open(file.as_ptr(), libc::O_WRONLY | libc::O_CLOEXEC); if fd >= 0 { @@ -643,10 +696,27 @@ impl Workload for Process { // The session id is the top pid (`setsid` above), and stays that // number for as long as anything is in it. let sid = self.running.get(id).map(|w| w.pid).expect("just inserted"); + // The join was unchecked in the child. A cgroup the child is not in + // would read as an empty launch, and a stop of it would do nothing, so + // it is dropped for the session-and-environment fallback. + let cgroup = match cgroup { + Some(path) if !joined(&path, sid) => { + tracing::warn!( + %id, + path = %path.display(), + "the launch did not join its cgroup; its members are found by environment" + ); + let _ = std::fs::remove_dir(&path); + None + } + other => other, + }; // Groups with nothing left in them are dropped here, as `running` is. self.groups.retain(|_, group| !group.members().is_empty()); let group = std::sync::Arc::new(Group::new(id, sid, cgroup)); - self.groups.insert(id.clone(), group.clone()); + if let Some(old) = self.groups.insert(id.clone(), group.clone()) { + old.superseded.store(true, std::sync::atomic::Ordering::SeqCst); + } Ok(Box::pin(async move { let exit = exit .await @@ -706,29 +776,38 @@ impl Workload for Process { Failure::new(format!("{} as {uid}:{gid}: {error}", exec.argv[0])) })?; let exit = watched.take_exit().expect("a new watch has its exit"); - // Said when it goes: with it gone no launch has anywhere to draw, and - // nothing else in the box would say why. - if let Ok(handle) = tokio::runtime::Handle::try_current() { - handle.spawn(async move { - if let Ok(exit) = exit.await { - tracing::error!(?exit, "the compositor has exited: no window in this box can draw"); - } - }); - } let pid = watched.pid; let flag = &watched; - crate::services::await_path( + if let Err(failure) = crate::services::await_path( "the compositor", &socket, before, &|| flag.running(), COMPOSITOR_READY, - )?; + ) { + // One that never served is not left running behind a box that + // reports it could not be made. + if watched.running() { + // SAFETY: two integers; the pid is still this child's. + unsafe { libc::kill(-pid, libc::SIGKILL) }; + unsafe { libc::kill(pid, libc::SIGKILL) }; + } + return Err(failure); + } tracing::info!(pid, %socket, "the compositor is serving"); + // Handed to the session, which reports it and ends the box: with it + // gone no launch has anywhere to draw. + self.compositor_exit = Some(Box::pin(async move { + exit.await.map_err(|_| io::Error::other("the compositor's exit was not delivered")) + })); self.compositor = Some(watched); Ok(()) } + fn compositor_exit(&mut self) -> Option { + self.compositor_exit.take() + } + fn stop_launch(&mut self, id: &LaunchId) { if let Some(group) = self.groups.get(id) { // Whether or not the top process is still there: its children may @@ -1465,6 +1544,22 @@ mod tests { }); } + /// A process whose cgroups are made in a directory of the test's own, so + /// no test touches the real `/sys/fs/cgroup`. It is a plain directory with + /// no `cgroup.procs`, so a join into it fails, which is the fallback path. + fn process_in_tmp(waiters: crate::reap::Waiters) -> (Process, std::path::PathBuf) { + static NEXT: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0); + let root = std::env::temp_dir().join(format!( + "nesinit-cg-{}-{}", + std::process::id(), + NEXT.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + )); + std::fs::create_dir_all(&root).unwrap(); + let mut process = Process::new(waiters); + process.cgroup_root = root.to_string_lossy().into_owned(); + (process, root) + } + fn me(script: &str) -> Exec { // SAFETY: no arguments. let (uid, gid) = unsafe { (libc::getuid(), libc::getgid()) }; @@ -1479,7 +1574,7 @@ mod tests { async fn started(script: &str) -> (Process, LaunchId, i32, Exited) { let waiters = crate::reap::Waiters::new(); - let mut process = Process::new(waiters.clone()); + let (mut process, _root) = process_in_tmp(waiters.clone()); static NEXT: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0); // Unique per test: members are found by the id in their environment. let id = LaunchId::new(format!( @@ -1541,11 +1636,51 @@ mod tests { assert!(group.members().is_empty()); } + /// A cgroup the child could not join is dropped, not trusted: it would + /// read as an empty launch and a stop would do nothing. + #[tokio::test] + async fn a_launch_that_cannot_join_its_cgroup_falls_back_and_is_still_stopped() { + let waiters = crate::reap::Waiters::new(); + let (mut process, root) = process_in_tmp(waiters.clone()); + let id = LaunchId::new(format!("join-{}", std::process::id())); + let exited = process.start(&id, &me("sleep 300 & wait")).unwrap(); + reap_for(waiters, process.running[&id].pid); + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + let group = process.groups[&id].clone(); + assert!(group.cgroup.is_none(), "the cgroup was kept"); + assert_eq!(std::fs::read_dir(&root).unwrap().count(), 0, "its directory was left"); + assert!(group.members().len() >= 2, "{:?}", group.members()); + process.stop_launch(&id); + tokio::time::timeout(std::time::Duration::from_secs(10), exited) + .await + .expect("the launch did not end") + .unwrap(); + assert!(group.members().is_empty()); + } + + /// A stop that is still waiting out its grace must not kill a launch that + /// has taken the same id since. + #[tokio::test] + async fn a_pending_stop_does_not_kill_the_launch_that_replaced_it() { + let (mut process, id, _top, _exited) = started("sleep 300 & wait").await; + let old = process.groups[&id].clone(); + process.stop_launch(&id); + let waiters = process.waiters.clone(); + let _again = process.start(&id, &me("sleep 300 & wait")).unwrap(); + reap_for(waiters, process.running[&id].pid); + assert!(old.is_superseded()); + tokio::time::sleep(STOP_GRACE + std::time::Duration::from_secs(1)).await; + let new = process.groups[&id].clone(); + let alive = new.members(); + new.kill(); + assert!(alive.len() >= 2, "the new launch was killed by the old stop: {alive:?}"); + } + /// Two launches are two groups: stopping one leaves the other running. #[tokio::test] async fn stopping_one_launch_leaves_another_running() { let waiters = crate::reap::Waiters::new(); - let mut process = Process::new(waiters.clone()); + let (mut process, _root) = process_in_tmp(waiters.clone()); let (a, b) = (LaunchId::new("iso-a"), LaunchId::new("iso-b")); let exited_a = process.start(&a, &me("sleep 300 & wait")).unwrap(); let _exited_b = process.start(&b, &me("sleep 300 & wait")).unwrap(); @@ -1945,6 +2080,8 @@ pub mod double { /// Every compositor asked for. pub compositors: Vec, pub compositor_failure: Option, + /// When the compositor "exits": the session takes this after it starts. + pub compositor_exit: Option>, pub mount_failure: Option, pub image_failure: Option, /// Like `image_failure`, but only for a call that names an image, so a @@ -1984,6 +2121,7 @@ pub mod double { stops: 0, compositors: Vec::new(), compositor_failure: None, + compositor_exit: None, mount_failure: None, image_failure: None, named_image_failure: None, @@ -2082,6 +2220,13 @@ pub mod double { } } + fn compositor_exit(&mut self) -> Option { + let rx = self.compositor_exit.take()?; + Some(Box::pin(async move { + rx.await.map_err(|_| io::Error::other("the compositor was dropped")) + })) + } + fn stop_launch(&mut self, id: &LaunchId) { self.stops += 1; self.stopped.push(Some(id.clone())); From d5c5a43d0f69f9c65c019bf452a455462643f091 Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Sun, 18 Oct 2026 07:58:03 +0300 Subject: [PATCH 6/7] feat(api): hold GET /machine/jobs open with ?wait= and wake it when work appears A host found new work on its next poll, 8 to 21 seconds on average. With wait (at most 25 s) a call that finds nothing is held until a run is requested, a stop is asked for, a launch is added or stopped, or a run ends and frees a slot, then answers. A held call also looks again every 5 s so a wake in another process costs seconds, not the job. Workers answer at once. Bun's idle timeout is raised so the held call is not cut. --- apps/api/app/routes/session.ts | 36 +++- apps/api/app/server.ts | 3 + apps/api/test/jobs-wait.test.ts | 303 +++++++++++++++++++++++++++++ packages/core/src/session/index.ts | 51 ++++- packages/core/src/session/wake.ts | 95 +++++++++ 5 files changed, 481 insertions(+), 7 deletions(-) create mode 100644 apps/api/test/jobs-wait.test.ts create mode 100644 packages/core/src/session/wake.ts diff --git a/apps/api/app/routes/session.ts b/apps/api/app/routes/session.ts index 293c5e15..45aae730 100644 --- a/apps/api/app/routes/session.ts +++ b/apps/api/app/routes/session.ts @@ -36,6 +36,9 @@ import { * credential is a long-lived secret on hardware in somebody's home, and what * one leaking can reach is decided here. */ +/** The longest a job call is held. Under the server's idle timeout and a tunnel's. */ +const MAX_WAIT_SECONDS = 25; + export namespace SessionApi { /** * One answer for "no such run" and "not your run". @@ -732,7 +735,7 @@ export namespace SessionApi { tags: ['Session'], summary: 'Ask for work', description: - 'Returns the work for the calling host, and only that host — it comes from its own credentials and the scope is the query, so a box cannot see work for another. `session.start` is a run waiting to be started; `session.stop` is a run to end now, because its owner asked, its last launch is over, or the terms it started under ran out, repeated on every poll until the host reports it ended. `launch.start` is another game to start in a box that is already live; `launch.stop` is one launch to end, both repeated until the host reports that launch. Poll at the cadence the heartbeat hands down. An agent skips a kind it does not know.', + 'Returns the work for the calling host, and only that host — it comes from its own credentials and the scope is the query, so a box cannot see work for another. `session.start` is a run waiting to be started; `session.stop` is a run to end now, because its owner asked, its last launch is over, or the terms it started under ran out, repeated on every poll until the host reports it ended. `launch.start` is another game to start in a box that is already live; `launch.stop` is one launch to end, both repeated until the host reports that launch. An agent skips a kind it does not know. With `wait`, a call that finds nothing is held open until work appears or the wait is up, then answers empty, so a host is handed work within a moment of it being asked for; poll again straight away. An API that cannot hold a call answers at once whatever `wait` says, and then the cadence the heartbeat hands down is the one to poll at.', responses: { 200: { content: { 'application/json': { schema: Result(z.array(Session.Job)) } }, @@ -741,8 +744,37 @@ export namespace SessionApi { 403: ErrorResponses[403] } }), + validator( + 'query', + z.object({ + wait: z.coerce + .number() + .int() + .min(0) + .optional() + .catch(undefined) + .meta({ + description: `Seconds to hold the call open when there is no work, at most ${MAX_WAIT_SECONDS}. A larger value is held for the most; absent or 0 answers at once`, + example: 25 + }) + }) + ), async (c) => { - return c.json({ data: await Session.listJobsForMachine(Actor.machineID) }); + const { wait } = c.req.valid('query'); + // A Worker has nowhere to keep a held call that another request can + // reach, so there the answer is the plain one and the host's own + // interval is what it polls at. Only a long-lived process holds. + const holds = typeof (globalThis as { Bun?: unknown }).Bun !== 'undefined'; + const waitMs = holds ? Math.min(wait ?? 0, MAX_WAIT_SECONDS) * 1000 : 0; + return c.json({ + data: + waitMs > 0 + ? await Session.holdJobsForMachine(Actor.machineID, { + waitMs, + signal: c.req.raw.signal + }) + : await Session.listJobsForMachine(Actor.machineID) + }); } ); } diff --git a/apps/api/app/server.ts b/apps/api/app/server.ts index 6f3d697b..380cb76e 100644 --- a/apps/api/app/server.ts +++ b/apps/api/app/server.ts @@ -31,6 +31,9 @@ const hostname = process.env.HOST ?? '127.0.0.1'; Bun.serve({ port, hostname, + // A host's job call is held open for up to 25 seconds with nothing sent, + // and Bun's default of ten would close it. See `GET /machine/jobs?wait=`. + idleTimeout: 60, // A Worker runtime hands the handler a context whose `waitUntil` keeps the // invocation alive past the response. A process does not need convincing to // stay alive, so the equivalent is to let the promise run — with a catch, diff --git a/apps/api/test/jobs-wait.test.ts b/apps/api/test/jobs-wait.test.ts new file mode 100644 index 00000000..40905d78 --- /dev/null +++ b/apps/api/test/jobs-wait.test.ts @@ -0,0 +1,303 @@ +import { afterAll, describe, expect, test } from 'bun:test'; + +import { AccessToken } from '@nestri/core/access-token/index'; +import { Box } from '@nestri/core/box/index'; +import { Fixtures } from '@nestri/core/db/fixtures'; +import { testDb } from '@nestri/core/db/test'; +import { Game } from '@nestri/core/game/index'; +import { Identifier } from '@nestri/core/id'; +import { Machine } from '@nestri/core/machine/index'; +import { Session } from '@nestri/core/session/index'; +import { JobWake } from '@nestri/core/session/wake'; +import { Library } from '@nestri/core/user/library'; +import { LinkedAccount } from '@nestri/core/user/linked-account'; + +import { app } from '../app/index'; +import './setup'; + +const sql = testDb(); + +const HOLDER = 'h'.repeat(32); + +const createdUserIds: string[] = []; +const createdGameIds: string[] = []; + +async function newGame(steamAppId: number): Promise { + const [row] = await Game.upsert({ + id: Identifier.ascending('game'), + steamAppId, + slug: `session-route-${steamAppId}`, + name: `Session Route ${steamAppId}` + }); + if (!row) throw new Error('expected a game row'); + createdGameIds.push(row.id); + return row.id; +} + +/** + * Everything one session needs, plus both sets of credentials that reach it. + * + * The person authenticates with a personal token, which is the one user + * credential a test can mint without an auth service; the host authenticates + * as itself with the secret registration hands back exactly once. + */ +async function scene(label: string, steamAppId: number) { + const owner = await Fixtures.owner(label); + createdUserIds.push(owner.userId); + + const registered = await Machine.register({ + id: Identifier.ascending('machine'), + ownerUserId: owner.userId, + teamId: owner.teamId, + label + }); + + const box = await Box.create({ + id: Identifier.ascending('box'), + userId: owner.userId, + machineId: registered.id, + label, + tier: 'sm' + }); + + const pat = await AccessToken.create({ + id: Identifier.ascending('accessToken'), + ownerUserId: owner.userId, + // Null on purpose: a token scoped to the user alone makes the caller a + // plain user actor, which is the credential a person browsing has. + teamId: null, + name: label + }); + + const gameId = await newGame(steamAppId); + // A run launches as a Steam account that owns the game, so the endpoint + // refuses one outside the caller's library. Every scene here is about + // something else, so the game is stocked. + await Library.upsert({ + id: Identifier.ascending('userLibrary'), + userId: owner.userId, + gameId, + playtime2w: null, + playtimeForever: null, + lastPlayed: null + }); + + return { + owner, + box, + machineId: registered.id, + gameId, + user: { + authorization: `Bearer ${pat.token}`, + 'content-type': 'application/json' + } as Record, + host: { + 'x-nestri-machine-id': registered.id, + 'x-nestri-machine-secret': registered.secret, + 'content-type': 'application/json' + } as Record + }; +} + +async function requestSession(s: Awaited>) { + const res = await app.request('/session', { + method: 'POST', + headers: s.user, + body: JSON.stringify({ + boxId: s.box.id, + gameId: s.gameId, + linkedAccountId: s.owner.linkedAccountId + }) + }); + const body = (await res.json()) as any; + return { res, body }; +} + +afterAll(async () => { + if (createdUserIds.length > 0) { + await sql`delete from "burn_segment" where session_id in ( + select s.id from "session" s + join "box" b on b.id = s.box_id + where b.user_id in ${sql(createdUserIds)} + )`; + await sql`delete from "box" where user_id in ${sql(createdUserIds)}`; + await sql`delete from "user" where id in ${sql(createdUserIds)}`; + } + if (createdGameIds.length > 0) { + await sql`delete from "game" where id in ${sql(createdGameIds)}`; + } +}); + +type Scene = Awaited>; + +const report = (s: Scene, id: string, state: string, extra: object = {}) => + app.request(`/session/${id}/state`, { + method: 'POST', + headers: s.host, + body: JSON.stringify({ state, claimToken: HOLDER, ...extra }) + }); + +const heldJobs = async (s: Scene, query: string) => { + const at = Date.now(); + const res = await app.request(`/machine/jobs${query}`, { headers: s.host }); + return { status: res.status, data: ((await res.json()) as any).data as any[], took: Date.now() - at }; +}; + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +const kinds = (data: any[]) => data.map((j) => j.kind); + +async function liveRun(label: string, appId: number) { + const s = await scene(label, appId); + const { body } = await requestSession(s); + const id = body.data.id as string; + await app.request('/machine/heartbeat', { + method: 'POST', + headers: s.host, + body: JSON.stringify({ capabilities: ['launches'] }) + }); + expect((await report(s, id, 'starting')).status).toBe(200); + expect((await report(s, id, 'live')).status).toBe(200); + const second = await newGame(appId + 1000); + await Library.upsert({ + id: Identifier.ascending('userLibrary'), + userId: s.owner.userId, + gameId: second, + playtime2w: null, + playtimeForever: null, + lastPlayed: null + }); + return { s, id, second }; +} + +describe('GET /machine/jobs?wait=', () => { + test('with no work, it is held for the wait and then answers empty', async () => { + const s = await scene('wait-empty', 8100); + const r = await heldJobs(s, '?wait=1'); + expect(r.status).toBe(200); + expect(r.data).toEqual([]); + expect(r.took).toBeGreaterThanOrEqual(900); + expect(r.took).toBeLessThan(3000); + expect(JobWake.held(s.machineId)).toBe(0); + }); + + test('without wait, or with one that is not a number, it answers at once', async () => { + const s = await scene('wait-absent', 8101); + expect((await heldJobs(s, '')).took).toBeLessThan(500); + const r = await heldJobs(s, '?wait=soon'); + expect(r.status).toBe(200); + expect(r.took).toBeLessThan(500); + expect((await heldJobs(s, '?wait=0')).took).toBeLessThan(500); + }); + + test('with work waiting, it answers at once', async () => { + const s = await scene('wait-ready', 8102); + await requestSession(s); + const r = await heldJobs(s, '?wait=20'); + expect(kinds(r.data)).toEqual(['session.start']); + expect(r.took).toBeLessThan(1000); + }); + + test('a run requested while held is handed over at once', async () => { + const s = await scene('wait-start', 8103); + const held = heldJobs(s, '?wait=20'); + await sleep(300); + expect(JobWake.held(s.machineId)).toBe(1); + const { body } = await requestSession(s); + const r = await held; + expect(r.data.map((j) => j.sessionId)).toEqual([body.data.id]); + expect(r.took).toBeLessThan(2000); + expect(JobWake.held(s.machineId)).toBe(0); + }); + + test('a stop asked for while held is handed over at once', async () => { + const { s, id } = await liveRun('wait-stop', 8104); + const held = heldJobs(s, '?wait=20'); + await sleep(300); + const stop = await app.request(`/session/${id}/stop`, { method: 'POST', headers: s.user }); + expect(stop.status).toBe(200); + const r = await held; + expect(r.data).toEqual([ + { kind: 'session.stop', sessionId: id, boxId: s.box.id, reason: 'user' } + ]); + expect(r.took).toBeLessThan(2000); + }); + + test('a launch added, and a launch stopped, while held are each handed over at once', async () => { + const { s, id, second } = await liveRun('wait-launch', 8105); + const addHeld = heldJobs(s, '?wait=20'); + await sleep(300); + const added = await app.request(`/session/${id}/launch`, { + method: 'POST', + headers: s.user, + body: JSON.stringify({ gameId: second }) + }); + expect(added.status).toBe(201); + const launch = ((await added.json()) as any).data; + const start = await addHeld; + expect(start.data.map((j) => [j.kind, j.launchId])).toEqual([['launch.start', launch.id]]); + expect(start.took).toBeLessThan(2000); + + const running = await app.request(`/session/${id}/launch/${launch.id}/state`, { + method: 'POST', + headers: s.host, + body: JSON.stringify({ state: 'running', claimToken: HOLDER }) + }); + expect(running.status).toBe(200); + + const stopHeld = heldJobs(s, '?wait=20'); + await sleep(300); + const stopped = await app.request(`/session/${id}/launch/${launch.id}/stop`, { + method: 'POST', + headers: s.user + }); + expect(stopped.status).toBe(200); + const stop = await stopHeld; + expect(stop.data.map((j) => [j.kind, j.launchId])).toEqual([['launch.stop', launch.id]]); + expect(stop.took).toBeLessThan(2000); + }); + + test('a run that ends frees a slot, and the next in line is handed it at once', async () => { + const { s, id } = await liveRun('wait-slot', 8106); + await app.request('/machine/heartbeat', { + method: 'POST', + headers: s.host, + body: JSON.stringify({ slots: 1 }) + }); + // The one slot is held by the live run, so a second box's run waits. + const other = await Box.create({ + id: Identifier.ascending('box'), + userId: s.owner.userId, + machineId: s.machineId, + label: 'second', + tier: 'sm' + }); + const queued = await Session.request({ + id: Identifier.ascending('session'), + boxId: other.id, + gameId: s.gameId, + linkedAccountId: s.owner.linkedAccountId + }); + const held = heldJobs(s, '?wait=20'); + await sleep(300); + expect(JobWake.held(s.machineId)).toBe(1); + expect((await report(s, id, 'ended')).status).toBe(200); + const r = await held; + expect(r.data.map((j) => j.sessionId)).toEqual([queued.id]); + expect(r.took).toBeLessThan(2000); + }); + + test('a host that hangs up stops being held', async () => { + const s = await scene('wait-abort', 8107); + const controller = new AbortController(); + const call = Promise.resolve( + app.request('/machine/jobs?wait=20', { headers: s.host, signal: controller.signal }) + ).catch(() => null); + await sleep(300); + expect(JobWake.held(s.machineId)).toBe(1); + controller.abort(); + await call; + await sleep(200); + expect(JobWake.held(s.machineId)).toBe(0); + }); +}); diff --git a/packages/core/src/session/index.ts b/packages/core/src/session/index.ts index bb1b4ffb..c186d3fd 100644 --- a/packages/core/src/session/index.ts +++ b/packages/core/src/session/index.ts @@ -16,6 +16,7 @@ import { Trial } from '../trial/index.js'; import { Launch } from './launch.js'; import { LaunchTable } from './launch.sql.js'; import { SessionState, SessionTable } from './session.sql.js'; +import { JobWake } from './wake.js'; /** * One run of one box, and the unit that gets billed. @@ -196,6 +197,9 @@ export namespace Session { gameId: session.gameId, first: true }); + // A run waiting is work for its host. After the commit, so the + // host's next look finds the row. + await Database.effect(() => JobWake.wakeBox(input.boxId)); return { ...session, launches: [first] }; }); } @@ -595,6 +599,34 @@ export namespace Session { return [...stops, ...launches.stops, ...starts, ...launches.starts]; }); + /** + * {@link listJobsForMachine}, held open when there is nothing. + * + * Answers at once when there is work. Otherwise waits for a wake (see + * {@link JobWake}) for up to `waitMs`, looking again on every wake and every + * {@link JobWake.RECHECK_MS}, and answers empty at the deadline. The wake + * is listened for before each look, so one that lands between the look and + * the wait is not lost. `signal` ends the wait early, when the host has + * hung up. No database connection is held while waiting. + */ + export async function holdJobsForMachine( + machineId: string, + options: { waitMs: number; signal?: AbortSignal } + ): Promise { + const deadline = Date.now() + options.waitMs; + for (;;) { + const listener = JobWake.listen(machineId); + try { + const jobs = await listJobsForMachine(machineId); + const remaining = deadline - Date.now(); + if (jobs.length > 0 || remaining <= 0 || options.signal?.aborted) return jobs; + await listener.until(Math.min(remaining, JobWake.RECHECK_MS), options.signal); + } finally { + listener.cancel(); + } + } + } + /** The runs a host is holding: claimed and not yet over. */ const holding = (machineId: string) => Database.use((tx) => @@ -854,7 +886,10 @@ export namespace Session { .set({ timeStopRequested: sql`coalesce(${SessionTable.timeStopRequested}, now())` }) .where(and(scope, inArray(SessionTable.state, ['starting', 'live']))) .returning(); - if (marked) return serialize(marked); + if (marked) { + await Database.effect(() => JobWake.wakeBox(marked.boxId)); + return serialize(marked); + } const [over] = await tx.select().from(SessionTable).where(scope); return over ? serialize(over) : null; }); @@ -949,8 +984,13 @@ export namespace Session { ) ) .returning() - .then((rows) => { + .then(async (rows) => { const row = rows.at(0); + // A run that is over frees a slot, and a run waiting for + // one is work the host can now be handed. + if (row && (input.to === 'ended' || input.to === 'failed')) { + await Database.effect(() => JobWake.wakeBox(row.boxId)); + } return row ? serialize(row) : null; }); }); @@ -1283,6 +1323,7 @@ export namespace Session { gameId: input.gameId, first: false }); + await Database.effect(() => JobWake.wakeBox(run.boxId)); return { outcome: 'created', launch }; }); } @@ -1336,9 +1377,9 @@ export namespace Session { } // A host that does not handle launches would never end this one. if (!(await Launch.hostTakesLaunches(run.id))) return 'hostTooOld'; - return ( - (await Launch.markStop(row.id)) ?? Launch.serialize((await Launch.row(run.id, row.id))!) - ); + const marked = await Launch.markStop(row.id); + if (marked) await JobWake.wakeBox(run.boxId); + return marked ?? Launch.serialize((await Launch.row(run.id, row.id))!); } ); diff --git a/packages/core/src/session/wake.ts b/packages/core/src/session/wake.ts new file mode 100644 index 00000000..f8052cde --- /dev/null +++ b/packages/core/src/session/wake.ts @@ -0,0 +1,95 @@ +import { eq } from 'drizzle-orm'; + +import { BoxTable } from '../box/box.sql.js'; +import { Database } from '../db/index.js'; + +/** + * Waking a host that is waiting for work. + * + * A host holds `GET /machine/jobs?wait=` open when there is nothing for it. + * Whatever creates or changes work for a machine calls {@link wake} after the + * write has committed, and the held call looks again at once. + * + * In this process only. A second API process would not hear it, which is why a + * held call also looks again every {@link RECHECK_MS} on its own: a wake that + * lands in another process costs a host a few seconds, never the job. The + * `redis` package is a dependency of the API but nothing uses it and no + * deployment configures one, so a shared channel is left until there is a + * second process to share it between. + */ +export namespace JobWake { + /** How often a held call looks again with nobody having woken it. */ + export const RECHECK_MS = 5_000; + + const waiting = new Map void>>(); + + /** Tell every call held for this machine to look again. */ + export function wake(machineId: string) { + const set = waiting.get(machineId); + if (!set) return; + waiting.delete(machineId); + for (const resolve of set) resolve(); + } + + /** + * Wake the machine a box is placed on. Looked up, because the places that + * change a run know its box and not its host. Never throws: a failed + * lookup must not fail the write it follows, and the recheck covers it. + */ + export async function wakeBox(boxId: string) { + try { + const machineId = await Database.use((tx) => + tx + .select({ machineId: BoxTable.machineId }) + .from(BoxTable) + .where(eq(BoxTable.id, boxId)) + .then((rows) => rows[0]?.machineId ?? null) + ); + if (machineId) wake(machineId); + } catch (err) { + console.error('[jobs] could not wake the host for a box:', err); + } + } + + /** + * Start listening for a wake. Taken *before* the caller reads the work, so + * a wake between the read and the wait is not lost. + */ + export function listen(machineId: string) { + let done!: () => void; + const woken = new Promise((resolve) => (done = resolve)); + let set = waiting.get(machineId); + if (!set) waiting.set(machineId, (set = new Set())); + set.add(done); + + return { + /** Resolves on a wake, after `ms`, or when `signal` aborts. */ + async until(ms: number, signal?: AbortSignal) { + let timer: ReturnType | undefined; + let onAbort: (() => void) | undefined; + await Promise.race([ + woken, + new Promise((resolve) => (timer = setTimeout(resolve, ms))), + new Promise((resolve) => { + if (!signal) return; + onAbort = () => resolve(); + if (signal.aborted) resolve(); + else signal.addEventListener('abort', onAbort, { once: true }); + }) + ]); + clearTimeout(timer); + if (signal && onAbort) signal.removeEventListener('abort', onAbort); + }, + cancel() { + const current = waiting.get(machineId); + current?.delete(done); + if (current?.size === 0) waiting.delete(machineId); + } + }; + } + + /** How many calls are held, for tests. */ + export function held(machineId: string) { + return waiting.get(machineId)?.size ?? 0; + } +} From 4fe1b5b343c41791ab3777666215ea6f048a22ea Mon Sep 17 00:00:00 2001 From: Wanjohi Date: Sun, 18 Oct 2026 08:06:39 +0300 Subject: [PATCH 7/7] feat(api): hold GET /machine/jobs while every job is one the host has seen (?seen=) --- apps/api/app/routes/session.ts | 17 +++++++++--- apps/api/test/jobs-wait.test.ts | 42 ++++++++++++++++++++++++++++++ packages/core/src/session/index.ts | 31 ++++++++++++++++------ 3 files changed, 79 insertions(+), 11 deletions(-) diff --git a/apps/api/app/routes/session.ts b/apps/api/app/routes/session.ts index 45aae730..ea0964cd 100644 --- a/apps/api/app/routes/session.ts +++ b/apps/api/app/routes/session.ts @@ -735,7 +735,7 @@ export namespace SessionApi { tags: ['Session'], summary: 'Ask for work', description: - 'Returns the work for the calling host, and only that host — it comes from its own credentials and the scope is the query, so a box cannot see work for another. `session.start` is a run waiting to be started; `session.stop` is a run to end now, because its owner asked, its last launch is over, or the terms it started under ran out, repeated on every poll until the host reports it ended. `launch.start` is another game to start in a box that is already live; `launch.stop` is one launch to end, both repeated until the host reports that launch. An agent skips a kind it does not know. With `wait`, a call that finds nothing is held open until work appears or the wait is up, then answers empty, so a host is handed work within a moment of it being asked for; poll again straight away. An API that cannot hold a call answers at once whatever `wait` says, and then the cadence the heartbeat hands down is the one to poll at.', + 'Returns the work for the calling host, and only that host — it comes from its own credentials and the scope is the query, so a box cannot see work for another. `session.start` is a run waiting to be started; `session.stop` is a run to end now, because its owner asked, its last launch is over, or the terms it started under ran out, repeated on every poll until the host reports it ended. `launch.start` is another game to start in a box that is already live; `launch.stop` is one launch to end, both repeated until the host reports that launch. An agent skips a kind it does not know. With `wait`, a call that finds nothing is held open until work appears or the wait is up, then answers empty, so a host is handed work within a moment of it being asked for; poll again straight away. Add `seen` and a call is held, too, while all the work it would answer is work the host already has, since a stop is repeated until reported. An API that cannot hold a call answers at once whatever `wait` says, and then the cadence the heartbeat hands down is the one to poll at.', responses: { 200: { content: { 'application/json': { schema: Result(z.array(Session.Job)) } }, @@ -756,11 +756,21 @@ export namespace SessionApi { .meta({ description: `Seconds to hold the call open when there is no work, at most ${MAX_WAIT_SECONDS}. A larger value is held for the most; absent or 0 answers at once`, example: 25 + }), + seen: z + .string() + .max(4096) + .optional() + .catch(undefined) + .meta({ + description: + 'Comma-separated keys of the jobs the host already has, from its last answer: `kind/sessionId`, and `kind/sessionId/launchId` for a launch job. With `wait`, work whose key is listed is not new, so a call is held while all of it is; at the end of the wait the answer is the full list. Absent, every job is new', + example: 'session.stop/ses_1,launch.stop/ses_1/lch_1' }) }) ), async (c) => { - const { wait } = c.req.valid('query'); + const { wait, seen } = c.req.valid('query'); // A Worker has nowhere to keep a held call that another request can // reach, so there the answer is the plain one and the host's own // interval is what it polls at. Only a long-lived process holds. @@ -771,7 +781,8 @@ export namespace SessionApi { waitMs > 0 ? await Session.holdJobsForMachine(Actor.machineID, { waitMs, - signal: c.req.raw.signal + signal: c.req.raw.signal, + seen: new Set(seen?.split(',').filter(Boolean)) }) : await Session.listJobsForMachine(Actor.machineID) }); diff --git a/apps/api/test/jobs-wait.test.ts b/apps/api/test/jobs-wait.test.ts index 40905d78..7bc04cb8 100644 --- a/apps/api/test/jobs-wait.test.ts +++ b/apps/api/test/jobs-wait.test.ts @@ -287,6 +287,48 @@ describe('GET /machine/jobs?wait=', () => { expect(r.took).toBeLessThan(2000); }); + test('a stop the host has seen is held, and answered in full at the deadline', async () => { + const { s, id } = await liveRun('wait-seen-hold', 8108); + await app.request(`/session/${id}/stop`, { method: 'POST', headers: s.user }); + const key = `session.stop/${id}`; + const unseen = await heldJobs(s, '?wait=20'); + expect(unseen.took).toBeLessThan(1000); + expect(kinds(unseen.data)).toEqual(['session.stop']); + const r = await heldJobs(s, `?wait=1&seen=${encodeURIComponent(key)}`); + expect(r.took).toBeGreaterThanOrEqual(900); + expect(r.took).toBeLessThan(3000); + expect(kinds(r.data)).toEqual(['session.stop']); + expect(JobWake.held(s.machineId)).toBe(0); + }); + + test('a new job among the seen ones is answered at once, with all of them', async () => { + const { s, id, second } = await liveRun('wait-seen-new', 8109); + const added = await app.request(`/session/${id}/launch`, { + method: 'POST', + headers: s.user, + body: JSON.stringify({ gameId: second }) + }); + const launch = ((await added.json()) as any).data; + const running = await app.request(`/session/${id}/launch/${launch.id}/state`, { + method: 'POST', + headers: s.host, + body: JSON.stringify({ state: 'running', claimToken: HOLDER }) + }); + expect(running.status).toBe(200); + await app.request(`/session/${id}/launch/${launch.id}/stop`, { + method: 'POST', + headers: s.user + }); + const seen = encodeURIComponent(`launch.stop/${id}/${launch.id}`); + const held = heldJobs(s, `?wait=20&seen=${seen}`); + await sleep(300); + expect(JobWake.held(s.machineId)).toBe(1); + await app.request(`/session/${id}/stop`, { method: 'POST', headers: s.user }); + const r = await held; + expect(r.took).toBeLessThan(2000); + expect(kinds(r.data).sort()).toEqual(['launch.stop', 'session.stop']); + }); + test('a host that hangs up stops being held', async () => { const s = await scene('wait-abort', 8107); const controller = new AbortController(); diff --git a/packages/core/src/session/index.ts b/packages/core/src/session/index.ts index c186d3fd..451c0a2d 100644 --- a/packages/core/src/session/index.ts +++ b/packages/core/src/session/index.ts @@ -600,18 +600,32 @@ export namespace Session { }); /** - * {@link listJobsForMachine}, held open when there is nothing. + * The identity of a job, stable across the repeats of it: its kind, its + * run, and for a launch job its launch. A host names the jobs it already + * has in `?seen=` by these. + */ + export function jobKey(job: Job): string { + return job.kind === 'launch.start' || job.kind === 'launch.stop' + ? `${job.kind}/${job.sessionId}/${job.launchId}` + : `${job.kind}/${job.sessionId}`; + } + + /** + * {@link listJobsForMachine}, held open when there is nothing new. * - * Answers at once when there is work. Otherwise waits for a wake (see - * {@link JobWake}) for up to `waitMs`, looking again on every wake and every - * {@link JobWake.RECHECK_MS}, and answers empty at the deadline. The wake - * is listened for before each look, so one that lands between the look and - * the wait is not lost. `signal` ends the wait early, when the host has + * Answers at once when there is work the host has not got. A stop, and a + * launch, are repeated until the host reports them, so work whose + * {@link jobKey} is in `seen` is not new: when every job is in `seen`, or + * there are none, the call waits for a wake (see {@link JobWake}) for up to + * `waitMs`, looking again on every wake and every {@link JobWake.RECHECK_MS}, + * and at the deadline answers with whatever there is, the full list. The + * wake is listened for before each look, so one that lands between the look + * and the wait is not lost. `signal` ends the wait early, when the host has * hung up. No database connection is held while waiting. */ export async function holdJobsForMachine( machineId: string, - options: { waitMs: number; signal?: AbortSignal } + options: { waitMs: number; signal?: AbortSignal; seen?: ReadonlySet } ): Promise { const deadline = Date.now() + options.waitMs; for (;;) { @@ -619,7 +633,8 @@ export namespace Session { try { const jobs = await listJobsForMachine(machineId); const remaining = deadline - Date.now(); - if (jobs.length > 0 || remaining <= 0 || options.signal?.aborted) return jobs; + const fresh = jobs.some((job) => !options.seen?.has(jobKey(job))); + if (fresh || remaining <= 0 || options.signal?.aborted) return jobs; await listener.until(Math.min(remaining, JobWake.RECHECK_MS), options.signal); } finally { listener.cancel();