hyperfile 0.10.2

#![cfg(not(feature = "blocking"))]
//! Compaction that is not mistaken for a change of content.
//!
//! A publish that only moved blocks — every dirty block came from
//! `HyperFile::relocate`, nothing else changed since the last publish, and its
//! number directly follows that publish — sets `InodeRaw::FLAG_COMPACTED`, and
//! `content_cno()` then names the checkpoint before it. Any publish that changes
//! content clears it. On memory staging.
//!
//! ```bash
//! cargo test --test functional_compaction
//! cargo test --features wal --test functional_compaction
//! ```

#![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::block::BlockState;
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;
const FMT: BlockPtrFormat = BlockPtrFormat::PartedSegment;

fn runtime() -> 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
}

fn config(staging: &MemoryStaging, rt: &HyperFileRuntimeConfig) -> 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()
}

/// A file of 40 blocks: 0..16 and 20..40 written, 16..20 a hole, 30 an explicit
/// zero block. Flushed.
async fn prepared(name: &str, rt: HyperFileRuntimeConfig) -> (MemoryStaging, MemFile<'static>) {
    let staging = MemoryStaging::new(StagingConfig::new_memory(name), rt.clone()).with_block_ptr_format(FMT);
    let mut f = MemFile::new(staging.clone(), staging.to_block_loader(), NullNodeCache, config(&staging, &rt),
        HyperFileFlags::from_flags(FileFlags::rdwr()), HyperFileMode::from_mode(FileMode::default_file()))
        .await.expect("create");
    f.write(0, &pattern(1, 16 * BS)).await.unwrap();
    f.write(20 * BS, &pattern(2, 20 * BS)).await.unwrap();
    f.write_zero(30 * BS, BS).await.unwrap();
    f.flush().await.expect("flush");
    (staging, f)
}

async fn reopen(staging: &MemoryStaging) -> MemFile<'static> {
    MemFile::open(staging.clone(), staging.to_block_loader(), NullNodeCache, config(staging, &runtime()),
        HyperFileFlags::from_flags(FileFlags::rdwr())).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>) -> Vec<u8> {
    let mut v = vec![0u8; 40 * BS];
    let n = f.read(0, &mut v).await.expect("read");
    v.truncate(n);
    v
}

/// The stored inode, as a reader that does not open the file sees it.
fn stored(staging: &MemoryStaging) -> InodeRaw {
    InodeRaw::from_u8_slice(&staging.inode_bytes().expect("an inode is stored"))
}

/// Refuses part-0 segment writes while armed.
#[derive(Clone, Default)]
struct RefuseCommit(Arc<Mutex<bool>>);

impl StagingIntercept<MemoryStaging> for RefuseCommit {
    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 refuse = *self.0.lock().unwrap() && segid.part_id() == Some(0);
        Box::pin(async move {
            if refuse { Err(Error::new(ErrorKind::ConnectionReset, "part 0 refused")) } else { Ok(()) }
        })
    }
}

/// A flush of relocated blocks only sets the flag: `content_cno` stays where it
/// was while `last_cno` moves on. The blocks read back the same, cold; holes and
/// zero blocks are skipped and stay as they were; mtime and ctime do not move.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn a_publish_of_relocated_blocks_only_keeps_content_cno() {
    let (s, mut f) = prepared("cmp-only", runtime()).await;
    let before = read_all(&mut f).await;
    let (last0, content0) = (f.last_cno(), f.content_cno().unwrap());
    assert_eq!(content0, last0, "a content flush names itself");
    let st0 = f.stat();
    let states0: Vec<BlockState> = { let mut v = Vec::new(); for b in 0..40 { v.push(f.block_state(b).await.unwrap()); } v };

    let n = f.relocate(&[(0, 40)]).await.expect("relocate");
    assert_eq!(n, 35, "16 + 20 written blocks, less the zero block; the hole skipped");
    f.flush().await.expect("flush");

    assert!(f.last_cno() > last0, "a relocation publishes a new checkpoint");
    assert_eq!(f.content_cno().unwrap(), content0, "content_cno moved on a relocation");
    assert!(stored(&s).is_compacted(), "the stored inode is not marked");
    assert_eq!(stored(&s).content_cno().unwrap(), content0);
    let st1 = f.stat();
    assert_eq!((st1.st_mtime, st1.st_mtime_nsec, st1.st_ctime, st1.st_ctime_nsec),
        (st0.st_mtime, st0.st_mtime_nsec, st0.st_ctime, st0.st_ctime_nsec), "relocate moved a timestamp");
    for b in 0..40 {
        assert_eq!(f.block_state(b).await.unwrap(), states0[b as usize], "block {} changed state", b);
    }
    let mut cold = reopen(&s).await;
    assert!(read_all(&mut cold).await == before, "relocated blocks read back differently");
    assert_eq!(cold.content_cno().unwrap(), content0, "a reopen sees another content_cno");
}

