liminal-server 0.24.1

//! DUR-027 R4: the registry holds only live and sweeping conversations, and a
//! retired id is tombstoned so it is never reopened.
//!
//! Registration, retirement and sweep rows cross a snapshot and restore into
//! the two sets; a store written before retirement (version 1 rows and a 1.0
//! snapshot) restores unchanged and is not rewritten; a row of an unknown
//! schema and a retire row for an id that is not live are refused by name.
//! The tombstone is the generation key at [`RETIRED_GENERATION`]: opening the
//! log refuses `ConversationRetired`, and an enrollment presenting the id is
//! answered `ConversationSealed` with no write to any stream.

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

use liminal::durability::bridge::block_on;
use liminal::durability::{DurableStore, RoutedBatchItem};
use liminal_protocol::wire::{
    ClientRequest, ConnectionIncarnation, EnrollmentRequest, EnrollmentToken,
};

use crate::server::mount::MountKind;
use crate::server::participant::{
    ParticipantConnectionContext, ParticipantConnectionConversations, ParticipantSemanticError,
    ParticipantSemanticHandler,
};

use super::ProductionParticipantHandler;
use super::log::{OperationLog, OperationLogError, RETIRED_GENERATION};
use super::log_error::{RegistryRowRefusal, StreamKeyRefusal};
use super::registry::ConversationRegistry;
use super::registry_snapshot::{RegistrySets, SNAPSHOT_KEY};
use super::tests::test_participant_config;
use super::tests_registry_snapshot_fixture::{
    KEY, Recording, TestResult, append_flat, ascending, flat_ids, memory, stored_snapshot,
};

fn row(schema_version: u8, kind: &str, conversation_id: u64) -> TestResult<Vec<u8>> {
    Ok(serde_json::to_vec(&serde_json::json!({
        "schema_version": schema_version,
        "kind": kind,
        "conversation_id": conversation_id,
    }))?)
}

/// 300 registrations cross the snapshot at 256; 299 retirements cross the
/// next at 512, whose snapshot holds the 212 ids retired below it as
/// sweeping and none of them live. A fresh registry restores the sets from
/// that snapshot and the tail; after the sweep, nothing is sweeping.
#[test]
fn registry_restores_live_and_sweeping_sets_across_a_snapshot() -> TestResult {
    let store = memory();
    let registry = ConversationRegistry::new(Arc::clone(&store));
    for id in 1..=300 {
        registry.register(id)?;
    }
    for id in 1..=299 {
        registry.retire(id)?;
    }
    let (snapshot, _) = stored_snapshot(store.as_ref())?.ok_or("no snapshot was written")?;
    assert_eq!(snapshot.covered_next_seq, 512);
    assert_eq!(snapshot.sets.live, ascending(213..301));
    assert_eq!(snapshot.sets.sweeping, ascending(1..213));

    let restored = ConversationRegistry::new(Arc::clone(&store));
    assert_eq!(
        restored.restore_sets()?,
        RegistrySets {
            live: BTreeSet::from([300]),
            sweeping: ascending(1..300),
        }
    );
    assert_eq!(
        ConversationRegistry::new(Arc::clone(&store)).restore()?,
        vec![300]
    );

    for id in 1..=299 {
        restored.swept(id)?;
    }
    assert_eq!(
        ConversationRegistry::new(Arc::clone(&store)).restore_sets()?,
        RegistrySets::of_live(BTreeSet::from([300]))
    );
    Ok(())
}

/// The 1.0 snapshot a store written before retirement holds: `covered` and
/// `ids`, closed by blake3 over its own domain, framed here outside the codec.
fn v1_snapshot(covered: u64, ids: &BTreeSet<u64>) -> TestResult<Vec<u8>> {
    let mut bytes = b"LSRS".to_vec();
    bytes.extend_from_slice(&[0, 1, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0]);
    bytes.extend_from_slice(&covered.to_be_bytes());
    bytes.extend_from_slice(&u64::try_from(ids.len())?.to_be_bytes());
    for id in ids {
        bytes.extend_from_slice(&id.to_be_bytes());
    }
    let mut hash = blake3::Hasher::new();
    hash.update(b"liminal-registry-snapshot-v1");
    hash.update(&bytes);
    bytes.extend_from_slice(hash.finalize().as_bytes());
    Ok(bytes)
}

