liminal-rs 0.16.1

//! The counting seam counts exactly the work forwarded through it and changes
//! no answer.

use crate::durability::bridge::block_on;
use crate::durability::tests_memory_store::{MemoryStore, lock_error};

use std::error::Error;
use std::time::Instant;

use super::*;

const STREAM: &str = "inbox:0";

#[test]
fn one_of_each_method_is_counted_exactly() -> Result<(), Box<dyn Error>> {
    let store = CountingStore::new(MemoryStore::default());

    block_on(store.append(STREAM, b"abc".to_vec(), 0))??;
    block_on(store.append_all(STREAM, vec![b"de".to_vec(), b"fghi".to_vec()], 1))??;
    let read = block_on(store.read_from(STREAM, 0, 10))??;
    let point = block_on(store.read_at(STREAM, 2))??;
    block_on(store.cas("cursor", 0, 7))??;
    let value = block_on(store.read_value("cursor"))??;
    let head = block_on(store.read_stream_next_seq(STREAM))??;
    let page = block_on(store.scan_page("inbox", None, 2))??;
    block_on(store.flush())??;

    assert_eq!(read.len(), 3);
    assert!(point.is_some());
    assert_eq!(value, Some(7));
    assert_eq!(head, Some(3));
    assert_eq!(page.entries.len(), 2);
    assert_eq!(
        store.snapshot(),
        StoreCounts {
            append_calls: 1,
            append_rows: 1,
            append_bytes: 3,
            append_all_calls: 1,
            append_all_rows: 2,
            append_all_bytes: 6,
            read_from_calls: 1,
            read_from_rows: 3,
            read_at_calls: 1,
            read_at_rows: 1,
            cas_calls: 1,
            read_value_calls: 1,
            read_stream_next_seq_calls: 1,
            scan_page_calls: 1,
            scan_page_rows: 2,
            flushes: 1,
            truncate_calls: 0,
            truncate_rows: 0,
            apply_routed_batch_calls: 0,
            apply_routed_batch_items: 0,
            get_routed_calls: 0,
            get_routed_rows: 0,
        }
    );
    Ok(())
}

#[test]
fn since_reports_only_the_work_between_two_snapshots() -> Result<(), Box<dyn Error>> {
    let store = CountingStore::new(MemoryStore::default());
    block_on(store.append(STREAM, b"before".to_vec(), 0))??;
    block_on(store.read_from(STREAM, 0, 10))??;

    let earlier = store.snapshot();
    for sequence in 1..=10 {
        block_on(store.append(STREAM, vec![0_u8; 4], sequence))??;
    }
    block_on(store.flush())??;
    let between = store.snapshot().since(&earlier);

    assert_eq!(
        between,
        StoreCounts {
            append_calls: 10,
            append_rows: 10,
            append_bytes: 40,
            flushes: 1,
            ..StoreCounts::default()
        }
    );
    Ok(())
}

#[test]
fn a_refused_call_is_counted_but_writes_and_returns_nothing() -> Result<(), Box<dyn Error>> {
    let store = CountingStore::new(MemoryStore::default());

    let refused = block_on(store.append(STREAM, b"late".to_vec(), 5))?;
    let refused_batch = block_on(store.append_all(STREAM, vec![b"x".to_vec()], 5))?;

    assert!(matches!(
        refused,
        Err(DurabilityError::SequenceConflict { .. })
    ));
    assert!(matches!(
        refused_batch,
        Err(DurabilityError::SequenceConflict { .. })
    ));
    assert_eq!(
        store.snapshot(),
        StoreCounts {
            append_calls: 1,
            append_all_calls: 1,
            ..StoreCounts::default()
        }
    );
    Ok(())
}

