haematite 0.7.0

//! COMMIT-COLLAPSE §6: the bounded, reused worker pool.
//!
//! The retired `run_indexed_parallel` helper spawned one scoped OS
//! thread PER item on EVERY call and joined them all — O(materialised) thread
//! creation per commit, repeated forever (the AION host-resource incident, §0).
//! This pool replaces it: a fixed set of named workers
//! (`haematite-exec-{n}`), created once at startup, parked on a condvar when
//! idle (zero CPU, no timer, no fd), reused across every global commit and both
//! sequence scans.
//!
//! # Admission (§6 queue, ⟨r3 M8⟩)
//! A shared `Arc<Database>` can be driven by arbitrarily many caller threads
//! (`tests/multi_reader_writer.rs`), so an unbounded submit path could park
//! `O(callers × materialised)` heap-resident jobs. The pool therefore admits at
//! most [`MAX_ADMITTED_BATCHES`] batches at once, whole-batch and atomically;
//! submitters beyond the cap BLOCK on the admission gate. Admission is acquired
//! ([`Executor::admit`]) BEFORE the caller materialises any handle or job (the
//! `validate → admit → materialise → run` ordering the three call sites follow),
//! so a blocked submitter holds no materialised handles. Because every batch is
//! capped at `shard_count` jobs by construction (the scoped scan validates its
//! ids as a true set — a duplicate is a typed refusal), the queue memory ceiling
//! is a genuine `MAX_ADMITTED_BATCHES × shard_count × job size`.
//!
//! # Doc-4 reuse constraint (§12)
//! RECOVERY-BOOT will REUSE this pool for materialisation fan-out. It MUST NOT
//! nest a batch admission inside an already-admitted batch: all
//! `MAX_ADMITTED_BATCHES` permits could be held by callers waiting to admit
//! inner work — a permit-exhaustion starvation shape. Boot fan-out submits as
//! ordinary sibling batches or through a dedicated startup path.
//!
//! # Failure and shutdown
//! A job that panics is caught in the worker ([`std::panic::catch_unwind`]); the
//! batch surfaces [`DatabaseError::ShardError`] (the old helper's contract) and
//! the worker SURVIVES for the next job. On shutdown every blocked submitter is
//! woken with the typed [`DatabaseError::ExecutorShutdown`], admitted batches
//! drain to completion, then the workers are joined — driven from
//! `Database::drop` STRICTLY BEFORE `router.shutdown_all` because jobs hold
//! cloned shard handles.

use std::collections::VecDeque;
use std::panic::{self, AssertUnwindSafe};
use std::sync::mpsc::{self, RecvError};
use std::sync::{Arc, Condvar, Mutex};
use std::thread::{self, JoinHandle};

use super::DatabaseError;

/// The admission ceiling (§6, ⟨r3 M8⟩; signed default per §9.2): one active
/// batch per worker burst plus headroom for the two scan callers and one queued
/// commit. Bounds memory, not correctness. Not configurable in v1.
const MAX_ADMITTED_BATCHES: usize = 4;

/// A unit of executor work: an owned, `Send`, run-once closure.
type Job = Box<dyn FnOnce() + Send + 'static>;

/// Worker-visible queue state, guarded by [`Shared::queue`].
struct QueueState {
    jobs: VecDeque<Job>,
    /// Set once at shutdown; a worker returns when the queue is drained AND this
    /// is set (admitted batches drain to completion before any worker exits).
    shutting_down: bool,
    /// Jobs currently executing on a worker — the running term of the
    /// queued-and-running gauge the admission-ceiling test asserts against.
    running: usize,
}

/// Submitter-visible admission state, guarded by [`Shared::admission`].
struct AdmissionState {
    admitted: usize,
    /// Set once at shutdown so blocked submitters wake and refuse typed.
    shutting_down: bool,
}

/// State shared between the [`Executor`] handle and its worker threads.
struct Shared {
    queue: Mutex<QueueState>,
    /// Signalled when a job is pushed OR shutdown begins.
    queue_ready: Condvar,
    admission: Mutex<AdmissionState>,
    /// Signalled when a permit is released OR shutdown begins.
    admission_free: Condvar,
    max_admitted_batches: usize,
}

impl Shared {
    fn new(max_admitted_batches: usize) -> Self {
        Self {
            queue: Mutex::new(QueueState {
                jobs: VecDeque::new(),
                shutting_down: false,
                running: 0,
            }),
            queue_ready: Condvar::new(),
            admission: Mutex::new(AdmissionState {
                admitted: 0,
                shutting_down: false,
            }),
            admission_free: Condvar::new(),
            max_admitted_batches,
        }
    }

