liminal/durability/

store.rs

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

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

use super::DurabilityError;
use super::counting::StoreCountsHandle;

use tempfile::TempDir;

/// A conditional key/value replacement in one routed atomic commit.
#[derive(Clone, Debug)]
pub struct RoutedBatchItem {
    /// The key routed with the entire operation.
    pub key: Vec<u8>,
    /// The original content hash; absence is an explicit precondition.
    pub expected: Option<haematite::tree::Hash>,
    /// The replacement bytes.
    pub value: Vec<u8>,
    /// The lifetime of the replacement, or no expiry.
    pub ttl: Option<std::time::Duration>,
}

/// 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,
}

/// Position a paged prefix scan resumes from.
///
/// Names one stream and the first sequence not yet delivered from it. Streams
/// sorting after `stream_key` are untouched by the resume and start at their
/// own sequence 0.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ScanCursor {
    /// Stream the next page resumes inside.
    pub stream_key: String,
    /// First sequence the next page reads from that stream.
    pub next_sequence: u64,
}

/// One bounded page of a prefix scan.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ScanPage {
    /// Entries this page delivered, in (stream key, sequence) order.
    pub entries: Vec<StoredEntry>,
    /// Position the next page resumes from, or `None` when the scan is done.
    pub next: Option<ScanCursor>,
}

