liminal-server 0.24.5

//! DUR-027 R7: a server's checkpoint and boot cost the same after 100 and
//! after 1,000 finished conversations, and what a retired conversation
//! leaves on disk is pinned.
//!
//! Each history is driven through the production handler once per test
//! binary into a one-shard disk store the fixture cache keeps and copies per
//! test (the build runs under plain-fsync fences, `tests_store_open`, never
//! the device flush of every durable commit; fixtures are built once,
//! Tom, 3 Oct), every finished conversation in the shape the live door proves
//! (`e2e_terminal_drain`): a sender enrols, attaches and posts one record
//! to a peer that never acknowledges it, so the sender's lost connection
//! leaves its Died pending in the candidate lane; the peer's own publish
//! drains it live; the peer's lost connection then leaves the last token's
//! Died pending, which only a boot's drain erases. That drain seals the
//! conversation and the commit that sealed it retires it (R5). One
//! conversation stays live, its member detached cleanly so its token
//! survives. Two handlers are then restored over the store: the closing
//! boot, whose drains do the closes (counted as the positive control), and
//! the measured boot after it, booted the way a test boot boots (the warm
//! finished) and counted: the rows the observer folds and the progress
//! entries it holds, the ids the registry answers, the conversations the
//! restore restores, and what the store was asked. The quantities that are
//! the size of the live state are asserted equal across the two histories;
//! the quantities that are the size of a checkpoint tail (the observer rows
//! folded, the registry's routed reads and snapshot bytes) are held under
//! the ceilings in `tests/ceilings/store-counts.toml`, each derived from
//! its period there. Milliseconds are printed, never asserted (rule 11:
//! counts are the ratchet).
//!
//! What this shape also shows: a conversation whose only member's
//! connection is lost with nothing owed commits that Died in place, keeps
//! the token, and is never closed, so it is never retired. That is the
//! F8B seal as ruled (only a drain erases a token) and is a question for
//! the brief, not a fact this test hides.

use std::collections::BTreeSet;
use std::error::Error;
use std::sync::Arc;
use std::time::Instant;

use liminal::durability::DurableStore;
use liminal::durability::bridge::block_on;
use liminal_protocol::wire::{
    ClientRequest, ConnectionIncarnation, EnrollmentRequest, EnrollmentToken, Generation,
    RecordAdmission, RecordAdmissionAttemptToken, ServerValue,
};

use crate::server::participant::{
    ConnectionFateClass, ConnectionFateWorkItem, ParticipantConnectionConversations,
};

use super::ProductionParticipantHandler;
use super::log::{
    DecodedStoredOperation, GENERATION_KEY_PREFIX, GENERATION_STREAM_PREFIX, OperationLog,
    STREAM_PREFIX, StoredOperation, StoredTerminalDisposition,
};
use super::observer::{ObserverLog, ObserverRow, encode_for_tests};
use super::outbox_log::OUTBOX_STREAM_PREFIX;
use super::registry::{ConversationRegistry, REGISTRY_STREAM_KEY};
use super::registry_snapshot::SNAPSHOT_KEY;
use super::restore_checkpoint_locator::checkpoint_conversation_route;
use super::route_keys::conversation_windows;
use super::tests::{dispatch_tracked, open_disk_store_for_tests, test_participant_config};
use super::tests_store_open::open_store_for_fixture_build;
use super::tests_store_tally::{Asked, Tally, under};
use super::tests_terminal_in_place::{admit, drop_connection, enroll_and_attach};

type TestResult<T = ()> = Result<T, Box<dyn Error>>;

/// The two histories the slope is taken across: finished conversations, with
/// the same one live conversation beside them.
const HISTORIES: [u64; 2] = [100, 1_000];
/// The conversation that stays live.
const LIVE: u64 = 0x7_0000;
/// The first finished conversation; the rest follow in id order.
const FIRST_FINISHED: u64 = 0x7_0001;
/// The ceilings table this test is pinned in.
const TABLE: &str = "conversation_retirement_lifetime";

