liminal-server 0.24.1

//! Registry snapshot pins, R2 (DUR-019): one routed batch per registration,
//! the crossing carries the snapshot, and a struck crossing is recovered.

use std::collections::BTreeSet;
use std::sync::Arc;
use std::sync::atomic::Ordering;

use liminal::durability::DurableStore;
use liminal::durability::bridge::block_on;

use super::registry::ConversationRegistry;
use super::registry_snapshot::SNAPSHOT_KEY;
use super::route_key_migration::MARKER_KEY;
use super::routed_stream::read_head;
use super::routed_stream_codec::{head_key, row_key};
use super::tests_outcome_unknown::{UnknownOutcomeStore, Write};
use super::tests_registry_snapshot_fixture::{
    KEY, Recording, TestResult, ascending, logged, memory, snapshot_in, stored_snapshot,
};

// ---------------------------------------------------------------------------
// R2: one routed batch per registration; the crossing carries the snapshot.
// ---------------------------------------------------------------------------

#[test]
fn registry_registration_is_one_batch_and_the_crossing_carries_the_snapshot() -> TestResult {
    let recording = Recording::new(memory());
    let registry = ConversationRegistry::new(recording.shared());
    assert!(registry.restore()?.is_empty());
    recording.take_batches()?;
    recording.other_writes.store(0, Ordering::SeqCst);
    let head = head_key(KEY)?;
    for id in 0..255 {
        registry.register(id)?;
    }
    let batches = recording.take_batches()?;
    assert_eq!(batches.len(), 255);
    for (sequence, (route, items)) in (0_u64..).zip(&batches) {
        assert_eq!(route, KEY);
        let keys: Vec<Vec<u8>> = items.iter().map(|item| item.key.clone()).collect();
        // The first row's batch carries the route-key migration marker
        // (DUR-028 R4): a store registered by this code never migrates.
        let expected = if sequence == 0 {
            vec![row_key(KEY, 0)?, MARKER_KEY.to_vec(), head.clone()]
        } else {
            vec![row_key(KEY, sequence)?, head.clone()]
        };
        assert_eq!(keys, expected);
    }
    assert_eq!(recording.flushes.load(Ordering::SeqCst), 0);
    // Every registration's write is its one batch: no append, cas, truncate
    // or removal beside it, and no flush (the store settles the batch).
    assert_eq!(recording.other_writes.load(Ordering::SeqCst), 0);
    assert!(stored_snapshot(recording.as_ref())?.is_none());

    registry.register(255)?;
    let batches = recording.take_batches()?;
    let [(route, items)] = batches.as_slice() else {
        return Err(format!("the crossing wrote {} batches", batches.len()).into());
    };
    assert_eq!(route, KEY);
    // OperationBatch::commit appends the routed head after every staged
    // item, so the snapshot stands between the row and the head.
    let keys: Vec<Vec<u8>> = items.iter().map(|item| item.key.clone()).collect();
    assert_eq!(keys, vec![row_key(KEY, 255)?, SNAPSHOT_KEY.to_vec(), head]);
    let (at_256, expected) = snapshot_in(items)?;
    assert_eq!(expected, None);
    assert_eq!(at_256.covered_next_seq, 256);
    assert_eq!(at_256.sets.live, ascending(0..256));
    let (_, hash_256) = stored_snapshot(recording.as_ref())?.ok_or("no snapshot at 256")?;

    for id in 256..511 {
        registry.register(id)?;
    }
    recording.take_batches()?;
    registry.register(511)?;
    let batches = recording.take_batches()?;
    let [(_, items)] = batches.as_slice() else {
        return Err("the second crossing wrote no single batch".into());
    };
    let (at_512, expected) = snapshot_in(items)?;
    assert_eq!(expected, Some(hash_256));
    assert_eq!(at_512.covered_next_seq, 512);
    assert_eq!(at_512.sets.live, ascending(0..512));
    assert_eq!(recording.flushes.load(Ordering::SeqCst), 0);
    assert_eq!(recording.other_writes.load(Ordering::SeqCst), 0);
    Ok(())
}

/// A memory store and a fault store over it striking `write` under the
/// registry key.
fn struck_store(write: Write) -> (Arc<dyn DurableStore>, Arc<UnknownOutcomeStore>) {
    let memory = memory();
    let fault = Arc::new(UnknownOutcomeStore::new(&memory, write, KEY));
    (memory, fault)
}

fn registered_255(store: &Arc<dyn DurableStore>) -> TestResult<ConversationRegistry> {
    let registry = ConversationRegistry::new(Arc::clone(store));
    registry.restore()?;
    for id in 0..255 {
        registry.register(id)?;
    }
    Ok(registry)
}

