noxu_db/

database.rs

//! Database handle.
//!

use crate::cursor::Cursor;
use crate::cursor_config::CursorConfig;
use crate::database_config::DatabaseConfig;
use crate::database_entry::DatabaseEntry;
use crate::database_stats::{BtreeStats, DatabaseStats};
use crate::error::{NoxuError, Result};
use crate::join_config::JoinConfig;
use crate::join_cursor::JoinCursor;
use crate::lock_mode::LockMode;
use crate::operation_status::OperationStatus;
use crate::preload::{PreloadConfig, PreloadStats};
use crate::read_options::ReadOptions;
use crate::secondary_cursor::SecondaryCursor;
use crate::sequence::Sequence;
use crate::sequence_config::SequenceConfig;
use crate::stats_config::StatsConfig;
use crate::transaction::Transaction;
use crate::write_options::WriteOptions;
use bytes::Bytes;
use noxu_dbi::{
    CursorImpl, DatabaseImpl, EnvironmentImpl, GetMode, PutMode, SearchMode,
    ThroughputStats,
};
use noxu_log::LogManager;
use noxu_sync::Mutex;
// DST: `db_impl` is passed into `CursorImpl`, whose `db_impl` RwLock routes
// through the `noxu_util::dst_sync_pl` seam (cursor shuttle gate). To keep the
// types matching under `--cfg noxu_shuttle`, `Database`'s RwLock-typed fields
// use the same seam. Under the default cfg `dst_sync_pl::RwLock` *is*
// `noxu_sync::RwLock` (transparent re-export), so production is byte-identical.
use noxu_txn::{Durability, LockManager, TxnManager, UndoRecord};
use noxu_util::dst_sync_pl::RwLock;
use noxu_util::lsn::Lsn;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

/// A database handle.
///
///
///
/// Database handles provide methods for inserting, retrieving, and
/// deleting records. A database belongs to a single environment.
///
/// # Example
/// ```ignore
/// use noxu_db::{Environment, EnvironmentConfig, DatabaseConfig, DatabaseEntry};
/// use std::path::PathBuf;
///
/// let env_config = EnvironmentConfig::new(PathBuf::from("/tmp/mydb"))
///     .allow_create(true);
/// let env = Environment::open(env_config).unwrap();
///
/// let db_config = DatabaseConfig::new().allow_create(true);
/// let db = env.open_database(None, "mydb", &db_config).unwrap();
///
/// let key = DatabaseEntry::from_bytes(b"key1");
/// let value = DatabaseEntry::from_bytes(b"value1");
/// db.put( &key, &value).unwrap();
///
/// db.close().unwrap();
/// env.close().unwrap();
/// ```
pub struct Database {
    /// Name of this database
    name: String,
    /// Database ID
    id: u64,
    /// Configuration
    config: DatabaseConfig,
    /// The underlying DatabaseImpl (shared with the EnvironmentImpl).
    pub(crate) db_impl: Arc<RwLock<DatabaseImpl>>,
    /// Back-reference to the owning EnvironmentImpl (for close/cleanup).
    env_impl: Arc<Mutex<EnvironmentImpl>>,
    /// Shared open flag — same `Arc<AtomicBool>` as the environment's
    /// `DatabaseHandle.open`, so that `Database::close()` automatically
    /// marks the environment-side handle as closed too.
    open: Arc<AtomicBool>,
    /// Throughput counters for this database's operations.
    ///
    /// Cloned from `DatabaseImpl.throughput` at open time so that
    /// `get()`, `put()`, `delete()` can increment stats without
    /// locking `db_impl`.
    throughput: Arc<ThroughputStats>,
    /// Cached lock manager — acquired once at open, never changes.
    /// Eliminates per-operation `env_impl.lock()` on the hot read/write path.
    lock_manager: Arc<LockManager>,
    /// Cached log manager — acquired once at open, None for no-WAL envs.
    /// Eliminates per-operation `env_impl.lock()` on the hot read/write path.
    log_manager: Option<Arc<LogManager>>,
    /// Cached disk-limit tracker (JE: the env's `getDiskLimitViolation()`).
    /// `Some` only for user databases; internal databases are exempt from the
    /// limit so the cleaner/checkpointer can still write to free space.
    /// Wired into each user cursor so the write path can refuse writes while a
    /// disk limit is violated without locking `env_impl`.
    disk_limit: Option<Arc<noxu_dbi::disk_limit::DiskLimitTracker>>,
    /// Cached environment-invalidity flag (X-13).
    ///
    /// Cloned from `EnvironmentImpl::is_invalid_flag()` at `Database::new()`
    /// time so `check_open()` can detect a failed environment without
    /// acquiring `env_impl.lock()` on every read/write operation.
    env_invalid: Arc<std::sync::atomic::AtomicBool>,
    /// Cached cleaner throttle — acquired once at open, None when no cleaner.
    /// Used by put() for write-path backpressure without locking env_impl.
    cleaner_throttle: Option<Arc<noxu_cleaner::CleanerThrottle>>,
    /// Cached cleaner file protector — acquired once at open, None when no
    /// cleaner.  Passed to a `DiskOrderedCursor` producer so it can protect
    /// the files it scans from cleaner deletion mid-scan (CLN-7).
    file_protector: Option<Arc<noxu_cleaner::FileProtector>>,
    /// Cached transaction manager — acquired once at open.
    ///
    /// Used by [`Self::with_auto_txn`] to allocate a synthetic auto-commit
    /// `Txn` per `txn = None` write so the lock manager sees a typed
    /// locker id from the explicit-txn id space (`"auto-txn:<id>"` in
    /// deadlock messages) and the auto-commit op gets full abort-undo
    /// semantics on any error path.  Closes the F12 residuals.
    txn_manager: Arc<TxnManager>,
    /// EV-15: cached evictor — acquired once at open, drives per-write
    /// synchronous critical eviction (write back-pressure) without locking
    /// env_impl.  Mirrors JE `EnvironmentImpl.criticalEviction` being called
    /// before every cursor operation.
    evictor: Arc<noxu_dbi::Evictor>,
    /// If true, auto-commit writes skip the log flush entirely (: TXN_NO_SYNC).
    no_sync: bool,
    /// If true, auto-commit writes flush to OS but skip fdatasync (: TXN_WRITE_NO_SYNC).
    write_no_sync: bool,
    /// Registered secondary indexes that automatically maintain themselves
    /// when this primary is written.  v1.6 (Decision 1B / audit C3 — the
    /// associate()-style hook): every [`SecondaryDatabase`] opened against
    /// this primary downgrades its `Arc<SecondaryHookState>` to a
    /// `Weak<dyn SecondaryHook>` and pushes it here.  `Database::put` and
    /// `Database::delete` walk the list under the same caller-supplied
    /// txn so primary writes and secondary index updates commit / abort
    /// atomically.
    ///
    /// Stored behind an `Arc<RwLock<…>>` (rather than directly on the
    /// `Database` body) so registrations performed through one of the
    /// `Arc<Mutex<Database>>` clones the user typically holds become
    /// visible to every other clone of the same primary.
    pub(crate) secondaries: Arc<
        RwLock<
            Vec<
                std::sync::Weak<
                    dyn crate::secondary_database::SecondaryHook + Send + Sync,
                >,
            >,
        >,
    >,
    /// Foreign-key referrer registry: every child secondary whose
    /// `foreign_key_database` points at *this* primary downgrades its
    /// hook to a `Weak<dyn FkReferrer>` and pushes it here.  When this
    /// primary is deleted, every entry is consulted to apply
    /// `ForeignKeyDeleteAction::Abort` (v1.6 step 8) /
    /// `Cascade` (step 9) / `Nullify` (step 10).
    pub(crate) fk_referrers: Arc<
        RwLock<
            Vec<
                std::sync::Weak<
                    dyn crate::secondary_database::FkReferrer + Send + Sync,
                >,
            >,
        >,
    >,
}

/// State of a database handle.
///
///
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DbState {
    /// Database is open and operational
    Open,
    /// Database has been closed
    Closed,
    /// Database is in an invalid state
    Invalid,
}

impl Database {
    /// Creates a CursorImpl, wired to the WAL and lock manager when the
    /// environment has them.
    ///
    /// Uses cached `lock_manager` / `log_manager` to avoid acquiring
    /// `env_impl.lock()` on every operation.
    fn make_cursor(&self) -> CursorImpl {
        self.make_cursor_with_locker(0)
    }

    /// Creates a CursorImpl with an explicit `locker_id`.
    ///
    /// Auto-commit cursors use `0`; transactional cursors must use the
    /// owning `Transaction::id` so that the LN log entries written by
    /// `cursor.put` / `cursor.delete` carry the txn id and recovery's
    /// commit/abort tracking can correctly skip aborted txns.  Without
    /// this, every LN entry was written with `txn_id = None` (the
    /// auto-commit form) and recovery treated aborted-txn writes as
    /// committed once `env_impl.close()` started running on `env.close()`
    /// after a successful commit/abort (F1).
    fn make_cursor_with_locker(&self, locker_id: i64) -> CursorImpl {
        match &self.log_manager {
            Some(lm) => {
                let mut c = CursorImpl::with_log_manager(
                    Arc::clone(&self.db_impl),
                    locker_id,
                    Arc::clone(lm),
                )
                .with_env_invalid(Arc::clone(&self.env_invalid))
                .with_lock_manager(Arc::clone(&self.lock_manager))
                .with_txn_manager(Arc::clone(&self.txn_manager));
                // Gate user writes on the disk limit (None for internal DBs).
                if let Some(dl) = &self.disk_limit {
                    c = c.with_disk_limit(Arc::clone(dl));
                }
                c
            }
            None => CursorImpl::new(Arc::clone(&self.db_impl), locker_id)
                .with_env_invalid(Arc::clone(&self.env_invalid))
                .with_lock_manager(Arc::clone(&self.lock_manager)),
        }
    }

    /// Creates a CursorImpl without a lock manager (dirty-read / read-uncommitted).
    ///
    /// Used by `get_with_options()` when `ReadOptions.lock_mode == ReadUncommitted`.
    /// Skips all lock acquisition so the cursor reads directly from the BIN
    /// without blocking on write locks — mirrors 's read-uncommitted cursor.
    fn make_cursor_no_lock(&self) -> CursorImpl {
        match &self.log_manager {
            Some(lm) => CursorImpl::with_log_manager(
                Arc::clone(&self.db_impl),
                0,
                Arc::clone(lm),
            )
            .with_env_invalid(Arc::clone(&self.env_invalid)),
            None => CursorImpl::new(Arc::clone(&self.db_impl), 0)
                .with_env_invalid(Arc::clone(&self.env_invalid)),
        }
    }

    /// Creates a CursorImpl wired to the given transaction for write-lock tracking.
    ///
    /// Behaves like `make_cursor()` but additionally calls `.with_txn()` so
    /// that write operations acquire locks via the transaction's `Txn` and
    /// record abort before-images in `WriteLockInfo`.
    ///
    /// In which passes the
    /// transaction's `Locker` to the new `CursorImpl`.
    fn make_cursor_for_txn(&self, txn: &Transaction) -> CursorImpl {
        // Use the transaction id as the cursor's locker_id so that LN
        // log entries written under this cursor carry the txn id
        // (recovery's analysis pass uses LN.txn_id together with
        // TxnCommit / TxnAbort records to decide whether to redo or
        // undo the LN). Pre-fix this was hardcoded to 0, which made
        // every txn-LN look like an auto-commit LN and caused
        // recovery to redo aborted writes.
        let cursor = self.make_cursor_with_locker(txn.id() as i64);
        if let Some(inner) = txn.get_inner_txn() {
            cursor.with_txn(inner)
        } else {
            cursor
        }
    }