/// Blocks already dirty are not relocated, and their checkpoint is a content one.
/// So is one where any content changed: a write, a write_zero of a whole block,
/// a truncate, a size set through the attributes — whether before or after the
/// relocation.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn any_content_change_in_the_checkpoint_clears_the_flag() {
    let changes: Vec<(&str, usize)> = vec![("write", 0), ("write_zero", 1), ("truncate", 2), ("size", 3), ("write first", 4)];
    for (what, k) in changes {
        let (s, mut f) = prepared(&format!("cmp-mixed-{}", k), runtime()).await;
        if k == 4 {
            f.write(5 * BS, &[9u8; 100]).await.unwrap();
            let n = f.relocate(&[(0, 16)]).await.unwrap();
            assert_eq!(n, 15, "{}: the dirty block was relocated", what);
        } else {
            f.relocate(&[(0, 16)]).await.unwrap();
        }
        match k {
            0 => { f.write(25 * BS, &[7u8; 10]).await.unwrap(); },
            1 => { f.write_zero(21 * BS, BS).await.unwrap(); },
            2 => { f.truncate(39 * BS + 1).await.unwrap(); },
            3 => { let mut st = f.stat(); st.st_size += 1; f.update_stat(&st).await.unwrap(); },
            _ => {},
        }
        f.flush().await.expect("flush");
        assert!(!stored(&s).is_compacted(), "{}: a content change was marked as moving blocks only", what);
        assert_eq!(f.content_cno().unwrap(), f.last_cno(), "{}: content_cno did not move", what);
    }
}

/// The flag describes the newest checkpoint: the next content flush clears it,
/// and an attribute-only write keeps it — unless that write publishes a change
/// of content, as a truncate that only extends the file does.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn the_flag_follows_the_newest_checkpoint() {
    let (s, mut f) = prepared("cmp-follow", runtime()).await;
    let c0 = f.content_cno().unwrap();
    f.relocate(&[(0, 40)]).await.unwrap();
    f.flush().await.unwrap();
    assert!(stored(&s).is_compacted());

    // attributes only: kept
    let mut st = f.stat();
    st.st_mode = (st.st_mode & !0o777) | 0o600;
    f.update_stat(&st).await.unwrap();
    assert!(stored(&s).is_compacted(), "a chmod cleared the flag");
    assert_eq!(f.content_cno().unwrap(), c0);

    // a truncate that only extends, published without a new checkpoint: cleared
    let last = f.last_cno();
    f.truncate(50 * BS).await.unwrap();
    assert_eq!(f.last_cno(), last, "an extend published a checkpoint after all");
    assert!(!stored(&s).is_compacted(), "an extend kept the flag");
    assert!(f.content_cno().unwrap() > c0, "content_cno did not move on an extend");

    // a relocation again, then a content flush: cleared
    f.relocate(&[(0, 16)]).await.unwrap();
    f.flush().await.unwrap();
    assert!(stored(&s).is_compacted());
    f.write(0, &[1u8; 10]).await.unwrap();
    f.flush().await.unwrap();
    assert!(!stored(&s).is_compacted(), "a content flush kept the flag");
    assert_eq!(f.content_cno().unwrap(), f.last_cno());
}

/// A content write in an earlier partial makes the whole checkpoint a content
/// one, even when what is dirty at the final flush was all relocated.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn a_content_write_in_an_earlier_partial_clears_the_flag() {
    let mut rt = runtime();
    rt.parted_segment_enabled = true;
    rt.data_cache_dirty_min_bytes_to_part = 4 * BS;
    rt.data_cache_dirty_max_blocks_threshold = 6;
    rt.data_cache_dirty_max_bytes_threshold = 6 * BS;
    let (s, mut f) = prepared("cmp-partial", rt).await;
    let parts_before = s.segment_count();
    for b in 0..8 {
        f.write(b * BS, &[0x55u8; BS]).await.unwrap();
    }
    f.relocate(&[(20, 4)]).await.unwrap();
    f.flush().await.unwrap();
    assert!(s.segment_count() > parts_before + 1, "no partial was written");
    assert!(!stored(&s).is_compacted(), "content in an earlier partial was missed");
    assert_eq!(f.content_cno().unwrap(), f.last_cno());
}

/// A flush that fails after a content write, retried with relocated blocks in
/// the same dirty set: still a content checkpoint. The bookkeeping lives until a
/// publish, not until a number is taken — here the checkpoint keeps its number
/// across the attempts, since it has partials, so only the bookkeeping stands
/// between it and the flag.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn a_failed_content_flush_retried_is_still_a_content_checkpoint() {
    let mut rt = runtime();
    rt.parted_segment_enabled = true;
    rt.data_cache_dirty_min_bytes_to_part = 4 * BS;
    rt.data_cache_dirty_max_blocks_threshold = 6;
    rt.data_cache_dirty_max_bytes_threshold = 6 * BS;
    let (s, mut f) = prepared("cmp-retry", rt).await;
    let refuse = RefuseCommit::default();
    f.staging_interceptor(refuse.clone());
    for b in 0..8 {
        f.write(b * BS, &[0x66u8; BS]).await.unwrap();
    }
    f.relocate(&[(20, 4)]).await.unwrap();
    *refuse.0.lock().unwrap() = true;
    assert!(f.flush().await.is_err(), "the flush was meant to fail");
    *refuse.0.lock().unwrap() = false;
    f.flush().await.expect("the retry");
    assert!(!stored(&s).is_compacted(), "a retried content flush was marked as moving blocks only");
}

