haematite 0.7.0

//! Lane-4 root-advance event seam (`docs/design/ROOT-ADVANCE-SEAM.md`, DESIGN r2).
//!
//! The engine's single push primitive: after a shard's committed root durably
//! advances, registered in-process subscribers are told
//! `(shard_id, prior_root, new_root, advance_gen)`. This is a DOORBELL, not a log
//! (design §0): correctness for every consumer lives in diff from the consumer's
//! own coverage root; the event's only obligation is to eventually fire after an
//! advance while the process lives. Nothing here touches disk, ever.
//!
//! The three opening obligations (design §1):
//! - **R1 — lock context.** Subscribers are invoked holding ZERO engine locks;
//!   the only lock held during a callback is this seam's own per-shard EMISSION
//!   mutex ([`ShardEmitState::emission`]) — a LEAF lock no engine path ever takes.
//!   The shard actor NEVER runs subscriber code: emission runs on the committer's
//!   or receiver's thread AFTER the durable commit result is in hand, never in the
//!   actor slice.
//! - **R2 — ordering.** Per-shard, delivered tells are STRICTLY gen-increasing,
//!   latest-wins; cross-shard EXPLICITLY unordered. Ordering is enforced by a
//!   MONOTONE DELIVERY FILTER ([`ShardEmitState::emission`] guards `last_told_gen`),
//!   not a turnstile: an arriving emitter with `gen <= last_told_gen` returns
//!   without telling (superseded), so a gen-regressing tell is STRUCTURALLY
//!   impossible. `advance_gen` itself is assigned in the shard's ordered actor
//!   slice (see `crate::shard::actor::native`), so gen order IS the shard's
//!   root-chain order.
//! - **R3 — no write-back.** Subscriber callbacks must not invoke engine write
//!   paths on the same [`Database`]: (a) unbounded commit->tell->commit recursion;
//!   (b) a same-thread write re-enters its own shard's emission mutex at its
//!   commit's tell — self-deadlock. Enforcement is a WALL, not advice: a
//!   thread-local in-emission flag ([`InEmissionGuard`]) is set around every
//!   callback and every shard write command refuses with a typed
//!   [`crate::db::DatabaseError::WriteDuringRootAdvanceEmission`] while it is set. Reads are
//!   permitted and expected.

use std::cell::Cell;
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex, PoisonError};

use crate::shard::commit_state::ShardCommitState;
use crate::tree::Hash;

use super::Database;

/// A durable committed-root advance for one shard (design §2).
///
/// `advance_gen` counts root ADVANCES of this shard only (`+1` iff
/// `prior_root != new_root`), assigned atomically with the root publish in the
/// shard's ordered actor slice; it is process-lifetime monotonic. Assigned gens
/// are gap-free, but DELIVERED tells strictly increase and MAY skip superseded
/// values: under latest-wins (§1 R2) a superseded advance delivers zero tells, so
/// consecutive delivered tells on a shard may skip intermediates (tell k's
/// `prior_root` need not equal tell k-1's `new_root`). Every delivered tell is
/// individually truthful — its `prior_root`/`new_root` are that commit's real
/// transition — and the highest gen can never be superseded, so the doorbell
/// guarantee holds. A consumer needing every transition is asking for a log; it
/// diffs any two roots instead (§7).
///
/// First-commit case: `prior_root` is the empty-root constant
/// ([`crate::tree::empty_root_hash`]), the baseline the first commit replaces — no
/// `Option`, no special variant; a subscriber diffs from the empty root and sees
/// the whole initial state, which is correct.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct RootAdvance {
    /// The shard whose committed root advanced.
    pub shard_id: usize,
    /// The root this commit replaced (the empty-root constant for the first).
    #[serde(with = "hash_serde")]
    pub prior_root: Hash,
    /// The durably committed root.
    #[serde(with = "hash_serde")]
    pub new_root: Hash,
    /// `+1` per root ADVANCE of this shard (iff `prior != new`), assigned
    /// atomically with the root publish; process-lifetime monotonic. Assigned gens
    /// are gap-free; DELIVERED tells strictly increase but may skip superseded
    /// values (latest-wins, §1 R2).
    pub advance_gen: u64,
}

/// The root transition an advancing shard command produced.
///
/// Carried back across the shard handle so the Database-side seam can emit AFTER
/// the durable result is in hand (never in the actor slice — R1). Crate-internal:
/// it is the reply payload the census rows (§3) converge on, not a public type.
#[derive(Debug, Clone, Copy)]
pub struct RootTransition {
    pub prior_root: Hash,
    pub new_root: Hash,
    pub advance_gen: u64,
}

