hyperfile 0.11.2

#![cfg(not(feature = "blocking"))]
// The WAL build sends nothing ahead and refuses relocate; what it leaves unused
// is for those tests.
#![cfg_attr(feature = "wal", allow(dead_code, unused_imports))]
//! The stream segment format (`BlockPtrFormat::StreamSegment`), on memory staging.
//!
//! Every object is laid out tail first and describes itself; a checkpoint larger
//! than one object is written as data parts, each with the leaves over its data,
//! and part 0 with the rest of the index and the inode.
//!
//! ```bash
//! cargo test --test functional_stream_segment
//! ```

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::meta_format::BlockPtrFormat;
use hyperfile::segment::SegmentReadWrite;
use hyperfile::staging::config::StagingConfig;
use hyperfile::staging::memory::MemoryStaging;
use hyperfile::staging::Staging;
use hyperfile::stream_segment;
use hyperfile::inode::FlushInodeFlag;
use hyperfile::ondisk::InodeRaw;
use hyperfile::staging::StagingIntercept;
use hyperfile::SegmentId;
use std::io::{Error, ErrorKind, Result};
use std::sync::{Arc, Mutex};

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

const BS: usize = 4096;
const FMT: BlockPtrFormat = BlockPtrFormat::StreamSegment;

fn runtime(part_bytes: 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 << 30;
    rt.segment_max_object_bytes = part_bytes;
    rt.data_cache_blocks = 0;
    // What the flush itself lays out; the tests of sending ahead turn it on.
    rt.stream_segment_bytes = 0;
    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()).with_block_ptr_format(FMT);
    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> {
    // A fresh loader: what is read comes from the stored objects.
    MemFile::open(s.clone(), s.to_block_loader(), NullNodeCache, config(s, rt), 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(f: &mut MemFile<'static>, off: usize, len: usize) -> Vec<u8> {
    let mut v = vec![0u8; len];
    let n = f.read(off, &mut v).await.expect("read");
    v.truncate(n);
    v
}

/// Each object parses from its own end, names itself, and its data blocks are
/// where its summary says.
fn check_objects(s: &MemoryStaging) -> usize {
    let mut n = 0;
    for id in s.segment_ids() {
        let bytes = s.segment_bytes(id).unwrap();
        let (layout, sum) = stream_segment::parse_tail(id, bytes.len(), &bytes)
            .unwrap_or_else(|e| panic!("{}: {}", id, e)).unwrap_or_else(|need| panic!("{}: short by {}", id, need));
        assert_eq!(layout.hdr.object, id);
        assert_eq!(sum.blocks.len(), layout.hdr.ndata as usize);
        // Every data block of the object is named by a leaf in the same object, at
        // the place its summary gives: the object describes itself.
        let mut in_leaves = std::collections::HashMap::new();
        for k in 0..layout.nmeta() {
            let at = layout.meta_off + k * layout.meta_block_size();
            let node = btree_ondisk::node::BtreeNode::<u64, u64, u64>::from_slice_ref(&bytes[at..at + layout.meta_block_size()]).unwrap();
            if node.get_level() == 1 {
                for c in 0..node.get_nchild() {
                    in_leaves.insert(*node.get_key(c), *node.get_val(c));
                }
            }
        }
        if layout.nmeta() > 0 || id.part_id() != Some(0) {
            // A slot whose block was written again after it was placed is dead,
            // and says so.
            for b in sum.blocks.iter().filter(|b| b.blkptr() != 0) {
                assert_eq!(in_leaves.get(&b.blkidx()), Some(&b.blkptr()), "{}: block {} has no leaf in its object", id, b.blkidx());
            }
        }
        n += 1;
    }
    n
}

/// A small checkpoint is one object, part 0, and reads back cold.
#[tokio::test]
async fn a_small_checkpoint_is_one_object_and_reads_back() {
    let rt = runtime(1 << 30);
    let (s, mut f) = create("ss-small", rt.clone()).await;
    for b in 0..20usize {
        f.write(b * BS, &block(b, 1)).await.unwrap();
    }
    let cno = f.flush().await.expect("flush").as_cno();
    assert_eq!(s.segment_ids(), vec![hyperfile::SegmentId::new_from_cno(cno).at_part(0)]);
    assert_eq!(check_objects(&s), 1);
    let mut g = reopen(&s, &rt).await;
    for b in 0..20usize {
        assert_eq!(read(&mut g, b * BS, BS).await, block(b, 1), "block {}", b);
    }
}

/// A checkpoint over many leaves, with objects capped near one leaf's worth, is
/// several data parts and part 0; every object describes itself, and the file
/// reads back cold — after a second checkpoint over part of it, too.
#[tokio::test]
async fn a_large_checkpoint_is_parts_that_each_describe_themselves() {
    let rt = runtime(2 << 20);
    let (s, mut f) = create("ss-large", rt.clone()).await;
    let blocks = 4096usize;
    for b in 0..blocks {
        f.write(b * BS, &block(b, 1)).await.unwrap();
    }
    let cno = f.flush().await.expect("flush").as_cno();
    let ids = s.segment_ids();
    assert!(ids.len() >= 8, "{} objects for 16 MiB in 2 MiB objects", ids.len());
    assert!(ids.iter().all(|i| i.as_cno() == cno));
    assert_eq!(check_objects(&s), ids.len());
    for id in &ids {
        assert!(s.segment_bytes(*id).unwrap().len() <= 2 << 20, "{} is {} bytes", id, s.segment_bytes(*id).unwrap().len());
    }
    for b in (0..blocks).step_by(97) {
        f.write(b * BS, &block(b, 2)).await.unwrap();
    }
    f.flush().await.expect("second flush");
    assert_eq!(check_objects(&s), s.segment_ids().len());
    let mut g = reopen(&s, &rt).await;
    for b in 0..blocks {
        let want = block(b, if b % 97 == 0 { 2 } else { 1 });
        assert_eq!(read(&mut g, b * BS, BS).await, want, "block {}", b);
    }
}

/// The summary any object's end holds is what `open` answers, whatever the
/// caller knows of the layout.
#[tokio::test]
async fn open_reads_a_summary_from_the_objects_end() {
    let rt = runtime(2 << 20);
    let (s, mut f) = create("ss-open", rt).await;
    for b in 0..2000usize {
        f.write(b * BS, &block(b, 1)).await.unwrap();
    }
    f.flush().await.unwrap();
    let mut total = 0;
    for id in s.segment_ids() {
        total += s.open(id).await.unwrap_or_else(|e| panic!("{}: {}", id, e)).blocks.len();
    }
    assert_eq!(total, 2000);
}

/// Refuses the next `n` part-0 writes with `ResourceBusy`, which a flush retries,
/// and every inode write while `inode_down`.
#[derive(Clone, Default)]
struct Refuse {
    part0: Arc<Mutex<usize>>,
    /// Data-part objects to refuse.
    data: Arc<Mutex<usize>>,
    inode_down: Arc<Mutex<bool>>,
}

impl StagingIntercept<MemoryStaging> for Refuse {
    fn before_flush_inode(&self, _: &MemoryStaging, _: &[u8], _: FlushInodeFlag)
        -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + '_ + Send>> {
        let down = *self.inode_down.lock().unwrap();
        Box::pin(async move { if down { Err(Error::new(ErrorKind::ConnectionReset, "inode refused")) } else { 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 counter = if segid.part_id() == Some(0) { &self.part0 } else { &self.data };
        let refuse = {
            let mut n = counter.lock().unwrap();
            if *n > 0 { *n -= 1; true } else { false }
        };
        Box::pin(async move { if refuse { Err(Error::new(ErrorKind::ResourceBusy, "part 0 refused")) } else { Ok(()) } })
    }
}

/// Past the dirty-data limits the writes go out as data parts, each its leaves
/// and its data, without publishing; the flush then publishes one checkpoint
/// under the number the partials took, and the file reads back cold.
#[tokio::test]
async fn partials_then_a_commit_are_one_checkpoint() {
    let mut rt = runtime(2 << 20);
    // Memory pressure only: what is sent ahead has its own test.
    rt.parted_segment_enabled = true;
    rt.data_cache_dirty_max_bytes_threshold = 3 << 20;
    rt.data_cache_dirty_max_blocks_threshold = (3 << 20) / BS;
    rt.data_cache_dirty_min_bytes_to_part = 2 << 20;
    let (s, mut f) = create("ss-partials", rt.clone()).await;
    let cno0 = f.last_cno();
    let blocks = 3000usize;
    let mut most = 0;
    for b in 0..blocks {
        f.write(b * BS, &block(b, 3)).await.unwrap();
        most = most.max(f.get_data_blocks_dirty().len());
    }
    let before = s.segment_ids();
    assert!(!before.is_empty(), "no partials were written");
    assert!(before.iter().all(|i| i.part_id().is_some_and(|p| p > 0)), "a partial was part 0: {:?}", before);
    assert_eq!(f.last_cno(), cno0, "the partials published");
    assert!(most * BS <= (3 << 20) + BS, "{} blocks dirty at once", most);
    let cno = f.flush().await.expect("flush").as_cno();
    assert_eq!(cno, cno0 + 1);
    assert!(s.segment_ids().iter().all(|i| i.as_cno() == cno), "the partials and the checkpoint have different numbers");
    assert_eq!(check_objects(&s), s.segment_ids().len());
    let mut g = reopen(&s, &rt).await;
    for b in 0..blocks {
        assert_eq!(read(&mut g, b * BS, BS).await, block(b, 3), "block {}", b);
    }
}

/// A checkpoint whose part 0 is refused once is written again under the same
/// number, its data parts under fresh part numbers, and published.
#[tokio::test]
async fn a_retry_keeps_the_number_and_takes_fresh_parts() {
    let mut rt = runtime(2 << 20);
    // The flush's own retry: nothing sent ahead, so every data part is the flush's.
    rt.stream_segment_bytes = 0;
    let data_parts = |s: &MemoryStaging| s.segment_ids().iter().filter(|i| i.part_id().is_some_and(|p| p > 0)).count();
    // The same writes, unrefused: how many data parts one attempt writes.
    let (s1, mut f1) = create("ss-retry-plain", rt.clone()).await;
    for b in 0..1500usize {
        f1.write(b * BS, &block(b, 4)).await.unwrap();
    }
    f1.flush().await.unwrap();
    let per_attempt = data_parts(&s1);
    assert!(per_attempt >= 2);

    let (s, mut f) = create("ss-retry", rt.clone()).await;
    let refuse = Refuse::default();
    f.staging_interceptor(refuse.clone());
    for b in 0..1500usize {
        f.write(b * BS, &block(b, 4)).await.unwrap();
    }
    let number = f.inode().get_last_seq().as_cno() + 1;
    *refuse.part0.lock().unwrap() = 1;
    let cno = f.flush().await.expect("flush, retried").as_cno();
    assert_eq!(cno, number, "the retry took another number");
    let ids = s.segment_ids();
    assert!(ids.iter().all(|i| i.as_cno() == cno), "{:?}", ids);
    assert_eq!(data_parts(&s), 2 * per_attempt, "the retry did not write its data parts under fresh part numbers: {:?}", ids);
    assert_eq!(ids.iter().filter(|i| i.part_id() == Some(0)).count(), 1);
    let mut g = reopen(&s, &rt).await;
    for b in (0..1500usize).step_by(7) {
        assert_eq!(read(&mut g, b * BS, BS).await, block(b, 4));
    }
}

/// A checkpoint whose part 0 landed and whose inode did not is taken up by the
/// next open: its header is found at its end.
#[tokio::test]
async fn a_checkpoint_without_its_inode_is_taken_up() {
    let rt = runtime(2 << 20);
    let (s, mut f) = create("ss-takeup", rt.clone()).await;
    let refuse = Refuse::default();
    f.staging_interceptor(refuse.clone());
    f.write(0, &block(0, 1)).await.unwrap();
    f.flush().await.unwrap();
    for b in 0..800usize {
        f.write(b * BS, &block(b, 5)).await.unwrap();
    }
    *refuse.inode_down.lock().unwrap() = true;
    let e = f.flush().await.expect_err("the inode write is refused");
    let pending = hyperfile::file::PublishIncomplete::of(&e).expect("says what it left");
    drop(f);
    let stored = InodeRaw::from_u8_slice(&s.inode_bytes().unwrap());
    assert!(stored.i_last_cno < pending.as_cno());
    let mut g = reopen(&s, &rt).await;
    assert_eq!(g.last_cno(), pending.as_cno(), "the checkpoint was not taken up");
    for b in (0..800usize).step_by(11) {
        assert_eq!(read(&mut g, b * BS, BS).await, block(b, 5));
    }
}

/// Moving blocks with `relocate` under this format still marks the checkpoint.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn a_relocation_is_marked() {
    let rt = runtime(2 << 20);
    let (s, mut f) = create("ss-relocate", rt.clone()).await;
    for b in 0..600usize {
        f.write(b * BS, &block(b, 6)).await.unwrap();
    }
    f.flush().await.unwrap();
    let content = f.content_cno().unwrap();
    assert_eq!(f.relocate(&[(0, 600)]).await.unwrap(), 600);
    f.flush().await.unwrap();
    let stored = InodeRaw::from_u8_slice(&s.inode_bytes().unwrap());
    assert!(stored.is_compacted());
    assert_eq!(f.content_cno().unwrap(), content);
    let mut g = reopen(&s, &rt).await;
    for b in (0..600usize).step_by(13) {
        assert_eq!(read(&mut g, b * BS, BS).await, block(b, 6));
    }
}

/// Objects sent ahead of `target` bytes, in 5 MiB pieces: filled a piece at a
/// time above one piece, sent whole at or below.
fn ahead(target: usize) -> HyperFileRuntimeConfig {
    let mut rt = runtime(1 << 30);
    rt.stream_segment_bytes = target;
    rt.segment_mpu_chunk_size = 5 << 20;
    rt
}

/// Data is sent ahead while it is written, before any flush, its blocks leaving
/// the dirty tier as each object completes; the flush writes what was not sent
/// and publishes one checkpoint under the number they took. Above one piece an
/// object is filled a piece at a time and closed near the target, 16 MiB here; at
/// one piece, 5 MiB, each is sent whole.
async fn sent_ahead(name: &str, target: usize) {
    let rt = ahead(target);
    let (s, mut f) = create(name, rt.clone()).await;
    let cno0 = f.last_cno();
    let blocks = 10 * 1024usize; // 40 MiB
    for b in 0..blocks {
        f.write(b * BS, &block(b, 7)).await.unwrap();
        tokio::task::yield_now().await;
    }
    let sent = s.segment_ids();
    let dirty = f.get_data_blocks_dirty().len();
    assert!(sent.len() >= (40 << 20) / target - 1, "{} objects of 40 MiB at {} bytes", sent.len(), target);
    assert!(sent.iter().all(|i| i.part_id().is_some_and(|p| p > 0)), "{:?}", sent);
    assert_eq!(f.last_cno(), cno0, "sending ahead published");
    assert!(dirty * BS <= target + (8 << 20), "{} blocks still dirty", dirty);
    for id in &sent {
        let len = s.segment_bytes(*id).unwrap().len();
        assert!(len >= target && len <= target + (6 << 20), "{} is {} bytes, not about {}", id, len, target);
    }
    // One piece or less: sent whole, no upload in pieces. More: each object one.
    if target <= rt.segment_mpu_chunk_size {
        assert_eq!(s.uploads_opened(), 0, "an object of one piece was uploaded in pieces");
    } else {
        assert!(s.uploads_opened() >= sent.len(), "{} uploads for {} objects", s.uploads_opened(), sent.len());
    }
    let cno = f.flush().await.expect("flush").as_cno();
    assert_eq!(cno, cno0 + 1);
    assert!(s.segment_ids().iter().all(|i| i.as_cno() == cno));
    assert_eq!(check_objects(&s), s.segment_ids().len());
    let mut g = reopen(&s, &rt).await;
    for b in 0..blocks {
        assert_eq!(read(&mut g, b * BS, BS).await, block(b, 7), "block {}", b);
    }
}

#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn objects_filled_in_pieces_are_closed_near_the_target() {
    sent_ahead("ss-ahead-pieces", 16 << 20).await;
}

#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn objects_of_one_piece_are_sent_whole() {
    sent_ahead("ss-ahead-whole", 5 << 20).await;
}

/// Holds every data-part object'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(()) })
    }
}

