liminal/durability/

counting.rs

//! A counting seam around any [`DurableStore`].
//!
//! Every durable operation liminal makes goes through a [`DurableStore`], so a
//! wrapper at that seam is where store work per journey can be counted: calls
//! per method, rows read, rows and bytes written, and flushes. The wrapper is
//! a mirror, not a filter: each method is forwarded once, explicitly, and its
//! answer is handed back untouched, so a caller cannot tell it is there.
//!
//! `append_all` and `read_at` are forwarded explicitly rather than left to the
//! trait's defaults. A store may override both (haematite commits a batch as
//! one write and answers a point read without a range read), and falling back
//! to the defaults here would turn one batch into one write per row.
//!
//! What the seam cannot see is work a store does inside one call: the engine
//! rows behind a short `read_from`, and the streams a `scan_page` lists. Both
//! happen below the trait, and an answer carries no record of them.
//!
//! Cost, measured once by `the_wrapper_cost_per_call_is_measured_over_a_hundred_thousand_calls`
//! on 6 October 2026 on Tom's Mac (debug build; Dean's laptop was offline):
//! median 230 ns per `read_value` wrapped against 140 ns bare, so 90 ns per
//! call for the seam, over 100,000 calls each. To be re-measured on Dean's
//! laptop.

use std::fmt;
use std::sync::{Arc, Mutex, OnceLock};
use std::sync::atomic::{AtomicU64, Ordering};

use super::{DurabilityError, DurableStore, ScanCursor, ScanPage, StoredEntry};

/// Every counter a [`CountingStore`] keeps, read at one moment.
///
/// Counters only grow, so the work between two snapshots is
/// [`StoreCounts::since`] of the later over the earlier.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct StoreCounts {
    /// Atomic routed batch calls, including refusals.
    pub apply_routed_batch_calls: u64,
    /// Conditional replacements in successful routed batches.
    pub apply_routed_batch_items: u64,
    /// Routed point-read calls, including errors.
    pub get_routed_calls: u64,
    /// Values returned by routed point reads.
    pub get_routed_rows: u64,

    /// `append` calls.
    pub append_calls: u64,
    /// Rows written by `append` calls that succeeded.
    pub append_rows: u64,
    /// Payload bytes written by `append` calls that succeeded.
    pub append_bytes: u64,
    /// `append_all` calls.
    pub append_all_calls: u64,
    /// Rows written by `append_all` calls that succeeded.
    pub append_all_rows: u64,
    /// Payload bytes written by `append_all` calls that succeeded.
    pub append_all_bytes: u64,
    /// `read_from` calls.
    pub read_from_calls: u64,
    /// Rows returned by `read_from`.
    pub read_from_rows: u64,
    /// `read_at` calls.
    pub read_at_calls: u64,
    /// Rows returned by `read_at`: one per call that found its row.
    pub read_at_rows: u64,
    /// `cas` calls.
    pub cas_calls: u64,
    /// `read_value` calls.
    pub read_value_calls: u64,
    /// `read_stream_next_seq` calls.
    pub read_stream_next_seq_calls: u64,
    /// `scan_page` calls.
    pub scan_page_calls: u64,
    /// Rows returned by `scan_page`.
    pub scan_page_rows: u64,
    /// `flush` calls.
    pub flushes: u64,
    /// `truncate_before` calls.
    pub truncate_calls: u64,
    /// Rows deleted by `truncate_before` calls that succeeded.
    pub truncate_rows: u64,
}

