hyperdir 0.5.0

//! End-to-end test against a real S3 bucket.
//!
//! Requires AWS credentials in the environment plus `S3_BUCKET` and
//! `S3_REGION`. Marked `#[ignore]`; run explicitly:
//!
//! ```ignore
//! source ../env.sh
//! cargo test --test e2e_s3 -- --ignored --nocapture
//! ```
//!
//! Each run uses a unique base prefix for isolation and deletes everything
//! under it on success. No environment values are hard-coded; the bucket and
//! region are read from the environment at run time.

use std::time::Duration;
use aws_sdk_s3::Client;
use uuid::Uuid;
use hyperdir::HyperDirLayout;
use hyperdir::ROOT_DIR_UUID;
use hyperdir::ScatterFirstInterceptor;
use hyperdir::hyper::HyperDir;
use hyperfile::file::flags::FileFlags;
use hyperfile::file::mode::FileMode;
use hyperfile::file::hyper::Hyper;
use hyperfile::staging::s3::S3Staging;
use hyperfile::staging::config::StagingConfig;
use hyperfile::config::HyperFileRuntimeConfig;
use hyperfile::inode::Inode;
use hyperfile::ondisk::InodeRaw;

fn bucket() -> String {
    std::env::var("S3_BUCKET").expect("S3_BUCKET not set (source ../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)
}

/// Delete every object under `prefix` (test cleanup).
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");
        if let Some(objs) = page.contents {
            for o in objs {
                if let Some(k) = o.key { keys.push(k); }
            }
        }
    }
    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;
    }
}

fn names(entries: &[hyperdir::file::DirFileEntry]) -> std::collections::BTreeSet<String> {
    entries.iter().map(|e| e.name.clone()).collect()
}

// Requires a large thread stack: debug-build async futures here are large
// enough to overflow the default ~2 MiB test-thread stack. `RUST_MIN_STACK`
// is exported by ../env.sh (sourced before running this test).
#[tokio::test]
#[ignore]
async fn e2e_directory_lifecycle() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);

    let result = run(&client, &bucket, &layout).await;
    // Always attempt cleanup before asserting the outcome.
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("e2e flow");
}

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

    // 2. mkdir alpha, beta under root (each emits a Create scatter)
    let (_a, _a_uuid) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, "alpha", FileFlags::rdwr(), dir_mode()).await?;
    let (_b, b_uuid) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, "beta", FileFlags::rdwr(), dir_mode()).await?;

    // 3. compact root: both children consolidated into the bmap
    let mut root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?;
    let stats = root.fs_compact().await?;
    assert_eq!(stats.entries_added, 2, "two children consolidated");
    assert_eq!(stats.tombstones_kept, 0);
    drop(root);

    // 4. read_dir sees alpha + beta
    let root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdonly()).await?;
    let n = names(&root.fs_read_dir().await?);
    assert!(n.contains("alpha") && n.contains("beta"), "got {:?}", n);
    drop(root);

    // 5. same-dir rename alpha -> gamma
    let mut root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?;
    root.fs_rename("alpha", "gamma").await?;
    let n = names(&root.fs_read_dir().await?);
    assert!(n.contains("gamma") && !n.contains("alpha") && n.contains("beta"), "got {:?}", n);
    drop(root);

    // 6. cross-dir rename: move beta into gamma as "beta2"
    //    (gamma is the renamed alpha; resolve its uuid first)
    let root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdonly()).await?;
    let gamma_uuid = root.fs_read_entry("gamma").await?.uuid;
    drop(root);
    HyperDir::fs_rename_across(client, layout, bucket, &ROOT_DIR_UUID, "beta", &gamma_uuid, "beta2", None).await?;

    let root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdonly()).await?;
    let n = names(&root.fs_read_dir().await?);
    assert!(!n.contains("beta"), "beta should have moved out of root, got {:?}", n);
    drop(root);

    let gamma = HyperDir::fs_open_dir(client, layout, bucket, &gamma_uuid, FileFlags::rdonly()).await?;
    let gn = names(&gamma.fs_read_dir().await?);
    assert!(gn.contains("beta2"), "beta2 should be under gamma, got {:?}", gn);
    drop(gamma);

    // 7. rmdir gamma's child beta2, then gamma; compact + gc
    HyperDir::fs_rmdir(client, layout, bucket, &gamma_uuid, "beta2", &b_uuid, None).await?;
    {
        let mut gamma = HyperDir::fs_open_dir(client, layout, bucket, &gamma_uuid, FileFlags::rdwr()).await?;
        gamma.fs_compact().await?;
        let gn = names(&gamma.fs_read_dir().await?);
        assert!(!gn.contains("beta2"), "beta2 removed, got {:?}", gn);
    }
    let stats = HyperDir::fs_gc(client, layout, bucket, &gamma_uuid).await?;
    assert!(stats.tombstones_reclaimed >= 1, "gc reclaimed the tombstone: {:?}", stats);

    Ok(())
}


// Create a regular file at FILE/<uuid> with a scatter interceptor toward
// `parent`, returning the file's UUID. Mirrors what hyperfs will do for
// file creation (hyperdir itself only allocates directory identities).
async fn create_file(
    client: &Client,
    layout: &HyperDirLayout,
    bucket: &str,
    parent: &Uuid,
    name: &str,
) -> std::io::Result<Uuid> {
    let uuid = Uuid::new_v4();
    let file_uri = layout.file_uri(bucket, &uuid);
    let parent_uri = layout.dir_uri(bucket, parent);
    let parent_staging = S3Staging::from(
        client,
        StagingConfig::new_s3_uri(&parent_uri, None),
        HyperFileRuntimeConfig::default(),
    ).await?;
    let interceptor = ScatterFirstInterceptor::new(parent_staging, name, uuid);
    let mode = FileMode::from((libc::S_IFREG | 0o644) as libc::mode_t);
    let mut f = Hyper::fs_create_with_interceptor(client, &file_uri, FileFlags::rdwr(), mode, interceptor).await?;
    let _ = f.fs_release().await?;
    Ok(uuid)
}

async fn file_nlink(client: &Client, layout: &HyperDirLayout, bucket: &str, uuid: &Uuid) -> std::io::Result<u64> {
    let st = HyperDir::fs_getattr_fast(client, &layout.file_uri(bucket, uuid)).await?;
    Ok(st.st_nlink)
}

async fn file_exists(client: &Client, layout: &HyperDirLayout, bucket: &str, uuid: &Uuid) -> bool {
    HyperDir::fs_getattr_fast(client, &layout.file_uri(bucket, uuid)).await.is_ok()
}