/// Per-shard emission state, owned by the seam and therefore PROCESS-LIFETIME.
///
/// The `Arc` is held beside the router (not inside the actor), so it SURVIVES an
/// actor restart (design §8, §10).
///
/// §8 BINDING OBLIGATION, now RESOLVED (COMMIT-COLLAPSE landed): the `advance_gen`
/// slot has been ABSORBED into the per-shard commit-state cell
/// ([`ShardCommitState`], `crate::shard::commit_state`) — one cell, one mutex,
/// every generation (root, dirty/committed, advance) published atomically. This
/// state now carries ONLY the delivery filter. The absorb is "semantics, not
/// address": the cell's `advance_gen` is the AUTHORITATIVE value the actor slice
/// assigns and this seam emits, DISTINCT from `dirty_gen`/`committed_gen` (which
/// count all commits).
#[derive(Debug)]
pub struct ShardEmitState {
    /// The per-shard EMISSION mutex (R1/R2). No ENGINE lock is ever acquired
    /// under it and no engine path takes it; the one lock nested under it is the
    /// seam's OWN registry lock (taken by `emit` to snapshot subscribers), and
    /// that nesting is strictly one-way — subscribe/cancel take the registry
    /// lock alone and never the emission mutex, so no cycle exists. Guards
    /// `last_told_gen`, the monotone delivery filter's high-water mark.
    /// Poison-adopting (house idiom) so one subscriber's panic does not
    /// permanently wedge a shard's emissions.
    emission: Mutex<u64>,
}

impl ShardEmitState {
    const fn new() -> Self {
        Self {
            emission: Mutex::new(0),
        }
    }
}

/// One registered subscriber: its cancellation id and its callback.
struct Subscriber {
    id: u64,
    callback: Arc<dyn Fn(RootAdvance) + Send + Sync + 'static>,
}

/// The process-local subscriber registry plus the id source for RAII cancel.
#[derive(Default)]
struct Registry {
    next_id: u64,
    subscribers: Vec<Subscriber>,
}

/// The root-advance seam.
///
/// A process-local subscriber registry (a `Vec` under its own registry lock, never
/// persisted, never replicated) plus per-shard emission state. Mounted-but-unused
/// it is inert data (design §5 Q1): zero subscribers is one is-empty check per
/// advance under the emission mutex — no thread, no timer, no fd, no allocation.
#[derive(Default)]
pub struct RootAdvanceSeam {
    registry: Mutex<Registry>,
    shards: Mutex<BTreeMap<usize, Arc<ShardEmitState>>>,
    /// The per-shard COMMIT-COLLAPSE cells (`crate::shard::commit_state`), created
    /// on first access and cached here so the same `Arc` — and its `advance_gen`,
    /// `incarnation`, and dirty classification — survives actor restart. The seam
    /// is the process-lifetime per-shard registry the design §8 absorb names.
    commit_states: Mutex<BTreeMap<usize, Arc<ShardCommitState>>>,
}

impl std::fmt::Debug for RootAdvanceSeam {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("RootAdvanceSeam")
            .finish_non_exhaustive()
    }
}

impl RootAdvanceSeam {
    pub fn new() -> Arc<Self> {
        Arc::new(Self::default())
    }

    /// The per-shard emission state, created on first access and cached. The `Arc`
    /// outlives any single actor incarnation (the seam is owned by the `Database`,
    /// shared into the router), so the `advance_gen` slot survives actor restart.
    pub fn shard_state(&self, shard_id: usize) -> Arc<ShardEmitState> {
        let mut shards = lock_adopt(&self.shards);
        Arc::clone(
            shards
                .entry(shard_id)
                .or_insert_with(|| Arc::new(ShardEmitState::new())),
        )
    }

    /// The per-shard COMMIT-COLLAPSE commit-state cell, created on first access and
    /// cached (design §2.2). The `Arc` outlives any single actor incarnation (the
    /// seam is owned by the `Database`), so the cell's `advance_gen` and
    /// `incarnation` survive actor restart and the router's commit-time classify
    /// reads the SAME cell the actor writes.
    pub fn commit_state(&self, shard_id: usize) -> Arc<ShardCommitState> {
        let mut cells = lock_adopt(&self.commit_states);
        Arc::clone(cells.entry(shard_id).or_insert_with(ShardCommitState::new))
    }

    /// Register `callback`; returns its cancellation id.
    fn subscribe(&self, callback: Arc<dyn Fn(RootAdvance) + Send + Sync + 'static>) -> u64 {
        let mut registry = lock_adopt(&self.registry);
        let id = registry.next_id;
        registry.next_id = registry.next_id.wrapping_add(1);
        registry.subscribers.push(Subscriber { id, callback });
        id
    }

    /// Remove the subscriber with `id` if present (idempotent — an explicit
    /// `.cancel()` followed by `Drop` removes once).
    fn cancel(&self, id: u64) {
        lock_adopt(&self.registry)
            .subscribers
            .retain(|subscriber| subscriber.id != id);
    }

