hyperfile 0.12.0

//! The stream segment format (`BlockPtrFormat::StreamSegment`) on S3.
//!
//! `functional_stream_segment` covers the layout and the bookkeeping on memory
//! staging. This covers what only S3 shows: objects read from their ends by ranged
//! and suffix requests, multipart uploads filled while written and completed
//! conditionally, and a checkpoint taken up from an object's end — in both ways of
//! sending ahead: objects sent whole, and objects filled a piece at a time.
//!
//! Run on a general purpose bucket and on a directory bucket.
//!
//! ```bash
//! cargo test --release --test integration_s3_stream_segment -- --ignored --test-threads=1
//! ```

mod common;
use common::*;

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;

use hyperfile::config::{HyperFileConfig, HyperFileConfigBuilder, HyperFileMetaConfig, HyperFileRuntimeConfig};
use hyperfile::file::flags::{FileFlags, HyperFileFlags};
use hyperfile::file::hyper::Hyper;
use hyperfile::file::mode::{FileMode, HyperFileMode};
use hyperfile::file::ReadPolicy;
use hyperfile::meta_format::BlockPtrFormat;
use hyperfile::staging::config::StagingConfig;

const BS: usize = 4096;
const MIB: usize = 1 << 20;

/// How data is sent ahead: none, objects sent whole, objects filled in pieces.
#[derive(Debug, Clone, Copy)]
enum Mode {
    Off,
    Whole,
    Pieces,
}

const MODES: [Mode; 3] = [Mode::Off, Mode::Whole, Mode::Pieces];

fn config(uri: &str, mode: Mode) -> HyperFileConfig {
    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.data_cache_blocks = 0;
    rt.segment_mpu_chunk_size = 5 * MIB;
    rt.segment_max_object_bytes = 24 * MIB;
    rt.stream_segment_bytes = match mode { Mode::Off => 0, Mode::Whole => 4 * MIB, Mode::Pieces => 16 * MIB };
    let mut meta = HyperFileMetaConfig::default();
    meta.block_ptr_format = BlockPtrFormat::StreamSegment;
    HyperFileConfigBuilder::new()
        .with_staging_config(&StagingConfig::new_s3_uri(uri, None))
        .with_meta_config(&meta)
        .with_runtime_config(&rt)
        .build()
}

fn data(off: usize, len: usize, seed: u8) -> Vec<u8> {
    (off..off + len).map(|i| ((i / BS) as u8).wrapping_mul(13) ^ seed ^ (i as u8)).collect()
}

async fn create(client: &aws_sdk_s3::Client, uri: &str, mode: Mode) -> Hyper<'static> {
    Hyper::create(client.clone(), config(uri, mode), HyperFileFlags::from_flags(FileFlags::rdwr()),
        HyperFileMode::from_mode(FileMode::default_file())).await.expect("create")
}

async fn reopen(client: &aws_sdk_s3::Client, uri: &str, mode: Mode) -> Hyper<'static> {
    Hyper::open(client.clone(), config(uri, mode), HyperFileFlags::from_flags(FileFlags::rdwr())).await.expect("open")
}

async fn objects(client: &aws_sdk_s3::Client, uri: &str) -> Vec<(String, i64)> {
    let (bucket, prefix) = uri.trim_start_matches("s3://").split_once('/').unwrap();
    let mut v = Vec::new();
    let mut pages = client.list_objects_v2().bucket(bucket).prefix(format!("{}/", prefix)).into_paginator().send();
    while let Some(page) = pages.next().await {
        for o in page.expect("list").contents() {
            let k = o.key().unwrap_or("");
            if !k.ends_with("/inode") {
                v.push((k.rsplit('/').next().unwrap().to_string(), o.size().unwrap_or(0)));
            }
        }
    }
    v
}

async fn open_uploads(client: &aws_sdk_s3::Client, uri: &str) -> usize {
    let (bucket, prefix) = uri.trim_start_matches("s3://").split_once('/').unwrap();
    client.list_multipart_uploads().bucket(bucket).prefix(format!("{}/", prefix)).send().await
        .map(|o| o.uploads().len()).unwrap_or(0)
}

