Refresh unit requests and preserve shared compilation across runtime shutdown

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();

host/tests/init.rs

@@ -154,6 +154,49 @@ fn lines(path: &Path) -> Vec<String> {
         .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;

userland/init/README.md

@@ -32,7 +32,7 @@ Each request uses a unique identifier.
 
 1. The client watches `/run/systemd/replies/<id>` before it submits the request.
 2. The client writes `/run/systemd/requests/<id>.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/<id>` 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.

userland/init/src/main.rs

@@ -339,6 +339,8 @@ fn controls(services: &mut BTreeMap<String, Service>) {
         let Ok(request) = serde_json::from_slice::<Request>(&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(),