liminal-server 0.23.1

//! Boot-replay measurement over a large durable log (opt-in: feature
//! `boot-replay-measurement`).
//!
//! Generates one conversation with at least 20,000 operation rows (record
//! admissions and acks through the live production seam), lands every
//! stream on a real on-disk haematite database in a temp dir, reopens that
//! database cold, and times cold replays of it by two routes, alternating on
//! the same store in the same process so a loaded host disturbs both alike:
//! the pre-merge two-pass shape (the standalone audit pass, then the replay)
//! and the production single pass. The numbers are printed with
//! `--nocapture`; the assertions pin only that both routes reach the row
//! count that was written.
//!
//! The seam is driven over [`MemoryStore`] rather than the disk store
//! because haematite fsyncs every node it publishes on every append, and
//! 20,000 rows of that is hours on a loaded box; the rows are then copied to
//! disk in batches (`EventStore::append_batch`, one commit per batch) so the
//! bytes replay reads are the bytes the seam wrote, stored and compressed
//! exactly as the production store stores them.

use crate::config::types::ParticipantConfig;

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

use liminal::durability::{DurableStore, HaematiteStore, bridge::block_on};

use liminal_protocol::wire::{
    ClientRequest, ConnectionIncarnation, EnrollmentRequest, EnrollmentToken, Generation,
    ParticipantAck, RecordAdmission, RecordAdmissionAttemptToken, ServerValue,
};

use std::error::Error;
use std::sync::Arc;
use std::time::{Duration, Instant};

use super::ProductionParticipantHandler;
use super::log::OperationLog;
use super::ops_session_replay::validate_operation_schema;
use super::outbox::ConversationOutboxLimits;
use super::outbox_log::OutboxLog;
use super::state::ConversationAuthority;
use super::tests::{dispatch, test_participant_config};
use super::tests_memory_store::MemoryStore;

const CONVERSATION: u64 = 0xB007;
const TARGET_ROWS: u64 = 20_000;
const PAYLOAD_BYTES: usize = 256;
const COPY_BATCH: usize = 32;

fn enroll(
    handler: &ProductionParticipantHandler,
    connection: ConnectionIncarnation,
    token: u8,
) -> Result<u64, Box<dyn Error>> {
    let enrolled = dispatch(
        handler,
        connection,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id: CONVERSATION,
            enrollment_token: EnrollmentToken::new([token; 16]),
        }),
    )?;
    let ServerValue::EnrollBound(receipt) = enrolled else {
        return Err(format!("enrollment {token:#x} did not bind: {enrolled:?}").into());
    };
    Ok(receipt.participant_id())
}

fn ack(
    handler: &ProductionParticipantHandler,
    connection: ConnectionIncarnation,
    participant_id: u64,
    through_seq: u64,
) -> Result<ServerValue, Box<dyn Error>> {
    dispatch(
        handler,
        connection,
        ClientRequest::ParticipantAck(ParticipantAck {
            conversation_id: CONVERSATION,
            participant_id,
            capability_generation: Generation::ONE,
            through_seq,
        }),
    )
}

/// The deployment-shaped test configuration with the closure retention
/// ceiling raised past what 10,000 record cycles accrue (the default 2,048
/// retained entries refuses the 4,072nd admission), so the log reaches the
/// target length on records and acks alone.
const fn measurement_config() -> ParticipantConfig {
    let mut config = test_participant_config();
    config.retained_capacity_entries = 1 << 20;
    config.retained_capacity_bytes = 1 << 40;
    config
}

/// Drives the seam until the conversation's operation log holds at least
/// `TARGET_ROWS` rows.
fn generate(store: Arc<MemoryStore>) -> Result<(), Box<dyn Error>> {
    let recipient_connection = ConnectionIncarnation::new(914, 1);
    let sender_connection = ConnectionIncarnation::new(914, 2);
    let handler = ProductionParticipantHandler::new(store, measurement_config())?;
    let recipient = enroll(&handler, recipient_connection, 0x91)?;
    let sender = enroll(&handler, sender_connection, 0x92)?;
    let acknowledged = ack(&handler, recipient_connection, recipient, 2)?;
    if !matches!(acknowledged, ServerValue::AckCommitted(_)) {
        return Err(format!("recipient marker ack did not commit: {acknowledged:?}").into());
    }
    let mut written = 0_u64;
    let mut cycle = 0_u64;
    while written < TARGET_ROWS {
        let mut token = [0_u8; 16];
        token[..8].copy_from_slice(&cycle.to_be_bytes());
        let committed = dispatch(
            &handler,
            sender_connection,
            ClientRequest::RecordAdmission(RecordAdmission {
                conversation_id: CONVERSATION,
                participant_id: sender,
                capability_generation: Generation::ONE,
                record_admission_attempt_token: RecordAdmissionAttemptToken::new(token),
                payload: vec![u8::try_from(cycle % 251)?; PAYLOAD_BYTES],
            }),
        )?;
        let ServerValue::RecordCommitted(committed) = committed else {
            return Err(format!("record {cycle} did not commit: {committed:?}").into());
        };
        let acknowledged = ack(
            &handler,
            recipient_connection,
            recipient,
            committed.delivery_seq(),
        )?;
        if !matches!(acknowledged, ServerValue::AckCommitted(_)) {
            return Err(format!("ack {cycle} did not commit: {acknowledged:?}").into());
        }
        written = written.checked_add(2).ok_or("row count overflow")?;
        cycle = cycle.checked_add(1).ok_or("cycle overflow")?;
    }
    Ok(())
}

