haematite 0.7.0

//! CORE-007: the shard's beamr native process body.
//!
//! [`ShardNativeHandler`] is a real [`beamr::NativeHandler`] — it runs as a
//! first-class, scheduler-supervised process (real pid, mailbox, factory
//! restart). When the host wakes it with the shard wake atom, it drains one
//! mailbox token per queued command, pops the command under a tight lock,
//! releases the lock, runs the storage op against the wrapped [`ShardActor`]
//! (which keeps the WAL-before-buffer and committed-root invariants), and
//! replies over the command's own channel. Binary never touches a term.
//!
//! Lock discipline (spec Landmine 4): the command-queue mutex is held ONLY for
//! the `pop_front` in [`lock_queue`] / [`pop_command`]; it is released before
//! any storage op and is never held across a [`NativeOutcome`] return.
//!
//! Wake-atom constraint: the handler never inspects the wake atom's VALUE — a
//! received mailbox token (`ctx.recv().is_some()`) simply means "drain one
//! command". That is WHY the host can intern the wake atom from a fresh local
//! `AtomTable` (see `ShardHandle::spawn`) without sharing the scheduler's table.
//! WARNING: if future code ever pattern-matches on the atom value in the mailbox
//! (e.g. to distinguish a wake from an exit signal), it MUST intern that atom
//! from the scheduler's own atom table, not a fresh local one — a fresh-table
//! atom has a different id and the match would silently fail.

use std::cmp::Ordering;
use std::collections::VecDeque;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::MutexGuard;
use std::sync::mpsc::SyncSender;

use beamr::process::ExitReason;
use beamr::term::Term;
use beamr::{NativeContext, NativeHandler, NativeOutcome};

use crate::db::root_advance::RootTransition;
use crate::shard::actor::ShardActor;
use crate::shard::commit_state::ShardCommitState;
use crate::store::DiskStore;
use crate::tree::{Cursor, Hash, empty_root_hash};
use crate::ttl::filter::{Visibility, visible_value};
use crate::wal::{DurableWal, FsyncPolicy, Mutation, WalRecovery};

use super::expiry_index::{ArmError, DeadlineScheduler, Generation};
use super::handle::{
    AppendReply, CommandQueue, CommitReply, MergeAdoptReply, RangeItem, ShardCommand,
    ShardCommandKind, ShardError, StreamSeq,
};
use super::{GroupOutcome, GroupWrite};

/// The reply channel shared by every groupable durable write (`Cas`,
/// `ApplyDurable`, `ApplyDurableTombstone`): each replies the optional
/// root-advance transition its group commit produced (`Some` iff the root moved).
type GroupReply = SyncSender<Result<Option<RootTransition>, ShardError>>;

/// The wrapped-storage state plus the bridge queue, run as a beamr process.
///
/// On a successful boot `state` is `Some`. If opening the store/WAL or
/// recovering failed, `state` is `None` and `startup_error` carries the cause:
/// the first slice replies-drains the queue with that error and stops cleanly
/// rather than panicking the scheduler.
pub struct ShardNativeHandler {
    state: Option<ShardState>,
    startup_error: Option<ShardError>,
    commands: CommandQueue,
}

/// The live storage owned by a booted shard process.
struct ShardState {
    actor: ShardActor,
    store: DiskStore,
    /// This shard's process-lifetime COMMIT-COLLAPSE cell. The ordered command loop
    /// (the shard's single ordered context) is the cell's actor-side writer: it
    /// publishes the dirty classification and — atomically — assigns this shard's
    /// `advance_gen` on a root advance. The seam's leaf emission mutex (delivery
    /// filter, Database side, R1) is a DIFFERENT lock this loop never touches.
    commit_state: Arc<ShardCommitState>,
}

impl ShardState {
    /// Publish a successful commit to the commit-state cell (COMMIT-COLLAPSE §3):
    /// the shard is CLEAN at its now-current root, and iff the root ADVANCED the
    /// cell bumps `advance_gen` by one — atomically with the publish — and returns
    /// the transition the root-advance seam tells with. `before` is the committed
    /// root observed BEFORE the command ran; the FIRST commit's `prior` is the
    /// empty-root constant (ROOT-ADVANCE-SEAM §2). A successful commit always clears
    /// the buffer, so this publishes clean unconditionally; the emitted
    /// `advance_gen` is SOURCED from the cell (the §8 absorb), never a seam-local
    /// counter. A non-advancing commit returns `None` and reaches no subscriber.
    fn publish_commit(&self, before: Option<Hash>) -> Option<RootTransition> {
        let after = self.actor.committed_root();
        let prior_root = before.unwrap_or_else(empty_root_hash);
        let new_root = after.unwrap_or_else(empty_root_hash);
        let advanced = prior_root != new_root;
        self.commit_state
            .publish_committed(after, advanced)
            .map(|advance_gen| RootTransition {
                prior_root,
                new_root,
                advance_gen,
            })
    }

    /// Publish the cell after a committing command SUCCEEDED. A real commit always
    /// flushes the buffer, so an empty buffer means "clean at the current root" —
    /// publish clean and, iff the root advanced, the tell. A NON-empty buffer means
    /// the command took an Ok early-return that committed NOTHING (an empty `append`
    /// or empty `apply_durable_batch`) and left a pre-existing dirty buffer intact —
    /// preserve that classification rather than falsely clearing it. Returns the
    /// root-advance transition to tell, if any.
    fn publish_after_commit(&self, before: Option<Hash>) -> Option<RootTransition> {
        if self.actor.buffer().is_empty() {
            self.publish_commit(before)
        } else {
            self.publish_from_buffer();
            None
        }
    }

    /// Publish the outcome of a single-key durable write (CAS / stamped value apply
    /// / stamped tombstone apply / batch apply) and shape its reply: on success the
    /// write committed, so publish clean + the optional root-advance transition; on
    /// error the buffer was rolled back, so republish the classification the
    /// restored buffer implies and pass the error through.
    fn commit_or_rollback(
        &self,
        result: Result<(), ShardError>,
        before: Option<Hash>,
    ) -> Result<Option<RootTransition>, ShardError> {
        match result {
            Ok(()) => Ok(self.publish_after_commit(before)),
            Err(error) => {
                self.publish_from_buffer();
                Err(error)
            }
        }
    }

