hyperdir 0.7.0

use std::io::{Result, Error, ErrorKind};
use std::time::{Instant, Duration};
use std::sync::Arc;
#[cfg(feature = "wal")]
use std::sync::Weak;
#[cfg(feature = "wal")]
use std::pin::Pin;
use std::ffi::OsStr;
use std::ffi::CStr;
use std::collections::{HashMap, HashSet, BTreeMap};
use log::warn;
use uuid::Uuid;
use ulid::Ulid;
use tokio::sync::{
    Semaphore, OwnedSemaphorePermit,
    Mutex, OwnedMutexGuard,
};
use btree_ondisk::{BlockLoader, bmap::BMap, NodeValue, NullNodeCache};
use btree_ondisk::btree::BtreeNodeDirty;
use btree_ondisk::DEFAULT_CACHE_UNLIMITED;
use hyperfile::file::{HyperTrait, DirtyDataBlocks, FlushTiming};
use hyperfile::file::file::PendingParts;
#[cfg(feature = "wal")]
use hyperfile::Cno;
use hyperfile::{BlockIndex, BlockPtr, SegmentId, SegmentOffset, BMapUserData};
use hyperfile::meta_format::BlockPtrFormat;
use hyperfile::inode::{Inode, FlushInodeFlag};
use hyperfile::config::{HyperFileConfig, HyperFileMetaConfig};
use hyperfile::staging::{StagingIntercept, Staging, config::StagingConfig};
use hyperfile::segment::SegmentReadWrite;
use hyperfile::file::flags::HyperFileFlags;
use hyperfile::file::mode::HyperFileMode;
use hyperfile::ondisk::{BMapRawType, InodeRaw};
use super::ondisk::{DirFileEntryRaw, DEFAULT_NAME_LEN};
use super::{DirStaging, DirScatterInode, DirScatterInodeOp};
use super::parse_predelete_body;

pub const DIR_FILE_ENTRY_RAW_SIZE: usize = std::mem::size_of::<DirFileEntryRaw>();

/// A directory entry as seen through the parent directory.
///
/// Authority model:
/// - The parent directory is authoritative for the **namespace**: the
///   `name -> uuid` mapping and whether the name exists.
/// - The file's own inode (at its `FILE/<uuid>` / `DIR/<uuid>` prefix, and
///   embedded in each of its segments) is authoritative for **file
///   metadata**: mode, size, times, and `nlink`.
/// - The `inode` carried here is a *cached snapshot* taken from the latest
///   scatter the parent consolidated. It is an advisory readdir/stat
///   optimization and may lag the authoritative file inode. For a
///   hard-linked file in particular, the cached `nlink` is only advisory:
///   the count is shared across all of the file's names and only the file's
///   own inode tracks it correctly.
///
/// Consequence for consumers: after opening a file through this entry
/// (by `uuid`, optionally pinning the version via `inode`'s last cno, which
/// resolves to the segment whose embedded inode is authoritative), refresh
/// the in-memory `nlink` (and any other metadata that matters) from the
/// opened file's inode rather than trusting this cached copy.
#[derive(Debug)]
pub struct DirFileEntry {
    pub inode: Inode,
    pub uuid: Uuid,
    pub name: String,
    /// For a fully-inlined small file/symlink (no `FILE/<uuid>` object), its
    /// bytes carried in the dirent itself; `None` for a normal segmented file.
    pub inline: Option<Vec<u8>>,
    /// For a second-level inline ("blob") file, the current version ULID; its
    /// bytes live in `FILE/<uuid>/<blob_ulid>.blob`. `None` for in-inode inline
    /// or a normal segmented file. See `docs/second-level-inline.md`.
    pub blob_ulid: Option<Ulid>,
}

impl DirFileEntry {
    fn from_raw(raw: &DirFileEntryRaw) -> Self {
        let name = match CStr::from_bytes_until_nul(&raw.name) {
            Ok(cstr) => {
                match cstr.to_str() {
                    Ok(s) => { s.to_string() },
                    Err(e) => {
                        // invalid filename
                        warn!("decode file name error: {}", e);
                        String::new()
                    },
                }
            },
            Err(e) => {
                // invalid filename
                warn!("decode file name error: {}", e);
                String::new()
            },
        };

        Self {
            inode: Inode::from_raw(&raw.inode, None),
            uuid: Uuid::from_bytes(raw.uuid),
            name,
            inline: if raw.inode.is_inline() { Some(raw.inode.inline_data().to_vec()) } else { None },
            blob_ulid: raw.blob_ulid(),
        }
    }
}

pub(crate) type EntryNameHash = u64;

/// Statistics returned from [`HyperDirFile::compact`].
#[derive(Default, Debug, Clone)]
pub struct CompactStats {
    /// Number of scatter objects listed and processed in this round.
    pub scatters_processed: usize,
    /// Number of bmap entries inserted/updated from Create/Update scatters.
    pub entries_added: usize,
    /// Number of bmap entries removed from Delete scatters.
    pub entries_removed: usize,
    /// Number of Delete scatters that won their filename group and were
    /// retained in S3 as tombstones (subject to retention; reclaimed by
    /// [`HyperDir::fs_gc`]).
    pub tombstones_kept: usize,
    /// True when the scatter list was empty; flush + scatter cleanup were
    /// skipped and this call had no S3 side effects.
    pub no_op: bool,
    /// Child UUIDs whose Delete was actually folded out of the bmap this pass.
    /// The caller decrements each child file's nlink once; doing it here (the
    /// single lease-serialized fold point) makes nlink exactly-once even under
    /// concurrent/duplicate unlinks.
    pub removed_children: Vec<Uuid>,
    /// PreDelete markers (phase-1 unlink) folded out of the bmap this pass, as
    /// `(name, child_uuid, is_dir, retention_until_unix_ms)`. For each, the
    /// caller (`fs_compact`, phase 2) reads the child's authoritative inode,
    /// writes the real Delete tombstone with that retention, and decrements a
    /// file's nlink once -- the single lease-serialized point that keeps the
    /// count exactly-once and the unlink path cheap (no child read at unlink).
    pub predeleted: Vec<(String, Uuid, bool, i64)>,
    /// Identities displaced under a name this pass by a different identity (a
    /// concurrent/sequential same-name create that won last-writer-wins), as
    /// `(displaced_uuid, is_dir, marker_ulid)`. The caller (`fs_compact`,
    /// phase 2) reclaims each via `reclaim_displaced` -- reference-safe (a still
    /// hard-linked file is kept via a `.unref` marker) -- so a losing same-name
    /// create never leaves a nameless orphan. `marker_ulid` is the displacing
    /// (or losing) scatter's ULID, making the reclaim idempotent across passes.
    pub displaced: Vec<(Uuid, bool, Ulid)>,
}

/// Internal: the result of listing + classifying + body-fetching scatter
/// objects, ready to be applied to either an in-memory view (`read_dir`) or
/// the persisted bmap (`compact`).
struct ScatterChanges {
    /// All scatter object keys that were observed in the LIST. A successful
    /// compact will delete this set after the bmap flush commits, *except*
    /// for any keys that appear in `tombstone_keys`. `read_dir` ignores
    /// this field.
    all_keys: Vec<String>,
    /// Filename -> (entry, winning scatter ULID), for filenames whose latest
    /// scatter is Create or Update. Keyed by the full filename (not its hash)
    /// so two distinct names that share a CRC64 are both preserved within one
    /// batch. The ULID identifies the winning create for idempotent reclaim of
    /// any bmap entry it displaces.
    upserts: HashMap<String, (DirFileEntryRaw, Ulid)>,
    /// Dead identities whose latest scatter is Delete, as `(name, child_uuid)`.
    /// A `Vec` because identity-aware resolution can surface several dead
    /// identities for one name (rapid create/delete churn before a compact).
    /// Compaction removes each only if it is still the bmap entry (uuid-checked)
    /// and decrements that child's nlink at most once (the lease-serialized
    /// removal point).
    deletes: Vec<(String, Uuid)>,
    /// Dead identities whose latest scatter is PreDelete (phase-1 unlink), as
    /// `(name, child_uuid, is_dir, retention_until_unix_ms)`. read_dir treats
    /// the matching bmap identity as absent; compaction folds out the matching
    /// one and surfaces it to `fs_compact` (phase 2) for the real tombstone +
    /// nlink decrement.
    predeletes: Vec<(String, Uuid, bool, i64)>,
    /// Subset of `all_keys`: the latest Delete scatter per filename. These
    /// must NOT be deleted by `compact`; they remain in S3 as tombstones
    /// until `fs_gc` reclaims them after their retention expires.
    tombstone_keys: HashSet<String>,
    /// Losing concurrent same-name creates resolved this pass, as
    /// `(loser_uuid, is_dir, loser_ulid)`. Compaction reclaims each (the name
    /// belongs to the winning create) so no nameless orphan is left.
    superseded_displaced: Vec<(Uuid, bool, Ulid)>,
}

