hyperfile 0.12.0

#![cfg(not(feature = "blocking"))]
//! Discarding changes not yet published: `HyperFile::discard_unflushed` and
//! `discard_range`, on memory staging, in every format.
//!
//! Each case keeps a model of what is published and checks, after the discard,
//! that the handle reads it — before a flush, after one, and cold — and that the
//! next publish holds only what was kept.
//!
//! ```bash
//! cargo test --test functional_discard
//! cargo test --features wal --test functional_discard
//! ```

#![cfg_attr(feature = "wal", allow(dead_code, unused_imports))]

use std::io::{ErrorKind, Result};
use std::sync::{Arc, Mutex};

use btree_ondisk::{MemoryBlockLoader, NullNodeCache};
use hyperfile::block_source::BlockSource;
use hyperfile::config::{HyperFileConfigBuilder, HyperFileMetaConfig, HyperFileRuntimeConfig};
use hyperfile::file::file::HyperFile;
use hyperfile::file::flags::{FileFlags, HyperFileFlags};
use hyperfile::file::mode::{FileMode, HyperFileMode};
use hyperfile::file::HyperTrait;
use hyperfile::inode::FlushInodeFlag;
use hyperfile::meta_format::BlockPtrFormat;
use hyperfile::ondisk::InodeRaw;
use hyperfile::staging::config::StagingConfig;
use hyperfile::staging::memory::MemoryStaging;
use hyperfile::staging::{Staging, StagingIntercept};
use hyperfile::SegmentId;

type MemFile<'a> = HyperFile<'a, MemoryStaging, MemoryBlockLoader<hyperfile::BlockPtr>, NullNodeCache>;

const BS: usize = 4096;

/// A format, and for the stream format how data is sent ahead.
#[derive(Debug, Clone, Copy)]
struct Shape {
    fmt: BlockPtrFormat,
    ahead: usize,
}

const SHAPES: [Shape; 4] = [
    Shape { fmt: BlockPtrFormat::PartedSegment, ahead: 0 },
    Shape { fmt: BlockPtrFormat::StreamSegment, ahead: 0 },
    Shape { fmt: BlockPtrFormat::StreamSegment, ahead: 5 << 20 },
    Shape { fmt: BlockPtrFormat::StreamSegment, ahead: 10 << 20 },
];

fn runtime(shape: Shape) -> HyperFileRuntimeConfig {
    let mut rt = HyperFileRuntimeConfig::default();
    rt.data_cache_dirty_max_bytes_threshold = 1 << 40;
    rt.data_cache_dirty_max_blocks_threshold = (1 << 40) / BS;
    rt.data_cache_dirty_max_flush_interval = u64::MAX / 2;
    rt.segment_buffer_size = if cfg!(feature = "wal") { 64 << 20 } else { 1 << 40 };
    rt.data_cache_blocks = 0;
    rt.stream_segment_bytes = shape.ahead;
    rt.segment_mpu_chunk_size = 5 << 20;
    rt
}

fn config(s: &MemoryStaging, shape: Shape) -> hyperfile::config::HyperFileConfig {
    let mut meta = HyperFileMetaConfig::default();
    meta.block_ptr_format = shape.fmt;
    HyperFileConfigBuilder::new().with_staging_config(s.config()).with_meta_config(&meta)
        .with_runtime_config(&runtime(shape)).build()
}

async fn create(name: &str, shape: Shape) -> (MemoryStaging, MemFile<'static>) {
    let s = MemoryStaging::new(StagingConfig::new_memory(name), runtime(shape)).with_block_ptr_format(shape.fmt);
    let f = MemFile::new(s.clone(), s.to_block_loader(), NullNodeCache, config(&s, shape),
        HyperFileFlags::from_flags(FileFlags::rdwr()), HyperFileMode::from_mode(FileMode::default_file()))
        .await.expect("create");
    (s, f)
}

async fn reopen(s: &MemoryStaging, shape: Shape) -> MemFile<'static> {
    MemFile::open(s.clone(), s.to_block_loader(), NullNodeCache, config(s, shape), HyperFileFlags::from_flags(FileFlags::rdwr()))
        .await.expect("open")
}

fn block(b: usize, g: u8) -> Vec<u8> {
    (0..BS).map(|i| (b as u8).wrapping_mul(29) ^ g ^ (i as u8)).collect()
}

async fn read_all(f: &mut MemFile<'static>, len: usize) -> Vec<u8> {
    let mut v = vec![0u8; len];
    let n = f.read(0, &mut v).await.expect("read");
    v.truncate(n);
    v
}

