haematite 0.7.0

//! COMMIT-COLLAPSE §2.2: the per-shard commit-state cell.
//!
//! A shard is DIRTY iff its `WalBuffer` holds mutations a tree commit has not yet
//! applied (design §2.1). The classification is published by the shard's serial
//! command execution (and, at spawn/restart, by the factory) into ONE coherent
//! [`CommitSnapshot`] under a single mutex — never a bag of atomics, so every read
//! is a consistent snapshot and every transition is atomic. Global commit
//! (`api/kv.rs`, COMMIT-COLLAPSE §5) reads the cell to address only dirty shards
//! while still returning the complete, byte-identical `ShardRoots` vector.
//!
//! ## Where the write happens
//! The cell is published from the shard's ordered native command loop
//! (`crate::shard::actor::native`), one publish per command, immediately after the
//! command mutated (or left untouched) the buffer — so from the cell's view every
//! buffer mutation is paired with a cell publish inside the same actor slice, with
//! no other command interleaving. The cell lock is a LEAF: nothing else is
//! acquired while it is held, on any path (the critical sections are pure field
//! writes), so it can never participate in a deadlock.
//!
//! ## Poison FAILS CLOSED (design §2.2, ⟨r3⟩)
//! A reader of a poisoned cell classifies the shard DIRTY (exactly like
//! `recovering`) and lets the actor path resolve it; no path ever substitutes a
//! default/clean snapshot on error. The actor's own next write recovers the
//! poisoned guard, overwrites the complete snapshot, and clears the poison
//! ([`Mutex::clear_poison`]) — writing through a recovered guard does not clear
//! std poison by itself.
//!
//! ## `advance_gen` — the ROOT-ADVANCE-SEAM §8 absorb
//! The cell carries `advance_gen`, the seam's per-shard advance counter (§8 binding
//! cross-lane obligation): bumped by exactly one iff `prior_root != new_root`,
//! atomically with the snapshot publish, and DISTINCT from
//! `dirty_gen`/`committed_gen` (which count all commits). This is the authoritative
//! value the root-advance seam emits — the cell ABSORBS the seam-owned per-shard
//! slot (semantics, not address). Because the cell is process-lifetime (held via
//! the seam), `advance_gen` survives actor restarts and stays monotone across
//! incarnations without persisting anything.

use std::sync::{Arc, Mutex, PoisonError};

use crate::tree::Hash;

/// One coherent, actor-written snapshot of a shard's commit classification.
///
/// `Copy` so a reader takes a cheap detached snapshot and releases the lock
/// immediately (no reference escapes the critical section).
#[derive(Clone, Copy, Debug)]
pub struct CommitSnapshot {
    /// The shard's current committed root; `None` only for a marker-less shard
    /// (fresh, or promise-only before its first commit).
    pub root: Option<Hash>,
    /// Monotonic pair with `committed_gen`: the shard is CLEAN iff the two are
    /// equal. `u64` overflow is unreachable (10⁹ mutations/s for 584 years).
    pub dirty_gen: u64,
    /// Advanced to `dirty_gen` by a successful commit (clean ⟺ `committed_gen ==
    /// dirty_gen`).
    pub committed_gen: u64,
    /// Bumped by each (re)spawn of the actor. DIAGNOSTIC-ONLY (design §2.2 ⟨r6⟩):
    /// no runtime decision reads it — its consumers are the restart tests (which
    /// assert reinitialisation by observing the bump) and operator debugging.
    pub incarnation: u64,
    /// True from the start of the restart protocol until the recovered snapshot is
    /// published; a reader that observes it classifies the shard DIRTY.
    pub recovering: bool,
    /// Set when a failure discarded buffer state without completing its publication
    /// (`merge_adopt` post-discard failure, §3); forces the full reconciling commit
    /// path regardless of buffer emptiness, and is never fast-pathed.
    pub unreconciled: bool,
    /// The root-advance counter (ROOT-ADVANCE-SEAM §8): `+1` per root ADVANCE of
    /// this shard, DISTINCT from the dirty/committed pair. Never reset across
    /// incarnations.
    pub advance_gen: u64,
}

