liminal/durability/

store.rs

use haematite::{ApiError, Database, DatabaseConfig, Event, EventStore};

use std::path::Path;
use std::sync::Arc;

use super::DurabilityError;

use tempfile::TempDir;

/// Entry read from a durable haematite stream.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredEntry {
    /// Opaque stored payload bytes.
    pub payload: Vec<u8>,
    /// Sequence number assigned by the stream.
    pub sequence: u64,
    /// Store timestamp associated with the entry.
    pub timestamp: u64,
}

/// Direct durability surface matching haematite's append/read/cas/scan API.
#[async_trait::async_trait]
pub trait DurableStore: std::fmt::Debug + Send + Sync {
    /// Appends `payload` to `stream_key` if `expected_seq` matches the stream head.
    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError>;

    /// Reads entries from `stream_key` beginning at `offset`, up to `limit` entries.
    async fn read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError>;

    /// Appends `payloads` to `stream_key` in order, the first at `expected_seq`,
    /// and returns the stream's next sequence (`expected_seq + payloads.len()`).
    ///
    /// A store that can commit several events in one write does so, and then
    /// either every payload is appended or none is. This default appends them
    /// one at a time: a failure part way leaves the leading payloads appended,
    /// so a caller that retries must read the stream head first.
    async fn append_all(
        &self,
        stream_key: &str,
        payloads: Vec<Vec<u8>>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        let mut next = expected_seq;
        for payload in payloads {
            let assigned = self.append(stream_key, payload, next).await?;
            if assigned != next {
                return Err(DurabilityError::SequenceConflict {
                    expected: next,
                    actual: assigned,
                });
            }
            next = next.checked_add(1).ok_or_else(|| {
                DurabilityError::ConfigError(format!(
                    "stream {stream_key} sequence overflowed during a multi-event append"
                ))
            })?;
        }
        Ok(next)
    }

    /// Reads exactly the event at `sequence` without traversing its suffix.
    async fn read_at(
        &self,
        stream_key: &str,
        sequence: u64,
    ) -> Result<Option<StoredEntry>, DurabilityError> {
        Ok(self
            .read_from(stream_key, sequence, 1)
            .await?
            .into_iter()
            .next())
    }

    /// Atomically replaces a stored numeric value if it equals `old_value`.
    ///
    /// An `old_value` of `0` matches a key that is currently *absent* as well as
    /// one explicitly stored as `0`: a fresh cursor is created on its first
    /// checkpoint without a prior write. See [`HaematiteStore::cas`] for how this
    /// "absent == 0" contract is preserved atomically over the real engine.
    async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError>;

    /// Reads a numeric value previously updated through compare-and-swap.
    async fn read_value(&self, key: &str) -> Result<Option<u64>, DurabilityError>;

    /// Scans entries by store prefix.
    async fn scan(&self, prefix: &str) -> Result<Vec<StoredEntry>, DurabilityError>;

    /// Flushes buffered writes so completed durable operations are persisted.
    ///
    /// # Errors
    /// Returns [`DurabilityError`] when the underlying store cannot complete the flush.
    async fn flush(&self) -> Result<(), DurabilityError>;
}

/// `DurableStore` implementation that delegates directly to haematite's `EventStore`.
///
/// The real [`EventStore`] is synchronous (every call blocks on the owning
/// shard actor's reply), so each `async` method below completes on its first
/// poll. The synchronous bridge in [`super::bridge`] relies on exactly that.
#[derive(Clone, Debug)]
pub struct HaematiteStore {
    event_store: Arc<EventStore>,
}

impl HaematiteStore {
    /// Wraps a haematite `EventStore` handle.
    #[must_use]
    pub const fn new(event_store: Arc<EventStore>) -> Self {
        Self { event_store }
    }

    /// Reads the half-open key window `[offset, offset + limit)` from one
    /// stream, or `None` when the window did not fill.
    ///
    /// `None` is not "empty": it is "this window cannot answer on its own",
    /// and the caller must fall through to the unbounded engine read. A window
    /// short by even one row may be short because the stream ended, because
    /// history was compacted, or because an entry inside it expired, and only
    /// the engine's own read distinguishes those.
    ///
    /// `limit` must be nonzero; a zero limit has no window to fill and is the
    /// caller's fall-through case.
    fn bounded_page(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Option<Vec<StoredEntry>>, DurabilityError> {
        const TIMESTAMP_WIDTH: usize = std::mem::size_of::<u64>();

        // Engine keys are 1-based; the public API is 0-based.
        let Some(engine_from) = offset.checked_add(1) else {
            return Ok(None);
        };
        let Some(engine_end) = u64::try_from(limit)
            .ok()
            .and_then(|limit| engine_from.checked_add(limit))
        else {
            return Ok(None);
        };
        let key = stream_key.as_bytes();
        let from = haematite::encode_stream_key(key, engine_from);
        let to = haematite::encode_stream_key(key, engine_end);
        let entries = self
            .event_store
            .database()
            .range_routed(key, &from, &to)
            .map_err(ApiError::from)
            .map_err(DurabilityError::from)?;
        if entries.len() != limit {
            return Ok(None);
        }

        let mut page = Vec::with_capacity(entries.len());
        for (encoded_key, value) in entries {
            let Some((decoded_key, engine_sequence)) = haematite::decode_stream_key(&encoded_key)
            else {
                return Err(DurabilityError::StoreError(ApiError::CorruptEvent(
                    format!("paged read key does not encode an event for stream {stream_key}"),
                )));
            };
            if decoded_key != key {
                return Err(DurabilityError::StoreError(ApiError::CorruptEvent(
                    format!("paged read key does not encode stream {stream_key}"),
                )));
            }
            let sequence = engine_sequence.checked_sub(1).ok_or_else(|| {
                DurabilityError::StoreError(ApiError::CorruptEvent(format!(
                    "paged read event key has zero seq for stream {stream_key}"
                )))
            })?;
            let Some(timestamp_bytes) = value.get(..TIMESTAMP_WIDTH) else {
                return Err(DurabilityError::StoreError(ApiError::CorruptEvent(
                    format!(
                        "paged read event value is shorter than its timestamp for stream {stream_key}"
                    ),
                )));
            };
            let timestamp = u64::from_be_bytes(timestamp_bytes.try_into().map_err(|_| {
                DurabilityError::StoreError(ApiError::CorruptEvent(format!(
                    "paged read event timestamp has the wrong width for stream {stream_key}"
                )))
            })?);
            let Some(payload) = value.get(TIMESTAMP_WIDTH..) else {
                return Err(DurabilityError::StoreError(ApiError::CorruptEvent(
                    format!("paged read event has no payload boundary for stream {stream_key}"),
                )));
            };
            page.push(StoredEntry {
                payload: payload.to_vec(),
                sequence,
                timestamp,
            });
        }
        Ok(Some(page))
    }
}

#[async_trait::async_trait]
impl DurableStore for HaematiteStore {
    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        // Contract bridge: liminal's `DurableStore::append` returns the *assigned
        // event sequence* (0-based position of the just-appended event), which is
        // exactly `expected_seq` for a single append. The real `EventStore::append`
        // instead returns the stream's new next-sequence (`expected_seq + 1`), so
        // subtract one to recover the assigned seq. A `0` next-seq is impossible
        // after a successful single append, so the `checked_sub` cannot saturate
        // silently; if it ever did the engine returned a contract-violating value.
        let next_seq = self
            .event_store
            .append(stream_key.as_bytes(), &payload, expected_seq)
            .map_err(DurabilityError::from)?;
        next_seq.checked_sub(1).ok_or_else(|| {
            DurabilityError::StoreError(ApiError::CorruptEvent(format!(
                "append returned next-seq 0 for stream {stream_key}"
            )))
        })
    }

