hyperfile 0.12.0

#![cfg(not(any(feature = "blocking", feature = "wal")))]
//! Compaction beside the writer, on memory staging with conditional writes: a
//! compactor handle writes copies and a map under a published number and
//! attaches it to the stored inode; the writer takes it in before its next
//! publish of data. In both formats with parts.
//!
//! ```bash
//! cargo test --release --test functional_online_compaction
//! ```

use btree_ondisk::{MemoryBlockLoader, NullNodeCache};
use hyperfile::config::{FlushConflictPolicy, 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::ondisk::{InodeRaw, SegmentHeader, SEGMENT_FLAG_CLEANER};
use hyperfile::staging::config::StagingConfig;
use hyperfile::staging::memory::MemoryStaging;
use hyperfile::staging::Staging;
use hyperfile::{stream_segment, SegmentId};
use std::io::ErrorKind;
use std::time::Duration;
use hyperfile::compaction::StoredCompaction;

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

const BS: usize = 4096;
const FORMATS: [BlockPtrFormat; 2] = [BlockPtrFormat::PartedSegment, BlockPtrFormat::StreamSegment];

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 = 1 << 40;
    rt.data_cache_blocks = 0;
    rt.stream_segment_bytes = 0;
    rt.flush_conflict_policy = FlushConflictPolicy::FailFast;
    rt
}

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

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

fn stored(s: &MemoryStaging) -> InodeRaw {
    InodeRaw::from_u8_slice(&s.inode_bytes().expect("an inode"))
}

/// Every block reads `want(b)` from a handle opened cold.
async fn check_cold(s: &MemoryStaging, fmt: BlockPtrFormat, rt: &HyperFileRuntimeConfig, name: &str, n: usize,
        want: impl Fn(usize) -> Vec<u8>) {
    let mut g = open(s, fmt, rt, FileFlags::rdonly()).await;
    let mut v = vec![0u8; BS];
    for b in 0..n {
        g.read(b * BS, &mut v).await.unwrap();
        assert!(v == want(b), "{}: block {} reads back wrong", name, b);
    }
}

/// Whether the object `id` is marked as a compactor's.
fn is_cleaners(s: &MemoryStaging, id: SegmentId, fmt: BlockPtrFormat) -> bool {
    let bytes = s.segment_bytes(id).expect("stored");
    if fmt.is_stream() {
        let (layout, _) = stream_segment::parse_tail(id, bytes.len(), &bytes).unwrap().unwrap();
        layout.hdr.is_cleaner()
    } else {
        SegmentHeader::from_slice(&bytes[..std::mem::size_of::<SegmentHeader>()]).s_flags & SEGMENT_FLAG_CLEANER != 0
    }
}

/// A file of `n` blocks written in four checkpoints, each over the last half
/// again: generation 1 for the first half, 4 for the second.
async fn written(name: &str, fmt: BlockPtrFormat, rt: &HyperFileRuntimeConfig, n: usize) -> (MemoryStaging, MemFile<'static>) {
    let (s, mut f) = create(name, fmt, rt).await;
    for g in 1..=4u8 {
        let from = if g == 1 { 0 } else { n / 2 };
        for b in from..n {
            f.write(b * BS, &block(b, g)).await.unwrap();
        }
        f.flush().await.expect("flush");
    }
    (s, f)
}

fn generation(b: usize, n: usize) -> u8 {
    if b < n / 2 { 1 } else { 4 }
}

const AGE: Duration = Duration::from_secs(3600);

/// What the stored inode object says of a compaction.
async fn attached(s: &MemoryStaging) -> StoredCompaction {
    let mut raw = InodeRaw::default();
    s.load_inode(raw.as_mut_u8_slice()).await.unwrap().map(|st| st.compaction).unwrap_or_default()
}

/// Compact `[0, n)` beside the writer and attach it.
async fn compact(s: &MemoryStaging, fmt: BlockPtrFormat, rt: &HyperFileRuntimeConfig, n: usize)
        -> hyperfile::file::file::CompactionAttached {
    let mut crt = rt.clone();
    crt.take_up_on_open = false;
    let mut c = open(s, fmt, &crt, FileFlags::rdwr()).await;
    c.begin_compaction().await.expect("begin");
    c.relocate(&[(0, n)]).await.unwrap();
    c.publish_compaction(AGE).await.expect("attach")
}

/// Where each block of `[0, n)` is, as a handle opened cold finds it.
async fn placement(s: &MemoryStaging, fmt: BlockPtrFormat, rt: &HyperFileRuntimeConfig, n: usize) -> Vec<SegmentId> {
    let mut g = open(s, fmt, rt, FileFlags::rdonly()).await;
    g.block_placement(0, n).await.unwrap().into_iter().map(|p| p.expect("a block in an object").0).collect()
}