/// The commit-time classification of a shard, read by global commit under the
/// router snapshot (COMMIT-COLLAPSE §5 step 2).
///
/// `dirty` drives the O(dirty) fan-out (only dirty shards are sent a `Commit`);
/// `root` fills a CLEAN shard's slot in the returned `ShardRoots` vector without
/// touching the actor. A dirty shard's returned root comes from its `Commit`
/// reply instead, so `root` here is only load-bearing for the clean case (where
/// the classification guarantees it is `Some`).
#[derive(Clone, Copy, Debug)]
pub struct Classification {
    /// DIRTY iff recovering || unreconciled || gens differ || root is None (and
    /// unconditionally on a poisoned cell — fail closed, §2.2).
    pub dirty: bool,
    /// The shard's committed root as the cell last published it — the value used
    /// to fill a clean shard's slot. `Some` whenever `dirty` is false.
    pub root: Option<Hash>,
}

impl CommitSnapshot {
    /// The fresh snapshot for a never-spawned shard: marker-less, clean-by-gens,
    /// zero incarnation. The restart protocol overwrites everything but
    /// `advance_gen` before the actor's first command.
    const fn initial() -> Self {
        Self {
            root: None,
            dirty_gen: 0,
            committed_gen: 0,
            incarnation: 0,
            recovering: false,
            unreconciled: false,
            advance_gen: 0,
        }
    }
}

/// The per-shard commit-state cell: one mutex-held [`CommitSnapshot`].
///
/// Process-lifetime (held via the seam's per-shard map), so the same `Arc` — and
/// therefore the same `advance_gen`/`incarnation` — is observed across every actor
/// incarnation of one shard id.
#[derive(Debug)]
pub struct ShardCommitState(Mutex<CommitSnapshot>);

impl ShardCommitState {
    /// A fresh cell wrapped in the process-lifetime `Arc` the seam caches.
    pub fn new() -> Arc<Self> {
        Arc::new(Self(Mutex::new(CommitSnapshot::initial())))
    }

    /// Run `edit` under the leaf lock, ADOPTING a poisoned guard, then clear the
    /// poison — the actor's next write is what heals a cell a subscriber/command
    /// panic left poisoned (design §2.2 ⟨r4⟩). The critical section is a pure field
    /// write, so adopting a poisoned guard can never expose a torn invariant.
    fn write<R>(&self, edit: impl FnOnce(&mut CommitSnapshot) -> R) -> R {
        let mut guard = self.0.lock().unwrap_or_else(PoisonError::into_inner);
        let out = edit(&mut guard);
        drop(guard);
        self.0.clear_poison();
        out
    }

    /// Set the dirty/clean gens to match the buffer's actual emptiness: CLEAN iff
    /// the buffer is empty. Leaves the root, incarnation, and `advance_gen` alone.
    const fn set_dirty(snapshot: &mut CommitSnapshot, buffer_empty: bool) {
        if buffer_empty {
            snapshot.committed_gen = snapshot.dirty_gen;
        } else if snapshot.committed_gen == snapshot.dirty_gen {
            snapshot.dirty_gen = snapshot.committed_gen.wrapping_add(1);
        }
    }

    /// Restart protocol step 1 (§2.2 M2): mark recovering and bump the incarnation
    /// BEFORE recovery runs, so a reader during rebuild classifies the shard dirty
    /// and the restart tests observe the reinitialisation.
    pub fn begin_recovery(&self) {
        self.write(|snapshot| {
            snapshot.recovering = true;
            snapshot.incarnation = snapshot.incarnation.wrapping_add(1);
        });
    }

    /// Restart protocol step 3 (§2.2 M2): publish the complete recovered snapshot
    /// before the actor accepts its first command. Dirty iff the recovered buffer
    /// is non-empty (design §2.1 recovery row). `advance_gen` is PRESERVED — the gen
    /// sequence continues across incarnations, never resets.
    pub fn publish_recovered(&self, root: Option<Hash>, buffer_empty: bool) {
        self.write(|snapshot| {
            snapshot.root = root;
            snapshot.recovering = false;
            snapshot.unreconciled = false;
            Self::set_dirty(snapshot, buffer_empty);
        });
    }

    /// Republish the classification the buffer's actual content implies (design
    /// §2.1): CLEAN iff `buffer_empty`. This is the transactional inverse the
    /// rollback/restore rows need — a once-failed op that restores an empty buffer
    /// returns the shard to clean — and the dirty-origin publish for a buffered
    /// put/delete/TTL-sweep (which leaves a non-empty buffer). The root is not
    /// touched: a non-committing path never changes `committed_root`.
    pub fn sync_from_buffer(&self, buffer_empty: bool) {
        self.write(|snapshot| Self::set_dirty(snapshot, buffer_empty));
    }