/// Hard links: two names share one file; nlink is authoritative in the file
/// inode; GC reclaims a file only once the last link is gone.
#[tokio::test]
#[ignore]
async fn e2e_hardlink_nlink() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = run_hardlink(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("hardlink flow");
}

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

    // create a file "f1" under root, consolidate, confirm nlink == 1
    let f = create_file(client, layout, bucket, &ROOT_DIR_UUID, "f1").await?;
    {
        let mut root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?;
        root.fs_compact().await?;
        assert!(names(&root.fs_read_dir().await?).contains("f1"));
    }
    assert_eq!(file_nlink(client, layout, bucket, &f).await?, 1, "fresh file nlink");

    // hard link "f2" -> same file; nlink == 2; both names resolve to f
    HyperDir::fs_link(client, layout, bucket, &ROOT_DIR_UUID, "f2", &f).await?;
    assert_eq!(file_nlink(client, layout, bucket, &f).await?, 2, "after link");
    {
        let root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdonly()).await?;
        assert_eq!(root.fs_read_entry("f1").await?.uuid, f);
        assert_eq!(root.fs_read_entry("f2").await?.uuid, f);
    }

    // linking onto an existing name fails with AlreadyExists and must NOT bump
    // nlink (a failed link never leaves a stale-high count).
    let dup = HyperDir::fs_link(client, layout, bucket, &ROOT_DIR_UUID, "f2", &f).await;
    assert_eq!(dup.unwrap_err().kind(), std::io::ErrorKind::AlreadyExists, "duplicate-name link -> AlreadyExists");
    assert_eq!(file_nlink(client, layout, bucket, &f).await?, 2, "nlink unchanged after failed link");

    // unlink f1 (no retention): nlink -> 1, GC must NOT reclaim (f2 still links)
    HyperDir::fs_unlink(client, layout, bucket, &ROOT_DIR_UUID, "f1", &f, false, None).await?;
    {
        let mut root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?;
        root.fs_compact().await?;
        let n = names(&root.fs_read_dir().await?);
        assert!(!n.contains("f1") && n.contains("f2"), "got {:?}", n);
    }
    assert_eq!(file_nlink(client, layout, bucket, &f).await?, 1, "after first unlink");
    HyperDir::fs_gc(client, layout, bucket, &ROOT_DIR_UUID).await?;
    assert!(file_exists(client, layout, bucket, &f).await, "file kept while a link remains");

    // unlink f2: nlink -> 0, GC reclaims the file prefix
    HyperDir::fs_unlink(client, layout, bucket, &ROOT_DIR_UUID, "f2", &f, false, None).await?;
    {
        let mut root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?;
        root.fs_compact().await?;
    }
    HyperDir::fs_gc(client, layout, bucket, &ROOT_DIR_UUID).await?;
    assert!(!file_exists(client, layout, bucket, &f).await, "file reclaimed after last unlink");
    Ok(())
}

/// Retention: GC leaves a tombstone in place until its retention expires.
#[tokio::test]
#[ignore]
async fn e2e_retention() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = run_retention(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("retention flow");
}

async fn run_retention(client: &Client, bucket: &str, layout: &HyperDirLayout) -> std::io::Result<()> {
    let _root = HyperDir::fs_create_root(client, layout, bucket, FileFlags::rdwr(), dir_mode()).await?;
    let f = create_file(client, layout, bucket, &ROOT_DIR_UUID, "keep").await?;
    {
        let mut root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?;
        root.fs_compact().await?;
    }

    // unlink with a short retention window
    HyperDir::fs_unlink(client, layout, bucket, &ROOT_DIR_UUID, "keep", &f, false, Some(Duration::from_secs(3))).await?;
    {
        let mut root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?;
        root.fs_compact().await?;
    }

    // GC before expiry: tombstone skipped, file still present
    let s1 = HyperDir::fs_gc(client, layout, bucket, &ROOT_DIR_UUID).await?;
    assert_eq!(s1.tombstones_reclaimed, 0, "nothing reclaimed before retention: {:?}", s1);
    assert!(s1.tombstones_skipped_retention >= 1, "tombstone skipped: {:?}", s1);
    assert!(file_exists(client, layout, bucket, &f).await, "file kept within retention");

    // after expiry: GC reclaims
    tokio::time::sleep(Duration::from_secs(4)).await;
    let s2 = HyperDir::fs_gc(client, layout, bucket, &ROOT_DIR_UUID).await?;
    assert!(s2.tombstones_reclaimed >= 1, "reclaimed after expiry: {:?}", s2);
    assert!(!file_exists(client, layout, bucket, &f).await, "file reclaimed after retention");
    Ok(())
}

/// Concurrent compaction: two compactors on one directory. The leader lease
/// plus hyperfile's inode OCC mean both report success or one backs off with
/// ResourceBusy; in all cases the directory ends consistent.
#[tokio::test]
#[ignore]
async fn e2e_concurrent_compact() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = run_concurrent_compact(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("concurrent compact flow");
}

async fn run_concurrent_compact(client: &Client, bucket: &str, layout: &HyperDirLayout) -> std::io::Result<()> {
    let _root = HyperDir::fs_create_root(client, layout, bucket, FileFlags::rdwr(), dir_mode()).await?;
    for n in 0..5 {
        let _ = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, &format!("d{n}"), FileFlags::rdwr(), dir_mode()).await?;
    }

    let mut a = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?;
    let mut b = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?;
    let (ra, rb) = tokio::join!(a.fs_compact(), b.fs_compact());

    // Each call either succeeds or backs off (ResourceBusy from the lease, or
    // AlreadyExists from the inode OCC under FailFast); neither corrupts.
    for r in [&ra, &rb] {
        if let Err(e) = r {
            assert!(
                matches!(e.kind(), std::io::ErrorKind::ResourceBusy | std::io::ErrorKind::AlreadyExists),
                "unexpected compact error: {e:?}"
            );
        }
    }
    assert!(ra.is_ok() || rb.is_ok(), "at least one compactor should succeed");

    // Final view is consistent: all five children present.
    let root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdonly()).await?;
    let n = names(&root.fs_read_dir().await?);
    for i in 0..5 {
        assert!(n.contains(&format!("d{i}")), "missing d{i}, got {:?}", n);
    }
    Ok(())
}


// A pair of distinct UTF-8 names with the SAME crc64 (crc64fast), constructed
// offline. They force two directory entries into the same bmap "home" slot,
// exercising open-addressing probing + backward-shift on delete.
const COLLIDE_A: &str = "A6bZ)*$e1W{";
const COLLIDE_B: &str = "B%'#]%J G|K";

/// CRC64 collision: two names hashing to the same slot must both be stored,
/// resolve to their own UUIDs, and survive deletion of the other.
#[tokio::test]
#[ignore]
async fn e2e_crc64_collision() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = run_collision(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("collision flow");
}

async fn run_collision(client: &Client, bucket: &str, layout: &HyperDirLayout) -> std::io::Result<()> {
    // sanity: the two names really do collide under the same hasher hyperdir uses
    let h = |s: &str| { let mut c = crc64fast::Digest::new(); c.write(s.as_bytes()); c.sum64() };
    assert_eq!(h(COLLIDE_A), h(COLLIDE_B), "test constants must collide");
    assert_ne!(COLLIDE_A, COLLIDE_B);

    let _root = HyperDir::fs_create_root(client, layout, bucket, FileFlags::rdwr(), dir_mode()).await?;
    let (_a, ua) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, COLLIDE_A, FileFlags::rdwr(), dir_mode()).await?;
    let (_b, ub) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, COLLIDE_B, FileFlags::rdwr(), dir_mode()).await?;
    assert_ne!(ua, ub);

    {
        let mut root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?;
        let stats = root.fs_compact().await?;
        assert_eq!(stats.entries_added, 2, "both colliding entries consolidated: {:?}", stats);
    }

    // both present and resolve to their own uuids despite the shared slot
    {
        let root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdonly()).await?;
        let n = names(&root.fs_read_dir().await?);
        assert!(n.contains(COLLIDE_A) && n.contains(COLLIDE_B), "got {:?}", n);
        assert_eq!(root.fs_read_entry(COLLIDE_A).await?.uuid, ua);
        assert_eq!(root.fs_read_entry(COLLIDE_B).await?.uuid, ub);
    }

    // delete A; B (which probed past A's slot) must remain reachable -> tests
    // the backward-shift repair of the probe chain
    HyperDir::fs_rmdir(client, layout, bucket, &ROOT_DIR_UUID, COLLIDE_A, &ua, None).await?;
    {
        let mut root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?;
        root.fs_compact().await?;
        let n = names(&root.fs_read_dir().await?);
        assert!(!n.contains(COLLIDE_A) && n.contains(COLLIDE_B), "after deleting A: {:?}", n);
        assert_eq!(root.fs_read_entry(COLLIDE_B).await?.uuid, ub, "B still resolves after A removed");
    }
    Ok(())
}

