hyperfile 0.11.3

#![cfg(not(any(feature = "blocking", feature = "wal")))]
//! Objects sent ahead under the stream format stay within the size one object
//! may have, whatever `stream_segment_bytes` is; and a panic in a task sending
//! one is not lost. On memory staging.
//!
//! The limit is `segment_max_object_bytes`, and never more than the 1 GiB a
//! pointer can address in one object. The data, the leaves over it, the summary
//! and the header all count.
//!
//! ```bash
//! cargo test --release --test functional_stream_object_limit
//! ```

use btree_ondisk::{MemoryBlockLoader, NullNodeCache};
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::staging::config::StagingConfig;
use hyperfile::staging::memory::MemoryStaging;
use hyperfile::staging::{Staging, StagingIntercept};
use hyperfile::SegmentId;
use std::io::Result;

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

const BS: usize = 4096;
const MIB: usize = 1 << 20;
const FMT: BlockPtrFormat = BlockPtrFormat::StreamSegment;

/// Sending ahead at `target`, in `piece`s, objects at most `limit`.
fn runtime(target: usize, piece: usize, limit: usize) -> 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 = 1 << 40;
    rt.segment_max_object_bytes = limit;
    rt.data_cache_blocks = 0;
    rt.stream_segment_bytes = target;
    rt.segment_mpu_chunk_size = piece;
    rt
}

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

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

async fn reopen(s: &MemoryStaging, rt: &HyperFileRuntimeConfig) -> MemFile<'static> {
    MemFile::open(s.clone(), s.to_block_loader(), NullNodeCache, config(s, rt), HyperFileFlags::from_flags(FileFlags::rdonly()))
        .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()
}

/// Every stored object's size, part 0 and the inode's checkpoint included.
fn largest_object(s: &MemoryStaging) -> (SegmentId, usize) {
    s.segment_ids().into_iter().map(|id| (id, s.segment_bytes(id).unwrap().len()))
        .max_by_key(|(_, n)| *n).unwrap()
}

/// Reads back `want(b)` for every block `b` in `blocks`, cold.
async fn check_cold(s: &MemoryStaging, rt: &HyperFileRuntimeConfig, name: &str, blocks: impl Iterator<Item = usize>, want: impl Fn(usize) -> Vec<u8>) {
    let mut g = reopen(s, rt).await;
    let mut v = vec![0u8; BS];
    for b in blocks {
        g.read(b * BS, &mut v).await.unwrap();
        assert!(v == want(b), "{}: block {} reads back wrong", name, b);
    }
}

/// Written in order, more than the limit: no object is larger than it, filled in
/// pieces or sent whole, none had to be given up, and the data reads back.
#[tokio::test]
async fn objects_sent_ahead_stay_within_the_limit() {
    for (name, target, piece, limit) in [
        ("ol-pieces", 16 * MIB, 5 * MIB, 16 * MIB),
        ("ol-pieces-under", 16 * MIB, 5 * MIB, 18 * MIB),
        ("ol-whole", 5 * MIB, 5 * MIB, 5 * MIB),
    ] {
        let rt = runtime(target, piece, limit);
        let (s, mut f) = create(name, &rt).await;
        let n = 64 * MIB / BS;
        for b in 0..n {
            f.write(b * BS, &block(b, 1)).await.unwrap();
            if b % 256 == 0 { tokio::task::yield_now().await; }
        }
        f.flush().await.expect("flush");
        assert_eq!(f.flush_timing().snapshot().stream_dead_bytes, 0, "{}: an object was given up", name);
        assert!(s.segment_ids().len() > 3, "{}: nothing was sent ahead", name);
        let (id, len) = largest_object(&s);
        assert!(len <= limit, "{}: object {} is {} bytes, past the limit of {}", name, id, len, limit);
        check_cold(&s, &rt, name, 0..n, |b| block(b, 1)).await;
    }
}

/// Blocks scattered one or two to a leaf: the leaves weigh as much as the data,
/// and still no object is past the limit, nor given up.
#[tokio::test]
async fn scattered_blocks_and_their_leaves_stay_within_the_limit() {
    for (name, target, piece, limit) in [
        ("ol-sparse-whole", 2 * MIB, 5 * MIB, 2 * MIB),
        ("ol-sparse-pieces", 8 * MIB, 5 * MIB, 8 * MIB),
    ] {
        let rt = runtime(0, piece, limit);
        let (s, mut f) = create(name, &rt).await;
        let n = 256 * MIB / BS;
        for b in 0..n {
            f.write(b * BS, &block(b, 1)).await.unwrap();
        }
        f.flush().await.expect("publish the base");
        drop(f);
        let rt = runtime(target, piece, limit);
        let mut f = MemFile::open(s.clone(), s.to_block_loader(), NullNodeCache, config(&s, &rt),
            HyperFileFlags::from_flags(FileFlags::rdwr())).await.unwrap();
        let before: std::collections::HashSet<SegmentId> = s.segment_ids().into_iter().collect();
        for b in (0..n).step_by(24) {
            f.write(b * BS, &block(b, 2)).await.unwrap();
            tokio::task::yield_now().await;
        }
        f.flush().await.expect("flush");
        assert_eq!(f.flush_timing().snapshot().stream_dead_bytes, 0, "{}: an object was given up", name);
        let sent: Vec<(SegmentId, usize)> = s.segment_ids().into_iter().filter(|id| !before.contains(id))
            .map(|id| (id, s.segment_bytes(id).unwrap().len())).collect();
        assert!(sent.len() > 2, "{}: nothing was sent ahead: {:?}", name, sent);
        for (id, len) in sent.iter() {
            assert!(*len <= limit, "{}: object {} is {} bytes, past the limit of {}", name, id, len, limit);
        }
        check_cold(&s, &rt, name, 0..n, |b| block(b, if b % 24 == 0 { 2 } else { 1 })).await;
    }
}

