liminal-server 0.24.1

//! DUR-020: the server listens before it restores any conversation.
//!
//! The participant handler is built before the listener binds. Its store work
//! must not grow with the number of registered conversations (R1), and a store
//! that holds no checkpoint, the shape of an install written before DUR-010,
//! applies no batch before bind: the conversation rebuilds on its first touch
//! instead (R1, R2).

use std::collections::BTreeSet;
use std::error::Error;
use std::sync::{Arc, Barrier};

use liminal::durability::bridge::block_on;
use liminal::durability::{CountingStore, DurableStore, StoreCounts};

use liminal_protocol::wire::{
    ClientRequest, ConnectionIncarnation, EnrollmentRequest, EnrollmentToken,
    IdentityCapacityExceeded, IdentityCapacityScope, ServerValue,
};

use crate::server::participant::ParticipantSemanticHandler;

use super::ProductionParticipantHandler;
use super::operation_batch::OperationBatch;
use super::registry_snapshot::{RegistrySets, write_registry_snapshot};
use super::route_key_migration::fresh_marker;
use super::tests::{dispatch, test_participant_config};
use super::tests_capacity::capacity_config;
use super::tests_memory_store::MemoryStore;
use super::tests_receipts::enroll;
use super::tests_restore_from_checkpoint_boot::{
    assert_checkpoint, assert_enrollment_unchanged, boot, legacy_fixture,
};
use super::tests_seed::Shared;

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

/// A store whose registry lists conversations `1..=count`, each registered
/// by its own row and all covered by one snapshot, as an install that has
/// migrated (DUR-019) holds them, with the route-key marker its first
/// registration would have carried (DUR-028 R4). No conversation has a row
/// of its own.
fn registered(count: u64) -> Result<Arc<dyn DurableStore>, Box<dyn Error>> {
    let store: Arc<dyn DurableStore> = Arc::new(MemoryStore::default());
    let payloads = (1..=count)
        .map(|conversation_id| {
            serde_json::to_vec(&serde_json::json!({
                "schema_version": 1,
                "conversation_id": conversation_id,
            }))
        })
        .collect::<Result<Vec<_>, _>>()?;
    assert_eq!(block_on(store.append_all(REGISTRY, payloads, 0))??, count);
    let ids: BTreeSet<u64> = (1..=count).collect();
    write_registry_snapshot(store.as_ref(), None, count, &RegistrySets::of_live(ids))?;
    let mut marker = OperationBatch::new(REGISTRY);
    marker.stage(fresh_marker())?;
    block_on(marker.commit(store.as_ref(), || {}))??;
    Ok(store)
}

/// The store work of building the handler and running the incarnation's
/// startup repair, which is all the participant service does before the
/// listener binds. The handler is built alone, so no warm thread starts and
/// nothing races the reading.
fn construction_counts(store: Arc<dyn DurableStore>) -> Result<StoreCounts, Box<dyn Error>> {
    let counted = Arc::new(CountingStore::new(Shared(store)));
    let handler = ProductionParticipantHandler::new(
        Arc::clone(&counted) as Arc<dyn DurableStore>,
        test_participant_config(),
    )?;
    handler.repair_unclean_server_restart(2)?;
    let counts = counted.snapshot();
    drop(handler);
    Ok(counts)
}

/// R1: ten registered conversations and ten thousand cost the handler the
/// same store work before bind, and none of it writes. The registered
/// conversations hold no rows of their own: before DUR-020 construction read
/// each one's log and checkpoint route even when empty, so the counts grew
/// with the registered count and this pin is red there.
#[test]
fn construction_reads_the_same_at_ten_and_ten_thousand_conversations() -> Result<(), Box<dyn Error>>
{
    let few = construction_counts(registered(10)?)?;
    let many = construction_counts(registered(10_000)?)?;
    assert_eq!(
        few.named(),
        many.named(),
        "the handler's store work before bind grew with the registered count"
    );
    for counts in [few, many] {
        assert_eq!(counts.apply_routed_batch_calls, 0);
        assert_eq!(counts.append_calls, 0);
        assert_eq!(counts.append_all_calls, 0);
        assert_eq!(counts.flushes, 0);
    }
    Ok(())
}

/// R1 and R2 over a store with no checkpoint: construction applies no batch
/// and logs no rebuild; the first touch rebuilds and publishes the checkpoint
/// once, answers exactly as before, and a second touch publishes nothing.
#[test]
fn an_install_with_no_checkpoint_rebuilds_on_first_touch_not_before_bind()
-> Result<(), Box<dyn Error>> {
    let (directory, store) = legacy_fixture()?;
    let counted = Arc::new(CountingStore::new(Shared(Arc::clone(&store))));
    let (handler, counts, diagnostics) = boot(Arc::clone(&counted) as Arc<dyn DurableStore>)?;
    assert_eq!(counts.apply_routed_batch_calls, 0, "{diagnostics}");
    assert!(!diagnostics.contains("CHECKPOINT_REBUILD"), "{diagnostics}");
    let before_touch = counted.snapshot();
    assert_enrollment_unchanged(&directory, &handler)?;
    assert_checkpoint(store.as_ref())?;
    let touched = counted.snapshot().since(&before_touch);
    assert_eq!(
        touched.apply_routed_batch_calls, 1,
        "the first touch publishes the rebuilt checkpoint in one batch and the replayed answers commit nothing"
    );
    let before_again = counted.snapshot();
    assert_enrollment_unchanged(&directory, &handler)?;
    assert_eq!(
        counted
            .snapshot()
            .since(&before_again)
            .apply_routed_batch_calls,
        0
    );
    Ok(())
}

