eidetic-engine 0.15.2

use std::collections::BTreeMap;
use std::ops::Deref;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, MutexGuard, OnceLock, RwLock, Weak};
use std::time::{Duration, Instant};

use super::{DatabaseConfig, DatabaseLocation, DbConnection, DbError, DbOperation, Result};

// SnapshotPin is intentionally a read transaction, not a write-owner transaction.
// The local adapter path is `src/db/mod.rs`: `begin_read_snapshot()` executes
// `BEGIN DEFERRED` through `execute_read_snapshot_raw()`, whose file-backed path
// retries SQLite contention but does not take `lock_file_write_owner_gate()`.
// The upstream SQLModel/FrankenSQLite path is
// `/dp/sqlmodel_rust/crates/sqlmodel-frankensqlite/src/connection.rs`:
// `FrankenConnection::execute_raw()` forwards the SQL batch to
// `fsqlite::Connection::execute()`. This is the primitive that lets a
// SnapshotPin hold a stable reader view while same-process writers commit.
const DEFAULT_MAX_SIZE: usize = 1;
const DEFAULT_IDLE_TIMEOUT: Duration = Duration::from_secs(30);
const DEFAULT_MAX_PIN_DURATION: Duration = Duration::from_secs(30);
const DEFAULT_ACQUIRE_TIMEOUT: Duration = Duration::from_secs(5);
const ACQUIRE_WAIT_SAMPLE_CAP: usize = 1024;
const ACQUIRE_SLEEP_STEP: Duration = Duration::from_millis(1);

pub const SNAPSHOT_PIN_EXPIRED_CODE: &str = "snapshot_pin_expired";
pub const SNAPSHOT_RELEASE_FAILED_CODE: &str = "snapshot_release_failed";
pub const SNAPSHOT_PIN_FORCE_RELEASED_CODE: &str = "snapshot_pin_force_released";
pub const READ_POOL_ACQUIRE_TIMEOUT_CODE: &str = "read_pool_acquire_timeout";
pub const READ_POOL_UNDERSIZED_CODE: &str = "read_pool_undersized";
pub const READ_POOL_UNDERSIZED_SAMPLE_FLOOR: usize = ACQUIRE_WAIT_SAMPLE_CAP;
pub const READ_POOL_UNDERSIZED_P99_THRESHOLD: Duration = Duration::from_millis(10);

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PoolConfig {
    max_size: usize,
    idle_timeout: Duration,
    max_pin_duration: Duration,
    acquire_timeout: Duration,
}

impl PoolConfig {
    #[must_use]
    pub const fn new(max_size: usize, idle_timeout: Duration) -> Self {
        Self {
            max_size,
            idle_timeout,
            max_pin_duration: DEFAULT_MAX_PIN_DURATION,
            acquire_timeout: DEFAULT_ACQUIRE_TIMEOUT,
        }
    }

    #[must_use]
    pub const fn default_single() -> Self {
        Self {
            max_size: DEFAULT_MAX_SIZE,
            idle_timeout: DEFAULT_IDLE_TIMEOUT,
            max_pin_duration: DEFAULT_MAX_PIN_DURATION,
            acquire_timeout: DEFAULT_ACQUIRE_TIMEOUT,
        }
    }

    #[must_use]
    pub const fn with_max_pin_duration(mut self, max_pin_duration: Duration) -> Self {
        self.max_pin_duration = max_pin_duration;
        self
    }

    #[must_use]
    pub const fn with_acquire_timeout(mut self, acquire_timeout: Duration) -> Self {
        self.acquire_timeout = acquire_timeout;
        self
    }

    #[must_use]
    pub const fn requested_max_size(&self) -> usize {
        self.max_size
    }

    #[must_use]
    pub const fn max_size(&self) -> usize {
        if self.max_size == 0 { 1 } else { self.max_size }
    }

    #[must_use]
    pub const fn size_was_zero(&self) -> bool {
        self.max_size == 0
    }

    #[must_use]
    pub const fn idle_timeout(&self) -> Duration {
        self.idle_timeout
    }

    #[must_use]
    pub const fn max_pin_duration(&self) -> Duration {
        self.max_pin_duration
    }

    #[must_use]
    pub const fn acquire_timeout(&self) -> Duration {
        self.acquire_timeout
    }
}

impl Default for PoolConfig {
    fn default() -> Self {
        Self::default_single()
    }
}

#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct PoolStats {
    pub active: usize,
    pub idle: usize,
    pub active_pins: usize,
    pub expired_pins: usize,
    pub max_size: usize,
    pub max_seen: usize,
    pub drops: u64,
    pub release_failures: u64,
    pub ad_hoc_bypass_count: u64,
    pub acquire_wait: AcquireWaitStats,
    pub size_was_zero: bool,
    /// Forward-looking checkpoint blocker attribution: if a
    /// `wal_checkpoint(PASSIVE|TRUNCATE)` were attempted at the moment `stats()`
    /// was called and returned BUSY, this is the most likely blocker — the
    /// oldest active `SnapshotPin` known to the pool. `None` when no pins are
    /// currently active.
    ///
    /// Pool state is process-local: a pin held by a different process is not
    /// visible here. Use this field as a same-process attribution signal; for
    /// cross-process readers, the SQLite WAL header itself carries the active
    /// reader frame range but not pin identity.
    pub checkpoint_blocked_by: Option<CheckpointBlocker>,
}

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct AcquireWaitStats {
    pub samples: usize,
    pub p50_ns: u128,
    pub p99_ns: u128,
}

pub struct ReadConnectionPool {
    database: DatabaseConfig,
    config: PoolConfig,
    state: Mutex<PoolState>,
}

// bd-fv0yn: RwLock (was Mutex) so cache-hit reads via
// `process_read_pool_for_database` and `registered_process_read_pool`'s
// Weak-upgrade fast path take `.read()` and run concurrently. Stale-
// Weak removal (formerly on the read path) is deferred to the next
// `registered_process_read_pool` write-path's opportunistic retain.
// Sibling to bd-1nan9 / bd-2lin9 / bd-25yao / bd-2r38i / bd-8tsi5 /
// bd-3mr0x.
static PROCESS_READ_POOL_REGISTRY: OnceLock<RwLock<BTreeMap<String, Weak<ReadConnectionPool>>>> =
    OnceLock::new();

#[derive(Clone, Debug)]
pub struct ReadConnectionPoolBuilder {
    database: DatabaseConfig,
    config: PoolConfig,
}

struct PoolState {
    active: usize,
    idle: Vec<IdleConnection>,
    acquire_wait_ns: Vec<u128>,
    next_slot_id: u64,
    next_pin_id: u64,
    active_pins: BTreeMap<u64, ActivePinRecord>,
    max_seen: usize,
    drops: u64,
    release_failures: u64,
    ad_hoc_bypass_count: u64,
}

fn allocate_next_read_pool_id(next_id: &mut u64, id_name: &str) -> u64 {
    let id = *next_id;
    *next_id = id
        .checked_add(1)
        .unwrap_or_else(|| panic!("read-pool {id_name} id exhausted at u64::MAX"));
    id
}

struct ActivePinRecord {
    slot_id: Option<u64>,
    metadata: SnapshotPinMetadata,
    acquired_at: Instant,
    max_pin_duration: Duration,
    poisoned: Arc<AtomicBool>,
}

struct IdleConnection {
    slot_id: u64,
    connection: DbConnection,
    returned_at: Instant,
}

pub struct PooledReadConnection<'pool> {
    pool: Option<&'pool ReadConnectionPool>,
    slot_id: Option<u64>,
    connection: Option<DbConnection>,
}

pub struct SnapshotPin<'pool> {
    pool: Option<&'pool ReadConnectionPool>,
    connection: Option<PooledReadConnection<'pool>>,
    snapshot_active: bool,
    pin_id: Option<u64>,
    poisoned: Arc<AtomicBool>,
    acquired_at: Instant,
    max_pin_duration: Duration,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ExpiredSnapshotPin {
    pub pin_id: u64,
    pub slot_id: Option<u64>,
    pub age: Duration,
}

#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct SnapshotPinMetadata {
    pub workflow_id: Option<String>,
    pub request_id: Option<String>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct SnapshotDrainReport {
    pub drained: bool,
    pub waited: Duration,
    pub active_pins_remaining: usize,
    pub force_poisoned: Vec<ExpiredSnapshotPin>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ActiveSnapshotPin {
    pub pin_id: u64,
    pub slot_id: Option<u64>,
    pub workflow_id: Option<String>,
    pub request_id: Option<String>,
    pub pin_age_ms: u128,
    pub max_pin_duration_ms: u128,
    pub poisoned: bool,
    pub release_state: SnapshotPinReleaseState,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum SnapshotPinReleaseState {
    Active,
    Expired,
    Poisoned,
}

/// Checkpoint-blocker attribution: the oldest active SnapshotPin holding a
/// WAL reader reference at the moment a WAL checkpoint reported BUSY.
///
/// Populated by [`ReadConnectionPool::oldest_active_pin_for_checkpoint_blocker`].
/// The workspace identifier is attached at the status-report layer; the pool
/// itself only owns the per-pin metadata.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct CheckpointBlocker {
    pub pin_id: u64,
    pub slot_id: Option<u64>,
    pub workflow_id: Option<String>,
    pub request_id: Option<String>,
    pub pin_age_ms: u128,
    pub max_pin_duration_ms: u128,
    pub poisoned: bool,
    pub release_state: SnapshotPinReleaseState,
}

/// Process-local checkpoint attribution from the read-pool registry.
///
/// This never claims cross-process visibility. A standalone `ee maintenance`
/// invocation can only see pools registered in the same OS process; readers in
/// other CLI processes may still pin WAL pages but do not expose SnapshotPin
/// identity through this registry.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ProcessCheckpointOutcome {
    pub checkpoint_busy: bool,
    pub process_pool_visible: bool,
    pub blocker: Option<CheckpointBlocker>,
}

impl From<ActiveSnapshotPin> for CheckpointBlocker {
    fn from(pin: ActiveSnapshotPin) -> Self {
        Self {
            pin_id: pin.pin_id,
            slot_id: pin.slot_id,
            workflow_id: pin.workflow_id,
            request_id: pin.request_id,
            pin_age_ms: pin.pin_age_ms,
            max_pin_duration_ms: pin.max_pin_duration_ms,
            poisoned: pin.poisoned,
            release_state: pin.release_state,
        }
    }
}

impl ReadConnectionPool {
    #[must_use]
    pub fn builder(database: DatabaseConfig) -> ReadConnectionPoolBuilder {
        ReadConnectionPoolBuilder::new(database)
    }

    #[must_use]
    pub fn new(database: DatabaseConfig, config: PoolConfig) -> Self {
        let database = read_pool_database_config(database);
        Self {
            database,
            config,
            state: Mutex::new(PoolState {
                active: 0,
                idle: Vec::new(),
                acquire_wait_ns: Vec::with_capacity(ACQUIRE_WAIT_SAMPLE_CAP),
                next_slot_id: 1,
                next_pin_id: 1,
                active_pins: BTreeMap::new(),
                max_seen: 0,
                drops: 0,
                release_failures: 0,
                ad_hoc_bypass_count: 0,
            }),
        }
    }

    #[must_use]
    pub fn config(&self) -> &PoolConfig {
        &self.config
    }

    pub fn warm(&self, count: usize) -> Result<usize> {
        let target = count.min(self.config.max_size());
        let mut warmed = Vec::with_capacity(target);
        for _ in 0..target {
            let connection = self.acquire()?;
            if connection.is_ad_hoc() {
                break;
            }
            warmed.push(connection);
        }
        let warmed_count = warmed.len();
        drop(warmed);
        Ok(warmed_count)
    }

    pub fn acquire(&self) -> Result<PooledReadConnection<'_>> {
        let max_size = self.config.max_size();
        let started = Instant::now();

