diff --git a/crates/tinymemory-module/Cargo.toml b/crates/tinymemory-module/Cargo.toml index dc4323e..6a017b1 100644 --- a/crates/tinymemory-module/Cargo.toml +++ b/crates/tinymemory-module/Cargo.toml @@ -77,11 +77,15 @@ log = "0.4" serde = { version = "1", features = ["derive"] } serde_json = "1" # The interface macro requires every method to be `async fn`, so a runtime has -# to exist. -tokio = { version = "1", features = ["macros", "rt-multi-thread", "sync"] } +# to exist. `time` is for the per-hook deadline in `host::run_shutdown_hooks`; +# it compiled without it only because another crate in the graph happened to +# enable the feature, which is not something to depend on. +tokio = { version = "1", features = ["macros", "rt-multi-thread", "sync", "time"] } [dev-dependencies] -tokio = { version = "1", features = ["macros", "rt-multi-thread", "time"] } +# `test-util` gives the paused clock the shutdown-deadline test advances, so it +# asserts the timeout without waiting out a real five seconds. +tokio = { version = "1", features = ["macros", "rt-multi-thread", "time", "test-util"] } # The host-side contract. A dev-dependency, not a normal one: the module serves # `tinymemory-api` types directly and needs nothing from this crate to run. What # it needs is the assertion — that the members it serves are exactly the ones diff --git a/crates/tinymemory-module/src/config_loader.rs b/crates/tinymemory-module/src/config_loader.rs index 4506648..3f19f9f 100644 --- a/crates/tinymemory-module/src/config_loader.rs +++ b/crates/tinymemory-module/src/config_loader.rs @@ -35,9 +35,12 @@ //! setting after this module loaded gets the old value from anything in this //! process until the host reloads the module. //! -//! That is a real degradation and it is reported once per process the first -//! time anything consults this loader — `report_unserved_once`, the same -//! latch-and-report the scheduler-gate and shutdown stubs use. Closing it +//! That is a real limit and it is logged once per process the first time +//! anything consults this loader — `warn_degraded_once`, which keeps the log +//! line the scheduler-gate stub emits but leaves the error reporter alone. It +//! is a documented design limit rather than something gone wrong, and the host +//! is the one that handed this module the frozen snapshot, so reporting it as a +//! defect only pages someone about a decision already made. Closing it //! properly means a host-pushed config signal (this module declares //! `signals = []`), not a bus *pull*: a pull would re-introduce the two-answers //! problem above while still being stale between ticks. @@ -79,7 +82,7 @@ use tinymemory_core::Config; use tinymemory_tinycortex::engine::EngineRuntimeConfig; use crate::config::ModuleConfig; -use crate::host::report_unserved_once; +use crate::host::warn_degraded_once; /// Latched so the degradation is named once per process rather than once per /// call — `ProviderContext::execute` reloads on *every* Composio action, and an @@ -152,7 +155,7 @@ impl ConfigLoader for ModuleConfigLoader { /// is no read to fail. The `Result` is the contract's, shaped for a host /// that reads a file. async fn load(&self) -> Result, String> { - report_unserved_once(&LOADER_REPORTED, CONFIG_LOADER_FROZEN, "config_loader"); + warn_degraded_once(&LOADER_REPORTED, CONFIG_LOADER_FROZEN); let owned: Box = Box::new((*self.snapshot).clone()); Ok(owned) } @@ -174,7 +177,7 @@ impl ConfigLoader for ModuleConfigLoader { /// another user's store. The paths are compared rather than the values /// because `config_path` is what the contract itself calls the anchor. async fn reload_snapshot(&self, snapshot: &Config) -> Result, String> { - report_unserved_once(&LOADER_REPORTED, CONFIG_LOADER_FROZEN, "config_loader"); + warn_degraded_once(&LOADER_REPORTED, CONFIG_LOADER_FROZEN); if snapshot.config_path() != self.snapshot.config_path() { return Err(FOREIGN_SNAPSHOT.to_string()); } diff --git a/crates/tinymemory-module/src/host.rs b/crates/tinymemory-module/src/host.rs index 7c76619..30373e7 100644 --- a/crates/tinymemory-module/src/host.rs +++ b/crates/tinymemory-module/src/host.rs @@ -197,11 +197,19 @@ pub(crate) fn install(connection: Connection) { // syncing within seconds. That half is benign; the paragraph above is not, and // closing it needs the same `SchedulerGate` bus interface named below. // -// Note also what is deliberately *not* built here: a module-local registry that -// banked shutdown hooks and drained them on the module's own `Shutdown` method. -// Nothing calls `MemoryProvider::shutdown()` on the way out of the host process, -// so those hooks would still never run — and they would stop reporting. That -// trades a loud gap for a quiet one. +// The module-local registry this paragraph used to argue against now exists — +// see `ModuleShutdownHost` — and the argument is worth recording because it was +// right about the part that has not changed. Banking hooks and draining them on +// the module's own `Shutdown` method is only useful if something calls it, and +// at the time of writing nothing in the host does on the way out. +// +// What tipped it is that banking is not worse than dropping in the case the +// paragraph worried about, and is better in every other: a dropped hook never +// runs, a banked one runs the moment the host wires the call. The objection was +// really to going quiet, not to banking, so the gap is still announced at +// `install_seams` — now naming the precise condition ("only when the host calls +// Shutdown") instead of a flat "unserved" that a host-side fix would leave +// stale. See tinyhumansai/tinymemory#133 for the host half. /// Latched so the gap is reported once per process, not once per job claim — /// `wait_for_capacity` is consulted before every claim, and an unlatched report @@ -209,8 +217,19 @@ pub(crate) fn install(connection: Connection) { /// storage-failure reports. static GATE_REPORTED: AtomicBool = AtomicBool::new(false); -/// Latched for the same reason: one hook dropped means every later one is too. -static SHUTDOWN_REPORTED: AtomicBool = AtomicBool::new(false); +/// Log a degradation once per process, without reporting it as a defect. +/// +/// The sibling of [`report_unserved_once`] for a gap that is a documented +/// design limit rather than something gone wrong. Both keep the log line; only +/// this one leaves the error reporter alone, so a limit the host already knows +/// about — it is the host that hands this module the frozen snapshot — stops +/// arriving as a classified error on the host's side. +pub(crate) fn warn_degraded_once(latch: &AtomicBool, message: &'static str) { + if latch.swap(true, Ordering::SeqCst) { + return; + } + log::warn!("[tinymemory:module] {message}"); +} /// What the missing scheduler gate costs, in the terms a reader of the log needs. const GATE_UNSERVED: &str = "scheduler gate unserved in module mode: background memory work in \ @@ -219,11 +238,6 @@ const GATE_UNSERVED: &str = "scheduler gate unserved in module mode: background the periodic sync loops here therefore also ignore the \ \"Memory Tree off\" and \"signed out\" pauses that would stop them"; -/// What the missing shutdown host costs. -const SHUTDOWN_UNSERVED: &str = "shutdown host unserved in module mode: a memory shutdown hook \ - was dropped, so in-flight queue job locks are not released on a \ - clean exit and the next launch waits out the lease instead"; - /// Log and report a seam degradation once per process. /// /// Shared with [`crate::composio`] and [`crate::config_loader`], which have @@ -448,18 +462,114 @@ impl tinymemory_core::scheduler_gate::SchedulerGate for UnservedSchedulerGate { /// receive the same handle for a wait on it to mean anything. static IDLE_NOTIFY: std::sync::OnceLock> = std::sync::OnceLock::new(); -/// The host's shutdown sequencer, which this module has no way to reach. +/// Hooks the engine registered, waiting for this module's `Shutdown` member. +/// +/// A plain `Mutex` because [`ShutdownHost::register`] is synchronous and the +/// engine registers from wherever the queue starts; the lock is held only long +/// enough to push or to take the list. +static BANKED_HOOKS: std::sync::Mutex> = + std::sync::Mutex::new(Vec::new()); + +/// Longest any one hook may hold up the module's shutdown. +/// +/// The bus call that triggers a shutdown carries the caller's deadline, so a +/// hook that wedges must not be able to outlast it — the caller would time out +/// and learn nothing. Releasing job locks is a handful of local SQLite writes; +/// anything past this is stuck, not slow, and the lease-expiry path at the next +/// startup is what it degrades to. +const HOOK_DEADLINE: std::time::Duration = std::time::Duration::from_secs(5); + +/// Banks the engine's shutdown hooks so [`run_shutdown_hooks`] can await them. +/// +/// # Why this is banked here rather than dropped +/// +/// The engine registers exactly one hook today: `queue::worker` releasing the +/// in-flight job locks so a clean restart re-claims that work immediately +/// instead of waiting out the lease. Dropping it guaranteed the slow path on +/// every launch. +/// +/// This module does have a moment to await them — the `Shutdown` member it +/// already serves. What it does **not** have is a guarantee that anyone calls +/// it: a host that exits without shutting the driver down still leaves the +/// locks to expire, which is why [`install_seams`] keeps saying so out loud. +/// Banking is strictly better than dropping either way — the hook runs when the +/// host asks, instead of never — but it is only half of the fix, and the other +/// half lives in the host. #[derive(Debug)] -pub(crate) struct UnservedShutdownHost; +pub(crate) struct ModuleShutdownHost; -impl tinymemory_core::shutdown::ShutdownHost for UnservedShutdownHost { +impl tinymemory_core::shutdown::ShutdownHost for ModuleShutdownHost { fn register(&self, hook: tinymemory_core::shutdown::ShutdownHook) { - // Dropped, not banked: there is no moment inside this module at which it - // could be awaited. The hard-kill path is what remains — leases expire - // and startup recovery reclaims them — so this is a degradation, not a - // loss of data, and the report is classified accordingly. - drop(hook); - report_unserved_once(&SHUTDOWN_REPORTED, SHUTDOWN_UNSERVED, "shutdown_host"); + match BANKED_HOOKS.lock() { + Ok(mut hooks) => hooks.push(hook), + // A poisoned lock means a previous holder panicked mid-push. The + // hook is dropped rather than recovered: shutdown is best-effort + // and the lease still expires, so this must not panic a caller. + Err(_) => log::warn!( + "[tinymemory:module] shutdown hook dropped: the hook registry is poisoned; \ + in-flight job locks will be reclaimed by lease expiry instead" + ), + } + } +} + +/// Run and clear every banked shutdown hook. +/// +/// Draining is what makes this idempotent, which the `Shutdown` member's +/// contract requires: a second call finds nothing banked and releases nothing +/// twice. +/// +/// Each hook runs in its own task under [`HOOK_DEADLINE`], so one that panics +/// or wedges costs its own deadline and not the whole shutdown. Both outcomes +/// degrade to the same place a dropped hook did — lease expiry at the next +/// startup — so neither is worth failing the call over. +pub(crate) async fn run_shutdown_hooks() { + // The guard is confined to this block deliberately. A `std::sync` + // `MutexGuard` is not `Send`, so merely being in scope across the awaits + // below would make this whole future non-`Send` — and the bus member that + // calls it has to be. Taking the list here also means a hook that registers + // another one cannot deadlock against a lock this function still holds. + let hooks = { + let Ok(mut banked) = BANKED_HOOKS.lock() else { + log::warn!("[tinymemory:module] shutdown hooks skipped: the hook registry is poisoned"); + return; + }; + std::mem::take(&mut *banked) + }; + if hooks.is_empty() { + return; + } + let total = hooks.len(); + log::info!("[tinymemory:module] running {total} banked shutdown hook(s)"); + for hook in hooks { + // Spawned so a panic inside a hook surfaces as a `JoinError` here + // rather than unwinding through the bus call. + let mut task = tokio::spawn(async move { hook().await }); + // `&mut` so the handle survives the timeout. Passing it by value would + // hand ownership to `timeout`, which drops it on expiry — and dropping + // a `JoinHandle` **detaches** the task rather than cancelling it. The + // hook would then still be running, still holding the store, while + // `provider.shutdown()` released the backend underneath it. + match tokio::time::timeout(HOOK_DEADLINE, &mut task).await { + Ok(Ok(())) => {} + Ok(Err(error)) => log::warn!( + "[tinymemory:module] a shutdown hook panicked: {error}; its work is left to \ + lease expiry at the next startup" + ), + Err(_) => { + task.abort(); + // `abort` only requests cancellation, so await the handle to + // know the task has actually stopped touching the store before + // this returns and the caller tears the backend down. The + // result is a `Cancelled` join error and carries nothing worth + // reporting past the warning below. + let _ = task.await; + log::warn!( + "[tinymemory:module] a shutdown hook exceeded {HOOK_DEADLINE:?} and was \ + cancelled; its work is left to lease expiry at the next startup" + ); + } + } } } @@ -486,16 +596,22 @@ pub(crate) fn install_seams(connection: Option) { tinymemory_core::scheduler_gate::set_scheduler_gate(Arc::new(UnservedSchedulerGate)); } } - tinymemory_core::shutdown::set_shutdown_host(Arc::new(UnservedShutdownHost)); + tinymemory_core::shutdown::set_shutdown_host(Arc::new(ModuleShutdownHost)); // One line, once per process — `setup` runs exactly once. A warning rather // than a debug line because a reader of the log should not have to diff // seam lists to find out which host behaviours are not in effect here. + // + // Shutdown is half-served and the wording says which half: the hooks are + // banked and this module runs them, but only if the host calls `Shutdown`. + // A host that exits without doing so still leaves the locks to expire, and + // that is not something this process can detect or fix from in here. log::warn!( - "[tinymemory:module] shutdown is unserved in module mode: a stub keeps the unwired \ - behaviour and reports once when consulted, so graceful queue-lock release is not \ - honoured inside this process. The scheduler gate is bus-backed when the host serves \ - SchedulerPolicy, and degrades to the unwired Policy::Normal stub behaviour when it \ - does not" + "[tinymemory:module] shutdown hooks are banked and run on this module's Shutdown \ + member, so graceful queue-lock release is honoured only when the host shuts the \ + driver down before exiting; a host that exits without calling Shutdown still leaves \ + in-flight locks to lease expiry. The scheduler gate is bus-backed when the host \ + serves SchedulerPolicy, and degrades to the unwired Policy::Normal stub behaviour \ + when it does not" ); } diff --git a/crates/tinymemory-module/src/host_test.rs b/crates/tinymemory-module/src/host_test.rs index ff6bea9..9f0d3d9 100644 --- a/crates/tinymemory-module/src/host_test.rs +++ b/crates/tinymemory-module/src/host_test.rs @@ -7,6 +7,22 @@ use tinymemory_api::host::{ ErrorReporter, MemoryEvent, MemoryEventSink, SpacyEntity, SpacyResponse, }; +/// Empty the shutdown-hook bank without running anything. +/// +/// The bank is process-wide, so a test that asserts on what ran needs a known +/// starting point — otherwise a hook another test registered would count. Pair +/// it with `seam_lock::hold_global_seams_async`, which is what stops two such +/// tests interleaving. +/// +/// Lives here rather than beside the bank because production sources carry no +/// test-only executable code; a child module still reaches its ancestor's +/// private statics. +fn drain_banked_hooks() { + if let Ok(mut hooks) = super::BANKED_HOOKS.lock() { + hooks.clear(); + } +} + struct HostSeamsRestore { event_sink: Option>, error_reporter: Option>, @@ -275,9 +291,8 @@ async fn install_wires_every_seam_this_module_can_supply() { } #[test] -fn the_unserved_stubs_answer_exactly_what_an_unwired_seam_answered() { +fn the_unserved_scheduler_gate_answers_exactly_what_an_unwired_seam_answered() { use tinymemory_core::scheduler_gate::SchedulerGate; - use tinymemory_core::shutdown::ShutdownHost; // Loud, not different. A stub that answered anything else would change // scheduling as a side effect of loading the module — and with no channel @@ -287,9 +302,142 @@ fn the_unserved_stubs_answer_exactly_what_an_unwired_seam_answered() { super::UnservedSchedulerGate.current_policy(), tinymemory_core::scheduler_gate::Policy::Normal ); - // Registering with nowhere to run reports and drops; it must never panic. - let hook: tinymemory_core::shutdown::ShutdownHook = Box::new(|| Box::pin(async {})); - super::UnservedShutdownHost.register(hook); +} + +/// A registered hook is banked and runs when the module is shut down. +/// +/// This is the whole point of the seam: the engine's one hook releases the +/// in-flight job locks, and before it was banked it was dropped, so a clean +/// restart always waited out the lease instead of re-claiming the work. +#[tokio::test] +async fn a_registered_hook_is_banked_and_runs_on_shutdown() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use tinymemory_core::shutdown::ShutdownHost; + + // Declared before any statement: the module crate denies items appearing + // after statements, and a process-wide counter is what lets the assertions + // below distinguish "ran once" from "ran twice". + static RUNS: AtomicUsize = AtomicUsize::new(0); + + let _seams = crate::seam_lock::hold_global_seams_async().await; + drain_banked_hooks(); + RUNS.store(0, Ordering::SeqCst); + + super::ModuleShutdownHost.register(Box::new(|| { + Box::pin(async { + RUNS.fetch_add(1, Ordering::SeqCst); + }) + })); + assert_eq!( + RUNS.load(Ordering::SeqCst), + 0, + "registering must not run it" + ); + + super::run_shutdown_hooks().await; + assert_eq!(RUNS.load(Ordering::SeqCst), 1, "the banked hook ran"); + + // Draining is what makes the `Shutdown` member idempotent, which its + // contract requires: a second call must not release the same locks twice. + super::run_shutdown_hooks().await; + assert_eq!( + RUNS.load(Ordering::SeqCst), + 1, + "a second shutdown runs nothing" + ); +} + +/// One bad hook costs its own deadline, not the shutdown. +/// +/// Both a panic and a wedge degrade to where a dropped hook already left the +/// work — lease expiry at the next startup — so neither may take the rest of +/// the shutdown sequence down with it. +#[tokio::test] +async fn a_panicking_hook_does_not_stop_the_others() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use tinymemory_core::shutdown::ShutdownHost; + + static SURVIVORS: AtomicUsize = AtomicUsize::new(0); + + let _seams = crate::seam_lock::hold_global_seams_async().await; + drain_banked_hooks(); + SURVIVORS.store(0, Ordering::SeqCst); + + super::ModuleShutdownHost.register(Box::new(|| Box::pin(async { panic!("hook exploded") }))); + super::ModuleShutdownHost.register(Box::new(|| { + Box::pin(async { + SURVIVORS.fetch_add(1, Ordering::SeqCst); + }) + })); + + super::run_shutdown_hooks().await; + assert_eq!( + SURVIVORS.load(Ordering::SeqCst), + 1, + "the hook after the panicking one still ran" + ); +} + +/// A hook that never finishes is cancelled at its deadline, and the next one +/// still runs. +/// +/// The deadline is the guard against a wedged hook outliving the bus call that +/// triggered the shutdown, so it is worth asserting rather than assuming. On a +/// paused clock the timeout fires the moment the runtime goes idle, so this +/// costs no real time and cannot flake on a loaded machine. +/// +/// It also pins the cancellation: the hook must not still be running when this +/// returns, because the caller releases the store next. +#[tokio::test(start_paused = true)] +async fn a_hook_that_never_finishes_is_cancelled_at_its_deadline() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use tinymemory_core::shutdown::ShutdownHost; + + static STARTED: AtomicUsize = AtomicUsize::new(0); + static FINISHED: AtomicUsize = AtomicUsize::new(0); + static AFTER: AtomicUsize = AtomicUsize::new(0); + + let _seams = crate::seam_lock::hold_global_seams_async().await; + drain_banked_hooks(); + STARTED.store(0, Ordering::SeqCst); + FINISHED.store(0, Ordering::SeqCst); + AFTER.store(0, Ordering::SeqCst); + + super::ModuleShutdownHost.register(Box::new(|| { + Box::pin(async { + STARTED.fetch_add(1, Ordering::SeqCst); + std::future::pending::<()>().await; + FINISHED.fetch_add(1, Ordering::SeqCst); + }) + })); + super::ModuleShutdownHost.register(Box::new(|| { + Box::pin(async { + AFTER.fetch_add(1, Ordering::SeqCst); + }) + })); + + super::run_shutdown_hooks().await; + + assert_eq!(STARTED.load(Ordering::SeqCst), 1, "the stalled hook ran"); + assert_eq!( + FINISHED.load(Ordering::SeqCst), + 0, + "it was cancelled rather than awaited to completion" + ); + assert_eq!( + AFTER.load(Ordering::SeqCst), + 1, + "one hook exceeding its deadline does not skip the rest" + ); +} + +#[tokio::test] +async fn shutting_down_with_nothing_banked_is_a_no_op() { + let _seams = crate::seam_lock::hold_global_seams_async().await; + drain_banked_hooks(); + // Reached whenever a host shuts a driver down before the queue ever + // started, which is ordinary rather than an error. + super::run_shutdown_hooks().await; } #[tokio::test] diff --git a/crates/tinymemory-module/src/lib.rs b/crates/tinymemory-module/src/lib.rs index b56ef58..00c4e0d 100644 --- a/crates/tinymemory-module/src/lib.rs +++ b/crates/tinymemory-module/src/lib.rs @@ -151,10 +151,12 @@ async fn setup(connection: Connection, mut config: ModuleConfig) -> BusResult<() // The scheduler gate is proxied to the host's SchedulerPolicy member — the // host's cron::scheduler_gate policy, polled and cached, so mode=off, // signed-out and battery pauses are honoured inside this process too. - // Shutdown stays a stub: no bus interface serves it and no local answer - // can honestly stand in for it (see `host::install_seams`). Installed with - // the rest, before the store exists, so nothing can consult a seam this - // process has not yet decided about. + // Shutdown banks the engine's hooks and runs them on this module's own + // `Shutdown` member, so graceful queue-lock release works whenever the host + // shuts the driver down before exiting (see `host::install_seams`). + // Installed with the rest, before the store exists, so nothing can consult + // a seam this process has not yet decided about — and so a hook registered + // by the queue pool below always finds the bank already there. host::install_seams(Some(connection.clone())); let client = tinymemory_core::store::factories::create_memory_client_with_local_ai( diff --git a/crates/tinymemory-module/src/service/mod.rs b/crates/tinymemory-module/src/service/mod.rs index 9ccbd0a..cad847b 100644 --- a/crates/tinymemory-module/src/service/mod.rs +++ b/crates/tinymemory-module/src/service/mod.rs @@ -368,6 +368,19 @@ impl MemoryService { /// module: `TinyBus` never unloads a library, so a host that shuts the /// driver down and rebinds gets a fresh engine inside the same mapped image. async fn shutdown(&self) -> BusResult<()> { + // Only the root object drains the hooks, and `opener` is what marks it: + // `OpenStore` hands back objects with `None` there. The bank is + // process-wide — the engine's hook releases the job locks for the whole + // queue, not for one subtree — so draining it when a single profile's + // store shuts down would run the release early and leave nothing banked + // for the shutdown that actually ends the process. + // + // Hooks before `provider.shutdown()`: releasing a lock is a write to the + // very store the provider is about to release, and the other order + // leaves it nothing to write through. + if self.opener.is_some() { + crate::host::run_shutdown_hooks().await; + } self.provider .shutdown() .await