liminal-server 0.14.3

//! 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 std::collections::BTreeMap;
use std::error::Error;
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::{Duration, Instant};

use haematite::{ApiError, Database, DatabaseConfig, DatabaseError, EventStore};
use liminal::durability::{
    DurabilityError, DurableStore, HaematiteStore, StoredEntry, bridge::block_on,
};
use liminal_protocol::wire::{
    ClientRequest, ConnectionIncarnation, EnrollmentRequest, EnrollmentToken, Generation,
    ParticipantAck, RecordAdmission, RecordAdmissionAttemptToken, ServerValue,
};

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

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};

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

type Streams = BTreeMap<String, Vec<Vec<u8>>>;
type Values = BTreeMap<String, u64>;

/// A `DurableStore` over process memory with the exact append/read/cas
/// contract the production seam relies on, for generating rows quickly.
#[derive(Debug, Default)]
struct MemoryStore {
    streams: Mutex<Streams>,
    values: Mutex<Values>,
}

impl MemoryStore {
    fn lock_streams(&self) -> Result<MutexGuard<'_, Streams>, DurabilityError> {
        self.streams
            .lock()
            .map_err(|_| DurabilityError::ConfigError("memory store stream lock poisoned".into()))
    }

    fn lock_values(&self) -> Result<MutexGuard<'_, Values>, DurabilityError> {
        self.values
            .lock()
            .map_err(|_| DurabilityError::ConfigError("memory store value lock poisoned".into()))
    }

    fn entries(
        payloads: &[Vec<u8>],
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        let start = usize::try_from(offset)
            .map_err(|_| DurabilityError::ConfigError(format!("offset {offset} exceeds memory")))?;
        payloads
            .iter()
            .enumerate()
            .skip(start)
            .take(limit)
            .map(|(index, payload)| {
                Ok(StoredEntry {
                    payload: payload.clone(),
                    sequence: u64::try_from(index).map_err(|_| {
                        DurabilityError::ConfigError(format!("sequence {index} exceeds u64"))
                    })?,
                    timestamp: 0,
                })
            })
            .collect()
    }
}

#[async_trait::async_trait]
impl DurableStore for MemoryStore {
    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        let mut streams = self.lock_streams()?;
        let stream = streams.entry(stream_key.to_owned()).or_default();
        let actual = u64::try_from(stream.len())
            .map_err(|_| DurabilityError::ConfigError("memory stream length overflow".into()))?;
        if actual != expected_seq {
            return Err(DurabilityError::SequenceConflict {
                expected: expected_seq,
                actual,
            });
        }
        stream.push(payload);
        drop(streams);
        Ok(actual)
    }

    async fn read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        let streams = self.lock_streams()?;
        streams.get(stream_key).map_or_else(
            || Ok(Vec::new()),
            |payloads| Self::entries(payloads, offset, limit),
        )
    }

    async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
        let mut values = self.lock_values()?;
        let stored = values.get(key).copied().unwrap_or(0);
        if stored != old_value {
            return Err(DurabilityError::CursorRegression {
                stored,
                attempted: old_value,
            });
        }
        values.insert(key.to_owned(), new_value);
        drop(values);
        Ok(())
    }

    async fn read_value(&self, key: &str) -> Result<Option<u64>, DurabilityError> {
        Ok(self.lock_values()?.get(key).copied())
    }

    async fn scan(&self, prefix: &str) -> Result<Vec<StoredEntry>, DurabilityError> {
        let streams = self.lock_streams()?;
        let mut entries = Vec::new();
        for (key, payloads) in streams.range(prefix.to_owned()..) {
            if !key.starts_with(prefix) {
                break;
            }
            entries.extend(Self::entries(payloads, 0, payloads.len())?);
        }
        drop(streams);
        Ok(entries)
    }

    async fn flush(&self) -> Result<(), DurabilityError> {
        Ok(())
    }
}

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),
    })?)
}

/// 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(
        CONVERSATION,
        &log,
        &outbox_log,
        config,
        limits,
    ))??;
    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)?,
    ))));
    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(())
}