impl StoreCounts {
    /// The work counted after `earlier` was taken.
    ///
    /// `earlier` must be a snapshot of the same store taken before `self`.
    /// Counters never decrease, so each difference is the work between the
    /// two; a counter that reads lower in `self` answers zero rather than
    /// wrapping.
    #[must_use]
    pub const fn since(&self, earlier: &Self) -> Self {
        Self {
            apply_routed_batch_calls: self
                .apply_routed_batch_calls
                .saturating_sub(earlier.apply_routed_batch_calls),
            apply_routed_batch_items: self
                .apply_routed_batch_items
                .saturating_sub(earlier.apply_routed_batch_items),
            get_routed_calls: self
                .get_routed_calls
                .saturating_sub(earlier.get_routed_calls),
            get_routed_rows: self.get_routed_rows.saturating_sub(earlier.get_routed_rows),

            append_calls: self.append_calls.saturating_sub(earlier.append_calls),
            append_rows: self.append_rows.saturating_sub(earlier.append_rows),
            append_bytes: self.append_bytes.saturating_sub(earlier.append_bytes),
            append_all_calls: self
                .append_all_calls
                .saturating_sub(earlier.append_all_calls),
            append_all_rows: self.append_all_rows.saturating_sub(earlier.append_all_rows),
            append_all_bytes: self
                .append_all_bytes
                .saturating_sub(earlier.append_all_bytes),
            read_from_calls: self.read_from_calls.saturating_sub(earlier.read_from_calls),
            read_from_rows: self.read_from_rows.saturating_sub(earlier.read_from_rows),
            read_at_calls: self.read_at_calls.saturating_sub(earlier.read_at_calls),
            read_at_rows: self.read_at_rows.saturating_sub(earlier.read_at_rows),
            cas_calls: self.cas_calls.saturating_sub(earlier.cas_calls),
            read_value_calls: self
                .read_value_calls
                .saturating_sub(earlier.read_value_calls),
            read_stream_next_seq_calls: self
                .read_stream_next_seq_calls
                .saturating_sub(earlier.read_stream_next_seq_calls),
            scan_page_calls: self.scan_page_calls.saturating_sub(earlier.scan_page_calls),
            scan_page_rows: self.scan_page_rows.saturating_sub(earlier.scan_page_rows),
            flushes: self.flushes.saturating_sub(earlier.flushes),
            truncate_calls: self.truncate_calls.saturating_sub(earlier.truncate_calls),
            truncate_rows: self.truncate_rows.saturating_sub(earlier.truncate_rows),
        }
    }

    /// The sum of two sets of counts, for work spread over two stores.
    #[must_use]
    pub const fn plus(&self, other: &Self) -> Self {
        Self {
            apply_routed_batch_calls: self
                .apply_routed_batch_calls
                .saturating_add(other.apply_routed_batch_calls),
            apply_routed_batch_items: self
                .apply_routed_batch_items
                .saturating_add(other.apply_routed_batch_items),
            get_routed_calls: self.get_routed_calls.saturating_add(other.get_routed_calls),
            get_routed_rows: self.get_routed_rows.saturating_add(other.get_routed_rows),

            append_calls: self.append_calls.saturating_add(other.append_calls),
            append_rows: self.append_rows.saturating_add(other.append_rows),
            append_bytes: self.append_bytes.saturating_add(other.append_bytes),
            append_all_calls: self.append_all_calls.saturating_add(other.append_all_calls),
            append_all_rows: self.append_all_rows.saturating_add(other.append_all_rows),
            append_all_bytes: self.append_all_bytes.saturating_add(other.append_all_bytes),
            read_from_calls: self.read_from_calls.saturating_add(other.read_from_calls),
            read_from_rows: self.read_from_rows.saturating_add(other.read_from_rows),
            read_at_calls: self.read_at_calls.saturating_add(other.read_at_calls),
            read_at_rows: self.read_at_rows.saturating_add(other.read_at_rows),
            cas_calls: self.cas_calls.saturating_add(other.cas_calls),
            read_value_calls: self.read_value_calls.saturating_add(other.read_value_calls),
            read_stream_next_seq_calls: self
                .read_stream_next_seq_calls
                .saturating_add(other.read_stream_next_seq_calls),
            scan_page_calls: self.scan_page_calls.saturating_add(other.scan_page_calls),
            scan_page_rows: self.scan_page_rows.saturating_add(other.scan_page_rows),
            flushes: self.flushes.saturating_add(other.flushes),
            truncate_calls: self.truncate_calls.saturating_add(other.truncate_calls),
            truncate_rows: self.truncate_rows.saturating_add(other.truncate_rows),
        }
    }

    /// Every counter with its name, in the one fixed order [`fmt::Display`]
    /// prints them in.
    #[must_use]
    pub const fn named(&self) -> [(&'static str, u64); 22] {
        [
            ("append_calls", self.append_calls),
            ("append_rows", self.append_rows),
            ("append_bytes", self.append_bytes),
            ("append_all_calls", self.append_all_calls),
            ("append_all_rows", self.append_all_rows),
            ("append_all_bytes", self.append_all_bytes),
            ("read_from_calls", self.read_from_calls),
            ("read_from_rows", self.read_from_rows),
            ("read_at_calls", self.read_at_calls),
            ("read_at_rows", self.read_at_rows),
            ("cas_calls", self.cas_calls),
            ("read_value_calls", self.read_value_calls),
            (
                "read_stream_next_seq_calls",
                self.read_stream_next_seq_calls,
            ),
            ("scan_page_calls", self.scan_page_calls),
            ("scan_page_rows", self.scan_page_rows),
            ("flushes", self.flushes),
            ("truncate_calls", self.truncate_calls),
            ("truncate_rows", self.truncate_rows),
            ("apply_routed_batch_calls", self.apply_routed_batch_calls),
            ("apply_routed_batch_items", self.apply_routed_batch_items),
            ("get_routed_calls", self.get_routed_calls),
            ("get_routed_rows", self.get_routed_rows),
        ]
    }
}

/// One counter per line, `name value`, in the order of [`StoreCounts::named`],
/// so two runs compare line by line.
impl fmt::Display for StoreCounts {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        for (name, value) in self.named() {
            writeln!(formatter, "{name} {value}")?;
        }
        Ok(())
    }
}

