hyperfile 0.10.2

#![cfg(not(feature = "blocking"))]
//! Writing by reference ([`HyperFile::write_ref`]): the bytes come from a
//! [`BlockSource`] when the checkpoint is built, and what is stored is exactly what
//! writing the bytes would have stored.
//!
//! On in-memory staging, so the stored segments can be compared byte for byte.
//!
//! ```bash
//! cargo test --test functional_write_ref
//! ```
//!
//! With the WAL, which keeps each segment's bytes in memory for replay, `write_ref`
//! is refused; that build runs only the last test.

#![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, HyperFileRuntimeConfig};
use hyperfile::file::file::HyperFile;
use hyperfile::file::flags::{FileFlags, HyperFileFlags};
use hyperfile::file::mode::{FileMode, HyperFileMode};
use hyperfile::file::HyperTrait;
use hyperfile::staging::config::StagingConfig;
use hyperfile::staging::memory::MemoryStaging;
use hyperfile::staging::Staging;
use hyperfile::SegmentId;

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

const BS: usize = 4096;

/// A source over a plain buffer that counts what it was asked for.
struct Buf {
    data: Mutex<Vec<u8>>,
    reads: Mutex<usize>,
}

impl Buf {
    fn new(size: usize) -> Arc<Buf> {
        Arc::new(Buf { data: Mutex::new(vec![0; size]), reads: Mutex::new(0) })
    }
    fn put(&self, off: usize, bytes: &[u8]) {
        self.data.lock().unwrap()[off..off + bytes.len()].copy_from_slice(bytes);
    }
}

impl BlockSource for Buf {
    fn read_at(&self, off: u64, buf: &mut [u8]) -> Result<()> {
        let d = self.data.lock().unwrap();
        let off = off as usize;
        if off + buf.len() > d.len() {
            return Err(Error::new(ErrorKind::UnexpectedEof, "past the source"));
        }
        buf.copy_from_slice(&d[off..off + buf.len()]);
        *self.reads.lock().unwrap() += 1;
        Ok(())
    }
}

fn runtime() -> HyperFileRuntimeConfig {
    let mut rt = HyperFileRuntimeConfig::default();
    // Nothing publishes on its own: the flushes below are the only ones.
    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
}

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

async fn reopen(staging: &MemoryStaging, name: &str) -> MemFile<'static> {
    let rt = runtime();
    let sc = StagingConfig::new_memory(name);
    let config = HyperFileConfigBuilder::new().with_staging_config(&sc).with_runtime_config(&rt).build();
    MemFile::open(staging.clone(), staging.to_block_loader(), NullNodeCache, config,
        HyperFileFlags::from_flags(FileFlags::rdwr())).await.expect("reopen")
}