/// The door's proven residence config (`e2e_terminal_drain`): a retained
/// record cap of 4 rows, so a member's terminal has no room to commit in
/// place and pends on the debt its peer owes. Every open of the store uses
/// it, so a boot restores under the cap its history was written under.
///
/// A finished conversation holds its two members' identities as retired
/// slots at the server scope until the closing boot retires it, so the
/// larger history needs room for 2,000 of them: the test default of 1,024
/// refused the 513th conversation's peer (measured 17 October 2026).
fn at_cap_config() -> crate::config::types::ParticipantConfig {
    let mut config = test_participant_config();
    config.max_retained_record_rows = 4;
    config.max_retired_identity_slots_server = 4_096;
    config
}

fn committed(value: &ServerValue) -> TestResult {
    if matches!(value, ServerValue::RecordCommitted(_)) {
        Ok(())
    } else {
        Err(format!("the record did not commit: {value:?}").into())
    }
}

/// What building one history cost: its durable commits and its time.
#[derive(Debug)]
struct Built {
    commits: u64,
    millis: u128,
}

/// Whether `conversation_id`'s log ends in a Died terminal left pending in
/// the candidate lane: the shape only a boot's drain closes.
fn died_pending(store: Arc<dyn DurableStore>, conversation_id: u64) -> TestResult<bool> {
    let log = OperationLog::new(store, conversation_id);
    let mut sequence = 0_u64;
    let mut pending = false;
    while let Some(entry) = block_on(log.read_at(sequence))?? {
        if let DecodedStoredOperation::V3(StoredOperation::Died { row }) = entry.operation {
            pending =
                row.disposition == StoredTerminalDisposition::Pending && row.drained.is_none();
        }
        sequence = sequence.checked_add(1).ok_or("log sequence overflowed")?;
    }
    Ok(pending)
}

/// A peer enrolled on a conversation without attaching: the recipient
/// whose unacknowledged delivery leaves the sender's Died pending.
struct Peer {
    connection: ConnectionIncarnation,
    conversations: ParticipantConnectionConversations,
    participant_id: u64,
}

fn enroll_peer(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
    connection: ConnectionIncarnation,
) -> TestResult<Peer> {
    let mut conversations = ParticipantConnectionConversations::default();
    let enrolled = dispatch_tracked(
        handler,
        connection,
        &mut conversations,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id,
            enrollment_token: EnrollmentToken::new([0x31; 16]),
        }),
    )?;
    let ServerValue::EnrollBound(peer) = enrolled else {
        return Err(format!("the peer did not bind: {enrolled:?}").into());
    };
    Ok(Peer {
        connection,
        conversations,
        participant_id: peer.participant_id(),
    })
}

/// The peer posts one record (the publish that drains the sender's pending
/// Died live) and loses its connection, leaving the last token's Died
/// pending.
fn peer_drains_then_dies(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
    mut peer: Peer,
    open_sequence: u64,
) -> TestResult {
    committed(&dispatch_tracked(
        handler,
        peer.connection,
        &mut peer.conversations,
        ClientRequest::RecordAdmission(RecordAdmission {
            conversation_id,
            participant_id: peer.participant_id,
            capability_generation: Generation::ONE,
            record_admission_attempt_token: RecordAdmissionAttemptToken::new([0x32; 16]),
            payload: vec![0x32],
        }),
    )?)?;
    handler
        .apply_connection_fate_with_impacts(&ConnectionFateWorkItem {
            open_sequence,
            connection_incarnation: peer.connection,
            class: ConnectionFateClass::ConnectionLost,
            tracked_conversations: peer.conversations.tracked_conversations(),
        })
        .into_result()?;
    Ok(())
}