#[inline]
/// Rebuild a `DirFileEntryRaw` from a Create/Update scatter body. The body is
/// the child's `InodeRaw`; for a **blob** entry (second-level inline) it is
/// followed by the 16-byte little-endian blob version ULID (see
/// `docs/second-level-inline.md`), which we lift into `blob_ulid` so
/// resolve/read_dir/compaction all surface the blob pointer identically.
fn entry_from_scatter_body(uuid: &Uuid, filename: &str, body: &[u8]) -> DirFileEntryRaw {
    let isz = std::mem::size_of::<InodeRaw>();
    let inode_raw = InodeRaw::from_u8_slice(&body[..isz]);
    let mut raw = DirFileEntryRaw::from(
        &inode_raw,
        uuid.as_bytes(),
        <str as AsRef<OsStr>>::as_ref(filename).as_encoded_bytes(),
    );
    if body.len() >= isz + 16 {
        let mut b = [0u8; 16];
        b.copy_from_slice(&body[isz..isz + 16]);
        raw.blob_ulid = b;
    }
    raw
}

fn hash_filename(name: &str) -> EntryNameHash {
    let mut c = crc64fast::Digest::new();
    c.write(name.as_bytes());
    c.sum64()
}

#[inline]
fn hash_filename_bytes(name: &[u8]) -> EntryNameHash {
    let mut c = crc64fast::Digest::new();
    c.write(name);
    c.sum64()
}

impl NodeValue for DirFileEntryRaw {
    fn is_invalid(&self) -> bool {
        self.name == [0u8; DEFAULT_NAME_LEN + 1]
    }

    fn invalid_value() -> DirFileEntryRaw {
        DirFileEntryRaw::default()
    }

    fn is_valid_extern_assign(&self) -> bool {
        false
    }
}

pub struct HyperDirFile<'a, T: Send + Clone, L: BlockLoader<BlockPtr>> {
    bmap: BMap<'a, BlockIndex, DirFileEntryRaw, BlockPtr, L, NullNodeCache>,
    staging: T,
    bmap_ud: BMapUserData,
    config: HyperFileConfig,
    flags: HyperFileFlags,
    last_flush: Instant,
    sema: Arc<Semaphore>,
    flush_lock: Arc<Mutex<()>>,
    inode: Inode,
    flush_timing: FlushTiming,
    /// The checkpoint being accumulated out of partial segments, if any.
    ///
    /// A directory has no dirty *data* blocks — `get_data_blocks_dirty` always
    /// answers empty, since its contents live in the bmap's leaves — so nothing
    /// here should ever start accumulating. This mirrors hyperfile's own state
    /// rather than hard-coding that conclusion: the flush path reads and writes
    /// it unconditionally, so tracking it honestly costs three lines and does
    /// not depend on a reachability argument staying true.
    pending_parts: Option<PendingParts>,
}

