hyperfile 0.9.0

#![cfg(not(feature = "blocking"))]
//! A checkpoint larger than one object is written as several.
//!
//! A block pointer addresses at most 1 GiB within one object, so a flush of more
//! than that used to wrap offsets into other blocks' places and store data that
//! read back wrong, while reporting success. Under the parted format a flush now
//! lays a checkpoint out in objects of at most `segment_max_object_bytes`:
//! data-only parts, then the part holding the index and the inode, uploaded last
//! because it is what makes the checkpoint. Formats that cannot name a second
//! object, and the WAL build, refuse instead, before anything is written.
//!
//! On memory staging, with `segment_max_object_bytes` set to 64 KiB so that a few
//! hundred blocks exercise what a real container reaches past 1 GiB. The same
//! code runs at full size against S3 in `integration_s3_large_checkpoint`.
//!
//! ```bash
//! cargo test --test functional_split_checkpoint
//! cargo test --features wal --test functional_split_checkpoint
//! ```

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

use std::io::{Error, 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::{SegmentHeader, SEGMENT_FLAG_PARTIAL};
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;
/// Small enough that a few hundred blocks take many objects.
const CAP: usize = 64 * 1024;

fn runtime(cap: usize) -> HyperFileRuntimeConfig {
    let mut rt = HyperFileRuntimeConfig::default();
    rt.segment_max_object_bytes = cap;
    // Nothing publishes or writes a partial on its own unless a test asks.
    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;
    // The WAL build allocates this up front for each segment, so it has to be a
    // size that can be allocated; nothing here comes near it.
    rt.segment_buffer_size = if cfg!(feature = "wal") { 64 << 20 } else { 1 << 40 };
    rt
}

fn config(staging: &MemoryStaging, rt: &HyperFileRuntimeConfig, fmt: BlockPtrFormat) -> hyperfile::config::HyperFileConfig {
    let mut meta = HyperFileMetaConfig::default();
    meta.block_ptr_format = fmt;
    HyperFileConfigBuilder::new()
        .with_staging_config(staging.config())
        .with_meta_config(&meta)
        .with_runtime_config(rt)
        .build()
}

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

async fn create(name: &str, rt: HyperFileRuntimeConfig) -> (MemoryStaging, MemFile<'static>) {
    create_fmt(name, rt, BlockPtrFormat::PartedSegment).await
}

/// A fresh handle on what is stored: nothing it reads comes from a cache.
async fn cold(staging: &MemoryStaging) -> MemFile<'static> {
    MemFile::open(staging.clone(), staging.to_block_loader(), NullNodeCache,
        config(staging, &HyperFileRuntimeConfig::default(), BlockPtrFormat::PartedSegment),
        HyperFileFlags::from_flags(FileFlags::rdonly())).await.expect("open")
}

fn pattern(seed: u32, len: usize) -> Vec<u8> {
    let mut x = seed | 1;
    (0..len).map(|_| { x ^= x << 13; x ^= x >> 17; x ^= x << 5; x 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
}

/// The objects stored for checkpoint `cno`, by part: `(part, header, bytes)`.
fn objects_of(staging: &MemoryStaging, cno: u64) -> Vec<(u16, SegmentHeader, usize)> {
    (0..4096u16).filter_map(|k| {
        let id = SegmentId::new_from_cno(cno).at_part(k);
        staging.segment_bytes(id).map(|b| (k, SegmentHeader::from_slice(&b[..SegmentHeader::size()]), b.len()))
    }).collect()
}

/// A block source over a buffer: referenced blocks never cross a threshold, so a
/// test can put more than one object's worth in front of a flush on a container
/// that writes partials.
struct Src(Vec<u8>);

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

/// What reached storage, in order: each segment object by name, and each inode
/// write as `None`. Optionally refuses the `n`th segment write.
#[derive(Clone, Default)]
struct Log {
    writes: Arc<Mutex<Vec<Option<SegmentId>>>>,
    refuse_segment: Arc<Mutex<Option<usize>>>,
    seen: Arc<Mutex<usize>>,
}

impl Log {
    fn writes(&self) -> Vec<Option<SegmentId>> {
        self.writes.lock().unwrap().clone()
    }
    fn segments(&self) -> Vec<SegmentId> {
        self.writes().into_iter().flatten().collect()
    }
}

impl StagingIntercept<MemoryStaging> for Log {
    fn before_flush_inode(&self, _: &MemoryStaging, _: &[u8], _: FlushInodeFlag)
        -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + '_ + Send>> {
        self.writes.lock().unwrap().push(None);
        Box::pin(async { Ok(()) })
    }
    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 n = { let mut s = self.seen.lock().unwrap(); let n = *s; *s += 1; n };
        let refuse = *self.refuse_segment.lock().unwrap() == Some(n);
        if !refuse {
            self.writes.lock().unwrap().push(Some(segid));
        }
        Box::pin(async move {
            if refuse { Err(Error::new(ErrorKind::ConnectionReset, format!("segment write {} refused", segid))) } else { Ok(()) }
        })
    }
}

