From 17d67aadcc32647020350ce99f8710fa2cb08f7e Mon Sep 17 00:00:00 2001 From: Surma Date: Fri, 6 Nov 2026 23:53:54 +0000 Subject: [PATCH] Refresh unit requests and preserve shared compilation across runtime shutdown --- host/src/process/cache.rs | 52 +++++++++++++++++++++++++++++++-------- host/tests/init.rs | 43 ++++++++++++++++++++++++++++++++ userland/init/README.md | 4 ++- userland/init/src/main.rs | 2 ++ 4 files changed, 90 insertions(+), 11 deletions(-) diff --git a/host/src/process/cache.rs b/host/src/process/cache.rs index df70a73..a5956ce 100644 --- a/host/src/process/cache.rs +++ b/host/src/process/cache.rs @@ -80,16 +80,10 @@ impl Compiler { .or_insert_with(|| { let (tx, rx) = watch::channel(None); let compiler = self.clone(); - // The job owns completion even if a command or a precompile waiter is cancelled. - tokio::spawn(async move { - let result = - tokio::task::spawn_blocking(move || compiler.compile(&bytes)).await; - tx.send_replace(Some( - result - .map_err(anyhow::Error::from) - .and_then(|r| r) - .map_err(|e| format!("{e:#}")), - )); + // Completion must survive the async runtime that started this shared job. + tokio::task::spawn_blocking(move || { + let result = compiler.compile(&bytes).map_err(|e| format!("{e:#}")); + tx.send_replace(Some(result)); }); rx }) @@ -192,6 +186,44 @@ mod tests { use crate::exec::{Command, Exec, Options}; use std::{future::Future, sync::atomic::Ordering}; + #[test] + fn compilation_survives_the_originating_runtime() { + let temp = tempfile::tempdir().unwrap(); + let compiler = Compiler::new(&temp.path().join("cache")).unwrap(); + let bytes = + wat::parse_str(r#"(module (memory (export "memory") 1) (func (export "_start")))"#) + .unwrap(); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (finish_tx, finish_rx) = std::sync::mpsc::channel(); + *compiler.before_compile.lock().unwrap() = Some(Box::new(move || { + started_tx.send(()).unwrap(); + finish_rx.recv().unwrap(); + })); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + let compiler = compiler.clone(); + let bytes = bytes.clone(); + tokio::spawn(async move { compiler.get(bytes).await }); + started_rx.await.unwrap(); + }); + runtime.shutdown_background(); + finish_tx.send(()).unwrap(); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + tokio::time::timeout(std::time::Duration::from_secs(10), compiler.get(bytes)) + .await + .unwrap() + .unwrap(); + }); + assert_eq!(compiler.compilations.load(Ordering::Relaxed), 1); + } + #[tokio::test(flavor = "multi_thread")] async fn command_joins_precompile_even_after_waiter_cancellation() { let temp = tempfile::tempdir().unwrap(); diff --git a/host/tests/init.rs b/host/tests/init.rs index 844320d..61fe19e 100644 --- a/host/tests/init.rs +++ b/host/tests/init.rs @@ -154,6 +154,49 @@ fn lines(path: &Path) -> Vec { .map(str::to_owned) .collect() } +#[tokio::test(flavor = "multi_thread")] +async fn immediate_start_reads_new_and_edited_units() { + let _serial = SERIAL.lock().await; + let (temp, runtime, root) = world(); + let supervisor = supervise(runtime.clone()); + let path = temp.path().join("world"); + wait(|| path.join("var/log/journal").exists()).await; + let unit = "/run/systemd/transient/immediate.service"; + write( + &root, + unit, + b"[Service]\nType=oneshot\nExecStart=probe oneshot /old.log\n", + ) + .await; + assert_eq!( + run(&runtime, &["/bin/systemctl.wasm", "start", "immediate"]) + .await + .0, + 0 + ); + wait(|| lines(&path.join("old.log")) == ["start"]).await; + write( + &root, + unit, + b"[Service]\nType=oneshot\nExecStart=probe oneshot /new.log\n", + ) + .await; + assert_eq!( + run(&runtime, &["/bin/systemctl.wasm", "start", "immediate"]) + .await + .0, + 0 + ); + wait(|| lines(&path.join("new.log")) == ["start"]).await; + assert_eq!(lines(&path.join("old.log")), ["start"]); + assert_eq!( + run(&runtime, &["/bin/systemctl.wasm", "start", "missing"]) + .await + .0, + 5 + ); + supervisor.shutdown().await; +} #[tokio::test] async fn unrelated_bursts_and_watched_overflow_only_reload_changed_units() { let _serial = SERIAL.lock().await; diff --git a/userland/init/README.md b/userland/init/README.md index 6cabd5e..8790bb7 100644 --- a/userland/init/README.md +++ b/userland/init/README.md @@ -32,7 +32,7 @@ Each request uses a unique identifier. 1. The client watches `/run/systemd/replies/` before it submits the request. 2. The client writes `/run/systemd/requests/.json`. -3. Init reads the request and applies the action. +3. Init reads the request, rescans the unit folders, and applies the action. 4. Init writes `/run/systemd/replies/` and removes the request. 5. The client reads and removes the reply. @@ -42,6 +42,8 @@ Request example: {"action":"start","unit":"example.service"} ``` +Requests observe unit files that the client writes before it submits the request. +A queued watcher event does not delay the definition that the request uses. Actions are `start`, `stop`, and `restart`. A reply acknowledges the action. It does not imply that the service completed its start or stop. The control tools then wait for the requested state. A failed start returns code 1. diff --git a/userland/init/src/main.rs b/userland/init/src/main.rs index 4a63394..92cb442 100644 --- a/userland/init/src/main.rs +++ b/userland/init/src/main.rs @@ -339,6 +339,8 @@ fn controls(services: &mut BTreeMap) { let Ok(request) = serde_json::from_slice::(&bytes) else { continue; }; + // A unit edit can arrive after the event loop's scan but before this request. + scan(services); let mut reply = Reply { code: 0, text: String::new(),