/// 48 MiB written in 1 MiB writes, part of it written again, flushed: cold, every
/// byte reads back, by `read` and by a plan of scattered ranges; every object is a
/// part of one checkpoint, within the object limit and a leaf; no upload is left
/// open.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn a_checkpoint_reads_back_cold_in_every_mode() {
    let _ = env_logger::try_init();
    let client = make_client().await;
    for mode in MODES {
        let tf = TestFile::new(&client).await;
        let mut h = create(&client, tf.uri(), mode).await;
        let len = 48 * MIB;
        for k in 0..48 {
            h.fs_write(k * MIB, &data(k * MIB, MIB, 1)).await.expect("write");
        }
        // Some of it again while sent ahead.
        for k in [3usize, 17, 40] {
            h.fs_write(k * MIB + 8192, &data(k * MIB + 8192, 64 * 1024, 2)).await.expect("rewrite");
        }
        let cno = h.fs_flush().await.expect("flush");
        let _ = h.fs_release().await;

        let mut want = data(0, len, 1);
        for k in [3usize, 17, 40] {
            let at = k * MIB + 8192;
            want[at..at + 64 * 1024].copy_from_slice(&data(at, 64 * 1024, 2));
        }
        let mut c = reopen(&client, tf.uri(), mode).await;
        let mut got = vec![0u8; len];
        let n = c.fs_read(0, &mut got).await.expect("read");
        let ranges: Vec<(u64, u64)> = (0..64u64).map(|k| (k * 750_000 % (len as u64 - 9000), 9000)).collect();
        let plan = c.fs_plan_reads(&ranges, &ReadPolicy::default()).await.expect("plan");
        let mut planned = vec![0u8; plan.len()];
        plan.execute(&mut planned, &tokio::sync::Semaphore::new(8)).await.expect("execute");
        let _ = c.fs_release().await;
        let objs = objects(&client, tf.uri()).await;
        let uploads = open_uploads(&client, tf.uri()).await;
        tf.cleanup(&client).await;

        assert_eq!(n, len, "{:?}", mode);
        assert!(got == want, "{:?}: the data read back differs", mode);
        let mut at = 0;
        for (off, l) in ranges {
            assert!(planned[at..at + l as usize] == want[off as usize..(off + l) as usize], "{:?}: planned read", mode);
            at += l as usize;
        }
        let name = format!("{:0>10}", cno);
        assert!(objs.iter().all(|(o, _)| o.starts_with(&format!("{}.", name))), "{:?}: {:?}", mode, objs);
        assert!(objs.iter().all(|(_, sz)| *sz as usize <= 30 * MIB), "{:?}: {:?}", mode, objs);
        // A build with the WAL sends nothing ahead: every mode is laid out by the flush.
        let mode = if cfg!(feature = "wal") { Mode::Off } else { mode };
        match mode {
            Mode::Off => assert!(objs.len() <= 4, "{:?}: {:?}", mode, objs),
            Mode::Whole => assert!(objs.len() >= 8, "{:?}: {:?}", mode, objs),
            Mode::Pieces => assert!(objs.len() >= 3 && objs.len() <= 6, "{:?}: {:?}", mode, objs),
        }
        assert_eq!(uploads, 0, "{:?}: an upload was left open", mode);
    }
}

/// Refuses the inode write while armed.
#[derive(Clone, Default)]
struct InodeDown(Arc<AtomicUsize>);

impl hyperfile::staging::StagingIntercept<hyperfile::staging::s3::S3Staging> for InodeDown {
    fn before_flush_inode(&self, _: &hyperfile::staging::s3::S3Staging, _: &[u8], _: hyperfile::inode::FlushInodeFlag)
        -> std::pin::Pin<Box<dyn std::future::Future<Output = std::io::Result<()>> + '_ + Send>> {
        let down = self.0.load(Ordering::SeqCst) > 0;
        Box::pin(async move {
            if down { Err(std::io::Error::new(std::io::ErrorKind::ConnectionReset, "inode refused")) } else { Ok(()) }
        })
    }
    fn after_flush_inode(&self, _: &hyperfile::staging::s3::S3Staging, _: &[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, _: &hyperfile::staging::s3::S3Staging)
        -> std::pin::Pin<Box<dyn std::future::Future<Output = std::io::Result<()>> + '_ + Send>> {
        Box::pin(async { Ok(()) })
    }
}

/// A checkpoint whose objects landed and whose inode write did not is taken up by
/// the next open, from part 0's end — data sent ahead included.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn a_checkpoint_whose_inode_did_not_land_is_taken_up() {
    let _ = env_logger::try_init();
    let client = make_client().await;
    for mode in MODES {
        let tf = TestFile::new(&client).await;
        let mut h = create(&client, tf.uri(), mode).await;
        h.fs_write(0, &data(0, MIB, 1)).await.unwrap();
        h.fs_flush().await.unwrap();
        let down = InodeDown::default();
        h.with_staging_interceptor(down.clone());
        for k in 0..24 {
            h.fs_write(k * MIB, &data(k * MIB, MIB, 5)).await.unwrap();
        }
        down.0.store(1, Ordering::SeqCst);
        let e = h.fs_flush().await.expect_err("the inode write is refused");
        let left = hyperfile::file::PublishIncomplete::of(&e).expect("says what it left");
        drop(h);
        // Past the wait a writable open gives a fresh commit's writer.
        tokio::time::sleep(std::time::Duration::from_secs(6)).await;
        let mut c = reopen(&client, tf.uri(), mode).await;
        let last = c.fs_last_cno();
        let mut got = vec![0u8; 24 * MIB];
        let n = c.fs_read(0, &mut got).await.expect("read");
        let _ = c.fs_release().await;
        tf.cleanup(&client).await;
        assert_eq!(last, left.as_cno(), "{:?}: the checkpoint was not taken up", mode);
        assert_eq!(n, 24 * MIB, "{:?}", mode);
        assert!(got == data(0, 24 * MIB, 5), "{:?}: taken-up data differs", mode);
    }
}