/// The crossing batch struck before it applied: nothing of it lands, no
/// warning stands in for the fault, and the registration answers Err.
fn strike_unlanded_crossing() -> TestResult<(
    Arc<dyn DurableStore>,
    Arc<UnknownOutcomeStore>,
    ConversationRegistry,
)> {
    let (memory, fault) = struck_store(Write::AppendAbsentIo);
    let store: Arc<dyn DurableStore> = fault.clone();
    let registry = registered_255(&store)?;
    fault.arm();
    let (answer, logs) = logged(|| registry.register(255))?;
    assert!(answer.is_err(), "the struck crossing answered {answer:?}");
    assert!(fault.struck(), "the crossing batch was never struck");
    assert!(
        !logs.contains("WARN"),
        "a warning stood in for the fault: {logs}"
    );
    fault.recover();
    assert_eq!(
        block_on(memory.get_routed(KEY, &row_key(KEY, 255)?))??,
        None
    );
    assert!(stored_snapshot(memory.as_ref())?.is_none());
    let routed = block_on(read_head(memory.as_ref(), KEY, KEY))??.ok_or("no routed head")?;
    assert_eq!(routed.next_sequence, 255);
    Ok((memory, fault, registry))
}

#[test]
fn registry_unlanded_crossing_is_recovered_at_its_original_head_after_restart() -> TestResult {
    let (memory, fault, _) = strike_unlanded_crossing()?;
    let store: Arc<dyn DurableStore> = fault;
    let restarted = ConversationRegistry::new(Arc::clone(&store));
    assert_eq!(restarted.restore()?, (0_u64..255).collect::<Vec<_>>());
    restarted.register(255)?;
    assert!(block_on(memory.get_routed(KEY, &row_key(KEY, 255)?))??.is_some());
    let (snapshot, _) = stored_snapshot(memory.as_ref())?.ok_or("no snapshot at 256")?;
    assert_eq!(snapshot.covered_next_seq, 256);
    assert_eq!(snapshot.sets.live, ascending(0..256));
    assert_eq!(
        ConversationRegistry::new(store).restore()?,
        (0_u64..256).collect::<Vec<_>>()
    );
    Ok(())
}

#[test]
fn registry_unlanded_crossing_is_recovered_on_the_same_handle() -> TestResult {
    let (memory, fault, registry) = strike_unlanded_crossing()?;
    registry.register(255)?;
    let (snapshot, _) = stored_snapshot(memory.as_ref())?.ok_or("no snapshot at 256")?;
    assert_eq!(snapshot.covered_next_seq, 256);
    assert_eq!(snapshot.sets.live, ascending(0..256));
    let store: Arc<dyn DurableStore> = fault;
    assert_eq!(
        ConversationRegistry::new(store).restore()?,
        (0_u64..256).collect::<Vec<_>>()
    );
    Ok(())
}

#[test]
fn registry_landed_crossing_under_an_unknown_outcome_keeps_its_id_and_hash() -> TestResult {
    let (memory, fault) = struck_store(Write::Append);
    let recording = Recording::new(fault.clone() as Arc<dyn DurableStore>);
    let store = recording.shared();
    let registry = registered_255(&store)?;
    fault.arm_unsettled();
    assert!(registry.register(255).is_err());
    assert!(fault.struck(), "the crossing batch was never struck");
    let (landed, landed_hash) =
        stored_snapshot(memory.as_ref())?.ok_or("the landed crossing left no snapshot")?;
    assert_eq!(landed.covered_next_seq, 256);
    registry.register(256)?;
    assert!(block_on(memory.get_routed(KEY, &row_key(KEY, 256)?))??.is_some());
    for id in 257..511 {
        registry.register(id)?;
    }
    recording.take_batches()?;
    registry.register(511)?;
    let batches = recording.take_batches()?;
    let [(_, items)] = batches.as_slice() else {
        return Err("the crossing at 512 wrote no single batch".into());
    };
    let (at_512, expected) = snapshot_in(items)?;
    assert_eq!(expected, Some(landed_hash));
    assert!(
        at_512.sets.live.contains(&255),
        "the snapshot at 512 misses the id of the row that landed unanswered"
    );
    assert_eq!(at_512.sets.live, ascending(0..512));
    Ok(())
}

#[test]
fn registry_duplicate_registrations_snapshot_one_id() -> TestResult {
    let memory = memory();
    let registry = ConversationRegistry::new(Arc::clone(&memory));
    registry.restore()?;
    for _ in 0..256 {
        registry.register(7)?;
    }
    let (snapshot, _) = stored_snapshot(memory.as_ref())?.ok_or("no snapshot at 256")?;
    assert_eq!(snapshot.covered_next_seq, 256);
    assert_eq!(snapshot.sets.live, BTreeSet::from([7]));
    assert_eq!(ConversationRegistry::new(memory).restore()?, vec![7]);
    Ok(())
}