fn pattern(seed: u8, len: usize) -> Vec<u8> {
    let mut x = seed as u32 | 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 same sequence of changes written by value into one container and partly by
/// reference into another leaves the same bytes in storage, and the stored
/// container opens and reads with no source at all.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn by_reference_stores_what_by_value_stores() {
    let size = 64 * BS;
    let (mut a, sa) = new_file("ref-eq-a").await;
    let (mut b, sb) = new_file("ref-eq-b").await;
    let src = Buf::new(size);
    b.set_block_source(Some(src.clone())).unwrap();

    // checkpoint 1: blocks 0..32 (by reference in b), a small byte write in the middle
    let one = pattern(1, 32 * BS);
    a.write(0, &one).await.unwrap();
    src.put(0, &one);
    b.write_ref(0, 32 * BS).await.unwrap();
    let small = pattern(2, 100);
    a.write(5 * BS + 10, &small).await.unwrap();
    b.write(5 * BS + 10, &small).await.unwrap();      // over a reference: read-modify-write
    let sid_a = a.flush().await.unwrap();
    let sid_b = b.flush().await.unwrap();
    assert_eq!(b.refs_len(), 0, "a published reference is gone");

    // checkpoint 2: rewrite 8..24 by reference, zero part of it, extend
    let two = pattern(3, 16 * BS);
    a.write(8 * BS, &two).await.unwrap();
    src.put(8 * BS, &two);
    b.write_ref(8 * BS, 16 * BS).await.unwrap();
    a.write_zero(10 * BS, 3 * BS).await.unwrap();
    b.write_zero(10 * BS, 3 * BS).await.unwrap();
    let tail = pattern(4, 8 * BS);
    a.write(40 * BS, &tail).await.unwrap();
    src.put(40 * BS, &tail);
    b.write_ref(40 * BS, 8 * BS).await.unwrap();
    // before the flush, a read of b is served from the source for the references
    assert_eq!(read_all(&mut b, 48 * BS).await, read_all(&mut a, 48 * BS).await,
        "unflushed reads agree");
    let sid_a2 = a.flush().await.unwrap();
    let sid_b2 = b.flush().await.unwrap();

    // the stored segments are the same bytes
    for (x, y) in [(sid_a, sid_b), (sid_a2, sid_b2)] {
        let get = |st: &MemoryStaging, id: SegmentId| st.segment_bytes(id)
            .or_else(|| st.segment_bytes(id.at_part(0)))
            .unwrap_or_else(|| panic!("no object for checkpoint {id}"));
        let (ba, bb) = (get(&sa, x), get(&sb, y));
        // Same size, and the same bytes everywhere but the segment header: that carries
        // each container's own inode inline (its timestamps, the state it was built on),
        // which two containers do not share. The block index, the metadata and the data
        // are identical.
        assert_eq!(ba.len(), bb.len(), "segment {x} / {y} differ in size");
        let diff: Vec<usize> = (0..ba.len()).filter(|i| ba[*i] != bb[*i]).collect();
        assert!(diff.iter().all(|i| *i < 256),
            "segment {x} / {y} differ past the header: {} bytes, first at {:?}", diff.len(), diff.iter().find(|i| **i >= 256));
    }
    assert_eq!(sa.segment_count(), sb.segment_count());

    // the source is no longer consulted, and a reopen with none reads the same
    src.put(0, &vec![0xEE; size]);
    let want = read_all(&mut a, 48 * BS).await;
    assert_eq!(read_all(&mut b, 48 * BS).await, want, "after the flush the source does not matter");
    let mut r = reopen(&sb, "ref-eq-b").await;
    assert_eq!(read_all(&mut r, 48 * BS).await, want, "the stored container reads with no source");
    let _: SegmentId = sid_b2;
}

/// A later write of a whole referenced block replaces it without reading the source;
/// a truncate cuts references past the new size.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn later_writes_and_truncates_supersede_references() {
    let (mut f, s) = new_file("ref-over").await;
    let src = Buf::new(16 * BS);
    f.set_block_source(Some(src.clone())).unwrap();
    src.put(0, &pattern(9, 16 * BS));
    f.write_ref(0, 16 * BS).await.unwrap();
    let newer = pattern(10, 4 * BS);
    f.write(4 * BS, &newer).await.unwrap();           // blocks 4..8, whole
    f.truncate(12 * BS + 100).await.unwrap();         // cuts 13..16, keeps part of 12
    // (the truncate or write may publish; either way the result below must hold)
    f.flush().await.unwrap();
    let mut want = pattern(9, 16 * BS);
    want[4 * BS..8 * BS].copy_from_slice(&newer);
    want.truncate(12 * BS + 100);
    let mut r = reopen(&s, "ref-over").await;
    assert_eq!(read_all(&mut r, 16 * BS).await, want);
}

#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn write_ref_refuses_what_it_cannot_do() {
    let (mut f, _) = new_file("ref-refuse").await;
    let e = f.write_ref(0, BS).await.unwrap_err();
    assert_eq!(e.kind(), ErrorKind::InvalidInput, "no source: {e}");
    let src = Buf::new(4 * BS);
    f.set_block_source(Some(src)).unwrap();
    assert_eq!(f.write_ref(100, BS).await.unwrap_err().kind(), ErrorKind::InvalidInput, "unaligned");
    assert_eq!(f.write_ref(0, 100).await.unwrap_err().kind(), ErrorKind::InvalidInput, "partial block");
    f.write_ref(0, BS).await.unwrap();
    assert!(f.set_block_source(None).is_err(), "cannot drop the source under a reference");
    f.flush().await.unwrap();
    f.set_block_source(None).expect("after the flush it can go");
}