impl<'a, T, L> HyperDirFile<'a, T, L>
    where
        T: Staging<L> + SegmentReadWrite + DirStaging + Send + Clone + 'static,
        L: BlockLoader<BlockPtr> + Clone,
{
    pub async fn new(staging: T, meta_block_loader: L, mut config: HyperFileConfig, flags: HyperFileFlags, mode: HyperFileMode) -> Result<Self>
    {
        // A directory's block pointers are `Flat` -- see the `bmap_ud` below,
        // which this crate has always hard-coded -- so the inode has to say the
        // same, because the two are read for different purposes and both must
        // agree. Pointer encoding follows the bmap's format; *object naming*
        // follows the inode's, via `Segment::object_of`.
        //
        // That was harmless until hyperfile 0.7.0: `Flat` and the old default
        // `MicroGroup` name a checkpoint's object identically (`<segid>`), the
        // micro-group field taking no part in addressing. `PartedSegment`, the
        // new default, names it `<segid>.<part>`. Left to disagree, a directory
        // writes its objects parted and then addresses them unparted, so every
        // read misses -- `read_dir` came back empty on a directory that had just
        // been compacted.
        //
        // Forcing `Flat` rather than adopting parts is the right way round:
        // parts exist so a container can stream dirty *data* out under memory
        // pressure, and a directory has no data blocks at all -- its contents
        // live in the bmap's leaves.
        config.meta.block_ptr_format = BlockPtrFormat::Flat;
        let meta_config = config.meta.clone();

        let bmap = BMap::<BlockIndex, DirFileEntryRaw, BlockPtr, L, NullNodeCache>::new(meta_config.root_size, meta_config.meta_block_size, meta_block_loader, NullNodeCache)?;

        let inode = Inode::default_dir()
            .with_mode(&mode)
            .with_meta_config(&meta_config);
        let bmap_ud = BMapUserData::new(BlockPtrFormat::Flat);
        bmap.set_userdata(bmap_ud.as_u32());

        let mut file = Self {
            bmap,
            staging,
            bmap_ud,
            config,
            flags,
            last_flush: Instant::now(),
            sema: Arc::new(Semaphore::new(1)),
            flush_lock: Arc::new(Mutex::new(())),
            inode,
            flush_timing: FlushTiming::default(),
            pending_parts: None,
        };
        // flush inode for hyper file new created
        file.flush_inode(FlushInodeFlag::Create).await?;
        Ok(file)
    }

    /// open a hyper file
    /// open by loading inode from staging,
    /// if inode is not found in staging, create hyper file from scratch
    pub async fn open(staging: T, meta_block_loader: L, config: HyperFileConfig, flags: HyperFileFlags) -> Result<Self>
    {
        Self::do_open(staging, meta_block_loader, config, flags, 0).await
    }

    /// open a hyper file with cno for read-only
    pub async fn open_cno(staging: T, meta_block_loader: L, config: HyperFileConfig, flags: HyperFileFlags, cno: u64) -> Result<Self>
    {
        if !flags.is_rdonly() {
            return Err(Error::new(ErrorKind::ReadOnlyFilesystem, "write access is not allowed for open specific cno"));
        }
        Self::do_open(staging, meta_block_loader, config, flags, cno).await
    }

    async fn do_open(staging: T, meta_block_loader: L, mut config: HyperFileConfig, flags: HyperFileFlags, cno: u64) -> Result<Self>
    {
        let mut raw_inode: InodeRaw = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
        let res_inode = if cno == 0 {
            staging.load_inode(raw_inode.as_mut_u8_slice()).await
        } else {
            staging.load_inode_from_segment(raw_inode.as_mut_u8_slice(), SegmentId::new_from_cno(cno)).await
        };
        let inode_state = match res_inode {
            Ok(od_state) => {
                /* if we load inode without error, we use inode as truth of metadata */
                od_state
            },
            Err(e) => {
                return Err(e);
            },
        };
        // get back meta config from inode raw
        let meta_config = HyperFileMetaConfig::from_u32(raw_inode.i_meta_config);
        // Pass the inode field directly: `BMap::read` peeks the root header via
        // `BtreeNode::from_slice_ref`, which REQUIRES an 8-byte-aligned buffer
        // (else `InvalidInput`). `i_bmap` sits at an 8-aligned offset inside
        // `repr(C, align(8))` InodeRaw, whereas a `let b = raw_inode.i_bmap`
        // copy is a bare `[u8; N]` with alignment 1 -- alignment only by luck.
        // (`BMap::read` copies the bytes into its own AlignedBuffer, so no
        // borrow of `raw_inode` is retained.) See the alignment guard in tests.
        let bmap = BMap::<BlockIndex, DirFileEntryRaw, BlockPtr, L, NullNodeCache>::read(&raw_inode.i_bmap, meta_config.meta_block_size, meta_block_loader, NullNodeCache)?;
        let bmap_ud = BMapUserData::from_u32(bmap.get_userdata());

        let permits = if flags.is_rdonly() {
            Semaphore::MAX_PERMITS
        } else {
            1
        };

        // overwrite the default meta config with the one we get from inode
        config.meta = meta_config;

        let mut file = Self {
            bmap,
            staging,
            bmap_ud,
            config,
            flags,
            last_flush: Instant::now(),
            sema: Arc::new(Semaphore::new(permits)),
            flush_lock: Arc::new(Mutex::new(())),
            inode: Inode::from_raw(&raw_inode, inode_state),
            flush_timing: FlushTiming::default(),
            pending_parts: None,
        };
        // refresh bmap if need to do recovery
        let _ = file.refresh_bmap().await?;
        Ok(file)
    }

    pub async fn release(&mut self) -> Result<SegmentId> {
        self.flush().await
    }

    pub fn stat(&self) -> libc::stat {
        // TODO: set dev and rdev here
        let dev = 0;
        let rdev = 0;
        self.inode.to_stat(dev, rdev)
    }

    // fast stat by read inode without open file
    pub async fn stat_fast(staging: T) -> Result<libc::stat> {
        let mut raw_inode: InodeRaw = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
        staging.load_inode(raw_inode.as_mut_u8_slice()).await?;
        let inode = Inode::from_raw(&raw_inode, None);
        Ok(inode.to_stat(0, 0))
    }

    // fast update stat by load inode and flush inode
    pub async fn update_stat_fast(staging: T, stat: &libc::stat) -> Result<libc::stat> {
        let mut raw_inode: InodeRaw = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
        let od_state = staging.load_inode(raw_inode.as_mut_u8_slice()).await?;
        let mut inode = Inode::from_raw(&raw_inode, od_state);
        inode.update_stat(stat);
        let raw = inode.to_raw(raw_inode.i_bmap);
        let od_state = inode.get_ondisk_state();
        let _ = staging.flush_inode(raw.as_u8_slice(), od_state, FlushInodeFlag::Update).await?;
        Ok(inode.to_stat(stat.st_dev, stat.st_rdev))
    }

    pub async fn update_stat(&mut self, stat: &libc::stat) -> Result<libc::stat> {
        let stat = self.inode.update_stat(stat);
        let _ = self.flush().await?;
        Ok(stat)
    }

    pub async fn flush_inode(&mut self, flag: FlushInodeFlag) -> Result<()> {
        // TODO update necessary inode fields
        let mut b: BMapRawType = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
        b.copy_from_slice(self.bmap.as_slice());
        let raw_inode = self.inode.to_raw(b);
        let od_state = self.staging.flush_inode(raw_inode.as_u8_slice(), self.inode.get_ondisk_state(), flag).await?;
        self.inode.clear_attr_dirty();
        self.inode.set_ondisk_state(od_state);
        self.inode.set_last_ondisk_cno(self.inode.get_last_cno());
        Ok(())
    }

    pub fn staging_config(&self) -> &StagingConfig {
        &self.config.staging
    }

    pub fn staging_interceptor(&mut self, i: impl StagingIntercept<T> + 'static) {
        self.staging.interceptor(i);
    }

    pub fn flags(&self) -> & HyperFileFlags {
        &self.flags
    }

    /// Look up a single directory entry by name.
    ///
    /// Resolves CRC64 collisions by open-addressing: probes from the name's
    /// home slot, comparing the full name stored in each entry until a match
    /// or an empty slot (NotFound) is found.
    pub async fn read_entry(&self, name: &str) -> Result<DirFileEntry> {
        let home = hash_filename(name);
        match self.probe_lookup_latest(name.as_bytes(), home).await? {
            Some((_, entry)) => Ok(DirFileEntry::from_raw(&entry)),
            None => Err(Error::new(ErrorKind::NotFound, format!("entry not found: {}", name))),
        }
    }

    /// Scatter-aware single-name resolve (the cheap lookup path).
    ///
    /// Lists only this name's scatter objects (`!/{base64(name)}/`, 0..N).
    /// If any exist, the latest wins (a Create/Update means present with that
    /// inode; a Delete means absent), exactly as `read_dir` would merge them.
    /// Only when the name has no pending scatter do we consult the compacted
    /// B-tree via a cheap point lookup. This avoids `read_dir`'s full B-tree
    /// walk and whole-`!/` fetch for the common (compacted) case, while never
    /// trusting the B-tree for a name that still has pending scatter.
    pub async fn resolve_entry(&self, name: &str) -> Result<Option<DirFileEntry>> {
        Self::resolve_fast(self.staging.clone(), self.bmap_get_block_loader(), name).await
    }

    /// Handle-less single-name resolve: the real logic behind `resolve_entry`,
    /// usable without first opening a `HyperDirFile`. Takes a `staging` (for
    /// the scatter LIST / body GET) and a block `loader` (for the fallback
    /// bmap), so the caller pays at most one inode GET (only on the fallback,
    /// and zero when the name still has pending scatter) instead of an
    /// open-the-handle inode GET plus the fallback's.
    pub async fn resolve_fast(staging: T, loader: L, name: &str) -> Result<Option<DirFileEntry>> {
        let scatters = staging.list_scatter_inodes_for_name(name).await?;
        // Dead identities seen in scatter (uuids). If the bmap fallback resolves
        // the name to one of these, the name is absent; if it resolves to a
        // DIFFERENT identity (a live entry already folded in), it is present.
        let mut dead_uuids: Vec<[u8; 16]> = Vec::new();
        if !scatters.is_empty() {
            // Identity-aware: a Delete/PreDelete for one identity must not hide a
            // same-name Create of another identity.
            let nv = DirScatterInode::resolve_name_view(scatters);
            if let Some(live) = nv.live {
                let fetched = staging.collect_scatter_inodes(vec![live]).await?;
                if let Some((s, body)) = fetched.into_iter().next() {
                    let raw = entry_from_scatter_body(&s.uuid, &s.filename, &body);
                    return Ok(Some(DirFileEntry::from_raw(&raw)));
                }
                // The live body was compacted away under us; the entry is now in
                // the B-tree (as the live identity), so fall through to the point
                // lookup with no dead filter.
            } else {
                // No live identity in scatter: only a bmap entry that is NOT one
                // of these dead identities counts as present.
                dead_uuids = nv.dead.iter().map(|d| *d.uuid.as_bytes()).collect();
            }
        }
        // Fallback: read the latest inode, build its bmap, and point-look-up.
        let mut raw_inode: InodeRaw = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
        staging.load_inode(raw_inode.as_mut_u8_slice()).await?;
        let meta_config = HyperFileMetaConfig::from_u32(raw_inode.i_meta_config);
        let bmap = BMap::<BlockIndex, DirFileEntryRaw, BlockPtr, L, NullNodeCache>::read(
            &raw_inode.i_bmap, meta_config.meta_block_size, loader, NullNodeCache)?;
        let home = hash_filename(name);
        match Self::probe_lookup_in(&bmap, name.as_bytes(), home).await? {
            Some((_, e)) if dead_uuids.iter().any(|u| u == &e.uuid) => Ok(None),
            Some((_, e)) => Ok(Some(DirFileEntry::from_raw(&e))),
            None => Ok(None),
        }
    }

    /// Look up the raw on-disk entry (inode + uuid + name) by name. Used by
    /// cross-directory rename to move an entry verbatim.
    pub async fn read_entry_raw(&self, name: &str) -> Result<DirFileEntryRaw> {
        let home = hash_filename(name);
        match self.probe_lookup(name.as_bytes(), home).await? {
            Some((_, entry)) => Ok(entry),
            None => Err(Error::new(ErrorKind::NotFound, format!("entry not found: {}", name))),
        }
    }

    /// Insert/update an entry under `name` and flush. Idempotent. Returns the
    /// entry that was overwritten in place under the same name (if any), so a
    /// caller can detect and reclaim a displaced different identity.
    pub async fn insert_entry(&mut self, name: &str, entry: DirFileEntryRaw) -> Result<Option<DirFileEntryRaw>> {
        // Mutate from the LATEST committed bmap, not this handle's open-time
        // snapshot: a concurrent compactor may have folded other entries into
        // a newer inode, and flushing our stale bmap would drop them.
        self.reload_to_latest().await?;
        let displaced = self.probe_upsert(name.as_bytes(), entry).await?;
        self.flush().await?;
        Ok(displaced)
    }

    /// Like [`insert_entry`] but refuses to overwrite an existing name:
    /// returns `Ok(false)` if `name` already exists in the latest committed
    /// bmap. A lost OCC flush race still surfaces as `Err(AlreadyExists)` for
    /// the caller to retry. Used by hard link so a concurrent same-name link
    /// cannot clobber the winner (which would also leak the loser's nlink) --
    /// `insert_entry` reloads-then-upserts, so without this check two racing
    /// inserts would both "succeed", the later overwriting the earlier.
    pub async fn insert_entry_excl(&mut self, name: &str, entry: DirFileEntryRaw) -> Result<bool> {
        self.reload_to_latest().await?;
        let home = hash_filename(name);
        if Self::probe_lookup_in(&self.bmap, name.as_bytes(), home).await?.is_some() {
            return Ok(false); // already present: don't overwrite
        }
        self.probe_upsert(name.as_bytes(), entry).await?;
        self.flush().await?;
        Ok(true)
    }

    /// Remove the entry named `name` and flush. Returns false if absent.
    pub async fn remove_entry(&mut self, name: &str) -> Result<bool> {
        self.reload_to_latest().await?;
        let removed = self.probe_delete(name.as_bytes()).await?;
        self.flush().await?;
        Ok(removed)
    }

    /// Pure-read directory enumeration.
    ///
    /// Lists outstanding scatter objects, walks the persisted bmap, and merges
    /// the two views in memory keyed by filename. Does not modify any S3
    /// state, so it is safe to call concurrently from multiple clients.
    ///
    /// The returned snapshot reflects "the latest scatter applied on top of
    /// the bmap as of the LIST/GET round-trips". Two consecutive calls may
    /// differ if other writers commit between them. Outstanding scatter
    /// objects are *not* deleted here; that is the job of [`compact`].
    pub async fn read_dir(&self) -> Result<Vec<DirFileEntry>> {
        // List scatter FIRST, then read the LATEST persisted bmap. Compact
        // always flushes the bmap before deleting the scatter objects it
        // folded, so reading the bmap *after* the scatter LIST guarantees any
        // scatter that has since vanished (deleted by a concurrent compactor)
        // is already reflected in this freshly-read bmap. We deliberately do
        // NOT use this handle's open-time bmap snapshot: a compact since open
        // could have advanced the persisted bmap and removed the very scatters
        // that carried those changes, which would make us miss/stale entries.
        let changes = self.collect_scatter_changes().await?;
        let latest_bmap = self.load_latest_bmap().await?;

        // start from the latest persisted bmap, keyed by filename
        let mut view: HashMap<String, DirFileEntryRaw> = HashMap::new();
        for entry in Self::walk_bmap(&latest_bmap).await? {
            view.insert(String::from_utf8_lossy(entry.name_bytes()).into_owned(), entry);
        }

        // overlay scatter changes (also keyed by filename); a live identity wins
        // over the bmap snapshot.
        let mut live_names: HashSet<String> = HashSet::new();
        for (name, (entry, _ulid)) in changes.upserts {
            live_names.insert(name.clone());
            view.insert(name, entry);
        }
        // A dead identity (Delete/PreDelete) retires a name only if no live
        // identity holds it AND the surviving bmap entry IS that same dead
        // identity -- so a (possibly late) tombstone for an old identity never
        // hides a same-name entry that is now a different, live identity.
        let mut dead: HashMap<String, Vec<[u8; 16]>> = HashMap::new();
        for (name, uuid) in &changes.deletes {
            dead.entry(name.clone()).or_default().push(*uuid.as_bytes());
        }
        for (name, uuid, _, _) in &changes.predeletes {
            dead.entry(name.clone()).or_default().push(*uuid.as_bytes());
        }
        for (name, uuids) in dead {
            if live_names.contains(&name) { continue; }
            if view.get(&name).map(|e| uuids.iter().any(|u| u == &e.uuid)).unwrap_or(false) {
                view.remove(&name);
            }
        }

        // Deterministic order, by name. The merge above is a HashMap, whose
        // iteration order is arbitrary AND changes as its contents change — so two
        // successive read_dir calls could return the same entries in a different
        // order. Callers that paginate by index (hyperfs's FUSE readdir hands the
        // kernel an index cookie, and the kernel always calls again to detect EOF)
        // would then re-send some entries and skip others: an `ls` of a directory
        // could show a name twice and omit another, even though membership never
        // changed. Sorting by name makes the cursor stable, since compaction
        // reorders entries but never renames them.
        let mut out: Vec<DirFileEntry> = view
            .into_values()
            .map(|raw| DirFileEntry::from_raw(&raw))
            .collect();
        out.sort_by(|a, b| a.name.cmp(&b.name));
        Ok(out)
    }

    /// Consolidate outstanding scatter objects into the persisted bmap.
    ///
    /// Acquires the directory's compactor leader lease before doing any work
    /// (see [`DirStaging::acquire_compact_lease`]). Concurrent callers on the
    /// same directory will see `Err(ErrorKind::ResourceBusy)` returned fast,
    /// rather than racing through the consolidate-and-flush path; correctness
    /// also relies on hyperfile's inode-flush OCC, but the lease avoids the
    /// duplicated I/O that OCC alone cannot.
    ///
    /// On a clean run the lease is released as soon as the consolidation
    /// commits; on any error or panic, the lease will be reclaimed by the
    /// next caller after [`DEFAULT_COMPACT_LEASE_TTL_MS`] passes.
    ///
    /// If the scatter LIST is empty this method is a no-op (no flush, no
    /// scatter delete) and returns `CompactStats { no_op: true, .. }`.
    pub async fn compact(&mut self) -> Result<CompactStats> {
        let lease = self.staging
            .acquire_compact_lease(crate::DEFAULT_COMPACT_LEASE_TTL_MS)
            .await?;
        let result = self.compact_inner(false).await;
        // Always best-effort release, regardless of compact_inner's outcome.
        // release_compact_lease itself only propagates errors that are not
        // expected as a side effect of the lease's natural lifecycle.
        let _ = self.staging.release_compact_lease(lease).await;
        result.map(|(stats, _)| stats)
    }

    /// [`compact_then_rename`](Self::compact_then_rename) under the compactor
    /// lease, mirroring [`compact`]'s lease handling.
    pub async fn compact_and_rename(&mut self, old_name: &str, new_name: &str) -> Result<CompactStats> {
        let lease = self.staging
            .acquire_compact_lease(crate::DEFAULT_COMPACT_LEASE_TTL_MS)
            .await?;
        let result = self.compact_then_rename(old_name, new_name).await;
        let _ = self.staging.release_compact_lease(lease).await;
        result
    }

    /// Body of [`compact`], assuming the caller has already acquired the
    /// compactor lease for this directory.
    ///
    /// With `defer_commit` the in-memory fold is left **uncommitted**: no flush
    /// and no scatter cleanup. The returned `to_strip` keys must then be removed
    /// by the caller *after* it flushes, which lets a caller that is going to
    /// mutate the bmap anyway (see
    /// [`compact_then_rename`](Self::compact_then_rename)) commit the fold and
    /// its own change in a single flush rather than two.
    async fn compact_inner(&mut self, defer_commit: bool) -> Result<(CompactStats, Vec<String>)> {
        // Fold into (and leave the handle on) the LATEST committed bmap, not
        // this handle's open-time snapshot. This matters both for correctness
        // of the fold under a concurrent compactor and because callers reuse
        // this handle right after (e.g. rename compacts then renames on the
        // same handle): a no-op compact must still leave the bmap current so
        // the following lookup doesn't miss an entry another compactor folded.
        self.reload_to_latest().await?;
        let changes = self.collect_scatter_changes().await?;
        if changes.all_keys.is_empty() {
            return Ok((CompactStats { no_op: true, ..Default::default() }, Vec::new()));
        }

        let scatters_processed = changes.all_keys.len();
        let tombstones_kept = changes.tombstone_keys.len();

        // Identities displaced under a name this pass, reclaimed in phase 2 so a
        // losing same-name create never orphans. Seed with the losing concurrent
        // creates resolved from scatter; add bmap entries overwritten below.
        let mut displaced: Vec<(Uuid, bool, Ulid)> = changes.superseded_displaced;

        let mut entries_added = 0;
        for (name, (entry, win_ulid)) in &changes.upserts {
            if let Some(old) = self.probe_upsert(name.as_bytes(), *entry).await? {
                // A different identity held this name (sequential create-over an
                // existing name, e.g. across mounts): the winner takes the name,
                // the old identity is displaced and must be reclaimed.
                if old.uuid != entry.uuid {
                    displaced.push((Uuid::from_bytes(old.uuid), old.is_dir(), *win_ulid));
                }
            }
            entries_added += 1;
        }

        let mut entries_removed = 0;
        let mut removed_children: Vec<Uuid> = Vec::new();
        for (name, child_uuid) in &changes.deletes {
            // uuid-checked: only retire the entry if it is STILL this identity.
            // After upserts above, a same-name live identity already holds the
            // slot, so a dead identity's tombstone is a no-op here (its storage
            // is reclaimed by fs_gc/orphan GC, not by clobbering the live entry).
            if self.probe_delete_if_uuid(name.as_bytes(), *child_uuid).await? {
                entries_removed += 1;
                // The name actually transitioned present -> absent here; the
                // caller decrements this child's nlink exactly once. A later
                // compact sees the (kept) tombstone again but probe_delete
                // returns false, so it is not double-counted.
                removed_children.push(*child_uuid);
            }
            // A tombstone for a name that was never persisted (created and
            // deleted before any compact ran, or already consolidated, or now
            // held by a different identity) just returns false; idempotent.
        }

        // Phase-1 PreDelete markers: fold the name out of the bmap and surface
        // (name, uuid, is_dir, retention) to fs_compact (phase 2), which reads
        // the child inode, writes the real Delete tombstone, and decrements
        // nlink. The PreDelete object itself is not a tombstone, so it is
        // stripped below with the other consolidated scatters.
        let mut predeleted: Vec<(String, Uuid, bool, i64)> = Vec::new();
        for (name, child_uuid, is_dir, retention) in &changes.predeletes {
            // uuid-checked, as for deletes: only fold out (and surface to
            // phase 2) when this identity is still the entry. If a same-name
            // live identity took the slot, the PreDelete'd identity is left to
            // orphan GC (safe -- a leak, never a wrongful delete of the new file).
            if self.probe_delete_if_uuid(name.as_bytes(), *child_uuid).await? {
                entries_removed += 1;
                predeleted.push((name.clone(), *child_uuid, *is_dir, *retention));
            }
        }

        // hyperfile's flush handles OCC retry per FlushConflictPolicy.
        // Deferred: the caller flushes (committing its own bmap edits together
        // with this fold) and strips the scatter afterwards.
        // Best-effort scatter cleanup. Keep tombstones (latest Delete scatter
        // per filename) so fs_gc can find and process them after retention;
        // strip everything else: stale scatters of any op, plus consolidated
        // Create/Update winners. Re-application on the bmap side is
        // idempotent if we ever see a tombstone twice.
        let to_delete: Vec<String> = changes.all_keys.into_iter()
            .filter(|k| !changes.tombstone_keys.contains(k))
            .collect();
        let mut to_strip = Vec::new();
        if defer_commit {
            to_strip = to_delete;
        } else {
            self.flush().await?;
            if !to_delete.is_empty() {
                self.staging.remove_scatter_inodes(to_delete).await?;
            }
        }

        Ok((CompactStats {
            scatters_processed,
            entries_added,
            entries_removed,
            tombstones_kept,
            no_op: false,
            removed_children,
            predeleted,
            displaced,
        }, to_strip))
    }

    /// Fold this directory's pending scatter **and** rename `old_name` to
    /// `new_name`, committed by a **single** flush.
    ///
    /// A same-directory rename has to fold first, because `rename_within` works
    /// on the B-tree and cannot see pending scatter. Doing that as
    /// `compact()` + `rename_within()` costs two commits — each writing a new
    /// segment plus an OCC inode PUT — for what is one logical change; the fold
    /// and the rename are applied to the same in-memory bmap here and committed
    /// once. It also removes the intermediate state where the fold is durable but
    /// the rename is not.
    ///
    /// The caller must hold the compactor lease (as [`compact`] does) and must
    /// run compaction's phase 2 on the returned stats.
    pub async fn compact_then_rename(&mut self, old_name: &str, new_name: &str) -> Result<CompactStats> {
        let (stats, to_strip) = self.compact_inner(true).await?;
        let renamed = self.rename_within_staged(old_name, new_name).await?;
        if renamed || !stats.no_op {
            self.flush().await?;
        }
        if !to_strip.is_empty() {
            self.staging.remove_scatter_inodes(to_strip).await?;
        }
        Ok(stats)
    }

    /// List scatter, classify by op, fetch bodies for Create/Update.
    ///
    /// Pure read; both `read_dir` and `compact` start from this.
    async fn collect_scatter_changes(&self) -> Result<ScatterChanges> {
        let scatters = self.staging.list_scatter_inodes().await?;
        let all_keys: Vec<String> = scatters.iter().map(|s| s.key.clone()).collect();

        // Identity-aware: resolve each name into its live entry + dead
        // identities, so a Delete/PreDelete for one identity never shadows or
        // strips a same-name Create of a different identity.
        let views = DirScatterInode::resolve_dir_view(scatters);

        let mut to_fetch = Vec::new();        // live Create/Update winners
        let mut to_fetch_pre = Vec::new();    // dead PreDelete markers
        let mut to_fetch_sup = Vec::new();    // losing concurrent live creates
        let mut deletes: Vec<(String, Uuid)> = Vec::new();
        let mut tombstone_keys: HashSet<String> = HashSet::new();
        for (_name, nv) in views {
            if let Some(live) = nv.live {
                // live.op is Create/Update by construction.
                to_fetch.push(live);
            }
            for d in nv.dead {
                match d.op {
                    DirScatterInodeOp::Delete => {
                        // Keep the dead identity's tombstone so fs_gc can reclaim
                        // its storage after retention; the fold removes it from
                        // the bmap only if it is still the entry there.
                        tombstone_keys.insert(d.key.clone());
                        deletes.push((d.filename, d.uuid));
                    },
                    DirScatterInodeOp::PreDelete => to_fetch_pre.push(d),
                    op => warn!("unexpected dead scatter op {:?}: {}", op, d.key),
                }
            }
            // Losing concurrent same-name creates: reclaim them (see below).
            to_fetch_sup.extend(nv.superseded);
        }

        let fetched = self.staging.collect_scatter_inodes(to_fetch).await?;
        let mut upserts: HashMap<String, (DirFileEntryRaw, Ulid)> = HashMap::new();
        for (s, body) in fetched {
            let entry = entry_from_scatter_body(&s.uuid, &s.filename, &body);
            upserts.insert(s.filename, (entry, s.ulid));
        }

        let mut predeletes: Vec<(String, Uuid, bool, i64)> = Vec::new();
        for (s, body) in self.staging.collect_scatter_inodes(to_fetch_pre).await? {
            let (is_dir, retention) = parse_predelete_body(&body);
            predeletes.push((s.filename, s.uuid, is_dir, retention));
        }

        // Losing live creates: fetch each loser's inode only to learn is_dir
        // (file vs directory prefix), so fs_compact can reclaim the right
        // storage. Cheap because this list is empty unless a same-name create
        // race actually happened.
        let mut superseded_displaced: Vec<(Uuid, bool, Ulid)> = Vec::new();
        for (s, body) in self.staging.collect_scatter_inodes(to_fetch_sup).await? {
            let inode_raw = InodeRaw::from_u8_slice(&body[..std::mem::size_of::<InodeRaw>()]);
            let is_dir = (inode_raw.i_mode & 0o170000) == 0o040000; // S_IFMT / S_IFDIR
            superseded_displaced.push((s.uuid, is_dir, s.ulid));
        }

        Ok(ScatterChanges { all_keys, upserts, deletes, predeletes, tombstone_keys, superseded_displaced })
    }

    /// Walk the persisted bmap read-only, returning all valid entries.
    /// Skips dummy values inserted by hyperfile as bmap placeholders.
    /// Build a fresh bmap from the LATEST persisted inode (re-read from
    /// staging), independent of this handle's open-time snapshot. Used by
    /// `read_dir` so it never walks a bmap a concurrent compactor has
    /// superseded.
    async fn load_latest_bmap(&self) -> Result<BMap<'a, BlockIndex, DirFileEntryRaw, BlockPtr, L, NullNodeCache>> {
        let mut raw_inode: InodeRaw = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
        let _ = self.staging.load_inode(raw_inode.as_mut_u8_slice()).await?;
        let meta_config = HyperFileMetaConfig::from_u32(raw_inode.i_meta_config);
        // `&raw_inode.i_bmap` (8-aligned field), not a bare `[u8; N]` copy: see
        // the note in `open` -- `BMap::read` rejects a misaligned buffer.
        BMap::<BlockIndex, DirFileEntryRaw, BlockPtr, L, NullNodeCache>::read(
            &raw_inode.i_bmap, meta_config.meta_block_size, self.bmap_get_block_loader(), NullNodeCache)
    }

    /// Replace this handle's in-memory inode + bmap with the LATEST persisted
    /// state, discarding the open-time snapshot. A write op that looks up an
    /// entry (rename, link) must do this first: a concurrent compactor may
    /// have folded the entry into a newer inode and deleted its scatter, so
    /// the open-time bmap would miss it. Only safe when the handle has no
    /// un-flushed changes (compaction calls it before folding).
    async fn reload_to_latest(&mut self) -> Result<()> {
        let mut raw_inode: InodeRaw = unsafe { std::mem::MaybeUninit::zeroed().assume_init() };
        let od_state = self.staging.load_inode(raw_inode.as_mut_u8_slice()).await?;
        let meta_config = HyperFileMetaConfig::from_u32(raw_inode.i_meta_config);
        let bmap = BMap::<BlockIndex, DirFileEntryRaw, BlockPtr, L, NullNodeCache>::read(
            &raw_inode.i_bmap, meta_config.meta_block_size, self.bmap_get_block_loader(), NullNodeCache)?;
        self.bmap = bmap;
        self.config.meta = meta_config;
        self.inode = Inode::from_raw(&raw_inode, od_state);
        Ok(())
    }

    /// Enumerate the live (non-dummy) entries of `bmap`.
    async fn walk_bmap(bmap: &BMap<'_, BlockIndex, DirFileEntryRaw, BlockPtr, L, NullNodeCache>) -> Result<Vec<DirFileEntryRaw>> {
        let mut out: Vec<DirFileEntryRaw> = Vec::new();
        let opt_last_key = match bmap.last_key().await {
            Ok(k) => Some(k),
            Err(e) if e.kind() == ErrorKind::NotFound => None,
            Err(e) => return Err(e),
        };
        if let Some(last_key) = opt_last_key {
            let mut n: BlockIndex = 0;
            loop {
                let key = match bmap.seek_key(&n).await {
                    Ok(k) => k,
                    Err(e) if e.kind() == ErrorKind::NotFound => break,
                    Err(e) => return Err(e),
                };
                if key > last_key {
                    break;
                }
                let val = bmap.lookup(&key).await?;
                if !val.is_dummy() {
                    out.push(val);
                }
                if key == BlockIndex::MAX {
                    break;
                }
                n = key + 1;
            }
        }
        Ok(out)
    }

    /// Probe from `home` for the slot holding `name`, returning its
    /// (slot, entry) or `None` if absent. Linear probing: a slot whose entry
    /// is a dummy or a colliding (different) name is skipped; the first
    /// absent slot (NotFound) terminates the chain.
    ///
    /// Probing does not wrap; with CRC64 spread uniformly over u64 and tiny
    /// clusters in practice, a chain reaching `u64::MAX` does not occur.
    async fn probe_lookup(&self, name: &[u8], home: BlockIndex) -> Result<Option<(BlockIndex, DirFileEntryRaw)>> {
        Self::probe_lookup_in(&self.bmap, name, home).await
    }

    /// Like [`probe_lookup`] but against the LATEST persisted bmap, re-read
    /// from staging rather than this handle's open-time snapshot. Read-path
    /// point lookups (`read_entry`, `resolve_entry`'s fallback) must use this:
    /// a concurrent compactor can fold a name into a new inode and delete its
    /// scatter, which would make an open-time-snapshot lookup miss it.
    async fn probe_lookup_latest(&self, name: &[u8], home: BlockIndex) -> Result<Option<(BlockIndex, DirFileEntryRaw)>> {
        let bmap = self.load_latest_bmap().await?;
        Self::probe_lookup_in(&bmap, name, home).await
    }

    async fn probe_lookup_in(bmap: &BMap<'_, BlockIndex, DirFileEntryRaw, BlockPtr, L, NullNodeCache>, name: &[u8], home: BlockIndex)
        -> Result<Option<(BlockIndex, DirFileEntryRaw)>>
    {
        let mut k = home;
        loop {
            match bmap.lookup(&k).await {
                Ok(entry) => {
                    if !entry.is_dummy() && entry.name_eq(name) {
                        return Ok(Some((k, entry)));
                    }
                },
                Err(e) if e.kind() == ErrorKind::NotFound => return Ok(None),
                Err(e) => return Err(e),
            }
            k = match k.checked_add(1) {
                Some(next) => next,
                None => return Ok(None),
            };
        }
    }

    /// Insert or update `entry` for `name`: write at the existing slot if the
    /// name is already present, else at the first empty slot in the probe
    /// chain from the name's home slot. Returns the entry that was overwritten
    /// in place (same name), if any -- the caller compares its uuid to detect a
    /// displaced identity that must be reclaimed.
    async fn probe_upsert(&mut self, name: &[u8], entry: DirFileEntryRaw) -> Result<Option<DirFileEntryRaw>> {
        let home = hash_filename_bytes(name);
        let mut k = home;
        let mut displaced: Option<DirFileEntryRaw> = None;
        loop {
            match self.bmap.lookup(&k).await {
                Ok(existing) => {
                    if !existing.is_dummy() && existing.name_eq(name) {
                        displaced = Some(existing);
                        break; // update in place at k
                    }
                },
                Err(e) if e.kind() == ErrorKind::NotFound => break, // empty slot at k
                Err(e) => return Err(e),
            }
            k = k.checked_add(1)
                .ok_or_else(|| Error::other("probe chain exhausted u64 key space"))?;
        }
        let _ = self.bmap.insert(k, entry).await?;
        Ok(displaced)
    }

    /// Delete `name`, repairing the probe chain with backward-shift so no gap
    /// breaks the lookup of entries that probed past the removed slot.
    /// Like [`probe_delete`], but only removes the entry if it currently
    /// resolves to `expected`. A tombstone/predelete for identity A must not
    /// delete a same-name entry that is now a DIFFERENT identity B (e.g. a file
    /// recreated under the same name while a late two-phase Delete for the old
    /// identity is still in flight). Returns true iff an entry was removed.
    async fn probe_delete_if_uuid(&mut self, name: &[u8], expected: Uuid) -> Result<bool> {
        let home = hash_filename_bytes(name);
        match self.probe_lookup(name, home).await? {
            Some((_, e)) if e.uuid == *expected.as_bytes() => self.probe_delete(name).await,
            _ => Ok(false),
        }
    }

    /// Returns false if `name` was not present.
    async fn probe_delete(&mut self, name: &[u8]) -> Result<bool> {
        let home = hash_filename_bytes(name);
        let slot = match self.probe_lookup(name, home).await? {
            Some((k, _)) => k,
            None => return Ok(false),
        };

        // Backward-shift: scan forward from the hole; move any following entry
        // whose home slot is at or before the hole into the hole.
        let mut hole = slot;
        let mut j = match slot.checked_add(1) { Some(v) => v, None => slot };
        while j != slot {
            let entry_j = match self.bmap.lookup(&j).await {
                Ok(e) => e,
                Err(e) if e.kind() == ErrorKind::NotFound => break, // cluster ended
                Err(e) => return Err(e),
            };
            // Dummies don't occur for directory bmaps in practice; treat one
            // as home == j so it never shifts.
            let home_j = if entry_j.is_dummy() { j } else { hash_filename_bytes(entry_j.name_bytes()) };
            if home_j <= hole {
                let _ = self.bmap.insert(hole, entry_j).await?;
                hole = j;
            }
            j = match j.checked_add(1) { Some(v) => v, None => break };
        }
        self.bmap.delete(&hole).await?;
        Ok(true)
    }

    /// Rename an entry within this directory.
    ///
    /// Same-directory rename is cheap under UUID identity: the child keeps its
    /// UUID and underlying storage; only the parent's name->entry mapping
    /// changes. Both bmap edits (remove old, insert new) are staged in memory
    /// and committed by a single inode flush, so the rename is atomic against
    /// concurrent writers via hyperfile's inode OCC.
    ///
    /// Source must exist (`NotFound`). If the destination already exists its
    /// entry is overwritten in place (the btree slot is replaced, so no
    /// tombstone is written and reads never see the name vanish); the caller
    /// is responsible for reclaiming the displaced child's storage. `old_name
    /// == new_name` is a no-op.
    pub async fn rename_within(&mut self, old_name: &str, new_name: &str) -> Result<()> {
        if !self.rename_within_staged(old_name, new_name).await? {
            return Ok(());
        }
        // One flush commits both bmap edits atomically.
        self.flush().await?;
        Ok(())
    }

    /// The bmap edits of a same-directory rename, **without flushing**: returns
    /// `false` when there was nothing to do (`old_name == new_name`), so the
    /// caller can skip the commit. Split out so a caller that is already going
    /// to flush — e.g. [`compact_then_rename`](Self::compact_then_rename) —
    /// commits the fold and the rename in a single flush instead of two.
    async fn rename_within_staged(&mut self, old_name: &str, new_name: &str) -> Result<bool> {
        if old_name == new_name {
            return Ok(false);
        }
        let old_home = hash_filename(old_name);
        let entry = match self.probe_lookup(old_name.as_bytes(), old_home).await? {
            Some((_, e)) => e,
            None => return Err(Error::new(ErrorKind::NotFound,
                format!("rename source not found: {}", old_name))),
        };

        // Same identity (uuid, inode AND blob pointer), new name. probe_upsert
        // overwrites an existing destination slot in place (replace-over-existing).
        let new_entry = entry.with_name(new_name.as_bytes());
        self.probe_delete(old_name.as_bytes()).await?;
        self.probe_upsert(new_name.as_bytes(), new_entry).await?;
        Ok(true)
    }

    /// Atomically exchange two entries in this directory (`RENAME_EXCHANGE`):
    /// the identities (uuid + inode) their names map to are swapped. Both names
    /// must already exist (else `NotFound`). Neither identity is created or
    /// removed, so nothing is reclaimed. One flush commits both upserts
    /// atomically (same as `rename_within`).
    pub async fn exchange_within(&mut self, a: &str, b: &str) -> Result<()> {
        if a == b {
            return Ok(());
        }
        let ea = match self.probe_lookup(a.as_bytes(), hash_filename(a)).await? {
            Some((_, e)) => e,
            None => return Err(Error::new(ErrorKind::NotFound,
                format!("exchange operand not found: {}", a))),
        };
        let eb = match self.probe_lookup(b.as_bytes(), hash_filename(b)).await? {
            Some((_, e)) => e,
            None => return Err(Error::new(ErrorKind::NotFound,
                format!("exchange operand not found: {}", b))),
        };

        // Swap: a now maps to b's identity and vice versa, each carrying its own
        // inode and blob pointer across.
        let new_a = eb.with_name(a.as_bytes());
        let new_b = ea.with_name(b.as_bytes());
        self.probe_upsert(a.as_bytes(), new_a).await?;
        self.probe_upsert(b.as_bytes(), new_b).await?;

        // One flush commits both bmap edits atomically.
        self.flush().await?;
        Ok(())
    }

    /// Every external storage location this directory's bmap references.
    ///
    /// A directory's B-tree leaves hold `DirFileEntryRaw` **inline payload**,
    /// not storage locations, so its metadata (B+tree) nodes are the *only*
    /// external storage the bmap occupies — which is exactly what this returns.
    /// The result is therefore the directory's complete reachable set, usable
    /// as-is by a segment reclaimer.
    ///
    /// Backed by `btree-ondisk`'s `collect_meta_ptrs`, which stops one level
    /// above the leaves (no leaf node is ever fetched). Note that
    /// `collect_ptrs` is **not** usable here: hyperdir's `V`
    /// (`DirFileEntryRaw`) differs from its `P` (`BlockPtr`), so that call
    /// returns `InvalidInput` by design.
    ///
    /// Returns an empty vector when there is no separately stored node: a
    /// direct (fully inline) root, or a B+tree shrunk to a single leaf root.
    ///
    /// # Errors
    ///
    /// `Err`, never a short result — a truncated reachable set is
    /// indistinguishable from a complete one and would let a caller delete live
    /// data. `InvalidData` means the bmap has un-assigned node ids (flush
    /// first); a loader error means a node could not be read. Treat any error
    /// as "reachability unknown" and reclaim nothing.
    pub async fn reachable_meta_ptrs(&self) -> Result<Vec<BlockPtr>> {
        self.bmap.collect_meta_ptrs().await
    }

    /// The set of **staging segments** this directory's bmap still references,
    /// decoded from [`reachable_meta_ptrs`](Self::reachable_meta_ptrs).
    ///
    /// Same fail-closed contract: on `Err`, reclaim nothing. Non-staging
    /// pointers (in-memory / zero / dummy) are skipped, so a segment is listed
    /// only when it actually holds one of this directory's metadata nodes.
    pub async fn reachable_segments(&self) -> Result<HashSet<SegmentId>> {
        let mut out = HashSet::new();
        for p in self.reachable_meta_ptrs().await? {
            if BlockPtrFormat::is_on_staging(&p) {
                out.insert(BlockPtrFormat::decode_segid(&p));
            }
        }
        Ok(out)
    }
}