    /// Begin shutdown: refuse further admissions, wake blocked submitters and
    /// parked workers. Admitted batches already in `jobs` still drain.
    fn begin_shutdown(&self) {
        if let Ok(mut admission) = self.admission.lock() {
            admission.shutting_down = true;
        }
        self.admission_free.notify_all();
        if let Ok(mut queue) = self.queue.lock() {
            queue.shutting_down = true;
        }
        self.queue_ready.notify_all();
    }
}

/// One admitted batch. Releasing it (drop — on success, error, or unwind) frees
/// an admission slot and wakes one blocked submitter, so the permit is released
/// on EVERY exit path (§11 permit-release pins).
struct BatchPermit {
    shared: Arc<Shared>,
}

impl Drop for BatchPermit {
    fn drop(&mut self) {
        if let Ok(mut admission) = self.shared.admission.lock() {
            admission.admitted = admission.admitted.saturating_sub(1);
        }
        self.shared.admission_free.notify_one();
    }
}

/// The bounded worker pool. Database-owned; outlived by nothing it borrows.
///
/// `pub` (not `pub(crate)`) because the enclosing `executor` module is already
/// `pub(crate)` in `db.rs`, which restricts this to the crate (the house
/// convention `db/helpers.rs` follows — a `pub(crate)` item here trips
/// `clippy::redundant_pub_crate`).
pub struct Executor {
    shared: Arc<Shared>,
    workers: Vec<JoinHandle<()>>,
    /// COMMIT-COLLAPSE §11 zero-dirty / one-dirty gauge: total jobs ever dispatched
    /// across the pool's lifetime. A zero-dirty global commit must not dispatch a
    /// single job (the classification hands `run_batch` an empty item list); a
    /// one-dirty commit dispatches exactly one. Test-only; compiled out of release.
    #[cfg(test)]
    dispatched_jobs: std::sync::atomic::AtomicUsize,
}

impl Executor {
    /// Spawn `worker_count` named workers. On a partial spawn failure the
    /// already-spawned workers are shut down and joined before the error returns
    /// (§6 partial startup failure).
    pub(crate) fn start(worker_count: usize) -> Result<Self, DatabaseError> {
        Self::start_with_admission(worker_count, MAX_ADMITTED_BATCHES)
    }

    /// Start with an explicit admission cap (test seam for the ceiling and
    /// shutdown pins, which need a small cap to reach the ceiling deterministically).
    #[cfg(test)]
    pub(crate) fn start_with_admission_for_test(
        worker_count: usize,
        max_admitted_batches: usize,
    ) -> Result<Self, DatabaseError> {
        Self::start_with_admission(worker_count, max_admitted_batches)
    }

    fn start_with_admission(
        worker_count: usize,
        max_admitted_batches: usize,
    ) -> Result<Self, DatabaseError> {
        let shared = Arc::new(Shared::new(max_admitted_batches));
        let mut workers = Vec::with_capacity(worker_count);
        for index in 0..worker_count {
            let worker_shared = Arc::clone(&shared);
            let spawned = thread::Builder::new()
                .name(format!("haematite-exec-{index}"))
                .spawn(move || worker_loop(&worker_shared));
            match spawned {
                Ok(handle) => workers.push(handle),
                Err(error) => {
                    shared.begin_shutdown();
                    for handle in workers {
                        drop(handle.join());
                    }
                    return Err(DatabaseError::ExecutorThreadsInvalid(format!(
                        "failed to spawn executor worker {index}: {error}"
                    )));
                }
            }
        }
        Ok(Self {
            shared,
            workers,
            #[cfg(test)]
            dispatched_jobs: std::sync::atomic::AtomicUsize::new(0),
        })
    }

    /// Acquire a whole-batch admission permit, BLOCKING while the ceiling is
    /// reached; refuses [`DatabaseError::ExecutorShutdown`] once shutdown began.
    /// Called BEFORE the caller materialises the batch's handles/jobs.
    fn admit(&self) -> Result<BatchPermit, DatabaseError> {
        let mut admission = self
            .shared
            .admission
            .lock()
            .map_err(|_| DatabaseError::ExecutorShutdown)?;
        loop {
            if admission.shutting_down {
                return Err(DatabaseError::ExecutorShutdown);
            }
            if admission.admitted < self.shared.max_admitted_batches {
                admission.admitted += 1;
                return Ok(BatchPermit {
                    shared: Arc::clone(&self.shared),
                });
            }
            admission = self
                .shared
                .admission_free
                .wait(admission)
                .map_err(|_| DatabaseError::ExecutorShutdown)?;
        }
    }