    async fn append_all(
        &self,
        stream_key: &str,
        payloads: Vec<Vec<u8>>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        let borrowed: Vec<&[u8]> = payloads.iter().map(Vec::as_slice).collect();
        self.event_store
            .append_batch(stream_key.as_bytes(), &borrowed, expected_seq)
            .map_err(DurabilityError::from)
    }

    async fn read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        // `EventStore::read_from` applies no limit: it materialises every event
        // with seq >= offset, key and value copied across the shard-actor
        // boundary, and truncating afterwards throws that work away. Paged
        // replay therefore costs O(N^2) engine rows to deliver N (#60).
        //
        // Ask the engine for the page instead. Event keys are
        // `stream_key || 0x00 || seq.to_be_bytes()` (haematite 0.8.1
        // `api/event_store.rs:375`), so byte order is sequence order and a
        // half-open key window names exactly one page. `range_routed` routes on
        // the stream key — the same co-location `EventStore` uses for its own
        // reads — and merges committed tree with WAL buffer, which is the
        // identical mechanism behind the unbounded read (`db.rs:212`).
        //
        // A FULL window is the same answer the unbounded read gave: it holds
        // `limit` live events, and key order makes those exactly the first
        // `limit` events at or after `offset`. Anything SHORT falls through to
        // the unbounded read, so the two answers the window cannot settle by
        // itself stay the engine's own: the `HistoryCompacted` verdict at
        // `offset == 0`, and the case where expiry or compaction leaves a hole
        // inside the window. The fall-through costs a suffix scan only where
        // the suffix is already shorter than a page — the end-of-stream read
        // that terminates every walk.
        if limit > 0 {
            if let Some(page) = self.bounded_page(stream_key, offset, limit)? {
                account_engine_read(page.len(), false);
                return Ok(page);
            }
        }
        let mut events = self
            .event_store
            .read_from(stream_key.as_bytes(), offset)
            .map_err(DurabilityError::from)?;
        account_engine_read(events.len(), true);
        events.truncate(limit);
        Ok(events.into_iter().map(StoredEntry::from).collect())
    }

    async fn read_at(
        &self,
        stream_key: &str,
        sequence: u64,
    ) -> Result<Option<StoredEntry>, DurabilityError> {
        const TIMESTAMP_WIDTH: usize = std::mem::size_of::<u64>();

        let engine_sequence = sequence.checked_add(1).ok_or_else(|| {
            DurabilityError::StoreError(ApiError::CorruptEvent(format!(
                "point read sequence overflow for stream {stream_key}"
            )))
        })?;
        let event_key = haematite::encode_stream_key(stream_key.as_bytes(), engine_sequence);
        let Some(value) = self
            .event_store
            .database()
            .get_routed(stream_key.as_bytes(), &event_key)
            .map_err(ApiError::from)
            .map_err(DurabilityError::from)?
        else {
            return Ok(None);
        };
        let Some(timestamp_bytes) = value.get(..TIMESTAMP_WIDTH) else {
            return Err(DurabilityError::StoreError(ApiError::CorruptEvent(
                format!(
                    "point-read event value is shorter than its timestamp for stream {stream_key}"
                ),
            )));
        };
        let timestamp = u64::from_be_bytes(timestamp_bytes.try_into().map_err(|_| {
            DurabilityError::StoreError(ApiError::CorruptEvent(format!(
                "point-read event timestamp has the wrong width for stream {stream_key}"
            )))
        })?);
        let Some(payload) = value.get(TIMESTAMP_WIDTH..) else {
            return Err(DurabilityError::StoreError(ApiError::CorruptEvent(
                format!("point-read event has no payload boundary for stream {stream_key}"),
            )));
        };
        Ok(Some(StoredEntry {
            payload: payload.to_vec(),
            sequence,
            timestamp,
        }))
    }

    async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
        // Preserve liminal's "absent == 0" cursor contract faithfully over an
        // engine that distinguishes `None` (absent) from `Some(0)` (a stored
        // zero). The invariant that makes the mapping below correct: we NEVER
        // persist a physical zero, so a logical value of 0 and physical absence
        // always coincide.
        //
        // A `cas` whose target `new_value` is 0 must therefore write nothing — it
        // only asserts the precondition. This is reachable as `cas(0, 0)` (a
        // cursor checkpoint at offset 0; offsets are monotonic so they never CAS
        // down to 0 from a higher value). Were we instead to let it store a
        // physical zero, the *next* `cas(0, n)` — mapped to expect-absent `None`
        // — would wrongly fail against the now-present key and permanently stall
        // the cursor. Asserting via a read is race-free here precisely because no
        // value is written, so there is no lost-update window.
        if new_value == 0 {
            return self
                .event_store
                .read_value(key.as_bytes())
                .map_err(DurabilityError::from)?
                .map_or(Ok(()), |stored| {
                    Err(DurabilityError::CursorRegression {
                        stored,
                        attempted: old_value,
                    })
                });
        }
        // With a physical zero never stored, `old_value == 0` is exactly the
        // expect-absent expectation. Any other `old_value` maps to `Some(_)`.
        // This is a single CAS routed to the owning shard actor, where read,
        // compare, and write run with no interleaving point (haematite's
        // `ShardActor::cas`) — the engine's atomicity is preserved end to end.
        let expected = if old_value == 0 {
            None
        } else {
            Some(old_value)
        };
        self.event_store
            .cas(key.as_bytes(), expected, new_value)
            .map_err(DurabilityError::from)
    }

    async fn read_value(&self, key: &str) -> Result<Option<u64>, DurabilityError> {
        self.event_store
            .read_value(key.as_bytes())
            .map_err(DurabilityError::from)
    }

    async fn scan(&self, prefix: &str) -> Result<Vec<StoredEntry>, DurabilityError> {
        // The real `scan` predicate yields stream *metadata* (key + next_seq),
        // not events. Liminal's contract is to return the events of every stream
        // whose key matches `prefix`, so collect the matching stream keys, then
        // read each stream's full event list and flatten the results.
        let prefix_bytes = prefix.as_bytes().to_vec();
        let matches = self
            .event_store
            .scan(|meta| meta.stream_key.starts_with(&prefix_bytes))
            .map_err(DurabilityError::from)?;
        let mut entries = Vec::new();
        for stream in matches {
            let events = self
                .event_store
                .read(&stream.stream_key)
                .map_err(DurabilityError::from)?;
            entries.extend(events.into_iter().map(StoredEntry::from));
        }
        Ok(entries)
    }

    async fn flush(&self) -> Result<(), DurabilityError> {
        self.event_store.flush().map_err(DurabilityError::from)
    }
}

