refactor(sync): make the fork check unavoidable by construction

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

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,35 +56,24 @@ pub struct SyncConsumer<T>
 where
     T: Send + Debug + Clone,
 {
-    waiter: Waiter,
-    reader: Reader<T>,
-    /// 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<Reader<T>>,
 }
 
 impl<T> SyncConsumer<T>
 where
     T: Send + Debug + Clone,
 {
     /// Create a new sync consumer (internal use only)
-    pub(crate) fn new(waiter: Waiter, reader: Reader<T>) -> Self {
+    pub(crate) fn new(rt: RuntimeRef, reader: Reader<T>) -> 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<T>) -> SyncResult<T> {
@@ -132,8 +121,8 @@ where
     /// # }
     /// ```
     pub fn get(&mut self) -> SyncResult<T> {
-        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<T> {
-        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<T> {
-        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<T> {
-        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<T> {
-        self.check_fork()?;
         // see internal comments for get_latest
         let deadline = Instant::now() + timeout;
         let oldest = self.get_catch_up(Some(deadline))?;