/// A checkpoint larger than one object is several, each within the limit, and
/// reads back cold exactly as written. Part 0 holds every meta block and is the
/// commit; the rest are data-only partials.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn a_checkpoint_past_one_object_is_written_as_several() {
    let (s, mut f) = create("split-basic", runtime(CAP)).await;
    let data = pattern(1, 300 * BS);
    f.write(0, &data).await.unwrap();
    let sid = f.flush().await.expect("flush");
    let objs = objects_of(&s, sid.as_cno());
    assert!(objs.len() > 5, "300 blocks in 64 KiB objects took only {} objects", objs.len());
    let total_meta: u32 = objs.iter().map(|(_, h, _)| h.s_nmetablk).sum();
    let total_data: u32 = objs.iter().map(|(_, h, _)| h.s_ndatablk).sum();
    assert_eq!(total_data, 300, "every block is in exactly one object");
    for (k, h, len) in &objs {
        assert!(*len <= CAP, "part {} is {} bytes, past the {} limit", k, len, CAP);
        if *k == 0 {
            assert_eq!(h.s_flags & SEGMENT_FLAG_PARTIAL, 0, "part 0 is the commit");
            assert_eq!(h.s_nmetablk, total_meta, "part 0 holds every meta block");
        } else {
            assert_ne!(h.s_flags & SEGMENT_FLAG_PARTIAL, 0, "data part {} is not marked partial", k);
            assert_eq!(h.s_nmetablk, 0, "data part {} holds meta blocks", k);
        }
    }
    let parts: Vec<u16> = objs.iter().map(|(k, _, _)| *k).collect();
    assert_eq!(parts, (0..objs.len() as u16).collect::<Vec<_>>(), "parts are numbered without gaps");
    let mut c = cold(&s).await;
    assert!(read_all(&mut c, data.len()).await == data, "the checkpoint read back wrong");
}

/// Laid out as several objects, a checkpoint reads back exactly as the same
/// writes stored as one object do — the layout changes where blocks are, never
/// what they hold. Also with a bmap deep enough to span several index levels.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn split_and_unsplit_checkpoints_read_the_same() {
    // The deep one has more index than 64 KiB, so it splits at a larger limit.
    for (name, blocks, split_cap) in [("same-small", 40usize, CAP), ("same-deep", 6000, 256 * 1024)] {
        let mut stored = Vec::new();
        for cap in [split_cap, 1 << 30] {
            let (s, mut f) = create(&format!("{}-{}", name, cap), runtime(cap)).await;
            f.write(0, &pattern(2, blocks * BS)).await.unwrap();
            f.flush().await.unwrap();
            // a second checkpoint over part of it, holes and zeroes included
            f.write(7 * BS + 100, &pattern(3, 20 * BS)).await.unwrap();
            f.write_zero(30 * BS, 3 * BS).await.unwrap();
            f.truncate((blocks - 5) * BS).await.unwrap();
            f.flush().await.unwrap();
            let mut c = cold(&s).await;
            stored.push(read_all(&mut c, blocks * BS).await);
        }
        assert!(stored[0] == stored[1], "{}: split and unsplit containers read differently", name);
        assert_eq!(stored[0].len(), (blocks - 5) * BS, "{}", name);
    }
}

/// Part 0 reaches storage after every data part and before the inode: an object
/// holding the inode is a commit to whoever finds it, so what it points at has
/// to be there first.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn part_0_lands_after_the_data_parts_and_before_the_inode() {
    let (_s, mut f) = create("split-order", runtime(CAP)).await;
    let log = Log::default();
    f.staging_interceptor(log.clone());
    f.write(0, &pattern(4, 200 * BS)).await.unwrap();
    let sid = f.flush().await.expect("flush");
    let w = log.writes();
    let last_segment = w.iter().rposition(|x| x.is_some()).unwrap();
    assert_eq!(w[last_segment], Some(SegmentId::new_from_cno(sid.as_cno()).at_part(0)),
        "part 0 is not the last object written: {:?}", w);
    assert_eq!(w.last(), Some(&None), "the inode is not written last: {:?}", w);
    assert!(w[..last_segment].iter().all(|x| x.is_some_and(|id| id.part_id() != Some(0))),
        "an inode or part 0 was written before the data parts: {:?}", w);
}