        loop {
            let mut state = self.lock_state();
            let stale = evict_expired_idle(&mut state, self.config.idle_timeout());
            let occupancy = match pool_occupancy(&state) {
                Ok(occupancy) => occupancy,
                Err(error) => {
                    drop(state);
                    drop_idle_connections(stale);
                    return Err(error);
                }
            };

            if let Some(idle) = state.idle.pop() {
                if let Err(error) = increment_pool_active(&mut state) {
                    drop(state);
                    drop_idle_connections(stale);
                    return Err(error);
                }
                state.max_seen = state.max_seen.max(occupancy);
                record_acquire_wait(&mut state, started.elapsed());
                drop(state);
                drop_idle_connections(stale);
                return Ok(PooledReadConnection {
                    pool: Some(self),
                    slot_id: Some(idle.slot_id),
                    connection: Some(idle.connection),
                });
            }

            if occupancy < max_size {
                let slot_id = allocate_next_read_pool_id(&mut state.next_slot_id, "slot");
                if let Err(error) = increment_pool_active(&mut state) {
                    drop(state);
                    drop_idle_connections(stale);
                    return Err(error);
                }
                state.max_seen = state.max_seen.max(occupancy + 1);
                record_acquire_wait(&mut state, started.elapsed());
                drop(state);

                drop_idle_connections(stale);
                return match DbConnection::open(self.database.clone()) {
                    Ok(connection) => Ok(PooledReadConnection {
                        pool: Some(self),
                        slot_id: Some(slot_id),
                        connection: Some(connection),
                    }),
                    Err(error) => {
                        let mut state = self.lock_state();
                        decrement_pool_active(&mut state, "open failure");
                        Err(error)
                    }
                };
            }

            let elapsed = started.elapsed();
            if elapsed >= self.config.acquire_timeout() {
                state.ad_hoc_bypass_count = checked_increment_pool_counter(
                    state.ad_hoc_bypass_count,
                    "ad_hoc_bypass_count",
                );
                record_acquire_wait(&mut state, elapsed);
                drop(state);
                drop_idle_connections(stale);
                return DbConnection::open(self.database.clone()).map(|connection| {
                    PooledReadConnection {
                        pool: None,
                        slot_id: None,
                        connection: Some(connection),
                    }
                });
            }

            drop(state);
            drop_idle_connections(stale);
            let remaining = self.config.acquire_timeout().saturating_sub(elapsed);
            std::thread::yield_now();
            std::thread::sleep(remaining.min(ACQUIRE_SLEEP_STEP));
        }
    }

    pub fn pin_snapshot(&self) -> Result<SnapshotPin<'_>> {
        self.acquire_snapshot(true)
    }

    pub fn pin_snapshot_with_metadata(
        &self,
        metadata: SnapshotPinMetadata,
    ) -> Result<SnapshotPin<'_>> {
        self.acquire_snapshot_with_metadata(true, metadata)
    }

    pub fn acquire_snapshot(&self, pin_snapshot: bool) -> Result<SnapshotPin<'_>> {
        self.acquire_snapshot_with_metadata(pin_snapshot, SnapshotPinMetadata::default())
    }

    fn acquire_snapshot_with_metadata(
        &self,
        pin_snapshot: bool,
        metadata: SnapshotPinMetadata,
    ) -> Result<SnapshotPin<'_>> {
        let connection = self.acquire()?;
        if pin_snapshot {
            if let Err(error) = connection.begin_read_snapshot() {
                if let Err(rollback_error) = connection.rollback_read_snapshot() {
                    tracing::error!(
                        phase = "db_read_pool_begin_snapshot",
                        error = %error,
                        rollback_error = %rollback_error,
                        "failed to rollback read snapshot after begin failure"
                    );
                }
                return Err(error);
            }
        }

        let acquired_at = Instant::now();
        let (pin_id, poisoned) = if pin_snapshot {
            let (pin_id, poisoned) = self.register_pin(
                connection.slot_id(),
                metadata,
                acquired_at,
                self.config.max_pin_duration(),
            );
            (Some(pin_id), poisoned)
        } else {
            (None, Arc::new(AtomicBool::new(false)))
        };

        Ok(SnapshotPin {
            pool: if pin_snapshot { Some(self) } else { None },
            connection: Some(connection),
            snapshot_active: pin_snapshot,
            pin_id,
            poisoned,
            acquired_at,
            max_pin_duration: self.config.max_pin_duration(),
        })
    }

    #[must_use]
    pub fn stats(&self) -> PoolStats {
        let state = self.lock_state();
        let now = Instant::now();
        // Compute the oldest active pin under the same lock the rest of the
        // stats are computed under, so the snapshot reflects a single point in
        // time. We deliberately re-use the per-pin lifecycle math instead of
        // calling `oldest_active_pin_for_checkpoint_blocker()` to avoid a
        // re-lock and to keep `stats()` consistent.
        let checkpoint_blocked_by = state
            .active_pins
            .iter()
            .map(|(pin_id, record)| active_snapshot_pin_from_record(*pin_id, record, now))
            .max_by_key(|pin| pin.pin_age_ms)
            .map(CheckpointBlocker::from);
        PoolStats {
            active: state.active,
            idle: state.idle.len(),
            active_pins: state.active_pins.len(),
            expired_pins: expired_pin_count(&state.active_pins),
            max_size: self.config.max_size(),
            max_seen: state.max_seen,
            drops: state.drops,
            release_failures: state.release_failures,
            ad_hoc_bypass_count: state.ad_hoc_bypass_count,
            acquire_wait: acquire_wait_stats(&state.acquire_wait_ns),
            size_was_zero: self.config.size_was_zero(),
            checkpoint_blocked_by,
        }
    }

    #[must_use]
    pub fn active_snapshot_pins(&self) -> Vec<ActiveSnapshotPin> {
        let now = Instant::now();
        let state = self.lock_state();
        state
            .active_pins
            .iter()
            .map(|(pin_id, record)| active_snapshot_pin_from_record(*pin_id, record, now))
            .collect()
    }

    /// Return the oldest active SnapshotPin as a checkpoint-blocker attribution.
    ///
    /// When a WAL checkpoint reports BUSY because a reader still holds a snapshot
    /// reference to an old WAL page, callers query this helper to identify the
    /// reader most likely responsible. The oldest pin is the one whose snapshot
    /// reference predates the most WAL activity, so it is the strongest blocker
    /// candidate. Returns `None` when no pins are currently active.
    #[must_use]
    pub fn oldest_active_pin_for_checkpoint_blocker(&self) -> Option<CheckpointBlocker> {
        let active = self.active_snapshot_pins();
        active
            .into_iter()
            .max_by_key(|pin| pin.pin_age_ms)
            .map(CheckpointBlocker::from)
    }

    /// Record the outcome of a real WAL checkpoint attempt.
    ///
    /// When SQLite reports a checkpoint as BUSY, the oldest active SnapshotPin
    /// is the strongest process-local attribution candidate. This helper keeps
    /// the "actual checkpoint was blocked" signal distinct from the
    /// forward-looking [`PoolStats::checkpoint_blocked_by`] status field.
    #[must_use]
    pub fn note_checkpoint_outcome(&self, busy: bool) -> Option<CheckpointBlocker> {
        if !busy {
            return None;
        }

        let blocker = self.oldest_active_pin_for_checkpoint_blocker();
        trace_checkpoint_blocked_by_pin(blocker.as_ref());
        blocker
    }

    fn release(&self, slot_id: u64, connection: DbConnection) {
        let mut to_close = None;
        {
            let mut state = self.lock_state();
            decrement_pool_active(&mut state, "release");
            if state.idle.len() < self.config.max_size() {
                state.idle.push(IdleConnection {
                    slot_id,
                    connection,
                    returned_at: Instant::now(),
                });
            } else {
                state.drops = checked_increment_pool_counter(state.drops, "drops");
                to_close = Some(connection);
            }
        }
        drop_idle_connections(to_close.into_iter().collect());
    }

    fn abandon(&self, connection: DbConnection) {
        {
            let mut state = self.lock_state();
            decrement_pool_active(&mut state, "abandon");
            state.drops = checked_increment_pool_counter(state.drops, "drops");
        }
        let _ = connection.close();
    }

    fn note_release_failure(&self) {
        let mut state = self.lock_state();
        state.release_failures =
            checked_increment_pool_counter(state.release_failures, "release_failures");
    }

    pub fn expire_stale_pins(&self) -> Vec<ExpiredSnapshotPin> {
        let now = Instant::now();
        let state = self.lock_state();
        let mut expired = Vec::new();

        for (pin_id, record) in &state.active_pins {
            let age = now
                .checked_duration_since(record.acquired_at)
                .unwrap_or(Duration::ZERO);
            if age >= record.max_pin_duration && !record.poisoned.swap(true, Ordering::AcqRel) {
                expired.push(ExpiredSnapshotPin {
                    pin_id: *pin_id,
                    slot_id: record.slot_id,
                    age,
                });
            }
        }

        expired
    }

    pub fn force_poison_active_pins(&self) -> Vec<ExpiredSnapshotPin> {
        let now = Instant::now();
        let state = self.lock_state();
        let mut poisoned = Vec::new();

        for (pin_id, record) in &state.active_pins {
            if !record.poisoned.swap(true, Ordering::AcqRel) {
                poisoned.push(ExpiredSnapshotPin {
                    pin_id: *pin_id,
                    slot_id: record.slot_id,
                    age: now
                        .checked_duration_since(record.acquired_at)
                        .unwrap_or(Duration::ZERO),
                });
            }
        }

        poisoned
    }

    pub fn drain_snapshot_pins(&self, timeout: Duration) -> SnapshotDrainReport {
        let started = Instant::now();

        loop {
            let active_pins = self.lock_state().active_pins.len();
            if active_pins == 0 {
                return SnapshotDrainReport {
                    drained: true,
                    waited: started.elapsed(),
                    active_pins_remaining: 0,
                    force_poisoned: Vec::new(),
                };
            }

            let elapsed = started.elapsed();
            if elapsed >= timeout {
                let force_poisoned = self.force_poison_active_pins();
                return SnapshotDrainReport {
                    drained: false,
                    waited: elapsed,
                    active_pins_remaining: self.lock_state().active_pins.len(),
                    force_poisoned,
                };
            }

            std::thread::yield_now();
            std::thread::sleep(timeout.saturating_sub(elapsed).min(ACQUIRE_SLEEP_STEP));
        }
    }

    fn register_pin(
        &self,
        slot_id: Option<u64>,
        metadata: SnapshotPinMetadata,
        acquired_at: Instant,
        max_pin_duration: Duration,
    ) -> (u64, Arc<AtomicBool>) {
        let mut state = self.lock_state();
        let pin_id = allocate_next_read_pool_id(&mut state.next_pin_id, "pin");
        let poisoned = Arc::new(AtomicBool::new(false));
        state.active_pins.insert(
            pin_id,
            ActivePinRecord {
                slot_id,
                metadata,
                acquired_at,
                max_pin_duration,
                poisoned: Arc::clone(&poisoned),
            },
        );
        (pin_id, poisoned)
    }

    fn unregister_pin(&self, pin_id: u64) {
        let mut state = self.lock_state();
        state.active_pins.remove(&pin_id);
    }

    fn lock_state(&self) -> MutexGuard<'_, PoolState> {
        self.state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }
}

/// Build or reuse a process-local read pool and register it weakly by database.
///
/// The registry stores only `Weak` handles, so it does not extend pool lifetime
/// or retain stale SnapshotPin state after the command-local owner drops the
/// pool. This is intentionally process-local and is not a cross-process WAL
/// blocker registry.
#[must_use]
pub fn registered_process_read_pool(
    database: DatabaseConfig,
    config: PoolConfig,
) -> Arc<ReadConnectionPool> {
    let key = process_read_pool_registry_key(&database);
    let registry = process_read_pool_registry();

    // bd-fv0yn: fast path — shared `.read()` lock for the cache-hit
    // upgrade. Concurrent callers against DIFFERENT database keys
    // parallelize at this layer.
    {
        let read_guard = registry
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if let Some(pool) = read_guard.get(&key).and_then(Weak::upgrade) {
            return pool;
        }
    }

    // Slow path: key missing or stale Weak. Take the write lock to
    // insert. Re-check after acquiring the write lock so a concurrent
    // creator's pool is honored (no double-construction for the same
    // key). The insert naturally overwrites any stale Weak for the
    // same key; opportunistically reap other stale Weaks here so the
    // map stays bounded by live pools, not by distinct-keys-ever-
    // observed.
    let mut write_guard = registry
        .write()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    if let Some(pool) = write_guard.get(&key).and_then(Weak::upgrade) {
        return pool;
    }
    write_guard.retain(|_, weak| weak.upgrade().is_some());
    let pool = Arc::new(ReadConnectionPool::new(database, config));
    write_guard.insert(key, Arc::downgrade(&pool));
    pool
}

/// Return same-process read-pool stats for a database if a live pool is
/// registered. `None` means no process-local pool is visible, not that no other
/// process has a WAL reader.
#[must_use]
pub fn process_read_pool_stats_for_database(database: &DatabaseConfig) -> Option<PoolStats> {
    process_read_pool_for_database(database).map(|pool| pool.stats())
}

/// Attribute a real WAL checkpoint outcome to the visible process-local pool.
///
/// If `checkpoint_busy` is true but `process_pool_visible` is false, callers
/// must report an honest null blocker with a cross-process visibility caveat.
#[must_use]
pub fn note_process_checkpoint_outcome(
    database: &DatabaseConfig,
    checkpoint_busy: bool,
) -> ProcessCheckpointOutcome {
    let Some(pool) = process_read_pool_for_database(database) else {
        return ProcessCheckpointOutcome {
            checkpoint_busy,
            process_pool_visible: false,
            blocker: None,
        };
    };

    ProcessCheckpointOutcome {
        checkpoint_busy,
        process_pool_visible: true,
        blocker: pool.note_checkpoint_outcome(checkpoint_busy),
    }
}