/// A block written again while the object holding its first bytes completes
/// stays dirty, and the flush writes the newer bytes; its slot in that object is
/// marked dead. One written again to the same bytes may leave the dirty tier.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn a_block_written_again_while_its_object_completes_keeps_its_newer_bytes() {
    for (name, target) in [("ss-rewrite-whole", 5 << 20), ("ss-rewrite-pieces", 10 << 20)] {
        rewritten_while_completing(name, target).await;
    }
}

async fn rewritten_while_completing(name: &str, target: usize) {
    let rt = ahead(target);
    let (s, mut f) = create(name, rt.clone()).await;
    f.staging_interceptor(Slow(300));
    let n = target / BS + 256;
    for b in 0..n {
        f.write(b * BS, &block(b, 1)).await.unwrap();
    }
    for b in 0..64usize {
        f.write(b * BS, &block(b, 9)).await.unwrap();
    }
    for b in 64..128usize {
        f.write(b * BS, &block(b, 1)).await.unwrap();
    }
    tokio::time::sleep(std::time::Duration::from_millis(700)).await;
    // Completed now; this write takes it in.
    f.write(4000 * BS, &block(4000, 1)).await.unwrap();
    assert!(!s.segment_ids().is_empty(), "{}: nothing was sent ahead", name);
    let dirty: std::collections::HashSet<u64> = f.get_data_blocks_dirty().data().keys().copied().collect();
    assert!((0..64u64).all(|b| dirty.contains(&b)), "{}: a block written again in flight left the dirty tier", name);
    f.flush().await.expect("flush");
    assert_eq!(check_objects(&s), s.segment_ids().len());
    let mut g = reopen(&s, &rt).await;
    for b in 0..n {
        let want = block(b, if b < 64 { 9 } else { 1 });
        assert_eq!(read(&mut g, b * BS, BS).await, want, "block {}", b);
    }
    assert_eq!(read(&mut g, 4000 * BS, BS).await, block(4000, 1));
}

