hyperdir 0.5.0

use std::io::{Result, Error, ErrorKind};
use std::time::Duration;
use log::{debug, warn};
use uuid::Uuid;
use ulid::Ulid;
use base64::{engine::general_purpose::STANDARD as B64, Engine as _};
use base64::engine::general_purpose::URL_SAFE_NO_PAD as B64URL;
use aws_sdk_s3::Client;
use aws_sdk_s3::primitives::SdkBody;
use crate::s3::s3_error_kind;
use hyperfile::file::flags::{FileFlags, HyperFileFlags};
use hyperfile::file::mode::{FileMode, HyperFileMode};
use hyperfile::file::hyper::Hyper as HyperFileHandle;
use hyperfile::config::HyperFileConfigBuilder;
use hyperfile::config::HyperFileRuntimeConfig;
use hyperfile::config::FlushConflictPolicy;
use hyperfile::staging::{Staging, config::StagingConfig, s3::S3Staging, StagingIntercept};
use hyperfile::meta_loader::s3_batch::S3BlockLoader;
use hyperfile::file::HyperTrait;
use hyperfile::ondisk::InodeRaw;
use hyperfile::inode::FlushInodeFlag;
use crate::hyper::HyperDir;
use crate::file::{DirFileEntry, CompactStats, HyperDirFile};
use crate::interceptor::ScatterFirstInterceptor;
use crate::layout::HyperDirLayout;
use crate::{
    DirStaging, DirScatterInodeOp,
    build_tombstone_body, build_predelete_body, parse_tombstone_body, unix_now_ms,
};

/// TTL (seconds) for open-lease liveness: an open-lease `_OPEN/<uuid>/<mount-id>`
/// whose owning mount's heartbeat (`heartbeat_key`) is older than this (or
/// absent) is treated as dead. Must exceed the maintenance/heartbeat cadence.
/// Process-wide and configurable: set once at startup (default 600), kept in
/// lock-step with hyperfs's heartbeat TTL so GC and the FUSE/reaper paths agree.
static OPEN_LEASE_TTL_SECS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(600);

/// Set the process-wide open-lease liveness TTL (seconds). Call once at startup.
pub fn set_open_lease_ttl_secs(secs: u64) {
    OPEN_LEASE_TTL_SECS.store(secs, std::sync::atomic::Ordering::Relaxed);
}

/// The configured open-lease liveness TTL (seconds); default 600.
pub fn open_lease_ttl_secs() -> u64 {
    OPEN_LEASE_TTL_SECS.load(std::sync::atomic::Ordering::Relaxed)
}

impl<'a> HyperDir<'a>
{
    pub async fn fs_create(client: &Client, uri: &str, flags: FileFlags, mode: FileMode) -> Result<Self>
    {
        debug!("fs_create - uri: {}, flags: {}", uri, flags);
        let staging_config = StagingConfig::new_s3_uri(uri, None);
        let file_config = HyperFileConfigBuilder::new()
                            .with_staging_config(&staging_config)
                            .build();
        let f = HyperFileFlags::from_flags(flags);
        let m = HyperFileMode::from_mode(mode);
        return Self::create(client.clone(), file_config, f, m).await;
    }