/// Cross-directory rename crash recovery: a crash right after the intent
/// commit (but before apply) leaves the move incomplete; fs_recover_renames
/// must finish it.
#[tokio::test]
#[ignore]
async fn e2e_rename_crash_recovery() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = run_rename_recovery(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("rename recovery flow");
}

async fn run_rename_recovery(client: &Client, bucket: &str, layout: &HyperDirLayout) -> std::io::Result<()> {
    let _root = HyperDir::fs_create_root(client, layout, bucket, FileFlags::rdwr(), dir_mode()).await?;
    // src dir A holds child "x"; dst dir B is empty
    let (_a, ua) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, "A", FileFlags::rdwr(), dir_mode()).await?;
    let (_b, ub) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, "B", FileFlags::rdwr(), dir_mode()).await?;
    {
        let mut root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?;
        root.fs_compact().await?;
    }
    let (_x, ux) = HyperDir::fs_create_default(client, layout, bucket, &ua, "x", FileFlags::rdwr(), dir_mode()).await?;
    {
        let mut a = HyperDir::fs_open_dir(client, layout, bucket, &ua, FileFlags::rdwr()).await?;
        a.fs_compact().await?;
        assert!(names(&a.fs_read_dir().await?).contains("x"));
    }

    // crash injection: commit the intent but do NOT apply it
    HyperDir::fs_rename_emit_intent_only(client, layout, bucket, &ua, "x", &ub, "y").await?;

    // mid-crash state: x still in A, y not yet in B (the move is committed but
    // not materialized)
    {
        let a = HyperDir::fs_open_dir(client, layout, bucket, &ua, FileFlags::rdonly()).await?;
        assert!(names(&a.fs_read_dir().await?).contains("x"), "x still in A before recovery");
        let b = HyperDir::fs_open_dir(client, layout, bucket, &ub, FileFlags::rdonly()).await?;
        assert!(!names(&b.fs_read_dir().await?).contains("y"), "y not in B before recovery");
    }

    // recovery completes the move
    let n = HyperDir::fs_recover_renames(client, layout, bucket).await?;
    assert!(n >= 1, "recovered at least one intent, got {}", n);

    {
        let mut a = HyperDir::fs_open_dir(client, layout, bucket, &ua, FileFlags::rdwr()).await?;
        a.fs_compact().await?;
        assert!(!names(&a.fs_read_dir().await?).contains("x"), "x removed from A after recovery");
        let mut b = HyperDir::fs_open_dir(client, layout, bucket, &ub, FileFlags::rdwr()).await?;
        b.fs_compact().await?;
        let bn = names(&b.fs_read_dir().await?);
        assert!(bn.contains("y"), "y present in B after recovery: {:?}", bn);
        // the moved child kept its identity
        assert_eq!(b.fs_read_entry("y").await?.uuid, ux);
    }

    // recovery is idempotent: a second pass finds nothing
    let n2 = HyperDir::fs_recover_renames(client, layout, bucket).await?;
    assert_eq!(n2, 0, "no intents left on second recovery");
    Ok(())
}


/// Crash recovery, harder case: a crash AFTER the destination-add but BEFORE
/// the source-remove leaves the child reachable under both the source and the
/// destination name (nlink still 1). fs_recover_renames must converge to the
/// single destination name -- the dangerous transient state same-source C1
/// guards against, here reached via a half-applied rename.
#[tokio::test]
#[ignore]
async fn e2e_rename_crash_partial_apply() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = run_rename_partial_apply(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("rename partial-apply recovery flow");
}

async fn run_rename_partial_apply(client: &Client, bucket: &str, layout: &HyperDirLayout) -> std::io::Result<()> {
    let _root = HyperDir::fs_create_root(client, layout, bucket, FileFlags::rdwr(), dir_mode()).await?;
    let (_a, ua) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, "A", FileFlags::rdwr(), dir_mode()).await?;
    let (_b, ub) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, "B", FileFlags::rdwr(), dir_mode()).await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; }
    let ux = create_file(client, layout, bucket, &ua, "x").await?;
    { let mut a = HyperDir::fs_open_dir(client, layout, bucket, &ua, FileFlags::rdwr()).await?; a.fs_compact().await?; }

    // crash injection: add to destination, do NOT remove from source.
    HyperDir::fs_rename_apply_partial(client, layout, bucket, &ua, "x", &ub, "y").await?;

    // mid-crash state: x still in A and y already in B (one child, two names).
    {
        let a = HyperDir::fs_open_dir(client, layout, bucket, &ua, FileFlags::rdonly()).await?;
        assert!(names(&a.fs_read_dir().await?).contains("x"), "x still in A pre-recovery");
        let b = HyperDir::fs_open_dir(client, layout, bucket, &ub, FileFlags::rdonly()).await?;
        assert!(names(&b.fs_read_dir().await?).contains("y"), "y already in B pre-recovery");
    }
    assert_eq!(file_nlink(client, layout, bucket, &ux).await?, 1, "nlink unchanged by a move");

    let n = HyperDir::fs_recover_renames(client, layout, bucket).await?;
    assert!(n >= 1, "recovered the half-applied rename, got {}", n);

    {
        let mut a = HyperDir::fs_open_dir(client, layout, bucket, &ua, FileFlags::rdwr()).await?;
        a.fs_compact().await?;
        assert!(!names(&a.fs_read_dir().await?).contains("x"), "x removed from A after recovery");
        let b = HyperDir::fs_open_dir(client, layout, bucket, &ub, FileFlags::rdonly()).await?;
        assert_eq!(b.fs_read_entry("y").await?.uuid, ux, "y in B resolves to the moved child");
    }
    assert_eq!(file_nlink(client, layout, bucket, &ux).await?, 1, "still exactly one link after recovery");
    Ok(())
}