impl<'a, T, L> HyperTrait<T, L, NullNodeCache, DirFileEntryRaw> for HyperDirFile<'a, T, L>
    where
        T: Staging<L> + SegmentReadWrite + Send + Clone + 'static,
        L: BlockLoader<BlockPtr> + Clone,
{
    fn blk_ptr_encode(&self, segid: SegmentId, offset: SegmentOffset, seq: usize) -> BlockPtr {
        BlockPtrFormat::encode(segid, offset, seq, &self.bmap_ud.blk_ptr_format)
    }

    fn blk_ptr_decode(&self, blk_ptr: &BlockPtr) -> (SegmentId, SegmentOffset) {
        BlockPtrFormat::decode(blk_ptr, &self.bmap_ud.blk_ptr_format)
    }

    fn blk_ptr_decode_display(&self, blk_ptr: &BlockPtr) -> String {
        if BlockPtrFormat::is_dummy_value(blk_ptr) {
            "[Dummy]".to_string()
        } else if BlockPtrFormat::is_invalid_value(blk_ptr) {
            "[Invalid]".to_string()
        } else if BlockPtrFormat::is_zero_block(blk_ptr) {
            "[Zero Block]".to_string()
        } else if BlockPtrFormat::is_on_staging(blk_ptr) {
            let (id, off) = self.blk_ptr_decode(blk_ptr);
            let group_id = BlockPtrFormat::decode_micro_group_id(blk_ptr);
            format!("[Staging: id {} - offset {} - group {}]", id, off, group_id)
        } else {
            format!("[Unknown: 0x{:x}]", blk_ptr)
        }
    }

    fn clear_data_blocks_cache(&mut self) {
        // do nothing
    }

    fn set_data_blocks_cache_unlimited(&mut self) {
        // do nothing
    }

    fn restore_data_blocks_cache_limit(&mut self) {
        // do nothing
    }

    fn get_data_blocks_dirty(&self) -> DirtyDataBlocks<'_> {
        DirtyDataBlocks { inner: Some(BTreeMap::new()), owned: None }
    }

    fn clear_data_blocks_dirty(&mut self) {
    }

    /// A directory has no dirty data blocks to demote — `get_data_blocks_dirty`
    /// answers empty above, so the flush path never names any. Empty for the
    /// same reason the three cache methods above are.
    fn demote_data_blocks_dirty(&mut self, _indexes: &[BlockIndex]) {
        // do nothing
    }

    fn pending_parts_get(&self) -> Option<PendingParts> {
        self.pending_parts
    }

    fn pending_parts_advance(&mut self, written: SegmentId) {
        let next = written.part_id()
            .expect("a partial segment names a part")
            .saturating_add(1);
        self.pending_parts = Some(PendingParts { segid: written.checkpoint(), next_part: next });
    }

    fn pending_parts_clear(&mut self) {
        self.pending_parts = None;
    }

    /// Always writable. hyperfile refuses writes once publishing has failed
    /// unrecoverably, which it tracks in its wal state; this crate has no wal
    /// implementation, so there is no such state to consult.
    fn check_writable(&self) -> Result<()> {
        Ok(())
    }

    async fn lock(&self) -> OwnedSemaphorePermit {
        let permit = self.sema.clone().acquire_owned().await.unwrap();
        permit
    }

    fn unlock(&self, permit: OwnedSemaphorePermit) {
        drop(permit);
    }

    async fn flush_lock(&self) -> OwnedMutexGuard<()> {
        self.flush_lock.clone().lock_owned().await
    }

    fn flush_unlock(&self, lock: OwnedMutexGuard<()>) {
        drop(lock);
    }

    fn bmap_as_slice(&self) -> &[u8] {
        self.bmap.as_slice()
    }

    fn bmap_get_block_loader(&self) -> L {
        self.bmap.get_block_loader()
    }

    fn bmap_get_node_cache(&self) -> NullNodeCache {
        NullNodeCache
    }

    fn bmap_dirty(&self) -> bool {
        self.bmap.dirty()
    }

    fn bmap_lookup_dirty(&self) -> Vec<BtreeNodeDirty<'_, BlockIndex, DirFileEntryRaw, BlockPtr>> {
        self.bmap.lookup_dirty()
    }

    async fn bmap_assign_meta_node(&self, blk_ptr: BlockPtr, node: BtreeNodeDirty<'_, BlockIndex, DirFileEntryRaw, BlockPtr>) -> Result<()> {
        self.bmap.assign_meta_node(blk_ptr, node).await
    }

    async fn bmap_assign_data_node(&self, blk_idx: &BlockIndex, blk_ptr: BlockPtr) -> Result<()> {
        self.bmap.assign_data_node(blk_idx, blk_ptr).await
    }

    fn bmap_clear_dirty(&mut self) {
        self.bmap.clear_dirty()
    }

    fn bmap_update(&mut self, bmap: BMap<'_, BlockIndex, DirFileEntryRaw, BlockPtr, L, NullNodeCache>) {
        self.bmap = unsafe {
            std::mem::transmute::<BMap<'_, BlockIndex, DirFileEntryRaw, BlockPtr, L, NullNodeCache>, BMap<'_, BlockIndex, DirFileEntryRaw, BlockPtr, L, NullNodeCache>>(bmap)
        };
    }

    async fn bmap_insert_dummy_value(bmap: &mut BMap<'_, BlockIndex, DirFileEntryRaw, BlockPtr, L, NullNodeCache>, blk_idx: &BlockIndex) -> Result<Option<DirFileEntryRaw>> {
        bmap.insert(*blk_idx, DirFileEntryRaw::dummy_value()).await
    }

    fn bmap_set_cache_unlimited(&self) -> usize {
        let limit = self.bmap.get_cache_limit();
        self.bmap.set_cache_limit(DEFAULT_CACHE_UNLIMITED);
        limit
    }

    fn bmap_set_cache_limit(&self, limit: usize) {
        self.bmap.set_cache_limit(limit);
    }

    fn inode(&self) -> &Inode {
        &self.inode
    }

    #[allow(mutable_transmutes)]
    fn inode_mut(&self) -> &mut Inode {
        unsafe {
            std::mem::transmute::<&Inode, &mut Inode>(&self.inode)
        }
    }

    fn staging(&self) -> &T {
        &self.staging
    }

    fn config(&self) -> &HyperFileConfig {
        &self.config
    }

    fn set_last_flush(&mut self) {
        self.last_flush = Instant::now();
    }

    async fn sleep(dur: Duration) {
        tokio::time::sleep(dur).await;
    }

    fn flush_timing(&self) -> &FlushTiming {
        &self.flush_timing
    }

    // ---- write-ahead log ----
    //
    // This crate implements no write-ahead log. The methods below are what "no
    // wal is configured" looks like, which is the same answer hyperfile gives
    // when its own `wal_mut()` is `None` -- see `wal_seal_barrier`, which
    // returns `Ok(())` on that branch.
    //
    // They were `todo!()` until now, and that was not safe: the flush path calls
    // `wal_drop_superseded_prefixes` gated only on `#[cfg(feature = "wal")]`,
    // never on whether a log exists, and its loop runs at least once even with
    // nothing deferred. So any directory flush with the feature on panicked.
    // Verified before this change: `fs_compact` on a 3-entry directory panicked
    // at the `wal_spawn_delete_segment` arm.
    //
    // Enabling `wal` on this crate therefore gives directory operations **no**
    // write-ahead logging. The feature exists so that a consumer which enables
    // `hyperfile/wal` for its *file* containers still compiles, since the trait
    // requires these whenever hyperfile has the feature on.

    #[cfg(feature = "wal")]
    fn wal_mut(&mut self) -> Option<&mut Box<dyn hyperfile::wal::WalReadWrite + Send>> {
        None
    }

    #[cfg(feature = "wal")]
    fn in_txn_trait(&self) -> bool {
        // No transactions here; a directory operation commits on its own.
        false
    }

    #[cfg(feature = "wal")]
    fn wal_may_defer_publish(&self) -> bool {
        // Deferring a publish is only safe when a log can replay what was
        // deferred. Without one, every flush must publish.
        false
    }

    #[cfg(feature = "wal")]
    fn wal_count_deferred_barrier(&mut self) {
        // Nothing defers, so there is nothing to count.
    }

    #[cfg(feature = "wal")]
    fn wal_deferred_count(&self) -> usize {
        0
    }

    #[cfg(feature = "wal")]
    fn wal_reset_deferred_count(&mut self) {
    }

    #[cfg(feature = "wal")]
    async fn wal_set_mem_segment(&self, _cno: Cno, _mem_segdata: Weak<Pin<Box<Vec<u8>>>>) {
        // Recovery reads the log to find in-memory segments; with no log there
        // is nothing to record them for.
    }

    #[cfg(feature = "wal")]
    async fn wal_clear_mem_segment(&self, _cno: Cno) {
    }

    #[cfg(feature = "wal")]
    fn wal_spawn_delete_segment(&self, _segid: SegmentId) {
        // No log chunks were written, so none are superseded.
    }
}