/// Several checkpoints over the same blocks, each sent ahead: the newest wins
/// everywhere, cold, and the earlier ones are still readable at their numbers.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn later_checkpoints_win_and_earlier_ones_stay_readable() {
    let _ = env_logger::try_init();
    let client = make_client().await;
    for mode in [Mode::Whole, Mode::Pieces] {
        let tf = TestFile::new(&client).await;
        let mut h = create(&client, tf.uri(), mode).await;
        let mut cnos = Vec::new();
        for round in 0..3u8 {
            for k in 0..20 {
                h.fs_write(k * MIB, &data(k * MIB, MIB, 10 + round)).await.unwrap();
            }
            cnos.push(h.fs_flush().await.unwrap());
        }
        let _ = h.fs_release().await;
        let mut c = reopen(&client, tf.uri(), mode).await;
        let mut got = vec![0u8; 20 * MIB];
        c.fs_read(0, &mut got).await.unwrap();
        let _ = c.fs_release().await;
        let mut first = Hyper::open_cno(client.clone(), config(tf.uri(), mode), HyperFileFlags::from_flags(FileFlags::rdonly()), cnos[0]).await.expect("open the first");
        let mut old = vec![0u8; 20 * MIB];
        first.fs_read(0, &mut old).await.unwrap();
        let _ = first.fs_release().await;
        tf.cleanup(&client).await;
        assert!(got == data(0, 20 * MIB, 12), "{:?}: the newest checkpoint does not win", mode);
        assert!(old == data(0, 20 * MIB, 10), "{:?}: the first checkpoint reads differently", mode);
    }
}

/// Turns the first successful answer to a write of an object whose name ends in
/// one of `suffixes` into a 500, after it landed; the SDK sends it again.
#[derive(Debug, Clone)]
struct LoseAnswers {
    suffixes: Arc<std::sync::Mutex<Vec<&'static str>>>,
    lost: Arc<AtomicUsize>,
}

#[derive(Debug, Clone)]
struct Marked;
impl aws_smithy_types::config_bag::Storable for Marked {
    type Storer = aws_smithy_types::config_bag::StoreReplace<Self>;
}

impl aws_sdk_s3::config::Intercept for LoseAnswers {
    fn name(&self) -> &'static str {
        "LoseAnswers"
    }
    fn read_before_execution(
        &self,
        ctx: &aws_sdk_s3::config::interceptors::BeforeSerializationInterceptorContextRef<'_>,
        cfg: &mut aws_sdk_s3::config::ConfigBag,
    ) -> Result<(), aws_sdk_s3::error::BoxError> {
        let input = ctx.input();
        let key = input.downcast_ref::<aws_sdk_s3::operation::put_object::PutObjectInput>().and_then(|i| i.key())
            .or_else(|| input.downcast_ref::<aws_sdk_s3::operation::complete_multipart_upload::CompleteMultipartUploadInput>()
                .and_then(|i| i.key()));
        if let Some(k) = key {
            let mut s = self.suffixes.lock().unwrap();
            if let Some(at) = s.iter().position(|x| k.ends_with(x)) {
                s.remove(at);
                cfg.interceptor_state().store_put(Marked);
            }
        }
        Ok(())
    }
    fn modify_before_deserialization(
        &self,
        ctx: &mut aws_sdk_s3::config::interceptors::BeforeDeserializationInterceptorContextMut<'_>,
        _rc: &aws_sdk_s3::config::RuntimeComponents,
        cfg: &mut aws_sdk_s3::config::ConfigBag,
    ) -> Result<(), aws_sdk_s3::error::BoxError> {
        if cfg.load::<Marked>().is_none() || !ctx.response().status().is_success() {
            return Ok(());
        }
        cfg.interceptor_state().unset::<Marked>();
        self.lost.fetch_add(1, Ordering::SeqCst);
        let resp = ctx.response_mut();
        *resp.status_mut() = 500u16.try_into().unwrap();
        resp.headers_mut().remove("content-length");
        *resp.body_mut() = aws_smithy_types::body::SdkBody::from(
            "<Error><Code>InternalError</Code><Message>answer lost by the test</Message></Error>");
        Ok(())
    }
}