/// Crash recovery of a replace-over-existing reclaim: the rename overwrote a
/// destination, recorded the displaced (now nameless) child in a `.reclaim`
/// intent, then crashed before reclaiming it. fs_recover_renames must reclaim
/// the orphan's storage. (And when the displaced name still resolves to it --
/// the rename never completed -- recovery must NOT reclaim.)
#[tokio::test]
#[ignore]
async fn e2e_reclaim_intent_recovery() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = run_reclaim_intent_recovery(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("reclaim intent recovery flow");
}

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

    // Case 1: the displaced name still resolves to the child -> NOT reclaimed.
    let kept = create_file(client, layout, bucket, &ROOT_DIR_UUID, "kept").await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; }
    let _ = HyperDir::fs_emit_reclaim_intent(client, layout, bucket, &ROOT_DIR_UUID, "kept", &kept, false).await?;
    HyperDir::fs_recover_renames(client, layout, bucket).await?;
    assert!(file_exists(client, layout, bucket, &kept).await, "a still-named child is not reclaimed");

    // Case 2: an orphaned (nameless, nlink 0) displaced child -> reclaimed.
    let orphan = create_file(client, layout, bucket, &ROOT_DIR_UUID, "tmp").await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; }
    HyperDir::fs_unlink(client, layout, bucket, &ROOT_DIR_UUID, "tmp", &orphan, false, None).await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; } // nlink -> 0, nameless
    let _ = HyperDir::fs_emit_reclaim_intent(client, layout, bucket, &ROOT_DIR_UUID, "victim", &orphan, false).await?;
    let n = HyperDir::fs_recover_renames(client, layout, bucket).await?;
    assert!(n >= 1, "recovered the reclaim intent, got {}", n);
    assert!(!file_exists(client, layout, bucket, &orphan).await, "the orphaned displaced child is reclaimed");

    // idempotent: a second pass finds nothing left.
    assert_eq!(HyperDir::fs_recover_renames(client, layout, bucket).await?, 0, "no intents left on second recovery");
    Ok(())
}


/// A: crash in the second phase of a delete -- the name was folded out and the
/// tombstone written, but nlink was not decremented (stale-high). fs_gc sees
/// the tombstone with nlink > 0 and refuses to reclaim; fs_gc_orphans, which is
/// reachability-based and ignores nlink, must reclaim the now-nameless file.
#[tokio::test]
#[ignore]
async fn e2e_unlink_crash_after_tombstone() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = run_unlink_crash_after_tombstone(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("unlink phase-2 crash flow");
}

async fn run_unlink_crash_after_tombstone(client: &Client, bucket: &str, layout: &HyperDirLayout) -> std::io::Result<()> {
    let _root = HyperDir::fs_create_root(client, layout, bucket, FileFlags::rdwr(), dir_mode()).await?;
    let f = create_file(client, layout, bucket, &ROOT_DIR_UUID, "x").await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; }

    // crash injection: name folded out + tombstone written, nlink NOT decremented
    HyperDir::fs_unlink_crash_after_tombstone(client, layout, bucket, &ROOT_DIR_UUID, "x", &f).await?;
    assert_eq!(file_nlink(client, layout, bucket, &f).await?, 1, "nlink left stale-high (>0)");

    // fs_gc cannot reclaim it: the tombstone's child still has nlink > 0.
    let _ = HyperDir::fs_gc(client, layout, bucket, &ROOT_DIR_UUID).await?;
    assert!(file_exists(client, layout, bucket, &f).await, "fs_gc must not reclaim a stale-high-nlink child");

    // fs_gc_orphans (reachability, ignores nlink) reclaims the nameless file.
    let n = HyperDir::fs_gc_orphans(client, layout, bucket, Duration::from_millis(0)).await?;
    assert!(n >= 1, "orphan sweep reclaimed the file, got {}", n);
    assert!(!file_exists(client, layout, bucket, &f).await, "nameless stale-nlink file reclaimed by orphan GC");
    Ok(())
}

/// B: crash in fs_link after the nlink bump but before the entry insert -- an
/// over-count. The file keeps its existing name; once that name is unlinked it
/// is nameless with nlink still 1 (should be 0), so fs_gc won't reclaim it and
/// fs_gc_orphans must.
#[tokio::test]
#[ignore]
async fn e2e_link_crash_overcount() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = run_link_crash_overcount(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("link crash over-count flow");
}

async fn run_link_crash_overcount(client: &Client, bucket: &str, layout: &HyperDirLayout) -> std::io::Result<()> {
    let _root = HyperDir::fs_create_root(client, layout, bucket, FileFlags::rdwr(), dir_mode()).await?;
    let f = create_file(client, layout, bucket, &ROOT_DIR_UUID, "x").await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; }

    // crash injection: nlink bumped, second name never inserted.
    HyperDir::fs_link_bump_only(client, layout, bucket, &f).await?;
    assert_eq!(file_nlink(client, layout, bucket, &f).await?, 2, "over-counted nlink");

    // unlink the only real name -> nlink folds 2->1, but the file is now nameless.
    HyperDir::fs_unlink(client, layout, bucket, &ROOT_DIR_UUID, "x", &f, false, None).await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; }
    assert_eq!(file_nlink(client, layout, bucket, &f).await?, 1, "stale-high nlink survives the unlink");

    // fs_gc can't reclaim (nlink > 0); fs_gc_orphans backstops the leak.
    let _ = HyperDir::fs_gc(client, layout, bucket, &ROOT_DIR_UUID).await?;
    assert!(file_exists(client, layout, bucket, &f).await, "fs_gc leaves the over-counted file");
    let n = HyperDir::fs_gc_orphans(client, layout, bucket, Duration::from_millis(0)).await?;
    assert!(n >= 1, "orphan sweep reclaimed the leak, got {}", n);
    assert!(!file_exists(client, layout, bucket, &f).await, "link-crash over-count reclaimed by orphan GC");
    Ok(())
}

/// C: a crashed mkdir (directory inode written, name never scattered) leaves a
/// nameless directory that fs_gc never sees (no tombstone). fs_gc_orphans must
/// reclaim an empty, unreferenced directory -- without it the dir leaks forever.
#[tokio::test]
#[ignore]
async fn e2e_orphan_dir_gc() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = run_orphan_dir_gc(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("orphan dir gc flow");
}

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

    // A real directory still referenced by root must NOT be collected.
    let (_kept, ukept) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, "kept", FileFlags::rdwr(), dir_mode()).await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; }

    // Crashed mkdir: create the directory inode WITHOUT scattering a name into
    // any parent (fs_create makes the DIR/<uuid> directly, no interceptor).
    let orphan = Uuid::new_v4();
    {
        let mut d = HyperDir::fs_create(client, &layout.dir_uri(bucket, &orphan), FileFlags::rdwr(), dir_mode()).await?;
        d.fs_flush().await?;
    }
    assert!(dir_exists(client, layout, bucket, &orphan).await, "orphan dir created");

    let n = HyperDir::fs_gc_orphans(client, layout, bucket, Duration::from_millis(0)).await?;
    assert!(n >= 1, "orphan sweep reclaimed the nameless dir, got {}", n);
    assert!(!dir_exists(client, layout, bucket, &orphan).await, "nameless empty dir reclaimed");
    assert!(dir_exists(client, layout, bucket, &ukept).await, "referenced dir left intact");
    Ok(())
}

async fn dir_exists(client: &Client, layout: &HyperDirLayout, bucket: &str, uuid: &Uuid) -> bool {
    HyperDir::fs_getattr_fast(client, &layout.dir_uri(bucket, uuid)).await.is_ok()
}

/// True if an object exists at `key`.
async fn object_exists(client: &Client, bucket: &str, key: &str) -> bool {
    client.head_object().bucket(bucket).key(key).send().await.is_ok()
}

/// Keys under `prefix` ending with `suffix` (test introspection).
async fn keys_with_suffix(client: &Client, bucket: &str, prefix: &str, suffix: &str) -> Vec<String> {
    let mut out = Vec::new();
    let mut stream = client.list_objects_v2().bucket(bucket).prefix(prefix).into_paginator().send();
    while let Some(page) = stream.next().await {
        let page = page.expect("list");
        if let Some(objs) = page.contents {
            for o in objs { if let Some(k) = o.key { if k.ends_with(suffix) { out.push(k); } } }
        }
    }
    out
}