/// A relocation published at a number that does not directly follow the last
/// publish — a failed attempt used one — is not marked: the checkpoint before it
/// is not the one `last_cno - 1` names. Reported as a change, once.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn a_relocation_after_a_skipped_number_is_not_marked() {
    let (s, mut f) = prepared("cmp-gap", runtime()).await;
    let refuse = RefuseCommit::default();
    f.staging_interceptor(refuse.clone());
    let last0 = f.last_cno();
    f.relocate(&[(0, 16)]).await.unwrap();
    *refuse.0.lock().unwrap() = true;
    assert!(f.flush().await.is_err());
    *refuse.0.lock().unwrap() = false;
    f.flush().await.expect("the retry");
    assert!(f.last_cno() > last0 + 1, "no number was skipped");
    assert!(!stored(&s).is_compacted(), "marked across a skipped number");
    assert_eq!(f.content_cno().unwrap(), f.last_cno());
}

/// A rollback brings back the persisted flag and forgets what it discarded: a
/// content write rolled back does not keep the next relocation from being marked.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn a_rollback_forgets_the_content_it_discarded() {
    let (s, mut f) = prepared("cmp-rollback", runtime()).await;
    f.relocate(&[(0, 16)]).await.unwrap();
    f.flush().await.unwrap();
    assert!(stored(&s).is_compacted());
    let c = f.content_cno().unwrap();
    let refuse = RefuseCommit::default();
    f.staging_interceptor(refuse.clone());
    f.write(0, &[3u8; 100]).await.unwrap();
    *refuse.0.lock().unwrap() = true;
    assert!(f.flush_with_rollback().await.is_err());
    *refuse.0.lock().unwrap() = false;
    assert_eq!(f.content_cno().unwrap(), c, "the rollback changed content_cno");
    // the abandoned number is behind us now; the next number follows the
    // persisted counter, which the rollback moved past it
    let _ = f.relocate(&[(20, 4)]).await.unwrap();
    f.flush().await.unwrap();
    let raw = stored(&s);
    assert_eq!(raw.content_cno().unwrap(), f.content_cno().unwrap());
    // Whether this one is marked depends on the skipped number; that it is not a
    // content checkpoint for the discarded write is what matters: the file reads
    // as it did before that write.
    let mut cold = reopen(&s).await;
    let mut b = vec![0u8; 100];
    cold.read(0, &mut b).await.unwrap();
    assert!(b == pattern(1, 16 * BS)[..100], "the rolled back write is in the file");
}

/// Blocks written by reference are not relocated, and keep the flag off.
#[cfg(not(feature = "wal"))]
#[tokio::test]
async fn references_are_not_relocated_and_are_content() {
    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(())
        }
    }
    let (s, mut f) = prepared("cmp-refs", runtime()).await;
    f.set_block_source(Some(Arc::new(Src(pattern(4, 40 * BS))))).unwrap();
    f.write_ref(20 * BS, 4 * BS).await.unwrap();
    let n = f.relocate(&[(20, 8)]).await.unwrap();
    assert_eq!(n, 4, "referenced blocks were relocated");
    f.flush().await.unwrap();
    assert!(!stored(&s).is_compacted());
}

/// `content_cno` is refused for what has no content checkpoint.
#[test]
fn content_cno_is_refused_for_a_device_node_and_an_inline_file() {
    let mut raw: InodeRaw = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
    raw.i_last_cno = 9;
    raw.i_mode = libc::S_IFREG as u32 | 0o644;
    assert_eq!(raw.content_cno().unwrap(), 9);
    raw.set_compacted(true);
    assert_eq!(raw.content_cno().unwrap(), 8);
    raw.i_mode = libc::S_IFCHR as u32 | 0o644;
    assert_eq!(raw.content_cno().unwrap_err().kind(), ErrorKind::InvalidInput);
    raw.i_mode = libc::S_IFREG as u32 | 0o644;
    raw.set_inline(b"tiny");
    assert_eq!(raw.content_cno().unwrap_err().kind(), ErrorKind::InvalidInput);
}

#[cfg(feature = "wal")]
#[tokio::test]
async fn relocate_is_refused_with_the_wal() {
    let (_s, mut f) = prepared("cmp-wal", runtime()).await;
    assert_eq!(f.relocate(&[(0, 4)]).await.unwrap_err().kind(), ErrorKind::Unsupported);
}