    /// Run `materialise` (which builds the batch's items) AFTER admission, then
    /// fan its items across the pool and collect their outputs, each tagged with
    /// its input index (`items[i]` -> `(i, work(items[i]))`), preserving the old
    /// helper's index↔result association and all-work-completes-before-return.
    ///
    /// The admission permit is held across materialisation, dispatch, and
    /// collection, and released when this returns — so a materialisation error
    /// releases it (RAII on the early `?`), and so does a caller unwind.
    pub(crate) fn submit<Item, Output, Materialise, Work>(
        &self,
        materialise: Materialise,
        work: Work,
    ) -> Result<Vec<(usize, Output)>, DatabaseError>
    where
        Item: Send + 'static,
        Output: Send + 'static,
        Materialise: FnOnce() -> Result<Vec<Item>, DatabaseError>,
        Work: Fn(Item) -> Output + Send + Sync + 'static,
    {
        let permit = self.admit()?;
        let outcome = materialise().and_then(|items| self.run_batch(items, work));
        drop(permit);
        outcome
    }

    /// Dispatch `items` across the pool and collect. A worker panic on any item
    /// converts to [`DatabaseError::ShardError`] for the whole batch (the old
    /// helper's contract) and the worker survives for later batches.
    fn run_batch<Item, Output, Work>(
        &self,
        items: Vec<Item>,
        work: Work,
    ) -> Result<Vec<(usize, Output)>, DatabaseError>
    where
        Item: Send + 'static,
        Output: Send + 'static,
        Work: Fn(Item) -> Output + Send + Sync + 'static,
    {
        let job_count = items.len();
        if job_count == 0 {
            return Ok(Vec::new());
        }
        #[cfg(test)]
        self.dispatched_jobs
            .fetch_add(job_count, std::sync::atomic::Ordering::Relaxed);
        let work = Arc::new(work);
        let (results_tx, results_rx) = mpsc::channel::<(usize, Result<Output, ()>)>();
        {
            let mut queue = self
                .shared
                .queue
                .lock()
                .map_err(|_| DatabaseError::ShardError("executor queue poisoned".to_owned()))?;
            for (index, item) in items.into_iter().enumerate() {
                let work = Arc::clone(&work);
                let results_tx = results_tx.clone();
                queue.jobs.push_back(Box::new(move || {
                    let outcome = panic::catch_unwind(AssertUnwindSafe(|| work(item)));
                    // A closed receiver means the collector already unwound; drop.
                    drop(results_tx.send((index, outcome.map_err(drop))));
                }));
            }
        }
        // Drop our own sender clone so the channel closes once every job replied.
        drop(results_tx);
        self.shared.queue_ready.notify_all();
        collect_batch(&results_rx, job_count)
    }

    /// The number of admitted batches right now (test gauge for the admission
    /// ceiling pin).
    #[cfg(test)]
    pub(crate) fn admitted_batches(&self) -> usize {
        self.shared
            .admission
            .lock()
            .map(|admission| admission.admitted)
            .unwrap_or(0)
    }

    /// Queued + currently-running jobs right now (test gauge: the value the
    /// admission ceiling bounds at `MAX_ADMITTED_BATCHES × shard_count`).
    #[cfg(test)]
    pub(crate) fn queued_and_running(&self) -> usize {
        self.shared
            .queue
            .lock()
            .map(|queue| queue.jobs.len() + queue.running)
            .unwrap_or(0)
    }

    /// The number of live workers (test gauge for the bounded-threads census).
    #[cfg(test)]
    pub(crate) const fn worker_count(&self) -> usize {
        self.workers.len()
    }

    /// The per-worker-burst admission cap (test gauge: the ceiling on
    /// queued+running is this × `shard_count`).
    #[cfg(test)]
    pub(crate) fn max_admitted_batches(&self) -> usize {
        self.shared.max_admitted_batches
    }

    /// Total jobs dispatched across the pool's lifetime (§11 zero-dirty /
    /// one-dirty gauge). Zero after a zero-dirty commit; exactly one more after a
    /// one-dirty commit.
    #[cfg(test)]
    pub(crate) fn dispatched_jobs(&self) -> usize {
        self.dispatched_jobs
            .load(std::sync::atomic::Ordering::Relaxed)
    }