    /// Clone the current subscriber callbacks under the registry lock, then release
    /// it. The registry lock is NEVER held across a callback (only the emission
    /// mutex is — R1), so a subscribe/cancel from within a callback cannot deadlock.
    fn snapshot(&self) -> Vec<Arc<dyn Fn(RootAdvance) + Send + Sync + 'static>> {
        lock_adopt(&self.registry)
            .subscribers
            .iter()
            .map(|subscriber| Arc::clone(&subscriber.callback))
            .collect()
    }

    /// THE single emission entry — every census row (§3) converges here; the actor
    /// never calls it (R1). Runs on the committer's/receiver's thread.
    ///
    /// Under the shard's leaf emission mutex: apply the monotone delivery filter
    /// (`gen <= last_told_gen` -> return, superseded), advance `last_told_gen`
    /// BEFORE invoking (so a mid-list panic can never enable a later gen
    /// regression), snapshot the subscriber list, then invoke in order with the R3
    /// wall raised. The emission mutex is held across the callbacks — that is the
    /// per-shard delivery serialization (R2/Q4), and it is the ONLY lock a callback
    /// runs under (R1).
    pub fn emit(&self, shard_id: usize, state: &ShardEmitState, transition: RootTransition) {
        let mut last_told = lock_adopt(&state.emission);
        if transition.advance_gen <= *last_told {
            // Superseded: a tell with gen >= this one has already delivered
            // (latest-wins). A superseded advance delivers ZERO tells (§1 R2).
            return;
        }
        *last_told = transition.advance_gen;
        let subscribers = self.snapshot();
        if subscribers.is_empty() {
            // Zero-subscriber fast path (§5 Q1): the guard drops here, right after
            // its last use — no callback, no wall, no allocation beyond the snapshot.
            return;
        }
        let event = RootAdvance {
            shard_id,
            prior_root: transition.prior_root,
            new_root: transition.new_root,
            advance_gen: transition.advance_gen,
        };
        // R3 wall: any engine write attempted on THIS thread while a callback runs
        // refuses typed. The guard resets on scope exit, including panic unwind.
        let _wall = InEmissionGuard::enter();
        for callback in &subscribers {
            callback(event);
        }
        // Explicit drop AFTER the callback loop: the emission mutex is
        // deliberately held across every callback — that hold IS the per-shard
        // delivery serialization (R2/Q4) — and this explicit release point says
        // so to both the reader and clippy (whose drop-tightening suggestion of
        // an early release would break R2's serialization).
        drop(last_told);
    }
}

/// Poison-adopting lock (house idiom): a prior panic while holding the lock does
/// not permanently wedge the seam. The emission mutex is a leaf, so adopting a
/// poisoned guard cannot expose a torn invariant — the only state under it is the
/// `last_told_gen` high-water mark, already advanced before any callback ran.
fn lock_adopt<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
    mutex.lock().unwrap_or_else(PoisonError::into_inner)
}

thread_local! {
    /// True while a root-advance callback runs ON THIS THREAD. The R3 write wall
    /// reads it; the seam sets it around every callback batch.
    static IN_EMISSION: Cell<bool> = const { Cell::new(false) };
}

/// Whether the current thread is inside a root-advance callback. Read by every
/// shard write command (`crate::shard::actor::handle::ShardHandle::enqueue`) to
/// refuse re-entrant writes (R3).
pub fn in_emission() -> bool {
    IN_EMISSION.with(Cell::get)
}

/// RAII guard raising the R3 in-emission flag for the duration of a callback batch.
/// Saves and restores the prior value so the flag is exact even if emission ever
/// nested on one thread (it cannot while the wall holds, but the guard is robust).
struct InEmissionGuard {
    previous: bool,
}

impl InEmissionGuard {
    fn enter() -> Self {
        let previous = IN_EMISSION.with(|flag| flag.replace(true));
        Self { previous }
    }
}

impl Drop for InEmissionGuard {
    fn drop(&mut self) {
        IN_EMISSION.with(|flag| flag.set(self.previous));
    }
}

/// An RAII subscription to a [`Database`]'s root-advance seam.
///
/// Unsubscribes on drop and by explicit [`RootAdvanceSubscription::cancel`].
/// Registration is process-local and never persisted; a subscriber re-arms after
/// restart by reading current roots once at attach (see
/// [`Database::subscribe_root_advance`]).
#[must_use = "dropping the subscription immediately unsubscribes; keep it alive to keep receiving tells"]
pub struct RootAdvanceSubscription {
    seam: Arc<RootAdvanceSeam>,
    id: u64,
    active: bool,
}

