liminal-server 0.25.4

//! Where the incarnation stream's history is bounded by policy.
//!
//! One estate-global append-only stream replayed from genesis makes a
//! regressing scalar header unrepresentable, and costs a boot every row the
//! estate ever wrote. This module keeps the first property and pays a bounded
//! price for it, by rotating each server incarnation into its OWN stream:
//!
//! ```text
//! liminal/participant/incarnation/v2                       genesis (pre-fix)
//! liminal/participant/incarnation/v2/incarnation/{n}        epoch of incarnation n
//! liminal/participant/incarnation/v2/anchor                 CAS value naming n
//! ```
//!
//! **No event in any epoch carries a header.** An epoch's resume seed is read
//! off its own KEY — epoch `n` replays from the header `(n - 1, no ordinal, not
//! exhausted)` and its first row is the Startup that the protocol's own
//! [`prepare_server_incarnation_startup`](liminal_protocol::lifecycle::prepare_server_incarnation_startup)
//! turns into exactly `n`. The header is still a consequence of protocol
//! transitions applied to canonical inputs; the rotation only moves where those
//! inputs start, so the grammar has gained no field a writer could regress.
//!
//! What names the key is the anchor, and what makes the anchor safe is that it
//! is only ever read as a HINT and only ever rolled FORWARD. A boot walks from
//! the anchor over every successor epoch that actually exists in the store and
//! adopts the highest, so an anchor that lags — a crash between the epoch's
//! fsync and the anchor's compare-and-swap — is repaired, and an anchor that
//! names an incarnation below durable history is overruled by that history.
//! Regressing the resume point therefore requires DESTROYING durable epochs,
//! not merely writing a value; and an epoch whose canonical inputs do not
//! produce the incarnation its key names is refused outright
//! ([`IncarnationStreamError::EpochHeaderMismatch`]).
//!
//! What is traded is stated in
//! `docs/design/INCARNATION-STREAM-BOUND-2026-08-29.md`: epochs below the
//! anchored one become dead history for the store vacuum to reclaim, and this
//! module never deletes a durable row.

use liminal::durability::settle::cas_settled;
use std::num::NonZeroU64;
use std::sync::Arc;

use liminal::durability::DurableStore;
use liminal_protocol::lifecycle::ConnectionIncarnationAllocatorRestore;

use super::error::IncarnationStreamError;

/// The pre-rotation estate-global stream. Anchor absent means this is the
/// state of record, which is true of every store written before this fix.
pub(super) const GENESIS_STREAM_KEY: &str = "liminal/participant/incarnation/v2";
/// Compare-and-swap value naming the server incarnation whose epoch is of
/// record. A hint, never an authority: see this module's own documentation.
pub(super) const EPOCH_ANCHOR_KEY: &str = "liminal/participant/incarnation/v2/anchor";
/// Prefix each epoch stream's key is built from.
const EPOCH_STREAM_KEY_PREFIX: &str = "liminal/participant/incarnation/v2/incarnation/";
/// The one sequence an epoch's Startup row may occupy.
///
/// Appending it at exactly this sequence is what makes a pre-existing epoch a
/// loud append conflict instead of a silent second history.
pub(super) const EPOCH_STARTUP_SEQUENCE: u64 = 0;

/// The durable stream a boot replays, and the seed it replays it from.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum EpochSource {
    /// Pre-rotation history: replayed from the fixed protocol genesis, exactly
    /// as it was before this fix, so existing v2 estates still open.
    Genesis,
    /// One server incarnation's own epoch, seeded from the key that names it.
    Epoch(NonZeroU64),
}

impl EpochSource {
    /// Selects the source an anchor value names. Zero — which the durable
    /// store's "absent == 0" contract also spells as an absent key — is the
    /// genesis stream, because no server incarnation is ever zero.
    pub(super) const fn from_anchor(anchor: u64) -> Self {
        match NonZeroU64::new(anchor) {
            Some(server_incarnation) => Self::Epoch(server_incarnation),
            None => Self::Genesis,
        }
    }

    /// The durable stream key this source replays.
    pub(super) fn stream_key(self) -> String {
        match self {
            Self::Genesis => GENESIS_STREAM_KEY.to_owned(),
            Self::Epoch(server_incarnation) => epoch_stream_key(server_incarnation.get()),
        }
    }

    /// The anchor value that names this source.
    pub(super) const fn anchor_value(self) -> u64 {
        match self {
            Self::Genesis => 0,
            Self::Epoch(server_incarnation) => server_incarnation.get(),
        }
    }

    /// The header replay starts from — derived from the KEY, never from bytes
    /// inside the stream.
    ///
    /// For an epoch that is the incarnation immediately before the one its key
    /// names, so the epoch's own Startup row is the transition that produces
    /// it. `NonZeroU64` is what makes that subtraction total.
    pub(super) const fn resume_seed(self) -> ConnectionIncarnationAllocatorRestore {
        let server_incarnation = match self {
            Self::Genesis => 0,
            Self::Epoch(named) => named.get() - 1,
        };
        ConnectionIncarnationAllocatorRestore {
            server_incarnation,
            last_examined_connection_ordinal: None,
            connection_ordinal_exhausted: false,
        }
    }
}

/// The durable stream key holding one server incarnation's canonical inputs.
pub(super) fn epoch_stream_key(server_incarnation: u64) -> String {
    format!("{EPOCH_STREAM_KEY_PREFIX}{server_incarnation}")
}

/// Reads the anchor hint. An absent key is zero by the store's own contract.
///
/// # Errors
///
/// Returns [`IncarnationStreamError`] when the store cannot be read.
pub(super) async fn read_anchor(
    store: &Arc<dyn DurableStore>,
) -> Result<u64, IncarnationStreamError> {
    Ok(store.read_value(EPOCH_ANCHOR_KEY).await?.unwrap_or(0))
}

