liminal-server 0.19.1

//! DUR-008 A3: the observer-recovery stream restores from its latest snapshot
//! row and folds only the rows after it.

use std::error::Error;
use std::fs;
use std::path::Path;
use std::sync::Arc;
use std::time::Instant;

use liminal::durability::{
    CountingStore, DurabilityError, DurableStore, ScanCursor, ScanPage, StoreCounts, StoredEntry,
    bridge::block_on,
};
use liminal_protocol::lifecycle::{
    ObserverProgressAdvanceDecision, ObserverProgressTrackDecision, ObserverRecoveryAggregate,
};

use super::observer::{
    OBSERVER_SNAPSHOT_EVERY, OBSERVER_SNAPSHOT_KEY, OBSERVER_STREAM_KEY, ObserverLog, ObserverRow,
    encode_for_tests, head_after, snapshot_before, snapshot_due, snapshot_record_failures,
};
use super::tests::open_disk_store_for_tests;
use super::tests_memory_store::MemoryStore;
use super::tests_seed::Shared;
use super::tests_w3_restore_fixture::new_store;

/// What every test here answers.
type TestResult = Result<(), Box<dyn Error>>;

/// Rows the history holds: far past one snapshot interval.
const HISTORY_ROWS: u64 = 5_000;
/// Conversations the history tracks before it advances them.
const CONVERSATIONS: u64 = 500;

/// Appends one row the way the handlers do (snapshot taken before the
/// aggregate moves, the head the log answers), leaving the flush to the end of
/// the history so 5,000 rows do not cost 5,000 flushes.
fn append(
    log: &ObserverLog,
    head: u64,
    aggregate: &ObserverRecoveryAggregate,
    row: &ObserverRow,
) -> Result<u64, Box<dyn Error>> {
    let snapshot = snapshot_before(head, aggregate);
    let next = block_on(log.append_unflushed_for_tests(row, head, snapshot))??;
    assert_eq!(Some(next), head_after(head, 1));
    Ok(next)
}

/// Writes `HISTORY_ROWS` operation rows (tracks, advances and arm plans)
/// through the live transitions, answering the live aggregate and head.
fn write_history(log: &ObserverLog) -> Result<(ObserverRecoveryAggregate, u64), Box<dyn Error>> {
    write_history_of(log, HISTORY_ROWS)
}

/// Writes `rows` operation rows the way [`write_history`] does.
fn write_history_of(
    log: &ObserverLog,
    rows: u64,
) -> Result<(ObserverRecoveryAggregate, u64), Box<dyn Error>> {
    let mut aggregate = ObserverRecoveryAggregate::new();
    let mut head = 0_u64;
    for written in 0..rows {
        let conversation_id = written % CONVERSATIONS + 1;
        let row = if written < CONVERSATIONS {
            let row = ObserverRow::Track {
                conversation_id,
                observer_progress: 0,
            };
            head = append(log, head, &aggregate, &row)?;
            aggregate = match aggregate.decide_track(conversation_id, 0) {
                ObserverProgressTrackDecision::Commit(transaction) => transaction.commit(),
                ObserverProgressTrackDecision::Refuse { .. } => return Err("track refused".into()),
            };
            continue;
        } else if written % 7 == 0 {
            let progress = aggregate
                .observer_progress(conversation_id)
                .ok_or("untracked")?;
            ObserverRow::Arms {
                arms: vec![(conversation_id, progress)],
            }
        } else {
            let progress = aggregate
                .observer_progress(conversation_id)
                .ok_or("untracked")?
                + 1;
            ObserverRow::Advance {
                conversation_id,
                observer_progress: progress,
            }
        };
        head = append(log, head, &aggregate, &row)?;
        aggregate = match row {
            ObserverRow::Arms { arms } => aggregate
                .install_restored_arms(&arms)
                .map_err(|_| "arms refused")?,
            ObserverRow::Advance {
                conversation_id,
                observer_progress,
            } => match aggregate.decide_progress_advance(conversation_id, observer_progress) {
                ObserverProgressAdvanceDecision::Commit(transaction) => transaction.commit().0,
                ObserverProgressAdvanceDecision::Refuse { .. } => {
                    return Err("advance refused".into());
                }
            },
            ObserverRow::Track { .. } | ObserverRow::Snapshot { .. } => {
                return Err("unexpected row".into());
            }
        };
    }
    block_on(log.flush_for_tests())??;
    Ok((aggregate, head))
}