#[cfg(test)]
mod reachability_tests {
    use super::*;
    use btree_ondisk::NullBlockLoader;

    /// hyperdir's leaf value is inline payload (`DirFileEntryRaw`), not a
    /// storage location, so `V != P`. btree-ondisk refuses `collect_ptrs` for
    /// that shape with `InvalidInput` (checked before any node is read) — which
    /// is why `reachable_meta_ptrs` uses `collect_meta_ptrs` instead. Pin both
    /// halves so a future refactor can't silently switch to the wrong call.
    #[tokio::test]
    async fn collect_ptrs_refused_meta_ptrs_served() {
        let bmap = BMap::<BlockIndex, DirFileEntryRaw, BlockPtr, NullBlockLoader, NullNodeCache>::new(
            56, 4096, NullBlockLoader, NullNodeCache).expect("bmap");

        let err = bmap.collect_ptrs().await.expect_err("collect_ptrs must refuse V != P");
        assert_eq!(err.kind(), ErrorKind::InvalidInput, "got {:?}: {}", err.kind(), err);

        // The correct call succeeds. A fresh (direct/inline) root references no
        // separately stored node, so the reachable set is legitimately empty.
        let ptrs = bmap.collect_meta_ptrs().await.expect("collect_meta_ptrs must work for V != P");
        assert!(ptrs.is_empty(), "a fresh inline root has no metadata nodes, got {:?}", ptrs);
    }
}