/// A compactor alone: its objects are under the published number, at parts the
/// writer did not use, marked; the stored inode is as it was but for the bit,
/// with the compaction in its metadata; the map readers see has not moved.
#[tokio::test]
async fn a_compaction_attaches_and_moves_nothing_yet() {
    for fmt in FORMATS {
        let name = format!("oc-attach-{:?}", fmt);
        let rt = runtime();
        let n = 3000;
        let (s, w) = written(&name, fmt, &rt, n).await;
        drop(w);
        let before = stored(&s);
        let top = *s.parts_of(before.i_last_cno).await.unwrap().iter().max().unwrap();
        let where_before = placement(&s, fmt, &rt, n).await;
        let out = compact(&s, fmt, &rt, n).await;
        let after = stored(&s);
        assert!(after.is_compaction_pending(), "{}", name);
        let mut bare = after;
        bare.set_compaction_pending(false);
        assert_eq!(bare.as_u8_slice(), before.as_u8_slice(), "{}: the inode changed but for the bit", name);
        let StoredCompaction::Pending(p) = attached(&s).await else { panic!("{}: no metadata", name) };
        assert_eq!((p.base, p.base_fingerprint, p.moved), (before.i_last_cno, before.fingerprint(), n as u64), "{}", name);
        assert_eq!(p.parts.0, top + 1, "{}", name);
        assert_eq!((out.attached_to, out.base, out.parts, out.moved, out.attempts),
            (before.i_last_cno, before.i_last_cno, p.parts, n as u64, 1), "{}", name);
        let ids = s.segment_ids();
        for part in p.parts.0..p.parts.1 {
            let id = SegmentId::new_from_cno(before.i_last_cno).at_part(part);
            if ids.contains(&id) {
                assert!(is_cleaners(&s, id, fmt), "{}: {} is not marked", name, id);
            }
        }
        assert!(ids.iter().all(|id| id.as_cno() <= before.i_last_cno), "{}: written past the head", name);
        assert_eq!(placement(&s, fmt, &rt, n).await, where_before, "{}: readers see moved blocks", name);
        check_cold(&s, fmt, &rt, &name, n, |b| block(b, generation(b, n))).await;
    }
}

/// A writer opened after the attach takes the compaction in at its first publish
/// of data, with no refusal: what it did not write is at the copies, what it wrote
/// is its own, and what it publishes carries neither bit nor metadata.
#[tokio::test]
async fn a_writer_opened_after_takes_it_in() {
    for fmt in FORMATS {
        let name = format!("oc-after-{:?}", fmt);
        let rt = runtime();
        let n = 3000;
        let (s, w) = written(&name, fmt, &rt, n).await;
        drop(w);
        let base = stored(&s).i_last_cno;
        let p = compact(&s, fmt, &rt, n).await;
        let mut w = open(&s, fmt, &rt, FileFlags::rdwr()).await;
        for b in 100..200 {
            w.write(b * BS, &block(b, 9)).await.unwrap();
        }
        let cno = w.flush().await.expect("publish").as_cno();
        assert_eq!(cno, base + 1, "{}: a number was given up", name);
        let after = stored(&s);
        assert!(!after.is_compaction_pending(), "{}", name);
        assert_eq!(attached(&s).await, StoredCompaction::None, "{}", name);
        let at = placement(&s, fmt, &rt, n).await;
        for (b, id) in at.iter().enumerate() {
            if (100..200).contains(&b) {
                assert_eq!(id.as_cno(), cno, "{}: block {} written here is in {}", name, b, id);
            } else {
                assert_eq!(id.as_cno(), base, "{}: block {} not at its copy: {}", name, b, id);
                assert!(id.part_id().unwrap() >= p.parts.0 && id.part_id().unwrap() < p.parts.1,
                    "{}: block {} in {}, not among the compaction's parts", name, b, id);
            }
        }
        drop(w);
        check_cold(&s, fmt, &rt, &name, n, |b| block(b, if (100..200).contains(&b) { 9 } else { generation(b, n) })).await;
    }
}

