liminal-server 0.24.5

//! Observer-progress witnesses are pruned once their progress is durable: the
//! state stays bounded however long the conversation runs, and a terminal
//! occurrence presented again at or below the pruned point is refused by name.

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

use liminal::durability::{DurableStore, open_ephemeral};
use liminal_protocol::lifecycle::ObserverProgressProjection;
use liminal_protocol::wire::{
    BindingEpoch, ClientRequest, ConnectionIncarnation, EnrollmentRequest, EnrollmentToken,
    Generation, ParticipantAck, RecordAdmission, RecordAdmissionAttemptToken, ServerValue,
};

use super::ProductionParticipantHandler;
use super::observer_progress::{
    ObserverProgressConformanceError, ObserverProgressSourceMetadata, ObserverProgressWitnessState,
};
use super::tests::{dispatch, test_participant_config};

const CONVERSATION: u64 = 0x0B5;
const CYCLES: u64 = 120;

fn enroll(
    handler: &ProductionParticipantHandler,
    connection: ConnectionIncarnation,
    token: u8,
) -> Result<u64, Box<dyn Error>> {
    let enrolled = dispatch(
        handler,
        connection,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id: CONVERSATION,
            enrollment_token: EnrollmentToken::new([token; 16]),
        }),
    )?;
    let ServerValue::EnrollBound(receipt) = enrolled else {
        return Err(format!("enrollment {token:#x} did not bind: {enrolled:?}").into());
    };
    Ok(receipt.participant_id())
}

fn ack(
    handler: &ProductionParticipantHandler,
    connection: ConnectionIncarnation,
    participant_id: u64,
    through_seq: u64,
) -> Result<(), Box<dyn Error>> {
    let answered = dispatch(
        handler,
        connection,
        ClientRequest::ParticipantAck(ParticipantAck {
            conversation_id: CONVERSATION,
            participant_id,
            capability_generation: Generation::ONE,
            through_seq,
        }),
    )?;
    if matches!(answered, ServerValue::AckCommitted(_)) {
        Ok(())
    } else {
        Err(format!("ack through {through_seq} did not commit: {answered:?}").into())
    }
}

fn admit(
    handler: &ProductionParticipantHandler,
    connection: ConnectionIncarnation,
    sender: u64,
    cycle: u64,
) -> Result<u64, Box<dyn Error>> {
    let mut token = [0_u8; 16];
    token[..8].copy_from_slice(&cycle.to_be_bytes());
    let committed = dispatch(
        handler,
        connection,
        ClientRequest::RecordAdmission(RecordAdmission {
            conversation_id: CONVERSATION,
            participant_id: sender,
            capability_generation: Generation::ONE,
            record_admission_attempt_token: RecordAdmissionAttemptToken::new(token),
            payload: vec![0x0B; 32],
        }),
    )?;
    let ServerValue::RecordCommitted(committed) = committed else {
        return Err(format!("record {cycle} did not commit: {committed:?}").into());
    };
    Ok(committed.delivery_seq())
}

/// Witnesses and occurrence keys held, and the pruned-through progress.
fn witness_state(
    handler: &ProductionParticipantHandler,
) -> Result<(usize, usize, u64), Box<dyn Error>> {
    let cell = handler.cell(CONVERSATION)?;
    let guard = cell.lock().map_err(|_| "conversation lock poisoned")?;
    let held = guard.as_ref().map(|authority| {
        let (witnesses, occurrences) = authority.observer_witness_state_held();
        (
            witnesses,
            occurrences,
            authority.observer_progress_pruned_through(),
        )
    });
    drop(guard);
    Ok(held.ok_or("conversation is not loaded")?)
}

#[test]
fn witness_state_stays_bounded_as_observer_progress_becomes_durable() -> Result<(), Box<dyn Error>>
{
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let handler = ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;
    let recipient_connection = ConnectionIncarnation::new(0x0B5, 1);
    let sender_connection = ConnectionIncarnation::new(0x0B5, 2);
    let recipient = enroll(&handler, recipient_connection, 0x51)?;
    let sender = enroll(&handler, sender_connection, 0x52)?;
    ack(&handler, recipient_connection, recipient, 2)?;
    let mut largest = (0, 0);
    let mut last_pruned = 0;
    for cycle in 0..CYCLES {
        let delivered = admit(&handler, sender_connection, sender, cycle)?;
        ack(&handler, recipient_connection, recipient, delivered)?;
        let (witnesses, occurrences, pruned_through) = witness_state(&handler)?;
        largest = (largest.0.max(witnesses), largest.1.max(occurrences));
        assert!(
            pruned_through >= last_pruned,
            "pruned progress moved backwards"
        );
        last_pruned = pruned_through;
    }
    assert!(
        last_pruned > CYCLES,
        "observer progress never became durable: pruned through {last_pruned}"
    );
    assert!(
        largest.0 <= 2 && largest.1 <= 4,
        "witness state grew with history: largest {largest:?} over {CYCLES} cycles"
    );

    drop(handler);
    let restarted =
        ProductionParticipantHandler::new_restored(Arc::clone(&store), test_participant_config())?;
    assert!(restarted.unloadable_conversations().is_empty());
    let (witnesses, occurrences, pruned_through) = witness_state(&restarted)?;
    assert_eq!(pruned_through, last_pruned);
    assert!(witnesses <= 2 && occurrences <= 4);
    let delivered = admit(&restarted, sender_connection, sender, CYCLES)?;
    ack(&restarted, recipient_connection, recipient, delivered)?;
    Ok(())
}

