diff --git a/PROTOCOL.md b/PROTOCOL.md index 9adef6d..966bdaa 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -19,7 +19,7 @@ Rust plugins use the SDK `sausage-plugin`. Its source is at | call | A JSON-RPC request. `notes.add` calls the method `add` of the plugin `notes`. | | interface | A set of methods. A plugin has one main interface. It can also declare endpoint interfaces. | | endpoint | An interface that a running process serves on a TCP port. An endpoint file tells others where it is. | -| service | A call in `/config/boot.yaml`. The core starts it at boot and again after an upgrade of its plugin. | +| service | A call in `/config/boot.yaml`. The core starts it at boot, restarts it after a crash, and starts it again after an upgrade of its plugin. | | import | An interface of another plugin that a plugin uses. | | provider, consumer | The plugin that has an interface, and a plugin that imports it. | | journal | The central log of the world. | @@ -159,8 +159,11 @@ active (section 10.3): continue to work. It can stop accepting work, finish its in-flight calls, and exit early. Ignoring the notice changes nothing. The grace period is the upper bound before the core stops the process (section 3.6); its - default is 60 seconds. A plugin can declare its own value in its old - `rpc.discover` document (`x-sausage-grace`). + duration is 5 seconds for every plugin. Discovery metadata, including an + explicit `x-sausage-grace`, does not change it. After grace, stdin closes + and the separate 5-second stop budget applies (section 3.6): at most two + 5-second windows. Discovery-only processes do not receive this notice or + enter upgrade grace; they follow the discovery lifetime (section 10.1). 3. When the old process of a service has ended, the core starts the service again with the new version. 4. When the last process of an older version has ended, the core starts a @@ -353,10 +356,29 @@ At start, the core starts each service in `/config/boot.yaml`, in order: - A service is a detached call with the parent `core`. The core waits for its result before it starts the next one. The result shows that the service is ready. -- If a service fails, the core writes the error to the journal and continues. -- After an upgrade of the plugin of a service, the core starts the service - again with the new version (section 3.7). -- The core does not restart other detached processes. +- If a service fails before answering, the core writes the error to the + journal and continues booting. A failure to admit a process (for example, + a missing plugin or invalid parameters) is logged, not retried in a loop. +- The core supervises the process lifetime, not its readiness response. A + nonzero exit, trap or failed host task restarts the same service with its + original method and parameters, including a crash before readiness. A + readiness response alone never triggers a restart; a clean exit does not + count as a crash or trigger an immediate restart. +- More than 5 crashes within 10 seconds gives up on that service and writes + `service gave up` to the journal. Each service has an independent sliding + window; crashes exactly 10 seconds old expire. There is no retry delay or + periodic liveness check. Giving up is latched for this host lifetime: + neither elapsed time nor an upgrade re-arms it. +- After an upgrade of the plugin of a service, the same supervisor starts + it again with the current version after its old process ends (section + 3.7). Superseded exits are retirement, not crashes. Admission and cleanup + reserve a single owner/attempt, so an upgrade racing a crash cannot start + two owners. A matching boot-service start from `rpc.retired` is ignored + (returns `null`), even while restart is pending or after giving up; older + plugins' fallback starts do not bypass supervision. +- An explicit `core.stop` or `core.kill` disables automatic attempts for + that service, including during an upgrade. Host shutdown starts no new + processes. The core does not restart other detached processes. - A singleton is a service with a fixed endpoint in `/run`. Only the core starts it, so two instances cannot start at the same time. @@ -424,7 +446,8 @@ Params are by name. The document can have these extension fields: `methods`. - `x-sausage-imports`: for each provider, the part of its document that the plugin uses. -- `x-sausage-grace`: the grace period in seconds. +- `x-sausage-grace`: legacy grace declaration in seconds. The core ignores + this field; upgrade grace is always 5 seconds (section 3.7). The schema of params describes what the provider can read. The schema of a result describes what the provider writes. Descriptions are part of the diff --git a/README.md b/README.md index f623eaa..5337d93 100644 --- a/README.md +++ b/README.md @@ -93,10 +93,16 @@ plugins//plugin.wasm the current version of a plugin plugins//source/ its source state// the state of a plugin, by convention state/core/journal.sqlite the journal: stderr of all processes, and core events -config/boot.yaml the services that start with the world +config/boot.yaml ordered boot services, supervised for crashes (PROTOCOL ยง8) config/agent.yaml the configuration of the agent ``` +Boot services restart on process crashes, not on readiness responses. More +than 5 crashes within 10 seconds stops automatic attempts for that service +and journals `service gave up`; an upgrade does not reset that decision. +Explicitly stopping or killing a service also leaves it stopped for this +host lifetime. See [Boot and services](PROTOCOL.md#8-boot-and-services). + Rust builds use the SDK in `plugins/cargo/sdk/` directly. The host seeds this folder only when it is missing; restarts and host upgrades preserve existing files. SDK edits take effect on the next `cargo.build`, without a restart. diff --git a/cli/src/activation.rs b/cli/src/activation.rs index 2729ff3..cbd561e 100644 --- a/cli/src/activation.rs +++ b/cli/src/activation.rs @@ -279,15 +279,10 @@ impl Host { .get(&candidate.plugin) .filter(|old| !old.hash.is_empty() && old.hash != candidate.interface.hash) { - let grace = old.document["x-sausage-grace"] - .as_f64() - .filter(|value| value.is_finite() && *value >= 0.0) - .unwrap_or(60.0); upgrades.push(self.snapshot_upgrade( &candidate.plugin, old.hash.clone(), &candidate.interface.hash, - grace, )); } self.0.journal.write( diff --git a/cli/src/graph.rs b/cli/src/graph.rs index fb551c7..38ed297 100644 --- a/cli/src/graph.rs +++ b/cli/src/graph.rs @@ -223,16 +223,7 @@ impl Host { if let Some(old) = old_plugins.get(plugin).filter(|old| { !old.hash.is_empty() && old.hash != current.hash && current.error.is_none() }) { - let grace = old.document["x-sausage-grace"] - .as_f64() - .filter(|value| value.is_finite() && *value >= 0.0) - .unwrap_or(60.0); - upgrades.push(self.snapshot_upgrade( - plugin, - old.hash.clone(), - ¤t.hash, - grace, - )); + upgrades.push(self.snapshot_upgrade(plugin, old.hash.clone(), ¤t.hash)); } } self.publish_graph(plugins); diff --git a/cli/src/process.rs b/cli/src/process.rs index 2ce92e1..bae0dcb 100644 --- a/cli/src/process.rs +++ b/cli/src/process.rs @@ -33,6 +33,8 @@ pub(crate) struct Process { pub(crate) id: Option, pub(crate) answer: Option>, pub(crate) ended: watch::Sender, + pub(crate) stopped: bool, + pub(crate) superseded: bool, } impl Host { /// Called only after the execution task ends. Wait for the store and file operations before releasing locks. @@ -57,6 +59,7 @@ impl Host { .map(|_| self.0.journal.query_later(json!({"pid":pid,"limit":20}))); self.0.journal.forget(pid); Self::release_locks(&mut state, pid); + self.service_ended(&mut state, pid, status, p.stopped, p.superseded); let status = status.to_string(); tokio::spawn(async move { if let Some(answer) = p.answer { @@ -84,13 +87,24 @@ impl Host { } /// Close stdin and start the same five-second deadline for each attached child. pub(crate) fn begin_stop(&self, pid: &str) -> Option> { + self.stop_process(pid, false) + } + /// Grace expiry is retirement, not an operator stopping the service. + pub(crate) fn begin_retire(&self, pid: &str) -> Option> { + self.stop_process(pid, true) + } + fn stop_process(&self, pid: &str, retirement: bool) -> Option> { let mut state = self.0.state.lock().unwrap(); let p = state.processes.get_mut(pid)?; + if !retirement { + p.superseded = false; + } let ended = p.ended.subscribe(); if p.info.state == "stopping" { return Some(ended); } p.info.state = "stopping".into(); + p.stopped = true; p.input.take(); self.0.journal.write(pid, &p.info.plugin, "core", "stop"); for lock in state.locks.values_mut() { @@ -110,17 +124,24 @@ impl Host { let pid = pid.to_string(); tokio::spawn(async move { tokio::time::sleep(std::time::Duration::from_secs(5)).await; - h.kill(&pid); + h.kill_process(&pid, retirement); }); Some(ended) } /// Abort a process task and perform its ended-state cleanup. pub(crate) fn kill(&self, pid: &str) { + self.kill_process(pid, false); + } + fn kill_process(&self, pid: &str, retirement: bool) { let mut state = self.0.state.lock().unwrap(); for lock in state.locks.values_mut() { lock.queue.retain(|(queued, _)| queued != pid); } if let Some(p) = state.processes.get_mut(pid) { + if !retirement { + p.superseded = false; + } + p.stopped = true; p.info.state = "stopping".into(); p.input.take(); self.0.journal.write(pid, &p.info.plugin, "core", "kill"); diff --git a/cli/src/routing.rs b/cli/src/routing.rs index 738e3f2..9489f9a 100644 --- a/cli/src/routing.rs +++ b/cli/src/routing.rs @@ -171,6 +171,33 @@ impl Host { { return Err(error(-32600, "caller stopped or ended")); } + // Older plugins may still try to start their boot service from + // rpc.retired. The supervisor owns it even while a restart is + // pending, or after it has given up. Do not create a second owner. + if state + .processes + .get(&caller) + .is_some_and(|p| p.info.method == "rpc.retired") + && state + .services + .iter() + .any(|s| s.method == format!("{plugin}.{method}") && s.params == params) + { + return Ok(Value::Null); + } + if caller == "core" + && let Some(service) = state + .services + .iter_mut() + .find(|s| Some(&s.request) == id.as_ref()) + { + if !service.enabled || service.owner.is_some() { + return Err(error(-32600, "service attempt is no longer current")); + } + service.owner = Some(pid.clone()); + service.pending = false; + service.version = Some(info.version.clone()); + } host.0 .journal .write(&pid, plugin, "core", &format!("start {method} {params}")); @@ -187,6 +214,8 @@ impl Host { id: id.clone(), answer: if id.is_some() { Some(tx) } else { None }, ended: watch::channel(false).0, + stopped: false, + superseded: false, }, ); } diff --git a/cli/src/services.rs b/cli/src/services.rs index f9e6bc2..9a7a684 100644 --- a/cli/src/services.rs +++ b/cli/src/services.rs @@ -1,12 +1,36 @@ -//! Ordered boot services and their identities for upgrade restarts (PROTOCOL.md section 8). -use crate::Host; +//! Ordered boot calls and the single restart owner for each service (PROTOCOL.md section 8). +use crate::{Host, host::State}; use serde_json::{Value, json}; -#[derive(Clone)] +use std::{collections::VecDeque, sync::atomic::Ordering, time::Duration}; +use tokio::time::Instant; + +// Agreed systemd-derived values: /docs/design/plugin-handover/design.md, +// https://www.man7.org/linux/man-pages/man5/systemd-system.conf.5.html. +const CRASH_LIMIT: usize = 5; +const CRASH_WINDOW: Duration = Duration::from_secs(10); + pub(crate) struct Service { pub(crate) method: String, pub(crate) params: Value, pub(crate) request: Value, - pub(crate) starting: std::sync::Arc, + pub(crate) owner: Option, + pub(crate) pending: bool, + pub(crate) version: Option, + pub(crate) enabled: bool, + crashes: VecDeque, +} +impl Service { + fn crashed(&mut self, now: Instant) -> bool { + while self + .crashes + .front() + .is_some_and(|t| now.duration_since(*t) >= CRASH_WINDOW) + { + self.crashes.pop_front(); + } + self.crashes.push_back(now); + self.crashes.len() <= CRASH_LIMIT + } } impl Host { /// Starts discovery in the background, then waits for each boot service in order. @@ -24,30 +48,119 @@ impl Host { } Ok(()) } - /// Records a unique request identity before the service can answer or exit. + /// Register once, before admission or readiness. Restarts reuse this registration. pub(crate) async fn start_service(&self, method: String, params: Value) { - if self.0.shutdown.load(std::sync::atomic::Ordering::SeqCst) { - return; - } let request = json!(uuid::Uuid::now_v7().to_string()); - let starting = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true)); - self.0.state.lock().unwrap().services.push(Service { - method: method.clone(), - params: params.clone(), - request: request.clone(), - starting: starting.clone(), - }); + { + let mut state = self.0.state.lock().unwrap(); + if self.0.shutdown.load(Ordering::SeqCst) { + return; + } + state.services.push(Service { + method, + params, + request: request.clone(), + owner: None, + enabled: true, + pending: true, + version: None, + crashes: VecDeque::new(), + }); + } + self.run_service(request).await; + } + /// Await readiness only; process cleanup, not this response, decides whether to restart. + async fn run_service(&self, request: Value) { + let (method, params) = { + let state = self.0.state.lock().unwrap(); + if self.0.shutdown.load(Ordering::SeqCst) { + return; + } + let Some(service) = state + .services + .iter() + .find(|s| s.request == request && s.enabled) + else { + return; + }; + (service.method.clone(), service.params.clone()) + }; self.0 .journal .write("core", "core", "core", &format!("service start {method}")); - if let Err(error) = self - .route("core".into(), Some(request), method, params, true) - .await + let result = self + .route("core".into(), Some(request.clone()), method, params, true) + .await; { + let mut state = self.0.state.lock().unwrap(); + if let Some(service) = state.services.iter_mut().find(|s| s.request == request) { + service.pending = false; + } + } + if let Err(error) = result { + // Admission failures have no process lifetime to supervise. A process crash + // before readiness is handled by finish just like one after readiness. self.0 .journal .write("core", "core", "core", &format!("service failed {error}")); } - starting.store(false, std::sync::atomic::Ordering::SeqCst); + } + /// Called under the state lock after the old owner and its files/locks are gone. + pub(crate) fn service_ended( + &self, + state: &mut State, + pid: &str, + status: &str, + stopped: bool, + superseded: bool, + ) { + let Some(service) = state + .services + .iter_mut() + .find(|s| s.owner.as_deref() == Some(pid)) + else { + return; + }; + service.owner = None; + if self.0.shutdown.load(Ordering::SeqCst) || !service.enabled { + return; + } + if !superseded { + if stopped { + service.enabled = false; + return; + } + if status == "exit 0" { + return; + } + if !service.crashed(Instant::now()) { + service.enabled = false; + self.0.journal.write( + "core", + "core", + "core", + &format!( + "service gave up {}: more than {CRASH_LIMIT} crashes within {} seconds", + service.method, + CRASH_WINDOW.as_secs(), + ), + ); + return; + } + } + self.queue_service(service); + } + /// Reserve the next attempt before spawning it; upgrades and crashes share this owner. + pub(crate) fn queue_service(&self, service: &mut Service) { + service.request = json!(uuid::Uuid::now_v7().to_string()); + service.pending = true; + let request = service.request.clone(); + let host = self.clone(); + tokio::spawn(async move { + host.run_service(request).await; + }); } } + +#[cfg(test)] +mod tests; diff --git a/cli/src/services/tests.rs b/cli/src/services/tests.rs new file mode 100644 index 0000000..2d38c86 --- /dev/null +++ b/cli/src/services/tests.rs @@ -0,0 +1,356 @@ +//! Deterministic service supervision tests: native tasks stand in for guest lifetimes. +use super::*; +use crate::{NativePlugin, runtime::Runtime}; +use std::sync::Arc; +use tokio::sync::{mpsc, oneshot, watch}; + +struct Rig { + host: Host, + starts: mpsc::UnboundedReceiver<(String, oneshot::Sender<()>)>, + _world: tempfile::TempDir, +} +impl Rig { + async fn new() -> Self { + let world = tempfile::tempdir().unwrap(); + let runtime = Runtime::new(world.path()).unwrap(); + let host = Host::test_world(world.path(), runtime).await.unwrap(); + let (tx, starts) = mpsc::unbounded_channel(); + let plugin: NativePlugin = Arc::new(move |ctx, method, _| { + let tx = tx.clone(); + Box::pin(async move { + if method != "serve" { + return Ok(Value::Null); + } + let (end, ended) = oneshot::channel(); + tx.send((ctx.pid, end)).unwrap(); + let _ = ended.await; + Ok(Value::Null) + }) + }); + host.register_native("demo", plugin.clone()); + host.register_native("other", plugin); + Self { + host, + starts, + _world: world, + } + } + async fn start(&mut self, method: &str) -> (String, oneshot::Sender<()>) { + let host = self.host.clone(); + let method = method.to_string(); + tokio::spawn(async move { + host.start_service(method, json!({"kept":true})).await; + }); + self.next().await + } + async fn next(&mut self) -> (String, oneshot::Sender<()>) { + // Reuse the host integration call bound (activation.rs::call). + tokio::time::timeout(Duration::from_secs(60), self.starts.recv()) + .await + .unwrap() + .unwrap() + } + fn ended(&self, pid: &str) -> watch::Receiver { + self.host.0.state.lock().unwrap().processes[pid] + .ended + .subscribe() + } + async fn crash(&self, pid: &str) { + let mut ended = self.ended(pid); + // Abort execution without asking the supervisor for an intentional stop. + self.host.0.state.lock().unwrap().processes[pid] + .abort + .as_ref() + .unwrap() + .abort(); + ended.wait_for(|done| *done).await.unwrap(); + } + async fn clean(&self, pid: &str, end: oneshot::Sender<()>) { + let mut ended = self.ended(pid); + end.send(()).unwrap(); + ended.wait_for(|done| *done).await.unwrap(); + } + fn assert_single(&self, pid: &str, registrations: usize) { + let state = self.host.0.state.lock().unwrap(); + assert_eq!(state.services.len(), registrations); + let p = &state.processes[pid]; + assert_eq!(p.info.params, json!({"kept":true})); + assert!(p.info.detached); + assert_eq!(p.caller, "core"); + assert_eq!( + state + .services + .iter() + .filter(|s| s.owner.as_deref() == Some(pid)) + .count(), + 1 + ); + } + fn no_start(&mut self) { + assert!(self.starts.try_recv().is_err()); + } +} + +#[tokio::test(start_paused = true)] +async fn crash_before_and_after_readiness_observes_process_not_response() { + let mut rig = Rig::new().await; + let (pid, _end) = rig.start("demo.serve").await; + rig.assert_single(&pid, 1); + rig.crash(&pid).await; // No ready RPC response was sent. + let (pid, end) = rig.next().await; + rig.assert_single(&pid, 1); + rig.host.answer(&pid, Ok(json!({"ready":true}))); + tokio::task::yield_now().await; + rig.no_start(); // Ready does not end the service lifetime. + rig.crash(&pid).await; + drop(end); + let (pid, end) = rig.next().await; + rig.assert_single(&pid, 1); + rig.clean(&pid, end).await; + rig.no_start(); // A clean, intentional return is not a crash. + rig.host.stop().await; +} + +#[tokio::test(start_paused = true)] +async fn fifth_restarts_sixth_gives_up_and_journals_once() { + let mut rig = Rig::new().await; + let (mut pid, mut end) = rig.start("demo.serve").await; + for crash in 1..=CRASH_LIMIT { + rig.crash(&pid).await; + drop(end); + (pid, end) = rig.next().await; + rig.assert_single(&pid, 1); + assert_eq!( + rig.host.0.state.lock().unwrap().services[0].crashes.len(), + crash + ); + } + rig.crash(&pid).await; + drop(end); + rig.no_start(); + let logs = rig + .host + .call("core.log", json!({"grep":"service gave up"})) + .await + .unwrap(); + assert_eq!(logs.as_array().unwrap().len(), 1); + assert!( + logs[0]["line"] + .as_str() + .unwrap() + .contains("more than 5 crashes within 10 seconds") + ); + assert!(!rig.host.0.state.lock().unwrap().services[0].enabled); + // A retired hook cannot bypass give-up while its supervisor has no owner. + rig.host.register_native( + "fallback", + Arc::new(move |ctx, _, _| { + Box::pin(async move { + ctx.call( + "core.call", + json!({"method":"demo.serve","params":{"kept":true},"detach":true}), + ) + .await + }) + }), + ); + assert_eq!( + rig.host + .call("fallback.rpc.retired", json!({})) + .await + .unwrap(), + Value::Null + ); + rig.no_start(); + // Neither time passing nor an upgrade re-arms a given-up service. + tokio::time::advance(CRASH_WINDOW).await; + let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new"); + rig.host.begin_upgrade(upgrade); + tokio::task::yield_now().await; + rig.no_start(); + rig.host.stop().await; +} + +#[test] +fn sliding_window_boundary_and_progress() { + let now = Instant::now(); + let service = || Service { + method: "demo.serve".into(), + params: json!({}), + request: json!("request"), + owner: None, + enabled: true, + pending: false, + version: None, + crashes: VecDeque::new(), + }; + let mut before = service(); + let mut exact = service(); + let mut after = service(); + for _ in 0..CRASH_LIMIT { + assert!(before.crashed(now)); + assert!(exact.crashed(now)); + assert!(after.crashed(now)); + } + assert!(!before.crashed(now + CRASH_WINDOW - Duration::from_nanos(1))); + assert!(exact.crashed(now + CRASH_WINDOW)); + assert_eq!(exact.crashes.len(), 1); + assert!(after.crashed(now + CRASH_WINDOW + Duration::from_nanos(1))); + // Repeated progress beyond the window never accumulates old failures. + for step in 2..=CRASH_LIMIT as u32 + 1 { + assert!(exact.crashed(now + CRASH_WINDOW * step)); + assert_eq!(exact.crashes.len(), 1); + } +} + +#[tokio::test(start_paused = true)] +async fn services_have_independent_crash_budgets() { + let mut rig = Rig::new().await; + let (mut a, mut a_end) = rig.start("demo.serve").await; + let (b, b_end) = rig.start("other.serve").await; + for _ in 0..CRASH_LIMIT { + rig.crash(&a).await; + drop(a_end); + (a, a_end) = rig.next().await; + } + rig.crash(&a).await; + drop(a_end); + rig.crash(&b).await; + drop(b_end); + let (b, b_end) = rig.next().await; + rig.assert_single(&b, 2); + rig.clean(&b, b_end).await; + rig.no_start(); + rig.host.stop().await; +} + +#[tokio::test(start_paused = true)] +async fn upgrade_crash_race_has_one_restart_and_retired_fallback_cannot_duplicate_it() { + let mut rig = Rig::new().await; + let (pid, end) = rig.start("demo.serve").await; + let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new"); + rig.crash(&pid).await; // Exits between snapshot and begin_upgrade. + drop(end); + rig.host.begin_upgrade(upgrade); + let (pid, end) = rig.next().await; + rig.assert_single(&pid, 1); + assert!( + rig.host.0.state.lock().unwrap().services[0] + .crashes + .is_empty() + ); + // A native retired hook exercises the same routing path as a guest fallback. + rig.host.register_native( + "fallback", + Arc::new(move |ctx, _, _| { + Box::pin(async move { + ctx.call( + "core.call", + json!({"method":"demo.serve","params":{"kept":true},"detach":true}), + ) + .await + }) + }), + ); + assert_eq!( + rig.host + .call("fallback.rpc.retired", json!({})) + .await + .unwrap(), + Value::Null + ); + rig.no_start(); + rig.clean(&pid, end).await; + rig.host.stop().await; +} + +#[tokio::test(start_paused = true)] +async fn crash_before_upgrade_snapshot_also_has_one_restart() { + let mut rig = Rig::new().await; + let (pid, end) = rig.start("demo.serve").await; + rig.crash(&pid).await; + drop(end); + // Restart may be pending admission or already admitted: snapshot handles both. + let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "native"); + rig.host.begin_upgrade(upgrade); + let (pid, end) = rig.next().await; + rig.assert_single(&pid, 1); + rig.no_start(); + rig.clean(&pid, end).await; + rig.host.stop().await; +} + +#[tokio::test(start_paused = true)] +async fn pending_admission_is_single_owned_across_upgrade() { + let mut rig = Rig::new().await; + let (pid, end) = rig.start("demo.serve").await; + let host = rig.host.clone(); + let admission = host.0.admission.write().await; + rig.crash(&pid).await; + drop(end); + let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "native"); + rig.host.begin_upgrade(upgrade); + assert!(rig.host.0.state.lock().unwrap().services[0].pending); + drop(admission); + let (pid, end) = rig.next().await; + rig.assert_single(&pid, 1); + rig.no_start(); + rig.clean(&pid, end).await; + rig.host.stop().await; +} + +#[tokio::test(start_paused = true)] +async fn shutdown_rejects_restart_already_waiting_for_admission() { + let mut rig = Rig::new().await; + let (pid, end) = rig.start("demo.serve").await; + let host = rig.host.clone(); + let admission = host.0.admission.write().await; + rig.crash(&pid).await; + drop(end); + rig.host.stop().await; + drop(admission); + tokio::task::yield_now().await; + rig.no_start(); + assert!(rig.host.processes().is_empty()); +} + +#[tokio::test(start_paused = true)] +async fn clean_service_stays_dormant_until_upgrade_and_explicit_stop_stays_disabled() { + let mut rig = Rig::new().await; + let (pid, end) = rig.start("demo.serve").await; + rig.clean(&pid, end).await; + rig.no_start(); + let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new"); + rig.host.begin_upgrade(upgrade); + let (pid, end) = rig.next().await; + rig.assert_single(&pid, 1); + // An explicit stop stays disabled even if the task subsequently exits cleanly. + let mut ended = rig.host.begin_stop(&pid).unwrap(); + end.send(()).unwrap(); + ended.wait_for(|done| *done).await.unwrap(); + let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "newer"); + rig.host.begin_upgrade(upgrade); + tokio::task::yield_now().await; + rig.no_start(); + rig.host.stop().await; +} + +#[tokio::test(start_paused = true)] +async fn intentional_stop_and_shutdown_never_restart_even_during_upgrade() { + let mut rig = Rig::new().await; + let (pid, end) = rig.start("demo.serve").await; + let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new"); + // An explicit kill after supersession overrides the automatic upgrade restart. + rig.host + .call("core.kill", json!({"pid":pid})) + .await + .unwrap(); + drop(end); + rig.host.begin_upgrade(upgrade); + tokio::task::yield_now().await; + rig.no_start(); + let (_pid, end) = rig.start("other.serve").await; + rig.host.stop().await; + drop(end); + rig.no_start(); +} diff --git a/cli/src/upgrades.rs b/cli/src/upgrades.rs index bfa1783..d3c8d09 100644 --- a/cli/src/upgrades.rs +++ b/cli/src/upgrades.rs @@ -1,9 +1,15 @@ //! Grace periods, service restarts, and retirement of replaced versions (PROTOCOL.md section 3.7). -use crate::{Host, services::Service}; +use crate::Host; use serde_json::json; use std::{collections::BTreeMap, sync::atomic::Ordering, time::Duration}; use tokio::sync::watch; +/// Uniform host policy (PROTOCOL 3.7), independent of discovery metadata. +const UPGRADE_GRACE: Duration = Duration::from_secs(5); + +#[cfg(test)] +mod tests; + pub(crate) struct OldProcess { pid: String, version: String, @@ -13,9 +19,7 @@ pub(crate) struct Upgrade { plugin: String, previous: String, current: String, - grace: Duration, processes: Vec, - services: Vec<(Service, Option>)>, } impl Host { /// Snapshots old admissions while activation holds the admission write guard. @@ -24,54 +28,48 @@ impl Host { plugin: &str, previous: String, current: &str, - grace: f64, ) -> Upgrade { let mut state = self.0.state.lock().unwrap(); let processes = state .processes - .iter() + .iter_mut() .filter(|(_, process)| { process.info.plugin == plugin && process.info.version != current && process.info.method != "rpc.discover" }) - .map(|(pid, process)| OldProcess { - pid: pid.clone(), - version: process.info.version.clone(), - ended: process.ended.subscribe(), + .map(|(pid, process)| { + // The service supervisor owns the restart, including an exit on + // the superseded notice before the grace timer is installed. + process.superseded = !process.stopped; + OldProcess { + pid: pid.clone(), + version: process.info.version.clone(), + ended: process.ended.subscribe(), + } }) .collect(); - let mut services = Vec::new(); - let mut retained = Vec::new(); - for service in std::mem::take(&mut state.services) { - if service - .method - .split_once('.') - .is_some_and(|(name, _)| name == plugin) + // A cleanly returned service is dormant until an upgrade. A pending + // admission will choose the current version itself. Disabled services + // (explicit stop or crash-limit give-up) must never be re-armed here. + for service in &mut state.services { + if service.enabled + && !service.pending + && service.owner.is_none() + && service.version.as_deref() != Some(current) + && service + .method + .split_once('.') + .is_some_and(|(name, _)| name == plugin) { - let process = state - .processes - .values() - .find(|process| process.id.as_ref() == Some(&service.request)); - if process.is_some_and(|process| process.info.version == current) - || (process.is_none() && service.starting.load(Ordering::SeqCst)) - { - retained.push(service); - } else { - services.push((service, process.map(|process| process.ended.subscribe()))); - } - } else { - retained.push(service); + self.queue_service(service); } } - state.services = retained; Upgrade { plugin: plugin.into(), previous, current: current.into(), - grace: Duration::from_secs_f64(grace), processes, - services, } } /// Lets old calls finish, stops them after grace, and starts each lifecycle hook once. @@ -92,7 +90,7 @@ impl Host { "core", &upgrade.plugin, "core", - &format!("upgrade grace {} seconds", upgrade.grace.as_secs_f64()), + &format!("upgrade grace {} seconds", UPGRADE_GRACE.as_secs()), ); let mut versions: BTreeMap>> = BTreeMap::new(); versions.entry(upgrade.previous).or_default(); @@ -102,23 +100,11 @@ impl Host { .or_default() .push(process.ended.clone()); let host = host.clone(); - let grace = upgrade.grace; tokio::spawn(async move { let mut ended = process.ended; tokio::select! { _ = ended.wait_for(|ended|*ended) => (), - _ = tokio::time::sleep(grace) => { host.begin_stop(&process.pid); } - } - }); - } - for (service, ended) in upgrade.services { - let host = host.clone(); - tokio::spawn(async move { - if let Some(mut ended) = ended { - let _ = ended.wait_for(|ended| *ended).await; - } - if !host.0.shutdown.load(Ordering::SeqCst) { - host.start_service(service.method, service.params).await; + _ = tokio::time::sleep(UPGRADE_GRACE) => { host.begin_retire(&process.pid); } } }); } diff --git a/cli/src/upgrades/tests.rs b/cli/src/upgrades/tests.rs new file mode 100644 index 0000000..3bf71d7 --- /dev/null +++ b/cli/src/upgrades/tests.rs @@ -0,0 +1,115 @@ +//! Controlled-clock coverage of both upgrade windows and discovery exclusion. +use super::*; +use crate::{ + process::{Process, ProcessInfo}, + runtime::Runtime, + world::World, +}; +use tokio::{ + sync::mpsc, + task::JoinHandle, + time::{Instant, advance}, +}; + +/// A process with active work that ignores both superseded and EOF. +fn process( + host: &Host, + pid: &str, + method: &str, +) -> (mpsc::UnboundedReceiver, JoinHandle<()>) { + let (input, incoming) = mpsc::unbounded_channel(); + let task = tokio::spawn(std::future::pending()); + host.0.state.lock().unwrap().processes.insert( + pid.into(), + Process { + info: ProcessInfo { + plugin: "notes".into(), + version: "old".into(), + method: method.into(), + params: json!({}), + parent: None, + detached: true, + start_time: 0, + state: "running".into(), + }, + memory: sausage_vfs::Memory::default(), + files: sausage_vfs::Process::default(), + store_ended: watch::channel(true).0, + input: Some(input), + abort: Some(task.abort_handle()), + caller: "console".into(), + id: None, + answer: None, + ended: watch::channel(false).0, + stopped: false, + superseded: false, + }, + ); + (incoming, task) +} + +/// PROTOCOL 3.7/3.6: five seconds running, then five seconds stopping; discovery is untouched. +#[tokio::test] +async fn two_five_second_windows_and_discovery_exclusion() { + let dir = tempfile::tempdir().unwrap(); + let root = dir.path().join("world"); + World::create(&root).unwrap(); + let host = Host::test_world(&root, Runtime::new(&dir.path().join("cache")).unwrap()) + .await + .unwrap(); + let (mut input, work) = process(&host, "old-work", "work"); + let (mut discovery_input, discovery) = process(&host, "old-discovery", "rpc.discover"); + let upgrade = host.snapshot_upgrade("notes", "old".into(), "new"); + assert_eq!(upgrade.processes.len(), 1); + tokio::time::pause(); + let start = Instant::now(); + host.begin_upgrade(upgrade); + assert_eq!( + input.recv().await.unwrap(), + json!({"jsonrpc":"2.0","method":"rpc.superseded","params":{"current":"new"}}) + ); + // Run the upgrade dispatcher, then its per-process task, so the deadline is armed. + tokio::task::yield_now().await; + tokio::task::yield_now().await; + let window = Duration::from_secs(5); + assert_eq!(UPGRADE_GRACE, window); + // tokio::time::sleep documents millisecond granularity; the paused driver + // may round a deadline up by one tick. Check each window independently. + let tick = Duration::from_millis(1); + let before_deadline = window - tick; + advance(before_deadline).await; + tokio::task::yield_now().await; + assert_eq!(host.processes()["old-work"].state, "running"); + assert!(!work.is_finished()); + advance(tick).await; + assert!( + input.recv().await.is_none(), + "EOF at grace, even with active work" + ); + let stopped = Instant::now(); + assert!((window..=window + tick).contains(&(stopped - start))); + assert_eq!(host.processes()["old-work"].state, "stopping"); + // begin_retire schedules the independent PROTOCOL 3.6 deadline. + tokio::task::yield_now().await; + advance(before_deadline).await; + tokio::task::yield_now().await; + assert!( + !work.is_finished(), + "on_stop still has its full five seconds" + ); + advance(tick).await; + assert!(work.await.unwrap_err().is_cancelled()); + assert!((window..=window + tick).contains(&(Instant::now() - stopped))); + assert_eq!(host.processes()["old-discovery"].state, "running"); + assert!(matches!( + discovery_input.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + )); + assert!(!discovery.is_finished()); + discovery.abort(); + let _ = discovery.await; + // Synthetic processes have no execution cleanup task; discard only this test table. + host.0.shutdown.store(true, Ordering::SeqCst); + host.0.state.lock().unwrap().processes.clear(); + host.stop().await; +} diff --git a/cli/tests/host/activation.rs b/cli/tests/host/activation.rs index dd20dab..ffb7bc8 100644 --- a/cli/tests/host/activation.rs +++ b/cli/tests/host/activation.rs @@ -160,9 +160,9 @@ impl Drop for TestWorld { let _ = std::fs::remove_dir_all(&self.0); } } -/// Builds a method-only document without an upgrade delay. +/// Builds a method-only document; all plugins use the host's five-second grace. fn document(name: &str) -> Value { - json!({"methods":[{"name":name,"params":[],"result":{"schema":{"type":"string"}}}],"x-sausage-grace":0}) + json!({"methods":[{"name":name,"params":[],"result":{"schema":{"type":"string"}}}]}) } /// Adds an import of a provider's method list. fn consumer(provider: &str, expected: Value) -> Value { @@ -484,7 +484,7 @@ async fn review_dropped_activation_completes() { host.stop().await; } /// PROTOCOL 10.1: discovery fails after its fixed ten-second deadline and receives EOF. -/// PROTOCOL 3.7: old processes retain their declared grace while new calls use new code. +/// PROTOCOL 3.7: old processes get five seconds regardless of declarations; new calls use new code. /// PROTOCOL 3.7: retirement waits for all live admissions, including reverted versions. /// PROTOCOL 3.7: rejected discovery candidates never receive retirement hooks. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -604,7 +604,7 @@ async fn source_changes_do_not_change_candidate() { ); host.stop().await; } -/// PROTOCOL 3.7: plugins without a declared grace period retain old calls before the default deadline. +/// PROTOCOL 3.7: missing metadata also retains old calls until the uniform deadline. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn default_grace_retains_old_process() { let world = TestWorld::new(); @@ -623,6 +623,104 @@ async fn default_grace_retains_old_process() { assert_eq!(host.processes()[&pid].state, "running"); host.stop().await; } +/// PROTOCOL 3.7: metadata cannot shorten or extend the actual grace for active requests. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn uniform_grace_ignores_discovery_declarations() { + let world = TestWorld::new(); + let declarations = [ + ("missing", None), + ("default", Some(json!(60))), + ("long", Some(json!(3600))), + ("zero", Some(json!(0))), + ("negative", Some(json!(-1))), + ("invalid", Some(json!("long"))), + ("huge", Some(json!(1e300))), + ]; + for (name, grace) in &declarations { + let mut old = document("work"); + if let Some(grace) = grace { + old["x-sausage-grace"] = grace.clone(); + } + world.plugin(name, old, document("work")); + } + let host = world.start().await; + let mut ready = host.subscribe(); + let mut calls = Vec::new(); + for (name, _) in &declarations { + let host = host.clone(); + let method = format!("{name}.work"); + calls.push(tokio::spawn(async move { + let result = host.call(&method, json!({})).await; + (result, std::time::Instant::now()) + })); + } + // Guest notifications are events, not process/file polling. + for _ in &declarations { + let event = tokio::time::timeout(Duration::from_secs(30), ready.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(event["params"]["name"], "ready"); + } + let before = std::time::Instant::now(); + let plugins: Vec<_> = declarations + .iter() + .map(|(name, _)| json!({"plugin":name,"wasm":format!("/state/{name}/next.wasm")})) + .collect(); + call(&host, "core.activate", json!({"plugins":plugins})).await; + // Use the existing discovery+stop test budget, not a new execution limit. + tokio::time::timeout(Duration::from_secs(15), async { + for task in calls { + let (result, ended) = task.await.unwrap(); + assert_eq!(result.unwrap_err()["code"], -32002); + assert!( + ended.duration_since(before) >= Duration::from_secs(5), + "every declaration gets the full grace" + ); + } + }) + .await + .unwrap(); + host.stop().await; +} + +/// PROTOCOL 10.4/3.7: a watcher refresh uses the same policy as explicit activation. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn manual_copy_uses_uniform_grace() { + let world = TestWorld::new(); + let mut old = document("work"); + old["x-sausage-grace"] = json!(3600); + world.plugin("notes", old, document("work")); + let host = world.start().await; + // Wait for initial discovery without changing the installed version. + call( + &host, + "core.activate", + json!({"plugin":"notes","wasm":"/plugins/notes/plugin.wasm"}), + ) + .await; + let mut ready = host.subscribe(); + let work_host = host.clone(); + let work = tokio::spawn(async move { work_host.call("notes.work", json!({})).await }); + let event = tokio::time::timeout(Duration::from_secs(30), ready.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(event["params"]["name"], "ready"); + let before = std::time::Instant::now(); + std::fs::copy(fixture(2), world.0.join("plugins/notes/plugin.wasm")).unwrap(); + assert_eq!( + tokio::time::timeout(Duration::from_secs(15), work) + .await + .unwrap() + .unwrap() + .unwrap_err()["code"], + -32002 + ); + assert!(before.elapsed() >= Duration::from_secs(5)); + host.stop().await; +} + /// PROTOCOL 10.3: candidate imports must fit installed providers. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn checks_candidate_imports() { @@ -818,11 +916,11 @@ async fn graph_missing_import() { assert!(dot.as_str().unwrap().contains("absent")); host.stop().await; } -/// Checks the declared upgrade grace and dispatch to the new version. +/// Checks that an explicit long declaration cannot extend grace; new calls use the new version. async fn old_process_grace_and_new_calls() { let world = TestWorld::new(); let mut old = document("add"); - old["x-sausage-grace"] = json!(0.5); + old["x-sausage-grace"] = json!(3600); world.plugin("notes", old, document("add")); let host = world.start().await; call(&host, "notes.hold", json!({})).await; @@ -868,7 +966,7 @@ async fn superseded_notifies_each_old_process() { json!({"jsonrpc":"2.0","method":"rpc.superseded","params":{"current":fixture_hash(2)}}) .to_string(); for pid in pids { - // marker's existing 30-second bound is shorter than the old 60-second grace. + // The notice arrives while the old process is still in its five-second window. marker( world.0.join(format!("state/notes/superseded-{pid}")), &expected, @@ -993,7 +1091,7 @@ async fn upgrades_enter_the_journal() { .as_array() .unwrap() .iter() - .any(|e| e["line"] == "upgrade grace 60 seconds") + .any(|e| e["line"] == "upgrade grace 5 seconds") { break entries; } @@ -1117,6 +1215,36 @@ async fn old_consumer_can_call_removed_provider_method() { host.stop().await; } +/// PROTOCOL 8: supervise actual WASI exits on either side of the readiness response. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn crashed_boot_service_restarts_before_and_after_ready() { + for when in ["before", "after"] { + let world = TestWorld::new(); + world.plugin("notes", document("serve"), document("serve")); + std::fs::write( + world.0.join("config/boot.yaml"), + json!([{"method":"notes.serve","params":{"crash_once":when}}]).to_string(), + ) + .unwrap(); + let host = world.start().await; + marker(world.0.join("state/notes/service"), "1").await; + let entries = call( + &host, + "core.log", + json!({"grep":"service start notes.serve"}), + ) + .await; + assert_eq!(entries.as_array().unwrap().len(), 2); + let processes = host.processes(); + let services: Vec<_> = processes.values().filter(|p| p.method == "serve").collect(); + assert_eq!(services.len(), 1); + assert_eq!(services[0].params, json!({"crash_once":when})); + let exits = call(&host, "core.log", json!({"plugin":"notes","grep":"exit 1"})).await; + assert_eq!(exits.as_array().unwrap().len(), 1); + host.stop().await; + } +} + /// PROTOCOL 10.2: activation and revert check interfaces but leave persistent state unchanged. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn state_is_outside_interface_checks() { diff --git a/cli/tests/host/sdk.rs b/cli/tests/host/sdk.rs index dae662c..f405086 100644 --- a/cli/tests/host/sdk.rs +++ b/cli/tests/host/sdk.rs @@ -178,6 +178,56 @@ async fn stop_runs_hook() { assert_eq!(std::fs::read(world.0.join("stopped")).unwrap(), b"yes"); host.stop().await; } +/// PROTOCOL 3.7: an idle SDK listener opts into drain and exits before EOF/grace. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn superseded_idle_listener_drains_before_grace() { + let world = TestWorld::new(); + let host = world.start().await; + let (notice, mut notices) = tokio::sync::mpsc::unbounded_channel(); + host.register_native( + "other", + std::sync::Arc::new(move |_, method, _| { + let notice = notice.clone(); + Box::pin(async move { + assert_eq!(method, "superseded"); + notice.send(std::time::Instant::now()).unwrap(); + Ok(json!({})) + }) + }), + ); + let pid = call(&host, "probe.drain_service", json!({})).await; + // Replace with a different real component; force only skips unrelated test consumers. + std::fs::copy( + fixtures().join("notes.wasm"), + world.0.join("state/next.wasm"), + ) + .unwrap(); + call( + &host, + "core.activate", + json!({"plugin":"probe","wasm":"/state/next.wasm","force":true}), + ) + .await; + // Activation preparation/discovery is outside upgrade grace. Time from the + // hook's outgoing call, not from the start of candidate compilation. + let before = tokio::time::timeout(Duration::from_secs(60), notices.recv()) + .await + .unwrap() + .unwrap(); + // Waiting on its lifetime lock is an exit event, not process-liveness polling. + call(&host, "core.lock", json!({"path":"/drain-owner"})).await; + assert!(before.elapsed() < Duration::from_secs(5)); + assert!(!host.processes().contains_key(pid.as_str().unwrap())); + assert!( + !world + .0 + .join(format!("stopped-{}", pid.as_str().unwrap())) + .exists(), + "drain must not run on_stop (discovery processes may run their own hooks)" + ); + host.stop().await; +} + /// SDK framing carries payloads above 64 KiB in both directions. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn large_messages() { diff --git a/cli/tests/plugins/activation/src/main.rs b/cli/tests/plugins/activation/src/main.rs index 60c0c05..a664212 100644 --- a/cli/tests/plugins/activation/src/main.rs +++ b/cli/tests/plugins/activation/src/main.rs @@ -84,8 +84,31 @@ fn main() { .unwrap(); json!({"retired":version}) } + "work" => { + // Keep an actual request unanswered through grace; notices do not drain it. + send( + json!({"jsonrpc":"2.0","method":"core.notify","params":{"name":"ready","params":{}}}), + ); + for _ in lines {} + return; + } "serve" | "hold" => { + // One genuine process failure, followed by a healthy service, without + // timers or liveness polling in either the fixture or supervisor. + let crash = request["params"]["crash_once"].as_str(); + let first = crash.is_some() + && std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(format!("/state/{plugin}/crashed-once")) + .is_ok(); + if first && crash == Some("before") { + std::process::exit(1); + } send(json!({"jsonrpc":"2.0","id":request["id"],"result":{"version":version}})); + if first && crash == Some("after") { + std::process::exit(1); + } if request["method"] == "serve" { std::fs::write(format!("/state/{plugin}/service"), version).unwrap(); } diff --git a/sdk/src/lib.rs b/sdk/src/lib.rs index d8865d9..582c79a 100644 --- a/sdk/src/lib.rs +++ b/sdk/src/lib.rs @@ -356,7 +356,9 @@ //! //! The interface check covers calls, not state. Old and new processes share files in `/state`. //! `core.revert` restores code, not state. Preserve compatibility or migrate the state. -//! `#[rpc(grace_seconds = N)]` declares the grace period for old processes. The default is 60 seconds. +//! Core gives every old process 5 seconds of upgrade grace, then closes stdin and allows +//! a separate 5-second stop budget before killing it. Discovery-only processes are excluded. +//! `#[rpc(grace_seconds = N)]` still emits legacy `x-sausage-grace` metadata, but core ignores it. //! New calls use the new version immediately. Old processes can still use old imports during their grace period. //! [`macro@on_superseded`] handles the notice at grace start, with normal calls still available. //! It is opt-in: without a hook the process continues as before until core stops it. diff --git a/sdk/tests/plugins/probe/src/main.rs b/sdk/tests/plugins/probe/src/main.rs index ba8caa4..5ee6049 100644 --- a/sdk/tests/plugins/probe/src/main.rs +++ b/sdk/tests/plugins/probe/src/main.rs @@ -128,6 +128,11 @@ impl Probe { sausage_plugin::listen("/state/control", std::rc::Rc::new(Control)).await?; Ok(sausage_plugin::pid()) } + /// Keep an idle listener and ownership lock until opt-in drain ends this process. + async fn drain_service(&self) -> Result { + lock("/drain-owner").await?.keep(); + self.service().await + } /// Add two numbers. /// /// The result is their sum. @@ -238,6 +243,7 @@ impl Probe { #[sausage_plugin::on_stop] async fn stop(&self) { std::fs::write("/stopped", b"yes").unwrap(); + std::fs::write(format!("/stopped-{}", sausage_plugin::pid()), b"yes").unwrap(); Ctx::default().notify("stopped", json!({})).unwrap(); assert!(call("other.no", json!({})).await.is_err()); }