    /// Republish the cell's dirty classification from the buffer's actual emptiness
    /// (COMMIT-COLLAPSE §2.1): CLEAN iff the buffer is empty. Used after a buffered
    /// put/delete/TTL-sweep (leaves the buffer non-empty → dirty) and after any
    /// rollback/restore or failed commit (clean iff the restored buffer is empty).
    fn publish_from_buffer(&self) {
        self.commit_state
            .sync_from_buffer(self.actor.buffer().is_empty());
    }

    /// Mark the cell UNRECONCILED after a `merge_adopt` failure (COMMIT-COLLAPSE §3):
    /// the disk may carry the adopted marker while the in-memory root is still the
    /// old one, so the next commit must run the FULL reconciling path.
    fn publish_unreconciled(&self) {
        self.commit_state.mark_unreconciled();
    }
}

/// Sealed-seam invariant (COMMIT-COLLAPSE §11, ⟨r3 M7⟩), asserted after every actor
/// command across the WHOLE shard test suite: a shard classified CLEAN must have an
/// empty buffer, no recovering/unreconciled flag, and a committed root. Any future
/// staging path that forgets the publication seam trips this in every test that
/// touches it, not just a bespoke one.
#[cfg(test)]
fn assert_sealed(state: &ShardState) {
    let snapshot = state.commit_state.snapshot();
    let clean = !(snapshot.recovering
        || snapshot.unreconciled
        || snapshot.committed_gen != snapshot.dirty_gen
        || snapshot.root.is_none());
    if clean {
        assert!(
            state.actor.buffer().is_empty(),
            "sealed seam: shard classified clean but its WAL buffer is non-empty"
        );
    }
}

/// Release build: the sealed-seam check compiles to nothing (§11 is a test-build
/// invariant). Kept as an unconditional call site so the drain path has a real
/// statement between binding the outcome and returning it in every configuration.
#[cfg(not(test))]
const fn assert_sealed(_state: &ShardState) {}

impl ShardNativeHandler {
    /// Build the restart-capable factory the scheduler stores and re-invokes.
    ///
    /// Each invocation re-opens the store + WAL and re-runs recovery against the
    /// SAME paths, so a re-spawn replays the durable WAL and resumes the shard.
    pub(super) fn make_factory(
        store_dir: PathBuf,
        wal_path: PathBuf,
        commands: CommandQueue,
        policy: crate::tree::TreePolicy,
        commit_state: Arc<ShardCommitState>,
    ) -> beamr::native::native_process::NativeHandlerFactory {
        Box::new(move || {
            Box::new(Self::build(
                &store_dir,
                &wal_path,
                Arc::clone(&commands),
                policy,
                Arc::clone(&commit_state),
            ))
        })
    }

    /// Open the store + WAL and recover. Any failure yields a sentinel handler
    /// that stops cleanly on its first slice.
    ///
    /// Restart protocol (COMMIT-COLLAPSE §2.2 M2): every factory invocation — first
    /// spawn AND restart — runs, in order, (1) mark the cell `recovering` and bump
    /// its `incarnation` BEFORE recovery, so a global commit racing the rebuild
    /// classifies this shard dirty and never clean-skips a recovered write;
    /// (2) recover; (3) publish the complete recovered snapshot before the first
    /// command (dirty iff the recovered buffer is non-empty), preserving
    /// `advance_gen` across the incarnation. A boot failure leaves the cell
    /// `recovering` (fail closed → classified dirty).
    fn build(
        store_dir: &Path,
        wal_path: &Path,
        commands: CommandQueue,
        policy: crate::tree::TreePolicy,
        commit_state: Arc<ShardCommitState>,
    ) -> Self {
        commit_state.begin_recovery();
        // COMMIT-COLLAPSE §11 (b): test-only recovery pause — a reader classifies
        // DIRTY throughout this window, and a commit queued while paused waits for
        // recovery to complete. No-op in release (the seam is compiled out).
        #[cfg(test)]
        crate::shard::commit_state::pause_at_seam_barrier(
            &commit_state,
            crate::shard::commit_state::SeamBarrier::Recovery,
        );
        match Self::boot(store_dir, wal_path, policy, commit_state) {
            Ok(state) => {
                state.commit_state.publish_recovered(
                    state.actor.committed_root(),
                    state.actor.buffer().is_empty(),
                );
                Self {
                    state: Some(state),
                    startup_error: None,
                    commands,
                }
            }
            Err(error) => Self {
                state: None,
                startup_error: Some(error),
                commands,
            },
        }
    }

    /// Open the store, recover the WAL against it, and seed a [`ShardActor`].
    ///
    /// FENCE-CHAIN r8 R1e (load-bearing ordering pin): the `DiskStore` is
    /// constructed BEFORE the `DurableWal`. The store uses D3's crate-private
    /// create-or-open-UNFENCED mode ([`DiskStore::new_unfenced`]) — a fresh shard
    /// must still CREATE its `store` root here, but the root's directory entry is
    /// fenced TRANSITIVELY by `DurableWal::new`'s fsync of the SHARED shard
    /// directory (`store` and `shard.wal` share the parent `shard-{id}`). Reorder
    /// these two lines and the fresh-shard `store` entry would ride unfenced past
    /// the WAL's first durable append. The `shard-{id}` entry itself is fenced by
    /// the L1 data-dir fence (R1b) before this handler ever spawns.
    fn boot(
        store_dir: &Path,
        wal_path: &Path,
        policy: crate::tree::TreePolicy,
        commit_state: Arc<ShardCommitState>,
    ) -> Result<ShardState, ShardError> {
        let store = DiskStore::new_unfenced(store_dir)?;
        let recovered = WalRecovery::recover_path(wal_path, &store)?;
        let wal = DurableWal::new(wal_path, FsyncPolicy::CommitOnly)?;
        let actor = ShardActor::from_recovered(wal, recovered, &store, policy)?;
        Ok(ShardState {
            actor,
            store,
            commit_state,
        })
    }