fn process_read_pool_for_database(database: &DatabaseConfig) -> Option<Arc<ReadConnectionPool>> {
    let key = process_read_pool_registry_key(database);
    // bd-fv0yn: pure read path — shared `.read()` lock; stale-Weak
    // removal is deferred to the next `registered_process_read_pool`
    // write-path retain rather than mutating the registry on every
    // observation.
    let registry = process_read_pool_registry()
        .read()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    let pool = registry.get(&key).and_then(Weak::upgrade);
    pool
}

fn process_read_pool_registry() -> &'static RwLock<BTreeMap<String, Weak<ReadConnectionPool>>> {
    PROCESS_READ_POOL_REGISTRY.get_or_init(|| RwLock::new(BTreeMap::new()))
}

fn process_read_pool_registry_key(database: &DatabaseConfig) -> String {
    match database.location() {
        super::DatabaseLocation::Memory => "memory".to_owned(),
        super::DatabaseLocation::File(path) => {
            let path = std::fs::canonicalize(path)
                .unwrap_or_else(|_| super::normalized_write_owner_file_key(path));
            format!("file:{}", path.display())
        }
    }
}

fn read_pool_database_config(database: DatabaseConfig) -> DatabaseConfig {
    match database.location().clone() {
        DatabaseLocation::Memory => database,
        DatabaseLocation::File(path) => DatabaseConfig::read_only_file(path),
    }
}

impl ReadConnectionPoolBuilder {
    #[must_use]
    pub fn new(database: DatabaseConfig) -> Self {
        Self {
            database,
            config: PoolConfig::default_single(),
        }
    }

    #[must_use]
    pub fn with_size(mut self, max_size: usize) -> Self {
        self.config.max_size = max_size;
        self
    }

    #[must_use]
    pub fn with_idle_timeout(mut self, idle_timeout: Duration) -> Self {
        self.config.idle_timeout = idle_timeout;
        self
    }

    #[must_use]
    pub fn with_max_pin_duration(mut self, max_pin_duration: Duration) -> Self {
        self.config.max_pin_duration = max_pin_duration;
        self
    }

    #[must_use]
    pub fn with_acquire_timeout(mut self, acquire_timeout: Duration) -> Self {
        self.config.acquire_timeout = acquire_timeout;
        self
    }

    #[must_use]
    pub fn build(self) -> ReadConnectionPool {
        ReadConnectionPool::new(self.database, self.config)
    }
}

impl PooledReadConnection<'_> {
    #[must_use]
    pub const fn slot_id(&self) -> Option<u64> {
        self.slot_id
    }

    #[must_use]
    pub const fn is_ad_hoc(&self) -> bool {
        self.pool.is_none()
    }

    fn abandon(&mut self) {
        if let Some(connection) = self.connection.take() {
            if let Some(pool) = self.pool {
                pool.abandon(connection);
            } else {
                let _ = connection.close();
            }
        }
    }
}

impl SnapshotPin<'_> {
    #[must_use]
    pub const fn is_pinned(&self) -> bool {
        self.snapshot_active
    }

    #[must_use]
    pub fn slot_id(&self) -> Option<u64> {
        self.connection().slot_id()
    }

    #[must_use]
    pub fn age(&self) -> Duration {
        self.acquired_at.elapsed()
    }

    #[must_use]
    pub fn is_expired(&self) -> bool {
        self.age() >= self.max_pin_duration
    }

    #[must_use]
    pub fn is_poisoned(&self) -> bool {
        self.poisoned.load(Ordering::Acquire)
    }

    pub fn checked_connection(&self) -> Result<&DbConnection> {
        if self.is_expired() {
            self.poisoned.store(true, Ordering::Release);
        }

        if self.is_poisoned() {
            return Err(self.poisoned_error());
        }

        Ok(self.connection().deref())
    }

    pub fn commit(mut self) -> Result<()> {
        if self.snapshot_active {
            if self.is_expired() {
                self.poisoned.store(true, Ordering::Release);
            }
            if self.is_poisoned() {
                let error = self.poisoned_error();
                self.rollback()?;
                return Err(error);
            }
            self.connection().commit_read_snapshot()?;
            self.snapshot_active = false;
            self.unregister_pin();
        }
        Ok(())
    }

    pub fn rollback(mut self) -> Result<()> {
        if self.snapshot_active {
            self.snapshot_active = false;
            self.unregister_pin();
            let mut rollback_error = None;
            if let Some(connection) = self.connection.as_mut() {
                if let Err(error) = connection.rollback_read_snapshot() {
                    connection.abandon();
                    rollback_error = Some(error);
                }
            }
            if let Some(error) = rollback_error {
                self.note_release_failure();
                return Err(error);
            }
        }
        Ok(())
    }

    fn rollback_on_drop(&mut self) {
        if !self.snapshot_active {
            return;
        }

        self.snapshot_active = false;
        self.unregister_pin();
        let mut release_failed = false;
        if let Some(connection) = self.connection.as_mut() {
            if connection.rollback_read_snapshot().is_err() {
                connection.abandon();
                release_failed = true;
            }
        }
        if release_failed {
            self.note_release_failure();
        }
    }

    fn unregister_pin(&mut self) {
        if let (Some(pin_id), Some(pool)) = (self.pin_id.take(), self.pool) {
            pool.unregister_pin(pin_id);
        }
    }

    fn note_release_failure(&self) {
        if let Some(pool) = self.pool {
            pool.note_release_failure();
        }
    }

    fn poisoned_error(&self) -> DbError {
        DbError::MalformedRow {
            operation: DbOperation::Query,
            message: format!(
                "snapshot pin {} was poisoned or expired by the read-pool lifecycle watchdog; release it and acquire a fresh snapshot",
                self.pin_id
                    .map(|pin_id| pin_id.to_string())
                    .unwrap_or_else(|| "<unpinned>".to_string())
            ),
        }
    }

    fn connection(&self) -> &PooledReadConnection<'_> {
        match self.connection.as_ref() {
            Some(connection) => connection,
            None => panic!("snapshot pin owns a connection until Drop"),
        }
    }
}

impl Deref for PooledReadConnection<'_> {
    type Target = DbConnection;

    fn deref(&self) -> &Self::Target {
        match self.connection.as_ref() {
            Some(connection) => connection,
            None => panic!("pooled read connection is present until Drop"), // ubs:ignore
        }
    }
}

impl Deref for SnapshotPin<'_> {
    type Target = DbConnection;

    fn deref(&self) -> &Self::Target {
        match self.checked_connection() {
            Ok(connection) => connection,
            Err(error) => {
                panic!("snapshot pin deref refused unavailable read snapshot: {error}")
            }
        }
    }
}

impl Drop for PooledReadConnection<'_> {
    fn drop(&mut self) {
        if let Some(connection) = self.connection.take() {
            match (self.pool, self.slot_id) {
                (Some(pool), Some(slot_id)) => pool.release(slot_id, connection),
                _ => {
                    let _ = connection.close();
                }
            }
        }
    }
}

impl Drop for SnapshotPin<'_> {
    fn drop(&mut self) {
        self.rollback_on_drop();
    }
}

fn evict_expired_idle(state: &mut PoolState, idle_timeout: Duration) -> Vec<DbConnection> {
    if state.idle.is_empty() {
        return Vec::new();
    }

    let now = Instant::now();
    let mut retained = Vec::with_capacity(state.idle.len());
    let mut expired = Vec::new();

    for idle in state.idle.drain(..) {
        let age = now
            .checked_duration_since(idle.returned_at)
            .unwrap_or(Duration::ZERO);
        if age >= idle_timeout {
            state.drops = checked_increment_pool_counter(state.drops, "drops");
            expired.push(idle.connection);
        } else {
            retained.push(idle);
        }
    }

    state.idle = retained;
    expired
}

fn checked_pool_count_add(left: usize, right: usize, label: &'static str) -> Result<usize> {
    left.checked_add(right)
        .ok_or_else(|| DbError::MalformedRow {
            operation: DbOperation::Query,
            message: format!("read-pool {label} count overflow"),
        })
}

fn checked_increment_pool_counter(count: u64, label: &'static str) -> u64 {
    count.checked_add(1).unwrap_or_else(|| {
        panic!("read-pool {label} counter exhausted at u64::MAX");
    })
}

fn pool_occupancy(state: &PoolState) -> Result<usize> {
    checked_pool_count_add(state.active, state.idle.len(), "occupancy")
}

fn increment_pool_active(state: &mut PoolState) -> Result<()> {
    state.active = checked_pool_count_add(state.active, 1, "active")?;
    Ok(())
}

fn decrement_pool_active(state: &mut PoolState, context: &'static str) {
    state.active = state.active.checked_sub(1).unwrap_or_else(|| {
        panic!("read-pool active count underflow during {context}");
    });
}

fn record_acquire_wait(state: &mut PoolState, duration: Duration) {
    if state.acquire_wait_ns.len() == ACQUIRE_WAIT_SAMPLE_CAP {
        state.acquire_wait_ns.remove(0);
    }
    state.acquire_wait_ns.push(duration.as_nanos());
}

fn acquire_wait_stats(samples: &[u128]) -> AcquireWaitStats {
    if samples.is_empty() {
        return AcquireWaitStats::default();
    }

    let mut sorted = samples.to_vec();
    sorted.sort_unstable();
    let p50_index = sorted.len() / 2;
    let p99_index = sorted.len().saturating_mul(99).saturating_sub(1) / 100;
    AcquireWaitStats {
        samples: sorted.len(),
        p50_ns: sorted[p50_index],
        p99_ns: sorted[p99_index.min(sorted.len() - 1)],
    }
}

fn expired_pin_count(active_pins: &BTreeMap<u64, ActivePinRecord>) -> usize {
    let now = Instant::now();
    active_pins
        .values()
        .filter(|record| {
            record.poisoned.load(Ordering::Acquire)
                || now
                    .checked_duration_since(record.acquired_at)
                    .unwrap_or(Duration::ZERO)
                    >= record.max_pin_duration
        })
        .count()
}

fn active_snapshot_pin_from_record(
    pin_id: u64,
    record: &ActivePinRecord,
    now: Instant,
) -> ActiveSnapshotPin {
    let age = now
        .checked_duration_since(record.acquired_at)
        .unwrap_or(Duration::ZERO);
    let poisoned = record.poisoned.load(Ordering::Acquire);
    let release_state = if poisoned {
        SnapshotPinReleaseState::Poisoned
    } else if age >= record.max_pin_duration {
        SnapshotPinReleaseState::Expired
    } else {
        SnapshotPinReleaseState::Active
    };

    ActiveSnapshotPin {
        pin_id,
        slot_id: record.slot_id,
        workflow_id: record.metadata.workflow_id.clone(),
        request_id: record.metadata.request_id.clone(),
        pin_age_ms: age.as_millis(),
        max_pin_duration_ms: record.max_pin_duration.as_millis(),
        poisoned,
        release_state,
    }
}

fn drop_idle_connections(connections: Vec<DbConnection>) {
    for connection in connections {
        let _ = connection.close();
    }
}