fn create_database(data_dir: &std::path::Path) -> Result<Database, Box<dyn Error>> {
    Ok(Database::create(
        DatabaseConfig {
            data_dir: data_dir.to_path_buf(),
            shard_count: 2,
            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),
            reclaim_start: haematite::ReclaimStart::AtOpen,
        },
    )?)
}

/// Whether `error` is haematite's shard REPLY timeout: the caller stopped
/// waiting, but the command is still in the shard actor's mailbox and will
/// run. haematite `=0.14.0` surfaces that timeout only as the text of
/// `DatabaseError::ShardError` (the `Display` of its `ShardError::ReplyTimeout`),
/// so the text is what is matched. Every other error — an unavailable actor,
/// a disconnected reply, a sequence conflict — is not a timeout and is
/// propagated.
fn is_reply_timeout(error: &ApiError) -> bool {
    matches!(
        error,
        ApiError::Storage(DatabaseError::ShardError(message))
            if message.starts_with("timed out waiting for shard actor")
    )
}

/// Runs an IDEMPOTENT shard command until its actor answers.
///
/// A command routed to a shard actor is answered in mailbox order, after
/// every command queued before it; a reply timeout means only that the
/// answer has not arrived yet. Only that timeout is asked again, each wait
/// is haematite's own, and each one is printed, so a wedged actor is visible
/// rather than silent.
fn until_answered<T>(
    what: &str,
    mut command: impl FnMut() -> Result<T, ApiError>,
) -> Result<T, Box<dyn Error>> {
    loop {
        match command() {
            Ok(answer) => return Ok(answer),
            Err(error) if is_reply_timeout(&error) => {
                eprintln!("BOOT-REPLAY COPY: {what} is still queued behind earlier work: {error}");
            }
            Err(error) => return Err(format!("{what} failed: {error}").into()),
        }
    }
}

/// Appends one batch at `next`, resuming across a shard reply timeout.
///
/// After a timeout the stream's head is read through the SAME shard actor
/// (`Database::read_stream_next_seq` routes on the stream key, as the append
/// does), so its answer comes after the timed-out commit has run: the head
/// says whether that commit landed (continue past it) or was refused (append
/// it again). Any other head is neither, and is an error naming the rows.
fn append_resuming(
    events: &EventStore,
    key: &str,
    batch: &[&[u8]],
    next: u64,
) -> Result<u64, Box<dyn Error>> {
    let landed = next
        .checked_add(u64::try_from(batch.len())?)
        .ok_or_else(|| format!("stream {key} row count overflowed at row {next}"))?;
    let refused = (next != 0).then_some(next);
    loop {
        match events.append_batch(key.as_bytes(), batch, next) {
            Ok(after) => return Ok(after),
            Err(error) if is_reply_timeout(&error) => {
                let head = until_answered(&format!("reading stream {key}'s head"), || {
                    events.read_stream_next_seq(key.as_bytes())
                })?;
                eprintln!(
                    "BOOT-REPLAY COPY: stream {key} commit of rows {next}..{landed} outlived \
                     haematite's reply wait; its head settled at {head:?}"
                );
                if head == Some(landed) {
                    return Ok(landed);
                }
                if head != refused {
                    return Err(format!(
                        "stream {key} head settled at {head:?} after a timed-out commit of rows \
                         {next}..{landed}: neither landed ({landed}) nor refused ({refused:?})"
                    )
                    .into());
                }
            }
            Err(error) => {
                return Err(format!(
                    "copying stream {key} from row {next} to disk failed: {error}"
                )
                .into());
            }
        }
    }
}

