hyperfile 0.10.2

//! Checkpoints larger than one object, on S3.
//!
//! A block pointer addresses at most 1 GiB within one object. A flush larger than
//! that used to report success and store blocks that read back wrong; it is now
//! written as several objects — data-only parts, then part 0 with the index and
//! the inode — and published as one checkpoint. `functional_split_checkpoint`
//! covers the layout on memory staging with a small limit; this runs it at the
//! real one, where the offsets that used to wrap are actually reached, and with a
//! smaller limit whose parts are themselves multipart uploads.
//!
//! The full-size test writes and reads 1100 MiB.
//!
//! ```bash
//! cargo test --release --test integration_s3_large_checkpoint -- --ignored --test-threads=1
//! ```

mod common;
use common::*;

use hyperfile::config::HyperFileRuntimeConfig;
use hyperfile::file::flags::FileFlags;
use hyperfile::file::hyper::Hyper;
use hyperfile::file::mode::FileMode;
use hyperfile::segment::Segment;
use hyperfile::SegmentId;

const MIB: usize = 1024 * 1024;

fn prefix_of(uri: &str) -> String {
    uri.trim_start_matches("s3://").splitn(2, '/').nth(1).unwrap().to_string()
}

/// Nothing publishes or writes a partial on its own: the flush is one checkpoint.
fn runtime(max_object: usize) -> HyperFileRuntimeConfig {
    let mut r = HyperFileRuntimeConfig::default();
    r.segment_max_object_bytes = max_object;
    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
}

/// Each MiB tagged with its own index, so a block read from the wrong place shows.
fn mib_of(i: usize) -> Vec<u8> {
    let mut v = vec![(i % 251) as u8 + 1; MIB];
    v[..8].copy_from_slice(&(i as u64).to_le_bytes());
    v
}

/// The sizes of checkpoint `cno`'s objects, by part, as S3 reports them.
async fn object_sizes(client: &aws_sdk_s3::Client, prefix: &str, cno: u64) -> Vec<(u16, i64)> {
    let mut out = Vec::new();
    for k in 0..64u16 {
        let key = format!("{}/{}", prefix, Segment::segid_to_staging_file_id(SegmentId::new_from_cno(cno).at_part(k)));
        match client.head_object().bucket(test_bucket()).key(&key).send().await {
            Ok(h) => out.push((k, h.content_length().unwrap_or(0))),
            Err(_) => break,
        }
    }
    out
}

async fn open_uploads(client: &aws_sdk_s3::Client, prefix: &str) -> usize {
    client.list_multipart_uploads().bucket(test_bucket()).prefix(format!("{}/", prefix))
        .send().await.expect("list multipart uploads").uploads().len()
}

/// Writes `mib` MiB in one checkpoint under `max_object`, then reads it back from a
/// fresh handle. Returns the flush's result, the objects' sizes, and the MiB that
/// came back wrong.
async fn write_and_read_back(mib: usize, max_object: usize) -> (std::io::Result<u64>, Vec<(u16, i64)>, Vec<usize>, usize) {
    let client = make_client().await;
    let tf = TestFile::new(&client).await;
    let p = prefix_of(tf.uri());
    {
        let mut h = Hyper::fs_open_or_create_with_default_opt(&client, tf.uri(), FileFlags::rdwr(), FileMode::default_file())
            .await.expect("create");
        let _ = h.fs_release().await;
    }
    let res = {
        let mut h = Hyper::fs_open_opt(&client, tf.uri(), FileFlags::rdwr(), &runtime(max_object)).await.expect("open");
        for i in 0..mib {
            h.fs_write(i * MIB, &mib_of(i)).await.expect("write");
        }
        let res = h.fs_flush().await;
        let _ = h.fs_release().await;
        res
    };
    let sizes = match &res { Ok(cno) => object_sizes(&client, &p, *cno).await, Err(_) => Vec::new() };
    let mut wrong = Vec::new();
    if res.is_ok() {
        let mut c = Hyper::fs_open(&client, tf.uri(), FileFlags::rdonly()).await.expect("reopen");
        let mut buf = vec![0u8; MIB];
        for i in 0..mib {
            let n = c.fs_read(i * MIB, &mut buf).await.expect("read");
            if n != MIB || buf != mib_of(i) {
                wrong.push(i);
            }
        }
        let _ = c.fs_release().await;
    }
    let open = open_uploads(&client, &p).await;
    tf.cleanup(&client).await;
    (res, sizes, wrong, open)
}

/// 1100 MiB in one flush: past what one object can address, so two objects, and
/// every MiB reads back from a cold handle. On 0.8.0 this flush succeeded and the
/// MiB past the first gigabyte read back wrong.
#[tokio::test]
#[ignore]
async fn a_checkpoint_past_1_gib_is_written_as_two_objects_and_reads_back() {
    let _ = env_logger::try_init();
    let (res, sizes, wrong, open) = write_and_read_back(1100, 1 << 30).await;
    res.expect("flush");
    assert_eq!(open, 0, "an upload was left open");
    assert_eq!(sizes.len(), 2, "1100 MiB was not two objects: {:?}", sizes);
    assert!(sizes.iter().all(|(_, n)| *n as usize <= 1 << 30), "an object is past 1 GiB: {:?}", sizes);
    assert!(wrong.is_empty(), "{} of 1100 MiB read back wrong, first {:?}", wrong.len(), wrong.first());
}

/// Under a 40 MiB limit, 150 MiB is four objects, each larger than a multipart
/// chunk, so every part is itself a multipart upload; it reads back.
#[tokio::test]
#[ignore]
async fn under_a_smaller_limit_each_part_is_its_own_multipart_upload() {
    let _ = env_logger::try_init();
    let (res, sizes, wrong, open) = write_and_read_back(150, 40 * MIB).await;
    res.expect("flush");
    assert_eq!(open, 0, "an upload was left open");
    assert_eq!(sizes.len(), 4, "150 MiB in 40 MiB objects: {:?}", sizes);
    assert!(sizes.iter().all(|(_, n)| *n as usize <= 40 * MIB), "an object is past the limit: {:?}", sizes);
    assert!(wrong.is_empty(), "{} of 150 MiB read back wrong, first {:?}", wrong.len(), wrong.first());
}