/// A writer with changes not yet published when the compaction is attached: its
/// publish is refused, it adopts the stored inode, takes the compaction in and
/// publishes on top under a fresh number.
#[tokio::test]
async fn a_running_writer_takes_it_in() {
    for fmt in FORMATS {
        let name = format!("oc-running-{:?}", fmt);
        let rt = runtime();
        let n = 3000;
        let (s, mut w) = written(&name, fmt, &rt, n).await;
        let base = stored(&s).i_last_cno;
        for b in 100..200 {
            w.write(b * BS, &block(b, 9)).await.unwrap();
        }
        compact(&s, fmt, &rt, n).await;
        let cno = w.flush().await.expect("publish on top").as_cno();
        assert!(cno > base, "{}", name);
        assert!(!stored(&s).is_compaction_pending(), "{}", name);
        let at = placement(&s, fmt, &rt, n).await;
        for (b, id) in at.iter().enumerate() {
            let mine = (100..200).contains(&b);
            assert_eq!(id.as_cno() == base, !mine, "{}: block {} in {}", name, b, id);
        }
        drop(w);
        check_cold(&s, fmt, &rt, &name, n, |b| block(b, if (100..200).contains(&b) { 9 } else { generation(b, n) })).await;
    }
}

/// The writer publishes five times between the compaction's begin and its
/// attach: the attach is one write, nothing redone; the writer's next publish
/// keeps every block it changed since and takes the copies for the rest.
#[tokio::test]
async fn the_writer_may_be_versions_ahead() {
    for fmt in FORMATS {
        let name = format!("oc-ahead-{:?}", fmt);
        let rt = runtime();
        let n = 3000;
        let (s, mut w) = written(&name, fmt, &rt, n).await;
        let base = stored(&s).i_last_cno;
        let mut crt = rt.clone();
        crt.take_up_on_open = false;
        let mut c = open(&s, fmt, &crt, FileFlags::rdwr()).await;
        c.begin_compaction().await.expect("begin");
        c.relocate(&[(0, n)]).await.unwrap();
        for k in 0..5usize {
            for b in k * 100..k * 100 + 50 {
                w.write(b * BS, &block(b, 10 + k as u8)).await.unwrap();
            }
            w.flush().await.expect("the writer goes on");
        }
        let out = c.publish_compaction(AGE).await.expect("attach");
        assert_eq!((out.base, out.attached_to, out.attempts), (base, base + 5, 1), "{}", name);
        w.write(2999 * BS, &block(2999, 30)).await.unwrap();
        w.flush().await.expect("take it in");
        let changed = |b: usize| (0..5).find(|k| (k * 100..k * 100 + 50).contains(&b));
        let at = placement(&s, fmt, &rt, n).await;
        for (b, id) in at.iter().enumerate() {
            let mine = changed(b).is_some() || b == 2999;
            assert_eq!(id.as_cno() == base, !mine, "{}: block {} in {}", name, b, id);
        }
        drop((w, c));
        check_cold(&s, fmt, &rt, &name, n, |b| block(b, match (b, changed(b)) {
            (2999, _) => 30,
            (_, Some(k)) => 10 + k as u8,
            _ => generation(b, n),
        })).await;
    }
}

/// The writer attached to one version publishes on top of a writer checkpoint
/// that changed blocks the compaction moved, and a truncate that dropped some:
/// the writer's content wins, and nothing truncated comes back.
#[tokio::test]
async fn what_the_writer_changed_or_dropped_stays_so() {
    for fmt in FORMATS {
        let name = format!("oc-dropped-{:?}", fmt);
        let rt = runtime();
        let n = 3000;
        let (s, mut w) = written(&name, fmt, &rt, n).await;
        let mut crt = rt.clone();
        crt.take_up_on_open = false;
        let mut c = open(&s, fmt, &crt, FileFlags::rdwr()).await;
        c.begin_compaction().await.expect("begin");
        c.relocate(&[(0, n)]).await.unwrap();
        w.truncate(2000 * BS).await.unwrap();
        w.truncate(n * BS).await.unwrap();
        for b in 0..10 {
            w.write(b * BS, &block(b, 6)).await.unwrap();
        }
        w.flush().await.expect("writer");
        c.publish_compaction(AGE).await.expect("attach");
        w.write(BS * 20, &block(20, 7)).await.unwrap();
        w.flush().await.expect("take it in");
        drop((w, c));
        check_cold(&s, fmt, &rt, &name, n, |b| match b {
            0..10 => block(b, 6),
            20 => block(20, 7),
            2000.. => vec![0u8; BS],
            _ => block(b, generation(b, n)),
        }).await;
    }
}