/// An object that cannot complete leaves its blocks dirty: the flush writes
/// them, and nothing is lost.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn an_object_that_fails_to_complete_loses_nothing() {
    for (name, target) in [("ss-fail-whole", 5 << 20), ("ss-fail-pieces", 10 << 20)] {
        fails_to_complete(name, target).await;
    }
}

async fn fails_to_complete(name: &str, target: usize) {
    let rt = ahead(target);
    let (s, mut f) = create(name, rt.clone()).await;
    let refuse = Refuse::default();
    f.staging_interceptor(refuse.clone());
    *refuse.data.lock().unwrap() = 1;
    let n = target / BS + 256;
    for b in 0..n {
        f.write(b * BS, &block(b, 3)).await.unwrap();
        tokio::task::yield_now().await;
    }
    f.write(4000 * BS, &block(4000, 3)).await.unwrap();
    f.flush().await.expect("flush");
    assert!(*refuse.data.lock().unwrap() == 0, "{}: the test did not refuse an object", name);
    let mut g = reopen(&s, &rt).await;
    for b in (0..n).step_by(5) {
        assert_eq!(read(&mut g, b * BS, BS).await, block(b, 3), "block {}", b);
    }
}

/// Nothing is sent ahead with `stream_segment_bytes` zero: the data goes out at the flush.
#[tokio::test]
async fn nothing_is_sent_ahead_when_off() {
    let rt = runtime(1 << 30);
    let (s, mut f) = create("ss-noahead", rt.clone()).await;
    for b in 0..4096usize {
        f.write(b * BS, &block(b, 2)).await.unwrap();
        tokio::task::yield_now().await;
    }
    assert!(s.segment_ids().is_empty());
    f.flush().await.unwrap();
    let mut g = reopen(&s, &rt).await;
    for b in (0..4096usize).step_by(31) {
        assert_eq!(read(&mut g, b * BS, BS).await, block(b, 2));
    }
}