/// Sets one copied scalar value, resuming across a shard reply timeout the
/// same way [`append_resuming`] does: the value is read back through the
/// key's own shard actor, after the timed-out swap has run.
fn cas_resuming(events: &EventStore, key: &str, value: u64) -> Result<(), Box<dyn Error>> {
    loop {
        match events.cas(key.as_bytes(), None, value) {
            Ok(()) => return Ok(()),
            Err(error) if is_reply_timeout(&error) => {
                match until_answered(&format!("reading value {key}"), || {
                    events.read_value(key.as_bytes())
                })? {
                    Some(stored) if stored == value => return Ok(()),
                    None => {}
                    Some(stored) => {
                        return Err(format!(
                            "value {key} settled at {stored} after a timed-out swap to {value}"
                        )
                        .into());
                    }
                }
            }
            Err(error) => return Err(format!("writing value {key} failed: {error}").into()),
        }
    }
}

/// Lands every generated stream and value on the disk database, one commit
/// per `COPY_BATCH` rows.
///
/// `COPY_BATCH` bounds the work of ONE commit. No batch size keeps every
/// commit inside haematite's fixed shard reply wait on a host this loaded —
/// 128 rows overran it at load average ~15 and 32 rows at ~70 — so a commit
/// that outlives the wait is not retried blind: [`append_resuming`] asks the
/// shard where the stream's head settled and continues from there. The
/// slowest commit per stream is printed.
fn copy_to_disk(memory: &MemoryStore, data_dir: &std::path::Path) -> Result<(), Box<dyn Error>> {
    let events = EventStore::new(create_database(data_dir)?);
    let streams = memory
        .streams
        .lock()
        .map_err(|_| "memory store stream lock poisoned")?;
    for (key, payloads) in streams.iter() {
        let started = Instant::now();
        let mut slowest_batch = Duration::ZERO;
        let mut next = 0_u64;
        for batch in payloads.chunks(COPY_BATCH) {
            let refs: Vec<&[u8]> = batch.iter().map(Vec::as_slice).collect();
            let batch_started = Instant::now();
            next = append_resuming(&events, key, &refs, next)?;
            slowest_batch = slowest_batch.max(batch_started.elapsed());
        }
        eprintln!(
            "BOOT-REPLAY COPY: stream {key} ({next} rows) landed in {:?}, slowest \
             {COPY_BATCH}-row commit {slowest_batch:?}",
            started.elapsed()
        );
    }
    drop(streams);
    let values = memory
        .values
        .lock()
        .map_err(|_| "memory store value lock poisoned")?;
    for (key, value) in values.iter() {
        cas_resuming(&events, key, *value)?;
    }
    drop(values);
    until_answered("flushing the copied store", || events.flush())?;
    Ok(())
}

/// Replays timed per route, alternating route by route on the same store.
/// Odd, so each route has a median; three is the fewest rounds that lets one
/// round disturbed by the loaded host fall out of that median.
const ROUNDS: usize = 3;

/// Which cold replay a timed round runs.
#[derive(Clone, Copy, Debug)]
enum Route {
    /// The pre-merge shape: the standalone audit pass reads the whole log,
    /// then the replay reads all of it again.
    TwoPass,
    /// The production single pass.
    OnePass,
}

fn timed_replay(
    route: Route,
    store: &Arc<dyn DurableStore>,
    config: &ParticipantConfig,
) -> Result<(Duration, u64), Box<dyn Error>> {
    let log = OperationLog::new(Arc::clone(store), CONVERSATION);
    let outbox_log = OutboxLog::new(Arc::clone(store), CONVERSATION);
    let limits =
        ConversationOutboxLimits::try_new(config.max_retained_record_rows, config.identity_slots)?;
    let started = Instant::now();
    if matches!(route, Route::TwoPass) {
        block_on(validate_operation_schema(&log, config.identity_slots))??;
    }
    let replayed = block_on(ConversationAuthority::replay_rewriting(
        CONVERSATION,
        &log,
        &outbox_log,
        config,
        limits,
        None,
    ))??;
    Ok((started.elapsed(), replayed.next_log_sequence))
}

fn median(mut samples: Vec<Duration>) -> Result<Duration, Box<dyn Error>> {
    samples.sort_unstable();
    samples
        .get(samples.len() / 2)
        .copied()
        .ok_or_else(|| "no timed rounds".into())
}