/// R4: the warm restores a conversation nobody touches. Over a store with no
/// checkpoint, construction leaves the conversation pending; the warm runs
/// its restore unit, which publishes the rebuilt checkpoint once, and a
/// second warm finds nothing pending and applies nothing.
#[test]
fn the_warm_restores_a_conversation_nobody_touches() -> Result<(), Box<dyn Error>> {
    let (_directory, store) = legacy_fixture()?;
    let counted = Arc::new(CountingStore::new(Shared(Arc::clone(&store))));
    let (handler, counts, _) = boot(Arc::clone(&counted) as Arc<dyn DurableStore>)?;
    assert_eq!(counts.apply_routed_batch_calls, 0);
    let before_warm = counted.snapshot();
    assert_eq!(
        handler.restore_remaining("the warm")?,
        0,
        "the warm refused a conversation"
    );
    assert_checkpoint(store.as_ref())?;
    assert_eq!(
        counted
            .snapshot()
            .since(&before_warm)
            .apply_routed_batch_calls,
        1
    );
    let before_again = counted.snapshot();
    assert_eq!(handler.restore_remaining("the warm")?, 0);
    assert_eq!(
        counted.snapshot().since(&before_again),
        StoreCounts::default(),
        "a finished warm still touched the store"
    );
    Ok(())
}

/// The store work of the first enrollment after a clean stop, over `count`
/// pending conversations, and the refusal it answers at a server identity
/// cap of one.
fn enrollment_after_clean_stop(count: u64) -> Result<StoreCounts, Box<dyn Error>> {
    let store = registered(count)?;
    let config = capacity_config(|config| config.max_retired_identity_slots_server = 1);
    let incarnation = ConnectionIncarnation::new(71, 1);
    let before_stop = ProductionParticipantHandler::new(Arc::clone(&store), config)?;
    enroll(&before_stop, incarnation, count + 1, [1; 16])?;
    before_stop.record_clean_stop_capacity();
    drop(before_stop);
    let counted = Arc::new(CountingStore::new(Shared(store)));
    let after_stop =
        ProductionParticipantHandler::new(Arc::clone(&counted) as Arc<dyn DurableStore>, config)?;
    let before_enrol = counted.snapshot();
    let refused = dispatch(
        &after_stop,
        incarnation,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id: count + 2,
            enrollment_token: EnrollmentToken::new([2; 16]),
        }),
    )?;
    let ServerValue::IdentityCapacityExceeded(IdentityCapacityExceeded {
        scope,
        limit,
        occupied,
        ..
    }) = refused
    else {
        return Err(format!("the seeded ledger did not refuse at the cap: {refused:?}").into());
    };
    assert_eq!(
        (scope, limit, occupied),
        (IdentityCapacityScope::Server, 1, 1)
    );
    Ok(counted.snapshot().since(&before_enrol))
}

/// R3 after a clean stop (Hermes 11:47): the next process seeds its capacity
/// ledger from the stop's snapshot, so its first enrollment waits for no warm.
/// It does the same store work at 10 and at 10,000 pending conversations,
/// one batch of it consuming the snapshot, and still refuses at the signed
/// server identity cap. After an unclean stop there is no snapshot and the
/// enrollment waits for the warm instead:
/// `tests_capacity::enrollment_identity_server_scope_refuses_and_survives_restart`
/// pins that it then answers right.
#[test]
fn after_a_clean_stop_enrollment_reads_the_same_at_ten_and_ten_thousand_conversations()
-> Result<(), Box<dyn Error>> {
    let few = enrollment_after_clean_stop(10)?;
    let many = enrollment_after_clean_stop(10_000)?;
    assert_eq!(
        few.named(),
        many.named(),
        "the first enrollment after a clean stop waited on the pending conversations"
    );
    assert_eq!(
        few.apply_routed_batch_calls, 1,
        "the snapshot is consumed once"
    );
    Ok(())
}

/// R2: a request and the warm reaching the same pending conversation run its
/// restore unit once between them. Both start behind one barrier; whichever
/// takes the conversation's gate first restores and publishes, and the other
/// finds it restored. The threads overlap only as the scheduler lets them, so
/// a run that happens to serialize them pins the sequential order; the
/// single-thread halves of
/// `an_install_with_no_checkpoint_rebuilds_on_first_touch_not_before_bind`
/// and `the_warm_restores_a_conversation_nobody_touches` pin each order alone.
#[test]
fn a_request_racing_the_warm_restores_the_conversation_once() -> Result<(), Box<dyn Error>> {
    let (directory, store) = legacy_fixture()?;
    let counted = Arc::new(CountingStore::new(Shared(Arc::clone(&store))));
    let (handler, _, _) = boot(Arc::clone(&counted) as Arc<dyn DurableStore>)?;
    let before = counted.snapshot();
    let barrier = Barrier::new(2);
    let warm_refused = std::thread::scope(|scope| {
        let warm = scope.spawn(|| {
            barrier.wait();
            handler.restore_remaining("the warm")
        });
        barrier.wait();
        let request = assert_enrollment_unchanged(&directory, &handler);
        let warm = warm.join().map_err(|_| "the warm thread panicked")?;
        request?;
        Ok::<_, Box<dyn Error>>(warm?)
    })?;
    assert_eq!(warm_refused, 0, "the warm refused the conversation");
    assert_checkpoint(store.as_ref())?;
    assert_eq!(
        counted.snapshot().since(&before).apply_routed_batch_calls,
        1,
        "the request and the warm each ran the restore unit"
    );
    Ok(())
}