redb 4.3.0

use crate::BackendError;
use crate::io;
use crate::transaction_tracker::{TransactionId, TransactionTracker};
#[cfg(feature = "experimental-multiprocess")]
use crate::transactions::AllocatorStateLatch;
#[cfg(feature = "experimental-multiprocess")]
use crate::tree_store::HeaderGuard;
#[cfg(not(redb_no_std))]
use crate::tree_store::ReadOnlyBackend;
#[cfg(feature = "experimental-multiprocess")]
use crate::tree_store::WriterLock;
use crate::tree_store::{
    AllocationPolicy, BtreeHeader, InternalTableDefinition, PAGE_SIZE, PageHint, PageNumber,
    PageResolver, ShrinkPolicy, TableTree, TableType, TransactionalMemory,
};
use crate::types::{Key, Value};
use crate::{
    CompactionError, DatabaseError, ReadOnlyTable, ReadableTable, StorageError, TableError,
};
use crate::{ReadTransaction, Result, WriteTransaction};
use alloc::boxed::Box;
use alloc::format;
use alloc::string::String;
use alloc::string::ToString;
use core::fmt::{Debug, Display, Formatter};

use alloc::sync::Arc;
use core::marker::PhantomData;
use core::ops::{Bound, Range};
#[cfg(not(redb_no_std))]
use std::fs::{File, OpenOptions};
#[cfg(not(redb_no_std))]
use std::path::Path;

use crate::error::TransactionError;
use crate::sealed::{Sealed, SealedInApi5};
use crate::transactions::{
    ALLOCATOR_STATE_TABLE_NAME, AllocatorStateKey, AllocatorStateTree, DATA_ALLOCATED_TABLE,
    DATA_FREED_TABLE, PageList, SYSTEM_FREED_TABLE, SystemTableDefinition,
    TransactionIdWithPagination,
};
#[cfg(not(redb_no_std))]
use crate::tree_store::file_backend::FileBackend;
#[cfg(feature = "logging")]
use log::{debug, warn};

#[allow(clippy::len_without_is_empty)]
/// Implements persistent storage for a database.
///
/// I/O failures are reported as [`io::Error`], which is [`std::io::Error`] whenever std is
/// available. Locking operations use [`BackendError`] to distinguish unsupported operations.
///
/// Locking is optional. Backends that implement it must override the lock methods. Locks belong to
/// a backend instance and must conflict with locks held by independently opened instances,
/// including those in the same process. All locks must be released by [`Self::close`].
/// redb never acquires overlapping locks on the same backend. Failed or refused acquisitions
/// must not acquire any locks or change existing locks.
///
/// Lock bounds refer to byte offsets and may extend beyond the current length of the storage.
/// An unbounded start means offset zero; an unbounded end includes all future growth. Bounds
/// must describe a nonempty range. Implementations may reject offsets or lengths that their
/// underlying locking API cannot represent with [`BackendError::Io`].
///
/// redb never queries backend-reserved ranges and only locks them during whole-storage fallback.
/// See the [design document](https://github.com/cberner/redb/blob/master/docs/design.md#lock-bytes).
///
/// Backends that support locking must support one of the following levels:
/// 1) All representable ranges requested by redb are supported. All concurrency modes will work.
/// 2) Only whole-storage locks (`Unbounded, Unbounded` or `Included(0), Unbounded`) are
///    supported. Other ranges return [`BackendError::Unsupported`]. Only the `ExclusiveWriter`
///    concurrency mode will work.
/// 3) All inputs return [`BackendError::Unsupported`]. Only the `ExclusiveWriter` concurrency
///    mode will work, and redb will log a warning on open if logging is enabled.
pub trait StorageBackend: 'static + Debug + Send + Sync {
    /// Gets the current length of the storage.
    fn len(&self) -> core::result::Result<u64, io::Error>;

    /// Reads the specified array of bytes from the storage.
    ///
    /// If `out.len()` + `offset` exceeds the length of the storage an appropriate `Error` must be returned.
    fn read(&self, offset: u64, out: &mut [u8]) -> core::result::Result<(), io::Error>;

    /// Sets the length of the storage.
    ///
    /// New positions in the storage must be initialized to zero.
    fn set_len(&self, len: u64) -> core::result::Result<(), io::Error>;

    /// Syncs all buffered data with the persistent storage.
    fn sync_data(&self) -> core::result::Result<(), io::Error>;

    /// Writes the specified array to the storage.
    fn write(&self, offset: u64, data: &[u8]) -> core::result::Result<(), io::Error>;

    /// Release any resources held by the backend
    ///
    /// Must release all locks acquired by one of the lock methods.
    ///
    /// Note: redb will not access the backend after calling this method and will call it exactly
    /// once: when the [`Database`] is dropped, or, if a [`WriteTransaction`] was live at that
    /// point, when that transaction completes, or if opening the database fails
    fn close(&self) -> core::result::Result<(), io::Error> {
        Ok(())
    }

    /// Attempts to acquire an exclusive byte-range lock.
    ///
    /// Returns `Ok(true)` on acquisition, `Ok(false)` on a conflicting lock, or an error.
    /// Defaults to [`BackendError::Unsupported`].
    fn try_lock_range(
        &self,
        _start: Bound<u64>,
        _end: Bound<u64>,
    ) -> core::result::Result<bool, BackendError> {
        Err(BackendError::Unsupported)
    }

    /// Attempts to acquire a shared byte-range lock.
    ///
    /// Returns `Ok(true)` on acquisition, `Ok(false)` on a conflicting lock, or an error.
    /// Defaults to [`BackendError::Unsupported`].
    fn try_lock_shared_range(
        &self,
        _start: Bound<u64>,
        _end: Bound<u64>,
    ) -> core::result::Result<bool, BackendError> {
        Err(BackendError::Unsupported)
    }

    /// Acquires an exclusive byte-range lock, waiting for conflicting locks to be released.
    ///
    /// Defaults to [`BackendError::Unsupported`].
    fn lock_range(
        &self,
        _start: Bound<u64>,
        _end: Bound<u64>,
    ) -> core::result::Result<(), BackendError> {
        Err(BackendError::Unsupported)
    }

    /// Acquires a shared byte-range lock, waiting for conflicting locks to be released.
    ///
    /// Defaults to [`BackendError::Unsupported`].
    fn lock_shared_range(
        &self,
        _start: Bound<u64>,
        _end: Bound<u64>,
    ) -> core::result::Result<(), BackendError> {
        Err(BackendError::Unsupported)
    }

    /// Releases the lock acquired over exactly this range.
    ///
    /// The range passed will exactly match a range successfully locked by one of the lock methods.
    ///
    /// Defaults to [`BackendError::Unsupported`].
    fn unlock_range(
        &self,
        _start: Bound<u64>,
        _end: Bound<u64>,
    ) -> core::result::Result<(), BackendError> {
        Err(BackendError::Unsupported)
    }

    /// Reports whether an exclusive lock over the range would conflict with a lock held elsewhere.
    ///
    /// The queried range will not overlap any lock currently held by this backend.
    /// Defaults to [`BackendError::Unsupported`].
    fn query_lock_range(
        &self,
        _start: Bound<u64>,
        _end: Bound<u64>,
    ) -> core::result::Result<bool, BackendError> {
        Err(BackendError::Unsupported)
    }
}

#[cfg_attr(redb_no_std, allow(dead_code))]
pub(crate) const FULL_RANGE: (Bound<u64>, Bound<u64>) = (Bound::Unbounded, Bound::Unbounded);

#[cfg_attr(not(any(windows, unix, target_os = "wasi")), allow(dead_code))]
const LOCK_BASE: u64 = 1 << 62;
/// Reserved for backend locking. Core only covers these bytes with a whole-storage fallback lock.
pub(crate) const BACKEND_LOCK_RANGE: Range<u64> = LOCK_BASE + 896..LOCK_BASE + 1024;

/// Held exclusively by the writing process in single-writer mode.
#[cfg(feature = "experimental-multiprocess")]
pub(crate) const WRITER_BYTE: u64 = LOCK_BASE;
/// Whether the database is open for a single writer (held exclusively by it) or for many (held
/// shared by each writing process while the database is open).
#[cfg_attr(not(any(windows, unix, target_os = "wasi")), allow(dead_code))]
pub(crate) const SHARED_WRITER_BYTE: u64 = LOCK_BASE + 1;
/// Held shared by every read-only multi-process handle while the database is open, so that a
/// exclusive-writer open conflicts with a live reader no matter what the reader is doing.
#[cfg(feature = "experimental-multiprocess")]
pub(crate) const SHARED_READER_BYTE: u64 = LOCK_BASE + 2;
/// Held shared by a read-only exclusive-writer handle as part of its whole-file lock. Such a
/// handle leaves `SHARED_WRITER_BYTE` free, so a multi-writer open, which takes only that byte,
/// must probe this one to find it.
#[cfg(feature = "experimental-multiprocess")]
pub(crate) const WHOLE_FILE_READER_BYTE: u64 = LOCK_BASE + 3;
/// Held shared by a writing process from the moment its open is complete -- past recovery and
/// the allocator load -- until it closes: the file is consistent, and the recovery flag, set from
/// here on, means only that a writer is live. `SHARED_WRITER_BYTE` is taken before recovery, so
/// it cannot say that; this byte can.
#[cfg(feature = "experimental-multiprocess")]
pub(crate) const CONSISTENT_BYTE: u64 = LOCK_BASE + 4;
/// Base of the "active transaction range": a handle reading transaction `t` holds `TXN_BASE + t`
/// shared for as long as it is reading it
#[cfg(feature = "experimental-multiprocess")]
pub(crate) const TXN_BASE: u64 = LOCK_BASE + 1024;

pub(crate) fn byte_range(offset: u64) -> (Bound<u64>, Bound<u64>) {
    (Bound::Included(offset), Bound::Included(offset))
}

pub trait TableHandle: Sealed {
    // Returns the name of the table
    fn name(&self) -> &str;
}

#[derive(Clone)]
pub struct UntypedTableHandle {
    name: String,
}

impl UntypedTableHandle {
    pub(crate) fn new(name: String) -> Self {
        Self { name }
    }
}

impl TableHandle for UntypedTableHandle {
    fn name(&self) -> &str {
        &self.name
    }
}

impl Sealed for UntypedTableHandle {}

pub trait MultimapTableHandle: Sealed {
    // Returns the name of the multimap table
    fn name(&self) -> &str;
}

#[derive(Clone)]
pub struct UntypedMultimapTableHandle {
    name: String,
}

impl UntypedMultimapTableHandle {
    pub(crate) fn new(name: String) -> Self {
        Self { name }
    }
}

impl MultimapTableHandle for UntypedMultimapTableHandle {
    fn name(&self) -> &str {
        &self.name
    }
}

impl Sealed for UntypedMultimapTableHandle {}

/// Defines the name and types of a table
///
/// A [`TableDefinition`] should be opened for use by calling [`ReadTransaction::open_table`] or [`WriteTransaction::open_table`]
///
/// Note that the lifetime of the `K` and `V` type parameters does not impact the lifetimes of the data
/// that is stored or retreived from the table
pub struct TableDefinition<'a, K: Key + 'static, V: Value + 'static> {
    name: &'a str,
    _key_type: PhantomData<K>,
    _value_type: PhantomData<V>,
}

impl<'a, K: Key + 'static, V: Value + 'static> TableDefinition<'a, K, V> {
    /// Construct a new table with given `name`
    ///
    /// # Panics
    ///
    /// Panics if `name` is empty. When `name` is a non-empty string literal
    /// this is checked at compile time, but callers that build the name at
    /// runtime are responsible for ensuring it is non-empty.
    pub const fn new(name: &'a str) -> Self {
        assert!(!name.is_empty());
        Self {
            name,
            _key_type: PhantomData,
            _value_type: PhantomData,
        }
    }
}

impl<K: Key + 'static, V: Value + 'static> TableHandle for TableDefinition<'_, K, V> {
    fn name(&self) -> &str {
        self.name
    }
}

impl<K: Key, V: Value> Sealed for TableDefinition<'_, K, V> {}

impl<K: Key + 'static, V: Value + 'static> Clone for TableDefinition<'_, K, V> {
    fn clone(&self) -> Self {
        *self
    }
}

impl<K: Key + 'static, V: Value + 'static> Copy for TableDefinition<'_, K, V> {}

impl<K: Key + 'static, V: Value + 'static> Display for TableDefinition<'_, K, V> {
    fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result {
        write!(
            f,
            "{}<{}, {}>",
            self.name,
            K::type_name().name(),
            V::type_name().name()
        )
    }
}

/// Defines the name and types of a multimap table
///
/// A [`MultimapTableDefinition`] should be opened for use by calling [`ReadTransaction::open_multimap_table`] or [`WriteTransaction::open_multimap_table`]
///
/// [Multimap tables](https://en.wikipedia.org/wiki/Multimap) may have multiple values associated with each key
///
/// Note that the lifetime of the `K` and `V` type parameters does not impact the lifetimes of the data
/// that is stored or retreived from the table
pub struct MultimapTableDefinition<'a, K: Key + 'static, V: Key + 'static> {
    name: &'a str,
    _key_type: PhantomData<K>,
    _value_type: PhantomData<V>,
}

impl<'a, K: Key + 'static, V: Key + 'static> MultimapTableDefinition<'a, K, V> {
    /// Construct a new multimap table with given `name`
    ///
    /// # Panics
    ///
    /// Panics if `name` is empty. When `name` is a non-empty string literal
    /// this is checked at compile time, but callers that build the name at
    /// runtime are responsible for ensuring it is non-empty.
    pub const fn new(name: &'a str) -> Self {
        assert!(!name.is_empty());
        Self {
            name,
            _key_type: PhantomData,
            _value_type: PhantomData,
        }
    }
}

impl<K: Key + 'static, V: Key + 'static> MultimapTableHandle for MultimapTableDefinition<'_, K, V> {
    fn name(&self) -> &str {
        self.name
    }
}

impl<K: Key, V: Key> Sealed for MultimapTableDefinition<'_, K, V> {}

impl<K: Key + 'static, V: Key + 'static> Clone for MultimapTableDefinition<'_, K, V> {
    fn clone(&self) -> Self {
        *self
    }
}

impl<K: Key + 'static, V: Key + 'static> Copy for MultimapTableDefinition<'_, K, V> {}

impl<K: Key + 'static, V: Key + 'static> Display for MultimapTableDefinition<'_, K, V> {
    fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result {
        write!(
            f,
            "{}<{}, {}>",
            self.name,
            K::type_name().name(),
            V::type_name().name()
        )
    }
}

/// Information regarding the usage of the in-memory cache
///
/// Note: these metrics are only collected when the "`cache_metrics`" feature is enabled
#[derive(Debug)]
pub struct CacheStats {
    pub(crate) evictions: u64,
    pub(crate) read_hits: u64,
    pub(crate) read_misses: u64,
    pub(crate) write_hits: u64,
    pub(crate) write_misses: u64,
    pub(crate) used_bytes: usize,
}

impl CacheStats {
    /// Number of times that data has been evicted, due to the cache being full
    ///
    /// To increase the cache size use [`Builder::set_cache_size`]
    pub fn evictions(&self) -> u64 {
        self.evictions
    }

    /// Number of times that unmodified data has been read from the cache
    pub fn read_hits(&self) -> u64 {
        self.read_hits
    }

    /// Number of times that unmodified data was not in the cache and was read from storage
    pub fn read_misses(&self) -> u64 {
        self.read_misses
    }

    /// Number of times that data modified in a transaction has been read from the cache
    pub fn write_hits(&self) -> u64 {
        self.write_hits
    }

    /// Number of times that data modified in a transaction was not in the cache and was read from storage
    pub fn write_misses(&self) -> u64 {
        self.write_misses
    }

    /// Number of bytes in the cache
    pub fn used_bytes(&self) -> usize {
        self.used_bytes
    }
}

// The write slot, taken ahead of the writer byte and held until the transaction ends. Releases
// it on drop, and performs the close a `Database::drop()` deferred to that end
pub(crate) struct WriteSlot {
    tracker: Arc<TransactionTracker>,
}

impl WriteSlot {
    fn take(tracker: Arc<TransactionTracker>) -> Self {
        tracker.take_write_slot();
        Self { tracker }
    }
}

impl Drop for WriteSlot {
    fn drop(&mut self) {
        if let Some(mem) = self.tracker.end_write_transaction() {
            close_database(&self.tracker, &mem);
        }
    }
}

pub(crate) enum TransactionGuard {
    Read {
        tracker: Arc<TransactionTracker>,
        transaction_id: TransactionId,
        // `Some` while this guard owns the transaction's reference, carrying what the tracker
        // needs to release it. `None` when a savepoint becomes persistent: the database
        // owns the reference from then on and releases it when the savepoint is deleted
        reference: Option<Arc<TransactionalMemory>>,
    },
    Write {
        transaction_id: TransactionId,
        // Dropped ahead of the slot, in this order: a thread waiting on the slot would take the
        // same byte on this file description, and this release would free theirs
        #[cfg(feature = "experimental-multiprocess")]
        _writer_lock: Arc<WriterLock>,
        slot: WriteSlot,
    },
    // Used for internal accesses that happen outside of any tracked transaction,
    // such as opening the database, repairing it, and running integrity checks.
    Untracked,
}

impl TransactionGuard {
    pub(crate) fn new_read(
        transaction_id: TransactionId,
        tracker: Arc<TransactionTracker>,
        mem: &Arc<TransactionalMemory>,
    ) -> Self {
        Self::Read {
            tracker,
            transaction_id,
            reference: Some(mem.clone()),
        }
    }

    // A guard for a transaction whose reference the database already owns
    pub(crate) fn new_read_unowned(
        transaction_id: TransactionId,
        tracker: Arc<TransactionTracker>,
    ) -> Self {
        Self::Read {
            tracker,
            transaction_id,
            reference: None,
        }
    }

    // Leak the reference to the transaction. The caller becomes responsible for decrementing the
    // reference count.
    // Dropping this guard then releases nothing.
    pub(crate) fn release_to_database(&mut self) {
        let Self::Read { reference, .. } = self else {
            unreachable!("only a read transaction's reference may be leaked")
        };
        drop(reference.take());
    }

    pub(crate) fn owns_reference(&self) -> bool {
        matches!(self, Self::Read { reference, .. } if reference.is_some())
    }

    pub(crate) fn tracker(&self) -> &Arc<TransactionTracker> {
        match self {
            Self::Read { tracker, .. } => tracker,
            Self::Write { slot, .. } => &slot.tracker,
            Self::Untracked => unreachable!("an untracked guard has no tracker"),
        }
    }