fn dead_slots(s: &MemoryStaging) -> usize {
    s.segment_ids().iter().map(|id| {
        let b = s.segment_bytes(*id).unwrap();
        let (_, sum) = stream_segment::parse_tail(*id, b.len(), &b).unwrap().unwrap();
        sum.blocks.iter().filter(|e| e.blkptr() == 0).count()
    }).sum()
}

/// Every way of changing a block, while the object holding its first bytes is in
/// flight: through the block API, by `write_zero` whole and in part, by a shrinking
/// `truncate` (which also publishes) and an extend past it. Cold, the file is what
/// those changes made it, and nothing cut by the truncate comes back.
async fn changed_every_way_in_flight(name: &str, target: usize) {
    let rt = ahead(target);
    let (s, mut f) = create(name, rt.clone()).await;
    f.staging_interceptor(Slow(300));
    let mut model: Vec<u8> = Vec::new();
    let 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);
    };
    let n = target / BS + 256;
    let piece = rt.segment_mpu_chunk_size / BS;
    // Past the first piece, then a change: in pieces, its object is still open.
    for b in 0..piece + 16 {
        f.write(b * BS, &block(b, 1)).await.unwrap();
        put(&mut model, b * BS, &block(b, 1));
    }
    for b in 10..20usize {
        let mut m = f.block_mut(b as u64, false).await.unwrap().expect("a written block");
        m.as_mut_slice().copy_from_slice(&block(b, 9));
        drop(m);
        put(&mut model, b * BS, &block(b, 9));
    }
    for b in piece + 16..n {
        f.write(b * BS, &block(b, 1)).await.unwrap();
        put(&mut model, b * BS, &block(b, 1));
    }
    // Closed now, completing slowly.
    f.write_zero(100 * BS, 10 * BS).await.unwrap();
    put(&mut model, 100 * BS, &vec![0u8; 10 * BS]);
    f.write_zero(200 * BS + 100, 1000).await.unwrap();
    put(&mut model, 200 * BS + 100, &vec![0u8; 1000]);
    // Changed in place: still dirty. Zeroed whole: out of the cache, the map naming
    // the zero block — so not where it was sent either way.
    let dirty: std::collections::HashSet<u64> = f.get_data_blocks_dirty().data().keys().copied().collect();
    for b in (10..20u64).chain([200]) {
        assert!(dirty.contains(&b), "{}: block {} changed in flight is not dirty", name, b);
    }
    for b in 100..110u64 {
        assert!(!dirty.contains(&b), "{}: block {} zeroed whole is still dirty", name, b);
    }
    // Cut through blocks in flight and blocks not sent, then extend past the cut.
    let cut = 700 * BS + 123;
    f.truncate(cut).await.expect("truncate");
    model.truncate(cut);
    f.write(900 * BS, &block(900, 4)).await.unwrap();
    put(&mut model, 900 * BS, &block(900, 4));
    tokio::time::sleep(std::time::Duration::from_millis(700)).await;
    f.flush().await.expect("flush");
    assert_eq!(check_objects(&s), s.segment_ids().len());
    if target > rt.segment_mpu_chunk_size {
        assert!(dead_slots(&s) >= 10, "{}: blocks changed before their object closed left no dead slot", name);
    }
    let mut g = reopen(&s, &rt).await;
    let got = read(&mut g, 0, model.len() + BS).await;
    assert_eq!(got.len(), model.len(), "{}: size", name);
    let first_bad = got.iter().zip(model.iter()).position(|(a, b)| a != b);
    assert!(first_bad.is_none(), "{}: differs at byte {:?} (block {:?})", name, first_bad, first_bad.map(|o| o / BS));
}

