builder: add native Cargo service and world CLI

This commit is contained in:
2026-11-07 11:59:28 +00:00
parent f74b16dea9
commit c96e59417a
26 changed files with 1278 additions and 36 deletions
Generated
+1
View File
@@ -5035,6 +5035,7 @@ dependencies = [
"hyper",
"hyper-util",
"if-addrs",
"libc",
"openssl",
"p256",
"rsa",
+1 -1
View File
@@ -39,7 +39,7 @@
moldEnv = pkgs.lib.optionalAttrs pkgs.stdenv.hostPlatform.isLinux {
${moldVar} = "-C link-arg=-fuse-ld=mold";
};
builderTools = [ kache.packages.${system}.default pkgs.bubblewrap pkgs.nushell pkgs.wasm-tools ];
builderTools = [ kache.packages.${system}.default pkgs.bubblewrap pkgs.nushell pkgs.wasm-tools pkgs.bash pkgs.coreutils ];
runtimePath = pkgs.lib.makeBinPath ([ rust pkgs.stdenv.cc ] ++ builderTools ++ mold);
# A WASI clang with wasi-libc, so that build scripts (the `cc` crate)
# can compile C for wasm32-wasip2. Its headers are WASI 0.1: C code
+1
View File
@@ -62,6 +62,7 @@ http-body-util = "0.1.5"
hyper-util = { version = "0.1.21", features = ["tokio"] }
hyper = { version = "1.12.0", features = ["client", "http1", "server"] }
tiny_http = "0.12"
libc = "0.2.190"
[dev-dependencies]
tempfile = "3"
+60
View File
@@ -0,0 +1,60 @@
//! A streaming Cargo response. Dropping the body cancels the builder's native process group.
use crate::builder::Event;
use bytes::Bytes;
use http::{Request, Response};
use http_body_util::BodyExt;
use std::{
pin::Pin,
task::{Context, Poll},
};
struct Body(tokio::sync::mpsc::Receiver<Event>);
impl http_body::Body for Body {
type Data = Bytes;
type Error = wasmtime_wasi_http::Error;
fn poll_frame(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<http_body::Frame<Bytes>, Self::Error>>> {
self.0.poll_recv(cx).map(|event| {
event.map(|event| {
let mut bytes = serde_json::to_vec(&event).expect("Cargo event");
bytes.push(b'\n');
Ok(http_body::Frame::data(Bytes::from(bytes)))
})
})
}
}
pub(crate) fn stream(
api: &super::Api,
request: Request<Bytes>,
) -> Response<wasmtime_wasi_http::WasiBody> {
let (tx, rx) = tokio::sync::mpsc::channel(16);
let builder = api.builder.get().cloned();
let request = serde_json::from_slice::<crate::builder::Request>(request.body());
tokio::spawn(async move {
let result = async {
builder
.ok_or_else(|| anyhow::anyhow!("builder unavailable"))?
.execute(request?, &tx)
.await
}
.await;
if let Err(error) = result {
let _ = tx
.send(Event::Stderr {
bytes: format!("cargo: {error:#}\n").into_bytes(),
})
.await;
let _ = tx
.send(Event::Exit {
code: 1,
executables: Vec::new(),
})
.await;
}
});
Response::builder()
.header("content-type", "application/x-ndjson")
.body(Body(rx).boxed_unsync())
.unwrap()
}
+4
View File
@@ -1,6 +1,8 @@
//! JSON routes for host.sausage. The transport supplies the trusted caller.
//! Host-owned pages use passkeys and typed proofs for every authorization ceremony.
mod auth;
mod cargo;
pub(crate) use cargo::stream;
mod pages;
mod server;
mod ssh;
@@ -19,6 +21,7 @@ use std::{
pub struct Api {
pub cred: Arc<Cred>,
pub(crate) builder: std::sync::OnceLock<Arc<crate::builder::Builder>>,
script: cred::quickjs::QuickJs,
pub passkeys: Mutex<Passkeys<rusqlite::Connection>>,
pub entry: EntryKey,
@@ -47,6 +50,7 @@ impl Api {
) -> Self {
Self {
cred: cred.into(),
builder: Default::default(),
script: cred::quickjs::QuickJs::default(),
passkeys: Mutex::new(passkeys),
entry,
+44
View File
@@ -0,0 +1,44 @@
# Rust builder
The world `cargo` sends `POST http://host.sausage/cargo` with `argv`, `cwd`, and optional `timeout_secs`.
The response contains newline-delimited JSON events.
`stdout` and `stderr` events contain byte arrays. The final `exit` event contains a code and executable paths.
The CLI forwards the bytes immediately. For `test` and `run`, it executes the artifacts through the process interface.
The host finds the Cargo workspace above the requested directory and copies its sources through the shared VFS.
It excludes `target` and `.git`. Each workspace owns a source copy, target directory, registry, and kache cache.
These directories live under `<dir>/host/builds/`.
Cargo fetches dependencies with network access. The command then runs offline inside bubblewrap namespaces.
Only the source copy, target, and private kache cache permit writes.
The registry and Nix store remain read-only. HOME and the host network are absent.
The PID namespace removes compiler descendants and the private kache daemon after Cargo exits.
The builder copies changed and new source files back through the VFS.
It publishes WASI executables into the world's workspace target directory.
Every target build links the shared cwd-init object, including its numeric exit-code interface.
The command deadline defaults to 30 minutes. A closed response cancels the native command and its process group.
The host retains at most 20 GiB of build directories and evicts the oldest workspace first.
The Nix wrapper supplies these tools on PATH:
- Rust and the native compiler
- The WASI C tools
- Bubblewrap
- Nushell
- wasm-tools
- Kache
- Bash and Coreutils
Kache comes from the upstream flake at revision `96052b4842239bb216f6c4812054203c21550788`.
The development shell supplies the same builder tools.
The builder resolves the wrapper and interpreter paths to their canonical Nix store paths.
The embedded cwd-init sources produce a startup object and a linker script inside each workspace's tool folder.
The linker script uses the canonical Nushell interpreter, so it needs no host `/usr/bin/env`.
The sandbox exposes Bash at `/bin/sh` for native build scripts.
It does not depend on a user-profile path or a host HOME file.
Development commands retain the global Cargo configuration and its kache wrapper.
The sandbox uses a private Cargo home and selects kache through `--config build.rustc-wrapper`.
For `cargo add`, the network phase performs a dry run to populate the registry index.
The offline command makes the actual manifest change inside the namespaces.
The disk cap applies after each completed command, including compiler failures.
+127
View File
@@ -0,0 +1,127 @@
//! Workspace discovery and source/artifact copies through the world's shared VFS.
use anyhow::{Context, Result};
use sausage_vfs::{Layer, Open};
use std::path::Path;
pub async fn read(root: &dyn Layer, path: &str) -> Result<Vec<u8>> {
let file = root
.open(
path,
Open {
follow: true,
..Default::default()
},
)
.await?;
let size = file.stat().await?.size;
Ok(file.read(0, usize::try_from(size)?).await?)
}
pub async fn workspace(root: &dyn Layer, cwd: &str) -> Result<String> {
let mut path = sausage_vfs::resolve("/", cwd);
let mut nearest = None;
loop {
let manifest = sausage_vfs::resolve(&path, "Cargo.toml");
if let Ok(bytes) = read(root, &manifest).await {
let value: toml::Value = toml::from_str(std::str::from_utf8(&bytes)?)?;
if value.get("workspace").is_some() {
return Ok(path);
}
if nearest.is_none() {
nearest = Some(path.clone());
}
}
if path == "/" {
break;
}
path = sausage_vfs::resolve(&path, "..");
}
nearest.context("no Cargo.toml above cwd")
}
pub async fn copy_out(root: &dyn Layer, source: &str, destination: &Path) -> Result<()> {
std::fs::create_dir_all(destination)?;
for entry in root.list(source).await? {
if matches!(entry.name.as_str(), "target" | ".git") {
continue;
}
let from = sausage_vfs::resolve(source, &entry.name);
let to = destination.join(&entry.name);
if entry.directory {
Box::pin(copy_out(root, &from, &to)).await?;
} else {
std::fs::write(to, read(root, &from).await?)?;
}
}
Ok(())
}
pub async fn mkdir(root: &dyn Layer, path: &str) -> Result<()> {
let mut current = String::new();
for part in path.split('/').filter(|p| !p.is_empty()) {
current.push('/');
current.push_str(part);
if let Err(error) = root.mkdir(&current).await
&& error.kind() != std::io::ErrorKind::AlreadyExists
{
return Err(error.into());
}
}
Ok(())
}
pub async fn write(root: &dyn Layer, path: &str, bytes: Vec<u8>) -> Result<()> {
if read(root, path).await.ok().as_deref() == Some(bytes.as_slice()) {
return Ok(());
}
mkdir(root, path.rsplit_once('/').unwrap().0).await?;
let file = root
.open(
path,
Open {
follow: true,
write: true,
create: true,
truncate: true,
..Default::default()
},
)
.await?;
let mut offset = 0;
while offset < bytes.len() {
let written = file.write(offset as u64, bytes[offset..].to_vec()).await?;
anyhow::ensure!(written != 0, "source write made no progress");
offset += written;
}
Ok(())
}
pub async fn copy_back(
root: &dyn Layer,
source: &Path,
destination: &str,
artifacts: bool,
) -> Result<()> {
for entry in std::fs::read_dir(source)? {
let entry = entry?;
let name = entry.file_name().to_string_lossy().into_owned();
if matches!(name.as_str(), "target" | ".git") {
continue;
}
let to = sausage_vfs::resolve(destination, &name);
if entry.file_type()?.is_dir() {
Box::pin(copy_back(root, &entry.path(), &to, artifacts)).await?;
} else if !artifacts || entry.path().extension().is_some_and(|e| e == "wasm") {
write(root, &to, std::fs::read(entry.path())?).await?;
}
}
Ok(())
}
pub fn size(path: &Path) -> Result<u64> {
let mut total = 0;
for entry in std::fs::read_dir(path)? {
let entry = entry?;
total += if entry.file_type()?.is_dir() {
size(&entry.path())?
} else {
entry.metadata()?.len()
};
}
Ok(total)
}
+410
View File
@@ -0,0 +1,410 @@
//! Native Cargo execution, isolated workspace copies, streamed output, and bounded disk use.
mod files;
#[cfg(test)]
mod tests;
use anyhow::{Context, Result, ensure};
use sausage_vfs::Layer;
use serde::{Deserialize, Serialize};
use std::{
path::{Path, PathBuf},
process::Stdio,
sync::Arc,
time::Duration,
};
use tokio::{
io::AsyncReadExt,
process::Command,
sync::{Mutex, mpsc},
time::Instant,
};
#[derive(Deserialize)]
pub struct Request {
pub argv: Vec<String>,
pub cwd: String,
pub timeout_secs: Option<u64>,
}
#[derive(Serialize)]
#[serde(tag = "event", rename_all = "snake_case")]
pub enum Event {
Stdout { bytes: Vec<u8> },
Stderr { bytes: Vec<u8> },
Exit { code: i32, executables: Vec<String> },
}
pub struct Builder {
root: Arc<dyn Layer>,
directory: PathBuf,
lock: Mutex<()>,
}
impl Builder {
pub fn new(root: Arc<dyn Layer>, host: &Path) -> Result<Self> {
Ok(Self {
root,
directory: std::fs::canonicalize(host)?.join("builds"),
lock: Mutex::new(()),
})
}
pub async fn execute(&self, request: Request, tx: &mpsc::Sender<Event>) -> Result<()> {
let _guard = self.lock.lock().await;
let action = request
.argv
.first()
.context("missing Cargo command")?
.as_str();
ensure!(
[
"build", "check", "clippy", "test", "run", "fmt", "fetch", "add", "remove",
"update", "tree", "metadata", "clean"
]
.contains(&action),
"unsupported Cargo command"
);
let seconds = request.timeout_secs.unwrap_or(1800);
ensure!(seconds > 0, "timeout must be positive");
let deadline = Instant::now()
.checked_add(Duration::from_secs(seconds))
.context("timeout too large")?;
let world_cwd = sausage_vfs::resolve("/", &request.cwd);
let workspace = files::workspace(self.root.as_ref(), &world_cwd).await?;
use std::hash::{Hash, Hasher};
let mut hash = std::collections::hash_map::DefaultHasher::new();
workspace.hash(&mut hash);
let id = format!("{:016x}", hash.finish());
let base = self.directory.join(id);
let source = base.join("source");
// Cargo clean must remove its directory, not a namespace mount point.
let target_mount = base.join("target");
let target = target_mount.join("cargo");
let registry = base.join("cargo");
let cache = base.join("kache");
if source.exists() {
std::fs::remove_dir_all(&source)?;
}
files::copy_out(self.root.as_ref(), &workspace, &source).await?;
for path in [&target, &registry, &cache, &target.join("tmp")] {
std::fs::create_dir_all(path)?;
}
std::fs::write(
target.join("CACHEDIR.TAG"),
b"Signature: 8a477f597d28d172789f06886806bc55\n",
)?;
std::fs::write(base.join("last-used"), [])?;
let cwd = source.join(
world_cwd
.strip_prefix(&workspace)
.expect("workspace is an ancestor of cwd")
.trim_start_matches('/'),
);
let mut fetch = Command::new("cargo");
if action == "add" {
fetch.args(&request.argv).arg("--dry-run");
} else {
fetch.arg("fetch");
}
fetch.current_dir(&cwd).env("CARGO_HOME", &registry);
let mut code = if matches!(action, "clean" | "fmt" | "remove") {
0
} else {
run(fetch, tx, deadline).await?.0
};
let mut executables = Vec::new();
if code == 0 {
let tools = base.join("tools");
std::fs::create_dir_all(&tools)?;
for (name, bytes) in [
(
"init.rs",
include_bytes!("../../../userland/cwd-init/init.rs").as_slice(),
),
(
"exit.wit",
include_bytes!("../../../userland/cwd-init/exit.wit").as_slice(),
),
(
"build.nu",
include_bytes!("../../../userland/cwd-init/build.nu").as_slice(),
),
(
"link.nu",
include_bytes!("../../../userland/cwd-init/link.nu").as_slice(),
),
] {
std::fs::write(tools.join(name), bytes)?;
}
let linker = std::fs::read_to_string(tools.join("link.nu"))?.replacen(
"#!/usr/bin/env nu",
&format!("#!{}", executable("nu")?.display()),
1,
);
std::fs::write(tools.join("link.nu"), linker)?;
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(
tools.join("link.nu"),
std::fs::Permissions::from_mode(0o755),
)?;
let mut startup = Command::new("nu");
startup.arg(tools.join("build.nu"));
let (startup_code, _) = run(startup, tx, deadline).await?;
if startup_code != 0 {
files::copy_back(self.root.as_ref(), &source, &workspace, false).await?;
self.evict(20 * 1024 * 1024 * 1024)?;
tx.send(Event::Exit {
code: startup_code,
executables,
})
.await?;
return Ok(());
}
let kache = executable("kache")?;
let mut cmd = Command::new("bwrap");
cmd.args([
"--unshare-user",
"--unshare-pid",
"--unshare-net",
"--die-with-parent",
"--ro-bind",
"/nix/store",
"/nix/store",
])
.arg("--bind")
.arg(target.join("tmp"))
.arg("/tmp")
.arg("--ro-bind")
.arg(executable("sh")?)
.arg("/bin/sh");
for path in [&source, &target_mount, &cache] {
cmd.arg("--bind").arg(path).arg(path);
}
for path in [&registry, &tools] {
cmd.arg("--ro-bind").arg(path).arg(path);
}
cmd.args([
"--proc",
"/proc",
"--dev",
"/dev",
"--remount-ro",
"/",
"--chdir",
])
.arg(&cwd)
.args(["--clearenv", "--setenv", "PATH"])
.arg(std::env::var("PATH")?)
.args(["--setenv", "HOME", "/nonexistent", "--setenv", "CARGO_HOME"])
.arg(&registry)
.args(["--setenv", "CARGO_TARGET_DIR"])
.arg(&target)
.args(["--setenv", "KACHE_CACHE_DIR"])
.arg(&cache)
.args(["--setenv", "SAUSAGE_CWD_INIT"])
.arg(tools.join("init.o"))
.args(["--setenv", "CARGO_BUILD_JOBS", "2"]);
for name in [
"CC_wasm32_wasip2",
"AR_wasm32_wasip2",
"CFLAGS_wasm32_wasip2",
] {
if let Ok(value) = std::env::var(name) {
cmd.args(["--setenv", name]).arg(value);
}
}
cmd.args(["cargo", "--offline", "--config"])
.arg(format!("build.rustc-wrapper={:?}", kache))
.arg("--config")
.arg(format!(
"target.wasm32-wasip2.linker={:?}",
tools.join("link.nu")
))
.args([
"--config",
"target.wasm32-wasip2.rustflags=['-C','link-arg=sausage-cwd-init']",
]);
let mut args = request.argv.clone();
if action == "run" {
args[0] = "build".into();
}
if let Some(index) = args.iter().position(|a| a == "--") {
args.truncate(index);
}
if action == "test" {
// Cargo's positional test filter belongs to the world runner, not the compiler.
args = test_build_args(&args);
if !args.iter().any(|a| a == "--no-run") {
args.push("--no-run".into());
}
}
cmd.args(args);
if ["build", "check", "clippy", "test", "run"].contains(&action) {
cmd.args([
"--target",
"wasm32-wasip2",
"--message-format=json-render-diagnostics",
]);
}
let (status, stdout) = run(cmd, tx, deadline).await?;
code = status;
let artifacts = target.join("wasm32-wasip2");
if artifacts.exists() {
files::copy_back(
self.root.as_ref(),
&artifacts,
&format!("{workspace}/target/wasm32-wasip2"),
true,
)
.await?;
}
if code == 0
&& matches!(action, "test" | "run")
&& !request.argv.iter().any(|a| a == "--no-run")
{
for line in stdout.split(|b| *b == b'\n') {
if let Ok(value) = serde_json::from_slice::<serde_json::Value>(line)
&& value["reason"] == "compiler-artifact"
&& (action == "run" || value["profile"]["test"] == true)
&& let Some(path) = value["executable"].as_str()
&& let Ok(relative) = Path::new(path).strip_prefix(&target)
{
executables.push(format!("{workspace}/target/{}", relative.display()));
}
}
executables.sort();
executables.dedup();
}
}
files::copy_back(self.root.as_ref(), &source, &workspace, false).await?;
self.evict(20 * 1024 * 1024 * 1024)?;
tx.send(Event::Exit { code, executables }).await?;
Ok(())
}
fn evict(&self, limit: u64) -> Result<()> {
let mut entries = Vec::new();
let mut total = 0;
for entry in std::fs::read_dir(&self.directory)? {
let path = entry?.path();
let size = files::size(&path)?;
total += size;
entries.push((
std::fs::metadata(path.join("last-used"))?.modified()?,
path,
size,
));
}
entries.sort_by_key(|e| e.0);
for (_, path, size) in entries {
if total <= limit {
break;
}
std::fs::remove_dir_all(path)?;
total -= size;
}
Ok(())
}
}
fn executable(name: &str) -> Result<PathBuf> {
let path = std::env::split_paths(&std::env::var_os("PATH").context("PATH missing")?)
.map(|p| p.join(name))
.find(|p| p.is_file())
.context(format!("missing {name}"))?;
Ok(std::fs::canonicalize(path)?)
}
fn test_build_args(args: &[String]) -> Vec<String> {
let mut out = vec!["test".into()];
let mut value = false;
for arg in &args[1..] {
if value || arg.starts_with('-') {
out.push(arg.clone());
}
value = [
"-p",
"--package",
"--test",
"--bin",
"--example",
"--features",
"--profile",
"--manifest-path",
"--exclude",
"-j",
"--jobs",
"-F",
"--config",
"--target",
"--target-dir",
"--message-format",
"--color",
"--lockfile-path",
"-Z",
]
.contains(&arg.as_str());
}
out
}
// The guard kills native descendants when Cargo exits or the caller cancels the command.
struct ProcessGroup(u32);
impl Drop for ProcessGroup {
fn drop(&mut self) {
unsafe {
libc::kill(-(self.0 as i32), libc::SIGKILL);
}
}
}
async fn pump(
mut reader: impl tokio::io::AsyncRead + Unpin,
tx: &mpsc::Sender<Event>,
stderr: bool,
) -> Result<Vec<u8>> {
let mut all = Vec::new();
let mut bytes = vec![0; 8192];
loop {
let n = reader.read(&mut bytes).await?;
if n == 0 {
return Ok(all);
}
if !stderr {
all.extend_from_slice(&bytes[..n]);
}
tx.send(if stderr {
Event::Stderr {
bytes: bytes[..n].to_vec(),
}
} else {
Event::Stdout {
bytes: bytes[..n].to_vec(),
}
})
.await?;
}
}
async fn run(
mut cmd: Command,
tx: &mpsc::Sender<Event>,
deadline: Instant,
) -> Result<(i32, Vec<u8>)> {
cmd.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.process_group(0)
.kill_on_drop(true);
let mut child = cmd.spawn()?;
let group = ProcessGroup(child.id().unwrap());
let out = child.stdout.take().unwrap();
let err = child.stderr.take().unwrap();
let collect = async {
let (status, out, _) = tokio::try_join!(
async { Ok::<_, anyhow::Error>(child.wait().await?) },
pump(out, tx, false),
pump(err, tx, true)
)?;
Ok::<_, anyhow::Error>((status.code().unwrap_or(1), out))
};
let result = tokio::select! {
result = collect => result,
_ = tx.closed() => anyhow::bail!("caller exited"),
_ = tokio::time::sleep_until(deadline) => {
tx.send(Event::Stderr { bytes: b"cargo: command timed out\n".to_vec() }).await?;
Ok((124, Vec::new()))
}
};
drop(group);
result
}
+85
View File
@@ -0,0 +1,85 @@
//! Focused tests for source discovery, argument forwarding, eviction, and process cancellation.
use super::*;
#[tokio::test]
async fn discovers_workspace_and_skips_target() -> Result<()> {
let temp = tempfile::tempdir()?;
let world = crate::world::World::open(temp.path())?;
files::write(
world.root.as_ref(),
"/project/Cargo.toml",
b"[workspace]\nmembers=['member']".to_vec(),
)
.await?;
files::write(
world.root.as_ref(),
"/project/member/Cargo.toml",
b"[package]\nname='member'\nversion='0.1.0'".to_vec(),
)
.await?;
files::write(
world.root.as_ref(),
"/project/target/junk",
b"junk".to_vec(),
)
.await?;
assert_eq!(
files::workspace(world.root.as_ref(), "/project/member/src").await?,
"/project"
);
let copy = temp.path().join("copy");
files::copy_out(world.root.as_ref(), "/project", &copy).await?;
assert!(copy.join("member/Cargo.toml").exists());
assert!(!copy.join("target").exists());
Ok(())
}
#[test]
fn relative_host_directory_uses_absolute_namespace_paths() -> Result<()> {
let temp = tempfile::tempdir()?;
let world = crate::world::World::open(temp.path())?;
let builder = Builder::new(world.root, Path::new("."))?;
assert!(builder.directory.is_absolute());
Ok(())
}
#[test]
fn test_filter_is_not_a_build_argument() {
let args = ["test", "-p", "example", "filter", "--release"].map(str::to_string);
assert_eq!(
test_build_args(&args),
["test", "-p", "example", "--release"]
);
}
#[test]
fn disk_cap_removes_workspaces() -> Result<()> {
let temp = tempfile::tempdir()?;
let world = crate::world::World::open(temp.path())?;
let builder = Builder::new(world.root, &world.host_dir)?;
for name in ["a", "b"] {
let path = builder.directory.join(name);
std::fs::create_dir_all(&path)?;
std::fs::write(path.join("last-used"), [])?;
std::fs::write(path.join("data"), [0; 32])?;
}
std::fs::File::open(builder.directory.join("a/last-used"))?
.set_times(std::fs::FileTimes::new().set_modified(std::time::UNIX_EPOCH))?;
builder.evict(32)?;
assert!(!builder.directory.join("a").exists());
assert!(builder.directory.join("b").exists());
assert_eq!(files::size(&builder.directory)?, 32);
Ok(())
}
#[tokio::test]
async fn closed_receiver_cancels_command() -> Result<()> {
let (tx, rx) = mpsc::channel(1);
drop(rx);
let mut cmd = Command::new("sh");
cmd.args(["-c", "exec sleep 1000"]);
assert!(
tokio::time::timeout(
Duration::from_secs(2),
run(cmd, &tx, Instant::now() + Duration::from_secs(60))
)
.await?
.is_err()
);
Ok(())
}
+1
View File
@@ -2,6 +2,7 @@
//! and the durable agent harness. See `~/research/sausage-v3/PLAN.md` until the docs move here.
pub mod api;
pub mod builder;
pub mod cred;
pub mod exec;
pub mod harness;
+8 -3
View File
@@ -45,9 +45,14 @@ impl WasiHttpHooks for Hooks {
.is_some_and(|h| h.eq_ignore_ascii_case("host.sausage"))
{
let bytes = body.collect().await?.to_bytes();
return Ok(response(
crate::api::handle(&router, caller, Request::from_parts(parts, bytes)).await,
));
let request = Request::from_parts(parts, bytes);
if request.method() == http::Method::POST && request.uri().path() == "/cargo" {
return Ok((
crate::api::stream(&router, request),
Box::new(async { Ok(()) }),
));
}
return Ok(response(crate::api::handle(&router, caller, request).await));
}
if !matches!(parts.uri.scheme_str(), Some("http" | "https")) {
while let Some(frame) = body.frame().await {
+4
View File
@@ -191,6 +191,10 @@ impl Runtime {
"http://localhost:8081",
)?),
};
let _ = router.builder.set(Arc::new(crate::builder::Builder::new(
world.root.clone(),
&world.host_dir,
)?));
let network = Arc::new(net::NetPolicy::open(&world.host_dir)?);
let engine = compiler.engine.clone();
let alive = Arc::new(AtomicBool::new(true));
+9
View File
@@ -0,0 +1,9 @@
[package]
name = "builder-fixture"
version = "0.1.0"
edition = "2024"
[workspace]
[dependencies]
serde = "1"
+39
View File
@@ -0,0 +1,39 @@
//! Assert the native build script's namespace boundaries.
fn main() {
assert!(!std::path::Path::new("/home/surma/.ssh").exists());
assert!(
!std::path::Path::new(&std::env::var("HOME").unwrap())
.join(".ssh")
.exists()
);
assert!(std::fs::write("/outside-probe", b"no").is_err());
assert!(
std::net::TcpStream::connect_timeout(
&"1.1.1.1:443".parse().unwrap(),
std::time::Duration::from_millis(100)
)
.is_err()
);
let out = std::path::PathBuf::from(std::env::var_os("OUT_DIR").unwrap());
let object = out.join("add.o");
assert!(
std::process::Command::new(std::env::var("CC_wasm32_wasip2").unwrap())
.args(["-c", "src/add.c", "-o"])
.arg(&object)
.status()
.unwrap()
.success()
);
assert!(
std::process::Command::new(std::env::var("AR_wasm32_wasip2").unwrap())
.arg("crs")
.arg(out.join("libadd.a"))
.arg(&object)
.status()
.unwrap()
.success()
);
println!("cargo:rustc-link-search=native={}", out.display());
println!("cargo:rustc-link-lib=static=add");
println!("cargo:warning=namespace assertions passed");
}
+1
View File
@@ -0,0 +1 @@
int c_add(int a, int b) { return a + b; }
+22
View File
@@ -0,0 +1,22 @@
//! An executable and tests that run only inside the world.
unsafe extern "C" {
fn c_add(a: i32, b: i32) -> i32;
}
fn main() {
if std::env::args().any(|arg| arg == "--exit") {
std::process::exit(23);
}
assert_eq!(unsafe { c_add(20, 22) }, 42);
println!(
"builder fixture cwd: {}",
std::env::current_dir().unwrap().display()
);
}
#[test]
fn pass() {
assert_eq!(std::env::current_dir().unwrap().to_str(), Some("/project"));
}
#[test]
fn fail() {
panic!("intentional fixture failure");
}
@@ -0,0 +1,5 @@
//! Cargo executes each integration test component through the world process interface.
#[test]
fn pass_integration() {
println!("INTEGRATION_CASE");
}
+262
View File
@@ -0,0 +1,262 @@
//! Real-world Cargo acceptance: native namespaces, source changes, cache, and WASI execution.
use sausage::{
exec::{Command, Exec, Options},
process::Runtime,
world::World,
};
use std::{
path::{Path, PathBuf},
sync::{Arc, OnceLock},
time::Duration,
};
fn programs() -> PathBuf {
static PATH: OnceLock<PathBuf> = OnceLock::new();
PATH.get_or_init(|| {
let root = Path::new(env!("CARGO_MANIFEST_DIR")).parent().unwrap();
assert!(
std::process::Command::new("nu")
.arg(root.join("userland/cwd-init/build.nu"))
.status()
.unwrap()
.success()
);
assert!(
std::process::Command::new("cargo")
.args(["build", "--release", "-p", "world-cargo"])
.current_dir(root.join("userland"))
.status()
.unwrap()
.success()
);
assert!(
std::process::Command::new("nu")
.arg(root.join("userland/brush/build.nu"))
.status()
.unwrap()
.success()
);
PathBuf::from(std::env::var_os("CARGO_TARGET_DIR").expect("executor target required"))
.join("wasm32-wasip2/release")
})
.clone()
}
async fn cargo(runtime: &Arc<Runtime>, args: &[&str]) -> anyhow::Result<(i32, String)> {
let mut bytes = Vec::new();
let code = runtime
.exec(
Command::Argv(
std::iter::once("/bin/cargo.wasm".into())
.chain(args.iter().map(|a| a.to_string()))
.collect(),
),
Options {
cwd: "/project".into(),
timeout: Some(Duration::from_secs(120)),
..Default::default()
},
&mut |_, b| bytes.extend_from_slice(b),
)
.await?;
Ok((code, String::from_utf8_lossy(&bytes).into_owned()))
}
fn cache_hits(base: &Path) -> anyhow::Result<u64> {
let stats = std::process::Command::new("kache")
.arg("stats")
.env("KACHE_CACHE_DIR", base.join("kache"))
.output()?;
assert!(stats.status.success());
let text = String::from_utf8_lossy(&stats.stdout);
println!("PRIVATE CACHE: {text}");
let hits = text
.lines()
.find(|line| line.contains("hits"))
.and_then(|line| {
line.split_whitespace()
.find_map(|word| word.replace(',', "").parse::<u64>().ok())
})
.unwrap_or(0);
assert!(
std::process::Command::new("kache")
.args(["daemon", "stop"])
.env("KACHE_CACHE_DIR", base.join("kache"))
.status()?
.success()
);
Ok(hits)
}
#[tokio::test(flavor = "multi_thread")]
async fn cargo_build_test_and_source_changes() -> anyhow::Result<()> {
let binaries = programs();
let temp = tempfile::tempdir()?;
let world = World::open(temp.path())?;
let bin = temp.path().join("world/bin");
std::fs::create_dir_all(&bin)?;
std::fs::copy(binaries.join("cargo.wasm"), bin.join("cargo.wasm"))?;
std::fs::copy(binaries.join("brush.wasm"), bin.join("sh.wasm"))?;
let source = temp.path().join("world/project");
std::fs::create_dir_all(source.join("src"))?;
std::fs::create_dir_all(source.join("tests"))?;
for (path, bytes) in [
(
"Cargo.toml",
include_bytes!("builder-fixture/Cargo.toml").as_slice(),
),
(
"build.rs",
include_bytes!("builder-fixture/build.rs").as_slice(),
),
(
"src/main.rs",
include_bytes!("builder-fixture/src/main.rs").as_slice(),
),
(
"tests/integration.rs",
include_bytes!("builder-fixture/tests/integration.rs").as_slice(),
),
(
"src/add.c",
include_bytes!("builder-fixture/src/add.c").as_slice(),
),
] {
std::fs::write(source.join(path), bytes)?;
}
let runtime = Runtime::new(world)?;
let (code, out) = cargo(&runtime, &["build"]).await?;
assert_eq!(code, 0, "{out}");
assert!(out.contains("namespace assertions passed"), "{out}");
assert!(source.join("Cargo.lock").exists());
let mut shell_output = Vec::new();
let shell_code = runtime
.exec(
Command::Argv(vec![
"/bin/sh.wasm".into(),
"-c".into(),
"./target/wasm32-wasip2/debug/builder-fixture.wasm".into(),
]),
Options {
cwd: "/project".into(),
..Default::default()
},
&mut |_, b| shell_output.extend_from_slice(b),
)
.await?;
assert_eq!(shell_code, 0, "{}", String::from_utf8_lossy(&shell_output));
assert!(String::from_utf8_lossy(&shell_output).contains("builder fixture cwd: /project"));
let (code, out) = cargo(&runtime, &["run"]).await?;
assert_eq!(code, 0, "{out}");
assert!(out.contains("builder fixture cwd: /project"), "{out}");
let (code, out) = cargo(&runtime, &["run", "--", "--exit"]).await?;
assert_eq!(code, 23, "{out}");
let (code, out) = cargo(&runtime, &["test", "pass", "--", "--nocapture"]).await?;
assert_eq!(code, 0, "{out}");
assert!(out.contains("1 passed"), "{out}");
assert!(out.contains("INTEGRATION_CASE"), "{out}");
let (code, out) = cargo(&runtime, &["test", "fail"]).await?;
assert_eq!(code, 101, "{out}");
assert!(out.contains("intentional fixture failure"), "{out}");
let (code, out) = cargo(&runtime, &["test", "--no-run"]).await?;
assert_eq!(code, 0, "{out}");
assert!(!out.contains("intentional fixture failure"), "{out}");
for action in ["check", "clippy", "fetch", "tree", "metadata", "update"] {
let (code, out) = cargo(&runtime, &[action]).await?;
assert_eq!(code, 0, "{action}: {out}");
}
let base = std::fs::read_dir(temp.path().join("host/builds"))?
.next()
.unwrap()?
.path();
let before = cache_hits(&base)?;
let (code, out) = cargo(&runtime, &["clean"]).await?;
assert_eq!(code, 0, "{out}");
let (code, out) = cargo(&runtime, &["build"]).await?;
assert_eq!(code, 0, "{out}");
let after = cache_hits(&base)?;
println!("CACHE AFTER CLEAN: {before} -> {after} hits");
assert!(
after > before,
"the second build did not restore cached actions"
);
std::fs::write(
source.join("src/main.rs"),
"fn main(){println!(\"format me\");}\n",
)?;
let (code, out) = cargo(&runtime, &["fmt"]).await?;
assert_eq!(code, 0, "{out}");
assert!(std::fs::read_to_string(source.join("src/main.rs"))?.contains("fn main() {"));
let (code, out) = cargo(&runtime, &["remove", "serde"]).await?;
assert_eq!(code, 0, "{out}");
let (code, out) = cargo(&runtime, &["add", "serde"]).await?;
assert_eq!(code, 0, "{out}");
assert!(std::fs::read_to_string(source.join("Cargo.toml"))?.contains("serde"));
std::fs::write(
source.join("src/main.rs"),
"fn main() { undefined_function(); }",
)?;
let (code, out) = cargo(&runtime, &["build"]).await?;
assert_ne!(code, 0, "{out}");
assert!(out.contains("undefined_function"), "{out}");
Ok(())
}
#[tokio::test]
async fn timeout_and_cancellation() -> anyhow::Result<()> {
use sausage::builder::{Builder, Event, Request};
let temp = tempfile::tempdir()?;
let world = World::open(temp.path())?;
let source = temp.path().join("world/project");
std::fs::create_dir_all(source.join("src"))?;
std::fs::write(
source.join("Cargo.toml"),
"[package]\nname='hang'\nversion='0.1.0'\n[workspace]\n",
)?;
std::fs::write(source.join("src/main.rs"), "fn main() {}")?;
std::fs::write(
source.join("build.rs"),
"fn main() { std::process::Command::new(\"sh\").args([\"-c\", \"echo $$ > child.pid; exec sleep 1000\"]).status().unwrap(); }",
)?;
let builder = Arc::new(Builder::new(world.root.clone(), &world.host_dir)?);
let (tx, mut rx) = tokio::sync::mpsc::channel(16);
let task = tokio::spawn(async move {
builder
.execute(
Request {
argv: vec!["build".into()],
cwd: "/project".into(),
timeout_secs: Some(3),
},
&tx,
)
.await
});
let mut timed_out = false;
while let Some(event) = rx.recv().await {
if let Event::Exit { code, .. } = event {
timed_out = code == 124;
}
}
task.await??;
assert!(timed_out);
let binaries = programs();
let bin = temp.path().join("world/bin");
std::fs::create_dir_all(&bin)?;
std::fs::copy(binaries.join("cargo.wasm"), bin.join("cargo.wasm"))?;
let runtime = Runtime::new(world)?;
let code = runtime
.exec(
Command::Argv(vec!["/bin/cargo.wasm".into(), "build".into()]),
Options {
cwd: "/project".into(),
timeout: Some(Duration::from_secs(1)),
..Default::default()
},
&mut |_, _| {},
)
.await?;
assert_ne!(code, 0);
let (code, out) =
tokio::time::timeout(Duration::from_secs(5), cargo(&runtime, &["metadata"])).await??;
assert_eq!(
code, 0,
"the canceled build still held the builder lock: {out}"
);
Ok(())
}
+9 -1
View File
@@ -24,7 +24,14 @@ fn git(cwd: &Path, args: &[&str]) {
async fn several_mb_git_push() {
static BUILT: OnceLock<std::path::PathBuf> = OnceLock::new();
let component = BUILT.get_or_init(|| {
let target = Path::new(env!("CARGO_TARGET_TMPDIR")).join("hook-git");
let target = std::env::var_os("CARGO_TARGET_DIR")
.map(std::path::PathBuf::from)
.unwrap_or_else(|| {
Path::new(env!("CARGO_TARGET_TMPDIR"))
.parent()
.unwrap()
.to_owned()
});
let userland = Path::new(env!("CARGO_MANIFEST_DIR")).join("../userland");
if !userland.join("cwd-init/init.o").is_file() {
assert!(
@@ -71,6 +78,7 @@ async fn several_mb_git_push() {
"-c",
"user.email=test@example.com",
"commit",
"--no-gpg-sign",
"-m",
"large commit",
],
+3 -2
View File
@@ -12,6 +12,7 @@ fn write(path: PathBuf, text: &str) -> std::io::Result<()> {
pub fn component() -> &'static Path {
static PATH: OnceLock<PathBuf> = OnceLock::new();
PATH.get_or_init(|| {
let target = std::env::var_os("CARGO_TARGET_DIR").map(PathBuf::from).unwrap_or_else(|| Path::new(env!("CARGO_TARGET_TMPDIR")).parent().unwrap().to_owned());
let directory = Path::new(env!("CARGO_TARGET_TMPDIR")).join("hook-guest");
std::fs::create_dir_all(directory.join("src")).unwrap();
std::fs::create_dir_all(directory.join("wit/deps")).unwrap();
@@ -67,7 +68,7 @@ async fn main() -> anyhow::Result<()> {
Ok(())
}
"#).unwrap();
assert!(std::process::Command::new("cargo").args(["build", "--manifest-path"]).arg(directory.join("Cargo.toml")).args(["--target", "wasm32-wasip2", "--release", "-j", "2"]).env("CARGO_TARGET_DIR", directory.join("target")).status().unwrap().success(), "test guest build failed");
directory.join("target/wasm32-wasip2/release/hook-guest.wasm")
assert!(std::process::Command::new("cargo").args(["build", "--manifest-path"]).arg(directory.join("Cargo.toml")).args(["--target", "wasm32-wasip2", "--release", "-j", "2"]).env("CARGO_TARGET_DIR", &target).status().unwrap().success(), "test guest build failed");
target.join("wasm32-wasip2/release/hook-guest.wasm")
})
}
+16 -28
View File
@@ -350,6 +350,16 @@ fn repo() -> PathBuf {
.unwrap()
.to_owned()
}
fn target() -> PathBuf {
std::env::var_os("CARGO_TARGET_DIR")
.map(PathBuf::from)
.unwrap_or_else(|| {
Path::new(env!("CARGO_TARGET_TMPDIR"))
.parent()
.unwrap()
.to_owned()
})
}
fn build() {
BUILD.call_once(|| {
let root = repo();
@@ -357,8 +367,7 @@ fn build() {
assert!(
std::process::Command::new(args[0])
.args(&args[1..])
.env_remove("CARGO_TARGET_DIR")
.env("CARGO_HOME", cwd.join(".cargo-home"))
.env("CARGO_TARGET_DIR", target())
.current_dir(cwd)
.status()
.unwrap()
@@ -403,16 +412,7 @@ fn build() {
&root.join("userland"),
);
run(
&[
"cargo",
"build",
"--release",
"-p",
"git",
"--target-dir",
"../target/process-userland",
"--locked",
],
&["cargo", "build", "--release", "-p", "git", "--locked"],
&root.join("userland"),
);
});
@@ -442,7 +442,7 @@ fn world() -> (tempfile::TempDir, Arc<Runtime>, sausage::process::Processes) {
let processes = world.processes.clone();
fixture(&temp.path().join("world"));
let artifacts = repo().join("userland/target/wasm32-wasip2/release");
let artifacts = target().join("wasm32-wasip2/release");
let bin = temp.path().join("world/bin");
for tool in ["cat", "tr", "ls", "wc", "yes", "head", "sleep"] {
std::fs::copy(
@@ -455,21 +455,9 @@ fn world() -> (tempfile::TempDir, Arc<Runtime>, sausage::process::Processes) {
std::fs::copy(artifacts.join("sausage-procps.wasm"), bin.join(tool)).unwrap();
}
std::fs::copy(artifacts.join("sausage-cwd-probe.wasm"), bin.join("probe")).unwrap();
std::fs::copy(
repo().join("userland/brush/source/target/wasm32-wasip2/release/brush.wasm"),
bin.join("sh"),
)
.unwrap();
std::fs::copy(
repo().join("userland/brush/probe/target/wasm32-wasip2/release/process-probe.wasm"),
bin.join("launcher"),
)
.unwrap();
std::fs::copy(
repo().join("target/process-userland/wasm32-wasip2/release/git.wasm"),
bin.join("git"),
)
.unwrap();
std::fs::copy(artifacts.join("brush.wasm"), bin.join("sh")).unwrap();
std::fs::copy(artifacts.join("process-probe.wasm"), bin.join("launcher")).unwrap();
std::fs::copy(artifacts.join("git.wasm"), bin.join("git")).unwrap();
(
temp,
Runtime::new_with_cache(world, &cache).unwrap(),
+10
View File
@@ -6742,6 +6742,16 @@ dependencies = [
"wasmparser",
]
[[package]]
name = "world-cargo"
version = "0.1.0"
dependencies = [
"sausage-spawn",
"serde",
"serde_json",
"wasi 0.14.7+wasi-0.2.4",
]
[[package]]
name = "write16"
version = "1.0.0"
+1 -1
View File
@@ -1,7 +1,7 @@
# The default CLIs of a world, built for wasm32-wasip2. Its own workspace, outside the host's.
[workspace]
resolver = "3"
members = ["git", "coreutils", "procps", "cwd-init/probe", "ripgrep", "diffutils", "jaq", "sed", "grep", "fd", "spawn", "findutils", "exec-tools", "ssh-client", "ssh", "curl", "init"]
members = ["git", "coreutils", "procps", "cwd-init/probe", "ripgrep", "diffutils", "jaq", "sed", "grep", "fd", "spawn", "findutils", "exec-tools", "ssh-client", "ssh", "curl", "init", "cargo"]
# WASI compatibility patches: git/vendor/PATCHES.md.
[patch.crates-io]
+14
View File
@@ -0,0 +1,14 @@
[package]
name = "world-cargo"
version = "0.1.0"
edition = "2024"
[[bin]]
name = "cargo"
path = "src/main.rs"
[dependencies]
wasi = "0.14"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
spawn = { package = "sausage-spawn", path = "../spawn" }
+131
View File
@@ -0,0 +1,131 @@
//! Stream native compiler events and run the resulting components through the process interface.
use std::io::{self, Write};
use wasi::http::{outgoing_handler, types::*};
#[derive(serde::Deserialize)]
#[serde(tag = "event", rename_all = "snake_case")]
enum Event {
Stdout { bytes: Vec<u8> },
Stderr { bytes: Vec<u8> },
Exit { code: i32, executables: Vec<String> },
}
fn error(value: impl std::fmt::Debug) -> io::Error {
io::Error::other(format!("{value:?}"))
}
pub fn run() -> io::Result<i32> {
let argv: Vec<String> = std::env::args().skip(1).collect();
let data = serde_json::to_vec(
&serde_json::json!({"argv": argv, "cwd": std::env::current_dir()?.to_string_lossy()}),
)?;
let fields = Fields::new();
fields
.set("content-length", &[data.len().to_string().into_bytes()])
.map_err(error)?;
fields
.set("content-type", &[b"application/json".to_vec()])
.map_err(error)?;
let request = OutgoingRequest::new(fields);
request.set_method(&Method::Post).map_err(error)?;
request.set_scheme(Some(&Scheme::Http)).map_err(error)?;
request.set_authority(Some("host.sausage")).map_err(error)?;
request.set_path_with_query(Some("/cargo")).map_err(error)?;
let outgoing = request.body().map_err(error)?;
let options = RequestOptions::new();
options
.set_between_bytes_timeout(Some(1860 * 1_000_000_000))
.map_err(error)?;
let future = outgoing_handler::handle(request, Some(options)).map_err(error)?;
let stream = outgoing.write().map_err(error)?;
let mut remaining = data.as_slice();
while !remaining.is_empty() {
stream.subscribe().block();
let n = (stream.check_write().map_err(error)? as usize).min(remaining.len());
stream.write(&remaining[..n]).map_err(error)?;
remaining = &remaining[n..];
}
drop(stream);
OutgoingBody::finish(outgoing, None).map_err(error)?;
future.subscribe().block();
let response = future
.get()
.ok_or_else(|| error("missing response"))?
.map_err(error)?
.map_err(error)?;
let incoming = response.consume().map_err(error)?;
let stream = incoming.stream().map_err(error)?;
let mut pending = Vec::new();
let mut result = None;
loop {
stream.subscribe().block();
match stream.read(65536) {
Ok(bytes) => pending.extend(bytes),
Err(wasi::io::streams::StreamError::Closed) => break,
Err(e) => return Err(error(e)),
}
while let Some(end) = pending.iter().position(|b| *b == b'\n') {
let line: Vec<_> = pending.drain(..=end).collect();
match serde_json::from_slice::<Event>(&line)? {
Event::Stdout { bytes } => {
io::stdout().write_all(&bytes)?;
io::stdout().flush()?;
}
Event::Stderr { bytes } => {
io::stderr().write_all(&bytes)?;
io::stderr().flush()?;
}
Event::Exit { code, executables } => result = Some((code, executables)),
}
}
}
drop(stream);
drop(IncomingBody::finish(incoming));
drop(response);
let (code, executables) = result.ok_or_else(|| error("missing Cargo exit event"))?;
if code != 0 {
return Ok(code);
}
let action = argv.first().map(String::as_str).unwrap_or("");
let args = execution_args(&argv);
for executable in executables {
let code = spawn::Command::new(executable).args(&args).status()?;
if code != 0 {
return Ok(if action == "test" { 101 } else { code });
}
}
Ok(0)
}
fn execution_args(argv: &[String]) -> Vec<String> {
let split = argv.iter().position(|a| a == "--").unwrap_or(argv.len());
let mut args = argv.get(split + 1..).unwrap_or_default().to_vec();
if argv.first().is_some_and(|a| a == "test") {
let mut value = false;
for arg in &argv[1..split] {
if !value && !arg.starts_with('-') {
args.insert(0, arg.clone());
break;
}
value = [
"-p",
"--package",
"--test",
"--bin",
"--example",
"--features",
"--profile",
"--manifest-path",
"--exclude",
"-j",
"--jobs",
"-F",
"--config",
"--target",
"--target-dir",
"--message-format",
"--color",
"--lockfile-path",
"-Z",
]
.contains(&arg.as_str());
}
}
args
}
+10
View File
@@ -0,0 +1,10 @@
//! Cargo's world entry point. Compile through HTTP, then execute tests and programs in the world.
fn main() {
match world_cargo::run() {
Ok(code) => std::process::exit(code),
Err(error) => {
eprintln!("cargo: {error}");
std::process::exit(1);
}
}
}