liminal-server 0.24.5

//! DUR-028 R6: the store census counts the shared key's damage and writes
//! nothing.

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

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

use super::super::prune::{HeldRecords, PruneReceipt, PrunedStream, receipt_key};
use super::super::registry::ConversationRegistry;
use super::super::restore_checkpoint_locator::{
    CheckpointBodyKey, CheckpointLocator, checkpoint_conversation_route,
};
use super::super::tests_memory_store::MemoryStore;
use super::store_census;

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

fn put(store: &dyn DurableStore, route: &str, key: Vec<u8>, value: Vec<u8>) -> TestResult {
    block_on(store.apply_routed_batch(
        route,
        vec![RoutedBatchItem {
            key,
            expected: None,
            value,
            ttl: None,
        }],
    ))??;
    Ok(())
}

/// Three registered conversations on one shard, as a store written before
/// DUR-028 leaves them: 10 has its own locator under its namespace; the
/// shard's bare locator names 10, so 11 and 12 restore against a locator
/// that is not theirs; 12's operation log was pruned at row 5.
fn shared_key_store() -> Result<Arc<dyn DurableStore>, Box<dyn Error>> {
    let counted = Arc::new(CountingStore::new(MemoryStore::default()));
    let store: Arc<dyn DurableStore> = counted;
    let registry = ConversationRegistry::new(Arc::clone(&store));
    registry.restore_sets()?;
    for id in [10, 11, 12] {
        registry.register(id)?;
    }
    let locator = CheckpointLocator::new(10, 1, 0, [7; 32], CheckpointBodyKey::for_body(b"body"))?
        .encode_canonical()
        .to_vec();
    put(
        store.as_ref(),
        &checkpoint_conversation_route(10),
        CheckpointLocator::routed_key(10),
        locator.clone(),
    )?;
    // The one bare key a shard held: haematite answers it on every route of
    // the shard; the memory store keys by route, so it is written on each.
    for id in [10, 11, 12] {
        put(
            store.as_ref(),
            &checkpoint_conversation_route(id),
            CheckpointLocator::legacy_routed_key().to_vec(),
            locator.clone(),
        )?;
    }
    let receipt = PruneReceipt {
        conversation_id: 12,
        stream: PrunedStream::Operations,
        cut_before: 5,
        covered_through: 5,
        checkpoint_position: 5,
        kept: 0,
        recipients: Vec::new(),
        held: HeldRecords::default(),
        at_ms: 0,
    };
    put(
        store.as_ref(),
        &checkpoint_conversation_route(12),
        receipt_key(12, PrunedStream::Operations, 5),
        serde_json::to_vec(&receipt)?,
    )?;
    Ok(store)
}

#[test]
fn the_census_names_who_the_shared_locator_hurt_and_writes_nothing() -> TestResult {
    let store = shared_key_store()?;
    let census = store_census(&store)?;
    assert_eq!((census.live, census.sweeping, census.shards), (3, 0, 1));
    assert_eq!(census.per_shard, BTreeMap::from([(0, 3)]));
    assert_eq!(census.legacy_locators, BTreeMap::from([(0, Ok(10))]));
    assert_eq!(census.names_another, vec![(11, 10), (12, 10)]);
    assert_eq!(census.pruned_below_rebuild, vec![12]);
    let counts = census.counts.as_ref().ok_or("the counting store counts")?;
    assert_eq!(
        counts.apply_routed_batch_calls + counts.append_calls + counts.append_all_calls,
        0,
        "the census wrote: {counts:?}"
    );
    assert!(counts.get_routed_calls > 0, "{counts:?}");
    let lines = census.lines();
    assert_eq!(
        lines[..3],
        [
            "store: 3 conversations (3 live, 0 sweeping) on 1 shards".to_owned(),
            "shard 0: 3 conversations".to_owned(),
            "legacy locator on shard 0: names conversation 10".to_owned(),
        ]
    );
    assert_eq!(
        lines[3],
        "conversations whose locator names another: 2 (11 names 10, 12 names 10)"
    );
    assert_eq!(lines[4], "of those, logs pruned below a rebuild: 1 (12)");
    assert!(lines[5].ends_with("writes=0"), "{}", lines[5]);
    Ok(())
}

#[test]
fn a_bare_locator_that_does_not_read_is_reported_not_decoded() -> TestResult {
    let store: Arc<dyn DurableStore> = Arc::new(MemoryStore::default());
    let registry = ConversationRegistry::new(Arc::clone(&store));
    registry.restore_sets()?;
    registry.register(3)?;
    put(
        store.as_ref(),
        &checkpoint_conversation_route(3),
        CheckpointLocator::legacy_routed_key().to_vec(),
        b"not a locator".to_vec(),
    )?;
    let census = store_census(&store)?;
    assert_eq!(census.names_another, Vec::new());
    let Some(Err(reason)) = census.legacy_locators.get(&0) else {
        return Err("the unreadable locator was not reported".into());
    };
    assert!(
        reason.contains("length") || reason.contains("128"),
        "{reason}"
    );
    assert!(
        census.counts.is_none(),
        "a plain memory store counts nothing"
    );
    Ok(())
}