    // Returns a read guard and the data root its registration protects.
    pub(crate) fn allocate_read(
        tracker: Arc<TransactionTracker>,
        mem: &Arc<TransactionalMemory>,
    ) -> Result<(Self, Option<BtreeHeader>)> {
        let (id, root) = tracker.register_read_transaction(mem)?;

        Ok((Self::new_read(id, tracker, mem), root))
    }

    pub(crate) fn new_write(
        transaction_id: TransactionId,
        slot: WriteSlot,
        #[cfg(feature = "experimental-multiprocess")] writer_lock: Arc<WriterLock>,
    ) -> Self {
        Self::Write {
            transaction_id,
            #[cfg(feature = "experimental-multiprocess")]
            _writer_lock: writer_lock,
            slot,
        }
    }

    pub(crate) fn untracked() -> Self {
        Self::Untracked
    }

    pub(crate) fn id(&self) -> TransactionId {
        match self {
            Self::Read { transaction_id, .. } | Self::Write { transaction_id, .. } => {
                *transaction_id
            }
            Self::Untracked => {
                panic!("TransactionGuard::id() called on an untracked guard")
            }
        }
    }
}

impl Drop for TransactionGuard {
    fn drop(&mut self) {
        match self {
            Self::Read {
                tracker,
                transaction_id,
                reference,
            } => {
                if let Some(mem) = reference {
                    tracker.deallocate_read_transaction(mem, *transaction_id);
                }
            }
            Self::Write {
                #[cfg(feature = "experimental-multiprocess")]
                slot,
                ..
            } => {
                // Leave `Live` before the fields drop the writer lock and then the slot.
                #[cfg(feature = "experimental-multiprocess")]
                slot.tracker.begin_finalizing();
            }
            Self::Untracked => {}
        }
    }
}

pub trait ReadableDatabase: SealedInApi5 {
    /// Begins a read transaction
    ///
    /// Captures a snapshot of the database, so that only data committed before calling this method
    /// is visible in the transaction
    ///
    /// Returns a [`ReadTransaction`] which may be used to read from the database. Read transactions
    /// may exist concurrently with writes
    fn begin_read(&self) -> Result<ReadTransaction, TransactionError>;

    /// Information regarding the usage of the in-memory cache
    ///
    /// Note: these metrics are only collected when the "`cache_metrics`" feature is enabled
    fn cache_stats(&self) -> CacheStats;
}

// Unavailable without std: every route to one goes through a path, and the file-backed API is
// gated out below.
/// A redb database opened in read-only mode
///
/// Use [`Self::begin_read`] to get a [`ReadTransaction`] object that can be used to read from the database
///
/// Multiple processes may open a [`ReadOnlyDatabase`], but it may not be opened concurrently
/// with a [`Database`].
#[cfg_attr(
    feature = "experimental-multiprocess",
    doc = "",
    doc = "`SingleWriter` and `MultiWriter` are the exception: a reader shares the file with the writer and follows its commits. See [`Builder::set_concurrency_mode`](crate::Builder::set_concurrency_mode)."
)]
///
/// # Examples
///
/// Basic usage:
///
/// ```rust
/// use redb::*;
/// # use tempfile::NamedTempFile;
/// const TABLE: TableDefinition<u64, u64> = TableDefinition::new("my_data");
///
/// # fn main() -> Result<(), Error> {
/// # #[cfg(not(target_os = "wasi"))]
/// # let tmpfile = NamedTempFile::new().unwrap();
/// # #[cfg(target_os = "wasi")]
/// # let tmpfile = NamedTempFile::new_in("/tmp").unwrap();
/// # let filename = tmpfile.path();
/// let db = Database::create(filename)?;
/// let txn = db.begin_write()?;
/// {
///     let mut table = txn.open_table(TABLE)?;
///     table.insert(&0, &0)?;
/// }
/// txn.commit()?;
/// drop(db);
///
/// let db = ReadOnlyDatabase::open(filename)?;
/// let txn = db.begin_read()?;
/// {
///     let mut table = txn.open_table(TABLE)?;
///     println!("{}", table.get(&0)?.unwrap().value());
/// }
/// # Ok(())
/// # }
/// ```
#[cfg(not(redb_no_std))]
pub struct ReadOnlyDatabase {
    mem: Arc<TransactionalMemory>,
    transaction_tracker: Arc<TransactionTracker>,
}

#[cfg(not(redb_no_std))]
impl Sealed for ReadOnlyDatabase {}

#[cfg(not(redb_no_std))]
impl ReadableDatabase for ReadOnlyDatabase {
    fn begin_read(&self) -> Result<ReadTransaction, TransactionError> {
        let (guard, root) =
            TransactionGuard::allocate_read(self.transaction_tracker.clone(), &self.mem)?;
        #[cfg(feature = "logging")]
        debug!("Beginning read transaction id={:?}", guard.id());

        ReadTransaction::new(self.mem.clone(), guard, root)
    }

    fn cache_stats(&self) -> CacheStats {
        self.mem.cache_stats()
    }
}

#[cfg(not(redb_no_std))]
impl ReadOnlyDatabase {
    /// Opens an existing redb database.
    #[cfg(not(redb_no_std))]
    pub fn open(path: impl AsRef<Path>) -> Result<ReadOnlyDatabase, DatabaseError> {
        Builder::new().open_read_only(path)
    }

    fn new(
        file: Box<dyn StorageBackend>,
        page_size: usize,
        region_size: Option<u64>,
        cache_size: usize,
        concurrency_mode: ConcurrencyMode,
    ) -> Result<Self, DatabaseError> {
        #[cfg(feature = "logging")]
        let file_path = format!("{:?}", &file);
        #[cfg(feature = "logging")]
        debug!("Opening database in read-only {:?}", &file_path);
        let (mem, _writer_lock) = TransactionalMemory::new(
            Box::new(ReadOnlyBackend::new(file)),
            false,
            page_size,
            region_size,
            cache_size,
            true,
            concurrency_mode,
        )?;
        let mem = Arc::new(mem);
        // A reader beside a multi-process writer never allocates, so it loads no allocator state
        #[cfg(feature = "experimental-multiprocess")]
        let multiprocess_writer = concurrency_mode.is_multi_process_writable();
        #[cfg(not(feature = "experimental-multiprocess"))]
        let multiprocess_writer = false;
        if !multiprocess_writer {
            // If the last transaction used 2-phase commit and updated the allocator state table, then
            // we can just load the allocator state from there. Otherwise, we need a full repair
            if let Some(tree) = Database::get_allocator_state_table(&mem)? {
                mem.load_allocator_state(&tree)?;
            } else {
                #[cfg(feature = "logging")]
                warn!(
                    "Database {:?} not shutdown cleanly. Repair required",
                    &file_path
                );
                return Err(DatabaseError::RepairAborted);
            }
        }

        let next_transaction_id = mem.get_last_committed_transaction_id()?.next();
        let db = Self {
            mem,
            transaction_tracker: Arc::new(TransactionTracker::new(next_transaction_id)),
        };

        Ok(db)
    }
}

/// Opened redb database file
///
/// Use [`Self::begin_read`] to get a [`ReadTransaction`] object that can be used to read from the database
/// Use [`Self::begin_write`] to get a [`WriteTransaction`] object that can be used to read or write to the database
///
/// Multiple reads may be performed concurrently, with each other, and with writes. Only a single write
/// may be in progress at a time.
///
/// # Close semantics
///
/// Dropping the [`Database`] closes the database: buffered data is flushed, the file lock is
/// released, and outstanding [`ReadTransaction`]s are invalidated (their operations will return
/// [`StorageError::DatabaseClosed`]).
///
/// A live [`WriteTransaction`] keeps the database open, however: if one exists when the
/// [`Database`] is dropped, the transaction remains fully usable and the close described above
/// is deferred until the transaction commits, aborts, or is dropped. Until then the database
/// file remains locked, so re-opening it fails with [`DatabaseError::DatabaseAlreadyOpen`].
///
/// # Examples
///
/// Basic usage:
///
/// ```rust
/// use redb::*;
/// # use tempfile::NamedTempFile;
/// const TABLE: TableDefinition<u64, u64> = TableDefinition::new("my_data");
///
/// # fn main() -> Result<(), Error> {
/// # #[cfg(not(target_os = "wasi"))]
/// # let tmpfile = NamedTempFile::new().unwrap();
/// # #[cfg(target_os = "wasi")]
/// # let tmpfile = NamedTempFile::new_in("/tmp").unwrap();
/// # let filename = tmpfile.path();
/// let db = Database::create(filename)?;
/// let write_txn = db.begin_write()?;
/// {
///     let mut table = write_txn.open_table(TABLE)?;
///     table.insert(&0, &0)?;
/// }
/// write_txn.commit()?;
/// # Ok(())
/// # }
/// ```
pub struct Database {
    mem: Arc<TransactionalMemory>,
    transaction_tracker: Arc<TransactionTracker>,
}

impl Sealed for Database {}

impl ReadableDatabase for Database {
    fn begin_read(&self) -> Result<ReadTransaction, TransactionError> {
        let (guard, root) =
            TransactionGuard::allocate_read(self.transaction_tracker.clone(), &self.mem)?;
        #[cfg(feature = "logging")]
        debug!("Beginning read transaction id={:?}", guard.id());
        ReadTransaction::new(self.get_memory(), guard, root)
    }

    fn cache_stats(&self) -> CacheStats {
        self.mem.cache_stats()
    }
}

impl Database {
    /// Opens the specified file as a redb database.
    /// * if the file does not exist, or is an empty file, a new database will be initialized in it
    /// * if the file is a valid redb database, it will be opened
    /// * otherwise this function will return an error
    #[cfg(not(redb_no_std))]
    pub fn create(path: impl AsRef<Path>) -> Result<Database, DatabaseError> {
        Self::builder().create(path)
    }

    /// Opens an existing redb database.
    #[cfg(not(redb_no_std))]
    pub fn open(path: impl AsRef<Path>) -> Result<Database, DatabaseError> {
        Self::builder().open(path)
    }

    pub(crate) fn get_memory(&self) -> Arc<TransactionalMemory> {
        self.mem.clone()
    }

    pub(crate) fn verify_primary_checksums(mem: Arc<TransactionalMemory>) -> Result<bool> {
        let data_root = mem.get_data_root();
        let system_root = mem.get_system_root();
        Self::verify_checksums(mem, data_root, system_root)
    }

    // Verifies the checksums reachable from the given data and system roots, reading pages through
    // `mem` (i.e. from disk if its cache was invalidated first).
    fn verify_checksums(
        mem: Arc<TransactionalMemory>,
        data_root: Option<BtreeHeader>,
        system_root: Option<BtreeHeader>,
    ) -> Result<bool> {
        let resolver = PageResolver::new(mem.clone());
        let table_tree = TableTree::new(
            data_root,
            PageHint::None,
            Arc::new(TransactionGuard::untracked()),
            resolver.clone(),
        )?;
        if !table_tree.verify_checksums()? {
            return Ok(false);
        }
        let system_table_tree = TableTree::new(
            system_root,
            PageHint::None,
            Arc::new(TransactionGuard::untracked()),
            resolver,
        )?;
        if !system_table_tree.verify_checksums()? {
            return Ok(false);
        }

        Ok(true)
    }