    /// Drain the next unit of work for one mailbox token, replying over each
    /// command's channel. Returns a stop outcome when a command requested process
    /// shutdown.
    ///
    /// Group commit (audit E): if the front of the queue is a groupable durable
    /// WRITE (`Cas` / `ApplyDurable` / `ApplyDurableTombstone`), a maximal run of
    /// consecutive such commands is popped under ONE tight lock and committed in
    /// ONE fsync via [`ShardActor::apply_group`], collapsing N fsyncs to one.
    /// Otherwise exactly one command is popped and run as before. (A group of K
    /// commands consumes K queued items but only this one token; the K-1 surplus
    /// tokens later find an empty queue — the existing `QueueEmpty` = spurious
    /// case — so no work is lost or double-run.)
    fn pop_and_execute(
        state: &mut ShardState,
        commands: &CommandQueue,
        ctx: &mut NativeContext<'_>,
    ) -> Option<NativeOutcome> {
        let outcome = match pop_next_unit(commands)? {
            DrainUnit::Group(group) => {
                state.run_group(group, ctx);
                None
            }
            DrainUnit::Single(command) => state.execute(command, ctx),
        };
        // COMMIT-COLLAPSE §11 sealed-seam invariant, checked after EVERY command
        // across the whole shard test suite (a no-op in release).
        assert_sealed(state);
        outcome
    }
}

/// One unit of work drained for a single mailbox token: either a coalesced run of
/// groupable durable writes, or one non-groupable command run on its own.
enum DrainUnit {
    Group(Vec<(GroupWrite, GroupReply)>),
    Single(ShardCommand),
}

/// Pop the next unit of work under a TIGHT lock (spec Landmine 4): the queue mutex
/// is held ONLY for the peek/pop here and released before any storage op or fsync.
///
/// If the front command is a groupable durable write, pop the maximal CONSECUTIVE
/// run of them (stopping at the first non-groupable command, which is left on the
/// queue and handled on its own next time). Otherwise pop exactly one command.
fn pop_next_unit(commands: &CommandQueue) -> Option<DrainUnit> {
    let mut queue = lock_queue(commands);
    if !queue
        .front()
        .is_some_and(|command| is_groupable(&command.kind))
    {
        return queue.pop_front().map(DrainUnit::Single);
    }
    let mut group = Vec::new();
    while queue
        .front()
        .is_some_and(|command| is_groupable(&command.kind))
    {
        // The front is groupable, so `pop_front` yields it and `into_group_write`
        // returns `Some` — but never `unwrap`: a (logically impossible) `None`
        // simply ends the run rather than panicking.
        if let Some(command) = queue.pop_front()
            && let Some(entry) = into_group_write(command.kind)
        {
            group.push(entry);
        }
    }
    Some(DrainUnit::Group(group))
}

/// Whether a command is a groupable durable WRITE that may be coalesced into a
/// group commit. ONLY single-key CAS / stamped value apply / stamped tombstone
/// apply qualify — they each fsync one root today. Everything else (promise-state
/// mutators, `Prepare`/`merge_adopt`, the all-or-nothing `ApplyDurableBatch`,
/// `Append`, reads, `Commit`, `Shutdown`) is NOT groupable and keeps its own
/// semantics and ordering.
const fn is_groupable(kind: &ShardCommandKind) -> bool {
    matches!(
        kind,
        ShardCommandKind::Cas { .. }
            | ShardCommandKind::ApplyDurable { .. }
            | ShardCommandKind::ApplyDurableTombstone { .. }
    )
}

/// Convert a groupable command kind into its [`GroupWrite`] descriptor plus reply
/// channel. Returns `None` for any non-groupable kind (the caller only ever passes
/// kinds for which [`is_groupable`] is true, so `None` never occurs in practice).
fn into_group_write(kind: ShardCommandKind) -> Option<(GroupWrite, GroupReply)> {
    match kind {
        ShardCommandKind::Cas {
            key,
            expected,
            new,
            reply,
        } => Some((GroupWrite::Cas { key, expected, new }, reply)),
        ShardCommandKind::ApplyDurable {
            key,
            expected,
            value,
            ttl,
            stamp,
            reply,
        } => Some((
            GroupWrite::ApplyValue {
                key,
                expected,
                value,
                ttl,
                stamp,
            },
            reply,
        )),
        ShardCommandKind::ApplyDurableTombstone {
            key,
            expected,
            stamp,
            reply,
        } => Some((
            GroupWrite::ApplyTombstone {
                key,
                expected,
                stamp,
            },
            reply,
        )),
        _ => None,
    }
}

struct NativeDeadlineScheduler<'context, 'process> {
    context: &'context mut NativeContext<'process>,
}

impl DeadlineScheduler for NativeDeadlineScheduler<'_, '_> {
    fn schedule(
        &mut self,
        delay: std::time::Duration,
        generation: Generation,
    ) -> Result<(), ArmError> {
        let value =
            i64::try_from(generation.value()).map_err(|_error| ArmError::TokenUnrepresentable)?;
        let token = Term::try_small_int(value).ok_or(ArmError::TokenUnrepresentable)?;
        self.context
            .schedule(delay, token)
            .map(drop)
            .ok_or(ArmError::SchedulerUnavailable)
    }
}

#[cfg(test)]
pub(super) fn schedule_probe(
    context: &mut NativeContext<'_>,
    delay: std::time::Duration,
    token: u64,
) -> Result<(), ArmError> {
    let generation = Generation::from_value(token).ok_or(ArmError::TokenUnrepresentable)?;
    NativeDeadlineScheduler { context }.schedule(delay, generation)
}

impl ShardState {
    fn arm_pending(&mut self, ctx: &mut NativeContext<'_>) {
        let mut scheduler = NativeDeadlineScheduler { context: ctx };
        if let Err(error) = self.actor.arm_expiry(&mut scheduler) {
            log::error!("shard TTL deadline arm refused: {error}");
        }
    }

