From 75bd67fb413d3c8cc8febbcfc654620140c2004f Mon Sep 17 00:00:00 2001 From: Joel Teply Date: Tue, 8 Sep 2026 18:59:13 -0500 Subject: [PATCH 1/4] feat(shutdown): a stop that SAVES, and says what reached disk MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `continuum stop` went straight to a kill tree. A kill runs no module's `save_state`, so an ordinary stop discarded every module's volatile state — and the CLI printed a success line and exited 0 either way, because a dead process and a cleanly stopped one are indistinguishable when the only thing you check is whether it is gone. `Runtime::shutdown` already did the right thing, returned `()`, and nothing on the stop path ever called it. WORSE, AND FOUND WHILE FIXING IT: the Windows signal arm never ran the broadcast at all. It killed sentinels, slept a flat 2 seconds, and `_exit`ed — verbatim 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 that day and this half was not, so on Windows no module has ever saved on a signal stop, and the node that runs the citizens is a Windows node. WHAT THIS ADDS - `ServiceModule::drain()`, broadcast BEFORE `save_state`, returning the count still in flight. Suspending a tick is not a drain: `quiesce_all` stops each mind's self-tick and deliberately leaves her reachable, so a save after a quiesce can still land underneath a turn halfway through writing. - `Runtime::shutdown` returns a `ShutdownReceipt`. Durability requires the drain to have been quiet AND every phase to have completed — including the join, because `shutdown` is contractually "release resources, FLUSH BUFFERS" and a join that did not finish is a flush that may not have happened. - `system/shutdown`, Privileged. A persona should not be able to stop the node it is thinking inside. Travels the socket path `ping` uses; no new transport. - `ShutdownOperation` owns the broadcast. A socket handler is cancelled when its client goes away, and a shutdown cancelled after ingress closed left the node refusing work with nothing saved — strictly worse than the kill it replaces. Idempotent, so a signal racing the verb joins rather than saving twice. - The verb no longer exits the process. A handler that kills its own process cannot also answer, and a timer guessing when its answer flushed is a guess. - `stop` asks before it kills and its exit code means "gone AND durable". `reboot` deliberately does not fail on an unsaved module. A core that predates this rail is `LegacyCore`, not a refusal: the one-time cost of the upgrade, named so a rollout does not look like a fault. ONE GATE, TWO USERS `AdmissionGate` is one word — high bit CLOSED, low bits the in-flight count, CAS admission. Written twice (turns, log queue) and the second copy reintroduced the race the first had removed: check a flag, a close lands, the drain reads zero, and the reservation then increments a queue whose writer is being joined. It is a TYPE because a process-wide one-way static cannot be tested — closing the real gate poisons every later test in the binary, so the tests reimplemented the logic instead of calling it and could not fail when production drifted. Same reason `ShutdownOperation` and the logger's failure counter are instances. REVIEW Astra and S6 found, in my code: the drain/join clobber, the admit/close race, an `await_shutdown` that burned a core on a closed channel, `core_is_up` reading a timeout as absence, a missing pidfile read as proof, two process snapshots where one was needed, Linux's 15-byte `comm` truncation defeating the enumerator, the logger's uncounted producers and its publish-then-count ordering, swallowed I/O failures under a Clean receipt, and three tests that could not fail for the reason they existed. All fixed. 21 tests, all driving production paths: the race written out step by step, a forced failed CAS via a hook between the load and the exchange, a close landing mid-retry, the receipt surviving every observer leaving, a second begin joining rather than restarting, and a real failed write followed by a real successful one ending in a non-clean stop. test result: ok. 21 passed; 0 failed; 0 ignored; 7906 filtered out NOT DONE: legacy checkpoint adoption with provenance. The outgoing core has no verb to call, so the first upgrade still loses its volatile state; that wants its own card. And no test has yet been WATCHED to go red on a mutation of the publisher — that remains a source-review claim. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Q4NU4VNiELPQfBpCacDZGc --- core/continuum-core/src/bin/continuum.rs | 272 +++++++- core/continuum-core/src/cognition/mod.rs | 1 + .../src/cognition/turn_ingress.rs | 106 +++ core/continuum-core/src/commands/log/write.rs | 6 +- .../src/commands/log/write_batch.rs | 4 +- core/continuum-core/src/commands/system.rs | 4 + .../src/commands/system/shutdown.rs | 148 ++++ core/continuum-core/src/main.rs | 15 +- core/continuum-core/src/modules/cognition.rs | 46 ++ core/continuum-core/src/modules/logger.rs | 383 +++++++++- .../src/persona/service_loop.rs | 22 + .../src/runtime/admission_gate.rs | 338 +++++++++ core/continuum-core/src/runtime/mod.rs | 8 +- core/continuum-core/src/runtime/runtime.rs | 657 +++++++++++++++++- .../src/runtime/service_module.rs | 22 + 15 files changed, 1971 insertions(+), 61 deletions(-) create mode 100644 core/continuum-core/src/cognition/turn_ingress.rs create mode 100644 core/continuum-core/src/commands/system/shutdown.rs create mode 100644 core/continuum-core/src/runtime/admission_gate.rs diff --git a/core/continuum-core/src/bin/continuum.rs b/core/continuum-core/src/bin/continuum.rs index 374e9bd9c2..4f8820d61e 100644 --- a/core/continuum-core/src/bin/continuum.rs +++ b/core/continuum-core/src/bin/continuum.rs @@ -902,7 +902,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() @@ -1839,6 +1843,61 @@ fn bound_elsewhere_hint(bound: &[(i32, String)], socket: &str) -> Option /// every platform, so there is one implementation instead of a Unix tool plus an unported gap. /// Matching on the executable NAME rather than `pgrep -f`'s full command line is also more precise /// here: it cannot accidentally match the bash process that is merely launching the core. +/// Both questions from ONE snapshot: was the process table readable, and which cores are +/// in it. +/// +/// Asking them separately reads two different moments of the world — a core can start or +/// exit between the calls, so "the table was readable" and "there were no cores" can be +/// true of different tables and neither statement constrains the other. That is the same +/// two-instruments-one-conclusion error as checking an open flag and then a count. +/// +/// The readability disambiguator is that THIS process always exists: a table with zero +/// entries in it is a table we did not read, not a machine with nothing running. +fn core_process_evidence() -> (bool, Vec) { + use sysinfo::{ProcessRefreshKind, ProcessesToUpdate, System, UpdateKind}; + let mut sys = System::new(); + // `.with_exe(...)` is REQUIRED, not an optimisation. Linux truncates `comm` to 15 + // bytes, and "continuum-core-server" is 21 — so `p.name()` there is + // "continuum-core-" and never contains the full string. `ProcessRefreshKind::nothing` + // does not request executable paths, so `p.exe()` would be `None` as well, and BOTH + // arms of the match would miss every core on the box. The enumeration would come back + // empty and be read as "nothing is running", which is the silent false-absence this + // whole function exists to prevent. Found by Astra. + sys.refresh_processes_specifics( + ProcessesToUpdate::All, + true, + ProcessRefreshKind::nothing().with_exe(UpdateKind::OnlyIfNotSet), + ); + + // READABILITY IS PROVEN BY FINDING OURSELVES, not by a non-empty table. A partial or + // permission-limited read can return some processes and still be missing the ones we + // care about; the one process we KNOW must be in a correct snapshot is this one. + let me = sysinfo::get_current_pid().ok(); + let readable = me.map(|pid| sys.process(pid).is_some()).unwrap_or(false); + if !readable { + return (false, Vec::new()); + } + + // Matched on the 15-byte-safe prefix as well as the full name, because the truncated + // `comm` is what Linux actually reports and it is a legitimate match, not a fallback. + const FULL: &str = "continuum-core-server"; + let truncated: &str = &FULL[..FULL.len().min(15)]; + let cores = sys + .processes() + .values() + .filter(|p| { + let name = p.name().to_string_lossy(); + name.contains(FULL) + || name.contains(truncated) + || p.exe() + .map(|e| e.to_string_lossy().contains(FULL)) + .unwrap_or(false) + }) + .map(|p| p.pid().as_u32() as i32) + .collect(); + (true, cores) +} + fn processes_named(fragment: &str) -> Vec { use sysinfo::{ProcessRefreshKind, ProcessesToUpdate, System}; let mut sys = System::new(); @@ -2573,8 +2632,188 @@ 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, +} + +/// 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 (table_readable, cores) = core_process_evidence(); + if !table_readable { + return GracefulStop::NoAnswer( + "the process table could not be read, so an empty result proves nothing about what is running" + .to_string(), + ); + } + let survivors = !cores.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 @@ -2587,7 +2826,30 @@ 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; + 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 @@ -2703,7 +2965,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; @@ -2729,7 +2991,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 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..f664200ad7 --- /dev/null +++ b/core/continuum-core/src/cognition/turn_ingress.rs @@ -0,0 +1,106 @@ +//! 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(); + +/// May a service loop begin a new turn? Prefer [`admit`], which answers this AND takes +/// ownership of the turn in one atomic step; a bare read is only safe for display. +pub fn is_open() -> bool { + GATE.is_open() +} + +/// 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() { + assert!(is_open(), "citizens must be able to take turns in a booted core"); + // Real admission through the real gate — not a reimplementation of 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..12ead5af34 --- /dev/null +++ b/core/continuum-core/src/commands/system/shutdown.rs @@ -0,0 +1,148 @@ +//! `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, then exit the process. Returns the receipt + /// BEFORE exiting, so the caller learns what was saved rather than only that the + /// process is gone. + 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 195618f964..a4ad6d369c 100644 --- a/core/continuum-core/src/modules/cognition.rs +++ b/core/continuum-core/src/modules/cognition.rs @@ -167,6 +167,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 778829c5e2..9059dc8fe2 100644 --- a/core/continuum-core/src/persona/service_loop.rs +++ b/core/continuum-core/src/persona/service_loop.rs @@ -513,6 +513,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/runtime.rs b/core/continuum-core/src/runtime/runtime.rs index 8f017eb925..2b31eb926b 100644 --- a/core/continuum-core/src/runtime/runtime.rs +++ b/core/continuum-core/src/runtime/runtime.rs @@ -670,50 +670,77 @@ 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 { + const PER_PHASE: std::time::Duration = std::time::Duration::from_secs(2); 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(); 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 +1419,311 @@ 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. State is durable; the + /// module's resources were abandoned to process exit. + 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's state is durable. Deliberately NOT "nothing went wrong" — a module + /// that saved and then failed to join is clean by this measure, because the thing at + /// stake is the citizen's state, not the tidiness of the exit. + 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. +pub struct ShutdownOperation { + result: tokio::sync::watch::Sender>, + started: std::sync::atomic::AtomicBool, +} + +impl ShutdownOperation { + pub 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. + pub 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() + } +} + +/// The process's one shutdown. `watch::channel` is not `const`, so the instance is built +/// on first use; the `OnceLock` being `static` is what lets `begin` take `&'static self` +/// and hand the receiver out with no lifetime attached to a caller. +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 +1731,254 @@ 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" + ); + } + } + + /// 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 @@ -1931,19 +2508,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 +2539,14 @@ 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 From 760c68b2948e1f54f1b62b813b71ae9a5b5543f4 Mon Sep 17 00:00:00 2001 From: Joel Teply Date: Tue, 8 Sep 2026 20:29:11 -0500 Subject: [PATCH 2/4] fix(shutdown): the reachability item was the TYPE, not the accessor; produced timeout outcomes; deterministic clocks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review corrections from Astra, S6 and IntelMac on #3929, none of which I found myself. THE RATCHET FIX WAS WRONG. I removed `turn_ingress::is_open()` and announced it as the fix for `unwired_public_machinery_never_increases` (108 vs 107). The scanner greps `pub struct ` / `pub enum ` — production_reachability.rs:125 — and does not count functions at all, so that could never have moved the number. The counted item is `pub struct ShutdownOperation`, now private along with `begin` and `shutdown_within`. `is_open` stays removed because it genuinely had no caller, but its comment no longer claims to be the ratchet fix. TIMEOUT OUTCOMES ARE NOW PRODUCED, NOT WRITTEN DOWN. `SaveTimedOut` and `JoinTimedOut` appeared only as literals in assertions; no test drove the runtime into emitting one, so both `Err(_)` arms were untaken and deleting the timeout handling would have left every receipt test green. `shutdown_within(per_phase)` takes the bound as a parameter — the constant was the reason the arms were unreachable — and three tests drive a real Runtime: a module that cannot save in time, one that cannot join in time, and a prompt module under the same bound so the first two cannot pass for the wrong reason. VIRTUAL TIME, because a duration margin is not a guarantee about poll order. Pinned Tokio's `Timeout::poll` polls the INNER future first, so a stalled scheduler could let a 400ms sleep complete before a 50ms timeout was polled. `start_paused` removes the wall clock: the timeout fires first on any machine. The flake's direction is a spurious FAILURE — the tests assert the timeout outcome, so a stall makes them go red, not green. An earlier version of this comment claimed the opposite and was wrong. Also: `ModuleContext` imported from `crate::runtime` rather than the module that merely uses it, and `handle_command` added to both stubs — the trait has no default and I had copied a nearby impl's shape without checking what was required. test result: ok. 24 passed; 0 failed; 0 ignored; 7906 filtered out NOT IN THIS COMMIT, and owned by root's integration worktree: the canary merge preserving typed `CoreProcessEvidence`, and the LegacyCore reboot guard. Mine's enumerator matched the truncated Linux name unconditionally, which fixes a false absence by risking a false presence; theirs treats a missing exe as uncertainty and says so. STILL MISSING: the five ts-rs outputs. The filtered test run did not execute the export tests, so they were not generated. Saying so rather than assuming the derive fired. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Q4NU4VNiELPQfBpCacDZGc --- .../src/cognition/turn_ingress.rs | 18 +- core/continuum-core/src/runtime/runtime.rs | 231 +++++++++++++++++- 2 files changed, 234 insertions(+), 15 deletions(-) diff --git a/core/continuum-core/src/cognition/turn_ingress.rs b/core/continuum-core/src/cognition/turn_ingress.rs index f664200ad7..5f6a80e2dd 100644 --- a/core/continuum-core/src/cognition/turn_ingress.rs +++ b/core/continuum-core/src/cognition/turn_ingress.rs @@ -31,12 +31,6 @@ use crate::runtime::AdmissionGate; /// reintroduced the race the first had just removed. static GATE: AdmissionGate = AdmissionGate::new(); -/// May a service loop begin a new turn? Prefer [`admit`], which answers this AND takes -/// ownership of the turn in one atomic step; a bare read is only safe for display. -pub fn is_open() -> bool { - GATE.is_open() -} - /// 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 @@ -96,8 +90,16 @@ mod tests { // it to the service loop, so that is all this asserts. #[test] fn the_process_turn_gate_starts_open_and_empty() { - assert!(is_open(), "citizens must be able to take turns in a booted core"); - // Real admission through the real gate — not a reimplementation of it. + // 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); diff --git a/core/continuum-core/src/runtime/runtime.rs b/core/continuum-core/src/runtime/runtime.rs index 2b31eb926b..b996332a74 100644 --- a/core/continuum-core/src/runtime/runtime.rs +++ b/core/continuum-core/src/runtime/runtime.rs @@ -678,7 +678,17 @@ impl Runtime { /// 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 { - const PER_PHASE: std::time::Duration = std::time::Duration::from_secs(2); + 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!( "Stopping {} modules (drain → save → join, parallel, 2s bound per phase)...", @@ -693,23 +703,23 @@ impl Runtime { // 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 { + 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()), + 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 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 { + match tokio::time::timeout(per_phase, module.shutdown()).await { Err(_) => ModuleStopOutcome::JoinTimedOut, Ok(Err(e)) => ModuleStopOutcome::JoinFailed { error: e }, Ok(Ok(())) => ModuleStopOutcome::Clean, @@ -1584,7 +1594,7 @@ impl ShutdownReceipt { /// 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. -pub struct ShutdownOperation { +struct ShutdownOperation { result: tokio::sync::watch::Sender>, started: std::sync::atomic::AtomicBool, } @@ -1601,7 +1611,7 @@ impl ShutdownOperation { /// 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. - pub fn begin(&self, rt: Option>) -> tokio::sync::watch::Receiver> { + fn begin(&self, rt: Option>) -> tokio::sync::watch::Receiver> { if !self.started.swap(true, std::sync::atomic::Ordering::AcqRel) { match rt { Some(rt) => { @@ -1868,6 +1878,213 @@ mod conditional_modules_tests { } } + /// 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 From d2f586f632efcdec3fd526a2da565aba77982a0b Mon Sep 17 00:00:00 2001 From: Joel Teply Date: Tue, 8 Sep 2026 21:51:49 -0500 Subject: [PATCH 3/4] fix(shutdown): restore three rationale comments the integration merge dropped MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The integration branch was based on 75bd67fb4, which is an ANCESTOR of the PR tip 760c68b29 — so the merge was made against a tree predating that commit and its content was re-applied by hand. No function and no assertion was lost (both files diffed line by line). Three comment blocks were, and each is load-bearing: shutdown_within's doc — WHY the per-phase bound is a parameter. Without it the timeout arms are unreachable: SaveTimedOut / JoinTimedOut existed only as literals in assertions, and deleting the timeout handling would have left every one of those tests green. Found by IntelMac. Losing this invites someone to "simplify" the parameter back to a constant and silently un-reach the arms. The virtual-time block on both timeout tests — including the correction that the flake is a FALSE NEGATIVE and not a false positive, and that the first version of that comment claimed the opposite. Losing it leaves the wrong reading available to be re-derived with nothing in the file to stop it. turn_ingress: that removing `is_open` was NOT the ratchet fix, because the scanner counts `pub struct` / `pub enum` and never functions. Same file, same night, second silent loss in a merge: #3925's with_exe fix also had to be checked for by hand. Review catches a deleted function; nobody diffs comments. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Q4NU4VNiELPQfBpCacDZGc --- .../src/cognition/turn_ingress.rs | 11 +++- core/continuum-core/src/runtime/runtime.rs | 57 +++++++++++++++++-- 2 files changed, 63 insertions(+), 5 deletions(-) diff --git a/core/continuum-core/src/cognition/turn_ingress.rs b/core/continuum-core/src/cognition/turn_ingress.rs index 55e6eaafc6..e6dc75326d 100644 --- a/core/continuum-core/src/cognition/turn_ingress.rs +++ b/core/continuum-core/src/cognition/turn_ingress.rs @@ -90,7 +90,16 @@ mod tests { // it to the service loop, so that is all this asserts. #[test] fn the_process_turn_gate_starts_open_and_empty() { - // Admission exercises the same atomic gate used by the service loop. + // 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(), diff --git a/core/continuum-core/src/runtime/runtime.rs b/core/continuum-core/src/runtime/runtime.rs index d46622becd..1737dc8d94 100644 --- a/core/continuum-core/src/runtime/runtime.rs +++ b/core/continuum-core/src/runtime/runtime.rs @@ -686,6 +686,13 @@ impl Runtime { } /// Share the production phase bounds with deterministic timeout tests. + /// `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!( @@ -1974,8 +1981,29 @@ mod conditional_modules_tests { // 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. - // Paused time makes the phase deadline fire before the longer inner sleep. - // A wall-clock scheduler stall can make both ready, producing a spurious failure. + // `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(); @@ -2001,8 +2029,29 @@ mod conditional_modules_tests { // 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. - // Paused time makes the phase deadline fire before the longer inner sleep. - // A wall-clock scheduler stall can make both ready, producing a spurious failure. + // `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(); From bc1369a24a2c95b552a1929389328cd8c5c4e9ff Mon Sep 17 00:00:00 2001 From: Joel Teply Date: Tue, 8 Sep 2026 22:18:26 -0500 Subject: [PATCH 4/4] fix(shutdown): declare SlowSaver / SlowJoiner as UNWIRED fixtures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI caught what my local validation could not: `every_service_module_is_registered_ or_declares_why_not` failed on the two test modules added to drive the timeout outcomes. They `impl ServiceModule` and are never registered — correctly, they are `#[cfg(test)]` fixtures — but the guard requires that to be DECLARED rather than merely true. Follows the convention already in the file (thirteen `why: "fixture: ..."` entries), naming what each one drives rather than just that it is a fixture: SlowSaver -> ModuleStopOutcome::SaveTimedOut, the outcome the stop exit code keys on SlowJoiner -> ModuleStopOutcome::JoinTimedOut, which before these tests existed only as a literal in an assertion WHY I DID NOT CATCH THIS LOCALLY, since it is the more useful half: I validated with `cargo check -p continuum-core --lib --bins`, which cannot see a test-only module. The guard lives in a `#[cfg(test)]` mod and only runs under `cargo test`. `check` and `test` are different scopes and I used the narrower one on a change whose whole substance was two new test modules. Verified with the scope that catches it: test runtime::registry::tests::every_service_module_is_registered_or_declares_why_not ... ok test result: ok. 133 passed; 0 failed Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Q4NU4VNiELPQfBpCacDZGc --- core/continuum-core/src/runtime/registry.rs | 8 ++++++++ 1 file changed, 8 insertions(+) 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",