/// An object left open close to the limit whose leaves then split: zeroed whole
/// between its blocks, which adds entries to the map and sends nothing. Closing
/// it would pass the limit, so it is given up, and its blocks are written again
/// elsewhere.
#[tokio::test]
async fn an_object_whose_leaves_outgrew_it_is_given_up() {
    let name = "ol-give-up";
    let rt = runtime(16 * MIB, 5 * MIB, 16 * MIB);
    let (s, mut f) = create(name, &rt).await;
    // Three pieces, every 16th block: the object is open at about 15 MiB.
    let placed = 3 * 5 * MIB / BS;
    for k in 0..placed {
        f.write(k * 16 * BS, &block(k * 16, 1)).await.unwrap();
        tokio::task::yield_now().await;
    }
    // The gaps, zeroed: the leaves over the placed blocks split many times.
    for k in 0..placed {
        f.write_zero((k * 16 + 1) * BS, 15 * BS).await.unwrap();
    }
    f.flush_timing().reset();
    f.flush().await.expect("flush");
    let dead = f.flush_timing().snapshot().stream_dead_bytes;
    let (id, len) = largest_object(&s);
    assert!(len <= 16 * MIB, "{}: object {} is {} bytes, past the limit", name, id, len);
    assert_eq!(dead, (placed * BS) as u64, "{}: the open object was not given up", name);
    check_cold(&s, &rt, name, 0..placed * 16, |b| if b % 16 == 0 { block(b, 1) } else { vec![0u8; BS] }).await;
}

/// Panics on completing a data part.
#[derive(Clone)]
struct PanicOnPart;

impl StagingIntercept<MemoryStaging> for PanicOnPart {
    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 part = segid.part_id().is_some_and(|p| p > 0);
        Box::pin(async move {
            if part { panic!("a send-ahead task panicked (the test's)"); }
            Ok(())
        })
    }
}

/// A panic in a task sending an object ahead reaches the caller, as if it had
/// happened in its own call, in both ways of sending ahead.
async fn a_panic_sending_ahead_reaches_the_caller(target: usize) {
    let rt = runtime(target, 5 * MIB, 1 << 30);
    let (_s, mut f) = create("ol-panic", &rt).await;
    f.staging_interceptor(PanicOnPart);
    for b in 0..(32 * MIB / BS) {
        f.write(b * BS, &block(b, 1)).await.unwrap();
        tokio::task::yield_now().await;
    }
    f.flush().await.unwrap();
}

#[tokio::test]
#[should_panic(expected = "a send-ahead task panicked (the test's)")]
async fn a_panic_sending_whole_objects_reaches_the_caller() {
    a_panic_sending_ahead_reaches_the_caller(5 * MIB).await;
}

#[tokio::test]
#[should_panic(expected = "a send-ahead task panicked (the test's)")]
async fn a_panic_sending_pieces_reaches_the_caller() {
    a_panic_sending_ahead_reaches_the_caller(16 * MIB).await;
}

/// What was reported: a checkpoint past 1 GiB with the target at or near the
/// 1 GiB one object can address, in 16 MiB and 5 MiB pieces, written as a
/// program would, from a multi-threaded runtime.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_checkpoint_past_one_gib_with_the_target_at_the_limit() {
    for (target, piece) in [(1024 * MIB, 16 * MIB), (1023 * MIB, 16 * MIB), (1000 * MIB, 16 * MIB), (1024 * MIB, 5 * MIB)] {
        let name = format!("ol-gib-{}-{}", target / MIB, piece / MIB);
        let rt = runtime(target, piece, 1 << 30);
        let (s, mut f) = create(&name, &rt).await;
        let chunk: Vec<u8> = (0..16 * MIB).map(|i| (i.wrapping_mul(2654435761) >> 13) as u8).collect();
        let total = 1100 * MIB;
        let mut off = 0;
        while off < total {
            f.write(off, &chunk).await.unwrap();
            off += chunk.len();
        }
        f.flush().await.expect("flush");
        drop(f);
        let (id, len) = largest_object(&s);
        assert!(len <= 1 << 30, "{}: object {} is {} bytes", name, id, len);
        let mut g = reopen(&s, &rt).await;
        let mut v = vec![0u8; chunk.len()];
        let mut off = 0;
        while off < total {
            assert_eq!(g.read(off, &mut v).await.unwrap(), v.len());
            assert!(v == chunk, "{}: 16 MiB at {} reads back wrong", name, off);
            off += chunk.len();
        }
    }
}