#[test]
fn the_wrapper_answers_exactly_what_the_wrapped_store_answers() -> Result<(), Box<dyn Error>> {
    let bare = MemoryStore::default();
    let counted = CountingStore::new(MemoryStore::default());
    for store in [&bare as &dyn DurableStore, &counted] {
        block_on(store.append(STREAM, b"one".to_vec(), 0))??;
        block_on(store.append_all("inbox:1", vec![b"two".to_vec(), b"three".to_vec()], 0))??;
        block_on(store.cas("cursor", 0, 3))??;
    }

    assert_eq!(
        block_on(counted.read_from(STREAM, 0, 5))??,
        block_on(bare.read_from(STREAM, 0, 5))??
    );
    assert_eq!(
        block_on(counted.read_at("inbox:1", 1))??,
        block_on(bare.read_at("inbox:1", 1))??
    );
    assert_eq!(block_on(counted.read_at(STREAM, 9))??, None);
    assert_eq!(
        block_on(counted.read_value("cursor"))??,
        block_on(bare.read_value("cursor"))??
    );
    assert_eq!(
        block_on(counted.scan_page("inbox", None, 2))??,
        block_on(bare.scan_page("inbox", None, 2))??
    );
    assert_eq!(
        block_on(counted.read_stream_next_seq(STREAM))??,
        block_on(bare.read_stream_next_seq(STREAM))??
    );
    let counted_conflict = block_on(counted.cas("cursor", 0, 4))?;
    let bare_conflict = block_on(bare.cas("cursor", 0, 4))?;
    assert_eq!(
        format!("{counted_conflict:?}"),
        format!("{bare_conflict:?}")
    );
    Ok(())
}

#[test]
fn display_prints_every_counter_one_per_line_in_a_fixed_order() {
    let counts = StoreCounts {
        append_calls: 1,
        read_from_rows: 22,
        flushes: 3,
        ..StoreCounts::default()
    };

    let printed = counts.to_string();

    assert_eq!(
        printed,
        "append_calls 1\n\
         append_rows 0\n\
         append_bytes 0\n\
         append_all_calls 0\n\
         append_all_rows 0\n\
         append_all_bytes 0\n\
         read_from_calls 0\n\
         read_from_rows 22\n\
         read_at_calls 0\n\
         read_at_rows 0\n\
         cas_calls 0\n\
         read_value_calls 0\n\
         read_stream_next_seq_calls 0\n\
         scan_page_calls 0\n\
         scan_page_rows 0\n\
         flushes 3\n\
         truncate_calls 0\n\
         truncate_rows 0\n\
         apply_routed_batch_calls 0\n\
         apply_routed_batch_items 0\n\
         get_routed_calls 0\n\
         get_routed_rows 0\n"
    );
}

/// `truncate_before` over a real haematite store deletes the rows below the
/// cut and counts the call and the rows; the same cut again deletes nothing
/// and is still counted as a call, so a caller can see it ran.
#[test]
fn truncate_before_counts_its_rows() -> Result<(), Box<dyn Error>> {
    let store = CountingStore::new(crate::durability::open_ephemeral(1)?);
    let rows: Vec<Vec<u8>> = (0_u8..10).map(|row| vec![row]).collect();
    block_on(store.append_all(STREAM, rows, 0))??;

    let before = store.snapshot();
    assert_eq!(block_on(store.truncate_before(STREAM, 4))??, 4);
    let first = store.snapshot().since(&before);
    assert_eq!((first.truncate_calls, first.truncate_rows), (1, 4));
    let kept = block_on(store.read_from(STREAM, 4, 20))??;
    assert_eq!(kept.first().map(|entry| entry.sequence), Some(4));
    assert_eq!(kept.len(), 6);

    let again = store.snapshot();
    assert_eq!(block_on(store.truncate_before(STREAM, 4))??, 0);
    let second = store.snapshot().since(&again);
    assert_eq!((second.truncate_calls, second.truncate_rows), (1, 0));

    // The head is kept: the next append lands where it would have.
    assert_eq!(block_on(store.append(STREAM, vec![10], 10))??, 10);
    Ok(())
}

/// A store that cannot delete refuses by name; it never answers 0 as if it
/// had cut.
#[test]
fn a_store_that_cannot_truncate_refuses_by_name() -> Result<(), Box<dyn Error>> {
    let store = CountingStore::new(MemoryStore::default());
    let refusal = block_on(store.truncate_before(STREAM, 4))?
        .err()
        .ok_or("a store without truncate must refuse")?;
    assert!(
        matches!(
            &refusal,
            DurabilityError::TruncateUnsupported { stream_key, before_seq: 4, .. }
                if stream_key == STREAM
        ),
        "got {refusal:?}"
    );
    let counts = store.snapshot();
    assert_eq!((counts.truncate_calls, counts.truncate_rows), (1, 0));
    Ok(())
}

