liminal-rs 0.11.0

//! The in-flight index's two ceiling pins over a long settled history.
//!
//! Each history is 1,280 settled claims, five index windows. It is seeded
//! through the real [`DedupCache`] onto a memory store, so the rows and the
//! index checkpoints every [`IN_FLIGHT_CHECKPOINT_EVERY`] are the cache's own,
//! each stream's rows in the cache's order. It is then landed on the haematite
//! store under test with one `append_all` for each stream, in no set order
//! across streams, then one `cas` for each value and one flush. A restart reads
//! no order across streams, since the index's start value lands last. Seeded
//! straight onto haematite, the same history costs one flushed write for each
//! operation: 3,847 writes in 232.6 s, against 1,284 operations in 27.2 s
//! landed and 13.7 ms through the cache in memory.
//! The restart and every assertion run on haematite.
//!
//! They run in the default suite, as liminal-server's `store_counts` ceilings
//! do, and each stays over 2 s (33.7 s and 23.5 s when measured). That is a
//! named defect on the work-list row for haematite's cross-stream batch:
//! haematite has no write that spans streams, so it ends only when one lands.

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

use super::in_flight::IN_FLIGHT_CHECKPOINT_EVERY;
use super::{ClaimEvidence, DedupCache, DedupDecision};
use crate::durability::bridge::block_on;
use crate::durability::store::EngineReadAccountingGuard;
use crate::durability::tests_memory_store::MemoryStore;
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])
}

/// 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)
}

/// Lands `seeded` on a fresh haematite store and answers it, printing the
/// seed's and the land's stage times beside the test's name.
fn land(
    test: &str,
    seeded: &MemoryStore,
    seed_started: Instant,
) -> Result<Arc<dyn DurableStore>, Box<dyn std::error::Error>> {
    let seed_ms = seed_started.elapsed().as_secs_f64() * 1000.0;
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let landing = Instant::now();
    block_on(seeded.land_on(store.as_ref()))??;
    println!(
        "{test}: seed through the cache in memory {seed_ms:.1} ms, land on haematite {:.1} ms",
        landing.elapsed().as_secs_f64() * 1000.0
    );
    Ok(store)
}

#[test]
fn ceiling_a_start_reads_the_index_window_not_every_claim_ever_made() -> TestResult {
    let seed_started = Instant::now();
    let seeded = Arc::new(MemoryStore::default());
    let running = DedupCache::new(Arc::clone(&seeded) as Arc<dyn DurableStore>, NAMESPACE);
    // One claim stays in flight the whole time, so the index must carry it.
    block_on(running.claim_with_evidence("held", evidence(), 1))??;
    let settled = settled_claims()?;
    for index in 0..settled {
        let key = format!("settled-{index}");
        assert_eq!(
            block_on(running.claim_with_evidence(&key, evidence(), 1))??,
            DedupDecision::Claimed
        );
        if index % 2 == 0 {
            block_on(running.complete_receipt(&key, ProcessingReceipt::for_position(1)))??;
        } else {
            block_on(running.release_claim(&key))??;
        }
    }
    drop(running);
    let store = land("a_start_reads_the_index_window", &seeded, seed_started)?;

    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!(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 {
    let seed_started = Instant::now();
    let seeded = Arc::new(MemoryStore::default());
    let running = DedupCache::new(Arc::clone(&seeded) as Arc<dyn DurableStore>, NAMESPACE);
    block_on(running.claim_with_evidence("k", evidence(), 1))??;
    block_on(running.release_claim("k"))??;
    block_on(running.claim_with_evidence("k", evidence(), 2))??;
    // Enough settled traffic after it that the start would have moved past
    // the second claim's row had the release forgotten it.
    for index in 0..settled_claims()? {
        let key = format!("settled-{index}");
        block_on(running.claim_with_evidence(&key, evidence(), 1))??;
        block_on(running.complete_receipt(&key, ProcessingReceipt::for_position(1)))??;
    }
    drop(running);
    let store = land(
        "a_claim_made_after_its_key_was_released",
        &seeded,
        seed_started,
    )?;

    let restarted = DedupCache::new(store, NAMESPACE);
    assert_eq!(
        block_on(restarted.in_flight_claims())??,
        [("k".to_owned(), Some(evidence()))]
    );
    Ok(())
}