liminal-server 0.24.9

//! A conversation's checkpoint keys are its own on a shard it shares.
//!
//! Haematite routes a routed key to a shard by its route and stores the key
//! as it is given (haematite 0.21.0 `api/kv.rs` `delete_routed_range`,
//! `range_routed`): the route names the shard, never a namespace. Two
//! conversations on one shard therefore share every bare key either writes,
//! and a window cut over one conversation's route cuts the other's keys.
//! These units run on a one-shard haematite store, the shape every new
//! liminal store has.

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

use liminal::durability::bridge::block_on;
use liminal::durability::{DurableStore, open_ephemeral};
use liminal_protocol::wire::{
    ClientRequest, ConnectionIncarnation, EnrollmentRequest, EnrollmentToken, Generation,
    RecordAdmission, RecordAdmissionAttemptToken, ServerValue,
};

use super::ProductionParticipantHandler;
use super::operation_batch::CHECKPOINT_EVERY;
use super::restore_checkpoint_locator::{CheckpointLocator, checkpoint_conversation_route};
use super::tests::{dispatch, test_participant_config};
use super::tests_store_tally::{Asked, Tally, under};

type TestResult = Result<(), Box<dyn Error>>;

/// Enrolls a sender and a recipient on `conversation_id`; answers the
/// sender's connection and participant id.
pub(super) fn enrolled(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
) -> Result<(ConnectionIncarnation, u64), Box<dyn Error>> {
    let sender = ConnectionIncarnation::new(conversation_id, 1);
    let mut tokens = [[0_u8; 16]; 2];
    tokens[0][..8].copy_from_slice(&conversation_id.to_be_bytes());
    tokens[1] = tokens[0];
    tokens[1][15] = 1;
    let bound = dispatch(
        handler,
        sender,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id,
            enrollment_token: EnrollmentToken::new(tokens[0]),
        }),
    )?;
    let ServerValue::EnrollBound(bound) = bound else {
        return Err(format!("enrollment did not bind: {bound:?}").into());
    };
    let recipient = dispatch(
        handler,
        ConnectionIncarnation::new(conversation_id, 2),
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id,
            enrollment_token: EnrollmentToken::new(tokens[1]),
        }),
    )?;
    if !matches!(recipient, ServerValue::EnrollBound(_)) {
        return Err(format!("the recipient did not bind: {recipient:?}").into());
    }
    Ok((sender, bound.participant_id()))
}

/// Admits `count` records on `conversation_id`, each answered committed.
pub(super) fn admit(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
    member: (ConnectionIncarnation, u64),
    count: u64,
) -> TestResult {
    admit_range(handler, conversation_id, member, 0..count)
}

/// Admits the records numbered `operations` on `conversation_id`, each
/// answered committed.
fn admit_range(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
    (connection, participant_id): (ConnectionIncarnation, u64),
    operations: std::ops::Range<u64>,
) -> TestResult {
    for operation in operations {
        let mut token = [0; 16];
        token[..8].copy_from_slice(&conversation_id.to_be_bytes());
        token[8..].copy_from_slice(&operation.to_be_bytes());
        let answer = dispatch(
            handler,
            connection,
            ClientRequest::RecordAdmission(RecordAdmission {
                conversation_id,
                participant_id,
                capability_generation: Generation::ONE,
                record_admission_attempt_token: RecordAdmissionAttemptToken::new(token),
                payload: operation.to_be_bytes().to_vec(),
            }),
        )?;
        if !matches!(answer, ServerValue::RecordCommitted(_)) {
            return Err(format!(
                "record {operation} on conversation {conversation_id} answered {answer:?}"
            )
            .into());
        }
    }
    Ok(())
}

/// The locator a conversation's own route answers, decoded for it.
pub(super) fn own_locator(
    store: &dyn DurableStore,
    conversation_id: u64,
) -> Result<CheckpointLocator, Box<dyn Error>> {
    let bytes = block_on(store.get_routed(
        &checkpoint_conversation_route(conversation_id),
        &CheckpointLocator::routed_key(conversation_id),
    ))??
    .ok_or_else(|| format!("conversation {conversation_id} has no checkpoint locator"))?;
    Ok(CheckpointLocator::decode_canonical(
        &bytes,
        conversation_id,
    )?)
}

/// Two conversations on one shard each cross their first checkpoint period:
/// each commit is answered committed, and each route answers its own
/// conversation's locator.
#[test]
fn two_conversations_on_one_shard_keep_their_own_checkpoints() -> TestResult {
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let handler = ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;
    let first = enrolled(&handler, 10)?;
    let second = enrolled(&handler, 11)?;
    admit(&handler, 10, first, 260)?;
    admit(&handler, 11, second, 260)?;
    assert!(own_locator(store.as_ref(), 10)?.covered_sequence() >= 255);
    assert!(own_locator(store.as_ref(), 11)?.covered_sequence() >= 255);
    Ok(())
}