fn put(model: &mut Vec<u8>, off: usize, bytes: &[u8]) {
    if model.len() < off + bytes.len() {
        model.resize(off + bytes.len(), 0);
    }
    model[off..off + bytes.len()].copy_from_slice(bytes);
}

fn same(name: &str, what: &str, got: &[u8], want: &[u8]) {
    assert_eq!(got.len(), want.len(), "{}: {}: size", name, what);
    let bad = got.iter().zip(want).position(|(a, b)| a != b);
    assert!(bad.is_none(), "{}: {}: differs at block {:?}", name, what, bad.map(|o| o / BS));
}

fn blocks_counted(f: &MemFile<'static>) -> i64 {
    f.stat().st_blocks
}

/// A source over a plain buffer.
struct Buf(Mutex<Vec<u8>>);

impl BlockSource for Buf {
    fn read_at(&self, off: u64, buf: &mut [u8]) -> Result<()> {
        let d = self.0.lock().unwrap();
        buf.copy_from_slice(&d[off as usize..off as usize + buf.len()]);
        Ok(())
    }
}

/// What a published file is: 1200 blocks, a hole at 300..340, a zero block at 500.
async fn published(name: &str, shape: Shape) -> (MemoryStaging, MemFile<'static>, Vec<u8>) {
    let (s, mut f) = create(name, shape).await;
    let mut model = Vec::new();
    for b in (0..1200usize).filter(|b| !(300..340).contains(b)) {
        f.write(b * BS, &block(b, 1)).await.unwrap();
        put(&mut model, b * BS, &block(b, 1));
    }
    f.write_zero(500 * BS, BS).await.unwrap();
    put(&mut model, 500 * BS, &vec![0u8; BS]);
    f.flush().await.expect("flush");
    (s, f, model)
}

/// Every change since the publish goes: writes over published blocks and holes,
/// a zero block, a reference, data sent ahead, a size grown — the handle reads what
/// is published, the next flush publishes only what follows, and so do later opens.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn discard_unflushed_goes_back_to_what_is_published() {
    for shape in SHAPES {
        let name = format!("dc-all-{:?}-{}", shape.fmt, shape.ahead);
        let (s, mut f, model) = published(&name, shape).await;
        let cno0 = f.last_cno();
        let blocks0 = blocks_counted(&f);
        let src: Vec<u8> = (0..1200).flat_map(|b| block(b, 7)).collect();
        f.set_block_source(Some(Arc::new(Buf(Mutex::new(src))))).unwrap();
        for b in 0..3000usize {
            f.write(b * BS, &block(b, 5)).await.unwrap();
        }
        f.write_zero(100 * BS, 4 * BS).await.unwrap();
        f.write_ref(200 * BS, 8 * BS).await.unwrap();
        f.discard_unflushed().await.expect("discard");
        same(&name, "after the discard", &read_all(&mut f, 4000 * BS).await, &model);
        assert_eq!(blocks_counted(&f), blocks0, "{}: i_blocks", name);
        assert_eq!(f.last_cno(), cno0);
        // The next flush holds only what follows.
        f.write(7 * BS, &block(7, 9)).await.unwrap();
        let mut want = model.clone();
        put(&mut want, 7 * BS, &block(7, 9));
        let cno = f.flush().await.expect("flush").as_cno();
        assert!(cno > cno0);
        same(&name, "after the next flush", &read_all(&mut f, 4000 * BS).await, &want);
        let mut g = reopen(&s, shape).await;
        same(&name, "cold", &read_all(&mut g, 4000 * BS).await, &want);
    }
}

/// Every number this session used is moved past on storage: a later session's
/// checkpoint does not run into what was sent ahead and then discarded.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn a_discard_moves_the_counter_past_what_was_sent_ahead() {
    for shape in SHAPES {
        let name = format!("dc-counter-{:?}-{}", shape.fmt, shape.ahead);
        let (s, mut f, model) = published(&name, shape).await;
        for b in 0..4000usize {
            f.write(b * BS, &block(b, 5)).await.unwrap();
            tokio::task::yield_now().await;
        }
        let used = f.inode().get_last_seq().as_cno();
        f.discard_unflushed().await.expect("discard");
        drop(f);
        let stored = InodeRaw::from_u8_slice(&s.inode_bytes().unwrap());
        assert!(stored.i_last_seq >= used, "{}: the counter was not moved past {}: {}", name, used, stored.i_last_seq);
        let mut g = reopen(&s, shape).await;
        same(&name, "reopened", &read_all(&mut g, 4000 * BS).await, &model);
        for b in 0..4000usize {
            g.write(b * BS, &block(b, 6)).await.unwrap();
        }
        g.flush().await.expect("a later session publishes past it");
        let mut h = reopen(&s, shape).await;
        let want: Vec<u8> = (0..4000).flat_map(|b| block(b, 6)).collect();
        same(&name, "the later session's checkpoint", &read_all(&mut h, 4000 * BS).await, &want);
    }
}