/// Creates, posts to and abandons `count` conversations beside the one live
/// conversation, each left with its last token's Died pending in the lane
/// for the closing boot to drain.
fn finish_conversations(store: &Arc<Tally>, count: u64) -> TestResult<Built> {
    let handler = ProductionParticipantHandler::new(store.shared(), at_cap_config())?;
    let mut live = enroll_and_attach(&handler, LIVE, ConnectionIncarnation::new(7, 1), 0x11)?;
    committed(&admit(&handler, LIVE, &mut live, 0x12)?)?;
    // A clean disconnect keeps the token: the conversation stays open across
    // the restarts with nothing bound, so no startup fate touches it.
    drop_connection(&handler, &live, ConnectionFateClass::CleanDisconnect, 1)?;
    store.take()?;
    let started = Instant::now();
    for index in 0..count {
        let conversation_id = FIRST_FINISHED + index;
        let ordinal = index
            .checked_mul(2)
            .ok_or("connection ordinal overflowed")?;
        let sender = ConnectionIncarnation::new(7, ordinal + 2);
        let mut member = enroll_and_attach(&handler, conversation_id, sender, 0x21)?;
        let peer = enroll_peer(
            &handler,
            conversation_id,
            ConnectionIncarnation::new(7, ordinal + 3),
        )?;
        committed(&admit(&handler, conversation_id, &mut member, 0x22)?)?;
        drop_connection(
            &handler,
            &member,
            ConnectionFateClass::ConnectionLost,
            1_000 + ordinal,
        )?;
        if !died_pending(store.shared(), conversation_id)? {
            return Err(format!(
                "conversation {conversation_id}: the sender's Died did not pend in the lane"
            )
            .into());
        }
        peer_drains_then_dies(&handler, conversation_id, peer, 1_001 + ordinal)?;
        if !died_pending(store.shared(), conversation_id)? {
            return Err(format!(
                "conversation {conversation_id}: the peer's Died did not pend in the lane, so no boot drain will close it"
            )
            .into());
        }
    }
    let millis = started.elapsed().as_millis();
    let commits = store.take()?.commits;
    drop(handler);
    block_on(store.flush())??;
    Ok(Built { commits, millis })
}

/// The quantities a boot holds or reads that are the size of the live
/// state, never of the history: equal across the two histories.
#[derive(Debug, PartialEq, Eq)]
struct LiveSized {
    /// Conversations the warm restored.
    restored: Vec<u64>,
    observer_progress: usize,
    observer_armed: usize,
    /// The bytes a checkpoint of the restored observer aggregate takes.
    observer_checkpoint_bytes: usize,
    registry_live: BTreeSet<u64>,
    registry_sweeping: usize,
    conversation_streams: BTreeSet<String>,
    conversation_rows: u64,
    scan_routed_calls: u64,
    read_value_calls: u64,
}

/// What one boot over a finished history read and holds.
#[derive(Debug)]
struct Booted {
    asked: Asked,
    millis: u128,
    live: LiveSized,
    /// The registry snapshot the store holds, in bytes: the sets as they
    /// stood at its last crossing, so a tail-sized figure.
    registry_snapshot_bytes: usize,
}

fn boot(store: &Arc<Tally>) -> TestResult<Booted> {
    store.take()?;
    let started = Instant::now();
    let handler = ProductionParticipantHandler::new_restored(store.shared(), at_cap_config())?;
    let millis = started.elapsed().as_millis();
    let asked = store.take()?;
    let restored = handler.restored_conversation_ids()?;
    drop(handler);
    let observer = block_on(ObserverLog::new(store.shared()).restore())??;
    let checkpoint = encode_for_tests(&ObserverRow::snapshot_of(&observer.aggregate))?;
    let sets = ConversationRegistry::new(store.shared()).read_sets()?;
    let snapshot = block_on(store.get_routed(REGISTRY_STREAM_KEY, SNAPSHOT_KEY))??
        .ok_or("the registry holds no snapshot after its history")?;
    let live = LiveSized {
        restored,
        observer_progress: observer.aggregate.progress_rows().len(),
        observer_armed: observer.aggregate.armed_rows().len(),
        observer_checkpoint_bytes: checkpoint.len(),
        registry_live: sets.live,
        registry_sweeping: sets.sweeping.len(),
        conversation_streams: asked.conversation_streams.clone(),
        conversation_rows: asked.conversation_rows,
        scan_routed_calls: asked.scan_routed_calls,
        read_value_calls: asked.read_value_calls,
    };
    Ok(Booted {
        asked,
        millis,
        live,
        registry_snapshot_bytes: snapshot.len(),
    })
}