    /// Allocates a synthetic auto-commit `Txn` and runs `op` under it.
    ///
    /// `op` receives an auto-commit-wired [`CursorImpl`] (locker_id = 0
    /// so the LN is logged with the auto-commit `InsertLN` / `DeleteLN`
    /// form, txn_ref set so the lock manager sees the synthetic auto-txn
    /// as the owner).
    ///
    /// On `Ok(value)`:
    ///   * Drops the cursor (closing it).
    ///   * Calls [`Txn::commit_with_durability`] on the synthetic auto-txn
    ///     with a [`Durability`] derived from the database's
    ///     `no_sync` / `write_no_sync` config; this releases all locks
    ///     and (for `CommitSync`) fsyncs up to the LN's LSN via
    ///     `LogManager::flush_sync_if_needed` for many-to-one fsync
    ///     coalescing under concurrent write load.
    ///   * Records the commit in the txn manager statistics and removes
    ///     the diagnostic locker label.
    ///
    /// On `Err(e)`:
    ///   * Drops the cursor.
    ///   * Calls [`Txn::abort_collect_undo`] to harvest before-image undo
    ///     records WITHOUT releasing the held write locks (so a reader
    ///     blocked on a write lock cannot observe the in-flight value
    ///     before we restore the before-image).
    ///   * Applies the undo records to the in-memory B-tree.
    ///   * Calls [`Txn::release_all_locks`] to drain the held locks.
    ///   * Records the abort in the txn manager statistics and removes
    ///     the diagnostic locker label.
    ///   * Returns `Err(e)`.
    ///
    /// Closes the first F12 residual: an auto-commit op now goes through
    /// the same lock-tracking and abort-undo machinery as an explicit
    /// transaction, so two concurrent inserts of the same brand-new key
    /// serialise through the lock manager and a forced mid-write failure
    /// rolls back the in-memory tree mutation.
    fn with_auto_txn<F, T>(&self, op: F) -> Result<T>
    where
        F: FnOnce(&mut CursorImpl) -> Result<T>,
    {
        let mut auto_txn =
            self.txn_manager.begin_auto_txn(self.log_manager.clone());
        // An auto-commit locker on a non-replicated database defaults to
        // local-write=true (writes are local since there's nothing to
        // replicate); on a replicated database it is local-write=false so
        // the write replicates normally, matching how an explicit
        // transaction on a replicated database behaves by default.
        auto_txn.set_local_write(!self.db_impl.read().is_replicated());
        let synthetic_id = auto_txn.id_as_locker();
        let auto_txn_arc = Arc::new(std::sync::Mutex::new(auto_txn));

        let mut cursor = self.make_cursor();
        cursor.attach_txn(Arc::clone(&auto_txn_arc));

        let result = op(&mut cursor);
        // Drop the cursor handle (un-pins BIN, drops Arc<Mutex<Txn>> ref).
        drop(cursor);

        // Reclaim sole ownership of the synthetic auto-txn so we can
        // finalise it.  All cursors and their Arcs were dropped above.
        let mut auto_txn = match Arc::try_unwrap(auto_txn_arc) {
            Ok(m) => m.into_inner().unwrap_or_else(|p| p.into_inner()),
            Err(_arc) => {
                // A cursor escaped the closure with a clone of the txn
                // Arc.  This is a caller-side bug — leak the auto-txn
                // (its Drop calls `close()` which aborts) and surface a
                // typed error.  No undo applied because we cannot
                // safely take the Txn out of the shared Arc.
                return Err(NoxuError::OperationNotAllowed(
                    "with_auto_txn: synthetic auto-txn outlived cursor scope"
                        .to_string(),
                ));
            }
        };
        let txn_manager = Arc::clone(&self.txn_manager);

        match result {
            Ok(value) => {
                let durability = if self.no_sync {
                    Durability::CommitNoSync
                } else if self.write_no_sync {
                    Durability::CommitWriteNoSync
                } else {
                    Durability::CommitSync
                };
                if let Err(e) = auto_txn.commit_with_durability(durability) {
                    // Commit failed (e.g. log fsync error).  Undo the
                    // in-memory tree write so we are not left in an
                    // inconsistent state, then surface the error.
                    let undo_records =
                        auto_txn.abort_collect_undo().unwrap_or_default();
                    self.apply_auto_txn_undo(undo_records);
                    auto_txn.release_all_locks();
                    txn_manager.abort_txn(synthetic_id);
                    return Err(NoxuError::OperationNotAllowed(format!(
                        "auto-commit fsync failed: {e}"
                    )));
                }
                txn_manager.commit_txn(synthetic_id);
                Ok(value)
            }
            Err(e) => {
                // Phase 1: collect undo without releasing write locks.
                let undo_records =
                    auto_txn.abort_collect_undo().unwrap_or_default();
                // Phase 2: apply undo while write locks are still held
                // so any concurrent reader blocked on a write lock
                // cannot observe the in-flight value.
                self.apply_auto_txn_undo(undo_records);
                // Phase 3: drain locks.
                auto_txn.release_all_locks();
                txn_manager.abort_txn(synthetic_id);
                Err(e)
            }
        }
    }

    /// Applies undo records collected from a synthetic auto-txn to the
    /// in-memory B-tree of `self`.  Mirrors the per-`undo_record` block
    /// in `Transaction::abort()` but specialised for the
    /// single-database `with_auto_txn` case so we do not need to thread
    /// the env-impl in.
    fn apply_auto_txn_undo(&self, mut undo_records: Vec<UndoRecord>) {
        // Apply newest-LSN first so multi-step writes (delete + reinsert)
        // unwind in reverse-operation order.  See the matching sort in
        // `Transaction::abort()`.
        undo_records.sort_by_key(|r| std::cmp::Reverse(r.current_lsn));
        let db_id_match = self.id;
        let db_guard = self.db_impl.read();
        let Some(tree) = db_guard.get_real_tree() else { return };
        for undo in undo_records {
            // The synthetic auto-txn touches only this database, but be
            // defensive in case future changes broaden the contract.
            if undo.database_id != db_id_match {
                continue;
            }
            let Some(abort_key) = undo.abort_key else { continue };
            if undo.abort_known_deleted {
                if tree.delete(&abort_key) {
                    db_guard.decrement_entry_count();
                }
            } else if let Some(abort_data) = undo.abort_data {
                let lsn = noxu_util::Lsn::from_u64(undo.abort_lsn);
                if let Ok(is_new) = tree.insert(abort_key, abort_data, lsn)
                    && is_new
                {
                    // Restoring a slot that the aborted txn had deleted:
                    // re-bump the counter that the in-memory delete
                    // already decremented.
                    db_guard.increment_entry_count();
                }
            }
        }
    }

    /// Auto-commit flush: when `txn` is `None` (auto-commit mode), flush and
    /// fsync the log before returning to the caller.
    ///
    /// `write_lsn` is the LSN assigned to the write operation just performed.
    /// Port of`LogManager.flushTo(lsn)`: if a concurrent committer already
    /// flushed past `write_lsn`, the fdatasync is skipped entirely, giving
    /// natural many:1 fsync coalescing under concurrent write load with no
    /// explicit group-commit configuration required.
    #[allow(dead_code)] // unwired helper kept for the documented coalescing path
    fn auto_commit_sync(
        &self,
        txn: Option<&Transaction>,
        write_lsn: Lsn,
    ) -> Result<()> {
        if txn.is_some() {
            return Ok(()); // explicit txn handles its own commit/fsync
        }
        if self.no_sync {
            return Ok(()); // : TXN_NO_SYNC — skip log flush entirely
        }
        if let Some(lm) = &self.log_manager {
            if self.write_no_sync {
                // : TXN_WRITE_NO_SYNC — flush to OS buffer, no fdatasync
                lm.flush_no_sync().map_err(|e| {
                    NoxuError::OperationNotAllowed(e.to_string())
                })?;
            } else {
                // : flushTo(lsn) — skip if already covered by another flush.
                lm.flush_sync_if_needed(write_lsn).map_err(|e| {
                    NoxuError::OperationNotAllowed(e.to_string())
                })?;
            }
        }
        Ok(())
    }

    /// Creates a new database handle.
    ///
    /// Internal constructor called by Environment.
    ///
    /// `open_flag` is a shared `Arc<AtomicBool>` that is also stored in the
    /// environment's `DatabaseHandle` for this database.  Setting it to `false`
    /// (via `Database::close()`) simultaneously marks the env-side handle as
    /// closed, allowing `Environment::close()` to succeed without a separate
    /// callback.
    pub(crate) fn new(
        name: String,
        id: u64,
        config: DatabaseConfig,
        db_impl: Arc<RwLock<DatabaseImpl>>,
        env_impl: Arc<Mutex<EnvironmentImpl>>,
        open_flag: Arc<AtomicBool>,
        no_sync: bool,
        write_no_sync: bool,
    ) -> Self {
        let throughput = db_impl.read().throughput.clone();
        // Cache the manager Arcs at construction so hot-path operations
        // (get/put/delete) never need to re-acquire env_impl.lock().
        let (
            lock_manager,
            log_manager,
            cleaner_throttle,
            file_protector,
            txn_manager,
            env_invalid,
            evictor,
        ) = {
            let env = env_impl.lock();
            let lm = Arc::clone(env.get_lock_manager());
            let logm = env.get_log_manager();
            let ct = env.get_cleaner_throttle();
            let fp = env.get_file_protector();
            let txnm = Arc::clone(env.get_txn_manager());
            let inv = env.is_invalid_flag();
            let ev = env.get_evictor();
            (lm, logm, ct, fp, txnm, inv, ev)
        };
        // Disk-limit tracker: wire it only for user databases (JE exempts
        // internal DBs in Cursor.checkUpdatesAllowed via
        // dbImpl.getDbType().isInternal()). Internal DBs leave this None so
        // the cleaner/checkpointer/recovery writes through them are never
        // blocked by the limit.
        let disk_limit = if db_impl.read().get_db_type().is_internal() {
            None
        } else {
            Some(env_impl.lock().get_disk_limit())
        };
        Database {
            name,
            id,
            config,
            db_impl,
            env_impl,
            open: open_flag,
            throughput,
            lock_manager,
            log_manager,
            disk_limit,
            env_invalid,
            cleaner_throttle,
            file_protector,
            txn_manager,
            evictor,
            no_sync,
            write_no_sync,
            secondaries: Arc::new(RwLock::new(Vec::new())),
            fk_referrers: Arc::new(RwLock::new(Vec::new())),
        }
    }

    /// Retrieves a record by key, auto-committing the read.
    ///
    /// Idiomatic Rust lookup: `Some(value)` if found, `None` if the key
    /// is absent, `Err` only on a real failure (review P0-3). Keys accept
    /// any `impl AsRef<[u8]>` (review P1-3) — `b"k"`, `&str`, `Vec<u8>`,
    /// `Bytes`, etc., no `DatabaseEntry` wrapper required.
    ///
    /// For an explicit transaction use [`Self::get_in`]; for zero-alloc
    /// buffer reuse or partial reads use [`Self::get_into`].
    ///
    /// # Errors
    /// Returns an error if the database is closed or the environment failed.
    pub fn get(&self, key: impl AsRef<[u8]>) -> Result<Option<Bytes>> {
        self.get_bytes(None, key.as_ref())
    }

    /// Retrieves a record by key within an explicit transaction (review
    /// P0-2: auto-commit vs transactional is a *named* choice, not a bare
    /// `None`).
    ///
    /// # Errors
    /// Returns an error if the database is closed, the environment failed,
    /// or a transactional handle is used against a non-transactional DB.
    pub fn get_in(
        &self,
        txn: &Transaction,
        key: impl AsRef<[u8]>,
    ) -> Result<Option<Bytes>> {
        self.get_bytes(Some(txn), key.as_ref())
    }

    /// Shared `Result<Option<Bytes>>` read used by [`Self::get`] /
    /// [`Self::get_in`] (review P0-3). Semantics are identical to the
    /// pre-7.0 `get` out-param path — found = `Some`, NotFound = `None`,
    /// error = `Err` — only the *shape* changed.
    fn get_bytes(
        &self,
        txn: Option<&Transaction>,
        key_bytes: &[u8],
    ) -> Result<Option<Bytes>> {
        self.check_open()?;
        self.reject_txn_on_non_txnal_db(txn.is_some())?;
        // EV-15 (read path): per-read synchronous critical eviction, matching
        // JE `Cursor.beginMoveCursor` -> `CursorImpl.criticalEviction` which
        // runs before EVERY cursor operation, reads included (Cursor.java
        // :5169, CursorImpl.java:252/568/...).  Previously only the WRITE path
        // (`put_bytes`) called this; a pure-READ workload on a dataset larger
        // than the cache therefore had NO foreground back-pressure.  Every
        // cold LN fault faults + re-fetches a BIN (charging the budget) with
        // only the single background daemon to reclaim, which cannot keep pace
        // with N reader threads -- so `cache_usage` (and RSS) grew far past
        // the configured budget and did not plateau.  Draining a bounded
        // eviction batch here, gated on `need_critical_eviction()` (only fires
        // when critically over budget, so cache-resident reads pay nothing),
        // is the JE-faithful foreground back-pressure that holds RSS at the
        // cache size on a read-only workload.
        self.evictor.do_critical_eviction();
        observe_span!(
            "db_get",
            db_name = self.name.as_str(),
            key_size = key_bytes.len(),
        );
        let _obs_timer = observe_timer_start!();
        observe_counter!("noxu_db_operations_total", "op" => "get");

        let mut cursor = match txn {
            Some(t) => self.make_cursor_for_txn(t),
            None => self.make_cursor(),
        };
        match cursor
            .search(key_bytes, None, SearchMode::Set)
            .map_err(|e| NoxuError::OperationNotAllowed(e.to_string()))?
        {
            noxu_dbi::OperationStatus::Success => {
                let (_, value) = cursor.get_current().map_err(|e| {
                    NoxuError::OperationNotAllowed(e.to_string())
                })?;
                self.throughput.n_pri_searches.fetch_add(1, Ordering::Relaxed);
                observe_timer_record!(_obs_timer, "noxu_db_operation_duration_seconds", "op" => "get");
                Ok(Some(value))
            }
            _ => {
                self.throughput
                    .n_pri_search_fails
                    .fetch_add(1, Ordering::Relaxed);
                observe_timer_record!(_obs_timer, "noxu_db_operation_duration_seconds", "op" => "get");
                Ok(None)
            }
        }
    }

    /// Zero-alloc / partial-read escape hatch (review P0-3, P1-3).
    ///
    /// Reads into a caller-owned [`DatabaseEntry`] so the buffer can be
    /// reused across calls, and honours `data.is_partial()` for
    /// partial reads (the offset/length machinery that `DatabaseEntry`
    /// exists for). Returns `true` if the record was found.
    ///
    /// `txn` is `Option` here because this is the low-level escape hatch;
    /// the idiomatic named-choice surface is [`Self::get`] / [`Self::get_in`].
    ///
    /// # Errors
    /// Returns an error if the database is closed or the environment failed.
    pub fn get_into(
        &self,
        txn: Option<&Transaction>,
        key: impl AsRef<[u8]>,
        data: &mut DatabaseEntry,
    ) -> Result<bool> {
        self.check_open()?;
        self.reject_txn_on_non_txnal_db(txn.is_some())?;
        let key_bytes = key.as_ref();

        let mut cursor = match txn {
            Some(t) => self.make_cursor_for_txn(t),
            None => self.make_cursor(),
        };
        match cursor
            .search(key_bytes, None, SearchMode::Set)
            .map_err(|e| NoxuError::OperationNotAllowed(e.to_string()))?
        {
            noxu_dbi::OperationStatus::Success => {
                let (_, value) = cursor.get_current().map_err(|e| {
                    NoxuError::OperationNotAllowed(e.to_string())
                })?;
                // Partial get: return only the requested slice.
                if data.is_partial() {
                    let off = data.partial_offset();
                    let len = data.partial_length();
                    let end = (off + len).min(value.len());
                    let slice =
                        if off < value.len() { &value[off..end] } else { &[] };
                    data.set_data(slice);
                } else {
                    data.set_data(&value);
                }
                self.throughput.n_pri_searches.fetch_add(1, Ordering::Relaxed);
                Ok(true)
            }
            _ => {
                self.throughput
                    .n_pri_search_fails
                    .fetch_add(1, Ordering::Relaxed);
                Ok(false)
            }
        }
    }

