liminal-rs 0.11.0

//! DUR-008 A12 at start: the in-flight index row carries its claim's sequence,
//! so a start reads one page of two rows per named key and never the key's
//! head or history. Also the index row's two forms in the codec.
//!
//! Key histories are written raw, one `append_all` per key, so no pin walks a
//! long history through per-row appends.

use std::sync::Arc;

use super::codec::{self, DedupRecord};
use super::{ClaimEvidence, DedupCache, DedupEntry};
use crate::durability::bridge::block_on;
use crate::durability::{
    CountingStore, DurableStore, ProcessingReceipt, StoreCounts, StoreCountsHandle, open_ephemeral,
};

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

/// Claims held in flight in each namespace a start is counted over.
const KEYS: u64 = 4;

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

fn index_stream(namespace: &str) -> String {
    format!("{namespace}/in-flight")
}

fn index_start_key(namespace: &str) -> String {
    format!("{namespace}/in-flight-start")
}

/// Writes `rounds` claim-and-release rounds to `key`'s stream in one append.
fn write_history(
    store: &dyn DurableStore,
    cache: &DedupCache,
    key: &str,
    rounds: u64,
) -> TestResult {
    let claim = DedupEntry::new(key, None, 1).serialize()?;
    let release = DedupRecord::tombstone(key.to_owned(), 1).serialize()?;
    let mut rows = Vec::new();
    for _ in 0..rounds {
        rows.push(claim.clone());
        rows.push(release.clone());
    }
    let stream = cache.stream_key_for(key);
    block_on(store.append_all(&stream, rows, 0))??;
    Ok(())
}

/// What one start over `namespace` costs, after each of its [`KEYS`] keys
/// was given `rounds` claim-and-release rounds and then one open claim.
fn start_counts(
    store: &Arc<dyn DurableStore>,
    counts: &StoreCountsHandle,
    namespace: &str,
    rounds: u64,
) -> Result<StoreCounts, Box<dyn std::error::Error>> {
    let writer = DedupCache::new_empty(Arc::clone(store), namespace);
    for index in 0..KEYS {
        let key = format!("key-{index}");
        write_history(store.as_ref(), &writer, &key, rounds)?;
        block_on(writer.claim_with_evidence(&key, evidence(), 1))??;
    }

    let restarted = DedupCache::new(Arc::clone(store), namespace);
    let before = counts.snapshot();
    let claims = block_on(restarted.in_flight_claims())??;
    let start = counts.snapshot().since(&before);
    assert_eq!(claims.len(), usize::try_from(KEYS)?, "{claims:?}");
    Ok(start)
}

/// A start over keys of history H and over keys of history 10H costs the
/// same: one read of the index start, one read of the index, and one page of
/// at most two rows per named key.
#[test]
fn a_start_reads_two_rows_per_claim_whatever_the_key_history() -> TestResult {
    let counting = Arc::new(CountingStore::new(open_ephemeral(1)?));
    let counts = counting.handle();
    let store: Arc<dyn DurableStore> = counting;

    let short = start_counts(&store, &counts, "a12-start-short", 20)?;
    let long = start_counts(&store, &counts, "a12-start-long", 200)?;

    assert_eq!(short, long, "a start's reads grew with the keys' history");
    assert_eq!(
        short.read_value_calls, 1,
        "a start reads no value but the index start: {short}"
    );
    assert_eq!(
        short.read_at_calls, 0,
        "a start reads no key's head: {short}"
    );
    assert_eq!(short.scan_page_calls, 0, "a start walks no namespace");
    // One read of the index, then one page per named key.
    assert_eq!(short.read_from_calls, 1 + KEYS, "{short}");
    // The index rows, then at most two rows per named key.
    assert!(
        short.read_from_rows <= KEYS + 2 * KEYS,
        "a start read {} rows over {KEYS} keys",
        short.read_from_rows
    );
    Ok(())
}