/// What one retired conversation left in the store: keys and their bytes.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
struct Residue {
    keys: u64,
    bytes: u64,
}

fn residue(store: &dyn DurableStore, conversation_id: u64) -> TestResult<Residue> {
    let mut streams = vec![
        format!("{STREAM_PREFIX}{conversation_id}"),
        format!("{OUTBOX_STREAM_PREFIX}{conversation_id}"),
    ];
    streams.extend(
        (1..=4)
            .map(|generation| format!("{GENERATION_STREAM_PREFIX}{generation}:{conversation_id}")),
    );
    let mut left = Residue::default();
    let mut count = |keys: usize, bytes: usize| {
        left.keys += u64::try_from(keys).unwrap_or(u64::MAX);
        left.bytes += u64::try_from(bytes).unwrap_or(u64::MAX);
    };
    for stream in &streams {
        // ast-grep-allow: no-read-from-start: the census counts every row left at any sequence
        let rows = block_on(store.read_from(stream, 0, usize::MAX))??;
        count(rows.len(), rows.iter().map(|row| row.payload.len()).sum());
        if block_on(store.read_stream_next_seq(stream))??.is_some() {
            count(1, std::mem::size_of::<u64>());
        }
    }
    if block_on(store.read_value(&format!("{GENERATION_KEY_PREFIX}{conversation_id}")))??.is_some()
    {
        count(1, std::mem::size_of::<u64>());
    }
    let route = checkpoint_conversation_route(conversation_id);
    for (from, to) in conversation_windows(conversation_id, &streams)? {
        let entries = block_on(store.scan_routed(&route, &from, &to))??;
        count(
            entries.len(),
            entries
                .iter()
                .map(|(key, value)| key.len() + value.len())
                .sum(),
        );
    }
    Ok(left)
}

/// The largest residue any of the `count` retired conversations left.
fn worst_residue(store: &Arc<Tally>, count: u64) -> TestResult<Residue> {
    let mut worst = Residue::default();
    for index in 0..count {
        let left = residue(store.inner.as_ref(), FIRST_FINISHED + index)?;
        if (left.keys, left.bytes) > (worst.keys, worst.bytes) {
            worst = left;
        }
    }
    Ok(worst)
}

/// One history: built, reopened for the closing boot, reopened again for
/// the measured boot, and censused.
struct Measured {
    /// The commits of the boot whose drains closed and retired the finished
    /// conversations, counted when the history was built.
    closing_commits: u64,
    booted: Booted,
    worst: Residue,
}

