From a953d80ce61f0314a9439083ee55de8ac08b1d97 Mon Sep 17 00:00:00 2001 From: SAU-358 implementer Date: Fri, 30 Oct 2026 19:49:06 +0000 Subject: [PATCH 1/2] Supervise crashed boot service lifetimes with a latched crash limit Keep one registration and restart owner across process cleanup and upgrades. Restart on failed process exit before or after readiness; give up after more than five crashes in a sliding ten-second window. Respect explicit stop and shutdown, and suppress boot-service fallbacks from retired hooks. Cover deterministic lifetime, boundary, give-up and admission races, plus real WASI failures on both sides of readiness. Document the host contract. Card: SAU-358 --- PROTOCOL.md | 29 +- README.md | 8 +- cli/src/process.rs | 23 +- cli/src/routing.rs | 29 ++ cli/src/services.rs | 151 ++++++++-- cli/src/services/tests.rs | 362 +++++++++++++++++++++++ cli/src/upgrades.rs | 66 ++--- cli/tests/host/activation.rs | 30 ++ cli/tests/plugins/activation/src/main.rs | 15 + 9 files changed, 646 insertions(+), 67 deletions(-) create mode 100644 cli/src/services/tests.rs diff --git a/PROTOCOL.md b/PROTOCOL.md index e9e6b6b..d7dfaf6 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. | @@ -322,10 +322,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. 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/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 4650c85..15a7a89 100644 --- a/cli/src/routing.rs +++ b/cli/src/routing.rs @@ -166,6 +166,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}")); @@ -182,6 +209,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..88bdad3 --- /dev/null +++ b/cli/src/services/tests.rs @@ -0,0 +1,362 @@ +//! 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", 0.0); + 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", 0.0); + 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", 0.0); + 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", 0.0); + 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", 0.0); + 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", 0.0); + 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", 0.0); + // 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..03b4cd0 100644 --- a/cli/src/upgrades.rs +++ b/cli/src/upgrades.rs @@ -1,5 +1,5 @@ //! 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; @@ -15,7 +15,6 @@ pub(crate) struct Upgrade { current: String, grace: Duration, processes: Vec, - services: Vec<(Service, Option>)>, } impl Host { /// Snapshots old admissions while activation holds the admission write guard. @@ -29,49 +28,45 @@ impl Host { 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. @@ -107,18 +102,7 @@ impl Host { 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(grace) => { host.begin_retire(&process.pid); } } }); } diff --git a/cli/tests/host/activation.rs b/cli/tests/host/activation.rs index dd20dab..1702b48 100644 --- a/cli/tests/host/activation.rs +++ b/cli/tests/host/activation.rs @@ -1117,6 +1117,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/plugins/activation/src/main.rs b/cli/tests/plugins/activation/src/main.rs index 60c0c05..9f91ab5 100644 --- a/cli/tests/plugins/activation/src/main.rs +++ b/cli/tests/plugins/activation/src/main.rs @@ -85,7 +85,22 @@ fn main() { json!({"retired":version}) } "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(); } From 9cbd4f94aabab440773756af2c3655206e6053a7 Mon Sep 17 00:00:00 2001 From: SAU-357 implementer Date: Fri, 30 Oct 2026 20:03:12 +0000 Subject: [PATCH 2/2] Use five-second upgrade grace for every plugin Ignore legacy grace metadata in both activation paths. Keep superseded delivery, admission ordering, discovery exclusion and the separate EOF stop budget. Document the uniform policy and cover both windows and opt-in drain. Card: SAU-357 --- PROTOCOL.md | 10 +- cli/src/activation.rs | 5 - cli/src/graph.rs | 11 +-- cli/src/upgrades.rs | 14 +-- cli/src/upgrades/tests.rs | 113 ++++++++++++++++++++++ cli/tests/host/activation.rs | 114 +++++++++++++++++++++-- cli/tests/host/sdk.rs | 50 ++++++++++ cli/tests/plugins/activation/src/main.rs | 8 ++ sdk/src/lib.rs | 4 +- sdk/tests/plugins/probe/src/main.rs | 6 ++ 10 files changed, 302 insertions(+), 33 deletions(-) create mode 100644 cli/src/upgrades/tests.rs diff --git a/PROTOCOL.md b/PROTOCOL.md index e9e6b6b..ca9b5bc 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -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 @@ -391,7 +394,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/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/upgrades.rs b/cli/src/upgrades.rs index bfa1783..3bbc866 100644 --- a/cli/src/upgrades.rs +++ b/cli/src/upgrades.rs @@ -4,6 +4,12 @@ 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,7 +19,6 @@ pub(crate) struct Upgrade { plugin: String, previous: String, current: String, - grace: Duration, processes: Vec, services: Vec<(Service, Option>)>, } @@ -24,7 +29,6 @@ impl Host { plugin: &str, previous: String, current: &str, - grace: f64, ) -> Upgrade { let mut state = self.0.state.lock().unwrap(); let processes = state @@ -69,7 +73,6 @@ impl Host { plugin: plugin.into(), previous, current: current.into(), - grace: Duration::from_secs_f64(grace), processes, services, } @@ -92,7 +95,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,12 +105,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); } + _ = tokio::time::sleep(UPGRADE_GRACE) => { host.begin_stop(&process.pid); } } }); } diff --git a/cli/src/upgrades/tests.rs b/cli/src/upgrades/tests.rs new file mode 100644 index 0000000..c2e6f81 --- /dev/null +++ b/cli/src/upgrades/tests.rs @@ -0,0 +1,113 @@ +//! 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, + }, + ); + (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_stop 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..5f5c692 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; } 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..2059faf 100644 --- a/cli/tests/plugins/activation/src/main.rs +++ b/cli/tests/plugins/activation/src/main.rs @@ -84,6 +84,14 @@ 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" => { send(json!({"jsonrpc":"2.0","id":request["id"],"result":{"version":version}})); if request["method"] == "serve" { 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 a4b013b..3de1c20 100644 --- a/sdk/tests/plugins/probe/src/main.rs +++ b/sdk/tests/plugins/probe/src/main.rs @@ -60,6 +60,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. @@ -170,6 +175,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()); }