liminal-server 0.23.1

//! Registry restore must depend on live state rather than append-only age.

use std::collections::{BTreeMap, BTreeSet};

use crate::boot_history_harness::{
    ReadCounts, ReadSource, StageReads, TestResult, during_stage, history_pair,
};
use haematite::tree::Hash;
use liminal::durability::bridge::block_on;

use super::registry::ConversationRegistry;
use super::registry_snapshot::{SNAPSHOT_KEY, write_registry_snapshot};
use super::routed_stream_codec::head_key;

const STREAM: &str = "liminal:participant-conversation-registry";

/// The registry stage reads the same rows and bytes at both history lengths.
///
/// The stage owes one snapshot read and two head reads (DUR-019 R4): the
/// snapshot (a routed value of one id, 64 + 8 bytes), the routed head (an
/// absent routed value, recorded as 0 rows), and the native head (8 bytes).
/// Both images hold flat rows `0..range.end`, no routed head and a snapshot
/// written at the native head by the production writer, so the tail is empty
/// and restore issues no row read: a cut every 256 rows would leave 232 tail
/// rows at 1,000 and 160 at 100,000, and the images would read differently.
#[test]
fn registry_boot_history_reads_do_not_grow_with_duplicate_creation_history() -> TestResult<()> {
    let payload = serde_json::to_vec(&serde_json::json!({
        "schema_version": 1,
        "conversation_id": 7,
    }))?;
    let pair = history_pair("registry-snapshot-v1", |store, range| {
        let expected = range.end;
        let start = range.start;
        let payloads = range.map(|_| payload.clone()).collect();
        let next = block_on(store.append_all(STREAM, payloads, start))??;
        assert_eq!(next, expected);
        let prior = block_on(store.get_routed(STREAM, SNAPSHOT_KEY))??;
        write_registry_snapshot(
            store,
            prior.as_deref().map(Hash::of),
            next,
            &BTreeSet::from([7]),
        )?;
        Ok(())
    })?;
    let expected = StageReads::from([(
        "registry".to_owned(),
        BTreeMap::from([
            (
                ReadSource::RoutedValue {
                    route: STREAM.to_owned(),
                    key: SNAPSHOT_KEY.to_vec(),
                },
                ReadCounts {
                    rows: 1,
                    bytes: 64 + 8,
                },
            ),
            (
                ReadSource::RoutedValue {
                    route: STREAM.to_owned(),
                    key: head_key(STREAM)?,
                },
                ReadCounts::default(),
            ),
            (
                ReadSource::StreamHead(STREAM.to_owned()),
                ReadCounts { rows: 1, bytes: 8 },
            ),
        ]),
    )]);
    pair.assert_stage("registry", &expected, |store| {
        let registry = ConversationRegistry::new(store);
        let restored = during_stage("registry", || registry.restore())?;
        assert_eq!(restored, vec![7]);
        Ok(())
    })?;
    Ok(())
}

/// The fixture's snapshot comes from the production writer, never from
/// bytes framed here, and the routed head key from the codec.
#[test]
fn registry_boot_history_fixture_uses_the_production_writer() {
    let source = include_str!("tests_boot_history.rs");
    assert!(!source.contains(concat!("LS", "RS")));
    assert!(source.contains("write_registry_snapshot(\n"));
    assert!(source.contains("key: head_key(STREAM)?"));
}

#[test]
fn harness_boot_history_rejects_a_full_registry_walk() -> TestResult<()> {
    let payload = serde_json::to_vec(&serde_json::json!({
        "schema_version": 1,
        "conversation_id": 7,
    }))?;
    let bytes = u64::try_from(payload.len())?;
    let pair = history_pair("registry-duplicate-v1", |store, range| {
        let expected = range.end;
        let start = range.start;
        let payloads = range.map(|_| payload.clone()).collect();
        assert_eq!(
            block_on(store.append_all(STREAM, payloads, start))??,
            expected
        );
        Ok(())
    })?;
    let expected = StageReads::from([(
        "registry".to_owned(),
        BTreeMap::from([(
            ReadSource::Stream(STREAM.to_owned()),
            ReadCounts {
                rows: 1_000,
                bytes: 1_000 * bytes,
            },
        )]),
    )]);
    let refusal = pair
        .assert_stage("registry", &expected, |store| {
            during_stage("registry", || -> TestResult<()> {
                let mut sequence = 0_u64;
                loop {
                    let rows = block_on(store.read_from(STREAM, sequence, 64))??;
                    if rows.is_empty() {
                        break;
                    }
                    for row in rows {
                        assert_eq!(row.sequence, sequence);
                        assert_eq!(row.payload, payload);
                        sequence = sequence.checked_add(1).ok_or("control sequence overflow")?;
                    }
                }
                assert!(crate::boot_history_harness::HISTORY_ROWS.contains(&sequence));
                Ok(())
            })
        })
        .err()
        .ok_or("boot history harness accepted a full registry walk")?
        .to_string();
    assert!(refusal.starts_with("boot stage registry: expected "));
    assert!(refusal.contains(&format!("rows: 1000, bytes: {}", 1_000 * bytes)));
    assert!(refusal.contains(&format!("rows: 100000, bytes: {}", 100_000 * bytes)));
    Ok(())
}