/// Calls per timed batch in the cost measurement.
const BATCH_CALLS: usize = 1_000;

/// Timed batches in the cost measurement: 100,000 calls each way.
const BATCHES: usize = 100;

/// Median nanoseconds per call of `read_value` on `store`, over [`BATCHES`]
/// batches of [`BATCH_CALLS`] calls.
fn median_nanos_per_call(store: &dyn DurableStore) -> Result<u128, Box<dyn Error>> {
    let mut per_call = Vec::with_capacity(BATCHES);
    for _ in 0..BATCHES {
        let started = Instant::now();
        for _ in 0..BATCH_CALLS {
            block_on(store.read_value("cursor"))??;
        }
        per_call.push(started.elapsed().as_nanos() / u128::from(u32::try_from(BATCH_CALLS)?));
    }
    per_call.sort_unstable();
    Ok(per_call[BATCHES / 2])
}

/// The wrapper's own cost per call, printed for the record in `counting.rs`.
///
/// It asserts the counts only: a time is a measurement of the machine, never a
/// pass or fail.
#[test]
fn the_wrapper_cost_per_call_is_measured_over_a_hundred_thousand_calls()
-> Result<(), Box<dyn Error>> {
    let bare = MemoryStore::default();
    let counted = CountingStore::new(MemoryStore::default());

    let bare_nanos = median_nanos_per_call(&bare)?;
    let counted_nanos = median_nanos_per_call(&counted)?;

    println!(
        "counting seam cost: median {counted_nanos} ns per call wrapped, {bare_nanos} ns bare, \
         difference {} ns, over {} calls each",
        counted_nanos.saturating_sub(bare_nanos),
        BATCHES * BATCH_CALLS
    );
    assert_eq!(
        counted.snapshot(),
        StoreCounts {
            read_value_calls: u64::try_from(BATCHES * BATCH_CALLS)?,
            ..StoreCounts::default()
        }
    );
    Ok(())
}

#[derive(Debug, Default)]
struct IncrementObserver {
    increments: AtomicU64,
    flushes: Mutex<Vec<(String, Option<u64>)>>,
    flush_failed: std::sync::atomic::AtomicBool,
}

impl StoreCountObserver for IncrementObserver {
    fn changed(&self) {
        self.increments.fetch_add(1, Ordering::Relaxed);
    }

    fn flushed(&self, name: &str, batch_rows: Option<u64>) {
        match self.flushes.lock() {
            Ok(mut flushes) => flushes.push((name.to_owned(), batch_rows)),
            Err(_) => self.flush_failed.store(true, Ordering::Release),
        }
    }
}

#[test]
fn observer_counts_actual_increments_and_one_event_per_flush() -> Result<(), Box<dyn Error>> {
    let store = CountingStore::new(MemoryStore::default());
    let observer = Arc::new(IncrementObserver::default());
    store
        .handle()
        .install_observer(&["durable"], observer.clone())?;
    assert_eq!(
        block_on(store.append_all(STREAM, vec![vec![1], vec![2]], 0))??,
        2
    );
    assert_eq!(observer.increments.load(Ordering::Relaxed), 3);
    assert_eq!(block_on(store.read_from(STREAM, 0, 2))??.len(), 2);
    assert_eq!(observer.increments.load(Ordering::Relaxed), 5);
    assert!(block_on(store.read_at(STREAM, 9))??.is_none());
    assert_eq!(observer.increments.load(Ordering::Relaxed), 6);
    block_on(store.flush())??;
    assert!(!observer.flush_failed.load(Ordering::Acquire));
    assert_eq!(observer.increments.load(Ordering::Relaxed), 7);
    let flushes = observer.flushes.lock().map_err(|_| lock_error())?;
    assert_eq!(*flushes, vec![("durable".to_owned(), None)]);
    drop(flushes);
    assert_eq!(store.snapshot().flushes, 1);
    assert_eq!(store.snapshot().append_all_rows, 2);
    Ok(())
}