#[test]
fn replay_of_twenty_thousand_entries_is_measured() -> Result<(), Box<dyn Error>> {
    let home = tempfile::tempdir()?;
    let data_dir = home.path().join("durability");
    let config = measurement_config();

    let generate_started = Instant::now();
    let memory = Arc::new(MemoryStore::default());
    generate(Arc::clone(&memory))?;
    let generated_in = generate_started.elapsed();
    eprintln!("BOOT-REPLAY GENERATED in {generated_in:?}");
    let copy_started = Instant::now();
    copy_to_disk(&memory, &data_dir)?;
    let copied_in = copy_started.elapsed();
    drop(memory);

    let store: Arc<dyn DurableStore> = Arc::new(HaematiteStore::new(Arc::new(EventStore::new(
        Database::open(
            &data_dir,
            haematite::RuntimeOptions {
                long_wait_interval: std::time::Duration::from_secs(5),
                reclaim_start: haematite::ReclaimStart::AtOpen,
            },
        )?,
    ))));
    let mut two_pass = Vec::with_capacity(ROUNDS);
    let mut one_pass = Vec::with_capacity(ROUNDS);
    let mut replayed_rows = None;
    for round in 0..ROUNDS {
        for route in [Route::TwoPass, Route::OnePass] {
            let (elapsed, rows) = timed_replay(route, &store, &config)?;
            assert!(
                rows >= TARGET_ROWS,
                "{route:?} replay reached {rows} rows, fewer than the {TARGET_ROWS} written"
            );
            assert_eq!(
                *replayed_rows.get_or_insert(rows),
                rows,
                "{route:?} replay disagreed on the log length"
            );
            eprintln!("BOOT-REPLAY ROUND {round} {route:?}: {rows} rows in {elapsed:?}");
            match route {
                Route::TwoPass => two_pass.push(elapsed),
                Route::OnePass => one_pass.push(elapsed),
            }
        }
    }
    eprintln!(
        "BOOT-REPLAY MEASUREMENT: {} rows generated in {generated_in:?}, copied to disk in \
         {copied_in:?}; median cold replay of the same store over {ROUNDS} alternating rounds: \
         two-pass {:?}, one-pass {:?}",
        replayed_rows.unwrap_or_default(),
        median(two_pass)?,
        median(one_pass)?
    );
    Ok(())
}

/// Two members sending to each other in turn, `records` records in all; with
/// `acking`, each member acknowledges every record delivered to it, so the
/// retention floor moves; without, both cursors stay at zero and every record
/// stays retained.
fn generate_conversation(
    store: Arc<MemoryStore>,
    records: u64,
    acking: bool,
) -> Result<(), Box<dyn Error>> {
    let connections = [
        ConnectionIncarnation::new(915, 1),
        ConnectionIncarnation::new(915, 2),
    ];
    let handler = ProductionParticipantHandler::new(store, measurement_config())?;
    let members = [
        enroll(&handler, connections[0], 0xA1)?,
        enroll(&handler, connections[1], 0xA2)?,
    ];
    for cycle in 0..records {
        let sender = usize::try_from(cycle % 2)?;
        let recipient = 1 - sender;
        let mut token = [0_u8; 16];
        token[..8].copy_from_slice(&cycle.to_be_bytes());
        let committed = dispatch(
            &handler,
            connections[sender],
            ClientRequest::RecordAdmission(RecordAdmission {
                conversation_id: CONVERSATION,
                participant_id: members[sender],
                capability_generation: Generation::ONE,
                record_admission_attempt_token: RecordAdmissionAttemptToken::new(token),
                payload: vec![u8::try_from(cycle % 251)?; PAYLOAD_BYTES],
            }),
        )?;
        let ServerValue::RecordCommitted(committed) = committed else {
            return Err(format!("record {cycle} did not commit: {committed:?}").into());
        };
        if acking {
            let acknowledged = ack(
                &handler,
                connections[recipient],
                members[recipient],
                committed.delivery_seq(),
            )?;
            if !matches!(acknowledged, ServerValue::AckCommitted(_)) {
                return Err(format!("ack {cycle} did not commit: {acknowledged:?}").into());
            }
        }
    }
    Ok(())
}

#[test]
fn boot_with_a_pinned_floor_and_a_moving_floor_is_measured() -> Result<(), Box<dyn Error>> {
    for records in [1_000_u64, 2_000, 4_000] {
        for acking in [false, true] {
            let memory = Arc::new(MemoryStore::default());
            generate_conversation(Arc::clone(&memory), records, acking)?;
            let store: Arc<dyn DurableStore> = memory;
            let mut samples = Vec::with_capacity(ROUNDS);
            let mut retained = 0;
            for _ in 0..ROUNDS {
                let started = Instant::now();
                let handler =
                    ProductionParticipantHandler::new(Arc::clone(&store), measurement_config())?;
                samples.push(started.elapsed());
                let cell = handler.cell(CONVERSATION)?;
                retained = cell
                    .lock()
                    .map_err(|_| "conversation lock poisoned")?
                    .as_ref()
                    .and_then(|authority| authority.frontier())
                    .map_or(0, |frontier| frontier.retained_charges().len());
            }
            eprintln!(
                "BOOT FLOOR {}: {records} records, {retained} retained, median boot {:?}",
                if acking { "moving" } else { "pinned" },
                median(samples)?
            );
        }
    }
    Ok(())
}