/// A source that fails makes the flush fail, the references stay, and a flush
/// after the source recovers publishes them.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn a_failing_source_fails_the_flush_and_a_retry_publishes() {
    struct Flaky { inner: Arc<Buf>, fail: Mutex<bool> }
    impl BlockSource for Flaky {
        fn read_at(&self, off: u64, buf: &mut [u8]) -> Result<()> {
            if *self.fail.lock().unwrap() {
                return Err(Error::other("injected"));
            }
            self.inner.read_at(off, buf)
        }
    }
    let (mut f, s) = new_file("ref-flaky").await;
    let buf = Buf::new(8 * BS);
    buf.put(0, &pattern(5, 8 * BS));
    let src = Arc::new(Flaky { inner: buf, fail: Mutex::new(true) });
    f.set_block_source(Some(src.clone())).unwrap();
    f.write_ref(0, 8 * BS).await.unwrap();
    assert!(f.flush().await.is_err(), "the flush fails with its source");
    assert_eq!(f.refs_len(), 8, "the references are still there");
    *src.fail.lock().unwrap() = false;
    f.flush().await.expect("the retry publishes");
    let mut r = reopen(&s, "ref-flaky").await;
    assert_eq!(read_all(&mut r, 8 * BS).await, pattern(5, 8 * BS));
}

#[cfg(feature = "wal")]
#[tokio::test]
async fn write_ref_is_refused_with_the_wal() {
    let (mut f, _) = new_file("ref-wal").await;
    f.set_block_source(Some(Buf::new(4 * BS))).unwrap();
    assert_eq!(f.write_ref(0, BS).await.unwrap_err().kind(), ErrorKind::Unsupported);
}

/// Every block of `bytes` of a file, as the same byte value per block or 0xEE
/// where a block is mixed; `n` blocks from 0.
async fn per_block(f: &mut MemFile<'static>, n: usize) -> Vec<Vec<u8>> {
    let all = read_all(f, n * BS).await;
    all.chunks(BS).map(|c| c.to_vec()).collect()
}

/// A block of `fill`, with `patch` written over `[at, at + patch.len())`.
fn block_with(fill: u8, at: usize, patch: &[u8]) -> Vec<u8> {
    let mut b = vec![fill; BS];
    b[at..at + patch.len()].copy_from_slice(patch);
    b
}

/// Every write path that can land on a referenced block takes the reference out
/// first: a whole-block write drops it, a partial one reads the block from the
/// source and modifies that. None of them may leave a block both referenced and
/// dirty, or referenced with its pointer unresolvable: `write_batch` over a
/// reference used to panic in the read path, and `block_mut` found nothing to
/// borrow.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn every_write_path_takes_a_reference_out_first() {
    use hyperfile::buffer::BatchDataBlockWrapper;
    let (mut f, s) = new_file("ref-paths").await;
    f.write(0, &vec![0x11u8; 8 * BS]).await.unwrap();
    f.flush().await.unwrap();
    let src = Buf::new(8 * BS);
    src.put(0, &vec![0x22u8; 8 * BS]);
    f.set_block_source(Some(src.clone())).unwrap();
    f.write_ref(0, 8 * BS).await.unwrap();

    // 1: a whole block by batch; 2: part of one by batch
    let full = BatchDataBlockWrapper::new(1, BS, false);
    full.as_mut_slice().fill(0x44);
    let part = BatchDataBlockWrapper::new_partial_block(2, BS, 100, 200, false);
    part.as_mut_slice().fill(0x45);
    f.write_batch(vec![full, part]).await.expect("write_batch over references");
    // 3: borrowed and modified in place
    {
        let mut m = f.block_mut(3, false).await.expect("block_mut").expect("a referenced block is data");
        m.as_mut_slice()[..10].fill(0x55);
    }
    // 4: borrowed for reading; the reference stays
    {
        let r = f.block(4).await.expect("block").expect("a referenced block is data");
        assert!(r.as_slice().iter().all(|b| *b == 0x22), "block() of a reference did not read the source");
    }
    assert_eq!(f.refs_len(), 5, "blocks 1, 2 and 3 are no longer references; 0 and 4-7 are");

    let want: Vec<Vec<u8>> = vec![
        vec![0x22; BS],
        vec![0x44; BS],
        block_with(0x22, 100, &[0x45; 200]),
        block_with(0x22, 0, &[0x55; 10]),
        vec![0x22; BS], vec![0x22; BS], vec![0x22; BS], vec![0x22; BS],
    ];
    assert!(per_block(&mut f, 8).await == want, "before the flush, the handle reads the wrong bytes");
    f.flush().await.expect("flush");
    let mut r = reopen(&s, "ref-paths").await;
    assert!(per_block(&mut r, 8).await == want, "the stored container is not what was written");
}