    /// Force a check of the integrity of the database file, and repair it if possible.
    ///
    /// Note: Calling this function is unnecessary during normal operation. redb will automatically
    /// detect and recover from crashes, power loss, and other unclean shutdowns. This function is
    /// quite slow and should only be used when you suspect the database file may have been modified
    /// externally to redb, or that a redb bug may have left the database in a corrupted state.
    ///
    /// Returns `Ok(true)` if the database passed integrity checks; `Ok(false)` if it failed but was repaired,
    /// and `Err(Corrupted)` if the check failed and the file could not be repaired.
    ///
    /// Returns [`DatabaseError::TransactionInProgress`] if any read or write transaction, or an
    /// ephemeral [`Savepoint`](crate::Savepoint), is still alive on this handle when this method
    /// is called.
    ///
    /// Transactions committed with [`Durability::None`](crate::Durability::None) that have not yet
    /// been made durable are made durable if the check passes, or rolled back if the database must
    /// be repaired.
    #[cfg_attr(
        feature = "experimental-multiprocess",
        doc = "",
        doc = "In the multi-process concurrency modes the check holds the writer lock: it waits for a write transaction in another process to end, and no other process writes to the file until it returns. A read transaction in another process neither blocks the check nor is disturbed by it. See [`Builder::set_concurrency_mode`](crate::Builder::set_concurrency_mode)."
    )]
    pub fn check_integrity(&mut self) -> Result<bool, DatabaseError> {
        if Arc::get_mut(&mut self.mem).is_none() {
            return Err(DatabaseError::TransactionInProgress);
        }
        // An ephemeral savepoint may pin a non-durable transaction whose pages the reload below
        // discards; restoring it afterwards could corrupt the database. Persistent savepoints are
        // durable, so they are unaffected.
        if self.transaction_tracker.any_ephemeral_savepoint_exists() {
            return Err(DatabaseError::TransactionInProgress);
        }

        // Report a latched I/O failure as such, not as the discarded allocator state it also causes
        self.mem.check_io_errors()?;
        // Once the allocator state has been discarded (by a failed commit or integrity check),
        // the database must be reopened to rebuild it; this check requires one to compare against
        if self.mem.allocator_state_invalidated() {
            return Err(StorageError::Corrupted(
                "Allocator state was discarded by a failed integrity check or commit; reopen the database to repair it".to_string(),
            )
            .into());
        }

        // Held until the check returns, so the file it reloads is the file it repairs
        #[cfg(feature = "experimental-multiprocess")]
        let writer_lock = self.mem.lock_writer()?;
        // Repairing rebuilds the allocator state, so a failure part way through leaves one that
        // describes neither the file nor anything else. Holding an allocator state must continue
        // to mean it describes the file.
        let result = self.check_integrity_inner(
            #[cfg(feature = "experimental-multiprocess")]
            &writer_lock,
        );
        if result.is_err() {
            self.mem.invalidate_allocator_state();
        }
        result
    }

    fn check_integrity_inner(
        &mut self,
        #[cfg(feature = "experimental-multiprocess")] writer_lock: &Arc<WriterLock>,
    ) -> Result<bool, DatabaseError> {
        // A pending Durability::None commit is acknowledged, live data that the reload below would
        // discard. If the live state verifies, promote it to durable rather than losing it -- even
        // if the durable state it replaces turns out to be corrupt, in which case we recover from
        // the live state and report not-clean.
        let mut rolling_back_non_durable = false;
        if self.mem.pending_non_durable_commit() {
            // Verify from disk, not the page cache, so external modification is detected.
            self.mem.clear_read_cache();
            // Don't promote over a truncated or extended file -- the committed layout would be
            // inconsistent with it. Fall through to reload + repair instead.
            if self.mem.file_len_matches_layout()?
                && let Some(live_allocator_clean) = self.repair_live_state()?
            {
                // The live tree is intact (its allocator state was rebuilt above if it was stale),
                // so promote the acknowledged commit rather than rolling it back. The result is
                // clean only if neither the allocator nor the durable state below needed repair.
                let durable_clean = self.durable_state_clean()?;
                let mut txn = self
                    .begin_write_with(
                        #[cfg(feature = "experimental-multiprocess")]
                        Some(writer_lock),
                        #[cfg(feature = "experimental-multiprocess")]
                        None,
                    )
                    .map_err(|e| DatabaseError::Storage(e.into_storage_error()))?;
                txn.disable_post_commit_free();
                txn.commit()
                    .map_err(|e| DatabaseError::Storage(e.into_storage_error()))?;
                // The rebuild above reclaimed any leaked pages, so the close may record a
                // clean shutdown again
                self.mem.clear_needs_repair();
                return Ok(live_allocator_clean && durable_clean);
            }
            // The live tree is corrupt (or the file size changed), so the reload rolls the commit
            // back -- not clean, even if the durable state it falls back to is intact.
            rolling_back_non_durable = true;
        }

        // No pending commit, or fall-through: verify and repair the durable state. Capture the
        // allocator hash to compare against the rebuild below; with the pending case handled above,
        // the live and durable states are identical here, so this is a valid check. When the reload
        // finds another process's commit, the snapshot it recorded is compared instead: that is
        // what the next open trusts.
        let own_allocator_hash = self.mem.allocator_hash();
        let mem = Arc::get_mut(&mut self.mem).unwrap();
        let (mut was_clean, peer_committed) = mem.clear_cache_and_reload()?;
        let allocator_hash = if peer_committed {
            match Self::get_allocator_state_table(&self.mem)? {
                Some(tree) => {
                    self.mem.load_allocator_state(&tree)?;
                    Some(self.mem.allocator_hash())
                }
                None => None,
            }
        } else {
            Some(own_allocator_hash)
        };
        // An interrupted compaction or repair may leave no allocator snapshot. Record one so
        // the next writer can load it without rebuilding again.
        #[cfg(feature = "experimental-multiprocess")]
        if peer_committed
            && allocator_hash.is_none()
            && self.mem.concurrency_mode() == ConcurrencyMode::MultiWriter
        {
            was_clean = false;
        }

        let old_roots = [self.mem.get_data_root(), self.mem.get_system_root()];

        let new_roots = Self::do_repair(&mut self.mem, &|_| {}).map_err(|err| match err {
            DatabaseError::Storage(storage_err) => storage_err,
            _ => unreachable!(),
        })?;

        if old_roots != new_roots
            || allocator_hash.is_some_and(|hash| hash != self.mem.allocator_hash())
            || rolling_back_non_durable
        {
            was_clean = false;
        }

        if !was_clean {
            let next_transaction_id = self.mem.get_last_committed_transaction_id()?.next();
            let [data_root, system_root] = new_roots;
            self.mem.commit(
                data_root,
                system_root,
                next_transaction_id,
                true,
                ShrinkPolicy::Never,
                #[cfg(feature = "experimental-multiprocess")]
                None,
            )?;
            // Reserve the id, or the next write transaction would commit with the same one,
            // which crash recovery could then resolve to the wrong slot
            self.transaction_tracker
                .reserve_repair_transaction_id(next_transaction_id);
        }

        // The rebuild reclaimed any leaked pages, so the close may record a clean shutdown
        // again
        self.mem.clear_needs_repair();
        self.mem.begin_writable()?;

        // The tracker holds the persistent savepoints the reloaded tables hold, as the open's does
        if peer_committed {
            Self::sync_persistent_savepoints(
                &self.transaction_tracker,
                &self.mem,
                #[cfg(feature = "experimental-multiprocess")]
                Some(writer_lock),
            )?;
        }
        // In multi-writer mode a repair ends with a commit recording the allocator state, as the
        // open's does; after the savepoints are held, since a commit frees what nothing pins
        #[cfg(feature = "experimental-multiprocess")]
        if !was_clean && self.mem.concurrency_mode() == ConcurrencyMode::MultiWriter {
            ensure_allocator_state_table_and_trim(
                &self.transaction_tracker,
                &self.mem,
                Some(writer_lock),
                None,
            )?;
        }

        Ok(was_clean)
    }

    // Synchronizes the tracker state for the file's persistent savepoints. The transaction this
    // begins is lent `writer_lock`, when the caller holds one
    fn sync_persistent_savepoints(
        transaction_tracker: &Arc<TransactionTracker>,
        mem: &Arc<TransactionalMemory>,
        #[cfg(feature = "experimental-multiprocess")] writer_lock: Option<&Arc<WriterLock>>,
    ) -> Result<(), DatabaseError> {
        let txn = begin_write_with_allocation_policy(
            transaction_tracker,
            mem,
            #[cfg(feature = "experimental-multiprocess")]
            writer_lock,
            #[cfg(feature = "experimental-multiprocess")]
            None,
            AllocationPolicy::Default,
        )
        .map_err(|e| e.into_storage_error())?;
        sync_persistent_savepoints(
            transaction_tracker,
            #[cfg(feature = "experimental-multiprocess")]
            mem,
            &txn,
        )?;
        txn.abort()?;

        Ok(())
    }

    // Verifies, and repairs in memory, the live (possibly non-durable) state. Returns:
    // - `None` if the live tree is corrupt and the commit must be rolled back;
    // - `Some(true)` if the live state is fully clean;
    // - `Some(false)` if the tree is intact but its allocator state was stale and has been rebuilt,
    //   so promoting it repairs the allocator while the check reports not-clean.
    // The allocator is rebuilt from the live roots -- a rebuild from the durable roots would
    // falsely differ when the live state is ahead of durable (e.g. a durable commit's free-page
    // epilogue).
    fn repair_live_state(&mut self) -> Result<Option<bool>, DatabaseError> {
        match Self::verify_primary_checksums(self.mem.clone()) {
            Ok(true) => {
                let live_allocator_hash = self.mem.allocator_hash();
                let live_roots = [self.mem.get_data_root(), self.mem.get_system_root()];
                match Self::rebuild_allocator_state(&self.mem, &|_| {}) {
                    // Only a durable root can be rewritten from here, so a live root whose table
                    // count was recomputed must be rolled back and repaired by the reload below
                    Ok(roots) if roots != live_roots => Ok(None),
                    Ok(_) => Ok(Some(live_allocator_hash == self.mem.allocator_hash())),
                    Err(DatabaseError::Storage(StorageError::Corrupted(_))) => Ok(None),
                    Err(err) => Err(err),
                }
            }
            Ok(false) | Err(StorageError::Corrupted(_)) => Ok(None),
            Err(err) => Err(err.into()),
        }
    }

    // Whether the durable (primary slot) state is intact. Any corruption -- a bad primary slot
    // checksum, a checksum mismatch, or an error raised while walking a malformed tree -- counts as
    // not-clean: the caller promotes the verified live commit to recover from it, so corruption
    // here must not abort the check. Only non-corruption errors (e.g. I/O) propagate.
    fn durable_state_clean(&self) -> Result<bool, DatabaseError> {
        match self.verify_durable_state() {
            Ok(clean) => Ok(clean),
            Err(DatabaseError::Storage(StorageError::Corrupted(_))) => Ok(false),
            Err(err) => Err(err),
        }
    }

    fn verify_durable_state(&self) -> Result<bool, DatabaseError> {
        if self.mem.durable_primary_slot_corrupt()? {
            return Ok(false);
        }
        let data_root = self.mem.get_durable_data_root();
        let system_root = self.mem.get_durable_system_root();
        Ok(Self::verify_checksums(
            self.mem.clone(),
            data_root,
            system_root,
        )?)
    }

    /// Compacts the database file
    ///
    /// Returns `true` if compaction was performed, and `false` if no futher compaction was possible
    pub fn compact(&mut self) -> Result<bool, CompactionError> {
        // These checks must run before begin_write(): the caller may legally hold an open
        // WriteTransaction (it is not lifetime-bound to the Database), and if that transaction
        // created a savepoint, blocking in begin_write() below would deadlock. Savepoints must
        // be diagnosed before read references, because every live savepoint also holds a read
        // reference. The tracker covers persistent savepoints created by previous Database
        // instances, because they are re-registered when the database is opened.
        // In multi-writer mode the tracker's savepoints go stale: a peer may have deleted one
        // since this handle last synced, and only a write transaction of this handle's own syncs
        // them again. Refresh before reporting one as a blocker, unless this process already has
        // a write transaction live, since beginning another would block on it.
        #[cfg(feature = "experimental-multiprocess")]
        if self.mem.concurrency_mode() == ConcurrencyMode::MultiWriter
            && self.transaction_tracker.any_persistent_savepoint_exists()
            && !self.transaction_tracker.write_transaction_live()
        {
            self.begin_write()
                .map_err(|e| e.into_storage_error())?
                .abort()?;
        }
        if self.transaction_tracker.any_persistent_savepoint_exists() {
            return Err(CompactionError::PersistentSavepointExists);
        }
        if self.transaction_tracker.any_savepoint_exists() {
            return Err(CompactionError::EphemeralSavepointExists);
        }
        if self.transaction_tracker.any_user_read_reference_exists() {
            return Err(CompactionError::TransactionInProgress);
        }
        // Where the file is shared, held until the compaction is over and lent to the
        // transactions it runs: no other process begins a transaction under them, and one that
        // already has is a transaction in progress. Taken once a write transaction begun before
        // this call has ended: its commit would wait on the holds, and in multi-writer mode its
        // end would release the writer byte from under a hold taken meanwhile
        #[cfg(feature = "experimental-multiprocess")]
        let (writer_lock, header_lock) = if self.mem.concurrency_mode().is_multi_process_writable()
        {
            self.begin_write()
                .map_err(|e| e.into_storage_error())?
                .abort()?;
            (
                Some(self.mem.lock_writer()?),
                Some(self.mem.lock_header_exclusive()?),
            )
        } else {
            (None, None)
        };
        // Use 2-phase commit to avoid any possible security issues. Plus this compaction is going to be so slow that it doesn't matter.
        // Once https://github.com/cberner/redb/issues/829 is fixed, we should upgrade this to use quick-repair -- that way the user
        // can cancel the compaction without requiring a full repair afterwards
        let txn = self
            .begin_write_with(
                #[cfg(feature = "experimental-multiprocess")]
                writer_lock.as_ref(),
                #[cfg(feature = "experimental-multiprocess")]
                header_lock.as_ref(),
            )
            .map_err(|e| e.into_storage_error())?;
        // Re-check inside the write transaction: a concurrent writer may have created a
        // savepoint between the checks above and the start of this transaction.
        if txn.list_persistent_savepoints()?.next().is_some() {
            return Err(CompactionError::PersistentSavepointExists);
        }
        if self.transaction_tracker.any_savepoint_exists() {
            return Err(CompactionError::EphemeralSavepointExists);
        }
        if self.transaction_tracker.any_user_read_reference_exists() {
            return Err(CompactionError::TransactionInProgress);
        }
        // No local read is left to bound the scan, so a pin it finds is a peer's
        #[cfg(feature = "experimental-multiprocess")]
        if let Some(header_lock) = &header_lock
            && self
                .mem
                .oldest_active_transaction(None, header_lock)?
                .is_some()
        {
            return Err(CompactionError::TransactionInProgress);
        }
        txn.abort()?;
        // Commit to free up any pending free pages
        self.drain_pending_free_pages(
            ShrinkPolicy::Maximum,
            #[cfg(feature = "experimental-multiprocess")]
            writer_lock.as_ref(),
            #[cfg(feature = "experimental-multiprocess")]
            header_lock.as_ref(),
        )?;

        let mut compacted = false;
        // Iteratively compact until no progress is made
        loop {
            let mut progress = false;

            let mut txn = self
                .begin_write_with(
                    #[cfg(feature = "experimental-multiprocess")]
                    writer_lock.as_ref(),
                    #[cfg(feature = "experimental-multiprocess")]
                    header_lock.as_ref(),
                )
                .map_err(|e| e.into_storage_error())?;
            txn.skip_allocator_state_record();
            if txn.compact_pages()? {
                progress = true;
                txn.commit_with(
                    #[cfg(feature = "experimental-multiprocess")]
                    header_lock.as_ref(),
                )
                .map_err(|e| e.into_storage_error())?;
            } else {
                txn.abort()?;
            }

            // Drain pages freed by compact_pages(), including system pages queued by any
            // post-commit cleanup root updates.
            self.drain_pending_free_pages(
                ShrinkPolicy::Maximum,
                #[cfg(feature = "experimental-multiprocess")]
                writer_lock.as_ref(),
                #[cfg(feature = "experimental-multiprocess")]
                header_lock.as_ref(),
            )?;

            if !progress {
                break;
            }

            compacted = true;
        }

        // In multi-writer mode the file's latest commit records the allocator state, for the
        // next writer, in any process, to load rather than rebuild: the commit the close makes.
        // Not in the other modes, where the close records and a second record would leave the
        // file one record larger, the pages of the one it replaces being freed only by the next
        // commit
        #[cfg(feature = "experimental-multiprocess")]
        if self.mem.concurrency_mode() == ConcurrencyMode::MultiWriter {
            ensure_allocator_state_table_and_trim(
                &self.transaction_tracker,
                &self.mem,
                writer_lock.as_ref(),
                header_lock.as_ref(),
            )?;
        }

        Ok(compacted)
    }

    fn drain_pending_free_pages(
        &self,
        shrink_policy: ShrinkPolicy,
        #[cfg(feature = "experimental-multiprocess")] writer_lock: Option<&Arc<WriterLock>>,
        #[cfg(feature = "experimental-multiprocess")] header_lock: Option<&HeaderGuard<'_>>,
    ) -> Result {
        // Preserve compact()'s empty durable commit, which also publishes pending
        // non-durable roots before checking for pending frees.
        let mut force_commit = true;
        loop {
            let mut txn = self
                .begin_write_with(
                    #[cfg(feature = "experimental-multiprocess")]
                    writer_lock,
                    #[cfg(feature = "experimental-multiprocess")]
                    header_lock,
                )
                .map_err(|e| e.into_storage_error())?;
            if !force_commit && !txn.pending_free_pages()? {
                txn.abort()?;
                return Ok(());
            }
            force_commit = false;
            txn.skip_allocator_state_record();
            txn.set_two_phase_commit(true);
            txn.set_shrink_policy(shrink_policy);
            txn.commit_with(
                #[cfg(feature = "experimental-multiprocess")]
                header_lock,
            )
            .map_err(|e| e.into_storage_error())?;
        }
    }

    #[cfg_attr(not(debug_assertions), expect(dead_code))]
    fn check_repaired_allocated_pages_table(
        system_root: Option<BtreeHeader>,
        mem: Arc<TransactionalMemory>,
    ) -> Result {
        let resolver = PageResolver::new(mem.clone());
        let table_tree = TableTree::new(
            system_root,
            PageHint::None,
            Arc::new(TransactionGuard::untracked()),
            resolver.clone(),
        )?;
        if let Some(table_def) = table_tree
            .get_table::<TransactionIdWithPagination, PageList>(
                DATA_ALLOCATED_TABLE.name(),
                TableType::Normal,
            )
            .map_err(|e| e.into_storage_error_or_corrupted("Allocated pages table corrupted"))?
        {
            let InternalTableDefinition::Normal { table_root, .. } = table_def else {
                unreachable!()
            };
            let table: ReadOnlyTable<TransactionIdWithPagination, PageList> = ReadOnlyTable::new(
                DATA_ALLOCATED_TABLE.name().to_string(),
                table_root,
                PageHint::None,
                Arc::new(TransactionGuard::untracked()),
                resolver,
            )?;
            for result in ReadableTable::iter(&table)? {
                let (_, pages) = result?;
                for i in 0..pages.value().len() {
                    assert!(mem.is_allocated(pages.value().get(i)));
                }
            }
        }

        Ok(())
    }

    fn visit_freed_tree<K: Key, V: Value, F>(
        system_root: Option<BtreeHeader>,
        table_def: SystemTableDefinition<K, V>,
        mem: Arc<TransactionalMemory>,
        mut visitor: F,
    ) -> Result
    where
        F: FnMut(PageNumber) -> Result,
    {
        let untracked_guard = Arc::new(TransactionGuard::untracked());
        let resolver = PageResolver::new(mem.clone());
        let system_tree = TableTree::new(
            system_root,
            PageHint::None,
            untracked_guard,
            resolver.clone(),
        )?;
        let table_name = table_def.name();
        let result = match system_tree.get_table::<K, V>(table_name, TableType::Normal) {
            Ok(result) => result,
            Err(TableError::Storage(err)) => {
                return Err(err);
            }
            Err(TableError::TableDoesNotExist(_)) => {
                return Ok(());
            }
            Err(_) => {
                return Err(StorageError::Corrupted(format!(
                    "Unable to open {table_name}"
                )));
            }
        };

        if let Some(definition) = result {
            let table_root = match definition {
                InternalTableDefinition::Normal { table_root, .. } => table_root,
                InternalTableDefinition::Multimap { .. } => unreachable!(),
            };
            let table: ReadOnlyTable<TransactionIdWithPagination, PageList<'static>> =
                ReadOnlyTable::new(
                    table_name.to_string(),
                    table_root,
                    PageHint::None,
                    Arc::new(TransactionGuard::untracked()),
                    resolver,
                )?;
            for result in ReadableTable::iter(&table)? {
                let (_, page_list) = result?;
                for i in 0..page_list.value().len() {
                    visitor(page_list.value().get(i))?;
                }
            }
        }

        Ok(())
    }

    #[cfg(debug_assertions)]
    fn mark_allocated_page_for_debug(mem: &Arc<TransactionalMemory>) -> Result {
        let data_root = mem.get_data_root();
        {
            let untracked = Arc::new(TransactionGuard::untracked());
            let tables = TableTree::new(
                data_root,
                PageHint::None,
                untracked,
                PageResolver::new(mem.clone()),
            )?;
            tables.visit_all_pages(|path| {
                mem.mark_debug_allocated_page(path.page_number());
                Ok(())
            })?;
        }

        let system_root = mem.get_system_root();
        {
            let untracked = Arc::new(TransactionGuard::untracked());
            let system_tables = TableTree::new(
                system_root,
                PageHint::None,
                untracked,
                PageResolver::new(mem.clone()),
            )?;
            system_tables.visit_all_pages(|path| {
                mem.mark_debug_allocated_page(path.page_number());
                Ok(())
            })?;
        }

        Self::visit_freed_tree(system_root, DATA_FREED_TABLE, mem.clone(), |page| {
            mem.mark_debug_allocated_page(page);
            Ok(())
        })?;
        Self::visit_freed_tree(system_root, SYSTEM_FREED_TABLE, mem.clone(), |page| {
            mem.mark_debug_allocated_page(page);
            Ok(())
        })?;

        Ok(())
    }

    // Whether the primary slot's trees verify. A Corrupted error counts as "they do not", so that
    // a torn slot carrying an invalid page number falls back to the secondary like any other bad
    // primary, rather than aborting the repair. Other errors (e.g. I/O) still propagate.
    fn primary_verifies(mem: &Arc<TransactionalMemory>) -> Result<bool> {
        match Self::verify_primary_checksums(mem.clone()) {
            Ok(verified) => Ok(verified),
            Err(StorageError::Corrupted(_)) => Ok(false),
            Err(err) => Err(err),
        }
    }

    fn do_repair(
        mem: &mut Arc<TransactionalMemory>, // Only &mut to ensure exclusivity
        repair_callback: &(dyn Fn(&mut RepairSession) + 'static),
    ) -> Result<[Option<BtreeHeader>; 2], DatabaseError> {
        if !Self::primary_verifies(mem)? {
            if mem.used_two_phase_commit() {
                return Err(DatabaseError::Storage(StorageError::Corrupted(
                    "Primary is corrupted despite 2-phase commit".to_string(),
                )));
            }

            // 0.3 because the repair takes 3 full scans and the first is done now
            let mut handle = RepairSession::new(0.3);
            repair_callback(&mut handle);
            if handle.aborted() {
                return Err(DatabaseError::RepairAborted);
            }

            mem.repair_primary_corrupted();
            // We need to invalidate the userspace cache, because walking the tree in verify_primary_checksums() may
            // have poisoned it with pages that just got rolled back by repair_primary_corrupted(), since
            // that rolls back a partially committed transaction.
            mem.clear_read_cache();
            if !Self::primary_verifies(mem)? {
                return Err(DatabaseError::Storage(StorageError::Corrupted(
                    "Failed to repair database. All roots are corrupted".to_string(),
                )));
            }
        }
        // 0.6 because the repair takes 3 full scans and the second is done now
        let mut handle = RepairSession::new(0.6);
        repair_callback(&mut handle);
        if handle.aborted() {
            return Err(DatabaseError::RepairAborted);
        }

        let [data_root, system_root] = Self::rebuild_allocator_state(mem, repair_callback)?;

        mem.clear_recovery_required()?;

        // We need to invalidate the userspace cache, because we're about to implicitly free the freed table
        // by storing an empty root during the below commit()
        mem.clear_read_cache();

        Ok([data_root, system_root])
    }

    // Rebuilds the in-memory allocator state by marking every page reachable from the current
    // roots (including the pages referenced by the freed-page tables) as allocated. Operates
    // purely on in-memory state and does not modify the file.
    //
    // The returned roots carry table counts recounted from the trees that were walked. These
    // counts are stored in the commit slot rather than in a page, so no page checksum covers
    // them; recounting here is what lets the rest of the codebase trust them.
    //
    // Callers must ensure nothing else uses the allocator meanwhile: either no transaction is
    // live, or the caller holds the write slot (read transactions never allocate or free).
    fn rebuild_allocator_state(
        mem: &Arc<TransactionalMemory>,
        repair_callback: &(dyn Fn(&mut RepairSession) + 'static),
    ) -> Result<[Option<BtreeHeader>; 2], DatabaseError> {
        mem.reset_allocator_state()?;

        let data_root = {
            let root = mem.get_data_root();
            let untracked = Arc::new(TransactionGuard::untracked());
            let tables = TableTree::new(
                root,
                PageHint::None,
                untracked,
                PageResolver::new(mem.clone()),
            )?;
            tables.visit_all_pages(|path| mem.mark_page_allocated(path.page_number()))?;
            Self::with_recounted_length(root, tables.count_tables()?)
        };

        // 0.9 because the repair takes 3 full scans and the third is done now. There is just some system tables left
        let mut handle = RepairSession::new(0.9);
        repair_callback(&mut handle);
        if handle.aborted() {
            return Err(DatabaseError::RepairAborted);
        }

        let system_root = {
            let root = mem.get_system_root();
            let untracked = Arc::new(TransactionGuard::untracked());
            let system_tables = TableTree::new(
                root,
                PageHint::None,
                untracked,
                PageResolver::new(mem.clone()),
            )?;
            system_tables.visit_all_pages(|path| mem.mark_page_allocated(path.page_number()))?;
            Self::with_recounted_length(root, system_tables.count_tables()?)
        };

        Self::visit_freed_tree(system_root, DATA_FREED_TABLE, mem.clone(), |page| {
            mem.mark_page_allocated(page)
        })?;
        Self::visit_freed_tree(system_root, SYSTEM_FREED_TABLE, mem.clone(), |page| {
            mem.mark_page_allocated(page)
        })?;
        // Non-durable commits hold their data-freed records in memory rather than in
        // DATA_FREED_TABLE, so the walk above does not reach those pages. They are still allocated
        // until a commit processes the records, so mark them like any other freed-table page.
        for page in mem.unpersisted_data_freed_pages() {
            mem.mark_page_allocated(page)?;
        }
        #[cfg(debug_assertions)]
        {
            Self::check_repaired_allocated_pages_table(system_root, mem.clone())?;
        }

        Ok([data_root, system_root])
    }

    fn with_recounted_length(root: Option<BtreeHeader>, length: u64) -> Option<BtreeHeader> {
        root.map(|header| BtreeHeader::new(header.root, header.checksum, length))
    }

    /// Loads the latest allocator snapshot, or rebuilds it if compaction or repair stopped before
    /// recording one. The caller holds the write slot and writer byte throughout.
    #[cfg(feature = "experimental-multiprocess")]
    fn load_synced_allocator_state(
        mem: &Arc<TransactionalMemory>,
        _writer_lock: &WriterLock,
    ) -> Result {
        if let Some(tree) = Self::get_allocator_state_table(mem)? {
            mem.load_allocator_state(&tree)?;
            #[cfg(debug_assertions)]
            Self::mark_allocated_page_for_debug(mem)?;
        } else {
            if !Self::verify_primary_checksums(mem.clone())? {
                return Err(StorageError::Corrupted(
                    "Cannot rebuild allocator state: committed trees are corrupted".to_string(),
                ));
            }
            // Freed-page tables retain the pages that readers and persistent savepoints still
            // reference. Rebuilding must keep those allocated as well as the current trees.
            let roots = Self::rebuild_allocator_state(mem, &|_| {})
                .map_err(DatabaseError::into_storage_error_or_corrupted)?;
            if roots != [mem.get_data_root(), mem.get_system_root()] {
                return Err(StorageError::Corrupted(
                    "Committed table counts do not match the database trees".to_string(),
                ));
            }
        }
        // Loaded or rebuilt, this state holds none of the pages leaked by a transaction that
        // was dropped while a panic unwound. The next commit can record it again.
        mem.clear_needs_repair();

        Ok(())
    }

    fn new(
        file: Box<dyn StorageBackend>,
        allow_initialize: bool,
        page_size: usize,
        region_size: Option<u64>,
        cache_size: usize,
        concurrency_mode: ConcurrencyMode,
        repair_callback: &(dyn Fn(&mut RepairSession) + 'static),
    ) -> Result<Self, DatabaseError> {
        #[cfg(feature = "logging")]
        let file_path = format!("{:?}", &file);
        #[cfg(feature = "logging")]
        debug!("Opening database {:?}", &file_path);
        // The open runs under this lock, which the multi-writer open took for the header read
        // below and the repair after it, and which the transactions it runs are lent
        let (mem, writer_lock) = TransactionalMemory::new(
            file,
            allow_initialize,
            page_size,
            region_size,
            cache_size,
            false,
            concurrency_mode,
        )?;
        let mut mem = Arc::new(mem);
        // If the last transaction used 2-phase commit and updated the allocator state table, then
        // we can just load the allocator state from there. Otherwise, we need a full repair
        let repaired = if let Some(tree) = Self::get_allocator_state_table(&mem)? {
            #[cfg(feature = "logging")]
            debug!("Found valid allocator state, full repair not needed");
            mem.load_allocator_state(&tree)?;
            #[cfg(debug_assertions)]
            Self::mark_allocated_page_for_debug(&mem)?;
            false
        } else {
            #[cfg(feature = "logging")]
            warn!("Database {:?} not shutdown cleanly. Repairing", &file_path);
            let mut handle = RepairSession::new(0.0);
            repair_callback(&mut handle);
            if handle.aborted() {
                return Err(DatabaseError::RepairAborted);
            }
            let [data_root, system_root] = Self::do_repair(&mut mem, repair_callback)?;
            let next_transaction_id = mem.get_last_committed_transaction_id()?.next();
            mem.commit(
                data_root,
                system_root,
                next_transaction_id,
                true,
                ShrinkPolicy::Never,
                #[cfg(feature = "experimental-multiprocess")]
                None,
            )?;
            true
        };

        mem.begin_writable()?;
        // Past recovery, so a reader finding this byte held knows the flag means a live writer
        #[cfg(feature = "experimental-multiprocess")]
        mem.mark_consistent()?;
        let next_transaction_id = mem.get_last_committed_transaction_id()?.next();

        let transaction_tracker = Arc::new(TransactionTracker::new(next_transaction_id));

        // Restore the tracker state for any persistent savepoints
        Self::sync_persistent_savepoints(
            &transaction_tracker,
            &mem,
            #[cfg(feature = "experimental-multiprocess")]
            writer_lock.as_ref(),
        )?;
        // In multi-writer mode a repair ends with a commit recording the allocator state, as
        // compaction and the integrity check do, so that the next open, in any process, loads it
        #[cfg(feature = "experimental-multiprocess")]
        if repaired && concurrency_mode == ConcurrencyMode::MultiWriter {
            ensure_allocator_state_table_and_trim(
                &transaction_tracker,
                &mem,
                writer_lock.as_ref(),
                None,
            )?;
        }
        #[cfg(not(feature = "experimental-multiprocess"))]
        let _ = (repaired, writer_lock);
        // Construct only after initialization succeeds: Database::drop takes the writer lock,
        // which the open still holds, and assumes the savepoint tracker is complete.
        Ok(Database {
            mem,
            transaction_tracker,
        })
    }

    fn get_allocator_state_table(
        mem: &Arc<TransactionalMemory>,
    ) -> Result<Option<AllocatorStateTree>> {
        // The allocator state table is only valid if the primary was written using 2-phase commit
        if !mem.used_two_phase_commit() {
            return Ok(None);
        }

        // See if it's present in the system table tree
        let resolver = PageResolver::new(mem.clone());
        let system_table_tree = TableTree::new(
            mem.get_system_root(),
            PageHint::None,
            Arc::new(TransactionGuard::untracked()),
            resolver.clone(),
        )?;
        let Some(allocator_state_table) = system_table_tree
            .get_table::<AllocatorStateKey, &[u8]>(ALLOCATOR_STATE_TABLE_NAME, TableType::Normal)
            .map_err(|e| e.into_storage_error_or_corrupted("Unexpected TableError"))?
        else {
            return Ok(None);
        };

        // Load the allocator state table
        let InternalTableDefinition::Normal { table_root, .. } = allocator_state_table else {
            unreachable!();
        };
        let tree = AllocatorStateTree::new(
            table_root,
            PageHint::None,
            Arc::new(TransactionGuard::untracked()),
            resolver,
        )?;

        // Make sure this isn't stale allocator state left over from a previous transaction
        if !mem.is_valid_allocator_state(&tree)? {
            return Ok(None);
        }

        Ok(Some(tree))
    }

    /// Convenience method for [`Builder::new`]
    pub fn builder() -> Builder {
        Builder::new()
    }

    /// Begins a write transaction
    ///
    /// Returns a [`WriteTransaction`] which may be used to read/write to the database. Only a single
    /// write may be in progress at a time. If a write is in progress, this function will block
    /// until it completes.
    ///
    /// The returned transaction is not lifetime-bound to this [`Database`] and keeps the
    /// database open: if the [`Database`] is dropped while the transaction is live, the
    /// transaction remains usable and the database closes when the transaction completes.
    pub fn begin_write(&self) -> Result<WriteTransaction, TransactionError> {
        self.begin_write_with(
            #[cfg(feature = "experimental-multiprocess")]
            None,
            #[cfg(feature = "experimental-multiprocess")]
            None,
        )
    }

    fn begin_write_with(
        &self,
        #[cfg(feature = "experimental-multiprocess")] writer_lock: Option<&Arc<WriterLock>>,
        #[cfg(feature = "experimental-multiprocess")] header_lock: Option<&HeaderGuard<'_>>,
    ) -> Result<WriteTransaction, TransactionError> {
        begin_write_with_allocation_policy(
            &self.transaction_tracker,
            &self.mem,
            #[cfg(feature = "experimental-multiprocess")]
            writer_lock,
            #[cfg(feature = "experimental-multiprocess")]
            header_lock,
            AllocationPolicy::Default,
        )
    }
}

// Syncs the file's persistent savepoints, which another process may have created or
// deleted since the tracker last saw the file, and continues savepoint ids past the file's.
fn sync_persistent_savepoints(
    transaction_tracker: &TransactionTracker,
    #[cfg(feature = "experimental-multiprocess")] mem: &TransactionalMemory,
    txn: &WriteTransaction,
) -> Result {
    if let Some(next_id) = txn.next_persistent_savepoint_id()? {
        transaction_tracker.restore_savepoint_counter_state(next_id);
    }
    let current = txn.persistent_savepoint_transactions()?;
    #[cfg(feature = "experimental-multiprocess")]
    for &transaction_id in current.values() {
        // The file names its persistent savepoints, so the transaction each points at is
        // untrusted: one outside the lock range would be tracked as a read this process holds
        mem.check_active_transaction_id(transaction_id)?;
    }
    transaction_tracker.sync_persistent_savepoints(&current)
}

/// Brings this handle up to the file's latest commit, in multi-writer mode, when it is not
/// already on it. That is the case when another process has committed since this handle last
/// read the file, and when a panic latched a repair, which leaves this handle's allocator state
/// holding pages that the file's latest commit does not.
///
/// `writer_lock` proves that the caller holds the writer byte, so that no other process commits
/// while the header is read and the allocator state loaded from it; `header_lock` is the header
/// lock the caller already holds, if any.
///
/// Returns `None` when the handle was already on the file's latest commit. Otherwise returns a
/// latch that discards the allocator state unless the caller finishes the sync.
#[cfg(feature = "experimental-multiprocess")]
fn sync_to_latest_commit(
    mem: &Arc<TransactionalMemory>,
    writer_lock: &WriterLock,
    header_lock: Option<&HeaderGuard<'_>>,
) -> Result<Option<AllocatorStateLatch>> {
    if mem.concurrency_mode() != ConcurrencyMode::MultiWriter {
        return Ok(None);
    }
    let synced = {
        let hold = mem.header_hold(header_lock)?;
        mem.reload_for_write(writer_lock, &hold)?
    };
    if !synced {
        return Ok(None);
    }
    // Armed before the load, because a load that stops part way leaves an allocator state that
    // describes neither the commit it came from nor the one it was going to
    let latch = AllocatorStateLatch::arm(mem.clone());
    Database::load_synced_allocator_state(mem, writer_lock)?;
    // Loading the allocator state cleared the recovery flag that a live writer keeps set, and a
    // peer's close may have cleared it in the file as well. Set it again, so that the next
    // commit writes it back and a crash of this handle is still recovered from.
    mem.mark_recovery_required();

    Ok(Some(latch))
}

// Takes the write slot and the writer byte, syncs to the file's latest commit, and builds the
// transaction on them.
//
// A caller that already holds the writer byte lends its lock as `writer_lock`. Taking the writer
// byte again from this file description would return that same lock, and the end of this
// transaction would then release the caller's. A caller that holds the header lock lends it as
// `header_lock`, because the in-process half of that lock cannot be taken twice.
//
// The allocation policy is fixed for the lifetime of the transaction; every page allocation the
// transaction makes goes through it.
fn begin_write_with_allocation_policy(
    transaction_tracker: &Arc<TransactionTracker>,
    mem: &Arc<TransactionalMemory>,
    #[cfg(feature = "experimental-multiprocess")] writer_lock: Option<&Arc<WriterLock>>,
    #[cfg(feature = "experimental-multiprocess")] header_lock: Option<&HeaderGuard<'_>>,
    allocation_policy: AllocationPolicy,
) -> Result<WriteTransaction, TransactionError> {
    // Fail early if there has been an I/O error -- nothing can be committed in that case
    mem.check_io_errors()?;
    // The slot ahead of the byte: locks on the byte from one file description are one lock, so
    // the slot is what orders this process's writers. An error return drops both
    let slot = WriteSlot::take(transaction_tracker.clone());
    #[cfg(feature = "experimental-multiprocess")]
    let writer_lock = writer_lock.map_or_else(|| mem.lock_writer(), |lent| Ok(lent.clone()))?;
    #[cfg(feature = "experimental-multiprocess")]
    let latch = sync_to_latest_commit(mem, &writer_lock, header_lock)?;
    // Re-checked after acquiring the write slot: the writer this call blocked on can fail its
    // commit, latching an I/O error and discarding the allocator state. The I/O check comes
    // first so a backend failure is not misreported as corruption
    mem.check_io_errors()?;
    if mem.allocator_state_invalidated() {
        return Err(StorageError::Corrupted(
            "Allocator state was discarded by a failed integrity check or commit; reopen the database to repair it".to_string(),
        )
        .into());
    }
    // Issued once the sync is done, so that the id follows the file's latest commit
    let transaction_id = transaction_tracker.issue_write_transaction_id(
        mem.get_last_committed_transaction_id()?,
        #[cfg(feature = "experimental-multiprocess")]
        &writer_lock,
    );
    let guard = TransactionGuard::new_write(
        transaction_id,
        slot,
        #[cfg(feature = "experimental-multiprocess")]
        writer_lock,
    );
    let transaction = WriteTransaction::new(
        guard,
        transaction_tracker.clone(),
        mem.clone(),
        allocation_policy,
    )?;
    // The file's persistent savepoints are synced here, before this transaction frees anything,
    // so that the transactions they point at stay pinned. If the sync fails, the transaction is
    // aborted and the latch then discards the allocator state.
    #[cfg(feature = "experimental-multiprocess")]
    if latch.is_some() {
        sync_persistent_savepoints(transaction_tracker, mem, &transaction)?;
    }
    #[cfg(feature = "experimental-multiprocess")]
    if let Some(latch) = latch {
        latch.disarm();
    }

    Ok(transaction)
}

// Records the allocator state in a commit that also trims the file: the close's last commit,
// compaction's, and the one ending a repair in the open or the integrity check. `writer_lock`
// and `header_lock` are the ones the caller lends, when it holds them; nothing is waiting on the
// write slot in any of these cases
fn ensure_allocator_state_table_and_trim(
    transaction_tracker: &Arc<TransactionTracker>,
    mem: &Arc<TransactionalMemory>,
    #[cfg(feature = "experimental-multiprocess")] writer_lock: Option<&Arc<WriterLock>>,
    #[cfg(feature = "experimental-multiprocess")] header_lock: Option<&HeaderGuard<'_>>,
) -> Result {
    // Make a new quick-repair commit to update the allocator state table
    #[cfg(feature = "logging")]
    debug!("Writing allocator state table");
    // If compact() left no free pages, the default allocator lands this
    // commit's writes at high page indices (see AllocationPolicy::Lowest)
    // and try_shrink can't reclaim the growth. See
    // https://github.com/cberner/redb/issues/1165
    let mut tx = begin_write_with_allocation_policy(
        transaction_tracker,
        mem,
        #[cfg(feature = "experimental-multiprocess")]
        writer_lock,
        #[cfg(feature = "experimental-multiprocess")]
        header_lock,
        AllocationPolicy::Lowest,
    )
    .map_err(|e| e.into_storage_error())?;
    tx.set_quick_repair(true);
    tx.disable_post_commit_free();
    tx.set_shrink_policy(ShrinkPolicy::Maximum);
    tx.commit_with(
        #[cfg(feature = "experimental-multiprocess")]
        header_lock,
    )
    .map_err(|e| e.into_storage_error())?;

    Ok(())
}

// Closes the database: persists the allocator state table, so that the next open does not
// require a repair, and closes the storage backend. Runs exactly once, when the database
// closes: from Database::drop, or from the end of the write transaction that was live at
// that point. In both cases the Database is being, or has been, dropped, so no new write
// transaction can be started concurrently and the commit in here cannot block on the
// write-transaction slot.
fn close_database(transaction_tracker: &Arc<TransactionTracker>, mem: &Arc<TransactionalMemory>) {
    // No saved allocator state when it needs repair: the next open must rebuild it instead
    // of trusting the saved one. Nor after a latched I/O failure, decided ahead of the lock,
    // which waits on a peer holding the writer byte
    let writing = !crate::panicking() && !mem.needs_repair() && mem.check_io_errors().is_ok();
    // One hold across the allocator state's commit and the shutdown header: a commit another
    // process made between them would be overwritten by the header. Without the hold, neither
    // is written: the commit would take one of its own and release it, and the header would
    // then overwrite a commit made between them
    #[cfg(feature = "experimental-multiprocess")]
    let writer_lock = if writing {
        mem.lock_writer().ok()
    } else {
        None
    };
    #[cfg(feature = "experimental-multiprocess")]
    let writing = writing && writer_lock.is_some();
    let recorded = writing
        && ensure_allocator_state_table_and_trim(
            transaction_tracker,
            mem,
            #[cfg(feature = "experimental-multiprocess")]
            writer_lock.as_ref(),
            #[cfg(feature = "experimental-multiprocess")]
            None,
        )
        .is_ok();
    if writing && !recorded {
        #[cfg(feature = "logging")]
        warn!("Failed to write allocator state table. Repair may be required at restart.");
    }
    // The shutdown header describes this handle, which matches the file only once the commit
    // above has synced to the file's latest commit. Without that commit, write no header, and
    // leave the file for the next open to recover.
    #[cfg(feature = "experimental-multiprocess")]
    let writer_lock = writer_lock.filter(|_| recorded);

    if mem
        .close(
            #[cfg(feature = "experimental-multiprocess")]
            writer_lock.as_ref(),
        )
        .is_err()
    {
        #[cfg(feature = "logging")]
        warn!("Failed to flush database file. Repair may be required at restart.");
    }
}

impl Drop for Database {
    fn drop(&mut self) {
        if self
            .transaction_tracker
            .defer_close_if_write_transaction_live(&self.mem)
        {
            // The write transaction holds the memory and tracker alive, so it remains fully
            // usable; TransactionGuard::drop performs the deferred close when it ends
            #[cfg(feature = "logging")]
            warn!(
                "Database dropped while a write transaction is in progress. The database will remain open until the write transaction completes."
            );
            return;
        }

        close_database(&self.transaction_tracker, &self.mem);
    }
}

pub struct RepairSession {
    progress: f64,
    aborted: bool,
}

impl RepairSession {
    pub(crate) fn new(progress: f64) -> Self {
        Self {
            progress,
            aborted: false,
        }
    }

    pub(crate) fn aborted(&self) -> bool {
        self.aborted
    }

    /// Abort the repair process. The coorresponding call to [`Builder::open`] or [`Builder::create`] will return an error
    pub fn abort(&mut self) {
        self.aborted = true;
    }

    /// Returns an estimate of the repair progress in the range [0.0, 1.0). At 1.0 the repair is complete.
    pub fn progress(&self) -> f64 {
        self.progress
    }
}

/// How processes share a database: the regime a writer operates under. Whether a given handle
/// writes is chosen by `open()` against `open_read_only()`, not by the mode.
///
/// Every process opening one database concurrently must use a compatible mode: one
/// `SingleWriter` writer or any number of `MultiWriter` writers, plus read-only handles in either
/// case. Those two modes need byte-range file locks, so they are supported on Linux, the Apple
/// platforms and Windows; elsewhere, opening a database in one of them fails. `ExclusiveWriter`
/// locks the whole file and works everywhere.
#[cfg_attr(
    not(feature = "experimental-multiprocess"),
    allow(dead_code, unreachable_pub, clippy::enum_variant_names)
)]
#[derive(Copy, Clone, Eq, PartialEq, Debug, Default)]
pub enum ConcurrencyMode {
    /// A writer excludes every other process; read-only handles share the file with each other
    /// and exclude any writer. Enforced by locking the whole file. The default.
    #[default]
    ExclusiveWriter,
    /// One process writes; any number of processes may, concurrently, open the database read-only
    /// and follow its commits.
    SingleWriter,
    /// Any number of processes may, concurrently, open the database for reading and writing.
    /// Only one write transaction may be open at a time.
    MultiWriter,
}