/// An object sent ahead, and part 0, whose writes landed with their answers lost
/// and were sent again: the flush publishes, and the data reads back.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn writes_whose_answers_were_lost_still_publish() {
    let _ = env_logger::try_init();
    let client = make_client().await;
    let sdk = aws_config::from_env().region(aws_config::Region::new(test_region())).load().await;
    for mode in [Mode::Whole, Mode::Pieces] {
        let lose = LoseAnswers { suffixes: Arc::new(std::sync::Mutex::new(vec![".1", ".0"])), lost: Arc::default() };
        let faulty = aws_sdk_s3::Client::from_conf(aws_sdk_s3::config::Builder::from(&sdk).interceptor(lose.clone()).build());
        let tf = TestFile::new(&client).await;
        let mut h = create(&faulty, tf.uri(), mode).await;
        for k in 0..24 {
            h.fs_write(k * MIB, &data(k * MIB, MIB, 6)).await.unwrap();
        }
        let flushed = h.fs_flush().await;
        let _ = h.fs_release().await;
        let mut c = reopen(&client, tf.uri(), mode).await;
        let mut got = vec![0u8; 24 * MIB];
        let n = c.fs_read(0, &mut got).await.expect("read");
        let _ = c.fs_release().await;
        tf.cleanup(&client).await;
        flushed.unwrap_or_else(|e| panic!("{:?}: {:?}: {}", mode, e.kind(), e));
        assert_eq!(lose.lost.load(Ordering::SeqCst), 2, "{:?}: the answers were not both lost", mode);
        assert_eq!(n, 24 * MIB);
        assert!(got == data(0, 24 * MIB, 6), "{:?}", mode);
    }
}

/// The objects of checkpoints, by name, with their tail-first headers.
async fn stream_objects(client: &aws_sdk_s3::Client, uri: &str) -> Vec<(String, hyperfile::stream_segment::StreamHeader)> {
    let (bucket, prefix) = uri.trim_start_matches("s3://").split_once('/').unwrap();
    let mut v = Vec::new();
    for (name, _) in objects(client, uri).await {
        let Some((num, _)) = name.split_once('.') else { continue };
        if num.len() != 10 || !num.bytes().all(|b| b.is_ascii_digit()) {
            continue;
        }
        let out = client.get_object().bucket(bucket).key(format!("{}/{}", prefix, name))
            .range(format!("bytes=-{}", hyperfile::stream_segment::HEADER_SIZE)).send().await.expect("get");
        let bytes = out.body.collect().await.unwrap().into_bytes();
        let hdr = hyperfile::stream_segment::StreamHeader::read_from(&bytes)
            .unwrap_or_else(|e| panic!("{} does not end in a stream header: {}", name, e));
        v.push((name, hdr));
    }
    v
}

#[cfg(feature = "wal")]
fn wal_config(uri: &str) -> HyperFileConfig {
    let mut c = config(uri, Mode::Off);
    c.wal = hyperfile::wal::config::HyperFileWalConfig::new(&format!("{}/wal", uri));
    c
}

/// With the WAL, a direct flush writes the checkpoint in the stream layout, one
/// object of it per checkpoint here, and it reads back with the WAL and without.
#[cfg(feature = "wal")]
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn with_the_wal_a_flush_writes_the_stream_layout() {
    let _ = env_logger::try_init();
    let client = make_client().await;
    let tf = TestFile::new(&client).await;
    let mut h = Hyper::create(client.clone(), wal_config(tf.uri()), HyperFileFlags::from_flags(FileFlags::rdwr()),
        HyperFileMode::from_mode(FileMode::default_file())).await.expect("create");
    for k in 0..12 {
        h.fs_write(k * MIB, &data(k * MIB, MIB, 21)).await.unwrap();
    }
    let cno = h.fs_flush().await.expect("flush");
    let _ = h.fs_release().await;
    let objs = stream_objects(&client, tf.uri()).await;
    let mut with = Hyper::open(client.clone(), wal_config(tf.uri()), HyperFileFlags::from_flags(FileFlags::rdonly())).await.unwrap();
    let mut a = vec![0u8; 12 * MIB];
    with.fs_read(0, &mut a).await.unwrap();
    let _ = with.fs_release().await;
    let mut without = reopen(&client, tf.uri(), Mode::Off).await;
    let mut b = vec![0u8; 12 * MIB];
    without.fs_read(0, &mut b).await.unwrap();
    let _ = without.fs_release().await;
    tf.cleanup(&client).await;
    assert!(objs.iter().any(|(n, h)| n == &format!("{:0>10}.0", cno) && !h.is_partial()), "{:?}",
        objs.iter().map(|o| &o.0).collect::<Vec<_>>());
    assert!(a == data(0, 12 * MIB, 21), "read back with the WAL");
    assert!(b == data(0, 12 * MIB, 21), "read back without the WAL");
}