    /// Publish a successful commit (design §3): the shard is CLEAN at `root`. If the
    /// root ADVANCED, bump `advance_gen` by one — atomically with this publish — and
    /// return it (the gen the root-advance seam tells with); otherwise return `None`
    /// (a non-advancing commit consumes no gen and reaches no subscriber).
    pub fn publish_committed(&self, root: Option<Hash>, advanced: bool) -> Option<u64> {
        self.write(|snapshot| {
            snapshot.root = root;
            snapshot.committed_gen = snapshot.dirty_gen;
            snapshot.recovering = false;
            snapshot.unreconciled = false;
            if advanced {
                snapshot.advance_gen = snapshot.advance_gen.wrapping_add(1);
                Some(snapshot.advance_gen)
            } else {
                None
            }
        })
    }

    /// Classify the shard for global commit (COMMIT-COLLAPSE §5 step 2), FAILING
    /// CLOSED: DIRTY iff `recovering || unreconciled || committed_gen != dirty_gen
    /// || root is None`, and a POISONED cell classifies DIRTY unconditionally —
    /// never toward "clean" (§2.2 the sealed law). This is the READER half of the
    /// poison law: it ADOPTS a poisoned guard to read the (stale) root for the
    /// dirty result but does NOT clear the poison — only the actor's next WRITE
    /// heals it (`write`). A false-DIRTY costs one actor message; a false-CLEAN is
    /// an unrecoverable correctness failure, so the asymmetry is deliberate.
    pub fn classify(&self) -> Classification {
        match self.0.lock() {
            Ok(snapshot) => Self::classify_snapshot(&snapshot, false),
            Err(poison) => Self::classify_snapshot(&poison.into_inner(), true),
        }
    }

    /// The pure classification predicate over one snapshot; `poisoned` forces
    /// DIRTY (fail closed) without touching the poison flag itself.
    const fn classify_snapshot(snapshot: &CommitSnapshot, poisoned: bool) -> Classification {
        Classification {
            dirty: poisoned
                || snapshot.recovering
                || snapshot.unreconciled
                || snapshot.committed_gen != snapshot.dirty_gen
                || snapshot.root.is_none(),
            root: snapshot.root,
        }
    }

    /// Whether the shard is UNRECONCILED (design §3). The clean fast path (§4)
    /// must NOT skip an unreconciled shard — its next commit runs the FULL
    /// reconciling path. Poison reads as unreconciled (fail closed: a poisoned
    /// cell never enables the zero-storage fast path).
    pub fn is_unreconciled(&self) -> bool {
        match self.0.lock() {
            Ok(snapshot) => snapshot.unreconciled,
            Err(_poison) => true,
        }
    }

    /// `merge_adopt` failure after the buffer discard (design §3): the disk may
    /// carry the adopted marker while the in-memory root is still the old one, so
    /// the shard is UNRECONCILED — its next commit runs the FULL reconciling path
    /// and it is never fast-pathed or clean-skipped until then.
    pub fn mark_unreconciled(&self) {
        self.write(|snapshot| snapshot.unreconciled = true);
    }

    /// A detached copy of the current snapshot (test observable + restart pins).
    #[cfg(test)]
    pub fn snapshot(&self) -> CommitSnapshot {
        match self.0.lock() {
            Ok(snapshot) => *snapshot,
            Err(poison) => *poison.into_inner(),
        }
    }

    /// Whether the leaf mutex is currently poisoned (test probe for the
    /// poison-adopt/clear pins).
    #[cfg(test)]
    pub fn is_poisoned(&self) -> bool {
        self.0.is_poisoned()
    }

    /// Deliberately poison the leaf mutex by unwinding while its guard is held
    /// (test seam). Diverges via an out-of-bounds index — never the `panic!` macro
    /// (crate-denied) — and catches the unwind so the caller keeps running.
    #[cfg(test)]
    pub fn force_poison_for_test(&self) {
        let poisoned = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let guard = self.0.lock().unwrap_or_else(PoisonError::into_inner);
            let empty: [u8; 0] = [];
            // Read the guard so it is genuinely held across the diverging access,
            // then index out of bounds to unwind while it is still in scope.
            let index = usize::try_from(guard.incarnation).unwrap_or(0);
            let _ = empty[index.wrapping_add(1)];
            drop(guard);
        }));
        debug_assert!(poisoned.is_err(), "force_poison_for_test must unwind");
    }
}