/// The writer's objects already on storage when the compaction is attached --
/// sent ahead under the stream format, partials under the parted one -- stay in
/// use: its blocks there keep their places.
#[tokio::test]
async fn what_the_writer_wrote_out_early_is_kept() {
    for fmt in FORMATS {
        let name = format!("oc-early-{:?}", fmt);
        let rt = runtime();
        let n = 3000;
        let (s, w) = written(&name, fmt, &rt, n).await;
        drop(w);
        let base = stored(&s);
        let mut wrt = runtime();
        if fmt.is_stream() {
            wrt.stream_segment_bytes = 5 << 20;
            wrt.segment_mpu_chunk_size = 5 << 20;
        } else {
            wrt.data_cache_dirty_max_bytes_threshold = 2 << 20;
            wrt.data_cache_dirty_max_blocks_threshold = (2 << 20) / BS;
            wrt.data_cache_dirty_min_bytes_to_part = 1 << 20;
        }
        let mut w = open(&s, fmt, &wrt, FileFlags::rdwr()).await;
        let more = 3000;
        for b in n..n + more {
            w.write(b * BS, &block(b, 3)).await.unwrap();
            tokio::task::yield_now().await;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
        let early: Vec<SegmentId> = s.segment_ids().into_iter().filter(|id| id.as_cno() > base.i_last_cno).collect();
        assert!(!early.is_empty(), "{}: nothing was written out before the flush", name);
        compact(&s, fmt, &rt, n).await;
        let cno = w.flush().await.expect("the writer on top").as_cno();
        assert!(cno > base.i_last_cno, "{}", name);
        let placed = w.block_placement(n as u64, more).await.unwrap();
        let in_early = placed.iter().filter(|p| early.contains(&p.unwrap().0)).count();
        assert!(in_early > 0, "{}: no block stayed where it was written early", name);
        drop(w);
        check_cold(&s, fmt, &rt, &name, n + more, |b| block(b, if b >= n { 3 } else { generation(b, n) })).await;
    }
}

/// A compactor past the dirty-data limits writes partials under the head's
/// number, marked like the rest, and attaches one compaction.
#[tokio::test]
async fn a_compactor_writes_partials_under_the_head() {
    for fmt in FORMATS {
        let name = format!("oc-compactor-partials-{:?}", fmt);
        let rt = runtime();
        let n = 3000;
        let (s, w) = written(&name, fmt, &rt, n).await;
        drop(w);
        let base = stored(&s);
        let mut crt = runtime();
        crt.take_up_on_open = false;
        crt.data_cache_dirty_max_bytes_threshold = 2 << 20;
        crt.data_cache_dirty_max_blocks_threshold = (2 << 20) / BS;
        crt.data_cache_dirty_min_bytes_to_part = 1 << 20;
        let mut c = open(&s, fmt, &crt, FileFlags::rdwr()).await;
        c.begin_compaction().await.expect("begin");
        for k in 0..(n / 500) {
            c.relocate(&[((k * 500) as u64, 500)]).await.unwrap();
        }
        let partials: Vec<SegmentId> = s.segment_ids().into_iter()
            .filter(|id| id.as_cno() == base.i_last_cno && is_cleaners(&s, *id, fmt)).collect();
        assert!(partials.len() >= 2, "{}: no partial written: {:?}", name, partials);
        let out = c.publish_compaction(AGE).await.expect("attach");
        assert_eq!(out.moved, n as u64, "{}", name);
        if !fmt.is_stream() {
            // Every object it wrote holds something: no summary of nothing once
            // the partials took the whole map.
            for part in out.parts.0..out.parts.1 {
                let id = SegmentId::new_from_cno(base.i_last_cno).at_part(part);
                let bytes = s.segment_bytes(id).expect("written");
                let h = SegmentHeader::from_slice(&bytes[..std::mem::size_of::<SegmentHeader>()]);
                assert!(h.s_ndatablk + h.s_nmetablk > 0, "{}: {} holds nothing", name, id);
            }
        }
        assert!(s.segment_ids().iter().all(|id| id.as_cno() <= base.i_last_cno), "{}: written past the head", name);
        drop(c);
        let mut w = open(&s, fmt, &rt, FileFlags::rdwr()).await;
        w.write(0, &block(0, 6)).await.unwrap();
        w.flush().await.expect("take it in");
        drop(w);
        check_cold(&s, fmt, &rt, &name, n, |b| block(b, if b == 0 { 6 } else { generation(b, n) })).await;
    }
}

/// Attributes alone keep the compaction attached; the publish of data after
/// takes it in. A counter-only write (`abandon_unpublished`, after a failed
/// flush) and `update_stat_fast` keep it too.
#[tokio::test]
async fn writes_that_keep_the_number_keep_it_attached() {
    for fmt in FORMATS {
        let name = format!("oc-keep-{:?}", fmt);
        let rt = runtime();
        let n = 300;
        let (s, mut w) = written(&name, fmt, &rt, n).await;
        compact(&s, fmt, &rt, n).await;
        let mut st = w.stat();
        st.st_mode = (st.st_mode & !0o777) | 0o640;
        w.update_stat(&st).await.expect("setattr");
        w.flush().await.expect("attributes");
        assert_eq!(stored(&s).i_mode & 0o777, 0o640, "{}", name);
        assert!(stored(&s).is_compaction_pending(), "{}: attributes dropped the compaction", name);
        assert!(matches!(attached(&s).await, StoredCompaction::Pending(_)), "{}", name);
        let mut st2 = st;
        st2.st_mode = (st2.st_mode & !0o777) | 0o600;
        MemFile::update_stat_fast(s.clone(), &st2).await.expect("update_stat_fast");
        assert!(stored(&s).is_compaction_pending() && matches!(attached(&s).await, StoredCompaction::Pending(_)),
            "{}: update_stat_fast dropped it", name);
        drop(w);
        let mut w = open(&s, fmt, &rt, FileFlags::rdwr()).await;
        w.write(0, &block(0, 6)).await.unwrap();
        w.flush().await.expect("data");
        assert!(!stored(&s).is_compaction_pending(), "{}", name);
        assert_eq!(stored(&s).i_mode & 0o777, 0o600, "{}", name);
        drop(w);
        check_cold(&s, fmt, &rt, &name, n, |b| block(b, if b == 0 { 6 } else { generation(b, n) })).await;
    }
}

/// One compaction at a time: another is refused while one is attached, and runs
/// once the writer has taken it in.
#[tokio::test]
async fn one_compaction_at_a_time() {
    for fmt in FORMATS {
        let name = format!("oc-one-{:?}", fmt);
        let rt = runtime();
        let n = 3000;
        let (s, mut w) = written(&name, fmt, &rt, n).await;
        compact(&s, fmt, &rt, n / 2).await;
        let mut crt = rt.clone();
        crt.take_up_on_open = false;
        let mut c = open(&s, fmt, &crt, FileFlags::rdwr()).await;
        assert_eq!(c.begin_compaction().await.unwrap_err().kind(), ErrorKind::ResourceBusy, "{}", name);
        drop(c);
        w.write(0, &block(0, 6)).await.unwrap();
        w.flush().await.expect("take it in");
        compact(&s, fmt, &rt, n).await;
        w.write(BS, &block(1, 6)).await.unwrap();
        w.flush().await.expect("take the second in");
        drop(w);
        check_cold(&s, fmt, &rt, &name, n, |b| block(b, if b < 2 { 6 } else { generation(b, n) })).await;
        let at = placement(&s, fmt, &rt, n).await;
        assert!(at[2..].iter().all(|id| is_cleaners(&s, *id, fmt)), "{}", name);
    }
}

/// Two compactors begun on one version: the second's objects find the names
/// taken, or the compaction attached, and it gives up; the content stays.
#[tokio::test]
async fn a_second_compactor_gives_up() {
    for fmt in FORMATS {
        let name = format!("oc-two-compactors-{:?}", fmt);
        let rt = runtime();
        let n = 3000;
        let (s, w) = written(&name, fmt, &rt, n).await;
        drop(w);
        let mut crt = rt.clone();
        crt.take_up_on_open = false;
        let mut a = open(&s, fmt, &crt, FileFlags::rdwr()).await;
        let mut b = open(&s, fmt, &crt, FileFlags::rdwr()).await;
        a.begin_compaction().await.expect("begin a");
        b.begin_compaction().await.expect("begin b");
        a.relocate(&[(0, n / 2)]).await.unwrap();
        b.relocate(&[((n / 2) as u64, n / 2)]).await.unwrap();
        a.publish_compaction(AGE).await.expect("a");
        let e = b.publish_compaction(AGE).await.expect_err("b gives up");
        assert!(matches!(e.kind(), ErrorKind::AlreadyExists | ErrorKind::ResourceBusy), "{}: {}", name, e);
        drop((a, b));
        check_cold(&s, fmt, &rt, &name, n, |b| block(b, generation(b, n))).await;
    }
}

/// A compaction older than it may be when it would attach is given up, nothing
/// attached.
#[tokio::test]
async fn a_compaction_too_old_is_given_up() {
    for fmt in FORMATS {
        let name = format!("oc-old-{:?}", fmt);
        let rt = runtime();
        let n = 300;
        let (s, w) = written(&name, fmt, &rt, n).await;
        drop(w);
        let mut crt = rt.clone();
        crt.take_up_on_open = false;
        let mut c = open(&s, fmt, &crt, FileFlags::rdwr()).await;
        c.begin_compaction().await.expect("begin");
        c.relocate(&[(0, n)]).await.unwrap();
        let e = c.publish_compaction(Duration::ZERO).await.expect_err("too old");
        assert_eq!(e.kind(), ErrorKind::TimedOut, "{}", name);
        assert!(!stored(&s).is_compaction_pending(), "{}", name);
        drop(c);
        check_cold(&s, fmt, &rt, &name, n, |b| block(b, generation(b, n))).await;
    }
}

/// Metadata that does not read, or none under the bit: the writer takes nothing
/// in and drops it at its next publish of data; the content is the writer's.
#[tokio::test]
async fn metadata_that_does_not_read_is_dropped() {
    for fmt in FORMATS {
        for meta in [Some("v1;base=nonsense"), None] {
            let name = format!("oc-unreadable-{:?}-{}", fmt, meta.is_some());
            let rt = runtime();
            let n = 300;
            let (s, w) = written(&name, fmt, &rt, n).await;
            drop(w);
            compact(&s, fmt, &rt, n).await;
            s.set_inode_compaction_metadata(meta);
            let mut w = open(&s, fmt, &rt, FileFlags::rdwr()).await;
            w.write(0, &block(0, 6)).await.unwrap();
            w.flush().await.expect("publish without it");
            assert!(!stored(&s).is_compaction_pending(), "{}", name);
            assert_eq!(attached(&s).await, StoredCompaction::None, "{}", name);
            let at = placement(&s, fmt, &rt, n).await;
            assert!(at[1..].iter().all(|id| !is_cleaners(&s, *id, fmt)), "{}: something was taken in", name);
            drop(w);
            check_cold(&s, fmt, &rt, &name, n, |b| block(b, if b == 0 { 6 } else { generation(b, n) })).await;
        }
    }
}

/// With the default conflict policy, too, an attached compaction is taken in
/// rather than retried over.
#[tokio::test]
async fn the_default_policy_takes_it_in() {
    for fmt in FORMATS {
        let name = format!("oc-default-policy-{:?}", fmt);
        let mut rt = runtime();
        rt.flush_conflict_policy = FlushConflictPolicy::RetryLastWriterWins;
        let n = 300;
        let (s, mut w) = written(&name, fmt, &rt, n).await;
        w.write(0, &block(0, 6)).await.unwrap();
        compact(&s, fmt, &rt, n).await;
        let t = std::time::Instant::now();
        w.flush().await.expect("writer");
        assert!(t.elapsed() < Duration::from_secs(2), "{}: retried with backoff", name);
        drop(w);
        check_cold(&s, fmt, &rt, &name, n, |b| block(b, if b == 0 { 6 } else { generation(b, n) })).await;
        let at = placement(&s, fmt, &rt, n).await;
        assert!(at[1..].iter().all(|id| is_cleaners(&s, *id, fmt)), "{}", name);
    }
}

/// Holds the first inode write until told to go on, or refuses it.
#[derive(Clone)]
struct FirstInode(std::sync::Arc<tokio::sync::Notify>, std::sync::Arc<std::sync::atomic::AtomicBool>, bool);

impl hyperfile::staging::StagingIntercept<MemoryStaging> for FirstInode {
    fn before_flush_inode(&self, _: &MemoryStaging, _: &[u8], _: hyperfile::inode::FlushInodeFlag)
        -> std::pin::Pin<Box<dyn std::future::Future<Output = std::io::Result<()>> + '_ + Send>> {
        let first = self.1.swap(false, std::sync::atomic::Ordering::SeqCst);
        let (go, refuse) = (self.0.clone(), self.2);
        Box::pin(async move {
            if first && refuse { return Err(std::io::Error::new(ErrorKind::ConnectionReset, "inode refused by the test")); }
            if first { go.notified().await; }
            Ok(())
        })
    }
    fn after_flush_inode(&self, _: &MemoryStaging, _: &[u8], _: hyperfile::inode::FlushInodeFlag)
        -> std::pin::Pin<Box<dyn std::future::Future<Output = std::io::Result<()>> + '_ + Send>> {
        Box::pin(async { Ok(()) })
    }
    fn after_remove_inode(&self, _: &MemoryStaging)
        -> std::pin::Pin<Box<dyn std::future::Future<Output = std::io::Result<()>> + '_ + Send>> {
        Box::pin(async { Ok(()) })
    }
}