    /// Begin shutdown WITHOUT joining (test seam): wakes blocked submitters with
    /// the typed error so the "shutdown wakes blocked submitter" pin can observe
    /// the wake before dropping the executor (whose [`Drop`] joins the workers).
    #[cfg(test)]
    pub(crate) fn trigger_shutdown(&self) {
        self.shared.begin_shutdown();
    }

    /// Drain admitted batches to completion and join every worker. Idempotent;
    /// called explicitly from `Database::drop` before `router.shutdown_all`, and
    /// again by [`Drop`].
    pub(crate) fn shutdown(&mut self) {
        self.shared.begin_shutdown();
        for handle in self.workers.drain(..) {
            drop(handle.join());
        }
    }
}

impl Drop for Executor {
    fn drop(&mut self) {
        self.shutdown();
    }
}

/// Collect exactly `job_count` positional results, converting a worker panic on
/// any job into a batch-level [`DatabaseError::ShardError`]. Outputs are returned
/// index-tagged and index-ordered (deterministic error/result order).
fn collect_batch<Output>(
    results_rx: &mpsc::Receiver<(usize, Result<Output, ()>)>,
    job_count: usize,
) -> Result<Vec<(usize, Output)>, DatabaseError> {
    let mut slots: Vec<Option<Output>> = (0..job_count).map(|_| None).collect();
    let mut panicked = false;
    for _ in 0..job_count {
        match results_rx.recv() {
            Ok((index, Ok(output))) => {
                if let Some(slot) = slots.get_mut(index) {
                    *slot = Some(output);
                }
            }
            Ok((_, Err(()))) => panicked = true,
            Err(RecvError) => {
                return Err(DatabaseError::ShardError(
                    "executor worker dropped a batch result before replying".to_owned(),
                ));
            }
        }
    }
    if panicked {
        return Err(DatabaseError::ShardError(
            "parallel worker thread panicked".to_owned(),
        ));
    }
    Ok(slots
        .into_iter()
        .enumerate()
        .filter_map(|(index, output)| output.map(|output| (index, output)))
        .collect())
}

/// The worker body: pop one job under the queue lock, release the lock, run the
/// job (its own `catch_unwind` keeps a panic from ever unwinding the worker),
/// repeat; return once the queue is drained and shutdown has begun.
fn worker_loop(shared: &Arc<Shared>) {
    loop {
        let job = {
            // Poison-exit: unreachable in practice — nothing under the queue lock
            // can panic (only `push_back`/`pop_front` and counter arithmetic run
            // while it is held, never user code), so the guard is never poisoned.
            // Exiting here (rather than adopting the guard) would strand queued
            // jobs and hang their collector, but that path cannot be taken.
            let Ok(mut queue) = shared.queue.lock() else {
                return;
            };
            loop {
                if let Some(job) = queue.jobs.pop_front() {
                    queue.running += 1;
                    break job;
                }
                if queue.shutting_down {
                    return;
                }
                queue = match shared.queue_ready.wait(queue) {
                    Ok(queue) => queue,
                    Err(_) => return,
                };
            }
        };
        job();
        if let Ok(mut queue) = shared.queue.lock() {
            queue.running = queue.running.saturating_sub(1);
        }
    }
}

/// Resolve the worker count from the configured knob and the shard count (§6
/// sizing, §9.2). `Some(0)` and an unavailable `available_parallelism()` are
/// typed refusals naming the remedy; `None` is the signed default
/// `min(shard_count, available_parallelism)`.
pub(super) fn resolve_worker_count(
    configured: Option<usize>,
    shard_count: usize,
) -> Result<usize, DatabaseError> {
    match configured {
        Some(0) => Err(DatabaseError::ExecutorThreadsInvalid(
            "executor_threads = 0 is not a valid worker count; set a positive integer or omit \
             the field for the signed default min(shard_count, available_parallelism)"
                .to_owned(),
        )),
        Some(threads) => Ok(threads),
        None => {
            let parallelism = thread::available_parallelism().map_err(|error| {
                DatabaseError::ExecutorThreadsInvalid(format!(
                    "available_parallelism could not be determined ({error}); set executor_threads \
                     explicitly to choose the worker count"
                ))
            })?;
            Ok(shard_count.min(parallelism.get()).max(1))
        }
    }
}

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