    fn receive_deadline(&mut self, generation: Generation, ctx: &mut NativeContext<'_>) {
        let Some(due) = self.actor.begin_expiry_deadline(generation) else {
            return;
        };
        for (position, (_deadline, key)) in due.iter().enumerate() {
            self.actor.inspect_expiry_key();
            let (reply, response) = std::sync::mpsc::sync_channel(1);
            let command = ShardCommand {
                id: 0,
                kind: ShardCommandKind::DeleteIfExpired {
                    key: key.clone(),
                    reply,
                },
            };
            let _outcome = self.execute(command, ctx);
            match response.recv() {
                Ok(Ok(_removed)) => {}
                Ok(Err(error)) => {
                    log::error!("shard TTL deadline delete failed: {error}");
                    self.actor.restore_detached(&due[position..]);
                    break;
                }
                Err(error) => {
                    log::error!("shard TTL deadline delete reply failed: {error}");
                    self.actor.restore_detached(&due[position..]);
                    break;
                }
            }
        }
        self.actor.finish_expiry_deadline();
        self.arm_pending(ctx);
    }

    /// Run a coalesced group of durable writes in ONE group commit and fan the
    /// per-write outcome to each command's reply channel (audit E).
    ///
    /// The writes are split from their reply senders, staged + committed once by
    /// [`ShardActor::apply_group`] (which returns one outcome per write, in order),
    /// then each outcome is mapped back to its sender: a survivor gets `Ok(())`, a
    /// rejected write gets its own CAS/fence error, and — if the shared commit
    /// failed — every survivor gets the retryable commit error.
    fn run_group(&mut self, group: Vec<(GroupWrite, GroupReply)>, ctx: &mut NativeContext<'_>) {
        let mut writes = Vec::with_capacity(group.len());
        let mut replies = Vec::with_capacity(group.len());
        for (write, reply) in group {
            writes.push(write);
            replies.push(reply);
        }
        let before = self.actor.committed_root();
        let outcomes = self.actor.apply_group(writes, &mut self.store);
        self.arm_pending(ctx);
        // ONE group commit => ONE root advance => ONE gen. `publish_after_commit`
        // publishes clean (buffer flushed) or preserves the post-rollback
        // classification (all rejected / commit failed), and returns the transition
        // iff the root advanced. Hand it to the FIRST committed reply (`take` leaves
        // the rest `None`); the seam's monotone filter would coalesce duplicates
        // anyway, but a single carrier keeps exactly one tell per group.
        let mut advance = self.publish_after_commit(before);
        for (reply, outcome) in replies.into_iter().zip(outcomes) {
            let payload = match outcome {
                GroupOutcome::Committed => Ok(advance.take()),
                GroupOutcome::Rejected(error) | GroupOutcome::CommitFailed(error) => Err(error),
            };
            drop(reply.send(payload));
        }
    }

    /// Flush the buffer and reply with the new root plus the optional root
    /// transition (`Some` iff the root advanced). Split out of [`Self::execute`] to
    /// keep that dispatch under the line budget.
    fn run_commit(&mut self, reply: &CommitReply) {
        // COMMIT-COLLAPSE §4 clean fast path (defense in depth): an empty buffer on
        // a marker-BEARING, reconciled shard commits ZERO storage and returns the
        // cached root. It backstops a false-DIRTY classification (one actor message,
        // no fsync, no root advance). A marker-LESS shard (committed_root `None`)
        // falls through so it publishes its committed-empty marker ONCE via the full
        // path — NEVER synthesised — preserving the observer contract; an
        // UNRECONCILED shard falls through so its next commit re-reconciles (§3, it
        // is never fast-pathed). `publish_from_buffer` republishes clean from the
        // empty buffer, correcting a false-dirty cell without touching the root.
        if self.actor.buffer().is_empty()
            && !self.commit_state.is_unreconciled()
            && let Some(root) = self.actor.committed_root()
        {
            self.publish_from_buffer();
            drop(reply.send(Ok((root, None))));
            return;
        }
        let before = self.actor.committed_root();
        let result = self.actor.commit(&mut self.store).map_err(ShardError::from);
        // COMMIT-COLLAPSE §11 (a): test-only barrier at the "marker durable,
        // publication pending" window — the WAL marker is durable but the cell has
        // NOT yet published, so a reader must still classify DIRTY. No-op in release.
        #[cfg(test)]
        if result.is_ok() {
            crate::shard::commit_state::pause_at_seam_barrier(
                &self.commit_state,
                crate::shard::commit_state::SeamBarrier::PostWalCommit,
            );
        }
        let payload = match result {
            Ok(root) => Ok((root, self.publish_after_commit(before))),
            Err(error) => {
                // Ordinary commit failure retains the buffer for retry (§3); the
                // cell tracks its (unchanged, non-empty) dirty classification.
                self.publish_from_buffer();
                Err(error)
            }
        };
        drop(reply.send(payload));
    }

    /// Append event entries (which stage AND commit once), publish the resulting
    /// classification, and reply the new sequence plus any root-advance transition.
    /// Split out of [`Self::execute`] to keep that dispatch under the line budget.
    fn run_append(
        &mut self,
        key: &[u8],
        entries: Vec<Vec<u8>>,
        expected_seq: u64,
        ttl: Option<std::time::Duration>,
        reply: &AppendReply,
        ctx: &mut NativeContext<'_>,
    ) {
        let before = self.actor.committed_root();
        let result = self
            .actor
            .append(key, entries, expected_seq, ttl, &mut self.store)
            .map_err(ShardError::from);
        self.arm_pending(ctx);
        let payload = match result {
            // Append commits on success => clean + (root advance); an empty append
            // committed nothing and keeps the pre-existing classification.
            Ok(new_seq) => Ok((new_seq, self.publish_after_commit(before))),
            // A failed append rolled the buffer back to its pre-append state.
            Err(error) => {
                self.publish_from_buffer();
                Err(error)
            }
        };
        drop(reply.send(payload));
    }