fn first_inode(refuse: bool) -> (FirstInode, std::sync::Arc<tokio::sync::Notify>) {
    let go = std::sync::Arc::new(tokio::sync::Notify::new());
    (FirstInode(go.clone(), std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true)), refuse), go)
}

/// A commit on storage past the inode holds a compaction off; a compactor opened
/// with `take_up_on_open` off leaves it to its writer, whose slow inode write
/// then lands.
#[tokio::test]
async fn a_compactor_leaves_a_slow_writers_commit_to_it() {
    for fmt in FORMATS {
        let name = format!("oc-slow-inode-{:?}", fmt);
        let rt = runtime();
        let n = 300;
        let (s, mut w) = written(&name, fmt, &rt, n).await;
        let base = stored(&s);
        let (hold, go) = first_inode(false);
        w.staging_interceptor(hold);
        w.write(0, &block(0, 6)).await.unwrap();
        let commit = SegmentId::new_from_cno(base.i_last_cno + 1).at_part(0);
        let compactor = async {
            while !s.segment_exists(commit).await.unwrap() {
                tokio::task::yield_now().await;
            }
            s.backdate_objects(60);
            let mut crt = runtime();
            crt.take_up_on_open = false;
            let mut c = open(&s, fmt, &crt, FileFlags::rdwr()).await;
            let untouched = stored(&s).as_u8_slice() == base.as_u8_slice();
            let begun = c.begin_compaction().await.map_err(|e| e.kind());
            go.notify_one();
            (untouched, begun)
        };
        let (flushed, (untouched, begun)) = tokio::join!(w.flush(), compactor);
        assert!(untouched, "{}: the compactor's open wrote the inode", name);
        assert_eq!(begun, Err(ErrorKind::ResourceBusy), "{}", name);
        flushed.unwrap_or_else(|e| panic!("{}: the writer's publish failed: {}", name, e));
        drop(w);
        check_cold(&s, fmt, &rt, &name, n, |b| block(b, if b == 0 { 6 } else { generation(b, n) })).await;
    }
}