/// Link-intent recovery (orphan source-(b)): a crash after `fs_link`'s nlink
/// bump and `.link` intent write but before the entry commit. `recover_link`
/// must record the phantom as an idempotent `.unref` marker (never a raw
/// decrement — two recoverers would double it), so `real = nlink - count(.unref)`
/// stays correct and `reclaim_if_unreferenced` purges the file once its real
/// names are gone. A *fresh* intent is deferred (never races a live `fs_link`).
#[tokio::test]
#[ignore]
async fn e2e_link_intent_recovery() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = run_link_intent_recovery(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("link intent recovery flow");
}

async fn run_link_intent_recovery(client: &Client, bucket: &str, layout: &HyperDirLayout) -> std::io::Result<()> {
    use hyperdir::Outcome;
    let _root = HyperDir::fs_create_root(client, layout, bucket, FileFlags::rdwr(), dir_mode()).await?;
    let f = create_file(client, layout, bucket, &ROOT_DIR_UUID, "x").await?;
    let g = create_file(client, layout, bucket, &ROOT_DIR_UUID, "g").await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; }

    // A FRESH link intent (on g) must be left untouched -- never race a live fs_link.
    let fresh = HyperDir::fs_link_crash_after_intent(client, layout, bucket, &ROOT_DIR_UUID, "gf", &g, 0).await?;
    assert_eq!(HyperDir::recover_link(client, layout, bucket, &fresh).await?, Outcome::Noop, "fresh intent deferred");
    assert!(object_exists(client, bucket, &fresh).await, "fresh intent retained");

    // Abandoned (stale) link on f: nlink bumped to 2, "y" never inserted.
    let key = HyperDir::fs_link_crash_after_intent(client, layout, bucket, &ROOT_DIR_UUID, "y", &f, 120_000).await?;
    assert_eq!(file_nlink(client, layout, bucket, &f).await?, 2, "phantom bump present");
    {
        let r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdonly()).await?;
        assert!(!names(&r.fs_read_dir().await?).contains("y"), "entry never committed");
    }

    let file_prefix = format!("{}/", layout.file_key(&f));
    // recover_link records a .unref marker and deletes the intent.
    assert_eq!(HyperDir::recover_link(client, layout, bucket, &key).await?, Outcome::Done, "stale link recovered");
    assert!(!object_exists(client, bucket, &key).await, "intent deleted");
    assert_eq!(keys_with_suffix(client, bucket, &file_prefix, ".unref").await.len(), 1, "one .unref marker");
    // real = nlink(2) - markers(1) = 1; f still has "x", so it is NOT reclaimed.
    assert_eq!(HyperDir::reclaim_if_unreferenced(client, layout, bucket, &f).await?, Outcome::Noop, "still referenced");
    assert!(file_exists(client, layout, bucket, &f).await, "file kept while named");

    // Idempotent: a second recovery is a Noop and adds no second marker.
    assert_eq!(HyperDir::recover_link(client, layout, bucket, &key).await?, Outcome::Noop, "recovery idempotent");
    assert_eq!(keys_with_suffix(client, bucket, &file_prefix, ".unref").await.len(), 1, "marker count unchanged");

    // Drop the real name: nlink folds 2->1, f nameless; real = 1 - 1 = 0 -> reclaimed.
    HyperDir::fs_unlink(client, layout, bucket, &ROOT_DIR_UUID, "x", &f, false, None).await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; }
    assert_eq!(HyperDir::reclaim_if_unreferenced(client, layout, bucket, &f).await?, Outcome::Done, "now unreferenced");
    assert!(!file_exists(client, layout, bucket, &f).await, "file reclaimed via .unref arithmetic");
    Ok(())
}

/// The per-key, event-driven recovery entries the keeper calls reach the same
/// end state as the full-scan `fs_recover_renames`: `recover_intent(key)`
/// finishes an interrupted cross-dir rename, and `process_reclaim(key)` reclaims
/// an orphaned displaced child. Both are idempotent (a second call is a Noop).
#[tokio::test]
#[ignore]
async fn e2e_per_key_recovery() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = run_per_key_recovery(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("per-key recovery flow");
}

async fn run_per_key_recovery(client: &Client, bucket: &str, layout: &HyperDirLayout) -> std::io::Result<()> {
    use hyperdir::Outcome;
    let _root = HyperDir::fs_create_root(client, layout, bucket, FileFlags::rdwr(), dir_mode()).await?;
    let (_a, ua) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, "A", FileFlags::rdwr(), dir_mode()).await?;
    let (_b, ub) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, "B", FileFlags::rdwr(), dir_mode()).await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; }
    let ux = create_file(client, layout, bucket, &ua, "x").await?;
    { let mut a = HyperDir::fs_open_dir(client, layout, bucket, &ua, FileFlags::rdwr()).await?; a.fs_compact().await?; }

    // recover_intent: an interrupted A/x -> B/y rename, recovered by its key.
    HyperDir::fs_rename_emit_intent_only(client, layout, bucket, &ua, "x", &ub, "y").await?;
    let ikey = keys_with_suffix(client, bucket, &layout.txn_prefix(), ".intent").await
        .into_iter().next().expect("an .intent key");
    assert_eq!(HyperDir::recover_intent(client, layout, bucket, &ikey).await?, Outcome::Done, "rename recovered by key");
    {
        let mut a = HyperDir::fs_open_dir(client, layout, bucket, &ua, FileFlags::rdwr()).await?; a.fs_compact().await?;
        assert!(!names(&a.fs_read_dir().await?).contains("x"), "x removed from A");
        let mut b = HyperDir::fs_open_dir(client, layout, bucket, &ub, FileFlags::rdwr()).await?; b.fs_compact().await?;
        assert_eq!(b.fs_read_entry("y").await?.uuid, ux, "y in B resolves to the moved child");
    }
    assert_eq!(HyperDir::recover_intent(client, layout, bucket, &ikey).await?, Outcome::Noop, "recover_intent idempotent");

    // process_reclaim: an orphaned displaced child, reclaimed by its intent key.
    let orphan = create_file(client, layout, bucket, &ROOT_DIR_UUID, "tmp").await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; }
    HyperDir::fs_unlink(client, layout, bucket, &ROOT_DIR_UUID, "tmp", &orphan, false, None).await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; }
    let rkey = HyperDir::fs_emit_reclaim_intent(client, layout, bucket, &ROOT_DIR_UUID, "victim", &orphan, false).await?;
    assert_eq!(HyperDir::process_reclaim(client, layout, bucket, &rkey).await?, Outcome::Done, "displaced orphan reclaimed by key");
    assert!(!file_exists(client, layout, bucket, &orphan).await, "orphan storage gone");
    assert_eq!(HyperDir::process_reclaim(client, layout, bucket, &rkey).await?, Outcome::Noop, "process_reclaim idempotent");
    Ok(())
}

/// Orphan GC must FAIL CLOSED when it can't read a directory's references: if a
/// referencing directory's read fails transiently, its files must NOT be
/// reclaimed (the round can't confirm they're unreferenced). Repro of the
/// data-loss bug: dir A references file F; we make A unreadable by corrupting
/// its inode object (A's reference to F still physically exists in its
/// segments), then run orphan GC. A buggy `fs_gc_orphans_from` silently skips A,
/// finds F "unreferenced", and DELETES a live file. This test asserts the
/// correct behavior (F survives) and so is RED until the fix lands.
#[tokio::test]
#[ignore]
async fn e2e_orphan_gc_failclosed_on_unreadable_dir() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = run_orphan_gc_failclosed(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("orphan gc fail-closed flow");
}