/// With the WAL on the reactor, a flush publishes from a pinned buffer — one
/// object in the stream layout — while reads go on: a read right after the flush
/// and a cold one both see the data.
#[cfg(all(feature = "wal", feature = "reactor"))]
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn with_the_wal_the_reactor_publishes_the_stream_layout() {
    use hyperfile::file::fh::HyperFileHandler;
    let _ = env_logger::try_init();
    let client = make_client().await;
    let tf = TestFile::new(&client).await;
    let reactor = hyperfile_reactor::Reactor::new_current().expect("reactor");
    let hyper = Hyper::create(client.clone(), wal_config(tf.uri()), HyperFileFlags::from_flags(FileFlags::rdwr()),
        HyperFileMode::from_mode(FileMode::default_file())).await.expect("create");
    let mut fh = HyperFileHandler::fh_from_hyper(&reactor, hyper).await.expect("handler");
    for k in 0..12 {
        fh.fh_write(k * MIB, &data(k * MIB, MIB, 22)).await.unwrap();
    }
    fh.fh_flush().await.expect("flush");
    let mut hot = vec![0u8; 12 * MIB];
    fh.fh_read(0, &mut hot).await.expect("read after the flush");
    // A second checkpoint over part of it.
    fh.fh_write(3 * MIB, &data(3 * MIB, MIB, 23)).await.unwrap();
    fh.fh_flush().await.expect("flush");
    let _ = fh.fh_release().await;
    drop(fh);
    let objs = stream_objects(&client, tf.uri()).await;
    let mut c = reopen(&client, tf.uri(), Mode::Off).await;
    let mut cold = vec![0u8; 12 * MIB];
    c.fs_read(0, &mut cold).await.unwrap();
    let _ = c.fs_release().await;
    tf.cleanup(&client).await;
    assert!(hot == data(0, 12 * MIB, 22), "read right after the flush");
    let mut want = data(0, 12 * MIB, 22);
    want[3 * MIB..4 * MIB].copy_from_slice(&data(3 * MIB, MIB, 23));
    assert!(cold == want, "cold read");
    assert!(objs.iter().filter(|(n, _)| n.ends_with(".0")).count() >= 2, "{:?}", objs.iter().map(|o| &o.0).collect::<Vec<_>>());
}

/// With the WAL, writes not flushed when the process dies are replayed at the next
/// open into a checkpoint in the stream layout.
#[cfg(feature = "wal")]
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn with_the_wal_a_replay_writes_the_stream_layout() {
    let _ = env_logger::try_init();
    let client = make_client().await;
    let tf = TestFile::new(&client).await;
    {
        let mut h = Hyper::create(client.clone(), wal_config(tf.uri()), HyperFileFlags::from_flags(FileFlags::rdwr()),
            HyperFileMode::from_mode(FileMode::default_file())).await.expect("create");
        h.fs_write(0, &data(0, MIB, 24)).await.unwrap();
        h.fs_flush().await.unwrap();
        for k in 1..6 {
            h.fs_write(k * MIB, &data(k * MIB, MIB, 24)).await.unwrap();
        }
        std::mem::forget(h);
    }
    let mut h = Hyper::open(client.clone(), wal_config(tf.uri()), HyperFileFlags::from_flags(FileFlags::rdwr())).await.expect("reopen");
    let last = h.fs_last_cno();
    let mut got = vec![0u8; 6 * MIB];
    let n = h.fs_read(0, &mut got).await.unwrap();
    let _ = h.fs_release().await;
    let objs = stream_objects(&client, tf.uri()).await;
    tf.cleanup(&client).await;
    assert_eq!(n, 6 * MIB, "the replay lost writes");
    assert!(got == data(0, 6 * MIB, 24));
    assert!(objs.iter().any(|(name, _)| name == &format!("{:0>10}.0", last)), "the replayed checkpoint: {:?}",
        objs.iter().map(|o| &o.0).collect::<Vec<_>>());
}

/// One container written by a build with the WAL and one without, in turn:
/// `STREAM_X_URI` names it, and `STREAM_X_STEP` is 1 and 3 in a WAL build, 2 and
/// 4 in a default one. Each step checks what the step before wrote; the last
/// removes it.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn interchange_step() {
    let (Ok(uri), Ok(step)) = (std::env::var("STREAM_X_URI"), std::env::var("STREAM_X_STEP")) else { return };
    let _ = env_logger::try_init();
    let client = make_client().await;
    let step: u8 = step.parse().unwrap();
    let mut want = vec![0u8; 32 * MIB];
    let gens: [(usize, usize, u8); 3] = [(0, 24, 31), (8, 32, 32), (16, 20, 33)];
    for (g, (from, to, seed)) in gens.iter().enumerate() {
        if g + 1 < step as usize {
            want[from * MIB..to * MIB].copy_from_slice(&data(from * MIB, (to - from) * MIB, *seed));
        }
    }
    let size = gens.iter().take(step as usize - 1).map(|g| g.1).max().unwrap_or(0) * MIB;
    #[cfg(feature = "wal")]
    let cfg = wal_config(&uri);
    #[cfg(not(feature = "wal"))]
    let cfg = config(&uri, Mode::Whole);
    assert_eq!(cfg!(feature = "wal"), step % 2 == 1, "step {} runs in the other build", step);
    let mut h = if step == 1 {
        Hyper::create(client.clone(), cfg, HyperFileFlags::from_flags(FileFlags::rdwr()),
            HyperFileMode::from_mode(FileMode::default_file())).await.expect("create")
    } else {
        Hyper::open(client.clone(), cfg, HyperFileFlags::from_flags(FileFlags::rdwr())).await.expect("open")
    };
    if size > 0 {
        let mut got = vec![0u8; size];
        let n = h.fs_read(0, &mut got).await.unwrap();
        assert_eq!(n, size, "step {}", step);
        assert!(got == want[..size], "step {}: what the other build wrote reads differently", step);
    }
    if let Some((from, to, seed)) = gens.get(step as usize - 1) {
        for k in *from..*to {
            h.fs_write(k * MIB, &data(k * MIB, MIB, *seed)).await.unwrap();
        }
        h.fs_flush().await.expect("flush");
    }
    let _ = h.fs_release().await;
    let objs = stream_objects(&client, &uri).await;
    println!("STEP {} objects {:?}", step, objs.iter().map(|o| &o.0).collect::<Vec<_>>());
    if step == 4 {
        let _ = Hyper::fs_unlink(&client, &uri).await;
    }
}