    /// Merge a promise majority's committed states and durably adopt the merged
    /// root, publishing the resulting classification. On adoption the buffer is
    /// discarded (clean); a failure after the discard may leave disk carrying the
    /// adopted marker while the in-memory root is stale (§3), so the shard is marked
    /// UNRECONCILED to force the next commit through the full reconciling path. Split
    /// out of [`Self::execute_extra`] to keep that dispatch under the line budget.
    fn run_merge_adopt(
        &mut self,
        promisers: &[(Option<Hash>, Vec<crate::sync::NodeTransfer>)],
        reply: &MergeAdoptReply,
        ctx: &mut NativeContext<'_>,
    ) {
        let before = self.actor.committed_root();
        let result = self
            .actor
            .merge_adopt(promisers, &mut self.store)
            .map_err(ShardError::from);
        self.arm_pending(ctx);
        let payload = match result {
            Ok(adopted) => Ok((adopted, self.publish_after_commit(before))),
            Err(error) => {
                self.publish_unreconciled();
                Err(error)
            }
        };
        drop(reply.send(payload));
    }

    /// Run one command against the wrapped storage and send the reply.
    fn execute(
        &mut self,
        command: ShardCommand,
        ctx: &mut NativeContext<'_>,
    ) -> Option<NativeOutcome> {
        match command.kind {
            ShardCommandKind::Get { key, reply } => {
                let result = self.actor.get(&key, &self.store).map_err(ShardError::from);
                drop(reply.send(result));
                None
            }
            ShardCommandKind::GetRaw { key, reply } => {
                let result = self
                    .actor
                    .get_raw(&key, &self.store)
                    .map_err(ShardError::from);
                drop(reply.send(result));
                None
            }
            ShardCommandKind::Put {
                key,
                value,
                ttl,
                reply,
            } => {
                let result = self
                    .actor
                    .put_with_ttl(key, value, ttl, &self.store)
                    .map_err(ShardError::from);
                self.arm_pending(ctx);
                // A buffered put leaves the buffer non-empty => dirty (§2.1); a
                // failed WAL append leaves it unchanged. Republish from the buffer.
                self.publish_from_buffer();
                drop(reply.send(result));
                None
            }
            ShardCommandKind::Delete { key, stamp, reply } => {
                let result = self
                    .actor
                    .delete(key, stamp, &self.store)
                    .map_err(ShardError::from);
                self.arm_pending(ctx);
                // A stamped tombstone is a buffered put (§2.1) => dirty.
                self.publish_from_buffer();
                drop(reply.send(result));
                None
            }
            ShardCommandKind::DeleteIfExpired { key, reply } => {
                let result = self
                    .actor
                    .delete_if_expired(&key, &self.store)
                    .map_err(ShardError::from);
                self.arm_pending(ctx);
                // TTL-sweep bare delete (§2.1) => dirty when it removed a key; a
                // no-op sweep leaves the buffer as it was.
                self.publish_from_buffer();
                drop(reply.send(result));
                None
            }
            ShardCommandKind::Commit { reply } => {
                self.run_commit(&reply);
                None
            }
            range @ (ShardCommandKind::Range { .. } | ShardCommandKind::HasLiveInRange(..)) => {
                self.execute_range(range)
            }
            ShardCommandKind::Append {
                key,
                entries,
                expected_seq,
                ttl,
                reply,
            } => {
                self.run_append(&key, entries, expected_seq, ttl, &reply, ctx);
                None
            }
            ShardCommandKind::ReadValue { key, reply } => {
                let result = self
                    .actor
                    .read_value(&key, &self.store)
                    .map_err(ShardError::from);
                drop(reply.send(result));
                None
            }
            ShardCommandKind::ReadPromiseState { reply } => {
                drop(reply.send(Ok(self.actor.promise_state())));
                None
            }
            extra @ (ShardCommandKind::Cas { .. }
            | ShardCommandKind::ApplyDurable { .. }
            | ShardCommandKind::ApplyDurableTombstone { .. }
            | ShardCommandKind::ApplyDurableBatch { .. }
            | ShardCommandKind::RecordPromise { .. }
            | ShardCommandKind::RecordOwnerEpoch { .. }
            | ShardCommandKind::ReserveMinted { .. }
            | ShardCommandKind::ExportReachable { .. }
            | ShardCommandKind::MergeAdopt { .. }) => self.execute_extra(extra, ctx),
            ShardCommandKind::ScanSequences { reply } => {
                drop(reply.send(self.scan_sequences()));
                None
            }
            ShardCommandKind::Shutdown { reply } => {
                self.actor.invalidate_expiry_on_shutdown();
                drop(reply.send(Ok(())));
                Some(NativeOutcome::Stop(ExitReason::Normal))
            }
        }
    }

    /// Run one CAS / durable-apply / AA-3-0 promise-state command against the
    /// actor (the promise mutators fsync before their reply). Split out of
    /// [`Self::execute`] to keep that dispatch under the line budget.
    fn execute_extra(
        &mut self,
        command: ShardCommandKind,
        ctx: &mut NativeContext<'_>,
    ) -> Option<NativeOutcome> {
        match command {
            ShardCommandKind::Cas {
                key,
                expected,
                new,
                reply,
            } => {
                let before = self.actor.committed_root();
                let result = self
                    .actor
                    .cas(&key, expected, new, &mut self.store)
                    .map_err(ShardError::from);
                self.arm_pending(ctx);
                let payload = self.commit_or_rollback(result, before);
                drop(reply.send(payload));
            }
            ShardCommandKind::ApplyDurable {
                key,
                expected,
                value,
                ttl,
                stamp,
                reply,
            } => {
                let before = self.actor.committed_root();
                let result = self
                    .actor
                    .apply_durable(&key, expected, value, ttl, stamp, &mut self.store)
                    .map_err(ShardError::from);
                self.arm_pending(ctx);
                let payload = self.commit_or_rollback(result, before);
                drop(reply.send(payload));
            }
            ShardCommandKind::ApplyDurableTombstone {
                key,
                expected,
                stamp,
                reply,
            } => {
                let before = self.actor.committed_root();
                let result = self
                    .actor
                    .apply_durable_tombstone(&key, expected, stamp, &mut self.store)
                    .map_err(ShardError::from);
                self.arm_pending(ctx);
                let payload = self.commit_or_rollback(result, before);
                drop(reply.send(payload));
            }
            ShardCommandKind::ApplyDurableBatch {
                items,
                stamp,
                reply,
            } => {
                let before = self.actor.committed_root();
                let result = self
                    .actor
                    .apply_durable_batch(items, stamp, &mut self.store)
                    .map_err(ShardError::from);
                self.arm_pending(ctx);
                let payload = self.commit_or_rollback(result, before);
                drop(reply.send(payload));
            }
            ShardCommandKind::RecordPromise { ballot, reply } => {
                let result = self.actor.record_promise(ballot).map_err(ShardError::from);
                drop(reply.send(result));
            }
            ShardCommandKind::RecordOwnerEpoch { ballot, reply } => {
                let result = self
                    .actor
                    .record_owner_epoch(ballot)
                    .map_err(ShardError::from);
                drop(reply.send(result));
            }
            ShardCommandKind::ReserveMinted { counter, reply } => {
                let result = self.actor.reserve_minted(counter).map_err(ShardError::from);
                drop(reply.send(result));
            }
            ShardCommandKind::ExportReachable { shard_id, reply } => {
                let result = self
                    .actor
                    .export_reachable(shard_id, &self.store)
                    .map_err(ShardError::from);
                drop(reply.send(result));
            }
            ShardCommandKind::MergeAdopt { promisers, reply } => {
                self.run_merge_adopt(&promisers, &reply, ctx);
            }
            _ => {}
        }
        None
    }