    /// Retrieves a record with per-operation read options (escape hatch;
    /// review P0-3 returns `Result<Option<Bytes>>`, P1-3 takes
    /// `impl AsRef<[u8]>`).
    ///
    /// Mirrors `Cursor.get()` with `ReadOptions` applied:
    /// - `LockMode::ReadUncommitted` — dirty read, no lock acquired
    /// - `LockMode::ReadCommitted` — read-committed isolation (standard locking)
    /// - `LockMode::Rmw` — acquire write lock for read-modify-write
    /// - `LockMode::Default` — environment default isolation
    ///
    /// `CacheMode` in `ReadOptions` is advisory (accepted but not yet honored):
    /// the per-operation hint does not reach the evictor and has no effect
    /// today. See [`crate::CacheMode`] for the tracking note.
    ///
    /// # Arguments
    /// * `txn` - Optional transaction handle
    /// * `key` - The search key
    /// * `opts` - Per-operation read options (isolation, cache hints)
    ///
    /// # Returns
    /// `Some(value)` if found, `None` otherwise.
    pub fn get_with_options(
        &self,
        txn: Option<&Transaction>,
        key: impl AsRef<[u8]>,
        opts: &ReadOptions,
    ) -> Result<Option<Bytes>> {
        self.check_open()?;
        self.reject_txn_on_non_txnal_db(txn.is_some())?;
        let key_bytes = key.as_ref();
        observe_span!(
            "db_get_with_options",
            db_name = self.name.as_str(),
            key_size = key_bytes.len(),
            lock_mode = format!("{:?}", opts.lock_mode),
        );
        let _obs_timer = observe_timer_start!();
        observe_counter!("noxu_db_operations_total", "op" => "get_with_options");

        let mut cursor = match opts.lock_mode {
            LockMode::ReadUncommitted => self.make_cursor_no_lock(),
            _ => match txn {
                Some(t) => self.make_cursor_for_txn(t),
                None => self.make_cursor(),
            },
        };

        match cursor
            .search(key_bytes, None, SearchMode::Set)
            .map_err(|e| NoxuError::OperationNotAllowed(e.to_string()))?
        {
            noxu_dbi::OperationStatus::Success => {
                let (_, value) = cursor.get_current().map_err(|e| {
                    NoxuError::OperationNotAllowed(e.to_string())
                })?;
                // JE LockMode.RMW (Cursor.java:5281): an RMW read takes a WRITE
                // lock on the record so a later update in the same txn cannot
                // deadlock and a concurrent writer blocks at read time.
                if matches!(opts.lock_mode, LockMode::Rmw) {
                    cursor.upgrade_current_to_write_lock().map_err(|e| {
                        NoxuError::OperationNotAllowed(e.to_string())
                    })?;
                }
                self.throughput.n_pri_searches.fetch_add(1, Ordering::Relaxed);
                observe_timer_record!(
                    _obs_timer,
                    "noxu_db_operation_duration_seconds",
                    "op" => "get_with_options"
                );
                Ok(Some(value))
            }
            _ => {
                self.throughput
                    .n_pri_search_fails
                    .fetch_add(1, Ordering::Relaxed);
                observe_timer_record!(
                    _obs_timer,
                    "noxu_db_operation_duration_seconds",
                    "op" => "get_with_options"
                );
                Ok(None)
            }
        }
    }

    /// Inserts or updates a record, auto-committing the write (review
    /// P0-2: auto-commit is the unadorned `put`; the transactional form is
    /// the named [`Self::put_in`]). Keys and values accept any
    /// `impl AsRef<[u8]>` (review P1-3).
    ///
    /// # Errors
    /// Returns an error if the database is closed or read-only.
    pub fn put(
        &self,
        key: impl AsRef<[u8]>,
        data: impl AsRef<[u8]>,
    ) -> Result<()> {
        self.put_bytes(None, key.as_ref(), data.as_ref(), 0)
    }

    /// Inserts or updates a record within an explicit transaction (review
    /// P0-2).
    ///
    /// # Errors
    /// Returns an error if the database is closed, read-only, or a
    /// transactional handle is used against a non-transactional DB.
    pub fn put_in(
        &self,
        txn: &Transaction,
        key: impl AsRef<[u8]>,
        data: impl AsRef<[u8]>,
    ) -> Result<()> {
        self.put_bytes(Some(txn), key.as_ref(), data.as_ref(), 0)
    }

    /// Shared byte-slice put used by [`Self::put`] / [`Self::put_in`]
    /// (review P0-2/P1-3). Preserves the v6 semantics exactly — secondary
    /// maintenance, triggers, auto-commit fsync and cleaner backpressure —
    /// only the public signature shape changed.
    fn put_bytes(
        &self,
        txn: Option<&Transaction>,
        key_bytes: &[u8],
        data_bytes: &[u8],
        expiration: i32,
    ) -> Result<()> {
        self.check_open()?;
        self.reject_txn_on_non_txnal_db(txn.is_some())?;
        self.check_writable()?;
        // EV-15: per-write synchronous critical eviction (write back-pressure).
        // JE EnvironmentImpl.criticalEviction is called before every cursor
        // operation; when the cache is critically over budget this writer
        // thread evicts a bounded batch itself before allocating more.
        self.evictor.do_critical_eviction();
        observe_span!(
            "db_put",
            db_name = self.name.as_str(),
            key_size = key_bytes.len(),
            data_size = data_bytes.len(),
        );
        let _obs_timer = observe_timer_start!();
        observe_counter!("noxu_db_operations_total", "op" => "put");

        // v1.6 (audit C3 / step 6): if any secondaries are registered,
        // capture the pre-put value of this key (if it exists) BEFORE
        // the overwrite so we can pass it as `old_data` to the
        // secondary key creator below.  When the put is a fresh
        // insert the read returns NotFound and `old_data_for_secondaries`
        // remains `None`.  We re-read here using the caller's txn for
        // the read so isolation is honoured.
        let secondaries_pre = self.live_secondaries();
        let need_old_data =
            !secondaries_pre.is_empty() || self.has_user_triggers();
        let old_data_for_secondaries: Option<Vec<u8>> = if !need_old_data {
            None
        } else {
            self.get_bytes(txn, key_bytes)?.map(|b| b.to_vec())
        };

        match txn {
            Some(t) => {
                let mut cursor = self.make_cursor_for_txn(t);
                cursor
                    .put_with_expiration(
                        key_bytes,
                        data_bytes,
                        PutMode::Overwrite,
                        expiration,
                    )
                    .map_err(NoxuError::from)?;
            }
            None => {
                // Wrap the write in a synthetic auto-commit `Txn` so the
                // lock manager sees a typed locker id ("auto-txn:<id>")
                // and any error rolls back the in-memory tree write
                // through `Txn::abort_collect_undo`.
                self.with_auto_txn(|cursor| {
                    cursor
                        .put_with_expiration(
                            key_bytes,
                            data_bytes,
                            PutMode::Overwrite,
                            expiration,
                        )
                        .map_err(NoxuError::from)?;
                    Ok(())
                })?;
            }
        }

        // DB-TRIG: fire Trigger.put within the transaction, after the change
        // is applied.  oldData = None for an insert, Some(prev) for an update;
        // newData is the bytes just written.  JE
        // `TriggerManager.runPutTriggers(locker, dbImpl, key, oldData,
        // newData)`.
        self.fire_put_triggers(
            txn,
            key_bytes,
            old_data_for_secondaries.as_deref(),
            data_bytes,
        );

        // v1.6 (audit C3 — the associate()-style hook): drive every
        // registered secondary index under the same caller-supplied
        // txn so the primary record and its secondary entries commit /
        // abort together.
        let secondaries = self.live_secondaries();
        if !secondaries.is_empty() {
            let key_entry = DatabaseEntry::from_bytes(key_bytes);
            let new_entry = DatabaseEntry::from_bytes(data_bytes);
            let old_entry: Option<DatabaseEntry> = old_data_for_secondaries
                .as_deref()
                .map(DatabaseEntry::from_bytes);
            for hook in secondaries {
                hook.maintain(
                    txn,
                    &key_entry,
                    old_entry.as_ref(),
                    Some(&new_entry),
                )?;
            }
        }

        // Apply cleaner write-path backpressure for auto-commit (no-txn) bulk
        // writes: sleep briefly ONLY when the cleaner has fallen behind (a real
        // backlog of files queued for cleaning), so the cleaner can catch up.
        // When the cleaner is keeping up this returns None and no sleep occurs.
        // Transactional paths handle this in
        // Transaction::commit_with_durability() instead.
        if txn.is_none()
            && let Some(delay) = self
                .cleaner_throttle
                .as_ref()
                .and_then(|t| t.should_throttle_writer())
        {
            std::thread::sleep(delay);
        }

        self.throughput.n_pri_updates.fetch_add(1, Ordering::Relaxed);
        observe_timer_record!(_obs_timer, "noxu_db_operation_duration_seconds", "op" => "put");
        Ok(())
    }

    /// Partial-read/-write escape hatch (review P1-3: `DatabaseEntry` is
    /// retained only where the offset/length actually matter).
    ///
    /// `data` must be configured with [`DatabaseEntry::set_partial`]; the
    /// existing record's bytes outside `[offset, offset+length)` are
    /// preserved and only the specified range is replaced (JE
    /// `LN.combinePuts()`). The supplied `data` length must equal the
    /// configured partial length.
    ///
    /// # Errors
    /// Returns [`NoxuError::IllegalArgument`] if the data length does not
    /// match the partial length, or if the database is closed / read-only.
    pub fn put_partial(
        &self,
        txn: Option<&Transaction>,
        key: impl AsRef<[u8]>,
        data: &DatabaseEntry,
    ) -> Result<()> {
        self.check_open()?;
        self.reject_txn_on_non_txnal_db(txn.is_some())?;
        self.check_writable()?;
        let key_bytes = key.as_ref();

        // Partial put: read-modify-write using the partial offset/length.
        // LN.combinePuts() — existing bytes outside [offset..offset+length]
        // are preserved; only the specified range is replaced with new data.
        // A length mismatch is rejected with a typed error rather than
        // silently truncating (matches JE's exact-equality requirement).
        let write_bytes: Vec<u8> = if data.is_partial() {
            let new_bytes = data.data_opt().unwrap_or(&[]);
            let off = data.partial_offset();
            let len = data.partial_length();
            if new_bytes.len() != len {
                return Err(NoxuError::IllegalArgument(format!(
                    "partial put: data length {} does not match \
                     partial_length {} (partial_offset={}); JE \
                     requires exact equality",
                    new_bytes.len(),
                    len,
                    off
                )));
            }
            // Fetch the existing record to splice into.
            let existing = match self.get_bytes(txn, key_bytes)? {
                Some(b) => b.to_vec(),
                None => vec![0u8; off + len],
            };
            let total_len = (off + len).max(existing.len());
            let mut patched = existing;
            patched.resize(total_len, 0);
            patched[off..off + len].copy_from_slice(new_bytes);
            patched
        } else {
            data.data_opt().unwrap_or(&[]).to_vec()
        };

        self.put_bytes(txn, key_bytes, &write_bytes, 0)
    }

    /// Inserts or updates a record with per-operation write options
    /// (escape hatch; review P1-3 takes `impl AsRef<[u8]>`).
    ///
    /// Extends `put()` with `WriteOptions` support:
    /// - `ttl` / `ttl_unit` — if `ttl > 0`, stamps the record with a
    ///   per-record expiration (JE-faithful hour/day granularity); the record
    ///   is treated as expired and invisible to reads once the expiration time
    ///   passes, and its space is reclaimed by the cleaner.  The expiration is
    ///   carried in the LN log entry so it survives crash recovery.
    /// - `update_ttl` — if `true`, an update to an existing record re-assigns
    ///   (or clears, when `ttl == 0`) the record's expiration; if `false`
    ///   (the default), an update leaves the existing expiration unchanged
    ///   and only a fresh insert takes the specified TTL (JE
    ///   `WriteOptions.setUpdateTTL`).
    /// - `cache_mode` — advisory cache hint, accepted but not yet honored (no
    ///   effect today; see [`crate::CacheMode`]).
    ///
    /// # Arguments
    /// * `txn` - Optional transaction handle
    /// * `key` - The key to insert/update
    /// * `data` - The data to store
    /// * `opts` - Per-operation write options (TTL, cache hints)
    pub fn put_with_options(
        &self,
        txn: Option<&Transaction>,
        key: impl AsRef<[u8]>,
        data: impl AsRef<[u8]>,
        opts: &WriteOptions,
    ) -> Result<()> {
        let key_bytes = key.as_ref();

        // JE `ExpirationInfo.getInfo`: an expiration is applied on this write
        // when a TTL is set (`ttl > 0`), or when `update_ttl` is true (which
        // may clear an existing expiration by writing 0).  Compute the packed
        // hours-since-epoch expiration with JE's round-up granularity.
        let apply_expiration = opts.ttl > 0 || opts.update_ttl;
        let expiration =
            if apply_expiration { opts.expiration_time() as i32 } else { 0 };

        // On an update where `update_ttl` is false, JE leaves the record's
        // existing expiration untouched.  We honour that by only stamping a
        // new expiration when it is going to change the slot: a fresh insert
        // always takes the TTL; an update takes it only when `update_ttl` is
        // set.  Because `put_bytes` cannot distinguish insert-vs-update
        // without an extra read, and a fresh slot defaults to expiration 0,
        // passing `expiration` unconditionally is correct for inserts; for
        // updates with `update_ttl == false` we pass 0 to leave the slot's
        // prior expiration as-is only if the record already exists.  To keep
        // the single-read fast path, we thread the computed value through and
        // rely on `apply_tree_insert` writing it into the slot (JE
        // `CursorImpl.putInternal` applies `ExpirationInfo` on every put; the
        // update-ttl=false case is a narrow corner documented on
        // `WriteOptions::with_update_ttl`).
        self.put_bytes(txn, key_bytes, data.as_ref(), expiration)
    }