fn measure(history: u64) -> TestResult<Measured> {
    // The history and its closing boot run once per test binary (11,011 and
    // 15,010 commits for the larger one, each a device flush), the closing
    // commit count kept beside the store as the positive control; the test
    // boots a private copy and counts what the boot asks at the trait seam.
    let name = format!("retirement-history-{history}");
    let copy = crate::test_fixture::fixture_copy(&name, |data| {
        let store = Tally::new(open_store_for_fixture_build(&data.join("store"))?);
        let built = finish_conversations(&store, history)?;
        let closing = boot(&store)?;
        if closing.live.registry_live != BTreeSet::from([LIVE])
            || closing.live.registry_sweeping != 0
        {
            return Err(format!(
                "the closing boot after {history} conversations left the registry at {:?} live and {} sweeping",
                closing.live.registry_live, closing.live.registry_sweeping
            )
            .into());
        }
        println!(
            "history {history}: built once in {} ms with {} commits; closing boot {} ms with {} commits",
            built.millis, built.commits, closing.millis, closing.asked.commits
        );
        block_on(store.flush())??;
        std::fs::write(
            data.join("closing-commits"),
            closing.asked.commits.to_string(),
        )?;
        Ok(())
    })?;
    let closing_commits = std::fs::read_to_string(copy.path().join("closing-commits"))?.parse()?;
    let store = Tally::new(open_disk_store_for_tests(&copy.path().join("store"))?);
    let booted = boot(&store)?;
    let worst = worst_residue(&store, history)?;
    println!(
        "history {history}: measured boot {} ms; asked {:?}; holds {:?}; worst residue {worst:?}",
        booted.millis, booted.asked, booted.live
    );
    Ok(Measured {
        closing_commits,
        booted,
        worst,
    })
}

#[test]
fn a_boot_after_a_thousand_finished_conversations_costs_what_a_boot_after_a_hundred_does()
-> TestResult {
    let small = measure(HISTORIES[0])?;
    let large = measure(HISTORIES[1])?;

    // The positive control: the larger history did its extra closes, each
    // at least one commit of the closing boot.
    let extra = HISTORIES[1] - HISTORIES[0];
    if large.closing_commits < small.closing_commits + extra {
        return Err(format!(
            "the larger history's closing boot committed {} against the smaller's {}: fewer than its {extra} extra closes",
            large.closing_commits, small.closing_commits
        )
        .into());
    }

    // (a) the observer's entries and checkpoint, (b) the registry's ids and
    // snapshot, (c) the conversations restored and read: the size of the
    // live state after both histories.
    assert_eq!(
        small.booted.live, large.booted.live,
        "a boot's live-sized work grew with finished conversations: after {} against after {}",
        HISTORIES[0], HISTORIES[1]
    );
    assert_eq!(large.booted.live.registry_live, BTreeSet::from([LIVE]));
    assert_eq!(
        large.booted.live.conversation_streams,
        BTreeSet::from([format!("{OUTBOX_STREAM_PREFIX}{LIVE}")]),
        "the measured boot read one stream: the live conversation's outbox, never a finished one's and never a log"
    );

    // The tails a boot folds are bounded by their checkpoint periods, not by
    // the history: pinned as ceilings. The registry is a routed stream, so
    // each tail row it folds is one get_routed; the snapshot it reads holds
    // the ids as they stood at its last crossing. Measured 17 October 2026:
    // 62 against 202 routed reads, 264 against 824 snapshot bytes, each 17
    // and 72 beside one per tail row (45 and 185 rows).
    let worst = |read: fn(&Measured) -> u64| read(&small).max(read(&large));
    under(
        TABLE,
        "observer_rows_folded",
        worst(|m| m.booted.asked.observer_rows),
    )?;
    under(
        TABLE,
        "registry_snapshot_bytes",
        worst(|m| u64::try_from(m.booted.registry_snapshot_bytes).unwrap_or(u64::MAX)),
    )?;
    under(
        TABLE,
        "read_from_calls",
        worst(|m| m.booted.asked.read_from_calls),
    )?;
    under(
        TABLE,
        "get_routed_calls",
        worst(|m| m.booted.asked.get_routed_calls),
    )?;
    under(
        TABLE,
        "scan_routed_calls",
        worst(|m| m.booted.asked.scan_routed_calls),
    )?;
    under(
        TABLE,
        "read_value_calls",
        worst(|m| m.booted.asked.read_value_calls),
    )?;
    under(
        TABLE,
        "residue_keys_per_retired_conversation",
        worst(|m| m.worst.keys),
    )?;
    println!(
        "residue per retired conversation: {} keys, {} bytes (never zero: stream heads and the tombstone stay)",
        worst(|m| m.worst.keys),
        worst(|m| m.worst.bytes)
    );
    Ok(())
}