From 4b95038cf9500015564c29ebfb77c731aabe71e1 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 19:14:12 +0000 Subject: [PATCH 01/11] refactor(sync): make the fork check unavoidable by construction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four defects were found in aimdb-sync in one review pass: an untested consumer fork guard, a ForkedChild variant missing from kind()'s test and the lib.rs list, a field added by #232 that both fork guards forgot to release, and a set_value family guarded only by the accident that it delegates. All four were the same defect — someone had to remember something and the compiler could not help. "Is this runtime usable after a fork()" is one fact. It was copied into three types and checked at nine call sites that each opted in, so a tenth public method reintroduced the bug for free. And the runtime thread was four loose fields, so releasing it meant remembering four takes in two guards. Now it is one value. Runtime holds the Tokio handle, the database and the fork generation; enter() and db() are the only routes to them, so a publish or a read cannot be written that skips the check. AimDbHandle drops from six fields to two, and release_inherited becomes owned = None — impossible to half-do. waiter.rs is retired: enter() returns the handle it wrapped. No signature and no behaviour changed. Producers and consumers still hold a weak reference, so a handle's lifetime still governs. An Arc was tried first, at the request to include it, and reverted. It bought one thing — a producer outliving its handle keeps working — and cost two that Weak gets free: the failed upgrade IS the liveness check, and dropping the database is what closes buffers and wakes a reader parked in get(). Rebuilding those took a liveness flag, a stop channel and a select around every blocking read, and a forgotten producer then kept an OS thread and a runtime alive with nobody owning them — the stranded thread #232 had just removed. Design 050 §6 now records that, since the note's own aside had favoured Arc. Because the stamp is ordinary data on a value, the refusal paths are unit tests against a stale-stamped Runtime: no thread, no fork, no sleep. The forking tests drop from five in two binaries to two in one — the pthread_atfork handler really being installed, and a destructor not joining a thread this process never had, neither of which a unit test can reach. The watchdog stays. One consequence worth naming: SyncConsumer holds the Tokio handle directly and treats a failed upgrade as "detached" rather than "refuse", because a Reader can still drain what is already buffered after the runtime is gone and the characterization tests pin that. Gating reads on a live Runtime broke it. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01DKAH7JNjPLPmG4mPTvWfth --- aimdb-sync/CHANGELOG.md | 14 ++ aimdb-sync/src/consumer.rs | 71 ++++-- aimdb-sync/src/handle.rs | 226 +++++++++-------- aimdb-sync/src/lib.rs | 2 +- aimdb-sync/src/producer.rs | 58 ++--- aimdb-sync/src/runtime.rs | 203 +++++++++++++++ aimdb-sync/src/waiter.rs | 18 -- aimdb-sync/tests/fork_post_attach_test.rs | 63 ----- aimdb-sync/tests/fork_safety_test.rs | 74 +----- docs/design/050-sync-runtime-ownership.md | 294 ++++++++++++++++++++++ 10 files changed, 709 insertions(+), 314 deletions(-) create mode 100644 aimdb-sync/src/runtime.rs delete mode 100644 aimdb-sync/src/waiter.rs delete mode 100644 aimdb-sync/tests/fork_post_attach_test.rs create mode 100644 docs/design/050-sync-runtime-ownership.md diff --git a/aimdb-sync/CHANGELOG.md b/aimdb-sync/CHANGELOG.md index e0335944..27f2f163 100644 --- a/aimdb-sync/CHANGELOG.md +++ b/aimdb-sync/CHANGELOG.md @@ -46,6 +46,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 A poisoned lock could panic out of the blocking surface, which across an FFI boundary is undefined behaviour rather than an error. +### Changed + +- **The runtime thread is one value, and the `fork()` check is unavoidable.** + `Runtime` (internal) now holds the Tokio handle, the database and the fork + generation, and `enter()`/`db()` are the only routes to them — so a publish or + a read cannot be written that skips the check. Previously the same fact was + copied into three types and checked at nine call sites that each had to opt + in; four defects in one review pass were all instances of someone forgetting + to. `AimDbHandle` drops from six loose fields to two, so releasing state + inherited across a `fork` can no longer be half-done. No API signature and no + behaviour changed: producers and consumers still hold a weak reference, so a + handle's lifetime still governs. `waiter.rs` is retired — `enter()` returns + the handle it existed to wrap. See design 050. + ### Added - **`fork()` safety.** A child of `fork` inherits every handle, producer and diff --git a/aimdb-sync/src/consumer.rs b/aimdb-sync/src/consumer.rs index f8b4fc5e..8858ea1e 100644 --- a/aimdb-sync/src/consumer.rs +++ b/aimdb-sync/src/consumer.rs @@ -2,8 +2,9 @@ use aimdb_core::{DbError, Reader}; -use crate::waiter::Waiter; +use crate::runtime::Runtime; use crate::{SyncError, SyncResult}; +use alloc::sync::Weak; use core::fmt::Debug; use core::time::Duration; use std::time::Instant; @@ -56,10 +57,23 @@ pub struct SyncConsumer where T: Send + Debug + Clone, { - waiter: Waiter, + /// The runtime this consumer belongs to, for the fork check only. + /// + /// `Weak`, and deliberately not required to be alive: a `Reader` can still + /// drain what is already buffered after the runtime is gone, and + /// [`try_get`](Self::try_get) returning that data is documented behaviour. + /// A failed upgrade therefore means "detached", not "refuse" — see + /// [`Self::guard`]. + rt: Weak, + + /// A way into the Tokio runtime that owns nothing. + /// + /// Held directly rather than reached through [`Runtime`] because a blocked + /// or draining read must not keep the database alive: that is what stops + /// `detach` from closing the buffers, and a reader parked in + /// [`get`](Self::get) would then never wake. + handle: tokio::runtime::Handle, reader: Reader, - /// The fork generation this consumer was made in. See [`crate::fork`]. - made_in: crate::fork::Generation, } impl SyncConsumer @@ -67,24 +81,28 @@ where T: Send + Debug + Clone, { /// Create a new sync consumer (internal use only) - pub(crate) fn new(waiter: Waiter, reader: Reader) -> Self { - Self { - waiter, - reader, - made_in: crate::fork::generation(), - } + pub(crate) fn new( + rt: Weak, + handle: tokio::runtime::Handle, + reader: Reader, + ) -> Self { + Self { rt, handle, reader } } - /// Refuse if this process has forked since the consumer was made. - /// - /// A forked child's reader would block forever: the buffer is there, but - /// the runtime thread that fills it is not. + /// Refuse a read this process has no runtime thread for. + /// + /// The fork check is only meaningful while the runtime is alive — and in a + /// forked child it *is* alive, because the `Arc` came across with the + /// address space, which is exactly why the check exists. A failed upgrade + /// means the handle was dropped in this process, which is not a fork: let + /// the read proceed and the buffer report for itself, so data already + /// queued is still delivered. #[inline] - fn check_fork(&self) -> SyncResult<()> { - if crate::fork::forked_since(self.made_in) { - return Err(SyncError::ForkedChild); + fn guard(&self) -> SyncResult<()> { + match self.rt.upgrade() { + Some(rt) => rt.check(), + None => Ok(()), } - Ok(()) } async fn get_impl(reader: &mut Reader) -> SyncResult { @@ -132,8 +150,10 @@ where /// # } /// ``` pub fn get(&mut self) -> SyncResult { - self.check_fork()?; - self.waiter.block_on(Self::get_impl(&mut self.reader)) + self.guard()?; + self.handle + .clone() + .block_on(Self::get_impl(&mut self.reader)) } /// Get a value with a timeout. @@ -176,9 +196,10 @@ where /// # } /// ``` pub fn get_with_timeout(&mut self, timeout: Duration) -> SyncResult { - self.check_fork()?; + self.guard()?; + let handle = self.handle.clone(); let fut = async { tokio::time::timeout(timeout, Self::get_impl(&mut self.reader)).await }; - let res = self.waiter.block_on(fut); + let res = handle.block_on(fut); res.unwrap_or_else(|_| Err(SyncError::GetTimeout)) } @@ -218,7 +239,7 @@ where /// # } /// ``` pub fn try_get(&mut self) -> SyncResult { - self.check_fork()?; + self.guard()?; let res = self.reader.try_recv(); res.map_err(|e| match e { DbError::BufferClosed { .. } => SyncError::RuntimeShutdown, @@ -269,7 +290,7 @@ where /// # } /// ``` pub fn get_latest(&mut self) -> SyncResult { - self.check_fork()?; + self.guard()?; // 1) can simply sequence get_catch_up and try_get - // no one else does it simultaneously thanks to &mut self // 2) if draining ends up with an error, we follow the previous impl @@ -321,7 +342,7 @@ where /// # } /// ``` pub fn get_latest_with_timeout(&mut self, timeout: Duration) -> SyncResult { - self.check_fork()?; + self.guard()?; // see internal comments for get_latest let deadline = Instant::now() + timeout; let oldest = self.get_catch_up(Some(deadline))?; diff --git a/aimdb-sync/src/handle.rs b/aimdb-sync/src/handle.rs index 0ec4003d..f081986d 100644 --- a/aimdb-sync/src/handle.rs +++ b/aimdb-sync/src/handle.rs @@ -1,6 +1,6 @@ //! AimDB handle for managing the sync API runtime thread. -use crate::waiter::Waiter; +use crate::runtime::{Runtime, ShutdownSignal}; use crate::{SyncError, SyncResult}; use aimdb_core::{log_error, log_warn, AimDb, AimDbBuilder, DbError}; use alloc::sync::Arc; @@ -110,17 +110,33 @@ impl AimDbSyncExt for AimDb { /// is dropped without calling `detach()`, a warning will be logged /// and an emergency shutdown will be attempted. pub struct AimDbHandle { - /// Thread handle for the runtime thread - thread_handle: Option>, - - /// Shutdown signal sender - shutdown_tx: Option>, + /// The runtime thread and everything reached through it, shared with every + /// producer and consumer made from this handle. See [`crate::runtime`]. + /// + /// Holding this keeps the thread running, so a producer that outlives its + /// handle keeps working. [`detach`](Self::detach) is what stops the thread + /// deliberately. + rt: Arc, + + /// What only the handle that started the thread may do: signal it, wait for + /// it, join it. + /// + /// `None` once detached, or once a `fork` proved the thread is not this + /// process's to reap. One field, so releasing it cannot be half-done — the + /// previous shape was four loose fields and a release path that forgot one + /// of them the moment a fifth was added. + owned: Option, +} - /// Tokio runtime handle for submitting async work - runtime_handle: tokio::runtime::Handle, +/// The parts of a runtime thread that only its owner may touch. +struct OwnedThread { + /// Joined by `detach`; released, never joined, by `Drop`. + join: JoinHandle<()>, - /// Shared reference to the database (protected by Arc for thread safety) - db: Arc, + /// Commands the thread to stop. Distinct from the keepalive clone every + /// [`Runtime`] holds: dropping a sender only decrements, sending *ends* the + /// thread even while other holders are alive. + shutdown: mpsc::Sender, /// Held open by the runtime thread for exactly as long as it runs. /// @@ -132,18 +148,10 @@ pub struct AimDbHandle { /// /// Behind a `Mutex` only to keep `AimDbHandle: Sync`, which `consumer()` /// relies on — a bare `Receiver` is `Send` but not `Sync`. It is never - /// locked: every access goes through `&mut self` and takes it by value. - thread_alive: Option>>, - - /// The fork generation this handle was created in. A `fork` copies this - /// struct but not `thread_handle`'s thread. See [`crate::fork`]. - made_in: crate::fork::Generation, + /// locked: every access takes it by value. + alive: std::sync::Mutex>, } -/// Signal to shut down the runtime thread. -#[derive(Debug, Clone, Copy)] -struct ShutdownSignal; - /// What the runtime thread reports back while starting up: the thing itself, /// or why it could not be produced. /// @@ -185,7 +193,9 @@ impl AimDbHandle { let (db_tx, mut db_rx) = mpsc::channel::>>(1); let (handle_tx, mut handle_rx) = mpsc::channel::>(1); - // See `thread_alive`: never sent on, only dropped when the thread ends. + // See `OwnedThread::alive`: never sent on, only dropped when the + // thread ends. The flag beside it answers the same question without + // blocking, which is what a producer on the publish path needs. let (alive_tx, thread_alive) = std::sync::mpsc::channel::<()>(); // Spawn the runtime thread @@ -201,14 +211,13 @@ impl AimDbHandle { let runtime_handle = recv_startup(&mut handle_rx, "runtime handle")?; let db = recv_startup(&mut db_rx, "database")?; - Ok(Self { - thread_handle: Some(thread_handle), - shutdown_tx: Some(shutdown_tx), - thread_alive: Some(std::sync::Mutex::new(thread_alive)), + Ok(Self::assemble( runtime_handle, db, - made_in: crate::fork::generation(), - }) + shutdown_tx, + thread_handle, + thread_alive, + )) } pub(crate) fn new(db: AimDb) -> SyncResult { @@ -221,7 +230,7 @@ impl AimDbHandle { // `new_from_builder` has always used. See `recv_startup`. let (handle_tx, mut handle_rx) = mpsc::channel::>(1); - // See `thread_alive`: never sent on, only dropped when the thread ends. + // See `OwnedThread::alive` and `RunningFlag`. let (alive_tx, thread_alive) = std::sync::mpsc::channel::<()>(); // Wrap database in Arc for sharing @@ -272,39 +281,32 @@ impl AimDbHandle { let runtime_handle = recv_startup(&mut handle_rx, "runtime handle")?; - Ok(Self { - thread_handle: Some(thread_handle), - shutdown_tx: Some(shutdown_tx), - thread_alive: Some(std::sync::Mutex::new(thread_alive)), + Ok(Self::assemble( runtime_handle, db, - made_in: crate::fork::generation(), - }) - } - - /// Drop everything tied to a runtime thread that does not exist here. - /// - /// A forked child inherited all of it: a `JoinHandle` for a thread this - /// process never had, the sender that would signal it to stop, and the - /// liveness channel that reports when it did. None of it means anything on - /// this side of the `fork`, and joining that handle panics inside `std`. - /// - /// One place rather than two, so a field added to this struct later is - /// released by both the `detach` and `Drop` guards or by neither — not by - /// whichever one its author happened to read. - fn release_inherited(&mut self) { - let _ = self.shutdown_tx.take(); - let _ = self.thread_handle.take(); - let _ = self.thread_alive.take(); + shutdown_tx, + thread_handle, + thread_alive, + )) } - /// Refuse if this process has forked since the handle was created. - #[inline] - fn check_fork(&self) -> SyncResult<()> { - if crate::fork::forked_since(self.made_in) { - return Err(SyncError::ForkedChild); + /// Both constructors end the same way: one shared [`Runtime`], one owned + /// thread. Written once so they cannot drift. + fn assemble( + runtime_handle: tokio::runtime::Handle, + db: Arc, + shutdown: mpsc::Sender, + join: JoinHandle<()>, + alive: std::sync::mpsc::Receiver<()>, + ) -> Self { + Self { + rt: Arc::new(Runtime::new(runtime_handle, db)), + owned: Some(OwnedThread { + join, + shutdown, + alive: std::sync::Mutex::new(alive), + }), } - Ok(()) } /// Create a synchronous producer for type `T`. @@ -334,8 +336,8 @@ impl AimDbHandle { where T: Send + 'static + Debug + Clone, { - self.check_fork()?; - Ok(crate::SyncProducer::new(Arc::downgrade(&self.db), key)) + self.rt.check()?; + Ok(crate::SyncProducer::new(Arc::downgrade(&self.rt), key)) } /// Create a synchronous consumer for type `T`. @@ -371,11 +373,18 @@ impl AimDbHandle { where T: Send + Sync + 'static + Debug + Clone, { - self.check_fork()?; + self.rt.check()?; let record_key = key.as_ref().to_string(); - let reader = self.db.subscribe::(&record_key).map_err(SyncError::Db)?; - let waiter = Waiter::new(self.runtime_handle.clone()); - Ok(crate::SyncConsumer::new(waiter, reader)) + let reader = self + .rt + .db_unchecked() + .subscribe::(&record_key) + .map_err(SyncError::Db)?; + Ok(crate::SyncConsumer::new( + Arc::downgrade(&self.rt), + self.rt.enter()?.clone(), + reader, + )) } /// Gracefully shut down the runtime thread. @@ -447,61 +456,64 @@ impl AimDbHandle { } /// Internal detach implementation. + /// + /// Deliberate shutdown: *sends* the signal rather than merely dropping a + /// sender, so the thread stops even though producers and consumers may + /// still hold [`Runtime`] clones keeping it alive. Those survivors then + /// fail with [`SyncError::RuntimeShutdown`], which is the point — `detach` + /// means "stop now", not "stop when everyone has finished". fn detach_internal(&mut self, timeout: Option) -> SyncResult<()> { // A forked child holds a `JoinHandle` for a thread that does not exist // here, and joining it is not merely useless: it panics inside `std` // with "threads should not terminate unexpectedly", which for an FFI // caller means a Rust backtrace on stderr from a destructor. Release - // the handle instead — the thread is the parent's to reap. - if crate::fork::forked_since(self.made_in) { - self.release_inherited(); + // it instead — the thread is the parent's to reap. + if self.rt.check().is_err() { + self.owned = None; return Err(SyncError::ForkedChild); } - // Send shutdown signal - if let Some(shutdown_tx) = self.shutdown_tx.take() { - // Try to send shutdown signal (non-blocking) - // If it fails, the runtime may have already stopped - let _ = shutdown_tx.try_send(ShutdownSignal); - } - - let Some(thread_handle) = self.thread_handle.take() else { + let Some(OwnedThread { + join, + shutdown, + alive, + }) = self.owned.take() + else { return Ok(()); }; + // Non-blocking. Failure means the thread has already stopped. + let _ = shutdown.try_send(ShutdownSignal); + if let Some(duration) = timeout { // `JoinHandle` has no timed join. Rather than park a helper thread // in `join()` — which could not be reclaimed when the wait expired, // stranding it for the life of the process — wait on the liveness - // channel the runtime thread holds open. See `thread_alive`. - if let Some(alive) = self.thread_alive.take() { - // Taken by value, so this cannot block and cannot fail; the - // poisoned arm is unreachable because nothing ever locks it. - let alive = alive.into_inner().unwrap_or_else(|e| e.into_inner()); - match alive.recv_timeout(duration) { - // The thread dropped its sender, so it is on its way out - // and the join below returns promptly. - Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {} - - // Still running. Release the `JoinHandle` instead of - // blocking on it: the shutdown signal was delivered, so the - // thread stops on its own and drops the database with it. - Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { - return Err(SyncError::DetachFailed { - message: format!( - "Runtime thread did not shut down within {:?}", - duration - ), - }); - } - - // Nothing is ever sent on this channel. - Ok(()) => {} + // channel the runtime thread holds open. See `OwnedThread::alive`. + // + // Taken by value, so this cannot block and cannot fail; the + // poisoned arm is unreachable because nothing ever locks it. + let alive = alive.into_inner().unwrap_or_else(|e| e.into_inner()); + match alive.recv_timeout(duration) { + // The thread dropped its sender, so it is on its way out and + // the join below returns promptly. + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {} + + // Still running. Release the `JoinHandle` instead of blocking + // on it: the shutdown signal was delivered, so the thread stops + // on its own and drops the database with it. + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { + return Err(SyncError::DetachFailed { + message: format!("Runtime thread did not shut down within {:?}", duration), + }); } + + // Nothing is ever sent on this channel. + Ok(()) => {} } } - thread_handle.join().map_err(|_| SyncError::DetachFailed { + join.join().map_err(|_| SyncError::DetachFailed { message: "Runtime thread panicked during shutdown".to_string(), })?; @@ -577,22 +589,18 @@ impl Drop for AimDbHandle { /// /// Call [`detach`](Self::detach) if you need to know that it did. fn drop(&mut self) { - // A child's handle owns nothing that runs. Releasing it quietly is - // correct; the warning below is for a *parent* that forgot to detach. - if crate::fork::forked_since(self.made_in) { - self.release_inherited(); + // A forked child's handle owns nothing that runs here: signalling would + // reach a thread this process never had. Release and say nothing. + if self.rt.check().is_err() { + self.owned = None; return; } - if self.thread_handle.is_some() { + if let Some(owned) = self.owned.take() { log_warn!("AimDbHandle dropped without calling detach()"); log_warn!("Shutdown was signalled; the runtime thread stops on its own"); - - if let Some(shutdown_tx) = self.shutdown_tx.take() { - let _ = shutdown_tx.try_send(ShutdownSignal); - } - // Released rather than joined — see the note above. - let _ = self.thread_handle.take(); + // Non-blocking. Released rather than joined — see the note above. + let _ = owned.shutdown.try_send(ShutdownSignal); } } } diff --git a/aimdb-sync/src/lib.rs b/aimdb-sync/src/lib.rs index 8327aeb1..a2c0dec6 100644 --- a/aimdb-sync/src/lib.rs +++ b/aimdb-sync/src/lib.rs @@ -220,7 +220,7 @@ mod handle; #[cfg(feature = "std")] mod producer; #[cfg(feature = "std")] -mod waiter; +mod runtime; #[cfg(feature = "std")] pub use consumer::SyncConsumer; diff --git a/aimdb-sync/src/producer.rs b/aimdb-sync/src/producer.rs index 8ab5bdbf..a98a9872 100644 --- a/aimdb-sync/src/producer.rs +++ b/aimdb-sync/src/producer.rs @@ -1,7 +1,8 @@ //! Synchronous producer for typed records. +use crate::runtime::Runtime; use crate::{SyncError, SyncResult}; -use aimdb_core::{AimDb, TryProduceError}; +use aimdb_core::TryProduceError; use alloc::sync::Weak; use core::fmt::Debug; use core::marker::PhantomData; @@ -40,10 +41,14 @@ pub struct SyncProducer where T: Send + 'static + Debug + Clone, { - db: Weak, + /// The runtime this producer publishes into. + /// + /// `Weak`, so the handle's lifetime still governs: when it goes, the + /// database goes with it and this upgrade fails. That failure *is* the + /// liveness check — see [`crate::runtime`] on why sharing ownership here + /// would cost more than it bought. + rt: Weak, key: String, - /// The fork generation this producer was made in. See [`crate::fork`]. - made_in: crate::fork::Generation, // same reasons as for Producer in aimdb-core/src/typed_api.rs _phantom: PhantomData T>, } @@ -53,26 +58,22 @@ where T: Send + 'static + Debug + Clone, { /// Create a new sync producer (internal use only) - pub(crate) fn new(db: Weak, key: impl AsRef) -> Self { + pub(crate) fn new(rt: Weak, key: impl AsRef) -> Self { Self { - db, + rt, key: key.as_ref().into(), - made_in: crate::fork::generation(), _phantom: PhantomData, } } - /// Refuse if this process has forked since the producer was made. + /// The runtime, or [`SyncError::RuntimeShutdown`] if it is gone. /// - /// Checked before the `Weak` upgrade, because a forked child's upgrade - /// *succeeds* — the `Arc` was copied with the address space — which is - /// exactly why the buffer would accept a value nobody will ever read. + /// Every publish goes through here and then through + /// [`Runtime::db`](crate::runtime::Runtime::db), so the fork check cannot + /// be skipped by adding a method and forgetting to guard it. #[inline] - fn check_fork(&self) -> SyncResult<()> { - if crate::fork::forked_since(self.made_in) { - return Err(SyncError::ForkedChild); - } - Ok(()) + fn runtime(&self) -> SyncResult> { + self.rt.upgrade().ok_or(SyncError::RuntimeShutdown) } /// Set the value, blocking until it can be sent. @@ -106,12 +107,8 @@ where /// # } /// ``` pub fn set(&self, value: T) -> SyncResult<()> { - self.check_fork()?; - if let Some(db) = self.db.upgrade() { - db.produce(&self.key, value).map_err(SyncError::Db) - } else { - Err(SyncError::RuntimeShutdown) - } + let rt = self.runtime()?; + rt.db()?.produce(&self.key, value).map_err(SyncError::Db) } /// Try to set the value without blocking. @@ -148,16 +145,13 @@ where /// # } /// ``` pub fn try_set(&self, value: T) -> SyncResult<()> { - self.check_fork()?; - if let Some(db) = self.db.upgrade() { - let producer = db.producer(&self.key)?; - producer.try_produce(value).map_err(|e| match e { - TryProduceError::Full(_) => SyncError::SetTimeout, - TryProduceError::Closed(_) => SyncError::RuntimeShutdown, - }) - } else { - Err(SyncError::RuntimeShutdown) - } + let rt = self.runtime()?; + let db = rt.db()?; + let producer = db.producer(&self.key)?; + producer.try_produce(value).map_err(|e| match e { + TryProduceError::Full(_) => SyncError::SetTimeout, + TryProduceError::Closed(_) => SyncError::RuntimeShutdown, + }) } } diff --git a/aimdb-sync/src/runtime.rs b/aimdb-sync/src/runtime.rs new file mode 100644 index 00000000..2eae118f --- /dev/null +++ b/aimdb-sync/src/runtime.rs @@ -0,0 +1,203 @@ +//! The runtime thread, as one value. +//! +//! `aimdb-sync` spawns an OS thread to own a Tokio runtime on the caller's +//! behalf. Everything that thread *is* — the way in, the database it built, and +//! the `fork` generation it belongs to — lives here, in one type, reached by +//! every object that needs it. +//! +//! # Why this is one type +//! +//! It used to be four fields on [`AimDbHandle`](crate::AimDbHandle) and a fork +//! stamp copied into three others. That shape cost four bugs in one review +//! pass, all of the same kind: someone had to remember something and the +//! compiler could not help. A guard was added to a new method — or wasn't. A +//! field was added to the handle — and the release path forgot it. +//! +//! # Why the check lives on the way in +//! +//! [`Runtime::enter`] is the only route to the Tokio handle and +//! [`Runtime::db`] the only route to the database, and both refuse a runtime +//! this process did not inherit a thread for. A publish or a read cannot be +//! written that skips the check, because it cannot be written without going +//! through one of them. Design 050 calls this "checked by construction rather +//! than by convention"; #231 gave panic-freedom the same treatment. +//! +//! The check is deliberately *before* the database is handed out. A forked +//! child's `Arc` is perfectly valid — it came across with the address +//! space — so a child that reached the database would publish into a buffer +//! nobody drains and be told `Ok`. That silence is the bug this whole mechanism +//! exists to prevent. +//! +//! # Who owns it +//! +//! The handle owns the `Arc`; producers and consumers hold a +//! [`Weak`](alloc::sync::Weak). So the database dies with the handle, exactly +//! as it did when producers held `Weak` directly, and two things stay +//! free that shared ownership would have made expensive: +//! +//! - **Liveness.** A failed upgrade *is* the check — no flag to keep in sync. +//! - **Waking a blocked reader.** Dropping the database closes its buffers, +//! which is what wakes a consumer parked in `get()`. `aimdb-core` has no +//! explicit close, so nothing else would. +//! +//! An `Arc` here was tried and reverted. It bought one thing — a producer +//! outliving its handle keeps working — and cost both of the above, each of +//! which had to be rebuilt by hand (a liveness flag, a level-triggered stop +//! channel, and a select around every blocking read). It also made a forgotten +//! producer keep an OS thread and a Tokio runtime alive with nobody owning +//! them, which is the stranded thread #232 had just removed. + +use alloc::sync::Arc; + +use aimdb_core::AimDb; + +use crate::error::{SyncError, SyncResult}; + +/// Signal to shut down the runtime thread. +#[derive(Debug, Clone, Copy)] +pub(crate) struct ShutdownSignal; + +/// A runtime thread and everything reached through it. +pub(crate) struct Runtime { + /// The way into the Tokio runtime. Reached only through [`Self::enter`]. + handle: tokio::runtime::Handle, + + /// The database the thread built. Reached only through [`Self::db`]. + db: Arc, + + /// The fork generation this runtime's thread was spawned in. + /// + /// Plain data, which is the point: every refusal path can be tested by + /// building a `Runtime` with a stale value, without a thread and without a + /// real `fork`. See the unit tests below. + made_in: crate::fork::Generation, +} + +impl Runtime { + pub(crate) fn new(handle: tokio::runtime::Handle, db: Arc) -> Self { + Self { + handle, + db, + made_in: crate::fork::generation(), + } + } + + /// Whether this process still owns the thread this runtime names. + /// + /// One relaxed atomic load and a comparison — no syscall, no lock. It sits + /// on the publish path, which is why it is not a `getpid` call. + #[inline] + pub(crate) fn check(&self) -> SyncResult<()> { + if crate::fork::forked_since(self.made_in) { + return Err(SyncError::ForkedChild); + } + Ok(()) + } + + /// The Tokio handle, or [`SyncError::ForkedChild`]. + /// + /// The only way to reach it. Blocking on a runtime whose thread did not + /// survive a `fork` would park forever. + #[inline] + pub(crate) fn enter(&self) -> SyncResult<&tokio::runtime::Handle> { + self.check()?; + Ok(&self.handle) + } + + /// The database, or [`SyncError::ForkedChild`]. + /// + /// The only way to reach it. See the module note on why the check must come + /// first: a forked child's handle to the database is valid, and that is + /// exactly the problem. + #[inline] + pub(crate) fn db(&self) -> SyncResult<&Arc> { + self.check()?; + Ok(&self.db) + } + + /// The database without the fork check, for the one caller that has already + /// made it: [`AimDbHandle::consumer`](crate::AimDbHandle::consumer) checks, + /// then subscribes. + /// + /// Not a hole in the guarantee — it is `pub(crate)` and its one use sits + /// directly after a [`Self::check`] — but it is the reason [`Self::db`] + /// exists as the ordinary path. + #[inline] + pub(crate) fn db_unchecked(&self) -> &Arc { + &self.db + } +} + +#[cfg(test)] +mod tests { + use super::*; + use aimdb_core::AimDbBuilder; + use aimdb_tokio_adapter::TokioAdapter; + + /// A `Runtime` over a real database, stamped `generations_behind` behind + /// the process. One behind is what a forked child's inherited runtime + /// looks like. + /// + /// No thread is spawned and no `fork` happens — which is the point of + /// moving the stamp onto a value. Before this, proving a refusal path meant + /// forking a real process from a parent holding a live Tokio runtime, the + /// least safe moment there is; that suite failed 11 runs in 60 until it was + /// mitigated (design 050 §8). + fn runtime_stamped(generations_behind: u64) -> (tokio::runtime::Runtime, Runtime) { + let tokio_rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("tokio runtime"); + + let (db, _runner) = tokio_rt + .block_on(async { + AimDbBuilder::new() + .runtime(Arc::new(TokioAdapter)) + .build() + .await + }) + .expect("build database"); + + let mut runtime = Runtime::new(tokio_rt.handle().clone(), Arc::new(db)); + runtime.made_in = crate::fork::generation().wrapping_sub(generations_behind); + + // Returned so the caller keeps it alive: the handle we cloned out of it + // must not outlive the runtime it came from. + (tokio_rt, runtime) + } + + #[test] + fn a_runtime_from_this_generation_is_usable() { + let (_guard, rt) = runtime_stamped(0); + assert!(rt.check().is_ok()); + assert!(rt.enter().is_ok()); + assert!(rt.db().is_ok()); + } + + /// Every route in refuses — not just the one someone remembered to guard. + /// This is the property the old per-object `check_fork` could not state. + #[test] + fn a_runtime_from_before_a_fork_refuses_every_route_in() { + let (_guard, rt) = runtime_stamped(1); + assert!(matches!(rt.check(), Err(SyncError::ForkedChild))); + assert!(matches!(rt.enter(), Err(SyncError::ForkedChild))); + assert!(matches!(rt.db(), Err(SyncError::ForkedChild))); + } + + /// A child that forked twice is no more usable than one that forked once. + #[test] + fn any_distance_behind_refuses() { + let (_guard, rt) = runtime_stamped(2); + assert!(matches!(rt.check(), Err(SyncError::ForkedChild))); + } + + /// Terminal for the same reason `RuntimeShutdown` is: the thread is gone + /// and will not come back in this process, so a caller must not retry. + #[test] + fn the_refusal_classifies_as_closed() { + use aimdb_core::DbErrorKind; + let (_guard, rt) = runtime_stamped(1); + let err = rt.enter().expect_err("must refuse"); + assert_eq!(err.kind(), DbErrorKind::Closed); + } +} diff --git a/aimdb-sync/src/waiter.rs b/aimdb-sync/src/waiter.rs deleted file mode 100644 index fb172141..00000000 --- a/aimdb-sync/src/waiter.rs +++ /dev/null @@ -1,18 +0,0 @@ -//! tokio-specific implementation of running the given future -//! on the current thread until completion - -use std::future::Future; - -pub struct Waiter { - handle: tokio::runtime::Handle, -} - -impl Waiter { - pub fn new(handle: tokio::runtime::Handle) -> Self { - Self { handle } - } - - pub fn block_on(&self, fut: F) -> F::Output { - self.handle.block_on(fut) - } -} diff --git a/aimdb-sync/tests/fork_post_attach_test.rs b/aimdb-sync/tests/fork_post_attach_test.rs deleted file mode 100644 index ba308f6a..00000000 --- a/aimdb-sync/tests/fork_post_attach_test.rs +++ /dev/null @@ -1,63 +0,0 @@ -//! A database the child attaches *after* forking is its own and must work. -//! -//! This is why the guard is a generation counter and not a poison flag: a -//! supervisor that forks per job and then does its own work would otherwise -//! find the API dead for no reason. -//! -//! # Why this has a binary to itself -//! -//! Its child allocates and spawns a thread — proving `attach` works cannot -//! avoid that — so it is the most exposed of the fork tests to a lock inherited -//! from a thread that did not survive. It deadlocked on the first CI run and -//! hung until the six-hour job ceiling killed it. -//! -//! What makes it safe here is not the separate binary as such: measurement -//! showed harness parallelism has no bearing on the hang rate (see the table in -//! `fork_child`). It is that this test attaches **nothing before forking**, so -//! the parent has no runtime thread that could be mid-allocation when the fork -//! happens — every other fork test must attach first in order to have something -//! to inherit. Its own binary is what keeps it that way: one `attach` anywhere -//! else in the process would undo it. The watchdog in `fork_child` bounds what -//! is left. -#![cfg(all(unix, feature = "std"))] -use aimdb_core::{buffer::BufferCfg, AimDbBuilder}; -use aimdb_sync::AimDbBuilderSyncExt; -use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; -use serde::{Deserialize, Serialize}; -use std::sync::Arc; - -mod fork_child; -use fork_child::in_forked_child; - -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] -struct Reading { - value: u32, -} - -#[test] -fn a_child_can_attach_its_own_database_after_forking() { - let code = in_forked_child(|| { - // Built inside the child on purpose: nothing is attached in the parent, - // so no runtime thread exists to be missing from this address space. - let mut builder = AimDbBuilder::new().runtime(Arc::new(TokioAdapter)); - builder.configure::("sensor.reading", |reg| { - reg.buffer(BufferCfg::SpmcRing { capacity: 8 }) - .tap(|_ctx, _consumer| async move {}); - }); - let handle = match builder.attach() { - Ok(h) => h, - Err(_) => return false, - }; - let producer = match handle.producer::("sensor.reading") { - Ok(p) => p, - Err(_) => return false, - }; - let published = producer.set(Reading { value: 7 }).is_ok(); - let detached = handle.detach().is_ok(); - published && detached - }); - assert_eq!( - code, 0, - "a post-fork attach belongs to the child and must work" - ); -} diff --git a/aimdb-sync/tests/fork_safety_test.rs b/aimdb-sync/tests/fork_safety_test.rs index 10490e27..8dad8ef7 100644 --- a/aimdb-sync/tests/fork_safety_test.rs +++ b/aimdb-sync/tests/fork_safety_test.rs @@ -1,5 +1,13 @@ //! What a `fork()`ed child is told about handles it inherited. //! +//! Only what a unit test cannot prove. Every *refusal* path is now checked in +//! `runtime.rs` against a `Runtime` stamped with a stale generation — no +//! thread, no fork, no sleep — because the stamp became ordinary data when the +//! runtime became a value (design 050 §8). What survives here needs a real +//! child process: that the `pthread_atfork` handler is genuinely installed and +//! fires, and that a destructor in that child does not join a thread this +//! process never had. +//! //! `fork` copies the address space but not the threads, so the child holds a //! handle whose runtime thread does not exist in this process. The failure this //! guards is not a crash but a silence: before the generation check, the child's @@ -10,7 +18,6 @@ use aimdb_sync::{AimDbBuilderSyncExt, SyncError}; use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; use serde::{Deserialize, Serialize}; use std::sync::Arc; -use std::time::Duration; mod fork_child; use fork_child::in_forked_child; @@ -68,71 +75,6 @@ fn a_forked_child_is_refused_rather_than_silently_dropped() { handle.detach().expect("detach"); } -/// An inherited *consumer* must refuse too, and must drop quietly. -/// -/// Its reader would otherwise block forever: the buffer came across with the -/// address space, but the runtime thread that fills it did not. -#[test] -fn a_forked_child_is_refused_by_an_inherited_consumer() { - let handle = attach(); - let mut inherited = handle - .consumer::("sensor.reading") - .expect("consumer"); - - let code = in_forked_child(move || { - let refused_try = matches!(inherited.try_get(), Err(SyncError::ForkedChild)); - // `get` blocks in the parent; in the child it must return, not park. - let refused_get = matches!(inherited.get(), Err(SyncError::ForkedChild)); - let refused_latest = matches!(inherited.get_latest(), Err(SyncError::ForkedChild)); - let refused_timeout = matches!( - inherited.get_with_timeout(Duration::from_millis(50)), - Err(SyncError::ForkedChild) - ); - let refused_latest_timeout = matches!( - inherited.get_latest_with_timeout(Duration::from_millis(50)), - Err(SyncError::ForkedChild) - ); - - // Leaked, not dropped — see the note in the producer test. This test - // asserts the five refusals above; `dropping_an_inherited_handle_does_ - // not_panic` is where the destructor itself is under test. - std::mem::forget(inherited); - - refused_try && refused_get && refused_latest && refused_timeout && refused_latest_timeout - }); - assert_eq!( - code, 0, - "every read on an inherited consumer should be refused" - ); - - handle.detach().expect("detach"); -} - -/// The handle must stop handing out new producers and consumers too, or the -/// guard is trivially bypassed by making a fresh one in the child. -#[test] -fn a_forked_child_cannot_make_new_producers_or_consumers() { - let handle = attach(); - - let code = in_forked_child(move || { - let no_producer = matches!( - handle.producer::("sensor.reading"), - Err(SyncError::ForkedChild) - ); - let no_consumer = matches!( - handle.consumer::("sensor.reading"), - Err(SyncError::ForkedChild) - ); - std::mem::forget(handle); - - no_producer && no_consumer - }); - assert_eq!( - code, 0, - "the child should get no new producers or consumers" - ); -} - /// Dropping an inherited handle must not join a thread this process does not /// have. That join panics inside `std` with "threads should not terminate /// unexpectedly" — from a destructor, which across an FFI boundary means a Rust diff --git a/docs/design/050-sync-runtime-ownership.md b/docs/design/050-sync-runtime-ownership.md new file mode 100644 index 00000000..73c21b92 --- /dev/null +++ b/docs/design/050-sync-runtime-ownership.md @@ -0,0 +1,294 @@ +# 050 — `aimdb-sync`: one runtime, one invariant + +**Status:** **implemented.** §6's open question is decided — `Weak`, not `Arc` +— and the reasoning below is amended where building it proved the note wrong. +Written against the tip of the #230 → #232 → #233 stack, which has merged. + +**Scope:** internal restructuring of `aimdb-sync`. No change to `aimdb-core`, no +change to the blocking API's shape (`attach` / `producer` / `consumer` / `set` / +`get` / `detach` all keep their signatures). One deliberate semantic decision is +open — §6 — and it is the reason this is a design note rather than a patch. + +**Vocabulary:** the **runtime thread** is the OS thread `aimdb-sync` spawns to +own a Tokio runtime on the caller's behalf. A **stamp** is the fork generation +recorded when an object is created, compared later to decide whether the object +predates a `fork()`. + +--- + +## 1. Where this sits + +`aimdb-sync` exists so an FFI or legacy caller gets a blocking API from a plain +`fn main()` with no `#[tokio::main]`. To do that it spawns and owns a runtime +thread the caller never sees. That decision is sound and is not in question +here — owning the thread is precisely why the crate, and not its callers, is the +layer that can notice the thread has gone. + +What follows is about how that owned thread is *represented*. + +## 2. What prompted this + +Four defects were found in `aimdb-sync` in a single review pass. They look +unrelated. They are the same defect. + +| # | Defect | Found by | +|---|---|---| +| 1 | `SyncConsumer`'s fork guard shipped with no test covering it | reading, not a failing test | +| 2 | `SyncError::ForkedChild` absent from `kind()`'s test and from the `lib.rs` variant list | reading; survived a PR that rewrote that exact test | +| 3 | `thread_alive` added to `AimDbHandle`, and both fork guards silently forgot to release it | reading | +| 4 | `set_value` / `try_set_value` / `set_value_at` are guarded only because they happen to delegate to `set` / `try_set` | reading; nothing enforces it | + +None was caught by a test, because in each case the code compiles, runs and +passes. Each is an instance of *someone had to remember something, and the +compiler could not help.* + +## 3. The two structural causes + +### 3.1 The invariant is enforced by convention, not construction + +"Is this runtime still usable after a `fork()`?" is **one fact**. Today it is +stored in three places and checked in nine: + +| | count | where | +|---|---|---| +| types carrying a copied `made_in` stamp | 3 | [`AimDbHandle`](../../aimdb-sync/src/handle.rs#L112), [`SyncProducer`](../../aimdb-sync/src/producer.rs#L39), [`SyncConsumer`](../../aimdb-sync/src/consumer.rs#L55) | +| hand-written `check_fork` implementations | 3 | `handle.rs`, `producer.rs`, `consumer.rs` | +| call sites that must *remember* to call it | 9 | 2 + 2 + 5 | +| public methods on `SyncProducer` / `SyncConsumer` | 5 + 5 | every one a chance to forget | + +The stamp is *copied per object* rather than being a property of the shared +runtime. Adding a tenth public method reintroduces the bug for free, and nothing +in the type system objects. Defects 1, 2 and 4 above are all this. + +This is the shape the repo already rejected once. #231's commit message reads +*"make panic-freedom a checked property, not a convention."* The fork guard is +the same class of problem and received the opposite treatment. + +### 3.2 The runtime thread is not a value + +`AimDbHandle` currently holds five fields, four of which are facets of one +concept: + +```rust +pub struct AimDbHandle { + thread_handle: Option>, // the thread + shutdown_tx: Option>, // how to stop it + runtime_handle: tokio::runtime::Handle, // how to enter it + db: Arc, + thread_alive: Option>>, // whether it still runs + made_in: crate::fork::Generation, // whether it is ours +} +``` + +"The runtime thread" is spelled out four times. Releasing it therefore means +remembering four `take()` calls in two separate guards — which is exactly how +defect 3 happened: #232 added `thread_alive` and the guards were not updated. + +The current mitigation, [`release_inherited`](../../aimdb-sync/src/handle.rs#L295), +centralises the *release* but leaves the four fields loose. It is a smaller +version of the same problem, not its removal. + +## 4. Proposed shape + +Group the thread into one owned value, and make the check unavoidable by making +it the only route to the runtime. + +```rust +/// One value, one OS thread. Constructed together, released together. +struct Runtime { + thread: JoinHandle<()>, + shutdown: mpsc::Sender, + alive: mpsc::Receiver<()>, + handle: tokio::runtime::Handle, + made_in: fork::Generation, +} + +impl Runtime { + /// The only way to reach the Tokio handle. The fork check lives here + /// because this is the chokepoint every operation must pass through. + fn enter(&self) -> SyncResult<&tokio::runtime::Handle> { + if fork::forked_since(self.made_in) { + return Err(SyncError::ForkedChild); + } + Ok(&self.handle) + } +} + +pub struct AimDbHandle { + rt: Option, // None once detached + db: Arc, +} +``` + +What this buys, concretely against §2: + +- **Defects 1 and 4 become unrepresentable.** A producer or consumer method + cannot publish or read without a `tokio::runtime::Handle`, and the only way to + get one is `enter()`. A tenth method inherits the guard by construction. +- **Defect 3 becomes unrepresentable.** Releasing the thread is `self.rt.take()` + — one field, impossible to half-do. +- **Defect 2 is unaffected** and stays a matter of discipline; it is a + documentation and test-coverage gap, not a structural one. + +Note the fork check must remain *before* the `Weak` upgrade, for the +reason #230 documented: in a forked child the upgrade **succeeds**, because the +`Arc` came across with the address space. That is exactly why a child's `set()` +would otherwise return `Ok` into a buffer nobody drains. `enter()` preserves +this ordering naturally — you reach the runtime before you reach the data. + +## 5. What it does not change + +- The public API's shape. `attach`, `producer`, `consumer`, `set`, `get`, + `detach`, `detach_timeout` keep their signatures. +- `aimdb-core`. Nothing crosses the crate boundary. +- The `pthread_atfork` mechanism, the generation-counter-not-poison-flag + reasoning, and the relaxed-atomic hot path — all of which #230 established and + measured, and none of which this touches. Detection stays as it is; only its + *plumbing* changes. +- `fork::generation` / `fork::forked_since` remain crate-private (#230 narrowed + them precisely so this refactor is free to choose a different shape). + +## 6. Decided: producers and consumers hold `Weak` + +This was the note's open question, and it recommended `Weak` while calling +`Arc` "arguably the better contract" for FFI callers. `Arc` was then tried, and +implementing it settled the question in the opposite direction from that aside. + +`Arc` buys exactly one thing: a producer outliving its handle keeps working +rather than failing with `RuntimeShutdown`. It costs two things that `Weak` +gets for free, because both fall out of ownership: + +- **Liveness.** With `Weak`, the failed upgrade *is* the check. With `Arc` it + has to be rebuilt — a flag the runtime thread flips on every exit path. +- **Waking a blocked reader.** Dropping the handle drops the database, which + closes its buffers, which is what wakes a consumer parked in `get()`. + `aimdb-core` has no explicit close, so with `Arc` nothing else does: a + consumer kept the database alive, `detach` closed nothing, and the reader + parked forever. That took a level-triggered stop channel and a `select` + around every blocking read to fix. + +So the ledger is one behaviour gained against three mechanisms added. And the +behaviour gained is itself questionable: under `Arc` a forgotten producer keeps +an OS thread and a Tokio runtime alive with nobody owning them — the stranded +thread #232 had just removed, reintroduced as a feature. + +| | `Weak` (chosen) | `Arc` (tried, reverted) | +|---|---|---| +| producer outliving its handle | `RuntimeShutdown` | keeps working | +| liveness check | free — the failed upgrade | explicit flag | +| waking a blocked reader | free — buffers close | stop channel + select per read | +| forgotten producer | harmless | strands a thread and a runtime | +| behaviour vs. today | identical | changed | + +`Weak` also keeps the whole change reviewable as "no behaviour changed", which +is worth more than the aside was. + +**One consequence to note.** `SyncConsumer` holds the Tokio handle directly +rather than reaching it through `Runtime`, and its fork check treats a failed +upgrade as "detached" rather than "refuse". Both are deliberate: a `Reader` can +still drain what is already buffered after the runtime is gone, and returning +that data is behaviour the characterization tests pin. Gating reads on a live +`Runtime` broke it. + +## 7. Secondary: the `Drop` / `detach` duality + +`AimDbHandle` has two shutdown paths for one resource: +[`detach_internal`](../../aimdb-sync/src/handle.rs#L450), reached from `detach` +and `detach_timeout`, and [`Drop`](../../aimdb-sync/src/handle.rs#L568). One can +report failure; the other cannot. + +#232 resolved the acute problem by making `Drop` non-blocking — the right +answer, arrived at as a bug fix rather than as a starting principle. With +`Runtime` as a value the residual duplication collapses naturally: `Drop` +becomes "signal and release", `detach` becomes "signal, release, and wait", and +both are expressed against one field instead of four. + +This is worth folding into the same change. It is not worth a separate one. + +## 8. Why this is also a testing problem + +The fork test suite in `aimdb-sync/tests/` had to fork real processes from a +parent holding a live, freshly started Tokio runtime — the least safe possible +moment, because the child inherits an allocator lock held by threads that did +not survive. Measured on the #230 branch, the suite failed **11 of 60 runs** +before mitigation, and required both a `mem::forget` of inherited state and a +settle period before the fork to reach zero. The settle is a duration, not a +handshake; it is a mitigation, not a guarantee. + +That difficulty is downstream of §3.2. Because `AimDbHandle` inseparably *means* +"a spawned OS thread", there is no way to construct the fork condition without +one. With `Runtime` as a value, `made_in` is ordinary data: the refusal paths — +every `SyncProducer` and `SyncConsumer` method, `producer()`, `consumer()`, +`detach()`, `Drop` — become unit-testable against a `Runtime` stamped with a +stale generation. No thread, no fork, no sleep. + +The end-to-end fork tests should not all be deleted; **one** genuine +`fork()`-and-assert case is worth keeping, because a unit test cannot prove the +`pthread_atfork` handler is really installed. But it should be one cheap case +instead of five expensive ones, and the watchdog in +`aimdb-sync/tests/fork_child/mod.rs` should stay regardless. + +## 9. How it was done + +Done after #230, #232 and #233 merged, as a single self-contained change. + +The steps, as executed: + +1. Introduce `Runtime`, move the four fields and the stamp into it. Internal + only; no signature changes. +2. Route every runtime access through `enter()`. Delete the three `check_fork` + implementations and the nine call sites. +3. Point `SyncProducer` / `SyncConsumer` at `Weak` (§6). +4. Collapse `Drop` / `detach_internal` onto the single field (§7). +5. Convert the refusal-path tests from fork tests to unit tests, keeping one + end-to-end fork case (§8). + +Outcome, measured against the merged stack: + +| | before | after | +|---|---|---| +| types carrying a copied fork stamp | 3 | 0 | +| hand-written `check_fork` implementations | 3 | 0 | +| call sites that must remember to guard | 9 | 0 | +| fields on `AimDbHandle` | 6 | 2 | +| forking tests | 5, in 2 binaries | 2, in 1 | +| modules | +`runtime.rs`, −`waiter.rs` | | + +Two forking tests remain rather than one. `dropping_an_inherited_handle_does_not_panic` +covers a destructor joining a thread this process never had, which panics +inside `std` — that needs a genuinely dead thread and no unit test can supply +one. The other proves the `pthread_atfork` handler is really installed. + +## 10. Risks and what this does not fix + +- **It is a refactor of working code.** Every defect in §2 is fixed on the + stack today; the guard is complete as it stands. The argument for this work is + that the *next* one is free to reappear, not that the crate is broken. +- **`enter()` is only a chokepoint if nothing else exposes the handle.** If a + future method returns `&tokio::runtime::Handle` or clones it out, the + guarantee is gone. That constraint needs stating in the type's docs, and it is + the one thing here still enforced by convention. +- **It does not remove the fork hazard from tests entirely** (§8) — one real + fork case remains, and it is the expensive one. +- **It does not address `fork` on non-Unix**, where `generation()` is + permanently `0`. That is correct today and stays correct; it is noted only so + the next reader does not mistake it for an oversight. + +## 11. Settled, and what is left + +1. **`Weak` or `Arc`?** Settled as `Weak` — see §6. The note's own aside + favouring `Arc` did not survive contact with the implementation. +2. **Should `Runtime` be exposed?** It is `pub(crate)`. Keep it private until an + FFI layer exists and can say what it needs — the reasoning that made + `fork::generation` crate-private in #230. +3. **Is the end-to-end fork coverage enough?** Two tests, for the two things a + unit test cannot reach (§9). The watchdog in `tests/fork_child/mod.rs` stays + regardless: it converts a deadlock from a six-hour CI hang into a bounded + failure, and that should not depend on how likely the deadlock is. + +Still open, unchanged by this work: + +- `enter()` is a chokepoint only while nothing else hands out the Tokio handle. + `SyncConsumer` now holds one directly (§6) for a documented reason, which + makes that constraint a live one rather than theoretical. It is the one thing + here still enforced by convention. From 663858b458f286e4049b7f9fd96537ead5c5b762 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 19:20:37 +0000 Subject: [PATCH 02/11] refactor(sync): gate the consumer's Tokio handle behind the check too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The consumer was the one place the fork check was still made by convention. It held a bare `tokio::runtime::Handle` field next to a hand-written `guard()`, called at five sites, which returned `Ok` when the `Weak` upgrade failed. That is the shape this whole change exists to remove — opt-in guards and a resource any method can reach without passing one — rebuilt on a single type. Naming it "the one thing still enforced by convention" documented the gap rather than closing it. The consumer genuinely needs a handle that outlives the runtime: a `Reader` can still drain what is already buffered after a `detach`, and delivering that data is behaviour the characterization tests pin. So the handle stays — but it moves into `RuntimeRef` in `runtime.rs`, where both fields are private to that module. `consumer.rs` can no longer obtain a handle except through `RuntimeRef::enter`, which checks first, so the blocking reads are checked by construction exactly as the publish path is. `try_get` still calls the check explicitly, and that is not a gap being glossed: it touches no runtime resource at all, so there is nothing to gate. `get_latest` and `get_latest_with_timeout` look like the same case but route through `get`/`get_with_timeout`, which are gated. One explicit check, not five. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01DKAH7JNjPLPmG4mPTvWfth --- aimdb-sync/CHANGELOG.md | 5 +- aimdb-sync/src/consumer.rs | 62 +++++------------------ aimdb-sync/src/handle.rs | 5 +- aimdb-sync/src/runtime.rs | 48 +++++++++++++++++- docs/design/050-sync-runtime-ownership.md | 33 ++++++++---- 5 files changed, 91 insertions(+), 62 deletions(-) diff --git a/aimdb-sync/CHANGELOG.md b/aimdb-sync/CHANGELOG.md index 27f2f163..243df1ce 100644 --- a/aimdb-sync/CHANGELOG.md +++ b/aimdb-sync/CHANGELOG.md @@ -57,7 +57,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 to. `AimDbHandle` drops from six loose fields to two, so releasing state inherited across a `fork` can no longer be half-done. No API signature and no behaviour changed: producers and consumers still hold a weak reference, so a - handle's lifetime still governs. `waiter.rs` is retired — `enter()` returns + handle's lifetime still governs. A consumer needs a Tokio handle that outlives + the runtime — buffered data stays readable after `detach` — so it holds a + `RuntimeRef` whose handle is private to the runtime module and reachable only + through a checked accessor, rather than a bare field beside a guard. `waiter.rs` is retired — `enter()` returns the handle it existed to wrap. See design 050. ### Added diff --git a/aimdb-sync/src/consumer.rs b/aimdb-sync/src/consumer.rs index 8858ea1e..f241bbcb 100644 --- a/aimdb-sync/src/consumer.rs +++ b/aimdb-sync/src/consumer.rs @@ -2,9 +2,8 @@ use aimdb_core::{DbError, Reader}; -use crate::runtime::Runtime; +use crate::runtime::RuntimeRef; use crate::{SyncError, SyncResult}; -use alloc::sync::Weak; use core::fmt::Debug; use core::time::Duration; use std::time::Instant; @@ -57,22 +56,13 @@ pub struct SyncConsumer where T: Send + Debug + Clone, { - /// The runtime this consumer belongs to, for the fork check only. - /// - /// `Weak`, and deliberately not required to be alive: a `Reader` can still - /// drain what is already buffered after the runtime is gone, and - /// [`try_get`](Self::try_get) returning that data is documented behaviour. - /// A failed upgrade therefore means "detached", not "refuse" — see - /// [`Self::guard`]. - rt: Weak, - - /// A way into the Tokio runtime that owns nothing. + /// The runtime this consumer reads from, reachable only through a check. /// - /// Held directly rather than reached through [`Runtime`] because a blocked - /// or draining read must not keep the database alive: that is what stops - /// `detach` from closing the buffers, and a reader parked in - /// [`get`](Self::get) would then never wake. - handle: tokio::runtime::Handle, + /// See [`RuntimeRef`](crate::runtime::RuntimeRef): the Tokio handle inside + /// it is private to that module, so no read here can reach one without + /// passing the fork check first. + rt: RuntimeRef, + reader: Reader, } @@ -81,28 +71,8 @@ where T: Send + Debug + Clone, { /// Create a new sync consumer (internal use only) - pub(crate) fn new( - rt: Weak, - handle: tokio::runtime::Handle, - reader: Reader, - ) -> Self { - Self { rt, handle, reader } - } - - /// Refuse a read this process has no runtime thread for. - /// - /// The fork check is only meaningful while the runtime is alive — and in a - /// forked child it *is* alive, because the `Arc` came across with the - /// address space, which is exactly why the check exists. A failed upgrade - /// means the handle was dropped in this process, which is not a fork: let - /// the read proceed and the buffer report for itself, so data already - /// queued is still delivered. - #[inline] - fn guard(&self) -> SyncResult<()> { - match self.rt.upgrade() { - Some(rt) => rt.check(), - None => Ok(()), - } + pub(crate) fn new(rt: RuntimeRef, reader: Reader) -> Self { + Self { rt, reader } } async fn get_impl(reader: &mut Reader) -> SyncResult { @@ -150,10 +120,7 @@ where /// # } /// ``` pub fn get(&mut self) -> SyncResult { - self.guard()?; - self.handle - .clone() - .block_on(Self::get_impl(&mut self.reader)) + self.rt.enter()?.block_on(Self::get_impl(&mut self.reader)) } /// Get a value with a timeout. @@ -196,8 +163,7 @@ where /// # } /// ``` pub fn get_with_timeout(&mut self, timeout: Duration) -> SyncResult { - self.guard()?; - let handle = self.handle.clone(); + let handle = self.rt.enter()?; let fut = async { tokio::time::timeout(timeout, Self::get_impl(&mut self.reader)).await }; let res = handle.block_on(fut); res.unwrap_or_else(|_| Err(SyncError::GetTimeout)) @@ -239,7 +205,7 @@ where /// # } /// ``` pub fn try_get(&mut self) -> SyncResult { - self.guard()?; + self.rt.check()?; let res = self.reader.try_recv(); res.map_err(|e| match e { DbError::BufferClosed { .. } => SyncError::RuntimeShutdown, @@ -290,7 +256,7 @@ where /// # } /// ``` pub fn get_latest(&mut self) -> SyncResult { - self.guard()?; + self.rt.check()?; // 1) can simply sequence get_catch_up and try_get - // no one else does it simultaneously thanks to &mut self // 2) if draining ends up with an error, we follow the previous impl @@ -342,7 +308,7 @@ where /// # } /// ``` pub fn get_latest_with_timeout(&mut self, timeout: Duration) -> SyncResult { - self.guard()?; + self.rt.check()?; // see internal comments for get_latest let deadline = Instant::now() + timeout; let oldest = self.get_catch_up(Some(deadline))?; diff --git a/aimdb-sync/src/handle.rs b/aimdb-sync/src/handle.rs index f081986d..8cedd902 100644 --- a/aimdb-sync/src/handle.rs +++ b/aimdb-sync/src/handle.rs @@ -1,6 +1,6 @@ //! AimDB handle for managing the sync API runtime thread. -use crate::runtime::{Runtime, ShutdownSignal}; +use crate::runtime::{Runtime, RuntimeRef, ShutdownSignal}; use crate::{SyncError, SyncResult}; use aimdb_core::{log_error, log_warn, AimDb, AimDbBuilder, DbError}; use alloc::sync::Arc; @@ -381,8 +381,7 @@ impl AimDbHandle { .subscribe::(&record_key) .map_err(SyncError::Db)?; Ok(crate::SyncConsumer::new( - Arc::downgrade(&self.rt), - self.rt.enter()?.clone(), + RuntimeRef::new(Arc::downgrade(&self.rt), self.rt.enter()?.clone()), reader, )) } diff --git a/aimdb-sync/src/runtime.rs b/aimdb-sync/src/runtime.rs index 2eae118f..1cc52731 100644 --- a/aimdb-sync/src/runtime.rs +++ b/aimdb-sync/src/runtime.rs @@ -47,7 +47,7 @@ //! producer keep an OS thread and a Tokio runtime alive with nobody owning //! them, which is the stranded thread #232 had just removed. -use alloc::sync::Arc; +use alloc::sync::{Arc, Weak}; use aimdb_core::AimDb; @@ -128,6 +128,52 @@ impl Runtime { } } +/// A borrowed view of a [`Runtime`] that outlives it on purpose. +/// +/// Held by [`SyncConsumer`](crate::SyncConsumer), which has a requirement the +/// handle and the producers do not: a `Reader` can still drain what is already +/// buffered after the runtime is gone, and delivering that data is behaviour +/// the characterization tests pin. So a dead runtime must not refuse a read — +/// but a `fork` still must. +/// +/// The Tokio handle is kept for that case and is **private to this module**, +/// which is the whole point of the type. `consumer.rs` cannot reach it except +/// through [`Self::enter`], so a read that skips the check cannot be written +/// there any more than it can anywhere else. A bare handle field beside a +/// hand-written guard — the shape this replaces — offered no such thing. +pub(crate) struct RuntimeRef { + rt: Weak, + handle: tokio::runtime::Handle, +} + +impl RuntimeRef { + pub(crate) fn new(rt: Weak, handle: tokio::runtime::Handle) -> Self { + Self { rt, handle } + } + + /// Refuse a forked child; let a detached one through. + /// + /// A failed upgrade means the handle was dropped in *this* process, which + /// is not a fork — the buffer is the right thing to answer for itself. + /// After a real `fork` the upgrade **succeeds**, because the `Arc` came + /// across with the address space, which is exactly why the check is worth + /// making at all. + #[inline] + pub(crate) fn check(&self) -> SyncResult<()> { + match self.rt.upgrade() { + Some(rt) => rt.check(), + None => Ok(()), + } + } + + /// The Tokio handle, checked. The only way to obtain one. + #[inline] + pub(crate) fn enter(&self) -> SyncResult { + self.check()?; + Ok(self.handle.clone()) + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/docs/design/050-sync-runtime-ownership.md b/docs/design/050-sync-runtime-ownership.md index 73c21b92..b3168f14 100644 --- a/docs/design/050-sync-runtime-ownership.md +++ b/docs/design/050-sync-runtime-ownership.md @@ -183,12 +183,20 @@ thread #232 had just removed, reintroduced as a feature. `Weak` also keeps the whole change reviewable as "no behaviour changed", which is worth more than the aside was. -**One consequence to note.** `SyncConsumer` holds the Tokio handle directly -rather than reaching it through `Runtime`, and its fork check treats a failed -upgrade as "detached" rather than "refuse". Both are deliberate: a `Reader` can -still drain what is already buffered after the runtime is gone, and returning -that data is behaviour the characterization tests pin. Gating reads on a live -`Runtime` broke it. +**One consequence, and how it is contained.** `SyncConsumer` needs a Tokio +handle that outlives the runtime: a `Reader` can still drain what is already +buffered after a `detach`, and delivering that data is behaviour the +characterization tests pin — gating reads on a live `Runtime` broke it. + +The first attempt stored a bare `handle` field beside a hand-written `guard()` +that returned `Ok` when the upgrade failed. That is the convention shape this +design exists to remove, rebuilt on one type: five opt-in call sites and a +field any of them could use without checking anything. + +It is now a `RuntimeRef` in `runtime.rs`, holding the `Weak` and the handle with +**both fields private to that module**. `consumer.rs` cannot obtain a handle +except through `RuntimeRef::enter`, which checks first. So the blocking reads +are checked by construction, exactly as the publish path is. ## 7. Secondary: the `Drop` / `detach` duality @@ -289,6 +297,13 @@ one. The other proves the `pthread_atfork` handler is really installed. Still open, unchanged by this work: - `enter()` is a chokepoint only while nothing else hands out the Tokio handle. - `SyncConsumer` now holds one directly (§6) for a documented reason, which - makes that constraint a live one rather than theoretical. It is the one thing - here still enforced by convention. + Nothing does: `Runtime` and `RuntimeRef` keep theirs private to + `runtime.rs`, so the constraint is enforced by module privacy rather than by + memory. Adding a `pub(crate)` accessor that returns one would silently undo + that, which is worth stating because it is the only way back to the old + shape. +- `SyncConsumer::try_get` reads without any runtime resource, so its fork check + is a call it makes rather than one it cannot avoid — the reader is plain data + and there is nothing to gate. `get_latest` and `get_latest_with_timeout` look + like the same case but are not: both route through `get`/`get_with_timeout`, + which are gated. One explicit check remains, not five. From a88dc59085dc6cac9997468dfb72030d78dab1bd Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 19:28:02 +0000 Subject: [PATCH 03/11] refactor(sync): gate the consumer's reader, removing the last opt-in check MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `try_get` was the one remaining place the fork check was a call rather than a consequence. The justification given for that — it touches no runtime resource, so there is nothing to gate — was wrong. It touches the `Reader`, and the `Reader` is exactly the thing to gate: something with no resource behind a checked accessor is something whose check is easy to leave out, which is the defect this design exists to remove. The reader now lives in `Guarded>`, whose value is private to `runtime.rs`. `consumer.rs` reaches it through `get()` (checked) or `enter()` (checked, and yields a Tokio handle for the reads that block). `SyncConsumer` is down to a single field and no longer names the runtime at all, so there is no longer a check to remember or forget: every read passes one because it cannot reach the buffer otherwise. `get_latest` and `get_latest_with_timeout` lose their explicit checks with nothing lost — both route through `get`/`get_with_timeout`/`try_get`, all of which are now gated at the point of access. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01DKAH7JNjPLPmG4mPTvWfth --- aimdb-sync/src/consumer.rs | 31 +++++++++--------- aimdb-sync/src/runtime.rs | 38 +++++++++++++++++++++++ docs/design/050-sync-runtime-ownership.md | 11 ++++--- 3 files changed, 59 insertions(+), 21 deletions(-) diff --git a/aimdb-sync/src/consumer.rs b/aimdb-sync/src/consumer.rs index f241bbcb..d2ee8fa8 100644 --- a/aimdb-sync/src/consumer.rs +++ b/aimdb-sync/src/consumer.rs @@ -2,7 +2,7 @@ use aimdb_core::{DbError, Reader}; -use crate::runtime::RuntimeRef; +use crate::runtime::{Guarded, RuntimeRef}; use crate::{SyncError, SyncResult}; use core::fmt::Debug; use core::time::Duration; @@ -56,14 +56,13 @@ pub struct SyncConsumer where T: Send + Debug + Clone, { - /// The runtime this consumer reads from, reachable only through a check. + /// The subscription, and with it the runtime. /// - /// See [`RuntimeRef`](crate::runtime::RuntimeRef): the Tokio handle inside - /// it is private to that module, so no read here can reach one without - /// passing the fork check first. - rt: RuntimeRef, - - reader: Reader, + /// Wrapped so that neither the reader nor a Tokio handle can be reached + /// without passing the fork check — see + /// [`Guarded`](crate::runtime::Guarded). Every read below therefore checks + /// because it must, not because someone remembered to. + reader: Guarded>, } impl SyncConsumer @@ -72,7 +71,9 @@ where { /// Create a new sync consumer (internal use only) pub(crate) fn new(rt: RuntimeRef, reader: Reader) -> Self { - Self { rt, reader } + Self { + reader: Guarded::new(rt, reader), + } } async fn get_impl(reader: &mut Reader) -> SyncResult { @@ -120,7 +121,8 @@ where /// # } /// ``` pub fn get(&mut self) -> SyncResult { - self.rt.enter()?.block_on(Self::get_impl(&mut self.reader)) + let (handle, reader) = self.reader.enter()?; + handle.block_on(Self::get_impl(reader)) } /// Get a value with a timeout. @@ -163,8 +165,8 @@ where /// # } /// ``` pub fn get_with_timeout(&mut self, timeout: Duration) -> SyncResult { - let handle = self.rt.enter()?; - let fut = async { tokio::time::timeout(timeout, Self::get_impl(&mut self.reader)).await }; + let (handle, reader) = self.reader.enter()?; + let fut = async { tokio::time::timeout(timeout, Self::get_impl(reader)).await }; let res = handle.block_on(fut); res.unwrap_or_else(|_| Err(SyncError::GetTimeout)) } @@ -205,8 +207,7 @@ where /// # } /// ``` pub fn try_get(&mut self) -> SyncResult { - self.rt.check()?; - let res = self.reader.try_recv(); + let res = self.reader.get()?.try_recv(); res.map_err(|e| match e { DbError::BufferClosed { .. } => SyncError::RuntimeShutdown, DbError::BufferEmpty => SyncError::GetTimeout, @@ -256,7 +257,6 @@ where /// # } /// ``` pub fn get_latest(&mut self) -> SyncResult { - self.rt.check()?; // 1) can simply sequence get_catch_up and try_get - // no one else does it simultaneously thanks to &mut self // 2) if draining ends up with an error, we follow the previous impl @@ -308,7 +308,6 @@ where /// # } /// ``` pub fn get_latest_with_timeout(&mut self, timeout: Duration) -> SyncResult { - self.rt.check()?; // see internal comments for get_latest let deadline = Instant::now() + timeout; let oldest = self.get_catch_up(Some(deadline))?; diff --git a/aimdb-sync/src/runtime.rs b/aimdb-sync/src/runtime.rs index 1cc52731..1e7c8dec 100644 --- a/aimdb-sync/src/runtime.rs +++ b/aimdb-sync/src/runtime.rs @@ -174,6 +174,44 @@ impl RuntimeRef { } } +/// A resource that cannot be touched without passing the fork check. +/// +/// The point is the field privacy, not the wrapper: `inner` is private to this +/// module, so a caller in `consumer.rs` has no way to reach the value except +/// through [`Self::get`] or [`Self::enter`], both of which check first. A plain +/// field beside a hand-written guard offers nothing — the guard is a call +/// someone can forget to make, and forgetting it is the defect this whole +/// design removes. +/// +/// Used for a `Reader`, which is the one thing a consumer touches that needs no +/// runtime: `try_get` reads straight out of the buffer. That is exactly why it +/// needs wrapping. Something with no resource to gate is something whose check +/// is easy to leave out. +pub(crate) struct Guarded { + rt: RuntimeRef, + inner: T, +} + +impl Guarded { + pub(crate) fn new(rt: RuntimeRef, inner: T) -> Self { + Self { rt, inner } + } + + /// The value, checked. For work that needs no runtime. + #[inline] + pub(crate) fn get(&mut self) -> SyncResult<&mut T> { + self.rt.check()?; + Ok(&mut self.inner) + } + + /// The value and a way to block, both checked. For work that waits. + #[inline] + pub(crate) fn enter(&mut self) -> SyncResult<(tokio::runtime::Handle, &mut T)> { + let handle = self.rt.enter()?; + Ok((handle, &mut self.inner)) + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/docs/design/050-sync-runtime-ownership.md b/docs/design/050-sync-runtime-ownership.md index b3168f14..aebc390b 100644 --- a/docs/design/050-sync-runtime-ownership.md +++ b/docs/design/050-sync-runtime-ownership.md @@ -302,8 +302,9 @@ Still open, unchanged by this work: memory. Adding a `pub(crate)` accessor that returns one would silently undo that, which is worth stating because it is the only way back to the old shape. -- `SyncConsumer::try_get` reads without any runtime resource, so its fork check - is a call it makes rather than one it cannot avoid — the reader is plain data - and there is nothing to gate. `get_latest` and `get_latest_with_timeout` look - like the same case but are not: both route through `get`/`get_with_timeout`, - which are gated. One explicit check remains, not five. +- Nothing. `SyncConsumer::try_get` was briefly the last opt-in check, on the + argument that it touches no runtime resource so there was nothing to gate. + That was wrong: it touches the `Reader`, which is precisely the thing to gate. + The reader now lives in a `Guarded>` whose value is private to + `runtime.rs`, so every read reaches it through a checked accessor. The + consumer has one field and no way to skip the check. From 873db60f9142a3e46c80d29809218496d7dafdf4 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 19:32:19 +0000 Subject: [PATCH 04/11] refactor(sync): delete the unchecked database accessor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `db_unchecked` existed for one caller — `AimDbHandle::consumer`, which checked and then subscribed — and was documented as "not a hole in the guarantee" because its single use sat next to a check. That is the argument every such accessor comes with, and it only holds until someone adds a second caller. `consumer()` now subscribes through `db()`, which checks, so the accessor has no reason to exist and is gone. No unchecked accessors remain in the crate. Also records what the producer factory's check now is. It is diagnostics, not the guarantee: a producer made in a forked child refuses on first use anyway, because it holds a `Weak` to the runtime the parent stamped and there is no fresh stamp to make it look current. That was not true when each producer copied a stamp at construction — a producer built in the child then took the child's generation and never refused, which is why the old code had to block that call. The refactor closed the bypass; the check is kept because failing at `producer()` beats failing at the first `set()` across an FFI boundary, and it now says so rather than implying it is load-bearing. Three explicit checks remain, all in handle.rs and none guarding a resource: the producer factory above, and the detach and Drop branches that decide to release a thread rather than join one this process never had. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01DKAH7JNjPLPmG4mPTvWfth --- aimdb-sync/src/handle.rs | 12 ++++++++++-- aimdb-sync/src/runtime.rs | 12 ------------ docs/design/050-sync-runtime-ownership.md | 15 ++++++++++++++- 3 files changed, 24 insertions(+), 15 deletions(-) diff --git a/aimdb-sync/src/handle.rs b/aimdb-sync/src/handle.rs index 8cedd902..f6f91008 100644 --- a/aimdb-sync/src/handle.rs +++ b/aimdb-sync/src/handle.rs @@ -336,6 +336,14 @@ impl AimDbHandle { where T: Send + 'static + Debug + Clone, { + // Diagnostics, not the guarantee. A producer made in a forked child + // would refuse on first use anyway: it holds a `Weak` to the + // runtime the *parent* stamped, so there is no fresh stamp to make it + // look current. That was not true when each producer copied a stamp at + // construction — then a producer built in the child got the child's + // generation and never refused, which is why the old code had to block + // this call. Kept because failing here beats failing at the first + // `set()`, especially across an FFI boundary. self.rt.check()?; Ok(crate::SyncProducer::new(Arc::downgrade(&self.rt), key)) } @@ -373,11 +381,11 @@ impl AimDbHandle { where T: Send + Sync + 'static + Debug + Clone, { - self.rt.check()?; let record_key = key.as_ref().to_string(); + // `db()` checks, so subscribing is gated the same way publishing is. let reader = self .rt - .db_unchecked() + .db()? .subscribe::(&record_key) .map_err(SyncError::Db)?; Ok(crate::SyncConsumer::new( diff --git a/aimdb-sync/src/runtime.rs b/aimdb-sync/src/runtime.rs index 1e7c8dec..f62d4739 100644 --- a/aimdb-sync/src/runtime.rs +++ b/aimdb-sync/src/runtime.rs @@ -114,18 +114,6 @@ impl Runtime { self.check()?; Ok(&self.db) } - - /// The database without the fork check, for the one caller that has already - /// made it: [`AimDbHandle::consumer`](crate::AimDbHandle::consumer) checks, - /// then subscribes. - /// - /// Not a hole in the guarantee — it is `pub(crate)` and its one use sits - /// directly after a [`Self::check`] — but it is the reason [`Self::db`] - /// exists as the ordinary path. - #[inline] - pub(crate) fn db_unchecked(&self) -> &Arc { - &self.db - } } /// A borrowed view of a [`Runtime`] that outlives it on purpose. diff --git a/docs/design/050-sync-runtime-ownership.md b/docs/design/050-sync-runtime-ownership.md index aebc390b..039a94c2 100644 --- a/docs/design/050-sync-runtime-ownership.md +++ b/docs/design/050-sync-runtime-ownership.md @@ -302,7 +302,20 @@ Still open, unchanged by this work: memory. Adding a `pub(crate)` accessor that returns one would silently undo that, which is worth stating because it is the only way back to the old shape. -- Nothing. `SyncConsumer::try_get` was briefly the last opt-in check, on the +- Three explicit `check()` calls remain, all in `handle.rs`, and none is a + guard on a resource: + + | site | what it is | + |---|---| + | `producer()` | diagnostics. A producer made in a forked child refuses on first use regardless — it holds a `Weak` to the runtime the *parent* stamped. Under the old per-object stamp it would have got the child's generation and never refused, which is why that call had to be blocked then. Failing here still beats failing at the first `set()`. | + | `detach_internal` | a branch, not a refusal: release the thread rather than join one this process never had. | + | `Drop` | the same branch. | + + `consumer()` no longer checks explicitly — it subscribes through `db()`, which + checks — and `db_unchecked`, added as a deliberate hole for it, is deleted. + There are now no unchecked accessors in the crate. + +- `SyncConsumer::try_get` was briefly the last opt-in check, on the argument that it touches no runtime resource so there was nothing to gate. That was wrong: it touches the `Reader`, which is precisely the thing to gate. The reader now lives in a `Guarded>` whose value is private to From 0fb05a5267a996022e2c8b98dfcdd9171a130305 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 19:37:21 +0000 Subject: [PATCH 05/11] test(sync): pin the producer factory's refusal in a forked child MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Asked why the producer factory still checks explicitly, the honest answer turned out to be the opposite of what the previous commit claimed. That commit called it "diagnostics, not the guarantee" — reasoning that a producer built in a child refuses on first use anyway, which is true but not the point. Constructing a producer touches nothing gated: it only downgrades an `Arc`. So without that line the call *succeeds* in a forked child, while `consumer()` right below it refuses, because subscribing goes through `db()`. A handle that hands out producers but not consumers is a worse contract than one that hands out neither, and the difference stays invisible until the first publish. The check is what makes the two factories agree, which is load-bearing enough. It was also untested — the test covering it went when the fork suite was cut from five tests to two. Rather than restore a binary, the two assertions join `dropping_an_inherited_handle_does_not_panic`, which already has handles in the child: no new fork, no new binary. Verified by removing the line and watching that test fail, then restoring it. The comment now says which of the two things it does: it is not what makes a child safe, it is what stops the two factories disagreeing. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01DKAH7JNjPLPmG4mPTvWfth --- aimdb-sync/src/handle.rs | 26 ++++++++++++++++------- aimdb-sync/tests/fork_safety_test.rs | 14 +++++++++++- docs/design/050-sync-runtime-ownership.md | 2 +- 3 files changed, 32 insertions(+), 10 deletions(-) diff --git a/aimdb-sync/src/handle.rs b/aimdb-sync/src/handle.rs index f6f91008..2c7ba096 100644 --- a/aimdb-sync/src/handle.rs +++ b/aimdb-sync/src/handle.rs @@ -336,14 +336,24 @@ impl AimDbHandle { where T: Send + 'static + Debug + Clone, { - // Diagnostics, not the guarantee. A producer made in a forked child - // would refuse on first use anyway: it holds a `Weak` to the - // runtime the *parent* stamped, so there is no fresh stamp to make it - // look current. That was not true when each producer copied a stamp at - // construction — then a producer built in the child got the child's - // generation and never refused, which is why the old code had to block - // this call. Kept because failing here beats failing at the first - // `set()`, especially across an FFI boundary. + // The one check in this crate that guards no resource, and it earns + // its place: constructing a producer touches nothing gated — it only + // downgrades an `Arc` — so without this line the call would succeed in + // a forked child. + // + // That asymmetry is the reason to keep it. `consumer()` below subscribes + // through `db()`, so it refuses in a child for free. A handle that + // hands out producers but not consumers would be a worse contract than + // one that hands out neither, and the difference would be invisible + // until first publish. + // + // It is no longer what makes a child *safe*, though. A producer built + // in a child holds a `Weak` to the runtime the parent stamped, + // so it refuses on first use regardless. Under the old per-object stamp + // it took the child's generation and never refused — that was a real + // bypass, and blocking this call was the only fix. Pinned by + // `dropping_an_inherited_handle_does_not_panic`, which fails if this + // line is removed. self.rt.check()?; Ok(crate::SyncProducer::new(Arc::downgrade(&self.rt), key)) } diff --git a/aimdb-sync/tests/fork_safety_test.rs b/aimdb-sync/tests/fork_safety_test.rs index 8dad8ef7..600ddf16 100644 --- a/aimdb-sync/tests/fork_safety_test.rs +++ b/aimdb-sync/tests/fork_safety_test.rs @@ -88,13 +88,25 @@ fn dropping_an_inherited_handle_does_not_panic() { // `detach` reports the situation rather than joining. let refused = matches!(to_detach.detach(), Err(SyncError::ForkedChild)); + // A handle in a child hands out nothing usable either. `consumer` gets + // this from `db()`, which checks; `producer` touches no gated resource, + // so its check is an explicit one — this is what pins it. + let no_producer = matches!( + to_drop.producer::("sensor.reading"), + Err(SyncError::ForkedChild) + ); + let no_consumer = matches!( + to_drop.consumer::("sensor.reading"), + Err(SyncError::ForkedChild) + ); + // This one is never detached: it is dropped when the closure returns, // which is the destructor path. It must return quietly rather than // join. A panic here would unwind out of the child instead of exiting // normally, and the parent's `WIFEXITED` assertion would catch it. drop(to_drop); - refused + refused && no_producer && no_consumer }); assert_eq!(code, 0, "detach in a child should be refused, not fatal"); } diff --git a/docs/design/050-sync-runtime-ownership.md b/docs/design/050-sync-runtime-ownership.md index 039a94c2..c153e2ce 100644 --- a/docs/design/050-sync-runtime-ownership.md +++ b/docs/design/050-sync-runtime-ownership.md @@ -307,7 +307,7 @@ Still open, unchanged by this work: | site | what it is | |---|---| - | `producer()` | diagnostics. A producer made in a forked child refuses on first use regardless — it holds a `Weak` to the runtime the *parent* stamped. Under the old per-object stamp it would have got the child's generation and never refused, which is why that call had to be blocked then. Failing here still beats failing at the first `set()`. | + | `producer()` | load-bearing, and the only check guarding no resource. Constructing a producer touches nothing gated — it just downgrades an `Arc` — so without it the call *succeeds* in a forked child, while `consumer()` refuses (it subscribes through `db()`). A handle that hands out producers but not consumers is a worse contract than one that hands out neither. It is not what makes a child safe — a producer built there refuses on first use regardless — but it is what makes the two factories agree. Removing it fails `dropping_an_inherited_handle_does_not_panic`. | | `detach_internal` | a branch, not a refusal: release the thread rather than join one this process never had. | | `Drop` | the same branch. | From 3e78754c293be4fa3cfcdc9abff697051969aae6 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 19:41:39 +0000 Subject: [PATCH 06/11] docs(sync): say why the producer is not a Guarded, as the consumer is MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The shapes differ and the asymmetry deserves a reason in the code rather than leaving the next reader to wonder whether the producer was simply missed. Guarded exists to put a check in front of a resource nothing else gates. A consumer owns one: its Reader is its own, and try_get reads straight out of it without touching the runtime. A producer owns a key and a PhantomData. The only resource it reaches is the database, which already sits behind Runtime::db — private to that module, checked on the way through. Wrapping the key would mean checking to reach a String and then checking again to reach the database: an extra hop for no extra guarantee. They also want opposite answers when the upgrade fails. A consumer must carry on, because its buffer may still hold data — that is why RuntimeRef::check returns Ok there, and why a detached consumer still delivers what was already queued. A producer must refuse: there is nothing left to publish into. One wrapper serving both would have to parameterise that policy, which costs more than the hop it saves. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01DKAH7JNjPLPmG4mPTvWfth --- aimdb-sync/src/producer.rs | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/aimdb-sync/src/producer.rs b/aimdb-sync/src/producer.rs index a98a9872..db96d928 100644 --- a/aimdb-sync/src/producer.rs +++ b/aimdb-sync/src/producer.rs @@ -47,6 +47,24 @@ where /// database goes with it and this upgrade fails. That failure *is* the /// liveness check — see [`crate::runtime`] on why sharing ownership here /// would cost more than it bought. + /// + /// # Why this is not a `Guarded`, as the consumer's reader is + /// + /// [`Guarded`](crate::runtime::Guarded) exists to put a check in front of a + /// resource nothing else gates. A consumer owns one: its `Reader` is its + /// own, and `try_get` reads straight out of it without touching the + /// runtime. A producer owns no such thing — a key and a `PhantomData`. The + /// only resource it reaches is the database, and that already lives behind + /// [`Runtime::db`](crate::runtime::Runtime::db), private to that module and + /// checked on the way through. Wrapping the key would mean checking to + /// reach a `String` and then checking again to reach the database: an extra + /// hop, no extra guarantee. + /// + /// The two also want opposite answers when the upgrade fails. A consumer + /// must carry on — its buffer may still hold data, which is why + /// `RuntimeRef::check` returns `Ok` there. A producer must refuse: there is + /// nothing left to publish into. Sharing one wrapper would mean + /// parameterising that policy, which costs more than the hop it saves. rt: Weak, key: String, // same reasons as for Producer in aimdb-core/src/typed_api.rs From 5faea0dc3d82f11e44017815c5dba82d3dd8a351 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 19:46:12 +0000 Subject: [PATCH 07/11] refactor(sync): drop the producer factory's check, which guarded nothing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Asked why `producer()` must call `check()` when there is nothing there to guard, the honest answer is that it must not. The question exposed a contradiction I had been defending rather than fixing: a check whose stated justification was that it protects no resource is a check with no job. Creating a producer touches nothing — not the database, not the runtime thread. That is deliberate and pinned: `test_error_propagation` asserts an unregistered key still yields a producer, and that `set()` reports the problem. A forked child is one more thing `set()` reports, through the `db()` it has to pass. So the check was making `fork` the single exception to this crate's own lazy-producer contract. It also left a category behind — "explicit checks that guard nothing" — and a category is an invitation. Every check now guards a resource: `db()` and `enter()` gate the database and the Tokio handle, `Guarded` gates the consumer's reader, and the two in `detach`/`Drop` gate the `JoinHandle`, where joining is the action and the check is a branch on state. The fork test is updated to assert what is actually true and actually matters: a child's `producer()` succeeds, and its first `set()` refuses. `consumer()` still refuses outright, because subscribing goes through `db()` — an asymmetry that predates this work, since an unregistered key already fails there and not at `producer()`. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01DKAH7JNjPLPmG4mPTvWfth --- aimdb-sync/src/handle.rs | 27 +++++++------------- aimdb-sync/tests/fork_safety_test.rs | 18 ++++++++------ docs/design/050-sync-runtime-ownership.md | 30 ++++++++++++++--------- 3 files changed, 37 insertions(+), 38 deletions(-) diff --git a/aimdb-sync/src/handle.rs b/aimdb-sync/src/handle.rs index 2c7ba096..6218bf09 100644 --- a/aimdb-sync/src/handle.rs +++ b/aimdb-sync/src/handle.rs @@ -336,25 +336,16 @@ impl AimDbHandle { where T: Send + 'static + Debug + Clone, { - // The one check in this crate that guards no resource, and it earns - // its place: constructing a producer touches nothing gated — it only - // downgrades an `Arc` — so without this line the call would succeed in - // a forked child. + // No check here, and deliberately none. Creating a producer touches + // nothing: not the database, not the runtime thread. `test_error_propagation` + // pins that — an unregistered key yields a producer, and `set()` reports + // the problem. A forked child is one more problem `set()` reports, via + // the `db()` it must pass through; making `fork` the single exception to + // this crate's own lazy-producer contract would be the odd thing. // - // That asymmetry is the reason to keep it. `consumer()` below subscribes - // through `db()`, so it refuses in a child for free. A handle that - // hands out producers but not consumers would be a worse contract than - // one that hands out neither, and the difference would be invisible - // until first publish. - // - // It is no longer what makes a child *safe*, though. A producer built - // in a child holds a `Weak` to the runtime the parent stamped, - // so it refuses on first use regardless. Under the old per-object stamp - // it took the child's generation and never refused — that was a real - // bypass, and blocking this call was the only fix. Pinned by - // `dropping_an_inherited_handle_does_not_panic`, which fails if this - // line is removed. - self.rt.check()?; + // `consumer()` below does refuse in a child, because subscribing needs + // the database and `db()` checks. That asymmetry is not new: an + // unregistered key already fails there and not here. Ok(crate::SyncProducer::new(Arc::downgrade(&self.rt), key)) } diff --git a/aimdb-sync/tests/fork_safety_test.rs b/aimdb-sync/tests/fork_safety_test.rs index 600ddf16..ad8ec74d 100644 --- a/aimdb-sync/tests/fork_safety_test.rs +++ b/aimdb-sync/tests/fork_safety_test.rs @@ -88,13 +88,15 @@ fn dropping_an_inherited_handle_does_not_panic() { // `detach` reports the situation rather than joining. let refused = matches!(to_detach.detach(), Err(SyncError::ForkedChild)); - // A handle in a child hands out nothing usable either. `consumer` gets - // this from `db()`, which checks; `producer` touches no gated resource, - // so its check is an explicit one — this is what pins it. - let no_producer = matches!( - to_drop.producer::("sensor.reading"), - Err(SyncError::ForkedChild) - ); + // A handle in a child hands out nothing *usable*. `consumer` refuses + // outright, because subscribing goes through the fork-checked `db()`. + // `producer` succeeds — it touches nothing, exactly as it does for an + // unregistered key — and the refusal lands on first use instead. Both + // are safe; what matters is that neither silently accepts a value. + let producer_defers = match to_drop.producer::("sensor.reading") { + Ok(p) => matches!(p.set(Reading { value: 9 }), Err(SyncError::ForkedChild)), + Err(_) => false, + }; let no_consumer = matches!( to_drop.consumer::("sensor.reading"), Err(SyncError::ForkedChild) @@ -106,7 +108,7 @@ fn dropping_an_inherited_handle_does_not_panic() { // normally, and the parent's `WIFEXITED` assertion would catch it. drop(to_drop); - refused && no_producer && no_consumer + refused && producer_defers && no_consumer }); assert_eq!(code, 0, "detach in a child should be refused, not fatal"); } diff --git a/docs/design/050-sync-runtime-ownership.md b/docs/design/050-sync-runtime-ownership.md index c153e2ce..1da41d67 100644 --- a/docs/design/050-sync-runtime-ownership.md +++ b/docs/design/050-sync-runtime-ownership.md @@ -302,18 +302,24 @@ Still open, unchanged by this work: memory. Adding a `pub(crate)` accessor that returns one would silently undo that, which is worth stating because it is the only way back to the old shape. -- Three explicit `check()` calls remain, all in `handle.rs`, and none is a - guard on a resource: - - | site | what it is | - |---|---| - | `producer()` | load-bearing, and the only check guarding no resource. Constructing a producer touches nothing gated — it just downgrades an `Arc` — so without it the call *succeeds* in a forked child, while `consumer()` refuses (it subscribes through `db()`). A handle that hands out producers but not consumers is a worse contract than one that hands out neither. It is not what makes a child safe — a producer built there refuses on first use regardless — but it is what makes the two factories agree. Removing it fails `dropping_an_inherited_handle_does_not_panic`. | - | `detach_internal` | a branch, not a refusal: release the thread rather than join one this process never had. | - | `Drop` | the same branch. | - - `consumer()` no longer checks explicitly — it subscribes through `db()`, which - checks — and `db_unchecked`, added as a deliberate hole for it, is deleted. - There are now no unchecked accessors in the crate. +- Two explicit `check()` calls remain, both in `handle.rs`, and both guard a + real resource — the `JoinHandle`. `detach_internal` and `Drop` use one to + decide whether to release the thread rather than join one this process never + had. Joining *is* the action there, so the check is a branch on state, not a + gate someone could forget. + + `producer()` briefly had a third. It was removed once the question "what does + it guard?" got a straight answer: nothing. Creating a producer touches neither + the database nor the runtime — `test_error_propagation` pins that an + unregistered key still yields a producer, with `set()` reporting the problem. + A forked child is one more thing `set()` reports, through the `db()` it must + pass. Keeping the check would have made `fork` the sole exception to this + crate's own lazy-producer contract, and would have left a category — + "checks that guard nothing" — for the next one to join. + + `consumer()` does refuse in a child, because subscribing goes through `db()`. + That asymmetry predates all of this: an unregistered key already fails at + `consumer()` and not at `producer()`. - `SyncConsumer::try_get` was briefly the last opt-in check, on the argument that it touches no runtime resource so there was nothing to gate. From d346348b6b95e4feac9ab144e907dd4871166467 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 19:52:25 +0000 Subject: [PATCH 08/11] docs: drop design note 050 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Removed at request. The reasoning it carried that the code still needs now stands on its own where it is used — why the fork check lives on the way in, why the consumer's reader is wrapped, and the 11-in-60 measurement that explains why the fork suite is two tests rather than five. Those comments cited the note by number; they no longer do, so nothing points at a file that is not there. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01DKAH7JNjPLPmG4mPTvWfth --- aimdb-sync/CHANGELOG.md | 2 +- aimdb-sync/src/runtime.rs | 6 +- aimdb-sync/tests/fork_safety_test.rs | 2 +- docs/design/050-sync-runtime-ownership.md | 329 ---------------------- 4 files changed, 5 insertions(+), 334 deletions(-) delete mode 100644 docs/design/050-sync-runtime-ownership.md diff --git a/aimdb-sync/CHANGELOG.md b/aimdb-sync/CHANGELOG.md index 243df1ce..3b33c8a4 100644 --- a/aimdb-sync/CHANGELOG.md +++ b/aimdb-sync/CHANGELOG.md @@ -61,7 +61,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 the runtime — buffered data stays readable after `detach` — so it holds a `RuntimeRef` whose handle is private to the runtime module and reachable only through a checked accessor, rather than a bare field beside a guard. `waiter.rs` is retired — `enter()` returns - the handle it existed to wrap. See design 050. + the handle it existed to wrap. ### Added diff --git a/aimdb-sync/src/runtime.rs b/aimdb-sync/src/runtime.rs index f62d4739..698251ee 100644 --- a/aimdb-sync/src/runtime.rs +++ b/aimdb-sync/src/runtime.rs @@ -19,8 +19,8 @@ //! [`Runtime::db`] the only route to the database, and both refuse a runtime //! this process did not inherit a thread for. A publish or a read cannot be //! written that skips the check, because it cannot be written without going -//! through one of them. Design 050 calls this "checked by construction rather -//! than by convention"; #231 gave panic-freedom the same treatment. +//! through one of them — checked by construction rather than by convention, +//! which is the treatment #231 gave panic-freedom. //! //! The check is deliberately *before* the database is handed out. A forked //! child's `Arc` is perfectly valid — it came across with the address @@ -214,7 +214,7 @@ mod tests { /// moving the stamp onto a value. Before this, proving a refusal path meant /// forking a real process from a parent holding a live Tokio runtime, the /// least safe moment there is; that suite failed 11 runs in 60 until it was - /// mitigated (design 050 §8). + /// mitigated. fn runtime_stamped(generations_behind: u64) -> (tokio::runtime::Runtime, Runtime) { let tokio_rt = tokio::runtime::Builder::new_current_thread() .enable_all() diff --git a/aimdb-sync/tests/fork_safety_test.rs b/aimdb-sync/tests/fork_safety_test.rs index ad8ec74d..e57810f7 100644 --- a/aimdb-sync/tests/fork_safety_test.rs +++ b/aimdb-sync/tests/fork_safety_test.rs @@ -3,7 +3,7 @@ //! Only what a unit test cannot prove. Every *refusal* path is now checked in //! `runtime.rs` against a `Runtime` stamped with a stale generation — no //! thread, no fork, no sleep — because the stamp became ordinary data when the -//! runtime became a value (design 050 §8). What survives here needs a real +//! runtime became a value. What survives here needs a real //! child process: that the `pthread_atfork` handler is genuinely installed and //! fires, and that a destructor in that child does not join a thread this //! process never had. diff --git a/docs/design/050-sync-runtime-ownership.md b/docs/design/050-sync-runtime-ownership.md deleted file mode 100644 index 1da41d67..00000000 --- a/docs/design/050-sync-runtime-ownership.md +++ /dev/null @@ -1,329 +0,0 @@ -# 050 — `aimdb-sync`: one runtime, one invariant - -**Status:** **implemented.** §6's open question is decided — `Weak`, not `Arc` -— and the reasoning below is amended where building it proved the note wrong. -Written against the tip of the #230 → #232 → #233 stack, which has merged. - -**Scope:** internal restructuring of `aimdb-sync`. No change to `aimdb-core`, no -change to the blocking API's shape (`attach` / `producer` / `consumer` / `set` / -`get` / `detach` all keep their signatures). One deliberate semantic decision is -open — §6 — and it is the reason this is a design note rather than a patch. - -**Vocabulary:** the **runtime thread** is the OS thread `aimdb-sync` spawns to -own a Tokio runtime on the caller's behalf. A **stamp** is the fork generation -recorded when an object is created, compared later to decide whether the object -predates a `fork()`. - ---- - -## 1. Where this sits - -`aimdb-sync` exists so an FFI or legacy caller gets a blocking API from a plain -`fn main()` with no `#[tokio::main]`. To do that it spawns and owns a runtime -thread the caller never sees. That decision is sound and is not in question -here — owning the thread is precisely why the crate, and not its callers, is the -layer that can notice the thread has gone. - -What follows is about how that owned thread is *represented*. - -## 2. What prompted this - -Four defects were found in `aimdb-sync` in a single review pass. They look -unrelated. They are the same defect. - -| # | Defect | Found by | -|---|---|---| -| 1 | `SyncConsumer`'s fork guard shipped with no test covering it | reading, not a failing test | -| 2 | `SyncError::ForkedChild` absent from `kind()`'s test and from the `lib.rs` variant list | reading; survived a PR that rewrote that exact test | -| 3 | `thread_alive` added to `AimDbHandle`, and both fork guards silently forgot to release it | reading | -| 4 | `set_value` / `try_set_value` / `set_value_at` are guarded only because they happen to delegate to `set` / `try_set` | reading; nothing enforces it | - -None was caught by a test, because in each case the code compiles, runs and -passes. Each is an instance of *someone had to remember something, and the -compiler could not help.* - -## 3. The two structural causes - -### 3.1 The invariant is enforced by convention, not construction - -"Is this runtime still usable after a `fork()`?" is **one fact**. Today it is -stored in three places and checked in nine: - -| | count | where | -|---|---|---| -| types carrying a copied `made_in` stamp | 3 | [`AimDbHandle`](../../aimdb-sync/src/handle.rs#L112), [`SyncProducer`](../../aimdb-sync/src/producer.rs#L39), [`SyncConsumer`](../../aimdb-sync/src/consumer.rs#L55) | -| hand-written `check_fork` implementations | 3 | `handle.rs`, `producer.rs`, `consumer.rs` | -| call sites that must *remember* to call it | 9 | 2 + 2 + 5 | -| public methods on `SyncProducer` / `SyncConsumer` | 5 + 5 | every one a chance to forget | - -The stamp is *copied per object* rather than being a property of the shared -runtime. Adding a tenth public method reintroduces the bug for free, and nothing -in the type system objects. Defects 1, 2 and 4 above are all this. - -This is the shape the repo already rejected once. #231's commit message reads -*"make panic-freedom a checked property, not a convention."* The fork guard is -the same class of problem and received the opposite treatment. - -### 3.2 The runtime thread is not a value - -`AimDbHandle` currently holds five fields, four of which are facets of one -concept: - -```rust -pub struct AimDbHandle { - thread_handle: Option>, // the thread - shutdown_tx: Option>, // how to stop it - runtime_handle: tokio::runtime::Handle, // how to enter it - db: Arc, - thread_alive: Option>>, // whether it still runs - made_in: crate::fork::Generation, // whether it is ours -} -``` - -"The runtime thread" is spelled out four times. Releasing it therefore means -remembering four `take()` calls in two separate guards — which is exactly how -defect 3 happened: #232 added `thread_alive` and the guards were not updated. - -The current mitigation, [`release_inherited`](../../aimdb-sync/src/handle.rs#L295), -centralises the *release* but leaves the four fields loose. It is a smaller -version of the same problem, not its removal. - -## 4. Proposed shape - -Group the thread into one owned value, and make the check unavoidable by making -it the only route to the runtime. - -```rust -/// One value, one OS thread. Constructed together, released together. -struct Runtime { - thread: JoinHandle<()>, - shutdown: mpsc::Sender, - alive: mpsc::Receiver<()>, - handle: tokio::runtime::Handle, - made_in: fork::Generation, -} - -impl Runtime { - /// The only way to reach the Tokio handle. The fork check lives here - /// because this is the chokepoint every operation must pass through. - fn enter(&self) -> SyncResult<&tokio::runtime::Handle> { - if fork::forked_since(self.made_in) { - return Err(SyncError::ForkedChild); - } - Ok(&self.handle) - } -} - -pub struct AimDbHandle { - rt: Option, // None once detached - db: Arc, -} -``` - -What this buys, concretely against §2: - -- **Defects 1 and 4 become unrepresentable.** A producer or consumer method - cannot publish or read without a `tokio::runtime::Handle`, and the only way to - get one is `enter()`. A tenth method inherits the guard by construction. -- **Defect 3 becomes unrepresentable.** Releasing the thread is `self.rt.take()` - — one field, impossible to half-do. -- **Defect 2 is unaffected** and stays a matter of discipline; it is a - documentation and test-coverage gap, not a structural one. - -Note the fork check must remain *before* the `Weak` upgrade, for the -reason #230 documented: in a forked child the upgrade **succeeds**, because the -`Arc` came across with the address space. That is exactly why a child's `set()` -would otherwise return `Ok` into a buffer nobody drains. `enter()` preserves -this ordering naturally — you reach the runtime before you reach the data. - -## 5. What it does not change - -- The public API's shape. `attach`, `producer`, `consumer`, `set`, `get`, - `detach`, `detach_timeout` keep their signatures. -- `aimdb-core`. Nothing crosses the crate boundary. -- The `pthread_atfork` mechanism, the generation-counter-not-poison-flag - reasoning, and the relaxed-atomic hot path — all of which #230 established and - measured, and none of which this touches. Detection stays as it is; only its - *plumbing* changes. -- `fork::generation` / `fork::forked_since` remain crate-private (#230 narrowed - them precisely so this refactor is free to choose a different shape). - -## 6. Decided: producers and consumers hold `Weak` - -This was the note's open question, and it recommended `Weak` while calling -`Arc` "arguably the better contract" for FFI callers. `Arc` was then tried, and -implementing it settled the question in the opposite direction from that aside. - -`Arc` buys exactly one thing: a producer outliving its handle keeps working -rather than failing with `RuntimeShutdown`. It costs two things that `Weak` -gets for free, because both fall out of ownership: - -- **Liveness.** With `Weak`, the failed upgrade *is* the check. With `Arc` it - has to be rebuilt — a flag the runtime thread flips on every exit path. -- **Waking a blocked reader.** Dropping the handle drops the database, which - closes its buffers, which is what wakes a consumer parked in `get()`. - `aimdb-core` has no explicit close, so with `Arc` nothing else does: a - consumer kept the database alive, `detach` closed nothing, and the reader - parked forever. That took a level-triggered stop channel and a `select` - around every blocking read to fix. - -So the ledger is one behaviour gained against three mechanisms added. And the -behaviour gained is itself questionable: under `Arc` a forgotten producer keeps -an OS thread and a Tokio runtime alive with nobody owning them — the stranded -thread #232 had just removed, reintroduced as a feature. - -| | `Weak` (chosen) | `Arc` (tried, reverted) | -|---|---|---| -| producer outliving its handle | `RuntimeShutdown` | keeps working | -| liveness check | free — the failed upgrade | explicit flag | -| waking a blocked reader | free — buffers close | stop channel + select per read | -| forgotten producer | harmless | strands a thread and a runtime | -| behaviour vs. today | identical | changed | - -`Weak` also keeps the whole change reviewable as "no behaviour changed", which -is worth more than the aside was. - -**One consequence, and how it is contained.** `SyncConsumer` needs a Tokio -handle that outlives the runtime: a `Reader` can still drain what is already -buffered after a `detach`, and delivering that data is behaviour the -characterization tests pin — gating reads on a live `Runtime` broke it. - -The first attempt stored a bare `handle` field beside a hand-written `guard()` -that returned `Ok` when the upgrade failed. That is the convention shape this -design exists to remove, rebuilt on one type: five opt-in call sites and a -field any of them could use without checking anything. - -It is now a `RuntimeRef` in `runtime.rs`, holding the `Weak` and the handle with -**both fields private to that module**. `consumer.rs` cannot obtain a handle -except through `RuntimeRef::enter`, which checks first. So the blocking reads -are checked by construction, exactly as the publish path is. - -## 7. Secondary: the `Drop` / `detach` duality - -`AimDbHandle` has two shutdown paths for one resource: -[`detach_internal`](../../aimdb-sync/src/handle.rs#L450), reached from `detach` -and `detach_timeout`, and [`Drop`](../../aimdb-sync/src/handle.rs#L568). One can -report failure; the other cannot. - -#232 resolved the acute problem by making `Drop` non-blocking — the right -answer, arrived at as a bug fix rather than as a starting principle. With -`Runtime` as a value the residual duplication collapses naturally: `Drop` -becomes "signal and release", `detach` becomes "signal, release, and wait", and -both are expressed against one field instead of four. - -This is worth folding into the same change. It is not worth a separate one. - -## 8. Why this is also a testing problem - -The fork test suite in `aimdb-sync/tests/` had to fork real processes from a -parent holding a live, freshly started Tokio runtime — the least safe possible -moment, because the child inherits an allocator lock held by threads that did -not survive. Measured on the #230 branch, the suite failed **11 of 60 runs** -before mitigation, and required both a `mem::forget` of inherited state and a -settle period before the fork to reach zero. The settle is a duration, not a -handshake; it is a mitigation, not a guarantee. - -That difficulty is downstream of §3.2. Because `AimDbHandle` inseparably *means* -"a spawned OS thread", there is no way to construct the fork condition without -one. With `Runtime` as a value, `made_in` is ordinary data: the refusal paths — -every `SyncProducer` and `SyncConsumer` method, `producer()`, `consumer()`, -`detach()`, `Drop` — become unit-testable against a `Runtime` stamped with a -stale generation. No thread, no fork, no sleep. - -The end-to-end fork tests should not all be deleted; **one** genuine -`fork()`-and-assert case is worth keeping, because a unit test cannot prove the -`pthread_atfork` handler is really installed. But it should be one cheap case -instead of five expensive ones, and the watchdog in -`aimdb-sync/tests/fork_child/mod.rs` should stay regardless. - -## 9. How it was done - -Done after #230, #232 and #233 merged, as a single self-contained change. - -The steps, as executed: - -1. Introduce `Runtime`, move the four fields and the stamp into it. Internal - only; no signature changes. -2. Route every runtime access through `enter()`. Delete the three `check_fork` - implementations and the nine call sites. -3. Point `SyncProducer` / `SyncConsumer` at `Weak` (§6). -4. Collapse `Drop` / `detach_internal` onto the single field (§7). -5. Convert the refusal-path tests from fork tests to unit tests, keeping one - end-to-end fork case (§8). - -Outcome, measured against the merged stack: - -| | before | after | -|---|---|---| -| types carrying a copied fork stamp | 3 | 0 | -| hand-written `check_fork` implementations | 3 | 0 | -| call sites that must remember to guard | 9 | 0 | -| fields on `AimDbHandle` | 6 | 2 | -| forking tests | 5, in 2 binaries | 2, in 1 | -| modules | +`runtime.rs`, −`waiter.rs` | | - -Two forking tests remain rather than one. `dropping_an_inherited_handle_does_not_panic` -covers a destructor joining a thread this process never had, which panics -inside `std` — that needs a genuinely dead thread and no unit test can supply -one. The other proves the `pthread_atfork` handler is really installed. - -## 10. Risks and what this does not fix - -- **It is a refactor of working code.** Every defect in §2 is fixed on the - stack today; the guard is complete as it stands. The argument for this work is - that the *next* one is free to reappear, not that the crate is broken. -- **`enter()` is only a chokepoint if nothing else exposes the handle.** If a - future method returns `&tokio::runtime::Handle` or clones it out, the - guarantee is gone. That constraint needs stating in the type's docs, and it is - the one thing here still enforced by convention. -- **It does not remove the fork hazard from tests entirely** (§8) — one real - fork case remains, and it is the expensive one. -- **It does not address `fork` on non-Unix**, where `generation()` is - permanently `0`. That is correct today and stays correct; it is noted only so - the next reader does not mistake it for an oversight. - -## 11. Settled, and what is left - -1. **`Weak` or `Arc`?** Settled as `Weak` — see §6. The note's own aside - favouring `Arc` did not survive contact with the implementation. -2. **Should `Runtime` be exposed?** It is `pub(crate)`. Keep it private until an - FFI layer exists and can say what it needs — the reasoning that made - `fork::generation` crate-private in #230. -3. **Is the end-to-end fork coverage enough?** Two tests, for the two things a - unit test cannot reach (§9). The watchdog in `tests/fork_child/mod.rs` stays - regardless: it converts a deadlock from a six-hour CI hang into a bounded - failure, and that should not depend on how likely the deadlock is. - -Still open, unchanged by this work: - -- `enter()` is a chokepoint only while nothing else hands out the Tokio handle. - Nothing does: `Runtime` and `RuntimeRef` keep theirs private to - `runtime.rs`, so the constraint is enforced by module privacy rather than by - memory. Adding a `pub(crate)` accessor that returns one would silently undo - that, which is worth stating because it is the only way back to the old - shape. -- Two explicit `check()` calls remain, both in `handle.rs`, and both guard a - real resource — the `JoinHandle`. `detach_internal` and `Drop` use one to - decide whether to release the thread rather than join one this process never - had. Joining *is* the action there, so the check is a branch on state, not a - gate someone could forget. - - `producer()` briefly had a third. It was removed once the question "what does - it guard?" got a straight answer: nothing. Creating a producer touches neither - the database nor the runtime — `test_error_propagation` pins that an - unregistered key still yields a producer, with `set()` reporting the problem. - A forked child is one more thing `set()` reports, through the `db()` it must - pass. Keeping the check would have made `fork` the sole exception to this - crate's own lazy-producer contract, and would have left a category — - "checks that guard nothing" — for the next one to join. - - `consumer()` does refuse in a child, because subscribing goes through `db()`. - That asymmetry predates all of this: an unregistered key already fails at - `consumer()` and not at `producer()`. - -- `SyncConsumer::try_get` was briefly the last opt-in check, on the - argument that it touches no runtime resource so there was nothing to gate. - That was wrong: it touches the `Reader`, which is precisely the thing to gate. - The reader now lives in a `Guarded>` whose value is private to - `runtime.rs`, so every read reaches it through a checked accessor. The - consumer has one field and no way to skip the check. From 172a89e5e35495351bb95c1cb147cdd989905547 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 20:16:11 +0000 Subject: [PATCH 09/11] fix(sync): refuse a forked child that released its inherited handle RuntimeRef inferred the fork check from a Weak upgrade and treated a failed upgrade as "detached, let the read through". That conflates two things which need opposite answers. A handle dropped in this process is a detach: the buffer may still hold data and delivering it is behaviour the characterization tests pin. A handle released in a forked child is not: the thread that fills that buffer does not exist here. The Weak cannot tell them apart, and releasing an inherited handle is precisely what a child is supposed to do. So a child that dropped its handle passed the check. try_get reported GetTimeout instead of ForkedChild, and get would park forever on the inherited Tokio handle. Verified against both revisions with the same reproducer: main reports ForkedChild, this branch reported GetTimeout. That is the silence the generation check exists to prevent, reintroduced by the refactor meant to make it unavoidable. RuntimeRef now carries its runtime's generation, copied at construction by the new Runtime::view - the only way to build one, and checked, so a view is not a route around enter(). The Weak goes with it: once the stamp is carried, the upgrade answers nothing, and liveness was never its job (a closed buffer is what reports RuntimeShutdown). Pinned three ways, since nothing covered RuntimeRef or Guarded before: unit tests for the detach case, the fork case and both routes through Guarded, against a stale-stamped view rather than a real fork; and the assertion folded into the existing child that already drops a handle, so no third forking binary and no new allocator-lock exposure. Mutating check() to the old semantics fails all of them. Also corrected four doc sites left describing the reverted Arc design - the rt field stated the ownership invariant backwards, two cited keepalive clones that producers and consumers do not hold, and one cited a RunningFlag that does not exist - and the CHANGELOG's "no behaviour changed", which missed that handle.producer() in a child now defers its refusal to set(). Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01BFpXuTKapm8XhFEGdMtCcr --- aimdb-sync/CHANGELOG.md | 24 ++++-- aimdb-sync/src/handle.rs | 34 ++++---- aimdb-sync/src/runtime.rs | 114 +++++++++++++++++++++++---- aimdb-sync/tests/fork_safety_test.rs | 20 ++++- 4 files changed, 154 insertions(+), 38 deletions(-) diff --git a/aimdb-sync/CHANGELOG.md b/aimdb-sync/CHANGELOG.md index 3b33c8a4..c35eebd7 100644 --- a/aimdb-sync/CHANGELOG.md +++ b/aimdb-sync/CHANGELOG.md @@ -55,21 +55,31 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 copied into three types and checked at nine call sites that each had to opt in; four defects in one review pass were all instances of someone forgetting to. `AimDbHandle` drops from six loose fields to two, so releasing state - inherited across a `fork` can no longer be half-done. No API signature and no - behaviour changed: producers and consumers still hold a weak reference, so a - handle's lifetime still governs. A consumer needs a Tokio handle that outlives + inherited across a `fork` can no longer be half-done. No API signature + changed, and producers and consumers still hold a weak reference, so a + handle's lifetime still governs. One behaviour did: `handle.producer()` in a + forked child now returns a producer instead of `Err(ForkedChild)`, and the + refusal lands on the first `set()`/`try_set()` instead. Creating a producer + touches neither the database nor the runtime thread, so this matches what an + unregistered key has always done there; nothing can be published either way. + A consumer needs a Tokio handle that outlives the runtime — buffered data stays readable after `detach` — so it holds a `RuntimeRef` whose handle is private to the runtime module and reachable only - through a checked accessor, rather than a bare field beside a guard. `waiter.rs` is retired — `enter()` returns + through a checked accessor, rather than a bare field beside a guard. That view + carries its runtime's fork generation rather than inferring it from a weak + reference: once the runtime is gone, a failed upgrade cannot tell a `detach` + in this process from a child that released its inherited handle, and the two + need opposite answers. `waiter.rs` is retired — `enter()` returns the handle it existed to wrap. ### Added - **`fork()` safety.** A child of `fork` inherits every handle, producer and consumer the parent held, and none of the runtime thread that makes them work - — so its `set()` used to return `Ok` into a buffer nobody drains. Handles, - producers and consumers now record a fork generation and refuse with the new - `SyncError::ForkedChild` once the process has forked since they were made. + — so its `set()` used to return `Ok` into a buffer nobody drains. The runtime + now records the fork generation it was built in, and every route to it — a + publish, a read, a subscribe — refuses with the new `SyncError::ForkedChild` + once the process has forked since then. `detach` and `Drop` release the runtime thread's `JoinHandle` rather than joining a thread this process does not have, which panicked inside `std`. Detection is a lazily registered `pthread_atfork` handler, so the check on the diff --git a/aimdb-sync/src/handle.rs b/aimdb-sync/src/handle.rs index 6218bf09..450d8d12 100644 --- a/aimdb-sync/src/handle.rs +++ b/aimdb-sync/src/handle.rs @@ -1,6 +1,6 @@ //! AimDB handle for managing the sync API runtime thread. -use crate::runtime::{Runtime, RuntimeRef, ShutdownSignal}; +use crate::runtime::{Runtime, ShutdownSignal}; use crate::{SyncError, SyncResult}; use aimdb_core::{log_error, log_warn, AimDb, AimDbBuilder, DbError}; use alloc::sync::Arc; @@ -113,9 +113,12 @@ pub struct AimDbHandle { /// The runtime thread and everything reached through it, shared with every /// producer and consumer made from this handle. See [`crate::runtime`]. /// - /// Holding this keeps the thread running, so a producer that outlives its - /// handle keeps working. [`detach`](Self::detach) is what stops the thread - /// deliberately. + /// This is the only strong reference: producers and consumers hold a + /// [`Weak`](alloc::sync::Weak), so the runtime dies with this handle and a + /// producer that outlives it fails with + /// [`SyncError::RuntimeShutdown`](crate::SyncError::RuntimeShutdown). + /// [`detach`](Self::detach) is what stops the thread deliberately, without + /// waiting for the survivors to finish. rt: Arc, /// What only the handle that started the thread may do: signal it, wait for @@ -133,9 +136,9 @@ struct OwnedThread { /// Joined by `detach`; released, never joined, by `Drop`. join: JoinHandle<()>, - /// Commands the thread to stop. Distinct from the keepalive clone every - /// [`Runtime`] holds: dropping a sender only decrements, sending *ends* the - /// thread even while other holders are alive. + /// Commands the thread to stop. Distinct from dropping it: the thread also + /// ends when every sender is gone, but *sending* ends it now, while + /// producers and consumers still hold views of the runtime. shutdown: mpsc::Sender, /// Held open by the runtime thread for exactly as long as it runs. @@ -230,7 +233,7 @@ impl AimDbHandle { // `new_from_builder` has always used. See `recv_startup`. let (handle_tx, mut handle_rx) = mpsc::channel::>(1); - // See `OwnedThread::alive` and `RunningFlag`. + // See `OwnedThread::alive`. let (alive_tx, thread_alive) = std::sync::mpsc::channel::<()>(); // Wrap database in Arc for sharing @@ -389,10 +392,7 @@ impl AimDbHandle { .db()? .subscribe::(&record_key) .map_err(SyncError::Db)?; - Ok(crate::SyncConsumer::new( - RuntimeRef::new(Arc::downgrade(&self.rt), self.rt.enter()?.clone()), - reader, - )) + Ok(crate::SyncConsumer::new(self.rt.view()?, reader)) } /// Gracefully shut down the runtime thread. @@ -466,10 +466,12 @@ impl AimDbHandle { /// Internal detach implementation. /// /// Deliberate shutdown: *sends* the signal rather than merely dropping a - /// sender, so the thread stops even though producers and consumers may - /// still hold [`Runtime`] clones keeping it alive. Those survivors then - /// fail with [`SyncError::RuntimeShutdown`], which is the point — `detach` - /// means "stop now", not "stop when everyone has finished". + /// sender, so the thread stops at once rather than when the last sender + /// goes. Producers and consumers hold only a + /// [`Weak`](alloc::sync::Weak), so they cannot keep it alive; what they can + /// do is still be mid-call when it stops, and they then fail with + /// [`SyncError::RuntimeShutdown`]. That is the point — `detach` means "stop + /// now", not "stop when everyone has finished". fn detach_internal(&mut self, timeout: Option) -> SyncResult<()> { // A forked child holds a `JoinHandle` for a thread that does not exist // here, and joining it is not merely useless: it panics inside `std` diff --git a/aimdb-sync/src/runtime.rs b/aimdb-sync/src/runtime.rs index 698251ee..355c3f3c 100644 --- a/aimdb-sync/src/runtime.rs +++ b/aimdb-sync/src/runtime.rs @@ -47,7 +47,7 @@ //! producer keep an OS thread and a Tokio runtime alive with nobody owning //! them, which is the stranded thread #232 had just removed. -use alloc::sync::{Arc, Weak}; +use alloc::sync::Arc; use aimdb_core::AimDb; @@ -114,6 +114,19 @@ impl Runtime { self.check()?; Ok(&self.db) } + + /// A view of this runtime that may outlive it. The only way to build one. + /// + /// Checked, because handing out the Tokio handle is what [`Self::enter`] + /// exists to gate — a view is not a way around it. The generation is copied + /// out here rather than read from the process, so it is *this* runtime's + /// stamp the view carries and not whatever the caller happened to be at. + pub(crate) fn view(&self) -> SyncResult { + Ok(RuntimeRef { + handle: self.enter()?.clone(), + made_in: self.made_in, + }) + } } /// A borrowed view of a [`Runtime`] that outlives it on purpose. @@ -129,29 +142,44 @@ impl Runtime { /// through [`Self::enter`], so a read that skips the check cannot be written /// there any more than it can anywhere else. A bare handle field beside a /// hand-written guard — the shape this replaces — offered no such thing. +/// +/// # Why the generation is copied rather than read through a `Weak` +/// +/// Everything else holds a [`Weak`](alloc::sync::Weak) and lets a failed upgrade answer the +/// question. That works only while the upgrade *means* something, and here it +/// does not: this view is built to outlive its runtime, so a failed upgrade is +/// the expected case rather than the interesting one. It also cannot +/// distinguish the two ways of getting there — a handle detached in this +/// process, where the buffer must still be drained, from one released in a +/// forked child, where the thread that fills that buffer does not exist and a +/// blocking read would park forever. Copying the stamp answers both without +/// consulting the `Arc` at all. pub(crate) struct RuntimeRef { - rt: Weak, handle: tokio::runtime::Handle, + + /// The generation of the runtime this views, copied at construction. + /// + /// The same value [`Runtime::made_in`] holds, so the two agree by + /// construction and this view keeps answering after that `Runtime` is + /// gone. + made_in: crate::fork::Generation, } impl RuntimeRef { - pub(crate) fn new(rt: Weak, handle: tokio::runtime::Handle) -> Self { - Self { rt, handle } - } - /// Refuse a forked child; let a detached one through. /// - /// A failed upgrade means the handle was dropped in *this* process, which - /// is not a fork — the buffer is the right thing to answer for itself. - /// After a real `fork` the upgrade **succeeds**, because the `Arc` came - /// across with the address space, which is exactly why the check is worth - /// making at all. + /// A detach is not a fork: the runtime is gone but this process is the one + /// that dropped it, so the buffer is the right thing to answer for itself + /// and the read carries on. A `fork` is refused whether or not the child + /// still holds the handle — which is the case a `Weak` upgrade got wrong, + /// because releasing an inherited handle is exactly what a child is + /// supposed to do. #[inline] pub(crate) fn check(&self) -> SyncResult<()> { - match self.rt.upgrade() { - Some(rt) => rt.check(), - None => Ok(()), + if crate::fork::forked_since(self.made_in) { + return Err(SyncError::ForkedChild); } + Ok(()) } /// The Tokio handle, checked. The only way to obtain one. @@ -272,4 +300,62 @@ mod tests { let err = rt.enter().expect_err("must refuse"); assert_eq!(err.kind(), DbErrorKind::Closed); } + + /// A view is not a way around [`Runtime::enter`]. + #[test] + fn a_view_of_a_forked_runtime_cannot_be_taken() { + let (_guard, rt) = runtime_stamped(1); + assert!(matches!(rt.view(), Err(SyncError::ForkedChild))); + } + + /// The detach case: the runtime is gone, but this process is what dropped + /// it. A consumer must still drain what its buffer already holds — the + /// behaviour the characterization tests pin. + #[test] + fn a_view_outliving_its_runtime_still_reads() { + let (_guard, rt) = runtime_stamped(0); + let view = rt.view().expect("view"); + drop(rt); + + assert!(view.check().is_ok()); + assert!(view.enter().is_ok()); + } + + /// The fork case, and the one a `Weak` upgrade got wrong. + /// + /// Releasing an inherited handle is what a forked child is *supposed* to + /// do, so the runtime being gone says nothing about whether this process + /// owns the thread. Before the stamp was carried, this view answered `Ok` + /// and a `get()` on it parked forever on a thread that does not exist here. + #[test] + fn a_view_outliving_its_runtime_in_a_forked_child_refuses() { + let (_guard, mut rt) = runtime_stamped(0); + let mut view = rt.view().expect("view"); + + // The fork happens after the view is taken, which is the order that + // matters: the view was legitimate when it was made. + view.made_in = crate::fork::generation().wrapping_sub(1); + rt.made_in = view.made_in; + drop(rt); + + assert!(matches!(view.check(), Err(SyncError::ForkedChild))); + assert!(matches!(view.enter(), Err(SyncError::ForkedChild))); + } + + /// Both routes through the wrapper check, including the one that needs no + /// runtime. `try_get` reads straight out of the buffer, which is exactly + /// why `get` has to check rather than lean on `enter`. + #[test] + fn a_guarded_value_is_unreachable_after_a_fork() { + let (_guard, rt) = runtime_stamped(0); + let mut guarded = Guarded::new(rt.view().expect("view"), 7u32); + + assert_eq!(*guarded.get().expect("current generation reads"), 7); + assert!(guarded.enter().is_ok()); + + guarded.rt.made_in = crate::fork::generation().wrapping_sub(1); + + assert!(matches!(guarded.get(), Err(SyncError::ForkedChild))); + assert!(matches!(guarded.enter(), Err(SyncError::ForkedChild))); + } } diff --git a/aimdb-sync/tests/fork_safety_test.rs b/aimdb-sync/tests/fork_safety_test.rs index e57810f7..45242fc9 100644 --- a/aimdb-sync/tests/fork_safety_test.rs +++ b/aimdb-sync/tests/fork_safety_test.rs @@ -84,6 +84,13 @@ fn dropping_an_inherited_handle_does_not_panic() { let to_detach = attach(); let to_drop = attach(); + // Made before the fork, so the child inherits a consumer that was + // legitimate when it was created. Allocated here, in the parent, where + // allocating is safe. + let mut inherited = to_drop + .consumer::("sensor.reading") + .expect("consumer"); + let code = in_forked_child(move || { // `detach` reports the situation rather than joining. let refused = matches!(to_detach.detach(), Err(SyncError::ForkedChild)); @@ -108,7 +115,18 @@ fn dropping_an_inherited_handle_does_not_panic() { // normally, and the parent's `WIFEXITED` assertion would catch it. drop(to_drop); - refused && producer_defers && no_consumer + // Releasing the handle must not release the *refusal*. Nothing about + // this process changed: the runtime thread was never here. A consumer + // that answered on its runtime's liveness would now say the buffer is + // merely empty — and a blocking read would park forever on a thread + // that does not exist — because dropping the handle above is the very + // thing a child is supposed to do. Only the generation the consumer + // carries can tell a detach from a fork once the runtime is gone. + let still_refused = matches!(inherited.try_get(), Err(SyncError::ForkedChild)); + // Leak rather than free, per the module note in `fork_child`. + std::mem::forget(inherited); + + refused && producer_defers && no_consumer && still_refused }); assert_eq!(code, 0, "detach in a child should be refused, not fatal"); } From 4c6943f5ffe1a37fe95d6ad302713a9913b760c2 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 20:29:11 +0000 Subject: [PATCH 10/11] docs(sync): tighten the comments added with the fork-check fix Same content, fewer lines. The RuntimeRef rationale loses its heading and becomes one paragraph, the four test docs and two handle.rs field docs drop to what they were actually saying, and the child's assertion comment stops restating the fix. 29 comment lines, no prose the code needed. One line of code goes with them: the fork test wrote a stale stamp back to the Runtime it then dropped, which nothing reads now that the view carries its own. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01BFpXuTKapm8XhFEGdMtCcr --- aimdb-sync/src/handle.rs | 20 ++++----- aimdb-sync/src/runtime.rs | 65 ++++++++++------------------ aimdb-sync/tests/fork_safety_test.rs | 14 +++--- 3 files changed, 35 insertions(+), 64 deletions(-) diff --git a/aimdb-sync/src/handle.rs b/aimdb-sync/src/handle.rs index 450d8d12..7611513c 100644 --- a/aimdb-sync/src/handle.rs +++ b/aimdb-sync/src/handle.rs @@ -113,12 +113,10 @@ pub struct AimDbHandle { /// The runtime thread and everything reached through it, shared with every /// producer and consumer made from this handle. See [`crate::runtime`]. /// - /// This is the only strong reference: producers and consumers hold a + /// The only strong reference: producers and consumers hold a /// [`Weak`](alloc::sync::Weak), so the runtime dies with this handle and a - /// producer that outlives it fails with - /// [`SyncError::RuntimeShutdown`](crate::SyncError::RuntimeShutdown). - /// [`detach`](Self::detach) is what stops the thread deliberately, without - /// waiting for the survivors to finish. + /// producer outliving it fails with `RuntimeShutdown`. + /// [`detach`](Self::detach) is what stops the thread deliberately. rt: Arc, /// What only the handle that started the thread may do: signal it, wait for @@ -137,8 +135,7 @@ struct OwnedThread { join: JoinHandle<()>, /// Commands the thread to stop. Distinct from dropping it: the thread also - /// ends when every sender is gone, but *sending* ends it now, while - /// producers and consumers still hold views of the runtime. + /// ends when every sender is gone, but *sending* ends it now. shutdown: mpsc::Sender, /// Held open by the runtime thread for exactly as long as it runs. @@ -467,11 +464,10 @@ impl AimDbHandle { /// /// Deliberate shutdown: *sends* the signal rather than merely dropping a /// sender, so the thread stops at once rather than when the last sender - /// goes. Producers and consumers hold only a - /// [`Weak`](alloc::sync::Weak), so they cannot keep it alive; what they can - /// do is still be mid-call when it stops, and they then fail with - /// [`SyncError::RuntimeShutdown`]. That is the point — `detach` means "stop - /// now", not "stop when everyone has finished". + /// goes. Producers and consumers hold only a [`Weak`](alloc::sync::Weak) + /// and cannot keep it alive, but one mid-call when it stops fails with + /// [`SyncError::RuntimeShutdown`] — `detach` means "stop now", not "stop + /// when everyone has finished". fn detach_internal(&mut self, timeout: Option) -> SyncResult<()> { // A forked child holds a `JoinHandle` for a thread that does not exist // here, and joining it is not merely useless: it panics inside `std` diff --git a/aimdb-sync/src/runtime.rs b/aimdb-sync/src/runtime.rs index 355c3f3c..bc70edb3 100644 --- a/aimdb-sync/src/runtime.rs +++ b/aimdb-sync/src/runtime.rs @@ -117,10 +117,8 @@ impl Runtime { /// A view of this runtime that may outlive it. The only way to build one. /// - /// Checked, because handing out the Tokio handle is what [`Self::enter`] - /// exists to gate — a view is not a way around it. The generation is copied - /// out here rather than read from the process, so it is *this* runtime's - /// stamp the view carries and not whatever the caller happened to be at. + /// Checked, so a view is not a way around [`Self::enter`], and it copies + /// the runtime's own stamp rather than reading the process's. pub(crate) fn view(&self) -> SyncResult { Ok(RuntimeRef { handle: self.enter()?.clone(), @@ -143,37 +141,25 @@ impl Runtime { /// there any more than it can anywhere else. A bare handle field beside a /// hand-written guard — the shape this replaces — offered no such thing. /// -/// # Why the generation is copied rather than read through a `Weak` -/// -/// Everything else holds a [`Weak`](alloc::sync::Weak) and lets a failed upgrade answer the -/// question. That works only while the upgrade *means* something, and here it -/// does not: this view is built to outlive its runtime, so a failed upgrade is -/// the expected case rather than the interesting one. It also cannot -/// distinguish the two ways of getting there — a handle detached in this -/// process, where the buffer must still be drained, from one released in a -/// forked child, where the thread that fills that buffer does not exist and a -/// blocking read would park forever. Copying the stamp answers both without -/// consulting the `Arc` at all. +/// The generation is copied rather than read through a +/// [`Weak`](alloc::sync::Weak): a failed upgrade is the expected case for a +/// view built to outlive its runtime, and it cannot tell a detach here — where +/// the buffer must still be drained — from a child that released its inherited +/// handle, where a blocking read would park forever. pub(crate) struct RuntimeRef { handle: tokio::runtime::Handle, - /// The generation of the runtime this views, copied at construction. - /// - /// The same value [`Runtime::made_in`] holds, so the two agree by - /// construction and this view keeps answering after that `Runtime` is - /// gone. + /// The runtime's own generation, copied at construction so this view keeps + /// answering after that runtime is gone. made_in: crate::fork::Generation, } impl RuntimeRef { /// Refuse a forked child; let a detached one through. /// - /// A detach is not a fork: the runtime is gone but this process is the one - /// that dropped it, so the buffer is the right thing to answer for itself - /// and the read carries on. A `fork` is refused whether or not the child - /// still holds the handle — which is the case a `Weak` upgrade got wrong, - /// because releasing an inherited handle is exactly what a child is - /// supposed to do. + /// A detach is not a fork — the buffer answers for itself — but a fork is + /// refused whether or not the child still holds the handle, which is what a + /// `Weak` upgrade got wrong. #[inline] pub(crate) fn check(&self) -> SyncResult<()> { if crate::fork::forked_since(self.made_in) { @@ -308,9 +294,8 @@ mod tests { assert!(matches!(rt.view(), Err(SyncError::ForkedChild))); } - /// The detach case: the runtime is gone, but this process is what dropped - /// it. A consumer must still drain what its buffer already holds — the - /// behaviour the characterization tests pin. + /// The detach case: the runtime is gone but this process dropped it, so the + /// buffer must still drain — what the characterization tests pin. #[test] fn a_view_outliving_its_runtime_still_reads() { let (_guard, rt) = runtime_stamped(0); @@ -321,36 +306,30 @@ mod tests { assert!(view.enter().is_ok()); } - /// The fork case, and the one a `Weak` upgrade got wrong. - /// - /// Releasing an inherited handle is what a forked child is *supposed* to - /// do, so the runtime being gone says nothing about whether this process - /// owns the thread. Before the stamp was carried, this view answered `Ok` - /// and a `get()` on it parked forever on a thread that does not exist here. + /// The fork case, and the one a `Weak` upgrade got wrong: releasing an + /// inherited handle is what a child is *supposed* to do, so a gone runtime + /// says nothing about who owns the thread. This answered `Ok` before. #[test] fn a_view_outliving_its_runtime_in_a_forked_child_refuses() { - let (_guard, mut rt) = runtime_stamped(0); + let (_guard, rt) = runtime_stamped(0); let mut view = rt.view().expect("view"); - // The fork happens after the view is taken, which is the order that - // matters: the view was legitimate when it was made. + // Forked after the view was taken — the order that matters. view.made_in = crate::fork::generation().wrapping_sub(1); - rt.made_in = view.made_in; drop(rt); assert!(matches!(view.check(), Err(SyncError::ForkedChild))); assert!(matches!(view.enter(), Err(SyncError::ForkedChild))); } - /// Both routes through the wrapper check, including the one that needs no - /// runtime. `try_get` reads straight out of the buffer, which is exactly - /// why `get` has to check rather than lean on `enter`. + /// Both routes check, including `get`, which needs no runtime — `try_get` + /// reads straight out of the buffer, so it cannot lean on `enter`. #[test] fn a_guarded_value_is_unreachable_after_a_fork() { let (_guard, rt) = runtime_stamped(0); let mut guarded = Guarded::new(rt.view().expect("view"), 7u32); - assert_eq!(*guarded.get().expect("current generation reads"), 7); + assert_eq!(*guarded.get().expect("reads before a fork"), 7); assert!(guarded.enter().is_ok()); guarded.rt.made_in = crate::fork::generation().wrapping_sub(1); diff --git a/aimdb-sync/tests/fork_safety_test.rs b/aimdb-sync/tests/fork_safety_test.rs index 45242fc9..7c4c0e68 100644 --- a/aimdb-sync/tests/fork_safety_test.rs +++ b/aimdb-sync/tests/fork_safety_test.rs @@ -85,8 +85,7 @@ fn dropping_an_inherited_handle_does_not_panic() { let to_drop = attach(); // Made before the fork, so the child inherits a consumer that was - // legitimate when it was created. Allocated here, in the parent, where - // allocating is safe. + // legitimate when created — and allocated in the parent, where that is safe. let mut inherited = to_drop .consumer::("sensor.reading") .expect("consumer"); @@ -115,13 +114,10 @@ fn dropping_an_inherited_handle_does_not_panic() { // normally, and the parent's `WIFEXITED` assertion would catch it. drop(to_drop); - // Releasing the handle must not release the *refusal*. Nothing about - // this process changed: the runtime thread was never here. A consumer - // that answered on its runtime's liveness would now say the buffer is - // merely empty — and a blocking read would park forever on a thread - // that does not exist — because dropping the handle above is the very - // thing a child is supposed to do. Only the generation the consumer - // carries can tell a detach from a fork once the runtime is gone. + // Releasing the handle must not release the *refusal*: the runtime + // thread was never here either way. Answering on the runtime's liveness + // instead would call the buffer merely empty — and park a blocking read + // forever — because the drop above is what a child is supposed to do. let still_refused = matches!(inherited.try_get(), Err(SyncError::ForkedChild)); // Leak rather than free, per the module note in `fork_child`. std::mem::forget(inherited); From a01e54b541a4f5f9519352540b58cc8d324e49ed Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 20:32:36 +0000 Subject: [PATCH 11/11] docs(sync): tighten runtime.rs, and fix what the fork fix made untrue The module note said producers and consumers hold a Weak. Consumers stopped doing that when RuntimeRef began carrying its own generation, so the ownership section now names the RuntimeRef and points at it for the reason. The rest is length. "Why this is one type" and "Why the check lives on the way in" argued the same point twice, so the history folds into the rule it justifies. The RuntimeRef and Guarded notes both explained that a plain field beside a hand-written guard is a call someone can forget; Guarded keeps it. enter() and db() each spent a summary line and a blank on "the only way to reach it". The test helper's 11-in-60 measurement stays, in one sentence. 108 comment lines, down from 141 and from the 122 this file carried before the fix, which added a type and four tests. No fact dropped. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01BFpXuTKapm8XhFEGdMtCcr --- aimdb-sync/src/runtime.rs | 137 +++++++++++++++----------------------- 1 file changed, 52 insertions(+), 85 deletions(-) diff --git a/aimdb-sync/src/runtime.rs b/aimdb-sync/src/runtime.rs index bc70edb3..ad9c4eff 100644 --- a/aimdb-sync/src/runtime.rs +++ b/aimdb-sync/src/runtime.rs @@ -2,50 +2,38 @@ //! //! `aimdb-sync` spawns an OS thread to own a Tokio runtime on the caller's //! behalf. Everything that thread *is* — the way in, the database it built, and -//! the `fork` generation it belongs to — lives here, in one type, reached by -//! every object that needs it. -//! -//! # Why this is one type -//! -//! It used to be four fields on [`AimDbHandle`](crate::AimDbHandle) and a fork -//! stamp copied into three others. That shape cost four bugs in one review -//! pass, all of the same kind: someone had to remember something and the -//! compiler could not help. A guard was added to a new method — or wasn't. A -//! field was added to the handle — and the release path forgot it. +//! the `fork` generation it belongs to — lives here, in one type. //! //! # Why the check lives on the way in //! -//! [`Runtime::enter`] is the only route to the Tokio handle and -//! [`Runtime::db`] the only route to the database, and both refuse a runtime -//! this process did not inherit a thread for. A publish or a read cannot be -//! written that skips the check, because it cannot be written without going -//! through one of them — checked by construction rather than by convention, -//! which is the treatment #231 gave panic-freedom. +//! [`Runtime::enter`] and [`Runtime::db`] are the only routes to the handle and +//! the database, and both refuse a runtime this process did not inherit a +//! thread for, so a publish or a read cannot be written that skips the check — +//! by construction rather than by convention, the treatment #231 gave +//! panic-freedom. It replaces a stamp copied into three types and guarded at +//! nine opt-in call sites, a shape that cost four bugs in one review pass, +//! every one of them someone forgetting to opt in. //! -//! The check is deliberately *before* the database is handed out. A forked +//! The check comes *before* the database is handed out, because a forked //! child's `Arc` is perfectly valid — it came across with the address -//! space — so a child that reached the database would publish into a buffer -//! nobody drains and be told `Ok`. That silence is the bug this whole mechanism -//! exists to prevent. +//! space — so a child that reached it would publish into a buffer nobody drains +//! and be told `Ok`. That silence is the bug this mechanism exists to prevent. //! //! # Who owns it //! -//! The handle owns the `Arc`; producers and consumers hold a -//! [`Weak`](alloc::sync::Weak). So the database dies with the handle, exactly -//! as it did when producers held `Weak` directly, and two things stay -//! free that shared ownership would have made expensive: +//! The handle owns the `Arc` and producers hold a +//! [`Weak`](alloc::sync::Weak), so the database dies with the handle and two +//! things stay free that shared ownership would have made expensive: a failed +//! upgrade *is* the liveness check, and dropping the database closes its +//! buffers, which is what wakes a consumer parked in `get()` — `aimdb-core` has +//! no explicit close. Consumers hold a [`RuntimeRef`] rather than a `Weak`, for +//! the reason given there. //! -//! - **Liveness.** A failed upgrade *is* the check — no flag to keep in sync. -//! - **Waking a blocked reader.** Dropping the database closes its buffers, -//! which is what wakes a consumer parked in `get()`. `aimdb-core` has no -//! explicit close, so nothing else would. -//! -//! An `Arc` here was tried and reverted. It bought one thing — a producer -//! outliving its handle keeps working — and cost both of the above, each of -//! which had to be rebuilt by hand (a liveness flag, a level-triggered stop -//! channel, and a select around every blocking read). It also made a forgotten -//! producer keep an OS thread and a Tokio runtime alive with nobody owning -//! them, which is the stranded thread #232 had just removed. +//! An `Arc` here was tried and reverted: it bought a producer outliving its +//! handle, cost both of the above (a liveness flag, a stop channel and a select +//! around every blocking read to rebuild), and left a forgotten producer +//! keeping an OS thread alive with nobody owning it — the stranded thread #232 +//! had just removed. use alloc::sync::Arc; @@ -67,9 +55,8 @@ pub(crate) struct Runtime { /// The fork generation this runtime's thread was spawned in. /// - /// Plain data, which is the point: every refusal path can be tested by - /// building a `Runtime` with a stale value, without a thread and without a - /// real `fork`. See the unit tests below. + /// Plain data, which is the point: every refusal path is a unit test over a + /// stale value — no thread, no `fork`. made_in: crate::fork::Generation, } @@ -84,8 +71,8 @@ impl Runtime { /// Whether this process still owns the thread this runtime names. /// - /// One relaxed atomic load and a comparison — no syscall, no lock. It sits - /// on the publish path, which is why it is not a `getpid` call. + /// One relaxed atomic load — no syscall, no lock. It sits on the publish + /// path, which is why it is not a `getpid` call. #[inline] pub(crate) fn check(&self) -> SyncResult<()> { if crate::fork::forked_since(self.made_in) { @@ -94,21 +81,18 @@ impl Runtime { Ok(()) } - /// The Tokio handle, or [`SyncError::ForkedChild`]. - /// - /// The only way to reach it. Blocking on a runtime whose thread did not - /// survive a `fork` would park forever. + /// The Tokio handle, or [`SyncError::ForkedChild`]. The only way to reach + /// it: blocking on a runtime whose thread did not survive a `fork` would + /// park forever. #[inline] pub(crate) fn enter(&self) -> SyncResult<&tokio::runtime::Handle> { self.check()?; Ok(&self.handle) } - /// The database, or [`SyncError::ForkedChild`]. - /// - /// The only way to reach it. See the module note on why the check must come - /// first: a forked child's handle to the database is valid, and that is - /// exactly the problem. + /// The database, or [`SyncError::ForkedChild`]. The only way to reach it — + /// a forked child's handle to the database is valid, which is exactly the + /// problem. #[inline] pub(crate) fn db(&self) -> SyncResult<&Arc> { self.check()?; @@ -130,22 +114,14 @@ impl Runtime { /// A borrowed view of a [`Runtime`] that outlives it on purpose. /// /// Held by [`SyncConsumer`](crate::SyncConsumer), which has a requirement the -/// handle and the producers do not: a `Reader` can still drain what is already -/// buffered after the runtime is gone, and delivering that data is behaviour -/// the characterization tests pin. So a dead runtime must not refuse a read — -/// but a `fork` still must. -/// -/// The Tokio handle is kept for that case and is **private to this module**, -/// which is the whole point of the type. `consumer.rs` cannot reach it except -/// through [`Self::enter`], so a read that skips the check cannot be written -/// there any more than it can anywhere else. A bare handle field beside a -/// hand-written guard — the shape this replaces — offered no such thing. +/// handle and producers do not: a `Reader` can still drain what is already +/// buffered after the runtime is gone, and the characterization tests pin that. +/// So a dead runtime must not refuse a read — but a `fork` still must. /// -/// The generation is copied rather than read through a -/// [`Weak`](alloc::sync::Weak): a failed upgrade is the expected case for a -/// view built to outlive its runtime, and it cannot tell a detach here — where -/// the buffer must still be drained — from a child that released its inherited -/// handle, where a blocking read would park forever. +/// Hence the generation is copied rather than reached through a +/// [`Weak`](alloc::sync::Weak), which cannot tell a detach here from a child +/// that released its inherited handle. The handle stays private to this module, +/// so `consumer.rs` reaches it only through [`Self::enter`]. pub(crate) struct RuntimeRef { handle: tokio::runtime::Handle, @@ -157,9 +133,9 @@ pub(crate) struct RuntimeRef { impl RuntimeRef { /// Refuse a forked child; let a detached one through. /// - /// A detach is not a fork — the buffer answers for itself — but a fork is - /// refused whether or not the child still holds the handle, which is what a - /// `Weak` upgrade got wrong. + /// A fork is refused whether or not the child still holds the handle — + /// what a `Weak` upgrade got wrong. A detach is not a fork; there the + /// buffer answers for itself. #[inline] pub(crate) fn check(&self) -> SyncResult<()> { if crate::fork::forked_since(self.made_in) { @@ -178,17 +154,11 @@ impl RuntimeRef { /// A resource that cannot be touched without passing the fork check. /// -/// The point is the field privacy, not the wrapper: `inner` is private to this -/// module, so a caller in `consumer.rs` has no way to reach the value except -/// through [`Self::get`] or [`Self::enter`], both of which check first. A plain -/// field beside a hand-written guard offers nothing — the guard is a call -/// someone can forget to make, and forgetting it is the defect this whole -/// design removes. -/// -/// Used for a `Reader`, which is the one thing a consumer touches that needs no -/// runtime: `try_get` reads straight out of the buffer. That is exactly why it -/// needs wrapping. Something with no resource to gate is something whose check -/// is easy to leave out. +/// The privacy is the point, not the wrapper: `inner` is unreachable from +/// `consumer.rs` except through [`Self::get`] or [`Self::enter`], which check +/// first, so the check stops being a call someone can forget. Used for a +/// `Reader` — the one thing a consumer touches that needs no runtime, and so +/// the one whose check is easiest to leave out. pub(crate) struct Guarded { rt: RuntimeRef, inner: T, @@ -221,14 +191,11 @@ mod tests { use aimdb_tokio_adapter::TokioAdapter; /// A `Runtime` over a real database, stamped `generations_behind` behind - /// the process. One behind is what a forked child's inherited runtime - /// looks like. + /// the process — one behind being what a forked child inherits. /// - /// No thread is spawned and no `fork` happens — which is the point of - /// moving the stamp onto a value. Before this, proving a refusal path meant - /// forking a real process from a parent holding a live Tokio runtime, the - /// least safe moment there is; that suite failed 11 runs in 60 until it was - /// mitigated. + /// No thread, no `fork`: the point of moving the stamp onto a value. Proving + /// a refusal used to mean forking from a parent holding a live Tokio + /// runtime, the least safe moment there is, and failed 11 runs in 60. fn runtime_stamped(generations_behind: u64) -> (tokio::runtime::Runtime, Runtime) { let tokio_rt = tokio::runtime::Builder::new_current_thread() .enable_all()