liminal-server 0.25.4

//! R3b review finding 21: the observer-recovery aggregate keeps a progress
//! row for every conversation it ever tracked, and every checkpoint row
//! (`ObserverRow::snapshot_of`) serializes all of them.
//!
//! `ObserverRecoveryAggregate` holds progress in a map with track and advance
//! transitions and no removal, so a server that has completed N conversations
//! and is now serving the same small live tail checkpoints N + tail rows
//! every `OBSERVER_SNAPSHOT_EVERY` operations, for its whole lifetime.
//!
//! DUR-027 rules the retirement: a closed, unarmed conversation retires in one
//! ordered transition, and its first step here is the aggregate's own
//! `decide_retire` (R2), written as `ObserverRow::Retire` and folded on
//! restore (R3, `tests_observer_retire`). `completed_then_live` retires each
//! completed conversation where the ruled event goes, and the pin asserts the
//! bound: equal rows and equal checkpoint bytes for the same live tail.
//!
//! What this pin does not cover: the server writing that row when a
//! conversation closes (DUR-027 R4) and cutting the streams below it (R5).
//! Those are pinned by R7's counted reds, which stay red until they land.
//!
//! Pure aggregate state and the production row encoder; no store is opened.

use std::error::Error;

use liminal_protocol::lifecycle::{
    ObserverProgressAdvanceDecision, ObserverProgressAdvanceError, ObserverProgressRetireDecision,
    ObserverProgressTrackDecision, ObserverProgressTrackError, ObserverRecoveryAggregate,
};

use super::observer::{ObserverRow, encode_for_tests};

/// Completed-conversation histories compared.
const HISTORIES: [u64; 2] = [100, 1_000];

/// The one conversation still live after either history. Its id is above
/// every completed id, so a bounded checkpoint is byte-for-byte the same at
/// both histories.
const LIVE_CONVERSATION: u64 = 1_000_000;

/// Final observer progress of each completed conversation.
const COMPLETED_PROGRESS: u64 = 2;

/// Observer progress of the live conversation when the checkpoint is taken.
const LIVE_PROGRESS: u64 = 3;

fn track(
    aggregate: ObserverRecoveryAggregate,
    conversation_id: u64,
) -> Result<ObserverRecoveryAggregate, Box<dyn Error>> {
    match aggregate.decide_track(conversation_id, 0) {
        ObserverProgressTrackDecision::Commit(transaction) => Ok(transaction.commit()),
        ObserverProgressTrackDecision::Refuse { error, .. } => {
            Err(format!("track of {conversation_id} refused: {error:?}").into())
        }
    }
}

fn advance(
    aggregate: ObserverRecoveryAggregate,
    conversation_id: u64,
    progress: u64,
) -> Result<ObserverRecoveryAggregate, Box<dyn Error>> {
    match aggregate.decide_progress_advance(conversation_id, progress) {
        ObserverProgressAdvanceDecision::Commit(transaction) => Ok(transaction.commit().0),
        ObserverProgressAdvanceDecision::Refuse { error, .. } => {
            Err(format!("advance of {conversation_id} to {progress} refused: {error:?}").into())
        }
    }
}

/// Completes `conversations` conversations (ids 1 to `conversations`), then
/// tracks and advances the one live conversation. Answers the aggregate and
/// the number of committed transitions, the positive control.
fn completed_then_live(
    conversations: u64,
) -> Result<(ObserverRecoveryAggregate, u64), Box<dyn Error>> {
    let mut aggregate = ObserverRecoveryAggregate::new();
    let mut committed = 0_u64;
    for conversation_id in 1..=conversations {
        aggregate = track(aggregate, conversation_id)?;
        committed += 1;
        for progress in 1..=COMPLETED_PROGRESS {
            aggregate = advance(aggregate, conversation_id, progress)?;
            committed += 1;
        }
        aggregate = retire(aggregate, conversation_id)?;
        committed += 1;
    }
    aggregate = track(aggregate, LIVE_CONVERSATION)?;
    committed += 1;
    for progress in 1..=LIVE_PROGRESS {
        aggregate = advance(aggregate, LIVE_CONVERSATION, progress)?;
        committed += 1;
    }
    Ok((aggregate, committed))
}