/// R3b F3: a truncate is store work like any other, so an installed
/// observer hears it once per counter it moves, the refused one included.
#[test]
fn observer_hears_every_truncate_counter_it_moves() -> Result<(), Box<dyn Error>> {
    let store = CountingStore::new(crate::durability::open_ephemeral(1)?);
    let observer = Arc::new(IncrementObserver::default());
    store
        .handle()
        .install_observer(&["durable"], observer.clone())?;
    let rows: Vec<Vec<u8>> = (0_u8..10).map(|row| vec![row]).collect();
    block_on(store.append_all(STREAM, rows, 0))??;
    let before = observer.increments.load(Ordering::Relaxed);
    assert_eq!(block_on(store.truncate_before(STREAM, 4))??, 4);
    assert_eq!(
        observer.increments.load(Ordering::Relaxed) - before,
        2,
        "a truncate that cut rows moved truncate_calls and truncate_rows unheard"
    );
    let before = observer.increments.load(Ordering::Relaxed);
    assert_eq!(block_on(store.truncate_before(STREAM, 4))??, 0);
    assert_eq!(
        observer.increments.load(Ordering::Relaxed) - before,
        1,
        "a truncate that cut nothing moved truncate_calls unheard"
    );

    // MemoryStore cannot delete, so its truncate is the refused case.
    let refusing = CountingStore::new(MemoryStore::default());
    let observer = Arc::new(IncrementObserver::default());
    refusing
        .handle()
        .install_observer(&["durable"], observer.clone())?;
    assert!(block_on(refusing.truncate_before(STREAM, 4))?.is_err());
    assert_eq!(
        observer.increments.load(Ordering::Relaxed),
        1,
        "a refused truncate moved truncate_calls unheard"
    );
    Ok(())
}

#[test]
fn observer_install_validation_is_atomic_across_counted_stores() -> Result<(), Box<dyn Error>> {
    let first = CountingStore::new(MemoryStore::default());
    let second = CountingStore::new(MemoryStore::default());
    let observer = Arc::new(IncrementObserver::default());
    let combined = first.handle().with(&second.handle());
    assert!(
        combined
            .install_observer(&["durable"], observer.clone())
            .is_err()
    );
    assert!(
        combined
            .install_observer(&["durable", " "], observer.clone())
            .is_err()
    );
    assert!(
        first
            .handle()
            .with(&first.handle())
            .install_observer(&["one", "two"], observer.clone())
            .is_err()
    );
    combined.install_observer(&["durable", "ephemeral-dedup"], observer.clone())?;
    assert!(
        combined
            .install_observer(&["durable", "ephemeral-dedup"], observer.clone())
            .is_err()
    );
    block_on(first.flush())??;
    block_on(second.flush())??;
    assert!(!observer.flush_failed.load(Ordering::Acquire));
    let flushes = observer.flushes.lock().map_err(|_| lock_error())?;
    assert_eq!(
        *flushes,
        vec![
            ("durable".to_owned(), None),
            ("ephemeral-dedup".to_owned(), None)
        ]
    );
    drop(flushes);
    Ok(())
}

#[test]
fn stream_heads_are_independent_of_numeric_values_and_counted_once() -> Result<(), Box<dyn Error>> {
    let store = CountingStore::new(MemoryStore::default());
    assert_eq!(block_on(store.read_stream_next_seq(STREAM))??, None);
    block_on(store.cas(STREAM, 0, 99))??;
    assert_eq!(block_on(store.read_stream_next_seq(STREAM))??, None);
    block_on(store.append_all(STREAM, vec![vec![1], vec![2]], 0))??;
    let before = store.snapshot();
    assert_eq!(block_on(store.read_stream_next_seq(STREAM))??, Some(2));
    let delta = store.snapshot().since(&before);
    assert_eq!(
        delta,
        StoreCounts {
            read_stream_next_seq_calls: 1,
            ..StoreCounts::default()
        }
    );
    assert_eq!(delta.plus(&delta).read_stream_next_seq_calls, 2);
    Ok(())
}