liminal-server 0.25.4

//! INCARNATION-STREAM-BOUND acceptance pin 1: the scale pin.
//!
//! `docs/design/INCARNATION-STREAM-BOUND-2026-08-29.md` states the law this
//! module measures: *history is bounded by policy, not accumulated until an
//! operator vacuums; boot never rebuilds anything proportional to the tree;
//! every such claim carries a pin that grows N and measures the time flat.*
//!
//! The measurement is COUNTS, never wall time. A duration on a shared machine
//! is a report about the machine; the row count a boot reads is a property of
//! the design and is identical on every box. Two counts are taken for each N:
//!
//! - **retained under policy** — the rows the durable stream of record holds
//!   after N rounds, which is what a store's occupancy grows with, and
//! - **replayed at the next startup** — every row a cold boot's reads actually
//!   deliver, counted at the [`DurableStore`] seam so a page that is fetched
//!   and discarded still counts.
//!
//! Both are asserted FLAT across a growing N. On the pre-fix tree both are
//! linear: one estate-global append-only stream replayed whole from genesis at
//! every boot.

use std::error::Error;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};

use liminal::durability::{
    DurabilityError, DurableStore, ScanCursor, ScanPage, StoredEntry, bridge::block_on,
    open_ephemeral,
};

use super::incarnation_stream::{
    ConnectionFateClass, IncarnationAllocation, IncarnationStartup, IncarnationStream,
};

/// Rows one full lifecycle round appends: Startup, Allocate, Open, Complete.
const ROWS_PER_ROUND: usize = 4;
/// Reads a bounded cold boot issues against the incarnation namespace: the
/// anchor walk's one successor probe, the page that delivers the epoch of
/// record, and the empty page that ends it.
const READS_PER_BOUNDED_BOOT: usize = 3;
/// Signed live-reference bound for the rounds below.
const REFERENCE_BOUND: usize = 4;
/// Signed conversation bound for the connection-fate Opens below.
const CONVERSATION_BOUND: usize = 2;
/// Round counts the pin grows through. The largest is far enough above the
/// smallest that a linear shape cannot be mistaken for a flat one.
const ROUND_COUNTS: [usize; 4] = [1, 4, 16, 64];
/// Read limit large enough to deliver any stream this pin builds in one call.
const WHOLE_STREAM: usize = 1 << 20;

/// Counts every durable row an incarnation-stream read DELIVERS.
///
/// Wrapping the store is the only place the count is honest: a reader that
/// pages and discards, or that fetches a suffix and truncates it, has already
/// paid for the rows by the time its caller sees them.
#[derive(Debug)]
struct CountingStore {
    inner: Arc<dyn DurableStore>,
    entries: AtomicUsize,
    reads: AtomicUsize,
}

impl CountingStore {
    fn new(inner: Arc<dyn DurableStore>) -> Self {
        Self {
            inner,
            entries: AtomicUsize::new(0),
            reads: AtomicUsize::new(0),
        }
    }

    /// Rows delivered from the incarnation namespace since construction.
    fn entries(&self) -> usize {
        self.entries.load(Ordering::SeqCst)
    }

    /// Reads issued against the incarnation namespace since construction.
    fn reads(&self) -> usize {
        self.reads.load(Ordering::SeqCst)
    }
}

#[async_trait::async_trait]
impl DurableStore for CountingStore {
    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<liminal::durability::RoutedBatchItem>,
    ) -> Result<(), liminal::durability::DurabilityError> {
        self.inner.apply_routed_batch(route_key, items).await
    }

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

    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.inner.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.inner
            .append_all(stream_key, payloads, expected_seq)
            .await
    }

    async fn read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        let entries = self.inner.read_from(stream_key, offset, limit).await?;
        if stream_key.starts_with(IncarnationStream::stream_key()) {
            self.reads.fetch_add(1, Ordering::SeqCst);
            self.entries.fetch_add(entries.len(), Ordering::SeqCst);
        }
        Ok(entries)
    }

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

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

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

    async fn scan_page(
        &self,
        prefix: &str,
        cursor: Option<ScanCursor>,
        limit: usize,
    ) -> Result<ScanPage, DurabilityError> {
        let page = self.inner.scan_page(prefix, cursor, limit).await?;
        if prefix.starts_with(IncarnationStream::stream_key()) {
            self.reads.fetch_add(1, Ordering::SeqCst);
            self.entries.fetch_add(page.entries.len(), Ordering::SeqCst);
        }
        Ok(page)
    }

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