/// A conversation's windows cut its own keys from a shared shard and no
/// other conversation's (DUR-028 R5).
#[test]
fn cutting_one_conversations_windows_leaves_the_other_conversations_keys() -> TestResult {
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let rows = |id: u64| {
        vec![
            CheckpointLocator::routed_key(id),
            super::route_keys::namespaced(id, b"checkpoint/body/v1/x"),
        ]
    };
    for id in [10, 11] {
        let items = rows(id)
            .into_iter()
            .map(|key| liminal::durability::RoutedBatchItem {
                key,
                expected: None,
                value: vec![1],
                ttl: None,
            })
            .collect();
        block_on(store.apply_routed_batch(&checkpoint_conversation_route(id), items))??;
    }
    let route = checkpoint_conversation_route(10);
    for (from, to) in super::route_keys::conversation_windows(10, &[])? {
        block_on(store.remove_routed_range(&route, &from, &to))??;
    }
    let left: Vec<Vec<u8>> = block_on(store.scan_routed(&route, &[], &[0xff]))??
        .into_iter()
        .map(|(key, _)| key)
        .collect();
    let mut kept = rows(11);
    kept.sort();
    assert_eq!(left, kept);
    Ok(())
}

/// DUR-028 R3: a conversation crossing three checkpoint periods holds one
/// body key and the parts its manifest lists under its namespace, and the
/// commit that crosses is one routed batch: no window is scanned, and the
/// only cuts are the prune's (DUR-018), none for the superseded body.
#[test]
fn three_checkpoint_periods_leave_one_body_and_its_parts_under_the_namespace() -> TestResult {
    const TABLE: &str = "checkpoint_crossing";
    let tally = Tally::new(Arc::new(open_ephemeral(1)?));
    let handler = ProductionParticipantHandler::new(tally.shared(), test_participant_config())?;
    let member = enrolled(&handler, 10)?;
    // One record at a time: the commit that crosses is the one after which
    // the locator's covered sequence moved. Every crossing is kept, so the
    // third's cost is read beside the second's (rule 11: never the history).
    let mut covered = 0;
    let mut crossings: Vec<Asked> = Vec::new();
    let mut ordinary = Asked::default();
    let started = Instant::now();
    tally.take()?;
    for operation in 0..(3 * CHECKPOINT_EVERY + 64) {
        admit_range(&handler, 10, member, operation..operation + 1)?;
        let asked = tally.take()?;
        let now =
            own_locator(tally.inner.as_ref(), 10).map_or(0, CheckpointLocator::covered_sequence);
        if now > covered {
            covered = now;
            crossings.push(asked);
            if crossings.len() == 3 {
                break;
            }
        } else {
            ordinary.get_routed_calls = ordinary.get_routed_calls.max(asked.get_routed_calls);
            ordinary.commits = ordinary.commits.max(asked.commits);
        }
    }
    println!(
        "three periods in {} ms; an ordinary record asks at most {} routed reads and {} commits; crossings {crossings:?}",
        started.elapsed().as_millis(),
        ordinary.get_routed_calls,
        ordinary.commits
    );
    let [first, second, third] = crossings.as_slice() else {
        return Err(format!("{} crossings, not three", crossings.len()).into());
    };
    for crossing in [first, second, third] {
        assert_eq!(
            crossing.apply_routed_batch_calls, 1,
            "one routed batch per crossing: {crossing:?}"
        );
        assert_eq!(crossing.scan_routed_calls, 0, "{crossing:?}");
    }
    println!(
        "crossing routed reads: first {} second {} third {} (a period of identities, each walking an index path whose depth grows with the log of the conversation's identities)",
        first.get_routed_calls, second.get_routed_calls, third.get_routed_calls
    );
    under(TABLE, "get_routed_calls", third.get_routed_calls)?;
    under(
        TABLE,
        "remove_routed_range_calls",
        third.remove_routed_range_calls,
    )?;
    let route = checkpoint_conversation_route(10);
    let (from, to) = super::route_keys::conversation_window(10);
    let keys: Vec<Vec<u8>> = block_on(tally.inner.scan_routed(&route, &from, &to))??
        .into_iter()
        .map(|(key, _)| key)
        .collect();
    let bodies = keys
        .iter()
        .filter(|key| key.starts_with(&super::route_keys::namespaced(10, b"checkpoint/body/v1/")))
        .count();
    let parts = keys
        .iter()
        .filter(|key| {
            key.starts_with(&super::route_keys::namespaced(
                10,
                b"checkpoint/body/part/v1/",
            ))
        })
        .count();
    let locator = own_locator(tally.inner.as_ref(), 10)?;
    let body = super::checkpoint_body_parts::read_body_for_tests(
        tally.inner.as_ref(),
        10,
        locator.body_key(),
    )?
    .ok_or("the locator names no body")?;
    assert_eq!(bodies, 1, "bodies under the namespace after three periods");
    assert_eq!(
        parts,
        super::checkpoint_body_parts::lay_out(10, body).parts.len(),
        "parts under the namespace are the manifest's"
    );
    Ok(())
}