#[cfg(test)]
mod block_ptr_format_tests {
    use super::*;
    use hyperfile::config::HyperFileMetaConfig;
    use hyperfile::segment::Segment;
    use hyperfile::SegmentId;

    /// Two things read a container's pointer format for different purposes and
    /// both must give the same answer: pointer encode/decode reads the bmap's
    /// userdata, while *object naming* reads the inode's `i_meta_config` through
    /// `Segment::object_of`. This crate hard-codes `Flat` for the bmap, so the
    /// inode has to record `Flat` too.
    ///
    /// They disagreed once hyperfile's default became `PartedSegment` (0.7.0):
    /// a directory wrote its objects as `<segid>.<part>` and then addressed them
    /// as `<segid>`, so every read missed and `read_dir` came back empty on a
    /// directory that had just been compacted — no error, just nothing. Pin both
    /// halves rather than the symptom.
    #[test]
    fn the_inode_records_the_format_the_bmap_encodes_with() {
        // What `HyperDirFile::new` writes into the inode.
        let meta = HyperFileMetaConfig {
            block_ptr_format: BlockPtrFormat::Flat,
            ..Default::default()
        };

        // What `blk_ptr_encode` / `blk_ptr_decode` use.
        let bmap_ud = BMapUserData::new(BlockPtrFormat::Flat);

        assert_eq!(meta.block_ptr_format, bmap_ud.blk_ptr_format,
            "the inode's format and the bmap's must agree");

        // The consequence that actually bit: an unparted format names a
        // checkpoint's object without a part suffix, which is what a `Flat`
        // pointer addresses.
        let named = Segment::object_of(SegmentId::new(7), &meta);
        assert_eq!(named.part_id(), None,
            "a directory's checkpoint must be a single unparted object, got {:?}", named);

        // And survives the round trip through the inode field.
        let decoded = HyperFileMetaConfig::from_u32(meta.as_u32());
        assert_eq!(decoded.block_ptr_format, BlockPtrFormat::Flat,
            "the format must round-trip through i_meta_config");
    }