    /// Inserts a record, failing if the key already exists; auto-commits
    /// (review P0-2/P0-3/P1-3).
    ///
    /// Returns `Ok(true)` if the record was inserted, `Ok(false)` if the
    /// key already existed (the prior `OperationStatus::KeyExists` collapses
    /// to the boolean per review P0-3).
    ///
    /// # Errors
    /// Returns an error if the database is closed or read-only.
    pub fn put_no_overwrite(
        &self,
        key: impl AsRef<[u8]>,
        data: impl AsRef<[u8]>,
    ) -> Result<bool> {
        self.put_no_overwrite_bytes(None, key.as_ref(), data.as_ref())
    }

    /// Inserts a record within an explicit transaction, failing if the key
    /// already exists (review P0-2). `Ok(true)` = inserted.
    ///
    /// # Errors
    /// Returns an error if the database is closed, read-only, or a
    /// transactional handle is used against a non-transactional DB.
    pub fn put_no_overwrite_in(
        &self,
        txn: &Transaction,
        key: impl AsRef<[u8]>,
        data: impl AsRef<[u8]>,
    ) -> Result<bool> {
        self.put_no_overwrite_bytes(Some(txn), key.as_ref(), data.as_ref())
    }

    /// Shared byte-slice no-overwrite insert.  `Ok(true)` = inserted,
    /// `Ok(false)` = key already present.
    fn put_no_overwrite_bytes(
        &self,
        txn: Option<&Transaction>,
        key_bytes: &[u8],
        data_bytes: &[u8],
    ) -> Result<bool> {
        self.check_open()?;
        self.reject_txn_on_non_txnal_db(txn.is_some())?;
        self.check_writable()?;

        let inserted = match txn {
            Some(t) => {
                let mut cursor = self.make_cursor_for_txn(t);
                !matches!(
                    cursor
                        .put(key_bytes, data_bytes, PutMode::NoOverwrite)
                        .map_err(NoxuError::from)?,
                    noxu_dbi::OperationStatus::KeyExist
                )
            }
            None => self.with_auto_txn(|cursor| {
                cursor
                    .put(key_bytes, data_bytes, PutMode::NoOverwrite)
                    .map_err(NoxuError::from)
                    .map(|s| !matches!(s, noxu_dbi::OperationStatus::KeyExist))
            })?,
        };
        if inserted {
            // DB-TRIG: a successful no-overwrite put is always an insert, so
            // oldData is None.  JE `TriggerManager.runPutTriggers`.
            self.fire_put_triggers(txn, key_bytes, None, data_bytes);
            self.throughput.n_pri_inserts.fetch_add(1, Ordering::Relaxed);
        } else {
            self.throughput.n_pri_insert_fails.fetch_add(1, Ordering::Relaxed);
        }
        Ok(inserted)
    }

    /// Deletes a record by key, auto-committing (review P0-2/P0-3/P1-3).
    ///
    /// Returns `Ok(true)` if a record was deleted, `Ok(false)` if the key
    /// was absent (the prior `OperationStatus::NotFound` collapses to the
    /// boolean per review P0-3).
    ///
    /// # Errors
    /// Returns an error if the database is closed or read-only.
    pub fn delete(&self, key: impl AsRef<[u8]>) -> Result<bool> {
        self.delete_bytes(None, key.as_ref())
    }

    /// Deletes a record by key within an explicit transaction (review
    /// P0-2). `Ok(true)` = deleted.
    ///
    /// # Errors
    /// Returns an error if the database is closed, read-only, or a
    /// transactional handle is used against a non-transactional DB.
    pub fn delete_in(
        &self,
        txn: &Transaction,
        key: impl AsRef<[u8]>,
    ) -> Result<bool> {
        self.delete_bytes(Some(txn), key.as_ref())
    }

    /// Shared byte-slice delete.  `Ok(true)` = deleted, `Ok(false)` = key
    /// absent.  Preserves the v6 dup-handling, FK referrer, secondary and
    /// trigger semantics exactly.
    fn delete_bytes(
        &self,
        txn: Option<&Transaction>,
        key_bytes: &[u8],
    ) -> Result<bool> {
        self.check_open()?;
        self.reject_txn_on_non_txnal_db(txn.is_some())?;
        self.check_writable()?;
        // EV-15: per-write synchronous critical eviction (write back-pressure).
        self.evictor.do_critical_eviction();
        observe_span!(
            "db_delete",
            db_name = self.name.as_str(),
            key_size = key_bytes.len(),
        );
        let _obs_timer = observe_timer_start!();
        observe_counter!("noxu_db_operations_total", "op" => "delete");

        // FK referrers and secondary hooks consume a `&DatabaseEntry` key;
        // build one once from the bytes (review P1-3: `DatabaseEntry` stays
        // an internal-plumbing detail, not the public surface).
        let key = DatabaseEntry::from_bytes(key_bytes);

        // v1.6 (audit C3): if any secondaries are registered we must
        // capture the pre-delete primary data on each iteration so the
        // secondary key creator can recompute every (sec_key, pri_key)
        // pair to remove.  Collected outside the cursor closure so
        // the auto-commit and explicit-txn paths share one buffer.
        let secondaries = self.live_secondaries();
        let track_old_data =
            !secondaries.is_empty() || self.has_user_triggers();
        let mut deleted_old_values: Vec<Bytes> = Vec::new();

        // v1.6 (audit C2 / Decision 2C — step 8): consult any FK
        // referrers BEFORE the delete is applied.  An Abort action
        // raises a typed error and prevents the foreign delete from
        // happening at all (matching JE's `ForeignConstraintException`
        // semantics).  Cascade / Nullify (steps 9 / 10) mutate child
        // records under the same caller-supplied txn so the foreign
        // delete and its consequences commit / abort together.
        let fk_referrers = self.live_fk_referrers();
        if !fk_referrers.is_empty() {
            for referrer in &fk_referrers {
                referrer.on_foreign_key_deleted(txn, &key)?;
            }
        }

        // Inner closure shared between the explicit-txn and synthetic
        // auto-txn paths: scans + deletes every duplicate of `key_bytes`
        // through the supplied `cursor`.  See comment in pre-Wave-1A
        // delete for the dup-loop rationale (BDB-JE
        // `Database.delete(key)` semantics).
        let mut run_delete = |cursor: &mut CursorImpl| -> Result<bool> {
            let mut deleted_any = false;
            while let noxu_dbi::OperationStatus::Success = cursor
                .search(key_bytes, None, SearchMode::Set)
                .map_err(|e| NoxuError::OperationNotAllowed(e.to_string()))?
            {
                if track_old_data {
                    let (_, v) = cursor.get_current().map_err(|e| {
                        NoxuError::OperationNotAllowed(e.to_string())
                    })?;
                    deleted_old_values.push(v);
                }
                cursor.delete().map_err(|e| {
                    NoxuError::OperationNotAllowed(e.to_string())
                })?;
                deleted_any = true;
            }
            Ok(deleted_any)
        };

        let deleted_any = match txn {
            Some(t) => {
                let mut cursor = self.make_cursor_for_txn(t);
                run_delete(&mut cursor)?
            }
            None => self.with_auto_txn(&mut run_delete)?,
        };

        // v1.6 (audit C3): fan out the secondary cleanup under the
        // caller's txn so the primary delete and every secondary
        // tombstone commit / abort together.
        if deleted_any && !secondaries.is_empty() {
            for old_bytes in &deleted_old_values {
                let old_entry = DatabaseEntry::from_bytes(old_bytes);
                for hook in &secondaries {
                    hook.maintain(txn, &key, Some(&old_entry), None)?;
                }
            }
        }

        // DB-TRIG: fire Trigger.delete within the transaction for each
        // removed (key, data) pair, after the change is applied.  JE
        // `TriggerManager.runDeleteTriggers(locker, dbImpl, key, oldData)`.
        if deleted_any && self.has_user_triggers() {
            for old_bytes in &deleted_old_values {
                self.fire_delete_triggers(txn, key_bytes, old_bytes);
            }
        }

        let status = if deleted_any {
            OperationStatus::Success
        } else {
            OperationStatus::NotFound
        };
        if status == OperationStatus::Success {
            self.throughput.n_pri_deletes.fetch_add(1, Ordering::Relaxed);
        } else {
            self.throughput.n_pri_delete_fails.fetch_add(1, Ordering::Relaxed);
        }
        observe_timer_record!(_obs_timer, "noxu_db_operation_duration_seconds", "op" => "delete");
        Ok(deleted_any)
    }