    pub async fn fs_create_with_interceptor(client: &Client, uri: &str, flags: FileFlags, mode: FileMode, interceptor: impl StagingIntercept<S3Staging> + 'static) -> Result<Self>
    {
        debug!("fs_create_with_interceptor - uri: {}, flags: {}", uri, flags);
        let staging_config = StagingConfig::new_s3_uri(uri, None);
        let file_config = HyperFileConfigBuilder::new()
                            .with_staging_config(&staging_config)
                            .build();
        let f = HyperFileFlags::from_flags(flags);
        let m = HyperFileMode::from_mode(mode);
        return Self::create_with_interceptor(client.clone(), file_config, f, m, interceptor).await;
    }

    /// Create a child directory under `parent_dir_uuid`, returning the new
    /// directory handle and the UUID allocated for it.
    ///
    /// hyperdir owns directory-identity allocation: a fresh v4 UUID is
    /// generated here and the directory's hyperfile is created at
    /// `layout.dir_uri(bucket, new_uuid)`. A [`ScatterFirstInterceptor`] is
    /// installed toward the parent so the first inode flush commits a scatter
    /// into the parent's `!/` namespace. The parent must already exist.
    ///
    /// The root directory has no parent; create it with [`HyperDir::create`]
    /// directly (no interceptor) at `layout.root_dir_uri(bucket)`.
    pub async fn fs_create_default(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        parent_dir_uuid: &Uuid,
        name: &str,
        flags: FileFlags,
        mode: FileMode,
    ) -> Result<(Self, Uuid)>
    {
        let uuid = Uuid::new_v4();
        let uri = layout.dir_uri(bucket, &uuid);
        let parent_dir_uri = layout.dir_uri(bucket, parent_dir_uuid);
        debug!("fs_create_default - uri: {}, parent: {}, name: {}", uri, parent_dir_uri, name);
        let parent_staging = S3Staging::from(
            client,
            StagingConfig::new_s3_uri(&parent_dir_uri, None),
            HyperFileRuntimeConfig::default(),
        ).await?;
        let interceptor = ScatterFirstInterceptor::new(parent_staging, name, uuid);
        let dir = Self::fs_create_with_interceptor(client, &uri, flags, mode, interceptor).await?;
        Ok((dir, uuid))
    }

    /// Create a regular file `name` under `parent_dir_uuid`, returning an open
    /// hyperfile handle for byte I/O and the UUID allocated for it.
    ///
    /// Mirrors [`fs_create_default`] but creates a FILE (`layout.file_uri`)
    /// and hands back hyperfile's `Hyper`: the namespace (name -> uuid) is
    /// hyperdir's via the scatter interceptor, while byte content is
    /// hyperfile's. An initial flush is issued here so the file's inode object
    /// and the parent's Create scatter both exist before returning — the file
    /// is immediately stat-able and visible to `read_dir`.
    pub async fn fs_create_file(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        parent_dir_uuid: &Uuid,
        name: &str,
        flags: FileFlags,
        mode: FileMode,
    ) -> Result<(HyperFileHandle<'static>, Uuid)>
    {
        let uuid = Uuid::new_v4();
        let uri = layout.file_uri(bucket, &uuid);
        let parent_dir_uri = layout.dir_uri(bucket, parent_dir_uuid);
        debug!("fs_create_file - uri: {}, parent: {}, name: {}", uri, parent_dir_uri, name);
        let parent_staging = S3Staging::from(
            client,
            StagingConfig::new_s3_uri(&parent_dir_uri, None),
            HyperFileRuntimeConfig::default(),
        ).await?;
        let interceptor = ScatterFirstInterceptor::new(parent_staging, name, uuid);
        let mut file = HyperFileHandle::fs_create_with_interceptor(client, &uri, flags, mode, interceptor).await?;
        file.fs_flush().await?;
        Ok((file, uuid))
    }

    /// Create a fully-inlined small file/symlink: emit a `Create` scatter into
    /// `parent` carrying `inode_raw` (which must already have its inline data set
    /// via `InodeRaw::set_inline`), and allocate its identity uuid -- **without**
    /// creating any `FILE/<uuid>` object. The bytes live entirely in the
    /// directory namespace (the dirent/scatter), so resolution returns them with
    /// no extra GET. Returns the allocated uuid.
    pub async fn fs_create_inline(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        parent_dir_uuid: &Uuid,
        name: &str,
        inode_raw: &InodeRaw,
    ) -> Result<Uuid>
    {
        let uuid = Uuid::new_v4();
        let parent_dir_uri = layout.dir_uri(bucket, parent_dir_uuid);
        let parent_staging = S3Staging::from(
            client,
            StagingConfig::new_s3_uri(&parent_dir_uri, None),
            HyperFileRuntimeConfig::default(),
        ).await?;
        parent_staging.emit_scatter_event(name, &uuid, inode_raw.as_u8_slice(), DirScatterInodeOp::Create).await?;
        Ok(uuid)
    }

    /// Re-emit an inlined file's bytes (an `Update` scatter for an existing
    /// inline identity `uuid`); used when an open inline file is flushed.
    pub async fn fs_update_inline(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        parent_dir_uuid: &Uuid,
        name: &str,
        uuid: &Uuid,
        inode_raw: &InodeRaw,
    ) -> Result<()>
    {
        let parent_dir_uri = layout.dir_uri(bucket, parent_dir_uuid);
        let parent_staging = S3Staging::from(
            client,
            StagingConfig::new_s3_uri(&parent_dir_uri, None),
            HyperFileRuntimeConfig::default(),
        ).await?;
        parent_staging.emit_scatter_event(name, uuid, inode_raw.as_u8_slice(), DirScatterInodeOp::Update).await
    }

    /// Create a real `FILE/<uuid>` object at a CALLER-supplied uuid (rather than
    /// a fresh one) under `parent/name` -- used to **spill** an inlined file to
    /// segmented storage once it grows past the inline cap, keeping its identity
    /// stable. Mirrors [`fs_create_file`] but with a given uuid.
    #[allow(clippy::too_many_arguments)]
    pub async fn fs_create_file_at(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        parent_dir_uuid: &Uuid,
        name: &str,
        uuid: &Uuid,
        flags: FileFlags,
        mode: FileMode,
    ) -> Result<HyperFileHandle<'static>>
    {
        let uri = layout.file_uri(bucket, uuid);
        let parent_dir_uri = layout.dir_uri(bucket, parent_dir_uuid);
        let parent_staging = S3Staging::from(
            client,
            StagingConfig::new_s3_uri(&parent_dir_uri, None),
            HyperFileRuntimeConfig::default(),
        ).await?;
        let interceptor = ScatterFirstInterceptor::new(parent_staging, name, *uuid);
        let mut file = HyperFileHandle::fs_create_with_interceptor(client, &uri, flags, mode, interceptor).await?;
        file.fs_flush().await?;
        Ok(file)
    }

    // ==== second-level inline ("blob") — see docs/second-level-inline.md ====

    /// Default max size for a blob-inlined file: files up to this many bytes are
    /// stored as a single ULID-versioned object `FILE/<uuid>/<ulid>.blob` rather
    /// than a full hyperfile. Overridable per-directory via the
    /// [`INLINE_MAX_XATTR`](Self::INLINE_MAX_XATTR) xattr.
    pub const DEFAULT_INLINE_MAX: usize = 128 * 1024;

    /// Per-directory xattr name carrying a decimal blob-threshold override.
    pub const INLINE_MAX_XATTR: &'static str = "hyperdir.inline_max";

    /// Effective blob threshold for a directory: its `hyperdir.inline_max` xattr
    /// if set and parseable, else [`DEFAULT_INLINE_MAX`](Self::DEFAULT_INLINE_MAX).
    /// Cheap enough to call at directory open and cache.
    pub async fn dir_inline_max(client: &Client, layout: &HyperDirLayout, bucket: &str, dir_uuid: &Uuid) -> usize {
        match Self::fs_getxattr(client, layout, bucket, dir_uuid, true, Self::INLINE_MAX_XATTR).await {
            Ok(Some(v)) => std::str::from_utf8(&v).ok()
                .and_then(|s| s.trim().parse::<usize>().ok())
                .unwrap_or(Self::DEFAULT_INLINE_MAX),
            _ => Self::DEFAULT_INLINE_MAX,
        }
    }

    /// Set a directory's blob threshold override (0 disables blob inlining for
    /// new files under it; callers still decide the tier — see the hyperfs
    /// requirement doc).
    pub async fn set_dir_inline_max(client: &Client, layout: &HyperDirLayout, bucket: &str, dir_uuid: &Uuid, max: usize) -> Result<()> {
        Self::fs_setxattr(client, layout, bucket, dir_uuid, true, Self::INLINE_MAX_XATTR, max.to_string().as_bytes()).await
    }

    /// PUT one write-once blob version object `FILE/<uuid>/<ulid>.blob`.
    async fn put_blob(client: &Client, layout: &HyperDirLayout, bucket: &str, file_uuid: &Uuid, blob_ulid: Ulid, bytes: &[u8]) -> Result<()> {
        let key = layout.blob_key(file_uuid, &blob_ulid.to_string());
        client.put_object()
            .bucket(bucket)
            .key(&key)
            .body(SdkBody::from(bytes.to_vec()).into())
            .if_none_match('*')
            .send()
            .await
            .map_err(|e| Error::new(s3_error_kind(&e), format!("PutObject blob s3://{}/{}: {}", bucket, key, e)))?;
        Ok(())
    }

    /// Create a **blob** file (80B < size ≤ threshold): allocate a fresh
    /// identity uuid + version ulid, write the data-first `<ulid>.blob` object,
    /// then emit the `Create` commit scatter into `parent` carrying
    /// `inode_raw` (which must have `i_size == bytes.len()` and **not**
    /// `FLAG_INLINE`) followed by the 16-byte version ULID. No `FILE/<uuid>`
    /// hyperfile is created. Returns `(uuid, blob_ulid)`.
    #[allow(clippy::too_many_arguments)]
    pub async fn fs_create_blob(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        parent_dir_uuid: &Uuid,
        name: &str,
        inode_raw: &InodeRaw,
        bytes: &[u8],
    ) -> Result<(Uuid, Ulid)>
    {
        let uuid = Uuid::new_v4();
        let blob_ulid = Ulid::new();
        // 1. data first (write-once), 2. namespace commit scatter.
        Self::put_blob(client, layout, bucket, &uuid, blob_ulid, bytes).await?;
        let parent_dir_uri = layout.dir_uri(bucket, parent_dir_uuid);
        let parent_staging = S3Staging::from(
            client, StagingConfig::new_s3_uri(&parent_dir_uri, None), HyperFileRuntimeConfig::default(),
        ).await?;
        let body = blob_scatter_body(inode_raw, blob_ulid);
        parent_staging.emit_scatter_event(name, &uuid, &body, DirScatterInodeOp::Create).await?;
        Ok((uuid, blob_ulid))
    }

    /// Write a new version of an existing blob file: write a fresh
    /// `<ulid>.blob` and emit an `Update` scatter refreshing the dirent inode +
    /// version pointer. Older versions remain (retention / undelete) until the
    /// version GC reclaims them. Returns the new version ulid.
    #[allow(clippy::too_many_arguments)]
    pub async fn fs_write_blob(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        parent_dir_uuid: &Uuid,
        name: &str,
        uuid: &Uuid,
        inode_raw: &InodeRaw,
        bytes: &[u8],
    ) -> Result<Ulid>
    {
        let blob_ulid = Ulid::new();
        Self::put_blob(client, layout, bucket, uuid, blob_ulid, bytes).await?;
        let parent_dir_uri = layout.dir_uri(bucket, parent_dir_uuid);
        let parent_staging = S3Staging::from(
            client, StagingConfig::new_s3_uri(&parent_dir_uri, None), HyperFileRuntimeConfig::default(),
        ).await?;
        let body = blob_scatter_body(inode_raw, blob_ulid);
        parent_staging.emit_scatter_event(name, uuid, &body, DirScatterInodeOp::Update).await?;
        Ok(blob_ulid)
    }

    /// Update a blob file's **dirent metadata only**, keeping its current
    /// version pointer: emits an `Update` scatter carrying `inode_raw` plus the
    /// EXISTING `blob_ulid`, without PUTting any bytes and without creating a
    /// new version. For an attribute-only change (mode/uid/gid/times) on a blob
    /// file, so a `chmod` costs one scatter instead of rewriting the whole blob.
    ///
    /// `inode_raw` must have the same shape [`fs_write_blob`] requires (i_size =
    /// the blob's byte length, no `FLAG_INLINE`); only the metadata fields
    /// should differ from what the current version was written with.
    #[allow(clippy::too_many_arguments)]
    pub async fn fs_update_blob_meta(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        parent_dir_uuid: &Uuid,
        name: &str,
        uuid: &Uuid,
        inode_raw: &InodeRaw,
        blob_ulid: Ulid,
    ) -> Result<()>
    {
        debug!("fs_update_blob_meta - {}/{} uuid {} ulid {}", parent_dir_uuid, name, uuid, blob_ulid);
        let parent_dir_uri = layout.dir_uri(bucket, parent_dir_uuid);
        let parent_staging = S3Staging::from(
            client, StagingConfig::new_s3_uri(&parent_dir_uri, None), HyperFileRuntimeConfig::default(),
        ).await?;
        let body = blob_scatter_body(inode_raw, blob_ulid);
        parent_staging.emit_scatter_event(name, uuid, &body, DirScatterInodeOp::Update).await
    }

    /// Read a blob file's current version bytes (one GET, no LIST). `blob_ulid`
    /// comes from the resolved dirent (`DirFileEntry::blob_ulid`).
    pub async fn fs_read_blob(client: &Client, layout: &HyperDirLayout, bucket: &str, file_uuid: &Uuid, blob_ulid: Ulid) -> Result<Vec<u8>> {
        let key = layout.blob_key(file_uuid, &blob_ulid.to_string());
        match client.get_object().bucket(bucket).key(&key).send().await {
            Ok(res) => {
                let bytes = res.body.collect().await
                    .map_err(|e| Error::other(format!("collect blob s3://{}/{}: {}", bucket, key, e)))?
                    .into_bytes();
                Ok(bytes.to_vec())
            },
            Err(e) => Err(Error::new(s3_error_kind(&e), format!("GetObject blob s3://{}/{}: {}", bucket, key, e))),
        }
    }

    /// Reconcile blob versions under `FILE/<uuid>/`: delete every `*.blob` whose
    /// ULID != `keep` (all of them if `keep` is `None`). Idempotent. Used to drop
    /// a now-segmented file's stray blobs (`keep = None`, e.g. after a spill) and,
    /// by the GC sweep, to prune superseded versions.
    pub async fn fs_reconcile_blobs(client: &Client, layout: &HyperDirLayout, bucket: &str, file_uuid: &Uuid, keep: Option<Ulid>) -> Result<usize> {
        let prefix = layout.blob_prefix(file_uuid);
        let keep_s = keep.map(|u| u.to_string());
        let mut deleted = 0;
        let mut stream = client.list_objects_v2().bucket(bucket).prefix(&prefix).into_paginator().send();
        while let Some(page) = stream.next().await {
            let page = page.map_err(|e| Error::new(s3_error_kind(&e), format!("ListObjectsV2 {}: {}", prefix, e)))?;
            for obj in page.contents() {
                let Some(key) = obj.key() else { continue };
                let Some(stem) = key.rsplit('/').next().and_then(|f| f.strip_suffix(".blob")) else { continue };
                if keep_s.as_deref() == Some(stem) { continue; }
                delete_object(client, bucket, key).await?;
                deleted += 1;
            }
        }
        Ok(deleted)
    }

    /// Delete every `*.blob` version object under `FILE/<uuid>/` (used after a
    /// spill). Idempotent. Thin wrapper over [`fs_reconcile_blobs`] (no keep).
    pub async fn delete_all_blob_versions(client: &Client, layout: &HyperDirLayout, bucket: &str, file_uuid: &Uuid) -> Result<usize> {
        Self::fs_reconcile_blobs(client, layout, bucket, file_uuid, None).await
    }

    /// Reclaim superseded versions of a **live** blob file: deletes every
    /// `FILE/<uuid>/*.blob` older than `keep` (the current version, from the
    /// dirent's `blob_ulid`) whose ULID timestamp is older than `retention`.
    /// The current version is always kept. The ULID timestamp is the write
    /// time, so no object read is needed. Returns the count deleted.
    pub async fn fs_gc_blob_versions(client: &Client, layout: &HyperDirLayout, bucket: &str, file_uuid: &Uuid, keep: Ulid, retention: Duration) -> Result<usize> {
        let prefix = layout.blob_prefix(file_uuid);
        let now_ms = unix_now_ms();
        let ret_ms = i64::try_from(retention.as_millis()).unwrap_or(i64::MAX);
        let keep_ts = keep.timestamp_ms();
        let mut deleted = 0;
        let mut stream = client.list_objects_v2().bucket(bucket).prefix(&prefix).into_paginator().send();
        while let Some(page) = stream.next().await {
            let page = page.map_err(|e| Error::new(s3_error_kind(&e), format!("ListObjectsV2 {}: {}", prefix, e)))?;
            for obj in page.contents() {
                let Some(key) = obj.key() else { continue };
                let Some(stem) = key.rsplit('/').next().and_then(|f| f.strip_suffix(".blob")) else { continue };
                let Ok(u) = Ulid::from_string(stem) else { continue };
                if u == keep { continue; } // never delete the current version
                if u.timestamp_ms() < keep_ts && now_ms - (u.timestamp_ms() as i64) >= ret_ms {
                    delete_object(client, bucket, key).await?;
                    deleted += 1;
                }
            }
        }
        Ok(deleted)
    }

    /// Spill a blob file to a segmented hyperfile at the **same uuid** (grew past
    /// threshold / random write / hard link). Reads the current blob bytes,
    /// materializes a real `FILE/<uuid>` hyperfile with those bytes in a single
    /// flush (the commit point: the dirent's `blob_ulid` is cleared to 0 by the
    /// segmented inode's scatter), then deletes the blob versions. Idempotent /
    /// crash-safe: whichever of {blob, hyperfile} the dirent points at wins, the
    /// other is reclaimable by GC. Returns the open hyperfile handle.
    #[allow(clippy::too_many_arguments)]
    pub async fn fs_spill_blob(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        parent_dir_uuid: &Uuid,
        name: &str,
        uuid: &Uuid,
        blob_ulid: Ulid,
        flags: FileFlags,
        mode: FileMode,
    ) -> Result<HyperFileHandle<'static>>
    {
        let bytes = Self::fs_read_blob(client, layout, bucket, uuid, blob_ulid).await?;
        // Build the hyperfile, write the bytes, then flush ONCE so the commit
        // scatter carries the full content (no empty-file window). This mirrors
        // fs_create_file_at but defers the flush until after the write.
        let uri = layout.file_uri(bucket, uuid);
        let parent_dir_uri = layout.dir_uri(bucket, parent_dir_uuid);
        let parent_staging = S3Staging::from(
            client, StagingConfig::new_s3_uri(&parent_dir_uri, None), HyperFileRuntimeConfig::default(),
        ).await?;
        let interceptor = ScatterFirstInterceptor::new(parent_staging, name, *uuid);
        let mut file = HyperFileHandle::fs_create_with_interceptor(client, &uri, flags, mode, interceptor).await?;
        if !bytes.is_empty() {
            file.fs_write(0, &bytes).await?;
        }
        file.fs_flush().await?;
        // Blob versions are now stray; drop them (best-effort — GC backstops).
        let _ = Self::delete_all_blob_versions(client, layout, bucket, uuid).await;
        Ok(file)
    }

    /// Create the root directory (`DIR/<nil-uuid>`).
    ///
    /// The root has no parent, so no scatter interceptor is installed: nothing
    /// observes the root's existence through a parent. Errors with
    /// `AlreadyExists` if the root already exists.
    pub async fn fs_create_root(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        flags: FileFlags,
        mode: FileMode,
    ) -> Result<Self>
    {
        let uri = layout.root_dir_uri(bucket);
        debug!("fs_create_root - uri: {}", uri);
        Self::fs_create(client, &uri, flags, mode).await
    }

    /// Open the root directory (`DIR/<nil-uuid>`).
    pub async fn fs_open_root(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        flags: FileFlags,
    ) -> Result<Self>
    {
        let uri = layout.root_dir_uri(bucket);
        debug!("fs_open_root - uri: {}", uri);
        Self::fs_open(client, &uri, flags).await
    }

    pub async fn fs_open(client: &Client, uri: &str, flags: FileFlags) -> Result<Self>
    {
        debug!("fs_open - uri: {}, flags: {}", uri, flags);
        let staging_config = StagingConfig::new_s3_uri(uri, None);
        let file_config = HyperFileConfigBuilder::new()
                            .with_staging_config(&staging_config)
                            .build();
        let f = HyperFileFlags::from_flags(flags);
        return Self::open(client.clone(), file_config, f).await;
    }

    pub async fn fs_open_opt(client: &Client, uri: &str, flags: FileFlags, runtime_config: &HyperFileRuntimeConfig) -> Result<Self>
    {
        debug!("fs_open_opt - uri: {}, flags: {}", uri, flags);
        let staging_config = StagingConfig::new_s3_uri(uri, None);
        let file_config = HyperFileConfigBuilder::new()
                            .with_staging_config(&staging_config)
                            .with_runtime_config(runtime_config)
                            .build();
        let f = HyperFileFlags::from_flags(flags);
        return Self::open(client.clone(), file_config, f).await;
    }

    pub async fn fs_open_or_create_with_default_opt(client: &Client, uri: &str, flags: FileFlags, mode: FileMode) -> Result<Self>
    {
        debug!("fs_open_or_create - uri: {}, flags: {}", uri, flags);
        let staging_config = StagingConfig::new_s3_uri(uri, None);
        let file_config = HyperFileConfigBuilder::new()
            .with_staging_config(&staging_config)
            .build();
        let f = HyperFileFlags::from_flags(flags);
        let m = HyperFileMode::from_mode(mode);
        return Self::do_open_or_create(client.clone(), file_config, f, m, true).await;
    }

    /// Delete a child by emitting a tombstone scatter to its parent directory.
    ///
    /// This method does **not** physically delete the child's S3 prefix. It
    /// opens the child briefly to capture its current `InodeRaw`, builds a
    /// tombstone body (`TombstoneHeader || InodeRaw`), and emits a `Delete`
    /// scatter into the parent directory's `!/` namespace as a conditional
    /// PUT (`If-None-Match: *`). The next `compact` on the parent removes
    /// the directory-entry mapping but keeps the tombstone scatter in S3 as
    /// the authoritative deletion record. Physical reclamation of the child
    /// prefix is the job of [`fs_gc`], which honours `retention`.
    ///
    /// `parent_dir_uuid` / `name` identify the entry in the parent;
    /// `child_uuid` is the child's own UUID and `child_is_dir` selects the
    /// `DIR/` vs `FILE/` namespace for the child's prefix.
    ///
    /// `retention`:
    /// - `None` => `retention_until_unix_ms = 0`. The next `fs_gc` may
    ///   immediately reclaim the child storage. From the user's perspective
    ///   this is "delete now (physical reclamation is asynchronous)".
    /// - `Some(d)` => the child storage is preserved for at least `d` past
    ///   `now`. Until the retention expires `fs_gc` will leave the child
    ///   prefix alone, which is what makes a future undelete possible.
    // Addressing context is passed as discrete components; this is a
    // provisional surface pending the consumer (hyperfs) driving it.
    #[allow(clippy::too_many_arguments)]
    pub async fn fs_unlink(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        parent_dir_uuid: &Uuid,
        name: &str,
        child_uuid: &Uuid,
        child_is_dir: bool,
        retention: Option<Duration>,
    ) -> Result<()>
    {
        let child_uri = if child_is_dir {
            layout.dir_uri(bucket, child_uuid)
        } else {
            layout.file_uri(bucket, child_uuid)
        };
        let parent_dir_uri = layout.dir_uri(bucket, parent_dir_uuid);
        debug!("fs_unlink - child: {}, parent: {}, name: {}, retention: {:?}",
               child_uri, parent_dir_uri, name, retention);

        // Two-phase delete, phase 1: emit a cheap PreDelete marker into the
        // parent's scatter namespace -- no child inode read, no full tombstone.
        // read_dir/resolve treat a PreDelete as absent immediately. Compaction
        // (the single lease-serialized fold point, phase 2 in fs_compact) reads
        // the child's authoritative inode, writes the real Delete tombstone with
        // this retention deadline, and decrements nlink exactly once. The body
        // carries is_dir + the retention deadline computed now.
        let now_ms = unix_now_ms();
        let retention_until_unix_ms = match retention {
            None => 0,
            Some(d) => {
                let dms = i64::try_from(d.as_millis()).unwrap_or(i64::MAX);
                now_ms.saturating_add(dms)
            },
        };
        let body = build_predelete_body(child_is_dir, retention_until_unix_ms);

        let runtime_config = HyperFileRuntimeConfig::default();
        let parent_staging = S3Staging::from(
            client,
            StagingConfig::new_s3_uri(&parent_dir_uri, None),
            runtime_config,
        ).await?;
        parent_staging.emit_scatter_event(name, child_uuid, &body, DirScatterInodeOp::PreDelete).await?;
        Ok(())
    }

    /// Create a hard link `new_name` in `parent_dir_uuid` to the existing file
    /// `target_file_uuid`.
    ///
    /// Bumps the target file's authoritative link count (stored in the file's
    /// own inode) and adds a directory entry pointing at the same UUID. Hard
    /// links to directories are rejected. Fails with `AlreadyExists` if
    /// `new_name` already exists in the parent.
    ///
    /// The link-count bump and the entry insert are not atomic: a crash
    /// between them over-counts (storage leak), never under-counts.
    pub async fn fs_link(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        parent_dir_uuid: &Uuid,
        new_name: &str,
        target_file_uuid: &Uuid,
    ) -> Result<()>
    {
        let child_uri = layout.file_uri(bucket, target_file_uuid);
        debug!("fs_link - parent: {}, name: {}, target: {}", parent_dir_uuid, new_name, child_uri);

        // Open the parent with FailFast: a concurrent same-name insert must
        // surface as AlreadyExists (not be silently overwritten by the default
        // RetryLastWriterWins flush), so the exclusive insert below can detect
        // the race and avoid clobbering the winner / leaking nlink.
        let dir_uri = layout.dir_uri(bucket, parent_dir_uuid);
        let rc = HyperFileRuntimeConfig { flush_conflict_policy: FlushConflictPolicy::FailFast, ..Default::default() };
        let mut parent = Self::fs_open_opt(client, &dir_uri, FileFlags::rdwr(), &rc).await?;
        if parent.inner.read_entry(new_name).await.is_ok() {
            return Err(Error::new(ErrorKind::AlreadyExists,
                format!("link target name exists: {}/{}", parent_dir_uuid, new_name)));
        }

        // Load the target inode; reject directory targets.
        let runtime_config = HyperFileRuntimeConfig::default();
        let target_staging = S3Staging::from(
            client,
            StagingConfig::new_s3_uri(&child_uri, None),
            runtime_config,
        ).await?;
        let mut raw_inode: InodeRaw = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
        let _ = target_staging.load_inode(raw_inode.as_mut_u8_slice()).await?;
        if is_dir_mode(raw_inode.i_mode) {
            return Err(Error::new(ErrorKind::InvalidInput,
                format!("hard link to a directory is not allowed: {}", child_uri)));
        }

        // Bump the authoritative nlink (its own OCC retry), then record a link
        // intent and add the new entry. Two hazards are handled so a failed/raced
        // link never leaves the target with a stale-high nlink (a storage leak):
        //   * a name created concurrently after our initial check must not be
        //     silently overwritten by the upsert -- re-check the latest dir
        //     state on every attempt and fail with AlreadyExists;
        //   * any path that doesn't commit the entry (lost OCC race exhausted,
        //     a concurrent same-name link, or a transient error) rolls the
        //     nlink bump back.
        // The link intent (written AFTER the bump, so its presence means the
        // bump happened) lets recovery undo a true crash before the entry
        // commit; the only residual leak is a crash in the narrow window between
        // the bump and the intent PUT, which the uuid-level orphan GC reclaims.
        let new_nlink = adjust_nlink(client, &child_uri, 1).await?;
        raw_inode.i_nlink = new_nlink;
        let entry = crate::ondisk::DirFileEntryRaw::from(&raw_inode, target_file_uuid.as_bytes(), new_name.as_bytes());

        let intent_key = layout.link_key(&Ulid::new().to_string());
        let intent = LinkIntent { file: *target_file_uuid, parent: *parent_dir_uuid, name: new_name.to_string() };
        if let Err(e) = client.put_object()
            .bucket(bucket).key(&intent_key)
            .body(SdkBody::from(intent.serialize().as_bytes()).into())
            .send().await
        {
            rollback_link(client, bucket, &intent_key, &child_uri).await; // undo the bump
            return Err(Error::new(s3_error_kind(&e), format!("PutObject link intent s3://{}/{}: {}", bucket, intent_key, e)));
        }

        for _ in 0..NLINK_RETRIES {
            match parent.inner.insert_entry_excl(new_name, entry).await {
                Ok(true) => { let _ = delete_object(client, bucket, &intent_key).await; return Ok(()); },
                Ok(false) => {
                    // The name was claimed by a concurrent op; undo our bump.
                    rollback_link(client, bucket, &intent_key, &child_uri).await;
                    return Err(Error::new(ErrorKind::AlreadyExists,
                        format!("link target name exists: {}/{}", parent_dir_uuid, new_name)));
                },
                Err(e) if e.kind() == ErrorKind::AlreadyExists => {
                    // Lost the dir-inode OCC race; re-open and retry (the
                    // exclusive insert re-checks presence on the latest state).
                    parent = Self::fs_open_opt(client, &dir_uri, FileFlags::rdwr(), &rc).await?;
                },
                Err(e) => {
                    rollback_link(client, bucket, &intent_key, &child_uri).await; // undo our bump
                    return Err(e);
                },
            }
        }
        rollback_link(client, bucket, &intent_key, &child_uri).await; // undo our bump
        Err(Error::new(ErrorKind::ResourceBusy, "fs_link: too many OCC conflicts"))
    }

    /// Remove an empty child directory.
    ///
    /// Opens the child directory and verifies it has no entries (merging its
    /// bmap with any outstanding scatter via `read_dir`); returns
    /// `ErrorKind::DirectoryNotEmpty` otherwise. On success it emits a Delete
    /// tombstone into the parent exactly like [`fs_unlink`], so physical
    /// reclamation is deferred to [`fs_gc`] under `retention`.
    ///
    /// The emptiness check and the tombstone are not atomic. The semantics is
    /// best-effort: a create into the child that lands in the window between
    /// the check and the tombstone is not seen, so rmdir succeeds and that
    /// child becomes a nameless orphan -- reclaimed by the GC chain (`fs_gc`
    /// removes this directory, then `fs_gc_orphans` the child), never a
    /// permanent leak or a dangling entry. A create visible at check time
    /// yields `DirectoryNotEmpty`. Strict ENOTEMPTY-always would require a
    /// per-directory seal checked by every entry-add path (not implemented).
    pub async fn fs_rmdir(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        parent_dir_uuid: &Uuid,
        name: &str,
        child_dir_uuid: &Uuid,
        retention: Option<Duration>,
    ) -> Result<()>
    {
        let child_uri = layout.dir_uri(bucket, child_dir_uuid);
        debug!("fs_rmdir - child: {}, name: {}", child_uri, name);

        let child = Self::fs_open(client, &child_uri, FileFlags::rdonly()).await?;
        let entries = child.fs_read_dir().await?;
        if !entries.is_empty() {
            return Err(Error::new(ErrorKind::DirectoryNotEmpty,
                format!("directory not empty: {} ({} entries)", name, entries.len())));
        }
        drop(child);

        Self::fs_unlink(client, layout, bucket, parent_dir_uuid, name, child_dir_uuid, true, retention).await
    }

    /// Reclaim a child that was displaced by a replace-over-existing rename.
    ///
    /// Unlike [`fs_unlink`], the displaced child has no name left in any
    /// directory (its entry was overwritten in place), so it cannot be
    /// reclaimed by the name-keyed tombstone/GC path. We reclaim it directly.
    ///
    /// This is **idempotent** so it is safe to run from both the foreground
    /// rename and crash recovery: a file is purged only when its stored
    /// `nlink <= 1` (the displaced name was the last/only one), never by a
    /// blind decrement. A still-hard-linked file (`nlink > 1`) is left in
    /// place — its `nlink` stays stale-high (a future leak on its final
    /// unlink, never data loss). Directories (single link) are purged.
    pub async fn fs_reclaim(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        child_uuid: &Uuid,
        child_is_dir: bool,
    ) -> Result<()>
    {
        if child_is_dir {
            let uri = layout.dir_uri(bucket, child_uuid);
            debug!("fs_reclaim - dir: {}", uri);
            return reclaim_child_prefix(client, &uri, HyperFileRuntimeConfig::default()).await;
        }
        let uri = layout.file_uri(bucket, child_uuid);
        debug!("fs_reclaim - file: {}", uri);
        match current_nlink(client, &uri).await? {
            None => Ok(()),                                                   // already reclaimed
            Some(n) if n <= 1 => reclaim_child_prefix(client, &uri, HyperFileRuntimeConfig::default()).await,
            Some(_) => Ok(()),                                                // still hard-linked elsewhere
        }
    }

    /// Reclaim a file iff it is no longer referenced, using the `.unref`-marker
    /// arithmetic `real_nlink = nlink - count(FILE/<uuid>/*.unref)` — **no
    /// directory scan**. `Done` if `real <= 0` and the prefix (incl. markers)
    /// was purged; `Noop` if still referenced or already gone. Safe: a missing
    /// marker only over-counts `real` (a leak the authoritative
    /// [`fs_gc_orphans`]/[`fs_gc_orphans_from`] backstop catches), never
    /// under-counts (which would mis-delete a live file). Drives the keeper's
    /// zero-scan orphan path on a `FILE/<uuid>/*.unref` event.
    pub async fn reclaim_if_unreferenced(client: &Client, layout: &HyperDirLayout, bucket: &str, file_uuid: &Uuid) -> Result<Outcome> {
        let uri = layout.file_uri(bucket, file_uuid);
        let Some(n) = current_nlink(client, &uri).await? else { return Ok(Outcome::Noop) };
        let markers = count_unref_markers(client, layout, bucket, file_uuid).await?;
        if (n as i64) - (markers as i64) <= 0 {
            reclaim_child_prefix(client, &uri, HyperFileRuntimeConfig::default()).await?;
            Ok(Outcome::Done)
        } else {
            Ok(Outcome::Noop)
        }
    }

    /// Reclaim a child displaced by a replace-over-existing rename, recording a
    /// key-visible `.unref` marker for the lost reference so orphaning is
    /// discoverable without a scan. The marker is keyed by `reclaim_key`'s ULID,
    /// so repeated/concurrent calls for the *same* displacement (foreground and
    /// recovery) write it at most once (`If-None-Match`). Directories (single
    /// link) are purged directly. A file with `nlink<=1` (the last reference was
    /// the displaced one) is purged immediately and needs no marker.
    pub async fn reclaim_displaced(client: &Client, layout: &HyperDirLayout, bucket: &str,
                                   displaced: &Uuid, displaced_is_dir: bool, reclaim_key: &str) -> Result<Outcome> {
        if displaced_is_dir {
            let uri = layout.dir_uri(bucket, displaced);
            reclaim_child_prefix(client, &uri, HyperFileRuntimeConfig::default()).await?;
            return Ok(Outcome::Done);
        }
        let uri = layout.file_uri(bucket, displaced);
        match current_nlink(client, &uri).await? {
            None => Ok(Outcome::Noop),                                        // already reclaimed
            Some(n) if n <= 1 => {                                            // last reference: purge, no marker
                reclaim_child_prefix(client, &uri, HyperFileRuntimeConfig::default()).await?;
                Ok(Outcome::Done)
            },
            Some(_) => {                                                      // still hard-linked: record the lost ref
                write_unref_marker(client, layout, bucket, displaced, txn_ulid(reclaim_key)).await?;
                Self::reclaim_if_unreferenced(client, layout, bucket, displaced).await
            },
        }
    }

    /// Durably record (commit point, `If-None-Match: *`) that `displaced` was
    /// replaced under `dst_parent/dst_name` by a rename, so a crash before the
    /// foreground reclaim doesn't orphan its storage. Returns the intent key,
    /// which the caller deletes once it has reclaimed. Left-behind intents are
    /// completed by [`fs_recover_renames`].
    pub async fn fs_emit_reclaim_intent(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        dst_parent: &Uuid,
        dst_name: &str,
        displaced: &Uuid,
        displaced_is_dir: bool,
    ) -> Result<String>
    {
        let intent = ReclaimIntent {
            dst_parent: *dst_parent,
            dst_name: dst_name.to_string(),
            displaced: *displaced,
            displaced_is_dir,
        };
        let key = layout.reclaim_key(&Ulid::new().to_string());
        client.put_object()
            .bucket(bucket)
            .key(&key)
            .body(SdkBody::from(intent.serialize().as_bytes()).into())
            .if_none_match('*')
            .send()
            .await
            .map_err(|e| Error::new(s3_error_kind(&e), format!("PutObject reclaim s3://{}/{}: {}", bucket, key, e)))?;
        Ok(key)
    }

    /// Delete an intent object (best-effort cleanup of a committed reclaim).
    pub async fn fs_delete_intent(client: &Client, bucket: &str, key: &str) -> Result<()> {
        delete_object(client, bucket, key).await
    }

    /// Set an extended attribute (last-write-wins plain PUT). Stored as a
    /// sidecar object `<child prefix>/_xattr/<base64url(name)>`; it lives under
    /// the inode's prefix, so hyperfile's prefix-wide unlink reclaims it too.
    /// Namespace policy (e.g. user.* only) is enforced by the caller.
    pub async fn fs_setxattr(client: &Client, layout: &HyperDirLayout, bucket: &str, uuid: &Uuid, is_dir: bool, name: &str, value: &[u8]) -> Result<()> {
        let key = xattr_key(layout, uuid, is_dir, name);
        client.put_object()
            .bucket(bucket)
            .key(&key)
            .body(SdkBody::from(value.to_vec()).into())
            .send()
            .await
            .map_err(|e| Error::new(s3_error_kind(&e), format!("PutObject xattr s3://{}/{}: {}", bucket, key, e)))?;
        Ok(())
    }

    /// Get an extended attribute's value, or `None` if it isn't set.
    pub async fn fs_getxattr(client: &Client, layout: &HyperDirLayout, bucket: &str, uuid: &Uuid, is_dir: bool, name: &str) -> Result<Option<Vec<u8>>> {
        let key = xattr_key(layout, uuid, is_dir, name);
        match client.get_object().bucket(bucket).key(&key).send().await {
            Ok(res) => {
                let bytes = res.body.collect().await
                    .map_err(|e| Error::other(format!("collect xattr s3://{}/{}: {}", bucket, key, e)))?
                    .into_bytes();
                Ok(Some(bytes.to_vec()))
            },
            Err(e) if e.as_service_error().is_some_and(|s| s.is_no_such_key()) => Ok(None),
            Err(e) => Err(Error::new(s3_error_kind(&e), format!("GetObject xattr s3://{}/{}: {}", bucket, key, e))),
        }
    }

    /// List the names of all set extended attributes.
    pub async fn fs_listxattr(client: &Client, layout: &HyperDirLayout, bucket: &str, uuid: &Uuid, is_dir: bool) -> Result<Vec<String>> {
        let prefix = xattr_prefix(layout, uuid, is_dir);
        let mut names = Vec::new();
        let mut stream = client.list_objects_v2()
            .bucket(bucket)
            .prefix(&prefix)
            .delimiter("/")
            .into_paginator()
            .send();
        while let Some(page) = stream.next().await {
            let page = page.map_err(|e| Error::new(s3_error_kind(&e), format!("ListObjectsV2 {}: {}", prefix, e)))?;
            for obj in page.contents() {
                let Some(seg) = obj.key().and_then(|k| k.strip_prefix(&prefix)) else { continue };
                if seg.is_empty() { continue; }
                if let Ok(bytes) = B64URL.decode(seg) {
                    if let Ok(s) = String::from_utf8(bytes) { names.push(s); }
                }
            }
        }
        Ok(names)
    }

    /// Remove an extended attribute. Returns `false` if it wasn't set.
    pub async fn fs_removexattr(client: &Client, layout: &HyperDirLayout, bucket: &str, uuid: &Uuid, is_dir: bool, name: &str) -> Result<bool> {
        if Self::fs_getxattr(client, layout, bucket, uuid, is_dir, name).await?.is_none() {
            return Ok(false);
        }
        delete_object(client, bucket, &xattr_key(layout, uuid, is_dir, name)).await?;
        Ok(true)
    }

    /// Reclaim physical storage for tombstoned children whose retention has
    /// expired.
    ///
    /// Lists scatter objects under the parent directory `parent_dir_uuid`,
    /// picks out the `Delete` (tombstone) scatters, reads each one's
    /// `TombstoneHeader`, and for every tombstone whose
    /// `retention_until_unix_ms <= now_ms`:
    ///   1. resolves the child URI from the child UUID carried in the scatter
    ///      and the file/dir mode read from the tombstone's `InodeRaw`,
    ///   2. opens the child staging and calls `unlink()` to remove the
    ///      child's prefix in full (inode + segments + scatter folder),
    ///   3. deletes the tombstone scatter object.
    ///
    /// Tombstones whose retention has not yet passed are skipped silently;
    /// they remain in S3 and a future `fs_gc` call after expiration will
    /// process them. If a child prefix is already gone (e.g. because a
    /// previous partial `fs_gc` run died between the unlink and the
    /// tombstone delete), the tombstone scatter is still removed so the
    /// directory's scatter listing converges.
    ///
    /// `fs_gc` is intentionally separate from `fs_compact`. Compact is
    /// expected to run at high frequency; GC walks tombstones, issues
    /// Does any mount currently hold `file_uuid` open (a live open-lease, i.e.
    /// `_OPEN/<uuid>/<mount-id>` whose owning mount's heartbeat is younger than
    /// `ttl_secs`)? Dead-owner leases are ignored. **Fail-safe**: returns `true`
    /// on any list/head uncertainty, so GC never reclaims under a possible open
    /// fd. Used to defer reclamation of a keep-open file (`FLAG_KEEP_OPEN`).
    pub async fn has_live_open_lease(client: &Client, layout: &HyperDirLayout, bucket: &str, file_uuid: &Uuid, ttl_secs: u64) -> bool {
        let prefix = layout.open_lease_prefix(file_uuid);
        let mut stream = client.list_objects_v2().bucket(bucket).prefix(&prefix).into_paginator().send();
        while let Some(page) = stream.next().await {
            let page = match page { Ok(p) => p, Err(_) => return true };
            for obj in page.contents() {
                let Some(key) = obj.key() else { continue };
                let Some(mount_id) = key.rsplit('/').next() else { continue };
                let hb = layout.heartbeat_key(mount_id);
                match client.head_object().bucket(bucket).key(&hb).send().await {
                    Ok(out) => {
                        let Some(lm) = out.last_modified() else { return true }; // unknown age -> assume alive
                        let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)
                            .map(|d| d.as_secs() as i64).unwrap_or(0);
                        if now - lm.secs() < ttl_secs as i64 { return true; }
                    },
                    Err(e) => {
                        if e.raw_response().map(|r| r.status().as_u16() == 404).unwrap_or(false) {
                            // dead/absent owner heartbeat: stale lease, ignore
                        } else {
                            return true; // unknown: fail safe
                        }
                    },
                }
            }
        }
        false
    }

    /// physical deletes, and is appropriate for cron/admin cadence.
    pub async fn fs_gc(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        parent_dir_uuid: &Uuid,
    ) -> Result<GcStats>
    {
        let parent_uri = layout.dir_uri(bucket, parent_dir_uuid);
        debug!("fs_gc - parent_uri: {}", parent_uri);
        let runtime_config = HyperFileRuntimeConfig::default();
        let parent_staging = S3Staging::from(
            client,
            StagingConfig::new_s3_uri(&parent_uri, None),
            runtime_config.clone(),
        ).await?;

        let scatters = parent_staging.list_scatter_inodes().await?;
        let now_ms = unix_now_ms();

        // Read handle used to check whether a tombstone's delete has been
        // folded into the persisted bmap yet (see the per-tombstone guard
        // below). Opened once and reused.
        let dir = Self::fs_open(client, &parent_uri, FileFlags::rdonly()).await?;

        let mut stats = GcStats::default();

        for scatter in scatters {
            if !matches!(scatter.op, DirScatterInodeOp::Delete) {
                continue;
            }
            stats.tombstones_visited += 1;

            // Read the tombstone body to extract the retention deadline and
            // the child's inode (whose mode tells us DIR vs FILE namespace).
            let body = match get_object_bytes(&parent_staging, &scatter.key).await {
                Ok(b) => b,
                Err(e) => {
                    warn!("fs_gc: failed to GET tombstone body s3://{}/{}: {}",
                          parent_staging.bucket, scatter.key, e);
                    stats.errors += 1;
                    continue;
                },
            };
            let (header, inode_raw_bytes) = match parse_tombstone_body(&body) {
                Ok(v) => v,
                Err(e) => {
                    warn!("fs_gc: malformed tombstone body s3://{}/{}: {}",
                          parent_staging.bucket, scatter.key, e);
                    stats.errors += 1;
                    continue;
                },
            };

            if header.retention_until_unix_ms > now_ms {
                stats.tombstones_skipped_retention += 1;
                continue;
            }

            // Fold guard: only finalize a tombstone whose delete has been
            // folded out of the persisted bmap. If the name still resolves in
            // the bmap to this same uuid, a compact hasn't folded the delete
            // yet; deleting the tombstone now would leave the bmap entry as a
            // ghost (read_dir would show it with no tombstone to mask it).
            // Leave it for a future GC pass after a compact folds the delete.
            // (A name resolving to a *different* uuid means this delete was
            // already superseded/folded, so it is safe to finalize.)
            match dir.fs_read_entry(&scatter.filename).await {
                Ok(entry) if entry.uuid == scatter.uuid => {
                    stats.tombstones_skipped_unfolded += 1;
                    continue;
                },
                Ok(_) => {}, // name now maps elsewhere: this delete is folded/superseded
                Err(e) if e.kind() == ErrorKind::NotFound => {}, // folded out: safe
                Err(e) => {
                    warn!("fs_gc: bmap fold-check failed for {}: {}", scatter.filename, e);
                    stats.errors += 1;
                    continue;
                },
            }

            // Resolve the child's prefix from its UUID and kind. The kind is
            // recovered from the inode mode stored in the tombstone.
            let inode_raw = InodeRaw::from_u8_slice(inode_raw_bytes);
            let is_dir = is_dir_mode(inode_raw.i_mode);
            let child_uri = if is_dir {
                layout.dir_uri(bucket, &scatter.uuid)
            } else {
                layout.file_uri(bucket, &scatter.uuid)
            };

            // Opt-in cross-mount open-but-unlinked: never reclaim a keep-open
            // file's storage while a mount still holds it open (a live
            // open-lease). This closes the open-vs-unlink race -- an unlink that
            // tombstoned the file because it missed a just-written lease must not
            // let GC pull the storage out from under that open fd. Keep the
            // tombstone; a later pass reclaims once the last lease drops.
            if !is_dir && inode_raw.is_keep_open()
                && Self::has_live_open_lease(client, layout, bucket, &scatter.uuid, open_lease_ttl_secs()).await
            {
                debug!("fs_gc: defer reclaim of keep-open {} (live open-lease)", scatter.uuid);
                continue;
            }

            // A file may still be reachable through other hard links; reclaim
            // only when its authoritative link count has reached zero. A
            // missing child inode means a previous GC already reclaimed it.
            // Directories carry exactly one name and are reclaimed outright.
            let should_reclaim = if is_dir {
                true
            } else {
                match current_nlink(client, &child_uri).await {
                    Ok(Some(n)) => n == 0,
                    Ok(None) => true,
                    Err(e) => {
                        warn!("fs_gc: failed to read nlink for {}: {}", child_uri, e);
                        stats.errors += 1;
                        continue;
                    },
                }
            };

            if should_reclaim {
                match reclaim_child_prefix(client, &child_uri, runtime_config.clone()).await {
                    Ok(_) => {},
                    Err(e) => {
                        warn!("fs_gc: failed to reclaim child {}: {}", child_uri, e);
                        stats.errors += 1;
                        continue;
                    },
                }
            }

            // Tombstone scatter object goes last: a partial GC that crashes
            // here will leave the tombstone behind, and a future fs_gc will
            // observe a missing child prefix (handled as already-cleaned)
            // and finish the job. The tombstone is removed whether or not the
            // child storage was reclaimed -- this name is gone either way.
            if let Err(e) = parent_staging.remove_scatter_inodes(vec![scatter.key.clone()]).await {
                warn!("fs_gc: failed to delete tombstone s3://{}/{}: {}",
                      parent_staging.bucket, scatter.key, e);
                stats.errors += 1;
                continue;
            }

            stats.tombstones_reclaimed += 1;
        }

        Ok(stats)
    }

    pub async fn fs_release(&mut self) -> Result<u64>
    {
        debug!("fs_release - ");
        self.inner.release().await
    }

    pub async fn fs_flush(&mut self) -> Result<u64>
    {
        debug!("fs_flush - ");
        self.inner.flush().await
    }

    pub fn fs_getattr(&self) -> Result<libc::stat>
    {
        debug!("fs_getattr - ");
        Ok(self.inner.stat())
    }

    pub async fn fs_getattr_fast(client: &Client, uri: &str) -> Result<libc::stat>
    {
        debug!("fs_getattr_fast - uri: {}", uri);
        let staging_config = StagingConfig::new_s3_uri(uri, None);
        let file_config = HyperFileConfigBuilder::new()
            .with_staging_config(&staging_config)
            .build();
        return Self::stat_fast(client.clone(), file_config).await;
    }

    pub async fn fs_chmod(&mut self, mode: libc::mode_t) -> Result<libc::stat>
    {
        debug!("fs_chmod - mode: {:#o}", mode);
        let mut stat = self.inner.stat();
        // update permission part only, don't change file type part
        stat.st_mode = (stat.st_mode & libc::S_IFMT) | (mode & !libc::S_IFMT);
        self.inner.update_stat(&stat).await
    }

    pub async fn fs_chown(&mut self, uid: libc::uid_t, gid: libc::gid_t) -> Result<libc::stat> {
        debug!("fs_chown - uid: {}, gid: {}", uid, gid);
        let mut stat = self.inner.stat();
        stat.st_uid = uid;
        stat.st_gid = gid;
        self.inner.update_stat(&stat).await
    }

    pub async fn fs_setattr(&mut self, stat: &libc::stat) -> Result<libc::stat> {
        debug!("fs_setattr - mode: {}, uid: {}, gid: {}", stat.st_mode, stat.st_uid, stat.st_gid);
        self.inner.update_stat(stat).await
    }

    pub async fn fs_setattr_fast(client: &Client, uri: &str, stat: &libc::stat) -> Result<libc::stat>
    {
        debug!("fs_setattr_fast - uri: {}", uri);
        let staging_config = StagingConfig::new_s3_uri(uri, None);
        let file_config = HyperFileConfigBuilder::new()
                            .with_staging_config(&staging_config)
                            .build();
        return Self::update_stat_fast(client.clone(), file_config, stat).await;
    }

    pub async fn fs_read_entry(&self, name: &str) -> Result<DirFileEntry>
    {
        debug!("fs_read_entry - name: {}", name);
        self.inner.read_entry(name).await
    }

    /// Scatter-aware single-name resolve (the cheap lookup path): see
    /// [`HyperDirFile::resolve_entry`]. Returns `None` if the name is absent.
    pub async fn fs_resolve_entry(&self, name: &str) -> Result<Option<DirFileEntry>>
    {
        debug!("fs_resolve_entry - name: {}", name);
        self.inner.resolve_entry(name).await
    }

    /// Handle-less single-name resolve: same result as [`fs_resolve_entry`] but
    /// without first opening a directory handle. The lookup hot path (one call
    /// per path component) thus avoids the open's redundant inode GET: zero
    /// inode GETs when the name still has pending scatter, one on the compacted
    /// fallback (vs. two for open-then-resolve).
    pub async fn fs_resolve_entry_fast(
        client: &Client, layout: &HyperDirLayout, bucket: &str, parent_uuid: &Uuid, name: &str,
    ) -> Result<Option<DirFileEntry>>
    {
        let uri = layout.dir_uri(bucket, parent_uuid);
        let staging_config = StagingConfig::new_s3_uri(&uri, None);
        let dir_config = S3Staging::to_dir_staging_config(&staging_config);
        let staging = S3Staging::from(client, dir_config, HyperFileRuntimeConfig::default()).await?;
        let loader = staging.to_block_loader();
        HyperDirFile::<S3Staging, S3BlockLoader>::resolve_fast(staging, loader, name).await
    }

    pub async fn fs_read_dir(&self) -> Result<Vec<DirFileEntry>>
    {
        debug!("fs_readdir - ");
        self.inner.read_dir().await
    }

    /// Consolidate outstanding scatter objects into the persisted bmap.
    ///
    /// Acquires the directory's compactor leader lease before doing any
    /// work, so concurrent callers on the same directory typically see
    /// `Err(ErrorKind::ResourceBusy)` returned fast rather than racing
    /// through the consolidate-and-flush path; correctness also relies on
    /// hyperfile's per-inode OCC, but the lease avoids the duplicated I/O
    /// that OCC alone cannot. The lease is best-effort released after
    /// either success or failure; on holder crash the lease is reclaimed
    /// after `DEFAULT_COMPACT_LEASE_TTL_MS`.
    /// Test-only: run just **phase 1** of compaction — fold scatter into the
    /// bmap and strip consolidated scatter (including folding PreDelete markers
    /// out of the bmap and stripping them), but WITHOUT phase 2 (reading child
    /// inodes, writing the real Delete tombstones, decrementing nlink). This
    /// deterministically opens the phase-1→phase-2 window: after it, a
    /// PreDelete'd name is absent (no bmap entry, no scatter) yet no Delete
    /// tombstone exists.
    #[doc(hidden)]
    pub async fn fs_compact_phase1_only(&mut self) -> Result<CompactStats> {
        self.inner.compact().await
    }

    pub async fn fs_compact(&mut self) -> Result<CompactStats>
    {
        debug!("fs_compact - ");
        let stats = self.inner.compact().await?;
        self.compact_phase2(&stats).await;
        Ok(stats)
    }

    /// Fold pending scatter and rename `old_name` to `new_name` in this
    /// directory, committed by a SINGLE flush (see
    /// `HyperDirFile::compact_then_rename`), then run compaction phase 2.
    ///
    /// A same-directory rename must fold first (rename works on the B-tree and
    /// cannot see scatter); doing it as compact-then-rename costs two commits
    /// for one logical change. Same semantics as `fs_compact` followed by
    /// `fs_rename`, with one commit instead of two.
    pub async fn fs_compact_and_rename(&mut self, old_name: &str, new_name: &str) -> Result<CompactStats>
    {
        debug!("fs_compact_and_rename - {} -> {}", old_name, new_name);
        let stats = self.inner.compact_and_rename(old_name, new_name).await?;
        self.compact_phase2(&stats).await;
        Ok(stats)
    }

    /// Compaction phase 2: the post-lease work driven by what phase 1 decided
    /// actually transitioned present->absent (nlink decrements, real Delete
    /// tombstones for folded PreDeletes, reclaim of displaced identities).
    /// Best-effort throughout: every step is idempotent across passes and a
    /// failure at worst leaves work for a later compaction or the orphan sweep.
    async fn compact_phase2(&mut self, stats: &CompactStats)
    {
        // Phase 2 of the two-phase delete plus the legacy raw-Delete path.
        // Compaction holds the per-directory lease, so probe_delete under it
        // decides -- exactly once, even under concurrent/duplicate unlinks --
        // which names actually transitioned present->absent this pass; the
        // post-lease work below (nlink decrement, tombstone write) is driven by
        // that decision. nlink lags until compaction, consistent with the
        // deferred (tombstone) deletion model.
        if !stats.removed_children.is_empty() || !stats.predeleted.is_empty() {
            let client = self.inner.staging().client.clone();
            let bucket = self.inner.staging().bucket.clone();
            let rp = self.inner.staging().root_path.clone();
            let base = &rp[..rp.rfind("DIR/").unwrap_or(0)]; // "{base}DIR/{uuid}" -> "{base}"

            // Legacy path: a raw Delete tombstone folded out of the bmap (rare;
            // normal unlinks go through the PreDelete path below). Just
            // decrement nlink once. A directory child has no FILE inode
            // (NotFound) and is skipped.
            for uuid in &stats.removed_children {
                let child_uri = format!("s3://{}/{}FILE/{}", bucket, base, uuid);
                match adjust_nlink(&client, &child_uri, -1).await {
                    Ok(_) => {},
                    Err(e) if e.kind() == ErrorKind::NotFound => {},
                    Err(e) => warn!("fs_compact: nlink decrement failed for {}: {}", child_uri, e),
                }
            }

            // Two-phase delete, phase 2: for each folded PreDelete, read the
            // child's authoritative inode now, write the real Delete tombstone
            // (with the retention deadline the unlink recorded) into this
            // directory's scatter, then decrement a file's nlink once. A later
            // compact folds that tombstone (probe_delete already false -> no
            // double dec); fs_gc reclaims the prefix once nlink hits 0 and
            // retention passes. A crash mid-step at worst leaves a nameless
            // file that the orphan GC reclaims.
            for (name, uuid, is_dir, retention_until) in &stats.predeleted {
                let child_uri = if *is_dir {
                    format!("s3://{}/{}DIR/{}", bucket, base, uuid)
                } else {
                    format!("s3://{}/{}FILE/{}", bucket, base, uuid)
                };
                let staging = match S3Staging::from(
                    &client, StagingConfig::new_s3_uri(&child_uri, None), HyperFileRuntimeConfig::default()).await {
                    Ok(s) => s,
                    Err(e) if e.kind() == ErrorKind::NotFound => continue, // child already gone
                    Err(e) => { warn!("fs_compact: open child {} failed: {}", child_uri, e); continue; },
                };
                let mut raw: InodeRaw = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
                if let Err(e) = staging.load_inode(raw.as_mut_u8_slice()).await {
                    warn!("fs_compact: load child inode {} failed: {}", child_uri, e);
                    continue;
                }
                let body = build_tombstone_body(unix_now_ms(), *retention_until, raw.as_u8_slice());
                if let Err(e) = self.inner.staging()
                    .emit_scatter_event(name, uuid, &body, DirScatterInodeOp::Delete).await {
                    warn!("fs_compact: write tombstone for {} failed: {}", child_uri, e);
                    continue; // don't decrement without a tombstone to drive GC
                }
                if !*is_dir {
                    match adjust_nlink(&client, &child_uri, -1).await {
                        Ok(_) => {},
                        Err(e) if e.kind() == ErrorKind::NotFound => {},
                        Err(e) => warn!("fs_compact: nlink decrement failed for {}: {}", child_uri, e),
                    }
                }
            }
        }

        // Reclaim identities displaced under a name this pass by a winning
        // same-name create (last-writer-wins). Each is reclaimed exactly like a
        // rename-over-existing displacement -- reference-safe via reclaim_displaced
        // (a still hard-linked file is kept behind a `.unref` marker) -- so a
        // losing/displaced create never leaves a nameless orphan for the orphan
        // sweep. Idempotent across passes: once the winner is folded into the
        // bmap, the displacement is no longer observed.
        if !stats.displaced.is_empty() {
            let client = self.inner.staging().client.clone();
            let bucket = self.inner.staging().bucket.clone();
            let rp = self.inner.staging().root_path.clone();
            let base = &rp[..rp.rfind("DIR/").unwrap_or(0)]; // "{base}DIR/{uuid}" -> "{base}"
            let layout = HyperDirLayout::with_base(base);
            for (uuid, is_dir, ulid) in &stats.displaced {
                // reclaim_key carries the displacing/losing scatter's ULID so a
                // `.unref` marker (only the still-hard-linked case) is written at
                // most once per displacement.
                let reclaim_key = layout.reclaim_key(&ulid.to_string());
                if let Err(e) = Self::reclaim_displaced(&client, &layout, &bucket, uuid, *is_dir, &reclaim_key).await {
                    warn!("fs_compact: reclaim displaced {} failed: {}", uuid, e);
                }
            }
        }
    }

    /// Rename an entry within this (already-open, writable) directory.
    ///
    /// Same-directory rename keeps the child's UUID and storage; only this
    /// directory's name->entry mapping changes, committed by a single inode
    /// flush (atomic via hyperfile OCC). The destination must not exist
    /// (otherwise `AlreadyExists`); `old_name == new_name` is a no-op.
    pub async fn fs_rename(&mut self, old_name: &str, new_name: &str) -> Result<()>
    {
        debug!("fs_rename - {} -> {}", old_name, new_name);
        self.inner.rename_within(old_name, new_name).await
    }

    /// Atomically exchange two entries in this directory (`RENAME_EXCHANGE`):
    /// the identities their names map to are swapped, committed by a single
    /// inode flush (atomic via hyperfile OCC). Both names must exist (otherwise
    /// `NotFound`); `name_a == name_b` is a no-op. Same-directory only —
    /// cross-directory exchange would need a bidirectional two-phase commit.
    pub async fn fs_exchange(&mut self, name_a: &str, name_b: &str) -> Result<()>
    {
        debug!("fs_exchange - {} <-> {}", name_a, name_b);
        self.inner.exchange_within(name_a, name_b).await
    }

    /// Rename a child across two different directories.
    ///
    /// The child keeps its UUID and storage; only the two parents' mappings
    /// change. Because the two parent inode flushes cannot be made atomic
    /// together, a rename intent object is written first as the single commit
    /// point: once it exists the rename is logically committed and is
    /// completed forward (add to destination, remove from source) idempotently,
    /// then the intent is deleted. A crash at any point leaves the intent
    /// behind for [`fs_recover_renames`] to finish.
    ///
    /// Source must exist (`NotFound`) and destination must not
    /// (`AlreadyExists`); both are checked before the commit. The pre-checks
    /// and the commit are not atomic, so a destination created concurrently
    /// between the check and the apply can be overwritten (a higher-layer
    /// concern).
    #[allow(clippy::too_many_arguments)]
    pub async fn fs_rename_across(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        src_parent_uuid: &Uuid,
        src_name: &str,
        dst_parent_uuid: &Uuid,
        dst_name: &str,
        reclaim_key: Option<&str>,
    ) -> Result<()>
    {
        debug!("fs_rename_across - {}/{} -> {}/{}",
               src_parent_uuid, src_name, dst_parent_uuid, dst_name);

        let (key, intent) = emit_rename_intent(
            client, layout, bucket, src_parent_uuid, src_name, dst_parent_uuid, dst_name).await?;

        // The destination-add (atomic on the bmap) tells us what identity, if
        // any, actually held the destination name -- including one the caller's
        // pre-check raced and missed. Reclaim a DIFFERENT displaced identity so
        // it is never orphaned (reference-safe via reclaim_displaced).
        let displaced = apply_rename_intent(client, layout, bucket, &intent).await?;
        if let Some(old) = displaced {
            if old.uuid != *intent.child.as_bytes() {
                let duuid = Uuid::from_bytes(old.uuid);
                match reclaim_key {
                    // Caller pre-recorded a reclaim intent: reclaim under its
                    // key so foreground and recovery converge on one `.unref`
                    // marker, and propagate errors so the intent (still held by
                    // the caller) drives recovery if this fails.
                    Some(k) => { Self::reclaim_displaced(client, layout, bucket, &duuid, old.is_dir(), k).await?; },
                    // Uncaught displacement (no intent): best-effort -- the move
                    // already committed, so a reclaim hiccup must not fail the
                    // rename; orphan GC backstops it. Keyed by the displaced uuid
                    // so repeats are idempotent.
                    None => {
                        let dk = layout.reclaim_key(&Ulid::from_bytes(old.uuid).to_string());
                        if let Err(e) = Self::reclaim_displaced(client, layout, bucket, &duuid, old.is_dir(), &dk).await {
                            warn!("fs_rename_across: best-effort reclaim of uncaught displaced {} failed: {}", duuid, e);
                        }
                    },
                }
            }
        }

        // Best-effort intent cleanup; a leftover intent is harmless and will
        // be re-applied (idempotently) and removed by fs_recover_renames.
        let _ = delete_object(client, bucket, &key).await;
        Ok(())
    }

    /// Test-only: perform a cross-directory rename's pre-checks and write the
    /// commit-point intent object, then return WITHOUT applying it -- i.e.
    /// simulate a crash immediately after the commit. A subsequent
    /// [`fs_recover_renames`] must complete the move.
    #[doc(hidden)]
    #[allow(clippy::too_many_arguments)]
    pub async fn fs_rename_emit_intent_only(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        src_parent_uuid: &Uuid,
        src_name: &str,
        dst_parent_uuid: &Uuid,
        dst_name: &str,
    ) -> Result<()>
    {
        let _ = emit_rename_intent(
            client, layout, bucket, src_parent_uuid, src_name, dst_parent_uuid, dst_name).await?;
        Ok(())
    }

    /// Test-only: pre-check a cross-directory rename, write the intent, and
    /// apply ONLY the destination-add -- i.e. simulate a crash after the child
    /// is added to the destination but before it is removed from the source, so
    /// it is momentarily reachable under both names. `fs_recover_renames` must
    /// complete the move (idempotently re-adding the destination and removing
    /// the source), converging to the single destination name.
    #[doc(hidden)]
    #[allow(clippy::too_many_arguments)]
    pub async fn fs_rename_apply_partial(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
        src_parent_uuid: &Uuid,
        src_name: &str,
        dst_parent_uuid: &Uuid,
        dst_name: &str,
    ) -> Result<()>
    {
        let (_key, intent) = emit_rename_intent(
            client, layout, bucket, src_parent_uuid, src_name, dst_parent_uuid, dst_name).await?;
        // Destination-add only (the first half of apply_rename_intent), without
        // removing the source -- so it is momentarily reachable under both names.
        let inode_raw = InodeRaw::from_u8_slice(&intent.inode);
        let mut entry = crate::ondisk::DirFileEntryRaw::from(&inode_raw, intent.child.as_bytes(), intent.dst_name.as_bytes());
        if let Some(u) = intent.blob_ulid { entry.set_blob_ulid(u); }
        let mut dst = Self::fs_open_dir(client, layout, bucket, &intent.dst_parent, FileFlags::rdwr()).await?;
        let _ = dst.inner.insert_entry(&intent.dst_name, entry).await?;
        Ok(())
    }

    /// Test-only: simulate a crash in the *second phase* of a delete -- the
    /// name has been folded out of the parent's bmap and the `Delete` tombstone
    /// written, but the child's `nlink` was NOT decremented yet. The child is
    /// then nameless with a stale-high `nlink`, so `fs_gc` (which only reclaims
    /// a file once `nlink == 0`) cannot reclaim it; `fs_gc_orphans` must.
    #[doc(hidden)]
    pub async fn fs_unlink_crash_after_tombstone(
        client: &Client, layout: &HyperDirLayout, bucket: &str,
        parent_dir_uuid: &Uuid, name: &str, child_uuid: &Uuid,
    ) -> Result<()>
    {
        let child_uri = layout.file_uri(bucket, child_uuid);
        let staging = S3Staging::from(client, StagingConfig::new_s3_uri(&child_uri, None),
            HyperFileRuntimeConfig::default()).await?;
        let mut raw: InodeRaw = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
        staging.load_inode(raw.as_mut_u8_slice()).await?;
        let body = build_tombstone_body(unix_now_ms(), 0, raw.as_u8_slice());
        let parent_staging = S3Staging::from(client,
            StagingConfig::new_s3_uri(&layout.dir_uri(bucket, parent_dir_uuid), None),
            HyperFileRuntimeConfig::default()).await?;
        parent_staging.emit_scatter_event(name, child_uuid, &body, DirScatterInodeOp::Delete).await?;
        let mut parent = Self::fs_open_dir(client, layout, bucket, parent_dir_uuid, FileFlags::rdwr()).await?;
        let _ = parent.inner.remove_entry(name).await?;
        Ok(()) // nlink deliberately NOT decremented
    }

    /// Test-only: simulate a crash in `fs_link` after the `nlink` bump but
    /// before the directory entry was inserted -- an over-count. The file keeps
    /// its existing name(s); once the last real name is unlinked it is a
    /// nameless file with a stale-high `nlink` that only `fs_gc_orphans` can
    /// reclaim.
    #[doc(hidden)]
    pub async fn fs_link_bump_only(
        client: &Client, layout: &HyperDirLayout, bucket: &str, target_file_uuid: &Uuid,
    ) -> Result<()>
    {
        let _ = adjust_nlink(client, &layout.file_uri(bucket, target_file_uuid), 1).await?;
        Ok(())
    }

    /// Test-only: simulate a crash in `fs_link` after the `nlink` bump AND after
    /// the `.link` intent was written, but before the directory entry was
    /// committed — the recoverable form of source-(b). The intent's ULID is
    /// backdated by `age_ms` so recovery treats it as stale at once (no real
    /// wait). Returns the intent key.
    #[doc(hidden)]
    pub async fn fs_link_crash_after_intent(
        client: &Client, layout: &HyperDirLayout, bucket: &str,
        parent_dir_uuid: &Uuid, new_name: &str, target_file_uuid: &Uuid, age_ms: u64,
    ) -> Result<String>
    {
        let child_uri = layout.file_uri(bucket, target_file_uuid);
        let _ = adjust_nlink(client, &child_uri, 1).await?;
        let ts = (unix_now_ms() as u64).saturating_sub(age_ms);
        let key = layout.link_key(&Ulid::from_parts(ts, 0).to_string());
        let li = LinkIntent { file: *target_file_uuid, parent: *parent_dir_uuid, name: new_name.to_string() };
        client.put_object().bucket(bucket).key(&key)
            .body(SdkBody::from(li.serialize().as_bytes()).into())
            .send().await
            .map_err(|e| Error::new(s3_error_kind(&e), format!("PutObject link intent s3://{}/{}: {}", bucket, key, e)))?;
        Ok(key)
    }

    /// Re-drive any rename intents left behind by an interrupted
    /// [`fs_rename_across`] (e.g. a crash between the commit and the intent
    /// delete). Each intent's forward steps are idempotent, so re-applying a
    /// fully-completed rename is a no-op.
    pub async fn fs_recover_renames(
        client: &Client,
        layout: &HyperDirLayout,
        bucket: &str,
    ) -> Result<usize>
    {
        let prefix = layout.txn_prefix();
        debug!("fs_recover_renames - prefix: {}", prefix);
        let mut recovered = 0;
        let mut stream = client.list_objects_v2()
            .bucket(bucket)
            .prefix(&prefix)
            .into_paginator()
            .send();
        while let Some(page) = stream.next().await {
            let page = page.map_err(|e| Error::new(s3_error_kind(&e), format!("ListObjectsV2 {}: {}", prefix, e)))?;
            if let Some(objects) = page.contents {
                for obj in objects.iter() {
                    let Some(key) = obj.key() else { continue };
                    if key.ends_with(".intent") {
                        let body = get_object_raw(client, bucket, key).await?;
                        let intent = match RenameIntent::parse(&body) {
                            Ok(i) => i,
                            Err(e) => { warn!("fs_recover_renames: bad intent {}: {}", key, e); continue; }
                        };
                        let _ = apply_rename_intent(client, layout, bucket, &intent).await?;
                        delete_object(client, bucket, key).await?;
                        recovered += 1;
                    } else if key.ends_with(".reclaim") {
                        let body = get_object_raw(client, bucket, key).await?;
                        let ri = match ReclaimIntent::parse(&body) {
                            Ok(i) => i,
                            Err(e) => { warn!("fs_recover_renames: bad reclaim {}: {}", key, e); continue; }
                        };
                        // Is the displaced child still named dst_name? (scatter-aware)
                        // Fail closed: a non-NotFound read error means we can't
                        // confirm whether dst_name still points to the displaced
                        // object; abort the round rather than reclaim a possibly
                        // still-named object (which would dangle dst_name). A
                        // NotFound parent cannot name it -> not still named.
                        let still = match Self::fs_open_dir(client, layout, bucket, &ri.dst_parent, FileFlags::rdonly()).await {
                            Ok(dir) => match dir.fs_read_dir().await {
                                Ok(es) => es.iter().any(|e| e.name == ri.dst_name && e.uuid == ri.displaced),
                                Err(e) if e.kind() == ErrorKind::NotFound => false,
                                Err(e) => return Err(e),
                            },
                            Err(e) if e.kind() == ErrorKind::NotFound => false,
                            Err(e) => return Err(e),
                        };
                        if still {
                            // Rename hasn't completed: the displaced child is still validly
                            // named, so don't reclaim. Only clean up a clearly-abandoned
                            // (stale) intent, to avoid racing an in-flight rename.
                            if txn_stale(key) { delete_object(client, bucket, key).await?; }
                        } else {
                            // Rename completed: the displaced child is orphaned. Idempotent.
                            Self::reclaim_displaced(client, layout, bucket, &ri.displaced, ri.displaced_is_dir, key).await?;
                            delete_object(client, bucket, key).await?;
                            recovered += 1;
                        }
                    } else if key.ends_with(".link") {
                        // Recover an interrupted fs_link (stale intents only;
                        // recover_link no-ops on fresh ones to avoid racing it).
                        if Self::recover_link(client, layout, bucket, key).await? == Outcome::Done {
                            recovered += 1;
                        }
                    }
                }
            }
        }
        Ok(recovered)
    }

    /// Open a directory by its UUID.
    pub async fn fs_open_dir(client: &Client, layout: &HyperDirLayout, bucket: &str, uuid: &Uuid, flags: FileFlags) -> Result<Self> {
        Self::fs_open(client, &layout.dir_uri(bucket, uuid), flags).await
    }

    /// Recover a single cross-directory rename intent by key — the per-key,
    /// event-driven counterpart of [`fs_recover_renames`]. Idempotent:
    /// - `Noop` if the intent is still **fresh** (younger than
    ///   [`TXN_INTENT_STALE_MS`]): a live rename normally applies and deletes it
    ///   within milliseconds, so only an abandoned one is recovered here (avoids
    ///   racing the live applier); the periodic/backstop pass catches a truly
    ///   crashed rename once its intent goes stale.
    /// - `Noop` if the intent object is already gone (applied + deleted).
    /// - `Done` if a stale intent was forward-applied and deleted.
    pub async fn recover_intent(client: &Client, layout: &HyperDirLayout, bucket: &str, intent_key: &str) -> Result<Outcome> {
        if !txn_stale(intent_key) { return Ok(Outcome::Noop); }
        let body = match get_object_raw(client, bucket, intent_key).await {
            Ok(b) => b,
            Err(e) if e.kind() == ErrorKind::NotFound => return Ok(Outcome::Noop),
            Err(e) => return Err(e),
        };
        let intent = RenameIntent::parse(&body)?;
        let _ = apply_rename_intent(client, layout, bucket, &intent).await?;
        delete_object(client, bucket, intent_key).await?;
        Ok(Outcome::Done)
    }

    /// Process a single displaced-child reclaim intent by key — the per-key,
    /// event-driven counterpart of the `.reclaim` branch of
    /// [`fs_recover_renames`]. Idempotent:
    /// - if the displaced child is still validly named at `dst_name`, the rename
    ///   hasn't completed → `Noop` (or delete a clearly-stale intent → `Done`);
    /// - otherwise the child is orphaned → reclaim it and delete the intent
    ///   (`Done`). `Noop` if the intent object is already gone.
    pub async fn process_reclaim(client: &Client, layout: &HyperDirLayout, bucket: &str, reclaim_key: &str) -> Result<Outcome> {
        let body = match get_object_raw(client, bucket, reclaim_key).await {
            Ok(b) => b,
            Err(e) if e.kind() == ErrorKind::NotFound => return Ok(Outcome::Noop),
            Err(e) => return Err(e),
        };
        let ri = ReclaimIntent::parse(&body)?;
        // Fail closed: if we can't read dst_parent to confirm whether it still
        // names the displaced object, return the error (retried) rather than
        // reclaim a possibly still-named object. A NotFound parent cannot name it.
        let still = match Self::fs_open_dir(client, layout, bucket, &ri.dst_parent, FileFlags::rdonly()).await {
            Ok(dir) => match dir.fs_read_dir().await {
                Ok(es) => es.iter().any(|e| e.name == ri.dst_name && e.uuid == ri.displaced),
                Err(e) if e.kind() == ErrorKind::NotFound => false,
                Err(e) => return Err(e),
            },
            Err(e) if e.kind() == ErrorKind::NotFound => false,
            Err(e) => return Err(e),
        };
        if still {
            if txn_stale(reclaim_key) {
                delete_object(client, bucket, reclaim_key).await?;
                return Ok(Outcome::Done);
            }
            return Ok(Outcome::Noop);
        }
        Self::reclaim_displaced(client, layout, bucket, &ri.displaced, ri.displaced_is_dir, reclaim_key).await?;
        delete_object(client, bucket, reclaim_key).await?;
        Ok(Outcome::Done)
    }

    /// Recover a single hard-link intent by key — the per-key, event-driven
    /// counterpart used by both [`fs_recover_renames`] and the keeper. The
    /// intent exists only between `fs_link`'s `nlink` bump and the entry commit,
    /// so its presence means the bump happened. Idempotent and safe:
    /// - `Noop` if the intent is still **fresh** (younger than
    ///   [`TXN_INTENT_STALE_MS`]) — never races a live `fs_link`;
    /// - `Noop` if the intent object is already gone;
    /// - if the link's directory entry **was** committed (the link actually
    ///   succeeded, crashing only before the intent delete), just delete the
    ///   intent (`Done`) — the bump is correct;
    /// - otherwise the link was abandoned: record the phantom `nlink` bump as a
    ///   `.unref` marker keyed by the intent's ULID, then delete the intent
    ///   (`Done`). The marker is the same write-once scheme used for displaced
    ///   hard links (source-(a)), so it is idempotent across concurrent
    ///   recoverers — `real = nlink − count(.unref)` then reflects the abandoned
    ///   reference, and the file is reclaimed when it falls to zero. (A raw
    ///   decrement could not be used: a recovered `_TXN` delete is idempotent in
    ///   S3, so two recoverers would each decrement.)
    pub async fn recover_link(client: &Client, layout: &HyperDirLayout, bucket: &str, link_key: &str) -> Result<Outcome> {
        if !txn_stale(link_key) { return Ok(Outcome::Noop); }
        let body = match get_object_raw(client, bucket, link_key).await {
            Ok(b) => b,
            Err(e) if e.kind() == ErrorKind::NotFound => return Ok(Outcome::Noop),
            Err(e) => return Err(e),
        };
        let li = LinkIntent::parse(&body)?;
        // Did the link ever commit its directory entry? (scatter-aware)
        let committed = match Self::fs_open_dir(client, layout, bucket, &li.parent, FileFlags::rdonly()).await {
            Ok(dir) => dir.fs_read_dir().await?.iter().any(|e| e.name == li.name && e.uuid == li.file),
            Err(e) if e.kind() == ErrorKind::NotFound => false, // parent gone -> entry can't exist
            Err(e) => return Err(e),                            // transient -> retry, never mis-account
        };
        if !committed {
            // Abandoned: record the phantom +1 idempotently, but only if the
            // file still exists (else it was already reclaimed — nothing to
            // offset, and a marker under a purged prefix would be a stray).
            match current_nlink(client, &layout.file_uri(bucket, &li.file)).await {
                Ok(Some(_)) => write_unref_marker(client, layout, bucket, &li.file, txn_ulid(link_key)).await?,
                Ok(None) => {}            // file already gone
                Err(e) => return Err(e), // transient -> retry, keep the intent
            }
        }
        delete_object(client, bucket, link_key).await?;
        Ok(Outcome::Done)
    }

    /// Enumerate every directory's UUID by listing the `DIR/` namespace with a
    /// `/` delimiter. Used by a maintenance pass to drive per-directory
    /// `fs_compact` / `fs_gc` across the whole tree.
    pub async fn fs_list_dir_uuids(client: &Client, layout: &HyperDirLayout, bucket: &str) -> Result<Vec<Uuid>> {
        list_namespace_uuids(client, bucket, &layout.dir_prefix()).await
    }

    /// Enumerate every file's UUID by listing the `FILE/` namespace.
    pub async fn fs_list_file_uuids(client: &Client, layout: &HyperDirLayout, bucket: &str) -> Result<Vec<Uuid>> {
        list_namespace_uuids(client, bucket, &layout.file_prefix()).await
    }

    /// Reclaim orphaned files: those no directory entry references anymore.
    ///
    /// The name-keyed tombstone GC ([`fs_gc`]) can't see a file that became
    /// nameless without a tombstone (e.g. the rare hard-linked child displaced
    /// by a replace-over-existing rename, left with a stale-high nlink). This
    /// is the backstop: mark every UUID referenced by any directory's
    /// `read_dir` (scatter-aware, so freshly created entries count), then purge
    /// any `FILE/<uuid>` not in that set whose inode is older than `grace`, and
    /// any EMPTY `DIR/<uuid>` not in that set (excluding the root) -- catching a
    /// crashed mkdir's nameless directory too. The grace window avoids
    /// collecting a child whose creation is still in flight. A non-empty orphan
    /// directory subtree is left in place (a documented limitation).
    pub async fn fs_gc_orphans(client: &Client, layout: &HyperDirLayout, bucket: &str, grace: Duration) -> Result<usize> {
        let dir_uuids = Self::fs_list_dir_uuids(client, layout, bucket).await?;
        let file_uuids = Self::fs_list_file_uuids(client, layout, bucket).await?;
        // Scanning sweep: also reconcile blob versions (prune superseded live
        // versions past grace, and drop stray `*.blob` under now-segmented files
        // left by a crashed spill — the O(referenced files) LIST cost is only
        // paid by this authoritative pass, not the Inventory-fed variant).
        Self::fs_gc_orphans_inner(client, layout, bucket, &file_uuids, &dir_uuids, grace, true).await
    }

    /// Inventory-fed variant of [`fs_gc_orphans`] (no `LIST` of `DIR/`/`FILE/`):
    /// the caller supplies the candidate `file_uuids` and the **complete**
    /// `dir_uuids` (the reference universe — a partial list would mis-delete a
    /// referenced file). Reads each directory's entries (scatter-aware) to build
    /// the referenced set, then purges any candidate file / empty candidate dir
    /// not referenced and older than `grace`. **Fail-closed**: if any directory
    /// can't be read (other than being gone, `NotFound`), the round aborts with
    /// that error and reclaims nothing — never deletes a file on an incomplete
    /// reference set. The authoritative, directory-content backstop behind the
    /// keeper's `.unref` fast path.
    pub async fn fs_gc_orphans_from(client: &Client, layout: &HyperDirLayout, bucket: &str,
                                    file_uuids: &[Uuid], dir_uuids: &[Uuid], grace: Duration) -> Result<usize> {
        // Inventory-fed callers stay zero-extra-scan: no per-file blob LIST here.
        // Blob version reconciliation is done by the scanning `fs_gc_orphans`, or
        // per-target by the keeper via `fs_gc_blob_versions` / `fs_reconcile_blobs`.
        Self::fs_gc_orphans_inner(client, layout, bucket, file_uuids, dir_uuids, grace, false).await
    }

    #[allow(clippy::too_many_arguments)]
    async fn fs_gc_orphans_inner(client: &Client, layout: &HyperDirLayout, bucket: &str,
                                 file_uuids: &[Uuid], dir_uuids: &[Uuid], grace: Duration,
                                 reconcile_blobs: bool) -> Result<usize> {
        use std::collections::{HashSet, HashMap};
        let mut referenced: HashSet<Uuid> = HashSet::new();
        let mut blob_current: HashMap<Uuid, Ulid> = HashMap::new();
        for d in dir_uuids {
            // Build the reference universe FAIL-CLOSED. A file is reclaimed below
            // iff it is absent from `referenced`, so an incomplete universe would
            // mis-classify a still-referenced file as an orphan and DELETE it. A
            // transient/unknown failure to read a directory therefore aborts the
            // whole round (reclaim nothing; a later pass retries — leaking is
            // safe, losing a live file is not). `NotFound` is the one error that
            // is certain information: the directory is gone, hence was empty
            // (rmdir / displaced-dir reclaim / orphan-dir reclaim all require
            // empty), hence referenced nothing live — skip it. Tolerating
            // `NotFound` also keeps the Inventory-fed caller working when its
            // directory list is slightly stale (reclaimed-since dirs).
            match Self::fs_open_dir(client, layout, bucket, d, FileFlags::rdonly()).await {
                Ok(dir) => match dir.fs_read_dir().await {
                    Ok(entries) => for e in entries {
                        referenced.insert(e.uuid);
                        if let Some(u) = e.blob_ulid { blob_current.insert(e.uuid, u); }
                    },
                    Err(e) if e.kind() == ErrorKind::NotFound => {},
                    Err(e) => return Err(e),
                },
                Err(e) if e.kind() == ErrorKind::NotFound => {},
                Err(e) => return Err(e),
            }
        }

        let now_ms = unix_now_ms();
        let grace_ms = i64::try_from(grace.as_millis()).unwrap_or(i64::MAX);
        let mut reclaimed = 0;
        for fu in file_uuids {
            if referenced.contains(fu) { continue; }
            let uri = layout.file_uri(bucket, fu);
            match Self::fs_getattr_fast(client, &uri).await {
                Ok(st) => {
                    // Segmented (or in-inode-inline) file: age-gate on inode ctime,
                    // then reclaim the whole prefix via hyperfile unlink.
                    if now_ms - st.st_ctime.saturating_mul(1000) >= grace_ms
                        && reclaim_child_prefix(client, &uri, HyperFileRuntimeConfig::default()).await.is_ok() {
                        reclaimed += 1;
                    }
                },
                Err(_) => {
                    // No hyperfile inode: a blob-only prefix (deleted/orphaned
                    // blob). `reclaim_child_prefix` opens via a staging that
                    // reports NotFound without an inode and short-circuits to
                    // Ok *without deleting anything*, so drop the blob versions
                    // directly (age-gated on the newest version's ULID
                    // timestamp), then best-effort clean any non-blob leftovers.
                    if let Some(ts) = newest_blob_ts_ms(client, layout, bucket, fu).await {
                        if now_ms - ts >= grace_ms {
                            if Self::fs_reconcile_blobs(client, layout, bucket, fu, None).await.is_ok() {
                                reclaimed += 1;
                            }
                            let _ = reclaim_child_prefix(client, &uri, HyperFileRuntimeConfig::default()).await;
                        }
                    }
                },
            }
        }

        // Orphaned directories: a `DIR/<uuid>` that no parent references (e.g. a
        // crashed mkdir that wrote the inode before scattering its name) leaks
        // otherwise -- `fs_gc` only reclaims tombstoned dirs. Reclaim only an
        // EMPTY, old-enough orphan (an empty dir strands no children); the root
        // has no parent and is never an orphan. A non-empty orphan subtree is
        // left in place (reclaiming it safely would need a reachability walk
        // from the root).
        for du in dir_uuids {
            if du.is_nil() || referenced.contains(du) { continue; }
            let empty = match Self::fs_open_dir(client, layout, bucket, du, FileFlags::rdonly()).await {
                Ok(dir) => dir.fs_read_dir().await.map(|e| e.is_empty()).unwrap_or(false),
                Err(_) => false,
            };
            if !empty { continue; }
            let uri = layout.dir_uri(bucket, du);
            match Self::fs_getattr_fast(client, &uri).await {
                Ok(st) if now_ms - st.st_ctime.saturating_mul(1000) >= grace_ms => {
                    if reclaim_child_prefix(client, &uri, HyperFileRuntimeConfig::default()).await.is_ok() {
                        reclaimed += 1;
                    }
                },
                _ => {},
            }
        }

        // Blob version reconciliation for REFERENCED files (docs/second-level-inline.md §8),
        // scanning sweep only. A live blob keeps its current version and drops
        // older-than-grace ones; a now-segmented file's stray `*.blob` (crashed
        // spill) is dropped. Unreferenced files are already whole-prefix-deleted
        // above, so they are skipped here.
        if reconcile_blobs {
            for fu in file_uuids {
                if !referenced.contains(fu) { continue; }
                match blob_current.get(fu) {
                    Some(cur) => { let _ = Self::fs_gc_blob_versions(client, layout, bucket, fu, *cur, grace).await; },
                    None => { let _ = Self::fs_reconcile_blobs(client, layout, bucket, fu, None).await; },
                }
            }
        }

        Ok(reclaimed)
    }
}

/// List the UUIDs of the prefixes directly under `prefix` (e.g. `DIR/` or
/// `FILE/`), using a `/` delimiter to get one common prefix per UUID.
async fn list_namespace_uuids(client: &Client, bucket: &str, prefix: &str) -> Result<Vec<Uuid>> {
    debug!("list_namespace_uuids - prefix: {}", prefix);
    let mut out = Vec::new();
    let mut stream = client.list_objects_v2()
        .bucket(bucket)
        .prefix(prefix)
        .delimiter("/")
        .into_paginator()
        .send();
    while let Some(page) = stream.next().await {
        let page = page.map_err(|e| Error::new(s3_error_kind(&e), format!("ListObjectsV2 {}: {}", prefix, e)))?;
        for cp in page.common_prefixes() {
            let Some(p) = cp.prefix() else { continue };
            let rest = p.strip_prefix(prefix).unwrap_or(p).trim_end_matches('/');
            if let Ok(u) = Uuid::parse_str(rest) {
                out.push(u);
            }
        }
    }
    Ok(out)
}

/// Statistics returned from [`HyperDir::fs_gc`].
#[derive(Default, Debug, Clone, Copy)]
pub struct GcStats {
    /// Number of tombstone scatter objects examined this round.
    pub tombstones_visited: usize,
    /// Tombstones whose retention had not yet expired and were left in
    /// place for a future GC pass.
    pub tombstones_skipped_retention: usize,
    /// Tombstones whose delete has not yet been folded into the persisted
    /// bmap (the name is still live there). Finalizing them now would orphan
    /// the bmap entry, so they are left for a future GC pass (after a compact
    /// folds the delete).
    pub tombstones_skipped_unfolded: usize,
    /// Tombstones whose child prefix was successfully unlinked and whose
    /// scatter object was successfully removed.
    pub tombstones_reclaimed: usize,
    /// Tombstones that hit a recoverable error (LIST/GET/PUT/DELETE failure,
    /// malformed body, etc.). These are left in S3 and a subsequent `fs_gc`
    /// call may retry them. The detailed reason is in the log at `warn` level.
    pub errors: usize,
}

/// Result of a per-key maintenance entry point ([`HyperDir::recover_intent`] /
/// [`HyperDir::process_reclaim`]): lets an event-driven caller decide whether to
/// ack the triggering event or retry it. Transient/busy conditions surface as
/// `Err(ResourceBusy)` (retry), not as a variant.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Outcome {
    /// Work was applied (and the intent object deleted).
    Done,
    /// Nothing to do: the intent is gone, already applied, or still in flight
    /// (too fresh) and deliberately deferred to the periodic/backstop pass.
    Noop,
}