/// One complete server lifetime through the REAL append path: the process
/// startup, one connection allocation, that connection's fate Open, and its
/// Complete. No fixture bytes and no seeded header — every row here is the row
/// production writes.
fn lifecycle_round(store: &Arc<dyn DurableStore>) -> Result<(), Box<dyn Error>> {
    let startup = block_on(IncarnationStream::new(Arc::clone(store), REFERENCE_BOUND).startup())??;
    let IncarnationStartup::Started(mut stream) = startup else {
        return Err("a lifecycle round could not start the incarnation stream".into());
    };
    let IncarnationAllocation::Allocated {
        connection_incarnation,
        ..
    } = block_on(stream.allocate(&[]))??
    else {
        return Err("a lifecycle round exhausted the connection-ordinal space".into());
    };
    let intent = block_on(stream.open_connection_fate(
        connection_incarnation,
        ConnectionFateClass::CleanDisconnect,
        CONVERSATION_BOUND,
        &[],
    ))??;
    block_on(stream.complete_connection_fate(intent.open_sequence))??;
    // The start thread settles the anchor once admission is open (SRV-048).
    block_on(stream.settle_anchor())??;
    Ok(())
}

/// Rows the durable stream of record holds — the occupancy policy bounds.
fn retained_rows(store: &Arc<dyn DurableStore>) -> Result<usize, Box<dyn Error>> {
    let key = block_on(IncarnationStream::stream_key_of_record(store))??;
    Ok(block_on(store.read_from(&key, 0, WHOLE_STREAM))??.len())
}

/// Rows one more cold boot's reads deliver, and how many reads it took.
fn next_startup_read_cost(inner: &Arc<dyn DurableStore>) -> Result<(usize, usize), Box<dyn Error>> {
    let counting = Arc::new(CountingStore::new(Arc::clone(inner)));
    let observed: Arc<dyn DurableStore> = Arc::clone(&counting) as Arc<dyn DurableStore>;
    let startup = block_on(IncarnationStream::new(observed, REFERENCE_BOUND).startup())??;
    let IncarnationStartup::Started(mut stream) = startup else {
        return Err("the measured boot could not start the incarnation stream".into());
    };
    let cost = (counting.entries(), counting.reads());
    block_on(stream.settle_anchor())??;
    Ok(cost)
}

/// Scale pin — law item 4. Both counts must be FLAT as N grows.
///
/// The pre-fix tree fails this on the first pair: one estate-global stream
/// retains `4N` rows and replays all `4N` of them at the next boot.
#[test]
fn bounded_incarnation_history_is_flat_as_lifecycle_rounds_grow() -> Result<(), Box<dyn Error>> {
    let mut measured: Vec<(usize, usize, usize, usize)> = Vec::new();
    for rounds in ROUND_COUNTS {
        let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
        for _ in 0..rounds {
            lifecycle_round(&store)?;
        }
        let retained = retained_rows(&store)?;
        let (replayed, reads) = next_startup_read_cost(&store)?;
        measured.push((rounds, retained, replayed, reads));
    }

    let Some(&(first_rounds, first_retained, first_replayed, first_reads)) = measured.first()
    else {
        return Err("the scale pin measured nothing".into());
    };
    for &(rounds, retained, replayed, reads) in &measured {
        assert_eq!(
            retained, first_retained,
            "rows retained under policy must not grow with lifecycle rounds: \
             {first_rounds} rounds retained {first_retained}, {rounds} rounds retained {retained} \
             (whole series: {measured:?})"
        );
        assert_eq!(
            replayed, first_replayed,
            "rows a cold boot replays must not grow with lifecycle rounds: \
             {first_rounds} rounds replayed {first_replayed}, {rounds} rounds replayed {replayed} \
             (whole series: {measured:?})"
        );
        assert_eq!(
            reads, first_reads,
            "reads a cold boot issues must not grow with lifecycle rounds \
             (whole series: {measured:?})"
        );
    }

    // Flat is necessary but not sufficient: a stream that retained everything
    // for every N would also be "equal", so pin the LEVEL at live state too.
    assert_eq!(
        first_retained, ROWS_PER_ROUND,
        "the stream of record must retain exactly one round's live state \
         (whole series: {measured:?})"
    );
    assert_eq!(
        first_replayed, ROWS_PER_ROUND,
        "a cold boot must replay exactly one round's live state \
         (whole series: {measured:?})"
    );
    assert_eq!(
        first_reads, READS_PER_BOUNDED_BOOT,
        "a cold boot's read count must be the bounded shape, not a page walk \
         (whole series: {measured:?})"
    );
    Ok(())
}