liminal-rs 0.16.3

//! Board #60. The page limit must be honoured by the ENGINE, not by a
//! truncation applied after the engine has already materialised the suffix.
//!
//! These pins are counts, never durations: the defect is a read shape, and
//! a shape is deterministic where a latency is not.

use super::{DurableStore, EngineReadAccountingGuard};
use crate::durability::bridge::block_on;
use crate::durability::tests_fixtures;

/// Page size used by both production replay readers (`READ_BATCH_SIZE` and
/// `UNIT2_OUTBOX_RESTORE_BATCH_ROWS` are both 64).
const PAGE: usize = 64;
/// Four pages. Enough that the quadratic and the linear shape differ by
/// more than a factor of two, small enough that seeding stays cheap.
const ROWS: u64 = 256;
const STREAM: &str = "liminal/p0-60/paged-read-shape";

/// Seeds `ROWS` events into one stream.
fn seeded() -> Result<impl DurableStore, Box<dyn std::error::Error>> {
    tests_fixtures::copied("paged-read")
}

/// Walks the whole stream one page at a time, exactly as replay does.
fn read_whole_stream(
    store: &impl DurableStore,
    page: usize,
) -> Result<usize, Box<dyn std::error::Error>> {
    let mut offset = 0_u64;
    let mut seen = 0_usize;
    loop {
        let entries = block_on(store.read_from(STREAM, offset, page))??;
        if entries.is_empty() {
            return Ok(seen);
        }
        for entry in &entries {
            assert_eq!(entry.sequence, offset, "paged read must stay contiguous");
            offset += 1;
        }
        seen = seen
            .checked_add(entries.len())
            .ok_or("row counter overflowed")?;
    }
}

/// Every read below is bounded by its `limit`, so no read costs more than
/// the rows it returns. One seeded store carries all four shapes.
#[test]
fn a_bounded_read_never_scans_beyond_its_page() -> Result<(), Box<dyn std::error::Error>> {
    let store = seeded()?;

    // 1. The whole stream, paged. O(N), not O(N^2).
    let accounting = EngineReadAccountingGuard::start();
    let seen = read_whole_stream(&store, PAGE)?;
    let walk = accounting.snapshot();
    drop(accounting);
    assert_eq!(
        u64::try_from(seen)?,
        ROWS,
        "the walk must deliver every row"
    );
    assert!(
        !walk.counter_overflow_observed,
        "a saturated counter is not a measurement"
    );
    assert!(walk.calls > 0, "the walk must have reached the store");
    assert_eq!(
        u64::try_from(walk.engine_entries)?,
        ROWS,
        "a full stream read must scan each row exactly once instead of \
         re-scanning every suffix once per page"
    );

    // 2. One page from the head.
    let accounting = EngineReadAccountingGuard::start();
    let head = block_on(store.read_from(STREAM, 0, PAGE))??;
    let head_read = accounting.snapshot();
    drop(accounting);
    assert_eq!(head.len(), PAGE, "a full page returns its limit");
    assert_eq!(
        head_read.engine_entries, PAGE,
        "the engine must be asked for one page, not for the whole stream"
    );

    // 3. One page from the MIDDLE. The rows after the page are the ones a
    //    suffix-scanning read would drag along; the rows before it are the
    //    ones the offset already excludes, so only a bounded upper edge can
    //    make this count come out at PAGE.
    let middle_offset = ROWS / 2;
    let accounting = EngineReadAccountingGuard::start();
    let middle = block_on(store.read_from(STREAM, middle_offset, PAGE))??;
    let middle_read = accounting.snapshot();
    drop(accounting);
    assert_eq!(
        middle.len(),
        PAGE,
        "a full page mid-stream returns its limit"
    );
    assert_eq!(
        middle_read.engine_entries, PAGE,
        "a mid-stream page must not scan the rows that follow it"
    );

    // 4. Past the head: end of stream, and no scan.
    let accounting = EngineReadAccountingGuard::start();
    let past = block_on(store.read_from(STREAM, ROWS, PAGE))??;
    let past_read = accounting.snapshot();
    drop(accounting);
    assert!(past.is_empty(), "past the head is end of stream");
    assert_eq!(
        past_read.engine_entries, 0,
        "an end-of-stream page must not scan the stream"
    );
    Ok(())
}

/// The equivalence the pushdown must preserve. This passes before and after
/// the fix by design: it is the control that says the fix changed the read
/// SHAPE and nothing else.
#[test]
fn page_size_never_changes_the_answer() -> Result<(), Box<dyn std::error::Error>> {
    let store = seeded()?;
    let whole = block_on(store.read_from(STREAM, 0, usize::MAX))??;
    assert_eq!(u64::try_from(whole.len())?, ROWS);

    for page in [1_usize, 7, 64, 255, 256, 257] {
        let mut offset = 0_u64;
        let mut collected = Vec::new();
        loop {
            let entries = block_on(store.read_from(STREAM, offset, page))??;
            if entries.is_empty() {
                break;
            }
            assert!(entries.len() <= page, "a page never exceeds its limit");
            offset = offset
                .checked_add(u64::try_from(entries.len())?)
                .ok_or("offset overflowed")?;
            collected.extend(entries);
        }
        assert_eq!(collected, whole, "page size {page} changed the answer");
    }

    // A zero limit is the one page size that must return nothing, and it
    // must not be answered by a bounded window that silently agrees.
    assert!(
        block_on(store.read_from(STREAM, 0, 0))??.is_empty(),
        "a zero limit reads nothing"
    );

    // Every suffix start agrees with the same suffix of the whole read.
    for offset in [0_u64, 1, 63, 64, 65, 128, 255] {
        let suffix = block_on(store.read_from(STREAM, offset, usize::MAX))??;
        assert_eq!(
            suffix,
            whole[usize::try_from(offset)?..],
            "suffix from {offset} diverged"
        );
    }
    Ok(())
}