haematite 0.7.0

//! COMMIT-COLLAPSE §6/§11 executor semantics pins.
//!
//! These exercise the bounded pool directly (no `Database`): index↔result
//! association, empty batches, worker-panic → `ShardError` with worker survival,
//! the admission ceiling holding under a submitter pile-up with no starvation,
//! and shutdown waking a blocked submitter with the typed error. The
//! multi-worker OVERLAP pin lives in `db_tests.rs`
//! (`executor_invokes_multiple_workers_concurrently`, the ported concurrency
//! pin).

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};

use super::{DatabaseError, Executor};

/// A one-shot release gate so a job can be held mid-flight deterministically
/// (no sleeps): workers park on it until the test opens it.
struct Gate {
    open: Mutex<bool>,
    ready: Condvar,
}

impl Gate {
    fn new() -> Arc<Self> {
        Arc::new(Self {
            open: Mutex::new(false),
            ready: Condvar::new(),
        })
    }

    fn wait(&self) {
        let mut open = match self.open.lock() {
            Ok(open) => open,
            Err(poisoned) => poisoned.into_inner(),
        };
        while !*open {
            open = match self.ready.wait(open) {
                Ok(open) => open,
                Err(poisoned) => poisoned.into_inner(),
            };
        }
        drop(open);
    }

    fn release(&self) {
        if let Ok(mut open) = self.open.lock() {
            *open = true;
        }
        self.ready.notify_all();
    }
}

#[test]
fn preserves_index_result_association_and_order() -> Result<(), DatabaseError> {
    let executor = Executor::start(3)?;
    assert_eq!(
        executor.worker_count(),
        3,
        "the pool has a fixed worker count"
    );
    let mut results = executor.submit(|| Ok(vec![10_u8, 20, 30, 40]), |value: u8| value + 1)?;
    results.sort_by_key(|(index, _)| *index);
    assert_eq!(results, vec![(0, 11), (1, 21), (2, 31), (3, 41)]);
    Ok(())
}

#[test]
fn empty_batch_returns_empty_without_dispatch() -> Result<(), DatabaseError> {
    let executor = Executor::start(2)?;
    let results: Vec<(usize, u8)> = executor.submit(|| Ok(Vec::new()), |value: u8| value)?;
    assert!(results.is_empty());
    Ok(())
}

#[test]
fn materialise_error_releases_permit() -> Result<(), DatabaseError> {
    let executor = Executor::start(1)?;
    // A batch whose materialisation fails must not leak an admission slot: the
    // permit is RAII-released on the early return, so the ceiling gauge returns
    // to baseline and a later batch still admits.
    let outcome: Result<Vec<(usize, u8)>, DatabaseError> = executor.submit(
        || Err(DatabaseError::DuplicateShardId { shard_id: 7 }),
        |value: u8| value,
    );
    assert!(matches!(
        outcome,
        Err(DatabaseError::DuplicateShardId { shard_id: 7 })
    ));
    assert_eq!(executor.admitted_batches(), 0);
    let results = executor.submit(|| Ok(vec![1_u8]), |value: u8| value)?;
    assert_eq!(results, vec![(0, 1)]);
    Ok(())
}

#[test]
fn worker_panic_becomes_shard_error_and_worker_survives() -> Result<(), DatabaseError> {
    let executor = Executor::start(2)?;
    // One job panics: the batch surfaces ShardError (the old helper's contract).
    let panicking: Result<Vec<(usize, u8)>, DatabaseError> = executor.submit(
        || Ok(vec![0_u8, 1]),
        |value: u8| {
            assert!(value < 1, "job {value} panics on purpose");
            value
        },
    );
    assert!(matches!(panicking, Err(DatabaseError::ShardError(_))));
    // The workers SURVIVE: a subsequent normal batch completes, and no permit
    // leaked from the panicking batch.
    assert_eq!(executor.admitted_batches(), 0);
    let results = executor.submit(|| Ok(vec![5_u8, 6]), |value: u8| value + 1)?;
    let mut results = results;
    results.sort_by_key(|(index, _)| *index);
    assert_eq!(results, vec![(0, 6), (1, 7)]);
    Ok(())
}