/// A checkpoint already part-way out as partial segments continues its part
/// numbers: the data parts of the final flush come after the partials, and no
/// name is written twice.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn a_split_flush_after_partials_continues_their_numbering() {
    let mut rt = runtime(CAP);
    rt.parted_segment_enabled = true;
    rt.data_cache_dirty_min_bytes_to_part = 8 * BS;
    rt.data_cache_dirty_max_blocks_threshold = 10;
    rt.data_cache_dirty_max_bytes_threshold = 10 * BS;
    let (s, mut f) = create("split-after-partials", rt).await;
    let log = Log::default();
    f.staging_interceptor(log.clone());
    let data = pattern(5, 120 * BS);
    // each write crosses the threshold, so partials are written along the way
    for i in 0..12 {
        f.write(i * 10 * BS, &data[i * 10 * BS..(i + 1) * 10 * BS]).await.unwrap();
    }
    // and a last batch big enough to be split by the flush itself, by reference
    // so that no threshold writes it out first
    let tail = pattern(6, 60 * BS);
    let mut backing = vec![0u8; 120 * BS];
    backing.extend_from_slice(&tail);
    f.set_block_source(Some(Arc::new(Src(backing)))).unwrap();
    f.write_ref(120 * BS, 60 * BS).await.unwrap();
    let before_flush = log.segments().len();
    assert!(before_flush > 0, "no partial was written");
    f.flush().await.expect("flush");
    let segs = log.segments();
    let mut seen = std::collections::HashSet::new();
    for id in &segs {
        assert!(seen.insert(*id), "{} was written twice: {:?}", id, segs);
    }
    let by_flush = &segs[before_flush..];
    assert!(by_flush.len() > 2, "the flush did not split: {:?}", by_flush);
    let last_partial = segs[..before_flush].iter().filter_map(|id| id.part_id()).max().unwrap();
    assert!(by_flush.iter().filter_map(|id| id.part_id()).filter(|p| *p != 0).all(|p| p > last_partial),
        "the flush's parts do not come after the partials: {:?}", segs);
    let mut want = data.clone();
    want.extend_from_slice(&tail);
    let mut c = cold(&s).await;
    assert!(read_all(&mut c, want.len()).await == want, "read back wrong");
}

/// A dirty set larger than one object, written out as a partial when a threshold
/// is crossed, is several partials too: data-only ones, then the one with the
/// index. Every one stays within the limit, and the checkpoint reads back.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn a_partial_past_one_object_is_written_as_several() {
    let mut rt = runtime(CAP);
    rt.parted_segment_enabled = true;
    rt.data_cache_dirty_min_bytes_to_part = 8 * BS;
    rt.data_cache_dirty_max_blocks_threshold = 100;
    rt.data_cache_dirty_max_bytes_threshold = 100 * BS;
    let (s, mut f) = create("split-partial", rt).await;
    let log = Log::default();
    f.staging_interceptor(log.clone());
    let data = pattern(7, 101 * BS);
    f.write(0, &data).await.unwrap();
    // one more write finds the threshold crossed and writes the partial
    f.write(101 * BS, &pattern(8, BS)).await.unwrap();
    let partials: Vec<SegmentId> = log.segments();
    assert!(partials.len() > 3 && partials.iter().all(|id| id.part_id() != Some(0)),
        "101 blocks were not written as several partials: {:?}", partials);
    for id in &partials {
        assert!(s.segment_bytes(*id).unwrap().len() <= CAP, "{} is past the limit", id);
    }
    // the last partial carries the index, the ones before it none
    let hdr = |id: &SegmentId| { let b = s.segment_bytes(*id).unwrap(); SegmentHeader::from_slice(&b[..SegmentHeader::size()]) };
    assert!(hdr(partials.last().unwrap()).s_nmetablk > 0, "the last partial has no meta blocks");
    assert!(partials[..partials.len() - 1].iter().all(|id| hdr(id).s_nmetablk == 0), "a data partial has meta blocks");
    f.flush().await.expect("flush");
    let mut want = data.clone();
    want.extend_from_slice(&pattern(8, BS));
    let mut c = cold(&s).await;
    assert!(read_all(&mut c, want.len()).await == want, "read back wrong");
}

/// Blocks written by reference are laid out with the rest and read from the
/// source as each object uploads, whichever object they land in.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn references_are_split_across_objects_like_any_block() {
    let src = pattern(9, 250 * BS);
    let (s, mut f) = create("split-refs", runtime(CAP)).await;
    f.set_block_source(Some(Arc::new(Src(src.clone())))).unwrap();
    f.write_ref(0, 200 * BS).await.unwrap();
    // by value, in between and after
    f.write(50 * BS, &pattern(10, 10 * BS)).await.unwrap();
    f.write(200 * BS, &src[200 * BS..]).await.unwrap();
    let sid = f.flush().await.expect("flush");
    assert!(objects_of(&s, sid.as_cno()).len() > 5, "no split happened");
    let mut want = src.clone();
    want[50 * BS..60 * BS].copy_from_slice(&pattern(10, 10 * BS));
    let mut c = cold(&s).await;
    assert!(read_all(&mut c, want.len()).await == want, "read back wrong");
}