/// A cross-directory rename, captured durably so the move can be completed
/// (or recovered) after the single intent-object commit.
struct RenameIntent {
    src_parent: Uuid,
    src_name: String,
    dst_parent: Uuid,
    dst_name: String,
    /// The child's UUID; preserved across the move (its storage never moves).
    child: Uuid,
    /// The source entry's raw inode bytes, moved verbatim to the destination.
    inode: Vec<u8>,
    /// The source entry's blob version pointer, if it is a second-level inline
    /// ("blob") file. It must travel with the entry: without it the destination
    /// entry would keep the size but point at nothing, so the file would read as
    /// empty. `None` for an in-inode inline or a segmented file.
    blob_ulid: Option<Ulid>,
}

impl RenameIntent {
    /// key=value text body; names and inode are base64 (values are otherwise
    /// arbitrary bytes). Mirrors the lease body's no-dependency style.
    fn serialize(&self) -> String {
        let mut s = format!(
            "src_parent={}\nsrc_name={}\ndst_parent={}\ndst_name={}\nchild={}\ninode={}\n",
            self.src_parent,
            B64.encode(self.src_name.as_bytes()),
            self.dst_parent,
            B64.encode(self.dst_name.as_bytes()),
            self.child,
            B64.encode(&self.inode),
        );
        // Optional line: absent means "not a blob", so an intent written by an
        // older build still parses.
        if let Some(u) = self.blob_ulid {
            s.push_str(&format!("blob_ulid={}\n", u));
        }
        s
    }