/// Drop shell enforcing "close the store, then remove its directory" as
/// explicit code rather than field declaration order.
///
/// Declaration order alone cannot express the unwind case: if dropping the
/// store panics (a haematite worker failing to join), Rust would still drop
/// the remaining fields during the unwind and remove the directory under
/// possibly-live workers. This `Drop` drops the store inside `catch_unwind`;
/// on unwind it DISARMS the directory guard — the directory is deliberately
/// leaked, because visible residue is diagnosable while removal under live
/// workers is filesystem corruption — logs the leaked path, and re-raises the
/// panic. On the clean path the directory is removed after the store, HERE,
/// by an explicit [`TempDir::close`] whose error is logged.
///
/// The explicitness is the point. Letting the `TempDir` field drop instead
/// would remove the directory via `tempfile`'s own `Drop`, which is
/// `let _ = remove_dir_all(..)` — the `io::Result` is discarded, so a removal
/// that FAILED would be indistinguishable from one that succeeded and this
/// doc's "the directory is removed" would be a claim no code could check.
/// `close()` returns that error; the clean path reports it and leaves the
/// residue where the log says it is. It never panics (a `Drop` that unwinds
/// during another unwind aborts the process) and never masks: a failure to
/// remove is a durability fact, not something to swallow.
///
/// Both fields are `Option` only so `drop` can move them out; they are `Some`
/// for the shell's entire life outside `drop`.
#[derive(Debug)]
struct EphemeralGuard<S> {
    store: Option<S>,
    dir: Option<TempDir>,
}

impl<S> Drop for EphemeralGuard<S> {
    fn drop(&mut self) {
        let store = self.store.take();
        // AssertUnwindSafe: the closure owns everything it touches (the moved
        // store), and the unwind path below observes no state the panicking
        // drop could have left broken — it only disarms the guard and re-raises.
        let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || drop(store)));
        if let Err(panic) = outcome {
            if let Some(dir) = self.dir.take() {
                let leaked = dir.keep();
                tracing::error!(
                    path = %leaked.display(),
                    "ephemeral store drop panicked; leaking its directory rather than \
                     removing it under possibly-live database workers"
                );
            }
            std::panic::resume_unwind(panic);
        }
        // Clean path: the store is closed, its workers are joined and the
        // writer lock is released, so removing the directory now is safe — and
        // removing it EXPLICITLY is what makes a failure sayable.
        if let Some(dir) = self.dir.take() {
            let path = dir.path().to_path_buf();
            if let Err(error) = dir.close() {
                tracing::error!(
                    path = %path.display(),
                    %error,
                    "ephemeral store directory removal failed; residue remains at the \
                     logged path"
                );
            }
        }
    }
}

/// Exclusive-ownership ephemeral durable store: the sole owner of both the
/// haematite database and the temporary directory that backs it.
///
/// [`HaematiteStore::new`] takes a *caller-supplied* `Arc<EventStore>`, so a
/// clone of that inner handle can outlive any guard placed merely beside it —
/// field declaration order proves nothing across that `Arc` boundary. This
/// wrapper instead owns the database outright: [`open_ephemeral`] constructs the
/// inner `Arc` itself, this type never exposes it (no getter) and is deliberately
/// **not `Clone`**, so the only handle a caller can hold is an
/// `Arc<dyn DurableStore>` over the whole wrapper. When the last such clone
/// drops, the [`EphemeralGuard`] drops the store FIRST — the database closes,
/// its shard actors join and the data-dir writer lock releases on fd close —
/// and only then removes the directory, logging the error if that removal
/// fails; if closing the database panics, the directory is deliberately leaked
/// instead (see [`EphemeralGuard`]).
#[derive(Debug)]
pub struct EphemeralHaematiteStore {
    guard: EphemeralGuard<HaematiteStore>,
}

impl EphemeralHaematiteStore {
    /// Takes an already-open ephemeral `Database` and the temporary directory it
    /// was opened under, becoming their single exclusive owner.
    ///
    /// The inner `Arc<EventStore>` is created here and never leaves this type, so
    /// no caller-supplied clone of it can exist to defeat the drop ordering.
    /// `ephemeral_dir` must be the directory `database` lives in and must have
    /// been created before the database was opened (so a failed open removed it
    /// via the guard's `Drop`, before this constructor was ever reached).
    fn new(database: Database, ephemeral_dir: TempDir) -> Self {
        Self {
            guard: EphemeralGuard {
                store: Some(HaematiteStore::new(Arc::new(EventStore::new(database)))),
                dir: Some(ephemeral_dir),
            },
        }
    }

    /// Store handle behind the guard's teardown-only `Option`.
    ///
    /// `None` exists only inside [`EphemeralGuard::drop`], which cannot overlap
    /// a `&self` call, so this error is unreachable by construction — it is a
    /// typed refusal in place of a panic the workspace forbids, not a state a
    /// caller can produce.
    fn store(&self) -> Result<&HaematiteStore, DurabilityError> {
        self.guard
            .store
            .as_ref()
            .ok_or(DurabilityError::EphemeralStoreDetached)
    }

    /// Path of the guarding temporary directory, for lifecycle assertions only.
    #[cfg(test)]
    pub(crate) fn ephemeral_dir_path(&self) -> Option<&Path> {
        self.guard.dir.as_ref().map(TempDir::path)
    }
}