#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn every_way_of_changing_a_block_in_flight_is_kept() {
    changed_every_way_in_flight("ss-change-whole", 5 << 20).await;
    changed_every_way_in_flight("ss-change-pieces", 10 << 20).await;
}

/// Leaves made clean by objects that completed, dropped from memory, are read
/// again from those objects — before the checkpoint is published, and after.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn clean_leaves_are_read_again_from_the_objects_they_went_to() {
    let mut rt = ahead(5 << 20);
    rt.node_cache_blocks = 1;
    let (s, mut f) = create("ss-leaves", rt.clone()).await;
    // Held back, so every leaf is dirty when counted.
    f.staging_interceptor(Slow(300));
    let n = 4 * 1280 + 300;
    for b in 0..n {
        f.write(b * BS, &block(b, 2)).await.unwrap();
    }
    let dirty_nodes_before = f.bmap_lookup_dirty().len();
    // Let everything sent complete, and take it in.
    tokio::time::sleep(std::time::Duration::from_millis(700)).await;
    f.write(n * BS, &block(n, 2)).await.unwrap();
    let dirty_nodes_after = f.bmap_lookup_dirty().len();
    assert!(dirty_nodes_after < dirty_nodes_before,
        "no leaf was made clean: {} dirty nodes before, {} after", dirty_nodes_before, dirty_nodes_after);
    // Clean nodes are dropped when the map's own dirty state is cleared or its
    // limit set, not as soon as one is cleaned: set it, so they go now.
    f.bmap_set_cache_limit(1);
    for b in 0..=n {
        assert_eq!(read(&mut f, b * BS, BS).await, block(b, 2), "block {} before the flush", b);
    }
    f.flush().await.expect("flush");
    let mut g = reopen(&s, &rt).await;
    for b in 0..=n {
        assert_eq!(read(&mut g, b * BS, BS).await, block(b, 2), "block {} cold", b);
    }
}