    fn parse(buf: &[u8]) -> Result<Self> {
        let s = std::str::from_utf8(buf)
            .map_err(|e| Error::new(ErrorKind::InvalidData, format!("intent not UTF-8: {}", e)))?;
        let mut src_parent = None;
        let mut src_name = None;
        let mut dst_parent = None;
        let mut dst_name = None;
        let mut child = None;
        let mut inode = None;
        let mut blob_ulid = None;
        let b64s = |v: &str| -> Result<String> {
            let bytes = B64.decode(v).map_err(|e| Error::new(ErrorKind::InvalidData, format!("intent b64: {}", e)))?;
            String::from_utf8(bytes).map_err(|e| Error::new(ErrorKind::InvalidData, format!("intent name: {}", e)))
        };
        let uuid = |v: &str| -> Result<Uuid> {
            Uuid::parse_str(v).map_err(|e| Error::new(ErrorKind::InvalidData, format!("intent uuid: {}", e)))
        };
        for line in s.lines() {
            if let Some(v) = line.strip_prefix("src_parent=") { src_parent = Some(uuid(v)?); }
            else if let Some(v) = line.strip_prefix("src_name=") { src_name = Some(b64s(v)?); }
            else if let Some(v) = line.strip_prefix("dst_parent=") { dst_parent = Some(uuid(v)?); }
            else if let Some(v) = line.strip_prefix("dst_name=") { dst_name = Some(b64s(v)?); }
            else if let Some(v) = line.strip_prefix("child=") { child = Some(uuid(v)?); }
            else if let Some(v) = line.strip_prefix("inode=") {
                inode = Some(B64.decode(v).map_err(|e| Error::new(ErrorKind::InvalidData, format!("intent inode b64: {}", e)))?);
            }
            else if let Some(v) = line.strip_prefix("blob_ulid=") {
                blob_ulid = Some(Ulid::from_string(v)
                    .map_err(|e| Error::new(ErrorKind::InvalidData, format!("intent blob_ulid: {}", e)))?);
            }
        }
        Ok(Self {
            src_parent: src_parent.ok_or_else(|| Error::new(ErrorKind::InvalidData, "intent missing src_parent"))?,
            src_name: src_name.ok_or_else(|| Error::new(ErrorKind::InvalidData, "intent missing src_name"))?,
            dst_parent: dst_parent.ok_or_else(|| Error::new(ErrorKind::InvalidData, "intent missing dst_parent"))?,
            dst_name: dst_name.ok_or_else(|| Error::new(ErrorKind::InvalidData, "intent missing dst_name"))?,
            child: child.ok_or_else(|| Error::new(ErrorKind::InvalidData, "intent missing child"))?,
            inode: inode.ok_or_else(|| Error::new(ErrorKind::InvalidData, "intent missing inode"))?,
            blob_ulid,
        })
    }
}