#[async_trait::async_trait]
impl DurableStore for EphemeralHaematiteStore {
    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.store()?
            .append(stream_key, payload, expected_seq)
            .await
    }

    async fn append_all(
        &self,
        stream_key: &str,
        payloads: Vec<Vec<u8>>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.store()?
            .append_all(stream_key, payloads, expected_seq)
            .await
    }

    async fn read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        self.store()?.read_from(stream_key, offset, limit).await
    }

    async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
        self.store()?.cas(key, old_value, new_value).await
    }

    async fn read_value(&self, key: &str) -> Result<Option<u64>, DurabilityError> {
        self.store()?.read_value(key).await
    }

    async fn scan(&self, prefix: &str) -> Result<Vec<StoredEntry>, DurabilityError> {
        self.store()?.scan(prefix).await
    }

    async fn flush(&self) -> Result<(), DurabilityError> {
        self.store()?.flush().await
    }
}

/// Opens a self-owning ephemeral haematite store under a fresh temporary
/// directory below the system temp dir.
///
/// The directory is created BEFORE [`Database::create`], so every failure path —
/// including a haematite open/create error — removes it when the guard drops on
/// the error return; the returned store owns the guard on success. The database
/// is created directly in the (empty) temporary directory: haematite's `create`
/// accepts an existing empty dir and, on failure, removes only a directory *it*
/// created, never this pre-existing guard dir (haematite 0.4.1
/// `db/startup.rs`), so the `TempDir` is the sole owner of directory lifetime on
/// every path.
///
/// # Errors
/// Returns [`DurabilityError::EphemeralStoreOpen`] if haematite cannot create the
/// database; the temporary directory is already removed when this returns.
pub fn open_ephemeral(shard_count: usize) -> Result<EphemeralHaematiteStore, DurabilityError> {
    open_ephemeral_in(ephemeral_tempdir(None)?, shard_count)
}

/// TEST SEAM: [`open_ephemeral`] with the temporary directory placed under
/// `root` instead of the system temp dir.
///
/// Rooting lets construction gates assert on an isolated directory instead of
/// scanning the shared temp dir. Same lifecycle contract as
/// [`open_ephemeral`] — the store owns and removes its directory; `root` must
/// already exist and must outlive the store.
///
/// That last requirement is why this is NOT a production API: the store's
/// exclusive ownership of its directory (the D3 invariant) says nothing about
/// the PARENT — a caller rooting the store inside a directory they own via
/// their own guard can drop that guard while the store is live, deleting the
/// database out from under its running workers. A general rooted API would
/// need a root-ownership token so parent cleanup cannot outrun the store;
/// that is deferred until a real embedder need arrives. Until then the
/// function is gated to tests (`cfg(test)` in this crate, the default-off
/// `test-support` feature for downstream test harnesses).
///
/// # Errors
/// Returns [`DurabilityError::EphemeralStoreOpen`] if the directory cannot be
/// created under `root` or haematite cannot create the database; no residue
/// remains under `root` when this returns an error.
#[cfg(any(test, feature = "test-support"))]
pub fn open_ephemeral_rooted(
    root: &Path,
    shard_count: usize,
) -> Result<EphemeralHaematiteStore, DurabilityError> {
    open_ephemeral_in(ephemeral_tempdir(Some(root))?, shard_count)
}

/// Creates the guard directory for an ephemeral store, under `root` when given
/// and under the system temp dir otherwise.
fn ephemeral_tempdir(root: Option<&Path>) -> Result<TempDir, DurabilityError> {
    let mut builder = tempfile::Builder::new();
    builder.prefix("liminal-durability-");
    root.map_or_else(|| builder.tempdir(), |root| builder.tempdir_in(root))
        .map_err(|error| {
            DurabilityError::EphemeralStoreOpen(format!(
                "could not create temporary directory: {error}"
            ))
        })
}

/// Opens an ephemeral store inside an already-created guard directory.
///
/// Split out so the guard exists before `Database::create` and so lifecycle
/// tests can inject an open failure into a directory they pre-populated.
fn open_ephemeral_in(
    ephemeral_dir: TempDir,
    shard_count: usize,
) -> Result<EphemeralHaematiteStore, DurabilityError> {
    let database = Database::create(DatabaseConfig {
        data_dir: ephemeral_dir.path().to_path_buf(),
        shard_count,
        distributed: None,
        executor_threads: None,
        // haematite 0.8.3 requires this field and refuses `None` at validation;
        // `Unlimited` is the crate's explicit spelling of the pre-budget
        // (0.8.1) behaviour, said out loud — the behaviour-preserving choice
        // for a dependency hop. Selecting a real byte ceiling is a deployment
        // decision (box size, shard count, leaf weight) that belongs to its
        // own measured lane, not to this bump.
        node_cache_budget: Some(haematite::NodeCacheBudget::Unlimited),
    })
    .map_err(|error| DurabilityError::EphemeralStoreOpen(error.to_string()))?;
    Ok(EphemeralHaematiteStore::new(database, ephemeral_dir))
}

/// Engine-read accounting for the paged-read shape (#60).
///
/// Counts what the ENGINE handed back, which is the quantity the page limit is
/// supposed to bound. A `DurableStore` decorator cannot see it: by the time a
/// wrapper observes the result it has already been cut to `limit`, so the
/// difference between "read one page" and "read the whole suffix and throw it
/// away" is invisible from outside this type.
#[cfg(test)]
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(crate) struct EngineReadAccounting {
    /// `read_from` calls made while the guard was live.
    pub(crate) calls: usize,
    /// Entries the engine returned, summed before any truncation to `limit`.
    pub(crate) engine_entries: usize,
    /// Calls that fell through to the unbounded engine read.
    pub(crate) unbounded_calls: usize,
    /// Set when a counter would have wrapped; a saturated count is never a pin.
    pub(crate) counter_overflow_observed: bool,
}

#[cfg(test)]
std::thread_local! {
    static ENGINE_READ_ACCOUNTING: std::cell::RefCell<Option<EngineReadAccounting>> =
        const { std::cell::RefCell::new(None) };
}

/// Scopes engine-read accounting to one thread and one measured region.
///
/// `!Send` so accounting cannot straddle a thread boundary and report a sum
/// whose addends came from different call stacks.
#[cfg(test)]
pub(crate) struct EngineReadAccountingGuard {
    _not_send: std::marker::PhantomData<*const ()>,
}