impl RootAdvanceSubscription {
    /// Explicitly unsubscribe now. Idempotent; a later drop is a no-op. Because
    /// emission is snapshot-then-invoke and serializes PER SHARD, a subscriber may
    /// receive at most one tell PER SHARD after `cancel` returns (a cancel landing
    /// while several shards are mid-snapshot is followed by at most one late tell
    /// from each — the bound is per shard, never one total; design §4). The
    /// synchronous-barrier alternative buys a lock-coupling this seam refuses.
    pub fn cancel(mut self) {
        self.deactivate();
    }

    fn deactivate(&mut self) {
        if self.active {
            self.active = false;
            self.seam.cancel(self.id);
        }
    }
}

impl std::fmt::Debug for RootAdvanceSubscription {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("RootAdvanceSubscription")
            .field("id", &self.id)
            .field("active", &self.active)
            .finish_non_exhaustive()
    }
}

impl Drop for RootAdvanceSubscription {
    fn drop(&mut self) {
        self.deactivate();
    }
}

impl Database {
    /// Subscribe to this database's committed-root advances.
    ///
    /// The callback is invoked `(shard_id, prior_root, new_root, advance_gen)` after
    /// a shard's committed root DURABLY advances, on the committer's or receiver's
    /// thread, holding ZERO engine locks (R1). Returns an RAII
    /// [`RootAdvanceSubscription`] that unsubscribes on drop.
    ///
    /// # Contract (design §1–§4)
    /// - The callback MUST NOT invoke engine WRITE paths on this same `Database`
    ///   (put/delete/append/cas/commit/merge): they refuse with
    ///   [`crate::db::DatabaseError::WriteDuringRootAdvanceEmission`] while a callback runs
    ///   (R3). READS (get/range/diff/checkout) are permitted and are precisely how
    ///   a subscriber acts on a tell. Do real work on YOUR OWN executor; the
    ///   callback should record-and-return.
    /// - A panicking callback surfaces on the emitting thread per normal Rust (the
    ///   engine adds no `catch_unwind`); it STARVES every LATER subscriber in that
    ///   one snapshot of that one tell, but the emission mutex stays healthy and the
    ///   next advance tells everyone (design §4).
    /// - Delivery is LATEST-WINS per shard: a superseded advance delivers zero
    ///   tells; the highest gen always delivers. Cross-shard tells are UNORDERED.
    ///
    /// # Attach protocol (loss-free)
    /// Subscribe FIRST, then read current roots, then diff-catch-up: an advance
    /// between subscribe and read is either in the read or arrives as a tell; both
    /// converge by diff. On each tell, read current roots and diff from YOUR OWN
    /// coverage root — never treat the tell's `new_root` as a sync target (under
    /// latest-wins plus the accepted starvation edges it can lag the shard's head).
    pub fn subscribe_root_advance(
        &self,
        callback: impl Fn(RootAdvance) + Send + Sync + 'static,
    ) -> RootAdvanceSubscription {
        let seam = self.seam();
        let id = seam.subscribe(Arc::new(callback));
        RootAdvanceSubscription {
            seam: Arc::clone(seam),
            id,
            active: true,
        }
    }
}

/// The remedy text carried in [`crate::db::DatabaseError::WriteDuringRootAdvanceEmission`]
/// (R3): hand the work to your own executor/queue rather than writing back inline.
pub const WRITE_DURING_EMISSION_REMEDY: &str = "a root-advance subscriber callback attempted an engine write on the same \
     Database; write-back from a callback is refused (it would recurse \
     commit->tell->commit and self-deadlock on the shard's emission mutex). \
     Remedy: record the tell and hand the write to your own executor/queue; \
     reads (get/range/diff/checkout) are permitted inside a callback";

#[cfg(test)]
#[path = "root_advance_tests.rs"]
mod tests;

/// serde for the [`Hash`] fields of [`RootAdvance`]. The seam never serializes on
/// its own (it is process-local, §0/§7); this exists only so an embedder that
/// BRIDGES tells to its own transport under its own contract (§7) can, honoring the
/// design's `+ serde` on the event type without adding a serde impl to the core
/// [`Hash`].
mod hash_serde {
    use crate::tree::Hash;
    use serde::{Deserialize, Deserializer, Serialize, Serializer};

    pub(super) fn serialize<S: Serializer>(hash: &Hash, serializer: S) -> Result<S::Ok, S::Error> {
        hash.as_bytes().serialize(serializer)
    }

    pub(super) fn deserialize<'de, D: Deserializer<'de>>(
        deserializer: D,
    ) -> Result<Hash, D::Error> {
        let bytes = <[u8; crate::tree::node::HASH_SIZE]>::deserialize(deserializer)?;
        Ok(Hash::from_bytes(bytes))
    }
}