diff --git a/aimdb-sync/CHANGELOG.md b/aimdb-sync/CHANGELOG.md index e0335944..c35eebd7 100644 --- a/aimdb-sync/CHANGELOG.md +++ b/aimdb-sync/CHANGELOG.md @@ -46,13 +46,40 @@ 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 + 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. 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/consumer.rs b/aimdb-sync/src/consumer.rs index f8b4fc5e..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::waiter::Waiter; +use crate::runtime::{Guarded, RuntimeRef}; use crate::{SyncError, SyncResult}; use core::fmt::Debug; use core::time::Duration; @@ -56,10 +56,13 @@ pub struct SyncConsumer where T: Send + Debug + Clone, { - waiter: Waiter, - reader: Reader, - /// The fork generation this consumer was made in. See [`crate::fork`]. - made_in: crate::fork::Generation, + /// The subscription, and with it the runtime. + /// + /// 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 @@ -67,24 +70,10 @@ where T: Send + Debug + Clone, { /// Create a new sync consumer (internal use only) - pub(crate) fn new(waiter: Waiter, reader: Reader) -> Self { + pub(crate) fn new(rt: RuntimeRef, reader: Reader) -> Self { Self { - waiter, - reader, - made_in: crate::fork::generation(), - } - } - - /// 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. - #[inline] - fn check_fork(&self) -> SyncResult<()> { - if crate::fork::forked_since(self.made_in) { - return Err(SyncError::ForkedChild); + reader: Guarded::new(rt, reader), } - Ok(()) } async fn get_impl(reader: &mut Reader) -> SyncResult { @@ -132,8 +121,8 @@ where /// # } /// ``` pub fn get(&mut self) -> SyncResult { - self.check_fork()?; - self.waiter.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. @@ -176,9 +165,9 @@ where /// # } /// ``` pub fn get_with_timeout(&mut self, timeout: Duration) -> SyncResult { - self.check_fork()?; - let fut = async { tokio::time::timeout(timeout, Self::get_impl(&mut self.reader)).await }; - let res = self.waiter.block_on(fut); + 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)) } @@ -218,8 +207,7 @@ where /// # } /// ``` pub fn try_get(&mut self) -> SyncResult { - self.check_fork()?; - 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, @@ -269,7 +257,6 @@ where /// # } /// ``` pub fn get_latest(&mut self) -> SyncResult { - self.check_fork()?; // 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 +308,6 @@ where /// # } /// ``` pub fn get_latest_with_timeout(&mut self, timeout: Duration) -> SyncResult { - self.check_fork()?; // 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..7611513c 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`]. + /// + /// The only strong reference: producers and consumers hold a + /// [`Weak`](alloc::sync::Weak), so the runtime dies with this handle and a + /// 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 + /// 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 dropping it: the thread also + /// 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. /// @@ -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`. 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,17 @@ impl AimDbHandle { where T: Send + 'static + Debug + Clone, { - self.check_fork()?; - Ok(crate::SyncProducer::new(Arc::downgrade(&self.db), key)) + // 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. + // + // `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)) } /// Create a synchronous consumer for type `T`. @@ -371,11 +382,14 @@ impl AimDbHandle { where T: Send + Sync + 'static + Debug + Clone, { - self.check_fork()?; 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)) + // `db()` checks, so subscribing is gated the same way publishing is. + let reader = self + .rt + .db()? + .subscribe::(&record_key) + .map_err(SyncError::Db)?; + Ok(crate::SyncConsumer::new(self.rt.view()?, reader)) } /// Gracefully shut down the runtime thread. @@ -447,61 +461,65 @@ impl AimDbHandle { } /// Internal detach implementation. + /// + /// 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) + /// 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` // 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 +595,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..db96d928 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,32 @@ 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. + /// + /// # 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, - /// 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 +76,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 +125,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 +163,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..ad9c4eff --- /dev/null +++ b/aimdb-sync/src/runtime.rs @@ -0,0 +1,307 @@ +//! 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. +//! +//! # Why the check lives on the way in +//! +//! [`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 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 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` 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. +//! +//! 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; + +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 is a unit test over a + /// stale value — no thread, no `fork`. + 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 — 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 — + /// 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()?; + Ok(&self.db) + } + + /// A view of this runtime that may outlive it. The only way to build one. + /// + /// 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(), + made_in: self.made_in, + }) + } +} + +/// A borrowed view of a [`Runtime`] that outlives it on purpose. +/// +/// Held by [`SyncConsumer`](crate::SyncConsumer), which has a requirement the +/// 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. +/// +/// 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, + + /// 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 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) { + return Err(SyncError::ForkedChild); + } + Ok(()) + } + + /// The Tokio handle, checked. The only way to obtain one. + #[inline] + pub(crate) fn enter(&self) -> SyncResult { + self.check()?; + Ok(self.handle.clone()) + } +} + +/// A resource that cannot be touched without passing the fork check. +/// +/// 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, +} + +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::*; + use aimdb_core::AimDbBuilder; + use aimdb_tokio_adapter::TokioAdapter; + + /// A `Runtime` over a real database, stamped `generations_behind` behind + /// the process — one behind being what a forked child inherits. + /// + /// 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() + .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); + } + + /// 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 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); + 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 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, rt) = runtime_stamped(0); + let mut view = rt.view().expect("view"); + + // Forked after the view was taken — the order that matters. + view.made_in = crate::fork::generation().wrapping_sub(1); + drop(rt); + + assert!(matches!(view.check(), Err(SyncError::ForkedChild))); + assert!(matches!(view.enter(), Err(SyncError::ForkedChild))); + } + + /// 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("reads before a fork"), 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/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..7c4c0e68 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. 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 @@ -142,17 +84,45 @@ 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 created — and allocated in the parent, where that 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)); + // 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) + ); + // 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 + // 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); + + refused && producer_defers && no_consumer && still_refused }); assert_eq!(code, 0, "detach in a child should be refused, not fatal"); }