async fn run_orphan_gc_failclosed(client: &Client, bucket: &str, layout: &HyperDirLayout) -> std::io::Result<()> {
    let _root = HyperDir::fs_create_root(client, layout, bucket, FileFlags::rdwr(), dir_mode()).await?;
    let (_a, ua) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, "A", FileFlags::rdwr(), dir_mode()).await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; }
    let f = create_file(client, layout, bucket, &ua, "f").await?;   // F's only name lives under A
    { let mut a = HyperDir::fs_open_dir(client, layout, bucket, &ua, FileFlags::rdwr()).await?; a.fs_compact().await?; }

    let dirs = [ROOT_DIR_UUID, ua];
    // Sanity: a clean, complete scan reclaims nothing and keeps the referenced file.
    let n = HyperDir::fs_gc_orphans_from(client, layout, bucket, &[f], &dirs, Duration::from_millis(0)).await?;
    assert_eq!(n, 0, "clean scan reclaims nothing");
    assert!(file_exists(client, layout, bucket, &f).await, "referenced file present after clean scan");

    // Simulate a transient read failure of A: corrupt its inode object. A's
    // reference to F still physically exists (in A's segments), so this models
    // "couldn't read A this round", not "A no longer references F".
    let a_inode = format!("{}/inode", layout.dir_key(&ua));
    client.put_object().bucket(bucket).key(&a_inode)
        .body(aws_sdk_s3::primitives::ByteStream::from_static(b"corrupt-not-an-inode"))
        .send().await.expect("corrupt A inode");
    assert!(HyperDir::fs_open_dir(client, layout, bucket, &ua, FileFlags::rdonly()).await.is_err(),
            "precondition: A is now unreadable");

    // Orphan GC must NOT delete F — it could not confirm F is unreferenced.
    let res = HyperDir::fs_gc_orphans_from(client, layout, bucket, &[f], &dirs, Duration::from_millis(0)).await;
    assert!(file_exists(client, layout, bucket, &f).await,
            "data loss: orphan GC deleted a referenced file after its parent dir read failed (res={:?})", res);
    Ok(())
}

/// Reclaim recovery must FAIL CLOSED when it can't read the destination parent:
/// the "is the displaced child still named dst_name?" check returned `false` on
/// an open/read_dir error, so a transient read failure made recovery treat an
/// un-applied rename as "completed" and reclaim a **still-named** object,
/// leaving a dangling directory entry. Repro: D/"victim" -> F with the rename
/// NOT applied (F must survive); corrupt D's inode so its read fails (D's entry
/// for F still physically exists in its segments); `process_reclaim` must not
/// delete F. Asserts the correct behavior, so RED until the fix lands.
#[tokio::test]
#[ignore]
async fn e2e_reclaim_recovery_failclosed_on_unreadable_dst() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = run_reclaim_failclosed(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("reclaim recovery fail-closed flow");
}

async fn run_reclaim_failclosed(client: &Client, bucket: &str, layout: &HyperDirLayout) -> std::io::Result<()> {
    let _root = HyperDir::fs_create_root(client, layout, bucket, FileFlags::rdwr(), dir_mode()).await?;
    let (_d, ud) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, "D", FileFlags::rdwr(), dir_mode()).await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; }
    let f = create_file(client, layout, bucket, &ud, "victim").await?;   // D/"victim" -> F
    { let mut d = HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdwr()).await?; d.fs_compact().await?; }

    // A reclaim intent naming F as the displaced child at D/"victim" — but the
    // rename has NOT applied: "victim" still resolves to F, so F is live and must
    // NOT be reclaimed.
    let rkey = HyperDir::fs_emit_reclaim_intent(client, layout, bucket, &ud, "victim", &f, false).await?;
    {
        let d = HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdonly()).await?;
        assert!(d.fs_read_dir().await?.iter().any(|e| e.name == "victim" && e.uuid == f), "victim still names F");
    }
    assert!(file_exists(client, layout, bucket, &f).await, "F present before recovery");
    assert!(object_exists(client, bucket, &rkey).await, "reclaim intent present before recovery");

    // Simulate a transient read failure of the destination parent D: corrupt its
    // inode. D's entry "victim" -> F still physically exists (in D's segments).
    let d_inode = format!("{}/inode", layout.dir_key(&ud));
    client.put_object().bucket(bucket).key(&d_inode)
        .body(aws_sdk_s3::primitives::ByteStream::from_static(b"corrupt-not-an-inode"))
        .send().await.expect("corrupt D inode");
    assert!(HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdonly()).await.is_err(),
            "precondition: D is now unreadable");

    // Recovery must NOT reclaim F — it could not confirm "victim" no longer names F.
    let res = HyperDir::process_reclaim(client, layout, bucket, &rkey).await;
    assert!(file_exists(client, layout, bucket, &f).await,
            "dangling: reclaim recovery deleted a still-named object after dst_parent read failed (res={:?})", res);
    Ok(())
}

/// Two-phase delete must not let a late Delete tombstone shadow a same-name file
/// recreated in the phase-1→phase-2 window. unlink("x"->A) emits a PreDelete;
/// compaction phase 1 folds it out (x now absent) but the real Delete tombstone
/// is written later (phase 2) with a fresh, *late* S3 timestamp. If a client
/// recreates "x"->B in that window, the late Delete(A) out-orders Create(B) in
/// the last-modified-ordered scatter view and hides/strips the live file B.
/// Asserts the correct outcome (x resolves to B, B survives), so RED until fixed.
#[tokio::test]
#[ignore]
async fn e2e_predelete_recreate_race_keeps_new_file() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = run_predelete_recreate_race(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("predelete/recreate race flow");
}

async fn run_predelete_recreate_race(client: &Client, bucket: &str, layout: &HyperDirLayout) -> std::io::Result<()> {
    let _root = HyperDir::fs_create_root(client, layout, bucket, FileFlags::rdwr(), dir_mode()).await?;
    let (_d, ud) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, "D", FileFlags::rdwr(), dir_mode()).await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; }
    let a = create_file(client, layout, bucket, &ud, "x").await?;     // "x" -> A
    { let mut d = HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdwr()).await?; d.fs_compact().await?; }

    // unlink "x" (phase-1 PreDelete marker), then run compaction PHASE 1 ONLY:
    // "x" is now absent (folded out + PreDelete stripped) but no Delete tombstone
    // has been written yet -- the phase-1->phase-2 window is open.
    HyperDir::fs_unlink(client, layout, bucket, &ud, "x", &a, false, None).await?;
    { let mut d = HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdwr()).await?; d.fs_compact_phase1_only().await?; }
    {
        let d = HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdonly()).await?;
        assert!(!names(&d.fs_read_dir().await?).contains("x"), "x absent in the window");
    }

    // A concurrent client recreates "x" -> B in the window (x looked absent).
    let b = create_file(client, layout, bucket, &ud, "x").await?;     // Create("x", B), later than the unlink
    assert_ne!(a, b);

    // The delayed phase 2 now writes the real Delete tombstone for A -- with a
    // FRESH, late timestamp (after B's create).
    HyperDir::fs_unlink_crash_after_tombstone(client, layout, bucket, &ud, "x", &a).await?;

    // The recreated file B must stay resolvable: a late tombstone for the OLD
    // identity A must not shadow the NEW identity B.
    {
        let d = HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdonly()).await?;
        let entries = d.fs_read_dir().await?;
        let x = entries.iter().find(|e| e.name == "x");
        assert!(x.map(|e| e.uuid) == Some(b),
            "recreated file B shadowed by a late tombstone for A (read_dir x = {:?})", x.map(|e| e.uuid));
    }

    // ...and survive compaction (B's Create must not be stripped as superseded).
    { let mut d = HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdwr()).await?; d.fs_compact().await?; }
    assert!(file_exists(client, layout, bucket, &b).await, "recreated file B lost after compaction");
    {
        let d = HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdonly()).await?;
        assert!(d.fs_read_dir().await?.iter().any(|e| e.name == "x" && e.uuid == b), "x must resolve to B after compaction");
    }
    Ok(())
}