#[cfg(feature = "experimental-multiprocess")]
impl ConcurrencyMode {
    /// Whether another process may have the database open, concurrently, and one process
    /// (possibly this one) is a writer
    pub(crate) fn is_multi_process_writable(self) -> bool {
        !matches!(self, ConcurrencyMode::ExclusiveWriter)
    }
}

/// Configuration builder of a redb [Database].
pub struct Builder {
    page_size: usize,
    region_size: Option<u64>,
    cache_size: usize,
    concurrency_mode: ConcurrencyMode,
    repair_callback: Box<dyn Fn(&mut RepairSession)>,
}

impl Builder {
    /// Construct a new [Builder] with sensible defaults.
    ///
    /// ## Defaults
    ///
    /// - `cache_size_bytes`: 1GiB
    #[allow(clippy::new_without_default)]
    pub fn new() -> Self {
        Self {
            // Default to 4k pages. Benchmarking showed that this was a good default on all platforms,
            // including MacOS with 16k pages. Therefore, users are not allowed to configure it at the moment.
            // It is part of the file format, so can be enabled in the future.
            page_size: PAGE_SIZE,
            region_size: None,
            concurrency_mode: ConcurrencyMode::ExclusiveWriter,
            cache_size: 1024 * 1024 * 1024,
            repair_callback: Box::new(|_| {}),
        }
    }

