hyperfile 0.11.2

//! Index nodes evicted and read back from S3 while others are changed.
//!
//! S3 staging's default loader (`meta_loader_batch`) returns, with each index
//! node asked for, the other nodes of the same stored segment. btree-ondisk
//! 0.20.0 installed those over nodes already in memory, including ones changed
//! since they were read, and the next flush failed while assigning index
//! pointers with "key not found through btree node lookup" — then failed again
//! on every retry. A downstream container hit it once flushes touched more index
//! nodes than the node cache holds, which checkpoints past 1 GiB made common.
//! Fixed in btree-ondisk 0.20.1.
//!
//! Built as that container builds its file — `S3Staging`, its own loader, and
//! `NullNodeCache`, so an evicted node is read back from S3 — with a node cache
//! of 16, so evictions come every flush. On btree-ondisk 0.20.0 the sixth flush
//! failed.
//!
//! ```bash
//! cargo test --release --test integration_s3_node_reload -- --ignored --test-threads=1
//! ```

mod common;
use common::*;

use btree_ondisk::NullNodeCache;
use hyperfile::config::{HyperFileConfigBuilder, HyperFileRuntimeConfig};
use hyperfile::file::file::HyperFile;
use hyperfile::file::flags::{FileFlags, HyperFileFlags};
use hyperfile::file::hyper::Hyper;
use hyperfile::file::mode::FileMode;
use hyperfile::file::HyperTrait;
use hyperfile::meta_loader::s3::S3BlockLoader;
use hyperfile::staging::{config::StagingConfig, s3::S3Staging, Staging};

type S3File = HyperFile<'static, S3Staging, S3BlockLoader, NullNodeCache>;

const MIB: usize = 1 << 20;

fn runtime() -> HyperFileRuntimeConfig {
    let mut r = HyperFileRuntimeConfig::default();
    // Every flush evicts, so later lookups read nodes back from S3.
    r.node_cache_blocks = 16;
    r.segment_max_object_bytes = 64 * MIB;
    // Only the flushes below publish.
    r.data_cache_dirty_max_bytes_threshold = 1 << 40;
    r.data_cache_dirty_max_blocks_threshold = (1 << 40) / 4096;
    r.data_cache_dirty_max_flush_interval = u64::MAX / 2;
    r.segment_buffer_size = 1 << 40;
    r.data_cache_blocks = 1;
    r
}

struct Rng(u64);

impl Rng {
    fn next(&mut self) -> u64 {
        self.0 ^= self.0 << 13;
        self.0 ^= self.0 >> 7;
        self.0 ^= self.0 << 17;
        self.0
    }
}

/// Sequential writes and then, flush after flush, 1 MiB overwrites, small
/// writes across a 16 GiB address space and 1 MiB zero writes far from the data:
/// every flush changes index nodes spread over several stored segments while
/// others are read back. Every flush succeeds, and the file reads back as
/// written from a fresh handle.
#[tokio::test]
#[ignore]
async fn flushes_succeed_while_index_nodes_are_read_back_from_s3() {
    let _ = env_logger::try_init();
    let client = make_client().await;
    let tf = TestFile::new(&client).await;
    const SIZE: usize = 16 << 30;
    {
        let mut h = Hyper::fs_open_or_create_with_default_opt(&client, tf.uri(), FileFlags::rdwr(), FileMode::default_file())
            .await.expect("create");
        h.fs_truncate(SIZE).await.expect("truncate");
        h.fs_flush().await.expect("flush");
        let _ = h.fs_release().await;
    }
    let rt = runtime();
    let sc = StagingConfig::new_s3_uri(tf.uri(), None);
    let config = HyperFileConfigBuilder::new().with_staging_config(&sc).with_runtime_config(&rt).build();
    let staging = S3Staging::from(&client, config.staging.clone(), config.runtime.clone()).await.expect("staging");
    let loader = staging.to_block_loader();
    let mut f = S3File::open(staging, loader, NullNodeCache, config, HyperFileFlags::from_flags(FileFlags::rdwr()))
        .await.expect("open");

    // What the file holds, 1 MiB at a time: the fill byte of each written MiB.
    let mut model = std::collections::BTreeMap::<usize, u8>::new();
    let mut failed = None;
    let mut rng = Rng(7);
    let mut buf = vec![0u8; MIB];
    'run: {
        for i in 0..128usize {
            buf.fill((i % 251) as u8 + 1);
            f.write(i * MIB, &buf).await.expect("write");
            model.insert(i, (i % 251) as u8 + 1);
            if i == 69 || i == 127 {
                if let Err(e) = f.flush().await { failed = Some(("sequential", e)); break 'run; }
            }
        }
        for round in 0..8 {
            for _ in 0..20 {
                let i = rng.next() as usize % 128;
                let v = (rng.next() % 250) as u8 + 1;
                buf.fill(v);
                f.write(i * MIB, &buf).await.expect("write");
                model.insert(i, v);
            }
            for _ in 0..40 {
                // Small writes beyond the data, where the model does not look.
                let len = 512 << (rng.next() % 4);
                let at = (128 * MIB + (rng.next() as usize % ((SIZE - 128 * MIB) >> 9)) * 512).min(SIZE - len);
                f.write(at, &buf[..len]).await.expect("small write");
            }
            for _ in 0..60 {
                let at = (8 << 30) + (rng.next() as usize % 6000) * MIB;
                f.write_zero(at.min(SIZE - MIB), MIB).await.expect("write_zero");
            }
            if let Err(e) = f.flush().await { failed = Some(("overwrite", e)); let _ = round; break 'run; }
        }
    }
    drop(f);
    let mut wrong = Vec::new();
    if failed.is_none() {
        let mut c = Hyper::fs_open(&client, tf.uri(), FileFlags::rdonly()).await.expect("reopen");
        for (i, v) in &model {
            let n = c.fs_read(i * MIB, &mut buf).await.expect("read");
            if n != MIB || buf.iter().any(|b| b != v) {
                wrong.push(*i);
            }
        }
        let _ = c.fs_release().await;
    }
    tf.cleanup(&client).await;
    if let Some((phase, e)) = failed {
        panic!("a flush in the {} phase failed: {}", phase, e);
    }
    assert!(wrong.is_empty(), "{} MiB read back wrong: {:?}", wrong.len(), wrong);
}