// COMMIT-COLLAPSE §11 (a)/(b) test-only deterministic failure-seam barriers.
//
// Two rendezvous points on the shard's real command/boot paths let a test observe
// (and hold) a shard EXACTLY at a window that is otherwise instantaneous:
//   * `PostWalCommit` (a): after `ShardActor::commit` made the marker durable but
//     BEFORE the cell publishes clean — the "marker durable, publication pending"
//     window (drives the restart (b) case and proves the cell stays conservatively
//     dirty until the publish).
//   * `Recovery` (b): after `begin_recovery` set `recovering` but BEFORE boot — a
//     reader must classify DIRTY throughout, and a commit submitted while paused
//     must not execute until recovery completes, then converge.
//
// Keyed by the cell's `Arc` address, which is stable per shard id for process
// lifetime and SHARED between the actor factory and the arming test (both resolve
// it from the seam) — so no path threading is needed, and, like the WAL failpoint,
// distinct shards/tests never collide. Cross-thread by construction (the actor
// boots/commits on a scheduler thread, not the arming test thread). Compiled out
// entirely in non-test builds.

/// Which deterministic failure seam a barrier holds a shard at.
#[cfg(test)]
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum SeamBarrier {
    /// After the WAL commit marker is durable, before the cell publishes clean.
    PostWalCommit,
    /// After `begin_recovery`, before boot runs.
    Recovery,
}

#[cfg(test)]
struct BarrierRendezvous {
    reached: std::sync::mpsc::SyncSender<()>,
    release: std::sync::mpsc::Receiver<()>,
}

#[cfg(test)]
static SEAM_BARRIERS: std::sync::LazyLock<
    std::sync::Mutex<std::collections::HashMap<(usize, SeamBarrier), BarrierRendezvous>>,
> = std::sync::LazyLock::new(|| std::sync::Mutex::new(std::collections::HashMap::new()));

/// Test-facing control for one armed barrier: wait until the shard REACHES it, then
/// RELEASE it to proceed.
#[cfg(test)]
pub struct BarrierHandle {
    reached: std::sync::mpsc::Receiver<()>,
    release: std::sync::mpsc::SyncSender<()>,
}

#[cfg(test)]
impl BarrierHandle {
    /// Block until the shard reaches the armed barrier (its actor thread is now
    /// parked inside the seam, holding no cell lock).
    pub fn wait_until_reached(&self) {
        let _ = self.reached.recv();
    }

    /// Release the parked shard so it proceeds past the barrier.
    pub fn release(&self) {
        let _ = self.release.send(());
    }
}

/// Arm `which` barrier for the shard whose cell is `cell`. The next time that
/// shard reaches the seam it parks until [`BarrierHandle::release`].
#[cfg(test)]
pub fn arm_seam_barrier(cell: &Arc<ShardCommitState>, which: SeamBarrier) -> BarrierHandle {
    let (reached_tx, reached_rx) = std::sync::mpsc::sync_channel(1);
    let (release_tx, release_rx) = std::sync::mpsc::sync_channel(1);
    let key = (std::sync::Arc::as_ptr(cell).addr(), which);
    if let Ok(mut barriers) = SEAM_BARRIERS.lock() {
        barriers.insert(
            key,
            BarrierRendezvous {
                reached: reached_tx,
                release: release_rx,
            },
        );
    }
    BarrierHandle {
        reached: reached_rx,
        release: release_tx,
    }
}

/// Production-path hook: if `which` is armed for `cell`, signal that the shard
/// reached the seam and block until the test releases it. A no-op when unarmed.
#[cfg(test)]
pub fn pause_at_seam_barrier(cell: &Arc<ShardCommitState>, which: SeamBarrier) {
    let key = (std::sync::Arc::as_ptr(cell).addr(), which);
    let rendezvous = SEAM_BARRIERS
        .lock()
        .ok()
        .and_then(|mut barriers| barriers.remove(&key));
    if let Some(rendezvous) = rendezvous {
        let _ = rendezvous.reached.send(());
        let _ = rendezvous.release.recv();
    }
}

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