coordinode-lsm-tree 6.0.0

use super::*;
use crate::{UserKey, UserValue, vlog::ValueHandle};

#[expect(clippy::unnecessary_wraps)]
fn entry(
    blob_file_id: BlobFileId,
    key: &[u8],
    offset: u64,
) -> crate::Result<(ScanEntry, BlobFileId)> {
    Ok((
        ScanEntry {
            key: UserKey::from(key),
            offset,
            seqno: 0,
            uncompressed_len: 0,
            value: UserValue::empty(),
            // These fixtures use ordinal offsets (0, 1, 2, ...), not real byte
            // positions, so the frame ends one ordinal past its start — never
            // at the start itself, which would mislead any future test that
            // inspects the consumed-frontier argument.
            frame_end: offset + 1,
            resynced: false,
        },
        blob_file_id,
    ))
}

#[test]
fn drain_blobs_simple() -> crate::Result<()> {
    let mut iter = [
        entry(0, b"a", 0),
        entry(0, b"a", 1),
        entry(0, b"a", 2),
        entry(0, b"a", 3),
        entry(0, b"a", 4),
    ]
    .into_iter()
    .peekable();

    drain_blobs(
        &mut iter,
        b"a",
        &BlobIndirection {
            size: 0,
            vhandle: ValueHandle {
                blob_file_id: 0,
                offset: 4,
                on_disk_size: 0,
            },
        },
        crate::comparator::default_comparator().as_ref(),
        &mut |_, _| {},
    )?;

    assert_eq!(entry(0, b"a", 4)?, iter.next().unwrap()?);

    Ok(())
}

#[test]
fn drain_blobs_multiple_keys() -> crate::Result<()> {
    let mut iter = [
        entry(0, b"a", 0),
        entry(0, b"b", 0),
        entry(0, b"c", 0),
        entry(0, b"d", 0),
        entry(0, b"e", 0),
    ]
    .into_iter()
    .peekable();

    drain_blobs(
        &mut iter,
        b"e",
        &BlobIndirection {
            size: 0,
            vhandle: ValueHandle {
                blob_file_id: 0,
                offset: 0,
                on_disk_size: 0,
            },
        },
        crate::comparator::default_comparator().as_ref(),
        &mut |_, _| {},
    )?;

    assert_eq!(entry(0, b"e", 0)?, iter.next().unwrap()?);

    Ok(())
}

/// Under a comparator that reverses byte order the scan runs from high keys
/// to low: draining towards a lower key passes higher ones, which sort before
/// it in the tree's order.
#[test]
fn drain_blobs_follows_the_tree_comparator() -> crate::Result<()> {
    struct Reverse;
    impl crate::comparator::UserComparator for Reverse {
        fn name(&self) -> &'static str {
            "reverse"
        }
        fn compare(&self, a: &[u8], b: &[u8]) -> core::cmp::Ordering {
            b.cmp(a)
        }
    }

    let mut iter = [entry(0, b"e", 0), entry(0, b"c", 1), entry(0, b"a", 2)]
        .into_iter()
        .peekable();
    drain_blobs(
        &mut iter,
        b"a",
        &BlobIndirection {
            size: 0,
            vhandle: ValueHandle {
                blob_file_id: 0,
                offset: 2,
                on_disk_size: 0,
            },
        },
        &Reverse,
        &mut |_, _| {},
    )?;

    assert_eq!(entry(0, b"a", 2)?, iter.next().unwrap()?);
    Ok(())
}

/// Draining a slice's unclaimed frames takes every frame below the bound,
/// records the proven ones and leaves the bound key's frames in place.
#[test]
#[expect(clippy::unwrap_used, reason = "test assertion")]
fn drain_unclaimed_below_takes_the_frames_below_the_bound() -> crate::Result<()> {
    let mut resync = entry(1, b"c", 7)?;
    resync.0.resynced = true;
    let mut iter = [
        entry(0, b"a", 0),
        entry(1, b"b", 4),
        Ok(resync),
        entry(0, b"d", 1),
    ]
    .into_iter()
    .peekable();

    let mut recorded = Vec::new();
    drain_unclaimed_below(
        &mut iter,
        b"d",
        crate::comparator::default_comparator().as_ref(),
        &mut |id, frame_end| recorded.push((id, frame_end)),
        &|| false,
    )?;

    assert_eq!(
        recorded,
        [(0, 1), (1, 5)],
        "the resynced frame moves nothing"
    );
    assert_eq!(entry(0, b"d", 1)?, iter.next().unwrap()?);
    Ok(())
}

/// A scan error met while draining ends the drain and is returned, not
/// skipped: the frontier never moves past a frame that could not be read.
#[test]
fn drain_unclaimed_below_returns_a_scan_error() {
    let mut iter = [
        entry(0, b"a", 0),
        Err(crate::Error::Io(crate::io::Error::other(
            "unreadable frame",
        ))),
        entry(0, b"b", 1),
    ]
    .into_iter()
    .peekable();

    let mut recorded = Vec::new();
    let result = drain_unclaimed_below(
        &mut iter,
        b"z",
        crate::comparator::default_comparator().as_ref(),
        &mut |id, frame_end| recorded.push((id, frame_end)),
        &|| false,
    );

    assert!(
        matches!(result, Err(crate::Error::Io(_))),
        "the scan error is returned: {result:?}"
    );
    assert_eq!(recorded, [(0, 1)], "only the frame before the error counts");
}

/// A stop requested while a slice drains its unclaimed frames ends the drain
/// with the compaction's cancellation error instead of scanning on.
#[test]
fn drain_unclaimed_below_stops_on_request() {
    let mut iter = [entry(0, b"a", 0), entry(0, b"b", 1)]
        .into_iter()
        .peekable();

    let mut recorded = 0usize;
    let result = drain_unclaimed_below(
        &mut iter,
        b"z",
        crate::comparator::default_comparator().as_ref(),
        &mut |_, _| recorded += 1,
        &|| true,
    );

    assert!(result.is_err(), "a stop must end the drain with an error");
    assert_eq!(recorded, 0, "nothing is drained after the stop");
    assert!(iter.next().is_some(), "the frames stay unread");
}

#[test]
fn drain_blobs_does_not_advance_the_frontier_past_a_resynced_frame() -> crate::Result<()> {
    // A RESYNCED entry's `frame_end` boundary is unproven (see
    // `ScanEntry::resynced`). Draining it must NOT advance the reclaim frontier,
    // or the tight-space loop could punch past that unanchored boundary and then
    // skip a later frame that is actually valid when the scan resumes there.
    let mut resync = entry(0, b"a", 3)?;
    resync.0.resynced = true;
    let mut iter = [entry(0, b"a", 0), entry(0, b"a", 1), Ok(resync)]
        .into_iter()
        .peekable();

    let mut max_recorded = 0u64;
    drain_blobs(
        &mut iter,
        b"a",
        &BlobIndirection {
            size: 0,
            vhandle: ValueHandle {
                blob_file_id: 0,
                offset: 9,
                on_disk_size: 0,
            },
        },
        crate::comparator::default_comparator().as_ref(),
        &mut |_, frame_end| max_recorded = max_recorded.max(frame_end),
    )?;

    // The last PROVEN frame (offset 1) ends at ordinal 2; the resynced frame
    // (frame_end 4) must not have advanced the frontier past it.
    assert_eq!(
        max_recorded, 2,
        "a resynced frame must not advance the reclaim frontier past the last proven boundary",
    );
    Ok(())
}