/// A reclaim intent for a child displaced by a replace-over-existing rename.
/// The displaced child's name was overwritten in place (no tombstone), so it
/// is recorded here durably; recovery reclaims it if the rename completed.
struct ReclaimIntent {
    dst_parent: Uuid,
    dst_name: String,
    displaced: Uuid,
    displaced_is_dir: bool,
}

impl ReclaimIntent {
    fn serialize(&self) -> String {
        format!(
            "dst_parent={}\ndst_name={}\ndisplaced={}\nis_dir={}\n",
            self.dst_parent,
            B64.encode(self.dst_name.as_bytes()),
            self.displaced,
            self.displaced_is_dir as u8,
        )
    }

    fn parse(buf: &[u8]) -> Result<Self> {
        let s = std::str::from_utf8(buf)
            .map_err(|e| Error::new(ErrorKind::InvalidData, format!("reclaim not UTF-8: {}", e)))?;
        let mut dst_parent = None;
        let mut dst_name = None;
        let mut displaced = None;
        let mut is_dir = false;
        for line in s.lines() {
            if let Some(v) = line.strip_prefix("dst_parent=") {
                dst_parent = Some(Uuid::parse_str(v).map_err(|e| Error::new(ErrorKind::InvalidData, format!("reclaim uuid: {}", e)))?);
            } else if let Some(v) = line.strip_prefix("dst_name=") {
                let bytes = B64.decode(v).map_err(|e| Error::new(ErrorKind::InvalidData, format!("reclaim b64: {}", e)))?;
                dst_name = Some(String::from_utf8(bytes).map_err(|e| Error::new(ErrorKind::InvalidData, format!("reclaim name: {}", e)))?);
            } else if let Some(v) = line.strip_prefix("displaced=") {
                displaced = Some(Uuid::parse_str(v).map_err(|e| Error::new(ErrorKind::InvalidData, format!("reclaim uuid: {}", e)))?);
            } else if let Some(v) = line.strip_prefix("is_dir=") {
                is_dir = v == "1";
            }
        }
        Ok(Self {
            dst_parent: dst_parent.ok_or_else(|| Error::new(ErrorKind::InvalidData, "reclaim missing dst_parent"))?,
            dst_name: dst_name.ok_or_else(|| Error::new(ErrorKind::InvalidData, "reclaim missing dst_name"))?,
            displaced: displaced.ok_or_else(|| Error::new(ErrorKind::InvalidData, "reclaim missing displaced"))?,
            displaced_is_dir: is_dir,
        })
    }
}