#[test]
fn registry_of_version_1_rows_and_a_version_1_snapshot_restores_unchanged() -> TestResult {
    let store = memory();
    let ids = ascending(0..300);
    let covered = flat_ids(store.as_ref(), ids.iter().copied())?;
    block_on(store.apply_routed_batch(
        KEY,
        vec![RoutedBatchItem {
            key: SNAPSHOT_KEY.to_vec(),
            expected: None,
            value: v1_snapshot(covered, &ids)?,
            ttl: None,
        }],
    ))??;
    let recording = Recording::new(Arc::clone(&store));
    let registry = ConversationRegistry::new(recording.shared());
    assert_eq!(registry.restore_sets()?, RegistrySets::of_live(ids));
    assert!(recording.take_batches()?.is_empty());
    assert_eq!(recording.other_writes.load(Ordering::SeqCst), 0);
    assert_eq!(recording.flushes.load(Ordering::SeqCst), 0);
    Ok(())
}

#[test]
fn registry_refuses_a_version_3_row_and_a_retire_of_an_id_not_live_by_name() -> TestResult {
    let future = memory();
    flat_ids(future.as_ref(), [1].into_iter())?;
    append_flat(future.as_ref(), vec![row(3, "register", 2)?])?;
    let refused = ConversationRegistry::new(future).restore_sets();
    assert!(
        matches!(refused, Err(OperationLogError::SchemaVersion(3))),
        "a version 3 row was answered {refused:?}"
    );

    let orphan = memory();
    flat_ids(orphan.as_ref(), [1].into_iter())?;
    append_flat(orphan.as_ref(), vec![row(2, "retire", 5)?])?;
    let refused = ConversationRegistry::new(orphan).restore_sets();
    assert!(
        matches!(
            refused,
            Err(OperationLogError::RegistryRow {
                sequence: 1,
                conversation_id: 5,
                refusal: RegistryRowRefusal::RetireNotLive,
            })
        ),
        "a retire of an id that is not live was answered {refused:?}"
    );
    Ok(())
}

/// Moves `conversation_id`'s generation key to the tombstone, as retirement
/// does, by compare-and-swap from the original generation.
fn tombstone(store: &Arc<dyn DurableStore>, conversation_id: u64) -> TestResult {
    block_on(
        OperationLog::new(Arc::clone(store), conversation_id)
            .switch_generation(0, RETIRED_GENERATION),
    )??;
    Ok(())
}

#[test]
fn opening_a_tombstoned_log_refuses_conversation_retired() -> TestResult {
    let store = memory();
    tombstone(&store, 77)?;
    let log = OperationLog::new(Arc::clone(&store), 77);
    let key = block_on(log.stream_key())?;
    assert!(
        matches!(
            key,
            Err(StreamKeyRefusal::Retired {
                conversation_id: 77
            })
        ),
        "the tombstoned log named a stream: {key:?}"
    );
    let page = block_on(log.read_raw_page(0))?;
    assert!(
        matches!(
            page,
            Err(OperationLogError::ConversationRetired {
                conversation_id: 77
            })
        ),
        "the tombstoned log was read: {page:?}"
    );
    Ok(())
}

/// A late enrollment presenting a retired id meets the sealed refusal by
/// type, and the store sees no batch, write or flush: no Genesis, no
/// registry row.
#[test]
fn an_enrollment_presenting_a_retired_id_is_sealed_and_writes_nothing() -> TestResult {
    let store = memory();
    tombstone(&store, 77)?;
    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);
    let flushes = recording.flushes.load(Ordering::SeqCst);

    let refused = handler.handle(
        ParticipantConnectionContext::new(ConnectionIncarnation::new(0x77, 1), MountKind::Tcp),
        &mut ParticipantConnectionConversations::default(),
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id: 77,
            enrollment_token: EnrollmentToken::new([0x77; 16]),
        }),
    );
    assert!(
        matches!(
            refused,
            Err(ParticipantSemanticError::ConversationSealed {
                conversation_id: 77
            })
        ),
        "an enrollment presenting a retired id was answered {refused:?}"
    );
    assert!(recording.take_batches()?.is_empty());
    assert_eq!(recording.other_writes.load(Ordering::SeqCst), writes);
    assert_eq!(recording.flushes.load(Ordering::SeqCst), flushes);
    Ok(())
}