liminal-server 0.16.0

//! 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::{
    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(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(())
}