    /// The upstream default is deliberately *not* what a directory uses, so if it
    /// ever becomes `Flat` this override is redundant and if it changes again the
    /// override still stands. Either way the test above is what matters; this one
    /// records why the override exists.
    #[test]
    fn the_upstream_default_is_parted_which_a_directory_must_not_inherit() {
        assert!(HyperFileMetaConfig::default().block_ptr_format.is_parted(),
            "hyperfile's default is expected to be parted; if this changed, \
             revisit the override in HyperDirFile::new");
    }
}

#[cfg(test)]
mod bmap_root_alignment_tests {
    use hyperfile::ondisk::InodeRaw;

    /// `BMap::read` peeks the root header through `BtreeNode::from_slice_ref`,
    /// which rejects a buffer that is not aligned to the node header (8 bytes)
    /// with `InvalidInput`. Every hyperdir call site therefore passes
    /// `&raw_inode.i_bmap` rather than a `[u8; N]` / `Vec<u8>` copy (alignment
    /// 1). That is only sound while `InodeRaw` is 8-aligned AND `i_bmap` sits at
    /// an 8-aligned offset within it -- both are upstream (hyperfile) layout
    /// properties, so assert them here instead of assuming them.
    #[test]
    fn i_bmap_is_eight_byte_aligned_within_inode() {
        assert!(std::mem::align_of::<InodeRaw>().is_multiple_of(8),
                "InodeRaw must be at least 8-aligned, got {}", std::mem::align_of::<InodeRaw>());
        let off = std::mem::offset_of!(InodeRaw, i_bmap);
        assert!(off.is_multiple_of(8), "i_bmap offset {} is not 8-aligned", off);

        // And the property that actually matters at the call site.
        let raw = InodeRaw::default();
        assert!((&raw.i_bmap as *const _ as usize).is_multiple_of(8), "&i_bmap not 8-aligned");
    }
}