/// Receives counter increments and explicit flush calls at the counting boundary.
///
/// The callback must not inspect history or perform work for ordinary increments.
/// Flush observations describe the boundary call, including a call that fails.
pub trait StoreCountObserver: fmt::Debug + Send + Sync {
    /// Called after an actual counter increment, with no publication origin.
    fn changed(&self);

    /// Called once per explicit flush; a row count is absent unless the boundary knows it.
    fn flushed(&self, store: &str, batch_rows: Option<u64>);
}

#[derive(Debug)]
struct ObserverBinding {
    name: String,
    observer: Arc<dyn StoreCountObserver>,
}

static OBSERVER_INSTALL: Mutex<()> = Mutex::new(());

/// The live counters behind a [`CountingStore`].
#[derive(Debug, Default)]
struct Counters {
    apply_routed_batch_calls: AtomicU64,
    apply_routed_batch_items: AtomicU64,
    get_routed_calls: AtomicU64,
    get_routed_rows: AtomicU64,

    append_calls: AtomicU64,
    append_rows: AtomicU64,
    append_bytes: AtomicU64,
    append_all_calls: AtomicU64,
    append_all_rows: AtomicU64,
    append_all_bytes: AtomicU64,
    read_from_calls: AtomicU64,
    read_from_rows: AtomicU64,
    read_at_calls: AtomicU64,
    read_at_rows: AtomicU64,
    cas_calls: AtomicU64,
    read_value_calls: AtomicU64,
    read_stream_next_seq_calls: AtomicU64,
    scan_page_calls: AtomicU64,
    scan_page_rows: AtomicU64,
    flushes: AtomicU64,
    truncate_calls: AtomicU64,
    truncate_rows: AtomicU64,
    observer: OnceLock<ObserverBinding>,
}

impl Counters {
    fn add(&self, counter: &AtomicU64, amount: u64) {
        if amount == 0 {
            return;
        }
        counter.fetch_add(amount, Ordering::Relaxed);
        if let Some(binding) = self.observer.get() {
            binding.observer.changed();
        }
    }

    fn flushed(&self) {
        if let Some(binding) = self.observer.get() {
            binding.observer.flushed(&binding.name, None);
        }
    }

    /// Every counter as it stands now, each read once without a lock.
    fn snapshot(&self) -> StoreCounts {
        let read = |counter: &AtomicU64| counter.load(Ordering::Relaxed);
        StoreCounts {
            apply_routed_batch_calls: read(&self.apply_routed_batch_calls),
            apply_routed_batch_items: read(&self.apply_routed_batch_items),
            get_routed_calls: read(&self.get_routed_calls),
            get_routed_rows: read(&self.get_routed_rows),
            append_calls: read(&self.append_calls),
            append_rows: read(&self.append_rows),
            append_bytes: read(&self.append_bytes),
            append_all_calls: read(&self.append_all_calls),
            append_all_rows: read(&self.append_all_rows),
            append_all_bytes: read(&self.append_all_bytes),
            read_from_calls: read(&self.read_from_calls),
            read_from_rows: read(&self.read_from_rows),
            read_at_calls: read(&self.read_at_calls),
            read_at_rows: read(&self.read_at_rows),
            cas_calls: read(&self.cas_calls),
            read_value_calls: read(&self.read_value_calls),
            read_stream_next_seq_calls: read(&self.read_stream_next_seq_calls),
            scan_page_calls: read(&self.scan_page_calls),
            scan_page_rows: read(&self.scan_page_rows),
            flushes: read(&self.flushes),
            truncate_calls: read(&self.truncate_calls),
            truncate_rows: read(&self.truncate_rows),
        }
    }
}

/// A cloneable reader of one or more [`CountingStore`]s' counters.
///
/// Handed out where a store is built, beside the store itself, so a test can
/// take a snapshot of a store it only holds as `Arc<dyn DurableStore>`. A
/// handle over several stores answers their sum: the work of one journey,
/// wherever it landed.
#[derive(Clone, Debug, Default)]
pub struct StoreCountsHandle {
    stores: Vec<Arc<Counters>>,
}