#[cfg(test)]
impl EngineReadAccountingGuard {
    pub(crate) fn start() -> Self {
        ENGINE_READ_ACCOUNTING.with(|accounting| {
            *accounting.borrow_mut() = Some(EngineReadAccounting::default());
        });
        Self {
            _not_send: std::marker::PhantomData,
        }
    }

    #[allow(clippy::unused_self)]
    pub(crate) fn snapshot(&self) -> EngineReadAccounting {
        ENGINE_READ_ACCOUNTING
            .with(|accounting| accounting.borrow().as_ref().copied().unwrap_or_default())
    }
}

#[cfg(test)]
impl Drop for EngineReadAccountingGuard {
    fn drop(&mut self) {
        ENGINE_READ_ACCOUNTING.with(|accounting| {
            *accounting.borrow_mut() = None;
        });
    }
}

/// Records one engine read. A no-op when no guard is live.
#[cfg(test)]
fn account_engine_read(engine_entries: usize, unbounded: bool) {
    ENGINE_READ_ACCOUNTING.with(|accounting| {
        if let Some(active) = accounting.borrow_mut().as_mut() {
            match (
                active.calls.checked_add(1),
                active.engine_entries.checked_add(engine_entries),
            ) {
                (Some(calls), Some(entries)) => {
                    active.calls = calls;
                    active.engine_entries = entries;
                }
                _ => active.counter_overflow_observed = true,
            }
            if unbounded {
                match active.unbounded_calls.checked_add(1) {
                    Some(unbounded_calls) => active.unbounded_calls = unbounded_calls,
                    None => active.counter_overflow_observed = true,
                }
            }
        }
    });
}

#[cfg(not(test))]
const fn account_engine_read(_engine_entries: usize, _unbounded: bool) {}

impl From<Event> for StoredEntry {
    fn from(event: Event) -> Self {
        Self {
            payload: event.payload,
            sequence: event.seq,
            timestamp: event.timestamp,
        }
    }
}

/// Maps a real-engine [`ApiError`] onto liminal's [`DurabilityError`].
///
/// The optimistic-concurrency variants route to their dedicated `DurabilityError`
/// cases (`SequenceConflict`, `CursorRegression`); everything else is a
/// store-level failure carried verbatim.
impl From<ApiError> for DurabilityError {
    fn from(error: ApiError) -> Self {
        match error {
            ApiError::SequenceConflict(conflict) => conflict.into(),
            ApiError::CasMismatch(mismatch) => mismatch.into(),
            other @ (ApiError::CorruptEvent(_)
            | ApiError::Storage(_)
            | ApiError::HistoryCompacted(_)) => Self::StoreError(other),
        }
    }
}

#[cfg(test)]
#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
mod ephemeral_lifecycle_tests {
    //! D3 §9 lifecycle gate. Each test names the gate it pins; all are permanent
    //! rule-1 assertions that the ephemeral store's directory has an enforced
    //! owner across every teardown path.

    use std::path::{Path, PathBuf};
    use std::sync::{Arc, Mutex};

    use super::super::bridge::block_on;
    use super::{
        DurableStore, EphemeralGuard, open_ephemeral, open_ephemeral_in, open_ephemeral_rooted,
    };

    const TEST_SHARD_COUNT: usize = 2;

    /// In-memory `tracing` sink, so a test can assert on what the teardown path
    /// LOGGED rather than on what it merely did.
    ///
    /// Every teardown assertion below runs against this one instrument, and
    /// [`panic_path_leak_is_logged_with_its_path`] is its positive control: it
    /// exercises the SAME predicate (`captured` contains the path and `ERROR`)
    /// against a log line that is emitted today. Without that control an empty
    /// capture would only measure the harness.
    #[derive(Clone, Default)]
    struct CapturedLog(Arc<Mutex<Vec<u8>>>);

    impl CapturedLog {
        /// Everything written to the sink so far, as text.
        fn text(&self) -> String {
            let bytes = self
                .0
                .lock()
                .expect("capture buffer is not poisoned")
                .clone();
            String::from_utf8(bytes).expect("tracing's fmt writer emits utf-8")
        }

        /// Runs `body` with this sink receiving everything the CURRENT THREAD
        /// logs, via one process-global subscriber and a thread-routed writer.
        ///
        /// Why not `tracing::subscriber::with_default`: a scoped subscriber
        /// registers a dispatcher on entry and deregisters it on exit, and
        /// tracing maintains global state (the per-callsite interest cache and
        /// the max-level hint) that is rebuilt on those edges. That produced a
        /// measured intermittently-EMPTY capture in this module — 3/40
        /// module-scoped runs raw; serializing the windows on a mutex cured
        /// the module-scoped loop (0/40) but the full-workspace battery still
        /// reproduced the empty capture with the mutex in place, so edge
        /// timing was not the whole mechanism. This design removes the CLASS:
        /// the global subscriber is installed exactly once and never
        /// deregistered, so no edge ever exists to re-poison the caches, and
        /// routing is thread-local so parallel tests cannot cross-capture.
        fn capturing<R>(&self, body: impl FnOnce() -> R) -> R {
            static INSTALL: std::sync::Once = std::sync::Once::new();
            /// Clears the thread's capture slot even when `body` unwinds.
            struct ResetOnDrop;
            impl Drop for ResetOnDrop {
                fn drop(&mut self) {
                    ACTIVE_CAPTURE.with(|slot| *slot.borrow_mut() = None);
                }
            }
            INSTALL.call_once(|| {
                let subscriber = tracing_subscriber::fmt()
                    .with_writer(RoutedWriter)
                    .with_ansi(false)
                    .finish();
                tracing::subscriber::set_global_default(subscriber)
                    .expect("no other global tracing subscriber is installed in this test binary");
            });
            ACTIVE_CAPTURE.with(|slot| *slot.borrow_mut() = Some(self.clone()));
            let _reset = ResetOnDrop;
            body()
        }
    }

    thread_local! {
        /// The capture buffer receiving THIS thread's log output, if a
        /// [`CapturedLog::capturing`] window is active on it.
        static ACTIVE_CAPTURE: std::cell::RefCell<Option<CapturedLog>> =
            const { std::cell::RefCell::new(None) };
    }

    /// The one writer the process-global subscriber owns: appends to the
    /// emitting thread's active capture buffer, and silently discards output
    /// from threads with no capture window open.
    #[derive(Clone, Copy, Default)]
    struct RoutedWriter;

    impl std::io::Write for RoutedWriter {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            ACTIVE_CAPTURE.with(|slot| {
                if let Some(capture) = slot.borrow().as_ref() {
                    capture
                        .0
                        .lock()
                        .map_err(|_| std::io::Error::other("capture buffer poisoned"))?
                        .extend_from_slice(buf);
                }
                Ok(buf.len())
            })
        }

        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    impl<'writer> tracing_subscriber::fmt::MakeWriter<'writer> for RoutedWriter {
        type Writer = Self;

        fn make_writer(&'writer self) -> Self::Writer {
            *self
        }
    }