/// What is sent ahead and what of it dies are counted: blocks written again while
/// their object completes are dead bytes, and are sent again.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn bytes_sent_ahead_and_dead_are_counted() {
    let rt = ahead(5 << 20);
    let (_s, mut f) = create("ss-count", rt.clone()).await;
    f.staging_interceptor(Slow(300));
    f.flush_timing().reset();
    for b in 0..1536usize {
        f.write(b * BS, &block(b, 1)).await.unwrap();
    }
    for b in 0..64usize {
        f.write(b * BS, &block(b, 9)).await.unwrap();
    }
    tokio::time::sleep(std::time::Duration::from_millis(700)).await;
    f.write(4000 * BS, &block(4000, 1)).await.unwrap();
    let t = f.flush_timing().snapshot();
    assert_eq!(t.stream_sent_bytes, (1280 * BS) as u64, "one object of 5 MiB was sent");
    assert_eq!(t.stream_dead_bytes, (64 * BS) as u64, "the 64 blocks written again in flight");
    assert!(t.stream_meta_bytes > 0);
}

/// Fails `.1` after `ms` milliseconds; everything else lands at once.
#[derive(Clone)]
struct SlowFailFirst(u64);

impl StagingIntercept<MemoryStaging> for SlowFailFirst {
    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 fail = segid.part_id() == Some(1);
        let ms = self.0;
        Box::pin(async move {
            if fail {
                tokio::time::sleep(std::time::Duration::from_millis(ms)).await;
                return Err(Error::new(ErrorKind::ConnectionReset, "object refused"));
            }
            Ok(())
        })
    }
}

