feat(sync): tell a forked child the truth about its inherited handles

Makefile

@@ -20,6 +20,13 @@ GREEN := \033[0;32m
 YELLOW := \033[0;33m
 BLUE := \033[0;34m
 RED := \033[0;31m
+
+# Crates that must never appear in aimdb-sync's --no-default-features graph.
+# `tokio` is the obvious one; `libc` is here because it defaults to a `std`
+# feature, so a target-specific dependency added for a std-only path (the
+# pthread_atfork fork detector) silently un-no_std's the crate if it is not
+# marked optional and gated behind `std`.
+SYNC_NO_STD_FORBIDDEN := tokio|libc
 NC := \033[0m # No Color
 
 ## Show available commands
@@ -89,16 +96,16 @@ build:
 	cargo build --package aimdb-sync
 	@printf "$(YELLOW)  → Building sync wrapper (no_std)$(NC)\n"
 	cargo build --package aimdb-sync --no-default-features
-	@printf "$(YELLOW)  → Asserting no tokio in sync wrapper (no_std)$(NC)\n"
+	@printf "$(YELLOW)  → Asserting no std-only crates in sync wrapper (no_std)$(NC)\n"
 	@out=$$(cargo tree -p aimdb-sync --no-default-features -e features,no-dev 2>&1) || { \
 		printf "$(RED)✗ cargo tree failed — refusing to pass vacuously:$(NC)\n"; \
 		printf '%s\n' "$$out"; exit 1; \
 	}; \
-	if printf '%s\n' "$$out" | grep -qi tokio; then \
-		printf "$(RED)✗ tokio leaked into the no_std build$(NC)\n"; \
-		printf '%s\n' "$$out" | grep -i tokio; exit 1; \
+	if printf '%s\n' "$$out" | grep -qiE '$(SYNC_NO_STD_FORBIDDEN)'; then \
+		printf "$(RED)✗ a std-only crate leaked into the no_std build$(NC)\n"; \
+		printf '%s\n' "$$out" | grep -iE '$(SYNC_NO_STD_FORBIDDEN)'; exit 1; \
 	fi
-	@printf "$(BLUE)✓ no_std graph is tokio-free$(NC)\n"
+	@printf "$(BLUE)✓ no_std graph is free of $(SYNC_NO_STD_FORBIDDEN)$(NC)\n"
 	@printf "$(YELLOW)  → Building codegen library$(NC)\n"
 	cargo build --package aimdb-codegen
 	@printf "$(YELLOW)  → Building CLI tools$(NC)\n"

aimdb-sync/CHANGELOG.md

@@ -30,6 +30,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### 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.
+  `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
+  publish path is one relaxed atomic load, and a program that never attaches
+  never installs a handler. A database the child attaches *itself* after
+  forking is unaffected — the guard is a generation counter, not a poison flag.
+  The detection is entirely internal: a facade built on this crate will have the
+  same problem for the same reason, but none exists yet, so exposing the
+  stamp-and-compare pair would commit the crate in semver to a model chosen
+  against no real caller.
 - **A panic-freedom contract on the blocking surface.** The crate is compiled
   under `deny(clippy::unwrap_used, clippy::expect_used, clippy::panic)` outside
   its own tests, so "a panic here is a bug, not an error channel" is checked

aimdb-sync/Cargo.toml

@@ -44,14 +44,20 @@ serde_json = "1.0"
 [features]
 default = ["std"]
 
-std = ["aimdb-core/std", "dep:tokio", "dep:aimdb-tokio-adapter"]
+std = ["aimdb-core/std", "dep:tokio", "dep:aimdb-tokio-adapter", "dep:libc"]
 
 # Enable tracing for debugging
 tracing = ["dep:tracing", "aimdb-core/tracing"]
 
 # SyncProducer::set_value / try_set_value / set_value_at for Settable types
 data-contracts = ["dep:aimdb-data-contracts"]
 
+# Fork detection only, and only where `pthread_atfork` exists. Optional and
+# activated by `std`, so the no_std build pulls neither libc nor its `std`
+# feature — see `src/fork.rs`.
+[target.'cfg(unix)'.dependencies]
+libc = { version = "0.2", default-features = false, optional = true }
+
 [package.metadata.docs.rs]
 all-features = true
 rustdoc-args = ["--cfg", "docsrs"]

aimdb-sync/src/consumer.rs

@@ -58,6 +58,8 @@ where
 {
     waiter: Waiter,
     reader: Reader<T>,
+    /// The fork generation this consumer was made in. See [`crate::fork`].
+    made_in: crate::fork::Generation,
 }
 
 impl<T> SyncConsumer<T>
@@ -66,7 +68,23 @@ where
 {
     /// Create a new sync consumer (internal use only)
     pub(crate) fn new(waiter: Waiter, reader: Reader<T>) -> Self {
-        Self { waiter, reader }
+        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);
+        }
+        Ok(())
     }
 
     async fn get_impl(reader: &mut Reader<T>) -> SyncResult<T> {
@@ -114,6 +132,7 @@ where
     /// # }
     /// ```
     pub fn get(&mut self) -> SyncResult<T> {
+        self.check_fork()?;
         self.waiter.block_on(Self::get_impl(&mut self.reader))
     }
 
@@ -157,6 +176,7 @@ 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);
         res.unwrap_or_else(|_| Err(SyncError::GetTimeout))
@@ -198,6 +218,7 @@ where
     /// # }
     /// ```
     pub fn try_get(&mut self) -> SyncResult<T> {
+        self.check_fork()?;
         let res = self.reader.try_recv();
         res.map_err(|e| match e {
             DbError::BufferClosed { .. } => SyncError::RuntimeShutdown,
@@ -248,6 +269,7 @@ 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
@@ -299,6 +321,7 @@ 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))?;

aimdb-sync/src/error.rs

@@ -40,6 +40,11 @@ pub enum SyncError {
     #[error("Runtime thread has shut down")]
     RuntimeShutdown,
 
+    /// This handle, producer or consumer was created before a `fork()`, and
+    /// this is the child. The runtime thread it needs did not survive.
+    #[error("created before a fork(); this process has no runtime thread for it")]
+    ForkedChild,
+
     /// Error from the underlying database.
     #[error(transparent)]
     Db(#[from] DbError),
@@ -62,7 +67,9 @@ impl SyncError {
 
             Self::SetTimeout | Self::GetTimeout => DbErrorKind::Retry,
 
-            Self::RuntimeShutdown => DbErrorKind::Closed,
+            // Terminal for the same reason RuntimeShutdown is: the runtime
+            // thread is gone and will not come back in this process.
+            Self::RuntimeShutdown | Self::ForkedChild => DbErrorKind::Closed,
 
             Self::Db(err) => err.kind(),
         }
@@ -89,6 +96,9 @@ mod tests {
         assert_eq!(SyncError::GetTimeout.kind(), DbErrorKind::Retry);
         assert_eq!(SyncError::SetTimeout.kind(), DbErrorKind::Retry);
         assert_eq!(SyncError::RuntimeShutdown.kind(), DbErrorKind::Closed);
+        // Terminal for the same reason: the runtime thread is gone and will
+        // not come back in this process, so a caller must not retry.
+        assert_eq!(SyncError::ForkedChild.kind(), DbErrorKind::Closed);
     }
 
     /// The point of returning `DbErrorKind` rather than a kind of this crate's

aimdb-sync/src/fork.rs

@@ -0,0 +1,188 @@
+//! Making a `fork()` visible to handles created before it.
+//!
+//! `fork` copies the address space but not the threads, so a child inherits
+//! every [`AimDbHandle`](crate::AimDbHandle), [`SyncProducer`](crate::SyncProducer)
+//! and [`SyncConsumer`](crate::SyncConsumer) the parent held — and none of the
+//! runtime thread that makes them work. Without the check below, the child's
+//! `set()` pushes into a buffer nobody drains and returns `Ok`.
+//!
+//! # Why a generation counter and not a flag
+//!
+//! A `bool` would poison the child permanently, including for a database the
+//! *child itself* attaches afterwards. A counter makes staleness relative: a
+//! handle is unusable when the process has forked since it was made, and one
+//! made after the fork is fine.
+//!
+//! # Why `pthread_atfork` and not `getpid`
+//!
+//! Because this sits on the publish path. Measured: `try_set` is 121 ns and
+//! `std::process::id()` is 321 ns, so reading the pid per call would cost more
+//! than twice the work it guards. A relaxed atomic load does not measurably
+//! cost anything.
+//!
+//! # The process-global caveat
+//!
+//! Registering a `pthread_atfork` handler is a process-wide decision, and a
+//! library making one on an application's behalf is normally a trespass. This
+//! is the exception: only the crate that owns the runtime thread knows the
+//! thread is gone, so nobody above can make this check — and the handler is
+//! registered lazily, on the first [`generation`] (which every `attach` takes),
+//! so a program that never uses the sync facade never gets one.
+
+use core::sync::atomic::{AtomicU64, Ordering};
+
+/// How many times this process has forked, as observed by the child. Parents
+/// never see it change.
+static GENERATION: AtomicU64 = AtomicU64::new(0);
+
+/// The generation a handle, producer or consumer was created in.
+///
+/// Compared, never interpreted: only equality with [`generation`] means
+/// anything.
+pub(crate) type Generation = u64;
+
+#[cfg(unix)]
+extern "C" fn on_fork_in_child() {
+    GENERATION.fetch_add(1, Ordering::Relaxed);
+}
+
+/// Register the fork handler, once per process.
+///
+/// Called from the constructors and from [`generation`] rather than a static
+/// initialiser, so a program that never uses the sync facade never installs a
+/// handler. Idempotent and cheap after the first call — a completed
+/// [`Once`](std::sync::Once) is one acquire load — which is why [`generation`]
+/// can afford to call it and [`forked_since`] does not have to.
+pub(crate) fn arm() {
+    #[cfg(unix)]
+    {
+        static ONCE: std::sync::Once = std::sync::Once::new();
+        ONCE.call_once(|| {
+            // SAFETY: `on_fork_in_child` is `extern "C"`, does not unwind, and
+            // performs one relaxed atomic add — permitted in a fork handler.
+            unsafe {
+                libc::pthread_atfork(None, None, Some(on_fork_in_child));
+            }
+        });
+    }
+}
+
+/// Read the counter. Never arms — see [`generation`] for why that split
+/// exists.
+#[inline]
+fn load() -> Generation {
+    GENERATION.load(Ordering::Relaxed)
+}
+
+/// The generation to stamp on something being created now.
+///
+/// Crate-private for now. A layer built *on* this crate has the same problem —
+/// an FFI door holds state of its own that a `fork` invalidates — but no such
+/// layer exists yet, and exposing this would commit us in semver to the
+/// stamp-and-compare model. Widen it when something real needs it, so the
+/// shape can be chosen against that caller rather than guessed at.
+///
+/// **Arms the handler**, because otherwise it hands out a number that cannot
+/// change. A caller above this crate stamps its own state before any database
+/// is attached — that is the normal order, an FFI door opens before it is used
+/// — and until the first `attach` there is no handler, so a `fork` in that
+/// window would go uncounted and the stamp would compare equal forever. That is
+/// the very bug this module exists to prevent, one layer up. Arming here closes
+/// it: taking a stamp is what makes the stamp meaningful.
+///
+/// This is the *construction-time* call and is cold. The hot path is
+/// [`forked_since`], which only loads.
+pub(crate) fn generation() -> Generation {
+    arm();
+    load()
+}
+
+/// Whether this process has forked since `made_in` was taken from
+/// [`generation`].
+///
+/// One relaxed load and a comparison — no syscall, no lock, and no arming: if
+/// `made_in` exists then [`generation`] already armed the handler. Safe to call
+/// from anywhere, including while the runtime thread is mid-shutdown.
+pub(crate) fn forked_since(made_in: Generation) -> bool {
+    load() != made_in
+}
+
+#[cfg(all(test, unix))]
+mod tests {
+    use super::*;
+    use std::time::{Duration, Instant};
+
+    /// A stamp taken before any `attach` must still be invalidated by a fork.
+    ///
+    /// The handler is armed lazily, so something has to trigger it. If arming
+    /// were left to the first `attach`, a stamp taken earlier would compare
+    /// equal forever and a `fork` in that window would go uncounted — the
+    /// silent-success failure this module exists to prevent, one layer up.
+    ///
+    /// This is a unit test rather than an integration test for two reasons.
+    /// [`generation`] is crate-private. And the precondition is that *nothing*
+    /// in the process has armed the handler yet: the lib test binary holds only
+    /// the compile-time `assert_send`/`assert_sync` checks and the
+    /// `SyncError::kind` tests, none of which attach a database, whereas any
+    /// binary sharing space with the fork suite would be armed by its first
+    /// `attach` and this test would then assert nothing.
+    #[test]
+    fn a_stamp_taken_before_any_attach_still_sees_a_fork() {
+        let before_any_attach = generation();
+        assert!(!forked_since(before_any_attach), "nothing has forked yet");
+
+        // SAFETY: the child reads one atomic and `_exit`s. It never allocates,
+        // so the usual "child deadlocks on an allocator lock inherited from a
+        // thread that did not survive the fork" hazard does not arise.
+        let pid = unsafe { libc::fork() };
+        assert_ne!(pid, -1, "fork failed");
+
+        if pid == 0 {
+            let saw_it = forked_since(before_any_attach);
+            // SAFETY: ends the child without running destructors, by design.
+            unsafe { libc::_exit(if saw_it { 0 } else { 1 }) }
+        }
+
+        let status = wait_briefly(pid);
+        assert!(libc::WIFEXITED(status), "child did not exit normally");
+        assert_eq!(
+            libc::WEXITSTATUS(status),
+            0,
+            "a stamp taken before the first attach must still see the fork"
+        );
+
+        // Only the child's counter moved; the parent is never poisoned.
+        assert!(
+            !forked_since(before_any_attach),
+            "the parent must not be poisoned by forking a child"
+        );
+    }
+
+    /// Reap `pid`, but fail rather than block forever.
+    ///
+    /// The child above cannot deadlock, but "cannot" is what was said about the
+    /// fork suite before it hung a CI job for six hours. A bounded wait costs
+    /// nothing and keeps that failure mode impossible here by construction.
+    fn wait_briefly(pid: libc::pid_t) -> libc::c_int {
+        let deadline = Instant::now() + Duration::from_secs(30);
+        loop {
+            let mut status: libc::c_int = 0;
+            // SAFETY: `pid` is our child and `status` is a valid out-pointer.
+            let waited = unsafe { libc::waitpid(pid, &mut status, libc::WNOHANG) };
+            if waited == pid {
+                return status;
+            }
+            assert_eq!(waited, 0, "waitpid failed");
+            if Instant::now() >= deadline {
+                // SAFETY: `pid` is our child; SIGKILL cannot be blocked.
+                unsafe {
+                    libc::kill(pid, libc::SIGKILL);
+                    let mut discard: libc::c_int = 0;
+                    libc::waitpid(pid, &mut discard, 0);
+                }
+                panic!("forked child did not finish within 30s");
+            }
+            std::thread::sleep(Duration::from_millis(5));
+        }
+    }
+}