    /// Sets `path`'s mode, used to make a parent directory unwritable so that
    /// removing a directory INSIDE it fails at the final `rmdir`.
    ///
    /// That is the observed production failure shape: the contents go, the
    /// directory itself stays, and the removal error is the only witness.
    #[cfg(unix)]
    fn set_mode(path: &Path, mode: u32) {
        use std::os::unix::fs::PermissionsExt;

        std::fs::set_permissions(path, std::fs::Permissions::from_mode(mode))
            .expect("test can set permissions on a directory it created");
    }

    /// Store stand-in whose `Drop` pins the guard's internal ordering: the
    /// directory must still exist at store-drop time, so this drop FAILS the
    /// test if the guard ever removes the directory first.
    struct OrderProbeStore {
        dir: PathBuf,
    }

    impl Drop for OrderProbeStore {
        fn drop(&mut self) {
            assert!(
                self.dir.exists(),
                "the guard must drop the store BEFORE removing the directory"
            );
        }
    }

    /// Store stand-in whose `Drop` panics, modelling a haematite worker failing
    /// to join while the database closes.
    struct PanickingProbeStore;

    impl Drop for PanickingProbeStore {
        fn drop(&mut self) {
            panic!("injected store-drop panic");
        }
    }

    /// Materialises shard directories and fds so the drop path actually has a
    /// live database to close before the guard removes the directory.
    fn write_one_event(store: &dyn DurableStore) {
        block_on(store.append("lifecycle/probe", b"payload".to_vec(), 0))
            .expect("bridge completes synchronously")
            .expect("append to a fresh ephemeral stream succeeds");
        block_on(store.flush())
            .expect("bridge completes synchronously")
            .expect("flush of a live ephemeral store succeeds");
    }

    /// §9 gate — normal drop: the directory is removed once the last (here, only)
    /// handle drops.
    #[test]
    fn ephemeral_dir_removed_after_last_handle_drops() {
        let store = open_ephemeral(TEST_SHARD_COUNT).expect("ephemeral open succeeds");
        let dir = store
            .ephemeral_dir_path()
            .expect("ephemeral store carries a guard dir")
            .to_path_buf();
        assert!(
            dir.exists(),
            "the guard directory exists while the store is live"
        );

        write_one_event(&store);
        drop(store);

        assert!(
            !dir.exists(),
            "the guard directory is removed on normal drop"
        );
    }

    /// §9 gate — teardown with store-handle clones alive: the directory survives
    /// until the LAST `Arc<dyn DurableStore>` clone drops, then is removed. This
    /// is the `Arc`-shared-into-channel-handles case: clones share one wrapper,
    /// so none can close the database early.
    #[test]
    fn ephemeral_dir_survives_until_last_store_clone_drops() {
        let store = open_ephemeral(TEST_SHARD_COUNT).expect("ephemeral open succeeds");
        let dir = store
            .ephemeral_dir_path()
            .expect("ephemeral store carries a guard dir")
            .to_path_buf();
        write_one_event(&store);

        let erased: Arc<dyn DurableStore> = Arc::new(store);
        let clone_a = Arc::clone(&erased);
        let clone_b = Arc::clone(&erased);

        drop(erased);
        assert!(
            dir.exists(),
            "directory survives while store clones remain alive"
        );
        drop(clone_a);
        assert!(
            dir.exists(),
            "directory survives while one store clone remains alive"
        );

        drop(clone_b);
        assert!(
            !dir.exists(),
            "the last store clone dropping removes the directory"
        );
    }

    /// §9 gate — startup rollback: an injected haematite open failure (a
    /// conflicting `config.json` pre-seeded into the guard dir) makes the
    /// constructor return `Err` AND leaves zero residue — the guard removes the
    /// directory independently of haematite's own cleanup.
    #[test]
    fn ephemeral_open_failure_rolls_back_directory() {
        let seeded = tempfile::Builder::new()
            .prefix("liminal-durability-test-")
            .tempdir()
            .expect("test can create a temp dir");
        let dir = seeded.path().to_path_buf();
        // A pre-existing `config.json` makes haematite refuse the create with
        // `DataDirAlreadyInitialised`; because the dir pre-existed the create,
        // haematite never removes it — only the guard does.
        std::fs::write(dir.join("config.json"), b"not-a-valid-config")
            .expect("test can seed a conflicting config");

        let result = open_ephemeral_in(seeded, TEST_SHARD_COUNT);

        assert!(result.is_err(), "an injected open failure returns Err");
        assert!(
            !dir.exists(),
            "the guard removes the directory on open failure — zero residue"
        );
    }

    /// §9 gate — repeated start/stop: each cycle owns a distinct directory and
    /// leaves zero residue after it drops.
    #[test]
    fn repeated_ephemeral_cycles_each_own_distinct_dir_zero_residue() {
        let mut seen: Vec<PathBuf> = Vec::new();
        for _ in 0..5 {
            let store = open_ephemeral(TEST_SHARD_COUNT).expect("ephemeral open succeeds");
            let dir = store
                .ephemeral_dir_path()
                .expect("ephemeral store carries a guard dir")
                .to_path_buf();
            assert!(
                dir.exists(),
                "the cycle's directory exists while its store is live"
            );
            assert!(!seen.contains(&dir), "each cycle owns a distinct directory");
            seen.push(dir.clone());

            write_one_event(&store);
            drop(store);
            assert!(
                !dir.exists(),
                "the cycle's directory is removed after its store drops"
            );
        }
    }

    /// §9 gate (drop-order pin): the guard drops the store strictly before it
    /// removes the directory. `OrderProbeStore::drop` asserts the directory
    /// still exists, so reversing the order inside [`EphemeralGuard`] fails this
    /// test rather than silently passing.
    #[test]
    fn guard_drops_store_before_removing_directory() {
        let dir = tempfile::tempdir().expect("test can create a temp dir");
        let path = dir.path().to_path_buf();
        let guard = EphemeralGuard {
            store: Some(OrderProbeStore { dir: path.clone() }),
            dir: Some(dir),
        };

        drop(guard);

        assert!(!path.exists(), "a clean drop still removes the directory");
    }