    /// Set a callback which will be invoked periodically in the event that the database file needs
    /// to be repaired.
    ///
    /// The [`RepairSession`] argument can be used to control the repair process.
    ///
    /// If the database file needs repair, the callback will be invoked at least once.
    /// There is no upper limit on the number of times it may be called.
    pub fn set_repair_callback(
        &mut self,
        callback: impl Fn(&mut RepairSession) + 'static,
    ) -> &mut Self {
        self.repair_callback = Box::new(callback);
        self
    }

    /// Set the internal page size of the database
    ///
    /// Valid values are powers of two, greater than or equal to 512
    ///
    /// ## Defaults
    ///
    /// Default to 4 Kib pages.
    #[cfg(any(fuzzing, test))]
    pub fn set_page_size(&mut self, size: usize) -> &mut Self {
        assert!(size.is_power_of_two());
        self.page_size = core::cmp::max(size, 512);
        self
    }

    /// Set the amount of memory (in bytes) used for caching data
    pub fn set_cache_size(&mut self, bytes: usize) -> &mut Self {
        self.cache_size = bytes;
        self
    }

    /// Set how processes may share this database. Defaults to [`ConcurrencyMode::ExclusiveWriter`].
    #[cfg(feature = "experimental-multiprocess")]
    pub fn set_concurrency_mode(&mut self, mode: ConcurrencyMode) -> &mut Self {
        self.concurrency_mode = mode;
        self
    }

    #[cfg(any(test, fuzzing))]
    pub fn set_region_size(&mut self, size: u64) -> &mut Self {
        assert!(size.is_power_of_two());
        self.region_size = Some(size);
        self
    }

    /// Opens the specified file as a redb database.
    /// * if the file does not exist, or is an empty file, a new database will be initialized in it
    /// * if the file is a valid redb database, it will be opened
    /// * otherwise this function will return an error
    #[cfg(not(redb_no_std))]
    pub fn create(&self, path: impl AsRef<Path>) -> Result<Database, DatabaseError> {
        let file = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(path)?;

        Database::new(
            Box::new(FileBackend::new(file)?),
            true,
            self.page_size,
            self.region_size,
            self.cache_size,
            self.concurrency_mode,
            &self.repair_callback,
        )
    }

    /// Opens an existing redb database.
    #[cfg(not(redb_no_std))]
    pub fn open(&self, path: impl AsRef<Path>) -> Result<Database, DatabaseError> {
        let file = OpenOptions::new().read(true).write(true).open(path)?;

        Database::new(
            Box::new(FileBackend::new(file)?),
            false,
            self.page_size,
            None,
            self.cache_size,
            self.concurrency_mode,
            &self.repair_callback,
        )
    }

    /// Opens an existing redb database.
    ///
    /// If the file has been opened for writing (i.e. as a [`Database`]) [`DatabaseError::DatabaseAlreadyOpen`]
    /// will be returned on platforms which support file locks (macOS, Windows, Linux). On other platforms,
    /// the caller MUST avoid calling this method when the database is open for writing.
    #[cfg_attr(
        feature = "experimental-multiprocess",
        doc = "",
        doc = "Only an `ExclusiveWriter` writer is refused: in `SingleWriter` and `MultiWriter` a reader opened in the same mode as the writer shares the file with it and picks up its commits. See [`Self::set_concurrency_mode`]."
    )]
    #[cfg(not(redb_no_std))]
    pub fn open_read_only(
        &self,
        path: impl AsRef<Path>,
    ) -> Result<ReadOnlyDatabase, DatabaseError> {
        let file = OpenOptions::new().read(true).open(path)?;

        ReadOnlyDatabase::new(
            Box::new(FileBackend::new(file)?),
            self.page_size,
            None,
            self.cache_size,
            self.concurrency_mode,
        )
    }

    /// Open an existing or create a new database in the given `file`.
    ///
    /// The file must be empty or contain a valid database.
    #[cfg(not(redb_no_std))]
    pub fn create_file(&self, file: File) -> Result<Database, DatabaseError> {
        Database::new(
            Box::new(FileBackend::new(file)?),
            true,
            self.page_size,
            self.region_size,
            self.cache_size,
            self.concurrency_mode,
            &self.repair_callback,
        )
    }

    /// Open an existing or create a new database with the given backend.
    pub fn create_with_backend(
        &self,
        backend: impl StorageBackend,
    ) -> Result<Database, DatabaseError> {
        Database::new(
            Box::new(backend),
            true,
            self.page_size,
            self.region_size,
            self.cache_size,
            self.concurrency_mode,
            &self.repair_callback,
        )
    }
}

impl core::fmt::Debug for Database {
    fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("Database").finish()
    }
}

#[cfg(test)]
mod test {
    use crate::backends::FileBackend;
    use crate::{
        CommitError, Database, DatabaseError, Durability, ReadableTable, StorageBackend,
        StorageError, TableDefinition, TransactionError,
    };
    use alloc::sync::Arc;
    use core::sync::atomic::{AtomicU64, Ordering};
    use std::fs::File;
    use std::io::{ErrorKind, Read, Seek, SeekFrom};

    #[derive(Debug)]
    struct FailingBackend {
        inner: FileBackend,
        countdown: Arc<AtomicU64>,
    }

    impl FailingBackend {
        fn new(backend: FileBackend, countdown: u64) -> Self {
            Self {
                inner: backend,
                countdown: Arc::new(AtomicU64::new(countdown)),
            }
        }

        fn check_countdown(&self) -> Result<(), std::io::Error> {
            if self.countdown.load(Ordering::SeqCst) == 0 {
                return Err(std::io::Error::from(ErrorKind::Other));
            }

            Ok(())
        }

        fn decrement_countdown(&self) -> Result<(), std::io::Error> {
            if self
                .countdown
                .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |x| {
                    if x > 0 { Some(x - 1) } else { None }
                })
                .is_err()
            {
                return Err(std::io::Error::from(ErrorKind::Other));
            }

            Ok(())
        }
    }

    impl StorageBackend for FailingBackend {
        fn len(&self) -> Result<u64, std::io::Error> {
            self.inner.len()
        }

        fn read(&self, offset: u64, out: &mut [u8]) -> Result<(), std::io::Error> {
            self.check_countdown()?;
            self.inner.read(offset, out)
        }

        fn set_len(&self, len: u64) -> Result<(), std::io::Error> {
            self.inner.set_len(len)
        }

        fn sync_data(&self) -> Result<(), std::io::Error> {
            self.check_countdown()?;
            self.inner.sync_data()
        }

        fn write(&self, offset: u64, data: &[u8]) -> Result<(), std::io::Error> {
            self.decrement_countdown()?;
            self.inner.write(offset, data)
        }
    }

    #[test]
    fn read_snapshot_survives_commits_before_construction() {
        use super::{ReadTransaction, TransactionGuard};

        const TABLE: TableDefinition<u64, u64> = TableDefinition::new("snapshot");
        for durability in [Durability::Immediate, Durability::None] {
            let tmpfile = crate::create_tempfile();
            let db = Database::create(tmpfile.path()).unwrap();
            let mut write = db.begin_write().unwrap();
            write.set_durability(durability).unwrap();
            write.open_table(TABLE).unwrap().insert(0, 0).unwrap();
            write.commit().unwrap();

            let (guard, root) =
                TransactionGuard::allocate_read(db.transaction_tracker.clone(), &db.mem).unwrap();
            assert_eq!(
                guard.id(),
                db.mem.get_last_committed_transaction_id().unwrap()
            );

            // Commits between registration and construction must not change this read's root.
            for value in 1..4 {
                let mut write = db.begin_write().unwrap();
                write.set_durability(durability).unwrap();
                write.open_table(TABLE).unwrap().insert(0, value).unwrap();
                write.commit().unwrap();
            }

            let read = ReadTransaction::new(db.mem.clone(), guard, root).unwrap();
            assert_eq!(
                read.open_table(TABLE)
                    .unwrap()
                    .get(0)
                    .unwrap()
                    .unwrap()
                    .value(),
                0
            );
        }
    }

    #[test]
    fn crash_regression4() {
        let tmpfile = crate::create_tempfile();
        let (file, path) = tmpfile.into_parts();

        let backend = FailingBackend::new(FileBackend::new(file).unwrap(), 20);
        let db = Database::builder()
            .set_cache_size(12686)
            .set_page_size(8 * 1024)
            .set_region_size(32 * 4096)
            .create_with_backend(backend)
            .unwrap();

        let table_def: TableDefinition<u64, &[u8]> = TableDefinition::new("x");

        let tx = db.begin_write().unwrap();
        let _savepoint = tx.ephemeral_savepoint().unwrap();
        let _persistent_savepoint = tx.persistent_savepoint().unwrap();
        tx.commit().unwrap();
        let tx = db.begin_write().unwrap();
        {
            let mut table = tx.open_table(table_def).unwrap();
            let _ = table.insert_reserve(118821, 360).unwrap();
        }
        let result = tx.commit();
        assert!(result.is_err());

        drop(db);
        Database::builder()
            .set_cache_size(1024 * 1024)
            .set_page_size(8 * 1024)
            .set_region_size(32 * 4096)
            .create(&path)
            .unwrap();
    }

    #[test]
    fn transient_io_error() {
        let tmpfile = crate::create_tempfile();
        let (file, path) = tmpfile.into_parts();

        let backend = FailingBackend::new(FileBackend::new(file).unwrap(), u64::MAX);
        let countdown = backend.countdown.clone();
        let db = Database::builder()
            .set_cache_size(0)
            .create_with_backend(backend)
            .unwrap();

        let table_def: TableDefinition<u64, u64> = TableDefinition::new("x");

        // Create some garbage
        let tx = db.begin_write().unwrap();
        {
            let mut table = tx.open_table(table_def).unwrap();
            table.insert(0, 0).unwrap();
        }
        tx.commit().unwrap();
        let tx = db.begin_write().unwrap();
        {
            let mut table = tx.open_table(table_def).unwrap();
            table.insert(0, 1).unwrap();
        }
        tx.commit().unwrap();

        let tx = db.begin_write().unwrap();
        // Cause an error in the commit
        countdown.store(0, Ordering::SeqCst);
        let result = tx.commit().err().unwrap();
        assert!(matches!(result, CommitError::Storage(StorageError::Io(_))));
        let result = db.begin_write().err().unwrap();
        assert!(matches!(
            result,
            TransactionError::Storage(StorageError::PreviousIo)
        ));
        // Simulate a transient error
        countdown.store(u64::MAX, Ordering::SeqCst);
        drop(db);

        // Check that recovery flag is set, even though the error has "cleared"
        let mut file = File::open(&path).unwrap();
        file.seek(SeekFrom::Start(9)).unwrap();
        let mut god_byte = vec![0u8];
        assert_eq!(file.read(&mut god_byte).unwrap(), 1);
        assert_ne!(god_byte[0] & 2, 0);
    }

    #[test]
    fn small_pages() {
        let tmpfile = crate::create_tempfile();

        let db = Database::builder()
            .set_page_size(512)
            .create(tmpfile.path())
            .unwrap();

        let table_definition: TableDefinition<u64, &[u8]> = TableDefinition::new("x");
        let txn = db.begin_write().unwrap();
        {
            txn.open_table(table_definition).unwrap();
        }
        txn.commit().unwrap();
    }

    #[test]
    fn small_pages2() {
        let tmpfile = crate::create_tempfile();

        let db = Database::builder()
            .set_page_size(512)
            .create(tmpfile.path())
            .unwrap();

        let table_def: TableDefinition<u64, &[u8]> = TableDefinition::new("x");

        let mut tx = db.begin_write().unwrap();
        tx.set_two_phase_commit(true);
        let savepoint0 = tx.ephemeral_savepoint().unwrap();
        {
            tx.open_table(table_def).unwrap();
        }
        tx.commit().unwrap();

        let mut tx = db.begin_write().unwrap();
        tx.set_two_phase_commit(true);
        let savepoint1 = tx.ephemeral_savepoint().unwrap();
        tx.restore_savepoint(&savepoint0).unwrap();
        tx.set_durability(Durability::None).unwrap();
        {
            let mut t = tx.open_table(table_def).unwrap();
            t.insert_reserve(&660503, 489).unwrap().as_mut().fill(0xFF);
            assert!(t.remove(&291295).unwrap().is_none());
        }
        tx.commit().unwrap();

        let mut tx = db.begin_write().unwrap();
        tx.set_two_phase_commit(true);
        tx.restore_savepoint(&savepoint0).unwrap();
        {
            tx.open_table(table_def).unwrap();
        }
        tx.commit().unwrap();

        let mut tx = db.begin_write().unwrap();
        tx.set_two_phase_commit(true);
        let savepoint2 = tx.ephemeral_savepoint().unwrap();
        drop(savepoint0);
        tx.restore_savepoint(&savepoint2).unwrap();
        {
            let mut t = tx.open_table(table_def).unwrap();
            assert!(t.get(&2059).unwrap().is_none());
            assert!(t.remove(&145227).unwrap().is_none());
            assert!(t.remove(&145227).unwrap().is_none());
        }
        tx.commit().unwrap();

        let mut tx = db.begin_write().unwrap();
        tx.set_two_phase_commit(true);
        let savepoint3 = tx.ephemeral_savepoint().unwrap();
        drop(savepoint1);
        tx.restore_savepoint(&savepoint3).unwrap();
        {
            tx.open_table(table_def).unwrap();
        }
        tx.commit().unwrap();

        let mut tx = db.begin_write().unwrap();
        tx.set_two_phase_commit(true);
        let savepoint4 = tx.ephemeral_savepoint().unwrap();
        drop(savepoint2);
        tx.restore_savepoint(&savepoint3).unwrap();
        tx.set_durability(Durability::None).unwrap();
        {
            let mut t = tx.open_table(table_def).unwrap();
            assert!(t.remove(&207936).unwrap().is_none());
        }
        tx.abort().unwrap();

        let mut tx = db.begin_write().unwrap();
        tx.set_two_phase_commit(true);
        let _savepoint5 = tx.ephemeral_savepoint().unwrap();
        drop(savepoint3);
        // savepoint4 was invalidated by the restore_savepoint(savepoint3) call
        // above, but that transaction was aborted, so the invalidation is
        // reversed and savepoint4 is valid again. Restoring it here invalidates
        // savepoint5 (which is newer), so the next transaction restores
        // savepoint4 again rather than savepoint5.
        tx.restore_savepoint(&savepoint4).unwrap();
        {
            tx.open_table(table_def).unwrap();
        }
        tx.commit().unwrap();

        let mut tx = db.begin_write().unwrap();
        tx.set_two_phase_commit(true);
        tx.restore_savepoint(&savepoint4).unwrap();
        tx.set_durability(Durability::None).unwrap();
        {
            tx.open_table(table_def).unwrap();
        }
        tx.commit().unwrap();
    }

    #[test]
    fn small_pages3() {
        let tmpfile = crate::create_tempfile();

        let db = Database::builder()
            .set_page_size(1024)
            .create(tmpfile.path())
            .unwrap();

        let table_def: TableDefinition<u64, &[u8]> = TableDefinition::new("x");

        let mut tx = db.begin_write().unwrap();
        let _savepoint0 = tx.ephemeral_savepoint().unwrap();
        tx.set_durability(Durability::None).unwrap();
        {
            let mut t = tx.open_table(table_def).unwrap();
            let value = vec![0; 306];
            t.insert(&539717, value.as_slice()).unwrap();
        }
        tx.abort().unwrap();

        let mut tx = db.begin_write().unwrap();
        let savepoint1 = tx.ephemeral_savepoint().unwrap();
        tx.restore_savepoint(&savepoint1).unwrap();
        tx.set_durability(Durability::None).unwrap();
        {
            let mut t = tx.open_table(table_def).unwrap();
            let value = vec![0; 2008];
            t.insert(&784384, value.as_slice()).unwrap();
        }
        tx.abort().unwrap();
    }

    #[test]
    fn small_pages4() {
        let tmpfile = crate::create_tempfile();

        let db = Database::builder()
            .set_cache_size(1024 * 1024)
            .set_page_size(1024)
            .create(tmpfile.path())
            .unwrap();

        let table_def: TableDefinition<u64, &[u8]> = TableDefinition::new("x");

        let tx = db.begin_write().unwrap();
        {
            tx.open_table(table_def).unwrap();
        }
        tx.commit().unwrap();

        let tx = db.begin_write().unwrap();
        {
            let mut t = tx.open_table(table_def).unwrap();
            assert!(t.get(&131072).unwrap().is_none());
            let value = vec![0xFF; 1130];
            t.insert(&42394, value.as_slice()).unwrap();
            t.insert_reserve(&744037, 3645).unwrap().as_mut().fill(0xFF);
            assert!(t.get(&0).unwrap().is_none());
        }
        tx.abort().unwrap();

        let tx = db.begin_write().unwrap();
        {
            let mut t = tx.open_table(table_def).unwrap();
            t.insert_reserve(&118749, 734).unwrap().as_mut().fill(0xFF);
        }
        tx.abort().unwrap();
    }

    #[test]
    fn dynamic_shrink() {
        let tmpfile = crate::create_tempfile();
        let table_definition: TableDefinition<u64, &[u8]> = TableDefinition::new("x");
        let big_value = vec![0u8; 1024];

        let db = Database::builder()
            .set_region_size(1024 * 1024)
            .create(tmpfile.path())
            .unwrap();

        let txn = db.begin_write().unwrap();
        {
            let mut table = txn.open_table(table_definition).unwrap();
            for i in 0..2048 {
                table.insert(&i, big_value.as_slice()).unwrap();
            }
        }
        txn.commit().unwrap();

        let file_size = tmpfile.as_file().metadata().unwrap().len();

        let txn = db.begin_write().unwrap();
        {
            let mut table = txn.open_table(table_definition).unwrap();
            for i in 0..2048 {
                table.remove(&i).unwrap();
            }
        }
        txn.commit().unwrap();

        // Perform a couple more commits to be sure the database has a chance to compact
        let txn = db.begin_write().unwrap();
        {
            let mut table = txn.open_table(table_definition).unwrap();
            table.insert(0, [].as_slice()).unwrap();
        }
        txn.commit().unwrap();
        let txn = db.begin_write().unwrap();
        {
            let mut table = txn.open_table(table_definition).unwrap();
            table.remove(0).unwrap();
        }
        txn.commit().unwrap();
        let txn = db.begin_write().unwrap();
        txn.commit().unwrap();

        let final_file_size = tmpfile.as_file().metadata().unwrap().len();
        assert!(final_file_size < file_size);
    }

    #[test]
    fn create_new_db_in_empty_file() {
        let tmpfile = crate::create_tempfile();

        let _db = Database::builder()
            .create_file(tmpfile.into_file())
            .unwrap();
    }

    #[test]
    fn open_missing_file() {
        let tmpfile = crate::create_tempfile();

        let err = Database::builder()
            .open(tmpfile.path().with_extension("missing"))
            .unwrap_err();

        match err {
            DatabaseError::Storage(StorageError::Io(err)) if err.kind() == ErrorKind::NotFound => {}
            err => panic!("Unexpected error for empty file: {err}"),
        }
    }

    #[test]
    fn open_empty_file() {
        let tmpfile = crate::create_tempfile();

        let err = Database::builder().open(tmpfile.path()).unwrap_err();

        match err {
            DatabaseError::Storage(StorageError::Io(err))
                if err.kind() == ErrorKind::InvalidData => {}
            err => panic!("Unexpected error for empty file: {err}"),
        }
    }
}