#[cfg(test)]
mod blob_tests {
    use super::*;
    use hyperfile::ondisk::InodeRaw;
    use ulid::Ulid;
    use uuid::Uuid;

    #[test]
    fn scatter_body_lifts_blob_ulid_and_roundtrips() {
        let uuid = Uuid::new_v4();
        let ulid = Ulid::new();
        let inode = InodeRaw::default();

        // A blob body is `InodeRaw ++ 16-byte LE ulid`: the entry becomes a blob
        // whose version pointer is that ULID.
        let mut body = inode.as_u8_slice().to_vec();
        body.extend_from_slice(&u128::from(ulid).to_le_bytes());
        let raw = entry_from_scatter_body(&uuid, "foo", &body);
        assert!(raw.is_blob());
        assert_eq!(raw.blob_ulid(), Some(ulid));
        assert_eq!(&raw.uuid, uuid.as_bytes());
        assert_eq!(raw.name_bytes(), b"foo");

        // A plain body (exactly `InodeRaw`) is NOT a blob.
        let plain = inode.as_u8_slice().to_vec();
        let raw2 = entry_from_scatter_body(&uuid, "foo", &plain);
        assert!(!raw2.is_blob());
        assert_eq!(raw2.blob_ulid(), None);

        // The blob pointer survives the fixed-size on-disk serialization used by
        // the B-tree value (`as_u8_slice` -> `from_slice`).
        let bytes = raw.as_u8_slice().to_vec();
        let back = DirFileEntryRaw::from_slice(&bytes);
        assert_eq!(back.blob_ulid(), Some(ulid));
        assert_eq!(back.name_bytes(), b"foo");
    }
}