liminal-rs 0.14.0

use std::sync::Arc;

use super::in_flight::IN_FLIGHT_CHECKPOINT_EVERY;
use super::{ClaimEvidence, DedupCache};
use crate::durability::DurableStore;
use crate::durability::bridge::block_on;
use crate::durability::store::EngineReadAccountingGuard;
use crate::durability::tests_fixtures;

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

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

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

/// Claims settled after the key held in flight: enough rows that a start
/// reading every one of them is unmistakable beside the index bound.
fn settled_claims() -> Result<usize, Box<dyn std::error::Error>> {
    Ok(usize::try_from(IN_FLIGHT_CHECKPOINT_EVERY)? * 5)
}

#[test]
fn ceiling_a_start_reads_the_index_window_not_every_claim_ever_made() -> TestResult {
    let settled = settled_claims()?;
    assert_eq!(tests_fixtures::settled_claims()?, settled);
    let store: Arc<dyn DurableStore> = Arc::new(tests_fixtures::copied("ceiling-held")?);

    let restarted = DedupCache::new(Arc::clone(&store), NAMESPACE);
    let accounting = EngineReadAccountingGuard::start();
    let claims = block_on(restarted.in_flight_claims())??;
    let start = accounting.snapshot();
    drop(accounting);

    assert_eq!(by_key(&claims), [("held".to_owned(), Some(evidence()))]);
    assert!(!start.counter_overflow_observed, "{start:?}");
    assert_eq!(
        start.streams_enumerated, 0,
        "a start never walks the namespace"
    );
    // At most two windows of index rows, each naming a key whose stream holds
    // a claim and its settlement.
    let bound = usize::try_from(IN_FLIGHT_CHECKPOINT_EVERY * 2)? * 3 + 8;
    assert!(
        start.engine_entries <= bound,
        "a start after {settled} settled claims read {} entries (bound {bound})",
        start.engine_entries
    );
    Ok(())
}

#[test]
fn ceiling_a_claim_made_after_its_key_was_released_stays_in_the_index() -> TestResult {
    assert_eq!(tests_fixtures::settled_claims()?, settled_claims()?);
    let store: Arc<dyn DurableStore> = Arc::new(tests_fixtures::copied("ceiling-reclaimed")?);

    let restarted = DedupCache::new(store, NAMESPACE);
    assert_eq!(
        by_key(&block_on(restarted.in_flight_claims())??),
        [("k".to_owned(), Some(evidence()))]
    );
    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()
}