/// The changes to a range go, and only those: the range reads what is published
/// — a published block, a hole, a zero block — whatever wrote it since, `write`,
/// `write_zero`, `write_ref` or `block_mut`; blocks outside keep their changes and
/// publish.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn discard_range_undoes_every_kind_of_write_and_only_there() {
    for shape in SHAPES {
        let name = format!("dc-range-{:?}-{}", shape.fmt, shape.ahead);
        let (s, mut f, model) = published(&name, shape).await;
        let src: Vec<u8> = (0..1200).flat_map(|b| block(b, 7)).collect();
        f.set_block_source(Some(Arc::new(Buf(Mutex::new(src.clone()))))).unwrap();
        let mut want = model.clone();
        // Changes in and around the range 280..520, which holds published blocks,
        // the hole 300..340 and the zero block at 500.
        for b in 250..550usize {
            f.write(b * BS, &block(b, 5)).await.unwrap();
        }
        f.write_zero(260 * BS, 4 * BS).await.unwrap();
        f.write_zero(400 * BS + 100, 50).await.unwrap();
        f.write_ref(290 * BS, 20 * BS).await.unwrap();
        f.write_ref(540 * BS, 4 * BS).await.unwrap();
        // In place over a published block, and created in the hole (from 280 on,
        // so inside the range), and in place outside it.
        for b in [450u64, 335, 530] {
            let mut m = f.block_mut(b, true).await.unwrap().expect("a block");
            m.as_mut_slice()[0] ^= 0xff;
        }
        let blocks_before = blocks_counted(&f);
        f.discard_range(280 * BS, 240 * BS).await.expect("discard_range");
        for b in 250..280usize {
            put(&mut want, b * BS, &block(b, 5));
        }
        put(&mut want, 260 * BS, &vec![0u8; 4 * BS]);
        for b in 520..550usize {
            put(&mut want, b * BS, &block(b, 5));
        }
        want[530 * BS] ^= 0xff;
        for b in 540..544usize {
            put(&mut want, b * BS, &src[b * BS..(b + 1) * BS]);
        }
        same(&name, "after the discard", &read_all(&mut f, 1200 * BS).await, &want);
        // The hole 300..340 had been filled and is a hole again.
        assert!(blocks_counted(&f) < blocks_before, "{}: i_blocks did not go down", name);
        f.flush().await.expect("flush");
        same(&name, "after the flush", &read_all(&mut f, 1200 * BS).await, &want);
        let mut g = reopen(&s, shape).await;
        same(&name, "cold", &read_all(&mut g, 1200 * BS).await, &want);
        let blocks_cold = blocks_counted(&g);
        assert_eq!(blocks_counted(&f), blocks_cold, "{}: i_blocks of the handle and of storage", name);
    }
}

/// Holds every data part's completion for `ms` milliseconds.
#[derive(Clone)]
struct Slow(u64);

impl StagingIntercept<MemoryStaging> for Slow {
    fn after_flush_inode(&self, _: &MemoryStaging, _: &[u8], _: FlushInodeFlag)
        -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + '_ + Send>> {
        Box::pin(async { Ok(()) })
    }
    fn after_remove_inode(&self, _: &MemoryStaging)
        -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + '_ + Send>> {
        Box::pin(async { Ok(()) })
    }
    fn before_segment_done(&self, _: &MemoryStaging, segid: SegmentId, _: &[u8], _: usize)
        -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + '_ + Send>> {
        let ms = if segid.part_id().is_some_and(|p| p > 0) { self.0 } else { 0 };
        Box::pin(async move { tokio::time::sleep(std::time::Duration::from_millis(ms)).await; Ok(()) })
    }
}