#[derive(Debug)]
struct HistoryFixture {
    store: Arc<dyn DurableStore>,
    live: ObserverRecoveryAggregate,
    head: u64,
    counts: SeedCounts,
}

#[derive(Debug, serde::Serialize, serde::Deserialize)]
struct SeedCounts {
    append_calls: u64,
    append_all_calls: u64,
    append_all_rows: u64,
    flushes: u64,
}

impl From<StoreCounts> for SeedCounts {
    fn from(counts: StoreCounts) -> Self {
        Self {
            append_calls: counts.append_calls,
            append_all_calls: counts.append_all_calls,
            append_all_rows: counts.append_all_rows,
            flushes: counts.flushes,
        }
    }
}

#[derive(Debug, serde::Serialize, serde::Deserialize)]
struct HistoryReference {
    progress: Vec<(u64, u64)>,
    armed: Vec<(u64, u64)>,
    head: u64,
    counts: SeedCounts,
}

fn build_history(data: &Path) -> TestResult {
    let start = Instant::now();
    let memory = Arc::new(MemoryStore::default());
    let store: Arc<dyn DurableStore> = memory.clone();
    let (live, head) = write_history(&ObserverLog::new(store))?;
    let rows = {
        let mut streams = memory
            .streams
            .lock()
            .map_err(|_| "fixture stream lock poisoned")?;
        let rows = streams
            .remove(OBSERVER_STREAM_KEY)
            .ok_or("fixture observer stream missing")?;
        assert!(streams.is_empty(), "fixture has an unexpected stream");
        drop(streams);
        rows
    };
    let recorded = {
        let mut values = memory
            .values
            .lock()
            .map_err(|_| "fixture value lock poisoned")?;
        let recorded = values
            .remove(OBSERVER_SNAPSHOT_KEY)
            .ok_or("fixture snapshot hint missing")?;
        assert!(values.is_empty(), "fixture has an unexpected value");
        drop(values);
        recorded
    };
    println!(
        "history row generation elapsed_ms={}",
        start.elapsed().as_secs_f64() * 1_000.0
    );

    let store = CountingStore::new(Shared(open_disk_store_for_tests(&data.join("store"))?));
    let counts = store.handle();
    let start = Instant::now();
    let committed = block_on(store.append_all(OBSERVER_STREAM_KEY, rows, 0))??;
    assert_eq!(committed, head);
    block_on(store.cas(OBSERVER_SNAPSHOT_KEY, 0, recorded))??;
    block_on(store.flush())??;
    let counts = counts.snapshot();
    println!(
        "history durable seed elapsed_ms={} counts={counts:?}",
        start.elapsed().as_secs_f64() * 1_000.0
    );
    assert_eq!(counts.cas_calls, 1);
    let reference = HistoryReference {
        progress: live.progress_rows(),
        armed: live.armed_rows(),
        head,
        counts: counts.into(),
    };
    pin_history_seed(&reference.counts, head);
    fs::write(data.join("reference.json"), serde_json::to_vec(&reference)?)?;
    drop(store);
    Ok(())
}

fn profiled_history() -> Result<(tempfile::TempDir, HistoryFixture), Box<dyn Error>> {
    let start = Instant::now();
    let directory = crate::test_fixture::fixture_copy("observer-snapshot-history", build_history)?;
    let builds = crate::test_fixture::fixture_builds("observer-snapshot-history")?;
    assert_eq!(builds, 1, "history is built once per test binary");
    let reference: HistoryReference =
        serde_json::from_slice(&fs::read(directory.path().join("reference.json"))?)?;
    let live = ObserverRecoveryAggregate::restore(&reference.progress, &reference.armed)
        .map_err(|error| format!("fixture reference refused: {error:?}"))?;
    let store = open_disk_store_for_tests(&directory.path().join("store"))?;
    println!(
        "history fixture elapsed_ms={} builds={builds} counts={:?}",
        start.elapsed().as_secs_f64() * 1_000.0,
        reference.counts
    );
    Ok((
        directory,
        HistoryFixture {
            store,
            live,
            head: reference.head,
            counts: reference.counts,
        },
    ))
}