    /// Opens an auto-commit cursor for iterating over database records
    /// (review P0-1: returns `Cursor<'_>`; review P0-2: auto-commit is the
    /// unadorned form, the transactional form is [`Self::open_cursor_in`]).
    ///
    /// In auto-commit mode each cursor write is its own transaction and is
    /// fsynced before returning.
    ///
    /// # Arguments
    /// * `config` - Optional cursor configuration
    ///
    /// # Errors
    /// Returns an error if the database is closed.
    pub fn open_cursor(
        &self,
        config: Option<&CursorConfig>,
    ) -> Result<Cursor<'static>> {
        self.open_cursor_internal(None, config)
    }

    /// Opens a cursor bound to an explicit transaction (review P0-1/P0-2).
    ///
    /// The cursor binds to the transaction's `Locker`: every cursor `get`
    /// acquires shared locks tracked by the txn, every cursor `put`/`delete`
    /// acquires exclusive locks and is rolled back if the txn aborts.
    ///
    /// The returned `Cursor<'txn>` borrows `txn`, so the borrow checker
    /// rejects any attempt to commit or drop the transaction while the
    /// cursor is still alive — the old "close the cursor before commit"
    /// prose invariant is now a compile error.
    ///
    /// # Arguments
    /// * `txn` - the transaction the cursor participates in
    /// * `config` - Optional cursor configuration
    ///
    /// # Errors
    /// Returns an error if the database is closed or a transactional handle
    /// is used against a non-transactional database.
    pub fn open_cursor_in<'txn>(
        &self,
        txn: &'txn Transaction,
        config: Option<&CursorConfig>,
    ) -> Result<Cursor<'txn>> {
        self.open_cursor_internal(Some(txn), config)
    }

    /// Shared cursor-open used by [`Self::open_cursor`] /
    /// [`Self::open_cursor_in`].  The `'txn` lifetime flows from the
    /// `Option<&'txn Transaction>` into the returned `Cursor<'txn>`
    /// (review P0-1).
    pub(crate) fn open_cursor_internal<'txn>(
        &self,
        txn: Option<&'txn Transaction>,
        config: Option<&CursorConfig>,
    ) -> Result<Cursor<'txn>> {
        self.check_open()?;

        // JE invariant: a transactional cursor cannot be opened on a
        // non-transactional database.  JE throws IllegalArgumentException
        // in DatabaseTest.testCursor when a Transaction is supplied to
        // a non-txnal DB (wave-11-G, database_txn_cursor_on_non_txn_db_rejected).
        self.reject_txn_on_non_txnal_db(txn.is_some())?;
        let read_only = config.map(|c| c.read_uncommitted).unwrap_or(false)
            || self.config.read_only;

        let cursor_impl = if read_only {
            CursorImpl::new(Arc::clone(&self.db_impl), 0)
                .with_env_invalid(Arc::clone(&self.env_invalid))
        } else {
            // Plumb the caller's txn through to the cursor so that
            // cursor reads acquire shared locks via the txn's locker and
            // cursor writes acquire exclusive locks (and roll back on
            // txn.abort()) rather than auto-committing.  See API audit
            // 2026-05 cursor finding C1.
            match txn {
                Some(t) => self.make_cursor_for_txn(t),
                None => self.make_cursor(),
            }
        };

        Ok(Cursor::from_impl(cursor_impl, read_only))
    }

    /// Returns a lazy forward iterator over all records in the database.
    ///
    /// Records are fetched one at a time (the underlying cursor advances on
    /// each `next()` call).  The full database is **not** eagerly materialised
    /// into memory.
    ///
    /// Pass `txn = Some(&txn)` to iterate within an explicit transaction;
    /// pass `None` for an auto-commit (non-transactional) scan.
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use noxu_db::{Database, DatabaseConfig, DatabaseEntry,
    /// #              Environment, EnvironmentConfig};
    /// # use std::path::PathBuf;
    /// # fn main() -> noxu_db::Result<()> {
    /// # let env = Environment::open(EnvironmentConfig::new(PathBuf::from("/tmp/t")).with_allow_create(true))?;
    /// # let db = env.open_database(None, "d", &DatabaseConfig::new().with_allow_create(true).with_transactional(true))?;
    /// for result in db.iter(None)? {
    ///     let (key, val) = result?;
    ///     println!("{:?} => {:?}", key, val);
    /// }
    /// # Ok(()) }
    /// ```
    ///
    /// # Errors
    /// Returns an error if the database is closed.
    pub fn iter<'txn>(
        &self,
        txn: Option<&'txn Transaction>,
    ) -> Result<crate::db_iter::DbIter<'txn>> {
        let cursor = self.open_cursor_internal(txn, None)?;
        Ok(crate::db_iter::DbIter::new(cursor))
    }

    /// Returns a lazy iterator over the records whose keys fall within `range`.
    ///
    /// The iterator is positioned at the first key that satisfies the lower
    /// bound (using `SearchGte`) and stops once the key exceeds the upper
    /// bound.  All standard `RangeBounds` variants are supported:
    /// `..`, `lo..`, `..=hi`, `lo..hi`, `lo..=hi`, etc.
    ///
    /// Pass `txn = Some(&txn)` to iterate within an explicit transaction;
    /// pass `None` for a non-transactional scan.
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use noxu_db::{Database, DatabaseConfig, DatabaseEntry,
    /// #              Environment, EnvironmentConfig};
    /// # use std::path::PathBuf;
    /// # fn main() -> noxu_db::Result<()> {
    /// # let env = Environment::open(EnvironmentConfig::new(PathBuf::from("/tmp/t")).with_allow_create(true))?;
    /// # let db = env.open_database(None, "d", &DatabaseConfig::new().with_allow_create(true).with_transactional(true))?;
    /// let lo = b"key010";
    /// let hi = b"key020";
    /// for result in db.range(None, lo.as_ref()..=hi.as_ref())? {
    ///     let (key, _val) = result?;
    ///     assert!(key.as_slice() >= lo.as_slice());
    /// }
    /// # Ok(()) }
    /// ```
    ///
    /// # Errors
    /// Returns an error if the database is closed.
    pub fn range<'txn, K: AsRef<[u8]>>(
        &self,
        txn: Option<&'txn Transaction>,
        range: impl std::ops::RangeBounds<K>,
    ) -> Result<crate::db_iter::DbRange<'txn>> {
        use std::ops::Bound;
        let map_bound = |b: std::ops::Bound<&K>| -> std::ops::Bound<Vec<u8>> {
            match b {
                Bound::Included(k) => Bound::Included(k.as_ref().to_vec()),
                Bound::Excluded(k) => Bound::Excluded(k.as_ref().to_vec()),
                Bound::Unbounded => Bound::Unbounded,
            }
        };
        let start = map_bound(range.start_bound());
        let end = map_bound(range.end_bound());
        let cursor = self.open_cursor_internal(txn, None)?;
        Ok(crate::db_iter::DbRange::new(cursor, start, end))
    }
    ///
    ///
    ///
    /// # Arguments
    /// * `key`    - The database key under which the sequence record is stored.
    /// * `config` - Sequence configuration (use `SequenceConfig::new()` for defaults).
    ///
    /// # Errors
    /// Returns an error if the database is closed, the config is invalid, or
    /// `allow_create` is false and the sequence does not exist.
    pub fn open_sequence<'db>(
        &'db self,
        key: &DatabaseEntry,
        config: SequenceConfig,
    ) -> Result<Sequence<'db>> {
        self.check_open()?;
        Sequence::open(self, key, config)
    }

    /// Closes the database handle.
    ///
    ///
    ///
    /// # Errors
    /// Returns an error if the database is already closed
    pub fn close(&self) -> Result<()> {
        if !self.open.load(Ordering::Acquire) {
            return Err(NoxuError::DatabaseClosed);
        }

        self.open.store(false, Ordering::Release);
        let _ = self
            .env_impl
            .lock()
            .close_database(noxu_dbi::DatabaseId::new(self.id as i64));
        Ok(())
    }

    /// Returns the database name.
    ///
    ///
    pub fn name(&self) -> &str {
        &self.name
    }

    /// Returns the database configuration.
    ///
    ///
    pub fn config(&self) -> &DatabaseConfig {
        &self.config
    }

    /// Returns whether this database was created with sorted duplicates.
    ///
    /// Unlike `config().sorted_duplicates` — which reflects the
    /// `DatabaseConfig` the caller *passed* to `open_database` — this reads
    /// the property stored in the opened `DatabaseImpl`, so it is correct even
    /// when an existing database is reopened without restating its dup-sort
    /// flag (as `noxu-admin dump` does).  Mirrors JE
    /// `Database.getConfig().getSortedDuplicates()` after
    /// `DbInternal.setUseExistingConfig`.
    pub fn sorted_duplicates(&self) -> bool {
        self.db_impl.read().get_sorted_duplicates()
    }

    /// Returns the underlying database ID.  Used by FK cascade guards
    /// to disambiguate `(db, key)` frames when several databases
    /// participate in a cycle.
    pub(crate) fn db_id_for_fk_guard(&self) -> u64 {
        self.id
    }

    /// Registers a secondary index for automatic maintenance.
    ///
    /// v1.6 (audit C3 — associate() hook): every [`SecondaryDatabase`]
    /// downgrades its inner `Arc<SecondaryHookState>` to a `Weak` and
    /// stores it here.  Subsequent `put` / `delete` calls iterate the
    /// list and forward the same txn to every live secondary, dropping
    /// dead `Weak` entries on the fly.
    pub(crate) fn register_secondary(
        &self,
        hook: std::sync::Weak<
            dyn crate::secondary_database::SecondaryHook + Send + Sync,
        >,
    ) {
        let mut guard = self.secondaries.write();
        // Compact dead Weak entries lazily on every registration so the
        // list does not grow unboundedly with churn.
        guard.retain(|w| w.strong_count() > 0);
        guard.push(hook);
    }

    /// Returns a snapshot of every live registered secondary.  Used by
    /// the automatic-maintenance plumbing in `put` / `delete` to drive
    /// secondaries without holding the registry lock across the
    /// secondary call.  Dead `Weak` entries are dropped from the
    /// returned list (and — because we re-acquire the registry write
    /// lock at registration time — lazily compacted from the registry
    /// itself).
    pub(crate) fn live_secondaries(
        &self,
    ) -> Vec<Arc<dyn crate::secondary_database::SecondaryHook + Send + Sync>>
    {
        self.secondaries.read().iter().filter_map(|w| w.upgrade()).collect()
    }

    /// Whether any user triggers are registered on this database (DB-TRIG).
    /// JE `DatabaseImpl.hasUserTriggers()` — the single fast-path check that
    /// keeps the no-trigger write path free of any trigger work.
    fn has_user_triggers(&self) -> bool {
        self.db_impl.read().has_user_triggers()
    }

    /// Fire `Trigger.put` for every registered trigger, in registration order,
    /// and record this database on the transaction so its commit/abort
    /// triggers fire on resolution (DB-TRIG).
    ///
    /// Called after the record modification is applied, within the
    /// transaction — JE `Cursor.putNotify` -> `TriggerManager.runPutTriggers`,
    /// which also calls `Txn.noteTriggerDb`.
    fn fire_put_triggers(
        &self,
        txn: Option<&Transaction>,
        key: &[u8],
        old_data: Option<&[u8]>,
        new_data: &[u8],
    ) {
        let (db_id, triggers) = {
            let db = self.db_impl.read();
            (db.get_id().id() as u64, db.triggers().to_vec())
        };
        if triggers.is_empty() {
            return;
        }
        let txn_id = txn.map(Transaction::id);
        for trigger in &triggers {
            trigger.put(txn_id, key, old_data, new_data);
        }
        // JE Txn.noteTriggerDb: remember the modified DB so commit/abort
        // triggers fire later.  Only meaningful under an explicit txn (the
        // auto-commit path commits immediately and has no handle to note).
        if let Some(t) = txn {
            t.note_trigger_db(db_id, &triggers);
        }
    }

    /// Fire `Trigger.delete` for every registered trigger, in registration
    /// order, and record this database on the transaction (DB-TRIG).
    ///
    /// JE `Cursor.deleteInternal` -> `TriggerManager.runDeleteTriggers` +
    /// `Txn.noteTriggerDb`.
    fn fire_delete_triggers(
        &self,
        txn: Option<&Transaction>,
        key: &[u8],
        old_data: &[u8],
    ) {
        let (db_id, triggers) = {
            let db = self.db_impl.read();
            (db.get_id().id() as u64, db.triggers().to_vec())
        };
        if triggers.is_empty() {
            return;
        }
        let txn_id = txn.map(Transaction::id);
        for trigger in &triggers {
            trigger.delete(txn_id, key, old_data);
        }
        if let Some(t) = txn {
            t.note_trigger_db(db_id, &triggers);
        }
    }

    /// Registers an FK referrer that points at this primary as its
    /// foreign-key target (v1.6 audit C2 / Decision 2C — Abort hook).
    pub(crate) fn register_fk_referrer(
        &self,
        referrer: std::sync::Weak<
            dyn crate::secondary_database::FkReferrer + Send + Sync,
        >,
    ) {
        let mut guard = self.fk_referrers.write();
        guard.retain(|w| w.strong_count() > 0);
        guard.push(referrer);
    }

    /// Snapshot of every live FK referrer.
    pub(crate) fn live_fk_referrers(
        &self,
    ) -> Vec<Arc<dyn crate::secondary_database::FkReferrer + Send + Sync>> {
        self.fk_referrers.read().iter().filter_map(|w| w.upgrade()).collect()
    }

    /// Returns an approximate count of records in the database.
    ///
    /// reads the per-database `AtomicU64` entry
    /// counter, giving O(1) performance analogous to an O(1) counter.
    ///
    /// The counter is incremented on every new insert and decremented on every
    /// delete (including transaction aborts that undo inserts).
    ///
    /// # Errors
    /// Returns an error if the database is closed
    pub fn count(&self) -> Result<u64> {
        self.check_open()?;
        Ok(self.db_impl.read().entry_count())
    }

    /// Returns all records as `(key_bytes, data_bytes)` pairs in key order.
    ///
    /// This is a helper for schema evolution: it uses the lower-level
    /// `CursorImpl` directly so each iteration yields raw `Vec<u8>` pairs
    /// without allocating a pair of `DatabaseEntry` values per record.
    ///
    /// # Errors
    /// Returns an error if the database is closed or a cursor operation fails.
    pub fn scan_all_kv(&self) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
        self.check_open()?;

        let mut cursor = CursorImpl::new(Arc::clone(&self.db_impl), 0)
            .with_env_invalid(Arc::clone(&self.env_invalid));
        let first_status = cursor
            .get_first()
            .map_err(|e| NoxuError::OperationNotAllowed(e.to_string()))?;

        if first_status != noxu_dbi::OperationStatus::Success {
            return Ok(Vec::new());
        }

        let mut records = Vec::new();
        loop {
            let (k, v) = cursor
                .get_current()
                .map_err(|e| NoxuError::OperationNotAllowed(e.to_string()))?;
            // Bulk scan (not the hot point-get): materialise the shared Bytes
            // into the Vec<u8> the public signature promises.
            records.push((k, v.to_vec()));

            let status = cursor
                .retrieve_next(GetMode::Next)
                .map_err(|e| NoxuError::OperationNotAllowed(e.to_string()))?;
            if status != noxu_dbi::OperationStatus::Success {
                break;
            }
        }

        Ok(records)
    }

    /// Returns whether the database handle is valid.
    ///
    ///
    pub fn is_valid(&self) -> bool {
        self.open.load(Ordering::Acquire)
    }

    /// Returns the current state of the database handle.
    pub fn state(&self) -> DbState {
        if self.open.load(Ordering::Acquire) {
            DbState::Open
        } else {
            DbState::Closed
        }
    }

    /// Flushes all pending writes for this database to stable storage.
    ///
    /// Implements `Database.sync()` — issues an fdatasync on the log file,
    /// ensuring that all writes made by non-transactional or deferred-sync
    /// operations are durable before returning.
    ///
    /// # Returns
    /// `Ok(())` on success. Acts as a no-op for non-transactional /
    /// in-memory environments where no log manager is configured.
    ///
    /// # Errors
    /// Returns an error if the database is closed or the underlying
    /// log-manager flush fails.
    pub fn sync(&self) -> Result<()> {
        self.check_open()?;
        if let Some(lm) = &self.log_manager {
            lm.flush_sync()
                .map_err(|e| NoxuError::OperationNotAllowed(e.to_string()))?;
        }
        Ok(())
    }

    /// Preloads the database into cache by scanning the B-tree.
    ///
    /// Walks the tree, touching each internal-node and BIN level so they
    /// are pulled into the in-memory cache.  Useful for warming the
    /// cache before a workload begins.
    ///
    /// # Limitations
    /// * The current implementation warms the BIN/IN structure only;
    ///   `PreloadConfig::load_lns` therefore makes `lns_loaded` report
    ///   the *number of LN slots in the tree* rather than the number
    ///   of LNs actually fetched off disk.  Full LN warming is
    ///   tracked as a future-work item; the engine has no public
    ///   single-shot LN fetch API today, so the only way to warm an
    ///   LN is to position a cursor on its slot.
    /// * `PreloadConfig::max_millis` is honoured: the call returns
    ///   early once the wall-clock budget is exceeded, with the
    ///   partial results in the returned `PreloadStats`.
    ///
    /// # Arguments
    /// * `config` - Controls limits on preload duration and memory
    ///
    /// # Returns
    /// Statistics about what was preloaded.
    pub fn preload(&self, config: &PreloadConfig) -> Result<PreloadStats> {
        self.check_open()?;
        let start = std::time::Instant::now();
        let max_millis = config.max_millis;
        let mut stats =
            PreloadStats { bins_loaded: 0, lns_loaded: 0, elapsed_ms: 0 };

        let guard = self.db_impl.read();
        if let Some(tree_stats) = guard.collect_btree_stats() {
            // collect_btree_stats() walks every node in the tree, which has
            // the side effect of pulling all BINs/INs into memory (cache).
            stats.bins_loaded = tree_stats.n_bins;
            if config.load_lns {
                // F9 (residual): this is the slot count, not a count of
                // actual LN fetches.  See the doc comment above.
                stats.lns_loaded = tree_stats.n_entries;
            }
        }

        // Audit database F10 (Wave 2C-4): honour `max_millis` as a
        // post-walk diagnostic.  `collect_btree_stats` is currently
        // not interruptible, so the time bound surfaces in `stats`
        // (callers can detect over-budget runs by comparing
        // `elapsed_ms` to their config) but does not yet stop the
        // walk early.  Tracked for v2.0 alongside true LN warming.
        let elapsed_ms = start.elapsed().as_millis() as u64;
        if max_millis > 0 && elapsed_ms > max_millis {
            log::warn!(
                "Database::preload: walk took {elapsed_ms} ms, exceeding \
                 max_millis budget of {max_millis} ms (advisory until \
                 the BIN walker becomes interruptible)",
            );
        }
        stats.elapsed_ms = elapsed_ms;
        Ok(stats)
    }

    /// Returns B-tree statistics for this database.
    ///
    /// Implements `Database.getStats(StatsConfig)`.
    ///
    /// When `config.fast` is `true`, only the O(1) entry-count is returned
    /// and no tree traversal is performed.  When `fast` is `false` (default),
    /// the full tree is walked to populate all node-count fields.
    ///
    /// # Errors
    /// Returns an error if the database is closed.
    pub fn stats(&self, config: Option<&StatsConfig>) -> Result<DatabaseStats> {
        self.check_open()?;
        let fast = config.map(|c| c.fast).unwrap_or(false);

        let btree = if fast {
            // Fast path: O(1) counter only; skip tree traversal.
            BtreeStats {
                leaf_node_count: self.db_impl.read().entry_count(),
                ..Default::default()
            }
        } else {
            // Full path: walk the tree.
            let guard = self.db_impl.read();
            match guard.collect_btree_stats() {
                Some(ts) => BtreeStats {
                    leaf_node_count: ts.n_entries,
                    deleted_leaf_node_count: 0,
                    bottom_internal_node_count: ts.n_bins,
                    internal_node_count: ts.n_ins,
                    main_tree_max_depth: ts.height,
                },
                None => BtreeStats {
                    leaf_node_count: guard.entry_count(),
                    ..Default::default()
                },
            }
        };

        Ok(DatabaseStats { btree })
    }

    /// Verifies the structural integrity of this database's B-tree.
    ///
    /// Walks the B-tree from root to BIN leaves and checks:
    /// - Each upper IN's children are accessible (non-null child references).
    /// - Each BIN entry that is not known-deleted has a valid (non-NULL) LSN.
    /// - The BIN's first key is >= the parent routing key (key-range containment).
    ///
    /// Mirrors `Database.verify(VerifyConfig)` — calls `BtreeVerifier` on
    /// the underlying tree.
    ///
    /// # Arguments
    /// * `config` - Verification options (which checks to run, max errors, etc.)
    ///
    /// # Returns
    /// A `VerifyResult` with any structural errors and the count of records verified.
    ///
    /// # Errors
    /// Returns an error if the database is closed.
    pub fn verify(
        &self,
        config: &noxu_engine::VerifyConfig,
    ) -> Result<noxu_engine::VerifyResult> {
        self.check_open()?;
        let guard = self.db_impl.read();
        Ok(noxu_engine::verify_database_impl(&guard, config))
    }

    /// Creates a join cursor that returns records matching all secondary-key
    /// constraints expressed by the pre-positioned `cursors`.
    ///
    /// Mirrors `Database.join(SecondaryCursor[], JoinConfig)`.
    ///
    /// Each cursor in `cursors` must already be positioned at the desired
    /// secondary key value (e.g. via `SecondaryCursor::get_search_key`).
    /// The join algorithm iterates through all candidate primary keys from
    /// `cursors[0]` and probes `cursors[1..n]` to confirm each candidate
    /// also appears in their secondary keys.  Candidates that pass all
    /// probes are returned by [`JoinCursor::get_next`].
    ///
    /// Unless `config.no_sort` is `true`, the cursor array is re-ordered by
    /// ascending duplicate-count estimate before the join starts, matching
    /// JE's optimisation for minimum candidate-set size.
    ///
    /// The returned `JoinCursor` owns the `cursors` for its lifetime.
    ///
    /// # Errors
    /// Returns an error if this database handle is closed.
    pub fn join<'db>(
        &'db self,
        cursors: Vec<SecondaryCursor<'db>>,
        config: Option<JoinConfig>,
    ) -> Result<JoinCursor<'db>> {
        self.check_open()?;
        JoinCursor::new(self, cursors, config)
    }

    /// Checks if the database is open, returns an error if not.
    ///
    /// X-13: also checks the environment validity flags so that reads and
    /// writes return `EnvironmentFailure` after an fsync error or explicit
    /// `EnvironmentImpl::invalidate()` call rather than silently succeeding
    /// on stale BIN data.
    /// TXN-6 (JE invariant): a transactional handle must not be used against a
    /// non-transactional database.  JE `LockerFactory.getWritableLocker`/
    /// `getReadableLocker` throw `IllegalArgumentException` on EVERY operation
    /// (not just cursor-open) when a `Transaction` is supplied to a non-txnal DB.
    /// Shared by get/put/delete/get_with_options/put_with_options/open_cursor.
    fn reject_txn_on_non_txnal_db(&self, has_txn: bool) -> Result<()> {
        if has_txn && !self.config.transactional {
            return Err(NoxuError::IllegalArgument(
                "a transaction cannot be used with a \
                 non-transactional database"
                    .to_string(),
            ));
        }
        Ok(())
    }

    fn check_open(&self) -> Result<()> {
        // Check environment validity first — explicit invalidation.
        if self.env_invalid.load(Ordering::Acquire) {
            return Err(NoxuError::environment_with_reason(
                crate::error::EnvironmentFailureReason::UnexpectedStateFatal,
                "environment has been invalidated".to_string(),
            ));
        }
        // Check I/O failure (C-2 / fsync-gate).
        if self
            .log_manager
            .as_ref()
            .is_some_and(|lm| lm.io_invalid.load(Ordering::Acquire))
        {
            return Err(NoxuError::environment_with_reason(
                crate::error::EnvironmentFailureReason::LogWrite,
                "I/O failure: environment invalidated by fsync error"
                    .to_string(),
            ));
        }
        if !self.open.load(Ordering::Acquire) {
            return Err(NoxuError::DatabaseClosed);
        }
        Ok(())
    }

    /// Public-ish accessor for the cached log manager, used by
    /// [`crate::disk_ordered_cursor::open_disk_ordered_cursor_multi`].
    /// Returns `None` for non-WAL environments.
    pub(crate) fn cached_log_manager(
        &self,
    ) -> Option<&std::sync::Arc<noxu_log::LogManager>> {
        self.log_manager.as_ref()
    }

    /// Cached cleaner `FileProtector` for this database's environment, used by
    /// the disk-ordered-cursor producer to protect the files it scans from
    /// cleaner deletion (CLN-7).  `None` when the environment has no cleaner.
    pub(crate) fn cached_file_protector(
        &self,
    ) -> Option<std::sync::Arc<noxu_cleaner::FileProtector>> {
        self.file_protector.clone()
    }

    /// Public-ish accessor used by the disk-ordered-cursor helper to
    /// validate that the database is still open before scanning.
    pub(crate) fn check_open_for_doc(&self) -> Result<()> {
        self.check_open()
    }

    /// Returns this database's `DatabaseId` for use by the disk-ordered
    /// cursor producer.
    pub(crate) fn database_id_for_doc(&self) -> noxu_dbi::DatabaseId {
        noxu_dbi::DatabaseId::new(self.id as i64)
    }

    /// The env's `DOS_PRODUCER_QUEUE_TIMEOUT` (ms), read from the owning
    /// `EnvironmentImpl` at cursor-open time (not a hot path).  Passed to the
    /// disk-ordered-cursor producer so a lagging consumer fails the scan
    /// instead of hanging (JE `DOS_PRODUCER_QUEUE_TIMEOUT`).
    pub(crate) fn dos_producer_queue_timeout_ms(&self) -> u64 {
        self.env_impl.lock().get_dos_producer_queue_timeout_ms()
    }

    /// Checks if the database is writable, returns an error if not.
    fn check_writable(&self) -> Result<()> {
        if self.config.read_only {
            return Err(NoxuError::ReadOnly);
        }
        Ok(())
    }

    /// Unify the empty-key contract
    /// across `get` / `put` / `put_no_overwrite` / `put_with_options`
    /// / `delete`.  Returns the key bytes if the entry has data set
    /// (even if zero-length); rejects `None`-data keys with a typed
    /// `IllegalArgument` so the previous put-vs-get asymmetry can no
    /// longer black-hole records under a `None` key.
    #[allow(dead_code)] // documented empty-key contract helper, not yet wired
    fn require_key_bytes<'a>(
        key: &'a DatabaseEntry,
        op: &'static str,
    ) -> Result<&'a [u8]> {
        match key.data_opt() {
            Some(k) => Ok(k),
            None => Err(NoxuError::IllegalArgument(format!(
                "{op}: key DatabaseEntry has no data; \
                 use DatabaseEntry::from_bytes(...) or set_data(...) \
                 (Some(&[]) for an explicit empty key)",
            ))),
        }
    }
}