#[test]
fn a_terminal_occurrence_at_or_below_the_pruned_progress_is_refused_by_name()
-> Result<(), Box<dyn Error>> {
    let mut state = ObserverProgressWitnessState::new();
    state.record(
        CONVERSATION,
        ObserverProgressProjection::for_test(CONVERSATION, 7),
        ObserverProgressSourceMetadata::attached(3, CONVERSATION, 1, 7),
    )?;
    state.prune_through(7)?;
    assert_eq!(state.held(), (0, 0));

    let again = state.record(
        CONVERSATION,
        ObserverProgressProjection::for_test(CONVERSATION, 7),
        ObserverProgressSourceMetadata::attached(9, CONVERSATION, 1, 7),
    );
    assert_eq!(
        again,
        Err(
            ObserverProgressConformanceError::OccurrenceAtOrBelowPrunedProgress {
                progress: 7,
                pruned_through: 7,
            }
        )
    );

    state.record(
        CONVERSATION,
        ObserverProgressProjection::for_test(CONVERSATION, 12),
        ObserverProgressSourceMetadata::attached(10, CONVERSATION, 1, 12),
    )?;
    assert_eq!(
        state.prune_through(5),
        Err(
            ObserverProgressConformanceError::DurableBelowPrunedProgress {
                durable: 5,
                pruned_through: 7,
            }
        )
    );
    Ok(())
}

#[test]
fn a_source_older_than_every_pruned_witness_is_still_refused_as_out_of_order()
-> Result<(), Box<dyn Error>> {
    let mut state = ObserverProgressWitnessState::new();
    state.record(
        CONVERSATION,
        ObserverProgressProjection::for_test(CONVERSATION, 4),
        ObserverProgressSourceMetadata::participant_ack(20, CONVERSATION, 1, 4),
    )?;
    state.prune_through(4)?;
    let out_of_order = state.record(
        CONVERSATION,
        ObserverProgressProjection::for_test(CONVERSATION, 3),
        ObserverProgressSourceMetadata::participant_ack(18, CONVERSATION, 2, 3),
    );
    assert_eq!(
        out_of_order,
        Err(ObserverProgressConformanceError::SourceOrder)
    );
    Ok(())
}

/// Participant whose fenced recoveries the recovered-fate pins record.
const RECOVERED: u64 = 7;

/// Source sequence and floor spacing between two fenced recoveries.
const RECOVERY_STRIDE: u64 = 10;

/// The binding epoch of fenced recovery `k`: a new connection and the next
/// capability generation each time.
fn recovery_epoch(k: u64) -> Result<BindingEpoch, Box<dyn Error>> {
    Ok(BindingEpoch::new(
        ConnectionIncarnation::new(0x0B5, k),
        Generation::new(k).ok_or("recovery generation is zero")?,
    ))
}

/// The recovered fate of fenced recovery `k`, presented by source `source`.
fn recovered_fate(source: u64, k: u64) -> Result<ObserverProgressSourceMetadata, Box<dyn Error>> {
    Ok(ObserverProgressSourceMetadata::recovered_binding_fate(
        source,
        CONVERSATION,
        RECOVERED,
        recovery_epoch(k)?,
        k * RECOVERY_STRIDE,
    ))
}

/// Records `recoveries` fenced recoveries of one participant, each at a
/// strictly later source, binding epoch and floor, and prunes through each
/// floor as soon as it is durable, as the live owner does.
fn after_recoveries(recoveries: u64) -> Result<ObserverProgressWitnessState, Box<dyn Error>> {
    let mut state = ObserverProgressWitnessState::new();
    for k in 1..=recoveries {
        let floor = k * RECOVERY_STRIDE;
        state.record(
            CONVERSATION,
            ObserverProgressProjection::for_test(CONVERSATION, floor),
            recovered_fate(floor, k)?,
        )?;
        state.prune_through(floor)?;
    }
    Ok(state)
}