fn trace_checkpoint_blocked_by_pin(blocker: Option<&CheckpointBlocker>) {
    match blocker {
        Some(blocker) => tracing::warn!(
            event = "read_pool.checkpoint_blocked_by_pin",
            surface = "read_pool",
            phase = "checkpoint_blocked_by_pin",
            blocker_present = true,
            pin_id = blocker.pin_id,
            slot_id = ?blocker.slot_id,
            workflow_id = blocker.workflow_id.as_deref().unwrap_or(""),
            request_id = blocker.request_id.as_deref().unwrap_or(""),
            age_ms = blocker.pin_age_ms,
            max_pin_duration_ms = blocker.max_pin_duration_ms,
            poisoned = blocker.poisoned,
            release_state = ?blocker.release_state,
            "WAL checkpoint reported BUSY while a read-pool SnapshotPin is active"
        ),
        None => tracing::warn!(
            event = "read_pool.checkpoint_blocked_by_pin",
            surface = "read_pool",
            phase = "checkpoint_blocked_by_pin",
            blocker_present = false,
            "WAL checkpoint reported BUSY but no process-local SnapshotPin was active"
        ),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::BTreeSet;
    use std::fs;
    use std::panic::{AssertUnwindSafe, catch_unwind};
    use std::path::{Path, PathBuf};
    use std::sync::{Arc, Barrier};
    use std::thread;

    fn memory_pool(max_size: usize, idle_timeout: Duration) -> ReadConnectionPool {
        ReadConnectionPool::new(
            DatabaseConfig::memory(),
            PoolConfig::new(max_size, idle_timeout),
        )
    }

    fn must<T, E: std::fmt::Display>(result: std::result::Result<T, E>, context: &str) -> T {
        match result {
            Ok(value) => value,
            Err(error) => panic!("{context}: {error}"),
        }
    }

    fn must_some<T>(value: Option<T>, context: &str) -> T {
        match value {
            Some(value) => value,
            None => panic!("{context}"),
        }
    }

    fn empty_pool_state() -> PoolState {
        PoolState {
            active: 0,
            idle: Vec::new(),
            acquire_wait_ns: Vec::with_capacity(ACQUIRE_WAIT_SAMPLE_CAP),
            next_slot_id: 1,
            next_pin_id: 1,
            active_pins: BTreeMap::new(),
            max_seen: 0,
            drops: 0,
            release_failures: 0,
            ad_hoc_bypass_count: 0,
        }
    }

    fn file_pool(max_size: usize) -> (tempfile::TempDir, PathBuf, ReadConnectionPool) {
        let tempdir = must(tempfile::tempdir(), "tempdir creates");
        let database_path = tempdir.path().join("read-pool-snapshot.db");
        seed_snapshot_database(&database_path);
        let pool = ReadConnectionPool::new(
            DatabaseConfig::file(database_path.clone()),
            PoolConfig::new(max_size, Duration::from_secs(30)),
        );
        (tempdir, database_path, pool)
    }

    fn seed_snapshot_database(database_path: &Path) {
        let connection = must(
            DbConnection::open_file(database_path),
            "seed database opens",
        );
        must(
            connection
                .execute_raw("CREATE TABLE snapshot_items (id INTEGER PRIMARY KEY, value TEXT)"),
            "snapshot table creates",
        );
        must(
            connection.execute_raw("INSERT INTO snapshot_items (id, value) VALUES (1, 'before')"),
            "initial row inserts",
        );
        must(connection.close(), "seed connection closes");
    }

    fn insert_snapshot_item(database_path: &Path, id: i64, value: &str) {
        let connection = must(DbConnection::open_file(database_path), "writer opens");
        must(
            connection.execute_raw(&format!(
                "INSERT INTO snapshot_items (id, value) VALUES ({id}, '{value}')"
            )),
            "writer inserts row",
        );
        close_writer_tolerating_pinned_checkpoint(connection);
    }

    /// Close a test writer while SnapshotPins are intentionally held open.
    ///
    /// FrankenSQLite's ordinary close runs a final passive WAL checkpoint
    /// (`sqlmodel-frankensqlite` `close_sync` -> `CloseMode::Checkpoint`);
    /// with active pinned readers that checkpoint reports "database is busy"
    /// while the handle is consumed regardless and the committed frames stay
    /// durable in the WAL, recovered and published by the next open. That
    /// contention outcome is the expected checkpoint-blocked-by-pin signal —
    /// exactly what `note_checkpoint_outcome` attributes — so ONLY the
    /// transient-contention class is tolerated here. Any other close failure
    /// still fails the test, and every snapshot/isolation assertion in the
    /// callers observes the row's durability through real reads afterwards.
    fn close_writer_tolerating_pinned_checkpoint(connection: DbConnection) {
        if let Err(error) = connection.close() {
            assert!(
                crate::db::db_error_is_transient_sqlite_contention(&error),
                "writer close failed with a non-contention error: {error}"
            );
        }
    }

    fn snapshot_item_count(connection: &DbConnection) -> i64 {
        let rows = must(
            connection.query("SELECT COUNT(*) FROM snapshot_items", &[]),
            "count query succeeds",
        );
        must_some(
            rows.first()
                .and_then(|row| row.get(0).and_then(|value| value.as_i64())),
            "count row present",
        )
    }

    fn join_reader_latency(handle: thread::JoinHandle<u128>) -> u128 {
        match handle.join() {
            Ok(value) => value,
            Err(payload) => {
                if let Some(message) = payload.downcast_ref::<&str>() {
                    panic!("reader thread panicked: {message}");
                }
                if let Some(message) = payload.downcast_ref::<String>() {
                    panic!("reader thread panicked: {message}");
                }
                panic!("reader thread panicked with non-string payload");
            }
        }
    }

    fn p50_latency_ms(values: &[u128]) -> u128 {
        let mut sorted = values.to_vec();
        sorted.sort_unstable();
        sorted[sorted.len() / 2]
    }

    fn read_repo_file(relative: &str) -> String {
        let path = Path::new(env!("CARGO_MANIFEST_DIR")).join(relative);
        must(
            fs::read_to_string(&path),
            &format!("{} reads", path.display()),
        )
    }

    fn pool_size_eight_batch_completion_latencies(
        database_path: PathBuf,
        readers: usize,
        per_reader_work: Duration,
    ) -> Vec<u128> {
        let pool = Arc::new(ReadConnectionPool::new(
            DatabaseConfig::file(database_path.clone()),
            PoolConfig::new(8, Duration::from_secs(30)),
        ));
        let readers_ready = Arc::new(Barrier::new(readers + 1));
        let release_readers = Arc::new(Barrier::new(readers + 1));
        let batch_start = Arc::new(Mutex::new(None::<Instant>));

        let handles: Vec<_> = (0..readers)
            .map(|_| {
                let pool = Arc::clone(&pool);
                let readers_ready = Arc::clone(&readers_ready);
                let release_readers = Arc::clone(&release_readers);
                let batch_start = Arc::clone(&batch_start);
                thread::spawn(move || {
                    let pin = must(pool.pin_snapshot(), "fanout reader snapshot opens");
                    assert_eq!(snapshot_item_count(&pin), 1);
                    readers_ready.wait();
                    release_readers.wait();
                    thread::sleep(per_reader_work);
                    assert_eq!(snapshot_item_count(&pin), 1);
                    batch_start
                        .lock()
                        .unwrap_or_else(std::sync::PoisonError::into_inner)
                        .expect("batch start set before readers release")
                        .elapsed()
                        .as_millis()
                })
            })
            .collect();

        readers_ready.wait();
        insert_snapshot_item(&database_path, 2, "during_pool_eight_readers");
        {
            let mut start = batch_start
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            *start = Some(Instant::now());
        }
        release_readers.wait();

        let latencies: Vec<u128> = handles.into_iter().map(join_reader_latency).collect();
        let fresh = must(pool.acquire(), "fresh reader opens after fanout batch");
        assert_eq!(snapshot_item_count(&fresh), 2);
        latencies
    }

    fn pool_size_one_batch_completion_latencies(
        database_path: PathBuf,
        readers: usize,
        per_reader_work: Duration,
    ) -> Vec<u128> {
        let pool = ReadConnectionPool::new(
            DatabaseConfig::file(database_path.clone()),
            PoolConfig::new(1, Duration::from_secs(30)),
        );
        let batch_start = Instant::now();
        let mut latencies = Vec::with_capacity(readers);

        for index in 0..readers {
            let pin = must(pool.pin_snapshot(), "serial reader snapshot opens");
            assert!(snapshot_item_count(&pin) >= 1);
            thread::sleep(per_reader_work);
            latencies.push(batch_start.elapsed().as_millis());
            drop(pin);
            if index == 0 {
                insert_snapshot_item(&database_path, 2, "during_pool_one_readers");
            }
        }

        latencies
    }

    #[test]
    fn read_pool_id_exhaustion_panics_instead_of_reusing_max() {
        let mut next_id = u64::MAX - 1;

        assert_eq!(
            allocate_next_read_pool_id(&mut next_id, "test"),
            u64::MAX - 1
        );
        assert_eq!(next_id, u64::MAX);

        let overflow = catch_unwind(AssertUnwindSafe(|| {
            let _ = allocate_next_read_pool_id(&mut next_id, "test");
        }));

        assert!(overflow.is_err());
        assert_eq!(next_id, u64::MAX);

        let slot_pool = memory_pool(1, Duration::from_secs(30));
        {
            let mut state = slot_pool.lock_state();
            state.next_slot_id = u64::MAX;
        }
        let slot_overflow = catch_unwind(AssertUnwindSafe(|| {
            let _ = slot_pool.acquire();
        }));
        assert!(slot_overflow.is_err());
        {
            let state = slot_pool.lock_state();
            assert_eq!(state.next_slot_id, u64::MAX);
            assert_eq!(state.active, 0);
        }

        let pin_pool = memory_pool(1, Duration::from_secs(30));
        {
            let mut state = pin_pool.lock_state();
            state.next_pin_id = u64::MAX;
        }
        let pin_overflow = catch_unwind(AssertUnwindSafe(|| {
            let _ = pin_pool.register_pin(
                None,
                SnapshotPinMetadata::default(),
                Instant::now(),
                Duration::from_secs(30),
            );
        }));
        assert!(pin_overflow.is_err());
        {
            let state = pin_pool.lock_state();
            assert_eq!(state.next_pin_id, u64::MAX);
            assert!(state.active_pins.is_empty());
        }
    }

    #[test]
    fn happy_path_pool_acquire_returns_distinct_connections_up_to_cap() {
        let pool = ReadConnectionPool::new(
            DatabaseConfig::memory(),
            PoolConfig::new(2, Duration::from_secs(30)).with_acquire_timeout(Duration::ZERO),
        );

        let first = must(pool.acquire(), "first connection opens");
        let second = must(pool.acquire(), "second connection opens");

        assert_ne!(first.slot_id(), second.slot_id());
        assert_eq!(
            pool.stats(),
            PoolStats {
                active: 2,
                idle: 0,
                active_pins: 0,
                expired_pins: 0,
                max_size: 2,
                max_seen: 2,
                drops: 0,
                release_failures: 0,
                ad_hoc_bypass_count: 0,
                acquire_wait: pool.stats().acquire_wait,
                size_was_zero: false,
                checkpoint_blocked_by: None,
            }
        );

        let third = must(pool.acquire(), "cap timeout opens ad-hoc connection");
        assert!(third.is_ad_hoc());
        assert_eq!(pool.stats().ad_hoc_bypass_count, 1);
    }

    #[test]
    fn happy_path_pool_release_returns_connection_to_lifo() {
        let pool = memory_pool(2, Duration::from_secs(30));

        let first = must(pool.acquire(), "first connection opens");
        let second = must(pool.acquire(), "second connection opens");
        let first_slot = first.slot_id();
        let second_slot = second.slot_id();
        drop(first);
        drop(second);

        assert_eq!(pool.stats().idle, 2);

        let reacquired = must(pool.acquire(), "idle connection reacquired");
        assert_eq!(reacquired.slot_id(), second_slot);
        assert_ne!(reacquired.slot_id(), first_slot);
    }

    #[test]
    fn file_pool_connections_open_read_only_and_reject_writes() {
        let (_tempdir, _database_path, pool) = file_pool(1);

        let connection = must(pool.acquire(), "file pool reader opens");
        assert_eq!(connection.mode(), crate::db::DatabaseOpenMode::ReadOnly);
        assert_eq!(snapshot_item_count(&connection), 1);

        let write_result = connection
            .execute_raw("INSERT INTO snapshot_items (id, value) VALUES (2, 'must_not_write')");
        assert!(matches!(
            write_result,
            Err(crate::db::DbError::InvalidMode {
                mode: crate::db::DatabaseOpenMode::ReadOnly,
                ..
            })
        ));
    }

    #[test]
    fn builder_overrides_config_for_test_construction() {
        let pool = ReadConnectionPool::builder(DatabaseConfig::memory())
            .with_size(4)
            .with_idle_timeout(Duration::from_millis(7))
            .with_max_pin_duration(Duration::from_millis(11))
            .with_acquire_timeout(Duration::from_millis(13))
            .build();

        assert_eq!(pool.config().requested_max_size(), 4);
        assert_eq!(pool.config().max_size(), 4);
        assert_eq!(pool.config().idle_timeout(), Duration::from_millis(7));
        assert_eq!(pool.config().max_pin_duration(), Duration::from_millis(11));
        assert_eq!(pool.config().acquire_timeout(), Duration::from_millis(13));
    }

    #[test]
    fn warm_creates_n_connections_synchronously() {
        let pool = ReadConnectionPool::builder(DatabaseConfig::memory())
            .with_size(3)
            .build();

        let warmed = must(pool.warm(2), "pool warm succeeds");

        assert_eq!(warmed, 2);
        let stats = pool.stats();
        assert_eq!(stats.active, 0);
        assert_eq!(stats.idle, 2);
        assert_eq!(stats.max_seen, 2);

        let first = must(pool.acquire(), "first warmed connection reacquires");
        let second = must(pool.acquire(), "second warmed connection reacquires");
        assert!(!first.is_ad_hoc());
        assert!(!second.is_ad_hoc());
        assert_ne!(first.slot_id(), second.slot_id());
    }

    #[test]
    fn warm_caps_at_pool_size_without_ad_hoc_bypass() {
        let pool = ReadConnectionPool::builder(DatabaseConfig::memory())
            .with_size(2)
            .with_acquire_timeout(Duration::ZERO)
            .build();

        let warmed = must(pool.warm(8), "pool warm succeeds");

        assert_eq!(warmed, 2);
        let stats = pool.stats();
        assert_eq!(stats.active, 0);
        assert_eq!(stats.idle, 2);
        assert_eq!(stats.max_seen, 2);
        assert_eq!(stats.ad_hoc_bypass_count, 0);
    }

    #[test]
    fn happy_path_snapshot_pin_holds_state_across_multiple_reads() {
        let (_tempdir, database_path, pool) = file_pool(2);
        let pin = must(pool.pin_snapshot(), "snapshot pin opens");

        assert!(pin.is_pinned());
        assert_eq!(snapshot_item_count(&pin), 1);

        insert_snapshot_item(&database_path, 2, "after");

        assert_eq!(snapshot_item_count(&pin), 1);
        drop(pin);

        let fresh = must(pool.acquire(), "fresh connection opens");
        assert_eq!(snapshot_item_count(&fresh), 2);
    }

    #[test]
    fn happy_path_two_concurrent_snapshot_pins_do_not_deadlock() {
        let (_tempdir, database_path, pool) = file_pool(2);
        let pool = Arc::new(pool);
        let start = Arc::new(Barrier::new(3));
        let (acquired_tx, acquired_rx) = std::sync::mpsc::channel();
        let mut release_txs = Vec::new();
        let mut handles = Vec::new();

        for reader in 0..2 {
            let pool = Arc::clone(&pool);
            let start = Arc::clone(&start);
            let acquired_tx = acquired_tx.clone();
            let (release_tx, release_rx) = std::sync::mpsc::channel();
            release_txs.push(release_tx);
            handles.push(thread::spawn(move || {
                start.wait();
                let pin = must(pool.pin_snapshot(), "reader snapshot pin opens");
                let slot_id = pin.slot_id();
                let before_insert = snapshot_item_count(&pin);
                must(
                    acquired_tx.send((reader, slot_id, before_insert)),
                    "reader reports acquired snapshot pin",
                );
                must(release_rx.recv(), "main releases reader snapshot pin");
                let after_insert = snapshot_item_count(&pin);
                (reader, slot_id, before_insert, after_insert)
            }));
        }
        drop(acquired_tx);

        start.wait();
        let mut acquired = Vec::new();
        for _ in 0..2 {
            acquired.push(must(
                acquired_rx.recv_timeout(Duration::from_secs(5)),
                "reader acquires snapshot pin before timeout",
            ));
        }

        assert_eq!(
            acquired
                .iter()
                .map(|(_reader, slot_id, _count)| *slot_id)
                .collect::<BTreeSet<_>>()
                .len(),
            2
        );
        assert!(
            acquired
                .iter()
                .all(|(_reader, _slot_id, before_insert)| *before_insert == 1)
        );

        insert_snapshot_item(&database_path, 2, "after");
        for release_tx in release_txs {
            must(release_tx.send(()), "main release signal sends");
        }

        let mut results = Vec::new();
        for handle in handles {
            match handle.join() {
                Ok(result) => results.push(result),
                Err(_) => panic!("reader thread panicked"),
            }
        }
        results.sort_by_key(|(reader, _slot_id, _before_insert, _after_insert)| *reader);
        assert_eq!(results.len(), 2);
        assert_ne!(results[0].1, results[1].1);
        for (_reader, _slot_id, before_insert, after_insert) in results {
            assert_eq!(before_insert, 1);
            assert_eq!(after_insert, 1);
        }
    }

    #[test]
    fn happy_path_disabled_snapshot_pin_preserves_unpinned_read_behavior() {
        let (_tempdir, database_path, pool) = file_pool(1);
        let unpinned = must(
            pool.acquire_snapshot(false),
            "unpinned snapshot handle opens",
        );

        assert!(!unpinned.is_pinned());
        assert_eq!(snapshot_item_count(&unpinned), 1);

        insert_snapshot_item(&database_path, 2, "after");
        assert_eq!(snapshot_item_count(&unpinned), 2);
    }

    #[test]
    fn happy_path_snapshot_pin_commit_releases_pool_connection() {
        let (_tempdir, _database_path, pool) = file_pool(1);
        let pin = must(pool.pin_snapshot(), "snapshot pin opens");

        assert_eq!(pool.stats().active, 1);
        assert_eq!(pool.stats().active_pins, 1);
        must(pin.commit(), "read snapshot commits");

        let stats = pool.stats();
        assert_eq!(stats.active, 0);
        assert_eq!(stats.idle, 1);
        assert_eq!(stats.active_pins, 0);
    }

    #[test]
    fn drop_snapshot_pin_abandons_connection_when_rollback_fails() {
        let (_tempdir, _database_path, pool) = file_pool(1);
        let pin = must(pool.pin_snapshot(), "snapshot pin opens");
        must(
            pin.connection().commit_read_snapshot(),
            "test commits behind pin",
        );

        drop(pin);

        let stats = pool.stats();
        assert_eq!(stats.active, 0);
        assert_eq!(stats.idle, 0);
        assert_eq!(stats.drops, 1);
        assert_eq!(stats.release_failures, 1);
        let fresh = must(pool.acquire(), "fresh connection opens after abandon");
        assert_ne!(fresh.slot_id(), Some(1));
    }

    #[test]
    fn drop_on_panic_path_releases_pin_idempotently() {
        let (_tempdir, _database_path, pool) = file_pool(1);
        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let _pin = must(pool.pin_snapshot(), "snapshot pin opens before panic");
            assert_eq!(pool.stats().active, 1);
            assert_eq!(pool.stats().active_pins, 1);
            panic!("force unwind across SnapshotPin");
        }));

        assert!(result.is_err());
        let stats = pool.stats();
        assert_eq!(stats.active, 0);
        assert_eq!(stats.idle, 1);
        assert_eq!(stats.active_pins, 0);
        assert_eq!(stats.drops, 0);
    }

    #[test]
    fn drop_double_drop_does_not_double_release() {
        let (_tempdir, _database_path, pool) = file_pool(1);
        let pin = must(pool.pin_snapshot(), "snapshot pin opens");
        let slot_id = pin.slot_id();

        must(pin.rollback(), "read snapshot rolls back explicitly");

        let stats = pool.stats();
        assert_eq!(stats.active, 0);
        assert_eq!(stats.idle, 1);
        assert_eq!(stats.active_pins, 0);
        assert_eq!(stats.drops, 0);

        let reacquired = must(pool.acquire(), "rolled back pin returns connection once");
        assert_eq!(reacquired.slot_id(), slot_id);
        assert_eq!(pool.stats().active, 1);
        assert_eq!(pool.stats().idle, 0);
    }

    #[test]
    fn drop_field_order_releases_pin_before_returning_connection_under_panic() {
        let (_tempdir, _database_path, pool) = file_pool(1);
        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let _pin = must(pool.pin_snapshot(), "snapshot pin opens before panic");
            panic!("force unwind across SnapshotPin");
        }));

        assert!(result.is_err());

        let stats = pool.stats();
        assert_eq!(stats.active, 0);
        assert_eq!(stats.idle, 1);
        assert_eq!(stats.active_pins, 0);

        let _reacquired = must(
            pool.acquire(),
            "connection is reusable only after pin metadata is cleared",
        );
        assert_eq!(pool.stats().active_pins, 0);
    }

    #[test]
    fn rollback_error_abandons_connection_and_prevents_lifo_reuse() {
        let (_tempdir, _database_path, pool) = file_pool(1);
        let pin = must(pool.pin_snapshot(), "snapshot pin opens");
        let slot_id = pin.slot_id();
        must(
            pin.connection().commit_read_snapshot(),
            "test commits behind pin before explicit rollback",
        );

        let error = match pin.rollback() {
            Ok(()) => panic!("rollback after manual commit should fail"),
            Err(error) => error,
        };
        assert!(error.operation().is_some());

        let stats = pool.stats();
        assert_eq!(stats.active, 0);
        assert_eq!(stats.idle, 0);
        assert_eq!(stats.active_pins, 0);
        assert_eq!(stats.drops, 1);
        assert_eq!(stats.release_failures, 1);

        let fresh = must(
            pool.acquire(),
            "fresh connection opens after rollback abandon",
        );
        assert_ne!(fresh.slot_id(), slot_id);
    }

    #[test]
    fn ad_hoc_snapshot_pin_unregisters_active_pin_on_drop() {
        let pool = ReadConnectionPool::new(
            DatabaseConfig::memory(),
            PoolConfig::new(1, Duration::from_secs(30)).with_acquire_timeout(Duration::ZERO),
        );
        let pooled = must(pool.acquire(), "first pooled connection opens");
        let ad_hoc_pin = must(
            pool.pin_snapshot(),
            "saturated pool opens ad-hoc snapshot pin",
        );

        assert!(ad_hoc_pin.is_pinned());
        assert!(ad_hoc_pin.slot_id().is_none());
        assert_eq!(pool.stats().active, 1);
        assert_eq!(pool.stats().active_pins, 1);
        assert_eq!(pool.stats().ad_hoc_bypass_count, 1);

        drop(ad_hoc_pin);

        let stats = pool.stats();
        assert_eq!(stats.active, 1);
        assert_eq!(stats.active_pins, 0);
        assert_eq!(stats.expired_pins, 0);
        assert_eq!(stats.release_failures, 0);

        drop(pooled);
        assert_eq!(pool.stats().active, 0);
    }

    #[test]
    fn cancel_during_wait_no_connection_acquired_no_leak() {
        let pool = ReadConnectionPool::new(
            DatabaseConfig::memory(),
            PoolConfig::new(1, Duration::from_secs(30)).with_acquire_timeout(Duration::ZERO),
        );
        let first = must(pool.acquire(), "first pooled connection opens");

        let timed_out = must(pool.acquire(), "timed-out acquire opens ad-hoc connection");
        assert!(timed_out.is_ad_hoc());
        let stats = pool.stats();
        assert_eq!(stats.active, 1);
        assert_eq!(stats.idle, 0);
        assert_eq!(stats.active_pins, 0);
        assert_eq!(stats.ad_hoc_bypass_count, 1);

        drop(timed_out);
        assert_eq!(pool.stats().active, 1);

        drop(first);
        let stats = pool.stats();
        assert_eq!(stats.active, 0);
        assert_eq!(stats.idle, 1);
        assert_eq!(stats.active_pins, 0);
    }

    #[test]
    fn cancel_just_after_acquire_connection_returns_to_lifo_no_pin_state() {
        let pool = memory_pool(1, Duration::from_secs(30));
        let acquired = must(pool.acquire(), "pooled connection opens");
        let slot_id = acquired.slot_id();

        drop(acquired);

        let stats = pool.stats();
        assert_eq!(stats.active, 0);
        assert_eq!(stats.idle, 1);
        assert_eq!(stats.active_pins, 0);

        let reacquired = must(pool.acquire(), "dropped connection returns to LIFO");
        assert_eq!(reacquired.slot_id(), slot_id);
    }

    #[test]
    fn cancel_during_pin_rollback_runs_connection_returns_to_lifo() {
        let (_tempdir, database_path, pool) = file_pool(1);
        let pin = must(pool.pin_snapshot(), "snapshot pin opens");
        let slot_id = pin.slot_id();
        assert_eq!(snapshot_item_count(&pin), 1);

        insert_snapshot_item(&database_path, 2, "during_pin");
        drop(pin);

        let stats = pool.stats();
        assert_eq!(stats.active, 0);
        assert_eq!(stats.idle, 1);
        assert_eq!(stats.active_pins, 0);

        let reacquired = must(pool.acquire(), "rolled-back pin returns connection");
        assert_eq!(reacquired.slot_id(), slot_id);
        assert_eq!(snapshot_item_count(&reacquired), 2);
    }

    #[test]
    fn watchdog_pin_within_max_duration_is_not_disturbed() {
        let (_tempdir, _database_path, pool) = file_pool(1);
        let pin = must(pool.pin_snapshot(), "snapshot pin opens");

        assert!(pool.expire_stale_pins().is_empty());
        assert!(!pin.is_poisoned());
        assert_eq!(pool.stats().active_pins, 1);
        assert_eq!(pool.stats().expired_pins, 0);
    }

    #[test]
    fn watchdog_pin_held_beyond_max_duration_is_marked_poisoned() {
        let tempdir = must(tempfile::tempdir(), "tempdir creates");
        let database_path = tempdir.path().join("read-pool-expired-pin.db");
        seed_snapshot_database(&database_path);
        let pool = ReadConnectionPool::new(
            DatabaseConfig::file(database_path),
            PoolConfig::new(1, Duration::from_secs(30)).with_max_pin_duration(Duration::ZERO),
        );
        let pin = must(pool.pin_snapshot(), "snapshot pin opens");
        let slot_id = pin.slot_id();

        let expired = pool.expire_stale_pins();

        assert_eq!(expired.len(), 1);
        assert_eq!(expired[0].slot_id, slot_id);
        assert!(pin.is_poisoned());
        assert_eq!(pool.stats().active_pins, 1);
        assert_eq!(pool.stats().expired_pins, 1);

        drop(pin);
        let stats = pool.stats();
        assert_eq!(stats.active, 0);
        assert_eq!(stats.idle, 1);
        assert_eq!(stats.active_pins, 0);
    }

    #[test]
    fn watchdog_expired_pin_is_reported_once_until_released() {
        let tempdir = must(tempfile::tempdir(), "tempdir creates");
        let database_path = tempdir.path().join("read-pool-expired-pin-once.db");
        seed_snapshot_database(&database_path);
        let pool = ReadConnectionPool::new(
            DatabaseConfig::file(database_path),
            PoolConfig::new(1, Duration::from_secs(30)).with_max_pin_duration(Duration::ZERO),
        );
        let pin = must(pool.pin_snapshot(), "snapshot pin opens");

        let first_scan = pool.expire_stale_pins();
        let second_scan = pool.expire_stale_pins();

        assert_eq!(first_scan.len(), 1);
        assert!(second_scan.is_empty());
        assert!(pin.is_poisoned());
        assert_eq!(pool.stats().active_pins, 1);
        assert_eq!(pool.stats().expired_pins, 1);

        drop(pin);
        assert_eq!(pool.stats().active_pins, 0);
        assert_eq!(pool.stats().expired_pins, 0);
    }

    #[test]
    fn active_snapshot_pins_report_ordered_lifecycle_metadata() {
        let tempdir = must(tempfile::tempdir(), "tempdir creates");
        let database_path = tempdir.path().join("read-pool-active-pins.db");
        seed_snapshot_database(&database_path);
        let pool = ReadConnectionPool::new(
            DatabaseConfig::file(database_path),
            PoolConfig::new(2, Duration::from_secs(30)).with_max_pin_duration(Duration::ZERO),
        );
        let first = must(pool.pin_snapshot(), "first snapshot pin opens");
        let second = must(pool.pin_snapshot(), "second snapshot pin opens");

        let active = pool.active_snapshot_pins();
        assert_eq!(active.len(), 2);
        assert!(active[0].pin_id < active[1].pin_id);
        assert_eq!(active[0].slot_id, first.slot_id());
        assert_eq!(active[1].slot_id, second.slot_id());
        assert_eq!(active[0].max_pin_duration_ms, 0);
        assert_eq!(active[1].max_pin_duration_ms, 0);
        assert_eq!(active[0].release_state, SnapshotPinReleaseState::Expired);
        assert_eq!(active[1].release_state, SnapshotPinReleaseState::Expired);
        assert!(!active[0].poisoned);
        assert!(!active[1].poisoned);

        let expired = pool.expire_stale_pins();
        assert_eq!(expired.len(), 2);
        let poisoned = pool.active_snapshot_pins();
        assert_eq!(poisoned[0].release_state, SnapshotPinReleaseState::Poisoned);
        assert_eq!(poisoned[1].release_state, SnapshotPinReleaseState::Poisoned);
        assert!(poisoned[0].poisoned);
        assert!(poisoned[1].poisoned);

        drop(first);
        drop(second);
        assert!(pool.active_snapshot_pins().is_empty());
    }

    #[test]
    fn active_snapshot_pins_preserve_workflow_and_request_metadata() {
        let (_tempdir, _database_path, pool) = file_pool(1);
        let pin = must(
            pool.pin_snapshot_with_metadata(SnapshotPinMetadata {
                workflow_id: Some("workflow-read-pool".to_owned()),
                request_id: Some("request-42".to_owned()),
            }),
            "metadata snapshot pin opens",
        );

        let active = pool.active_snapshot_pins();
        assert_eq!(active.len(), 1);
        assert_eq!(active[0].workflow_id.as_deref(), Some("workflow-read-pool"));
        assert_eq!(active[0].request_id.as_deref(), Some("request-42"));
        assert_eq!(active[0].release_state, SnapshotPinReleaseState::Active);
        assert!(!active[0].poisoned);

        drop(pin);
        assert!(pool.active_snapshot_pins().is_empty());
    }

    #[test]
    fn pool_stats_reports_oldest_pin_as_checkpoint_blocker() {
        let (_tempdir, _database_path, pool) = file_pool(2);

        // No pins → stats reports None.
        assert!(pool.stats().checkpoint_blocked_by.is_none());

        let older = must(
            pool.pin_snapshot_with_metadata(SnapshotPinMetadata {
                workflow_id: Some("workflow-older".to_owned()),
                request_id: Some("request-older".to_owned()),
            }),
            "older pin opens",
        );
        thread::sleep(Duration::from_millis(5));
        let newer = must(
            pool.pin_snapshot_with_metadata(SnapshotPinMetadata {
                workflow_id: Some("workflow-newer".to_owned()),
                request_id: Some("request-newer".to_owned()),
            }),
            "newer pin opens",
        );

        let stats = pool.stats();
        let blocker = must_some(
            stats.checkpoint_blocked_by,
            "stats reports oldest pin as checkpoint blocker",
        );
        assert_eq!(blocker.workflow_id.as_deref(), Some("workflow-older"));
        assert_eq!(blocker.request_id.as_deref(), Some("request-older"));
        assert_eq!(blocker.slot_id, older.slot_id());
        assert_eq!(blocker.release_state, SnapshotPinReleaseState::Active);

        drop(older);
        drop(newer);
        assert!(pool.stats().checkpoint_blocked_by.is_none());
    }

    #[test]
    fn checkpoint_blocked_reports_oldest_pin_as_blocker() {
        let (_tempdir, _database_path, pool) = file_pool(3);

        // Empty pool: no candidate blocker.
        assert!(pool.oldest_active_pin_for_checkpoint_blocker().is_none());

        let first = must(
            pool.pin_snapshot_with_metadata(SnapshotPinMetadata {
                workflow_id: Some("workflow-oldest".to_owned()),
                request_id: Some("request-oldest".to_owned()),
            }),
            "first (oldest) pin opens",
        );
        // The first pin must be observably older than later acquisitions. The
        // pool surfaces `pin_age_ms` derived from a monotonic clock and the
        // oldest-pin tie-breaker is age, not pin_id, so sleep briefly to keep
        // the age separation deterministic on fast CI hardware.
        thread::sleep(Duration::from_millis(5));
        let second = must(
            pool.pin_snapshot_with_metadata(SnapshotPinMetadata {
                workflow_id: Some("workflow-mid".to_owned()),
                request_id: Some("request-mid".to_owned()),
            }),
            "second (middle) pin opens",
        );
        thread::sleep(Duration::from_millis(5));
        let third = must(
            pool.pin_snapshot_with_metadata(SnapshotPinMetadata {
                workflow_id: Some("workflow-newest".to_owned()),
                request_id: Some("request-newest".to_owned()),
            }),
            "third (newest) pin opens",
        );

        let blocker = must_some(
            pool.oldest_active_pin_for_checkpoint_blocker(),
            "oldest pin is reported as checkpoint blocker",
        );
        assert_eq!(blocker.workflow_id.as_deref(), Some("workflow-oldest"));
        assert_eq!(blocker.request_id.as_deref(), Some("request-oldest"));
        assert_eq!(blocker.slot_id, first.slot_id());
        assert_eq!(blocker.release_state, SnapshotPinReleaseState::Active);
        assert!(!blocker.poisoned);
        // Oldest pin's age is strictly greater than newer pins' ages.
        let active = pool.active_snapshot_pins();
        let mid_age = active
            .iter()
            .find(|pin| pin.slot_id == second.slot_id())
            .expect("middle pin present")
            .pin_age_ms;
        let newest_age = active
            .iter()
            .find(|pin| pin.slot_id == third.slot_id())
            .expect("newest pin present")
            .pin_age_ms;
        assert!(blocker.pin_age_ms >= mid_age);
        assert!(mid_age >= newest_age);

        // Drop the oldest pin: the next-oldest pin becomes the new blocker.
        drop(first);
        let blocker = must_some(
            pool.oldest_active_pin_for_checkpoint_blocker(),
            "next-oldest pin becomes the blocker after oldest is dropped",
        );
        assert_eq!(blocker.workflow_id.as_deref(), Some("workflow-mid"));
        assert_eq!(blocker.slot_id, second.slot_id());

        drop(second);
        drop(third);
        assert!(pool.oldest_active_pin_for_checkpoint_blocker().is_none());
    }

    #[test]
    fn checkpoint_outcome_notes_only_busy_attempts() {
        let (_tempdir, _database_path, pool) = file_pool(1);
        let _pin = must(
            pool.pin_snapshot_with_metadata(SnapshotPinMetadata {
                workflow_id: Some("workflow-checkpoint".to_owned()),
                request_id: Some("request-checkpoint".to_owned()),
            }),
            "snapshot pin opens",
        );

        assert!(
            pool.note_checkpoint_outcome(false).is_none(),
            "non-busy checkpoints should not report a blocker"
        );

        let blocker = must_some(
            pool.note_checkpoint_outcome(true),
            "busy checkpoint reports active pin blocker",
        );
        assert_eq!(blocker.workflow_id.as_deref(), Some("workflow-checkpoint"));
        assert_eq!(blocker.request_id.as_deref(), Some("request-checkpoint"));
        assert_eq!(blocker.release_state, SnapshotPinReleaseState::Active);
    }

    #[test]
    fn checkpoint_outcome_handles_busy_without_process_local_pin() {
        let (_tempdir, _database_path, pool) = file_pool(1);

        assert!(
            pool.note_checkpoint_outcome(true).is_none(),
            "busy checkpoint without a process-local pin reports no blocker"
        );
    }

    #[test]
    fn process_registry_reports_visible_pool_blocker_for_busy_checkpoint() {
        let tempdir = must(tempfile::tempdir(), "tempdir creates");
        let database_path = tempdir.path().join("read-pool-registry-visible.db");
        seed_snapshot_database(&database_path);
        let database = DatabaseConfig::file(database_path);
        let pool = registered_process_read_pool(
            database.clone(),
            PoolConfig::new(1, Duration::from_secs(30)),
        );
        let _pin = must(
            pool.pin_snapshot_with_metadata(SnapshotPinMetadata {
                workflow_id: Some("workflow-registry".to_owned()),
                request_id: Some("request-registry".to_owned()),
            }),
            "registered pool snapshot pin opens",
        );

        let outcome = note_process_checkpoint_outcome(&database, true);

        assert!(outcome.checkpoint_busy);
        assert!(outcome.process_pool_visible);
        let blocker = must_some(outcome.blocker, "registered pool exposes blocker");
        assert_eq!(blocker.workflow_id.as_deref(), Some("workflow-registry"));
        assert_eq!(blocker.request_id.as_deref(), Some("request-registry"));
    }

    #[test]
    fn process_registry_reports_no_pool_visibility_without_registered_pool() {
        let tempdir = must(tempfile::tempdir(), "tempdir creates");
        let database = DatabaseConfig::file(tempdir.path().join("not-registered.db"));

        let outcome = note_process_checkpoint_outcome(&database, true);

        assert!(outcome.checkpoint_busy);
        assert!(!outcome.process_pool_visible);
        assert!(outcome.blocker.is_none());
        assert!(process_read_pool_stats_for_database(&database).is_none());
    }

    #[test]
    fn process_registry_does_not_extend_pool_lifetime() {
        let tempdir = must(tempfile::tempdir(), "tempdir creates");
        let database_path = tempdir.path().join("read-pool-registry-lifetime.db");
        seed_snapshot_database(&database_path);
        let database = DatabaseConfig::file(database_path);
        {
            let pool = registered_process_read_pool(
                database.clone(),
                PoolConfig::new(1, Duration::from_secs(30)),
            );
            assert!(process_read_pool_stats_for_database(&database).is_some());
            drop(pool);
        }

        assert!(
            process_read_pool_stats_for_database(&database).is_none(),
            "registry must store Weak handles so command-local pools can drop"
        );
    }

    #[test]
    fn process_registry_key_normalizes_missing_file_paths_lexically() {
        let tempdir = must(tempfile::tempdir(), "tempdir creates");
        let direct = DatabaseConfig::file(tempdir.path().join("read-pool-missing.db"));
        let indirect = DatabaseConfig::file(
            tempdir
                .path()
                .join("not-created")
                .join("..")
                .join("read-pool-missing.db"),
        );

        assert_eq!(
            process_read_pool_registry_key(&direct),
            process_read_pool_registry_key(&indirect)
        );
    }

    #[test]
    fn workspace_close_force_poisons_pins_after_drain_timeout() {
        let (_tempdir, _database_path, pool) = file_pool(2);
        let first = must(pool.pin_snapshot(), "first snapshot pin opens");
        let second = must(pool.pin_snapshot(), "second snapshot pin opens");
        let first_slot = first.slot_id();
        let second_slot = second.slot_id();

        let poisoned = pool.force_poison_active_pins();

        assert_eq!(poisoned.len(), 2);
        assert_eq!(poisoned[0].slot_id, first_slot);
        assert_eq!(poisoned[1].slot_id, second_slot);
        assert!(poisoned[0].pin_id < poisoned[1].pin_id);
        assert!(first.is_poisoned());
        assert!(second.is_poisoned());
        assert_eq!(pool.stats().active_pins, 2);
        assert_eq!(pool.stats().expired_pins, 2);

        drop(first);
        drop(second);
        let stats = pool.stats();
        assert_eq!(stats.active, 0);
        assert_eq!(stats.idle, 2);
        assert_eq!(stats.active_pins, 0);
        assert_eq!(stats.expired_pins, 0);
    }

    #[test]
    fn workspace_close_drains_readers_before_writer_shutdown() {
        let tempdir = must(tempfile::tempdir(), "tempdir creates");
        let database_path = tempdir.path().join("read-pool-drain.db");
        seed_snapshot_database(&database_path);
        let pool = Arc::new(ReadConnectionPool::new(
            DatabaseConfig::file(database_path),
            PoolConfig::new(1, Duration::from_secs(30)),
        ));
        let pin_acquired = Arc::new(Barrier::new(2));

        let reader = {
            let pool = Arc::clone(&pool);
            let pin_acquired = Arc::clone(&pin_acquired);
            thread::spawn(move || {
                let pin = must(pool.pin_snapshot(), "reader snapshot pin opens");
                assert_eq!(snapshot_item_count(&pin), 1);
                pin_acquired.wait();
                thread::sleep(Duration::from_millis(10));
                drop(pin);
            })
        };

        pin_acquired.wait();
        let report = pool.drain_snapshot_pins(Duration::from_secs(1));
        must(reader.join().map_err(|_| "reader panicked"), "reader joins");

        assert!(report.drained);
        assert_eq!(report.active_pins_remaining, 0);
        assert!(report.force_poisoned.is_empty());
        assert_eq!(pool.stats().active_pins, 0);
        assert_eq!(pool.stats().active, 0);
        assert_eq!(pool.stats().idle, 1);
    }

    #[test]
    fn workspace_close_force_poisons_pins_when_drain_timeout_elapses() {
        let (_tempdir, _database_path, pool) = file_pool(1);
        let pin = must(pool.pin_snapshot(), "snapshot pin opens");
        let slot_id = pin.slot_id();

        let report = pool.drain_snapshot_pins(Duration::ZERO);

        assert!(!report.drained);
        assert_eq!(report.active_pins_remaining, 1);
        assert_eq!(report.force_poisoned.len(), 1);
        assert_eq!(report.force_poisoned[0].slot_id, slot_id);
        assert!(pin.is_poisoned());
        assert_eq!(pool.stats().expired_pins, 1);

        drop(pin);
        assert_eq!(pool.stats().active_pins, 0);
    }

    #[test]
    fn force_poison_reports_each_active_pin_once_until_released() {
        let (_tempdir, _database_path, pool) = file_pool(1);
        let pin = must(pool.pin_snapshot(), "snapshot pin opens");

        let first = pool.force_poison_active_pins();
        let second = pool.force_poison_active_pins();

        assert_eq!(first.len(), 1);
        assert_eq!(first[0].slot_id, pin.slot_id());
        assert!(second.is_empty());
        assert!(pin.is_poisoned());
        assert_eq!(pool.stats().expired_pins, 1);

        drop(pin);
        assert_eq!(pool.stats().active_pins, 0);
        assert_eq!(pool.stats().expired_pins, 0);
    }

    #[test]
    fn force_poison_reports_active_pins_in_stable_order() {
        let (_tempdir, _database_path, pool) = file_pool(3);
        let first_pin = must(pool.pin_snapshot(), "first snapshot pin opens");
        let second_pin = must(pool.pin_snapshot(), "second snapshot pin opens");
        let third_pin = must(pool.pin_snapshot(), "third snapshot pin opens");

        let expected: Vec<(u64, Option<u64>)> = pool
            .active_snapshot_pins()
            .iter()
            .map(|pin| (pin.pin_id, pin.slot_id))
            .collect();
        assert_eq!(
            expected,
            vec![
                (1, first_pin.slot_id()),
                (2, second_pin.slot_id()),
                (3, third_pin.slot_id()),
            ]
        );

        let poisoned = pool.force_poison_active_pins();
        let actual: Vec<(u64, Option<u64>)> = poisoned
            .iter()
            .map(|pin| (pin.pin_id, pin.slot_id))
            .collect();

        assert_eq!(actual, expected);
        assert!(pool.force_poison_active_pins().is_empty());
        assert!(first_pin.is_poisoned());
        assert!(second_pin.is_poisoned());
        assert!(third_pin.is_poisoned());
        assert_eq!(pool.stats().expired_pins, 3);

        drop(first_pin);
        drop(second_pin);
        drop(third_pin);
        assert_eq!(pool.stats().active_pins, 0);
        assert_eq!(pool.stats().expired_pins, 0);
    }

    #[test]
    fn poisoned_snapshot_pin_checked_connection_returns_clean_error() {
        let (_tempdir, _database_path, pool) = file_pool(1);
        let pin = must(pool.pin_snapshot(), "snapshot pin opens");

        assert_eq!(
            snapshot_item_count(must(pin.checked_connection(), "pin is usable")),
            1
        );
        let poisoned = pool.force_poison_active_pins();
        assert_eq!(poisoned.len(), 1);

        let error = match pin.checked_connection() {
            Ok(_) => panic!("poisoned pin should not return checked connection"),
            Err(error) => error,
        };
        assert_eq!(error.operation(), Some(DbOperation::Query));
        assert!(error.to_string().contains("snapshot pin"));
        assert!(error.to_string().contains("poisoned"));
        assert!(pin.is_poisoned());

        drop(pin);
        let stats = pool.stats();
        assert_eq!(stats.active, 0);
        assert_eq!(stats.idle, 1);
        assert_eq!(stats.active_pins, 0);
    }

    #[test]
    fn snapshot_pin_deref_refuses_poisoned_pin_to_prevent_silent_stale_reads() {
        let (_tempdir, _database_path, pool) = file_pool(1);
        let pin = must(pool.pin_snapshot(), "snapshot pin opens");

        assert_eq!(snapshot_item_count(&pin), 1);
        let poisoned = pool.force_poison_active_pins();
        assert_eq!(poisoned.len(), 1);

        let panic = catch_unwind(AssertUnwindSafe(|| {
            let _ = snapshot_item_count(&pin);
        }));
        assert!(panic.is_err(), "poisoned snapshot deref must fail closed");
        assert!(pin.is_poisoned());

        drop(pin);
        let stats = pool.stats();
        assert_eq!(stats.active, 0);
        assert_eq!(stats.idle, 1);
        assert_eq!(stats.active_pins, 0);
    }

    #[test]
    fn poisoned_snapshot_pin_commit_returns_clean_error_and_releases_connection() {
        let (_tempdir, _database_path, pool) = file_pool(1);
        let pin = must(pool.pin_snapshot(), "snapshot pin opens");
        let slot_id = pin.slot_id();

        let poisoned = pool.force_poison_active_pins();
        assert_eq!(poisoned.len(), 1);

        let error = match pin.commit() {
            Ok(()) => panic!("poisoned pin should not commit as a clean release"),
            Err(error) => error,
        };

        assert_eq!(error.operation(), Some(DbOperation::Query));
        assert!(error.to_string().contains("snapshot pin"));
        assert!(error.to_string().contains("poisoned"));

        let stats = pool.stats();
        assert_eq!(stats.active, 0);
        assert_eq!(stats.idle, 1);
        assert_eq!(stats.active_pins, 0);
        assert_eq!(stats.expired_pins, 0);

        let reacquired = must(pool.acquire(), "poisoned commit releases connection");
        assert_eq!(reacquired.slot_id(), slot_id);
    }

    #[test]
    fn checked_connection_expires_over_age_pin_without_pool_scan() {
        let tempdir = must(tempfile::tempdir(), "tempdir creates");
        let database_path = tempdir.path().join("read-pool-checked-expiry.db");
        seed_snapshot_database(&database_path);
        let pool = ReadConnectionPool::new(
            DatabaseConfig::file(database_path),
            PoolConfig::new(1, Duration::from_secs(30)).with_max_pin_duration(Duration::ZERO),
        );
        let pin = must(pool.pin_snapshot(), "snapshot pin opens");

        let error = match pin.checked_connection() {
            Ok(_) => panic!("expired pin should not return checked connection"),
            Err(error) => error,
        };
        assert_eq!(error.operation(), Some(DbOperation::Query));
        assert!(error.to_string().contains("expired"));
        assert!(pin.is_poisoned());
        assert_eq!(pool.stats().active_pins, 1);
        assert_eq!(pool.stats().expired_pins, 1);

        drop(pin);
        assert_eq!(pool.stats().active_pins, 0);
    }

    #[test]
    fn snapshot_pin_reports_age_and_expiry_against_configured_limit() {
        let pool = ReadConnectionPool::new(
            DatabaseConfig::memory(),
            PoolConfig::new(1, Duration::from_secs(30))
                .with_max_pin_duration(Duration::from_secs(60)),
        );
        let pin = must(pool.pin_snapshot(), "snapshot pin opens");

        assert!(!pin.is_expired());
        assert!(pin.age() < Duration::from_secs(60));
    }

    #[test]
    fn empty_or_boundary_pool_size_zero_falls_back_to_size_one_with_warning() {
        let config = PoolConfig::new(0, Duration::from_secs(30));
        assert_eq!(config.requested_max_size(), 0);
        assert_eq!(config.max_size(), 1);
        assert!(config.size_was_zero());

        let pool = ReadConnectionPool::new(
            DatabaseConfig::memory(),
            config.with_acquire_timeout(Duration::ZERO),
        );
        let _first = must(pool.acquire(), "normalized first acquire opens");
        let second = must(pool.acquire(), "normalized pool opens ad-hoc on timeout");
        assert!(second.is_ad_hoc());

        let stats = pool.stats();
        assert_eq!(stats.max_size, 1);
        assert!(stats.size_was_zero);
    }

    #[test]
    fn empty_or_boundary_expired_idle_connection_is_evicted_lazily() {
        let pool = memory_pool(1, Duration::ZERO);
        assert_eq!(pool.stats().idle, 0);

        let first = must(pool.acquire(), "first connection opens");
        let first_slot = first.slot_id();
        drop(first);

        let second = must(pool.acquire(), "expired idle connection replaced");
        assert_ne!(second.slot_id(), first_slot);

        let stats = pool.stats();
        assert_eq!(stats.active, 1);
        assert_eq!(stats.idle, 0);
        assert_eq!(stats.drops, 1);
    }

    #[test]
    fn error_or_invalid_pool_acquire_when_db_is_unopenable_returns_error_not_panic() {
        let current_exe = match std::env::current_exe() {
            Ok(path) => path,
            Err(error) => panic!("current test binary path: {error}"),
        };
        let database_path = current_exe.join("read-pool.db");
        let pool = ReadConnectionPool::new(
            DatabaseConfig::file(database_path),
            PoolConfig::new(1, Duration::from_secs(30)),
        );

        let error = match pool.acquire() {
            Ok(_) => panic!("missing parent should fail"),
            Err(error) => error,
        };
        assert!(error.operation().is_some());
        assert_eq!(pool.stats().active, 0);
    }

    #[test]
    fn read_pool_active_count_overflow_is_rejected() {
        let mut state = empty_pool_state();
        state.active = usize::MAX;

        let error =
            increment_pool_active(&mut state).expect_err("overflowing active count must fail");
        assert_eq!(state.active, usize::MAX);
        assert!(
            error
                .to_string()
                .contains("read-pool active count overflow")
        );

        let occupancy_error = checked_pool_count_add(usize::MAX, 1, "occupancy")
            .expect_err("overflowing occupancy count must fail");
        assert!(
            occupancy_error
                .to_string()
                .contains("read-pool occupancy count overflow")
        );
    }

    #[test]
    fn read_pool_metric_counter_overflow_panics() {
        let panic = catch_unwind(AssertUnwindSafe(|| {
            let _ = checked_increment_pool_counter(u64::MAX, "drops");
        }));

        assert!(
            panic.is_err(),
            "read-pool metric counter overflow must fail loudly"
        );
    }

    #[test]
    fn read_pool_active_count_underflow_panics() {
        let panic = catch_unwind(AssertUnwindSafe(|| {
            let mut state = empty_pool_state();
            decrement_pool_active(&mut state, "test");
        }));

        assert!(panic.is_err(), "active count underflow must fail loudly");
    }

    #[test]
    fn error_or_invalid_snapshot_pin_begin_error_returns_connection_to_pool() {
        let pool = memory_pool(1, Duration::from_secs(30));
        let connection = must(pool.acquire(), "connection opens");
        must(connection.begin(), "manual transaction begins");
        drop(connection);

        let error = match pool.pin_snapshot() {
            Ok(_) => panic!("nested snapshot begin should fail"),
            Err(error) => error,
        };
        assert!(error.operation().is_some());
        assert_eq!(pool.stats().active, 0);
        assert_eq!(pool.stats().idle, 1);

        let connection = must(pool.acquire(), "connection reacquires");
        must(
            connection.execute_raw("CREATE TABLE pin_error_reuse (id INTEGER PRIMARY KEY)"),
            "connection remains reusable after failed pin",
        );
    }

    #[test]
    fn in_process_fanout_pool_size_eight_holds_eight_stable_snapshots_while_writer_commits() {
        let (_tempdir, database_path, pool) = file_pool(8);
        let readers = 8usize;

        let pins: Vec<_> = (0..readers)
            .map(|_| must(pool.pin_snapshot(), "reader snapshot pin opens"))
            .collect();
        let mut records: Vec<(u64, i64)> = pins
            .iter()
            .map(|pin| {
                (
                    pin.slot_id().expect("fanout pins are pooled"),
                    snapshot_item_count(pin),
                )
            })
            .collect();
        records.sort_by_key(|(slot_id, _)| *slot_id);

        assert_eq!(records.len(), readers);
        let unique_slots: BTreeSet<u64> = records.iter().map(|(slot_id, _)| *slot_id).collect();
        assert_eq!(unique_slots.len(), readers);
        for (_slot_id, before) in &records {
            assert_eq!(*before, 1);
        }

        insert_snapshot_item(&database_path, 2, "after");

        for pin in &pins {
            assert_eq!(snapshot_item_count(pin), 1);
        }

        drop(pins);
        let fresh = must(pool.acquire(), "fresh connection opens after fanout");
        assert_eq!(snapshot_item_count(&fresh), 2);
        drop(fresh);

        let stats = pool.stats();
        assert_eq!(stats.active, 0);
        assert_eq!(stats.max_seen, readers);
        assert!(stats.idle <= readers);
    }

    #[test]
    fn acquire_timeout_emits_stats_and_serves_via_ad_hoc_connection() {
        let tempdir = must(tempfile::tempdir(), "tempdir creates");
        let database_path = tempdir.path().join("read-pool-ad-hoc.db");
        seed_snapshot_database(&database_path);
        let pool = ReadConnectionPool::new(
            DatabaseConfig::file(database_path),
            PoolConfig::new(1, Duration::from_secs(30)).with_acquire_timeout(Duration::ZERO),
        );
        let first = must(pool.pin_snapshot(), "first snapshot pin opens");

        let second = must(
            pool.pin_snapshot(),
            "pool_size=1 timeout should open ad-hoc snapshot",
        );
        assert_eq!(second.slot_id(), None);
        assert!(second.connection().is_ad_hoc());
        assert_eq!(snapshot_item_count(&second), 1);
        assert_eq!(pool.stats().active, 1);
        assert_eq!(pool.stats().ad_hoc_bypass_count, 1);
        assert_eq!(pool.stats().acquire_wait.samples, 2);

        drop(first);
        drop(second);
        assert_eq!(pool.stats().active, 0);
        assert_eq!(pool.stats().idle, 1);
    }

    #[test]
    fn snapshot_primitive_documented_source_paths_match_current_adapter() {
        let read_pool = read_repo_file("src/db/read_pool.rs");
        let db_mod = read_repo_file("src/db/mod.rs");
        let storage_docs = read_repo_file("docs/configuration/storage.md");

        assert!(read_pool.contains("src/db/mod.rs"));
        assert!(
            read_pool.contains("/dp/sqlmodel_rust/crates/sqlmodel-frankensqlite/src/connection.rs")
        );
        assert!(storage_docs.contains("src/db/mod.rs"));
        assert!(storage_docs.contains("sqlmodel-frankensqlite/src/connection.rs"));
        assert!(db_mod.contains("pub(crate) fn begin_read_snapshot(&self) -> Result<()>"));
        assert!(db_mod.contains(
            "self.execute_read_snapshot_raw(DbOperation::BeginTransaction, \"BEGIN DEFERRED\")"
        ));

        let read_snapshot_raw = db_mod
            .split("fn execute_read_snapshot_raw")
            .nth(1)
            .and_then(|tail| tail.split("fn execute_for").next())
            .expect("execute_read_snapshot_raw block is present");
        assert!(
            !read_snapshot_raw.contains("lock_file_write_owner_gate"),
            "read snapshots must not acquire the write-owner gate"
        );
    }

    #[test]
    fn property_read_pool_determinism() {
        for pool_size in [1, 2, 4, 8] {
            for pin_snapshot in [true, false] {
                let (_tempdir, database_path, pool) = file_pool(pool_size);
                let snapshot = must(
                    pool.acquire_snapshot(pin_snapshot),
                    "configured snapshot acquisition succeeds",
                );
                assert_eq!(snapshot_item_count(&snapshot), 1);

                insert_snapshot_item(&database_path, 2, "after");
                let expected_count = if pin_snapshot { 1 } else { 2 };
                assert_eq!(
                    snapshot_item_count(&snapshot),
                    expected_count,
                    "pool_size={pool_size} pin_snapshot={pin_snapshot}"
                );

                drop(snapshot);
                let fresh = must(pool.acquire(), "fresh read opens after snapshot release");
                assert_eq!(snapshot_item_count(&fresh), 2);
            }
        }
    }

    #[test]
    fn property_read_pool_snapshot_isolation() {
        let (_tempdir, database_path, pool) = file_pool(3);
        let first = must(pool.pin_snapshot(), "first pin opens");
        let second = must(pool.pin_snapshot(), "second pin opens");

        assert_eq!(snapshot_item_count(&first), 1);
        assert_eq!(snapshot_item_count(&second), 1);

        insert_snapshot_item(&database_path, 2, "after_two_pins");

        assert_eq!(snapshot_item_count(&first), 1);
        assert_eq!(snapshot_item_count(&second), 1);

        drop(first);
        drop(second);

        let later = must(pool.pin_snapshot(), "later pin opens");
        assert_eq!(snapshot_item_count(&later), 2);
        insert_snapshot_item(&database_path, 3, "after_later_pin");
        assert_eq!(snapshot_item_count(&later), 2);
    }

    #[test]
    fn concurrent_same_process_write_during_read_pin_write_commits_read_unaffected() {
        let (_tempdir, database_path, pool) = file_pool(2);
        let pin = must(pool.pin_snapshot(), "reader pin opens");
        let writer_start = Arc::new(Barrier::new(2));
        let writer_start_for_thread = Arc::clone(&writer_start);
        let writer_path = database_path.clone();

        assert_eq!(snapshot_item_count(&pin), 1);
        let writer = thread::spawn(move || {
            writer_start_for_thread.wait();
            insert_snapshot_item(&writer_path, 2, "writer_committed");
        });
        writer_start.wait();
        match writer.join() {
            Ok(()) => {}
            Err(_) => panic!("writer thread panicked"),
        }

        assert_eq!(snapshot_item_count(&pin), 1);
        drop(pin);

        let fresh = must(pool.acquire(), "fresh read opens after writer commit");
        assert_eq!(snapshot_item_count(&fresh), 2);
    }

    #[test]
    fn acquire_timeout_ad_hoc_connection_is_dropped_after_use() {
        let tempdir = must(tempfile::tempdir(), "tempdir creates");
        let database_path = tempdir.path().join("read-pool-ad-hoc-drop.db");
        seed_snapshot_database(&database_path);
        let pool = ReadConnectionPool::new(
            DatabaseConfig::file(database_path),
            PoolConfig::new(1, Duration::from_secs(30)).with_acquire_timeout(Duration::ZERO),
        );
        let first = must(pool.acquire(), "first pooled connection opens");
        let ad_hoc = must(pool.acquire(), "ad-hoc connection opens");
        assert!(ad_hoc.is_ad_hoc());
        drop(ad_hoc);

        let stats = pool.stats();
        assert_eq!(stats.active, 1);
        assert_eq!(stats.idle, 0);
        assert_eq!(stats.drops, 0);
        assert_eq!(stats.ad_hoc_bypass_count, 1);

        drop(first);
        assert_eq!(pool.stats().idle, 1);
    }

    #[test]
    fn acquire_under_contention_waits_for_pooled_release_before_bypass() {
        let pool = Arc::new(ReadConnectionPool::new(
            DatabaseConfig::memory(),
            PoolConfig::new(1, Duration::from_secs(30))
                .with_acquire_timeout(Duration::from_millis(250)),
        ));
        let first = must(pool.acquire(), "first pooled connection opens");
        let first_slot = first.slot_id();
        let waiter_ready = Arc::new(Barrier::new(2));

        let handle = {
            let pool = Arc::clone(&pool);
            let waiter_ready = Arc::clone(&waiter_ready);
            thread::spawn(move || {
                waiter_ready.wait();
                let acquired = must(pool.acquire(), "waiter acquires released pooled connection");
                (acquired.is_ad_hoc(), acquired.slot_id())
            })
        };

        waiter_ready.wait();
        thread::sleep(Duration::from_millis(20));
        drop(first);

        let (was_ad_hoc, slot_id) = match handle.join() {
            Ok(result) => result,
            Err(_) => panic!("waiter thread panicked"),
        };
        assert!(!was_ad_hoc);
        assert_eq!(slot_id, first_slot);

        let stats = pool.stats();
        assert_eq!(stats.ad_hoc_bypass_count, 0);
        assert!(stats.acquire_wait.samples >= 2);
        assert!(stats.acquire_wait.p99_ns > 0);
    }

    #[test]
    fn pool_grow_race_concurrent_acquire_at_empty_pool_respects_cap() {
        let max_size = 3;
        let readers = 8;
        let pool = Arc::new(ReadConnectionPool::new(
            DatabaseConfig::memory(),
            PoolConfig::new(max_size, Duration::from_secs(30)).with_acquire_timeout(Duration::ZERO),
        ));
        let start = Arc::new(Barrier::new(readers + 1));
        let release = Arc::new(Barrier::new(readers + 1));

        let handles: Vec<_> = (0..readers)
            .map(|_| {
                let pool = Arc::clone(&pool);
                let start = Arc::clone(&start);
                let release = Arc::clone(&release);
                thread::spawn(move || {
                    start.wait();
                    let acquired = must(pool.acquire(), "concurrent acquire succeeds");
                    let result = (acquired.is_ad_hoc(), acquired.slot_id());
                    release.wait();
                    result
                })
            })
            .collect();

        start.wait();
        release.wait();

        let mut pooled_slots = BTreeSet::new();
        let mut ad_hoc_count = 0usize;
        for handle in handles {
            let (is_ad_hoc, slot_id) = match handle.join() {
                Ok(result) => result,
                Err(_) => panic!("reader thread panicked"),
            };
            if is_ad_hoc {
                ad_hoc_count += 1;
                assert_eq!(slot_id, None);
            } else {
                pooled_slots.insert(must_some(slot_id, "pooled slot id present"));
            }
        }

        assert_eq!(pooled_slots.len(), max_size);
        assert_eq!(ad_hoc_count, readers - max_size);

        let stats = pool.stats();
        assert_eq!(stats.max_seen, max_size);
        assert_eq!(stats.ad_hoc_bypass_count, (readers - max_size) as u64);
        assert_eq!(stats.active, 0);
        assert!(stats.idle <= max_size);
    }

    #[test]
    fn metrics_acquire_wait_histogram_records_p50_and_p99() {
        let samples = acquire_wait_stats(&[10, 30, 20, 50, 40]);

        assert_eq!(samples.samples, 5);
        assert_eq!(samples.p50_ns, 30);
        assert_eq!(samples.p99_ns, 50);
    }

    #[test]
    fn metrics_acquire_wait_window_keeps_latest_1024_samples() {
        let mut state = PoolState {
            active: 0,
            idle: Vec::new(),
            acquire_wait_ns: Vec::with_capacity(ACQUIRE_WAIT_SAMPLE_CAP),
            next_slot_id: 1,
            next_pin_id: 1,
            active_pins: BTreeMap::new(),
            max_seen: 0,
            drops: 0,
            release_failures: 0,
            ad_hoc_bypass_count: 0,
        };

        for sample in 0..=ACQUIRE_WAIT_SAMPLE_CAP {
            record_acquire_wait(&mut state, Duration::from_nanos(sample as u64));
        }

        assert_eq!(state.acquire_wait_ns.len(), ACQUIRE_WAIT_SAMPLE_CAP);
        assert_eq!(state.acquire_wait_ns.first().copied(), Some(1));
        assert_eq!(
            state.acquire_wait_ns.last().copied(),
            Some(ACQUIRE_WAIT_SAMPLE_CAP as u128)
        );

        let stats = acquire_wait_stats(&state.acquire_wait_ns);
        assert_eq!(stats.samples, ACQUIRE_WAIT_SAMPLE_CAP);
        assert_eq!(stats.p50_ns, (ACQUIRE_WAIT_SAMPLE_CAP / 2) as u128 + 1);
        assert_eq!(stats.p99_ns, 1014);
    }

    #[test]
    fn in_process_fanout_latency_pool_eight_meets_speedup_budget_under_writer_load() {
        let readers = 8;
        let per_reader_work = Duration::from_millis(15);
        let (_single_tempdir, single_database_path, _single_pool) = file_pool(1);
        let (_fanout_tempdir, fanout_database_path, _fanout_pool) = file_pool(8);

        let single_latencies = pool_size_one_batch_completion_latencies(
            single_database_path,
            readers,
            per_reader_work,
        );
        let fanout_latencies = pool_size_eight_batch_completion_latencies(
            fanout_database_path,
            readers,
            per_reader_work,
        );
        let single_p50 = p50_latency_ms(&single_latencies);
        let fanout_p50 = p50_latency_ms(&fanout_latencies);

        assert!(
            fanout_p50.saturating_mul(100) <= single_p50.saturating_mul(60),
            "pool_size=8 p50 {fanout_p50}ms should be <= 60% of pool_size=1 p50 {single_p50}ms; single={single_latencies:?} fanout={fanout_latencies:?}",
        );
    }
}