perf(run): do file-backend I/O on the main thread (LAB-12103)

Cargo.toml

@@ -32,7 +32,7 @@ unsafe_code = "deny"
 [dependencies]
 blake2 = "0.11"
 cachekit-core = { version = "0.7", default-features = false, features = ["encryption"] }
-cachekit-rs = { version = "0.13.0", default-features = false, features = ["cachekitio", "encryption", "file"] }
+cachekit-rs = { version = "0.13.0", default-features = false, features = ["cachekitio", "encryption", "file", "unsync"] }
 getrandom = "0.4"
 hex = "0.4"
 libc = "0.2"

src/child.rs

@@ -431,10 +431,11 @@ fn forward(pid: Pid, sig: i32) {
 }
 
 /// Block [`HANDLED`] on the calling thread. ck calls this on every thread it
-/// starts (the signal thread and tokio's blocking pool), so their handlers
-/// always run on the main thread: a signal delivered to ck is recorded before
-/// the main thread's next `read` or `waitid` returns. Whether it is
-/// delivered before the child exits depends on the kernel; see
+/// starts (the signal thread, and the blocking-pool threads that run the
+/// saas backend's DNS lookups; the file backend starts none), so their
+/// handlers always run on the main thread: a signal delivered to ck is
+/// recorded before the main thread's next `read` or `waitid` returns.
+/// Whether it is delivered before the child exits depends on the kernel; see
 /// [`Supervisor::received_after`].
 pub(crate) fn block_handled_signals() {
     sigmask(libc::SIG_BLOCK, None);

src/store.rs

@@ -49,9 +49,10 @@ impl Store {
                 data_dir.display()
             ))
         })?;
-        // The file backend runs its I/O on tokio's blocking pool; a
-        // current-thread runtime with no drivers is the cheapest one that
-        // provides it, and keeps the hit path off the network stack.
+        // cachekit's `unsync` feature runs the file backend's I/O inline on
+        // this thread, so a hit starts no thread. A current-thread runtime
+        // with no drivers is the cheapest one that can drive its futures,
+        // and keeps the hit path off the network stack.
         let runtime = runtime(&mut tokio::runtime::Builder::new_current_thread())?;
         Self::new(runtime, Box::new(backend), master_key, None)
     }

tests/bench.rs

@@ -1,8 +1,9 @@
 //! The hit-path budget: `ck run -- true` may add at most 5 ms over `true`,
 //! against a cache directory holding about 10,000 entries.
 //!
-//! The populated directory matters: the file backend scans its whole
-//! directory each time it is opened, and nothing evicts unread entries.
+//! The populated directory is the realistic one, since nothing evicts unread
+//! entries. A hit reads one entry and never lists the directory: only a
+//! process's first write sweeps it for orphaned temp files.
 //! Ignored by default because it needs a release build to mean anything:
 //! `cargo test --release --test bench -- --ignored --nocapture`.
 

tests/run.rs

@@ -506,6 +506,46 @@ fn output_past_20_mib_is_streamed_and_not_cached() {
     }
 }
 
+/// A miss does no backend I/O while the command runs, so the output past the
+/// cap reaches the caller before the command exits, not after.
+#[test]
+fn output_past_20_mib_is_streamed_while_the_command_runs() {
+    use std::io::Read;
+
+    let s = Sandbox::new();
+    let size = 21 * 1024 * 1024;
+    let done = s.home().join("done");
+    let script = s.bin().join("big");
+    fs::write(
+        &script,
+        format!(
+            "#!/bin/sh\nhead -c {size} /dev/zero | tr '\\0' x\nsleep 3\ntouch '{}'\n",
+            done.display()
+        ),
+    )
+    .unwrap();
+    fs::set_permissions(&script, fs::Permissions::from_mode(0o755)).unwrap();
+
+    let mut child = s.ck(&["run", "--ttl", "1h", "--", "big"]).spawn().unwrap();
+    let mut pipe = child.stdout.take().unwrap();
+    let mut got = vec![0u8; size];
+    pipe.read_exact(&mut got).unwrap();
+    assert!(
+        !done.exists(),
+        "the output was held until the command exited"
+    );
+    let mut rest = Vec::new();
+    pipe.read_to_end(&mut rest).unwrap();
+    let out = child.wait_with_output().unwrap();
+
+    assert!(
+        rest.is_empty() && got.iter().all(|&b| b == b'x'),
+        "the output changed"
+    );
+    assert_eq!(code(&out), 0);
+    assert!(stderr(&out).contains("20 MiB"), "{}", stderr(&out));
+}
+
 fn signal(child: &std::process::Child, sig: rustix::process::Signal) {
     let pid = rustix::process::Pid::from_raw(child.id() as i32).unwrap();
     rustix::process::kill_process(pid, sig).unwrap();
@@ -893,8 +933,9 @@ fn helper_threads_block_the_handled_signals() {
             helpers += 1;
         }
     }
-    // The signal thread and tokio's blocking-pool thread.
-    assert!(helpers >= 2, "found {helpers} helper threads");
+    // The signal thread alone: the file backend does its I/O inline, so the
+    // reads before the run start no blocking-pool thread.
+    assert_eq!(helpers, 1, "found {helpers} helper threads");
     child.kill().unwrap();
     let _ = child.wait();
 }

tests/saas.rs

@@ -295,7 +295,7 @@ impl Fake {
     }
 }
 
-#[async_trait]
+#[async_trait(?Send)]
 impl Backend for Fake {
     async fn get(&self, key: &str) -> Result<Option<Vec<u8>>, BackendError> {
         match self.answer("get", key).await {