/// Fails the first inode write, so the flush that tries it rolls back.
struct RefuseFirstInode(std::sync::atomic::AtomicUsize);

impl hyperfile::staging::StagingIntercept<MemoryStaging> for RefuseFirstInode {
    fn before_flush_inode(&self, _: &MemoryStaging, _: &[u8], _: hyperfile::inode::FlushInodeFlag)
        -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + '_ + Send>> {
        let n = self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        Box::pin(async move {
            if n == 0 { Err(Error::new(ErrorKind::PermissionDenied, "inode write refused")) } else { Ok(()) }
        })
    }
    fn after_flush_inode(&self, _: &MemoryStaging, _: &[u8], _: hyperfile::inode::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(()) })
    }
}

/// A flush that rolls back discards references along with the data written by
/// value: both were changes since the persisted state. Keeping the references
/// left the handle reading the source's bytes while the next flush, having
/// nothing it saw as dirty, published the old ones — the handle and storage
/// disagreeing without an error.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn a_rolled_back_flush_discards_references_with_the_rest() {
    let (mut f, s) = new_file("ref-rollback").await;
    f.write(0, &vec![0x11u8; 16 * BS]).await.unwrap();
    f.flush().await.unwrap();
    let src = Buf::new(16 * BS);
    src.put(0, &vec![0x22u8; 16 * BS]);
    f.set_block_source(Some(src.clone())).unwrap();
    f.write_ref(0, 8 * BS).await.unwrap();
    f.write(8 * BS, &vec![0x33u8; 8 * BS]).await.unwrap();
    f.staging_interceptor(RefuseFirstInode(Default::default()));
    f.flush_with_rollback().await.expect_err("the inode write is refused");

    assert_eq!(f.refs_len(), 0, "the rollback kept references");
    assert!(read_all(&mut f, 16 * BS).await == vec![0x11u8; 16 * BS],
        "after the rollback the handle does not read the persisted state");
    f.flush_with_rollback().await.expect("flush after the rollback");
    let mut r = reopen(&s, "ref-rollback").await;
    assert!(read_all(&mut r, 16 * BS).await == vec![0x11u8; 16 * BS],
        "storage does not hold what the handle read");
}

