hyperdir 0.7.0

//! Deterministic fault-injection tests (real S3 + the `fault-injection` feature):
//!
//!   set -a; . ../hyperdir-env.sh; set +a
//!   cargo test --features fault-injection --test e2e_fault -- --ignored --test-threads=1
//!
//! These pin down behaviour that otherwise only shows up when S3 happens to fail
//! mid-operation. The scatter listing used to swallow a failed LIST page and
//! return a PARTIAL list, which made a deleted name resolve as live; the contract
//! is that a failed listing must surface as an error so the caller can retry,
//! never as a wrong answer.
//!
//! The whole file compiles away without the feature, so `--all-targets` builds
//! (and clippy) work in a default checkout.
#![cfg(feature = "fault-injection")]

use std::time::Duration;

use aws_sdk_s3::Client;
use hyperdir::fault::S3FaultInterceptor;
use hyperdir::hyper::HyperDir;
use hyperdir::HyperDirLayout;
use hyperfile::file::flags::FileFlags;
use hyperfile::file::mode::FileMode;
use hyperfile::inode::Inode;
use uuid::Uuid;

async fn client_with(fault: Option<S3FaultInterceptor>) -> Client {
    let region = std::env::var("S3_REGION").expect("S3_REGION not set");
    let cfg = aws_config::from_env().region(aws_config::Region::new(region)).load().await;
    let mut b = aws_sdk_s3::config::Builder::from(&cfg).interceptor(hyperdir::S3ClockInterceptor);
    if let Some(f) = fault {
        b = b.interceptor(f);
    }
    // Retries would paper over an injected fault; this test is about what the
    // caller sees when the request ultimately fails.
    b = b.retry_config(aws_sdk_s3::config::retry::RetryConfig::disabled());
    Client::from_conf(b.build())
}

async fn purge(client: &Client, bucket: &str, prefix: &str) {
    let mut stream = client.list_objects_v2().bucket(bucket).prefix(format!("{prefix}/")).into_paginator().send();
    let mut keys = Vec::new();
    while let Some(page) = stream.next().await {
        let Ok(p) = page else { break };
        for o in p.contents() {
            if let Some(k) = o.key() { keys.push(k.to_string()); }
        }
    }
    for k in keys {
        let _ = client.delete_object().bucket(bucket).key(k).send().await;
    }
}

/// A failed LIST page during resolve must surface as an error, not as a stale
/// answer. Without the fix the failing page was skipped, the (partial) listing
/// omitted the name's PreDelete, and an unlinked file resolved as still live.
#[tokio::test(flavor = "multi_thread")]
#[ignore]
async fn e2e_fault_resolve_errors_when_listing_fails() {
    let (Ok(bkt), Ok(_)) = (std::env::var("S3_BUCKET"), std::env::var("S3_REGION")) else {
        eprintln!("skipping: S3_BUCKET/S3_REGION not set");
        return;
    };
    let fs = format!("hyperdir-fault/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&fs);
    let clean = client_with(None).await;

    HyperDir::fs_create_root(&clean, &layout, &bkt, FileFlags::rdwr(), FileMode::from(0o755))
        .await
        .expect("create root");
    let root = Uuid::nil();

    // A file that exists, then is unlinked: its scatter holds a Create and then a
    // PreDelete for the same identity.
    let mut inl = Inode::default_file().to_raw(unsafe { std::mem::zeroed() });
    inl.i_mode = libc::S_IFREG | 0o644;
    inl.set_inline(b"hello");
    let uuid = HyperDir::fs_create_inline(&clean, &layout, &bkt, &root, "f", &inl)
        .await
        .expect("create inline");
    assert!(HyperDir::fs_resolve_entry_fast(&clean, &layout, &bkt, &root, "f").await.unwrap().is_some(),
        "present before unlink");
    HyperDir::fs_unlink(&clean, &layout, &bkt, &root, "f", &uuid, false, None)
        .await
        .expect("unlink");
    assert!(HyperDir::fs_resolve_entry_fast(&clean, &layout, &bkt, &root, "f").await.unwrap().is_none(),
        "absent after unlink (no faults)");

    // Now fail every ListObjectsV2. Resolve must report the failure; it must not
    // fall back to a partial listing and resurrect the deleted name.
    let fault = S3FaultInterceptor::new(&["ListObjectsV2"], 1, 0);
    let faulty = client_with(Some(fault.clone())).await;
    let res = HyperDir::fs_resolve_entry_fast(&faulty, &layout, &bkt, &root, "f").await;
    assert!(fault.injected() > 0, "the fault interceptor must have fired");
    match res {
        Err(e) => eprintln!("  resolve surfaced the listing failure as expected: {e}"),
        Ok(Some(_)) => panic!("resolve resurrected a deleted name from a partial listing"),
        Ok(None) => panic!("resolve reported a definitive answer despite a failed listing"),
    }

    // With faults removed the answer is still correct (nothing was corrupted).
    assert!(HyperDir::fs_resolve_entry_fast(&clean, &layout, &bkt, &root, "f").await.unwrap().is_none(),
        "still absent once listing works again");

    purge(&clean, &bkt, &fs).await;
    let _ = Duration::from_secs(0);
}