/// Whether an epoch for `server_incarnation` was ever durably started.
///
/// One point read of one row. The Startup row at
/// [`EPOCH_STARTUP_SEQUENCE`] is what "this incarnation exists" MEANS, so this
/// asks the store rather than trusting the anchor.
///
/// # Errors
///
/// Returns [`IncarnationStreamError`] when the store cannot be read.
pub(super) async fn epoch_exists(
    store: &Arc<dyn DurableStore>,
    server_incarnation: u64,
) -> Result<bool, IncarnationStreamError> {
    Ok(store
        .read_at(
            &epoch_stream_key(server_incarnation),
            EPOCH_STARTUP_SEQUENCE,
        )
        .await?
        .is_some())
}

/// Moves the anchor from `from` to `to`, with no flush of its own.
///
/// Compare-and-swap, so a concurrent mover is a refusal rather than a lost
/// update, and every caller here supplies a strictly greater `to`. The swap
/// becomes durable with the store's next flush (SRV-047 R1): the anchor is a
/// hint, and a swap lost in a crash is the lagging anchor every boot already
/// repairs by rolling forward over the epochs the store holds, so its own
/// fsync would buy nothing and cost one on every restart. The store's flush is
/// store-wide, so any later flush that makes another write durable makes this
/// swap durable with it.
///
/// # Errors
///
/// Returns [`IncarnationStreamError`] when the store refuses the swap —
/// including when another process has already moved the anchor.
pub(super) async fn advance_anchor(
    store: &Arc<dyn DurableStore>,
    from: u64,
    to: u64,
) -> Result<(), IncarnationStreamError> {
    cas_settled(store.as_ref(), EPOCH_ANCHOR_KEY, from, to).await?;
    Ok(())
}

/// The anchor swap a restart's rotation leaves unrun (SRV-048 R1): from the
/// source the start replayed to the epoch it opened.
///
/// The start publishes on its flushed Startup row and this swap is settled
/// after, off the path to admission: a lone compare-and-swap is a whole
/// durable commit of the store, and the anchor is a hint nothing reads while
/// this process serves. A swap never settled is the crash case the next boot's
/// forward walk already adopts.
#[derive(Debug)]
#[must_use = "an unsettled swap leaves the anchor for the next boot to roll forward"]
pub(in crate::server) struct AnchorSwap {
    store: Arc<dyn DurableStore>,
    from: u64,
    to: u64,
}

impl AnchorSwap {
    pub(super) const fn new(store: Arc<dyn DurableStore>, from: u64, to: u64) -> Self {
        Self { store, from, to }
    }

    /// The anchor value the swap expects to replace.
    pub(in crate::server) const fn from(&self) -> u64 {
        self.from
    }

    /// The incarnation the swap names.
    pub(in crate::server) const fn to(&self) -> u64 {
        self.to
    }

    /// Swaps the anchor to name this start's epoch, with its milliseconds.
    ///
    /// # Errors
    ///
    /// Returns [`IncarnationStreamError`] when the store refuses the swap,
    /// including when another writer has already moved the anchor.
    pub(in crate::server) async fn settle(self) -> Result<(), IncarnationStreamError> {
        let settling = std::time::Instant::now();
        advance_anchor(&self.store, self.from, self.to).await?;
        tracing::info!(
            from = self.from,
            to = self.to,
            anchor_ms = u64::try_from(settling.elapsed().as_millis()).unwrap_or(u64::MAX),
            "incarnation-stream anchor settled"
        );
        Ok(())
    }
}

impl super::StartedIncarnationStream {
    /// Takes the anchor swap this start left unrun, for its owner to settle
    /// once admission is open; none once taken (SRV-048 R1).
    pub(in crate::server) const fn take_anchor_swap(&mut self) -> Option<AnchorSwap> {
        self.anchor_swap.take()
    }

    /// Settles the anchor swap this start left unrun, as the start thread
    /// does; nothing once taken.
    ///
    /// # Errors
    ///
    /// Returns [`IncarnationStreamError`] when the store refuses the swap.
    #[cfg(test)]
    pub(in crate::server) async fn settle_anchor(&mut self) -> Result<(), IncarnationStreamError> {
        match self.take_anchor_swap() {
            Some(swap) => swap.settle().await,
            None => Ok(()),
        }
    }
}

/// Walks the anchor up over every successor epoch the store actually holds.
///
/// This is the whole of the non-regression argument at boot, and it is told by
/// the store: it stops at the first successor that does not exist, and it never
/// moves down. A crash between an epoch's Startup fsync and its anchor swap is
/// repaired here — the incarnation that epoch published is ADOPTED, never
/// re-minted — and an anchor that has somehow come to name an incarnation below
/// durable history is overruled by the epochs above it.
///
/// # Errors
///
/// Returns [`IncarnationStreamError`] when the store cannot be read or the
/// anchor swap is refused.
pub(super) async fn roll_anchor_forward(
    store: &Arc<dyn DurableStore>,
    anchor: u64,
) -> Result<u64, IncarnationStreamError> {
    let mut anchored = anchor;
    while anchored > 0 {
        let Some(successor) = anchored.checked_add(1) else {
            return Ok(anchored);
        };
        if !epoch_exists(store, successor).await? {
            return Ok(anchored);
        }
        advance_anchor(store, anchored, successor).await?;
        anchored = successor;
    }
    Ok(anchored)
}

#[cfg(test)]
#[path = "epoch_tests.rs"]
mod tests;