//! Containers API, raw output archives, and bounded world transfers. pub mod cli; #[derive(Debug)] pub struct Error { pub message: String, pub data: Option, } impl std::fmt::Display for Error { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.write_str(&self.message) } } impl std::error::Error for Error {} impl From for Error { fn from(e: std::io::Error) -> Self { error(e.to_string()) } } impl From for Error { fn from(e: serde_json::Error) -> Self { error(e.to_string()) } } pub type Result = std::result::Result; fn error(s: impl Into) -> Error { Error { message: s.into(), data: None, } } fn pid() -> String { std::env::var("SAUSAGE_PID").unwrap_or_else(|_| { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() .as_nanos() .to_string() }) } use serde_json::{Value, json}; use std::{ fs::File, io::{Read, Write}, path::Path, }; use wasi::http::types::Method; mod archive; mod excerpt; mod http; mod retention; mod transfer; const DEFAULT_HOST: &str = "citadel.vpn.surma.technology:8093"; const MAX_FILE: u64 = 64 << 20; pub const NODE_MODULE: &str = r#"{ pkgs, ... }: { environment.systemPackages = with pkgs; [ nodejs git chromium fontconfig gnutar ]; fonts.packages = [ pkgs.dejavu_fonts ]; environment.variables = { PLAYWRIGHT_SKIP_BROWSER_DOWNLOAD = "1"; PLAYWRIGHT_CHROMIUM_EXECUTABLE_PATH = "${pkgs.chromium}/bin/chromium"; FONTCONFIG_FILE = "${pkgs.fontconfig.out}/etc/fonts/fonts.conf"; FONTCONFIG_PATH = "${pkgs.fontconfig.out}/etc/fonts"; }; }"#; // The service has two genuine JSON unions. Keep these wire types separate from // #[derive(serde::Serialize, serde::Deserialize)] structs; the SDK subset cannot express unions. No arbitrary result JSON. #[derive(serde::Serialize, serde::Deserialize)] #[serde(untagged)] pub enum Command { Shell(String), Argv(Vec), } #[derive(serde::Serialize, serde::Deserialize)] #[serde(untagged)] pub enum Nixos { Text(String), Options(std::collections::BTreeMap), } /// Exactly one Nix configuration source. #[derive(serde::Serialize, serde::Deserialize)] pub struct Config { /// Flake reference, optionally #configuration (default container). flake: Option, /// NixOS module text or JSON option object. nixos: Option, /// Already built /nix/store system path on the service host. system: Option, } /// Create a disposable container and wait for readiness. #[derive(serde::Serialize, serde::Deserialize)] pub struct Create { /// Exactly one of flake, nixos, system. config: Config, /// Lifetime from creation, service default 3600 seconds. ttl_secs: Option, /// Informational owner label. owner: Option, /// Service hostname or hostname:port, default citadel:8093. host: Option, } /// Select a service for listing boxes. #[derive(serde::Serialize, serde::Deserialize)] pub struct List { /// Hostname or hostname:port; default citadel:8093. host: Option, } /// Select a box; remembered host from create is used by default. #[derive(serde::Serialize, serde::Deserialize)] pub struct Id { /// Service-issued tc-xxxxxx ID. id: String, /// Override remembered host; default citadel:8093 for unknown IDs. host: Option, } /// Service metadata, with build-log tail attached to failed creates. #[derive(serde::Serialize, serde::Deserialize)] pub struct Container { /// Container ID. id: String, /// creating, running, updating or failed. state: String, /// Private container IP, accessible from service host. address: String, /// UTC creation timestamp. created_at: String, /// UTC expiry timestamp. expires_at: String, /// Informational owner. owner: String, /// Configuration used to create this container. config: Config, /// Service failure message. error: Option, /// Bounded build-log tail when creation fails; full available bytes are at build_log_path. #[serde(default)] build_log_tail: Option, /// Permanent archive of the build diagnostic returned by the service. #[serde(default)] build_log_path: Option, /// Exact archived byte count, not a total for old-service tails. #[serde(default)] build_log_bytes: Option, /// True when the legacy endpoint was used and may have returned only a tail. #[serde(default)] build_log_service_truncated: Option, /// Inline-tail byte range or compatibility warning, if applicable. #[serde(default)] build_log_notice: Option, } /// Service inventory. #[derive(serde::Serialize, serde::Deserialize)] pub struct Containers { /// Containers currently known to this host. containers: Vec, } /// Extend expiry to now plus ttl_secs (not additive). #[derive(serde::Serialize, serde::Deserialize)] pub struct Extend { /// Container ID. id: String, /// Positive remaining lifetime. ttl_secs: u32, /// Override remembered host. host: Option, } /// Execute as root inside a running box. #[derive(serde::Serialize, serde::Deserialize)] pub struct Exec { /// Container ID. id: String, /// Shell string (sh -c) or nonempty argv array. cmd: Command, /// UTF-8 standard input. stdin: Option, /// Default 60, maximum 1800 seconds. timeout_secs: Option, /// Inline UTF-8 byte budget per stream (default 16384), excluding the omission marker. Zero omits stream text. Full archives are unaffected. inline_bytes: Option, /// Override remembered host. host: Option, } /// Command completion, including timeout/output-limit status. #[derive(serde::Serialize, serde::Deserialize)] pub struct Executed { /// Process exit code. exit: i32, /// Bounded stdout head and tail; full available bytes are in stdout_path. stdout: String, /// Bounded stderr head and tail; full available bytes are in stderr_path. stderr: String, /// Service killed the command after its timeout. timed_out: bool, /// Inline output was cut by the service or plugin; service_truncated distinguishes upstream loss. truncated: bool, /// Permanent archive of service result metadata and inline excerpts. #[serde(default)] output_log: Option, /// Service execution identifier (new services only). #[serde(default)] exec_id: Option, /// Exact raw stdout total on new services; available UTF-8 bytes on old services (which may have lost output). #[serde(default)] stdout_bytes: Option, /// Exact raw stderr total on new services; available UTF-8 bytes on old services (which may have lost output). #[serde(default)] stderr_bytes: Option, /// Service download reference when stdout exceeded the inline limit. #[serde(default)] stdout_url: Option, /// Service download reference when stderr exceeded the inline limit. #[serde(default)] stderr_url: Option, /// Permanent full raw stdout archive (available UTF-8 text on old services). #[serde(default)] stdout_path: Option, /// Permanent full raw stderr archive (available UTF-8 text on old services). #[serde(default)] stderr_path: Option, /// True if the old service discarded output beyond its cap. #[serde(default)] service_truncated: bool, /// Output availability or archival failure; execution already completed, do not replay. #[serde(default)] output_notice: Option, } /// A successful mutation. #[derive(serde::Serialize, serde::Deserialize)] pub struct OkResult { /// True on success. ok: bool, } /// Build log response. #[derive(serde::Serialize, serde::Deserialize)] pub struct Log { /// At most the last 16 KiB of raw build bytes, displayed with UTF-8 replacement; full bytes are at path. text: String, /// Permanent world archive path. path: String, /// Exact byte count in the archived file. #[serde(default)] bytes: u64, /// Legacy endpoint used; the full build log may be unavailable. #[serde(default)] service_truncated: bool, /// Inline-tail byte range or compatibility warning, if applicable. #[serde(default)] notice: Option, } /// Upload UTF-8 text or stream a VFS file. #[derive(serde::Serialize, serde::Deserialize)] pub struct PutFile { /// Container ID. id: String, /// Absolute path inside the box. path: String, /// Inline text; exactly one of text and from_vfs is required. text: Option, /// World file to stream, including binary data, at most 64 MiB. from_vfs: Option, /// Override remembered host. host: Option, } /// Download a file, optionally streaming to the world. #[derive(serde::Serialize, serde::Deserialize)] pub struct GetFile { /// Container ID. id: String, /// Absolute file path inside the box (not a directory). path: String, /// World destination; atomically replaced only on successful complete download. to_vfs: Option, /// Override remembered host. host: Option, } /// File transfer result; payload omitted for VFS transfers. #[derive(serde::Serialize, serde::Deserialize)] pub struct FileResult { /// Bytes transferred. bytes: u64, /// UTF-8 contents for inline downloads, capped at 64 KiB. text: Option, /// Destination world path, if any. to_vfs: Option, } /// Copy a world directory into the box using an uncompressed tar. #[derive(serde::Serialize, serde::Deserialize)] pub struct PutTree { /// Container ID. id: String, /// World source directory. from_vfs_dir: String, /// Absolute destination inside the box. Existing files may be overwritten. to_dir: String, /// Override remembered host. host: Option, } /// Copy a box directory into a new world directory using tar. #[derive(serde::Serialize, serde::Deserialize)] pub struct GetTree { /// Container ID. id: String, /// Absolute source directory inside the box. from_dir: String, /// New world destination directory; must not already exist. to_vfs_dir: String, /// Override remembered host. host: Option, } /// Ready-to-use JavaScript and Chromium container. #[derive(serde::Serialize, serde::Deserialize)] pub struct NodeBox { /// Lifetime in seconds; service default 3600. ttl_secs: Option, } /// Disposable build/test boxes. No background workers or startup side effects. pub struct BoxPlugin; fn host(input: Option<&str>) -> Result { let h = input.unwrap_or(DEFAULT_HOST); let h = if h.contains(':') { h.to_owned() } else { format!("{h}:8093") }; let u = url::Url::parse(&format!("http://{h}")).map_err(|e| error(format!("invalid host: {e}")))?; if u.host_str().is_none() || u.path() != "/" || u.query().is_some() || u.fragment().is_some() { return Err(error("host must be hostname or hostname:port")); } Ok(h) } fn base(id: &str) -> Result { if id.len() != 9 || !id.starts_with("tc-") || !id[3..] .bytes() .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit()) { return Err(error("invalid box ID; expected tc-xxxxxx")); } Ok(format!("/containers/{id}")) } const HOSTS: &str = "/var/lib/box/hosts"; fn remember(id: &str, h: &str) -> Result<()> { base(id)?; std::fs::create_dir_all(HOSTS)?; let path = Path::new(HOSTS).join(id); if let Ok(old) = std::fs::read_to_string(&path) { if old != h { return Err(error(format!( "box ID collision for {id}; specify --host explicitly" ))); } return Ok(()); } let temp = path.with_extension(pid()); std::fs::write(&temp, h)?; std::fs::rename(temp, path)?; Ok(()) } fn resolve(id: &str, h: Option<&str>) -> Result { base(id)?; if let Some(h) = h { return host(Some(h)); } match std::fs::read_to_string(Path::new(HOSTS).join(id)) { Ok(h) => Ok(h), Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(DEFAULT_HOST.into()), Err(e) => Err(e.into()), } } fn get(h: &str, id: &str) -> Result { http::json(h, Method::Get, &base(id)?, None, 30) } fn log(h: &str, id: &str) -> Result { archive::build(h, id, http::now().saturating_add(60_000_000_000)) } // Shared stdout/stderr budget: 60 s overhead plus 1 s per 256 KiB, capped at // 30 min. Unknown totals get 5 min. HTTP idle timeouts still catch stalled peers. fn archive_budget_ns(stdout: Option, stderr: Option) -> u64 { let seconds = match (stdout, stderr) { (Some(out), Some(err)) => { 60u64.saturating_add(out.saturating_add(err).div_ceil(256 * 1024)) } _ => 300, }; seconds.min(1800) * 1_000_000_000 } fn execute(h: &str, id: &str, cmd: Value, stdin: Option, secs: u32) -> Result { let mut result: Executed = http::json( h, Method::Post, &format!("{}/exec", base(id)?), Some( json!({"cmd":cmd,"stdin":stdin.unwrap_or_default(),"timeout_secs":secs.clamp(1,1800)}), ), secs.clamp(1, 1800), )?; result.service_truncated = result.truncated && result.exec_id.is_none(); if result.service_truncated { result.output_notice = Some("Old service cut output at its stream cap; bytes beyond the inline output are unavailable.".into()); } // Fetch raw streams, never concatenate JSON excerpts (which can replace binary/UTF-8 bytes). if let Some(exec_id) = result.exec_id.as_deref() { let deadline = http::now().saturating_add(archive_budget_ns(result.stdout_bytes, result.stderr_bytes)); for (stream, url, count, path) in [ ( "stdout", &result.stdout_url, &mut result.stdout_bytes, &mut result.stdout_path, ), ( "stderr", &result.stderr_url, &mut result.stderr_bytes, &mut result.stderr_path, ), ] { // Known-empty streams need no network roundtrip (even if a URL was supplied). // Keep the usual on-disk contract for callers reading stdout_path. let name = format!("{id}-{exec_id}"); let saved = if *count == Some(0) { retention::create(&name, stream) .map(|(path, _)| archive::Download { path, bytes: 0 }) .map_err(Into::into) } else { // Small nonempty streams still need raw bytes: JSON may replace invalid UTF-8. let endpoint = url .clone() .unwrap_or_else(|| format!("/containers/{id}/exec/{exec_id}/{stream}")); archive::exec_download(h, &endpoint, &name, stream, deadline, *count) }; match saved { Ok(d) => { *path = Some(d.path); *count = Some(d.bytes); } Err(e) => { eprintln!( "box {id} {stream} archive failed: {e}; command completed, do not replay" ); let notice = result.output_notice.get_or_insert_with(String::new); notice.push_str(&format!( "{stream} archive unavailable: {}; command completed, do not replay. ", e.message )); } } } } if result.exec_id.is_none() { // Legacy services expose only JSON text. Preserve all available text before // bounding presentation, without claiming bytes discarded upstream exist. for (stream, text, path, count) in [ ( "stdout", &result.stdout, &mut result.stdout_path, &mut result.stdout_bytes, ), ( "stderr", &result.stderr, &mut result.stderr_path, &mut result.stderr_bytes, ), ] { *count = Some(text.len() as u64); match retention::save("box", &format!("{id}-{stream}"), text.as_bytes()) { Ok(saved) => *path = Some(saved), Err(e) => { let notice = result.output_notice.get_or_insert_with(String::new); notice.push_str(&format!( "{stream} archive unavailable: {e}; command completed, do not replay. " )); } } } } let bytes = serde_json::to_vec_pretty(&result)?; match retention::save("box", id, &bytes) { Ok(path) => result.output_log = Some(path), Err(e) => eprintln!( "box {id} output archive failed: {e}; command completed, full result returned; do not replay" ), } Ok(result) } fn checked(h: &str, id: &str, cmd: Value) -> Result { let r = execute(h, id, cmd, None, 300)?; if r.exit != 0 || r.timed_out || r.service_truncated { return Err(error(format!( "box command failed (exit {}, timeout {}, truncated {}): {}; {}", r.exit, r.timed_out, r.truncated, r.stderr, r.output_notice.as_deref().unwrap_or("") ))); } if let Some(path) = r.stdout_path { return Ok(std::fs::read_to_string(path)?); } // Archival failure is not execution failure. Helpers can use complete inline // text (especially empty mutation output); never parse a cut/binary excerpt. if r.exec_id.is_some() && (r.stdout_bytes != Some(r.stdout.len() as u64) || r.stdout.contains('\u{fffd}')) { return Err(error(format!( "command completed; required stdout unavailable; {}; do not replay", r.output_notice .as_deref() .unwrap_or("incomplete inline stdout") ))); } Ok(r.stdout) } // The person explicitly approved bounded polling within create/node_box because // this service has no readiness events. No worker or polling survives the RPC. #[derive(Clone, Copy)] pub struct Readiness { timeout_ns: u64, step_ns: u64, max_sleep_ns: u64, } const READINESS: Readiness = Readiness { timeout_ns: 240_000_000_000, step_ns: 1_000_000_000, max_sleep_ns: 3_000_000_000, }; fn accepted_error(mut e: Error, id: &str, h: &str) -> Error { e.message = format!( "box {id} on {h}: {}; continue with box.get, do not retry create", e.message ); e.data = Some(json!({"id":id,"host":h})); e } fn wait_ready(h: &str, mut c: Container, timing: Readiness, deadline: u64) -> Result { let mut sleep_ns = timing.step_ns; loop { match c.state.as_str() { "running" => return Ok(c), "failed" => { // Preserve terminal metadata even if the log endpoint is unavailable. let tail = archive::build(h, &c.id, deadline); match tail { Ok(log) => { c.build_log_tail = Some(log.text); c.build_log_path = Some(log.path); c.build_log_bytes = Some(log.bytes); c.build_log_service_truncated = Some(log.service_truncated); c.build_log_notice = log.notice; } Err(e) => { c.build_log_tail = Some(format!("build log unavailable: {}", e.message)) } } return Ok(c); } _ => {} } let now = http::now(); if now >= deadline { return Err(error("readiness timed out after 240 s")); } wasi::clocks::monotonic_clock::subscribe_instant( now.saturating_add(sleep_ns).min(deadline), ) .block(); if http::now() >= deadline { return Err(error("readiness timed out after 240 s")); } c = http::json_until(h, Method::Get, &base(&c.id)?, None, 30, Some(deadline))?; sleep_ns = sleep_ns .saturating_add(timing.step_ns) .min(timing.max_sleep_ns); } } impl BoxPlugin { pub fn create_box(&self, p: Create) -> Result { self.create_box_with_wait(p, READINESS) } fn create_box_with_wait(&self, p: Create, timing: Readiness) -> Result { let n = usize::from(p.config.flake.is_some()) + usize::from(p.config.nixos.is_some()) + usize::from(p.config.system.is_some()); if n != 1 || p.ttl_secs == Some(0) { return Err(error( "provide exactly one config source and a positive TTL", )); } let h = host(p.host.as_deref())?; let deadline = http::now().saturating_add(timing.timeout_ns); let c: Container = http::json_until( &h, Method::Post, "/containers", Some( json!({"config":p.config,"ttl_secs":p.ttl_secs,"owner":p.owner.unwrap_or_default()}), ), 30, Some(deadline), )?; let id = c.id.clone(); remember(&id, &h).map_err(|e| accepted_error(e, &id, &h))?; eprintln!("box {id} accepted on {h}; waiting up to 240 s for readiness"); wait_ready(&h, c, timing, deadline).map_err(|e| accepted_error(e, &id, &h)) } } impl BoxPlugin { /// Create a box and wait up to 240 seconds for running/failed. Timeout errors carry id/host for box.get; never retry create blindly. pub fn create(&self, p: Create) -> Result { self.create_box(p) } /// List boxes known to one service host. pub fn list(&self, p: List) -> Result { http::json( &host(p.host.as_deref())?, Method::Get, "/containers", None, 30, ) } /// Fetch current metadata; expired/deleted IDs yield a clear service 404. pub fn get(&self, p: Id) -> Result { get(&resolve(&p.id, p.host.as_deref())?, &p.id) } /// Fetch and archive the full build log; old services fall back to their tail with a warning. pub fn log(&self, p: Id) -> Result { log(&resolve(&p.id, p.host.as_deref())?, &p.id) } /// Reset remaining lifetime to ttl_secs from now. pub fn extend(&self, p: Extend) -> Result { if p.ttl_secs == 0 { return Err(error("ttl_secs must be positive")); } http::json( &resolve(&p.id, p.host.as_deref())?, Method::Post, &format!("{}/extend", base(&p.id)?), Some(json!({"ttl_secs":p.ttl_secs})), 30, ) } /// Destroy a box. Unknown IDs error even though upstream delete is idempotent. pub fn delete(&self, p: Id) -> Result { let h = resolve(&p.id, p.host.as_deref())?; get(&h, &p.id)?; // Best-effort byte preservation must never make explicit cleanup impossible. if let Err(e) = log(&h, &p.id) { eprintln!( "box {} build log not saved before deletion: {}", p.id, e.message ); } http::json(&h, Method::Delete, &base(&p.id)?, None, 30) } /// Run a shell string or argv command as root inside the box. pub fn exec(&self, p: Exec) -> Result { match &p.cmd { Command::Shell(s) if s.is_empty() => return Err(error("cmd must not be empty")), Command::Argv(a) if a.is_empty() => return Err(error("argv must not be empty")), _ => {} } let mut result = execute( &resolve(&p.id, p.host.as_deref())?, &p.id, serde_json::to_value(p.cmd)?, p.stdin, p.timeout_secs.unwrap_or(60), )?; excerpt::apply( &mut result, p.inline_bytes.unwrap_or(excerpt::DEFAULT_BYTES), ); Ok(result) } /// Stream text or a world file to a box, enforcing the service's 64 MiB cap. pub fn put_file(&self, p: PutFile) -> Result { let h = resolve(&p.id, p.host.as_deref())?; let bytes = match (p.text, p.from_vfs) { (Some(text), None) => { transfer::put(&h, &p.id, &p.path, &mut text.as_bytes(), text.len() as u64)? } (None, Some(path)) => { let mut f = File::open(path)?; let len = f.metadata()?.len(); transfer::put(&h, &p.id, &p.path, &mut f, len)? } _ => return Err(error("provide exactly one of text and from_vfs")), }; Ok(FileResult { bytes, text: None, to_vfs: None, }) } /// Download into VFS without returning file contents; inline text is capped at 64 KiB. pub fn get_file(&self, p: GetFile) -> Result { transfer::get_file(&resolve(&p.id, p.host.as_deref())?, p) } /// Tar a world directory and stream it to the box. Merges into to_dir. pub fn put_tree(&self, p: PutTree) -> Result { transfer::put_tree(&resolve(&p.id, p.host.as_deref())?, p) } /// Stream a tar from a box and unpack into a new world directory. pub fn get_tree(&self, p: GetTree) -> Result { transfer::get_tree(&resolve(&p.id, p.host.as_deref())?, p) } /// Create a node/npm/git/Chromium/fontconfig box with Playwright environment defaults. pub fn node_box(&self, p: NodeBox) -> Result { self.create_box(Create { config: Config { flake: None, nixos: Some(Nixos::Text(NODE_MODULE.into())), system: None, }, ttl_secs: p.ttl_secs, owner: Some("box.node_box".into()), host: None, }) } } #[cfg(test)] mod tests { use super::*; #[test] fn retention_deadline_scales_without_a_byte_cap() { assert_eq!(archive_budget_ns(Some(0), Some(0)), 60_000_000_000); assert_eq!(archive_budget_ns(Some(256 * 1024), Some(1)), 62_000_000_000); assert_eq!(archive_budget_ns(None, Some(0)), 300_000_000_000); assert_eq!( archive_budget_ns(Some(u64::MAX), Some(u64::MAX)), 1_800_000_000_000 ); } }