/// A claim the index names that was completed, or released, after its join
/// is found settled by the row after it and dropped.
#[test]
fn a_settled_claim_named_by_the_index_is_dropped_at_start() -> TestResult {
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let namespace = "a12-start-settled";
    let running = DedupCache::new_empty(Arc::clone(&store), namespace);
    block_on(running.claim_with_evidence("completed", evidence(), 1))??;
    block_on(running.complete_receipt("completed", ProcessingReceipt::for_position(1)))??;
    block_on(running.claim_with_evidence("released", evidence(), 1))??;
    block_on(running.release_claim("released"))??;
    block_on(running.claim_with_evidence("held", evidence(), 1))??;

    let restarted = DedupCache::new(store, namespace);
    assert_eq!(
        block_on(restarted.in_flight_claims())??,
        [("held".to_owned(), Some(evidence()))]
    );
    // Once loaded, the held claim is read again from its sequence.
    assert_eq!(
        block_on(restarted.in_flight_claims())??,
        [("held".to_owned(), Some(evidence()))]
    );
    Ok(())
}

/// An index row from a binary older than the claim sequence names its key
/// alone; the start looks that key up in full and still finds it in flight.
#[test]
fn an_index_row_from_before_the_claim_sequence_falls_back_to_the_full_lookup() -> TestResult {
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let namespace = "a12-start-ldx1";
    let cache = DedupCache::new(Arc::clone(&store), namespace);
    write_history(store.as_ref(), &cache, "legacy", 3)?;
    let mut claim = DedupEntry::new("legacy", None, 1);
    claim.evidence = Some(evidence());
    let stream = cache.stream_key_for("legacy");
    block_on(store.append(&stream, claim.serialize()?, 6))??;
    block_on(store.append(
        &index_stream(namespace),
        codec::encode_join_without_claim("legacy")?,
        0,
    ))??;
    block_on(store.cas(&index_start_key(namespace), 0, 1))??;

    assert_eq!(
        block_on(cache.in_flight_claims())??,
        [("legacy".to_owned(), Some(evidence()))]
    );
    // The claim held is then read from the sequence the full look-up found.
    assert_eq!(
        block_on(cache.in_flight_claims())??,
        [("legacy".to_owned(), Some(evidence()))]
    );
    Ok(())
}

/// The crash window after the index row and before the claim: the key's
/// stream holds no row at the claim the index names, so the key is dropped,
/// whether its stream is empty or holds earlier settled claims.
#[test]
fn a_claim_never_written_after_its_join_is_dropped() -> TestResult {
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let namespace = "a12-start-unwritten";
    let cache = DedupCache::new(Arc::clone(&store), namespace);
    write_history(store.as_ref(), &cache, "after-history", 1)?;
    let rows = vec![
        codec::encode_join("never", 0)?,
        codec::encode_join("after-history", 2)?,
    ];
    block_on(store.append_all(&index_stream(namespace), rows, 0))??;
    block_on(store.cas(&index_start_key(namespace), 0, 1))??;

    assert!(block_on(cache.in_flight_claims())??.is_empty());
    Ok(())
}

#[test]
fn an_index_row_carries_its_key_and_claim_sequence() -> TestResult {
    let row = codec::encode_join("key-a", 41)?;
    assert_eq!(&row[..4], b"LDX2");
    assert_eq!(codec::decode_join(&row)?, ("key-a".to_owned(), Some(41)));
    Ok(())
}

#[test]
fn an_index_row_from_an_older_binary_decodes_without_a_claim_sequence() -> TestResult {
    let row = codec::encode_join_without_claim("key-a")?;
    assert_eq!(&row[..4], b"LDX1");
    assert_eq!(codec::decode_join(&row)?, ("key-a".to_owned(), None));
    Ok(())
}

#[test]
fn a_malformed_index_row_is_refused() -> TestResult {
    let mut truncated = codec::encode_join("key-a", 41)?;
    truncated.pop();
    assert!(codec::decode_join(&truncated).is_err());
    let mut trailing = codec::encode_join_without_claim("key-a")?;
    trailing.push(0);
    assert!(codec::decode_join(&trailing).is_err());
    assert!(codec::decode_join(b"LDX9").is_err());
    Ok(())
}