liminal-rs 0.16.2

//! DUR-008 A12: a dedup lookup never reads a key's whole history. A short key
//! is read in one page, as before A12, with no head; a long key is read past
//! its first page from its verified recorded head.
//!
//! Long 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::DedupRecord;
use super::{DedupCache, DedupDecision, DedupEntry};
use crate::durability::bridge::block_on;
use crate::durability::head_checkpoint::{HEAD_CHECKPOINT_EVERY, READ_BATCH_SIZE, advance_head};
use crate::durability::{
    CountingStore, DurableStore, ProcessingReceipt, StoreCounts, head_checkpoint_key,
    open_ephemeral,
};

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

const NAMESPACE: &str = "a12-lookup";

/// Rows past the recorded head in both keys of the H and 10H pin.
const TAIL: u64 = 5;

/// Writes `rounds` claim-and-release rounds and one open claim to `key`'s
/// stream in one append, answering the stream's row count.
fn write_long_history(
    store: &dyn DurableStore,
    cache: &DedupCache,
    key: &str,
    rounds: u64,
) -> Result<u64, Box<dyn std::error::Error>> {
    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());
    }
    rows.push(claim);
    Ok(block_on(store.append_all(
        &cache.stream_key_for(key),
        rows,
        0,
    ))??)
}

/// The reads one lookup of `key` costs on a fresh cache, over a stream of
/// `rounds` claim-and-release rounds and one open claim whose head is
/// recorded [`TAIL`] rows before its end.
fn long_lookup_counts(rounds: u64) -> Result<StoreCounts, Box<dyn std::error::Error>> {
    let counting = Arc::new(CountingStore::new(open_ephemeral(1)?));
    let counts = counting.handle();
    let store: Arc<dyn DurableStore> = counting;
    let cache = DedupCache::new_empty(Arc::clone(&store), NAMESPACE);
    let rows = write_long_history(store.as_ref(), &cache, "key", rounds)?;
    block_on(advance_head(
        store.as_ref(),
        &cache.stream_key_for("key"),
        rows - TAIL,
    ))??;

    let before = counts.snapshot();
    let found = block_on(cache.lookup("key"))??;
    let lookup = counts.snapshot().since(&before);
    assert_eq!(found, Some(DedupDecision::InFlight));
    Ok(lookup)
}

/// A key of history H and one of history 10H, both longer than one page and
/// each with its head recorded [`TAIL`] rows before its end, cost one lookup
/// the same: every counter is equal (the same calls of each kind and the same
/// rows, one full first page plus the [`TAIL`] rows past the head), and
/// neither reads its history between the first page and the head.
#[test]
fn a_lookup_reads_the_same_rows_for_a_key_of_history_h_and_10h() -> TestResult {
    let short = long_lookup_counts(600)?;
    let long = long_lookup_counts(6_000)?;
    assert_eq!(short, long, "lookup reads grew with the key's history");
    let page = u64::try_from(READ_BATCH_SIZE)?;
    assert_eq!(short.read_from_rows, page + TAIL, "{short}");
    assert!(short.read_from_rows <= page + HEAD_CHECKPOINT_EVERY);
    assert_eq!(short.read_value_calls, 1, "{short}");
    assert_eq!(short.read_at_calls, 1, "{short}");
    assert_eq!(short.cas_calls, 0, "a short tail records no head: {short}");
    assert_eq!(short.scan_page_calls, 0);
    Ok(())
}

/// A key of two records is read in one page: no head is read, verified or
/// recorded.
#[test]
fn a_short_key_lookup_reads_one_page_and_no_head() -> TestResult {
    let counting = Arc::new(CountingStore::new(open_ephemeral(1)?));
    let counts = counting.handle();
    let store: Arc<dyn DurableStore> = counting;
    let cache = DedupCache::new(Arc::clone(&store), NAMESPACE);
    block_on(cache.claim_or_get("key", 1))??;
    block_on(cache.complete_receipt("key", ProcessingReceipt::for_position(1)))??;

    let before = counts.snapshot();
    let found = block_on(cache.lookup("key"))??;
    let lookup = counts.snapshot().since(&before);
    assert!(
        matches!(found, Some(DedupDecision::Completed(_))),
        "{found:?}"
    );
    assert_eq!(
        lookup,
        StoreCounts {
            read_from_calls: 1,
            read_from_rows: 2,
            ..StoreCounts::default()
        }
    );
    Ok(())
}

/// A long key stream with no recorded head is walked once, its head recorded,
/// and every lookup after that reads its first page and the head's row only.
#[test]
fn a_stream_without_a_recorded_head_is_walked_once() -> TestResult {
    let counting = Arc::new(CountingStore::new(open_ephemeral(1)?));
    let counts = counting.handle();
    let store: Arc<dyn DurableStore> = counting;
    let cache = DedupCache::new(Arc::clone(&store), NAMESPACE);
    let rows = write_long_history(store.as_ref(), &cache, "key", 600)?;
    let head = head_checkpoint_key(&cache.stream_key_for("key"));
    assert_eq!(block_on(store.read_value(&head))??, None);

    let before = counts.snapshot();
    assert_eq!(
        block_on(cache.lookup("key"))??,
        Some(DedupDecision::InFlight)
    );
    let first = counts.snapshot().since(&before);
    assert_eq!(
        first.read_from_rows, rows,
        "the first lookup walks: {first}"
    );
    assert_eq!(first.cas_calls, 1, "the walk records the head: {first}");
    assert_eq!(block_on(store.read_value(&head))??, Some(rows));

    let before = counts.snapshot();
    assert_eq!(
        block_on(cache.lookup("key"))??,
        Some(DedupDecision::InFlight)
    );
    let second = counts.snapshot().since(&before);
    assert_eq!(
        second.read_from_rows,
        u64::try_from(READ_BATCH_SIZE)?,
        "the second lookup reads its first page and nothing past the head: {second}"
    );
    assert_eq!(second.cas_calls, 0, "{second}");
    Ok(())
}