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(); }