liminal-server 0.24.1

//! DUR-028 R4: a store written before the namespace is migrated once at
//! boot, resumably, and a boot that finds the marker reads only it.
//!
//! The legacy store is made as the old code made it: a conversation is
//! driven through the production handler past a checkpoint, then every key
//! under its namespace is written bare on its route and removed from the
//! namespace, and the marker its first registration wrote is removed, so
//! the store holds exactly the bare keys a pre-brief liminal left.

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

use liminal::durability::bridge::block_on;
use liminal::durability::{DurableStore, RoutedBatchItem, open_ephemeral};

use super::ProductionParticipantHandler;
use super::registry::REGISTRY_STREAM_KEY;
use super::restore_checkpoint_locator::checkpoint_conversation_route;
use super::route_key_migration::{MARKER_KEY, legacy_windows, migrate_route_keys, past};
use super::route_keys;
use super::tests::test_participant_config;
use super::tests_checkpoint_route_keys::{admit, enrolled, own_locator};
use super::tests_store_tally::{Tally, ceiling, disk_store, under};

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

const TABLE: &str = "route_key_migration";

/// Records past one checkpoint period.
const CHECKPOINTED: u64 = 260;

/// Writes `id`'s namespaced checkpoint keys bare on its route and removes
/// them from the namespace: the shape the code before DUR-028 R2 left.
fn make_legacy(store: &dyn DurableStore, id: u64) -> TestResult {
    let route = checkpoint_conversation_route(id);
    let (from, to) = route_keys::conversation_window(id);
    let rows = block_on(store.scan_routed(&route, &from, &to))??;
    if rows.is_empty() {
        return Err(format!("conversation {id} has no namespaced keys to make legacy").into());
    }
    let prefix = route_keys::conversation_namespace(id).len();
    let items = rows
        .into_iter()
        .map(|(key, value)| RoutedBatchItem {
            key: key[prefix..].to_vec(),
            expected: None,
            value,
            ttl: None,
        })
        .collect();
    block_on(store.apply_routed_batch(&route, items))??;
    block_on(store.remove_routed_range(&route, &from, &to))??;
    Ok(())
}

/// Removes the marker the first registration wrote, so the store reads as
/// one written before the namespace and never booted since.
fn remove_marker(store: &dyn DurableStore) -> TestResult {
    let (from, to) = past(MARKER_KEY);
    block_on(store.remove_routed_range(REGISTRY_STREAM_KEY, &from, &to))??;
    Ok(())
}

/// Every bare key of the four legacy families on `id`'s route.
fn legacy_keys(store: &dyn DurableStore, id: u64) -> TestResult<Vec<Vec<u8>>> {
    let route = checkpoint_conversation_route(id);
    let mut keys = Vec::new();
    for (from, to) in legacy_windows() {
        keys.extend(
            block_on(store.scan_routed(&route, &from, &to))??
                .into_iter()
                .map(|(key, _)| key),
        );
    }
    Ok(keys)
}

fn marker(store: &dyn DurableStore) -> TestResult<Option<String>> {
    Ok(
        block_on(store.get_routed(REGISTRY_STREAM_KEY, MARKER_KEY))??
            .map(String::from_utf8)
            .transpose()?,
    )
}

/// Drives `ids` through a handler on `store`: the first past a checkpoint,
/// the rest a few records each.
fn written(store: &Arc<Tally>, ids: &[u64]) -> TestResult {
    let handler = ProductionParticipantHandler::new(store.shared(), test_participant_config())?;
    for (index, &id) in ids.iter().enumerate() {
        let member = enrolled(&handler, id)?;
        admit(
            &handler,
            id,
            member,
            if index == 0 { CHECKPOINTED } else { 4 },
        )?;
    }
    Ok(())
}

/// The marker-present boot: one read and nothing else.
fn marker_boot_reads_only_the_marker(store: &Arc<Tally>, ids: &[u64]) -> TestResult {
    store.take()?;
    assert_eq!(migrate_route_keys(&store.shared(), ids)?, None);
    let asked = store.take()?;
    assert_eq!(asked.get_routed_calls, 1, "{asked:?}");
    assert_eq!(
        (
            asked.commits,
            asked.scan_routed_calls,
            asked.remove_routed_range_calls,
            asked.read_from_calls,
            asked.read_value_calls,
        ),
        (0, 0, 0, 0, 0),
        "{asked:?}"
    );
    Ok(())
}

/// Red first: two conversations on one shard, one owning the legacy
/// locator; after the boot the owner restores from its migrated
/// checkpoint, the other rebuilds from its log, and no legacy key remains.
#[test]
fn a_boot_brings_the_legacy_owner_under_its_namespace_and_the_other_restores_from_its_log()
-> TestResult {
    let tally = Tally::new(Arc::new(open_ephemeral(1)?));
    written(&tally, &[10, 11])?;
    make_legacy(tally.inner.as_ref(), 10)?;
    remove_marker(tally.inner.as_ref())?;
    assert!(own_locator(tally.inner.as_ref(), 10).is_err());
    assert!(!legacy_keys(tally.inner.as_ref(), 10)?.is_empty());

    let handler =
        ProductionParticipantHandler::new_restored(tally.shared(), test_participant_config())?;
    let mut restored = handler.restored_conversation_ids()?;
    restored.sort_unstable();
    assert_eq!(restored, vec![10, 11]);
    assert!(own_locator(tally.inner.as_ref(), 10)?.covered_sequence() >= CHECKPOINTED - 8);
    // The other conversation rebuilt from its log at the restore and
    // published a checkpoint of its own there: one that covers its four
    // records, never the owner's.
    assert!(
        own_locator(tally.inner.as_ref(), 11)?.covered_sequence() < CHECKPOINTED - 8,
        "the conversation the legacy locator did not name holds the owner's checkpoint"
    );
    for id in [10, 11] {
        assert_eq!(
            legacy_keys(tally.inner.as_ref(), id)?,
            Vec::<Vec<u8>>::new()
        );
    }
    let line = marker(tally.inner.as_ref())?.ok_or("the boot wrote no marker")?;
    assert!(
        line.contains("1 migrated, 0 skipped, 1 without a checkpoint, 0 refused"),
        "{line}"
    );
    drop(handler);
    marker_boot_reads_only_the_marker(&tally, &[10, 11])
}