/// With references outstanding on a container that writes partial segments, a
/// write by value that crosses the dirty threshold is not refused: the partial it
/// writes holds the references too, reading their bytes as it uploads, and once it
/// has landed they are blocks on storage like the rest — the flush does not store
/// them again.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn a_write_past_the_threshold_is_not_refused_while_references_are_out() {
    let mut rt = runtime();
    rt.parted_segment_enabled = true;
    rt.data_cache_dirty_min_bytes_to_part = 16 * BS;
    rt.data_cache_dirty_max_blocks_threshold = 32;
    rt.data_cache_dirty_max_bytes_threshold = 32 * BS;
    let sc = StagingConfig::new_memory("ref-threshold");
    let config = HyperFileConfigBuilder::new().with_staging_config(&sc).with_runtime_config(&rt).build();
    let s = MemoryStaging::new(sc, rt);
    let mut f = MemFile::new(s.clone(), s.to_block_loader(), NullNodeCache, config,
        HyperFileFlags::from_flags(FileFlags::rdwr()), HyperFileMode::from_mode(FileMode::default_file()))
        .await.expect("create");
    let src = Buf::new(8 * BS);
    src.put(0, &pattern(9, 8 * BS));
    f.set_block_source(Some(src.clone())).unwrap();
    f.write_ref(0, 8 * BS).await.unwrap();
    let by_value = pattern(10, 48 * BS);
    f.write(8 * BS, &by_value).await.expect("a write past the threshold");
    f.flush().await.expect("flush");
    let mut r = reopen(&s, "ref-threshold").await;
    let mut want = pattern(9, 8 * BS);
    want.extend_from_slice(&by_value);
    assert!(read_all(&mut r, 56 * BS).await == want, "the stored container is not what was written");
    // every block stored once, across the partials and the final part
    let stored: u32 = (0..64u16).filter_map(|k| s.segment_bytes(SegmentId::new_from_cno(1).at_part(k)))
        .map(|b| hyperfile::ondisk::SegmentHeader::from_slice(&b[..hyperfile::ondisk::SegmentHeader::size()]).s_ndatablk)
        .sum();
    assert_eq!(stored, 56, "blocks were stored more than once");
}

/// A source that can only answer once another task on the same runtime has run.
/// Reading it on the runtime's own thread — a current-thread runtime here, as
/// `#[tokio::test]` gives — blocks that task out for good; read on the blocking
/// pool, it answers. It gives up after a while rather than hang the suite.
struct NeedsTheRuntime {
    inner: Arc<Buf>,
    ready: Mutex<Option<std::sync::mpsc::Receiver<()>>>,
}

impl BlockSource for NeedsTheRuntime {
    fn read_at(&self, off: u64, buf: &mut [u8]) -> Result<()> {
        if let Some(rx) = self.ready.lock().unwrap().as_ref() {
            rx.recv_timeout(std::time::Duration::from_secs(2))
                .map_err(|_| Error::new(ErrorKind::TimedOut, "read on the runtime's thread"))?;
        }
        self.inner.read_at(off, buf)
    }
}

fn needs_the_runtime(inner: Arc<Buf>) -> Arc<NeedsTheRuntime> {
    let (tx, rx) = std::sync::mpsc::channel();
    tokio::spawn(async move { loop { if tx.send(()).is_err() { break; } tokio::task::yield_now().await; } });
    Arc::new(NeedsTheRuntime { inner, ready: Mutex::new(Some(rx)) })
}

/// Reads from the source for a read, and for a write that modifies part of a
/// referenced block, run on the blocking pool, as the upload's do: a source does
/// ordinary blocking I/O.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn the_source_is_read_off_the_runtime_thread() {
    let (mut f, s) = new_file("ref-blocking").await;
    let buf = Buf::new(8 * BS);
    buf.put(0, &vec![0x22u8; 8 * BS]);
    f.set_block_source(Some(needs_the_runtime(buf))).unwrap();
    f.write_ref(0, 8 * BS).await.unwrap();

    let mut one = vec![0u8; BS];
    f.read(0, &mut one).await.expect("a read of a referenced block");
    assert!(one.iter().all(|b| *b == 0x22));
    let mut some = vec![0u8; 100];
    f.read(BS + 10, &mut some).await.expect("a read of part of a referenced block");
    f.write(2 * BS + 10, &[0x66; 10]).await.expect("a write of part of a referenced block");
    f.flush().await.expect("the flush reads the rest");
    let mut r = reopen(&s, "ref-blocking").await;
    assert!(per_block(&mut r, 3).await[2] == block_with(0x22, 10, &[0x66; 10]));
}