/// Probed directly, since the public API reports every lock conflict the same way and so
/// cannot say which byte caused one.
#[cfg(all(
    test,
    feature = "experimental-multiprocess",
    any(target_os = "linux", target_vendor = "apple", windows)
))]
mod writer_byte_test {
    use super::{ConcurrencyMode, Database, WRITER_BYTE, byte_range};
    use crate::TableDefinition;
    use crate::tree_store::HEADER_LOCK;
    use crate::tree_store::file_backend::range_lock::RangeLock;
    use std::fs::{File, OpenOptions};
    use std::path::Path;

    const TABLE: TableDefinition<u64, u64> = TableDefinition::new("x");

    fn create(path: &Path, mode: ConcurrencyMode) -> Database {
        let mut builder = Database::builder();
        builder.set_concurrency_mode(mode);
        builder.create(path).unwrap()
    }

    /// A separate description, so its locks conflict with the database's exactly as another
    /// process's would
    fn probe(path: &Path) -> File {
        OpenOptions::new()
            .read(true)
            .write(true)
            .open(path)
            .unwrap()
    }

    fn commit_one(db: &Database) {
        let write = db.begin_write().unwrap();
        {
            let mut table = write.open_table(TABLE).unwrap();
            table.insert(0, 0).unwrap();
        }
        write.commit().unwrap();
    }

    /// Whether the file is unclean, as a shared reader sees it: it is refused an unclean file
    /// no live writer holds
    fn left_unclean(path: &Path) -> bool {
        let mut builder = Database::builder();
        builder.set_concurrency_mode(ConcurrencyMode::MultiWriter);
        match builder.open_read_only(path) {
            Err(super::DatabaseError::RepairAborted) => true,
            Err(err) => panic!("{err}"),
            Ok(_) => false,
        }
    }

    /// The shutdown header is written under the writer byte, or not at all: written without it,
    /// it would overwrite a commit another process makes meanwhile
    #[test]
    fn the_shutdown_header_is_written_under_the_writer_byte_or_not_at_all() {
        // The memory's close alone, without the database's, which would write the header itself
        fn close(path: &Path, under_the_byte: bool) {
            let db = create(path, ConcurrencyMode::MultiWriter);
            commit_one(&db);
            let mem = db.mem.clone();
            std::mem::forget(db);
            if under_the_byte {
                let writer_lock = mem.lock_writer().unwrap();
                mem.close(Some(&writer_lock)).unwrap();
            } else {
                mem.close(None).unwrap();
            }
        }

        let tmpfile = crate::create_tempfile();
        close(tmpfile.path(), false);
        assert!(
            left_unclean(tmpfile.path()),
            "a header was written without the byte"
        );

        let tmpfile = crate::create_tempfile();
        close(tmpfile.path(), true);
        assert!(
            !left_unclean(tmpfile.path()),
            "no header was written under the byte"
        );
    }

    /// The close takes the byte once, for its commit and for the header after it
    #[test]
    fn a_close_leaves_a_clean_file() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        commit_one(&db);
        drop(db);
        assert!(
            !left_unclean(tmpfile.path()),
            "the close left the file unclean"
        );
    }

    /// A multi-writer repair, the create's included, ends with a commit recording the allocator
    /// state it rebuilt, for the next open to load
    #[test]
    fn a_multi_writer_repair_records_the_allocator_state() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        assert!(
            Database::get_allocator_state_table(&db.mem)
                .unwrap()
                .is_some()
        );
    }

    /// Rebuilding keeps live readers and the peer's persistent savepoints usable, even if the
    /// first transaction using that allocator aborts.
    #[test]
    fn a_sync_rebuilds_a_missing_allocator_state() {
        use crate::{ReadableDatabase, ReadableTable};

        let tmpfile = crate::create_tempfile();
        let mut builder = Database::builder();
        builder
            .set_concurrency_mode(ConcurrencyMode::MultiWriter)
            .set_cache_size(0);
        let db = builder.create(tmpfile.path()).unwrap();
        commit_one(&db);
        let old_read = db.begin_read().unwrap();
        let peer = builder.open(tmpfile.path()).unwrap();
        let mut write = peer.begin_write().unwrap();
        let savepoint_id = write.persistent_savepoint().unwrap();
        write.skip_allocator_state_record();
        write.open_table(TABLE).unwrap().insert(0, 1).unwrap();
        write.commit().unwrap();

        let mut write = db.begin_write().unwrap();
        {
            let table = write.open_table(TABLE).unwrap();
            assert_eq!(table.get(0).unwrap().unwrap().value(), 1);
        }
        let savepoint = write.get_persistent_savepoint(savepoint_id).unwrap();
        write.restore_savepoint(&savepoint).unwrap();
        write.abort().unwrap();

        let write = db.begin_write().unwrap();
        {
            let mut table = write.open_table(TABLE).unwrap();
            assert_eq!(table.get(0).unwrap().unwrap().value(), 1);
            table.insert(0, 2).unwrap();
        }
        write.commit().unwrap();
        assert!(
            Database::get_allocator_state_table(&db.mem)
                .unwrap()
                .is_some()
        );
        let old_table = old_read.open_table(TABLE).unwrap();
        assert_eq!(old_table.get(0).unwrap().unwrap().value(), 0);

        let mut write = peer.begin_write().unwrap();
        let savepoint = write.get_persistent_savepoint(savepoint_id).unwrap();
        write.restore_savepoint(&savepoint).unwrap();
        write.commit().unwrap();
        let read = peer.begin_read().unwrap();
        let table = read.open_table(TABLE).unwrap();
        assert_eq!(table.get(0).unwrap().unwrap().value(), 0);
    }

    /// Exit after compaction's initial drain, before its final allocator snapshot. The surviving
    /// writer must resume without reopening, including while a peer reads the published commit.
    #[test]
    fn a_writer_recovers_after_interrupted_compaction() {
        use crate::tree_store::ShrinkPolicy;
        use crate::{ReadableDatabase, ReadableTable};
        use std::process::Command;

        const CHILD_PATH: &str = "REDB_TEST_INTERRUPTED_COMPACTION_PATH";
        let mut builder = Database::builder();
        builder
            .set_concurrency_mode(ConcurrencyMode::MultiWriter)
            .set_cache_size(0);
        if let Some(path) = std::env::var_os(CHILD_PATH) {
            let db = builder.open(path).unwrap();
            let write = db.begin_write().unwrap();
            write.open_table(TABLE).unwrap().insert(1, 1).unwrap();
            write.commit().unwrap();
            let writer = db.mem.lock_writer().unwrap();
            let header = db.mem.lock_header_exclusive().unwrap();
            db.drain_pending_free_pages(ShrinkPolicy::Maximum, Some(&writer), Some(&header))
                .unwrap();
            assert!(
                Database::get_allocator_state_table(&db.mem)
                    .unwrap()
                    .is_none()
            );
            std::process::exit(42);
        }

        let tmpfile = crate::create_tempfile();
        let db = builder.create(tmpfile.path()).unwrap();
        commit_one(&db);
        let child = Command::new(std::env::current_exe().unwrap())
            .args([
                "--exact",
                "db::writer_byte_test::a_writer_recovers_after_interrupted_compaction",
                "--nocapture",
            ])
            .env(CHILD_PATH, tmpfile.path())
            .output()
            .unwrap();
        assert_eq!(child.status.code(), Some(42), "{child:?}");

        let reader = builder.open_read_only(tmpfile.path()).unwrap();
        let old_read = reader.begin_read().unwrap();
        let write = db.begin_write().unwrap();
        {
            let mut table = write.open_table(TABLE).unwrap();
            assert_eq!(table.get(1).unwrap().unwrap().value(), 1);
            table.insert(2, 2).unwrap();
        }
        write.commit().unwrap();
        assert!(
            Database::get_allocator_state_table(&db.mem)
                .unwrap()
                .is_some()
        );
        let old_table = old_read.open_table(TABLE).unwrap();
        assert_eq!(old_table.get(1).unwrap().unwrap().value(), 1);
        assert!(old_table.get(2).unwrap().is_none());

        builder
            .set_repair_callback(|_| panic!("the recovered writer did not record its allocator"));
        let peer = builder.open(tmpfile.path()).unwrap();
        let read = peer.begin_read().unwrap();
        let table = read.open_table(TABLE).unwrap();
        for key in 0..3u64 {
            assert_eq!(table.get(key).unwrap().unwrap().value(), key);
        }
    }

    /// A missing snapshot permits rebuilding the allocator, but never trusting damaged roots.
    #[test]
    fn a_sync_refuses_corruption_without_an_allocator_snapshot() {
        use crate::tree_store::ShrinkPolicy;
        use crate::{StorageError, TransactionError};

        for corrupt_checksum in [true, false] {
            let tmpfile = crate::create_tempfile();
            let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
            let peer = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
            commit_one(&peer);
            let writer = peer.mem.lock_writer().unwrap();
            let mut root = peer.mem.get_data_root().unwrap();
            if corrupt_checksum {
                root.checksum ^= 1;
            } else {
                root.length += 1;
            }
            peer.mem
                .commit(
                    Some(root),
                    peer.mem.get_system_root(),
                    peer.mem.get_last_committed_transaction_id().unwrap().next(),
                    true,
                    ShrinkPolicy::Never,
                    None,
                )
                .unwrap();
            peer.mem.invalidate_allocator_state();
            drop(writer);

            assert!(matches!(
                db.begin_write(),
                Err(TransactionError::Storage(StorageError::Corrupted(_)))
            ));
            assert!(!db.mem.allocator_state_loaded());
            assert!(matches!(
                db.begin_write(),
                Err(TransactionError::Storage(StorageError::Corrupted(_)))
            ));
        }
    }

    /// A transaction dropped while a panic unwinds leaks its pages, which latches a repair.
    /// Syncing to a peer's commit replaces the allocator state with the file's, which does not
    /// hold those pages, so the next commit records the state again.
    #[test]
    fn syncing_to_a_peers_commit_releases_the_repair_a_panic_latched() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        let mut builder = Database::builder();
        builder.set_concurrency_mode(ConcurrencyMode::MultiWriter);
        let peer = builder.open(tmpfile.path()).unwrap();
        let unwound = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let write = db.begin_write().unwrap();
            write.open_table(TABLE).unwrap().insert(1, 1).unwrap();
            panic!("unwinding through the transaction");
        }));
        assert!(unwound.is_err());
        assert!(db.mem.needs_repair());

        commit_one(&peer);
        commit_one(&db);
        assert!(!db.mem.needs_repair());
        assert!(
            Database::get_allocator_state_table(&db.mem)
                .unwrap()
                .is_some()
        );
    }

    /// The first transaction after a panic releases the repair latch, so its commit records the
    /// allocator state without making a peer rebuild it.
    #[test]
    fn a_transaction_releases_the_repair_a_panic_latched_without_a_peers_commit() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        let mut builder = Database::builder();
        builder.set_concurrency_mode(ConcurrencyMode::MultiWriter);
        let peer = builder.open(tmpfile.path()).unwrap();
        let unwound = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let write = db.begin_write().unwrap();
            write.open_table(TABLE).unwrap().insert(1, 1).unwrap();
            panic!("unwinding through the transaction");
        }));
        assert!(unwound.is_err());
        assert!(db.mem.needs_repair());

        commit_one(&db);
        assert!(!db.mem.needs_repair());
        peer.begin_write()
            .expect("the peer syncs to a commit that recorded the allocator state")
            .abort()
            .unwrap();
    }

    /// An open reads the file's length under the header lock, so a create holding that lock is
    /// waited for and its result read, rather than the file being sampled part way through it.
    /// Before, the length was sampled outside any lock: an empty file was called empty and a
    /// resized one not a redb database, however far the create had got by the time it was read
    #[test]
    fn an_open_waits_behind_a_creator_instead_of_calling_the_file_empty() {
        use std::io::{Read, Seek, SeekFrom, Write};
        use std::sync::mpsc;
        use std::thread;
        use std::time::Duration;

        // What the create will have produced by the time it releases the lock
        let initialized = crate::create_tempfile();
        drop(create(initialized.path(), ConcurrencyMode::MultiWriter));
        let mut bytes = Vec::new();
        File::open(initialized.path())
            .unwrap()
            .read_to_end(&mut bytes)
            .unwrap();

        let tmpfile = crate::create_tempfile();
        let probe = probe(tmpfile.path());
        // Stands in for a create holding the header lock across its initialization
        assert!(probe.try_lock_range(HEADER_LOCK).unwrap());

        let path = tmpfile.path().to_path_buf();
        let (opened, opens) = mpsc::channel();
        let opening = thread::spawn(move || {
            let mut builder = Database::builder();
            builder.set_concurrency_mode(ConcurrencyMode::MultiWriter);
            let result = builder.open(&path);
            opened.send(()).unwrap();
            result
        });

        assert!(
            opens.recv_timeout(Duration::from_millis(500)).is_err(),
            "the open read the file while the creator held the header lock"
        );
        // The create finishes: from here the file is a database, as it would be at the release.
        // Written through the handle that holds the lock, as a create writes under its own: on
        // Windows a byte-range lock is mandatory, and excludes writes from every other handle
        (&probe).seek(SeekFrom::Start(0)).unwrap();
        (&probe).write_all(&bytes).unwrap();
        probe.sync_all().unwrap();
        probe.unlock_range(HEADER_LOCK).unwrap();

        opens
            .recv_timeout(Duration::from_secs(10))
            .expect("the open never took the header lock");
        // A length sampled before the lock would still be the zero it read at the start
        opening
            .join()
            .unwrap()
            .expect("the open used a length it read before the creator finished");
    }

    /// A multi-writer open runs under the writer byte, so that two processes do not repair the
    /// same file at once and commit over each other
    #[test]
    fn a_multi_writer_open_waits_for_the_writer_byte() {
        use std::sync::mpsc;
        use std::thread;
        use std::time::Duration;

        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        // A commit recording nothing leaves the file for the next open to repair. The handle
        // stays open, since its close would record
        let mut write = db.begin_write().unwrap();
        write.skip_allocator_state_record();
        write.open_table(TABLE).unwrap().insert(1, 1).unwrap();
        write.commit().unwrap();

        let probe = probe(tmpfile.path());
        assert!(probe.try_lock_range(byte_range(WRITER_BYTE)).unwrap());
        let path = tmpfile.path().to_path_buf();
        let (repairing, repairs) = mpsc::channel();
        let opening = thread::spawn(move || {
            let mut builder = Database::builder();
            builder.set_concurrency_mode(ConcurrencyMode::MultiWriter);
            builder.set_repair_callback(move |_| {
                let _ = repairing.send(());
            });
            builder.open(&path).unwrap()
        });

        assert!(
            repairs.recv_timeout(Duration::from_millis(500)).is_err(),
            "the open repaired while the writer byte was held"
        );
        probe.unlock_range(byte_range(WRITER_BYTE)).unwrap();
        repairs
            .recv_timeout(Duration::from_secs(10))
            .expect("the repair runs once the byte is free");
        opening.join().unwrap();
    }

    /// A peer can commit while an open waits for the writer byte. The open reads the file only
    /// once it holds the byte, so it sees that commit and repairs nothing it recorded.
    #[test]
    fn a_multi_writer_open_reads_the_file_a_peer_committed_while_it_waited() {
        use std::sync::mpsc;
        use std::thread;
        use std::time::Duration;

        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        let mut write = db.begin_write().unwrap();
        write.skip_allocator_state_record();
        write.open_table(TABLE).unwrap().insert(1, 1).unwrap();
        write.commit().unwrap();
        // Holds the byte across the open, and records the allocator state when it commits
        let held = db.begin_write().unwrap();
        held.open_table(TABLE).unwrap().insert(2, 2).unwrap();

        let path = tmpfile.path().to_path_buf();
        let (repairing, repairs) = mpsc::channel();
        let (opened, opens) = mpsc::channel();
        let opening = thread::spawn(move || {
            let mut builder = Database::builder();
            builder.set_concurrency_mode(ConcurrencyMode::MultiWriter);
            builder.set_repair_callback(move |_| {
                let _ = repairing.send(());
            });
            let db = builder.open(&path).unwrap();
            opened.send(()).unwrap();
            db
        });

        assert!(
            opens.recv_timeout(Duration::from_millis(500)).is_err(),
            "the open did not wait for the byte"
        );
        held.commit().unwrap();
        opens
            .recv_timeout(Duration::from_secs(10))
            .expect("the open takes the byte the commit released");
        let peer = opening.join().unwrap();

        assert!(
            repairs.try_recv().is_err(),
            "the open repaired a file the commit had recorded"
        );
        // A repair from the header read before that commit would have committed over it
        let read = super::ReadableDatabase::begin_read(&peer).unwrap();
        let table = read.open_table(TABLE).unwrap();
        assert_eq!(
            crate::ReadableTable::get(&table, 2)
                .unwrap()
                .unwrap()
                .value(),
            2
        );
    }
    /// A multi-writer compaction ends with a commit recording the allocator state, for the next
    /// writer, in any process, to load rather than rebuild
    #[test]
    fn a_multi_writer_compaction_records_the_allocator_state() {
        let tmpfile = crate::create_tempfile();
        let mut db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        commit_one(&db);

        db.compact().unwrap();
        assert!(
            Database::get_allocator_state_table(&db.mem)
                .unwrap()
                .is_some()
        );
    }

    /// Held for the transaction and no longer, which is what lets the next writer in
    #[test]
    fn a_multi_writer_transaction_holds_the_byte_and_gives_it_back() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        let probe = probe(tmpfile.path());

        let write = db.begin_write().unwrap();
        assert!(
            !probe.try_lock_range(byte_range(WRITER_BYTE)).unwrap(),
            "the byte was free while a write transaction was open"
        );
        write.commit().unwrap();

        assert!(
            probe.try_lock_range(byte_range(WRITER_BYTE)).unwrap(),
            "the byte was still held after the transaction ended"
        );
        probe.unlock_range(byte_range(WRITER_BYTE)).unwrap();
    }

    /// This mode settles who writes at open instead, so the byte is held from then until the
    /// database closes. A transaction taking it again would convert that hold and drop what
    /// remained on release, so it holds the open's instead.
    #[test]
    fn a_single_writer_open_holds_the_byte_for_its_lifetime() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::SingleWriter);
        let probe = probe(tmpfile.path());

        assert!(
            !probe.try_lock_range(byte_range(WRITER_BYTE)).unwrap(),
            "the open did not take the writer byte"
        );
        commit_one(&db);
        assert!(
            !probe.try_lock_range(byte_range(WRITER_BYTE)).unwrap(),
            "a write transaction punctured the open's hold on the writer byte"
        );

        drop(db);
        assert!(
            probe.try_lock_range(byte_range(WRITER_BYTE)).unwrap(),
            "the byte was still held after the database closed"
        );
        probe.unlock_range(byte_range(WRITER_BYTE)).unwrap();
    }

    /// A transaction that outlives the database keeps the lock, and the deferred close writes
    /// the allocator state under it
    #[test]
    fn a_transaction_outliving_the_database_keeps_the_writer_lock() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::SingleWriter);
        let probe = probe(tmpfile.path());

        let write = db.begin_write().unwrap();
        drop(db);
        assert!(
            !probe.try_lock_range(byte_range(WRITER_BYTE)).unwrap(),
            "dropping the database released the byte from under a live transaction"
        );

        write.commit().unwrap();
        assert!(
            probe.try_lock_range(byte_range(WRITER_BYTE)).unwrap(),
            "the byte was still held after the deferred close"
        );
        probe.unlock_range(byte_range(WRITER_BYTE)).unwrap();
    }
}

