liminal-server 0.24.9

//! DUR-027 R5: a closed, unarmed conversation is retired by the commit that
//! closed it, in one ordered transition, and leaves nothing a reader or a boot
//! visits.
//!
//! The closing commit is the boot drain that erases the conversation's last
//! enrollment token (F8B R-SEAL): `last_identity_store` mints those bytes
//! through real transitions, and the first touch of a handler over them runs
//! the restore unit whose drain seals it.

use std::error::Error;
use std::sync::Arc;
use std::sync::atomic::Ordering;

use liminal::durability::DurableStore;
use liminal::durability::bridge::block_on;
use liminal_protocol::wire::ConversationId;

use super::ProductionParticipantHandler;
use super::log::{GENERATION_STREAM_PREFIX, RETIRED_GENERATION, STREAM_PREFIX};
use super::observer::{OBSERVER_STREAM_KEY, ObserverLog, ObserverRow, encode_for_tests};
use super::outbox_log::OUTBOX_STREAM_PREFIX;
use super::registry::ConversationRegistry;
use super::registry_snapshot::RegistrySets;
use super::restore_checkpoint_locator::checkpoint_conversation_route;
use super::retirement::Retirement;
use super::route_keys::conversation_windows;
use super::tests::test_participant_config;
use super::tests_f8b_seal::last_identity_store;
use super::tests_registry_snapshot_fixture::{Recording, memory};

type TestResult = Result<(), Box<dyn Error>>;

/// The bound the last-identity fixture was written under; every reopen of
/// its bytes presents the same one, or replay diverges by design.
fn fixture_config() -> crate::config::types::ParticipantConfig {
    let mut config = test_participant_config();
    config.max_retained_record_rows = 4;
    config
}

/// Every row the conversation's streams hold and every key in its own
/// windows of its route; the shard's other families (the registry, the
/// observer) are not the conversation's and are not counted.
fn residue(
    store: &dyn DurableStore,
    conversation_id: ConversationId,
) -> Result<usize, Box<dyn Error>> {
    let mut streams = vec![
        format!("{STREAM_PREFIX}{conversation_id}"),
        format!("{OUTBOX_STREAM_PREFIX}{conversation_id}"),
    ];
    streams.extend(
        (1..=4)
            .map(|generation| format!("{GENERATION_STREAM_PREFIX}{generation}:{conversation_id}")),
    );
    let mut rows = 0;
    for stream in &streams {
        // ast-grep-allow: no-read-from-start: the assertion is that no row is left at any sequence
        rows += block_on(store.read_from(stream, 0, usize::MAX))??.len();
    }
    let route = checkpoint_conversation_route(conversation_id);
    for (from, to) in conversation_windows(conversation_id, &streams)? {
        rows += block_on(store.scan_routed(&route, &from, &to))??.len();
    }
    Ok(rows)
}

#[test]
fn a_conversation_sealed_by_its_boot_drain_is_retired_by_that_commit() -> TestResult {
    let minted = last_identity_store()?;
    let id = minted.conversation_id;
    assert!(
        residue(minted.store.as_ref(), id)? > 0,
        "the fixture minted no rows"
    );

    let booted = ProductionParticipantHandler::new(Arc::clone(&minted.store), fixture_config())?;
    // The first touch runs the restore unit: its drain seals and retires.
    assert!(
        booted.existing_cell(id)?.is_none(),
        "the retired conversation kept a cell"
    );

    let sets = ConversationRegistry::new(Arc::clone(&minted.store)).restore_sets()?;
    assert!(
        !sets.live.contains(&id),
        "the retired id is still live: {sets:?}"
    );
    assert!(
        sets.sweeping.is_empty(),
        "the retirement stopped sweeping: {sets:?}"
    );
    let observer = block_on(ObserverLog::new(Arc::clone(&minted.store)).restore())??;
    assert_eq!(observer.aggregate.observer_progress(id), None);
    assert_eq!(residue(minted.store.as_ref(), id)?, 0);
    let generation =
        block_on(super::log::OperationLog::new(Arc::clone(&minted.store), id).generation())??;
    assert_eq!(generation, RETIRED_GENERATION);
    Ok(())
}

