diff --git a/core/continuum-core/src/bin/continuum.rs b/core/continuum-core/src/bin/continuum.rs index 616e776fd5..f71f675007 100644 --- a/core/continuum-core/src/bin/continuum.rs +++ b/core/continuum-core/src/bin/continuum.rs @@ -1087,7 +1087,11 @@ async fn reboot(options: RebootOptions) -> Result<(), String> { .map(|p| p.build_sha.clone()) .or_else(git_head_short_sha); let _deploy_claim = DeployClaimGuard::take(target_sha.as_deref().unwrap_or("unknown")); - stop_with(true).await?; + // Reboot deliberately does NOT fail on an unsaved module: the caller's goal is a + // running core, and refusing to continue would leave the node down over a module that + // could not flush. The warning is printed by `stop_with`; `stop` is the verb whose + // exit code carries it. + let _ = stop_with(true).await?; // Keep the launcher's wait as the honesty check that teardown actually took. let source = prebuilt .as_ref() @@ -2845,8 +2849,205 @@ fn start_log_report(logfile: &str) -> String { } /// `continuum stop` — stop the running core (the detached session started by `continuum start`). +/// How long the core gets to drain, save and join before the CLI stops waiting. +/// +/// `Runtime::shutdown` runs three 2s-bounded phases per module in parallel, so a healthy +/// stop is ~6s worst case; the extra room is for the response to travel back. A stop that +/// exceeds this is not assumed dead — it is assumed UNKNOWN, and the caller says so. +const GRACEFUL_STOP_BUDGET: std::time::Duration = std::time::Duration::from_secs(20); + +/// What the graceful request achieved, if anything. +enum GracefulStop { + /// The core ran the broadcast and every module's state reached disk. + Durable(String), + /// The core ran the broadcast and something did NOT save. The message names it. + /// The process is still stopping; the operator needs the names, not a retry. + Incomplete(String), + /// No answer — a core too wedged to answer, or a response that did not arrive inside + /// the budget. NOT the same as "it stopped": the caller must still tear the process + /// down, and must not report a clean stop. + NoAnswer(String), + /// The running core PREDATES this rail: it has no `system/shutdown` verb, so it was + /// never going to save and no fix in this binary can change that. + /// + /// Distinct from `NoAnswer` on purpose. Both are non-durable, but they are different + /// facts and an operator needs to tell them apart: this one is the expected, one-time + /// cost of the FIRST upgrade — the core being replaced was built before the rail + /// existed — whereas `NoAnswer` is a core that HAS the capability and would not use + /// it, which is a fault. Collapsing them would make every first rollout look like a + /// malfunction, and every malfunction look like a rollout. + LegacyCore(String), + /// Nothing was listening in the first place. Distinct from `NoAnswer` on purpose: a + /// core that never ran lost nothing, so `stop` on an already-stopped node must exit 0 + /// rather than announce a data loss that did not happen. The sweep below still runs — + /// an unanswering socket is not proof that no process survives. + NothingRunning, +} + +impl GracefulStop { + /// A legacy core cannot attest a final checkpoint. Keep it running until the + /// operator can stop it explicitly and preserve/adopt its selected checkpoint. + fn ensure_teardown_supported(&self, reboot: bool) -> Result<(), String> { + if let (true, Self::LegacyCore(reason)) = (reboot, self) { + return Err(format!( + "reboot refused before teardown: {reason}. The existing core is still running. \ + Use `continuum stop`, preserve the outgoing checkpoints, and use \ + `continuum checkpoint inspect` / `continuum checkpoint adopt` where recovery \ + is needed before running `continuum reboot` again." + )); + } + Ok(()) + } +} + +/// Ask the running core to stop ITSELF, so every module's `save_state` runs. +/// +/// This is the whole point of the rail. A kill — `taskkill /F` on Windows, a signal-less +/// tree kill elsewhere — runs no module's save, so the citizens' volatile state is lost +/// on every ordinary `stop`, and the CLI reported success because a dead process is +/// indistinguishable from a cleanly stopped one when the only thing you check is whether +/// it is gone. +/// +/// The request travels the same socket path `ping` uses, so there is no new transport and +/// no Windows-specific arrangement. +async fn request_graceful_stop() -> GracefulStop { + // Ask whether anything is listening BEFORE spending the stop budget on a socket + // nobody holds. Without this, `stop` on an already-stopped node waits the full + // graceful budget and then reports state loss — 20 seconds to be told, wrongly, that + // a core which never ran failed to save. + // + // BUT A PING TIMEOUT IS NOT PROOF OF ABSENCE. `core_is_up` returns false for a core + // that is running and too wedged to answer — which is the case where state is most + // likely to be lost, and reporting it as "nothing was running" would exit 0 and tell + // the operator nothing was at stake. So absence must be corroborated by the pidfile: + // no answer AND no pidfile is an empty node; no answer WITH a pidfile is a core that + // would not speak, and that is `NoAnswer`. (The pidfile can also be stale, which is + // why this decides the LABEL only — the sweep below runs either way.) + if !core_is_up().await { + // A MISSING PIDFILE IS NOT PROOF THAT NOTHING IS RUNNING. It was, in the first + // version of this, and that is the same error as reading a ping timeout as an + // empty node — one layer down. A core started outside the pidfile's owner, one + // whose file was removed, or a second core that never wrote one, all present as + // "no record" while still holding the socket and a citizen's unsaved state. + // + // So absence is only claimed when the sweep can also find no process. That is + // what `running_core_pids` answers, and it is the same enumeration the sweep + // below acts on — one instrument, so the label and the action cannot disagree. + // + // It can still be wrong in the SAFE direction: `processes_named` returning empty + // because the enumerator itself failed reads as "nothing running". That is worth + // naming rather than hiding — it is the residual case, and it is why the teardown + // below runs regardless of what this decides. + // THREE OUTCOMES, not two. Each piece of evidence can say "yes", "no", or "I + // could not look", and only the last of those may not be rounded to "no". + let pidfile = pidfile_for(&socket_path()); + let recorded = match std::fs::read_to_string(&pidfile) { + Ok(c) => c.trim().parse::().is_ok(), + // The ONLY error that means absence. A permission error, a busy file, a + // path on a filesystem that went away — those mean we did not find out, and + // `.ok()` used to flatten every one of them into "no record". + Err(e) if e.kind() == std::io::ErrorKind::NotFound => false, + Err(e) => { + return GracefulStop::NoAnswer(format!( + "could not read {}: {e} — whether a core is running is unknown, so this stop is not being called clean", + pidfile + )); + } + }; + // ONE snapshot answers both questions, so they cannot describe different moments. + let evidence = match core_process_evidence() { + Ok(evidence) => evidence, + Err(error) => { + return GracefulStop::NoAnswer(format!( + "could not establish whether a core is running: {error}" + )); + } + }; + let survivors = !evidence.core_pids.is_empty(); + return if recorded || survivors { + GracefulStop::NoAnswer( + "a core process is present but did not answer a ping".to_string(), + ) + } else { + // Both instruments looked, and both found nothing. + GracefulStop::NothingRunning + }; + } + let conn = connection(); + let cmds = conn.commands(); + let req = cmds.execute_value("system/shutdown", Value::Object(Default::default())); + match tokio::time::timeout(GRACEFUL_STOP_BUDGET, req).await { + Ok(Ok(value)) => { + let durable = value + .get("state_is_durable") + .and_then(Value::as_bool) + // A response whose shape we do not recognise is not a durable stop. An + // older core answering `system/shutdown` with something else must not be + // read as a clean save. + .unwrap_or(false); + let summary = value + .get("summary") + .and_then(Value::as_str) + .unwrap_or("the core answered without a summary") + .to_string(); + if durable { + GracefulStop::Durable(summary) + } else { + GracefulStop::Incomplete(summary) + } + } + Ok(Err(e)) => { + // A core that does not KNOW the verb is not a core that refused it. The + // command surface answers an unknown name with a "no handler"-shaped error, + // and during the first upgrade that is exactly what the outgoing core says — + // it was built before this rail existed. + let unknown = { + // `e` is a ClientError, not a String — formatted first so the match is on + // the rendered message the transport actually produced. + let m = format!("{e}").to_lowercase(); + m.contains("unknown command") + || m.contains("no handler") + || m.contains("not found") + || m.contains("unsupported") + }; + if unknown { + GracefulStop::LegacyCore(format!( + "the running core has no system/shutdown verb ({e}) — it predates this rail, so its modules were never going to save and this stop cannot be called durable" + )) + } else { + GracefulStop::NoAnswer(format!("the core refused the request: {e}")) + } + } + Err(_) => GracefulStop::NoAnswer(format!( + "no answer within {}s", + GRACEFUL_STOP_BUDGET.as_secs() + )), + } +} + async fn stop() -> Result<(), String> { - stop_with(false).await + // The operator verb answers with its EXIT CODE. `stop` returning 0 has meant only + // "the process is gone"; it now means "the process is gone AND every module's state + // reached disk", which is the question anyone typing `stop` before an upgrade is + // actually asking. A stop that could not save is a failure the shell can see. + match stop_with(false).await? { + // Nothing ran, so nothing was lost. Exiting non-zero here would report a data + // loss that did not happen, and `stop` is used in scripts that would then treat a + // clean no-op as a failure. + GracefulStop::Durable(_) | GracefulStop::NothingRunning => Ok(()), + GracefulStop::Incomplete(summary) => Err(format!( + "the core stopped but its state is not durable: {summary}" + )), + GracefulStop::NoAnswer(why) => Err(format!( + "the core was forced down without saving ({why}) — volatile state since the last save-on-write is gone" + )), + // Also a failure, and deliberately so: the operator asked for a stop and did not + // get a durable one. It is EXPECTED once, on the upgrade that installs the rail, + // and the message says which it is rather than making a rollout look like a fault. + GracefulStop::LegacyCore(why) => Err(format!( + "the outgoing core could not stop gracefully ({why}); this is the one-time cost of installing the shutdown rail, and it is still not a durable stop" + )), + } } /// The one teardown, parameterized by lane fate. `keep_lanes: true` is the @@ -2859,7 +3060,36 @@ async fn stop() -> Result<(), String> { /// core swap alone (Joel 2026-08-23: "if it's taking so long we need to fix /// that first"). The standalone `stop` verb keeps FULL teardown — an operator /// who says stop means everything. -async fn stop_with(keep_lanes: bool) -> Result<(), String> { +async fn stop_with(keep_lanes: bool) -> Result { + // ASK BEFORE KILLING. Everything below this point is a kill, and a kill runs no + // module's `save_state` — so before it, the core gets the chance to stop itself and + // report what reached disk. The kill still runs afterwards either way: a core that + // answered is already exiting and the sweep finds nothing, and a core that did not + // answer still has to go. What changes is that the operator is told which of those + // happened instead of reading the same success line for both. + let graceful = request_graceful_stop().await; + // `keep_lanes` IS the reboot flag — see this function's doc: "`keep_lanes: true` is the + // REBOOT path". Named `reboot` on the guard because that is the property it reasons about, + // and passed `keep_lanes` because today the two callers are exactly reboot(true)/stop(false). + // A third caller that wants lanes kept for some OTHER reason must pass the reboot-ness + // separately rather than reuse this argument. + graceful.ensure_teardown_supported(keep_lanes)?; + match &graceful { + GracefulStop::Durable(summary) => println!("core stopped gracefully — {summary}"), + GracefulStop::Incomplete(summary) => { + eprintln!("core stopped WITH UNSAVED STATE — {summary}") + } + GracefulStop::NoAnswer(why) => { + eprintln!("no graceful stop ({why}); forcing — modules did not save") + } + GracefulStop::NothingRunning => { + println!("no core is listening; sweeping for survivors") + } + GracefulStop::LegacyCore(why) => { + eprintln!("FIRST UPGRADE — {why}; forcing, and its volatile state is lost") + } + } + let socket = socket_path(); let pidfile = pidfile_for(&socket); // Resolve identity BEFORE any tree is killed. live_lane requires a matching @@ -2975,7 +3205,7 @@ async fn stop_with(keep_lanes: bool) -> Result<(), String> { if keep_lanes { println!(" leaving serving lane(s) up for adoption by the next core (reboot path)"); let _ = std::fs::remove_file(&socket); // socket cleanup still ours — only the lane fate changed - return Ok(()); + return Ok(graceful); } for outcome in continuum_core::inference::lane_registry::sweep_all() { use continuum_core::inference::lane_registry::SweepOutcome as S; @@ -3001,7 +3231,7 @@ async fn stop_with(keep_lanes: bool) -> Result<(), String> { } let _ = std::fs::remove_file(&socket); - Ok(()) + Ok(graceful) } /// Kill every owned engine process not descended from `keep`, reporting each @@ -3192,6 +3422,30 @@ fn usage() -> String { #[cfg(test)] mod tests { + // Regression for #3929: a first upgrade must leave the legacy core available + // for explicit checkpoint recovery, while normal reboot outcomes may continue. + #[test] + fn legacy_core_requires_explicit_recovery_before_reboot_teardown() { + use super::GracefulStop; + let legacy = GracefulStop::LegacyCore("no shutdown handler".into()); + let error = legacy.ensure_teardown_supported(true).unwrap_err(); + assert!( + legacy.ensure_teardown_supported(false).is_ok(), + "explicit stop remains available" + ); + assert!(error.contains("before teardown")); + assert!(error.contains("checkpoint inspect")); + assert!(error.contains("checkpoint adopt")); + for outcome in [ + GracefulStop::NothingRunning, + GracefulStop::Durable("saved".into()), + GracefulStop::Incomplete("save failed".into()), + GracefulStop::NoAnswer("timed out".into()), + ] { + assert!(outcome.ensure_teardown_supported(true).is_ok()); + } + } + // What this catches (card 9f160b78): missing/truncated process evidence and // a live PID-file process must never authorize offline memory replacement. #[test] diff --git a/core/continuum-core/src/cognition/mod.rs b/core/continuum-core/src/cognition/mod.rs index 73ae624112..9a37517b88 100644 --- a/core/continuum-core/src/cognition/mod.rs +++ b/core/continuum-core/src/cognition/mod.rs @@ -59,6 +59,7 @@ pub mod deliberation_parse; pub mod deliberation_prompt; pub mod dispatch_listener; pub mod activity_gate; +pub mod turn_ingress; pub mod dream_consolidation; pub mod embedding; pub mod eval; diff --git a/core/continuum-core/src/cognition/turn_ingress.rs b/core/continuum-core/src/cognition/turn_ingress.rs new file mode 100644 index 0000000000..e6dc75326d --- /dev/null +++ b/core/continuum-core/src/cognition/turn_ingress.rs @@ -0,0 +1,112 @@ +//! THE DOOR A STOPPING NODE CLOSES. +//! +//! One process-wide flag: may a persona's service loop begin a NEW turn? +//! +//! # Why this is not `quiesced` +//! +//! `PersonaAircRuntimeRegistry`'s per-persona `quiesced` flag suspends the autonomic +//! SELF-TICK and deliberately leaves her reachable — its contract is "she stops wandering +//! and still picks up the phone", because a measurement lease must not make a citizen +//! unreachable. That is the right behaviour for a benchmark and the wrong one for a stop: +//! a quiesced citizen still starts turns when addressed, so a `save_state` taken after a +//! quiesce can still land underneath a turn halfway through writing. +//! +//! So this is a different question with a different answer, and folding it into `quiesced` +//! would break the lease's promise. Closing this door stops EVERY new turn — directed, +//! self-directed, forked — and lets the ones already running finish. +//! +//! # One-way, on purpose +//! +//! There is no `open()`. The only caller is the shutdown drain, and the process exits +//! immediately afterwards; a reopen verb would exist solely to be called by mistake. A +//! node that wants turns again starts a core. + +use crate::runtime::AdmissionGate; + +/// The process's turn gate. One [`AdmissionGate`]: open/closed and the in-flight count in +/// one word, so an admission cannot slip between a close and the drain's read. +/// +/// The gate logic lives in `runtime::admission_gate` rather than here because the log +/// queue needs exactly the same invariant, and when it was written twice the second copy +/// reintroduced the race the first had just removed. +static GATE: AdmissionGate = AdmissionGate::new(); + +/// SERVICE-LOOP turns running right now, across every persona in this process. +/// +/// # What this number does NOT include, stated because a drain keys on it +/// +/// Only work that passed through [`admit`] is counted, and today that is the persona +/// service loop. A `cognition/eval` fork, or any caller that reaches the cognition +/// faculties directly rather than through a citizen's loop, runs UNCOUNTED — so a drain +/// can report the node quiet while such a call is mid-flight. +/// +/// That is a narrowed contract, not an oversight, and it is narrowed rather than widened +/// because the alternative is worse: an eval fork holding a permit would keep the whole +/// node from draining for the length of a benchmark, and a permit taken somewhere that +/// does not release it on every path leaks a phantom turn that no drain can ever clear. +/// The service loop is the one caller whose entry and exit are a single bounded scope. +/// +/// The consequence a reader must not be surprised by: a stop taken DURING an eval saves a +/// consistent citizen but may cut the eval. Widening this is a real piece of work — +/// permits at the faculty boundary, with the same RAII discipline — and it belongs to +/// whoever gives `cognition/eval` a lifecycle, not to a shutdown rail. Raised in review +/// by Astra. +pub fn in_flight() -> u64 { + GATE.in_flight() +} + +/// ADMIT a turn, or refuse because the node is stopping. +/// +/// # Why not read `activity_gate`'s engaged flag +/// +/// `persona_engaged` is stamped where a serving LANE is acquired, deliberately — a room +/// wake alone is not wakefulness, or a busy room would cancel every dream. So a turn that +/// has been admitted and is still composing context, or is queued for a lane, is not yet +/// `engaged`. Draining on that flag would walk past exactly the turns that have taken +/// input and not yet written anything, which are the ones whose loss is invisible. +#[must_use = "the permit IS the turn's admission; dropping it immediately ends the turn"] +pub fn admit() -> Option> { + GATE.admit() +} + +/// Shut the door. Returns whether THIS call closed it, so a second drain — a signal +/// racing the `system/shutdown` verb, since both reach the same broadcast — can tell it is +/// re-entering rather than report a fresh close. +pub fn close() -> bool { + GATE.close() +} + +#[cfg(test)] +mod tests { + use super::*; + + // what this catches: the process gate being wired to something other than an open + // AdmissionGate — a gate that defaulted closed would make every citizen refuse to take + // a turn, a total silent outage no shutdown test would exercise. + // + // The gate's BEHAVIOUR — the admit/close race, the CAS retry, the shared word — is + // tested against the real type in `runtime::admission_gate`, on instances a test can + // close without poisoning the process. This module's job is only to hold one and hand + // it to the service loop, so that is all this asserts. + #[test] + fn the_process_turn_gate_starts_open_and_empty() { + // Openness is asserted by ADMITTING, not by a separate reader: a bare `is_open` + // on this module had no caller at all — the service loop takes a permit — so it + // was dead weight and is gone. + // + // It was NOT the reachability-ratchet failure, though I first claimed it was. The + // scanner greps `pub struct ` / `pub enum ` (production_reachability.rs:125) and + // does not count functions at all, so removing one could never have moved the + // number. The counted item was `pub struct ShutdownOperation`, now private. + // Corrected by Astra, who READ the scanner rather than inferring what it measures + // — which is what I should have done before claiming a fix for it. + let permit = admit().expect("an open gate admits"); + assert_eq!( + in_flight(), + 1, + "an admitted turn must be visible to a drain" + ); + drop(permit); + assert_eq!(in_flight(), 0, "and invisible once it ends"); + } +} diff --git a/core/continuum-core/src/commands/log/write.rs b/core/continuum-core/src/commands/log/write.rs index 721613b40b..94cbd78d68 100644 --- a/core/continuum-core/src/commands/log/write.rs +++ b/core/continuum-core/src/commands/log/write.rs @@ -40,9 +40,9 @@ crate::action_command! { params: WriteLogPayload, output: WriteLogResult, run(this, _ctx, p) => { - this.state - .log_tx - .send(p) + // Through the counting choke point, not the raw sender: the shutdown drain waits + // on that depth, and an entry enqueued around it is one the drain cannot see. + crate::modules::logger::enqueue_log_blocking(&this.state.log_tx, p) .map_err(|e| format!("Queue send failed: {e}"))?; this.state.requests_processed.fetch_add(1, Ordering::Relaxed); Ok(WriteLogResult { bytes_written: 0 }) diff --git a/core/continuum-core/src/commands/log/write_batch.rs b/core/continuum-core/src/commands/log/write_batch.rs index 879cc907df..3822729ee9 100644 --- a/core/continuum-core/src/commands/log/write_batch.rs +++ b/core/continuum-core/src/commands/log/write_batch.rs @@ -51,7 +51,9 @@ crate::action_command! { let entries_queued = p.entries.len(); for entry in p.entries { // Best-effort: drop on a full queue rather than block the submitter. - let _ = this.state.log_tx.try_send(entry); + // Counted: see `enqueue_log`. A batch that bypassed it hid its whole size + // from the drain. + let _ = crate::modules::logger::enqueue_log(&this.state.log_tx, entry); } this.state.requests_processed.fetch_add(1, Ordering::Relaxed); Ok(WriteLogBatchResult { entries_queued }) diff --git a/core/continuum-core/src/commands/system.rs b/core/continuum-core/src/commands/system.rs index 2d1bab5ab9..2e50f7bff8 100644 --- a/core/continuum-core/src/commands/system.rs +++ b/core/continuum-core/src/commands/system.rs @@ -28,6 +28,10 @@ pub mod memory_gate; pub mod pressure; pub mod pressure_broker_state; pub mod resources; +// `system/shutdown` holds no deps, so `action_command!`'s stateless arm registers it +// onto the one registry by itself — it is deliberately NOT in `command_objects` below, +// which is only for commands that need construction with a service. +pub mod shutdown; /// Shared params for the no-argument `system/*` reads (cpu, memory, pressure, /// memory-gate, memory-budget, docker-tier-stats). One empty contract reused across diff --git a/core/continuum-core/src/commands/system/shutdown.rs b/core/continuum-core/src/commands/system/shutdown.rs new file mode 100644 index 0000000000..cddb48251f --- /dev/null +++ b/core/continuum-core/src/commands/system/shutdown.rs @@ -0,0 +1,147 @@ +//! `system/shutdown` — ask the RUNNING core to stop itself gracefully and say what +//! reached disk. +//! +//! # Why a resident verb rather than a signal +//! +//! `continuum stop` reached for `taskkill /F` on Windows and a kill tree elsewhere. A +//! forced kill runs no module's `save_state`, so a stop that looked identical to a clean +//! one lost every module's volatile state — and the CLI printed a success line and exited +//! 0 either way, because nothing came back from the kill except "the process is gone". +//! +//! Signals do exist and `main.rs` wires them, but they are not a request/response: the +//! handler cannot hand an exit code or a receipt back to the operator who typed `stop`, +//! and on Windows the console-control arms are best-effort. This verb travels the socket +//! request path every other command already uses, so the answer comes back the same way +//! `ping`'s does — no new transport, no Windows-specific rail. +//! +//! `Privileged`, never `AiSafe`: stopping the node is not a thing a persona reasoning +//! about its own workload should be able to reach for. + +use serde::{Deserialize, Serialize}; +use ts_rs::TS; + +use crate::runtime::ShutdownReceipt; + +use super::SystemQuery; + +/// How the core answered a stop request. Carries the receipt so the caller can exit +/// non-zero and NAME what did not save, rather than reporting a success it never +/// established. +#[derive(Debug, Clone, Serialize, Deserialize, TS)] +#[ts( + export, + export_to = "../../../protocol/typescript/system/ShutdownResult.ts" +)] +pub struct ShutdownResult { + /// Every module's durable state reached disk. The one field a caller may key an exit + /// code on. + pub state_is_durable: bool, + /// One line naming what did not save, if anything. + pub summary: String, + pub receipt: ShutdownReceipt, +} + +crate::action_command! { + /// Drain, save and join every module. Return the retained shutdown receipt before + /// the calling CLI tears down the process. + pub struct SystemShutdown; + name: "system/shutdown", + access: Privileged, + params: SystemQuery, + output: ShutdownResult, + run(_this, _ctx, _p) => { + // TRIGGER AND OBSERVE — the operation itself belongs to the runtime. + // + // This handler used to run the broadcast inline, and a socket handler is + // cancelled when its client goes away. A CLI that died or was interrupted + // mid-request therefore left the node with turn ingress CLOSED and no shutdown: + // every citizen refusing work, nothing saved, nothing to restart it. Strictly + // worse than the forced kill this verb exists to replace. `begin_shutdown` runs + // it in a task no connection owns, and is idempotent, so a signal racing this + // request joins the same broadcast rather than saving twice over one state. + // Refuse BEFORE triggering when no runtime is installed. `begin_shutdown` + // publishes an empty receipt in that case so no observer hangs — but an empty + // receipt is "all zero modules were durable", which is true and says nothing, and + // the caller reads a success here as permission to stop hard-killing. A core in + // its boot window has state at stake and no one to save it; that is a refusal, + // not a clean stop. + if crate::runtime::signal_runtime().is_none() { + return Err(crate::sdk_codegen::CommandError::Internal( + "no runtime is installed yet — the core cannot stop gracefully, so nothing here says its state is durable" + .to_string(), + )); + } + let rx = crate::runtime::begin_shutdown(); + + // Bounded so a wedged module cannot hold the connection open forever. A timeout + // here does NOT cancel the shutdown — it is still running, un-cancelled, in the + // runtime's task — so the honest answer is that durability is unknown, not that + // it failed. + const OBSERVE: std::time::Duration = std::time::Duration::from_secs(15); + let Some(receipt) = crate::runtime::await_shutdown(rx, OBSERVE).await else { + return Err(crate::sdk_codegen::CommandError::Internal(format!( + "shutdown is still running after {}s — it was NOT cancelled, and whether every module saved is unknown from here", + OBSERVE.as_secs() + ))); + }; + + // The process is deliberately still alive. Every module has drained, saved and + // joined; what a stopped node is FOR is the caller's decision — the CLI tears it + // down having read this receipt, and the signal path exits because it must. + // + // The previous version exited here on a 400ms timer, which proved nothing about + // whether this response had reached the wire: a handler that kills its own + // process cannot also answer, and a timer guessing when its answer flushed is a + // guess. Handing the teardown back to the caller removes the question instead of + // estimating it. + Ok(ShutdownResult { + state_is_durable: receipt.state_is_durable(), + summary: receipt.summary(), + receipt, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::sdk_codegen::{ActionCommand, Ctx}; + + // what this catches: the access level. `system/shutdown` on the AiSafe surface would + // let any persona stop the node it is thinking inside. + #[test] + fn stopping_the_node_is_privileged_not_ai_safe() { + assert_eq!(SystemShutdown::NAME, "system/shutdown"); + assert!(matches!( + SystemShutdown::ACCESS, + crate::sdk_codegen::AccessLevel::Privileged + )); + } + + // what this catches: a core with no runtime answering "stopped cleanly". The caller + // treats a successful response as permission to stop hard-killing, so a success here + // would be read as "state is safe" about a core that never ran a save. + #[tokio::test] + async fn without_a_runtime_it_refuses_rather_than_claiming_a_clean_stop() { + // `signal_runtime()` is a process-wide OnceLock. In a test binary that has not + // installed one this is None; if another test in the same process installed it, + // this assertion would be about a live runtime, so it checks the branch it can. + if crate::runtime::signal_runtime().is_none() { + let err = SystemShutdown + .run(&Ctx::default(), SystemQuery {}) + .await + .expect_err("no runtime must refuse"); + // Matched on the VARIANT, not on formatted text: `Internal` is the category + // that says "the node could not do this", and a future refactor that turned + // this into `Invalid` (a caller error) would be a real behaviour change the + // test should catch rather than paper over with a substring match. + let crate::sdk_codegen::CommandError::Internal(msg) = err else { + panic!("a core that cannot stop gracefully is an Internal failure, not a caller error"); + }; + assert!( + msg.contains("durable"), + "the refusal must say what is unestablished, got: {msg}" + ); + } + } +} diff --git a/core/continuum-core/src/main.rs b/core/continuum-core/src/main.rs index 6518549059..451acceba2 100644 --- a/core/continuum-core/src/main.rs +++ b/core/continuum-core/src/main.rs @@ -140,9 +140,20 @@ fn install_shutdown_handlers() { _ = close_fut => {} _ = shutdown_fut => {} } - eprintln!("[continuum-core] shutdown signal — killing sentinel process groups"); + eprintln!("[continuum-core] shutdown signal — save-and-join broadcast, then exit"); + // THE SAME BROADCAST THE UNIX ARMS RUN. This arm killed sentinels, slept a + // flat 2 seconds, and `_exit`ed — which is precisely the behaviour the SIGTERM + // arm's own comment says was replaced on 2026-09-02 ("a flat 2s sleep during + // which NOTHING saved"). The unix half was fixed and this half was not, so on + // Windows no module has ever saved on a signal stop: not the citizens' + // workspaces, not the log queue, nothing. The node that runs the citizens is a + // Windows node. + // + // Ordering matches unix deliberately: modules drain and save FIRST, sentinels + // are killed after (they are children, not modules), then the fast `_exit` + // that skips llama.cpp's double-free-prone static destructors. + continuum_core::runtime::run_signal_shutdown().await; continuum_core::modules::sentinel::shutdown_all_sentinels(); - tokio::time::sleep(std::time::Duration::from_secs(2)).await; unsafe { libc::_exit(0) }; }); } diff --git a/core/continuum-core/src/modules/cognition.rs b/core/continuum-core/src/modules/cognition.rs index 1b9d62ee49..251947c1b9 100644 --- a/core/continuum-core/src/modules/cognition.rs +++ b/core/continuum-core/src/modules/cognition.rs @@ -216,6 +216,52 @@ impl ServiceModule for CognitionModule { } } + /// CLOSE THE DOOR AND WAIT FOR THE CITIZENS TO FINISH THEIR TURNS. + /// + /// This is the module the drain phase exists for. Everything else in a stopping node + /// either holds no work or holds a buffer; cognition holds a citizen mid-thought, and + /// a `save_state` taken across that boundary writes a working memory that is half of + /// one turn and half of the next. + /// + /// Two steps, in this order: + /// + /// 1. `turn_ingress::close()` — no service loop admits another turn. NOT + /// `quiesce_all`, whose contract is the opposite: a quiesced citizen stops + /// wandering and deliberately STAYS reachable, so she would keep starting turns + /// while we tried to save her. + /// 2. Wait for the turns already admitted to finish, bounded. + /// + /// The count comes from the permits the loops hold, not from `activity_gate`'s + /// `engaged` flag: engagement is stamped at serving-lane acquisition, so a turn that + /// has taken a message and is still composing context is not engaged, and draining on + /// that flag would walk past exactly the turns whose loss leaves no trace. + /// + /// Returns the number still running when the budget expired — a citizen who is still + /// thinking is a fact the receipt reports, not one it waits forever for. + async fn drain(&self) -> Result { + const POLL: std::time::Duration = std::time::Duration::from_millis(50); + // Inside the runtime's 2s phase bound, so the deadline that fires is this one, + // with a count, rather than the outer timeout, which produces no number. + const BUDGET: std::time::Duration = std::time::Duration::from_millis(1_800); + + let first_to_close = crate::cognition::turn_ingress::close(); + let deadline = std::time::Instant::now() + BUDGET; + while std::time::Instant::now() < deadline { + if crate::cognition::turn_ingress::in_flight() == 0 { + return Ok(0); + } + tokio::time::sleep(POLL).await; + } + let left = crate::cognition::turn_ingress::in_flight(); + crate::probe!( + class = "cognition.drain.incomplete", + in_flight = left, + first_to_close = first_to_close, + "citizens were still inside turns when the drain budget expired — whatever is saved next is a snapshot taken mid-turn, and the receipt says so" + ); + Ok(left.min(u32::MAX as u64) as u32) + } + async fn initialize(&self, _ctx: &ModuleContext) -> Result<(), String> { // No init needed. Recipes are JSON data walked by the host // (TS recipe loader for the chat path today; future Rust diff --git a/core/continuum-core/src/modules/logger.rs b/core/continuum-core/src/modules/logger.rs index 26d858ce78..e4535db6e6 100644 --- a/core/continuum-core/src/modules/logger.rs +++ b/core/continuum-core/src/modules/logger.rs @@ -45,6 +45,96 @@ use ts_rs::TS; /// Uses SyncSender with try_send() for GUARANTEED non-blocking. static GLOBAL_LOG_SENDER: OnceLock> = OnceLock::new(); +/// The log queue's admission gate: is the queue accepting, and how many entries are in +/// it, in ONE word. +/// +/// Depth is distinct from `LoggerCommandState::pending_writes`, which counts entries the +/// writer has already WRITTEN and not yet flushed. Both are needed to answer "is this +/// module drained": one covers the channel, the other the file buffers, and a stop that +/// checked only the second would flush an empty buffer and report success while entries +/// were still queued behind it. `SyncSender` exposes no length, so depth is counted +/// rather than inferred. +/// +/// This was a separate `ADMITTING` bool beside a separate counter, which is the race the +/// turn gate had just been fixed for — check admission, close lands, drain reads zero, +/// reservation increments a queue whose writer is being joined. It is the SAME type as +/// the turn gate now, so the invariant has one implementation instead of two that must be +/// kept agreeing. +static QUEUE: crate::runtime::AdmissionGate = crate::runtime::AdmissionGate::new(); + +/// Stop accepting new log entries. Returns whether THIS call closed admission. +pub fn close_log_admission() -> bool { + QUEUE.close() +} + +/// Entries sitting in the channel right now. Read by the drain phase of shutdown. +pub fn queued_log_entries() -> u64 { + QUEUE.in_flight() +} + +/// The WRITER's release: one entry has left the channel AND been written. +/// +/// Paired with the `forget` in the enqueue path — the producer hands the count to the +/// writer, and the writer gives it back only after the write, so an entry being written +/// is still in flight and a drain waiting on zero cannot race the write it is waiting for. +fn release_queued_entry() { + // Constructing a permit to drop it would be clearer, but `Permit` borrows the gate and + // exists only to be RAII; the writer's release is a plain decrement of the same word. + QUEUE.release_one(); +} + +/// THE ONLY WAY INTO THE LOG QUEUE. Returns whether the entry was accepted. +/// +/// One choke point on purpose: the depth was first counted inside `queue_log`, and +/// `log/write` and `log/write-batch` send straight down `state.log_tx`, so two of the +/// three producers were invisible and the drain would call the module quiet with entries +/// still queued behind it. Anything that can enqueue must come through here, or the +/// counter is decoration. +pub fn enqueue_log_blocking( + sender: &mpsc::SyncSender, + payload: WriteLogPayload, +) -> Result<(), mpsc::SendError> { + // The BLOCKING form, for `log/write`, whose caller is waiting on a result and must + // not have its entry silently dropped when the queue is full. Same counter, same + // rule — count only what the channel accepted. + // ADMISSION AND RESERVATION ARE ONE STEP. Checking a flag and then incrementing a + // separate counter lets a close land between them: the drain reads zero, declares the + // module quiet, and this entry then joins a queue whose writer is being joined. + // + // Reserved BEFORE the send, too: the writer runs concurrently, so an entry published + // before its count lands can be popped and decremented first, underflowing the depth + // and making the drain wait forever. + // ADMISSION AND RESERVATION ARE ONE STEP. Checking a flag and then incrementing a + // separate counter lets a close land between them: the drain reads zero, declares the + // module quiet, and this entry then joins a queue whose writer is being joined. + let Some(permit) = QUEUE.admit() else { + return Err(mpsc::SendError(payload)); + }; + if let Err(e) = sender.send(payload) { + return Err(e); // permit drops here: the entry was never queued, so release it + } + // The entry is now the WRITER's to release, not ours — it stays counted until the + // writer has written it. `forget` transfers that ownership; dropping here would + // uncount an entry that is still in the channel. + std::mem::forget(permit); + Ok(()) +} + +pub fn enqueue_log(sender: &mpsc::SyncSender, payload: WriteLogPayload) -> bool { + // Counted only on a SUCCESSFUL send: a dropped entry was never queued, and counting + // it leaves a depth that never returns to zero and a drain that can never complete. + // Same single-step admission+reservation as the blocking form, for the same reason. + let Some(permit) = QUEUE.admit() else { + return false; + }; + if sender.try_send(payload).is_ok() { + std::mem::forget(permit); // handed to the writer; see the blocking form + true + } else { + false // permit drops: refused by the channel, so it was never queued + } +} + /// Channel capacity - if full, new messages dropped (NEVER blocks) const CLOG_CHANNEL_CAPACITY: usize = 4096; @@ -74,7 +164,7 @@ pub fn queue_log(category: &str, level: LogLevel, component: &str, message: &str }; // GUARANTEED NON-BLOCKING: try_send returns immediately // If channel full, message dropped - NEVER blocks caller - let _ = sender.try_send(payload); + let _ = enqueue_log(sender, payload); } // If GLOBAL_LOG_SENDER not set, silently drop (LoggerModule not initialized yet) } @@ -443,7 +533,61 @@ fn format_log_entry(payload: &WriteLogPayload, timestamp: &str) -> String { } } -fn flush_all(file_cache: &FileCache) { +/// Failures the writer could not report, counted so a STOP can. +/// +/// The writer thread runs with nowhere to return an error to, so a failed write or flush +/// was printed to stderr and dropped. That is defensible while the process runs — a log +/// line is not worth killing a node over — and indefensible at shutdown, where the module +/// then returned `Ok(())` and the receipt said `Clean` over writes that never landed. +/// +/// PER-INSTANCE, not a global. As a `static` it could not be tested: driving a real write +/// failure would have left the count non-zero for the whole test binary, so every later +/// `shutdown()` would report a dirty stop. That is the third time in this rail that a +/// process-wide global was the reason a test had to be fake rather than the reason it was +/// hard — see `AdmissionGate` and `ShutdownOperation`. +pub type WriteFailures = Arc; + +/// Whether a stop may be reported clean, given what this module failed to write. +/// +/// Pure, and separate from `shutdown` so it can be tested against a real failure count +/// without constructing a module around env vars. The decision is the load-bearing part: +/// with only a fully completed stop supporting durability, a module that cannot confirm +/// its content reached disk must fail the stop rather than let the CLI's exit code claim +/// the citizens' logs are safe. +/// WRITE AND LATCH: attempt one log write and record the failure if it does not land. +/// +/// One operation, shared by the writer thread and by tests, because a test that performs +/// the latch ITSELF proves only that a counter can be incremented — not that production +/// increments it. Deleting the production increments would have left the earlier version +/// of the logger regression green. Astra's discriminator, and she was right. +fn write_and_latch( + payload: &WriteLogPayload, + log_dir: &str, + continuum_root: &str, + file_cache: &FileCache, + headers_written: &HeaderTracker, + failures: &WriteFailures, +) -> bool { + match write_log_message(payload, log_dir, continuum_root, file_cache, headers_written) { + Ok(_) => true, + Err(e) => { + failures.fetch_add(1, Ordering::Relaxed); + eprintln!("❌ LoggerModule write error: {e}"); + false + } + } +} + +fn stop_outcome(failures: u64) -> Result<(), String> { + if failures > 0 { + return Err(format!( + "{failures} log write(s)/flush(es) failed in this process — some log content did not reach disk" + )); + } + Ok(()) +} + +fn flush_all(file_cache: &FileCache, failures: &WriteFailures) { let handles: Vec = { let cache = file_cache.lock().unwrap_or_else(|e| e.into_inner()); cache.values().cloned().collect() @@ -451,7 +595,13 @@ fn flush_all(file_cache: &FileCache) { for locked_file in handles { let mut file = locked_file.lock().unwrap_or_else(|e| e.into_inner()); - let _ = file.flush(); + // A flush that failed at shutdown is the whole reason `shutdown` exists for this + // module. Swallowing it here and returning Ok() above made the receipt claim + // durability for content still sitting in a buffer that never reached disk. + if let Err(e) = file.flush() { + failures.fetch_add(1, Ordering::Relaxed); + eprintln!("❌ LoggerModule flush error: {e}"); + } } } @@ -477,6 +627,9 @@ pub struct LoggerCommandState { pub started_at: Instant, pub requests_processed: AtomicU64, pub pending_writes: Arc, + /// Writes and flushes this module could not complete. Non-zero means some log content + /// is gone, and a stop must not be reported as clean. + pub write_failures: WriteFailures, } #[cfg(test)] @@ -492,12 +645,25 @@ impl LoggerCommandState { started_at: Instant::now(), requests_processed: AtomicU64::new(0), pending_writes: Arc::new(AtomicU64::new(0)), + write_failures: Arc::new(AtomicU64::new(0)), }); (state, rx) } } impl LoggerModule { + /// Build a module around an existing state, so a test can drive the REAL `shutdown` + /// over a failure count it produced through the real write path. `new()` reads env + /// vars and spawns a writer thread; a test needs neither, and needs the state it can + /// see. + #[cfg(test)] + pub(crate) fn with_state(state: Arc) -> Self { + Self { + log_dir: String::new(), + state, + } + } + pub fn new() -> Self { let continuum_root = std::env::var("CONTINUUM_ROOT").unwrap_or_else(|_| { let home = dirs::home_dir().expect("Failed to resolve home directory"); @@ -516,6 +682,7 @@ impl LoggerModule { let file_cache = Arc::new(Mutex::new(HashMap::new())); let headers_written = Arc::new(Mutex::new(HashSet::new())); let pending_writes = Arc::new(AtomicU64::new(0)); + let write_failures: WriteFailures = Arc::new(AtomicU64::new(0)); // Create BOUNDED sync_channel for GUARANTEED non-blocking // try_send() returns immediately - if full, message dropped (NEVER blocks) @@ -530,6 +697,7 @@ impl LoggerModule { let writer_log_dir = log_dir.clone(); let writer_continuum_root = continuum_root.clone(); let writer_pending = pending_writes.clone(); + let writer_failures = write_failures.clone(); thread::spawn(move || { const FLUSH_INTERVAL: Duration = Duration::from_millis(250); @@ -542,15 +710,14 @@ impl LoggerModule { |payload: &WriteLogPayload, limiter: &mut RateLimiter, pending: &mut usize| { match limiter.check(&payload.category) { RateDecision::Allow => { - if let Err(e) = write_log_message( + write_and_latch( payload, &writer_log_dir, &writer_continuum_root, &writer_file_cache, &writer_headers, - ) { - eprintln!("❌ LoggerModule write error: {e}"); - } + &writer_failures, + ); *pending += 1; } RateDecision::Drop => {} @@ -565,22 +732,26 @@ impl LoggerModule { ), args: None, }; - let _ = write_log_message( + // Counted like every other write. It is the rate-limit + // warning — the line that explains why other lines are + // missing — so losing it silently is the one loss that also + // erases the evidence of the losses. + write_and_latch( &warning, &writer_log_dir, &writer_continuum_root, &writer_file_cache, &writer_headers, + &writer_failures, ); - if let Err(e) = write_log_message( + write_and_latch( payload, &writer_log_dir, &writer_continuum_root, &writer_file_cache, &writer_headers, - ) { - eprintln!("❌ LoggerModule write error: {e}"); - } + &writer_failures, + ); *pending += 2; } } @@ -589,20 +760,26 @@ impl LoggerModule { loop { match log_rx.recv_timeout(FLUSH_INTERVAL) { Ok(payload) => { + // Decremented AFTER the write, not on receipt: an entry popped + // off the channel and still being written is in flight, and a + // drain that stopped waiting at the pop would race the write it + // was waiting for. process_payload(&payload, &mut limiter, &mut pending); + release_queued_entry(); // Drain remaining messages non-blocking while pending < MAX_BATCH_BEFORE_FLUSH { match log_rx.try_recv() { Ok(payload) => { process_payload(&payload, &mut limiter, &mut pending); + release_queued_entry(); } Err(_) => break, } } if pending >= MAX_BATCH_BEFORE_FLUSH { - flush_all(&writer_file_cache); + flush_all(&writer_file_cache, &writer_failures); writer_pending.store(0, Ordering::Relaxed); pending = 0; } else { @@ -611,14 +788,14 @@ impl LoggerModule { } Err(mpsc::RecvTimeoutError::Timeout) => { if pending > 0 { - flush_all(&writer_file_cache); + flush_all(&writer_file_cache, &writer_failures); writer_pending.store(0, Ordering::Relaxed); pending = 0; } } Err(mpsc::RecvTimeoutError::Disconnected) => { if pending > 0 { - flush_all(&writer_file_cache); + flush_all(&writer_file_cache, &writer_failures); } break; } @@ -632,6 +809,7 @@ impl LoggerModule { started_at: Instant::now(), requests_processed: AtomicU64::new(0), pending_writes, + write_failures, }); Self { log_dir, state } @@ -677,10 +855,50 @@ impl ServiceModule for LoggerModule { crate::commands::log::command_objects(self.state.clone()) } + /// Wait for the log queue to empty, bounded. + /// + /// The first real implementation of the drain contract, and it is here because the + /// loss is concrete: `queue_log` hands entries to a writer THREAD, and `shutdown` + /// only ever flushed the open files. Anything still in the channel when the core + /// stopped was never written and never counted — including, on a bad stop, the log + /// lines explaining why it was stopping. + /// + /// Polls rather than signals: the writer is a plain `std::thread` on a blocking + /// `recv_timeout`, so there is no async completion to await, and its own flush + /// interval is 250ms. Returning the residual count rather than an error is what lets + /// the receipt say how many lines were lost instead of only that some were. + async fn drain(&self) -> Result { + // CLOSE ADMISSION FIRST. Waiting for the depth to reach zero while producers are + // still free to enqueue measures a moment, not a drain — the count can be zero + // and one `clog_*` later be one again, with the writer already joined. + close_log_admission(); + const POLL: std::time::Duration = std::time::Duration::from_millis(25); + // Bounded strictly inside the runtime's 2s phase bound, so the deadline that + // fires is this one — with a count — rather than the outer timeout, which + // produces no number at all. + const BUDGET: std::time::Duration = std::time::Duration::from_millis(1_500); + let deadline = std::time::Instant::now() + BUDGET; + while std::time::Instant::now() < deadline { + if queued_log_entries() == 0 && self.state.pending_writes.load(Ordering::Relaxed) == 0 { + return Ok(0); + } + tokio::time::sleep(POLL).await; + } + // Both halves count: entries still in the channel, plus writes the writer has + // made but not flushed. A drain that reported only the channel would call a + // module drained while its file buffers still held lines. + let left = queued_log_entries() + self.state.pending_writes.load(Ordering::Relaxed); + Ok(left.min(u32::MAX as u64) as u32) + } + async fn shutdown(&self) -> Result<(), String> { // Flush any pending writes - flush_all(&self.state.file_cache); - Ok(()) + flush_all(&self.state.file_cache, &self.state.write_failures); + // AND SAY SO IF ANY OF IT FAILED. Returning Ok() here made the receipt report + // `Clean` for a module whose writes and flushes had been failing all along — the + // errors went to stderr, which on a stopping node is nobody. A module that cannot + // confirm its content reached disk must not let the stop be called durable. + stop_outcome(self.state.write_failures.load(Ordering::Relaxed)) } fn as_any(&self) -> &dyn Any { @@ -737,4 +955,135 @@ mod tests { assert!(matches!(rl.check("test"), RateDecision::Allow)); assert!(matches!(rl.check("test"), RateDecision::Drop)); } + + /// The stop must not be called clean when log content did not reach disk. + mod write_failures_reach_the_receipt { + use super::*; + + // what this catches: a write failure being swallowed while the stop still reports + // clean. The writer thread has nowhere to return an error to, so failures went to + // stderr — which on a stopping node is nobody — and `shutdown()` returned `Ok(())` + // regardless. With only a fully completed stop supporting durability, that made + // the CLI's exit code claim the citizens' logs were safe when they were not. + // + // Drives the REAL `write_log_message` against a real unwritable destination, a + // real successful one after it, and the REAL decision `shutdown` delegates to. + #[tokio::test] + async fn a_failed_write_then_a_successful_one_still_ends_in_a_non_clean_stop() { + let tmp = tempfile::tempdir().unwrap(); + let failures: WriteFailures = Arc::new(AtomicU64::new(0)); + let cache: FileCache = Default::default(); + let headers: HeaderTracker = Default::default(); + let payload = WriteLogPayload { + category: "test".into(), + level: LogLevel::Warn, + component: "regression".into(), + message: "this write cannot land".into(), + args: None, + }; + + // A FILE where the log directory must be: the write genuinely cannot land. + let blocked = tmp.path().join("blocked"); + std::fs::write(&blocked, b"in the way").unwrap(); + // THE SHARED OPERATION — the same `write_and_latch` the writer thread calls. + // The earlier version called `write_log_message` and then did the + // `fetch_add` ITSELF, which proved a counter can be incremented rather than + // that production increments it: deleting the production latch would have + // left it green. + let landed = write_and_latch( + &payload, + &blocked.to_string_lossy(), + &tmp.path().to_string_lossy(), + &cache, + &headers, + &failures, + ); + assert!( + !landed, + "a log directory that is a FILE must fail the write, not silently succeed" + ); + assert_eq!( + failures.load(Ordering::Relaxed), + 1, + "the shared write+latch must record the failure — no manual increment here" + ); + + // A LATER write SUCCEEDS. This is the case that used to read as clean: the + // last write worked and the buffers flushed, and nothing carried the earlier + // loss forward. + let ok_dir = tmp.path().join("writable"); + std::fs::create_dir_all(&ok_dir).unwrap(); + let good = write_and_latch( + &payload, + &ok_dir.to_string_lossy(), + &tmp.path().to_string_lossy(), + &cache, + &headers, + &failures, + ); + assert!(good, "a writable destination must still work"); + assert_eq!( + failures.load(Ordering::Relaxed), + 1, + "a SUCCESSFUL write must not add a failure — the count is the earlier loss" + ); + + // THE POINT: the REAL `shutdown()` on a REAL module holding that count is not + // clean. Reading `stop_outcome` directly, as the earlier version did, would + // have stayed green if `shutdown` stopped delegating to it. + let (log_tx, _rx) = mpsc::sync_channel::(8); + let state = Arc::new(LoggerCommandState { + log_tx, + file_cache: cache.clone(), + started_at: Instant::now(), + requests_processed: AtomicU64::new(0), + pending_writes: Arc::new(AtomicU64::new(0)), + write_failures: failures.clone(), + }); + let module = LoggerModule::with_state(state); + let err = module + .shutdown() + .await + .expect_err("a module that lost log content must not stop clean"); + assert!( + err.contains('1') && err.contains("did not reach disk"), + "the failure must be counted and named, got: {err}" + ); + } + + // what this catches: the opposite error — a module with no failures refusing to + // stop cleanly, which would make every ordinary shutdown report data loss and + // train the operator to ignore the signal entirely. + // what this catches: a module with no failures refusing to stop cleanly, which + // would make every ordinary shutdown report data loss and train the operator to + // ignore the signal. Also through the REAL `shutdown`, so it fails if the + // delegation inverts. + #[tokio::test] + async fn a_module_that_wrote_everything_stops_clean() { + let (log_tx, _rx) = mpsc::sync_channel::(8); + let state = Arc::new(LoggerCommandState { + log_tx, + file_cache: Default::default(), + started_at: Instant::now(), + requests_processed: AtomicU64::new(0), + pending_writes: Arc::new(AtomicU64::new(0)), + write_failures: Arc::new(AtomicU64::new(0)), + }); + assert!( + LoggerModule::with_state(state).shutdown().await.is_ok(), + "no failures means a clean stop" + ); + } + + // what this catches: the counter being process-global again. Two modules must not + // see each other's failures — which is exactly why this was moved off a `static`, + // and why a test could not drive a real failure before. + #[test] + fn one_modules_failures_do_not_condemn_anothers_stop() { + let mine: WriteFailures = Arc::new(AtomicU64::new(3)); + let theirs: WriteFailures = Arc::new(AtomicU64::new(0)); + assert!(stop_outcome(mine.load(Ordering::Relaxed)).is_err()); + assert!(stop_outcome(theirs.load(Ordering::Relaxed)).is_ok()); + } + } } diff --git a/core/continuum-core/src/persona/service_loop.rs b/core/continuum-core/src/persona/service_loop.rs index adb5fd24bf..7e7ec29ad7 100644 --- a/core/continuum-core/src/persona/service_loop.rs +++ b/core/continuum-core/src/persona/service_loop.rs @@ -528,6 +528,28 @@ async fn serve_persona_loop_inner( Wake::Stop => {} } } + // ADMIT THE TURN, or leave because the node is stopping. + // + // Taken the moment a wake is produced, BEFORE the match, because the + // `Wake::Tick` arm does its self-tick deliberation INLINE and then `continue`s — + // it never reaches the code below it. A permit taken after the match would have + // counted inbound turns and silently missed every self-directed one, which is the + // same blindness as draining on `activity_gate`'s `engaged` flag: that one is + // stamped at serving-lane acquisition, so a turn still composing its context does + // not appear there either. Both gaps hide turns that have consumed input and + // written nothing. + // + // Held to the end of the loop body, so it is released on EVERY exit path — the + // `continue` at the end of the tick arm included. A manual decrement would have to + // be repeated at each of them, and would be missed at the next one added. + let Some(_turn) = crate::cognition::turn_ingress::admit() else { + crate::probe!( + class = "persona.ingress.closed", + persona = ctx.identity.peer_id.as_uuid().to_string(), + "turn ingress closed for shutdown — this citizen takes no further turns and leaves her loop" + ); + break; + }; let msg = match wake { Wake::Stop => break, Wake::Tick => { diff --git a/core/continuum-core/src/runtime/admission_gate.rs b/core/continuum-core/src/runtime/admission_gate.rs new file mode 100644 index 0000000000..aaa326404b --- /dev/null +++ b/core/continuum-core/src/runtime/admission_gate.rs @@ -0,0 +1,338 @@ +//! ONE ADMISSION GATE, two users. +//! +//! A stopping node has the same question in more than one place: *may new work be +//! accepted, and how much accepted work is still running?* Both halves must be read and +//! written together — a flag beside a counter is a race, because a close can land between +//! the check and the increment, and a drain then reads zero while a producer is about to +//! make it one. +//! +//! This is that pair in a single word: high bit CLOSED, low 63 bits the in-flight count, +//! admission by `compare_exchange`. It exists as a TYPE rather than a pattern because it +//! was written twice — once for persona turns, once for the log queue — and the second +//! copy reintroduced the exact race the first had just removed. Two implementations of an +//! invariant are two chances to get it wrong; the review that caught the second copy said +//! "use one-word CAS like the turn gate", which is this. +//! +//! Instance methods, not free functions over a global, so a test can drive the REAL +//! admission and the REAL close on its own gate. The previous shape was a process-wide +//! one-way static, which meant any test exercising a real close poisoned every later test +//! in the binary — so the tests reimplemented the logic instead of calling it, and could +//! not fail when the production copy regressed. + +use std::sync::atomic::{AtomicU64, Ordering}; + +/// High bit. Leaves 63 bits of count, which is not a limit anyone reaches: the count is +/// concurrent units of work, bounded by citizens or queue capacity. +const CLOSED: u64 = 1 << 63; + +/// Admission state for one resource: open/closed plus the in-flight count, in one word. +pub struct AdmissionGate { + word: AtomicU64, +} + +impl AdmissionGate { + /// A new, OPEN gate with nothing in flight. `const` so it can be a `static` without + /// a lazy initialiser. + pub const fn new() -> Self { + Self { + word: AtomicU64::new(0), + } + } + + /// Is the gate still accepting? Prefer [`admit`](Self::admit), which answers this AND + /// takes ownership in the same atomic step; a bare read is only safe for display. + pub fn is_open(&self) -> bool { + self.word.load(Ordering::Acquire) & CLOSED == 0 + } + + /// Units of work admitted and not yet released. + pub fn in_flight(&self) -> u64 { + self.word.load(Ordering::Acquire) & !CLOSED + } + + /// ADMIT one unit of work, or refuse because the gate is closed. + /// + /// The returned permit owns it: counted on creation, uncounted on drop, so the count + /// is right on every exit path including `?`, `continue`, `break` and unwind. A manual + /// decrement has to be repeated at each of those and will be missed at the next one + /// added, and a leaked count is a drain that waits its whole budget for work that + /// finished long ago. + /// + /// The CAS is what makes this safe: an admission either lands before a close — and is + /// therefore in the count that close carries, so a drain waits for it — or observes + /// the close and refuses. There is no interleaving in which work starts unseen. + pub fn admit(&self) -> Option> { + // A no-op hook, monomorphised away: production runs the identical loop. + self.admit_hooked(|| {}) + } + + /// `admit`, with a hook run between the LOAD and the compare-exchange. + /// + /// The hook exists so a test can make the CAS genuinely fail. A test that simply + /// admits twice in sequence never contends, so the retry branch is never taken and a + /// test named for it proves nothing — which is what the first version of the retry + /// regression did. Mutating the word from inside the hook guarantees the first + /// exchange sees a stale value, deterministically, with no threads and no timing. + /// + /// Private: the hook is a test seam, not an API. Production reaches this only through + /// [`admit`](Self::admit) with an empty closure. + fn admit_hooked(&self, mut between_load_and_cas: impl FnMut()) -> Option> { + let mut cur = self.word.load(Ordering::Acquire); + loop { + if cur & CLOSED != 0 { + return None; + } + between_load_and_cas(); + // The count is the low bits, so `cur + 1` increments it and leaves the (here + // necessarily clear) closed bit alone. + match self + .word + .compare_exchange(cur, cur + 1, Ordering::AcqRel, Ordering::Acquire) + { + Ok(_) => return Some(Permit { gate: self }), + // Someone else moved the word: re-read and retry against what is there + // now, rather than against the stale value we loaded. Re-reading is the + // whole point — retrying with `cur` unchanged would either spin forever or + // clobber the other writer's increment. + Err(actual) => cur = actual, + } + } + } + + /// Release one unit of work whose permit was FORGOTTEN because ownership crossed a + /// thread boundary. + /// + /// The RAII permit is the right tool whenever admission and completion happen in one + /// scope, and it should be used wherever it can be. It cannot span the log queue: the + /// producer admits an entry, hands it to a writer THREAD, and the writer is what knows + /// when the entry has actually been written. A permit dropped at the send would uncount + /// an entry still sitting in the channel, and a drain would then call the module quiet + /// while the writer still had work. + /// + /// So the contract for this method is narrow and must stay narrow: **it is only + /// correct when a permit was `mem::forget`ed at a hand-off, and it must be called + /// exactly once for each one.** Anywhere both ends are in the same scope, use the + /// permit and let it drop — a manual decrement has to be repeated at every exit path + /// and will be missed at the next one added. + /// + /// Saturating, so a double release cannot wrap the count into the closed bit and + /// reopen a shut gate. A double release is still a bug; this only bounds the damage. + pub fn release_one(&self) { + let _ = self + .word + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |cur| { + let count = cur & !CLOSED; + if count == 0 { + // Nothing to release. Refuse rather than wrap: `cur - 1` here would + // borrow from the closed bit and silently REOPEN the gate. + None + } else { + Some((cur & CLOSED) | (count - 1)) + } + }); + } + + /// Shut the gate. Returns whether THIS call closed it, so a second caller — a signal + /// racing a stop verb, a retried request — can tell it is re-entering rather than + /// report a fresh close. + /// + /// One-way by design: the only caller is a stop, and a reopen would exist to be called + /// by mistake. A gate is re-opened by constructing a new one. + pub fn close(&self) -> bool { + self.word.fetch_or(CLOSED, Ordering::AcqRel) & CLOSED == 0 + } +} + +impl Default for AdmissionGate { + fn default() -> Self { + Self::new() + } +} + +/// Ownership of one admitted unit of work. See [`AdmissionGate::admit`]. +#[must_use = "the permit IS the admission; dropping it immediately releases the work"] +pub struct Permit<'a> { + gate: &'a AdmissionGate, +} + +impl Drop for Permit<'_> { + fn drop(&mut self) { + // Subtracting 1 from the whole word decrements the low-bit count and cannot reach + // the closed bit, because a permit only exists when the count was at least 1. + self.gate.word.fetch_sub(1, Ordering::AcqRel); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + // what this catches: THE RACE, driven through the REAL admit/close rather than a + // reimplementation of them. With an open flag and a count as separate atomics this + // order was reachable: admit reads OPEN, close lands, drain reads 0 and saves, admit + // then increments and starts work behind the drain's back — the torn save the drain + // exists to prevent, caused by the drain. + #[test] + fn work_can_never_be_admitted_after_the_gate_reads_closed() { + let gate = AdmissionGate::new(); + assert!(gate.close(), "first close is the closer"); + assert!( + gate.admit().is_none(), + "an admission after close must be refused, not deferred" + ); + assert_eq!( + gate.in_flight(), + 0, + "and the drain's zero is permanently true, because the refusal cannot become work" + ); + } + + // what this catches: an admission that WON the race being invisible to the close that + // follows it — the drain would then wait for nothing and save over live work. + #[test] + fn an_admission_that_wins_is_carried_by_the_close_that_follows() { + let gate = AdmissionGate::new(); + let permit = gate.admit().expect("open gate admits"); + assert!(gate.close()); + assert_eq!( + gate.in_flight(), + 1, + "the close must carry the work admitted just before it" + ); + assert!(gate.admit().is_none()); + drop(permit); + assert_eq!(gate.in_flight(), 0, "the drain's wait ends when the work does"); + assert!(!gate.is_open(), "releasing the last permit must not reopen the gate"); + } + + // what this catches: the closed bit and the count disturbing each other. They share + // one word, so an arithmetic slip either way reopens a shut gate or corrupts the + // number a drain waits on. + #[test] + fn the_count_and_the_closed_bit_survive_each_other_in_any_order() { + let gate = AdmissionGate::new(); + let a = gate.admit().expect("open"); + let b = gate.admit().expect("open"); + let c = gate.admit().expect("open"); + assert_eq!(gate.in_flight(), 3); + drop(b); // one finishes BEFORE the close + gate.close(); + assert_eq!(gate.in_flight(), 2, "the close carries what was still running"); + drop(a); + drop(c); // and the rest finish after it, out of admission order + assert_eq!(gate.in_flight(), 0); + assert!(!gate.is_open()); + } + + // what this catches: `close` reporting a fresh close every time. Both a signal handler + // and a stop verb reach the same broadcast, so close is called twice on an ordinary + // stop; the second must say it was already shut rather than look like a new event. + #[test] + fn a_second_close_reports_that_it_was_already_shut() { + let gate = AdmissionGate::new(); + let _held = gate.admit().expect("open"); + assert!(gate.close(), "first close closed it"); + assert!(!gate.close(), "second close must report it was already shut"); + assert_eq!(gate.in_flight(), 1, "and must not disturb the count"); + } + + // what this catches: the CAS retry path dropping or clobbering an admission. + // + // The FIRST version of this test admitted several times in sequence and asserted the + // count — which never contends, so the retry branch it was named for was never taken. + // A test named for an invariant that never drives the invariant is worse than a + // missing one, because it is counted. Astra caught it. + // + // This one FORCES a failed exchange: the hook fires between the load and the CAS and + // moves the word, so the first attempt is guaranteed to see a stale value. No threads, + // no timing, no flake — the interleaving is driven, not hoped for. + #[test] + fn a_failed_exchange_retries_against_the_current_word_not_the_stale_one() { + let gate = AdmissionGate::new(); + let interferences = std::cell::Cell::new(0u32); + + // Steal the word exactly once, on the first attempt only. A hook that fired every + // time would spin forever, which is itself worth knowing: the retry must make + // progress against a word that STOPS moving. + let permit = gate + .admit_hooked(|| { + if interferences.get() == 0 { + interferences.set(1); + // Another admission lands between our load and our exchange. + gate.word.fetch_add(1, Ordering::AcqRel); + } + }) + .expect("an open gate admits even when it has to retry"); + + assert_eq!(interferences.get(), 1, "the hook must have contended once"); + assert_eq!( + gate.in_flight(), + 2, + "the retry must ADD to the interfering write, not replace it — a retry against the stale word would have stored 1 and erased the other admission" + ); + drop(permit); + assert_eq!(gate.in_flight(), 1, "and only OUR admission is released"); + } + + // what this catches: a close that lands during the retry being ignored. The loop + // re-reads on failure, so it must re-check CLOSED on the new value too — retrying + // blindly would admit work into a gate that shut while we were contending. + #[test] + fn a_close_that_lands_during_a_retry_refuses_the_admission() { + let gate = AdmissionGate::new(); + let fired = std::cell::Cell::new(false); + let outcome = gate.admit_hooked(|| { + if !fired.get() { + fired.set(true); + gate.close(); // shuts between our load and our exchange + } + }); + assert!(fired.get(), "the hook must have run"); + assert!( + outcome.is_none(), + "an admission whose CAS lost to a CLOSE must be refused, not retried into a closed gate" + ); + assert_eq!(gate.in_flight(), 0); + assert!(!gate.is_open()); + } + + // what this catches: a double release wrapping the count and REOPENING a closed gate. + // The count and the closed bit share a word, so `cur - 1` at zero would borrow from + // the high bit — turning a stopping node back into an accepting one, which is the + // worst possible direction for this bug. + #[test] + fn releasing_more_than_was_admitted_cannot_reopen_the_gate() { + let gate = AdmissionGate::new(); + gate.close(); + assert_eq!(gate.in_flight(), 0); + gate.release_one(); // unpaired: a bug, but it must not be a catastrophic one + gate.release_one(); + assert_eq!(gate.in_flight(), 0, "the count must not wrap"); + assert!(!gate.is_open(), "a closed gate must stay closed"); + assert!(gate.admit().is_none()); + } + + // what this catches: the hand-off path miscounting. The log queue forgets its permit + // at the send and the writer releases after the write, so an entry being written is + // still in flight — a drain that stopped at the hand-off would race the write. + #[test] + fn a_forgotten_permit_stays_counted_until_the_holder_releases_it() { + let gate = AdmissionGate::new(); + let permit = gate.admit().expect("open"); + std::mem::forget(permit); // handed to another thread + assert_eq!(gate.in_flight(), 1, "the work is still in flight after the hand-off"); + gate.release_one(); // the holder finished it + assert_eq!(gate.in_flight(), 0); + } + + // what this catches: a gate defaulting CLOSED, which would make every producer in a + // normally-booted core refuse — a total, silent outage that no test of the shutdown + // path would exercise. + #[test] + fn a_new_gate_is_open_and_empty() { + let gate = AdmissionGate::new(); + assert!(gate.is_open()); + assert_eq!(gate.in_flight(), 0); + assert!(gate.admit().is_some()); + } +} diff --git a/core/continuum-core/src/runtime/mod.rs b/core/continuum-core/src/runtime/mod.rs index ad7e43c617..2e385a6f35 100644 --- a/core/continuum-core/src/runtime/mod.rs +++ b/core/continuum-core/src/runtime/mod.rs @@ -24,6 +24,7 @@ use dashmap::DashMap; use std::sync::Arc; use std::sync::OnceLock; +pub mod admission_gate; pub mod airc_interceptor; pub mod artifact_handle; pub mod boot_mode; @@ -105,7 +106,12 @@ pub use provided_provider::{ pub use ready_buffer::{DashMapReadyBuffer, ReadyBuffer}; pub use region_telemetry::RegionTelemetry; pub use registry::ModuleRegistry; -pub use runtime::{install_signal_shutdown, run_signal_shutdown, Runtime}; +pub use admission_gate::{AdmissionGate, Permit}; +pub use runtime::{ + await_shutdown, begin_shutdown, install_signal_shutdown, run_signal_shutdown, + signal_runtime, DrainOutcome, ModuleStop, + ModuleStopOutcome, Runtime, ShutdownReceipt, +}; pub use service_module::{ CommandResult, CommandSchema, ModuleConfig, ModulePriority, ParamSchema, ServiceModule, }; diff --git a/core/continuum-core/src/runtime/registry.rs b/core/continuum-core/src/runtime/registry.rs index 0b002a8d4e..a01dd826be 100644 --- a/core/continuum-core/src/runtime/registry.rs +++ b/core/continuum-core/src/runtime/registry.rs @@ -690,6 +690,14 @@ mod tests { module: "RecorderModule", why: "fixture: genome local_manager call recorder", }, + Unwired { + module: "SlowJoiner", + why: "fixture: a module that saves and then cannot be joined inside the phase bound — drives ModuleStopOutcome::JoinTimedOut, which before it existed only as a literal in assertions", + }, + Unwired { + module: "SlowSaver", + why: "fixture: a module that cannot save inside the phase bound — drives ModuleStopOutcome::SaveTimedOut, the outcome the stop exit code keys on", + }, Unwired { module: "StubAircModule", why: "fixture: ChatModule's airc stand-in", diff --git a/core/continuum-core/src/runtime/runtime.rs b/core/continuum-core/src/runtime/runtime.rs index 8f017eb925..465d1fb6fe 100644 --- a/core/continuum-core/src/runtime/runtime.rs +++ b/core/continuum-core/src/runtime/runtime.rs @@ -7,9 +7,9 @@ //! that owns the CBP_Analyzer pipeline and orchestrates frame flow. use super::boot_mode::BootMode; +use super::command_interceptor::{CommandInterceptor, InterceptorOutcome}; use super::message_bus::MessageBus; use super::module_context::ModuleContext; -use super::command_interceptor::{CommandInterceptor, InterceptorOutcome}; use super::registry::ModuleRegistry; use super::service_module::{CommandResult, ServiceModule}; use super::shared_compute::SharedCompute; @@ -416,7 +416,10 @@ impl Runtime { .unwrap_or_else(|e| e.into_inner()) // unwrap_or_else: a poisoned lock still holds the list; read it .clone(); for interceptor in chain { - match interceptor.try_route(command, ¶ms, caller.as_ref()).await { + match interceptor + .try_route(command, ¶ms, caller.as_ref()) + .await + { Ok(InterceptorOutcome::Handled(result)) => return Some(Ok(result)), Ok(InterceptorOutcome::Decline) => continue, Err(e) => { @@ -670,50 +673,93 @@ impl Runtime { /// all: the SIGTERM handler killed sentinels, slept a flat 2s, and /// `_exit`ed — so no module ever saved on a real stop. The handler now /// runs this first ([`install_signal_shutdown`]). - pub async fn shutdown(&self) { - const PER_MODULE: std::time::Duration = std::time::Duration::from_secs(2); + /// Stop every module: DRAIN, then save, then join — each phase bounded, all modules + /// in parallel — and return a receipt saying what actually reached disk. + /// + /// The drain phase is new and it is the point. Suspending a module's tick is not a + /// drain (`quiesce_all` stops each mind's self-tick and leaves room input arriving and + /// active turns running), so without it `save_state` could be taken underneath a turn + /// halfway through writing, and the result was indistinguishable from a clean save. + pub async fn shutdown(&self) -> ShutdownReceipt { + self.shutdown_within(std::time::Duration::from_secs(2)) + .await + } + + /// `shutdown`, with the per-phase bound passed IN. + /// + /// A parameter rather than a constant because the TIMEOUT arms could not otherwise be + /// reached: no test module takes two seconds, so `SaveTimedOut` and `JoinTimedOut` + /// were only ever constructed as literals and asserted on. Deleting the timeout + /// handling here would have left every one of those tests green — the outcomes were + /// described by the suite and never produced by it. Found by IntelMac. + async fn shutdown_within(&self, per_phase: std::time::Duration) -> ShutdownReceipt { let modules = self.registry.list_modules(); - info!("Shutting down {} modules (parallel, 2s bound each)...", modules.len()); + info!( + "Stopping {} modules (drain → save → join, parallel, 2s bound per phase)...", + modules.len() + ); let started = std::time::Instant::now(); let futs = modules.iter().filter_map(|name| { self.registry.get_by_name(name).map(|module| { - let name = name.clone(); + // Deref, not clone: `list_modules` yields `Vec<&'static str>`, so `name` + // here is `&&'static str` and `.clone()` copied the OUTER reference — + // it never cloned a string, and the borrow it produced only compiled + // because the inner `&str` is 'static. Carried in from the pre-refactor + // `shutdown()`; saying `*name` is what was always meant. + let name: &'static str = *name; async move { let t = std::time::Instant::now(); - // Save first, then join — each half under the bound. - let saved = tokio::time::timeout(PER_MODULE, module.save_state()).await; - let joined = tokio::time::timeout(PER_MODULE, module.shutdown()).await; - let outcome = match (saved, joined) { - (Ok(Ok(())), Ok(Ok(()))) => "ok", - (Err(_), _) | (_, Err(_)) => "timeout", - _ => "error", + // 1. Stop taking new work and let in-flight work finish. A module + // that cannot drain in time still gets its save attempted — a + // mid-turn snapshot beats no snapshot — but the receipt says so. + let drain = match tokio::time::timeout(per_phase, module.drain()).await { + Ok(Ok(0)) => DrainOutcome::Drained, + Ok(Ok(in_flight)) => DrainOutcome::Incomplete { in_flight }, + Ok(Err(e)) => DrainOutcome::Unknown { reason: e }, + Err(_) => DrainOutcome::Unknown { + reason: format!("drain exceeded {}s", per_phase.as_secs()), + }, + }; + + // 2. Save. This is the phase whose failure is DATA, not tidiness. + let saved = tokio::time::timeout(per_phase, module.save_state()).await; + let outcome = match saved { + Err(_) => ModuleStopOutcome::SaveTimedOut, + Ok(Err(e)) => ModuleStopOutcome::SaveFailed { error: e }, + Ok(Ok(())) => { + // 3. Join, only meaningful once the state is durable. + match tokio::time::timeout(per_phase, module.shutdown()).await { + Err(_) => ModuleStopOutcome::JoinTimedOut, + Ok(Err(e)) => ModuleStopOutcome::JoinFailed { error: e }, + Ok(Ok(())) => ModuleStopOutcome::Clean, + } + } }; crate::probe!( class = "shutdown.step", module = %name, - outcome = outcome, + outcome = format!("{outcome:?}"), + drain = format!("{drain:?}"), ms = t.elapsed().as_millis() as u64, - "module save-and-join" + "module drain-save-join" ); - (name, outcome) + ModuleStop { + // `list_modules` yields &'static str; the receipt owns its names + // because it outlives the registry borrow and crosses the wire. + module: name.to_string(), + drain, + outcome, + ms: t.elapsed().as_millis() as u64, + } } }) }); - let reports = futures::future::join_all(futs).await; - let slow: Vec = reports - .iter() - .filter(|(_, o)| *o != "ok") - .map(|(n, _)| n.to_string()) - .collect(); - info!( - "All modules shut down in {}ms{}", - started.elapsed().as_millis(), - if slow.is_empty() { - String::new() - } else { - format!(" (non-ok: {})", slow.join(", ")) - } - ); + let receipt = ShutdownReceipt { + modules: futures::future::join_all(futs).await, + total_ms: started.elapsed().as_millis() as u64, + }; + info!("{}", receipt.summary()); + receipt } /// Verify all required modules are registered for the given @@ -1392,9 +1438,316 @@ pub fn install_signal_shutdown(rt: Arc) { /// Run the save-and-join broadcast from a signal handler, bounded overall — /// a stop must complete in seconds even if the runtime misbehaves. +/// What one module did when the node was told to stop. +/// +/// Three phases, in the order the runtime broadcasts them, each with its own outcome — +/// because they fail differently and a caller that only learns "shutdown finished" cannot +/// tell a clean stop from one that abandoned a half-written turn. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, ts_rs::TS)] +#[ts( + export, + export_to = "../../../protocol/typescript/system/ModuleStopOutcome.ts" +)] +pub enum ModuleStopOutcome { + /// Drained, saved and joined within the bound. + Clean, + /// `save_state` did not finish inside its bound. THE STATE IS UNKNOWN, not merely + /// old: the write may have been half-applied. This is the case a caller must not + /// report as success. + SaveTimedOut, + /// `save_state` returned an error and said why. + SaveFailed { error: String }, + /// Saved, but `shutdown` did not return inside its bound. Its final flush is + /// unconfirmed, so the receipt cannot claim durable state. + JoinTimedOut, + /// `shutdown` returned an error after a successful save. + JoinFailed { error: String }, +} + +impl ModuleStopOutcome { + /// Did EVERY phase this module ran complete? The only outcome that can support a + /// durability claim. + /// + /// An earlier version answered `true` for `JoinTimedOut` and `JoinFailed`, on the + /// reasoning that a module which saved and then failed to let go is untidy rather + /// than lossy. **That was wrong, and the trait's own contract says so:** `shutdown` + /// is documented as "release resources, FLUSH BUFFERS", and the logger's + /// implementation is literally `flush_all`. A module whose join failed is a module + /// whose final flush may not have happened, so its last writes are exactly as gone as + /// a failed save's. Found by Astra, who read the contract I had written and I had not. + /// + /// The phase distinction is still carried in the variant, because it tells a reader + /// WHERE the stop broke. It just cannot decide whether the state is on disk. + pub fn completed(&self) -> bool { + matches!(self, ModuleStopOutcome::Clean) + } +} + +/// What the DRAIN phase achieved, kept separate from the save/join outcome. +/// +/// These were one enum and it lost information: an incomplete drain followed by a join +/// timeout produced `JoinTimedOut`, the drain result was discarded, and the module then +/// reported its state durable — a save taken mid-turn, described as clean. They are +/// orthogonal facts about the same module and neither may overwrite the other. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, ts_rs::TS)] +#[ts( + export, + export_to = "../../../protocol/typescript/system/DrainOutcome.ts" +)] +pub enum DrainOutcome { + /// Nothing was in flight when the module stopped taking work. + Drained, + /// The bound expired with work outstanding, so whatever was saved next is a snapshot + /// taken mid-turn. + Incomplete { in_flight: u32 }, + /// `drain` itself errored or timed out. How much was in flight is UNKNOWN — which is + /// why this is not `Incomplete { in_flight: 0 }`, a number nobody measured. + Unknown { reason: String }, +} + +impl DrainOutcome { + /// Did the module stop taking work with nothing outstanding? Anything else means the + /// save that followed may be torn. + pub fn is_quiet(&self) -> bool { + matches!(self, DrainOutcome::Drained) + } +} + +/// One module's line in the receipt. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, ts_rs::TS)] +#[ts( + export, + export_to = "../../../protocol/typescript/system/ModuleStop.ts" +)] +pub struct ModuleStop { + pub module: String, + pub drain: DrainOutcome, + pub outcome: ModuleStopOutcome, + #[ts(type = "number")] + pub ms: u64, +} + +impl ModuleStop { + /// Did this module's state reach disk INTACT? Both halves must hold: every phase + /// completed, AND the module was quiet when its state was taken. + /// + /// A save that succeeded over a still-running turn wrote something — it just did not + /// write a consistent something. A join that failed may have skipped the flush that + /// puts the save on disk. Either way the answer is no, and reporting either as + /// durable is the failure this whole receipt exists to prevent. + pub fn state_is_durable(&self) -> bool { + self.outcome.completed() && self.drain.is_quiet() + } +} + +/// WHAT ACTUALLY HAPPENED when the node stopped. +/// +/// `Runtime::shutdown` used to return `()`. It logged non-ok modules to the core's own +/// log and told the caller nothing, so `continuum stop` printed a success line and exited +/// 0 whether every module had saved or none had. A stop whose result only exists in the +/// log of the process that just exited is not a result anybody can act on. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, ts_rs::TS)] +#[ts( + export, + export_to = "../../../protocol/typescript/system/ShutdownReceipt.ts" +)] +pub struct ShutdownReceipt { + pub modules: Vec, + #[ts(type = "number")] + pub total_ms: u64, +} + +impl ShutdownReceipt { + /// Modules whose durable state did NOT reach disk intact. Empty is the only value + /// that justifies reporting a clean stop. + pub fn unsaved(&self) -> Vec<&ModuleStop> { + self.modules + .iter() + .filter(|m| !m.state_is_durable()) + .collect() + } + + /// Every module drained, saved, and completed its final shutdown flush. + pub fn state_is_durable(&self) -> bool { + self.unsaved().is_empty() + } + + /// One line a human can act on, without reading the process's log after it exited. + pub fn summary(&self) -> String { + let unsaved = self.unsaved(); + if unsaved.is_empty() { + return format!( + "{} modules stopped, all state durable, {}ms", + self.modules.len(), + self.total_ms + ); + } + let names: Vec = unsaved + .iter() + .map(|m| format!("{} (drain {:?}, save {:?})", m.module, m.drain, m.outcome)) + .collect(); + format!( + "{} of {} modules did NOT save: {}", + unsaved.len(), + self.modules.len(), + names.join(", ") + ) + } +} + +/// ONE shutdown, owned. +/// +/// This was a bare `OnceLock` plus a free function, and the tests for it had to build +/// their own `watch` channel — so they asserted a property of `tokio::sync::watch` and +/// would have stayed green if the production publisher regressed from `send_replace` back +/// to `send`. Established by Astra reading the source, NOT by an executed mutation run — +/// nobody has yet watched that test go red, and the weaker claim is the true one. +/// +/// As an instance, a test can construct the SAME owner over a real `Runtime` and drive the +/// real `begin` / publisher, instead of a look-alike. Same reason `AdmissionGate` is a +/// type: an invariant that can only be reached through a global is an invariant whose +/// tests drift into testing something adjacent. +struct ShutdownOperation { + result: tokio::sync::watch::Sender>, + started: std::sync::atomic::AtomicBool, +} + +impl ShutdownOperation { + fn new() -> Self { + Self { + result: tokio::sync::watch::channel(None).0, + started: std::sync::atomic::AtomicBool::new(false), + } + } + + /// Start the broadcast over `rt` if it has not started, and return a view of the + /// result. Idempotent: a second caller — a signal racing the stop verb, a retried + /// request — joins the first broadcast rather than running `save_state` twice over + /// the same state. + fn begin( + &self, + rt: Option>, + ) -> tokio::sync::watch::Receiver> { + if !self.started.swap(true, std::sync::atomic::Ordering::AcqRel) { + match rt { + Some(rt) => { + let tx = self.result.clone(); + // Detached ON PURPOSE: this is the task whose whole point is that no + // connection owns it. A socket handler is cancelled when its client + // goes away, and a shutdown cancelled after ingress closed leaves the + // node refusing work with nothing saved. + tokio::spawn(async move { + let receipt = rt.shutdown().await; + // send_replace, NEVER send: `send` returns Err and DROPS the value + // when the last receiver has gone — and the receiver that goes + // away is the disconnecting CLI, so `send` loses the terminal + // receipt in precisely the case this design exists for. A later + // subscriber would then wait forever on a `None` that never + // changes. + let _ = tx.send_replace(Some(receipt)); + }); + } + None => { + // No runtime: nothing to save AND nothing that could have saved. + // Publishing an empty receipt says exactly that, rather than leaving + // every observer waiting on a broadcast that will never run. + let _ = self.result.send_replace(Some(ShutdownReceipt { + modules: Vec::new(), + total_ms: 0, + })); + } + } + } + self.result.subscribe() + } +} + +impl Default for ShutdownOperation { + fn default() -> Self { + Self::new() + } +} + +/// Initialize the process owner on first use; observers receive an owned watch handle. +static SHUTDOWN: std::sync::OnceLock = std::sync::OnceLock::new(); + +fn process_shutdown() -> &'static ShutdownOperation { + SHUTDOWN.get_or_init(ShutdownOperation::new) +} + +/// START the node's shutdown if it has not started, and hand back a view of its result. +/// +/// Delegates to the process's single [`ShutdownOperation`]. See that type for why the +/// operation is owned rather than free-standing. +pub fn begin_shutdown() -> tokio::sync::watch::Receiver> { + process_shutdown().begin(signal_runtime()) +} + +/// Wait for the shutdown to finish, up to `budget`. +/// +/// `None` means it is still running — NOT that it failed and not that state is durable. +/// The distinction matters to the caller's exit code: a stop we stopped waiting for is +/// unknown, and unknown is not success. +pub async fn await_shutdown( + mut rx: tokio::sync::watch::Receiver>, + budget: std::time::Duration, +) -> Option { + if let Some(r) = rx.borrow_and_update().clone() { + return Some(r); + } + let deadline = tokio::time::Instant::now() + budget; + loop { + match tokio::time::timeout_at(deadline, rx.changed()).await { + Ok(Ok(())) => { + if let Some(r) = rx.borrow_and_update().clone() { + return Some(r); + } + // A change that is still `None`: keep waiting. + } + // THE SENDER IS GONE. `changed()` returns `Err(RecvError)` immediately and + // forever once the owner drops, so testing only for the TIMEOUT spun this loop + // at full speed until the deadline — burning a core for the whole budget on + // the one path where nothing can ever arrive. Found in source review by Astra. + Ok(Err(_)) => return None, + // Budget spent. The operation is still running, un-cancelled; unknown is not + // failure and not success. + Err(_) => return None, + } + } +} + +/// The runtime the signal handlers stop, for callers that need the SAME one — the +/// resident `system/shutdown` verb runs the identical broadcast, so it must not reach +/// for a second registry. One runtime, one stop path, two triggers. +pub fn signal_runtime() -> Option> { + SIGNAL_RUNTIME.get().cloned() +} + pub async fn run_signal_shutdown() { - if let Some(rt) = SIGNAL_RUNTIME.get() { - let _ = tokio::time::timeout(std::time::Duration::from_secs(3), rt.shutdown()).await; + // Through the SAME idempotent operation the `system/shutdown` verb triggers, so a + // signal arriving while the verb is running joins that broadcast instead of starting + // a second one over the same state. The bound is deliberately larger than one phase: + // three 2s phases run in parallel across modules, so a healthy stop is ~6s worst case + // and a tighter cap here would cut off the save phase on any node slow enough to + // need it. + let rx = begin_shutdown(); + match await_shutdown(rx, std::time::Duration::from_secs(8)).await { + Some(receipt) => { + // A signal handler cannot return an exit code to anyone, so the receipt's only + // readers are the log and the probe ledger — which is exactly why it is + // written here rather than dropped. `_exit` follows immediately. + if !receipt.state_is_durable() { + eprintln!( + "[continuum-core] STOPPED WITH UNSAVED STATE: {}", + receipt.summary() + ); + } + } + None => { + eprintln!( + "[continuum-core] shutdown did not finish within 8s — state durability is UNKNOWN" + ); + } } } @@ -1402,6 +1755,466 @@ pub async fn run_signal_shutdown() { mod conditional_modules_tests { use super::*; + /// The shutdown result must survive the disappearance of everyone watching it. + mod receipt_retention { + use super::*; + // The ONE recording fixture, borrowed from the dispatch tests rather than + // reimplemented here. + use crate::runtime::runtime::piece_2_pr3_dispatch_tests::RecordingModule; + + // what this catches: `watch::Sender::send` dropping the terminal receipt when the + // last receiver has gone. That receiver is the CLI, and its disconnect is the case + // this whole design exists to survive — so `send` loses the receipt in exactly the + // situation it matters, and the next subscriber (a retried request, or the signal + // path joining the same broadcast) waits forever on a `None` that never changes. + // + // DRIVEN THROUGH THE REAL OWNER. The first version of this test built its own + // `watch` and called `send_replace` itself, so it asserted a property of tokio and + // would have stayed GREEN if the production publisher regressed to `send` — a + // SOURCE-REVIEW finding by Astra, reasoned from the code rather than executed as a + // mutation run, and labelled that way because she insisted on the distinction when + // a relay upgraded it to "she ran it". S6 supplied the shape of the fix — + // make the operation an instance so a test can hold the same one production holds. + // This constructs a real `ShutdownOperation` over a real `Runtime` with the + // existing `RecordingModule`, drops every caller, and only then resubscribes. + #[tokio::test] + async fn the_receipt_survives_every_observer_going_away() { + // A plain local: `begin` needs only `&self` — it clones the owned sender + // before the spawn and the returned Receiver borrows nothing. My first version + // leaked a `'static` instance to satisfy a lifetime the code never required. + let op = ShutdownOperation::new(); + + let runtime = Arc::new(Runtime::new()); + let (module, _received) = RecordingModule::new("stop-recorder", Vec::new()); + runtime.register(module); + + // The only observer disconnects — the CLI died, was interrupted, timed out. + let rx = op.begin(Some(runtime)); + drop(rx); + + // Wait for the owner to publish. It is a detached task, so this polls a FRESH + // subscription rather than the one we dropped. + let mut seen = None; + for _ in 0..200 { + if let Some(r) = op.result.borrow().clone() { + seen = Some(r); + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + + let receipt = seen.expect("the owner must publish even with no receivers left"); + assert!( + receipt.modules.iter().any(|m| m.module == "stop-recorder"), + "the receipt must describe the module the runtime actually stopped, got {:?}", + receipt.modules + ); + + // And a subscriber arriving AFTER every observer left still learns what + // happened — which is the property `send` would have destroyed. + let late = op.result.subscribe(); + assert_eq!( + late.borrow().clone(), + Some(receipt), + "a late subscriber must still see the terminal receipt" + ); + } + + // what this catches: `begin` running the broadcast twice. A signal racing the stop + // verb, or a retried request, must JOIN the first operation — running `save_state` + // twice over the same state is the corruption the idempotence exists to prevent. + #[tokio::test] + async fn a_second_begin_joins_the_first_instead_of_restarting_it() { + let op = ShutdownOperation::new(); + let runtime = Arc::new(Runtime::new()); + let (module, _r) = RecordingModule::new("join-recorder", Vec::new()); + let saves = module.saves.clone(); + runtime.register(module); + + let _first = op.begin(Some(runtime.clone())); + // A second caller with a runtime it would otherwise stop. + let _second = op.begin(Some(runtime)); + + let mut receipt = None; + for _ in 0..200 { + if let Some(r) = op.result.borrow().clone() { + receipt = Some(r); + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + let _receipt = receipt.expect("the operation must complete"); + // ONE BROADCAST, not one receipt row. Counting rows would pass even if the + // second `begin` ran a parallel broadcast, because each publication produces + // one row per module regardless of how many times the module was stopped. + // What must be true is that the module's `save_state` ran ONCE — saving twice + // over the same state is the corruption the idempotence exists to prevent. + assert_eq!( + saves.load(std::sync::atomic::Ordering::Relaxed), + 1, + "a second begin must JOIN the first, not stop the module again" + ); + } + + // what this catches: `await_shutdown` treating a still-running stop as a finished + // one. Its timeout does not cancel anything — the operation is in a task no + // connection owns — so `None` means UNKNOWN, and a caller that read it as failure + // or as success would be wrong in opposite directions. + #[tokio::test] + async fn an_unfinished_shutdown_reports_unknown_rather_than_a_verdict() { + let (tx, rx) = tokio::sync::watch::channel(None::); + // A SECOND receiver, held here, is what makes the liveness assertion mean + // anything: `await_shutdown` consumes the one it is given and drops it, so + // with only that one alive `is_closed()` becomes true purely because WE + // stopped watching — which is not the fact under test and would have made + // this assertion fail (or, worse, pass for the wrong reason on a later + // refactor). Astra caught it at runtime.rs:1754. + let _observer = tx.subscribe(); + let out = await_shutdown(rx, std::time::Duration::from_millis(30)).await; + assert!( + out.is_none(), + "a stop still in flight must not produce a receipt" + ); + // Nothing was cancelled by our giving up: the operation can still publish, + // and `_observer` proves the channel is live rather than merely unobserved. + assert!( + !tx.is_closed(), + "giving up watching must not end the shutdown" + ); + let receipt = ShutdownReceipt { + modules: Vec::new(), + total_ms: 3, + }; + tx.send_replace(Some(receipt.clone())); + assert_eq!( + _observer.borrow().clone(), + Some(receipt), + "the operation we stopped waiting for must still be able to report" + ); + } + } + + /// Outcomes the runtime PRODUCES, not ones a test writes down. + /// + /// Every other receipt test constructs `ModuleStopOutcome` values as literals and + /// asserts on them, which describes the enum rather than the code that fills it — so + /// deleting the timeout handling in `shutdown` would have left them all green. These + /// drive a real `Runtime` with modules that genuinely exceed the phase bound. Found by + /// IntelMac. + mod outcomes_the_runtime_actually_produces { + use super::*; + use crate::runtime::runtime::piece_2_pr3_dispatch_tests::RecordingModule; + // The ServiceModule surface is not in this mod's `use super::*` reach; naming the + // imports beats a glob here because the two stub modules below implement the trait + // and a missing one shows up as four unrelated-looking errors. + use crate::runtime::service_module::{ModuleConfig, ModulePriority, ServiceModule}; + // NOT from `service_module` — it does not export this; the trait itself names + // `super::ModuleContext`. + use crate::runtime::ModuleContext; + use async_trait::async_trait; + use std::any::Any; + + /// A module whose SAVE never finishes inside the bound. + struct SlowSaver; + #[async_trait] + impl ServiceModule for SlowSaver { + fn config(&self) -> ModuleConfig { + ModuleConfig { + name: "slow-saver", + priority: ModulePriority::Normal, + command_prefixes: &[], + event_subscriptions: &[], + needs_dedicated_thread: false, + max_concurrency: 0, + tick_interval: None, + } + } + async fn initialize(&self, _ctx: &ModuleContext) -> Result<(), String> { + Ok(()) + } + async fn handle_command( + &self, + _command: &str, + _params: serde_json::Value, + ) -> Result { + // Required: `ServiceModule::handle_command` has no default. These stubs + // exist to exceed a phase bound, not to serve commands. + Err("not handled".to_string()) + } + async fn save_state(&self) -> Result<(), String> { + tokio::time::sleep(std::time::Duration::from_millis(400)).await; + Ok(()) + } + fn as_any(&self) -> &dyn Any { + self + } + } + + /// A module that saves promptly and then never lets go. + struct SlowJoiner; + #[async_trait] + impl ServiceModule for SlowJoiner { + fn config(&self) -> ModuleConfig { + ModuleConfig { + name: "slow-joiner", + priority: ModulePriority::Normal, + command_prefixes: &[], + event_subscriptions: &[], + needs_dedicated_thread: false, + max_concurrency: 0, + tick_interval: None, + } + } + async fn initialize(&self, _ctx: &ModuleContext) -> Result<(), String> { + Ok(()) + } + async fn handle_command( + &self, + _command: &str, + _params: serde_json::Value, + ) -> Result { + // Required: `ServiceModule::handle_command` has no default. These stubs + // exist to exceed a phase bound, not to serve commands. + Err("not handled".to_string()) + } + async fn shutdown(&self) -> Result<(), String> { + tokio::time::sleep(std::time::Duration::from_millis(400)).await; + Ok(()) + } + fn as_any(&self) -> &dyn Any { + self + } + } + + // what this catches: the save timeout arm being deleted or inverted. A module that + // cannot save inside the bound must produce `SaveTimedOut` and make the stop + // non-durable — this is the outcome the CLI's exit code keys on, and until now no + // test had ever caused one. + // `start_paused` — time is VIRTUAL here, and that is not a speed optimisation. + // + // With a real clock, a 400ms sleep against a 50ms timeout can still return + // `Clean`: pinned Tokio's `Timeout::poll` polls the INNER future FIRST, so if the + // scheduler stalls long enough that the sleep has already completed by the time + // the timeout is polled, the inner future wins. Under CI contention that is not + // hypothetical. + // + // The flake is a FALSE NEGATIVE, not a false positive: these tests ASSERT + // `SaveTimedOut`, so a stall that yields `Clean` makes the assertion FAIL and the + // test go red. Spurious failure on a loaded machine — annoying, worth removing, + // and it never certifies anything wrongly. + // + // Saying so precisely because the first version of this comment claimed the + // opposite ("fails open", certifying durability on a machine that cannot tell). + // That was wrong: it conflated the production path RETURNING Clean with the TEST + // PASSING, and the assertion sits between them. Mechanism from Astra and Popper; + // the retraction of the stronger claim from IntelMac, who traced it rather than + // just accepting the correction. + // + // Paused, the clock advances only when every task is idle and only to the nearest + // deadline, so the 50ms timeout fires before the 400ms sleep can complete, on any + // machine, every time. + #[tokio::test(start_paused = true)] + async fn a_module_that_cannot_save_in_time_produces_save_timed_out() { + let runtime = Runtime::new(); + runtime.register(Arc::new(SlowSaver)); + let receipt = runtime + .shutdown_within(std::time::Duration::from_millis(50)) + .await; + let row = receipt + .modules + .iter() + .find(|m| m.module == "slow-saver") + .expect("the slow module must appear in the receipt"); + assert_eq!(row.outcome, ModuleStopOutcome::SaveTimedOut); + assert!( + !row.state_is_durable(), + "a save that timed out leaves the state UNKNOWN, not merely old" + ); + assert!(!receipt.state_is_durable()); + assert!(receipt.summary().contains("slow-saver")); + } + + // what this catches: the join timeout arm. The module SAVED, so an implementation + // that stopped bounding the join would report `Clean` — and with `shutdown` + // contractually being "release resources, FLUSH BUFFERS", that would claim + // durability over a flush that may not have happened. + // `start_paused` — time is VIRTUAL here, and that is not a speed optimisation. + // + // With a real clock, a 400ms sleep against a 50ms timeout can still return + // `Clean`: pinned Tokio's `Timeout::poll` polls the INNER future FIRST, so if the + // scheduler stalls long enough that the sleep has already completed by the time + // the timeout is polled, the inner future wins. Under CI contention that is not + // hypothetical. + // + // The flake is a FALSE NEGATIVE, not a false positive: these tests ASSERT + // `SaveTimedOut`, so a stall that yields `Clean` makes the assertion FAIL and the + // test go red. Spurious failure on a loaded machine — annoying, worth removing, + // and it never certifies anything wrongly. + // + // Saying so precisely because the first version of this comment claimed the + // opposite ("fails open", certifying durability on a machine that cannot tell). + // That was wrong: it conflated the production path RETURNING Clean with the TEST + // PASSING, and the assertion sits between them. Mechanism from Astra and Popper; + // the retraction of the stronger claim from IntelMac, who traced it rather than + // just accepting the correction. + // + // Paused, the clock advances only when every task is idle and only to the nearest + // deadline, so the 50ms timeout fires before the 400ms sleep can complete, on any + // machine, every time. + #[tokio::test(start_paused = true)] + async fn a_module_that_cannot_join_in_time_produces_join_timed_out() { + let runtime = Runtime::new(); + runtime.register(Arc::new(SlowJoiner)); + let receipt = runtime + .shutdown_within(std::time::Duration::from_millis(50)) + .await; + let row = receipt + .modules + .iter() + .find(|m| m.module == "slow-joiner") + .expect("the slow module must appear in the receipt"); + assert_eq!(row.outcome, ModuleStopOutcome::JoinTimedOut); + assert!(!row.state_is_durable()); + } + + // what this catches: the ordinary path degrading. A module that drains, saves and + // joins promptly must still come back Clean and durable under the same bound — or + // the two tests above would pass for the wrong reason. + // Paused for the same reason, and it matters MORE here: a control that ran on a + // different clock from the cases it controls is not a control. + #[tokio::test(start_paused = true)] + async fn a_prompt_module_is_clean_under_the_same_bound() { + let runtime = Runtime::new(); + let (module, _r) = RecordingModule::new("prompt", Vec::new()); + runtime.register(module); + let receipt = runtime + .shutdown_within(std::time::Duration::from_millis(50)) + .await; + let row = receipt + .modules + .iter() + .find(|m| m.module == "prompt") + .expect("present"); + assert_eq!(row.outcome, ModuleStopOutcome::Clean); + assert!(row.state_is_durable()); + assert!(receipt.state_is_durable()); + } + } + + /// The shutdown receipt's semantics. These are pure and cheap, and they exist because + /// the whole rail turns on ONE distinction: a module that failed to SAVE lost state, + /// and a module that failed to JOIN merely exited untidily. Collapse them and + /// `continuum stop` goes back to reporting success for a stop that lost a citizen's + /// working memory. + mod shutdown_receipt { + use super::*; + + fn stop(module: &str, outcome: ModuleStopOutcome) -> ModuleStop { + ModuleStop { + module: module.to_string(), + drain: DrainOutcome::Drained, + outcome, + ms: 1, + } + } + + // what this catches: a save timeout treated as a successful stop. This is the + // exact value the CLI's exit code keys on. + // what this catches: a failed JOIN being read as durable state. I originally + // ruled that a module which saved and then failed to let go was untidy rather + // than lossy — but `ServiceModule::shutdown` is contractually "release resources, + // FLUSH BUFFERS", and the logger's implementation is `flush_all`. A join that did + // not finish is a flush that may not have happened, which makes the last writes + // exactly as gone as a failed save's. Astra read the contract I had written and + // I had not. + #[test] + fn only_a_fully_completed_stop_can_support_a_durability_claim() { + assert!(ModuleStopOutcome::Clean.completed()); + assert!(!ModuleStopOutcome::SaveTimedOut.completed()); + assert!(!ModuleStopOutcome::SaveFailed { + error: "disk full".into() + } + .completed()); + // The two that used to pass, and must not. + assert!(!ModuleStopOutcome::JoinTimedOut.completed()); + assert!(!ModuleStopOutcome::JoinFailed { + error: "task refused to join".into() + } + .completed()); + } + + // what this catches: THE clobber. When drain and save/join shared one enum, an + // incomplete drain followed by a join timeout produced `JoinTimedOut`, the drain + // result was discarded, and the module reported its state durable — a save taken + // mid-turn, described as clean. Found in review by Astra, not by me. + #[test] + fn an_incomplete_drain_survives_a_failing_join_and_stays_non_durable() { + let torn = ModuleStop { + module: "cognition".into(), + drain: DrainOutcome::Incomplete { in_flight: 3 }, + outcome: ModuleStopOutcome::JoinTimedOut, + ms: 1, + }; + // The join outcome alone would once have said "saved, just untidy". + assert!(!torn.outcome.completed()); + // The whole module is NOT durable, because the save was taken over a turn. + assert!(!torn.state_is_durable()); + assert_eq!(torn.drain, DrainOutcome::Incomplete { in_flight: 3 }); + } + + // what this catches: a drain that FAILED being reported as a measured zero. + // "I could not find out how much was in flight" is not "nothing was in flight". + #[test] + fn a_failed_drain_is_unknown_not_a_measured_zero() { + let unknown = DrainOutcome::Unknown { + reason: "drain exceeded 2s".into(), + }; + assert!(!unknown.is_quiet()); + assert_ne!(unknown, DrainOutcome::Incomplete { in_flight: 0 }); + assert_ne!(unknown, DrainOutcome::Drained); + } + + // what this catches: a summary that hides which module lost state. "3 modules did + // not save" without names sends the reader to a log that the exiting process may + // not have flushed. + #[test] + fn the_summary_names_the_modules_that_did_not_save() { + let receipt = ShutdownReceipt { + modules: vec![ + stop("logger", ModuleStopOutcome::Clean), + stop("cognition", ModuleStopOutcome::SaveTimedOut), + ], + total_ms: 42, + }; + assert!(!receipt.state_is_durable()); + assert_eq!(receipt.unsaved().len(), 1); + let summary = receipt.summary(); + assert!(summary.contains("cognition"), "got: {summary}"); + assert!( + !summary.contains("logger"), + "clean modules are noise here: {summary}" + ); + } + + // what this catches: an EMPTY receipt reading as a successful stop. A core that + // registered no modules, or a shutdown that never ran one, produces an empty list + // — and "all state durable" over zero modules is true but says nothing. The + // summary must show the count so a reader can see it was zero. + #[test] + fn an_empty_receipt_reports_the_count_it_actually_stopped() { + let receipt = ShutdownReceipt { + modules: Vec::new(), + total_ms: 0, + }; + assert!(receipt.state_is_durable()); + assert!( + receipt.summary().contains("0 modules"), + "an empty stop must say so: {}", + receipt.summary() + ); + } + } + /// Card 506a388c: the socket route (uu / IPC / MCP / desktop) dispatched /// typed and legacy commands directly and never consulted the interceptor /// chain, so a peer-addressed `ai/generate {aircPeer}` from the CLI ran @@ -1481,7 +2294,10 @@ mod conditional_modules_tests { .await .expect("an error is an answer, not None"); let err = out.expect_err("the refusal must surface"); - assert!(err.contains("refuses") && err.contains("the address was bad"), "{err}"); + assert!( + err.contains("refuses") && err.contains("the address was bad"), + "{err}" + ); } // what this catches: the card itself — the real airc interceptor on the @@ -1931,19 +2747,27 @@ mod piece_2_pr3_dispatch_tests { use std::any::Any; use std::sync::Arc; - struct RecordingModule { + /// `pub(super)` so sibling test mods can use the ONE recording fixture instead of + /// each writing their own — the duplication CLAUDE.md's task #155 exists to stop. + pub(super) struct RecordingModule { name: &'static str, subscriptions: Vec, received: Arc>>, + /// How many times the runtime asked this module to SAVE. Counting broadcasts + /// rather than receipt rows is the difference between "the receipt mentions this + /// module once" and "the module was stopped once" — a second broadcast could stop + /// it again and still produce one row per publication. + pub(super) saves: Arc, } impl RecordingModule { - fn new( + pub(super) fn new( name: &'static str, subscriptions: Vec, ) -> (Arc, Arc>>) { let received = Arc::new(Mutex::new(Vec::new())); let module = Arc::new(Self { + saves: Arc::new(std::sync::atomic::AtomicUsize::new(0)), name, subscriptions, received: received.clone(), @@ -1954,6 +2778,15 @@ mod piece_2_pr3_dispatch_tests { #[async_trait] impl ServiceModule for RecordingModule { + /// Counts the runtime's save BROADCASTS at this module. The idempotence test + /// asserts on this rather than on receipt rows: a second `begin` that ran a + /// parallel broadcast would call this twice while still publishing one receipt. + async fn save_state(&self) -> Result<(), String> { + self.saves + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + Ok(()) + } + fn config(&self) -> ModuleConfig { ModuleConfig { name: self.name, diff --git a/core/continuum-core/src/runtime/service_module.rs b/core/continuum-core/src/runtime/service_module.rs index 76e443adce..7f53d7a392 100644 --- a/core/continuum-core/src/runtime/service_module.rs +++ b/core/continuum-core/src/runtime/service_module.rs @@ -308,6 +308,28 @@ pub trait ServiceModule: Send + Sync + Any { None } + /// STOP TAKING NEW WORK and let what is already in flight finish, within the + /// caller's bound. Broadcast by [`Runtime::shutdown`] BEFORE `save_state`. + /// + /// Returns the number of items still in flight when the method returns — `0` means + /// drained, non-zero means the bound expired with work outstanding and whatever + /// `save_state` writes next is a snapshot taken mid-turn. Reporting the count rather + /// than a bool is what lets the receipt say how much was lost instead of only that + /// something was. + /// + /// This exists because suspending a module's own tick is NOT a drain. The persona + /// registry's `quiesce_all` stops each mind's autonomic self-tick, which is what a + /// measurement lease needs, and leaves room input arriving and active turns running — + /// so a save that follows a quiesce can still be taken underneath a turn that is + /// halfway through writing. + /// + /// Default is `Ok(0)`, honest for a module with no producer of its own. A module + /// that accepts work from outside itself implements this or its shutdown saves a + /// torn state BY CONTRACT. + async fn drain(&self) -> Result { + Ok(0) + } + /// SAVE this node's volatile state to its durable home — the explicit half /// of the CBAR contract (Joel 2026-09-02: "I can call all nodes and tell /// them to save or load state"). Broadcast by [`Runtime::shutdown`] before