/// With room for one index node in memory: blocks that left memory once their
/// objects completed are read from those objects before the checkpoint is
/// published; after the flush, which drops clean index nodes, the leaves sent
/// ahead are read again from those objects too; and the data reads cold.
#[cfg(not(feature = "wal"))]
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn what_was_sent_ahead_is_read_back_from_where_it_went() {
    let _ = env_logger::try_init();
    let client = make_client().await;
    let tf = TestFile::new(&client).await;
    let mut cfg = config(tf.uri(), Mode::Whole);
    cfg.runtime.node_cache_blocks = 1;
    let mut h = Hyper::create(client.clone(), cfg.clone(), HyperFileFlags::from_flags(FileFlags::rdwr()),
        HyperFileMode::from_mode(FileMode::default_file())).await.expect("create");
    for k in 0..24 {
        h.fs_write(k * MIB, &data(k * MIB, MIB, 41)).await.unwrap();
    }
    // Let what was sent complete, and take it in.
    tokio::time::sleep(std::time::Duration::from_secs(2)).await;
    h.fs_write(24 * MIB, &data(24 * MIB, BS, 41)).await.unwrap();
    h.read_timing().reset();
    let mut before = vec![0u8; 24 * MIB];
    let n = h.fs_read(0, &mut before).await.expect("read before the flush");
    let reads_before = h.read_timing().snapshot();
    h.fs_flush().await.expect("flush");
    h.read_timing().reset();
    let mut after = vec![0u8; 24 * MIB];
    h.fs_read(0, &mut after).await.expect("read after the flush");
    let reads_after = h.read_timing().snapshot();
    let _ = h.fs_release().await;
    let mut c = Hyper::open(client.clone(), cfg, HyperFileFlags::from_flags(FileFlags::rdonly())).await.expect("open");
    let mut cold = vec![0u8; 24 * MIB];
    c.fs_read(0, &mut cold).await.unwrap();
    let _ = c.fs_release().await;
    tf.cleanup(&client).await;
    let want = data(0, 24 * MIB, 41);
    assert_eq!(n, 24 * MIB);
    assert!(before == want, "read before the flush");
    assert!(reads_before.data_gets > 0, "no block was read from the objects sent ahead: {:?}", reads_before);
    assert!(after == want, "read after the flush");
    assert!(reads_after.meta_gets > 0, "no leaf was read again: {:?}", reads_after);
    assert!(cold == want, "cold read");
}

/// Through the reactor's handler, in both ways of sending ahead: data is sent
/// while `fh_write` goes on, reads in the meantime see what was written, blocks
/// written again in flight keep their newer bytes, and after `fh_flush` the file
/// reads back cold, as one checkpoint of several objects.
#[cfg(all(feature = "reactor", not(feature = "wal")))]
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn the_reactor_sends_ahead_and_reads_meanwhile() {
    use hyperfile::file::fh::HyperFileHandler;
    let _ = env_logger::try_init();
    let client = make_client().await;
    for mode in [Mode::Whole, Mode::Pieces] {
        let tf = TestFile::new(&client).await;
        let reactor = hyperfile_reactor::Reactor::new_current().expect("reactor");
        let hyper = create(&client, tf.uri(), mode).await;
        let mut fh = HyperFileHandler::fh_from_hyper(&reactor, hyper).await.expect("handler");
        let mut want = data(0, 40 * MIB, 51);
        for k in 0..40 {
            fh.fh_write(k * MIB, &data(k * MIB, MIB, 51)).await.unwrap();
        }
        // Again, in part, while what was sent is in flight.
        for k in [2usize, 9, 30] {
            let at = k * MIB + 4096;
            let bytes = data(at, 128 * 1024, 52);
            fh.fh_write(at, &bytes).await.unwrap();
            want[at..at + bytes.len()].copy_from_slice(&bytes);
        }
        let mut meanwhile = vec![0u8; 40 * MIB];
        let n = fh.fh_read(0, &mut meanwhile).await.expect("read before the flush");
        let sent = fh.fh_flush_timing().await.unwrap().stream_sent_bytes;
        let cno = fh.fh_flush().await.expect("flush");
        let _ = fh.fh_release().await;
        drop(fh);
        let objs = objects(&client, tf.uri()).await;
        let mut c = reopen(&client, tf.uri(), mode).await;
        let mut cold = vec![0u8; 40 * MIB];
        c.fs_read(0, &mut cold).await.unwrap();
        let _ = c.fs_release().await;
        let uploads = open_uploads(&client, tf.uri()).await;
        tf.cleanup(&client).await;
        assert_eq!(n, 40 * MIB, "{:?}", mode);
        assert!(meanwhile == want, "{:?}: a read before the flush differs", mode);
        assert!(sent > 0, "{:?}: nothing was sent ahead", mode);
        assert!(cold == want, "{:?}: the cold read differs", mode);
        let name = format!("{:0>10}.", cno);
        assert!(objs.len() >= 3 && objs.iter().all(|(o, _)| o.starts_with(&name)), "{:?}: {:?}", mode, objs);
        assert_eq!(uploads, 0, "{:?}: an upload was left open", mode);
    }
}