fn pin_history_seed(counts: &SeedCounts, head: u64) {
    assert_eq!(counts.append_calls, 0);
    assert_eq!(
        counts.append_all_calls, 1,
        "fixture history is one durable batch"
    );
    assert_eq!(counts.append_all_rows, head);
    assert_eq!(counts.flushes, 1);
}

/// A restore from the latest snapshot plus its tail yields the same aggregate
/// as a fold over the whole stream, and reads at most one snapshot interval
/// of rows to do it.
#[test]
fn a_tail_fold_from_the_snapshot_equals_the_whole_fold_on_5000_rows() -> Result<(), Box<dyn Error>>
{
    let (
        directory,
        HistoryFixture {
            store,
            live,
            head,
            counts: seed_counts,
        },
    ) = profiled_history()?;
    let snapshots = head - HISTORY_ROWS;
    assert_eq!(snapshots, HISTORY_ROWS / OBSERVER_SNAPSHOT_EVERY);

    let start = Instant::now();
    let whole = block_on(ObserverLog::new(Arc::clone(&store)).restore_whole_for_tests())??;
    println!(
        "whole replay elapsed_ms={}",
        start.elapsed().as_secs_f64() * 1_000.0
    );
    assert_eq!(whole.aggregate, live);
    assert_eq!(whole.next_sequence, head);

    let counting = Arc::new(CountingStore::new(Shared(Arc::clone(&store))));
    let counts = counting.handle();
    let counted: Arc<dyn DurableStore> = counting;
    let start = Instant::now();
    let restored = block_on(ObserverLog::new(counted).restore())??;
    println!(
        "snapshot-tail replay elapsed_ms={}",
        start.elapsed().as_secs_f64() * 1_000.0
    );
    assert_eq!(restored.aggregate, whole.aggregate);
    assert_eq!(restored.next_sequence, head);
    let counted = counts.snapshot();
    assert_eq!(counted.read_at_calls, 1, "the snapshot is one point read");
    assert!(
        counted.read_from_rows <= OBSERVER_SNAPSHOT_EVERY,
        "restore read {} rows past its snapshot",
        counted.read_from_rows
    );
    assert_eq!(counted.cas_calls, 0, "a clean restore records nothing");
    pin_history_seed(&seed_counts, head);
    drop(store);
    drop(directory);
    Ok(())
}

/// The appender places a snapshot exactly at the due positions, so the rows
/// after the latest one never exceed the interval.
#[test]
fn snapshots_fall_once_per_interval() {
    let due: Vec<u64> = (0..3 * (OBSERVER_SNAPSHOT_EVERY + 1))
        .filter(|head| snapshot_due(*head))
        .collect();
    assert_eq!(
        due,
        vec![
            OBSERVER_SNAPSHOT_EVERY,
            2 * OBSERVER_SNAPSHOT_EVERY + 1,
            3 * OBSERVER_SNAPSHOT_EVERY + 2
        ]
    );
    assert_eq!(
        head_after(0, OBSERVER_SNAPSHOT_EVERY),
        Some(OBSERVER_SNAPSHOT_EVERY)
    );
    assert_eq!(
        head_after(0, OBSERVER_SNAPSHOT_EVERY + 1),
        Some(OBSERVER_SNAPSHOT_EVERY + 2)
    );
    assert_eq!(head_after(u64::MAX, 1), None);
}