/// A crash after the first conversation's batch (the store refuses the
/// second batch once) leaves the first migrated and the second untouched;
/// the next boot finishes it and writes the marker; a third reads only the
/// marker. Two shards, so each holds a legacy locator of its own.
#[test]
fn a_crash_after_the_first_batch_is_finished_by_the_next_boot() -> TestResult {
    let home = tempfile::tempdir()?;
    let tally = disk_store(home.path(), 2)?;
    let spine = tally
        .inner
        .reclaim_spine()
        .ok_or("the disk store has no spine")?;
    let shard_of = |id: u64| {
        spine
            .database()
            .shard_for(checkpoint_conversation_route(id).as_bytes())
    };
    let first = 10;
    let second = (11..64)
        .find(|&id| shard_of(id) != shard_of(first))
        .ok_or("no id routes to the other shard")?;
    let ids = [first, second];
    {
        let handler = ProductionParticipantHandler::new(tally.shared(), test_participant_config())?;
        for id in ids {
            let member = enrolled(&handler, id)?;
            admit(&handler, id, member, CHECKPOINTED)?;
        }
    }
    for id in ids {
        make_legacy(tally.inner.as_ref(), id)?;
    }
    remove_marker(tally.inner.as_ref())?;

    tally.take()?;
    tally.refuse_commit(2)?;
    let refused = migrate_route_keys(&tally.shared(), &ids);
    assert!(
        refused.is_err(),
        "the second batch was not refused: {refused:?}"
    );
    assert!(own_locator(tally.inner.as_ref(), first).is_ok());
    assert!(own_locator(tally.inner.as_ref(), second).is_err());
    assert_eq!(marker(tally.inner.as_ref())?, None);

    let finished = migrate_route_keys(&tally.shared(), &ids)?
        .ok_or("the marker stood before the migration finished")?;
    assert_eq!(finished.skipped, 1, "{finished:?}");
    assert_eq!(finished.migrated, vec![second], "{finished:?}");
    for id in ids {
        assert!(own_locator(tally.inner.as_ref(), id)?.covered_sequence() >= CHECKPOINTED - 8);
        assert_eq!(
            legacy_keys(tally.inner.as_ref(), id)?,
            Vec::<Vec<u8>>::new()
        );
    }
    assert!(marker(tally.inner.as_ref())?.is_some());
    marker_boot_reads_only_the_marker(&tally, &ids)
}

/// The migration's store counts over N conversations are a function of N,
/// held under `[route_key_migration]`. On a disk store, whose spine names
/// each route's shard, the legacy windows are cut once per shard reached;
/// a store without a spine cuts them once per conversation instead.
#[test]
fn the_migrations_store_counts_are_a_function_of_the_conversations() -> TestResult {
    const N: u64 = 8;
    let ids: Vec<u64> = (10..10 + N).collect();
    let home = tempfile::tempdir()?;
    let tally = disk_store(home.path(), 1)?;
    written(&tally, &ids)?;
    make_legacy(tally.inner.as_ref(), ids[0])?;
    remove_marker(tally.inner.as_ref())?;

    tally.take()?;
    let report = migrate_route_keys(&tally.shared(), &ids)?
        .ok_or("the marker stood before the migration ran")?;
    let asked = tally.take()?;
    assert_eq!(report.migrated, vec![ids[0]], "{report:?}");
    assert_eq!(report.without_checkpoint, ids[1..].to_vec(), "{report:?}");
    assert_eq!(report.skipped, 0, "{report:?}");
    assert_eq!(report.refused, Vec::new(), "{report:?}");
    let bound = ceiling(TABLE, "get_routed_calls_per_conversation")? * N
        + ceiling(TABLE, "get_routed_calls_fixed")?;
    assert!(
        asked.get_routed_calls <= bound,
        "get_routed over {N} conversations: ceiling {bound}, measured {}",
        asked.get_routed_calls
    );
    println!(
        "route key migration over {N} conversations: get_routed {} (ceiling {bound}); asked {asked:?}",
        asked.get_routed_calls
    );
    under(TABLE, "scan_routed_calls", asked.scan_routed_calls)?;
    under(TABLE, "commits", asked.commits)?;
    under(
        TABLE,
        "remove_routed_range_calls",
        asked.remove_routed_range_calls,
    )?;
    assert_eq!(
        asked.read_from_calls, 0,
        "the migration reads no stream: {asked:?}"
    );
    Ok(())
}