    /// §9 gate (unwind pin): a panic while the store drops leaves the directory
    /// LEAKED, never removed under possibly-live workers, and the panic still
    /// propagates.
    #[test]
    fn guard_leaks_directory_when_store_drop_panics() {
        let dir = tempfile::tempdir().expect("test can create a temp dir");
        let path = dir.path().to_path_buf();
        let guard = EphemeralGuard {
            store: Some(PanickingProbeStore),
            dir: Some(dir),
        };

        let unwound = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || drop(guard)));

        assert!(unwound.is_err(), "the injected store-drop panic propagates");
        assert!(
            path.exists(),
            "a panicking store drop leaks the directory instead of removing it"
        );
        std::fs::remove_dir_all(&path).expect("test cleans up the deliberately leaked directory");
    }

    /// The rooted factory places (and removes) the guard directory under the
    /// caller-supplied root, which is what lets construction gates assert on an
    /// isolated root instead of scanning the system temp dir.
    #[test]
    fn rooted_ephemeral_store_lives_and_dies_under_the_given_root() {
        let root = tempfile::tempdir().expect("test can create a temp root");
        let store =
            open_ephemeral_rooted(root.path(), TEST_SHARD_COUNT).expect("rooted open succeeds");
        let dir = store
            .ephemeral_dir_path()
            .expect("ephemeral store carries a guard dir")
            .to_path_buf();
        assert!(
            dir.starts_with(root.path()),
            "the guard directory is created under the supplied root"
        );

        write_one_event(&store);
        drop(store);

        assert!(!dir.exists(), "the rooted directory is removed on drop");
    }

    /// Clean-teardown gate (keepalive-honest shape): the guard directory is
    /// present for the store's WHOLE life — re-checked between unrelated
    /// operations that each succeed — and gone once the store drops cleanly.
    ///
    /// The "unrelated ops proceed" leg is what makes the final absence mean
    /// something: a directory that vanished early would take the appends,
    /// reads and CAS down with it, so this cannot pass by removing the
    /// directory too soon and cannot pass by never having created it.
    #[test]
    fn ephemeral_dir_persists_across_unrelated_work_then_goes_on_clean_drop() {
        let store = open_ephemeral(TEST_SHARD_COUNT).expect("ephemeral open succeeds");
        let dir = store
            .ephemeral_dir_path()
            .expect("ephemeral store carries a guard dir")
            .to_path_buf();
        assert!(
            dir.exists(),
            "the directory exists as soon as the store does"
        );

        for round in 0..3_u64 {
            block_on(store.append("clean-teardown/probe", b"payload".to_vec(), round))
                .expect("bridge completes synchronously")
                .expect("append to a live ephemeral store succeeds");
            assert!(
                dir.exists(),
                "the directory is still there after append round {round}"
            );
        }
        block_on(store.cas("clean-teardown/counter", 0, 7))
            .expect("bridge completes synchronously")
            .expect("cas on a live ephemeral store succeeds");
        let entries = block_on(store.read_from("clean-teardown/probe", 0, 10))
            .expect("bridge completes synchronously")
            .expect("read from a live ephemeral store succeeds");
        assert_eq!(entries.len(), 3, "every appended entry is readable back");
        assert!(
            dir.exists(),
            "the directory is still there after unrelated cas and read work"
        );

        block_on(store.flush())
            .expect("bridge completes synchronously")
            .expect("flush of a live ephemeral store succeeds");
        drop(store);

        assert!(
            !dir.exists(),
            "the clean drop removes the directory it kept alive throughout"
        );
    }

    /// Clean-teardown gate: when removal FAILS on the clean path the guard
    /// LOGS the failure and its path, and does not panic.
    ///
    /// Injected the way it fails in production: the parent is made unwritable,
    /// so `remove_dir_all` clears the contents and then cannot unlink the
    /// directory itself. `tempfile`'s own `Drop` discards that error
    /// (`let _ = remove_dir_all(..)`), which is why this pin is red until the
    /// clean path calls `close()` and reports what it returns.
    #[cfg(unix)]
    #[test]
    fn clean_drop_removal_failure_is_logged_and_never_panics() {
        let parent = tempfile::tempdir().expect("test can create a temp parent");
        let dir = tempfile::Builder::new()
            .prefix("liminal-durability-")
            .tempdir_in(parent.path())
            .expect("test can create a guard dir under the parent");
        let path = dir.path().to_path_buf();
        let guard = EphemeralGuard {
            store: Some(()),
            dir: Some(dir),
        };

        set_mode(parent.path(), 0o500);
        let captured = CapturedLog::default();
        let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            captured.capturing(|| drop(guard));
        }));
        // Restored before the assertions so a failing assertion still leaves a
        // parent the outer `TempDir` can clean up.
        set_mode(parent.path(), 0o700);

        assert!(
            outcome.is_ok(),
            "a removal failure is reported, never raised as a panic"
        );
        let logged = captured.text();
        assert!(
            logged.contains("ERROR"),
            "the removal failure is logged at error level; captured: {logged:?}"
        );
        assert!(
            logged.contains(&path.display().to_string()),
            "the log names the directory that survived; captured: {logged:?}"
        );
        assert!(
            path.exists(),
            "the residue is left where the log says it is, not silently claimed removed"
        );
    }

    /// Clean-teardown gate (negative control for the capture instrument): a
    /// removal that SUCCEEDS logs nothing, so the assertion above discriminates
    /// failure from success rather than matching any teardown at all.
    #[test]
    fn clean_drop_that_succeeds_logs_nothing() {
        let dir = tempfile::tempdir().expect("test can create a temp dir");
        let path = dir.path().to_path_buf();
        let guard = EphemeralGuard {
            store: Some(()),
            dir: Some(dir),
        };

        let captured = CapturedLog::default();
        captured.capturing(|| drop(guard));

        assert!(!path.exists(), "the successful clean drop removed the dir");
        assert!(
            captured.text().is_empty(),
            "a successful removal is silent; captured: {:?}",
            captured.text()
        );
    }

    /// Positive control for the capture instrument: the panic path's sanctioned
    /// leak line IS captured, path and all, by the same predicate the
    /// removal-failure gate uses.
    ///
    /// Without this, an empty capture would be a measurement of the harness
    /// rather than of the code under test.
    #[test]
    fn panic_path_leak_is_logged_with_its_path() {
        let dir = tempfile::tempdir().expect("test can create a temp dir");
        let path = dir.path().to_path_buf();
        let guard = EphemeralGuard {
            store: Some(PanickingProbeStore),
            dir: Some(dir),
        };

        let captured = CapturedLog::default();
        let unwound = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            captured.capturing(|| drop(guard));
        }));

        assert!(unwound.is_err(), "the injected store-drop panic propagates");
        let logged = captured.text();
        assert!(
            logged.contains("ERROR"),
            "the sanctioned leak is logged at error level; captured: {logged:?}"
        );
        assert!(
            logged.contains(&path.display().to_string()),
            "the leak log names the leaked directory; captured: {logged:?}"
        );
        std::fs::remove_dir_all(&path).expect("test cleans up the deliberately leaked directory");
    }
}