/// Pin for R3b review finding 11. After 1 and after 100 fenced recoveries of
/// one participant, each pruned through as soon as it is durable, the live
/// state is the same (no witness, the latest epoch's floor) and so must be
/// the number of occurrence keys held.
///
/// `prunable` still excludes `RecoveredFate`, which kept one key per recovery
/// (100 against 1). A recovered-fate key is now forgotten when the
/// participant's next recovered fate, at a later binding epoch, is recorded:
/// one attach mints one fate token and its recovered fate consumes it, so no
/// route presents an earlier epoch's fate again, and one that did is refused
/// by name. The control below holds through it.
#[test]
fn recovered_fate_keys_stay_bounded_across_recoveries() -> Result<(), Box<dyn Error>> {
    let one = after_recoveries(1)?;
    let hundred = after_recoveries(100)?;
    let (one_witnesses, one_keys) = one.held();
    let (hundred_witnesses, hundred_keys) = hundred.held();
    assert_eq!(
        (hundred.pruned_through(), one.pruned_through()),
        (100 * RECOVERY_STRIDE, RECOVERY_STRIDE),
        "positive control failed: the recoveries were not recorded and pruned through"
    );
    assert_eq!(
        (hundred_witnesses, one_witnesses),
        (0, 0),
        "every recovered witness is durable and pruned, so none is held"
    );
    assert_eq!(
        hundred_keys, one_keys,
        "R3b review finding 11: occurrence keys {hundred_keys} after 100 fenced recoveries vs \
         {one_keys} after 1, expected equal; recovered-fate keys survive every prune"
    );
    Ok(())
}

/// Control for R3b review finding 11, with and without the bound: once
/// 100 recoveries have been pruned through, the current epoch's recovered
/// fate presented again is still refused as a duplicate, an older source is
/// still out of order, a terminal at the pruned progress is still refused by
/// name, and the next recovery is still accepted.
#[test]
fn a_recovered_fate_presented_again_after_pruning_is_still_refused() -> Result<(), Box<dyn Error>> {
    let mut state = after_recoveries(100)?;
    let floor = 100 * RECOVERY_STRIDE;
    let duplicate = state.record(
        CONVERSATION,
        ObserverProgressProjection::for_test(CONVERSATION, floor),
        recovered_fate(floor + 1, 100)?,
    );
    assert_eq!(
        duplicate,
        Err(ObserverProgressConformanceError::DuplicateOccurrenceProducer)
    );
    let next_floor = 101 * RECOVERY_STRIDE;
    let older = state.record(
        CONVERSATION,
        ObserverProgressProjection::for_test(CONVERSATION, next_floor),
        recovered_fate(floor - 1, 101)?,
    );
    assert_eq!(older, Err(ObserverProgressConformanceError::SourceOrder));
    let terminal = state.record(
        CONVERSATION,
        ObserverProgressProjection::for_test(CONVERSATION, floor),
        ObserverProgressSourceMetadata::attached(floor + 2, CONVERSATION, RECOVERED + 1, floor),
    );
    assert_eq!(
        terminal,
        Err(
            ObserverProgressConformanceError::OccurrenceAtOrBelowPrunedProgress {
                progress: floor,
                pruned_through: floor,
            }
        )
    );
    state.record(
        CONVERSATION,
        ObserverProgressProjection::for_test(CONVERSATION, next_floor),
        recovered_fate(next_floor, 101)?,
    )?;
    Ok(())
}

/// The recovered fate of participant `RECOVERED` at the binding epoch of
/// fenced recovery `k`, naming `floor` and presented by source `source`.
fn recovered_fate_at(
    source: u64,
    k: u64,
    floor: u64,
) -> Result<ObserverProgressSourceMetadata, Box<dyn Error>> {
    Ok(ObserverProgressSourceMetadata::recovered_binding_fate(
        source,
        CONVERSATION,
        RECOVERED,
        recovery_epoch(k)?,
        floor,
    ))
}