impl StoreCountsHandle {
    /// The number of independently counted stores read by this handle.
    #[must_use]
    pub fn store_count(&self) -> usize {
        self.stores.len()
    }

    /// Installs one observer with one nonempty name for each counted store.
    ///
    /// # Errors
    /// Refuses absent stores, a name count mismatch, empty names, duplicate stores,
    /// an already installed observer, or a poisoned installation lock. Refusals
    /// detected during validation leave every store unchanged.
    pub fn install_observer(
        &self,
        names: &[&str],
        observer: Arc<dyn StoreCountObserver>,
    ) -> Result<(), DurabilityError> {
        if self.stores.is_empty() || names.len() != self.stores.len() {
            return Err(observer_error(
                "one name is required for each counted store",
            ));
        }
        if names.iter().any(|name| name.trim().is_empty()) {
            return Err(observer_error("store name is empty"));
        }
        for (index, counters) in self.stores.iter().enumerate() {
            if self.stores[..index]
                .iter()
                .any(|earlier| Arc::ptr_eq(earlier, counters))
            {
                return Err(observer_error(
                    "the handle contains a duplicate counted store",
                ));
            }
        }
        let installation = OBSERVER_INSTALL
            .lock()
            .map_err(|_| observer_error("installation lock poisoned"))?;
        if self
            .stores
            .iter()
            .any(|counters| counters.observer.get().is_some())
        {
            return Err(observer_error("observer already installed"));
        }
        for ((counters, name), observer) in self
            .stores
            .iter()
            .zip(names)
            .zip(std::iter::repeat_n(observer, self.stores.len()))
        {
            counters
                .observer
                .set(ObserverBinding {
                    name: (*name).to_owned(),
                    observer,
                })
                .map_err(|_| observer_error("observer installation invariant failed"))?;
        }
        drop(installation);
        Ok(())
    }

    /// One handle answering the sum of `self` and `other`.
    #[must_use]
    pub fn with(mut self, other: &Self) -> Self {
        self.stores.extend(other.stores.iter().cloned());
        self
    }

    /// Every counter, summed over the stores this handle reads, as it stands
    /// now. See [`CountingStore::snapshot`] for what a snapshot taken during
    /// a call may miss.
    #[must_use]
    pub fn snapshot(&self) -> StoreCounts {
        self.stores
            .iter()
            .fold(StoreCounts::default(), |total, counters| {
                total.plus(&counters.snapshot())
            })
    }
}

/// Databases opened or created in this process through liminal's open paths.
static STORES_OPENED: AtomicU64 = AtomicU64::new(0);

/// How many durable databases this process has opened or created through
/// liminal's open paths, ephemeral and persistent alike. Only grows.
#[must_use]
pub fn stores_opened() -> u64 {
    STORES_OPENED.load(Ordering::Relaxed)
}

/// Records one durable database opened or created. Every open path calls it
/// once, after the database is open, so a failed open is never counted.
pub fn record_store_opened() {
    add(&STORES_OPENED, 1);
}

/// Adds `amount` to `counter`.
///
/// Relaxed ordering: each counter is a tally on its own, read only by a
/// snapshot, and orders nothing else.
fn add(counter: &AtomicU64, amount: u64) {
    counter.fetch_add(amount, Ordering::Relaxed);
}

fn observer_error(message: &str) -> DurabilityError {
    DurabilityError::ConfigError(format!("store count observer: {message}"))
}

/// A row or byte total as a counter increment; a `usize` always fits.
fn widen(amount: usize) -> u64 {
    u64::try_from(amount).unwrap_or(u64::MAX)
}

/// A [`DurableStore`] that forwards every method to `S` and counts the work.
///
/// Counter increments take no lock and allocate nothing. An installed observer
/// receives the increment and explicit flush call; every await is the inner store's.
#[derive(Debug)]
pub struct CountingStore<S> {
    inner: S,
    counters: Arc<Counters>,
}

impl<S: DurableStore> CountingStore<S> {
    /// Wraps `inner`, with every counter at zero.
    #[must_use]
    pub fn new(inner: S) -> Self {
        Self {
            inner,
            counters: Arc::default(),
        }
    }

    /// A handle that reads this store's counters, to be kept beside the store
    /// once it is held only as `Arc<dyn DurableStore>`.
    #[must_use]
    pub fn handle(&self) -> StoreCountsHandle {
        StoreCountsHandle {
            stores: vec![Arc::clone(&self.counters)],
        }
    }

