fix(scene): a browser helper exits with the studio, even while CEF starts

quest/m0/README.md

@@ -103,7 +103,6 @@ repository.
 - [Browser sources on the GPU under Mesa](/quest/m0/cef-headless.md) - shared-texture pages paint on Intel and AMD under Wayland, and a page that cannot paint fails loudly
 - [The program reaches a relay on every runner](/quest/m0/relay-test-flake.md) - the relay engine test stops failing the nightly on GitHub's runner
 - [Browser sources make no outside requests](/quest/m0/browser-offline.md) - no component updates, spellcheck downloads, or network time from the helper
-- [A browser helper exits with the studio](/quest/m0/browser-helper-exit.md) - no orphaned CEF helper after a crash, kill, or timed-out test
 - [The Apple signing identity is available](/quest/m0/signing.md) - the Apple credentials packaging and the macOS camera need; Windows ships unsigned
 - [Packaging](/quest/m0/packaging.md) - a signed macOS app, a Windows installer, and a Flatpak for Linux
 - [The persistable portal restore token is pinned](/quest/m0/pin-portal-token.md) - moq-video's restore token, for Wayland sources

quest/m0/browser-source.md

@@ -35,6 +35,12 @@ directory instead. What the other platforms need on top:
   with CEF's helper apps (`cef-rs`'s `bundle-cef-app`) and its own
   NSApplication. An IOSurface crosses processes only as a Mach port
   (`IOSurfaceCreateMachPort`), so the channel needs a Mach port rendezvous.
+- **Exiting with the studio:** the studio holds the channel open until the
+  helper exits and stops it with `Command::Stop`, so the helper reads the
+  channel's end, from before CEF starts, as the studio dying, and kills the
+  process group it leads with CEF's processes. macOS keeps that as is.
+  Windows has no process groups: start the helper in a job object with
+  `JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE`, which the studio's handle keeps alive.
 
 ## Required
 

rs/moq-scene/src/browser/host.rs

@@ -11,7 +11,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, mpsc};
 
 use cef::*;
-use rustix::process::{Resource, Rlimit, getrlimit, setrlimit};
+use rustix::process::{Resource, Rlimit, Signal, getpid, getrlimit, kill_process_group, setrlimit};
 
 use super::outbox::Commands;
 use super::protocol::{Area, Channel, Command, Display, Event, Format, Id, Layer, Open, Plane, Raster, Start, Texture};
@@ -32,7 +32,7 @@ pub(super) fn subprocess() -> ExitCode {
 	ExitCode::from(code.clamp(0, 255) as u8)
 }
 