/// A leaf is not made clean while blocks under it are in an object still in
/// flight, though an object that completed holds a copy of it: here the first
/// object fails after the second, holding that copy, completed. A clean leaf is
/// not written again, and setting a pointer in one does not make it dirty, so
/// cleaning it would publish its copy pointing into the object that failed.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn a_leaf_with_blocks_still_in_flight_stays_dirty() {
    let rt = ahead(5 << 20);
    let (s, mut f) = create("ss-leaf-in-flight", rt.clone()).await;
    f.staging_interceptor(SlowFailFirst(400));
    // The first object: 0..1280, failing slowly.
    for b in 0..1280usize {
        f.write(b * BS, &block(b, 1)).await.unwrap();
    }
    // Block 5 again: its leaf now goes with the next object.
    f.write(5 * BS, &block(5, 7)).await.unwrap();
    // The second object, holding block 5 and its leaf, completes at once.
    for b in 20000..21280usize {
        f.write(b * BS, &block(b, 1)).await.unwrap();
    }
    tokio::task::yield_now().await;
    f.write(30000 * BS, &block(30000, 1)).await.unwrap();
    // The first fails now.
    tokio::time::sleep(std::time::Duration::from_millis(600)).await;
    f.write(30001 * BS, &block(30001, 1)).await.unwrap();
    f.flush().await.expect("flush");
    let mut g = reopen(&s, &rt).await;
    for b in (0..1280usize).chain(20000..21280) {
        let want = block(b, if b == 5 { 7 } else { 1 });
        assert_eq!(read(&mut g, b * BS, BS).await, want, "block {}", b);
    }
}