/// A recorded position that names an operation row is refused, and restore
/// falls back to the whole fold with the same answer.
#[test]
fn a_record_naming_a_non_snapshot_row_falls_back_to_the_whole_fold() -> Result<(), Box<dyn Error>> {
    let (
        directory,
        HistoryFixture {
            store,
            live,
            head,
            counts: seed_counts,
        },
    ) = profiled_history()?;
    let recorded = block_on(store.read_value(super::observer::OBSERVER_SNAPSHOT_KEY))??
        .ok_or("no snapshot recorded")?;
    block_on(store.cas(
        super::observer::OBSERVER_SNAPSHOT_KEY,
        recorded,
        recorded + 1,
    ))??;
    let start = Instant::now();
    let restored = block_on(ObserverLog::new(Arc::clone(&store)).restore())??;
    println!(
        "invalid-hint whole replay elapsed_ms={}",
        start.elapsed().as_secs_f64() * 1_000.0
    );
    assert_eq!(restored.aggregate, live);
    assert_eq!(restored.next_sequence, head);
    pin_history_seed(&seed_counts, head);
    drop(store);
    drop(directory);
    Ok(())
}

/// A store whose snapshot head key can be neither read nor written; every
/// other call goes through unchanged.
#[derive(Debug)]
struct SnapshotKeyFails(Arc<dyn DurableStore>);

impl SnapshotKeyFails {
    fn refuse(key: &str) -> Result<(), DurabilityError> {
        if key == OBSERVER_SNAPSHOT_KEY {
            Err(DurabilityError::ConfigError(
                "injected snapshot key failure".to_owned(),
            ))
        } else {
            Ok(())
        }
    }
}

#[async_trait::async_trait]
impl DurableStore for SnapshotKeyFails {
    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.0.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.0.append_all(stream_key, payloads, expected_seq).await
    }

    async fn read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        self.0.read_from(stream_key, offset, limit).await
    }

    async fn read_at(
        &self,
        stream_key: &str,
        sequence: u64,
    ) -> Result<Option<StoredEntry>, DurabilityError> {
        self.0.read_at(stream_key, sequence).await
    }

    async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
        Self::refuse(key)?;
        self.0.cas(key, old_value, new_value).await
    }

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

    async fn read_value(&self, key: &str) -> Result<Option<u64>, DurabilityError> {
        Self::refuse(key)?;
        self.0.read_value(key).await
    }

    async fn scan_page(
        &self,
        prefix: &str,
        cursor: Option<ScanCursor>,
        limit: usize,
    ) -> Result<ScanPage, DurabilityError> {
        self.0.scan_page(prefix, cursor, limit).await
    }

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

/// The snapshot head key is a hint (review item 1, Hermes 20:18): when it
/// cannot be written, an append whose snapshot and row landed still answers
/// the head past them, the failure is counted, and a restore that cannot use
/// the key folds the stream to the same aggregate.
#[test]
fn a_snapshot_record_that_fails_is_counted_and_never_a_failed_append() -> TestResult {
    let store = new_store()?;
    let failing: Arc<dyn DurableStore> = Arc::new(SnapshotKeyFails(Arc::clone(&store)));
    let before = snapshot_record_failures();
    let rows = OBSERVER_SNAPSHOT_EVERY + 44;
    let (live, head) = write_history_of(&ObserverLog::new(failing), rows)?;
    assert_eq!(Some(head), head_after(0, rows));
    assert_eq!(
        snapshot_record_failures() - before,
        1,
        "the one due snapshot's record failure was not counted"
    );
    assert_eq!(block_on(store.read_value(OBSERVER_SNAPSHOT_KEY))??, None);
    let restored = block_on(ObserverLog::new(store).restore())??;
    assert_eq!(restored.aggregate, live);
    assert_eq!(restored.next_sequence, head);
    Ok(())
}

