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