/// The writer's commit lands and its inode write fails as a compaction is
/// attached to the version before. The commit was never acknowledged: a reopen
/// does not take it up past the attachment, and the same handle, flushing again,
/// takes the compaction in and publishes its writes.
#[tokio::test]
async fn a_commit_left_behind_by_an_attachment() {
    for fmt in FORMATS {
        for carry_on in [false, true] {
            let name = format!("oc-left-behind-{:?}-{}", fmt, carry_on);
            let rt = runtime();
            let n = 300;
            let (s, mut w) = written(&name, fmt, &rt, n).await;
            let base = stored(&s);
            let mut crt = rt.clone();
            crt.take_up_on_open = false;
            let mut c = open(&s, fmt, &crt, FileFlags::rdwr()).await;
            c.begin_compaction().await.expect("begin, before the commit lands");
            c.relocate(&[(0, n)]).await.unwrap();
            let (refuse, _) = first_inode(true);
            w.staging_interceptor(refuse);
            w.write(0, &block(0, 6)).await.unwrap();
            assert!(w.flush().await.is_err(), "{}", name);
            assert!(s.segment_exists(SegmentId::new_from_cno(base.i_last_cno + 1).at_part(0)).await.unwrap(), "{}", name);
            c.publish_compaction(AGE).await.expect("attach to the version before");
            drop(c);
            let want0 = if carry_on {
                w.flush().await.expect("the writer flushes again");
                6
            } else {
                drop(w);
                s.backdate_objects(60);
                let w = open(&s, fmt, &rt, FileFlags::rdwr()).await;
                drop(w);
                generation(0, n)
            };
            check_cold(&s, fmt, &rt, &name, n, |b| block(b, if b == 0 { want0 } else { generation(b, n) })).await;
        }
    }
}