/// Direct durability surface matching haematite's append/read/cas/paged-scan API.
#[async_trait::async_trait]
pub trait DurableStore: std::fmt::Debug + Send + Sync {
    /// Commits all conditional replacements on one route, or publishes none.
    ///
    /// No implementation may compose this operation from single writes.
    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<RoutedBatchItem>,
    ) -> Result<(), DurabilityError>;

    /// Reads one key from the route used by the atomic writer.
    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, DurabilityError>;

    /// 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>;

    /// Atomically appends all payloads in order, starting at the expected head,
    /// and returns the stream's next sequence.
    ///
    /// A conflict or a refusal before commit leaves every payload absent.
    /// If the commit's reply is lost, the whole batch may be present; readers
    /// must never observe a prefix. Implementations must submit one atomic
    /// batch rather than compose individual appends.
    async fn append_all(
        &self,
        stream_key: &str,
        payloads: Vec<Vec<u8>>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError>;

    /// 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 the exclusive append head without reading stream rows.
    ///
    /// `None` means the stream has never held an appended row. A present
    /// head is nonzero and is preserved when rows are pruned. Numeric values
    /// written through compare-and-swap do not constitute a stream head.
    ///
    /// # Errors
    /// Returns the store's failure or an explicit unsupported-operation refusal.
    async fn read_stream_next_seq(&self, stream_key: &str) -> Result<Option<u64>, DurabilityError>;

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

    /// Reads one page of at most `limit` entries from the streams whose keys
    /// begin with `prefix`, in byte order of stream key and then sequence.
    ///
    /// `cursor` resumes a walk: `None` starts at the first matching stream,
    /// `Some` starts at that cursor's position. `limit` must be nonzero.
    ///
    /// A returned `next` of `Some` means only that this page FILLED, never that
    /// more entries exist: a follow-up call that finds nothing returns an empty
    /// page with `next` of `None`, and that empty page is a legal terminator.
    ///
    /// The contract is resume-at-position, not a point-in-time snapshot. There
    /// is no read view held across calls, so entries appended at or after the
    /// cursor's position between two pages WILL appear in the later page.
    async fn scan_page(
        &self,
        prefix: &str,
        cursor: Option<ScanCursor>,
        limit: usize,
    ) -> Result<ScanPage, 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>;

    /// Deletes every entry of `stream_key` with sequence below `before_seq`,
    /// in one commit, and returns how many were deleted. Later entries are
    /// untouched, the stream's head is kept (an append continues where it
    /// would have, and no sequence is reused), and a second identical call
    /// deletes nothing and returns 0.
    ///
    /// A store that cannot delete refuses by name rather than answering 0,
    /// so a caller pruning a bounded stream is never told a cut happened
    /// when nothing was removed.
    ///
    /// # Errors
    /// Returns [`DurabilityError::TruncateUnsupported`] from a store that
    /// cannot delete, or the store's own failure.
    async fn truncate_before(
        &self,
        stream_key: &str,
        before_seq: u64,
    ) -> Result<u64, DurabilityError> {
        Err(DurabilityError::TruncateUnsupported {
            store: std::any::type_name::<Self>(),
            stream_key: stream_key.to_owned(),
            before_seq,
        })
    }

    /// Reads the routed keys of one window on a route, with their values.
    ///
    /// The window is the half-open `[from, to)` in key order, and it is
    /// physical: the caller derives it from the keys it wrote through
    /// [`Self::apply_routed_batch`] on the same route (DUR-018 R3).
    ///
    /// # Errors
    /// Returns [`DurabilityError::RoutedScanUnsupported`] from a store that
    /// cannot list routed keys, or the store's own failure.
    async fn scan_routed(
        &self,
        route_key: &str,
        _from: &[u8],
        _to: &[u8],
    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>, DurabilityError> {
        Err(DurabilityError::RoutedScanUnsupported {
            store: std::any::type_name::<Self>(),
            route_key: route_key.to_owned(),
        })
    }

    /// Deletes the routed keys of one window on a route in one commit.
    ///
    /// The window is the half-open `[from, to)`; the answer is how many keys
    /// went. Keys outside the window, the route's other records included,
    /// are untouched, and a second identical call deletes nothing and
    /// returns 0.
    ///
    /// A store that cannot do it in one commit refuses by name rather than
    /// composing single deletes (Tom, 3 October 2026: a batch is one durable
    /// operation, never leaf by leaf), so a caller pruning a routed stream is
    /// never told a cut happened when nothing was removed, and never pays a
    /// commit per row.
    ///
    /// # Errors
    /// Returns [`DurabilityError::RoutedRemoveUnsupported`] from a store that
    /// cannot delete a window in one commit, or the store's own failure.
    async fn remove_routed_range(
        &self,
        route_key: &str,
        _from: &[u8],
        _to: &[u8],
    ) -> Result<u64, DurabilityError> {
        Err(DurabilityError::RoutedRemoveUnsupported {
            store: std::any::type_name::<Self>(),
            route_key: route_key.to_owned(),
        })
    }

    /// A reader of the counters of the work done through this store, when the
    /// store counts it.
    ///
    /// Only a [`super::CountingStore`] counts; every other store answers
    /// `None`. A store held as `Arc<dyn DurableStore>` is reachable no other
    /// way, which is why the reader is asked of the store itself.
    fn store_counts(&self) -> Option<StoreCountsHandle> {
        None
    }
}

/// `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 apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<RoutedBatchItem>,
    ) -> Result<(), DurabilityError> {
        let items = items
            .into_iter()
            .map(|item| haematite::RoutedBatchItem {
                key: item.key,
                expected: item.expected,
                value: item.value,
                ttl: item.ttl,
            })
            .collect();
        self.event_store
            .database()
            .apply_routed_batch(route_key.as_bytes(), items)
            .map_err(ApiError::from)
            .map_err(DurabilityError::from)
    }

    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, DurabilityError> {
        self.event_store
            .database()
            .get_routed(route_key.as_bytes(), key)
            .map_err(ApiError::from)
            .map_err(DurabilityError::from)
    }

    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_stream_next_seq(&self, stream_key: &str) -> Result<Option<u64>, DurabilityError> {
        self.event_store
            .read_stream_next_seq(stream_key.as_bytes())
            .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_page(
        &self,
        prefix: &str,
        cursor: Option<ScanCursor>,
        limit: usize,
    ) -> Result<ScanPage, DurabilityError> {
        if limit == 0 {
            return Err(DurabilityError::ConfigError(
                "scan_page limit must be nonzero".to_owned(),
            ));
        }

        let mut entries: Vec<StoredEntry> = Vec::new();
        let mut resume_at: Option<ScanCursor> = None;
        // The cursor's own stream resumes where the last page stopped; every
        // stream after it starts at sequence 0.
        let mut after: Option<String> = if let Some(cursor) = cursor {
            if cursor.stream_key.starts_with(prefix) {
                let page = self
                    .read_from(&cursor.stream_key, cursor.next_sequence, limit)
                    .await?;
                resume_at = scan_resume_position(&cursor.stream_key, &page)?.or(resume_at);
                entries.extend(page);
            }
            Some(cursor.stream_key)
        } else {
            None
        };

        // Haematite's `scan_prefix` enumerates streams in key order from a
        // position, a bounded page at a time (G1), so a page lists only the
        // streams after its cursor and never the whole prefix again: a walk
        // drained page by page enumerates each matching stream once. A live
        // stream holds at least one live event, so asking for as many streams
        // as the page has room for never enumerates more than it can read.
        while let Some(remaining) = limit
            .checked_sub(entries.len())
            .and_then(std::num::NonZeroUsize::new)
        {
            let streams = self
                .event_store
                .scan_prefix(
                    prefix.as_bytes(),
                    remaining,
                    after.as_deref().map(str::as_bytes),
                )
                .map_err(DurabilityError::from)?;
            account_stream_enumeration(streams.streams.len());
            let exhausted = streams.next_after.is_none() || streams.streams.is_empty();
            for stream in streams.streams {
                // Every key liminal writes comes from a `&str`; a key that is
                // not UTF-8 was written by something else and cannot be named
                // in a cursor, so refuse rather than skip it into silent data
                // loss.
                let stream_key = String::from_utf8(stream.stream_key).map_err(|error| {
                    DurabilityError::StoreError(ApiError::CorruptEvent(format!(
                        "scanned stream key is not utf-8: {error}"
                    )))
                })?;
                let Some(remaining) = limit.checked_sub(entries.len()).filter(|rest| *rest > 0)
                else {
                    break;
                };
                // The bounded `read_from` above, never `EventStore::read`:
                // asking for exactly the remaining budget is what keeps a
                // page's cost proportional to the page and not to the stream
                // behind it.
                let page = self.read_from(&stream_key, 0, remaining).await?;
                resume_at = scan_resume_position(&stream_key, &page)?.or(resume_at);
                entries.extend(page);
                after = Some(stream_key);
            }
            if exhausted {
                break;
            }
        }

        // `next` reports only that the page FILLED. A stream that returned
        // fewer entries than its budget has already been passed over by the
        // loop, so a short page has nothing left to resume from.
        let next = if entries.len() == limit {
            resume_at
        } else {
            None
        };
        Ok(ScanPage { entries, next })
    }

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

    /// One haematite `truncate_stream` call. liminal's sequence is
    /// haematite's API sequence, so `before_seq` passes through unchanged.
    async fn truncate_before(
        &self,
        stream_key: &str,
        before_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.event_store
            .truncate_stream(stream_key.as_bytes(), before_seq)
            .map_err(DurabilityError::from)
    }

    /// One haematite `range_routed` call on the route's shard.
    ///
    /// It reads the committed tree merged with the WAL buffer, the same read
    /// behind every routed lookup.
    async fn scan_routed(
        &self,
        route_key: &str,
        from: &[u8],
        to: &[u8],
    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>, DurabilityError> {
        self.event_store
            .database()
            .range_routed(route_key.as_bytes(), from, to)
            .map_err(ApiError::from)
            .map_err(DurabilityError::from)
    }

    /// Refused by name until haematite can delete a window in one commit.
    ///
    /// Haematite 0.15.1 (a3a7749d) deletes a routed key only one at a time
    /// (`delete_routed`, one stamp and one commit each) and has no routed
    /// range delete. Composing the window from those
    /// would be a commit per row, the leaf-by-leaf vacuuming Tom ruled out
    /// on 3 October 2026, so the cut waits for a one-commit range delete in
    /// haematite (DUR-018 R3 names the ask) and the prune receipt stands
    /// unclaimed until then.
    async fn remove_routed_range(
        &self,
        route_key: &str,
        _from: &[u8],
        _to: &[u8],
    ) -> Result<u64, DurabilityError> {
        Err(DurabilityError::RoutedRemoveUnsupported {
            store: std::any::type_name::<Self>(),
            route_key: route_key.to_owned(),
        })
    }
}

/// The position a scan resumes from after delivering `page` from
/// `stream_key`, or `None` when the page delivered nothing from it.
fn scan_resume_position(
    stream_key: &str,
    page: &[StoredEntry],
) -> Result<Option<ScanCursor>, DurabilityError> {
    let Some(last) = page.last() else {
        return Ok(None);
    };
    let next_sequence = last.sequence.checked_add(1).ok_or_else(|| {
        DurabilityError::StoreError(ApiError::CorruptEvent(format!(
            "scan page sequence overflow for stream {stream_key}"
        )))
    })?;
    Ok(Some(ScanCursor {
        stream_key: stream_key.to_owned(),
        next_sequence,
    }))
}

/// 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> {
    ownership: Option<super::ephemeral_root::EphemeralOwnership>,
    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"
                );
            }
            if let Some(ownership) = self.ownership.take() {
                std::mem::forget(ownership);
            }
            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: [`super::EphemeralRoot::open_store`] 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 {
                ownership: None,
                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)
    }

    fn ephemeral_dir_path_owned(&self) -> Result<&Path, DurabilityError> {
        self.guard
            .dir
            .as_ref()
            .map(TempDir::path)
            .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 apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<crate::durability::RoutedBatchItem>,
    ) -> Result<(), crate::durability::DurabilityError> {
        self.store()?.apply_routed_batch(route_key, items).await
    }

    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, crate::durability::DurabilityError> {
        self.store()?.get_routed(route_key, key).await
    }

    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_stream_next_seq(&self, stream_key: &str) -> Result<Option<u64>, DurabilityError> {
        self.store()?.read_stream_next_seq(stream_key).await
    }

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

    async fn scan_page(
        &self,
        prefix: &str,
        cursor: Option<ScanCursor>,
        limit: usize,
    ) -> Result<ScanPage, DurabilityError> {
        self.store()?.scan_page(prefix, cursor, limit).await
    }

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

    async fn truncate_before(
        &self,
        stream_key: &str,
        before_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.store()?.truncate_before(stream_key, before_seq).await
    }

    async fn scan_routed(
        &self,
        route_key: &str,
        from: &[u8],
        to: &[u8],
    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>, DurabilityError> {
        self.store()?.scan_routed(route_key, from, to).await
    }

    async fn remove_routed_range(
        &self,
        route_key: &str,
        from: &[u8],
        to: &[u8],
    ) -> Result<u64, DurabilityError> {
        self.store()?.remove_routed_range(route_key, from, to).await
    }
}