fn retire(
    aggregate: ObserverRecoveryAggregate,
    conversation_id: u64,
) -> Result<ObserverRecoveryAggregate, Box<dyn Error>> {
    match aggregate.decide_retire(conversation_id) {
        ObserverProgressRetireDecision::Commit(transaction) => Ok(transaction.commit()),
        ObserverProgressRetireDecision::Refuse { error, .. } => {
            Err(format!("retire of {conversation_id} refused: {error:?}").into())
        }
    }
}

/// Progress rows held and the encoded bytes of the checkpoint row the
/// appender would write now.
fn checkpoint_of(aggregate: &ObserverRecoveryAggregate) -> Result<(usize, usize), Box<dyn Error>> {
    let encoded = encode_for_tests(&ObserverRow::snapshot_of(aggregate))?;
    Ok((aggregate.progress_rows().len(), encoded.len()))
}

/// Control, green today and after the bound: both histories built what they
/// claim, the live conversation answers the same progress, and the legal
/// refusals over it are unchanged.
#[test]
fn completed_conversations_leave_the_live_answers_unchanged() -> Result<(), Box<dyn Error>> {
    let [small, large] = HISTORIES;
    let (smaller, small_commits) = completed_then_live(small)?;
    let (larger, large_commits) = completed_then_live(large)?;
    // a track, its advances and its retirement
    let per_conversation = COMPLETED_PROGRESS + 2;
    assert_eq!(
        large_commits - small_commits,
        (large - small) * per_conversation,
        "positive control failed: the larger history did not commit its extra transitions"
    );
    for aggregate in [&smaller, &larger] {
        assert_eq!(
            aggregate.observer_progress(LIVE_CONVERSATION),
            Some(LIVE_PROGRESS),
            "the live conversation's progress differs with the history behind it"
        );
    }
    let ObserverProgressTrackDecision::Refuse { aggregate, error } =
        larger.decide_track(LIVE_CONVERSATION, 0)
    else {
        return Err("a second track of the live conversation was not refused".into());
    };
    assert_eq!(
        error,
        ObserverProgressTrackError::AlreadyTracked {
            conversation_id: LIVE_CONVERSATION
        }
    );
    let ObserverProgressAdvanceDecision::Refuse { error, .. } =
        aggregate.decide_progress_advance(LIVE_CONVERSATION, LIVE_PROGRESS)
    else {
        return Err("a non-advancing advance of the live conversation was not refused".into());
    };
    assert_eq!(
        error,
        ObserverProgressAdvanceError::NotAdvancing {
            conversation_id: LIVE_CONVERSATION,
            current_observer_progress: LIVE_PROGRESS,
            presented_progress: LIVE_PROGRESS,
        }
    );
    Ok(())
}

/// RED pin for R3b review finding 21. With the same live tail of one
/// conversation, the aggregate after 100 and after 1,000 completed
/// conversations must hold the same progress rows and checkpoint the same
/// encoded bytes.
///
/// Red until DUR-027 R2: 1,001 progress rows against 101, and 7,974
/// checkpoint bytes against 773, because nothing removed a completed
/// conversation's row. Green with the aggregate's retirement (R2), which the
/// observer stream carries as a retire row (R3).
#[test]
fn completed_conversations_leave_the_observer_checkpoint() -> Result<(), Box<dyn Error>> {
    let [small, large] = HISTORIES;
    let (smaller, _) = completed_then_live(small)?;
    let (larger, _) = completed_then_live(large)?;
    let (small_rows, small_bytes) = checkpoint_of(&smaller)?;
    let (large_rows, large_bytes) = checkpoint_of(&larger)?;
    let mut grown = Vec::new();
    if large_rows != small_rows {
        grown.push(format!(
            "progress rows {large_rows} at {large} vs {small_rows} at {small}"
        ));
    }
    if large_bytes != small_bytes {
        grown.push(format!(
            "checkpoint encoded bytes {large_bytes} at {large} vs {small_bytes} at {small}"
        ));
    }
    assert!(
        grown.is_empty(),
        "R3b review finding 21: with the same live tail the observer checkpoint grew with \
         completed conversations, expected equal: {}",
        grown.join("; ")
    );
    Ok(())
}