    /// Every counter as it stands now.
    ///
    /// Each counter is read once, without a lock, so a snapshot taken while
    /// other calls are in flight may count a call whose rows it does not yet
    /// count. Between two steps of one journey nothing is in flight.
    #[must_use]
    pub fn snapshot(&self) -> StoreCounts {
        self.counters.snapshot()
    }
}

#[async_trait::async_trait]
impl<S: DurableStore> DurableStore for CountingStore<S> {
    fn reclaim_spine(&self) -> Option<std::sync::Arc<haematite::EventStore>> {
        self.inner.reclaim_spine()
    }

    fn max_request_bytes(&self) -> Option<usize> {
        self.inner.max_request_bytes()
    }

    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<super::RoutedBatchItem>,
    ) -> Result<(), DurabilityError> {
        self.counters
            .add(&self.counters.apply_routed_batch_calls, 1);
        let count = widen(items.len());
        self.inner.apply_routed_batch(route_key, items).await?;
        self.counters
            .add(&self.counters.apply_routed_batch_items, count);
        Ok(())
    }

    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, DurabilityError> {
        self.counters.add(&self.counters.get_routed_calls, 1);
        let value = self.inner.get_routed(route_key, key).await?;
        if value.is_some() {
            self.counters.add(&self.counters.get_routed_rows, 1);
        }
        Ok(value)
    }

    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.counters.add(&self.counters.append_calls, 1);
        let bytes = widen(payload.len());
        let assigned = self.inner.append(stream_key, payload, expected_seq).await?;
        self.counters.add(&self.counters.append_rows, 1);
        self.counters.add(&self.counters.append_bytes, bytes);
        Ok(assigned)
    }

    async fn read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        self.counters.add(&self.counters.read_from_calls, 1);
        let entries = self.inner.read_from(stream_key, offset, limit).await?;
        self.counters
            .add(&self.counters.read_from_rows, widen(entries.len()));
        Ok(entries)
    }

    async fn append_all(
        &self,
        stream_key: &str,
        payloads: Vec<Vec<u8>>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.counters.add(&self.counters.append_all_calls, 1);
        let rows = widen(payloads.len());
        let bytes = payloads
            .iter()
            .map(|payload| widen(payload.len()))
            .fold(0_u64, u64::saturating_add);
        let next = self
            .inner
            .append_all(stream_key, payloads, expected_seq)
            .await?;
        self.counters.add(&self.counters.append_all_rows, rows);
        self.counters.add(&self.counters.append_all_bytes, bytes);
        Ok(next)
    }

    async fn read_at(
        &self,
        stream_key: &str,
        sequence: u64,
    ) -> Result<Option<StoredEntry>, DurabilityError> {
        self.counters.add(&self.counters.read_at_calls, 1);
        let entry = self.inner.read_at(stream_key, sequence).await?;
        if entry.is_some() {
            self.counters.add(&self.counters.read_at_rows, 1);
        }
        Ok(entry)
    }

    async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
        self.counters.add(&self.counters.cas_calls, 1);
        self.inner.cas(key, old_value, new_value).await
    }

    async fn read_stream_next_seq(&self, stream_key: &str) -> Result<Option<u64>, DurabilityError> {
        self.counters
            .add(&self.counters.read_stream_next_seq_calls, 1);
        self.inner.read_stream_next_seq(stream_key).await
    }

    async fn read_value(&self, key: &str) -> Result<Option<u64>, DurabilityError> {
        self.counters.add(&self.counters.read_value_calls, 1);
        self.inner.read_value(key).await
    }

    async fn scan_page(
        &self,
        prefix: &str,
        cursor: Option<ScanCursor>,
        limit: usize,
    ) -> Result<ScanPage, DurabilityError> {
        self.counters.add(&self.counters.scan_page_calls, 1);
        let page = self.inner.scan_page(prefix, cursor, limit).await?;
        self.counters
            .add(&self.counters.scan_page_rows, widen(page.entries.len()));
        Ok(page)
    }

    async fn flush(&self) -> Result<(), DurabilityError> {
        self.counters.add(&self.counters.flushes, 1);
        self.counters.flushed();
        self.inner.flush().await
    }

    async fn truncate_before(
        &self,
        stream_key: &str,
        before_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.counters.add(&self.counters.truncate_calls, 1);
        let deleted = self.inner.truncate_before(stream_key, before_seq).await?;
        self.counters.add(&self.counters.truncate_rows, deleted);
        Ok(deleted)
    }

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

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

    fn store_counts(&self) -> Option<StoreCountsHandle> {
        Some(self.handle())
    }
}

#[cfg(test)]
mod tests;