/// A hard-link in progress, recorded durably *after* the `nlink` bump so a
/// crash before the directory entry is committed (orphan source-(b)) can be
/// recovered instead of leaking a stale-high `nlink`. The intent existing means
/// the bump definitely happened.
struct LinkIntent {
    file: Uuid,
    parent: Uuid,
    name: String,
}

impl LinkIntent {
    fn serialize(&self) -> String {
        format!("file={}\nparent={}\nname={}\n", self.file, self.parent, B64.encode(self.name.as_bytes()))
    }

    fn parse(buf: &[u8]) -> Result<Self> {
        let s = std::str::from_utf8(buf)
            .map_err(|e| Error::new(ErrorKind::InvalidData, format!("link not UTF-8: {}", e)))?;
        let (mut file, mut parent, mut name) = (None, None, None);
        for line in s.lines() {
            if let Some(v) = line.strip_prefix("file=") {
                file = Some(Uuid::parse_str(v).map_err(|e| Error::new(ErrorKind::InvalidData, format!("link uuid: {}", e)))?);
            } else if let Some(v) = line.strip_prefix("parent=") {
                parent = Some(Uuid::parse_str(v).map_err(|e| Error::new(ErrorKind::InvalidData, format!("link uuid: {}", e)))?);
            } else if let Some(v) = line.strip_prefix("name=") {
                let bytes = B64.decode(v).map_err(|e| Error::new(ErrorKind::InvalidData, format!("link b64: {}", e)))?;
                name = Some(String::from_utf8(bytes).map_err(|e| Error::new(ErrorKind::InvalidData, format!("link name: {}", e)))?);
            }
        }
        Ok(Self {
            file: file.ok_or_else(|| Error::new(ErrorKind::InvalidData, "link missing file"))?,
            parent: parent.ok_or_else(|| Error::new(ErrorKind::InvalidData, "link missing parent"))?,
            name: name.ok_or_else(|| Error::new(ErrorKind::InvalidData, "link missing name"))?,
        })
    }
}