/// Concurrent same-name create: two clients create "x" with distinct uuids
/// before either sees the other (no per-name storage exclusion; stale-negative
/// dentry across mounts). Last-writer-wins for the NAME is acceptable, but the
/// LOSING identity's file object must not be left as a nameless orphan waiting
/// for the orphan sweep -- it must be deterministically reclaimed at the
/// lease-serialized compaction point. Asserts the loser is reclaimed, so RED
/// until fixed.
#[tokio::test]
#[ignore]
async fn e2e_concurrent_create_same_name_no_orphan() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = run_concurrent_create_no_orphan(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("concurrent same-name create flow");
}

async fn run_concurrent_create_no_orphan(client: &Client, bucket: &str, layout: &HyperDirLayout) -> std::io::Result<()> {
    let _root = HyperDir::fs_create_root(client, layout, bucket, FileFlags::rdwr(), dir_mode()).await?;
    let (_d, ud) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, "D", FileFlags::rdwr(), dir_mode()).await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; }

    // Two racing same-name creates with distinct uuids (no per-name exclusion).
    let a = create_file(client, layout, bucket, &ud, "x").await?;
    let b = create_file(client, layout, bucket, &ud, "x").await?;
    assert_ne!(a, b);
    assert!(file_exists(client, layout, bucket, &a).await && file_exists(client, layout, bucket, &b).await,
        "both file objects exist right after the racing creates");

    // Compaction is the single lease-serialized point that resolves the name.
    { let mut d = HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdwr()).await?; d.fs_compact().await?; }

    // LWW: exactly one entry named "x", resolving to one of the two identities.
    let (winner, loser) = {
        let d = HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdonly()).await?;
        let entries = d.fs_read_dir().await?;
        let xs: Vec<_> = entries.iter().filter(|e| e.name == "x").collect();
        assert_eq!(xs.len(), 1, "exactly one entry named x after compaction");
        let w = xs[0].uuid;
        assert!(w == a || w == b, "x resolves to one of the two created identities");
        (w, if w == a { b } else { a })
    };

    // The winner's file object stays; the loser must be reclaimed, NOT orphaned.
    assert!(file_exists(client, layout, bucket, &winner).await, "winner file object must survive");
    assert!(!file_exists(client, layout, bucket, &loser).await,
        "losing same-name create left a nameless orphan (file object {} still present)", loser);
    Ok(())
}

/// Same-name create over an ALREADY-COMMITTED name (sequential, e.g. a stale
/// negative dentry on another mount): "x"->A is folded into the bmap, then a
/// create "x"->B arrives. LWW gives the name to B, but the displaced A must be
/// reclaimed (like a rename-over-existing), not left as a nameless orphan.
#[tokio::test]
#[ignore]
async fn e2e_create_over_committed_name_no_orphan() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = run_create_over_committed_no_orphan(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("create-over-committed flow");
}

async fn run_create_over_committed_no_orphan(client: &Client, bucket: &str, layout: &HyperDirLayout) -> std::io::Result<()> {
    let _root = HyperDir::fs_create_root(client, layout, bucket, FileFlags::rdwr(), dir_mode()).await?;
    let (_d, ud) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, "D", FileFlags::rdwr(), dir_mode()).await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; }

    // "x" -> A, committed into the bmap.
    let a = create_file(client, layout, bucket, &ud, "x").await?;
    { let mut d = HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdwr()).await?; d.fs_compact().await?; }
    assert!(names(&HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdonly()).await?.fs_read_dir().await?).contains("x"));

    // A create that didn't see "x" makes a second "x" -> B.
    let b = create_file(client, layout, bucket, &ud, "x").await?;
    assert_ne!(a, b);
    { let mut d = HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdwr()).await?; d.fs_compact().await?; }

    // LWW: x resolves to B; the displaced A is reclaimed (no orphan).
    {
        let d = HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdonly()).await?;
        let entries = d.fs_read_dir().await?;
        assert!(entries.iter().any(|e| e.name == "x" && e.uuid == b), "x must resolve to B");
    }
    assert!(file_exists(client, layout, bucket, &b).await, "winner B must survive");
    assert!(!file_exists(client, layout, bucket, &a).await,
        "displaced A left a nameless orphan (file object {} still present)", a);
    Ok(())
}

/// Cross-directory rename whose destination name is held by a *different*
/// identity the rename's pre-check didn't catch (a concurrent client committed
/// "y"->C). Today the destination-add is a blind bmap upsert (insert_entry),
/// so it overwrites C in the bmap and leaves C a nameless orphan. The moved
/// child must take the name AND the displaced C must be reclaimed (no orphan),
/// exactly like any other same-name displacement. RED until rename's
/// destination-add participates in the identity-aware scatter plane (option B).
#[tokio::test]
#[ignore]
async fn e2e_rename_over_uncaught_target_no_orphan() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = run_rename_over_uncaught_target(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("rename-over-uncaught-target flow");
}

async fn run_rename_over_uncaught_target(client: &Client, bucket: &str, layout: &HyperDirLayout) -> std::io::Result<()> {
    let _root = HyperDir::fs_create_root(client, layout, bucket, FileFlags::rdwr(), dir_mode()).await?;
    let (_s, us) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, "S", FileFlags::rdwr(), dir_mode()).await?;
    let (_d, ud) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, "D", FileFlags::rdwr(), dir_mode()).await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; }

    // Source S/x -> child, committed.
    let child = create_file(client, layout, bucket, &us, "x").await?;
    { let mut s = HyperDir::fs_open_dir(client, layout, bucket, &us, FileFlags::rdwr()).await?; s.fs_compact().await?; }

    // A concurrent client already created and committed D/y -> C.
    let c = create_file(client, layout, bucket, &ud, "y").await?;
    { let mut d = HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdwr()).await?; d.fs_compact().await?; }
    assert!(file_exists(client, layout, bucket, &c).await);
    assert_ne!(child, c);

    // Cross-dir rename S/x -> D/y. (fs_rename_across is the hyperdir rename; it
    // does no destination pre-check, exactly the case where do_rename's check
    // raced and missed C.)
    HyperDir::fs_rename_across(client, layout, bucket, &us, "x", &ud, "y", None).await?;
    { let mut d = HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdwr()).await?; d.fs_compact().await?; }

    // y takes the moved child; the displaced C must be reclaimed, not orphaned.
    {
        let d = HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdonly()).await?;
        let entries = d.fs_read_dir().await?;
        assert!(entries.iter().any(|e| e.name == "y" && e.uuid == child), "y must resolve to the moved child");
    }
    assert!(file_exists(client, layout, bucket, &child).await, "moved child must survive");
    assert!(!file_exists(client, layout, bucket, &c).await,
        "rename displaced target C left a nameless orphan (file object {} still present)", c);
    {
        let s = HyperDir::fs_open_dir(client, layout, bucket, &us, FileFlags::rdonly()).await?;
        assert!(!names(&s.fs_read_dir().await?).contains("x"), "source name removed");
    }
    Ok(())
}