/// Test construction seam for 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.
#[cfg(any(test, feature = "test-support"))]
pub fn open_ephemeral(shard_count: usize) -> Result<EphemeralHaematiteStore, DurabilityError> {
    open_ephemeral_in(ephemeral_tempdir(&std::env::temp_dir())?, 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.
///
/// This test seam leaves ownership of the parent with its caller. Production
/// stores instead retain an [`super::EphemeralRoot`] and its lock until every
/// store has closed, so the parent cannot be swept under a live store.
///
/// # 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(root)?, shard_count)
}

pub(super) fn open_ephemeral_owned(
    root: Arc<super::EphemeralRoot>,
    shard_count: usize,
) -> Result<EphemeralHaematiteStore, DurabilityError> {
    let mut store = open_ephemeral_in(ephemeral_tempdir(root.path())?, shard_count)?;
    let path = store.ephemeral_dir_path_owned()?;
    let holder = super::ephemeral_root::lock_file(&path.join("holder.lock"))?;
    rustix::fs::flock(
        &holder,
        rustix::fs::FlockOperation::NonBlockingLockExclusive,
    )
    .map_err(super::ephemeral_root::root_error)?;
    store.guard.ownership = Some(super::ephemeral_root::EphemeralOwnership { root, holder });
    Ok(store)
}

/// Creates the guard directory for an ephemeral store under its chosen root.
fn ephemeral_tempdir(root: &Path) -> Result<TempDir, DurabilityError> {
    let mut builder = tempfile::Builder::new();
    builder.prefix("liminal-durability-");
    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),
            retention: haematite::Retention::reclaim(0),
        },
        haematite::RuntimeOptions {
            long_wait_interval: std::time::Duration::from_secs(5),
        },
    )
    .map_err(|error| DurabilityError::EphemeralStoreOpen(error.to_string()))?;
    super::counting::record_store_opened();
    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,
    /// Stream metadata entries the engine enumerated for prefix scans.
    pub(crate) streams_enumerated: 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) {}