/// A reader opened before the attach, and one opened after it, read what the
/// writer published, through the objects the compaction left in place; and after
/// the writer takes it in.
#[tokio::test]
async fn readers_read_on() {
    for fmt in FORMATS {
        let name = format!("oc-reader-{:?}", fmt);
        let rt = runtime();
        let n = 3000;
        let (s, mut w) = written(&name, fmt, &rt, n).await;
        let mut before = open(&s, fmt, &rt, FileFlags::rdonly()).await;
        compact(&s, fmt, &rt, n).await;
        let mut after = open(&s, fmt, &rt, FileFlags::rdonly()).await;
        w.write(0, &block(0, 6)).await.unwrap();
        w.flush().await.unwrap();
        let mut v = vec![0u8; BS];
        for b in (1..n).step_by(7) {
            before.read(b * BS, &mut v).await.unwrap();
            assert!(v == block(b, generation(b, n)), "{}: block {} before", name, b);
            after.read(b * BS, &mut v).await.unwrap();
            assert!(v == block(b, generation(b, n)), "{}: block {} after", name, b);
        }
    }
}

/// Two writers are still two writers.
#[tokio::test]
async fn a_second_writer_is_still_refused() {
    for fmt in FORMATS {
        let name = format!("oc-two-writers-{:?}", fmt);
        let rt = runtime();
        let n = 300;
        let (s, mut w) = written(&name, fmt, &rt, n).await;
        let mut other = open(&s, fmt, &rt, FileFlags::rdwr()).await;
        other.write(0, &block(0, 5)).await.unwrap();
        other.flush().await.expect("the other writer");
        w.write(BS, &block(1, 6)).await.unwrap();
        let e = w.flush().await.expect_err("refused");
        assert_eq!(e.kind(), ErrorKind::AlreadyExists, "{}: {}", name, e);
    }
}

