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

host: wasip1 processes, wasi:http, identity, stdio options, services (T12)

Browse Source 
- Core modules run on the host-side preview1 over the shared VFS (D41).
- wasi:http default handler in the linker; anonymous HTTPS git clone works.
  Sockets stay denied until P2.
- Each process carries the conversation (from Exec, inherited by children,
  shown in /proc/processes) and a host-assigned SAUSAGE_PID.
- Spawn requests choose pipe, inherit or null per stream.
- /config/services.yaml supervisor: live edits, restarts, crash limit 5/10 s,
  graceful stop (stdin EOF, then kill after 5 s) (D44).
- Compile caches keyed by content; PATH=/usr/local/bin:/bin.

world.rs also carries T13's optional read-only /bin mount (D43).
This commit is contained in:
surma
2026-11-06 17:21:51 +00:00
parent afff3ccda5
commit 665b8a2a03
13 changed files with 1385 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
host/src/process/mod.rs
+145 -38
View File
@@ -1,4 +1,4 @@
//! Runs components in independent stores over the world root. Owns process groups, stdio, and Exec.
//! Runs WASI programs in independent stores over the world root. Owns process groups, stdio, and Exec.
use crate::{
exec::{Command, Exec, Options, Stream},
world::World,
@@ -26,7 +26,7 @@ use wasmtime::{
};
use wasmtime_wasi::{
WasiCtx, WasiCtxBuilder, WasiCtxView, WasiView,
cli::{AsyncStdinStream, AsyncStdoutStream},
cli::AsyncStdoutStream,
p2::{
DynInputStream, DynOutputStream,
bindings::CommandPre,
@@ -44,7 +44,17 @@ mod bindings {
},
});
}
use bindings::sausage::process::host::{Host, HostChild, Request};
use bindings::sausage::process::host::{Host, HostChild, Request, Stdio};
mod services;
mod stdio;
pub use services::supervise;
use stdio::Io;
#[derive(Clone)]
enum Compiled {
Component(CommandPre<State>),
Module(wasmtime::InstancePre<State>),
}
/// A finite snapshot of process metadata exposed through /proc/processes.
#[derive(Clone, Debug)]
@@ -53,6 +63,7 @@ pub struct ProcessInfo {
pub argv: Vec<String>,
pub cwd: String,
pub state: String,
pub conversation: Option<String>,
}
pub type Processes = Arc<Mutex<std::collections::BTreeMap<u32, ProcessInfo>>>;
@@ -66,7 +77,7 @@ struct Entry {
pub struct Runtime {
world: World,
engine: Engine,
cache: Mutex<HashMap<Vec<u8>, CommandPre<State>>>,
cache: Mutex<HashMap<Vec<u8>, Compiled>>,
entries: Mutex<HashMap<u32, Entry>>,
next: AtomicU32,
alive: Arc<AtomicBool>,
@@ -84,6 +95,19 @@ struct State {
pid: u32,
command: u32,
group: u32,
conversation: Option<String>,
stdio: Io,
preview1: Option<sausage_vfs::preview1::State>,
http: wasmtime_wasi_http::WasiHttpCtx,
}
impl wasmtime_wasi_http::WasiHttpView for State {
fn http(&mut self) -> wasmtime_wasi_http::WasiHttpCtxView<'_> {
wasmtime_wasi_http::WasiHttpCtxView {
ctx: &mut self.http,
table: &mut self.table,
hooks: Default::default(),
}
}
}
impl WasiView for State {
fn ctx(&mut self) -> WasiCtxView<'_> {
@@ -105,15 +129,17 @@ struct Started {
pid: u32,
group: u32,
code: watch::Receiver<Option<i32>>,
input: DuplexStream,
output: DuplexStream,
error: DuplexStream,
input: Option<DuplexStream>,
output: Option<DuplexStream>,
error: Option<DuplexStream>,
}
impl Runtime {
pub fn open(path: &Path) -> anyhow::Result<Arc<Self>> {
Self::new(World::open(path)?)
}
pub fn new(world: World) -> anyhow::Result<Arc<Self>> {
// Deno and WASI HTTP enable different providers. Select one before HTTPS creates a client.
let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
let mut config = wasmtime::Config::new();
config.epoch_interruption(true);
let mut cache = wasmtime::CacheConfig::new();
@@ -138,6 +164,10 @@ impl Runtime {
next: AtomicU32::new(1),
}))
}
/// Number of distinct program contents compiled by this runtime.
pub fn compiled_programs(&self) -> usize {
self.cache.lock().unwrap().len()
}
fn kill_command(&self, command: u32) {
for entry in self.entries.lock().unwrap().values() {
if entry.command == command {
@@ -170,6 +200,8 @@ impl Runtime {
parent: u32,
command: u32,
inherited_group: u32,
conversation: Option<String>,
inherited_io: Option<Io>,
) -> anyhow::Result<Started> {
anyhow::ensure!(!request.argv.is_empty(), "empty argv");
let cancelled = if parent == 0 {
@@ -182,6 +214,11 @@ impl Runtime {
.map(|entry| entry.cancelled.clone())
.ok_or_else(|| anyhow::anyhow!("parent exited"))?
};
if !request.env.iter().any(|(k, _)| k == "PATH") {
request
.env
.push(("PATH".into(), "/usr/local/bin:/bin".into()));
}
request.cwd = sausage_vfs::resolve("/", &request.cwd);
request.env.retain(|(name, _)| name != "PWD");
request.env.push(("PWD".into(), request.cwd.clone()));
@@ -205,19 +242,29 @@ impl Runtime {
if let Some(pre) = cache.get(&bytes) {
pre.clone()
} else {
let component = Component::new(&self.engine, &bytes)?;
let mut linker = Linker::new(&self.engine);
wasmtime_wasi::p2::add_to_linker_async(&mut linker)?;
// The library linker includes exit-with-code. Only Wasmtime's CLI gates it.
adapter::add_to_linker(&mut linker, |s: &mut State| View {
adapter: &mut s.fs,
table: &mut s.table,
})?;
bindings::sausage::process::host::add_to_linker::<
_,
wasmtime::component::HasSelf<_>,
>(&mut linker, |s| s)?;
let pre = CommandPre::new(linker.instantiate_pre(&component)?)?;
let pre = if bytes.get(..8) == Some(b"\0asm\x01\0\0\0") {
let module = wasmtime::Module::new(&self.engine, &bytes)?;
let mut linker = wasmtime::Linker::new(&self.engine);
sausage_vfs::preview1::add_to_linker(&mut linker, |s: &mut State| {
s.preview1.as_mut().unwrap()
})?;
Compiled::Module(linker.instantiate_pre(&module)?)
} else {
let component = Component::new(&self.engine, &bytes)?;
let mut linker = Linker::new(&self.engine);
wasmtime_wasi::p2::add_to_linker_async(&mut linker)?;
// The library linker includes exit-with-code. Only Wasmtime's CLI gates it.
wasmtime_wasi_http::p2::add_only_http_to_linker_async(&mut linker)?;
adapter::add_to_linker(&mut linker, |s: &mut State| View {
adapter: &mut s.fs,
table: &mut s.table,
})?;
bindings::sausage::process::host::add_to_linker::<
_,
wasmtime::component::HasSelf<_>,
>(&mut linker, |s| s)?;
Compiled::Component(CommandPre::new(linker.instantiate_pre(&component)?)?)
};
cache.insert(bytes, pre.clone());
pre
}
@@ -230,20 +277,45 @@ impl Runtime {
None if inherited_group != 0 => inherited_group,
None => pid,
};
request.env.retain(|(name, _)| name != "SAUSAGE_PID");
request.env.push(("SAUSAGE_PID".into(), pid.to_string()));
request.env.push(("PID".into(), pid.to_string()));
request.env.push(("PPID".into(), parent.to_string()));
let (input, child_input) = tokio::io::duplex(8192);
let (output, child_output) = tokio::io::duplex(8192);
let (error, child_error) = tokio::io::duplex(8192);
let (input, child_input) = stdio::input(request.stdin, inherited_io.as_ref());
let (output, child_output) =
stdio::output(request.stdout, inherited_io.as_ref().map(|io| &io.output));
let (error, child_error) =
stdio::output(request.stderr, inherited_io.as_ref().map(|io| &io.error));
let stdio = Io {
input: child_input,
output: child_output,
error: child_error,
};
let preview1 = if matches!(pre, Compiled::Module(_)) {
Some(sausage_vfs::preview1::State::new(
self.world.root.clone(),
&request.cwd,
request.argv.clone(),
request.env.clone(),
stdio.input(),
Box::new(stdio.output.clone()),
Box::new(stdio.error.clone()),
))
} else {
None
};
let mut mounts = Mounts::default();
mounts.mount("/", self.world.root.clone(), false);
let mut wasi = WasiCtxBuilder::new();
wasi.args(&request.argv)
.envs(&request.env)
.initial_cwd(&request.cwd)
.stdin(AsyncStdinStream::new(child_input))
.stdout(AsyncStdoutStream::new(8192, child_output))
.stderr(AsyncStdoutStream::new(8192, child_error));
.stdin(stdio.input.clone())
.stdout(AsyncStdoutStream::new(8192, stdio.output.clone()))
.stderr(AsyncStdoutStream::new(8192, stdio.error.clone()))
.allow_tcp(false)
.allow_udp(false)
.allow_ip_name_lookup(false);
let mut store = Store::new(
&self.engine,
State {
@@ -254,6 +326,10 @@ impl Runtime {
pid,
command,
group,
conversation: conversation.clone(),
stdio,
preview1,
http: wasmtime_wasi_http::WasiHttpCtx::new(),
},
);
store.epoch_deadline_async_yield_and_update(1);
@@ -264,6 +340,7 @@ impl Runtime {
argv: request.argv,
cwd: request.cwd,
state: "running".into(),
conversation,
},
);
let (tx, code) = watch::channel(None);
@@ -281,8 +358,21 @@ impl Runtime {
return;
}
let result = async {
let command = pre.instantiate_async(&mut store).await?;
match command.wasi_cli_run().call_run(&mut store).await {
let result = match pre {
Compiled::Component(pre) => {
let command = pre.instantiate_async(&mut store).await?;
command.wasi_cli_run().call_run(&mut store).await
}
Compiled::Module(pre) => {
let module = pre.instantiate_async(&mut store).await?;
module
.get_typed_func::<(), ()>(&mut store, "_start")?
.call_async(&mut store, ())
.await
.map(|()| Ok(()))
}
};
match result {
Ok(Ok(())) => Ok(0),
Ok(Err(())) => Ok(1),
Err(e) => match e.downcast_ref::<wasmtime_wasi::I32Exit>() {
@@ -348,7 +438,14 @@ impl Host for State {
) -> wasmtime::Result<Result<Resource<Child>, String>> {
let child = match self
.runtime
.start(request, self.pid, self.command, self.group)
.start(
request,
self.pid,
self.command,
self.group,
self.conversation.clone(),
Some(self.stdio.clone()),
)
.await
{
Ok(child) => child,
@@ -358,9 +455,15 @@ impl Host for State {
pid: child.pid,
group: child.group,
code: child.code,
input: Some(Box::new(AsyncWriteStream::new(8192, child.input))),
output: Some(Box::new(AsyncReadStream::new(child.output))),
error: Some(Box::new(AsyncReadStream::new(child.error))),
input: child
.input
.map(|s| Box::new(AsyncWriteStream::new(8192, s)) as DynOutputStream),
output: child
.output
.map(|s| Box::new(AsyncReadStream::new(s)) as DynInputStream),
error: child
.error
.map(|s| Box::new(AsyncReadStream::new(s)) as DynInputStream),
})?))
}
async fn kill(&mut self, pid: u32) -> wasmtime::Result<Result<(), String>> {
@@ -463,10 +566,7 @@ impl Exec for Arc<Runtime> {
],
Command::Argv(argv) => argv,
};
let mut env = options.env;
if !env.iter().any(|(k, _)| k == "PATH") {
env.push(("PATH".into(), "/bin".into()));
}
let env = options.env;
let start = self.start(
Request {
program: argv.first().cloned().unwrap_or_default(),
@@ -474,10 +574,15 @@ impl Exec for Arc<Runtime> {
env,
cwd: options.cwd,
group: None,
stdin: Stdio::Pipe,
stdout: Stdio::Pipe,
stderr: Stdio::Pipe,
},
0,
0,
0,
options.conversation,
None,
);
let mut child = if let Some(deadline) = deadline {
tokio::time::timeout_at(deadline, start)
@@ -491,13 +596,15 @@ impl Exec for Arc<Runtime> {
group: child.pid,
};
drop(child.input);
let mut stdout = child.output.take().unwrap();
let mut stderr = child.error.take().unwrap();
let run = async {
let (mut out_open, mut err_open) = (true, true);
let (mut out, mut err) = ([0; 8192], [0; 8192]);
while out_open || err_open {
tokio::select! {
n = child.output.read(&mut out), if out_open => { let n = n?; out_open = n != 0; if n != 0 { output(Stream::Stdout, &out[..n]); } }
n = child.error.read(&mut err), if err_open => { let n = n?; err_open = n != 0; if n != 0 { output(Stream::Stderr, &err[..n]); } }
n = stdout.read(&mut out), if out_open => { let n = n?; out_open = n != 0; if n != 0 { output(Stream::Stdout, &out[..n]); } }
n = stderr.read(&mut err), if err_open => { let n = n?; err_open = n != 0; if n != 0 { output(Stream::Stderr, &err[..n]); } }
}
}
loop {
host/src/process/services.rs
+169
View File
@@ -0,0 +1,169 @@
//! Supervises the world's services list, live edits, crash limits, and graceful stops.
use super::{GroupGuard, Request, Runtime, Started, Stdio};
use serde::Deserialize;
use std::{
collections::{BTreeMap, VecDeque},
sync::Arc,
time::Duration,
};
use tokio::time::Instant;
#[derive(Clone, Deserialize, PartialEq, Eq)]
struct Spec {
name: String,
argv: Vec<String>,
#[serde(default = "root")]
cwd: String,
#[serde(default)]
env: BTreeMap<String, String>,
}
fn root() -> String {
"/".into()
}
struct Live {
child: Started,
_guard: GroupGuard,
}
struct Service {
spec: Spec,
live: Option<Live>,
crashes: VecDeque<Instant>,
stopped: bool,
}
impl Service {
async fn start(&mut self, runtime: &Arc<Runtime>) {
match runtime
.start(
Request {
program: self.spec.argv.first().cloned().unwrap_or_default(),
argv: self.spec.argv.clone(),
cwd: self.spec.cwd.clone(),
env: self
.spec
.env
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
.collect(),
group: None,
stdin: Stdio::Pipe,
stdout: Stdio::Null,
stderr: Stdio::Null,
},
0,
0,
0,
None,
None,
)
.await
{
Ok(child) => {
let guard = GroupGuard {
runtime: runtime.clone(),
group: child.pid,
};
self.live = Some(Live {
child,
_guard: guard,
});
}
Err(_) => self.crashed(),
}
}
fn crashed(&mut self) {
let now = Instant::now();
while self
.crashes
.front()
.is_some_and(|t| now.duration_since(*t) >= Duration::from_secs(10))
{
self.crashes.pop_front();
}
self.crashes.push_back(now);
self.stopped = self.crashes.len() > 5;
}
async fn stop(&mut self) {
if let Some(mut live) = self.live.take() {
drop(live.child.input.take());
let _ = tokio::time::timeout(Duration::from_secs(5), async {
loop {
if live.child.code.borrow().is_some() {
break;
}
if live.child.code.changed().await.is_err() {
break;
}
}
})
.await;
// The guard stops remaining descendants, or the process after its grace period.
}
}
}
/// Start the supervisor. Aborting its task stops its service processes.
/// The YAML file contains name, argv, optional cwd, and an optional map of environment variables.
pub fn supervise(runtime: Arc<Runtime>) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
let mut services: BTreeMap<String, Service> = BTreeMap::new();
let mut previous = Vec::new();
let mut interval = tokio::time::interval(Duration::from_millis(100));
loop {
interval.tick().await;
let content = match runtime
.world
.root
.open("/config/services.yaml", Default::default())
.await
{
Ok(file) => match file.stat().await {
Ok(stat) => file.read(0, stat.size as usize).await.ok(),
Err(_) => None,
},
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Some(b"[]".to_vec()),
Err(_) => None,
};
if let Some(content) = content.filter(|c| *c != previous)
&& let Ok(specs) = serde_yaml_ng::from_slice::<Vec<Spec>>(&content)
{
let specs: BTreeMap<_, _> =
specs.into_iter().map(|s| (s.name.clone(), s)).collect();
let removed: Vec<_> = services
.keys()
.filter(|name| !specs.contains_key(*name))
.cloned()
.collect();
for name in removed {
services.remove(&name).unwrap().stop().await;
}
for (name, spec) in specs {
let service = services.entry(name).or_insert_with(|| Service {
spec: spec.clone(),
live: None,
crashes: VecDeque::new(),
stopped: false,
});
if service.spec != spec {
service.stop().await;
}
service.spec = spec;
service.crashes.clear();
service.stopped = false;
}
previous = content;
}
for service in services.values_mut() {
if service
.live
.as_ref()
.is_some_and(|live| live.child.code.borrow().is_some())
{
service.live = None;
service.crashed();
}
if service.live.is_none() && !service.stopped {
service.start(&runtime).await;
}
}
}
})
}
host/src/process/stdio.rs
+73
View File
@@ -0,0 +1,73 @@
//! Shares process stdio across stores without a parent-side IO pump.
use super::bindings::sausage::process::host::Stdio;
use std::{
pin::Pin,
sync::{Arc, Mutex},
task::{Context, Poll},
};
use tokio::io::AsyncWrite;
use wasmtime_wasi::cli::{AsyncStdinStream, StdinStream};
#[derive(Clone)]
pub(super) struct Io {
pub input: Arc<AsyncStdinStream>,
pub output: Writer,
pub error: Writer,
}
impl Io {
pub fn input(&self) -> sausage_vfs::preview1::Input {
Box::new(Pin::from(self.input.async_stream()))
}
}
#[derive(Clone)]
pub(super) struct Writer(Arc<Mutex<Box<dyn AsyncWrite + Send + Sync + Unpin>>>);
impl Writer {
pub fn new(writer: impl AsyncWrite + Send + Sync + Unpin + 'static) -> Self {
Self(Arc::new(Mutex::new(Box::new(writer))))
}
}
impl AsyncWrite for Writer {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
bytes: &[u8],
) -> Poll<std::io::Result<usize>> {
Pin::new(&mut **self.0.lock().unwrap()).poll_write(cx, bytes)
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
Pin::new(&mut **self.0.lock().unwrap()).poll_flush(cx)
}
fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
// Closing one inherited handle must not close the other handles.
self.poll_flush(cx)
}
}
pub(super) fn input(
mode: Stdio,
parent: Option<&Io>,
) -> (Option<tokio::io::DuplexStream>, Arc<AsyncStdinStream>) {
match mode {
Stdio::Pipe => {
let (host, guest) = tokio::io::duplex(8192);
(Some(host), Arc::new(AsyncStdinStream::new(guest)))
}
Stdio::Inherit => (
None,
parent.expect("inherit requires a parent").input.clone(),
),
Stdio::Null => (None, Arc::new(AsyncStdinStream::new(tokio::io::empty()))),
}
}
pub(super) fn output(
mode: Stdio,
parent: Option<&Writer>,
) -> (Option<tokio::io::DuplexStream>, Writer) {
match mode {
Stdio::Pipe => {
let (host, guest) = tokio::io::duplex(8192);
(Some(host), Writer::new(guest))
}
Stdio::Inherit => (None, parent.expect("inherit requires a parent").clone()),
Stdio::Null => (None, Writer::new(tokio::io::sink())),
}
}