/// Pin for R3b review row 11. After 100 fenced recoveries
/// are pruned through, a recovered fate at an earlier binding epoch, from a
/// newer source and at a floor the lineage allows, is refused, and the
/// newer epoch's key is still held: its fate presented again is still a
/// duplicate.
#[test]
fn an_earlier_epoch_recovered_fate_at_a_newer_source_is_refused() -> Result<(), Box<dyn Error>> {
    let mut state = after_recoveries(100)?;
    let floor = 100 * RECOVERY_STRIDE;
    let earlier = state.record(
        CONVERSATION,
        ObserverProgressProjection::for_test(CONVERSATION, floor),
        recovered_fate_at(floor + 1, 99, floor)?,
    );
    assert_eq!(
        earlier,
        Err(
            ObserverProgressConformanceError::RecoveredFateEpochNotAbove {
                participant_id: RECOVERED,
                held: recovery_epoch(100)?,
                presented: recovery_epoch(99)?,
            }
        ),
        "R3b review row 11: a recovered fate at epoch 99 after epoch 100 was accepted"
    );
    let again = state.record(
        CONVERSATION,
        ObserverProgressProjection::for_test(CONVERSATION, floor),
        recovered_fate(floor + 2, 100)?,
    );
    assert_eq!(
        again,
        Err(ObserverProgressConformanceError::DuplicateOccurrenceProducer),
        "R3b review row 11: epoch 100's key is no longer held"
    );
    Ok(())
}

/// Pin for R3b review row 11. A second recovered fate at the
/// participant's held recovered epoch, naming another floor, is refused:
/// one attach is recovered once.
#[test]
fn a_second_recovered_fate_at_the_held_epoch_is_refused() -> Result<(), Box<dyn Error>> {
    let mut state = after_recoveries(100)?;
    let next_floor = 101 * RECOVERY_STRIDE;
    let second = state.record(
        CONVERSATION,
        ObserverProgressProjection::for_test(CONVERSATION, next_floor),
        recovered_fate_at(next_floor, 100, next_floor)?,
    );
    assert_eq!(
        second,
        Err(
            ObserverProgressConformanceError::RecoveredFateEpochNotAbove {
                participant_id: RECOVERED,
                held: recovery_epoch(100)?,
                presented: recovery_epoch(100)?,
            }
        ),
        "R3b review row 11: a second recovered fate at epoch 100 was accepted"
    );
    Ok(())
}

/// Pin for R3b review row 11: "later" is the capability
/// generation; a binding epoch has no order. A participant may rebind on a
/// connection opened before the one it last bound on, so its next recovered
/// fate can name an earlier connection at the next generation. That fate is
/// accepted and forgets the held one: the keys held equal those after one
/// recovery.
#[test]
fn a_next_generation_recovered_fate_on_an_earlier_connection_supersedes()
-> Result<(), Box<dyn Error>> {
    let mut state = after_recoveries(100)?;
    let next_floor = 101 * RECOVERY_STRIDE;
    let earlier_connection = BindingEpoch::new(
        ConnectionIncarnation::new(0x0B5, 1),
        Generation::new(101).ok_or("recovery generation is zero")?,
    );
    state.record(
        CONVERSATION,
        ObserverProgressProjection::for_test(CONVERSATION, next_floor),
        ObserverProgressSourceMetadata::recovered_binding_fate(
            next_floor,
            CONVERSATION,
            RECOVERED,
            earlier_connection,
            next_floor,
        ),
    )?;
    state.prune_through(next_floor)?;
    let (_, one_keys) = after_recoveries(1)?.held();
    let (witnesses, keys) = state.held();
    assert_eq!(
        (witnesses, keys),
        (0, one_keys),
        "R3b review row 11: the generation-101 fate did not forget generation 100's key"
    );
    Ok(())
}

/// Pin for R3b review row 11: a second recovered fate at the
/// held generation is refused whichever connection it names, here one
/// opened before the held binding's.
#[test]
fn a_same_generation_recovered_fate_on_another_connection_is_refused() -> Result<(), Box<dyn Error>>
{
    let mut state = after_recoveries(100)?;
    let next_floor = 101 * RECOVERY_STRIDE;
    let other_connection = BindingEpoch::new(
        ConnectionIncarnation::new(0x0B5, 1),
        Generation::new(100).ok_or("recovery generation is zero")?,
    );
    let second = state.record(
        CONVERSATION,
        ObserverProgressProjection::for_test(CONVERSATION, next_floor),
        ObserverProgressSourceMetadata::recovered_binding_fate(
            next_floor,
            CONVERSATION,
            RECOVERED,
            other_connection,
            next_floor,
        ),
    );
    assert_eq!(
        second,
        Err(
            ObserverProgressConformanceError::RecoveredFateEpochNotAbove {
                participant_id: RECOVERED,
                held: recovery_epoch(100)?,
                presented: other_connection,
            }
        ),
        "R3b review row 11: a second recovered fate at generation 100 on connection 1 was \
         accepted"
    );
    Ok(())
}

#[path = "tests_observer_witness_pruning/checkpoint.rs"]
mod checkpoint;