/// Writes `rows` Track rows the way a binary from before snapshots wrote its
/// stream (schema-1 operation rows, no snapshot row, no record), in one
/// `append_all` and one flush, answering the aggregate they fold to.
fn legacy_stream(
    store: &Arc<dyn DurableStore>,
    rows: u64,
) -> Result<ObserverRecoveryAggregate, Box<dyn Error>> {
    let mut aggregate = ObserverRecoveryAggregate::new();
    let mut payloads = Vec::new();
    for conversation_id in 1..=rows {
        payloads.push(encode_for_tests(&ObserverRow::Track {
            conversation_id,
            observer_progress: 0,
        })?);
        aggregate = match aggregate.decide_track(conversation_id, 0) {
            ObserverProgressTrackDecision::Commit(transaction) => transaction.commit(),
            ObserverProgressTrackDecision::Refuse { .. } => return Err("track refused".into()),
        };
    }
    block_on(store.append_all(OBSERVER_STREAM_KEY, payloads, 0))??;
    block_on(store.flush())??;
    Ok(aggregate)
}

/// Review item 6 (Hermes 20:18): a stream longer than one interval with no
/// snapshot (written before snapshots) is folded whole once; that restore
/// writes a snapshot at the head and records it, so the next restore folds
/// no row.
#[test]
fn a_long_stream_without_a_snapshot_takes_one_at_its_first_restore() -> TestResult {
    let store = new_store()?;
    let rows = OBSERVER_SNAPSHOT_EVERY + 44;
    let live = legacy_stream(&store, rows)?;
    let first = block_on(ObserverLog::new(Arc::clone(&store)).restore())??;
    assert_eq!(first.aggregate, live);
    assert_eq!(first.next_sequence, rows + 1, "no snapshot row was written");
    assert_eq!(
        block_on(store.read_value(OBSERVER_SNAPSHOT_KEY))??,
        Some(rows + 1)
    );

    let counting = Arc::new(CountingStore::new(Shared(Arc::clone(&store))));
    let counts = counting.handle();
    let counted: Arc<dyn DurableStore> = counting;
    let second = block_on(ObserverLog::new(counted).restore())??;
    assert_eq!(second.aggregate, live);
    assert_eq!(second.next_sequence, rows + 1);
    let counted = counts.snapshot();
    assert_eq!(counted.read_from_rows, 0, "the second restore folded rows");
    assert_eq!(counted.append_all_calls, 0, "the second restore wrote");
    Ok(())
}

/// Review item 5 (Hermes 20:18): the upgrade path. A stream shaped as 0.16
/// wrote it (schema-1 operation rows, no snapshot row, no record, shorter
/// than one interval) restores by a whole fold that writes nothing. The
/// appender then takes its first snapshot at the first due position, and a
/// restore from that snapshot equals the live aggregate.
#[test]
fn a_stream_from_before_snapshots_restores_and_takes_its_first_snapshot() -> TestResult {
    let store = new_store()?;
    let legacy = OBSERVER_SNAPSHOT_EVERY - 56;
    let mut aggregate = legacy_stream(&store, legacy)?;
    let log = ObserverLog::new(Arc::clone(&store));
    let restored = block_on(log.restore())??;
    assert_eq!(restored.aggregate, aggregate);
    assert_eq!(
        restored.next_sequence, legacy,
        "a short stream was written to"
    );
    assert_eq!(block_on(store.read_value(OBSERVER_SNAPSHOT_KEY))??, None);

    let mut head = restored.next_sequence;
    for conversation_id in legacy + 1..=legacy + 100 {
        let row = ObserverRow::Track {
            conversation_id,
            observer_progress: 0,
        };
        let snapshot = snapshot_before(head, &aggregate);
        let next = block_on(log.append(&row, head, snapshot))??;
        assert_eq!(Some(next), head_after(head, 1));
        head = next;
        aggregate = match aggregate.decide_track(conversation_id, 0) {
            ObserverProgressTrackDecision::Commit(transaction) => transaction.commit(),
            ObserverProgressTrackDecision::Refuse { .. } => return Err("track refused".into()),
        };
    }
    assert_eq!(
        block_on(store.read_value(OBSERVER_SNAPSHOT_KEY))??,
        Some(OBSERVER_SNAPSHOT_EVERY + 1),
        "the first snapshot was not taken at the first due position"
    );
    let again = block_on(ObserverLog::new(store).restore())??;
    assert_eq!(again.aggregate, aggregate);
    assert_eq!(again.next_sequence, head);
    Ok(())
}