/// Holds the first inode write, saying when it got there.
#[derive(Clone)]
struct HeldAttach(std::sync::Arc<tokio::sync::Notify>, std::sync::Arc<std::sync::atomic::AtomicBool>, std::sync::Arc<std::sync::atomic::AtomicBool>);

impl hyperfile::staging::StagingIntercept<MemoryStaging> for HeldAttach {
    fn before_flush_inode(&self, _: &MemoryStaging, _: &[u8], _: hyperfile::inode::FlushInodeFlag)
        -> std::pin::Pin<Box<dyn std::future::Future<Output = std::io::Result<()>> + '_ + Send>> {
        let first = self.1.swap(false, std::sync::atomic::Ordering::SeqCst);
        self.2.store(true, std::sync::atomic::Ordering::SeqCst);
        let go = self.0.clone();
        Box::pin(async move { if first { go.notified().await; } Ok(()) })
    }
    fn after_flush_inode(&self, _: &MemoryStaging, _: &[u8], _: hyperfile::inode::FlushInodeFlag)
        -> std::pin::Pin<Box<dyn std::future::Future<Output = std::io::Result<()>> + '_ + Send>> {
        Box::pin(async { Ok(()) })
    }
    fn after_remove_inode(&self, _: &MemoryStaging)
        -> std::pin::Pin<Box<dyn std::future::Future<Output = std::io::Result<()>> + '_ + Send>> {
        Box::pin(async { Ok(()) })
    }
}

/// Two compactors on distinct parts: the one whose attach lands first wins, and
/// the other, finding a compaction attached when it would write its own, gives
/// up rather than replace it.
#[tokio::test]
async fn an_attach_never_replaces_another() {
    use std::sync::{Arc, atomic::{AtomicBool, Ordering}};
    for fmt in FORMATS {
        let name = format!("oc-attach-race-{:?}", fmt);
        let rt = runtime();
        let n = 3000;
        let (s, w) = written(&name, fmt, &rt, n).await;
        drop(w);
        let mut crt = rt.clone();
        crt.take_up_on_open = false;
        let mut a = open(&s, fmt, &crt, FileFlags::rdwr()).await;
        let go = Arc::new(tokio::sync::Notify::new());
        let there = Arc::new(AtomicBool::new(false));
        a.staging_interceptor(HeldAttach(go.clone(), Arc::new(AtomicBool::new(true)), there.clone()));
        a.begin_compaction().await.expect("begin a");
        a.relocate(&[(0, n / 2)]).await.unwrap();
        let b_side = async {
            while !there.load(Ordering::SeqCst) {
                tokio::task::yield_now().await;
            }
            // a's objects are written and its attach held: b takes parts past them.
            let mut b = open(&s, fmt, &crt, FileFlags::rdwr()).await;
            b.begin_compaction().await.expect("begin b");
            b.relocate(&[((n / 2) as u64, n / 2)]).await.unwrap();
            let r = b.publish_compaction(AGE).await;
            go.notify_one();
            r
        };
        let (ra, rb) = tokio::join!(a.publish_compaction(AGE), b_side);
        let b = rb.expect("b attaches first");
        let e = ra.expect_err("a gives up");
        assert_eq!(e.kind(), ErrorKind::ResourceBusy, "{}: {}", name, e);
        let StoredCompaction::Pending(p) = attached(&s).await else { panic!("{}: nothing attached", name) };
        assert_eq!(p.parts, b.parts, "{}: the attached compaction is not b's", name);
        drop(a);
        check_cold(&s, fmt, &rt, &name, n, |b| block(b, generation(b, n))).await;
    }
}