surma/sausage
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Merge branch 'v3' into v3-t59

Browse Source
This commit is contained in:
surma
2026-11-08 01:08:16 +00:00
parent 994ff9da55 1aaaf91f57
commit ad795d8528
11 changed files with 432 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
host/src/process/mod.rs
+39 -10
View File
@@ -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<Runtime>,
pid: u32,
tx: watch::Sender<Option<i32>>,
creations: Option<sausage_vfs::observe::Creations>,
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 {
host/tests/git_runtime.rs
+241
View File
@@ -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<PathBuf> = 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<Runtime>,
root: Arc<dyn Layer>,
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;
}
userland/brush/build.nu
+1 -1
View File
@@ -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'} }
}
userland/brush/process.rs
+3 -1
View File
@@ -114,7 +114,9 @@ pub(crate) fn spawn(mut request: LaunchRequest) -> std::io::Result<ChildProcess>
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(
userland/git/README.md
+1 -1
View File
@@ -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.
userland/git/src/lib.rs
+4 -1
View File
@@ -209,6 +209,8 @@ enum Command {
oneline: bool,
#[arg(short = 'n')]
limit: Option<usize>,
#[arg(short = '1')]
one: bool,
#[arg(long)]
format: Option<String>,
#[arg(long)]
@@ -958,6 +960,7 @@ pub fn run(cli: Cli) -> Result<i32> {
diff_merges,
oneline,
limit,
one,
format,
pretty,
patch,
@@ -965,7 +968,7 @@ pub fn run(cli: Cli) -> Result<i32> {
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))?;
userland/git/test.nu
+2
View File
@@ -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}
vfs/src/journal/layer.rs
+3
View File
@@ -157,6 +157,9 @@ impl Handle {
}
#[async_trait]
impl File for Handle {
fn unlink_if_named(&self, path: String) -> Option<crate::observe::Unlink> {
self.file.unlink_if_named(path)
}
async fn stat(&self) -> Result<Stat> {
self.file.stat().await
}
vfs/src/lib.rs
+4
View File
@@ -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<observe::Unlink> {
None
}
async fn stat(&self) -> Result<Stat>;
async fn read(&self, offset: u64, length: usize) -> Result<Vec<u8>>;
async fn write(&self, offset: u64, bytes: Vec<u8>) -> Result<usize>;
vfs/src/mounts.rs
+5 -3
View File
@@ -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<dyn Layer>, 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<dyn Layer>, 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<Arc<dyn File>> {
let (m, p) = self.route(path, true)?;
vfs/src/observe.rs
+129 -1
View File
@@ -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<Box<dyn Future<Output = Result<()>> + Send>>;
/// Exclusive creations in one process namespace, retained after descriptor closure.
#[derive(Clone, Default)]
pub struct Creations(Arc<Mutex<Vec<(String, Unlink)>>>);
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<BTreeMap<String, Weak<RwLock<()>>>>;
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<Namespace>,
creations: Arc<Mutex<Option<Creations>>>,
}
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<dyn File>,
layer: Arc<dyn Layer>,
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<Arc<dyn File>> {
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<Unlink> {
// 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<Stat> {
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<dyn Layer> = 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(),