liminal-server 0.17.0

//! 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::sync::Arc;

use liminal::durability::{
    CountingStore, DurabilityError, DurableStore, ScanCursor, ScanPage, 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_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))
}

/// 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 store = new_store()?;
    let (live, head) = write_history(&ObserverLog::new(Arc::clone(&store)))?;
    let snapshots = head - HISTORY_ROWS;
    assert_eq!(snapshots, HISTORY_ROWS / OBSERVER_SNAPSHOT_EVERY);

    let whole = block_on(ObserverLog::new(Arc::clone(&store)).restore_whole_for_tests())??;
    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 restored = block_on(ObserverLog::new(counted).restore())??;
    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");
    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 store = new_store()?;
    let (live, head) = write_history(&ObserverLog::new(Arc::clone(&store)))?;
    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 restored = block_on(ObserverLog::new(Arc::clone(&store)).restore())??;
    assert_eq!(restored.aggregate, live);
    assert_eq!(restored.next_sequence, head);
    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_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(())
}