/// The consistent byte, probed directly: a byte-range lock is invisible through the public API.
#[cfg(all(
    test,
    feature = "experimental-multiprocess",
    any(target_os = "linux", target_vendor = "apple", windows)
))]
mod consistent_byte_test {
    use super::{CONSISTENT_BYTE, ConcurrencyMode, Database, byte_range};
    use crate::backends::FileBackend;
    use crate::tree_store::file_backend::range_lock::RangeLock;
    use crate::{StorageBackend, TableDefinition};
    use std::fs::{File, OpenOptions};
    use std::path::Path;
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::sync::{Arc, Mutex};

    const TABLE: TableDefinition<u64, u64> = TableDefinition::new("x");

    fn builder(mode: ConcurrencyMode) -> crate::Builder {
        let mut builder = Database::builder();
        builder.set_concurrency_mode(mode);
        builder
    }

    /// A separate description, so its locks conflict with the database's exactly as another
    /// process's would
    fn probe(path: &Path) -> File {
        OpenOptions::new()
            .read(true)
            .write(true)
            .open(path)
            .unwrap()
    }

    /// True when some writer asserts the database is consistent
    fn held(probe: &File) -> bool {
        let taken = probe.try_lock_range(byte_range(CONSISTENT_BYTE)).unwrap();
        if taken {
            probe.unlock_range(byte_range(CONSISTENT_BYTE)).unwrap();
        }
        !taken
    }

    /// Power loss: once armed, every write silently does nothing, so the close writes neither
    /// the allocator record nor the clean-shutdown header
    #[derive(Debug)]
    struct CrashBackend {
        inner: FileBackend,
        dead: Arc<AtomicBool>,
    }

    impl StorageBackend for CrashBackend {
        fn len(&self) -> Result<u64, std::io::Error> {
            self.inner.len()
        }
        fn read(&self, offset: u64, out: &mut [u8]) -> Result<(), std::io::Error> {
            self.inner.read(offset, out)
        }
        fn set_len(&self, len: u64) -> Result<(), std::io::Error> {
            if self.dead.load(Ordering::SeqCst) {
                return Ok(());
            }
            self.inner.set_len(len)
        }
        fn sync_data(&self) -> Result<(), std::io::Error> {
            if self.dead.load(Ordering::SeqCst) {
                return Ok(());
            }
            self.inner.sync_data()
        }
        fn write(&self, offset: u64, data: &[u8]) -> Result<(), std::io::Error> {
            if self.dead.load(Ordering::SeqCst) {
                return Ok(());
            }
            self.inner.write(offset, data)
        }
    }

    /// Leaves the file as a crashed writer would: a committed transaction, no allocator record
    /// and the recovery flag still set, so the next open has to repair it. Built through a
    /// caller-supplied backend, which takes no locks, so nothing of this handle outlives it.
    fn dirty(path: &Path) {
        let file = OpenOptions::new()
            .read(true)
            .write(true)
            .open(path)
            .unwrap();
        let dead = Arc::new(AtomicBool::new(false));
        let db = Database::builder()
            .create_with_backend(CrashBackend {
                inner: FileBackend::new(file).unwrap(),
                dead: Arc::clone(&dead),
            })
            .unwrap();
        let write = db.begin_write().unwrap();
        {
            let mut table = write.open_table(TABLE).unwrap();
            table.insert(0, 0).unwrap();
        }
        write.commit().unwrap();
        // Nothing this handle does from here on reaches the file, its close included
        dead.store(true, Ordering::SeqCst);
        drop(db);
    }

    #[test]
    fn a_shared_writable_open_asserts_consistency_until_it_closes() {
        for mode in [ConcurrencyMode::SingleWriter, ConcurrencyMode::MultiWriter] {
            let tmpfile = crate::create_tempfile();
            let db = builder(mode).create(tmpfile.path()).unwrap();
            let probe = probe(tmpfile.path());
            assert!(held(&probe), "{mode:?} open did not take the byte");

            drop(db);
            assert!(!held(&probe), "{mode:?} close did not release the byte");
        }
    }

    /// The byte's whole purpose. `SHARED_WRITER_BYTE` is taken before recovery runs, so during a
    /// repair it says a writer is here while the file is still the one the last writer left. This
    /// byte is taken after, so it says nothing until the file is consistent.
    #[test]
    fn a_repairing_open_asserts_consistency_only_once_the_repair_is_done() {
        let tmpfile = crate::create_tempfile();
        dirty(tmpfile.path());

        let probe = Arc::new(Mutex::new(probe(tmpfile.path())));
        let during: Arc<Mutex<Option<bool>>> = Arc::new(Mutex::new(None));

        let db = {
            let probe = probe.clone();
            let during = during.clone();
            let mut builder = builder(ConcurrencyMode::MultiWriter);
            builder.set_repair_callback(move |_| {
                let mut during = during.lock().unwrap();
                if during.is_none() {
                    *during = Some(held(&probe.lock().unwrap()));
                }
            });
            builder.open(tmpfile.path()).unwrap()
        };

        assert_eq!(
            during.lock().unwrap().take(),
            Some(false),
            "the byte was held while the open was still repairing"
        );
        assert!(held(&probe.lock().unwrap()), "the open never took the byte");
        drop(db);
    }
}

/// The "active transaction range" locks a read transaction publishes, probed directly: a
/// byte-range lock is invisible through the public API
#[cfg(all(
    test,
    feature = "experimental-multiprocess",
    any(target_os = "linux", target_vendor = "apple", windows)
))]
mod active_transaction_test {
    use super::{ConcurrencyMode, Database, ReadableDatabase, TXN_BASE, WRITER_BYTE, byte_range};
    use crate::tree_store::HEADER_LOCK;
    use crate::tree_store::file_backend::range_lock::RangeLock;
    use crate::{Durability, SavepointError, SetDurabilityError, TableDefinition};
    use std::fs::{File, OpenOptions};
    use std::path::Path;

    const TABLE: TableDefinition<u64, u64> = TableDefinition::new("x");
    // The ids a freshly created database can have committed, with room for a few more commits
    const SEARCHED: std::ops::Range<u64> = 0..16;

    fn create(path: &Path, mode: ConcurrencyMode) -> Database {
        let mut builder = Database::builder();
        builder.set_concurrency_mode(mode);
        let db = builder.create(path).unwrap();
        let write = db.begin_write().unwrap();
        {
            let mut table = write.open_table(TABLE).unwrap();
            table.insert(0, 0).unwrap();
        }
        write.commit().unwrap();
        db
    }

    /// A separate description, so its locks conflict with the database's exactly as another
    /// process's would
    fn probe(path: &Path) -> File {
        OpenOptions::new()
            .read(true)
            .write(true)
            .open(path)
            .unwrap()
    }

    /// Probed exclusively, since the byte is held shared and another process may hold the same
    /// one. Reports the whole-file lock a exclusive-writer open holds as well, which is what
    /// `an_exclusive_writer_read_does_not_puncture_the_whole_file_lock` relies on
    fn is_held(probe: &File, id: u64) -> bool {
        let free = probe.try_lock_range(byte_range(TXN_BASE + id)).unwrap();
        if free {
            probe.unlock_range(byte_range(TXN_BASE + id)).unwrap();
        }
        !free
    }

    fn held_ids(probe: &File) -> Vec<u64> {
        SEARCHED.filter(|id| is_held(probe, *id)).collect()
    }

    #[test]
    fn a_read_transaction_locks_the_id_it_reads_until_it_ends() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        let probe = probe(tmpfile.path());
        assert!(held_ids(&probe).is_empty());

        let read = db.begin_read().unwrap();
        assert_eq!(
            held_ids(&probe).len(),
            1,
            "a read transaction locked {:?}",
            held_ids(&probe)
        );