/// Under the stream format, a range discarded while its blocks are sent ahead —
/// in flight, and landed and out of memory — reads what is published, and the
/// checkpoint does not take those copies.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn discard_range_abandons_what_was_sent_ahead() {
    for shape in &SHAPES[2..] {
        for (when, wait) in [("in flight", 0u64), ("landed", 700)] {
            let name = format!("dc-sent-{}-{}", shape.ahead, when.replace(' ', "-"));
            let (s, mut f, model) = published(&name, *shape).await;
            f.staging_interceptor(Slow(300));
            let n = shape.ahead / BS + 600;
            let mut want = model.clone();
            for b in 0..n {
                f.write(b * BS, &block(b, 5)).await.unwrap();
                put(&mut want, b * BS, &block(b, 5));
            }
            tokio::time::sleep(std::time::Duration::from_millis(wait)).await;
            // Taken in now if landed.
            f.write(n * BS, &block(n, 5)).await.unwrap();
            put(&mut want, n * BS, &block(n, 5));
            f.discard_range(100 * BS, 300 * BS).await.expect("discard_range");
            want[100 * BS..400 * BS].copy_from_slice(&model[100 * BS..400 * BS]);
            same(&name, "after the discard", &read_all(&mut f, (n + 1) * BS).await, &want);
            tokio::time::sleep(std::time::Duration::from_millis(700)).await;
            f.flush().await.expect("flush");
            let mut g = reopen(&s, *shape).await;
            same(&name, "cold", &read_all(&mut g, (n + 1) * BS).await, &want);
        }
    }
}

/// A discard changes no size: past what is published, a discarded range of a
/// grown file reads zeros, and `truncate` sets the size back.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn a_discarded_range_past_the_published_end_reads_zeros() {
    let shape = SHAPES[1];
    let (s, mut f, mut model) = published("dc-size", shape).await;
    let end = model.len();
    for b in 1200..1300usize {
        f.write(b * BS, &block(b, 5)).await.unwrap();
    }
    f.discard_range(end, 100 * BS).await.unwrap();
    put(&mut model, end, &vec![0u8; 100 * BS]);
    same("dc-size", "grown and discarded", &read_all(&mut f, 1400 * BS).await, &model);
    f.truncate(end).await.unwrap();
    model.truncate(end);
    same("dc-size", "truncated back", &read_all(&mut f, 1400 * BS).await, &model);
    f.flush().await.unwrap();
    let mut g = reopen(&s, shape).await;
    same("dc-size", "cold", &read_all(&mut g, 1400 * BS).await, &model);
}

/// Several discards and writes in turn, as a client re-sending parts would: the
/// result is what the last writes that were kept made it.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn repeated_writes_and_discards_keep_what_was_kept() {
    for shape in SHAPES {
        let name = format!("dc-repeat-{:?}-{}", shape.fmt, shape.ahead);
        let (s, mut f, model) = published(&name, shape).await;
        let mut want = model.clone();
        let mut seed = 12345u64;
        for round in 0..30u8 {
            seed = seed.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407);
            let start = (seed >> 33) as usize % 2000;
            let len = 1 + (seed >> 20) as usize % 200;
            for b in start..start + len {
                f.write(b * BS, &block(b, round)).await.unwrap();
                put(&mut want, b * BS, &block(b, round));
            }
            if round % 3 == 2 {
                let d = start + len / 3;
                let dl = len / 2;
                f.discard_range(d * BS, dl * BS).await.unwrap();
                // What is published there.
                for b in d..d + dl {
                    let published = if b * BS < model.len() { model[b * BS..(b + 1) * BS].to_vec() } else { vec![0u8; BS] };
                    put(&mut want, b * BS, &published);
                }
            }
            tokio::task::yield_now().await;
        }
        let len = want.len();
        same(&name, "before the flush", &read_all(&mut f, len).await, &want);
        f.flush().await.unwrap();
        let mut g = reopen(&s, shape).await;
        same(&name, "cold", &read_all(&mut g, len).await, &want);
    }
}

/// What is refused: a range not on block boundaries, a handle not open for
/// writing.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn misuse_is_refused() {
    let (s, mut f, _) = published("dc-misuse", SHAPES[0]).await;
    assert_eq!(f.discard_range(100, BS).await.unwrap_err().kind(), ErrorKind::InvalidInput);
    assert_eq!(f.discard_range(BS, 100).await.unwrap_err().kind(), ErrorKind::InvalidInput);
    f.discard_range(BS, 0).await.expect("an empty range is nothing to do");
    let mut r = MemFile::open(s.clone(), s.to_block_loader(), NullNodeCache, config(&s, SHAPES[0]),
        HyperFileFlags::from_flags(FileFlags::rdonly())).await.unwrap();
    assert!(r.discard_unflushed().await.is_err());
    assert!(r.discard_range(0, BS).await.is_err());
}