    /// Decode every stream's sequence metadata from the actor-owned index.
    fn scan_sequences(&self) -> Result<Vec<StreamSeq>, ShardError> {
        self.actor.scan_sequences()
    }

    /// Run range-shaped commands against the merged tree+buffer read view.
    fn execute_range(&self, command: ShardCommandKind) -> Option<NativeOutcome> {
        match command {
            ShardCommandKind::Range { from, to, reply } => {
                drop(reply.send(self.collect_range(&from, &to)));
            }
            ShardCommandKind::HasLiveInRange(from, to, reply) => {
                let result = super::liveness::has_live_in_range(
                    &self.store,
                    self.actor.committed_root(),
                    self.actor.buffer(),
                    from.as_slice(),
                    to.as_slice(),
                );
                drop(reply.send(result));
            }
            _ => {}
        }
        None
    }

    /// Merge committed-tree entries with the live buffer for `[from, to)`.
    fn collect_range(&self, from: &[u8], to: &[u8]) -> Result<Vec<RangeItem>, ShardError> {
        let tree_entries = match self.actor.committed_root() {
            Some(root) => Cursor::new(&self.store, root)
                .range(from, to)
                .collect::<Result<Vec<_>, _>>()
                .map_err(ShardError::from)?,
            None => Vec::new(),
        };
        let buffer_entries: Vec<&Mutation> = self
            .actor
            .buffer()
            .iter()
            .filter(|mutation| in_range(mutation, from, to))
            .collect();
        merge_range(&tree_entries, &buffer_entries)
    }
}

impl NativeHandler for ShardNativeHandler {
    fn handle(&mut self, ctx: &mut NativeContext<'_>) -> NativeOutcome {
        let Some(state) = self.state.as_mut() else {
            return self.fail_startup();
        };
        state.arm_pending(ctx);
        while let Some(message) = ctx.recv() {
            if let Some(raw_generation) = message.as_small_int()
                && let Ok(value) = u64::try_from(raw_generation)
                && let Some(generation) = Generation::from_value(value)
            {
                state.receive_deadline(generation, ctx);
            } else if let Some(outcome) = Self::pop_and_execute(state, &self.commands, ctx) {
                return outcome;
            }
        }
        NativeOutcome::Wait
    }
}

impl ShardNativeHandler {
    /// Sentinel slice: a shard that failed to boot drains the queue with the
    /// startup error so callers fail fast, then stops cleanly.
    fn fail_startup(&self) -> NativeOutcome {
        let message = self
            .startup_error
            .as_ref()
            .map_or_else(|| "shard startup failed".to_owned(), ToString::to_string);
        while let Some(command) = pop_command(&self.commands) {
            super::startup::reply_startup_error(command, &message);
        }
        NativeOutcome::Stop(ExitReason::Error)
    }
}

/// Lock the command queue. The guard is held only by the immediate caller for a
/// `push_back` / `retain` / `pop_front`; it must never span a storage op.
pub(super) fn lock_queue(commands: &CommandQueue) -> MutexGuard<'_, VecDeque<ShardCommand>> {
    commands
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
}

/// Pop one command under a tight lock, releasing it before returning.
fn pop_command(commands: &CommandQueue) -> Option<ShardCommand> {
    lock_queue(commands).pop_front()
}

/// True when `mutation`'s key lies in `[from, to)`.
fn in_range(mutation: &Mutation, from: &[u8], to: &[u8]) -> bool {
    let key = mutation.key();
    from <= key && key < to
}

/// Two-way merge of committed-tree entries and buffered mutations into an
/// ordered range result. A buffered key shadows the tree; a buffered delete
/// removes the key from the result. Ported from the CORE-007 reference.
fn merge_range(
    tree_entries: &[(Vec<u8>, Vec<u8>)],
    buffer_entries: &[&Mutation],
) -> Result<Vec<RangeItem>, ShardError> {
    // Upper bound: every tree entry plus every buffered mutation can yield at
    // most one item, plus the trailing `Done` sentinel. Both counts are local
    // (not attacker-controlled), so this single allocation never reallocates.
    let capacity = tree_entries
        .len()
        .saturating_add(buffer_entries.len())
        .saturating_add(1);
    let mut items = Vec::with_capacity(capacity);
    let mut tree_index = 0;
    let mut buffer_index = 0;
    loop {
        match (
            tree_entries.get(tree_index),
            buffer_entries.get(buffer_index),
        ) {
            (Some((tree_key, tree_value)), Some(buffer_mutation)) => {
                match tree_key.as_slice().cmp(buffer_mutation.key()) {
                    Ordering::Less => {
                        push_entry(&mut items, tree_key, tree_value)?;
                        tree_index = tree_index.saturating_add(1);
                    }
                    Ordering::Equal => {
                        push_mutation(&mut items, buffer_mutation)?;
                        tree_index = tree_index.saturating_add(1);
                        buffer_index = buffer_index.saturating_add(1);
                    }
                    Ordering::Greater => {
                        push_mutation(&mut items, buffer_mutation)?;
                        buffer_index = buffer_index.saturating_add(1);
                    }
                }
            }
            (Some((tree_key, tree_value)), None) => {
                push_entry(&mut items, tree_key, tree_value)?;
                tree_index = tree_index.saturating_add(1);
            }
            (None, Some(buffer_mutation)) => {
                push_mutation(&mut items, buffer_mutation)?;
                buffer_index = buffer_index.saturating_add(1);
            }
            (None, None) => break,
        }
    }
    items.push(RangeItem::Done);
    Ok(items)
}

