diff --git a/PROTOCOL.md b/PROTOCOL.md index 312c692..9f0399b 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -1,7 +1,8 @@ # The Sausage protocol This document is the contract between the host and the actors. An actor in -any language that follows it works in Sausage. +any language that follows it works in Sausage. The host publishes this file at +`/actor/broker/public/PROTOCOL.md`, so every actor can read it. ## Actors @@ -43,10 +44,14 @@ A line on **stdout**: mailbox keeps messages while the actor restarts or is upgraded. - If the broker cannot deliver a message, the sender gets `{"type": "undeliverable", "reason": "...", "to": "..."}` from `broker`. +- An actor must keep reading stdin. If a message waits 30 seconds for the + actor to read it, the broker kills and restarts the actor, and the sender + gets `undeliverable`. Before the broker stops an instance, it sends `{"id": "broker/shutdown", "from": "broker", "body": {"type": "shutdown"}}` -and then closes stdin. After 5 seconds, it kills the instance. +and then closes stdin. The broker still routes what the actor writes until +it exits. After 5 seconds, it kills the instance. ## Filesystem @@ -93,7 +98,9 @@ sender's). has `"rejected"` in its `meta.json`. **Tests.** Before a spawn or an upgrade, the broker runs the candidate on a -copy of the actor's `/data` and `/public`, without HTTP. The tests are JSON +copy of the actor's `/data` and `/public` (empty for a new actor), without +HTTP. The candidate sees its future `SAUSAGE_VERSION` and +`SAUSAGE_CANDIDATE=1`. The tests are JSON files in `source/tests/*.json`: ```json diff --git a/cli/src/broker.rs b/cli/src/broker.rs index be929d1..cab4f01 100644 --- a/cli/src/broker.rs +++ b/cli/src/broker.rs @@ -241,9 +241,17 @@ impl Broker { let dir = self.world.version_dir(id, meta.version); std::fs::create_dir_all(&dir)?; std::fs::copy(wasm, dir.join("actor.wasm")).with_context(|| format!("copy {}", wasm.display()))?; - world::replace_tree(source.unwrap_or(&dir.join("no-source")), &dir.join("source"))?; + world::replace_source(source.unwrap_or(&dir.join("no-source")), &dir.join("source"))?; world::write_atomic(&dir.join("meta.json"), serde_json::to_string_pretty(meta)?.as_bytes())?; - self.runtime.load(&dir.join("actor.wasm")) + self.load_version(id, meta.version) + } + + /// Compile `versions//actor.wasm`. + pub fn load_version(&self, id: &str, version: u64) -> Result { + let started = Instant::now(); + let component = self.runtime.load(&self.world.version_dir(id, version).join("actor.wasm"))?; + self.log(id, &format!("compiled version {version} in {} ms", started.elapsed().as_millis())); + Ok(component) } /// Record in `versions//meta.json` why the broker refused a version. @@ -282,11 +290,11 @@ impl Broker { if !world::valid_id(id) { bail!("invalid actor ID {id:?}: use a-z, 0-9, '-' and '_'"); } + let lock = self.op_lock(id); + let _guard = lock.lock().await; if NATIVE.contains(&id) || self.is_supervised(id) || self.world.wasm_path(id).exists() { bail!("an actor with ID {id} already exists"); } - let lock = self.op_lock(id); - let _guard = lock.lock().await; let version = self.world.next_version(id); let meta = VersionMeta { version, ..meta }; self.create_version(id, version, &meta, wasm, source, start).await @@ -295,7 +303,8 @@ impl Broker { async fn create_version(self: &Arc, id: &str, version: u64, meta: &VersionMeta, wasm: &Path, source: Option<&Path>, start: bool) -> Result { let tested = async { let component = self.store_version(id, meta, wasm, source)?; - candidate::test(self, id, &component, &self.world.version_dir(id, version).join("source"), None, None).await + let source = self.world.version_dir(id, version).join("source"); + candidate::test(self, id, version, &component, &source, &self.world.data_dir(id), &self.world.public_dir(id)).await } .await; let report = match tested { @@ -328,7 +337,7 @@ impl Broker { let tested = async { let component = self.store_version(id, meta, wasm, source)?; let source = self.world.version_dir(id, version).join("source"); - candidate::test(self, id, &component, &source, Some(&self.world.data_dir(id)), Some(&self.world.public_dir(id))).await + candidate::test(self, id, version, &component, &source, &self.world.data_dir(id), &self.world.public_dir(id)).await } .await; let report = match tested { @@ -405,10 +414,13 @@ impl Broker { let source = source.is_dir().then_some(source.as_path()); let result = match installed { None => self.create_version(id, version, &vmeta, &wasm, source, true).await, - Some(current) => { + Some(_) => { let r = self.upgrade_version(id, &vmeta, &wasm, source).await; - if r.is_err() { - // Put the installed version back in place. + // Put the installed version back in place. Read it again: an + // automatic rollback can have changed it meanwhile. + if r.is_err() + && let Some(current) = self.meta(id).and_then(|m| m.version) + { let _ = self.restore_files(id, current); } r @@ -527,10 +539,18 @@ impl Broker { } } - /// The native `broker` actor. + /// The native `broker` actor. Requests run concurrently. Aborting this + /// task aborts the requests that still run. pub async fn run_actor(self: Arc, mut mailbox: mpsc::Receiver) { - while let Some(env) = mailbox.recv().await { - tokio::spawn(self.clone().handle(env)); + let mut requests = tokio::task::JoinSet::new(); + loop { + tokio::select! { + env = mailbox.recv() => match env { + Some(env) => { requests.spawn(self.clone().handle(env)); } + None => break, + }, + Some(_) = requests.join_next() => {} + } } } } @@ -586,6 +606,8 @@ struct Supervisor { swapped_at: Option, crashes_since_swap: usize, upgrade_requester: Option, + /// A control request that arrived during a blocked delivery. + next_control: Option, } async fn recv_line(instance: &mut Option) -> Option { @@ -632,6 +654,7 @@ impl Supervisor { swapped_at: None, crashes_since_swap: 0, upgrade_requester: None, + next_control: None, } } @@ -658,9 +681,15 @@ impl Supervisor { async fn run(mut self) { self.status(); loop { + if let Some(ctl) = self.next_control.take() { + if self.handle_control(ctl).await { + break; + } + continue; + } if self.desired_running && self.instance.is_none() && self.restart_at.is_none_or(|t| Instant::now() >= t) { self.restart_at = None; - self.start_instance().await; + let _ = self.start_instance().await; } let has_instance = self.instance.is_some(); let waiting = !has_instance && self.desired_running; @@ -690,36 +719,65 @@ impl Supervisor { } } + /// Write a message to the instance's stdin, and route its output + /// meanwhile. An instance that does not read its stdin for + /// `STDIN_WRITE_TIMEOUT`, or while a control request waits, is killed and + /// restarted: a partly written line would corrupt the next message. async fn deliver(&mut self, msg: Envelope) { let Some(inst) = self.instance.as_mut() else { return }; - let line = msg.to_line(); - match tokio::time::timeout(STDIN_WRITE_TIMEOUT, inst.send_line(&line)).await { - Ok(Ok(())) => self.last_delivered = Some(msg), - Ok(Err(_)) => {} // The instance is ending. The exit path handles it. - Err(_) => { - let note = format!("actor did not read stdin for {STDIN_WRITE_TIMEOUT:?}; message {} dropped", msg.id); - self.broker.log(&self.id, ¬e); + let (broker, id, line) = (self.broker.clone(), self.id.clone(), msg.to_line()); + let write = tokio::time::timeout(STDIN_WRITE_TIMEOUT, inst.send_line(&line, |l| broker.route_line(&id, &l))); + let failure = tokio::select! { + written = write => match written { + Ok(Ok(())) => { + self.last_delivered = Some(msg); + return; + } + // The instance is ending. The main loop sees its closed stdout. + Ok(Err(_)) => None, + Err(_) => Some(format!("the actor did not read stdin for {STDIN_WRITE_TIMEOUT:?}")), + }, + Some(ctl) = self.control.recv() => { + self.next_control = Some(ctl); + Some("the actor did not read stdin when the broker needed it".to_string()) } + }; + if msg.from != "broker" { + let reason = failure.as_deref().unwrap_or("the actor stopped"); + let body = json!({"type": "undeliverable", "reason": reason, "to": self.id}); + self.broker.send("broker", Some(&msg.from), None, Some(&msg.id), body); + } + if let (Some(failure), Some(inst)) = (failure, &self.instance) { + // The main loop sees the end of the instance and restarts it. + self.broker.log(&self.id, &format!("killed: {failure}")); + inst.kill(); } } - async fn start_instance(&mut self) { + /// Start an instance of the current version. On failure, the actor is + /// "broken" until the next start, upgrade, or rollback. + async fn start_instance(&mut self) -> Result<()> { + let started = self.try_start(); + if let Err(e) = &started { + self.broker.log(&self.id, &format!("cannot start version {}: {e:#}", self.version)); + self.state = "broken"; + self.desired_running = false; + self.status(); + return started; + } + self.state = "running"; + self.started_at = Instant::now(); + self.status(); + if let Some(msg) = self.redeliver.take() { + self.broker.log(&self.id, &format!("delivering parked message {} again", msg.id)); + self.deliver(msg).await; + } + Ok(()) + } + + fn try_start(&mut self) -> Result<()> { if self.component.is_none() { - let path = self.world().version_dir(&self.id, self.version).join("actor.wasm"); - let t = Instant::now(); - match self.broker.runtime.load(&path) { - Ok(c) => { - self.broker.log(&self.id, &format!("compiled version {} in {} ms", self.version, t.elapsed().as_millis())); - self.component = Some(c); - } - Err(e) => { - self.broker.log(&self.id, &format!("cannot load version {}: {e:#}", self.version)); - self.state = "broken"; - self.desired_running = false; - self.status(); - return; - } - } + self.component = Some(self.broker.load_version(&self.id, self.version)?); } let cfg = InstanceConfig { mounts: self.broker.mounts(&self.id), @@ -730,29 +788,15 @@ impl Supervisor { ("SAUSAGE_VERSION".into(), self.version.to_string()), ], }; - match self.broker.runtime.start(self.component.as_ref().unwrap(), cfg) { - Ok(inst) => { - self.instance = Some(inst); - self.state = "running"; - self.started_at = Instant::now(); - self.status(); - if let Some(msg) = self.redeliver.take() { - self.broker.log(&self.id, &format!("delivering parked message {} again", msg.id)); - self.deliver(msg).await; - } - } - Err(e) => { - self.broker.log(&self.id, &format!("cannot start: {e:#}")); - self.state = "broken"; - self.desired_running = false; - self.status(); - } - } + self.instance = Some(self.broker.runtime.start(self.component.as_ref().unwrap(), cfg)?); + Ok(()) } + /// Stop the instance gracefully. Its last messages still get routed. async fn stop_instance(&mut self) -> Option { let inst = self.instance.take()?; - let exit = inst.stop(&shutdown_line(), SHUTDOWN_DEADLINE).await; + let (broker, id) = (self.broker.clone(), self.id.clone()); + let exit = inst.stop(&shutdown_line(), SHUTDOWN_DEADLINE, |line| broker.route_line(&id, &line)).await; self.broker.log(&self.id, &format!("stopped: {exit}")); self.last_exit = Some(exit.to_string()); Some(exit) @@ -801,37 +845,48 @@ impl Supervisor { Ok(()) } - /// Make `version` the current files: `actor.wasm`, `source/`, `actor.json`. + /// Make `version` current: compile it first, then write `actor.wasm`, + /// `source/`, and `actor.json`. If compiling fails, nothing changes. fn install(&mut self, version: u64) -> Result<()> { + let component = self.broker.load_version(&self.id, version)?; let dir = self.world().version_dir(&self.id, version); world::write_atomic(&self.world().wasm_path(&self.id), &std::fs::read(dir.join("actor.wasm"))?)?; world::replace_tree(&dir.join("source"), &self.world().source_dir(&self.id))?; - self.version = version; - self.component = None; self.broker.update_meta(&self.id, |m| { m.version = Some(version); m.running = true; - }) + })?; + self.version = version; + self.component = Some(component); + Ok(()) } fn set_running(&self, running: bool) -> Result<()> { self.broker.update_meta(&self.id, |m| m.running = running) } + /// Clear the restart and crash-loop state before a swap. + fn reset_restarts(&mut self) { + self.restarts = 0; + self.restart_at = None; + self.crashes_since_swap = 0; + self.desired_running = true; + } + async fn upgrade(&mut self, version: u64, requester: String) -> Result { self.stop_instance().await; self.snapshot_current()?; let previous = self.version; self.install(version)?; + self.reset_restarts(); self.previous = Some(previous); self.swapped_at = Some(Instant::now()); - self.crashes_since_swap = 0; - self.restarts = 0; - self.restart_at = None; self.upgrade_requester = Some(requester); self.redeliver = self.parked.take(); - self.desired_running = true; - self.start_instance().await; + if let Err(e) = self.start_instance().await { + self.rollback(Some(previous)).await?; + bail!("version {version} did not start ({e:#}); rolled back to {previous}"); + } Ok(json!({"type": "upgraded", "actor": self.id, "version": version, "previous": previous})) } @@ -847,6 +902,7 @@ impl Supervisor { if !dir.join("actor.wasm").is_file() { bail!("{} has no version {target}", self.id); } + self.broker.load_version(&self.id, target)?; self.stop_instance().await; self.snapshot_current()?; let from = self.version; @@ -856,12 +912,10 @@ impl Supervisor { world::replace_tree(&snap.join("public"), &self.world().public_dir(&self.id))?; } self.install(target)?; + self.reset_restarts(); self.previous = None; self.swapped_at = None; - self.restarts = 0; - self.restart_at = None; - self.desired_running = true; - self.start_instance().await; + self.start_instance().await?; Ok(json!({"type": "rolled_back", "actor": self.id, "from": from, "to": target, "restored_state": snap.is_dir()})) } @@ -894,8 +948,12 @@ impl Supervisor { self.desired_running = false; self.stop_instance().await; let r = self.archive(); + let removed = r.is_ok(); let _ = reply.send(r); - return true; + if removed { + return true; + } + self.state = "stopped"; } Control::Shutdown { reply } => { self.stop_instance().await; @@ -913,7 +971,8 @@ impl Supervisor { fn archive(&self) -> Result { let archive = self.world().root.join("archive"); std::fs::create_dir_all(&archive)?; - let dest = archive.join(format!("{}-{}", self.id, world::now())); + let nanos = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_nanos(); + let dest = archive.join(format!("{}-{nanos}", self.id)); std::fs::rename(self.world().actor_dir(&self.id), &dest)?; Ok(json!({"type": "removed", "actor": self.id, "archive": dest.display().to_string()})) } diff --git a/cli/src/candidate.rs b/cli/src/candidate.rs index e6a4eda..a479d5d 100644 --- a/cli/src/candidate.rs +++ b/cli/src/candidate.rs @@ -18,7 +18,7 @@ //! also go to that recipient. //! //! Each file runs in a fresh instance on a copy of the actor's `/data` and -//! `/public`. The candidate has no HTTP access. Without test files, the +//! `/public` (empty folders for a new actor). The candidate has no HTTP access. Without test files, the //! candidate must start and stay up for `SMOKE_TIME`. use std::path::Path; @@ -66,14 +66,26 @@ fn stderr_tail(path: &Path) -> String { text[s..].to_string() } +/// The version under test and the state that its tests start from. +struct Candidate<'a> { + broker: &'a Broker, + id: &'a str, + version: u64, + component: &'a Component, + data: &'a Path, + public: &'a Path, +} + pub async fn test( broker: &Broker, id: &str, + version: u64, component: &Component, source: &Path, - data: Option<&Path>, - public: Option<&Path>, + data: &Path, + public: &Path, ) -> Result { + let candidate = Candidate { broker, id, version, component, data, public }; let mut files: Vec<_> = std::fs::read_dir(source.join("tests")) .map(|rd| { rd.filter_map(|e| e.ok()) @@ -85,7 +97,7 @@ pub async fn test( files.sort(); if files.is_empty() { - let (mut inst, scratch) = start(broker, id, component, data, public)?; + let (mut inst, scratch) = candidate.start()?; tokio::time::sleep(SMOKE_TIME).await; if inst.is_finished() { let exit = inst.wait().await; @@ -94,94 +106,93 @@ pub async fn test( stderr_tail(&scratch.dir.join("stderr.log")) ); } - inst.stop(&shutdown_line(), SHUTDOWN_DEADLINE).await; + inst.stop(&shutdown_line(), SHUTDOWN_DEADLINE, |_| {}).await; return Ok(json!({"smoke_test": "passed", "files": 0})); } let mut passed = Vec::new(); for file in &files { let name = file.file_name().unwrap().to_string_lossy().to_string(); - run_file(broker, id, component, data, public, file) - .await - .map_err(|e| anyhow!("test {name} failed: {e:#}"))?; + candidate.run_file(file).await.map_err(|e| anyhow!("test {name} failed: {e:#}"))?; passed.push(name); } Ok(json!({"passed": passed})) } -fn start(broker: &Broker, id: &str, component: &Component, data: Option<&Path>, public: Option<&Path>) -> Result<(Instance, Scratch)> { - static N: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); - let n = N.fetch_add(1, std::sync::atomic::Ordering::Relaxed); - let dir = broker.world.scratch_dir().join(format!("{id}-{}-{n}", std::process::id())); - let scratch = Scratch { dir: dir.clone() }; - std::fs::create_dir_all(dir.join("data"))?; - std::fs::create_dir_all(dir.join("public"))?; - if let Some(d) = data.filter(|d| d.is_dir()) { - world::copy_tree(d, &dir.join("data"))?; - } - if let Some(p) = public.filter(|p| p.is_dir()) { - world::copy_tree(p, &dir.join("public"))?; - } - let cfg = InstanceConfig { - mounts: vec![ - Mount { guest: "/actor".into(), host: broker.world.actors_root(), writable: false }, - Mount { guest: "/public".into(), host: dir.join("public"), writable: true }, - Mount { guest: "/data".into(), host: dir.join("data"), writable: true }, - ], - allow_http: false, - stderr_log: dir.join("stderr.log"), - env: vec![("SAUSAGE_ACTOR".into(), id.to_string()), ("SAUSAGE_CANDIDATE".into(), "1".into())], - }; - Ok((broker.runtime.start(component, cfg)?, scratch)) -} - -async fn run_file( - broker: &Broker, - id: &str, - component: &Component, - data: Option<&Path>, - public: Option<&Path>, - file: &Path, -) -> Result<()> { - let test: Value = serde_json::from_str(&std::fs::read_to_string(file)?).map_err(|e| anyhow!("invalid JSON: {e}"))?; - let steps = test["steps"].as_array().ok_or_else(|| anyhow!("missing \"steps\" list"))?; - let (mut inst, scratch) = start(broker, id, component, data, public)?; - let stderr = scratch.dir.join("stderr.log"); - for (n, step) in steps.iter().enumerate() { - let n = n + 1; - if let Some(body) = step.get("send") { - let from = step["from"].as_str().unwrap_or("test"); - let env = json!({"id": format!("test/{n}"), "from": from, "body": body}); - inst.send_line(&env.to_string()) - .await - .map_err(|e| anyhow!("step {n}: candidate does not read stdin: {e}"))?; +impl Candidate<'_> { + /// Start an instance on a fresh copy of the state, without HTTP. + fn start(&self) -> Result<(Instance, Scratch)> { + static N: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + let n = N.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let dir = self.broker.world.scratch_dir().join(format!("{}-{}-{n}", self.id, std::process::id())); + let scratch = Scratch { dir: dir.clone() }; + for (from, name) in [(self.data, "data"), (self.public, "public")] { + std::fs::create_dir_all(dir.join(name))?; + if from.is_dir() { + world::copy_tree(from, &dir.join(name))?; + } } - let expected = match step.get("expect") { - None => vec![], - Some(Value::Array(list)) => list.clone(), - Some(one) => vec![one.clone()], + let cfg = InstanceConfig { + mounts: vec![ + Mount { guest: "/actor".into(), host: self.broker.world.actors_root(), writable: false }, + Mount { guest: "/public".into(), host: dir.join("public"), writable: true }, + Mount { guest: "/data".into(), host: dir.join("data"), writable: true }, + ], + allow_http: false, + stderr_log: dir.join("stderr.log"), + env: vec![ + ("SAUSAGE_ACTOR".into(), self.id.to_string()), + ("SAUSAGE_VERSION".into(), self.version.to_string()), + ("SAUSAGE_CANDIDATE".into(), "1".into()), + ], }; - for expected in expected { - let line = match tokio::time::timeout(EXPECT_TIMEOUT, inst.stdout.recv()).await { - Err(_) => bail!("step {n}: no message within {EXPECT_TIMEOUT:?}; expected {expected}"), - Ok(None) => { - let exit = inst.wait().await; - bail!("step {n}: candidate ended ({exit}); expected {expected}. stderr:\n{}", stderr_tail(&stderr)) - } - Ok(Some(l)) => l, - }; - let msg: Value = serde_json::from_str(&line) - .map_err(|_| anyhow!("step {n}: candidate wrote a non-JSON line: {}", truncate(&line, 300)))?; - if !subset_match(&expected, &msg["body"]) { - bail!("step {n}: expected body {expected}, got {}", truncate(&msg["body"].to_string(), 1000)); + Ok((self.broker.runtime.start(self.component, cfg)?, scratch)) + } + + async fn run_file(&self, file: &Path) -> Result<()> { + let test: Value = serde_json::from_str(&std::fs::read_to_string(file)?).map_err(|e| anyhow!("invalid JSON: {e}"))?; + let steps = test["steps"].as_array().ok_or_else(|| anyhow!("missing \"steps\" list"))?; + let (mut inst, scratch) = self.start()?; + let stderr = scratch.dir.join("stderr.log"); + // Lines that the candidate wrote while a step was sent. + let mut early = std::collections::VecDeque::new(); + for (n, step) in steps.iter().enumerate() { + let n = n + 1; + if let Some(body) = step.get("send") { + let from = step["from"].as_str().unwrap_or("test"); + let env = json!({"id": format!("test/{n}"), "from": from, "body": body}); + inst.send_line(&env.to_string(), |line| early.push_back(line)) + .await + .map_err(|e| anyhow!("step {n}: candidate does not read stdin: {e}"))?; } - if let Some(to) = step.get("to") - && msg.get("to") != Some(to) - { - bail!("step {n}: expected recipient {to}, got {}", msg.get("to").unwrap_or(&Value::Null)); + let expected = match step.get("expect") { + None => vec![], + Some(Value::Array(list)) => list.clone(), + Some(one) => vec![one.clone()], + }; + for expected in expected { + let next = async { if early.is_empty() { inst.stdout.recv().await } else { early.pop_front() } }; + let line = match tokio::time::timeout(EXPECT_TIMEOUT, next).await { + Err(_) => bail!("step {n}: no message within {EXPECT_TIMEOUT:?}; expected {expected}"), + Ok(None) => { + let exit = inst.wait().await; + bail!("step {n}: candidate ended ({exit}); expected {expected}. stderr:\n{}", stderr_tail(&stderr)) + } + Ok(Some(l)) => l, + }; + let msg: Value = serde_json::from_str(&line) + .map_err(|_| anyhow!("step {n}: candidate wrote a non-JSON line: {}", truncate(&line, 300)))?; + if !subset_match(&expected, &msg["body"]) { + bail!("step {n}: expected body {expected}, got {}", truncate(&msg["body"].to_string(), 1000)); + } + if let Some(to) = step.get("to") + && msg.get("to") != Some(to) + { + bail!("step {n}: expected recipient {to}, got {}", msg.get("to").unwrap_or(&Value::Null)); + } } } + inst.stop(&shutdown_line(), SHUTDOWN_DEADLINE, |_| {}).await; + Ok(()) } - inst.stop(&shutdown_line(), SHUTDOWN_DEADLINE).await; - Ok(()) } diff --git a/cli/src/cargo_actor.rs b/cli/src/cargo_actor.rs index 1e423c5..01610fb 100644 --- a/cli/src/cargo_actor.rs +++ b/cli/src/cargo_actor.rs @@ -18,7 +18,7 @@ use std::time::Instant; use anyhow::{Context, Result, anyhow, bail}; use serde_json::{Value, json}; use tokio::process::Command; -use tokio::sync::{Mutex, mpsc}; +use tokio::sync::mpsc; use include_dir::{Dir, include_dir}; @@ -38,17 +38,16 @@ pub struct BuildOutput { pub warnings: String, } +/// All builds share one target folder, so they run one at a time. pub struct Cargo { world: World, - /// One build at a time: all builds share one target folder. - lock: Mutex<()>, } impl Cargo { /// Create the actor and install the SDK of this host into `data/sdk/`. pub fn new(world: World) -> Result { write_dir(&SDK, &world.data_dir(ID).join("sdk"))?; - Ok(Self { world, lock: Mutex::new(()) }) + Ok(Self { world }) } /// Build the crate in `source` (a host path) and publish the `.wasm` file @@ -57,12 +56,11 @@ impl Cargo { if !source.join("Cargo.toml").is_file() { bail!("{} has no Cargo.toml", source.display()); } - let _guard = self.lock.lock().await; let started = Instant::now(); let data = self.world.data_dir(ID); let key = format!("{:016x}", fxhash(source.to_string_lossy().as_bytes())); let build_dir = data.join("work").join(&key); - world::replace_tree(source, &build_dir)?; + world::replace_source(source, &build_dir)?; let manifest = build_dir.join("Cargo.toml"); std::fs::write(&manifest, prepare_manifest(&std::fs::read_to_string(&manifest)?, Some(&data.join("sdk")))?)?; @@ -100,16 +98,13 @@ impl Cargo { Ok(BuildOutput { wasm: dest, seconds: started.elapsed().as_secs_f64(), warnings: strip_progress(&stderr) }) } - pub async fn run_actor(self: Arc, broker: Arc, mut mailbox: mpsc::Receiver) { + /// The native `cargo` actor. Builds run one at a time, in order. + pub async fn run_actor(self, broker: Arc, mut mailbox: mpsc::Receiver) { while let Some(env) = mailbox.recv().await { - let this = self.clone(); - let broker = broker.clone(); - tokio::spawn(async move { - let body = this.handle(&broker, &env).await.unwrap_or_else(|e| { - json!({"type": "build_failed", "source": env.body.get("source"), "errors": format!("{e:#}")}) - }); - broker.send(ID, Some(&env.from), None, Some(&env.id), body); + let body = self.handle(&broker, &env).await.unwrap_or_else(|e| { + json!({"type": "build_failed", "source": env.body.get("source"), "errors": format!("{e:#}")}) }); + broker.send(ID, Some(&env.from), None, Some(&env.id), body); } } diff --git a/cli/src/fs.rs b/cli/src/fs.rs index 2f0d4bf..33173c7 100644 --- a/cli/src/fs.rs +++ b/cli/src/fs.rs @@ -316,6 +316,8 @@ impl HostDescriptor for ActorFs<'_> { None => tmp_name.clone(), }; let tmp = target.with_file_name(&tmp_name); + // The temporary file is always new, also for `truncate` without `create`. + let oflags = oflags | types::OpenFlags::CREATE; let r = HostDescriptor::open_at(&mut self.inner, fd, path_flags, tmp_rel, oflags, flags).await?; self.ext.pending.insert(r.rep(), Pending { tmp, target }); return Ok(r); @@ -363,7 +365,25 @@ impl HostDescriptor for ActorFs<'_> { if is_tmp_name(&old_path) || is_tmp_name(&new_path) { return Err(ErrorCode::NoEntry.into()); } - HostDescriptor::rename_at(&mut self.inner, fd, old_path, new_fd, new_path).await + let host = |ext: &FsExt, fd: u32, path: &str| ext.dirs.get(&fd).and_then(|d| resolve(d, path)); + let moved = host(self.ext, fd.rep(), &old_path).zip(host(self.ext, new_fd.rep(), &new_path)); + HostDescriptor::rename_at(&mut self.inner, fd, old_path, new_fd, new_path).await?; + // Open writes and directories below a renamed folder move with it. + if let Some((from, to)) = moved { + let rebase = |p: &mut PathBuf| { + if let Ok(rest) = p.strip_prefix(&from) { + *p = to.join(rest); + } + }; + for pending in self.ext.pending.values_mut() { + rebase(&mut pending.tmp); + rebase(&mut pending.target); + } + for dir in self.ext.dirs.values_mut() { + rebase(&mut dir.host); + } + } + Ok(()) } async fn symlink_at( diff --git a/cli/src/lib.rs b/cli/src/lib.rs index 2cfa971..de6598e 100644 --- a/cli/src/lib.rs +++ b/cli/src/lib.rs @@ -20,10 +20,15 @@ use tokio::sync::mpsc; pub use broker::{Broker, Envelope}; pub use world::World; +/// The contract between the host and the actors. The host publishes it at +/// `/actor/broker/public/PROTOCOL.md`, so that every actor can read it. +const PROTOCOL: &str = include_str!("../../PROTOCOL.md"); + /// A running world. pub struct Host { pub world: World, pub broker: Arc, + native: Vec>, _lock: std::fs::File, } @@ -40,16 +45,17 @@ impl Host { bail!("another sausage process runs {}", root.display()); } let broker = Broker::new(world.clone(), runtime::Runtime::new()?)?; + world::write_atomic(&world.public_dir("broker").join("PROTOCOL.md"), PROTOCOL.as_bytes())?; - let rx = broker.open_mailbox("broker"); - tokio::spawn(broker.clone().run_actor(rx)); - let rx = broker.open_mailbox(cargo_actor::ID); - tokio::spawn(Arc::new(cargo_actor::Cargo::new(world.clone())?).run_actor(broker.clone(), rx)); - + let cargo = cargo_actor::Cargo::new(world.clone())?; + let native = vec![ + tokio::spawn(broker.clone().run_actor(broker.open_mailbox("broker"))), + tokio::spawn(cargo.run_actor(broker.clone(), broker.open_mailbox(cargo_actor::ID))), + ]; for change in broker.scan().await { eprintln!("sausage: {change}"); } - Ok(Host { world, broker, _lock: lock }) + Ok(Host { world, broker, native, _lock: lock }) } /// Register a native actor with the ID `id` and return its mailbox. @@ -57,8 +63,15 @@ impl Host { self.broker.open_mailbox(id) } - /// Stop every actor gracefully. + /// Stop the broker and `cargo` (with the requests they still run), then + /// every actor, gracefully. pub async fn stop(self) { + for task in &self.native { + task.abort(); + } + for task in self.native { + let _ = task.await; + } self.broker.shutdown_all().await; } } diff --git a/cli/src/runtime.rs b/cli/src/runtime.rs index fb050ea..7986123 100644 --- a/cli/src/runtime.rs +++ b/cli/src/runtime.rs @@ -131,19 +131,42 @@ pub struct InstanceConfig { pub env: Vec<(String, String)>, } +/// A running guest. Dropping it kills the guest. pub struct Instance { - stdin: DuplexStream, + stdin: Option, /// Lines that the guest writes to stdout. Closed when the guest ends. pub stdout: mpsc::Receiver, task: JoinHandle, } +impl Drop for Instance { + fn drop(&mut self) { + self.task.abort(); + } +} + impl Instance { - /// Write one line to the guest's stdin. - pub async fn send_line(&mut self, line: &str) -> std::io::Result<()> { - self.stdin.write_all(line.as_bytes()).await?; - self.stdin.write_all(b"\n").await?; - self.stdin.flush().await + /// Write one line to the guest's stdin, in one write. Meanwhile, pass the + /// lines that the guest writes to `route`: a guest that blocks on a full + /// stdout would never read its stdin. + pub async fn send_line(&mut self, line: &str, mut route: impl FnMut(String)) -> std::io::Result<()> { + let stdin = self.stdin.as_mut().ok_or(std::io::ErrorKind::BrokenPipe)?; + let bytes = format!("{line}\n"); + let write = async { + stdin.write_all(bytes.as_bytes()).await?; + stdin.flush().await + }; + tokio::pin!(write); + let mut open = true; + loop { + tokio::select! { + done = &mut write => return done, + out = self.stdout.recv(), if open => match out { + Some(out) => route(out), + None => open = false, + }, + } + } } /// Wait until the guest ends. @@ -159,19 +182,35 @@ impl Instance { self.task.is_finished() } - /// Send `shutdown`, close stdin, and wait. Kill the guest after `deadline`. - pub async fn stop(mut self, shutdown_line: &str, deadline: Duration) -> Exit { - let _ = self.send_line(shutdown_line).await; - let _ = self.stdin.shutdown().await; - drop(self.stdin); - match tokio::time::timeout(deadline, &mut self.task).await { - Ok(Ok(exit)) => exit, - Ok(Err(e)) => Exit::Trap(format!("host task failed: {e}")), - Err(_) => { - self.task.abort(); - let _ = (&mut self.task).await; - Exit::Trap("killed after shutdown deadline".into()) - } + /// Kill the guest. Its stdout closes when it has ended. + pub fn kill(&self) { + self.task.abort(); + } + + /// Send `shutdown_line`, close stdin, and wait until the guest ends. + /// Pass every line that the guest writes meanwhile to `route`. Kill the + /// guest when `deadline` has passed. + pub async fn stop(mut self, shutdown_line: &str, deadline: Duration, mut route: impl FnMut(String)) -> Exit { + let stdin = self.stdin.take(); + let stdout = &mut self.stdout; + let finished = tokio::time::timeout(deadline, async { + let close = async { + if let Some(mut stdin) = stdin { + let _ = stdin.write_all(format!("{shutdown_line}\n").as_bytes()).await; + let _ = stdin.shutdown().await; + } + }; + let drain = async { + while let Some(line) = stdout.recv().await { + route(line); + } + }; + tokio::join!(close, drain); + }) + .await; + match finished { + Ok(()) => self.wait().await, + Err(_) => Exit::Trap("killed after shutdown deadline".into()), } } } @@ -299,6 +338,6 @@ impl Runtime { exit }); - Ok(Instance { stdin: host_in, stdout: line_rx, task }) + Ok(Instance { stdin: Some(host_in), stdout: line_rx, task }) } } diff --git a/cli/src/world.rs b/cli/src/world.rs index 1c78d1a..ffadd91 100644 --- a/cli/src/world.rs +++ b/cli/src/world.rs @@ -227,8 +227,15 @@ pub fn write_atomic(path: &Path, bytes: &[u8]) -> Result<()> { Ok(()) } -/// Copy a directory tree. Skips `target/` folders and uncommitted files. +/// Folders that a copy of a source folder leaves out. +const BUILD_DIRS: &[&str] = &["target", ".git"]; + +/// Copy a directory tree, except uncommitted files. pub fn copy_tree(from: &Path, to: &Path) -> Result<()> { + copy_filtered(from, to, &[]) +} + +fn copy_filtered(from: &Path, to: &Path, skip: &[&str]) -> Result<()> { std::fs::create_dir_all(to)?; for entry in std::fs::read_dir(from).with_context(|| format!("read {}", from.display()))? { let entry = entry?; @@ -239,8 +246,8 @@ pub fn copy_tree(from: &Path, to: &Path) -> Result<()> { } let dst = to.join(&*name); let ty = entry.file_type()?; - if ty.is_dir() && name != "target" && name != ".git" { - copy_tree(&entry.path(), &dst)?; + if ty.is_dir() && !skip.contains(&&*name) { + copy_filtered(&entry.path(), &dst, skip)?; } else if ty.is_file() { std::fs::copy(entry.path(), &dst)?; } @@ -251,6 +258,15 @@ pub fn copy_tree(from: &Path, to: &Path) -> Result<()> { /// Replace the directory `dst` with a copy of `src` (an empty folder if `src` /// does not exist): copy to a sibling, then swap with two renames. pub fn replace_tree(src: &Path, dst: &Path) -> Result<()> { + replace_filtered(src, dst, &[]) +} + +/// Like `replace_tree`, for a source folder: leaves out `target/` and `.git/`. +pub fn replace_source(src: &Path, dst: &Path) -> Result<()> { + replace_filtered(src, dst, BUILD_DIRS) +} + +fn replace_filtered(src: &Path, dst: &Path, skip: &[&str]) -> Result<()> { let parent = dst.parent().context("path has no parent")?; let name = dst.file_name().and_then(|n| n.to_str()).unwrap_or("dir"); let fresh = parent.join(format!("{TMP_PREFIX}new-{name}")); @@ -258,7 +274,7 @@ pub fn replace_tree(src: &Path, dst: &Path) -> Result<()> { let _ = std::fs::remove_dir_all(&fresh); let _ = std::fs::remove_dir_all(&old); if src.is_dir() { - copy_tree(src, &fresh)?; + copy_filtered(src, &fresh, skip)?; } else { std::fs::create_dir_all(&fresh)?; } diff --git a/cli/tests/actors/toolbox/src/main.rs b/cli/tests/actors/toolbox/src/main.rs index 133fad1..95bc8d5 100644 --- a/cli/tests/actors/toolbox/src/main.rs +++ b/cli/tests/actors/toolbox/src/main.rs @@ -1,19 +1,21 @@ use std::collections::HashMap; use std::io::{self, BufRead, Write}; -use std::time::Duration; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use serde_json::{Value, json}; -fn reply(out: &mut impl Write, message: &Value, body: Value) { - writeln!( - out, - "{}", - json!({"to": message["from"], "reply_to": message["id"], "body": body}) - ) - .unwrap(); +fn send(out: &mut impl Write, message: Value) { + writeln!(out, "{message}").unwrap(); out.flush().unwrap(); } +fn reply(out: &mut impl Write, message: &Value, body: Value) { + send( + out, + json!({"to": message["from"], "reply_to": message["id"], "body": body}), + ); +} + fn io_error(error: io::Error) -> Value { json!({"type": "error", "error": error.to_string()}) } @@ -21,7 +23,8 @@ fn io_error(error: io::Error) -> Value { fn main() { let actor = std::env::var("SAUSAGE_ACTOR").unwrap_or_default(); let version = std::env::var("SAUSAGE_VERSION").unwrap_or_else(|_| "candidate".to_string()); - if actor == "crashloop" && version == "2" { + let candidate = std::env::var_os("SAUSAGE_CANDIDATE").is_some(); + if actor == "crashloop" && version == "2" && !candidate { std::fs::write("/data/state", "bad-v2").unwrap(); panic!("version 2 crashes on startup"); } @@ -29,6 +32,14 @@ fn main() { let mut open = HashMap::::new(); let stdin = io::stdin(); let mut out = io::stdout().lock(); + let mut emit_shutdown = false; + if actor == "spawn-race" && !candidate { + let stamp = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + std::fs::write(format!("/data/start-{stamp}"), "started").unwrap(); + } for line in stdin.lock().lines() { let Ok(line) = line else { break }; let Ok(message) = serde_json::from_str::(&line) else { @@ -43,17 +54,21 @@ fn main() { let _ = std::fs::write("/data/shutdown", "stopping"); std::thread::sleep(Duration::from_millis(800)); } + if emit_shutdown { + send( + &mut out, + json!({"to": "console", "reply_to": message["id"], "body": {"type": "shutdown-reply"}}), + ); + send(&mut out, json!({"body": {"type": "shutdown-publish"}})); + } break; } Some("echo") => json!({"type": "echo", "value": body["value"]}), Some("publish") => { - writeln!( - out, - "{}", - json!({"body": {"type": "published", "value": body["value"]}}) - ) - .unwrap(); - out.flush().unwrap(); + send( + &mut out, + json!({"body": {"type": "published", "value": body["value"]}}), + ); continue; } Some("version") => json!({"type": "version", "version": version}), @@ -62,6 +77,15 @@ fn main() { Ok(()) => json!({"type": "ok"}), Err(error) => io_error(error), }, + Some("mkdir") => match std::fs::create_dir_all(&path) { + Ok(()) => json!({"type": "ok"}), + Err(error) => io_error(error), + }, + Some("rename") => match std::fs::rename(&path, body["to"].as_str().unwrap_or_default()) + { + Ok(()) => json!({"type": "ok"}), + Err(error) => io_error(error), + }, Some("open") => { let result = (|| -> io::Result<()> { let mut file = std::fs::File::create(&path)?; @@ -74,6 +98,21 @@ fn main() { Err(error) => io_error(error), } } + Some("open-truncate") => { + let result = (|| -> io::Result<()> { + let mut file = std::fs::OpenOptions::new() + .write(true) + .truncate(true) + .open(&path)?; + file.write_all(text.as_bytes())?; + open.insert(path.clone(), file); + Ok(()) + })(); + match result { + Ok(()) => json!({"type": "opened"}), + Err(error) => io_error(error), + } + } Some("append") => { let result = (|| -> io::Result<()> { let mut file = std::fs::OpenOptions::new().append(true).open(&path)?; @@ -87,6 +126,11 @@ fn main() { } } Some("close") => json!({"type": "closed", "closed": open.remove(&path).is_some()}), + Some("arm-shutdown") => { + emit_shutdown = true; + json!({"type": "armed"}) + } + Some("ignore-stdin") => json!({"type": "ignoring"}), Some("read") => match std::fs::read_to_string(&path) { Ok(text) => json!({"type": "content", "text": text}), Err(error) => io_error(error), @@ -107,5 +151,10 @@ fn main() { _ => json!({"type": "error", "error": "unknown message"}), }; reply(&mut out, &message, response); + if body["type"] == "ignore-stdin" { + loop { + std::thread::sleep(Duration::from_secs(60)); + } + } } } diff --git a/cli/tests/common/mod.rs b/cli/tests/common/mod.rs index cc695cc..b5b7096 100644 --- a/cli/tests/common/mod.rs +++ b/cli/tests/common/mod.rs @@ -142,6 +142,28 @@ pub async fn receive_reply( .unwrap_or_else(|_| panic!("timed out waiting for reply to {id}")) } +pub async fn receive_replies( + mailbox: &mut mpsc::Receiver, + ids: &[String], + timeout: Duration, +) -> Vec { + let mut replies = Vec::with_capacity(ids.len()); + tokio::time::timeout(timeout, async { + while replies.len() < ids.len() { + let message = mailbox.recv().await.expect("console mailbox closed"); + assert!( + ids.iter() + .any(|id| message.reply_to.as_deref() == Some(id.as_str())), + "unexpected message while waiting for replies: {message:?}" + ); + replies.push(message); + } + }) + .await + .expect("timed out waiting for replies"); + replies +} + pub async fn ask( host: &Host, mailbox: &mut mpsc::Receiver, @@ -152,16 +174,24 @@ pub async fn ask( receive_reply(mailbox, &id, REPLY_TIMEOUT).await } -pub async fn spawn_actor( +pub async fn ask_type( host: &Host, mailbox: &mut mpsc::Receiver, actor: &str, - wasm: &Path, - source: &Path, + body: Value, + expected: &str, ) -> Envelope { - request( - host, - mailbox, + let reply = ask(host, mailbox, actor, body).await; + assert_eq!(reply.body["type"], expected, "unexpected reply: {reply:?}"); + reply +} + +pub fn spawn_request(host: &Host, actor: &str, wasm: &Path, source: &Path) -> String { + host.broker.send( + "console", + Some("broker"), + None, + None, json!({ "type": "spawn", "actor": actor, @@ -169,7 +199,32 @@ pub async fn spawn_actor( "source": source.display().to_string() }), ) - .await +} + +pub async fn spawn_actor( + host: &Host, + mailbox: &mut mpsc::Receiver, + actor: &str, + wasm: &Path, + source: &Path, +) -> Envelope { + let id = spawn_request(host, actor, wasm, source); + receive_reply(mailbox, &id, REPLY_TIMEOUT).await +} + +pub async fn spawn_type( + host: &Host, + mailbox: &mut mpsc::Receiver, + actor: &str, + wasm: &Path, + source: &Path, +) -> Envelope { + let reply = spawn_actor(host, mailbox, actor, wasm, source).await; + assert_eq!( + reply.body["type"], "spawned", + "unexpected spawn reply: {reply:?}" + ); + reply } pub async fn upgrade_actor( diff --git a/cli/tests/host.rs b/cli/tests/host.rs index 3ba74d6..7192837 100644 --- a/cli/tests/host.rs +++ b/cli/tests/host.rs @@ -1,6 +1,6 @@ mod common; -use std::time::Duration; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use common::*; use sausage::fs::TMP_PREFIX; @@ -20,6 +20,41 @@ async fn filesystem_world(label: &str) -> (World, Host, mpsc::Receiver (world, host, mailbox) } +async fn receive_shutdown_messages( + mailbox: &mut mpsc::Receiver, + id: &str, + expected: &str, +) { + let (mut result, mut reply, mut publish) = (false, false, false); + tokio::time::timeout(Duration::from_secs(10), async { + while !result || !reply || !publish { + let message = mailbox.recv().await.expect("console mailbox closed"); + if message.reply_to.as_deref() == Some(id) { + assert_eq!(message.body["type"], expected); + result = true; + } else if message.reply_to.as_deref() == Some("broker/shutdown") { + assert_eq!(message.from, "shutdown-output"); + assert_eq!(message.body["type"], "shutdown-reply"); + reply = true; + } else { + assert_eq!(message.from, "shutdown-output"); + assert_eq!(message.reply_to, None); + assert_eq!(message.body["type"], "shutdown-publish"); + publish = true; + } + } + }) + .await + .expect("shutdown output or lifecycle reply did not arrive"); +} + +fn epoch_second() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs() +} + #[tokio::test] async fn routes_replies_publishes_and_reports_undeliverable() { let world = create_world("routing"); @@ -639,3 +674,295 @@ async fn rejects_parent_paths() { ); host.stop().await; } + +#[tokio::test] +async fn concurrent_spawns_of_one_id_leave_one_live_instance() { + let world = create_world("concurrent-spawn"); + let wasm = toolbox_wasm().clone(); + let (host, mut mailbox) = start(&world).await; + let source = passing_source(&world, "source"); + let ids = [ + spawn_request(&host, "spawn-race", &wasm, &source), + spawn_request(&host, "spawn-race", &wasm, &source), + ]; + let replies = receive_replies(&mut mailbox, &ids, Duration::from_secs(20)).await; + let spawned = replies + .iter() + .filter(|r| r.body["type"] == "spawned") + .count(); + let failed = replies.iter().filter(|r| r.body["type"] == "error").count(); + assert_eq!((spawned, failed), (1, 1)); + + let echo = ask_type( + &host, + &mut mailbox, + "spawn-race", + json!({"type": "echo", "value": "live"}), + "echo", + ) + .await; + assert_eq!(echo.body["value"], "live"); + tokio::time::sleep(Duration::from_millis(100)).await; + let list = ask_type( + &host, + &mut mailbox, + "spawn-race", + json!({"type": "list", "path": "/data"}), + "entries", + ) + .await; + let starts = list.body["names"] + .as_array() + .unwrap() + .iter() + .filter(|n| n.as_str().unwrap().starts_with("start-")) + .count(); + assert_eq!(starts, 1); + host.stop().await; +} + +#[tokio::test] +async fn rollback_preserves_target_and_git_files_in_data_and_public() { + let world = create_world("snapshot-special-directories"); + let wasm = toolbox_wasm().clone(); + let (host, mut mailbox) = start(&world).await; + let source = passing_source(&world, "source"); + spawn_type(&host, &mut mailbox, "snapshot", &wasm, &source).await; + let files = [ + ("/data/target", "/data/target/x", "data-target"), + ("/data/.git", "/data/.git/x", "data-git"), + ("/public/target", "/public/target/x", "public-target"), + ("/public/.git", "/public/.git/x", "public-git"), + ]; + for (dir, path, text) in files { + let mkdir = json!({"type": "mkdir", "path": dir}); + ask_type(&host, &mut mailbox, "snapshot", mkdir, "ok").await; + let write = json!({"type": "write", "path": path, "text": text}); + ask_type(&host, &mut mailbox, "snapshot", write, "ok").await; + } + let upgrade = json!({"type": "upgrade", "actor": "snapshot", "wasm": wasm.display().to_string(), "source": source.display().to_string()}); + ask_type(&host, &mut mailbox, "broker", upgrade, "upgraded").await; + let rollback = json!({"type": "rollback", "actor": "snapshot", "version": 1}); + ask_type(&host, &mut mailbox, "broker", rollback, "rolled_back").await; + for (_, path, text) in files { + let read = json!({"type": "read", "path": path}); + let reply = ask_type(&host, &mut mailbox, "snapshot", read, "content").await; + assert_eq!(reply.body["text"], text); + } + host.stop().await; +} + +#[tokio::test] +async fn stop_finishes_when_actor_stdin_is_full() { + let world = create_world("full-stdin-stop"); + let wasm = toolbox_wasm().clone(); + let (host, mut mailbox) = start(&world).await; + let source = passing_source(&world, "source"); + spawn_type(&host, &mut mailbox, "full-stdin", &wasm, &source).await; + let ignore = json!({"type": "ignore-stdin"}); + ask_type(&host, &mut mailbox, "full-stdin", ignore, "ignoring").await; + let large = json!({"type": "echo", "value": "x".repeat(256 * 1024)}); + host.broker + .send("console", Some("full-stdin"), None, None, large); + tokio::time::sleep(Duration::from_millis(100)).await; + let stop = json!({"type": "stop", "actor": "full-stdin"}); + let id = host + .broker + .send("console", Some("broker"), None, None, stop); + let stopped = tokio::time::timeout( + Duration::from_secs(6), + receive_reply(&mut mailbox, &id, Duration::from_secs(10)), + ) + .await + .expect("stop exceeded the five-second shutdown deadline"); + assert_eq!(stopped.body["type"], "stopped"); + host.stop().await; +} + +#[tokio::test] +async fn shutdown_reply_and_publish_survive_stop_and_upgrade() { + let world = create_world("shutdown-output"); + let wasm = toolbox_wasm().clone(); + let (host, mut mailbox) = start(&world).await; + let source = passing_source(&world, "source"); + spawn_type(&host, &mut mailbox, "shutdown-output", &wasm, &source).await; + let subscribe = json!({"type": "subscribe", "actor": "shutdown-output"}); + ask_type(&host, &mut mailbox, "broker", subscribe, "subscribed").await; + + for operation in ["stop", "upgrade"] { + let arm = json!({"type": "arm-shutdown"}); + ask_type(&host, &mut mailbox, "shutdown-output", arm, "armed").await; + let body = match operation { + "stop" => json!({"type": "stop", "actor": "shutdown-output"}), + _ => { + json!({"type": "upgrade", "actor": "shutdown-output", "wasm": wasm.display().to_string(), "source": source.display().to_string()}) + } + }; + let id = host + .broker + .send("console", Some("broker"), None, None, body); + receive_shutdown_messages( + &mut mailbox, + &id, + if operation == "stop" { + "stopped" + } else { + "upgraded" + }, + ) + .await; + if operation == "stop" { + let start = json!({"type": "start", "actor": "shutdown-output"}); + ask_type(&host, &mut mailbox, "broker", start, "started").await; + } + } + host.stop().await; +} + +#[tokio::test] +async fn rapid_remove_recreate_remove_keeps_actor_manageable() { + let world = create_world("rapid-remove"); + let wasm = toolbox_wasm().clone(); + let (host, mut mailbox) = start(&world).await; + let source = passing_source(&world, "source"); + spawn_type(&host, &mut mailbox, "rapid", &wasm, &source).await; + let stop = json!({"type": "stop", "actor": "rapid"}); + ask_type(&host, &mut mailbox, "broker", stop, "stopped").await; + + while SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .subsec_millis() + >= 100 + { + tokio::time::sleep(Duration::from_millis(10)).await; + } + let archive_second = epoch_second(); + let remove = json!({"type": "remove", "actor": "rapid"}); + let first = ask_type(&host, &mut mailbox, "broker", remove.clone(), "removed").await; + assert_eq!( + epoch_second(), + archive_second, + "first remove crossed an epoch-second boundary" + ); + + // Restore the stopped actor from its archive to avoid a candidate smoke test. + sausage::world::copy_tree( + first.body["archive"] + .as_str() + .map(std::path::Path::new) + .unwrap(), + &world.actor_dir("rapid"), + ) + .unwrap(); + let scan = json!({"type": "scan"}); + ask_type(&host, &mut mailbox, "broker", scan.clone(), "scanned").await; + let second = ask_type(&host, &mut mailbox, "broker", remove.clone(), "removed").await; + assert_eq!( + epoch_second(), + archive_second, + "remove and recreate crossed an epoch-second boundary" + ); + + sausage::world::copy_tree( + second.body["archive"] + .as_str() + .map(std::path::Path::new) + .unwrap(), + &world.actor_dir("rapid"), + ) + .unwrap(); + ask_type(&host, &mut mailbox, "broker", scan, "scanned").await; + let start = json!({"type": "start", "actor": "rapid"}); + ask_type(&host, &mut mailbox, "broker", start, "started").await; + let echo = ask_type( + &host, + &mut mailbox, + "rapid", + json!({"type": "echo", "value": "managed"}), + "echo", + ) + .await; + assert_eq!(echo.body["value"], "managed"); + host.stop().await; +} + +#[tokio::test] +async fn close_commits_open_file_after_parent_directory_rename() { + let (_, host, mut mailbox) = filesystem_world("rename-open-file").await; + let mkdir = json!({"type": "mkdir", "path": "/public/a"}); + ask_type(&host, &mut mailbox, "writer", mkdir, "ok").await; + let open = json!({"type": "open", "path": "/public/a/file", "text": "content"}); + ask_type(&host, &mut mailbox, "writer", open, "opened").await; + let rename = json!({"type": "rename", "path": "/public/a", "to": "/public/b"}); + ask_type(&host, &mut mailbox, "writer", rename, "ok").await; + let close = json!({"type": "close", "path": "/public/a/file"}); + assert_eq!( + ask(&host, &mut mailbox, "writer", close).await.body["closed"], + true + ); + let read = json!({"type": "read", "path": "/public/b/file"}); + assert_eq!( + ask_type(&host, &mut mailbox, "writer", read, "content") + .await + .body["text"], + "content" + ); + host.stop().await; +} + +#[tokio::test] +async fn truncate_without_create_commits_existing_file_on_close() { + let (_, host, mut mailbox) = filesystem_world("truncate-without-create").await; + let write = json!({"type": "write", "path": "/public/file", "text": "old"}); + ask_type(&host, &mut mailbox, "writer", write, "ok").await; + let open = json!({"type": "open-truncate", "path": "/public/file", "text": "new"}); + ask_type(&host, &mut mailbox, "writer", open, "opened").await; + let path = "/actor/writer/public/file"; + let read = json!({"type": "read", "path": path}); + assert_eq!( + ask_type(&host, &mut mailbox, "reader", read, "content") + .await + .body["text"], + "old" + ); + let close = json!({"type": "close", "path": "/public/file"}); + assert_eq!( + ask(&host, &mut mailbox, "writer", close).await.body["closed"], + true + ); + let read = json!({"type": "read", "path": path}); + assert_eq!( + ask_type(&host, &mut mailbox, "reader", read, "content") + .await + .body["text"], + "new" + ); + host.stop().await; +} + +#[tokio::test] +async fn candidates_receive_version_and_preexisting_spawn_state() { + let world = create_world("candidate-context"); + let wasm = toolbox_wasm().clone(); + let (host, mut mailbox) = start(&world).await; + let version_step = + json!({"send": {"type": "version"}, "expect": {"type": "version", "version": "1"}}); + let version_source = source_with_test(&world, "version", json!({"steps": [version_step]})); + let state_step = json!({"send": {"type": "read", "path": "/data/state.json"}, "expect": {"type": "content", "text": "seeded"}}); + let state_source = source_with_test(&world, "state", json!({"steps": [state_step]})); + let state_dir = world.data_dir("stateful"); + std::fs::create_dir_all(&state_dir).unwrap(); + std::fs::write(state_dir.join("state.json"), "seeded").unwrap(); + let ids = [ + spawn_request(&host, "versioned", &wasm, &version_source), + spawn_request(&host, "stateful", &wasm, &state_source), + ]; + let replies = receive_replies(&mut mailbox, &ids, Duration::from_secs(20)).await; + let successful = replies + .iter() + .filter(|reply| reply.body["type"] == "spawned") + .count(); + assert_eq!(successful, 2); + host.stop().await; +}