liminal-rs 0.14.0

//! The in-flight index against the real haematite-backed store: a start reads
//! the index window and the keys it names, never the namespace, and the one
//! walk that builds the index of an older store is not made again.
//!
//! These pins are counts, never durations, taken from the engine-read
//! accounting the store keeps.

use std::sync::Arc;

use super::{ClaimEvidence, DedupCache};
use crate::durability::bridge::block_on;
use crate::durability::store::{EngineReadAccounting, EngineReadAccountingGuard};
use crate::durability::{DurableStore, ProcessingReceipt, open_ephemeral};

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

const NAMESPACE: &str = "liminal:in-flight-test";

fn evidence() -> ClaimEvidence {
    ClaimEvidence::new("orders", vec![0])
}

#[test]
fn a_store_without_an_index_is_walked_once_and_then_never_again() -> TestResult {
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    // What liminal-server 0.16.0 leaves: claims without evidence, some
    // completed, one in flight, and no index.
    let legacy = DedupCache::new(Arc::clone(&store), NAMESPACE);
    for index in 0..16 {
        let key = format!("done-{index}");
        block_on(legacy.claim_or_get(&key, 1))??;
        block_on(legacy.complete_receipt(&key, ProcessingReceipt::new(Vec::new())))??;
    }
    block_on(legacy.claim_or_get("stranded", 1))??;

    let first = DedupCache::new(Arc::clone(&store), NAMESPACE);
    let accounting = EngineReadAccountingGuard::start();
    let found = block_on(first.in_flight_claims())??;
    let migration = accounting.snapshot();
    drop(accounting);
    assert_eq!(
        found,
        [super::InFlightClaim {
            idempotency_key: "stranded".to_owned(),
            claimed_at_millis: 1,
            evidence: None,
        }]
    );
    assert!(
        migration.streams_enumerated > 0,
        "the first start walks once"
    );

    let second = DedupCache::new(Arc::clone(&store), NAMESPACE);
    let accounting = EngineReadAccountingGuard::start();
    let found = block_on(second.in_flight_claims())??;
    let restart = accounting.snapshot();
    drop(accounting);
    assert_eq!(
        found,
        [super::InFlightClaim {
            idempotency_key: "stranded".to_owned(),
            claimed_at_millis: 1,
            evidence: None,
        }]
    );
    assert_eq!(
        restart.streams_enumerated, 0,
        "the walk is recorded and not repeated"
    );
    Ok(())
}

#[test]
fn a_namespace_begun_empty_is_read_by_no_start_and_walked_by_none() -> TestResult {
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let created = DedupCache::new_empty(Arc::clone(&store), NAMESPACE);

    let accounting = EngineReadAccountingGuard::start();
    let found = block_on(created.in_flight_claims())??;
    let start = accounting.snapshot();
    drop(accounting);
    assert!(found.is_empty());
    assert_eq!(
        start,
        EngineReadAccounting::default(),
        "a namespace begun empty is not read at start"
    );
    assert_eq!(
        block_on(store.read_value(&format!("{NAMESPACE}/in-flight-start")))??,
        None,
        "nothing is recorded until a claim needs the index"
    );

    // The first join records the start, so the next start finds an index and
    // never mistakes the store for one written before the index existed.
    block_on(created.claim_with_evidence("held", evidence(), 1))??;
    let restarted = DedupCache::new(Arc::clone(&store), NAMESPACE);
    let accounting = EngineReadAccountingGuard::start();
    let found = block_on(restarted.in_flight_claims())??;
    let restart = accounting.snapshot();
    drop(accounting);
    assert_eq!(by_key(&found), [("held".to_owned(), Some(evidence()))]);
    assert_eq!(
        restart.streams_enumerated, 0,
        "a start after the first join reads the index, never the namespace"
    );
    Ok(())
}

/// The keys and evidence of `claims`, the shape a pin compares when the
/// claim time is not its subject.
fn by_key(claims: &[super::InFlightClaim]) -> Vec<(String, Option<super::ClaimEvidence>)> {
    claims
        .iter()
        .map(|claim| (claim.idempotency_key.clone(), claim.evidence.clone()))
        .collect()
}