/// A checkpoint past 1 GiB sent ahead at the default target and pieces: every
/// object within the 1 GiB one object can address, closed early rather than
/// past it; the data reads back cold, and no upload is left open.
#[cfg(not(feature = "wal"))]
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn a_checkpoint_past_one_gib_at_the_default_target() {
    let _ = env_logger::try_init();
    let client = make_client().await;
    let tf = TestFile::new(&client).await;
    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;
    let mut meta = HyperFileMetaConfig::default();
    meta.block_ptr_format = BlockPtrFormat::StreamSegment;
    let cfg = HyperFileConfigBuilder::new().with_staging_config(&StagingConfig::new_s3_uri(tf.uri(), None))
        .with_meta_config(&meta).with_runtime_config(&rt).build();
    let mut h = Hyper::create(client.clone(), cfg.clone(), HyperFileFlags::from_flags(FileFlags::rdwr()),
        HyperFileMode::from_mode(FileMode::default_file())).await.expect("create");
    let total = 1100 * MIB;
    let chunk = 16 * MIB;
    let mut off = 0;
    while off < total {
        h.fs_write(off, &data(off, chunk, 1)).await.expect("write");
        off += chunk;
    }
    let flushed = h.fs_flush().await;
    let _ = h.fs_release().await;
    let objs = objects(&client, tf.uri()).await;
    let mut bad = Vec::new();
    if flushed.is_ok() {
        let mut g = Hyper::open(client.clone(), cfg, HyperFileFlags::from_flags(FileFlags::rdonly())).await.expect("open");
        let mut v = vec![0u8; chunk];
        let mut off = 0;
        while off < total {
            let n = g.fs_read(off, &mut v).await.expect("read");
            if n != chunk || v != data(off, chunk, 1) { bad.push(off / MIB); }
            off += chunk;
        }
        let _ = g.fs_release().await;
    }
    let uploads = open_uploads(&client, tf.uri()).await;
    tf.cleanup(&client).await;
    flushed.expect("flush");
    let largest = objs.iter().map(|(_, n)| *n).max().unwrap_or(0);
    assert!(objs.len() >= 3, "nothing was sent ahead: {:?}", objs);
    assert!(largest <= 1 << 30, "an object of {} bytes: {:?}", largest, objs);
    assert!(bad.is_empty(), "MiB {:?} read back wrong", bad);
    assert_eq!(uploads, 0, "an upload was left open");
}

/// A source over a buffer, counting the bytes read from it.
#[cfg(not(feature = "wal"))]
struct CountingSource(Vec<u8>, Arc<AtomicUsize>);

#[cfg(not(feature = "wal"))]
impl hyperfile::block_source::BlockSource for CountingSource {
    fn read_at(&self, off: u64, buf: &mut [u8]) -> std::io::Result<()> {
        buf.copy_from_slice(&self.0[off as usize..off as usize + buf.len()]);
        self.1.fetch_add(buf.len(), Ordering::Relaxed);
        Ok(())
    }
}