/// Records stream metadata entries a prefix scan enumerated. A no-op when no
/// guard is live.
#[cfg(test)]
fn account_stream_enumeration(streams: usize) {
    ENGINE_READ_ACCOUNTING.with(|accounting| {
        if let Some(active) = accounting.borrow_mut().as_mut() {
            match active.streams_enumerated.checked_add(streams) {
                Some(enumerated) => active.streams_enumerated = enumerated,
                None => active.counter_overflow_observed = true,
            }
        }
    });
}

#[cfg(not(test))]
const fn account_stream_enumeration(_streams: usize) {}

/// Reference paging over an in-memory stream map, for the crate's fake stores.
///
/// Shared rather than re-derived per fixture so the fakes cannot drift from
/// each other — or from [`HaematiteStore::scan_page`], whose cursor, ordering
/// and short-page rules this mirrors exactly. Entries are selected by their
/// `sequence` field rather than by position, so a fixture seeding a sparse or
/// offset stream is paged honestly.
#[cfg(test)]
pub(crate) fn page_in_memory_streams(
    streams: &std::collections::HashMap<String, Vec<StoredEntry>>,
    prefix: &str,
    cursor: Option<&ScanCursor>,
    limit: usize,
) -> Result<ScanPage, DurabilityError> {
    if limit == 0 {
        return Err(DurabilityError::ConfigError(
            "scan_page limit must be nonzero".to_owned(),
        ));
    }
    let mut stream_keys: Vec<&String> = streams
        .keys()
        .filter(|stream_key| stream_key.starts_with(prefix))
        .collect();
    stream_keys.sort_unstable();

    let mut entries: Vec<StoredEntry> = Vec::new();
    let mut resume_at: Option<ScanCursor> = None;
    for stream_key in stream_keys {
        let Some(remaining) = limit.checked_sub(entries.len()).filter(|rest| *rest > 0) else {
            break;
        };
        let offset = match cursor {
            Some(cursor) if *stream_key < cursor.stream_key => continue,
            Some(cursor) if *stream_key == cursor.stream_key => cursor.next_sequence,
            _ => 0,
        };
        let Some(stream) = streams.get(stream_key) else {
            continue;
        };
        let page: Vec<StoredEntry> = stream
            .iter()
            .filter(|entry| entry.sequence >= offset)
            .take(remaining)
            .cloned()
            .collect();
        if let Some(last) = page.last() {
            let next_sequence = last.sequence.checked_add(1).ok_or_else(|| {
                DurabilityError::ConfigError(format!(
                    "fake scan page sequence overflow for stream {stream_key}"
                ))
            })?;
            resume_at = Some(ScanCursor {
                stream_key: stream_key.clone(),
                next_sequence,
            });
        }
        entries.extend(page);
    }

    let next = if entries.len() == limit {
        resume_at
    } else {
        None
    };
    Ok(ScanPage { entries, next })
}

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(),
            ApiError::Storage(haematite::DatabaseError::OutcomeUnknown { pid }) => {
                Self::OutcomeUnknown { pid }
            }
            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 {
            ownership: None,
            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 {
            ownership: None,
            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 {
            ownership: None,
            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 {
            ownership: None,
            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 {
            ownership: None,
            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};
    use crate::durability::bridge::block_on;
    use crate::durability::tests_fixtures;

    /// 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>> {
        tests_fixtures::copied("paged-read")
    }

    /// 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(())
    }
}

#[cfg(test)]
#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
mod paged_scan_tests {
    //! The prefix scan's bounded contract, against the real haematite-backed
    //! ephemeral store rather than a fake, so the pins exercise the engine path
    //! the production callers take.
    //!
    //! `scan_page` bounds the entry axis by its limit and the stream axis by
    //! its cursor: each page enumerates only the streams after the one it
    //! resumes in, so a drained walk lists every matching stream once.

    use crate::durability::tests_fixtures;
    use std::collections::HashSet;

    use super::{
        DurabilityError, DurableStore, EngineReadAccountingGuard, ScanCursor, StoredEntry,
        page_in_memory_streams,
    };
    use crate::durability::bridge::block_on;

    const PREFIX: &str = "liminal/scan-bound/";
    /// Three streams, so a page boundary can fall inside one of them and the
    /// stream-ordering rule has something to order.
    const STREAMS: [&str; 3] = [
        "liminal/scan-bound/alpha",
        "liminal/scan-bound/bravo",
        "liminal/scan-bound/charlie",
    ];
    /// Entries per stream. Deliberately not equal across streams so a page that
    /// silently assumed uniform stream length would be caught.
    const ROWS_PER_STREAM: [u64; 3] = [7, 11, 5];

    /// Payload naming the stream and sequence it was written at, so a drained
    /// page can be checked for identity, not merely for count.
    fn payload(stream: &str, sequence: u64) -> Vec<u8> {
        format!("{stream}#{sequence}").into_bytes()
    }

    /// Every payload the seed writes, as the census a full drain must equal.
    fn seed_census() -> Vec<Vec<u8>> {
        let mut census = Vec::new();
        for (stream, rows) in STREAMS.iter().zip(ROWS_PER_STREAM) {
            for sequence in 0..rows {
                census.push(payload(stream, sequence));
            }
        }
        census
    }

    fn seeded() -> Result<impl DurableStore, Box<dyn std::error::Error>> {
        tests_fixtures::copied("paged-scan")
    }

    /// A page never exceeds its limit, and a page that fills reports a resume
    /// position. The seed spans three streams and more rows than the limit, so
    /// the limit is the only thing that can stop this page.
    #[test]
    fn a_full_scan_page_stops_at_its_limit_and_reports_where_to_resume()
    -> Result<(), Box<dyn std::error::Error>> {
        const LIMIT: usize = 9;
        let store = seeded()?;

        let page = block_on(store.scan_page(PREFIX, None, LIMIT))??;

        assert_eq!(
            page.entries.len(),
            LIMIT,
            "a seed larger than the limit must fill exactly one page"
        );
        let cursor = page
            .next
            .ok_or("a filled page must report a resume position")?;
        // 9 entries with alpha holding 7 means the boundary lands inside bravo,
        // two entries in — the mid-stream split the resume rule exists for.
        assert_eq!(cursor.stream_key, STREAMS[1]);
        assert_eq!(cursor.next_sequence, 2);
        Ok(())
    }

    /// Draining every page reproduces the seed exactly: no losses, no
    /// duplicates, and each stream's entries still in sequence order.
    #[test]
    fn draining_every_page_delivers_the_seed_exactly_once() -> Result<(), Box<dyn std::error::Error>>
    {
        let store = seeded()?;
        let census = seed_census();

        // Limits chosen to split mid-stream in different places, including 1
        // (every page boundary is a mid-stream boundary) and limits larger than
        // a single stream but smaller than the whole seed.
        for limit in [1_usize, 2, 4, 7, 8, 13, 23, 24] {
            let mut drained: Vec<Vec<u8>> = Vec::new();
            let mut per_stream: Vec<(String, u64)> = Vec::new();
            let mut cursor: Option<ScanCursor> = None;
            let mut pages = 0_usize;
            loop {
                let page = block_on(store.scan_page(PREFIX, cursor, limit))??;
                assert!(
                    page.entries.len() <= limit,
                    "limit {limit} page {pages} exceeded its limit"
                );
                for entry in &page.entries {
                    let text = String::from_utf8(entry.payload.clone())?;
                    let stream = text
                        .split_once('#')
                        .ok_or("seeded payload names its stream")?
                        .0
                        .to_owned();
                    per_stream.push((stream, entry.sequence));
                }
                drained.extend(page.entries.into_iter().map(|entry| entry.payload));
                pages = pages.checked_add(1).ok_or("page counter overflowed")?;
                assert!(pages <= 64, "limit {limit} did not terminate in 64 pages");
                let Some(next) = page.next else {
                    break;
                };
                cursor = Some(next);
            }

            let drained_set: HashSet<Vec<u8>> = drained.iter().cloned().collect();
            assert_eq!(
                drained.len(),
                drained_set.len(),
                "limit {limit} delivered a duplicate entry"
            );
            assert_eq!(
                drained_set,
                census.iter().cloned().collect::<HashSet<_>>(),
                "limit {limit} did not deliver exactly the seeded entries"
            );

            for stream in STREAMS {
                let sequences: Vec<u64> = per_stream
                    .iter()
                    .filter(|(seen, _)| seen == stream)
                    .map(|(_, sequence)| *sequence)
                    .collect();
                assert!(
                    sequences.windows(2).all(|pair| pair[0] < pair[1]),
                    "limit {limit} broke sequence order within {stream}: {sequences:?}"
                );
            }
        }
        Ok(())
    }

    /// A page shorter than its limit ends the scan, and the empty page a
    /// filled-to-the-last-entry walk produces is a legal terminator too.
    #[test]
    fn a_short_page_ends_the_scan() -> Result<(), Box<dyn std::error::Error>> {
        let store = seeded()?;
        let total = usize::try_from(ROWS_PER_STREAM.iter().sum::<u64>())?;

        let whole =
            block_on(store.scan_page(PREFIX, None, total.checked_add(1).ok_or("overflow")?))??;
        assert_eq!(
            whole.entries.len(),
            total,
            "the whole seed fits in one page"
        );
        assert!(
            whole.next.is_none(),
            "a page short of its limit terminates the scan"
        );

        // Exactly-full: the page cannot know it is the last, so it reports a
        // resume position, and the follow-up empty page terminates instead.
        let exact = block_on(store.scan_page(PREFIX, None, total))??;
        assert_eq!(exact.entries.len(), total);
        let cursor = exact
            .next
            .ok_or("an exactly-full page reports a resume position")?;
        let terminator = block_on(store.scan_page(PREFIX, Some(cursor), total))??;
        assert!(
            terminator.entries.is_empty(),
            "the resume past the last entry finds nothing"
        );
        assert!(
            terminator.next.is_none(),
            "a final empty page is a legal terminator"
        );

        // A prefix nothing matches is the same terminator on the first call.
        let unmatched = block_on(store.scan_page("liminal/no-such-prefix/", None, 8))??;
        assert!(unmatched.entries.is_empty());
        assert!(unmatched.next.is_none());
        Ok(())
    }

    /// Page order follows stream key byte order, not the order the streams were
    /// discovered in — the property the cursor's "streams before mine are done"
    /// rule rests on.
    ///
    /// Pinned against [`page_in_memory_streams`] rather than the engine on
    /// purpose. The real impl takes its order from haematite's `scan_prefix`,
    /// whose contract is ascending key order, so a pin over the real store
    /// would test the engine rather than this rule. A `HashMap`'s iteration order is
    /// genuinely arbitrary and reseeded per process, so this pin has a source of
    /// disorder to discriminate against; with 32 streams an accidentally sorted
    /// traversal is not a case worth budgeting for.
    #[test]
    fn a_page_orders_streams_by_key_not_by_discovery_order()
    -> Result<(), Box<dyn std::error::Error>> {
        const STREAM_COUNT: usize = 32;
        let mut streams = std::collections::HashMap::new();
        for index in 0..STREAM_COUNT {
            let stream_key = format!("{PREFIX}stream-{index:02}");
            let entry = StoredEntry {
                payload: stream_key.clone().into_bytes(),
                sequence: 0,
                timestamp: 0,
            };
            streams.insert(stream_key, vec![entry]);
        }

        let page = page_in_memory_streams(&streams, PREFIX, None, STREAM_COUNT)?;

        let delivered: Vec<String> = page
            .entries
            .into_iter()
            .map(|entry| String::from_utf8(entry.payload))
            .collect::<Result<_, _>>()?;
        let mut expected = delivered.clone();
        expected.sort_unstable();
        assert_eq!(
            delivered, expected,
            "a page must deliver its streams in key order"
        );
        Ok(())
    }

    /// A walk drained page by page enumerates each matching stream once, not
    /// once per page: a page resumes the stream enumeration after its cursor
    /// instead of listing the whole prefix again, so a walk of K streams costs
    /// one metadata pass over them however many pages it takes.
    #[test]
    fn a_drained_walk_enumerates_each_stream_once() -> Result<(), Box<dyn std::error::Error>> {
        const WALK_STREAMS: usize = 48;
        const LIMIT: usize = 5;
        let store = tests_fixtures::copied("paged-walk")?;

        let accounting = EngineReadAccountingGuard::start();
        let mut cursor = None;
        let mut delivered = 0_usize;
        let mut pages = 0_usize;
        loop {
            let page = block_on(store.scan_page(PREFIX, cursor, LIMIT))??;
            pages += 1;
            delivered += page.entries.len();
            match page.next {
                Some(next) => cursor = Some(next),
                None => break,
            }
        }
        let walk = accounting.snapshot();
        drop(accounting);

        assert_eq!(delivered, WALK_STREAMS, "the walk delivers every stream");
        assert!(!walk.counter_overflow_observed, "{walk:?}");
        assert!(
            walk.streams_enumerated <= WALK_STREAMS + pages,
            "a walk of {WALK_STREAMS} streams in {pages} pages enumerated {} streams",
            walk.streams_enumerated
        );
        Ok(())
    }

    /// A zero limit has no page to fill and is a typed refusal, never a silent
    /// empty answer that a drain loop would read as the end of the scan.
    #[test]
    fn a_zero_limit_is_refused() -> Result<(), Box<dyn std::error::Error>> {
        let store = seeded()?;

        let refused = block_on(store.scan_page(PREFIX, None, 0))?;

        assert!(
            matches!(refused, Err(DurabilityError::ConfigError(_))),
            "a zero limit is refused as a configuration error, got {refused:?}"
        );
        Ok(())
    }
}

#[cfg(test)]
mod truncate_cost_tests {
    //! A prune far into a stream's history does engine work in proportion to
    //! the rows it deletes, not to the sequence it cuts at (SRV-014 R7).
    //!
    //! haematite 0.15.0 staged a truncate by walking every sequence from 0 to
    //! the cut, so the self streams' prune grew with their history. The pinned
    //! rev reads only the stream's present entries. The pin counts the engine's
    //! node reads (cache hits included) across one prune of a few rows at a
    //! sequence far past them.

    use std::error::Error;
    use std::sync::Arc;

    use super::super::bridge::block_on;
    use super::{Database, DatabaseConfig, DurableStore, EventStore, HaematiteStore};

    const STREAM: &str = "truncate-cost";
    /// The stream's history before the prune under test.
    const HISTORY: u64 = 10_000;
    /// Rows appended per batch while the history is written.
    const BATCH: u64 = 1_000;
    /// Rows still present when the prune under test runs.
    const PRESENT: u64 = 10;
    /// Node reads one prune may take beyond one per deleted row: the path
    /// down the tree, the commit, and the reclaim of the nodes it replaces.
    const OVERHEAD: u64 = 64;

    #[test]
    fn a_prune_far_into_history_reads_only_what_it_deletes() -> Result<(), Box<dyn Error>> {
        let dir = tempfile::tempdir()?;
        let database = Database::create(
            DatabaseConfig {
                data_dir: dir.path().to_path_buf(),
                shard_count: 1,
                distributed: None,
                executor_threads: None,
                node_cache_budget: Some(haematite::NodeCacheBudget::Unlimited),
                retention: haematite::Retention::reclaim(0),
            },
            haematite::RuntimeOptions {
                long_wait_interval: std::time::Duration::from_secs(5),
            },
        )?;
        let events = Arc::new(EventStore::new(database));
        let store = HaematiteStore::new(Arc::clone(&events));
        let batch: Vec<Vec<u8>> = (0..BATCH).map(|_| vec![0x07]).collect();
        let mut next = 0;
        while next < HISTORY {
            block_on(store.append_all(STREAM, batch.clone(), next))??;
            next += BATCH;
        }
        block_on(store.truncate_before(STREAM, HISTORY - PRESENT))??;

        let before = events.database().store_counts().node_gets;
        let deleted = block_on(store.truncate_before(STREAM, HISTORY))??;
        let reads = events.database().store_counts().node_gets - before;
        assert_eq!(
            (deleted, reads <= PRESENT + OVERHEAD),
            (PRESENT, true),
            "a prune of {deleted} rows at sequence {HISTORY} read {reads} nodes; at most \
             {PRESENT} plus {OVERHEAD} is owed: the prune walked the history, not the rows"
        );
        drop(store);
        drop(events);
        drop(dir);
        Ok(())
    }
}

#[cfg(test)]
#[path = "store_batch_tests.rs"]
mod batch_tests;