/// A memory staging not told the container's format still reads a stream
/// container back: a stream object says what it is, and its index nodes are
/// keyed by that. Only the format given at creation decided it, and without it
/// every read past the first leaf's worth found holes, without an error.
#[tokio::test]
async fn memory_staging_need_not_be_told_the_stream_format() {
    for target in [0usize, 5 << 20] {
        let mut rt = runtime(1 << 30);
        rt.stream_segment_bytes = target;
        rt.segment_mpu_chunk_size = 5 << 20;
        let s = MemoryStaging::new(StagingConfig::new_memory(&format!("ss-untold-{}", target)), rt.clone());
        let f_cfg = config(&s, &rt);
        let mut f = MemFile::new(s.clone(), s.to_block_loader(), NullNodeCache, f_cfg.clone(),
            HyperFileFlags::from_flags(FileFlags::rdwr()), HyperFileMode::from_mode(FileMode::default_file()))
            .await.expect("create");
        for b in 0..6000usize {
            f.write(b * BS, &block(b, 3)).await.unwrap();
        }
        f.flush().await.expect("flush");
        let mut g = MemFile::open(s.clone(), s.to_block_loader(), NullNodeCache, f_cfg, HyperFileFlags::from_flags(FileFlags::rdonly()))
            .await.expect("open");
        for b in 0..6000usize {
            assert_eq!(read(&mut g, b * BS, BS).await, block(b, 3), "sent ahead {}: block {}", target, b);
        }
    }
}