/// Blocks written by reference are sent ahead, whole and in pieces, with the
/// dirty-data limits at their defaults: objects are stored before the flush,
/// the flush reads from the source only what was not sent, and the data reads
/// back cold. No upload is left open.
#[cfg(not(feature = "wal"))]
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn references_are_sent_ahead() {
    use btree_ondisk::NullNodeCache;
    use hyperfile::file::file::HyperFile;
    use hyperfile::file::HyperTrait;
    use hyperfile::staging::s3::S3Staging;
    use hyperfile::staging::Staging;
    let _ = env_logger::try_init();
    let client = make_client().await;
    for (mode, target) in [(Mode::Whole, 5 * MIB), (Mode::Pieces, 16 * MIB)] {
        let tf = TestFile::new(&client).await;
        let mut rt = HyperFileRuntimeConfig::default();
        rt.data_cache_blocks = 0;
        rt.segment_mpu_chunk_size = 5 * MIB;
        rt.stream_segment_bytes = target;
        let mut meta = HyperFileMetaConfig::default();
        meta.block_ptr_format = BlockPtrFormat::StreamSegment;
        let cfg = HyperFileConfigBuilder::new().with_staging_config(&StagingConfig::new_s3_uri(tf.uri(), None))
            .with_meta_config(&meta).with_runtime_config(&rt).build();
        let mut h = Hyper::create(client.clone(), cfg.clone(), HyperFileFlags::from_flags(FileFlags::rdwr()),
            HyperFileMode::from_mode(FileMode::default_file())).await.expect("create");
        let _ = h.fs_release().await;
        drop(h);
        let staging = S3Staging::from(&client, cfg.staging.clone(), cfg.runtime.clone()).await.expect("staging");
        let loader = staging.to_block_loader();
        let mut f = HyperFile::open(staging, loader, NullNodeCache, cfg.clone(), HyperFileFlags::from_flags(FileFlags::rdwr()))
            .await.expect("open");
        let total = 64 * MIB;
        let read = Arc::new(AtomicUsize::new(0));
        f.set_block_source(Some(Arc::new(CountingSource(data(0, total, 7), read.clone())))).unwrap();
        for k in 0..total / MIB {
            f.write_ref(k * MIB, MIB).await.expect("write_ref");
        }
        tokio::time::sleep(std::time::Duration::from_secs(2)).await;
        let before = objects(&client, tf.uri()).await.len();
        let read_ahead = read.load(Ordering::Relaxed);
        let flushed = f.flush().await;
        let read_all = read.load(Ordering::Relaxed);
        drop(f);
        let mut got = Vec::new();
        if flushed.is_ok() {
            let mut g = Hyper::open(client.clone(), cfg, HyperFileFlags::from_flags(FileFlags::rdonly())).await.expect("open");
            got = vec![0u8; total];
            g.fs_read(0, &mut got).await.expect("read");
            let _ = g.fs_release().await;
        }
        let uploads = open_uploads(&client, tf.uri()).await;
        tf.cleanup(&client).await;
        flushed.expect("flush");
        assert!(before >= 2, "{:?}: nothing was stored before the flush ({} objects)", mode, before);
        assert!(read_ahead >= 32 * MIB, "{:?}: {} bytes read ahead", mode, read_ahead);
        assert!(read_all <= total + 16 * MIB, "{:?}: the source was read for {} bytes of {}", mode, read_all, total);
        assert!(got == data(0, total, 7), "{:?}: read back wrong", mode);
        assert_eq!(uploads, 0, "{:?}: an upload was left open", mode);
    }
}

/// Sent whole with the target at the default piece size, 16 MiB, each object is
/// one PutObject: none is a multipart upload, though each has its leaves,
/// summary and header on top of its data. The data reads back cold.
#[cfg(not(feature = "wal"))]
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn objects_sent_whole_at_the_piece_size_are_one_request_each() {
    let _ = env_logger::try_init();
    let client = make_client().await;
    let tf = TestFile::new(&client).await;
    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 = rt.segment_mpu_chunk_size;
    let mut meta = HyperFileMetaConfig::default();
    meta.block_ptr_format = BlockPtrFormat::StreamSegment;
    let cfg = HyperFileConfigBuilder::new().with_staging_config(&StagingConfig::new_s3_uri(tf.uri(), None))
        .with_meta_config(&meta).with_runtime_config(&rt).build();
    let mut h = Hyper::create(client.clone(), cfg.clone(), HyperFileFlags::from_flags(FileFlags::rdwr()),
        HyperFileMode::from_mode(FileMode::default_file())).await.expect("create");
    let total = 96 * MIB;
    for off in (0..total).step_by(MIB) {
        h.fs_write(off, &data(off, MIB, 3)).await.expect("write");
        if off % (4 * MIB) == 0 { tokio::time::sleep(std::time::Duration::from_millis(10)).await; }
    }
    let flushed = h.fs_flush().await;
    let _ = h.fs_release().await;
    let (bucket, prefix) = tf.uri().trim_start_matches("s3://").split_once('/').unwrap();
    let mut multipart = Vec::new();
    let mut parts = 0;
    for (name, _) in objects(&client, tf.uri()).await {
        // Data parts: `<cno>.<part>` with a part past 0.
        let Some((_, p)) = name.rsplit_once('.') else { continue };
        if p.parse::<u32>().map(|p| p == 0).unwrap_or(true) { continue; }
        parts += 1;
        let head = client.head_object().bucket(bucket).key(format!("{}/{}", prefix, name)).part_number(1).send().await;
        if head.map(|h| h.parts_count().unwrap_or(1) > 1).unwrap_or(false) {
            multipart.push(name);
        }
    }
    let mut got = Vec::new();
    if flushed.is_ok() {
        let mut g = Hyper::open(client.clone(), cfg, HyperFileFlags::from_flags(FileFlags::rdonly())).await.expect("open");
        got = vec![0u8; total];
        g.fs_read(0, &mut got).await.expect("read");
        let _ = g.fs_release().await;
    }
    tf.cleanup(&client).await;
    flushed.expect("flush");
    assert!(parts >= 4, "only {} data parts", parts);
    assert!(multipart.is_empty(), "sent as multipart uploads: {:?}", multipart);
    assert!(got == data(0, total, 3), "read back wrong");
}