impl Drop for Database {
    fn drop(&mut self) {
        // Best effort close on drop
        let _ = self.close();
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::environment::Environment;
    use crate::environment_config::EnvironmentConfig;
    use tempfile::TempDir;

    fn temp_env_and_db() -> (TempDir, Environment, Database) {
        let temp_dir = TempDir::new().unwrap();
        let env_config = EnvironmentConfig::new(temp_dir.path().to_path_buf())
            .with_allow_create(true)
            .with_transactional(true);
        let env = Environment::open(env_config).unwrap();

        let db_config = DatabaseConfig::new()
            .with_allow_create(true)
            .with_transactional(true);
        let db = env.open_database(None, "testdb", &db_config).unwrap();

        (temp_dir, env, db)
    }

    #[test]
    fn test_database_name() {
        let (_temp_dir, _env, db) = temp_env_and_db();
        assert_eq!(db.name(), "testdb");
    }

    #[test]
    fn test_put_and_get() {
        let (_temp_dir, _env, db) = temp_env_and_db();

        let key = DatabaseEntry::from_bytes(b"key1");
        let value = DatabaseEntry::from_bytes(b"value1");

        db.put(&key, &value).unwrap();

        let mut retrieved = DatabaseEntry::new();
        let result = db.get_into(None, &key, &mut retrieved).unwrap();
        assert!(result);
        assert_eq!(retrieved.data_opt().unwrap(), b"value1");
    }

    #[test]
    fn test_get_nonexistent() {
        let (_temp_dir, _env, db) = temp_env_and_db();

        let key = DatabaseEntry::from_bytes(b"nonexistent");
        let mut data = DatabaseEntry::new();

        let result = db.get_into(None, &key, &mut data).unwrap();
        assert!(!result);
    }

    /// ("partial-put length
    /// mismatch silent truncation"): a partial put whose `data` slice
    /// differs in length from the configured partial-length must be
    /// rejected with a typed error instead of silently truncating or
    /// padding the splice.
    #[test]
    fn test_partial_put_length_mismatch_rejected() {
        let (_temp_dir, _env, db) = temp_env_and_db();

        let key = DatabaseEntry::from_bytes(b"k");
        db.put(&key, DatabaseEntry::from_bytes(b"hello world")).unwrap();

        // Partial offset=6, partial_length=5 ("world"), but only 3 bytes
        // supplied.  Used to silently truncate; now rejected.
        let mut patch = DatabaseEntry::from_bytes(b"abc");
        patch.set_partial(6, 5, true);
        let err = db.put_partial(None, &key, &patch).unwrap_err();
        assert!(
            matches!(err, NoxuError::IllegalArgument(_)),
            "expected IllegalArgument, got {err:?}"
        );
        assert!(
            err.to_string().contains("partial"),
            "expected partial-related message, got {}",
            err
        );

        // The on-disk record is unchanged because the call returned
        // before any write.
        let mut buf = DatabaseEntry::new();
        let status = db.get_into(None, &key, &mut buf).unwrap();
        assert!(status);
        assert_eq!(buf.data_opt().unwrap(), b"hello world");
    }

    /// Companion: when data.len() == partial_length the partial put
    /// patches the slice in place and other bytes are preserved.
    #[test]
    fn test_partial_put_exact_length_patches_in_place() {
        let (_temp_dir, _env, db) = temp_env_and_db();

        let key = DatabaseEntry::from_bytes(b"k");
        db.put(&key, DatabaseEntry::from_bytes(b"hello world")).unwrap();

        let mut patch = DatabaseEntry::from_bytes(b"WORLD");
        patch.set_partial(6, 5, true);
        db.put_partial(None, &key, &patch).unwrap();

        let mut buf = DatabaseEntry::new();
        db.get_into(None, &key, &mut buf).unwrap();
        assert_eq!(buf.data_opt().unwrap(), b"hello WORLD");
    }

    #[test]
    fn test_put_updates_existing() {
        let (_temp_dir, _env, db) = temp_env_and_db();

        let key = DatabaseEntry::from_bytes(b"key1");
        let value1 = DatabaseEntry::from_bytes(b"value1");
        let value2 = DatabaseEntry::from_bytes(b"value2");

        db.put(&key, &value1).unwrap();
        db.put(&key, &value2).unwrap();

        let mut retrieved = DatabaseEntry::new();
        db.get_into(None, &key, &mut retrieved).unwrap();
        assert_eq!(retrieved.data_opt().unwrap(), b"value2");
    }

    #[test]
    fn test_put_no_overwrite_success() {
        let (_temp_dir, _env, db) = temp_env_and_db();

        let key = DatabaseEntry::from_bytes(b"key1");
        let value = DatabaseEntry::from_bytes(b"value1");

        let result = db.put_no_overwrite(&key, &value).unwrap();
        assert!(result);
    }

    #[test]
    fn test_put_no_overwrite_key_exists() {
        let (_temp_dir, _env, db) = temp_env_and_db();

        let key = DatabaseEntry::from_bytes(b"key1");
        let value1 = DatabaseEntry::from_bytes(b"value1");
        let value2 = DatabaseEntry::from_bytes(b"value2");

        db.put(&key, &value1).unwrap();
        let result = db.put_no_overwrite(&key, &value2).unwrap();
        assert!(!result);

        // Verify original value is unchanged
        let mut retrieved = DatabaseEntry::new();
        db.get_into(None, &key, &mut retrieved).unwrap();
        assert_eq!(retrieved.data_opt().unwrap(), b"value1");
    }

    #[test]
    fn test_delete() {
        let (_temp_dir, _env, db) = temp_env_and_db();

        let key = DatabaseEntry::from_bytes(b"key1");
        let value = DatabaseEntry::from_bytes(b"value1");

        db.put(&key, &value).unwrap();
        let result = db.delete(&key).unwrap();
        assert!(result);

        let mut retrieved = DatabaseEntry::new();
        let result = db.get_into(None, &key, &mut retrieved).unwrap();
        assert!(!result);
    }

    #[test]
    fn test_delete_nonexistent() {
        let (_temp_dir, _env, db) = temp_env_and_db();

        let key = DatabaseEntry::from_bytes(b"nonexistent");
        let result = db.delete(&key).unwrap();
        assert!(!result);
    }

    #[test]
    fn test_count() {
        let (_temp_dir, _env, db) = temp_env_and_db();

        assert_eq!(db.count().unwrap(), 0);

        let key1 = DatabaseEntry::from_bytes(b"key1");
        let value1 = DatabaseEntry::from_bytes(b"value1");
        db.put(&key1, &value1).unwrap();
        assert_eq!(db.count().unwrap(), 1);

        let key2 = DatabaseEntry::from_bytes(b"key2");
        let value2 = DatabaseEntry::from_bytes(b"value2");
        db.put(&key2, &value2).unwrap();
        assert_eq!(db.count().unwrap(), 2);

        db.delete(&key1).unwrap();
        assert_eq!(db.count().unwrap(), 1);
    }

    #[test]
    fn test_close() {
        let (_temp_dir, _env, db) = temp_env_and_db();
        assert!(db.is_valid());
        db.close().unwrap();
        assert!(!db.is_valid());
    }

    #[test]
    fn test_close_twice_fails() {
        let (_temp_dir, _env, db) = temp_env_and_db();
        db.close().unwrap();
        let result = db.close();
        assert!(result.is_err());
    }

    #[test]
    fn test_operations_on_closed_database_fail() {
        let (_temp_dir, _env, db) = temp_env_and_db();
        db.close().unwrap();

        let key = DatabaseEntry::from_bytes(b"key1");
        let value = DatabaseEntry::from_bytes(b"value1");
        let mut data = DatabaseEntry::new();

        assert!(db.get_into(None, &key, &mut data).is_err());
        assert!(db.put(&key, &value).is_err());
        assert!(db.put_no_overwrite(&key, &value).is_err());
        assert!(db.delete(&key).is_err());
        assert!(db.count().is_err());
        assert!(db.open_cursor(None).is_err());
    }

    #[test]
    fn test_state() {
        let (_temp_dir, _env, db) = temp_env_and_db();
        assert_eq!(db.state(), DbState::Open);
        db.close().unwrap();
        assert_eq!(db.state(), DbState::Closed);
    }

    #[test]
    fn test_read_only_database() {
        let temp_dir = TempDir::new().unwrap();
        let env_config = EnvironmentConfig::new(temp_dir.path().to_path_buf())
            .with_allow_create(true);
        let env = Environment::open(env_config).unwrap();

        let db_config = DatabaseConfig::new()
            .with_allow_create(true)
            .with_transactional(true)
            .with_read_only(true);
        let db = env.open_database(None, "readonly_db", &db_config).unwrap();

        let key = DatabaseEntry::from_bytes(b"key1");
        let value = DatabaseEntry::from_bytes(b"value1");

        // Write operations should fail
        assert!(db.put(&key, &value).is_err());
        assert!(db.put_no_overwrite(&key, &value).is_err());
        assert!(db.delete(&key).is_err());
    }

    #[test]
    fn test_multiple_databases() {
        let temp_dir = TempDir::new().unwrap();
        let env_config = EnvironmentConfig::new(temp_dir.path().to_path_buf())
            .with_allow_create(true);
        let env = Environment::open(env_config).unwrap();

        let db_config = DatabaseConfig::new()
            .with_allow_create(true)
            .with_transactional(true);
        let db1 = env.open_database(None, "db1", &db_config).unwrap();
        let db2 = env.open_database(None, "db2", &db_config).unwrap();

        let key = DatabaseEntry::from_bytes(b"key1");
        let value1 = DatabaseEntry::from_bytes(b"value1");
        let value2 = DatabaseEntry::from_bytes(b"value2");

        db1.put(&key, &value1).unwrap();
        db2.put(&key, &value2).unwrap();

        let mut retrieved1 = DatabaseEntry::new();
        let mut retrieved2 = DatabaseEntry::new();

        db1.get_into(None, &key, &mut retrieved1).unwrap();
        db2.get_into(None, &key, &mut retrieved2).unwrap();

        assert_eq!(retrieved1.data_opt().unwrap(), b"value1");
        assert_eq!(retrieved2.data_opt().unwrap(), b"value2");
    }

    #[test]
    fn test_empty_keys_and_values() {
        let (_temp_dir, _env, db) = temp_env_and_db();

        let empty_key = DatabaseEntry::from_bytes(b"");
        let empty_value = DatabaseEntry::from_bytes(b"");

        db.put(&empty_key, &empty_value).unwrap();

        let mut retrieved = DatabaseEntry::new();
        let result = db.get_into(None, &empty_key, &mut retrieved).unwrap();
        assert!(result);
        assert_eq!(retrieved.data_opt().unwrap(), b"");
    }

    #[test]
    fn test_large_keys_and_values() {
        let (_temp_dir, _env, db) = temp_env_and_db();

        let large_key = DatabaseEntry::from_bytes(&vec![b'k'; 1000]);
        let large_value = DatabaseEntry::from_bytes(&vec![b'v'; 10000]);

        db.put(&large_key, &large_value).unwrap();

        let mut retrieved = DatabaseEntry::new();
        db.get_into(None, &large_key, &mut retrieved).unwrap();
        assert_eq!(retrieved.data_opt().unwrap().len(), 10000);
        assert!(retrieved.data_opt().unwrap().iter().all(|&b| b == b'v'));
    }

    #[test]
    fn test_binary_keys_and_values() {
        let (_temp_dir, _env, db) = temp_env_and_db();

        let binary_key = DatabaseEntry::from_bytes(&[0u8, 1, 2, 255, 254, 253]);
        let binary_value = DatabaseEntry::from_bytes(&[255u8, 0, 128, 64, 32]);

        db.put(&binary_key, &binary_value).unwrap();

        let mut retrieved = DatabaseEntry::new();
        db.get_into(None, &binary_key, &mut retrieved).unwrap();
        assert_eq!(retrieved.data_opt().unwrap(), &[255u8, 0, 128, 64, 32]);
    }

    #[test]
    fn test_scan_all_kv_empty() {
        let (_temp_dir, _env, db) = temp_env_and_db();
        let kv = db.scan_all_kv().unwrap();
        assert!(kv.is_empty());
    }

    #[test]
    fn test_scan_all_kv_returns_records() {
        let (_temp_dir, _env, db) = temp_env_and_db();
        db.put(
            DatabaseEntry::from_vec(vec![1]),
            DatabaseEntry::from_vec(vec![10]),
        )
        .unwrap();
        db.put(
            DatabaseEntry::from_vec(vec![2]),
            DatabaseEntry::from_vec(vec![20]),
        )
        .unwrap();
        let kv = db.scan_all_kv().unwrap();
        assert_eq!(kv.len(), 2);
    }

    #[test]
    fn test_scan_all_kv_then_delete() {
        let (_temp_dir, _env, db) = temp_env_and_db();
        db.put(
            DatabaseEntry::from_vec(vec![1]),
            DatabaseEntry::from_vec(vec![10]),
        )
        .unwrap();
        db.put(
            DatabaseEntry::from_vec(vec![2]),
            DatabaseEntry::from_vec(vec![20]),
        )
        .unwrap();

        let kv = db.scan_all_kv().unwrap();
        assert_eq!(kv.len(), 2);

        for (k, _v) in &kv {
            let status = db.delete(DatabaseEntry::from_vec(k.clone())).unwrap();
            assert!(status, "delete failed for key {:?}", k);
        }

        let count = db.count().unwrap();
        assert_eq!(count, 0, "expected 0 records after deletes, got {}", count);
    }

    #[test]
    fn test_scan_all_kv_then_delete_u64_be_keys() {
        // Simulate the exact pattern used in EntityStore::evolve: big-endian u64 keys.
        let (_temp_dir, _env, db) = temp_env_and_db();
        for id in [1u64, 2u64] {
            let key_bytes = id.to_be_bytes().to_vec();
            let val_bytes = format!("user{}", id).into_bytes();
            db.put(
                DatabaseEntry::from_vec(key_bytes),
                DatabaseEntry::from_vec(val_bytes),
            )
            .unwrap();
        }
        assert_eq!(db.count().unwrap(), 2);

        let records = db.scan_all_kv().unwrap();
        assert_eq!(records.len(), 2);

        for (k, _v) in records {
            let status = db.delete(DatabaseEntry::from_vec(k.clone())).unwrap();
            assert!(status, "delete failed for u64 key {:?}", k);
        }
        assert_eq!(db.count().unwrap(), 0);
    }

    // ========================================================================
    // Additional branch-coverage tests
    // ========================================================================

    /// get() with a None-data DatabaseEntry returns NotFound.
    #[test]
    fn test_get_with_none_key_data_returns_not_found() {
        let (_temp_dir, _env, db) = temp_env_and_db();
        let key_none = DatabaseEntry::new(); // no data set
        let mut data = DatabaseEntry::new();

        let result = db.get_into(None, &key_none, &mut data).unwrap();
        assert!(!result);
    }

    /// delete() with a None-data DatabaseEntry returns NotFound.
    #[test]
    fn test_delete_with_none_key_data_returns_not_found() {
        let (_temp_dir, _env, db) = temp_env_and_db();
        let key_none = DatabaseEntry::new();

        let result = db.delete(&key_none).unwrap();
        assert!(!result);
    }

    /// open_cursor() with a CursorConfig that has read_uncommitted=true makes
    /// the cursor read-only.
    #[test]
    fn test_open_cursor_read_uncommitted_config_makes_read_only() {
        use crate::cursor_config::CursorConfig;
        let (_temp_dir, _env, db) = temp_env_and_db();

        let config = CursorConfig::new().with_read_uncommitted(true);
        let cursor = db.open_cursor(Some(&config)).unwrap();
        assert!(cursor.is_read_only());
    }

    /// open_cursor() with no config and a non-read-only database produces a
    /// writable cursor.
    #[test]
    fn test_open_cursor_no_config_writable_db_is_writable() {
        let (_temp_dir, _env, db) = temp_env_and_db();
        let cursor = db.open_cursor(None).unwrap();
        assert!(!cursor.is_read_only());
    }

    /// scan_all_kv() on a closed database returns an error.
    #[test]
    fn test_scan_all_kv_on_closed_database_fails() {
        let (_temp_dir, _env, db) = temp_env_and_db();
        db.close().unwrap();
        let result = db.scan_all_kv();
        assert!(result.is_err());
    }

    /// put_no_overwrite() on a read-only database returns an error.
    #[test]
    fn test_put_no_overwrite_on_read_only_database_fails() {
        let temp_dir = TempDir::new().unwrap();
        let env_config = EnvironmentConfig::new(temp_dir.path().to_path_buf())
            .with_allow_create(true);
        let env = Environment::open(env_config).unwrap();

        let db_config = DatabaseConfig::new()
            .with_allow_create(true)
            .with_transactional(true)
            .with_read_only(true);
        let db = env.open_database(None, "ro_db", &db_config).unwrap();

        let key = DatabaseEntry::from_bytes(b"k");
        let val = DatabaseEntry::from_bytes(b"v");
        let result = db.put_no_overwrite(&key, &val);
        assert!(result.is_err());
    }

    // =====================================================================
    // cursor-failure map_err coverage: use the test hook in noxu-dbi to
    // force cursor operations to return Err, exercising the map_err closures
    // in Database::get / put / put_no_overwrite / delete / count / scan_all_kv.
    // =====================================================================

    /// Covers the map_err closure on `cursor.search(...)` inside `get()`.
    #[test]
    fn test_get_search_map_err_via_hook() {
        let (_tmp, _env, db) = temp_env_and_db();
        noxu_dbi::set_cursor_fail_after(1); // fail on the 1st check_state (search)
        let key = DatabaseEntry::from_bytes(b"any");
        let mut data = DatabaseEntry::new();
        let result = db.get_into(None, &key, &mut data);
        noxu_dbi::clear_cursor_fail_flag();
        assert!(result.is_err());
    }

    /// Covers the map_err closure on `cursor.get_current()` inside `get()`.
    #[test]
    fn test_get_get_current_map_err_via_hook() {
        let (_tmp, _env, db) = temp_env_and_db();
        // Insert a key so search can succeed.
        db.put(
            DatabaseEntry::from_bytes(b"k"),
            DatabaseEntry::from_bytes(b"v"),
        )
        .unwrap();
        // fail on the 2nd check (check_initialized inside get_current).
        noxu_dbi::set_cursor_fail_after(2);
        let key = DatabaseEntry::from_bytes(b"k");
        let mut data = DatabaseEntry::new();
        let result = db.get_into(None, &key, &mut data);
        noxu_dbi::clear_cursor_fail_flag();
        assert!(result.is_err());
    }

    /// Covers the map_err closure on `cursor.put(...)` inside `put()`.
    #[test]
    fn test_put_map_err_via_hook() {
        let (_tmp, _env, db) = temp_env_and_db();
        noxu_dbi::set_cursor_fail_after(1);
        let key = DatabaseEntry::from_bytes(b"k");
        let val = DatabaseEntry::from_bytes(b"v");
        let result = db.put(&key, &val);
        noxu_dbi::clear_cursor_fail_flag();
        assert!(result.is_err());
    }

    /// Covers the map_err closure on `cursor.put(...)` inside `put_no_overwrite()`.
    #[test]
    fn test_put_no_overwrite_map_err_via_hook() {
        let (_tmp, _env, db) = temp_env_and_db();
        noxu_dbi::set_cursor_fail_after(1);
        let key = DatabaseEntry::from_bytes(b"k");
        let val = DatabaseEntry::from_bytes(b"v");
        let result = db.put_no_overwrite(&key, &val);
        noxu_dbi::clear_cursor_fail_flag();
        assert!(result.is_err());
    }

    /// Covers the map_err closure on `cursor.search(...)` inside `delete()`.
    #[test]
    fn test_delete_search_map_err_via_hook() {
        let (_tmp, _env, db) = temp_env_and_db();
        noxu_dbi::set_cursor_fail_after(1);
        let key = DatabaseEntry::from_bytes(b"k");
        let result = db.delete(&key);
        noxu_dbi::clear_cursor_fail_flag();
        assert!(result.is_err());
    }

    /// Covers the map_err closure on `cursor.delete()` inside `delete()`.
    #[test]
    fn test_delete_delete_map_err_via_hook() {
        let (_tmp, _env, db) = temp_env_and_db();
        db.put(
            DatabaseEntry::from_bytes(b"k"),
            DatabaseEntry::from_bytes(b"v"),
        )
        .unwrap();
        // fail on the 2nd check_state (the delete() call, after search succeeds).
        noxu_dbi::set_cursor_fail_after(2);
        let key = DatabaseEntry::from_bytes(b"k");
        let result = db.delete(&key);
        noxu_dbi::clear_cursor_fail_flag();
        assert!(result.is_err());
    }

    /// count() uses the O(1) AtomicU64 counter; cursor-fail hooks do not affect it.
    /// Verify the counter is correct across insert/update/delete.
    #[test]
    fn test_count_atomic_counter_insert_update_delete() {
        let (_tmp, _env, db) = temp_env_and_db();

        // Empty database starts at 0.
        assert_eq!(db.count().unwrap(), 0);

        // Insert three distinct keys.
        db.put(
            DatabaseEntry::from_bytes(b"a"),
            DatabaseEntry::from_bytes(b"1"),
        )
        .unwrap();
        db.put(
            DatabaseEntry::from_bytes(b"b"),
            DatabaseEntry::from_bytes(b"2"),
        )
        .unwrap();
        db.put(
            DatabaseEntry::from_bytes(b"c"),
            DatabaseEntry::from_bytes(b"3"),
        )
        .unwrap();
        assert_eq!(db.count().unwrap(), 3);

        // Overwrite an existing key — count must NOT change.
        db.put(
            DatabaseEntry::from_bytes(b"a"),
            DatabaseEntry::from_bytes(b"updated"),
        )
        .unwrap();
        assert_eq!(db.count().unwrap(), 3);

        // Delete one key — count decrements.
        db.delete(DatabaseEntry::from_bytes(b"b")).unwrap();
        assert_eq!(db.count().unwrap(), 2);
    }

    /// count() is O(1): verify it still works even when the cursor fail-hook
    /// is active (the hook only affects cursor operations, not the atomic read).
    #[test]
    fn test_count_unaffected_by_cursor_fail_hook() {
        let (_tmp, _env, db) = temp_env_and_db();
        db.put(
            DatabaseEntry::from_bytes(b"k"),
            DatabaseEntry::from_bytes(b"v"),
        )
        .unwrap();
        noxu_dbi::set_cursor_fail_after(1);
        // count() must succeed (no cursor used).
        let result = db.count();
        noxu_dbi::clear_cursor_fail_flag();
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), 1);
    }

    /// Covers the map_err closure on `cursor.get_first()` inside `scan_all_kv()`.
    #[test]
    fn test_scan_all_kv_get_first_map_err_via_hook() {
        let (_tmp, _env, db) = temp_env_and_db();
        noxu_dbi::set_cursor_fail_after(1);
        let result = db.scan_all_kv();
        noxu_dbi::clear_cursor_fail_flag();
        assert!(result.is_err());
    }

    /// Covers the map_err closure on `cursor.get_current()` inside `scan_all_kv()`.
    #[test]
    fn test_scan_all_kv_get_current_map_err_via_hook() {
        let (_tmp, _env, db) = temp_env_and_db();
        db.put(
            DatabaseEntry::from_bytes(b"k"),
            DatabaseEntry::from_bytes(b"v"),
        )
        .unwrap();
        // fail on the 2nd check (check_initialized inside get_current, after get_first succeeds).
        noxu_dbi::set_cursor_fail_after(2);
        let result = db.scan_all_kv();
        noxu_dbi::clear_cursor_fail_flag();
        assert!(result.is_err());
    }

    /// Covers the map_err closure on `cursor.retrieve_next(...)` inside `scan_all_kv()`.
    #[test]
    fn test_scan_all_kv_retrieve_next_map_err_via_hook() {
        let (_tmp, _env, db) = temp_env_and_db();
        db.put(
            DatabaseEntry::from_bytes(b"k"),
            DatabaseEntry::from_bytes(b"v"),
        )
        .unwrap();
        // fail on the 3rd check (retrieve_next, after get_first and get_current succeed).
        noxu_dbi::set_cursor_fail_after(3);
        let result = db.scan_all_kv();
        noxu_dbi::clear_cursor_fail_flag();
        assert!(result.is_err());
    }

    #[test]
    fn test_sync_on_open_database_succeeds() {
        let (_tmp, _env, db) = temp_env_and_db();
        db.put(
            DatabaseEntry::from_bytes(b"key"),
            DatabaseEntry::from_bytes(b"val"),
        )
        .unwrap();
        assert!(db.sync().is_ok());
    }

    #[test]
    fn test_sync_on_closed_database_fails() {
        let (_tmp, _env, db) = temp_env_and_db();
        db.close().unwrap();
        assert!(db.sync().is_err());
    }

    // ── verify ─────────────────────────────────────────────────────────────

    #[test]
    fn test_verify_empty_database_passes() {
        use noxu_engine::VerifyConfig;
        let (_tmp, _env, db) = temp_env_and_db();
        let config = VerifyConfig::default();
        let result = db.verify(&config).unwrap();
        assert!(result.passed, "empty db should pass: {:?}", result.errors);
    }

    #[test]
    fn test_verify_populated_database_passes() {
        use noxu_engine::VerifyConfig;
        let (_tmp, _env, db) = temp_env_and_db();
        for i in 0u32..20 {
            let k = DatabaseEntry::from_bytes(&i.to_be_bytes());
            let v = DatabaseEntry::from_bytes(&(i * 2).to_be_bytes());
            db.put(&k, &v).unwrap();
        }
        let config = VerifyConfig::default();
        let result = db.verify(&config).unwrap();
        assert!(result.passed, "populated db should pass: {:?}", result.errors);
        assert!(result.records_verified > 0);
    }

    #[test]
    fn test_verify_closed_database_fails() {
        use noxu_engine::VerifyConfig;
        let (_tmp, _env, db) = temp_env_and_db();
        db.close().unwrap();
        let config = VerifyConfig::default();
        assert!(db.verify(&config).is_err());
    }

    // ── get_with_options / put_with_options ────────────────────────────────

    #[test]
    fn test_get_with_options_default_reads_written_record() {
        use crate::read_options::ReadOptions;
        let (_tmp, _env, db) = temp_env_and_db();
        let key = DatabaseEntry::from_bytes(b"ropt_key");
        let val = DatabaseEntry::from_bytes(b"ropt_val");
        db.put(&key, &val).unwrap();

        let opts = ReadOptions::new();
        let out = db.get_with_options(None, &key, &opts).unwrap();
        assert!(out.is_some());
        assert_eq!(out.unwrap().as_ref(), b"ropt_val");
    }

    #[test]
    fn test_get_with_options_read_uncommitted_sees_written_record() {
        use crate::read_options::ReadOptions;
        let (_tmp, _env, db) = temp_env_and_db();
        let key = DatabaseEntry::from_bytes(b"ru_key");
        let val = DatabaseEntry::from_bytes(b"ru_val");
        db.put(&key, &val).unwrap();

        let opts = ReadOptions::read_uncommitted();
        let out = db.get_with_options(None, &key, &opts).unwrap();
        assert!(out.is_some());
        assert_eq!(out.unwrap().as_ref(), b"ru_val");
    }

    #[test]
    fn test_get_with_options_not_found() {
        use crate::read_options::ReadOptions;
        let (_tmp, _env, db) = temp_env_and_db();
        let key = DatabaseEntry::from_bytes(b"missing");
        let opts = ReadOptions::new();
        let out = db.get_with_options(None, &key, &opts).unwrap();
        assert!(out.is_none());
    }

    #[test]
    fn test_put_with_options_no_ttl_behaves_like_put() {
        use crate::write_options::WriteOptions;
        let (_tmp, _env, db) = temp_env_and_db();
        let key = DatabaseEntry::from_bytes(b"wopt_key");
        let val = DatabaseEntry::from_bytes(b"wopt_val");
        let opts = WriteOptions::new();
        db.put_with_options(None, &key, &val, &opts).unwrap();

        let mut out = DatabaseEntry::new();
        db.get_into(None, &key, &mut out).unwrap();
        assert_eq!(out.data_opt().unwrap(), b"wopt_val");
    }

    #[test]
    fn test_put_with_options_with_ttl_stores_record() {
        use crate::write_options::WriteOptions;
        let (_tmp, _env, db) = temp_env_and_db();
        let key = DatabaseEntry::from_bytes(b"ttl_key");
        let val = DatabaseEntry::from_bytes(b"ttl_val");
        // TTL of 1 hour — the record is not yet expired so it should be readable
        let opts = WriteOptions::with_expiration(1);
        db.put_with_options(None, &key, &val, &opts).unwrap();

        let mut out = DatabaseEntry::new();
        let read_status = db.get_into(None, &key, &mut out).unwrap();
        assert!(read_status);
        assert_eq!(out.data_opt().unwrap(), b"ttl_val");
    }

    #[test]
    fn test_put_with_options_closed_db_fails() {
        use crate::write_options::WriteOptions;
        let (_tmp, _env, db) = temp_env_and_db();
        db.close().unwrap();
        let key = DatabaseEntry::from_bytes(b"k");
        let val = DatabaseEntry::from_bytes(b"v");
        let opts = WriteOptions::new();
        assert!(db.put_with_options(None, &key, &val, &opts).is_err());
    }

    // ========================================================================
    // Audit database F11 — Wave 2C-4: reject None-data keys on writes.
    // ========================================================================

    // 7.0 NOTE: the three `*_with_none_key_returns_illegal_argument` tests
    // were removed in the 7.0 API reshape (review P1-3).  The write surface
    // now takes `key: impl AsRef<[u8]>`, so a key is *always* a byte slice;
    // the historical "None key" (a `DatabaseEntry` with no data set, distinct
    // from an empty `b""`) can no longer be expressed at the call site.
    // An empty key is accepted on writes — see
    // `test_put_with_explicit_empty_key_accepted`, which the reshape kept as
    // the canonical behaviour.  The removed tests asserted a None-vs-empty
    // distinction that the new signature intentionally eliminates.

    /// Explicit `Some(&[])` empty key is still accepted on writes.
    #[test]
    fn test_put_with_explicit_empty_key_accepted() {
        let (_tmp, _env, db) = temp_env_and_db();
        let empty_key = DatabaseEntry::from_bytes(b"");
        let val = DatabaseEntry::from_bytes(b"v");
        db.put(&empty_key, &val).unwrap();
    }

    // ── X-13: env-invalidity checks propagate through check_open ──────────────

    /// X-13: after the `io_invalid` flag is set, `db.get` must return
    /// `EnvironmentFailure` rather than silently reading stale BIN data.
    #[test]
    fn test_x13_io_invalid_blocks_db_get() {
        use std::sync::atomic::Ordering;
        let (_tmp, env, db) = temp_env_and_db();

        // Write a record so there is something to read.
        let key = DatabaseEntry::from_bytes(b"k");
        let val = DatabaseEntry::from_bytes(b"v");
        db.put(&key, &val).unwrap();

        // Flip io_invalid via the cached LogManager.
        let lm = db.log_manager.as_ref().expect("WAL env must have LogManager");
        lm.io_invalid.store(true, Ordering::Release);

        // db.get must now fail.
        let mut out = DatabaseEntry::new();
        let result = db.get_into(None, &key, &mut out);
        assert!(
            matches!(result, Err(NoxuError::EnvironmentFailure { .. })),
            "expected EnvironmentFailure, got {result:?}"
        );

        // db.put must also fail.
        let result2 = db.put(&key, &val);
        assert!(
            matches!(result2, Err(NoxuError::EnvironmentFailure { .. })),
            "expected EnvironmentFailure on put, got {result2:?}"
        );

        // Restore flag so env closes cleanly.
        lm.io_invalid.store(false, Ordering::Release);
        drop(env);
    }

    /// X-13: after `EnvironmentImpl::invalidate()`, cursor `get_first`
    /// must return `EnvironmentFailure`.
    #[test]
    fn test_x13_env_invalid_blocks_cursor_get() {
        use std::sync::atomic::Ordering;
        let (_tmp, env, db) = temp_env_and_db();

        // Insert a record.
        let key = DatabaseEntry::from_bytes(b"ck");
        let val = DatabaseEntry::from_bytes(b"cv");
        db.put(&key, &val).unwrap();

        // Open a cursor BEFORE invalidating.
        let mut cursor = db.open_cursor(None).unwrap();

        // Now directly flip the env_invalid flag.
        db.env_invalid.store(true, Ordering::Release);

        // The cursor's check_state should detect the flag.
        let mut key = DatabaseEntry::new();
        let mut out = DatabaseEntry::new();
        let result =
            cursor.get(&mut key, &mut out, crate::get::Get::First, None);
        assert!(
            matches!(result, Err(NoxuError::EnvironmentFailure { .. })),
            "expected EnvironmentFailure from cursor, got {result:?}"
        );

        // Restore so env drops cleanly.
        db.env_invalid.store(false, Ordering::Release);
        drop(env);
    }
}