/// Append a visible tree entry to the range result.
fn push_entry(items: &mut Vec<RangeItem>, key: &[u8], value: &[u8]) -> Result<(), ShardError> {
    match visible_value(value)
        .map_err(|error| ShardError::Wal(crate::wal::WalError::TreeError(error.to_string())))?
    {
        Visibility::Live(value) => items.push(RangeItem::Entry {
            key: key.to_vec(),
            value,
        }),
        Visibility::Expired => {}
    }
    Ok(())
}

/// Append a buffered mutation: a put becomes an entry, a delete is skipped.
fn push_mutation(items: &mut Vec<RangeItem>, mutation: &Mutation) -> Result<(), ShardError> {
    if let Mutation::Put { key, value } = mutation {
        push_entry(items, key, value)?;
    }
    Ok(())
}

/// Pre-flight (spec Phase 0): a [`DiskStore`] must be `Send` so it can live
/// inside a [`NativeHandler`] (which is `Send + 'static`). `RefCell<LruCache>`
/// is `Send` when its contents are `Send`, so no store change is needed; this
/// fails to compile if that ever regresses.
#[cfg(test)]
const fn assert_disk_store_send() {
    const fn require_send<T: Send>() {}
    require_send::<DiskStore>();
    require_send::<ShardNativeHandler>();
    let _ = require_send::<DiskStore>;
}

#[cfg(test)]
const _: () = assert_disk_store_send();

#[cfg(test)]
mod boot_failure_tests {
    use super::ShardNativeHandler;
    use crate::shard::actor::handle::{
        CommandQueue, RangeItem, ShardCommand, ShardCommandKind, ShardError,
    };
    use beamr::NativeOutcome;
    use beamr::process::ExitReason;
    use std::collections::VecDeque;
    use std::error::Error;
    use std::sync::mpsc;
    use std::sync::{Arc, Mutex};

    /// Build a handler whose boot DETERMINISTICALLY fails: the store path is a
    /// regular file, so `DiskStore::new` errors and the handler boots into the
    /// sentinel (`state = None`, `startup_error = Some`). The `TempDir` is
    /// returned so the caller keeps it alive.
    fn boot_failed_handler(
        commands: CommandQueue,
    ) -> Result<(ShardNativeHandler, tempfile::TempDir), Box<dyn Error>> {
        let dir = tempfile::tempdir()?;
        let store_path = dir.path().join("store-is-a-file");
        std::fs::write(&store_path, b"not a directory")?;
        let wal_path = dir.path().join("shard.wal");
        let seam = crate::db::root_advance::RootAdvanceSeam::new();
        let handler = ShardNativeHandler::build(
            &store_path,
            &wal_path,
            commands,
            crate::tree::TreePolicy::V1_DEFAULT,
            seam.commit_state(0),
        );
        Ok((handler, dir))
    }

    /// The queued-at-boot drain path: a command already on the queue when a
    /// boot-failed shard runs its sentinel slice must fail fast with
    /// [`ShardError::Spawn`] (never hang, never a storage error), the sentinel
    /// must stop the process, and the queue must be fully drained. One command of
    /// EACH kind exercises every `reply_startup_error` arm. This covers the path
    /// that is racy to force through the live scheduler (the host runs the first
    /// slice before an external command can be enqueued), so it is asserted here
    /// against the sentinel directly.
    #[test]
    fn fail_startup_drains_every_command_kind_with_spawn_then_stops() -> Result<(), Box<dyn Error>>
    {
        let commands: CommandQueue = Arc::new(Mutex::new(VecDeque::new()));
        let (get_tx, get_rx) = mpsc::sync_channel(1);
        let (put_tx, put_rx) = mpsc::sync_channel(1);
        let (del_tx, del_rx) = mpsc::sync_channel(1);
        let (commit_tx, commit_rx) = mpsc::sync_channel(1);
        let (range_tx, range_rx) = mpsc::sync_channel::<Result<Vec<RangeItem>, ShardError>>(1);
        {
            let mut queue = commands.lock().map_err(|_| "queue poisoned")?;
            queue.push_back(ShardCommand {
                id: 1,
                kind: ShardCommandKind::Get {
                    key: b"k".to_vec(),
                    reply: get_tx,
                },
            });
            queue.push_back(ShardCommand {
                id: 2,
                kind: ShardCommandKind::Put {
                    key: b"k".to_vec(),
                    value: b"v".to_vec(),
                    ttl: None,
                    reply: put_tx,
                },
            });
            queue.push_back(ShardCommand {
                id: 3,
                kind: ShardCommandKind::Delete {
                    key: b"k".to_vec(),
                    stamp: crate::sync::ballot::Stamp::bottom(),
                    reply: del_tx,
                },
            });
            queue.push_back(ShardCommand {
                id: 4,
                kind: ShardCommandKind::Commit { reply: commit_tx },
            });
            queue.push_back(ShardCommand {
                id: 5,
                kind: ShardCommandKind::Range {
                    from: b"a".to_vec(),
                    to: b"z".to_vec(),
                    reply: range_tx,
                },
            });
        }

        let (handler, _dir) = boot_failed_handler(Arc::clone(&commands))?;
        assert!(handler.state.is_none(), "boot must have failed");
        assert!(handler.startup_error.is_some(), "startup_error must be set");

        let outcome = handler.fail_startup();

        assert!(
            matches!(get_rx.try_recv(), Ok(Err(ShardError::Spawn(_)))),
            "Get arm"
        );
        assert!(
            matches!(put_rx.try_recv(), Ok(Err(ShardError::Spawn(_)))),
            "Put arm"
        );
        assert!(
            matches!(del_rx.try_recv(), Ok(Err(ShardError::Spawn(_)))),
            "Delete arm"
        );
        assert!(
            matches!(commit_rx.try_recv(), Ok(Err(ShardError::Spawn(_)))),
            "Commit arm"
        );
        assert!(
            matches!(range_rx.try_recv(), Ok(Err(ShardError::Spawn(_)))),
            "Range arm"
        );

        assert!(
            matches!(outcome, NativeOutcome::Stop(ExitReason::Error)),
            "sentinel stops"
        );
        assert!(
            commands.lock().map_err(|_| "queue poisoned")?.is_empty(),
            "queue fully drained"
        );
        Ok(())
    }
}

