diff --git a/host/src/process/mod.rs b/host/src/process/mod.rs index 921b08d..3181925 100644 --- a/host/src/process/mod.rs +++ b/host/src/process/mod.rs @@ -430,7 +430,13 @@ impl Runtime { None }; let mut mounts = Mounts::default(); - mounts.mount("/", root, false); + let operations = mounts.mount("/", root, false); + let creations = + matches!(executable.rsplit('/').next(), Some("git" | "git.wasm")).then(|| { + sausage_vfs::Layer::observer(&mounts) + .unwrap() + .exclusive_creations() + }); let mut wasi = WasiCtxBuilder::new(); wasi.args(&request.argv) .envs(&request.env) @@ -490,6 +496,8 @@ impl Runtime { runtime: self.clone(), pid, tx, + creations, + operations, }; // The start gate makes the entry visible before a guest can spawn children or exit. let (gate, ready) = tokio::sync::oneshot::channel(); @@ -559,19 +567,40 @@ struct Cleanup { runtime: Arc, pid: u32, tx: watch::Sender>, + creations: Option, + operations: sausage_vfs::Process, +} +impl Cleanup { + fn reaped(runtime: &Runtime, pid: u32) { + runtime.entries.lock().unwrap().remove(&pid); + runtime.world.processes.lock().unwrap().remove(&pid); + } } impl Drop for Cleanup { fn drop(&mut self) { - if self.tx.borrow().is_none() { - self.tx.send_replace(Some(137)); + if self.tx.borrow().is_none() + && let Some(creations) = self.creations.clone() + { + let (runtime, pid, tx, operations) = ( + self.runtime.clone(), + self.pid, + self.tx.clone(), + self.operations.clone(), + ); + tokio::spawn(async move { + operations.idle().await; + // A cancelled guest cannot run Git's lockfile destructors. Retain + // ownership across descriptor closure, but never remove a replacement. + let result = creations.remove(|path| path.ends_with("/index.lock")).await; + Cleanup::reaped(&runtime, pid); + tx.send_replace(Some(if result.is_ok() { 137 } else { 125 })); + }); + } else { + if self.tx.borrow().is_none() { + self.tx.send_replace(Some(137)); + } + Cleanup::reaped(&self.runtime, self.pid); } - self.runtime.entries.lock().unwrap().remove(&self.pid); - self.runtime - .world - .processes - .lock() - .unwrap() - .remove(&self.pid); } } impl Host for State { diff --git a/host/tests/git_runtime.rs b/host/tests/git_runtime.rs new file mode 100644 index 0000000..f0178dc --- /dev/null +++ b/host/tests/git_runtime.rs @@ -0,0 +1,241 @@ +//! Git stays responsive through the shell, and cancellation removes only its own index lock. +use sausage::{ + exec::{Command, Exec, Options}, + process::Runtime, + world::World, +}; +use sausage_vfs::{Layer, Open}; +use std::{ + path::{Path, PathBuf}, + sync::{Arc, OnceLock}, + time::Duration, +}; + +fn guests() -> &'static Path { + static BUILT: OnceLock = OnceLock::new(); + BUILT.get_or_init(|| { + let repo = Path::new(env!("CARGO_MANIFEST_DIR")).parent().unwrap(); + let target = PathBuf::from(std::env::var("CARGO_TARGET_DIR").unwrap()); + for script in ["userland/cwd-init/build.nu", "userland/brush/build.nu"] { + if script == "userland/cwd-init/build.nu" + && repo.join("userland/cwd-init/init.o").is_file() + { + continue; + } + assert!( + std::process::Command::new("nu") + .args(["--no-config-file", script]) + .current_dir(repo) + .status() + .unwrap() + .success() + ); + } + assert!( + std::process::Command::new("cargo") + .args([ + "build", + "--release", + "--locked", + "--target", + "wasm32-wasip2", + "-p", + "git" + ]) + .current_dir(repo.join("userland")) + .status() + .unwrap() + .success() + ); + target.join("wasm32-wasip2/release") + }) +} +fn native(cwd: &Path, args: &[&str]) { + let output = std::process::Command::new("git") + .args(args) + .current_dir(cwd) + .output() + .unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); +} +struct Fixture { + _temp: tempfile::TempDir, + runtime: Arc, + root: Arc, + processes: sausage::process::Processes, +} +fn fixture() -> Fixture { + let programs = guests(); + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("world"); + for path in ["bin", "src/demo/src", "worktrees"] { + std::fs::create_dir_all(root.join(path)).unwrap(); + } + for (source, name) in [("brush", "sh"), ("git", "git")] { + std::fs::copy( + programs.join(format!("{source}.wasm")), + root.join(format!("bin/{name}.wasm")), + ) + .unwrap(); + } + let repo = root.join("src/demo"); + std::fs::write(repo.join("src/main.rs"), "old\n").unwrap(); + native(&repo, &["init", "-b", "main"]); + native(&repo, &["add", "."]); + native( + &repo, + &[ + "-c", + "user.name=Test", + "-c", + "user.email=test@example.com", + "commit", + "--no-gpg-sign", + "-m", + "initial", + ], + ); + native( + &repo, + &[ + "worktree", + "add", + "-b", + "worker", + root.join("worktrees/160").to_str().unwrap(), + ], + ); + std::fs::write( + root.join("worktrees/160/.git"), + "gitdir: /src/demo/.git/worktrees/160\n", + ) + .unwrap(); + std::fs::write( + repo.join(".git/worktrees/160/gitdir"), + "/worktrees/160/.git\n", + ) + .unwrap(); + let world = World::open(temp.path()).unwrap(); + let root = world.root.clone(); + let processes = world.processes.clone(); + let runtime = Runtime::new(world).unwrap(); + Fixture { + _temp: temp, + runtime, + root, + processes, + } +} +fn options() -> Options { + Options { + cwd: "/worktrees/160".into(), + env: [ + ("GIT_AUTHOR_NAME", "Test"), + ("GIT_AUTHOR_EMAIL", "test@example.com"), + ("GIT_COMMITTER_NAME", "Test"), + ("GIT_COMMITTER_EMAIL", "test@example.com"), + ] + .map(|(k, v)| (k.into(), v.into())) + .into(), + ..Default::default() + } +} +async fn run(fixture: &Fixture, script: &str) -> String { + let mut output = Vec::new(); + let code = fixture + .runtime + .exec(Command::Shell(script.into()), options(), &mut |_, bytes| { + output.extend_from_slice(bytes) + }) + .await + .unwrap(); + assert_eq!(code, 0, "{}", String::from_utf8_lossy(&output)); + String::from_utf8(output).unwrap() +} +async fn edit(fixture: &Fixture) { + let file = fixture + .root + .open( + "/worktrees/160/src/main.rs", + Open { + write: true, + truncate: true, + ..Default::default() + }, + ) + .await + .unwrap(); + file.write(0, b"changed\n".to_vec()).await.unwrap(); +} +#[cfg(not(debug_assertions))] +#[tokio::test(flavor = "current_thread")] +async fn linked_worktree_commands_stay_under_two_seconds() { + let fixture = fixture(); + run(&fixture, "git status --short").await; + edit(&fixture).await; + for script in [ + "git status --short", + "git log -1", + "git add src/main.rs && git commit --no-gpg-sign -m changed", + ] { + let start = std::time::Instant::now(); + let output = run(&fixture, script).await; + let elapsed = start.elapsed(); + println!("{script}: {elapsed:?}\n{output}"); + assert!(elapsed < Duration::from_secs(2), "{script}: {elapsed:?}"); + } +} +#[tokio::test(flavor = "current_thread")] +async fn cancelled_add_releases_its_index_lock() { + let fixture = fixture(); + run(&fixture, "git status --short").await; + edit(&fixture).await; + let path = "/src/demo/.git/worktrees/160/index.lock"; + let mut changes = fixture + .root + .observer() + .unwrap() + .subscribe(vec![path.into()]); + let mut output = Vec::new(); + let mut collect = |_, bytes: &[u8]| output.extend_from_slice(bytes); + let mut command = Box::pin(fixture.runtime.exec( + Command::Argv(vec![ + "/bin/git.wasm".into(), + "add".into(), + "src/main.rs".into(), + ]), + options(), + &mut collect, + )); + tokio::time::timeout(Duration::from_secs(30), async { + tokio::select! { + biased; + change = changes.recv() => { assert_eq!(change.unwrap().0, path); }, + result = &mut command => panic!("add completed before cancellation: {result:?}"), + } + }) + .await + .unwrap(); + assert!( + !fixture.processes.lock().unwrap().is_empty(), + "Git must still run when cancellation occurs" + ); + drop(command); + tokio::time::timeout(Duration::from_secs(30), async { + while !fixture.processes.lock().unwrap().is_empty() { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!(fixture.root.stat(path, false).await.is_err()); + run( + &fixture, + "git add src/main.rs && git commit --no-gpg-sign -m recovered", + ) + .await; +} diff --git a/userland/brush/build.nu b/userland/brush/build.nu index da7908b..f5318d7 100644 --- a/userland/brush/build.nu +++ b/userland/brush/build.nu @@ -39,6 +39,6 @@ def main [] { $env.SAUSAGE_CWD_INIT = ($here | path join .. cwd-init init.o | path expand) $env.CARGO_TARGET_WASM32_WASIP2_LINKER = ($here | path join .. cwd-init link.nu | path expand) $env.CARGO_TARGET_WASM32_WASIP2_RUSTFLAGS = '-C link-arg=sausage-cwd-init' - ^cargo build --manifest-path ($source | path join Cargo.toml) -p brush-shell --release --features minimal --no-default-features --target wasm32-wasip2 --locked + ^cargo build --manifest-path ($source | path join Cargo.toml) -p brush-shell --release --features minimal,tokio/time --no-default-features --target wasm32-wasip2 --locked if $env.LAST_EXIT_CODE != 0 { error make {msg: 'brush build failed'} } } diff --git a/userland/brush/process.rs b/userland/brush/process.rs index 9050413..178742d 100644 --- a/userland/brush/process.rs +++ b/userland/brush/process.rs @@ -114,7 +114,9 @@ pub(crate) fn spawn(mut request: LaunchRequest) -> std::io::Result return Ok(ExecutionResult::new(code as u8)); } } - tokio::task::yield_now().await; + // A guest-only yield keeps this WASM invocation runnable and starves + // the host tasks that supply the child's filesystem and stream IO. + tokio::time::sleep(std::time::Duration::from_millis(1)).await; } }); Ok( diff --git a/userland/git/README.md b/userland/git/README.md index 13e2f39..0512101 100644 --- a/userland/git/README.md +++ b/userland/git/README.md @@ -56,7 +56,7 @@ It checks multiple revisions, detached status, branch deletion messages, and mer It also checks these forms: - `status -sb` and `status --porcelain=v1`. -- `log -n`, `--oneline`, `-p`, `--stat`, and one literal path after `--`. +- `log -1`, `-n`, `--oneline`, `-p`, `--stat`, and one literal path after `--`. - `log --format` and `--pretty=format:` with `%H %h %s %an %ae %ad %d`. - `log A..B`, `--name-only`, `--no-renames`, and `--diff-merges=first-parent`. - `%x00`, `%x1e`, and `%(trailers:key=Sausage-Conversation,valueonly)` in log formats. diff --git a/userland/git/src/lib.rs b/userland/git/src/lib.rs index f4bcf27..6689eec 100644 --- a/userland/git/src/lib.rs +++ b/userland/git/src/lib.rs @@ -209,6 +209,8 @@ enum Command { oneline: bool, #[arg(short = 'n')] limit: Option, + #[arg(short = '1')] + one: bool, #[arg(long)] format: Option, #[arg(long)] @@ -958,6 +960,7 @@ pub fn run(cli: Cli) -> Result { diff_merges, oneline, limit, + one, format, pretty, patch, @@ -965,7 +968,7 @@ pub fn run(cli: Cli) -> Result { revision, paths, } => { - let limit = limit.unwrap_or(usize::MAX); + let limit = if one { 1 } else { limit.unwrap_or(usize::MAX) }; let (tip, excluded) = plumbing::range(&repository, revision.as_deref().unwrap_or("HEAD"))?; let commit = repository.find_commit(tip).map_err(|e| error("log", e))?; diff --git a/userland/git/test.nu b/userland/git/test.nu index 97ffa39..b1ce75d 100644 --- a/userland/git/test.nu +++ b/userland/git/test.nu @@ -56,6 +56,8 @@ def main [component: path, server: path = '/tmp/p1-git/server-target/debug/git-t {args: [branch --show-current], compare: true} {args: [branch -a], compare: true} {args: [log -n 1], compare: true} + {args: [log -1], compare: true} + {args: [log --oneline -1], compare: true} {args: [log -n 1 '--format=%H %h %s %an %ae %ad %d'], compare: true} {args: [log -n 1 '--pretty=format:%H %h %s %an %ae %ad %d'], compare: true} {args: [log -n 1 -p], compare: true} diff --git a/vfs/src/journal/layer.rs b/vfs/src/journal/layer.rs index 7c2669c..bb0af93 100644 --- a/vfs/src/journal/layer.rs +++ b/vfs/src/journal/layer.rs @@ -157,6 +157,9 @@ impl Handle { } #[async_trait] impl File for Handle { + fn unlink_if_named(&self, path: String) -> Option { + self.file.unlink_if_named(path) + } async fn stat(&self) -> Result { self.file.stat().await } diff --git a/vfs/src/lib.rs b/vfs/src/lib.rs index 4ef66b0..9245c6a 100644 --- a/vfs/src/lib.rs +++ b/vfs/src/lib.rs @@ -52,6 +52,10 @@ pub struct Open { /// An open file keeps its identity after rename or unlink. #[async_trait] pub trait File: Send + Sync { + /// Prepare an unlink that does nothing if this file no longer has the given name. + fn unlink_if_named(&self, _path: String) -> Option { + None + } async fn stat(&self) -> Result; async fn read(&self, offset: u64, length: usize) -> Result>; async fn write(&self, offset: u64, bytes: Vec) -> Result; diff --git a/vfs/src/mounts.rs b/vfs/src/mounts.rs index d062cc9..576a683 100644 --- a/vfs/src/mounts.rs +++ b/vfs/src/mounts.rs @@ -19,11 +19,13 @@ impl Default for Mounts { } } impl Mounts { - /// Add or replace a mount. Read-only mounts reject every mutation. - pub fn mount(&mut self, path: &str, layer: Arc, readonly: bool) { + /// Add or replace a mount and return its operation tracker. Read-only mounts reject every mutation. + pub fn mount(&mut self, path: &str, layer: Arc, readonly: bool) -> Process { let path = resolve("/", path); - let layer = self.observer.layer(layer, &path, Process::default()); + let process = Process::default(); + let layer = self.observer.layer(layer, &path, process.clone()); self.mounts.insert(path, Mount { layer, readonly }); + process } pub(crate) async fn metadata_file(&self, path: &str, follow: bool) -> Result> { let (m, p) = self.route(path, true)?; diff --git a/vfs/src/observe.rs b/vfs/src/observe.rs index 4a5008b..89138a7 100644 --- a/vfs/src/observe.rs +++ b/vfs/src/observe.rs @@ -14,6 +14,23 @@ pub enum ChangeKind { Removed, } +/// Deferred removal of one observed file, checked against its current name. +pub type Unlink = std::pin::Pin> + Send>>; +/// Exclusive creations in one process namespace, retained after descriptor closure. +#[derive(Clone, Default)] +pub struct Creations(Arc>>); +impl Creations { + /// Remove selected files that still have their original names. First drain the mount's operations. + pub async fn remove(self, selected: fn(&str) -> bool) -> Result<()> { + let entries = std::mem::take(&mut *self.0.lock().unwrap()); + for (path, unlink) in entries { + if selected(&path) { + unlink.await?; + } + } + Ok(()) + } +} type Namespace = Mutex>>>; enum Held { Read { _guard: OwnedRwLockReadGuard<()> }, @@ -37,6 +54,7 @@ pub struct Observer { // Only open-registration versus rename/remove needs ordering. Lock the affected // subtrees, not all IO: a blocked open must not stall unrelated namespaces. namespace: Arc, + creations: Arc>>, } impl Observer { pub fn new(changed: impl Fn(&str) + Send + Sync + 'static) -> Self { @@ -45,8 +63,15 @@ impl Observer { events: Arc::default(), paths: Arc::default(), namespace: Arc::default(), + creations: Arc::default(), } } + /// Track exclusive creations in this namespace for optional cancellation cleanup. + pub fn exclusive_creations(&self) -> Creations { + let creations = Creations::default(); + *self.creations.lock().unwrap() = Some(creations.clone()); + creations + } /// Subscribe before the first mutation of interest. Lag means callers must rescan. pub fn subscribe( &self, @@ -169,6 +194,8 @@ impl Observed { } struct Handle { file: Arc, + layer: Arc, + prefix: String, path: Path, observer: Observer, process: Process, @@ -203,6 +230,7 @@ impl Layer for Observed { self.layer.list(path).await } async fn open(&self, path: &str, options: Open) -> Result> { + let prefix = self.prefix.clone(); let (layer, local, path, observer, process) = ( self.layer.clone(), path.to_owned(), @@ -217,10 +245,19 @@ impl Layer for Observed { let file = layer.open(&local, options).await?; let handle = Arc::new(Handle { file, + layer, + prefix, path: observer.path(path.clone()), observer: observer.clone(), process, }); + if options.create + && options.exclusive + && let Some(creations) = observer.creations.lock().unwrap().as_ref() + && let Some(unlink) = handle.unlink_if_named(path.clone()) + { + creations.0.lock().unwrap().push((path.clone(), unlink)); + } drop(order); if options.create || options.truncate { observer.notify( @@ -316,6 +353,29 @@ impl Observed { } #[async_trait] impl File for Handle { + fn unlink_if_named(&self, expected: String) -> Option { + // Prefer the innermost observer: its namespace sees renames by every process. + if let Some(unlink) = self.file.unlink_if_named(expected.clone()) { + return Some(unlink); + } + let observer = Observer { + creations: Arc::default(), + ..self.observer.clone() + }; + let (path, layer, prefix) = (self.path.clone(), self.layer.clone(), self.prefix.clone()); + Some(Box::pin(async move { + let order = observer.order(&[(&expected, true)]).await; + if path.lock().unwrap().as_deref() != Some(expected.as_str()) { + return Ok(()); + } + let local = expected.strip_prefix(&prefix).unwrap(); + layer.remove_file(local).await?; + observer.moved(&expected, None); + drop(order); + observer.notify(&expected, ChangeKind::Removed); + Ok(()) + })) + } async fn stat(&self) -> Result { self.file.stat().await } @@ -378,6 +438,72 @@ mod tests { use super::*; use tokio::sync::mpsc; + #[tokio::test] + async fn exclusive_cleanup_survives_close_and_preserves_replacements() { + for renamed in [false, true] { + let mut shared = Mounts::default(); + shared.mount("/", Arc::new(Memory::default()), false); + let shared: Arc = Arc::new(shared); + let mut process = Mounts::default(); + process.mount("/", shared.clone(), false); + let creations = process.observer().unwrap().exclusive_creations(); + let file = process + .open( + "/index.lock", + Open { + create: true, + exclusive: true, + write: true, + ..Default::default() + }, + ) + .await + .unwrap(); + file.write(0, b"old".to_vec()).await.unwrap(); + drop(file); + if renamed { + // Another namespace publishes the old file and creates a new lock. + shared.rename("/index.lock", "/index").await.unwrap(); + let file = shared + .open( + "/index.lock", + Open { + create: true, + exclusive: true, + write: true, + ..Default::default() + }, + ) + .await + .unwrap(); + file.write(0, b"new".to_vec()).await.unwrap(); + } + creations.remove(|_| true).await.unwrap(); + assert_eq!(shared.stat("/index.lock", false).await.is_ok(), renamed); + if renamed { + assert_eq!( + shared + .open("/index.lock", Open::default()) + .await + .unwrap() + .read(0, 3) + .await + .unwrap(), + b"new" + ); + assert_eq!( + shared + .open("/index", Open::default()) + .await + .unwrap() + .read(0, 3) + .await + .unwrap(), + b"old" + ); + } + } + } #[test] fn component_matching() { assert!(under("/mailbox/a", "/mailbox")); @@ -577,7 +703,7 @@ mod tests { "/run", Process::default(), ); - let other = observer.layer(memory, "/run", Process::default()); + let other = observer.layer(memory.clone(), "/run", Process::default()); let opening_task = tokio::spawn(async move { slow.open( "/slow/file", @@ -627,6 +753,8 @@ mod tests { // this write is held in flight. Its notification uses the rebased path. let handle = Handle { file: raw.clone(), + layer: memory.clone(), + prefix: "/run".into(), path: observer.path("/run/moved/file".into()), observer, process: Process::default(),