#[cfg(test)]
#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
mod paged_read_shape_tests {
    //! Board #60. The page limit must be honoured by the ENGINE, not by a
    //! truncation applied after the engine has already materialised the suffix.
    //!
    //! These pins are counts, never durations: the defect is a read shape, and
    //! a shape is deterministic where a latency is not.

    use super::{DurableStore, EngineReadAccountingGuard, open_ephemeral};
    use crate::durability::bridge::block_on;

    /// Page size used by both production replay readers (`READ_BATCH_SIZE` and
    /// `UNIT2_OUTBOX_RESTORE_BATCH_ROWS` are both 64).
    const PAGE: usize = 64;
    /// Four pages. Enough that the quadratic and the linear shape differ by
    /// more than a factor of two, small enough that seeding stays cheap.
    const ROWS: u64 = 256;
    const STREAM: &str = "liminal/p0-60/paged-read-shape";

    /// Seeds `ROWS` events into one stream.
    fn seeded() -> Result<impl DurableStore, Box<dyn std::error::Error>> {
        let store = open_ephemeral(1)?;
        for sequence in 0..ROWS {
            block_on(store.append(STREAM, sequence.to_be_bytes().to_vec(), sequence))??;
        }
        block_on(store.flush())??;
        Ok(store)
    }

    /// Walks the whole stream one page at a time, exactly as replay does.
    fn read_whole_stream(
        store: &impl DurableStore,
        page: usize,
    ) -> Result<usize, Box<dyn std::error::Error>> {
        let mut offset = 0_u64;
        let mut seen = 0_usize;
        loop {
            let entries = block_on(store.read_from(STREAM, offset, page))??;
            if entries.is_empty() {
                return Ok(seen);
            }
            for entry in &entries {
                assert_eq!(entry.sequence, offset, "paged read must stay contiguous");
                offset += 1;
            }
            seen = seen
                .checked_add(entries.len())
                .ok_or("row counter overflowed")?;
        }
    }

    /// Every read below is bounded by its `limit`, so no read costs more than
    /// the rows it returns. One seeded store carries all four shapes.
    #[test]
    fn a_bounded_read_never_scans_beyond_its_page() -> Result<(), Box<dyn std::error::Error>> {
        let store = seeded()?;

        // 1. The whole stream, paged. O(N), not O(N^2).
        let accounting = EngineReadAccountingGuard::start();
        let seen = read_whole_stream(&store, PAGE)?;
        let walk = accounting.snapshot();
        drop(accounting);
        assert_eq!(
            u64::try_from(seen)?,
            ROWS,
            "the walk must deliver every row"
        );
        assert!(
            !walk.counter_overflow_observed,
            "a saturated counter is not a measurement"
        );
        assert!(walk.calls > 0, "the walk must have reached the store");
        assert_eq!(
            u64::try_from(walk.engine_entries)?,
            ROWS,
            "a full stream read must scan each row exactly once instead of \
             re-scanning every suffix once per page"
        );

        // 2. One page from the head.
        let accounting = EngineReadAccountingGuard::start();
        let head = block_on(store.read_from(STREAM, 0, PAGE))??;
        let head_read = accounting.snapshot();
        drop(accounting);
        assert_eq!(head.len(), PAGE, "a full page returns its limit");
        assert_eq!(
            head_read.engine_entries, PAGE,
            "the engine must be asked for one page, not for the whole stream"
        );

        // 3. One page from the MIDDLE. The rows after the page are the ones a
        //    suffix-scanning read would drag along; the rows before it are the
        //    ones the offset already excludes, so only a bounded upper edge can
        //    make this count come out at PAGE.
        let middle_offset = ROWS / 2;
        let accounting = EngineReadAccountingGuard::start();
        let middle = block_on(store.read_from(STREAM, middle_offset, PAGE))??;
        let middle_read = accounting.snapshot();
        drop(accounting);
        assert_eq!(
            middle.len(),
            PAGE,
            "a full page mid-stream returns its limit"
        );
        assert_eq!(
            middle_read.engine_entries, PAGE,
            "a mid-stream page must not scan the rows that follow it"
        );

        // 4. Past the head: end of stream, and no scan.
        let accounting = EngineReadAccountingGuard::start();
        let past = block_on(store.read_from(STREAM, ROWS, PAGE))??;
        let past_read = accounting.snapshot();
        drop(accounting);
        assert!(past.is_empty(), "past the head is end of stream");
        assert_eq!(
            past_read.engine_entries, 0,
            "an end-of-stream page must not scan the stream"
        );
        Ok(())
    }

    /// The equivalence the pushdown must preserve. This passes before and after
    /// the fix by design: it is the control that says the fix changed the read
    /// SHAPE and nothing else.
    #[test]
    fn page_size_never_changes_the_answer() -> Result<(), Box<dyn std::error::Error>> {
        let store = seeded()?;
        let whole = block_on(store.read_from(STREAM, 0, usize::MAX))??;
        assert_eq!(u64::try_from(whole.len())?, ROWS);

        for page in [1_usize, 7, 64, 255, 256, 257] {
            let mut offset = 0_u64;
            let mut collected = Vec::new();
            loop {
                let entries = block_on(store.read_from(STREAM, offset, page))??;
                if entries.is_empty() {
                    break;
                }
                assert!(entries.len() <= page, "a page never exceeds its limit");
                offset = offset
                    .checked_add(u64::try_from(entries.len())?)
                    .ok_or("offset overflowed")?;
                collected.extend(entries);
            }
            assert_eq!(collected, whole, "page size {page} changed the answer");
        }

        // A zero limit is the one page size that must return nothing, and it
        // must not be answered by a bounded window that silently agrees.
        assert!(
            block_on(store.read_from(STREAM, 0, 0))??.is_empty(),
            "a zero limit reads nothing"
        );

        // Every suffix start agrees with the same suffix of the whole read.
        for offset in [0_u64, 1, 63, 64, 65, 128, 255] {
            let suffix = block_on(store.read_from(STREAM, offset, usize::MAX))??;
            assert_eq!(
                suffix,
                whole[usize::try_from(offset)?..],
                "suffix from {offset} diverged"
            );
        }
        Ok(())
    }
}