/// Group commit (audit E): the native drain loop's grouping/forwarding logic,
/// tested against a hand-built queue (deterministic, no scheduler race). These
/// prove `pop_next_unit` coalesces a CONSECUTIVE run of groupable writes and STOPS
/// at the first non-groupable command — so promise-state mutators (`RecordPromise`)
/// and friends are never swallowed into a group and keep their own slice.
#[cfg(test)]
mod group_drain_tests {
    use super::{DrainUnit, lock_queue, pop_next_unit};
    use crate::shard::actor::handle::{CommandQueue, ShardCommand, ShardCommandKind};
    use crate::sync::ballot::{Ballot, Stamp};
    use crate::sync::topology::SyncNodeId;
    use std::collections::VecDeque;
    use std::error::Error;
    use std::sync::mpsc;
    use std::sync::{Arc, Mutex};

    fn stamp() -> Stamp {
        Stamp::new(Ballot::new(1, SyncNodeId::new("owner")), 0)
    }

    fn apply_command(id: u64, key: &[u8]) -> ShardCommand {
        let (reply, _rx) = mpsc::sync_channel(1);
        ShardCommand {
            id,
            kind: ShardCommandKind::ApplyDurable {
                key: key.to_vec(),
                expected: None,
                value: b"v".to_vec(),
                ttl: None,
                stamp: stamp(),
                reply,
            },
        }
    }

    fn cas_command(id: u64, key: &[u8]) -> ShardCommand {
        let (reply, _rx) = mpsc::sync_channel(1);
        ShardCommand {
            id,
            kind: ShardCommandKind::Cas {
                key: key.to_vec(),
                expected: None,
                new: 1,
                reply,
            },
        }
    }

    fn promise_command(id: u64) -> ShardCommand {
        let (reply, _rx) = mpsc::sync_channel(1);
        ShardCommand {
            id,
            kind: ShardCommandKind::RecordPromise {
                ballot: Ballot::new(2, SyncNodeId::new("owner")),
                reply,
            },
        }
    }

    /// A consecutive run of groupable writes is coalesced into ONE group, the run
    /// STOPS at the non-groupable `RecordPromise` (which is left on the queue and
    /// drained on its own next), and the trailing groupable write forms its own
    /// group. This is the wiring-level proof that Prepare/promise are NOT coalesced.
    #[test]
    fn drain_groups_consecutive_writes_and_stops_at_non_groupable() -> Result<(), Box<dyn Error>> {
        let commands: CommandQueue = Arc::new(Mutex::new(VecDeque::new()));
        {
            let mut queue = lock_queue(&commands);
            queue.push_back(apply_command(1, b"a")); // groupable
            queue.push_back(cas_command(2, b"b")); // groupable
            queue.push_back(promise_command(3)); // NOT groupable — stops the run
            queue.push_back(apply_command(4, b"c")); // groupable (own group)
        }

        // First unit: a group of exactly the first TWO groupable writes.
        match pop_next_unit(&commands).ok_or("expected a first unit")? {
            DrainUnit::Group(group) => assert_eq!(group.len(), 2, "coalesce the leading run"),
            DrainUnit::Single(_) => return Err("expected a group, got a single".into()),
        }

        // Second unit: the non-groupable RecordPromise, on its own (NOT coalesced).
        match pop_next_unit(&commands).ok_or("expected a second unit")? {
            DrainUnit::Single(command) => assert!(
                matches!(command.kind, ShardCommandKind::RecordPromise { .. }),
                "RecordPromise must be handled as a single, never grouped"
            ),
            DrainUnit::Group(_) => return Err("RecordPromise must not be grouped".into()),
        }

        // Third unit: the trailing groupable write as its own group of one.
        match pop_next_unit(&commands).ok_or("expected a third unit")? {
            DrainUnit::Group(group) => assert_eq!(group.len(), 1, "trailing write groups alone"),
            DrainUnit::Single(_) => return Err("expected a trailing group".into()),
        }

        // Queue is now empty.
        assert!(pop_next_unit(&commands).is_none(), "queue fully drained");
        Ok(())
    }

    /// A non-groupable command at the FRONT is popped as a single, never wrapped in
    /// a group — even when groupable writes follow it.
    #[test]
    fn non_groupable_front_is_single_even_with_groupable_behind() -> Result<(), Box<dyn Error>> {
        let commands: CommandQueue = Arc::new(Mutex::new(VecDeque::new()));
        {
            let mut queue = lock_queue(&commands);
            queue.push_back(promise_command(1)); // NOT groupable, at the front
            queue.push_back(apply_command(2, b"a")); // groupable, behind it
        }
        match pop_next_unit(&commands).ok_or("expected a unit")? {
            DrainUnit::Single(command) => assert!(matches!(
                command.kind,
                ShardCommandKind::RecordPromise { .. }
            )),
            DrainUnit::Group(_) => return Err("front non-groupable must be single".into()),
        }
        match pop_next_unit(&commands).ok_or("expected a unit")? {
            DrainUnit::Group(group) => assert_eq!(group.len(), 1),
            DrainUnit::Single(_) => return Err("trailing write should group".into()),
        }
        Ok(())
    }
}