/// Cross-directory rename that replaces an existing COMMITTED destination (the
/// normal POSIX replace), driven the way do_rename does it: record a reclaim
/// intent for the displaced target, then fs_rename_across with that key. The
/// displaced target must be reclaimed exactly once (no orphan, no double-marker
/// over-count) and y must resolve to the moved child.
#[tokio::test]
#[ignore]
async fn e2e_rename_replace_committed_reclaims_once() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = run_rename_replace_committed(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("rename-replace-committed flow");
}

async fn run_rename_replace_committed(client: &Client, bucket: &str, layout: &HyperDirLayout) -> std::io::Result<()> {
    let _root = HyperDir::fs_create_root(client, layout, bucket, FileFlags::rdwr(), dir_mode()).await?;
    let (_s, us) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, "S", FileFlags::rdwr(), dir_mode()).await?;
    let (_d, ud) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, "D", FileFlags::rdwr(), dir_mode()).await?;
    { let mut r = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?; r.fs_compact().await?; }

    let child = create_file(client, layout, bucket, &us, "x").await?;
    { let mut s = HyperDir::fs_open_dir(client, layout, bucket, &us, FileFlags::rdwr()).await?; s.fs_compact().await?; }
    // Pre-existing committed destination D/y -> dold.
    let dold = create_file(client, layout, bucket, &ud, "y").await?;
    { let mut d = HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdwr()).await?; d.fs_compact().await?; }
    assert_ne!(child, dold);

    // do_rename's shape: record the reclaim intent for the caught target, then
    // rename under that key.
    let rkey = HyperDir::fs_emit_reclaim_intent(client, layout, bucket, &ud, "y", &dold, false).await?;
    HyperDir::fs_rename_across(client, layout, bucket, &us, "x", &ud, "y", Some(&rkey)).await?;
    let _ = HyperDir::fs_delete_intent(client, bucket, &rkey).await;
    { let mut d = HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdwr()).await?; d.fs_compact().await?; }

    {
        let d = HyperDir::fs_open_dir(client, layout, bucket, &ud, FileFlags::rdonly()).await?;
        assert!(d.fs_read_dir().await?.iter().any(|e| e.name == "y" && e.uuid == child), "y resolves to the moved child");
    }
    assert!(file_exists(client, layout, bucket, &child).await, "moved child survives");
    assert!(!file_exists(client, layout, bucket, &dold).await, "replaced committed target must be reclaimed");
    Ok(())
}

/// Build an inlined regular-file inode (`FLAG_INLINE`, bytes embedded, `i_size`
/// = len), the way `hyperfs::do_create`/an open inline flush does.
fn inline_inode(data: &[u8]) -> InodeRaw {
    let mut raw = Inode::default_file().to_raw(unsafe { std::mem::zeroed() });
    raw.i_mode = 0o100_644; // S_IFREG | 0644
    raw.set_inline(data);
    raw
}

/// An inlined small file (bytes live in its parent dirent, no `FILE/<uuid>`
/// object) must keep its bytes across (a) a compaction that folds the inline
/// Create/Update scatter into the bmap, and (b) a cross-directory rename that
/// carries the inode through a rename intent. Verified deterministically with
/// an explicit `fs_compact` (no reliance on the background sweep timing).
#[tokio::test]
#[ignore]
async fn e2e_inline_survives_compaction_and_rename() {
    let client = make_client().await;
    let bucket = bucket();
    let base = format!("hyperdir-e2e/{}", Uuid::new_v4());
    let layout = HyperDirLayout::with_base(&base);
    let result = inline_durability_flow(&client, &bucket, &layout).await;
    purge_prefix(&client, &bucket, &format!("{}/", base)).await;
    result.expect("inline durability flow");
}

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

    // Create an inlined file: no FILE/<uuid> object, bytes carried in the dirent.
    let v1 = b"inline payload v1".to_vec();
    let uuid = HyperDir::fs_create_inline(client, layout, bucket, &ROOT_DIR_UUID, "inl", &inline_inode(&v1)).await?;

    // Pre-compaction the entry resolves from the (uncompacted) scatter.
    let root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdonly()).await?;
    let ent = root.fs_read_dir().await?.into_iter().find(|e| e.name == "inl")
        .expect("inl present pre-compaction");
    assert_eq!(ent.inline.as_deref(), Some(&v1[..]), "inline bytes visible pre-compaction");
    drop(root);

    // Simulate an open-file flush updating the inline bytes (still <= cap).
    let v2 = b"inline payload v2 (still tiny <=80B)".to_vec();
    assert!(v2.len() <= InodeRaw::inline_cap());
    HyperDir::fs_update_inline(client, layout, bucket, &ROOT_DIR_UUID, "inl", &uuid, &inline_inode(&v2)).await?;

    // Compact root: the inline Create + Update fold into the bmap entry.
    let mut root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdwr()).await?;
    root.fs_compact().await?;
    drop(root);

    // Post-compaction the bytes (and size) are read from the bmap entry, intact.
    let root = HyperDir::fs_open_root(client, layout, bucket, FileFlags::rdonly()).await?;
    let ent = root.fs_read_dir().await?.into_iter().find(|e| e.name == "inl")
        .expect("inl present post-compaction");
    assert_eq!(ent.inline.as_deref(), Some(&v2[..]), "inline bytes survive compaction");
    assert_eq!(ent.inode.to_stat(0, 0).st_size as usize, v2.len(), "inline size survives compaction");
    drop(root);

    // No FILE/<uuid> object was ever created for an inline file.
    let r = HyperDir::fs_getattr_fast(client, &layout.file_uri(bucket, &uuid)).await;
    assert!(matches!(&r, Err(e) if e.kind() == std::io::ErrorKind::NotFound),
            "inline file must have no FILE object, got {r:?}");

    // Cross-directory rename carries the inline bytes (no FILE object involved).
    let (_d, d_uuid) = HyperDir::fs_create_default(client, layout, bucket, &ROOT_DIR_UUID, "sub", FileFlags::rdwr(), dir_mode()).await?;
    HyperDir::fs_rename_across(client, layout, bucket, &ROOT_DIR_UUID, "inl", &d_uuid, "moved", None).await?;
    let sub = HyperDir::fs_open_dir(client, layout, bucket, &d_uuid, FileFlags::rdonly()).await?;
    let ent = sub.fs_read_dir().await?.into_iter().find(|e| e.name == "moved")
        .expect("moved present after cross-dir rename");
    assert_eq!(ent.inline.as_deref(), Some(&v2[..]), "inline bytes survive cross-dir rename");
    drop(sub);
    let r = HyperDir::fs_getattr_fast(client, &layout.file_uri(bucket, &uuid)).await;
    assert!(matches!(&r, Err(e) if e.kind() == std::io::ErrorKind::NotFound),
            "still inline (no FILE object) after rename, got {r:?}");

    Ok(())
}