/// DUR-027 R6: a retirement that committed its registry row and then stopped
/// (the id sweeping, every stream and key still there) is finished by the
/// next boot, before anything is served, and boot never lists it again.
#[test]
fn boot_finishes_a_retirement_a_stop_interrupted() -> TestResult {
    let minted = last_identity_store()?;
    let id = minted.conversation_id;
    let registry = ConversationRegistry::new(Arc::clone(&minted.store));
    registry.restore_sets()?;
    // Step 1 alone: the retirement's commit point, then the stop.
    registry.retire(id)?;
    assert!(
        residue(minted.store.as_ref(), id)? > 0,
        "the fixture minted no rows"
    );

    let booted = ProductionParticipantHandler::new(Arc::clone(&minted.store), fixture_config())?;
    let sets = ConversationRegistry::new(Arc::clone(&minted.store)).restore_sets()?;
    assert!(
        sets.sweeping.is_empty() && !sets.live.contains(&id),
        "boot left the interrupted retirement: {sets:?}"
    );
    assert_eq!(residue(minted.store.as_ref(), id)?, 0);
    let generation =
        block_on(super::log::OperationLog::new(Arc::clone(&minted.store), id).generation())??;
    assert_eq!(generation, RETIRED_GENERATION);
    assert!(
        booted.existing_cell(id)?.is_none(),
        "a retired conversation was restored by the boot that finished it"
    );
    Ok(())
}

/// Every step reads the store before it writes, so finishing a retired id
/// again (as boot does for a sweeping one) writes and flushes nothing.
#[test]
fn finishing_a_retired_conversation_again_writes_nothing() -> TestResult {
    let minted = last_identity_store()?;
    let id = minted.conversation_id;
    let recording = Recording::new(Arc::clone(&minted.store));
    let booted = ProductionParticipantHandler::new(recording.shared(), fixture_config())?;
    assert!(
        booted.existing_cell(id)?.is_none(),
        "the retired conversation kept a cell"
    );
    recording.take_batches()?;
    let writes = recording.other_writes.load(Ordering::SeqCst);
    let flushes = recording.flushes.load(Ordering::SeqCst);

    let mut observer = booted.loaded_observer()?;
    assert_eq!(
        booted.finish_retirement(id, &mut observer),
        Retirement::Retired
    );
    drop(observer);
    assert!(recording.take_batches()?.is_empty());
    assert_eq!(recording.other_writes.load(Ordering::SeqCst), writes);
    assert_eq!(recording.flushes.load(Ordering::SeqCst), flushes);
    Ok(())
}

/// An installed arm is an observer still waiting: retiring under it would
/// answer it by deleting what it waits on, so nothing is written.
#[test]
fn a_closed_conversation_with_an_arm_installed_is_not_retired() -> TestResult {
    let store = memory();
    let rows = [
        ObserverRow::Track {
            conversation_id: 5,
            observer_progress: 0,
        },
        ObserverRow::Arms { arms: vec![(5, 0)] },
    ]
    .iter()
    .map(encode_for_tests)
    .collect::<Result<Vec<_>, _>>()?;
    block_on(store.append_all(OBSERVER_STREAM_KEY, rows, 0))??;
    ConversationRegistry::new(Arc::clone(&store)).register(5)?;
    let recording = Recording::new(Arc::clone(&store));
    let handler = ProductionParticipantHandler::new(recording.shared(), test_participant_config())?;
    recording.take_batches()?;
    let writes = recording.other_writes.load(Ordering::SeqCst);

    assert_eq!(handler.retire_for_test(5), Retirement::Armed);
    assert!(recording.take_batches()?.is_empty());
    assert_eq!(recording.other_writes.load(Ordering::SeqCst), writes);
    assert_eq!(
        ConversationRegistry::new(store).restore_sets()?,
        RegistrySets::of_live([5].into())
    );
    Ok(())
}

/// The retirement is attempted on commits alone: its source names no sleep,
/// timer, clock or wait.
#[test]
fn retirement_waits_on_nothing() {
    // Code only: the module's own prose says what it never does.
    let source: String = include_str!("retirement.rs")
        .lines()
        .filter(|line| !line.trim_start().starts_with("//"))
        .collect::<Vec<_>>()
        .join("\n");
    for word in [
        "sleep", "Instant", "Duration", "interval", "timeout", "spawn", "loop {",
    ] {
        assert!(!source.contains(word), "retirement.rs names {word:?}");
    }
}