/// A split flush that fails part-way has written data parts but not part 0, so
/// there is no commit: a reopen takes nothing up and the inode is untouched. The
/// next flush publishes, and when the checkpoint was already part-way out as
/// partials — which keeps its number across attempts — it does not reuse a part
/// name the failed attempt wrote.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn a_split_flush_that_fails_part_way_leaves_no_commit_and_no_reused_name() {
    let mut rt = runtime(CAP);
    rt.parted_segment_enabled = true;
    rt.data_cache_dirty_min_bytes_to_part = 8 * BS;
    rt.data_cache_dirty_max_blocks_threshold = 10;
    rt.data_cache_dirty_max_bytes_threshold = 10 * BS;
    let (s, mut f) = create("split-fail", rt).await;
    f.write(0, b"before").await.unwrap();
    f.flush().await.unwrap();
    let inode_before = s.inode_bytes();
    let log = Log::default();
    f.staging_interceptor(log.clone());
    // a partial first, so the checkpoint keeps its number across attempts
    f.write(0, &pattern(11, 11 * BS)).await.unwrap();
    f.write(11 * BS, &pattern(12, BS)).await.unwrap();
    assert!(!log.segments().is_empty(), "no partial was written");
    // then enough for the flush to split, by reference so that no threshold
    // writes it out first, and the flush's third object refused
    let more = pattern(13, 80 * BS);
    let mut backing = vec![0u8; 12 * BS];
    backing.extend_from_slice(&more);
    f.set_block_source(Some(Arc::new(Src(backing)))).unwrap();
    f.write_ref(12 * BS, 80 * BS).await.unwrap();
    *log.refuse_segment.lock().unwrap() = Some(*log.seen.lock().unwrap() + 2);
    f.flush().await.expect_err("an object write is refused");
    assert_eq!(s.inode_bytes(), inode_before, "a failed flush wrote the inode");
    {
        let mut c = cold(&s).await;
        assert_eq!(read_all(&mut c, 6).await, b"before", "a reopen took up a checkpoint with no part 0");
    }
    f.flush().await.expect("the next flush publishes");
    let segs = log.segments();
    let mut seen = std::collections::HashSet::new();
    for id in &segs {
        assert!(seen.insert(*id), "{} was written twice: {:?}", id, segs);
    }
    let mut want = pattern(11, 11 * BS);
    want.extend_from_slice(&pattern(12, BS));
    want.extend_from_slice(&more);
    let mut c = cold(&s).await;
    assert!(read_all(&mut c, want.len()).await == want, "read back wrong");
}

/// Where a checkpoint cannot be split, a flush that needs more than one object is
/// refused before anything is written, and the file is left to try again: the
/// older formats name no second object, and a checkpoint whose index alone does
/// not fit one object is refused under any format.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn what_cannot_be_split_is_refused_before_anything_is_written() {
    // MicroGroup: a checkpoint is one object
    let (s, mut f) = create_fmt("split-refuse-mg", runtime(CAP), BlockPtrFormat::MicroGroup).await;
    f.write(0, &pattern(14, 100 * BS)).await.unwrap();
    let before = s.segment_count();
    let e = f.flush().await.expect_err("MicroGroup cannot split");
    assert_eq!(e.kind(), ErrorKind::InvalidInput, "{}", e);
    assert_eq!(s.segment_count(), before, "something was written");

    // an index larger than one object, at the smallest limit
    let (s, mut f) = create("split-refuse-meta", runtime(16 * 1024)).await;
    f.write(0, &pattern(15, 3000 * BS)).await.unwrap();
    let before = s.segment_count();
    let e = f.flush().await.expect_err("the index does not fit 16 KiB");
    assert_eq!(e.kind(), ErrorKind::InvalidInput, "{}", e);
    assert!(e.to_string().contains("index"), "{}", e);
    assert_eq!(s.segment_count(), before, "something was written");
    let mut c = cold(&s).await;
    assert_eq!(read_all(&mut c, BS).await.len(), 0, "a refused flush published");
}

/// With the WAL, which keeps a segment in one buffer for replay, a checkpoint is
/// always one object, and one that would need more is refused.
#[cfg(feature = "wal")]
#[tokio::test]
async fn with_the_wal_a_checkpoint_past_one_object_is_refused() {
    let (s, mut f) = create("split-wal", runtime(CAP)).await;
    f.write(0, &pattern(16, 100 * BS)).await.unwrap();
    let before = s.segment_count();
    let e = f.flush().await.expect_err("the WAL build cannot split");
    assert_eq!(e.kind(), ErrorKind::InvalidInput, "{}", e);
    assert!(e.to_string().contains("WAL"), "{}", e);
    assert_eq!(s.segment_count(), before, "something was written");
}