/// Wall-clock age beyond which a `_TXN/` intent (rename `.intent`, reclaim
/// `.reclaim`, or link `.link`) is treated as abandoned rather than the
/// in-flight tail of a live op. Exposed so an external maintainer can pick stale
/// intents off an S3 Inventory listing by the object's ULID/`mtime`.
pub const TXN_INTENT_STALE_MS: i64 = 60_000;

/// Whether a `_TXN/<ulid>.{intent,reclaim,link}` object's ULID timestamp is
/// older than [`TXN_INTENT_STALE_MS`] (or its stem isn't a parseable ULID).
fn txn_stale(key: &str) -> bool {
    let stem = key.rsplit('/').next().unwrap_or(key);
    let stem = stem.strip_suffix(".reclaim")
        .or_else(|| stem.strip_suffix(".intent"))
        .or_else(|| stem.strip_suffix(".link"))
        .unwrap_or(stem);
    match Ulid::from_string(stem) {
        Ok(u) => unix_now_ms() - (u.timestamp_ms() as i64) > TXN_INTENT_STALE_MS,
        Err(_) => true,
    }
}

/// Pre-check a cross-directory rename and write its commit-point intent
/// object (`If-None-Match: *`), returning the intent's S3 key and the parsed
/// intent. Source must exist (`NotFound`); destination must not
/// (`AlreadyExists`). Shared by `fs_rename_across` and the crash-injection
/// test entry.
async fn emit_rename_intent(
    client: &Client,
    layout: &HyperDirLayout,
    bucket: &str,
    src_parent_uuid: &Uuid,
    src_name: &str,
    dst_parent_uuid: &Uuid,
    dst_name: &str,
) -> Result<(String, RenameIntent)> {
    // Source-scoped commit point: every concurrent rename of the SAME source
    // maps to one deterministic key, so `If-None-Match: *` admits exactly one
    // winner. This gives cross-directory rename the same move-once guarantee
    // same-directory rename gets from a single bmap OCC flush -- without it two
    // concurrent renames of one source would each commit a distinct intent and
    // leave the child reachable under two names with nlink unchanged (a dangling
    // entry once one name is unlinked). The recovery path is key-agnostic (it
    // parses the body), so the key format is free to change.
    let txn_id = format!("{}_{}", src_parent_uuid, B64URL.encode(src_name));
    let key = layout.txn_key(&txn_id);
    for attempt in 0..RENAME_CLAIM_RETRIES {
        // (Re)read the source entry. This is also the move-once check: if a
        // concurrent winner already moved the source away, read_entry_raw is
        // NotFound and we surface that (the loser sees ENOENT, exactly as if the
        // two renames had been serialized).
        let src = HyperDir::fs_open_dir(client, layout, bucket, src_parent_uuid, FileFlags::rdonly()).await?;
        let entry_raw = src.inner.read_entry_raw(src_name).await?;
        drop(src);

        // An existing destination is overwritten: apply_rename_intent's
        // destination-add is an upsert, so the displaced entry's btree slot is
        // replaced in place (no tombstone). The caller reclaims the displaced
        // child's storage.
        let intent = RenameIntent {
            src_parent: *src_parent_uuid,
            src_name: src_name.to_string(),
            dst_parent: *dst_parent_uuid,
            dst_name: dst_name.to_string(),
            child: Uuid::from_bytes(entry_raw.uuid),
            inode: entry_raw.inode.as_u8_slice().to_vec(),
            // Carry the blob pointer across, or the moved file reads as empty.
            blob_ulid: entry_raw.blob_ulid(),
        };

        match client.put_object()
            .bucket(bucket)
            .key(&key)
            .body(SdkBody::from(intent.serialize().as_bytes()).into())
            .if_none_match('*')
            .send()
            .await
        {
            Ok(_) => {
                // Won the exclusive claim. Re-verify the source survived the
                // race: winning the claim means any prior holder has finished,
                // and a holder removes the source before releasing its claim
                // (foreground apply and fs_recover_renames both do), so a
                // concurrently-moved source now reads absent. This closes the
                // read-vs-claim TOCTOU -- without the recheck the loser could
                // read the source as present, then win the just-freed claim and
                // re-apply a stale move (the source under two names).
                let src = HyperDir::fs_open_dir(client, layout, bucket, src_parent_uuid, FileFlags::rdonly()).await?;
                let still = src.inner.read_entry_raw(src_name).await;
                drop(src);
                match still {
                    Ok(e) if e.uuid == entry_raw.uuid => return Ok((key, intent)),
                    Ok(_) => { // source replaced by a different file
                        let _ = delete_object(client, bucket, &key).await;
                        return Err(Error::new(ErrorKind::NotFound, "rename: source moved concurrently"));
                    },
                    Err(e) if e.kind() == ErrorKind::NotFound => {
                        let _ = delete_object(client, bucket, &key).await;
                        return Err(e);
                    },
                    Err(e) => { let _ = delete_object(client, bucket, &key).await; return Err(e); },
                }
            },
            Err(e) if e.raw_response().map(|r| matches!(r.status().as_u16(), 412 | 409)).unwrap_or(false) => {
                // Another rename of this source holds the claim: back off and
                // recheck. It will either finish (source gone -> NotFound above)
                // or release the claim so we can win.
                tokio::time::sleep(Duration::from_millis(40 * (attempt as u64 + 1))).await;
            },
            Err(e) => return Err(Error::new(s3_error_kind(&e), format!("PutObject intent s3://{}/{}: {}", bucket, key, e))),
        }
    }
    Err(Error::new(ErrorKind::ResourceBusy, "rename: source has a concurrent rename in progress"))
}