        drop(read);
        assert!(
            held_ids(&probe).is_empty(),
            "a lock outlived the read transaction that took it"
        );
    }

    /// Two readers of one snapshot take the byte once, since it does not nest through a single
    /// file description: it is released only as the last of them ends
    #[test]
    fn concurrent_readers_of_one_snapshot_share_the_lock() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        let probe = probe(tmpfile.path());

        let first = db.begin_read().unwrap();
        let second = db.begin_read().unwrap();
        let held = held_ids(&probe);
        assert_eq!(held.len(), 1, "expected one active id: {held:?}");

        drop(first);
        assert!(is_held(&probe, held[0]), "released early");
        drop(second);
        assert!(!is_held(&probe, held[0]), "never released");
    }

    /// A read-only handle is a participant like any other: its snapshot is exactly what a
    /// writer in another process must not reclaim
    #[test]
    fn a_read_only_handle_locks_what_it_reads() {
        let tmpfile = crate::create_tempfile();
        // Dropped, so the read-only open is the only handle: it takes SHARED_READER_BYTE, and
        // the writer that created the file would otherwise hold the whole storage
        drop(create(tmpfile.path(), ConcurrencyMode::MultiWriter));
        let probe = probe(tmpfile.path());

        let mut builder = Database::builder();
        builder.set_concurrency_mode(ConcurrencyMode::MultiWriter);
        let db = builder.open_read_only(tmpfile.path()).unwrap();
        let read = db.begin_read().unwrap();
        assert_eq!(
            held_ids(&probe).len(),
            1,
            "a read-only handle locked {:?}",
            held_ids(&probe)
        );

        drop(read);
        assert!(held_ids(&probe).is_empty(), "the lock outlived the reader");
    }

    #[test]
    fn read_only_snapshot_survives_new_read_before_construction() {
        use super::{ReadTransaction, TransactionGuard};
        use crate::ReadableTable;

        for mode in [ConcurrencyMode::SingleWriter, ConcurrencyMode::MultiWriter] {
            let tmpfile = crate::create_tempfile();
            let writer = create(tmpfile.path(), mode);
            let db = Database::builder()
                .set_concurrency_mode(mode)
                .set_cache_size(0)
                .open_read_only(tmpfile.path())
                .unwrap();
            let (guard, root) =
                TransactionGuard::allocate_read(db.transaction_tracker.clone(), &db.mem).unwrap();

            for value in 1..4 {
                let write = writer.begin_write().unwrap();
                write.open_table(TABLE).unwrap().insert(0, value).unwrap();
                write.commit().unwrap();
            }
            // Another read observes the peer's commit before the first read is constructed.
            let latest = db.begin_read().unwrap();
            assert_eq!(
                latest
                    .open_table(TABLE)
                    .unwrap()
                    .get(0)
                    .unwrap()
                    .unwrap()
                    .value(),
                3
            );

            let read = ReadTransaction::new(db.mem.clone(), guard, root).unwrap();
            assert_eq!(
                read.open_table(TABLE)
                    .unwrap()
                    .get(0)
                    .unwrap()
                    .unwrap()
                    .value(),
                0
            );
        }
    }

    #[cfg(feature = "cache_metrics")]
    #[test]
    fn read_only_file_snapshots_preserve_memory_state_and_reuse_cache() {
        use super::TransactionGuard;
        use crate::ReadableTable;

        for mode in [ConcurrencyMode::SingleWriter, ConcurrencyMode::MultiWriter] {
            let tmpfile = crate::create_tempfile();
            let writer = create(tmpfile.path(), mode);
            let db = Database::builder()
                .set_concurrency_mode(mode)
                .open_read_only(tmpfile.path())
                .unwrap();
            let original_id = db.mem.get_last_committed_transaction_id().unwrap();
            let original_root = db.mem.get_data_root();
            let read = db.begin_read().unwrap();
            assert_eq!(
                read.open_table(TABLE)
                    .unwrap()
                    .get(0)
                    .unwrap()
                    .unwrap()
                    .value(),
                0
            );
            let cached_bytes = db.cache_stats().used_bytes();
            assert!(cached_bytes > 0);

            let (same, _) =
                TransactionGuard::allocate_read(db.transaction_tracker.clone(), &db.mem).unwrap();
            assert_eq!(same.id(), original_id);
            assert_eq!(db.cache_stats().used_bytes(), cached_bytes);

            let write = writer.begin_write().unwrap();
            write.open_table(TABLE).unwrap().insert(0, 1).unwrap();
            write.commit().unwrap();

            let (latest, _) =
                TransactionGuard::allocate_read(db.transaction_tracker.clone(), &db.mem).unwrap();
            assert!(latest.id() > original_id);
            assert_eq!(db.cache_stats().used_bytes(), 0);
            assert_eq!(
                db.mem.get_last_committed_transaction_id().unwrap(),
                original_id
            );
            assert_eq!(db.mem.get_data_root(), original_root);

            let current = db.begin_read().unwrap();
            assert_eq!(
                current
                    .open_table(TABLE)
                    .unwrap()
                    .get(0)
                    .unwrap()
                    .unwrap()
                    .value(),
                1
            );
            let cached_bytes = db.cache_stats().used_bytes();
            assert!(cached_bytes > 0);
            let (repeated, _) =
                TransactionGuard::allocate_read(db.transaction_tracker.clone(), &db.mem).unwrap();
            assert_eq!(repeated.id(), latest.id());
            assert_eq!(db.cache_stats().used_bytes(), cached_bytes);
            assert_eq!(
                read.open_table(TABLE)
                    .unwrap()
                    .get(0)
                    .unwrap()
                    .unwrap()
                    .value(),
                0
            );
        }
    }

    #[cfg(feature = "cache_metrics")]
    #[test]
    fn a_multi_writer_commit_keeps_its_read_cache() {
        use crate::ReadableTable;

        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        {
            let read = db.begin_read().unwrap();
            assert_eq!(
                read.open_table(TABLE)
                    .unwrap()
                    .get(0)
                    .unwrap()
                    .unwrap()
                    .value(),
                0
            );
        }

        let write = db.begin_write().unwrap();
        write.open_table(TABLE).unwrap().insert(1, 1).unwrap();
        write.commit().unwrap();
        let cached_bytes = db.cache_stats().used_bytes();
        assert!(cached_bytes > 0);

        let _read = db.begin_read().unwrap();
        assert_eq!(db.cache_stats().used_bytes(), cached_bytes);
    }

    /// A savepoint holds a read transaction live, so its snapshot stays active for as long as
    /// the savepoint does. In the shared mode that supports an ephemeral one
    #[test]
    fn an_ephemeral_savepoint_locks_the_snapshot_it_references() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::SingleWriter);
        let probe = probe(tmpfile.path());
        assert!(held_ids(&probe).is_empty());

        let write = db.begin_write().unwrap();
        let savepoint = write.ephemeral_savepoint().unwrap();
        write.commit().unwrap();
        assert_eq!(
            held_ids(&probe).len(),
            1,
            "a savepoint locked {:?}",
            held_ids(&probe)
        );

        drop(savepoint);
        assert!(
            held_ids(&probe).is_empty(),
            "the lock outlived the savepoint"
        );
    }

    /// A persistent savepoint bounds this process's reclamation, but takes no byte: the file
    /// names it, so every process protects it by syncing, and a byte would go on pinning the
    /// transaction after another process deleted the savepoint, until this one next synced
    #[test]
    fn a_persistent_savepoint_takes_no_active_transaction_byte() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        let probe = probe(tmpfile.path());

        // Nothing else references the savepoint's transaction, so the byte is the savepoint's to
        // take or leave
        let pinned = db.mem.get_last_committed_transaction_id().unwrap();
        let write = db.begin_write().unwrap();
        let savepoint = write.persistent_savepoint().unwrap();
        write.commit().unwrap();
        assert!(
            held_ids(&probe).is_empty(),
            "the savepoint locked {:?}",
            held_ids(&probe)
        );
        assert_eq!(
            db.transaction_tracker.oldest_local_referenced_transaction(),
            Some(pinned),
            "the savepoint stopped bounding this process's reclamation"
        );
        // The probe does see a byte where there is one to see
        let read = db.begin_read().unwrap();
        assert_eq!(held_ids(&probe).len(), 1);
        drop(read);
        assert!(held_ids(&probe).is_empty());

        let write = db.begin_write().unwrap();
        assert!(write.delete_persistent_savepoint(savepoint).unwrap());
        write.commit().unwrap();
        assert_eq!(
            db.transaction_tracker.oldest_local_referenced_transaction(),
            None,
            "the deleted savepoint kept its reference"
        );
        assert!(held_ids(&probe).is_empty());
    }

    /// The same for a savepoint this handle adopted rather than created: the open syncs the
    /// file's persistent savepoints, and references them without locking them either
    #[test]
    fn a_synced_persistent_savepoint_takes_no_active_transaction_byte() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        let write = db.begin_write().unwrap();
        write.persistent_savepoint().unwrap();
        write.commit().unwrap();
        drop(db);

        let probe = probe(tmpfile.path());
        assert!(
            held_ids(&probe).is_empty(),
            "the closed database left a lock"
        );

        let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        assert!(
            held_ids(&probe).is_empty(),
            "the open locked {:?} for the file's savepoint",
            held_ids(&probe)
        );
        assert!(
            db.transaction_tracker
                .oldest_local_referenced_transaction()
                .is_some(),
            "the open did not adopt the file's savepoint"
        );
        // The probe does see a byte where there is one to see
        let read = db.begin_read().unwrap();
        assert_eq!(held_ids(&probe).len(), 1);
        drop(read);
    }

    /// A non-durable commit exists only in this process's memory, so a shared mode refuses it
    #[test]
    fn a_shared_mode_refuses_durability_none() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::SingleWriter);

        let mut write = db.begin_write().unwrap();
        assert!(matches!(
            write.set_durability(Durability::None),
            Err(SetDurabilityError::NonDurableCommitUnsupported)
        ));
    }

    /// Every multi-writer commit records the allocator state, for the next writer, in any
    /// process, to load rather than rebuild: turning quick-repair off has no effect there
    #[test]
    fn a_multi_writer_commit_records_the_allocator_state_regardless() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);

        let mut write = db.begin_write().unwrap();
        write.set_quick_repair(false);
        write.open_table(TABLE).unwrap().insert(1, 1).unwrap();
        write.commit().unwrap();
        assert!(
            Database::get_allocator_state_table(&db.mem)
                .unwrap()
                .is_some()
        );
    }

    /// An ephemeral savepoint would be known to this process alone, and a persistent savepoint
    /// a peer creates could take its id, so multi-writer mode refuses it
    #[test]
    fn a_multi_writer_mode_refuses_an_ephemeral_savepoint() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);

        let write = db.begin_write().unwrap();
        assert!(matches!(
            write.ephemeral_savepoint(),
            Err(SavepointError::EphemeralSavepointUnsupported)
        ));
        write.persistent_savepoint().unwrap();
        write.commit().unwrap();
    }

    /// An interrupted repair may leave no allocator snapshot. The check records it so the next
    /// writer can load it without rebuilding again.
    #[test]
    fn an_integrity_check_records_a_peers_unrecorded_commit() {
        let tmpfile = crate::create_tempfile();
        let mut db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        let peer = Database::builder()
            .set_concurrency_mode(ConcurrencyMode::MultiWriter)
            .open(tmpfile.path())
            .unwrap();
        // A commit recording no allocator state, as a repair's does. The peer stays open, since
        // its close would record
        let mut write = peer.begin_write().unwrap();
        write.skip_allocator_state_record();
        write.open_table(TABLE).unwrap().insert(1, 1).unwrap();
        write.commit().unwrap();

        assert!(!db.check_integrity().unwrap());
        assert!(
            Database::get_allocator_state_table(&db.mem)
                .unwrap()
                .is_some()
        );
        drop(peer);
    }

    #[test]
    fn a_write_transaction_releases_the_writer_byte_and_slot() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        let probe = probe(tmpfile.path());

        let txn = db.begin_write().unwrap();
        assert!(db.transaction_tracker.holds_writer_byte());
        assert!(db.transaction_tracker.write_transaction_live());
        assert!(!probe.try_lock_range(byte_range(WRITER_BYTE)).unwrap());

        txn.abort().unwrap();
        assert!(!db.transaction_tracker.holds_writer_byte());
        assert!(!db.transaction_tracker.write_transaction_live());
        assert!(probe.try_lock_range(byte_range(WRITER_BYTE)).unwrap());
        probe.unlock_range(byte_range(WRITER_BYTE)).unwrap();
    }

    /// A leak another process left in the file is repaired, and the repair records the allocator
    /// state for the next writer to load
    #[test]
    fn an_integrity_check_repairs_a_leak_a_peer_recorded() {
        use crate::tree_store::{AllocationPolicy, PageAllocator, PageTracker};

        let tmpfile = crate::create_tempfile();
        let mut db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        let peer = Database::builder()
            .set_concurrency_mode(ConcurrencyMode::MultiWriter)
            .open(tmpfile.path())
            .unwrap();
        // A page the peer allocates and neither frees nor references: a leak its close's snapshot
        // records as allocated
        let allocator = PageAllocator::new(peer.mem.clone(), AllocationPolicy::Default);
        drop(allocator.allocate(64, &PageTracker::ignore()).unwrap());
        drop(allocator);
        drop(peer);

        assert!(!db.check_integrity().unwrap());
        // The repair recorded the allocator state, for a peer to load
        assert!(
            Database::get_allocator_state_table(&db.mem)
                .unwrap()
                .is_some()
        );
        assert!(db.check_integrity().unwrap());
    }

    /// A read-only participant sees what another process committed, not the header it read at open
    #[test]
    fn a_read_only_participant_picks_up_a_peers_commit() {
        let tmpfile = crate::create_tempfile();
        let writer = create(tmpfile.path(), ConcurrencyMode::MultiWriter);

        let mut builder = Database::builder();
        builder.set_concurrency_mode(ConcurrencyMode::MultiWriter);
        let reader = builder.open_read_only(tmpfile.path()).unwrap();
        let probe = probe(tmpfile.path());

        let read = reader.begin_read().unwrap();
        let before = held_ids(&probe);
        assert_eq!(before.len(), 1, "expected one active id: {before:?}");
        let before = before[0];
        drop(read);

        let write = writer.begin_write().unwrap();
        {
            let mut table = write.open_table(TABLE).unwrap();
            table.insert(1, 1).unwrap();
        }
        write.commit().unwrap();
        drop(writer);
        assert!(
            held_ids(&probe).is_empty(),
            "the closed writer left {:?} locked",
            held_ids(&probe)
        );

        let read = reader.begin_read().unwrap();
        let after = held_ids(&probe);
        assert_eq!(after.len(), 1, "expected one active id: {after:?}");
        assert!(
            after[0] > before,
            "the reader stayed at {before} after the writer committed"
        );
        drop(read);
    }

    /// A read-only participant takes the primary as recorded: choosing between the slots is a
    /// repairing writer's job, and a newer secondary it can see is a commit whose pages are not in
    /// the file yet, or one a repair has just rolled back
    #[test]
    fn a_read_only_participant_takes_the_primary_as_recorded() {
        use std::io::{Read, Write};

        let tmpfile = crate::create_tempfile();
        drop(create(tmpfile.path(), ConcurrencyMode::ExclusiveWriter));

        let mut builder = Database::builder();
        builder.set_concurrency_mode(ConcurrencyMode::MultiWriter);
        let reader = builder.open_read_only(tmpfile.path()).unwrap();

        // A newer commit whose pages this file never received: made in a copy, then its slot
        // spliced into this file's secondary slot
        let copy = crate::create_tempfile();
        std::fs::copy(tmpfile.path(), copy.path()).unwrap();
        let db = Database::open(copy.path()).unwrap();
        let write = db.begin_write().unwrap();
        {
            let mut table = write.open_table(TABLE).unwrap();
            table.insert(0, 1).unwrap();
        }
        write.commit().unwrap();
        drop(db);

        let header = |path: &Path| -> Vec<u8> {
            let mut header = vec![0u8; 320];
            File::open(path).unwrap().read_exact(&mut header).unwrap();
            header
        };
        let slot = |index: usize| 64 + index * 128..64 + (index + 1) * 128;
        let id = |header: &[u8], index: usize| {
            u64::from_le_bytes(header[slot(index)][104..112].try_into().unwrap())
        };
        let newer = header(copy.path());
        let mut spliced = header(tmpfile.path());
        let primary = usize::from(spliced[9] & 1);
        spliced[slot(1 - primary)].copy_from_slice(&newer[slot(usize::from(newer[9] & 1))]);
        // The god byte a repair leaves on a 1-phase history: recovery clear, 2-phase clear
        spliced[9] &= !6;
        OpenOptions::new()
            .write(true)
            .open(tmpfile.path())
            .unwrap()
            .write_all(&spliced)
            .unwrap();
        assert!(id(&spliced, 1 - primary) > id(&spliced, primary));
        let probe = probe(tmpfile.path());

        // Reloading into that state, and opening in it
        let read = reader.begin_read().unwrap();
        assert_eq!(
            held_ids(&probe),
            vec![id(&spliced, primary)],
            "the reader adopted a slot whose pages the file never received"
        );
        drop(read);
        let opened = builder.open_read_only(tmpfile.path()).unwrap();
        let read = opened.begin_read().unwrap();
        assert_eq!(held_ids(&probe), vec![id(&spliced, primary)]);
        drop(read);
    }

    /// Sharing the file forces 2-phase, whatever the transaction asked for
    #[test]
    fn a_shared_commit_is_two_phase() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);

        let mut write = db.begin_write().unwrap();
        write.set_two_phase_commit(false);
        {
            let mut table = write.open_table(TABLE).unwrap();
            table.insert(1, 1).unwrap();
        }
        write.commit().unwrap();

        assert!(
            db.mem.used_two_phase_commit(),
            "a shared commit published without the flush between the header writes"
        );
    }

    /// With no writer holding the file, an unclean one is refused: nothing is coming to repair it
    #[test]
    fn a_shared_read_refuses_an_unclean_file_with_no_writer() {
        use std::io::{Read, Seek, SeekFrom, Write};

        let tmpfile = crate::create_tempfile();
        drop(create(tmpfile.path(), ConcurrencyMode::MultiWriter));

        // The god byte's recovery-required bit, which a clean shutdown clears
        {
            let mut file = OpenOptions::new()
                .read(true)
                .write(true)
                .open(tmpfile.path())
                .unwrap();
            let mut byte = [0u8; 1];
            file.seek(SeekFrom::Start(9)).unwrap();
            file.read_exact(&mut byte).unwrap();
            byte[0] |= 2;
            file.seek(SeekFrom::Start(9)).unwrap();
            file.write_all(&byte).unwrap();
        }

        let mut builder = Database::builder();
        builder.set_concurrency_mode(ConcurrencyMode::MultiWriter);
        assert!(matches!(
            builder.open_read_only(tmpfile.path()),
            Err(crate::DatabaseError::RepairAborted)
        ));
    }

    /// A exclusive-writer open holds the whole file, which covers these bytes. Locking one and
    /// releasing it would punch a hole in that lock, so this mode locks nothing
    #[test]
    fn an_exclusive_writer_read_does_not_puncture_the_whole_file_lock() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::ExclusiveWriter);
        let probe = probe(tmpfile.path());

        drop(db.begin_read().unwrap());

        assert_eq!(
            held_ids(&probe),
            SEARCHED.collect::<Vec<_>>(),
            "a read transaction punctured the whole-file lock"
        );
    }

    /// The floor is the older of this process's oldest read and a peer's pin, and a peer's pin
    /// only matters below the former, which is where the scan looks
    #[test]
    fn the_oldest_active_transaction_is_the_lower_of_ours_and_a_peers() {
        for mode in [ConcurrencyMode::SingleWriter, ConcurrencyMode::MultiWriter] {
            let tmpfile = crate::create_tempfile();
            let db = create(tmpfile.path(), mode);
            let scan = |local| {
                let header_lock = db.mem.lock_header_exclusive().unwrap();
                db.mem
                    .oldest_active_transaction(local, &header_lock)
                    .unwrap()
            };
            let first = db.mem.get_last_committed_transaction_id().unwrap();
            assert_eq!(scan(None), None);

            // A peer pins the last committed transaction, as its read of the header would
            let probe = probe(tmpfile.path());
            probe
                .lock_shared_range(byte_range(TXN_BASE + first.raw_id()))
                .unwrap();
            assert_eq!(scan(None), Some(first), "{mode:?}");

            let write = db.begin_write().unwrap();
            {
                let mut table = write.open_table(TABLE).unwrap();
                table.insert(1, 1).unwrap();
            }
            write.commit().unwrap();
            let second = db.mem.get_last_committed_transaction_id().unwrap();
            assert!(first < second);
            assert_eq!(scan(Some(second)), Some(first), "{mode:?}");
            probe
                .unlock_range(byte_range(TXN_BASE + first.raw_id()))
                .unwrap();
            assert_eq!(scan(Some(second)), Some(second), "{mode:?}");
        }
    }

    /// A transaction that is lent the exclusive header lock syncs to a peer's commit under that
    /// lock, savepoints included, rather than trying to take the lock a second time.
    #[test]
    fn a_lent_header_lock_covers_syncing_to_a_peers_commit() {
        use std::sync::mpsc;
        use std::thread;
        use std::time::Duration;

        fn open(path: &Path) -> Database {
            let mut builder = Database::builder();
            builder.set_concurrency_mode(ConcurrencyMode::MultiWriter);
            builder.open(path).unwrap()
        }

        let tmpfile = crate::create_tempfile();
        create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        let peer = open(tmpfile.path());
        // The handle syncs to the commit under holds it took itself, as compaction does, in a
        // thread, so that taking the header lock again fails rather than hangs
        let path = tmpfile.path().to_path_buf();
        let (opened, wait_for_open) = mpsc::channel();
        let (committed, wait_for_commit) = mpsc::channel();
        let (synced, wait_for_sync) = mpsc::channel();
        let syncing = thread::spawn(move || {
            let db = open(&path);
            opened.send(()).unwrap();
            wait_for_commit.recv().unwrap();
            let writer_lock = db.mem.lock_writer().unwrap();
            let header_lock = db.mem.lock_header_exclusive().unwrap();
            let txn = db
                .begin_write_with(Some(&writer_lock), Some(&header_lock))
                .unwrap();
            let savepoints: Vec<u64> = txn.list_persistent_savepoints().unwrap().collect();
            txn.abort().unwrap();
            drop(header_lock);
            drop(writer_lock);
            synced.send(savepoints).unwrap();
        });
        wait_for_open.recv().unwrap();
        let txn = peer.begin_write().unwrap();
        let peers_savepoint = txn.persistent_savepoint().unwrap();
        txn.commit().unwrap();
        // Closed first: otherwise its close would wait on the writer byte that a hung
        // transaction still holds
        drop(peer);
        committed.send(()).unwrap();

        let savepoints = wait_for_sync
            .recv_timeout(Duration::from_secs(10))
            .expect("the transaction took the header lock its caller holds");
        assert_eq!(savepoints, vec![peers_savepoint]);
        syncing.join().unwrap();
    }

    /// A transaction that is lent the exclusive header lock begins and commits under it, and
    /// leaves the lock held afterwards.
    #[test]
    fn a_lent_header_hold_outlives_the_commit() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        let probe = probe(tmpfile.path());

        let header_lock = db.mem.lock_header_exclusive().unwrap();
        let txn = db.begin_write_with(None, Some(&header_lock)).unwrap();
        txn.open_table(TABLE).unwrap().insert(1, 1).unwrap();
        txn.commit_with(Some(&header_lock)).unwrap();
        assert!(!probe.try_lock_shared_range(HEADER_LOCK).unwrap());

        drop(header_lock);
        assert!(probe.try_lock_shared_range(HEADER_LOCK).unwrap());
        probe.unlock_range(HEADER_LOCK).unwrap();
    }

    /// A transaction lent the writer byte leaves it held when it ends
    #[test]
    fn a_lent_writer_byte_outlives_the_transaction() {
        let tmpfile = crate::create_tempfile();
        let db = create(tmpfile.path(), ConcurrencyMode::MultiWriter);
        let probe = probe(tmpfile.path());

        let writer_lock = db.mem.lock_writer().unwrap();
        let txn = db.begin_write_with(Some(&writer_lock), None).unwrap();
        txn.abort().unwrap();
        assert!(!probe.try_lock_range(byte_range(WRITER_BYTE)).unwrap());

        drop(writer_lock);
        assert!(probe.try_lock_range(byte_range(WRITER_BYTE)).unwrap());
        probe.unlock_range(byte_range(WRITER_BYTE)).unwrap();
    }
}