-/// Run the helper until the studio closes the socket or the page fails.
+/// Run the helper until the studio stops it or the page fails.
 pub(super) fn run() -> ExitCode {
 	match serve() {
 		Ok(()) => ExitCode::SUCCESS,
@@ -87,7 +87,7 @@ impl Paged {
 thread_local! {
 	/// The open pages, on CEF's UI thread, where input is delivered.
 	static PAGES: RefCell<HashMap<Id, (Browser, Arc<Paged>)>> = RefCell::new(HashMap::new());
-	/// Whether the studio is gone, so the helper quits once its last page
+	/// Whether the studio stopped the helper, so it quits once its last page
 	/// closes.
 	static QUITTING: Cell<bool> = const { Cell::new(false) };
 	/// The number of the last copy asked of a renderer.
@@ -125,7 +125,7 @@ fn serve() -> Result<(), String> {
 }
 
 /// Run CEF on the profile `start` names, opening and closing pages as the
-/// studio asks, until it closes the socket and the last page closes.
+/// studio asks, until it stops the helper and the last page closes.
 fn browse(shared: &Arc<Shared>, start: Start, reader: Channel, painted_tx: mpsc::Sender<()>) -> Result<(), String> {
 	// A crashed process is reported only once it is gone, and one dumping
 	// core is not gone yet: systemd-coredump took 20 to 70 seconds over a
@@ -145,6 +145,21 @@ fn browse(shared: &Arc<Shared>, start: Start, reader: Channel, painted_tx: mpsc:
 	)
 	.map_err(|err| format!("cannot turn off core dumps: {err}"))?;
 
+	// Read before CEF starts, which can block for good, as on a display that
+	// never answers, so the studio's end is heard even then.
+	let inbox = Arc::new(Inbox {
+		pending: Commands::default(),
+		posted: AtomicBool::new(true),
+		up: AtomicBool::new(false),
+	});
+	{
+		let (shared, inbox) = (shared.clone(), inbox.clone());
+		std::thread::Builder::new()
+			.name("browser-commands".into())
+			.spawn(move || commands(shared, reader, painted_tx, inbox))
+			.map_err(|err| format!("cannot start the helper's command thread: {err}"))?;
+	}
+
 	let _ = api_hash(sys::CEF_API_VERSION_LAST, 0);
 	let args = args::Args::new();
 	let mut app = HelperApp::new(HelperProcess::new(), start.raster);
@@ -180,62 +195,83 @@ fn browse(shared: &Arc<Shared>, start: Start, reader: Channel, painted_tx: mpsc:
 		});
 	}
 
-	let commands = {
-		let shared = shared.clone();
-		std::thread::Builder::new()
-			.name("browser-commands".into())
-			.spawn(move || commands(shared, reader, painted_tx))
-	};
-	if let Err(err) = commands {
-		shutdown();
-		return Err(format!("cannot start the helper's command thread: {err}"));
-	}
-
+	// Pages asked for meanwhile open now.
+	inbox.up.store(true, Ordering::Release);
+	inbox.post(shared);
 	run_message_loop();
 	PAGES.with_borrow_mut(HashMap::clear);
 	// Saves the profile, such as cookies not yet written.
 	shutdown();
 	Ok(())
 }
 
-/// Read the studio's commands until it closes the socket, then close every
+/// The studio's commands the UI thread has yet to apply.
+///
+/// They wait coalesced, as in the studio, with one task at a time posted to
+/// apply them, so a stalled UI thread holds what the pages open need rather
+/// than a task per command. Input past the backlog drops, as for a hung page.
+struct Inbox {
+	pending: Commands,
+	/// Whether a task applying `pending` is posted, or CEF is not up to run
+	/// one.
+	posted: AtomicBool,
+	/// Whether CEF is up.
+	up: AtomicBool,
+}
+
+impl Inbox {
+	/// Have the UI thread apply what is pending, unless a task already will.
+	fn schedule(self: &Arc<Self>, shared: &Arc<Shared>) {
+		if !self.posted.swap(true, Ordering::AcqRel) {
+			self.post(shared);
+		}
+	}
+
+	/// Post a task applying what is pending.
+	fn post(self: &Arc<Self>, shared: &Arc<Shared>) {
+		let (inbox, shared) = (self.clone(), shared.clone());
+		post(move || {
+			// Cleared first, so anything queued from here on posts again.
+			inbox.posted.store(false, Ordering::Release);
+			apply(&shared, inbox.pending.take());
+		});
+	}
+}
+
+/// Read the studio's commands until it stops the helper, then close every
 /// page, which ends the message loop.
 ///
-/// What the UI thread has yet to apply waits coalesced, as in the studio, with
-/// one task at a time posted to apply it, so a stalled UI thread holds what
-/// the pages open need rather than a task per command. Input past the
-/// backlog drops, as for a hung page.
-fn commands(shared: Arc<Shared>, reader: Channel, painted: mpsc::Sender<()>) {
-	let pending = Commands::default();
-	let posted = Arc::new(AtomicBool::new(false));
-	let schedule = || {
-		if !posted.swap(true, Ordering::AcqRel) {
-			let (shared, pending, posted) = (shared.clone(), pending.clone(), posted.clone());
-			post(move || {
-				// Cleared first, so anything queued from here on posts again.
-				posted.store(false, Ordering::Release);
-				apply(&shared, pending.take());
-			});
-		}
-	};
+/// The studio holds its end of the socket until the helper has exited, so
+/// the socket ends early only once the studio is gone, crashed or killed.
+/// Nobody is left then to show a page to or to wait for the profile, so the
+/// helper exits at once with every process CEF started, wherever CEF is.
+fn commands(shared: Arc<Shared>, reader: Channel, painted: mpsc::Sender<()>, inbox: Arc<Inbox>) {
 	loop {
-		match reader.recv::<Command>() {
-			Ok(Some((Command::Painted, _))) => {
+		match recv(&reader) {
+			Command::Painted => {
 				let _ = painted.send(());
 				continue;
 			}
-			Ok(Some((Command::Open(id, open), _))) => pending.open(id, open),
-			Ok(Some((Command::Close(id), _))) => pending.close(id),
-			Ok(Some((Command::Input(id, input), _))) => _ = pending.input(id, input),
-			Ok(Some((Command::FrameRate(id, rate), _))) => _ = pending.rate(id, rate),
-			Ok(Some((Command::Start(_), _))) | Ok(None) | Err(_) => break,
+			Command::Open(id, open) => inbox.pending.open(id, open),
+			Command::Close(id) => inbox.pending.close(id),
+			Command::Input(id, input) => _ = inbox.pending.input(id, input),
+			Command::FrameRate(id, rate) => _ = inbox.pending.rate(id, rate),
+			Command::Stop => break,
+			Command::Start(_) => {
+				eprintln!("browser source helper: a second Start");
+				exit_now();
+			}
 		}
-		schedule();
+		inbox.schedule(&shared);
+	}
+	// No page opens before CEF is up, so nothing waits to be saved.
+	if !inbox.up.load(Ordering::Acquire) {
+		exit_now();
 	}
-	// Unblocks a paint waiting on the studio.
+	// Unblocks a paint waiting on the studio, which answers no more.
 	drop(painted);
 	post(move || {
-		apply(&shared, pending.take());
+		apply(&shared, inbox.pending.take());
 		QUITTING.set(true);
 		let ids: Vec<Id> = PAGES.with_borrow(|pages| pages.keys().copied().collect());
 		if ids.is_empty() {
@@ -245,6 +281,31 @@ fn commands(shared: Arc<Shared>, reader: Channel, painted: mpsc::Sender<()>) {
 			close(id);
 		}
 	});
+	let command = recv(&reader);
+	eprintln!("browser source helper: {command:?} after Stop");
+	exit_now();
+}
+
+/// The studio's next command, or [`exit_now`] once the socket ends.
+fn recv(reader: &Channel) -> Command {
+	use std::io::ErrorKind::*;
+
+	match reader.recv::<Command>() {
+		Ok(Some((command, _))) => return command,
+		Ok(None) => {}
+		Err(err) if matches!(err.kind(), BrokenPipe | ConnectionReset | UnexpectedEof) => {}
+		Err(err) => eprintln!("browser source helper: the studio's socket: {err}"),
+	}
+	exit_now()
+}
+
+/// End the helper and every process CEF started, at once.
+fn exit_now() -> ! {
+	// The studio starts the helper leading a group of its own, which CEF's
+	// processes join; a group with the helper's id is only ever that one.
+	// The kill ends the helper too, before returning.
+	let _ = kill_process_group(getpid(), Signal::KILL);
+	std::process::abort()
 }
 
 /// Apply the studio's commands. Runs on the UI thread.
@@ -255,7 +316,7 @@ fn apply(shared: &Arc<Shared>, batch: Vec<Command>) {
 			Command::Close(id) => close(id),
 			Command::Input(id, input) => deliver(id, input),
 			Command::FrameRate(id, rate) => retime(id, rate),
-			Command::Start(_) | Command::Painted => unreachable!("never pending"),
+			Command::Start(_) | Command::Painted | Command::Stop => unreachable!("never pending"),
 		}
 	}
 }

rs/moq-scene/src/browser/outbox.rs

@@ -73,6 +73,8 @@ struct Outbox {
 	painted: usize,
 	/// Open pages to close.
 	closes: Vec<Id>,
+	/// Whether the helper is to exit, after everything else.
+	quit: bool,
 	/// Whether the writer stopped, so nothing more is taken.
 	stopped: bool,
 }
@@ -90,9 +92,13 @@ impl Commands {
 		true
 	}
 
-	/// Open page `id`.
+	/// Open page `id`, unless the helper is to exit, which would only close it.
 	pub(crate) fn open(&self, id: Id, open: Open) {
-		self.push(|outbox| outbox.opens.push((id, open)));
+		self.push(|outbox| {
+			if !outbox.quit {
+				outbox.opens.push((id, open));
+			}
+		});
 	}
 
 	/// Close page `id`, which drops what it has waiting, its open too if
@@ -142,6 +148,12 @@ impl Commands {
 		Ok(())
 	}
 
+	/// Have the helper close every page and exit, after what is queued, and
+	/// open no more.
+	pub(crate) fn quit(&self) {
+		self.push(|outbox| outbox.quit = true);
+	}
+
 	/// Stop taking commands, so [`next`](Self::next) returns `None`.
 	pub(crate) fn stop(&self) {
 		let (outbox, ready) = &*self.0;
@@ -175,6 +187,7 @@ impl Commands {
 			&& outbox.owed.is_empty()
 			&& outbox.painted == 0
 			&& outbox.closes.is_empty()
+			&& !outbox.quit
 	}
 
 	/// The commands waiting, without waiting for any.
@@ -185,7 +198,8 @@ impl Commands {
 }
 
 impl Outbox {
-	/// Everything waiting, each page's open ahead of anything else for it.
+	/// Everything waiting, each page's open ahead of anything else for it,
+	/// and the helper's exit last.
 	fn take(&mut self) -> Vec<Command> {
 		let mut batch = Vec::new();
 		for (id, open) in self.opens.drain(..) {
@@ -200,6 +214,10 @@ impl Outbox {
 			self.open.remove(&id);
 			batch.push(Command::Close(id));
 		}
+		if self.quit {
+			batch.push(Command::Stop);
+			self.stopped = true;
+		}
 		batch
 	}
 }
@@ -303,4 +321,14 @@ mod tests {
 		commands.close(1);
 		assert!(matches!(commands.take()[..], [Command::Close(1)]));
 	}
+
+	/// A page opened once the helper is to exit never opens: the helper would
+	/// only close it again.
+	#[test]
+	fn a_page_opened_after_quit_never_opens() {
+		let commands = Commands::default();
+		commands.quit();
+		commands.open(0, open("late"));
+		assert!(matches!(commands.take()[..], [Command::Stop]));
+	}
 }

rs/moq-scene/src/browser/protocol.rs

@@ -25,7 +25,7 @@ pub(crate) const MAX_PLANES: usize = 4;
 /// The longest message either side accepts: the largest page's pixels, with
 /// room for the rest of the message. Bounds what a misbehaving peer can make
 /// the other allocate.
-const MAX_MESSAGE: usize = (super::MAX_SIDE * super::MAX_SIDE * 4) as usize + 1024;
+pub(crate) const MAX_MESSAGE: usize = (super::MAX_SIDE * super::MAX_SIDE * 4) as usize + 1024;
 
 /// A page's number, unique among those one helper opens.
 pub(crate) type Id = u64;
@@ -45,6 +45,10 @@ pub(crate) enum Command {
 	Input(Id, Input),
 	/// The canvas's frame rate changed, so the page paints at this one.
 	FrameRate(Id, u32),
+	/// The last message: close every page, save the profile, and exit. The
+	/// studio holds the socket open until then, so the helper reads its end
+	/// as the studio dying and exits at once.
+	Stop,
 }
 
 /// How a helper runs every page it opens.