Merge v3-t39: Rust builder behind host.sausage/cargo and the world cargo CLI (T39, P4)
# Conflicts: # flake.nix # host/src/process/mod.rs # host/tests/hook_git.rs # host/tests/hook_guest.rs # host/tests/process.rs # userland/Cargo.toml
This commit is contained in:
@@ -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()
|
||||
}
|
||||
@@ -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;
|
||||
@@ -20,6 +22,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,
|
||||
@@ -49,6 +52,7 @@ impl Api {
|
||||
) -> Self {
|
||||
Self {
|
||||
cred: cred.into(),
|
||||
builder: Default::default(),
|
||||
script: cred::quickjs::QuickJs::default(),
|
||||
passkeys: Mutex::new(passkeys),
|
||||
entry,
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
# Rust builder
|
||||
|
||||
The world `cargo` calls `POST http://host.sausage/cargo` with `argv` and `cwd`.
|
||||
The request also accepts 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 component locations.
|
||||
Each location contains an executable path and working directory.
|
||||
|
||||
The CLI forwards the bytes immediately.
|
||||
For `test` and `run`, it executes the artifacts through the process interface.
|
||||
Cargo selects run binaries through a capture runner, so `--bin` and `default-run` retain their normal behavior.
|
||||
The test runner uses each package's root directory, even when the caller runs Cargo from a subdirectory.
|
||||
For `test` and `run`, arguments after `--` belong to the executable, not the host build.
|
||||
Other commands retain their `--` arguments, such as `fmt -- --check` and Clippy lint options.
|
||||
|
||||
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 and target directory.
|
||||
It also owns a 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.
|
||||
|
||||
The source copy and target directory permit writes.
|
||||
The private kache cache also permits writes.
|
||||
The registry and Nix store remain read-only.
|
||||
Host home directories 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 preserves other source files, including concurrent edits in the world.
|
||||
It publishes WASI executables into the world's workspace target directory.
|
||||
After `cargo clean`, it removes the published artifacts that Cargo removed from the native target.
|
||||
Every target build uses `wasm32-wasip2` and links the shared cwd-init object, including its numeric exit-code interface.
|
||||
The host rejects other build targets and `--target-dir` overrides.
|
||||
|
||||
The command deadline defaults to 30 minutes.
|
||||
The host kills timed-out commands before it sends their timeout message.
|
||||
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.
|
||||
@@ -0,0 +1,156 @@
|
||||
//! Workspace discovery and source/artifact copies through the world's shared VFS.
|
||||
use anyhow::{Context, Result};
|
||||
use sausage_vfs::{Layer, Open};
|
||||
use std::{collections::HashMap, path::Path};
|
||||
|
||||
/// Source hashes before the native command changes its copy.
|
||||
pub type Snapshot = HashMap<String, [u8; 32]>;
|
||||
|
||||
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<Snapshot> {
|
||||
std::fs::create_dir_all(destination)?;
|
||||
let mut before = Snapshot::new();
|
||||
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 {
|
||||
before.extend(Box::pin(copy_out(root, &from, &to)).await?);
|
||||
} else {
|
||||
let bytes = read(root, &from).await?;
|
||||
before.insert(from, openssl::sha::sha256(&bytes));
|
||||
std::fs::write(to, bytes)?;
|
||||
}
|
||||
}
|
||||
Ok(before)
|
||||
}
|
||||
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(¤t).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(())
|
||||
}
|
||||
/// A snapshot selects changed sources. Without a snapshot, publish only WASM artifacts.
|
||||
pub async fn copy_back(
|
||||
root: &dyn Layer,
|
||||
source: &Path,
|
||||
destination: &str,
|
||||
before: Option<&Snapshot>,
|
||||
) -> 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, before)).await?;
|
||||
} else if before.is_some() || entry.path().extension().is_some_and(|e| e == "wasm") {
|
||||
let bytes = std::fs::read(entry.path())?;
|
||||
if before.is_none_or(|before| before.get(&to) != Some(&openssl::sha::sha256(&bytes))) {
|
||||
write(root, &to, bytes).await?;
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Remove published artifacts that Cargo clean removed from the native target.
|
||||
pub async fn clean_artifacts(root: &dyn Layer, native: &Path, world: &str) -> Result<()> {
|
||||
let entries = match root.list(world).await {
|
||||
Ok(entries) => entries,
|
||||
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
|
||||
Err(error) => return Err(error.into()),
|
||||
};
|
||||
for entry in entries {
|
||||
let path = sausage_vfs::resolve(world, &entry.name);
|
||||
let native = native.join(&entry.name);
|
||||
if entry.directory {
|
||||
Box::pin(clean_artifacts(root, &native, &path)).await?;
|
||||
} else if entry.name.ends_with(".wasm") && !native.exists() {
|
||||
root.remove_file(&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)
|
||||
}
|
||||
@@ -0,0 +1,492 @@
|
||||
//! 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, PartialEq, Eq, PartialOrd, Ord)]
|
||||
pub struct Executable {
|
||||
pub path: String,
|
||||
pub cwd: String,
|
||||
}
|
||||
#[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<Executable>,
|
||||
},
|
||||
}
|
||||
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 compiles = ["build", "check", "clippy", "test", "run"].contains(&action);
|
||||
let mut args = request.argv.iter().take_while(|arg| arg.as_str() != "--");
|
||||
while let Some(arg) = args.next() {
|
||||
ensure!(
|
||||
arg != "--target-dir" && !arg.starts_with("--target-dir="),
|
||||
"the builder controls the target directory"
|
||||
);
|
||||
let target = if arg == "--target" {
|
||||
args.next().map(String::as_str)
|
||||
} else {
|
||||
arg.strip_prefix("--target=")
|
||||
};
|
||||
if compiles && let Some(target) = target {
|
||||
ensure!(
|
||||
target == "wasm32-wasip2",
|
||||
"the build target is always wasm32-wasip2"
|
||||
);
|
||||
}
|
||||
}
|
||||
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?;
|
||||
let world_target = sausage_vfs::resolve(&workspace, "target");
|
||||
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");
|
||||
// Keep /tmp's namespace mountpoints outside the directory Cargo clean removes.
|
||||
let temporary = target_mount.join("tmp");
|
||||
let registry = base.join("cargo");
|
||||
let cache = base.join("kache");
|
||||
if source.exists() {
|
||||
std::fs::remove_dir_all(&source)?;
|
||||
}
|
||||
let before = files::copy_out(self.root.as_ref(), &workspace, &source).await?;
|
||||
for path in [&target, ®istry, &cache, &temporary] {
|
||||
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 relative = world_cwd
|
||||
.strip_prefix(&workspace)
|
||||
.expect("workspace is an ancestor of cwd")
|
||||
.trim_start_matches('/');
|
||||
let cwd = if relative.is_empty() {
|
||||
source.clone()
|
||||
} else {
|
||||
source.join(relative)
|
||||
};
|
||||
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", ®istry);
|
||||
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(),
|
||||
),
|
||||
("run.nu", include_bytes!("run.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, Some(&before)).await?;
|
||||
self.evict(20 * 1024 * 1024 * 1024)?;
|
||||
tx.send(Event::Exit {
|
||||
code: startup_code,
|
||||
executables,
|
||||
})
|
||||
.await?;
|
||||
return Ok(());
|
||||
}
|
||||
let kache = executable("kache")?;
|
||||
let run_path = target_mount.join("run-path");
|
||||
if action == "run" && run_path.exists() {
|
||||
std::fs::remove_file(&run_path)?;
|
||||
}
|
||||
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(&temporary)
|
||||
.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 [®istry, &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(®istry)
|
||||
.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", "SAUSAGE_RUN_PATH"])
|
||||
.arg(&run_path)
|
||||
.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 matches!(action, "run" | "test")
|
||||
&& 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());
|
||||
}
|
||||
}
|
||||
let split = args.iter().position(|a| a == "--").unwrap_or(args.len());
|
||||
if action == "run" {
|
||||
cmd.arg("--config").arg(format!(
|
||||
"target.wasm32-wasip2.runner=[{:?},{:?}]",
|
||||
executable("nu")?,
|
||||
tools.join("run.nu")
|
||||
));
|
||||
}
|
||||
// Keep all global config flags before the subcommand so Cargo retains them.
|
||||
cmd.args(&args[..split]);
|
||||
if compiles {
|
||||
cmd.args([
|
||||
"--target",
|
||||
"wasm32-wasip2",
|
||||
"--message-format=json-render-diagnostics",
|
||||
]);
|
||||
}
|
||||
cmd.args(&args[split..]);
|
||||
let (status, stdout) = run(cmd, tx, deadline).await?;
|
||||
code = status;
|
||||
if action == "clean" && code == 0 {
|
||||
files::clean_artifacts(self.root.as_ref(), &target, &world_target).await?;
|
||||
}
|
||||
let artifacts = target.join("wasm32-wasip2");
|
||||
if artifacts.exists() {
|
||||
files::copy_back(
|
||||
self.root.as_ref(),
|
||||
&artifacts,
|
||||
&format!("{world_target}/wasm32-wasip2"),
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
if code == 0 && action == "run" && run_path.exists() {
|
||||
let path = std::fs::read_to_string(&run_path)?;
|
||||
let relative = Path::new(&path).strip_prefix(&target)?;
|
||||
executables.push(Executable {
|
||||
path: format!("{world_target}/{}", relative.display()),
|
||||
cwd: world_cwd.clone(),
|
||||
});
|
||||
}
|
||||
if code == 0
|
||||
&& action == "test"
|
||||
&& !request
|
||||
.argv
|
||||
.iter()
|
||||
.take_while(|a| a.as_str() != "--")
|
||||
.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"
|
||||
&& value["profile"]["test"] == true
|
||||
&& let Some(path) = value["executable"].as_str()
|
||||
&& let Ok(relative) = Path::new(path).strip_prefix(&target)
|
||||
{
|
||||
let manifest = Path::new(
|
||||
value["manifest_path"]
|
||||
.as_str()
|
||||
.context("missing test manifest")?,
|
||||
);
|
||||
let package = manifest.parent().unwrap().strip_prefix(&source)?;
|
||||
let cwd = sausage_vfs::resolve(&workspace, &package.to_string_lossy());
|
||||
executables.push(Executable {
|
||||
path: format!("{world_target}/{}", relative.display()),
|
||||
cwd,
|
||||
});
|
||||
}
|
||||
}
|
||||
executables.sort();
|
||||
executables.dedup();
|
||||
}
|
||||
}
|
||||
files::copy_back(self.root.as_ref(), &source, &workspace, Some(&before)).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, timed_out) = tokio::select! {
|
||||
result = collect => (result, false),
|
||||
_ = tx.closed() => anyhow::bail!("caller exited"),
|
||||
_ = tokio::time::sleep_until(deadline) => (Ok((124, Vec::new())), true),
|
||||
};
|
||||
// Kill the command before a full output channel can delay its timeout message.
|
||||
drop(group);
|
||||
if timed_out {
|
||||
tx.send(Event::Stderr {
|
||||
bytes: b"cargo: command timed out\n".to_vec(),
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
result
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
# Record Cargo's selected component without executing it on the host.
|
||||
def main [executable: string] {
|
||||
$executable | save --raw --force $env.SAUSAGE_RUN_PATH
|
||||
}
|
||||
@@ -0,0 +1,236 @@
|
||||
//! 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", ©).await?;
|
||||
assert!(copy.join("member/Cargo.toml").exists());
|
||||
assert!(!copy.join("target").exists());
|
||||
Ok(())
|
||||
}
|
||||
#[tokio::test]
|
||||
async fn copies_only_changed_and_new_sources() -> Result<()> {
|
||||
let temp = tempfile::tempdir()?;
|
||||
let world = crate::world::World::open(temp.path())?;
|
||||
for name in ["unchanged.rs", "changed.rs"] {
|
||||
files::write(
|
||||
world.root.as_ref(),
|
||||
&format!("/project/{name}"),
|
||||
b"initial".to_vec(),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
let copy = temp.path().join("copy");
|
||||
let before = files::copy_out(world.root.as_ref(), "/project", ©).await?;
|
||||
files::write(
|
||||
world.root.as_ref(),
|
||||
"/project/unchanged.rs",
|
||||
b"concurrent edit".to_vec(),
|
||||
)
|
||||
.await?;
|
||||
std::fs::write(copy.join("changed.rs"), b"changed by command")?;
|
||||
std::fs::write(copy.join("Cargo.lock"), b"created by command")?;
|
||||
files::copy_back(world.root.as_ref(), ©, "/project", Some(&before)).await?;
|
||||
for (name, expected) in [
|
||||
("unchanged.rs", "concurrent edit"),
|
||||
("changed.rs", "changed by command"),
|
||||
("Cargo.lock", "created by command"),
|
||||
] {
|
||||
assert_eq!(
|
||||
files::read(world.root.as_ref(), &format!("/project/{name}")).await?,
|
||||
expected.as_bytes()
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
#[tokio::test]
|
||||
async fn clean_removes_only_deleted_artifacts() -> Result<()> {
|
||||
let temp = tempfile::tempdir()?;
|
||||
let world = crate::world::World::open(temp.path())?;
|
||||
let native = temp.path().join("native");
|
||||
std::fs::create_dir_all(&native)?;
|
||||
std::fs::write(native.join("keep.wasm"), b"keep")?;
|
||||
for name in ["keep.wasm", "remove.wasm", "note.txt"] {
|
||||
files::write(
|
||||
world.root.as_ref(),
|
||||
&format!("/target/{name}"),
|
||||
b"data".to_vec(),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
files::clean_artifacts(world.root.as_ref(), &native, "/target").await?;
|
||||
assert!(
|
||||
files::read(world.root.as_ref(), "/target/remove.wasm")
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
for name in ["keep.wasm", "note.txt"] {
|
||||
assert_eq!(
|
||||
files::read(world.root.as_ref(), &format!("/target/{name}")).await?,
|
||||
b"data"
|
||||
);
|
||||
}
|
||||
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(())
|
||||
}
|
||||
#[tokio::test]
|
||||
async fn target_options_preserve_the_platform_target_and_cache() -> Result<()> {
|
||||
let temp = tempfile::tempdir()?;
|
||||
let world = crate::world::World::open(temp.path())?;
|
||||
let builder = Builder::new(world.root, &world.host_dir)?;
|
||||
let (tx, _rx) = mpsc::channel(1);
|
||||
for options in [
|
||||
vec!["--target", "aarch64-unknown-linux-gnu"],
|
||||
vec!["--target=aarch64-unknown-linux-gnu"],
|
||||
vec!["--target-dir", "/other"],
|
||||
vec!["--target-dir=/other"],
|
||||
] {
|
||||
let error = builder
|
||||
.execute(
|
||||
Request {
|
||||
argv: std::iter::once("build")
|
||||
.chain(options)
|
||||
.map(str::to_string)
|
||||
.collect(),
|
||||
cwd: "/".into(),
|
||||
timeout_secs: None,
|
||||
},
|
||||
&tx,
|
||||
)
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(error.to_string().contains("target"), "{error}");
|
||||
}
|
||||
for (action, options) in [
|
||||
("run", vec!["--target", "wasm32-wasip2"]),
|
||||
("run", vec!["--target=wasm32-wasip2"]),
|
||||
("run", vec!["--", "--target-dir", "/argument"]),
|
||||
("tree", vec!["--target", "all"]),
|
||||
] {
|
||||
let error = builder
|
||||
.execute(
|
||||
Request {
|
||||
argv: std::iter::once(action)
|
||||
.chain(options)
|
||||
.map(str::to_string)
|
||||
.collect(),
|
||||
cwd: "/".into(),
|
||||
timeout_secs: None,
|
||||
},
|
||||
&tx,
|
||||
)
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert_eq!(error.to_string(), "no Cargo.toml above cwd");
|
||||
}
|
||||
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(())
|
||||
}
|
||||
#[cfg(target_os = "linux")]
|
||||
#[tokio::test]
|
||||
async fn timeout_kills_before_output_can_drain() -> Result<()> {
|
||||
use std::os::fd::{FromRawFd, OwnedFd};
|
||||
let (tx, mut rx) = mpsc::channel(1);
|
||||
let output = tx.clone();
|
||||
let mut cmd = Command::new("sh");
|
||||
cmd.args(["-c", "echo $$; exec sleep 1000"]);
|
||||
let task =
|
||||
tokio::spawn(
|
||||
async move { run(cmd, &output, Instant::now() + Duration::from_secs(1)).await },
|
||||
);
|
||||
let Some(Event::Stdout { bytes }) =
|
||||
tokio::time::timeout(Duration::from_secs(2), rx.recv()).await?
|
||||
else {
|
||||
anyhow::bail!("missing native PID");
|
||||
};
|
||||
let pid: u32 = std::str::from_utf8(&bytes)?.trim().parse()?;
|
||||
let fd = unsafe { libc::syscall(libc::SYS_pidfd_open, pid, 0) };
|
||||
ensure!(fd >= 0, "pidfd_open: {}", std::io::Error::last_os_error());
|
||||
// OwnedFd keeps the PID descriptor valid and unchanged until AsyncFd drops it.
|
||||
let exited = unsafe { tokio::io::unix::AsyncFd::register(OwnedFd::from_raw_fd(fd as i32)) }?;
|
||||
// Fill the output channel, so the timeout message cannot enter it.
|
||||
tx.send(Event::Stdout { bytes: Vec::new() }).await?;
|
||||
let _ready = tokio::time::timeout(Duration::from_secs(2), exited.readable()).await??;
|
||||
assert!(matches!(rx.recv().await, Some(Event::Stdout { .. })));
|
||||
assert!(matches!(rx.recv().await, Some(Event::Stderr { bytes })
|
||||
if bytes.as_slice() == b"cargo: command timed out\n"));
|
||||
assert_eq!(task.await??.0, 124);
|
||||
Ok(())
|
||||
}
|
||||
#[tokio::test]
|
||||
async fn closed_receiver_cancels_command() -> Result<()> {
|
||||
let (tx, mut rx) = mpsc::channel(1);
|
||||
let mut cmd = Command::new("sh");
|
||||
cmd.args(["-c", "printf ready; exec sleep 1000"]);
|
||||
let task =
|
||||
tokio::spawn(async move { run(cmd, &tx, Instant::now() + Duration::from_secs(60)).await });
|
||||
// Output must arrive before the command exits, not after collection completes.
|
||||
let event = tokio::time::timeout(Duration::from_secs(2), rx.recv())
|
||||
.await?
|
||||
.context("missing streamed output")?;
|
||||
assert!(matches!(event, Event::Stdout { bytes } if bytes == b"ready"));
|
||||
drop(rx);
|
||||
assert!(
|
||||
tokio::time::timeout(Duration::from_secs(2), task)
|
||||
.await??
|
||||
.is_err()
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
@@ -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(()) }) as Io,
|
||||
));
|
||||
}
|
||||
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 {
|
||||
|
||||
@@ -193,6 +193,10 @@ impl Runtime {
|
||||
)?),
|
||||
};
|
||||
router.attach_journal(world.root.journal());
|
||||
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));
|
||||
|
||||
Reference in New Issue
Block a user