#[test]
fn admission_ceiling_holds_and_all_submitters_complete() -> Result<(), DatabaseError> {
    // Cap = 2, workers = 4, six concurrent submitters each holding their job on
    // the gate. At most 2 batches are ever admitted, so queued+running never
    // exceeds 2 × batch-size; releasing the gate lets every submitter finish.
    let executor = Arc::new(Executor::start_with_admission_for_test(4, 2)?);
    let gate = Gate::new();
    let done = Arc::new(AtomicUsize::new(0));
    let mut submitters = Vec::new();
    for _ in 0..6 {
        let executor = Arc::clone(&executor);
        let gate = Arc::clone(&gate);
        let done = Arc::clone(&done);
        submitters.push(std::thread::spawn(move || {
            let outcome = executor.submit(
                || Ok(vec![()]),
                move |()| {
                    gate.wait();
                },
            );
            if outcome.is_ok() {
                done.fetch_add(1, Ordering::SeqCst);
            }
        }));
    }

    // Poll the ceiling for a while: it must never exceed the cap.
    let deadline = Instant::now() + Duration::from_secs(2);
    let mut observed_at_cap = false;
    while Instant::now() < deadline {
        let admitted = executor.admitted_batches();
        assert!(
            admitted <= 2,
            "admitted batches {admitted} exceeded the cap"
        );
        assert!(
            executor.queued_and_running() <= 2,
            "queued+running exceeded cap × batch-size"
        );
        if admitted == 2 {
            observed_at_cap = true;
        }
        if observed_at_cap {
            break;
        }
        std::thread::yield_now();
    }
    assert!(
        observed_at_cap,
        "the ceiling was never reached under 6 submitters"
    );

    gate.release();
    for submitter in submitters {
        submitter
            .join()
            .map_err(|_| DatabaseError::ShardError("submitter thread panicked".to_owned()))?;
    }
    assert_eq!(
        done.load(Ordering::SeqCst),
        6,
        "every submitter must complete"
    );
    assert_eq!(executor.admitted_batches(), 0, "all permits released");
    Ok(())
}

#[test]
fn shutdown_wakes_blocked_submitter_with_typed_error() -> Result<(), DatabaseError> {
    // Cap = 1, worker = 1: the first submitter holds the only permit (its job
    // parks on the gate); a second submitter blocks on admission; shutdown wakes
    // it with the typed ExecutorShutdown while the admitted batch drains.
    let executor = Arc::new(Executor::start_with_admission_for_test(1, 1)?);
    let gate = Gate::new();

    let holder_executor = Arc::clone(&executor);
    let holder_gate = Arc::clone(&gate);
    let holder = std::thread::spawn(move || {
        holder_executor.submit(
            || Ok(vec![()]),
            move |()| {
                holder_gate.wait();
            },
        )
    });

    // Wait until the holder has taken the single admission slot.
    let deadline = Instant::now() + Duration::from_secs(2);
    while executor.admitted_batches() < 1 {
        assert!(Instant::now() < deadline, "holder never admitted");
        std::thread::yield_now();
    }

    let blocked_executor = Arc::clone(&executor);
    let blocked = std::thread::spawn(move || blocked_executor.submit(|| Ok(vec![()]), |()| {}));

    // Give the blocked submitter a moment to park on admission, then shut down.
    std::thread::yield_now();
    executor.trigger_shutdown();

    let blocked_outcome = blocked
        .join()
        .map_err(|_| DatabaseError::ShardError("blocked submitter panicked".to_owned()))?;
    assert!(
        matches!(blocked_outcome, Err(DatabaseError::ExecutorShutdown)),
        "a submitter blocked at shutdown must receive the typed ExecutorShutdown"
    );

    // Release the drained batch so the holder finishes and the pool joins.
    gate.release();
    let holder_outcome = holder
        .join()
        .map_err(|_| DatabaseError::ShardError("holder submitter panicked".to_owned()))?;
    assert!(
        holder_outcome.is_ok(),
        "the admitted batch drains to completion"
    );
    Ok(())
}