hyperdir 0.5.0

//! End-to-end test for the directory **reachability** API (btree-ondisk 0.20's
//! `collect_meta_ptrs`, exposed as `HyperDir::fs_reachable_meta_ptrs` /
//! `fs_reachable_segments`).
//!
//! Requires AWS credentials plus `S3_BUCKET` and `S3_REGION`. `#[ignore]`; run:
//!
//! ```ignore
//! source ../hyperdir-env.sh
//! cargo test --test e2e_reachability -- --ignored --nocapture
//! ```

use aws_sdk_s3::Client;
use uuid::Uuid;
use hyperdir::{HyperDirLayout, ROOT_DIR_UUID};
use hyperdir::hyper::HyperDir;
use hyperfile::file::flags::FileFlags;
use hyperfile::file::mode::FileMode;

fn bucket() -> String {
    std::env::var("S3_BUCKET").expect("S3_BUCKET not set (source ../hyperdir-env.sh)")
}

async fn make_client() -> Client {
    let region = std::env::var("S3_REGION").expect("S3_REGION not set");
    let config = aws_config::from_env().region(aws_config::Region::new(region)).load().await;
    Client::new(&config)
}

fn dir_mode() -> FileMode { FileMode::from(0o755) }

async fn purge_prefix(client: &Client, bucket: &str, prefix: &str) {
    let mut stream = client.list_objects_v2().bucket(bucket).prefix(prefix).into_paginator().send();
    let mut keys: Vec<String> = Vec::new();
    while let Some(page) = stream.next().await {
        let page = page.expect("list for cleanup");
        for o in page.contents() { if let Some(k) = o.key() { keys.push(k.to_string()); } }
    }
    for chunk in keys.chunks(1000) {
        let ids: Vec<_> = chunk.iter().map(|k| {
            aws_sdk_s3::types::ObjectIdentifier::builder().key(k).build().unwrap()
        }).collect();
        let del = aws_sdk_s3::types::Delete::builder().set_objects(Some(ids)).build().unwrap();
        let _ = client.delete_objects().bucket(bucket).delete(del).send().await;
    }
}

#[tokio::test]
#[ignore]
async fn e2e_directory_reachability() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e-reach/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);

    let result = run(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("e2e reachability flow");
}

async fn run(client: &Client, bucket: &str, layout: &HyperDirLayout) -> std::io::Result<()> {
    let _root = HyperDir::fs_create_root(client, layout, bucket, FileFlags::rdwr(), dir_mode()).await?;

    // A directory entry is ~448 bytes and the inline root area is 56 bytes, so
    // the root is an internal node from the first insert (design.md §13) --
    // i.e. real B+tree metadata nodes exist as soon as the dir is populated.
    // Create enough children to span several 4 KiB meta-block leaves.
    for i in 0..40 {
        let name = format!("child{:03}", i);
        let _ = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, &name,
                                            FileFlags::rdwr(), dir_mode()).await?;
    }
    let mut root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?;
    let stats = root.fs_compact().await?;
    assert_eq!(stats.entries_added, 40, "all children folded into the bmap");
    drop(root);

    // Reopen from persisted state: the bmap is now a flushed B+tree whose
    // metadata nodes have externally assigned locations.
    let root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdonly()).await?;

    let ptrs = root.fs_reachable_meta_ptrs().await?;
    assert!(!ptrs.is_empty(),
            "a populated directory must reference at least one metadata node, got {:?}", ptrs);

    // Every reported location must be a real one (not dummy/invalid), and the
    // set must be duplicate-free -- a reclaimer subtracts this from what exists.
    let unique: std::collections::HashSet<_> = ptrs.iter().copied().collect();
    assert_eq!(unique.len(), ptrs.len(), "reachable ptrs must not repeat: {:?}", ptrs);

    // Segment view: what a reclaimer would keep.
    let segs = root.fs_reachable_segments().await?;
    assert!(!segs.is_empty(), "metadata nodes must live in at least one staging segment");
    println!("reachable: {} meta node(s) across {} segment(s): {:?}", ptrs.len(), segs.len(), segs);

    // Stable across repeated calls (pure read, no mutation).
    let again = root.fs_reachable_meta_ptrs().await?;
    assert_eq!(again.len(), ptrs.len(), "reachability is a pure read and must be stable");

    // Growing the directory can only add metadata nodes, never lose the ones
    // already referenced (the regression the upstream data loss came from).
    drop(root);
    for i in 40..80 {
        let name = format!("child{:03}", i);
        let _ = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, &name,
                                            FileFlags::rdwr(), dir_mode()).await?;
    }
    let mut root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?;
    let _ = root.fs_compact().await?;
    drop(root);
    let root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdonly()).await?;
    let grown = root.fs_reachable_meta_ptrs().await?;
    assert!(grown.len() >= ptrs.len(),
            "a grown directory must not report fewer metadata nodes ({} -> {})", ptrs.len(), grown.len());
    assert_eq!(root.fs_read_dir().await?.len(), 80, "all 80 children readable");
    println!("after growth: {} meta node(s)", grown.len());

    Ok(())
}