/// Forward-apply a committed rename intent: add the entry to the destination
/// directory, then remove it from the source. Both steps are idempotent, so
/// this is safe to re-run during recovery. Destination-add happens first so
/// the child is never unreachable by name mid-rename.
async fn apply_rename_intent(client: &Client, layout: &HyperDirLayout, bucket: &str, intent: &RenameIntent) -> Result<Option<crate::ondisk::DirFileEntryRaw>> {
    let inode_raw = InodeRaw::from_u8_slice(&intent.inode);
    let mut entry = crate::ondisk::DirFileEntryRaw::from(&inode_raw, intent.child.as_bytes(), intent.dst_name.as_bytes());
    // A blob entry must arrive with its version pointer, else the destination
    // keeps the size but points at nothing and the file reads as empty.
    if let Some(u) = intent.blob_ulid { entry.set_blob_ulid(u); }

    // Destination-add first so the child is never unreachable by name mid-rename.
    // insert_entry returns the entry it overwrote in place: if that is a
    // DIFFERENT identity (a destination the rename's pre-check didn't catch --
    // e.g. a name a concurrent client created), the caller reclaims it instead
    // of leaving it a nameless orphan.
    let mut dst = HyperDir::fs_open_dir(client, layout, bucket, &intent.dst_parent, FileFlags::rdwr()).await?;
    let displaced = dst.inner.insert_entry(&intent.dst_name, entry).await?;
    drop(dst);

    let mut src = HyperDir::fs_open_dir(client, layout, bucket, &intent.src_parent, FileFlags::rdwr()).await?;
    let _ = src.inner.remove_entry(&intent.src_name).await?;
    Ok(displaced)
}

/// GET an object's raw bytes by key.
async fn get_object_raw(client: &Client, bucket: &str, key: &str) -> Result<bytes::Bytes> {
    let res = client.get_object().bucket(bucket).key(key).send().await
        .map_err(|e| Error::new(s3_error_kind(&e), format!("GetObject s3://{}/{}: {}", bucket, key, e)))?;
    let bytes = res.body.collect().await
        .map_err(|e| Error::other(format!("collect body s3://{}/{}: {}", bucket, key, e)))?
        .into_bytes();
    Ok(bytes)
}

/// DELETE an object by key.
async fn delete_object(client: &Client, bucket: &str, key: &str) -> Result<()> {
    client.delete_object().bucket(bucket).key(key).send().await
        .map_err(|e| Error::new(s3_error_kind(&e), format!("DeleteObject s3://{}/{}: {}", bucket, key, e)))?;
    Ok(())
}

/// Undo a hard link that did not commit its directory entry: delete the link
/// intent and, **only if that delete succeeds**, decrement the phantom `nlink`
/// bump. The guard makes the decrement at-most-once across processes — a crash
/// between the delete and the decrement leaves a stale-high `nlink` that
/// `fs_gc_orphans` reclaims, never a double decrement (which would drop a live
/// reference). `delete_object` is idempotent (S3 DELETE succeeds on a missing
/// key), so this is also safe when the intent PUT itself failed.
async fn rollback_link(client: &Client, bucket: &str, intent_key: &str, child_uri: &str) {
    if delete_object(client, bucket, intent_key).await.is_ok() {
        let _ = adjust_nlink(client, child_uri, -1).await;
    }
}

/// Count this file's `.unref` markers (one per displaced hard link not yet
/// folded into `nlink`). A scoped LIST of the single file's prefix.
async fn count_unref_markers(client: &Client, layout: &HyperDirLayout, bucket: &str, file_uuid: &Uuid) -> Result<usize> {
    let prefix = format!("{}/", layout.file_key(file_uuid));
    let mut n = 0usize;
    let mut stream = client.list_objects_v2().bucket(bucket).prefix(&prefix).into_paginator().send();
    while let Some(page) = stream.next().await {
        let page = page.map_err(|e| Error::new(s3_error_kind(&e), format!("ListObjectsV2 {}: {}", prefix, e)))?;
        for obj in page.contents() {
            if obj.key().is_some_and(|k| k.ends_with(".unref")) { n += 1; }
        }
    }
    Ok(n)
}

/// Write the `.unref` marker for one displaced reference, keyed by the
/// displacement ULID so it is write-once (`If-None-Match`); a 412/409 means it
/// was already recorded (idempotent).
async fn write_unref_marker(client: &Client, layout: &HyperDirLayout, bucket: &str, file_uuid: &Uuid, ulid: &str) -> Result<()> {
    let key = layout.unref_key(file_uuid, ulid);
    match client.put_object().bucket(bucket).key(&key)
        .body(SdkBody::from(ulid.as_bytes()).into())
        .if_none_match('*')
        .send().await
    {
        Ok(_) => Ok(()),
        Err(sdk_err) => {
            if sdk_err.raw_response().map(|r| matches!(r.status().as_u16(), 412 | 409)).unwrap_or(false) {
                Ok(())
            } else {
                Err(Error::new(s3_error_kind(&sdk_err), format!("PutObject unref s3://{}/{}: {}", bucket, key, sdk_err)))
            }
        }
    }
}

/// The ULID stem of a `_TXN/<ulid>.{intent,reclaim,link}` key, used to key the
/// matching `.unref` marker so foreground and recovery agree on one marker.
fn txn_ulid(key: &str) -> &str {
    let stem = key.rsplit('/').next().unwrap_or(key);
    stem.strip_suffix(".reclaim")
        .or_else(|| stem.strip_suffix(".intent"))
        .or_else(|| stem.strip_suffix(".link"))
        .unwrap_or(stem)
}

/// True if a Unix mode word denotes a directory (`S_IFDIR`).
fn is_dir_mode(mode: u32) -> bool {
    (mode & libc::S_IFMT) == libc::S_IFDIR
}

/// Body of a blob Create/Update scatter: the child's `InodeRaw` followed by the
/// 16-byte little-endian version ULID. `entry_from_scatter_body` (file.rs) lifts
/// the trailing ULID into `DirFileEntryRaw::blob_ulid`. A normal (non-blob)
/// scatter body is exactly `InodeRaw`, so the extra 16 bytes are unambiguous.
fn blob_scatter_body(inode_raw: &InodeRaw, blob_ulid: Ulid) -> Vec<u8> {
    let mut body = inode_raw.as_u8_slice().to_vec();
    body.extend_from_slice(&u128::from(blob_ulid).to_le_bytes());
    body
}

/// Newest blob version's ULID timestamp (ms since epoch) under `FILE/<uuid>/`,
/// or `None` if there are no `*.blob` objects. Used as the age gate for
/// reclaiming a blob-only prefix (no hyperfile inode to read a ctime from).
async fn newest_blob_ts_ms(client: &Client, layout: &HyperDirLayout, bucket: &str, file_uuid: &Uuid) -> Option<i64> {
    let prefix = layout.blob_prefix(file_uuid);
    let mut newest: Option<u64> = None;
    let mut stream = client.list_objects_v2().bucket(bucket).prefix(&prefix).into_paginator().send();
    while let Some(page) = stream.next().await {
        let page = match page { Ok(p) => p, Err(_) => return None };
        for obj in page.contents() {
            let Some(key) = obj.key() else { continue };
            let Some(stem) = key.rsplit('/').next().and_then(|f| f.strip_suffix(".blob")) else { continue };
            if let Ok(u) = Ulid::from_string(stem) {
                let ts = u.timestamp_ms();
                newest = Some(newest.map_or(ts, |n| n.max(ts)));
            }
        }
    }
    newest.map(|ts| ts as i64)
}

/// S3 key of one extended-attribute sidecar object under the child's prefix.
fn xattr_key(layout: &HyperDirLayout, uuid: &Uuid, is_dir: bool, name: &str) -> String {
    format!("{}{}", xattr_prefix(layout, uuid, is_dir), B64URL.encode(name))
}

/// LIST prefix covering a child's extended-attribute sidecar objects.
fn xattr_prefix(layout: &HyperDirLayout, uuid: &Uuid, is_dir: bool) -> String {
    let base = if is_dir { layout.dir_key(uuid) } else { layout.file_key(uuid) };
    format!("{}/_xattr/", base)
}

/// Maximum optimistic-concurrency retries for an nlink read-modify-write.
const NLINK_RETRIES: usize = 5;

/// Bounded attempts to claim a source's cross-directory rename commit point
/// when a concurrent rename of the same source holds it (see
/// `emit_rename_intent`).
const RENAME_CLAIM_RETRIES: usize = 8;

/// Read-modify-write a file's authoritative `i_nlink` by `delta`, returning
/// the new value. Uses a staging with no scatter interceptor (an nlink change
/// is not a directory-listing event) and retries on the inode's OCC conflict.
/// The count saturates at zero.
async fn adjust_nlink(client: &Client, child_uri: &str, delta: i64) -> Result<u64> {
    let staging = S3Staging::from(
        client,
        StagingConfig::new_s3_uri(child_uri, None),
        HyperFileRuntimeConfig::default(),
    ).await?;
    let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default();
    for _ in 0..NLINK_RETRIES {
        let mut raw: InodeRaw = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
        let od_state = staging.load_inode(raw.as_mut_u8_slice()).await?;
        let new_nlink = (raw.i_nlink as i64 + delta).max(0) as u64;
        raw.i_nlink = new_nlink;
        // POSIX: changing the link count updates ctime. (For a hard link this is
        // eager; for an unlink the decrement -- and so this bump -- happens when
        // compaction folds the delete, so the file's ctime is eventual there.)
        raw.i_ctime = now.as_secs();
        raw.i_ctime_nsec = now.subsec_nanos();
        match staging.flush_inode(raw.as_u8_slice(), &od_state, FlushInodeFlag::Update).await {
            Ok(_) => return Ok(new_nlink),
            Err(e) if e.kind() == ErrorKind::AlreadyExists => continue,
            Err(e) => return Err(e),
        }
    }
    Err(Error::new(ErrorKind::ResourceBusy, "adjust_nlink: too many OCC conflicts"))
}

/// Read a file's current authoritative `i_nlink`. Returns `None` if the child
/// inode no longer exists (already reclaimed).
async fn current_nlink(client: &Client, child_uri: &str) -> Result<Option<u64>> {
    match S3Staging::from(client, StagingConfig::new_s3_uri(child_uri, None), HyperFileRuntimeConfig::default()).await {
        Ok(staging) => {
            let mut raw: InodeRaw = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
            match staging.load_inode(raw.as_mut_u8_slice()).await {
                Ok(_) => Ok(Some(raw.i_nlink)),
                Err(e) if e.kind() == ErrorKind::NotFound => Ok(None), // reclaimed concurrently
                Err(e) => Err(e),
            }
        },
        Err(e) if e.kind() == ErrorKind::NotFound => Ok(None),
        Err(e) => Err(e),
    }
}

/// GET a single scatter / tombstone object, returning its body bytes.
async fn get_object_bytes(staging: &S3Staging, key: &str) -> Result<bytes::Bytes> {
    let res = staging.client
        .get_object()
        .bucket(&staging.bucket)
        .key(key)
        .send()
        .await
        .map_err(|e| Error::new(s3_error_kind(&e),
            format!("GetObject s3://{}/{}: {}", staging.bucket, key, e)))?;
    let bytes = res.body.collect().await
        .map_err(|e| Error::other(
            format!("collect body s3://{}/{}: {}", staging.bucket, key, e)))?
        .into_bytes();
    Ok(bytes)
}

/// Open the child staging at `child_uri` and call hyperfile's `unlink()` to
/// remove its inode + segments. A `NotFound` from the open path is treated
/// as "already cleaned": that means a previous partial GC removed the child
/// prefix but failed to delete the tombstone scatter, and the current call
/// can finish the job.
async fn reclaim_child_prefix(
    client: &Client,
    child_uri: &str,
    runtime_config: HyperFileRuntimeConfig,
) -> Result<()> {
    let child_config = StagingConfig::new_s3_uri(child_uri, None);
    match S3Staging::from(client, child_config, runtime_config).await {
        // A concurrent reclaim of the same prefix may delete the inode/segments
        // mid-flight; treat a NotFound from unlink as "already reclaimed" so
        // reclaim is idempotent under contention.
        Ok(child_staging) => match child_staging.unlink().await {
            Ok(()) => Ok(()),
            Err(e) if e.kind() == ErrorKind::NotFound => Ok(()),
            Err(e) => Err(e),
        },
        Err(e) if e.kind() == ErrorKind::NotFound => Ok(()),
        Err(e) => Err(e),
    }
}