coordinode-lsm-tree 7.0.0

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2024-present, fjall-rs
// Copyright (c) 2026-present, Dmitry Prudnikov

#[cfg(feature = "metrics")]
use crate::metrics::Metrics;
#[cfg(not(feature = "std"))]
use alloc::vec::Vec;

use super::{
    block_index::BlockIndexImpl, block_layout::BlockLayoutMap, meta::ParsedMeta,
    regions::ParsedRegions,
};
use crate::deletion_pause::DeletionPause;
use crate::{
    Checksum, GlobalTableId, SeqNo, TableId,
    cache::Cache,
    comparator::SharedComparator,
    encryption::EncryptionProvider,
    file_accessor::FileAccessor,
    fs::Fs,
    range_tombstone::RangeTombstone,
    table::{IndexBlock, filter::block::FilterBlock, writer::LinkedFile},
    tree::inner::TreeId,
};
use alloc::boxed::Box;
use alloc::sync::Arc;
use core::sync::atomic::AtomicBool;

use crate::path::PathBuf;
use portable_atomic::AtomicU64;

pub struct Inner {
    pub path: Arc<PathBuf>,

    pub(crate) tree_id: TreeId,

    #[doc(hidden)]
    pub(crate) file_accessor: FileAccessor,

    /// Filesystem backend for file operations (open, remove, etc.).
    pub(crate) fs: Arc<dyn Fs>,

    /// Parsed metadata
    #[doc(hidden)]
    pub metadata: ParsedMeta,

    /// The bytes the file holds, as the table was opened: its whole length,
    /// unlike the metadata's `file_size`, which stops where the metadata was
    /// written and leaves out the sections after it.
    pub(crate) file_len: u64,

    /// Parsed region block handles
    #[doc(hidden)]
    pub regions: ParsedRegions,

    /// Translates key (first item of a block) to block offset (address inside file) and (compressed) size
    #[doc(hidden)]
    pub block_index: Arc<BlockIndexImpl>,

    /// Block cache
    ///
    /// Stores index and data blocks
    #[doc(hidden)]
    pub cache: Arc<Cache>,

    /// Pinned filter index (in case of partitioned filters)
    pub(super) pinned_filter_index: Option<IndexBlock>,

    /// Pinned AMQ filter
    pub pinned_filter_block: Option<FilterBlock>,

    /// The extractor this table was opened under, held only when its filter
    /// indexes that extractor's prefixes, so a prefix scan may skip the table
    /// on its answer. A view reopened over the same file carries it on.
    pub(super) prefix_filter: Option<alloc::sync::Arc<dyn crate::prefix::PrefixExtractor>>,

    /// True when the table was compacted away or dropped
    ///
    /// May be kept alive until all Arcs to the table have been dropped (to facilitate snapshots)
    pub is_deleted: AtomicBool,

    /// Whether the directory entry naming the file was durable before the
    /// table was written into it (a pooled file), so the install that first
    /// names the table need not sync its folder.
    pub(crate) entry_durable: AtomicBool,

    /// Tight-space punch-on-drop offset, or [`u64::MAX`] (the default) for "no
    /// punch". When a tight-space compaction restricts a table to `[K, hi)`, the
    /// PRIOR (unrestricted) view of the same physical SST is marked here with
    /// `offset(K)`: once every reader holding that old view has dropped its
    /// `Arc` (so no read can touch the prefix), this view's [`Drop`] reclaims the
    /// `[0, offset)` byte range via [`Fs::punch_hole`](crate::fs::Fs::punch_hole)
    /// while LEAVING the file in place (the restricted view, a distinct `Inner`,
    /// still serves the suffix). Distinct from [`Self::is_deleted`]: a punched
    /// view is not deleted. Using the `Drop`-at-refcount-zero signal makes the
    /// punch safe without an explicit snapshot gate — it fires exactly when the
    /// old view is unreachable. Cumulative across slices and idempotent.
    pub(crate) punch_on_drop: AtomicU64,

    /// How far this file's data region is known punched, shared by the views
    /// of the file a tight-space pass restricts it to one after another: each
    /// view punches only above what an earlier one finished, rather than the
    /// whole prefix again on every slice. Raised only by a punch that
    /// completed; one deferred or stopped by a failure leaves it, so the next
    /// view punches that range too.
    pub(crate) punched_through: once_cell::race::OnceBox<Arc<AtomicU64>>,

    pub(super) checksum: Checksum,

    pub(super) global_seqno: SeqNo,

    /// L0 recency key (see [`super::Table::l0_recency`]): the manifest's, or
    /// the file's own when opened without one.
    pub(super) l0_recency: TableId,

    pub(crate) comparator: SharedComparator,

    #[cfg(feature = "metrics")]
    pub(crate) metrics: Arc<Metrics>,

    /// The table's `linked_blob_files` records, read on first use and kept, so
    /// compaction decisions and the storage statistics do no I/O for them
    /// after the first. Empty for a table that references no blob file. The
    /// section is immutable after open, so two racing first readers parse the
    /// same records and either result is the one kept.
    pub(crate) blob_links: once_cell::race::OnceBox<Box<[LinkedFile]>>,

    /// Cumulative point reads that consulted this segment's data, bumped
    /// (`Relaxed`) once a read passes the segment's seqno-range and bloom gates
    /// (a bloom miss or out-of-seqno-range probe is NOT counted, so the figure
    /// tracks data hotness, not raw probe frequency). A monotonic counter, not a
    /// rate: a tiering consumer derives a read-rate / EMA from successive polls
    /// (delta over elapsed time). Reset to `0` only when the segment is created.
    pub(crate) read_count: AtomicU64,

    /// Cumulative range scans that opened a reader over this segment, bumped
    /// (`Relaxed`) once per scan as it reaches the segment. Beside
    /// [`Self::read_count`] it tells how a segment is read: by point reads,
    /// whose cost a filter keeps to one run, or by scans, which read every
    /// run they cross. Reset to `0` only when the segment is created.
    pub(crate) scan_count: AtomicU64,

    /// Unix seconds when this segment began counting its reads in this
    /// process, when it was written or opened: [`Self::read_count`] and
    /// [`Self::scan_count`] are reads since then, which a rate divides by.
    pub(crate) counting_since_secs: u64,

    /// Unix seconds of the most recent data-consulting read (same gating as
    /// [`Self::read_count`]), or `0` if never read. Maintained only on `std`
    /// builds (system clock); a no-std build leaves it at `0`. The recency
    /// signal for demotion / tiering decisions.
    pub(crate) last_access_secs: AtomicU64,

    /// This table's filter probes, installed when its tree allocates filter
    /// memory by probe load and absent otherwise, so a tree that does not
    /// pays no counting on the probe path.
    pub(crate) probe_stats: once_cell::race::OnceBox<crate::table::probe_stats::ProbeStats>,

    /// Range tombstones stored in this table. Loaded on open.
    pub(crate) range_tombstones: Vec<RangeTombstone>,

    /// The greatest end of `range_tombstones` in the tree's key order, so a
    /// point read rules the table out in two compares.
    pub(crate) range_tombstones_end: Option<crate::UserKey>,

    /// Inner zstd-block layout index, loaded on open from the optional
    /// `block_layout` section. Empty (no entries) when the table has no
    /// multi-inner-block data blocks. Lets a range query partial-decode only
    /// the inner blocks covering a key range in a large cold block.
    // Read only by the zstd partial-decode path; in a no-zstd build it is still
    // loaded (always empty there, since no zstd blocks ever split) but unread.
    #[cfg_attr(
        not(feature = "zstd"),
        expect(
            dead_code,
            reason = "consumed only by the zstd partial-decode read path"
        )
    )]
    pub(crate) block_layout: BlockLayoutMap,

    /// Per-data-block seqno bounds, loaded on open from the optional
    /// `seqno_bounds` section. Empty when the table has none (legacy or
    /// `seqno_in_index` off), so `scan_since_seqno` falls back to a full
    /// per-entry filter. Read only by the seqno-scoped scan path.
    pub(crate) seqno_bounds: crate::table::seqno_bounds::SeqnoBoundsMap,

    /// Per-data-block zone map, loaded on open from the optional `zone_map`
    /// section. Empty when the table has none (zone-map policy off), so a
    /// predicate scan falls back to reading every block. Read by the
    /// range-cardinality estimate and the block-skip scan path.
    pub(crate) zone_map: crate::table::zone_map::ZoneMap,

    /// Per-segment positional delete-bitmap, loaded on open from the optional
    /// `delete_bitmap` section. Empty when the segment has no materialized
    /// deletes, so a read applies no mask. `Arc` so a reader can hold it cheaply.
    pub(crate) delete_bitmap: alloc::sync::Arc<crate::table::delete_bitmap::DeleteBitmap>,

    /// Each data block's first global row position (file offset -> start row, in
    /// block-index = writer order), built on open from the zone map when the
    /// delete-bitmap is non-empty. `None` when there are no deletes. Lets the
    /// read paths resolve a block's start position in O(1) for positional
    /// masking, instead of recomputing the cumulative row counts per read.
    pub(crate) delete_block_starts: Option<alloc::sync::Arc<crate::HashMap<u64, u32>>>,

    /// Whether a SALVAGE-MODE open degraded this segment's delete masking: its
    /// `delete_bitmap` section exists but was unreadable (so the bitmap was
    /// reset to empty) or readable-but-unpositionable (its zone map was
    /// unreadable, so it was ignored). Reads then show every row live,
    /// resurrecting positionally-deleted rows — the salvage walk consults this
    /// to fail closed unless the caller explicitly opted into that
    /// degradation. Always `false` on a normal (non-salvage) open, which fails
    /// instead of degrading.
    pub(crate) delete_bitmap_degraded: bool,

    /// Whether a SALVAGE-MODE open degraded a REBUILDABLE side section
    /// (filter / `filter_tli`, seqno bounds, zone map, locator) because its
    /// block did not decode as the claimed type. Salvage re-derives every such
    /// section from the recovered entries, so a section that is present but does
    /// not decode may be a `range_tombstones` / `delete_bitmap` relabeled to a
    /// rebuildable name and re-roled — which salvage would discard, resurrecting
    /// the suppressed rows. The salvage walk consults this to fail closed when
    /// the table exposes no deletion metadata. Purely STRUCTURAL (each decode
    /// reads its own section's bytes, independent of the data blocks), so a
    /// corrupt DATA block does not trip it.
    pub(crate) rebuildable_section_degraded: bool,

    /// Retrieval-ribbon locator, loaded on open from the optional `locator`
    /// section. `Some` only when the table was written with a locator policy
    /// enabled; lets a point read resolve a key to its data block in O(1),
    /// skipping the index-block binary search. `None` (default) leaves the
    /// point read on the sorted-index path.
    pub(crate) locator_index: Option<crate::table::locator::LoadedLocator>,

    /// Block encryption provider for encryption at rest.
    pub(crate) encryption: Option<Arc<dyn EncryptionProvider>>,

    /// Pre-trained zstd dictionary for dictionary decompression.
    #[cfg(zstd_any)]
    pub(crate) zstd_dictionary: Option<Arc<crate::compression::ZstdDictionary>>,

    /// Tree-wide file-deletion gate. Installed once by
    /// [`Table::install_deletion_pause`](super::Table::install_deletion_pause)
    /// after the table is registered with a tree. When `Some` and active,
    /// the [`Drop`] impl defers the underlying `remove_file` call so that
    /// an in-progress [`Tree::create_checkpoint`](crate::AbstractTree::create_checkpoint)
    /// can hard-link the file before it disappears.
    // `once_cell::race::OnceBox` rather than `std::sync::OnceLock` so
    // this field doesn't pin the type to `std` — OnceBox is no-std +
    // alloc by construction. The slot is set once after recovery /
    // compaction and read on every Drop; CAS-based race semantics are
    // fine for this single-publisher, many-reader pattern.
    pub(crate) deletion_pause: once_cell::race::OnceBox<Arc<DeletionPause>>,

    /// Tree-wide background file deleter. Installed once by
    /// [`Table::install_background_deleter`](super::Table::install_background_deleter)
    /// after the table is registered with a tree. When present (and no
    /// checkpoint pause is active), the [`Drop`] impl reclaims the SST's blocks
    /// synchronously via [`Fs::truncate_file`](crate::fs::Fs::truncate_file) and
    /// hands the directory-entry `unlink` to this deleter's worker, off the
    /// foreground path. Absent (e.g. orphan cleanup before a tree owns the
    /// table) the Drop falls back to a synchronous `remove_file`.
    // std-only (the deleter spawns a thread); `no_std` builds never install
    // one and keep the synchronous Drop path. OnceBox keeps the field itself
    // alloc-friendly, matching `deletion_pause`.
    #[cfg(feature = "std")]
    pub(crate) background_deleter: once_cell::race::OnceBox<Arc<crate::BackgroundDeleter>>,

    /// Tree-wide ECC heal-hint sink. Installed once by
    /// [`Table::install_heal_hints`](super::Table::install_heal_hints) after the
    /// table joins a tree. When present, a block read that is ECC-corrected and
    /// confirmed persistent (via a cache-bypassing re-read) records this SST's
    /// id so the compaction picker can rewrite it clean. Absent before the table
    /// is tree-owned: corrections are still returned to the caller, they are
    /// just not scheduled for healing.
    // alloc-friendly `OnceBox` (not `std::sync::OnceLock`) so the field does not
    // pin `Inner` to std, matching `deletion_pause`. The hint set itself is
    // `no_std` + alloc (see `crate::heal_hints`).
    pub(crate) heal_hints: once_cell::race::OnceBox<Arc<crate::heal_hints::HealHints>>,

    /// How columnar reads of this table fetch their pages, installed once the
    /// table joins a tree; a table read before it is bound reads under the
    /// default budget. Not a property of the file, so it is the tree's to set.
    pub(crate) read_budget: once_cell::race::OnceBox<crate::config::ReadBudget>,

    /// Serializes concurrent in-place heal passes of THIS table, held by the
    /// patrol scrub across the WHOLE scan-to-reconcile span. Two overlapping
    /// heals race in two ways without it: through the link-count probe (one
    /// detaches the live path onto a private copy, leaving the other's
    /// already-open handle on the OLD inode — whose count then reads 1 even
    /// though only checkpoint links remain, so that heal would write through
    /// the snapshot), and through the digest reconciliation (A computes a
    /// digest, B heals a fresh fault and installs its own, A installs the
    /// stale one).
    // std+page_ecc: only the heal-mode patrol scrub takes it; `parking_lot`
    // (not `spin`) because a heal pass is a long blocking operation.
    //
    // Held behind a shared `Arc` inside an `OnceBox` so it serializes by STABLE
    // table identity, not by this replaceable `Inner`: tight-space compaction
    // re-opens a table as a DISTINCT `Inner` (a different physical view of the
    // same file), and `reopen_restricted` propagates this lock into it so two
    // patrols cannot heal + reconcile the same SST concurrently. Lazily created
    // on first heal for an ordinary table (the tree does not install it).
    #[cfg(all(feature = "std", feature = "page_ecc"))]
    pub(crate) heal_lock: once_cell::race::OnceBox<Arc<parking_lot::Mutex<()>>>,
}

impl Inner {
    /// Gets the global table ID.
    #[must_use]
    pub(super) fn global_id(&self) -> GlobalTableId {
        (self.tree_id, self.metadata.id).into()
    }

    /// Walks the block index for maintenance that rewrites or reshapes the
    /// table: compaction input, heal, salvage and the geometry of a restricted
    /// view. A cold index is read from disk as it is walked; those reads stay
    /// out of the read counters, which describe reads, and are cached like any
    /// other, since the table is about to be read in full.
    pub(crate) fn maintenance_index_walk(&self) -> super::block_index::BlockIndexIterImpl {
        use super::block_index::BlockIndex;
        self.block_index
            .iter()
            .with_charge(super::util::ReadCharge::Maintenance)
    }

    /// Walks the block index for work that must leave no trace: the patrol
    /// scrub, verification gates and the space reclaim of a dropped table.
    /// Uncounted like maintenance, and it neither fills the block cache nor
    /// promotes what is cached, so a patrol over every table cannot evict the
    /// workload's blocks.
    pub(crate) fn untraced_index_walk(&self) -> super::block_index::BlockIndexIterImpl {
        use super::block_index::BlockIndex;
        self.block_index
            .iter()
            .with_charge(super::util::ReadCharge::Untraced)
    }
}

impl Drop for Inner {
    fn drop(&mut self) {
        let global_id = self.global_id();

        if self.is_deleted.load(core::sync::atomic::Ordering::Acquire) {
            log::trace!("Cleanup deleted table {global_id:?} at {:?}", self.path);

            // A pin may list the file through another view of it (the one
            // before a tight-space restriction): it stays, with its sidecars,
            // until that pin drops. The pin's consumer reads the heal
            // attestation beside the table.
            #[cfg(feature = "std")]
            let pinned = self.deletion_pause.get().is_some_and(|pause| {
                pause.defer_removal_if_pinned(
                    &self.fs,
                    &self.path,
                    crate::deletion_pause::PinnedKind::Table,
                )
            });
            #[cfg(not(feature = "std"))]
            let pinned = false;

            // Reclaim any pending `.heal-attest` sidecar: a retired table can
            // never be reconciled, so its attestation is dead weight. Done here
            // (before the deferred / background unlink paths return) so every
            // deletion route reclaims it. Best-effort: a missing sidecar (the
            // common case) is a no-op, and the recovery scan sweeps any straggler.
            #[cfg(feature = "std")]
            if !pinned {
                crate::scrub::heal_attest::remove(&*self.fs, &self.path);
            }

            // Reclaim any `.restrict-bound` sidecar the same way: a retired table's
            // tight-space restriction bound is dead weight, and left behind it would
            // linger as an orphan (swept by the recovery scan, but a leak until
            // then). Best-effort; done before the deferred / background unlink paths
            // return so every deletion route reclaims it. A concurrent checkpoint has
            // already linked its OWN copy of the sidecar, so removing this original is
            // safe even on the deferred-deletion (checkpoint-active) branch below.
            #[cfg(feature = "std")]
            if !pinned {
                crate::restrict_bound::remove(&*self.fs, &self.path, crate::fs::SyncMode::Normal);
            }

            // Move the accessor and block index out so all file handles
            // (including clones held by the block index) are closed before
            // attempting deletion. On Windows, remove_file fails while any
            // handle is open.
            let file_accessor = core::mem::replace(&mut self.file_accessor, FileAccessor::Closed);
            let block_index =
                core::mem::replace(&mut self.block_index, Arc::new(BlockIndexImpl::Closed));

            // Evict cached FD from the descriptor table.
            file_accessor.as_descriptor_table().inspect(|d| {
                d.remove_for_table(&global_id);
            });

            // Drop the accessor and block index (releases all Arc<dyn FsFile>).
            drop(file_accessor);
            drop(block_index);

            if pinned {
                return;
            }

            // If a checkpoint is active, defer the physical deletion so the
            // file remains hard-linkable until the checkpoint releases its
            // pause. Falls through to immediate removal when no pause is
            // installed or the pause is inactive.
            // Short-circuit on the common no-checkpoint path: skip
            // the Arc<dyn Fs> bump and PathBuf clone unless a pause is
            // both installed AND currently active. `try_enqueue` still
            // re-checks `is_active()` under the queue lock to close
            // the publish-then-release race, so the outer check is a
            // pure perf gate, not a correctness one.
            if let Some(pause) = self.deletion_pause.get()
                && pause.is_active()
                && pause.try_enqueue(Arc::clone(&self.fs), (*self.path).clone())
            {
                log::trace!(
                    "Deferred deletion of table {global_id:?} at {:?} (checkpoint active)",
                    self.path,
                );
                return;
            }

            // Off-foreground reclaim: return the SST's blocks to the filesystem
            // synchronously (so a footprint scan reflects the reclaim at once)
            // and hand the directory-entry unlink to the background deleter.
            // Falls through to a synchronous remove_file when no deleter is
            // installed (e.g. orphan cleanup before a tree owns the table).
            #[cfg(feature = "std")]
            if let Some(deleter) = self.background_deleter.get() {
                // Truncate (instant block-free) only when we own the sole hard
                // link. A completed checkpoint may have hard-linked this SST;
                // truncating the shared inode would zero the checkpoint's copy
                // too. When the link is shared (or the count is unknown), skip
                // the truncate and just unlink our directory entry — the data
                // survives via the other link and its blocks free once the last
                // link is gone. (An in-progress checkpoint is already handled by
                // the deletion-pause branch above.)
                if self.fs.hard_link_count(&self.path).is_ok_and(|n| n <= 1)
                    && let Err(e) = self.fs.truncate_file(&self.path)
                {
                    log::warn!(
                        "Failed to truncate deleted table {global_id:?} at {:?}: {e:?}",
                        self.path,
                    );
                }
                deleter.enqueue(Arc::clone(&self.fs), (*self.path).clone());
                return;
            }

            if let Err(e) = self.fs.remove_file(&self.path) {
                log::warn!(
                    "Failed to cleanup deleted table {global_id:?} at {:?}: {e:?}",
                    self.path,
                );
            }
        } else {
            // Not deleted, but possibly marked for tight-space prefix reclaim:
            // this (old, unrestricted) view's last Arc is dropping, so no read
            // can touch the prefix anymore. Reclaim the consumed prefix's DATA
            // blocks and LEAVE the file — the restricted view (a distinct Inner)
            // still serves the suffix. `punch_hole` opens the path itself, so it
            // is fine that this view's own handles drop right after this body.
            //
            // Punch the data blocks below the boundary in calls of many adjacent
            // blocks, not one call per block: a filesystem deallocates only the
            // pages wholly inside one call and zero-fills the edges, so blocks
            // of about a page punched one by one at unaligned offsets would keep
            // nearly every page allocated. A call takes whole blocks up to about
            // [`PUNCH_CHUNK`], so each edge between two calls keeps at most one
            // page allocated. The writer lays the data blocks out contiguously
            // from offset 0 and the block index yields only data-block handles,
            // so the calls cover exactly the reclaimable data (the index, filter
            // and meta sections all sit past the data region).
            //
            // Punch TOP-DOWN (highest call first) and STOP at the first failure.
            // This keeps the resulting hole pattern CLASSIFIABLE for a
            // sidecar-less manifest repair: every call that lands zeroes at
            // least one whole block, so any failure (or crash) leaves intact
            // blocks strictly BELOW zeroed ones (the irregular signature repair
            // fails closed on), while a fully successful pass leaves the clean
            // zeroed prefix the classical geometry bound is sound for. A call
            // that ended inside a block would zero only its tail, and if no call
            // below it landed, the file would show no zeroed block at all: it
            // would read as unpunched, with consumed rows intact. Bottom-up with
            // continue-past-failures could instead leave intact-but-consumed
            // blocks ABOVE a clean zeroed prefix, indistinguishable from a live
            // suffix, and a sidecar-less repair would then restrict to a bound
            // that resurrects their superseded rows. Stopping reclaims less
            // space on a failure, but reclaim is best-effort; classification
            // soundness is not.
            let off = self
                .punch_on_drop
                .load(core::sync::atomic::Ordering::Acquire);
            if off == u64::MAX {
                return;
            }
            // `[start, end)` of each data block below `off`, in file order.
            let mut blocks: alloc::vec::Vec<(u64, u64)> = alloc::vec::Vec::new();
            // The table is being dropped: caching its index would only evict
            // blocks someone still reads.
            for handle in self.untraced_index_walk() {
                // Log why reclaim stopped instead of silently swallowing the
                // block-index read error (this is an integrity-sensitive path).
                let handle = match handle {
                    Ok(handle) => handle,
                    Err(e) => {
                        log::warn!(
                            "Failed to iterate block index while punching table {global_id:?} at {:?}: {e:?}",
                            self.path,
                        );
                        return;
                    }
                };
                let block_off = handle.offset().0;
                if block_off < off {
                    // Bounded by the file's length.
                    blocks.push((block_off, block_off + u64::from(handle.size())));
                }
            }
            // An earlier view of this file punched everything below the floor;
            // from the 64 KiB edge under it, so the page its punch ended in is
            // freed now that the rest of it is consumed too. The block cut at
            // the floor lies wholly below that punch's end, already zeroed.
            let floor = self.punched_through.get().map_or(0, |punched| {
                punched.load(core::sync::atomic::Ordering::Acquire)
            });
            let floor = floor - floor % PUNCH_ALIGN;
            blocks.retain_mut(|block| {
                block.0 = block.0.max(floor);
                block.0 < block.1
            });
            let reclaimable = punch_extents_top_down(&blocks);

            // An IN-PROGRESS checkpoint's window covers the link step, so a
            // punch cannot race a link that is about to appear — but the intent
            // lives in this view, which is dropping, so DEFER it onto the pause
            // rather than lose it (the release re-probes and punches then).
            // The extents are CLONED into the enqueue (not moved): a pause
            // released between the `is_active` check and the enqueue's own
            // re-check returns `false`, and the reclaim must then still be in
            // hand for the immediate path below.
            #[cfg(feature = "std")]
            if let Some(pause) = self.deletion_pause.get()
                && pause.is_active()
                && pause.try_enqueue_punch(
                    Arc::clone(&self.fs),
                    (*self.path).clone(),
                    reclaimable.clone(),
                )
            {
                log::trace!(
                    "Deferred tight-space punch of table {global_id:?} at {:?} (checkpoint active)",
                    self.path,
                );
                return;
            }

            // A pinned file keeps its bytes until its last pin drops, which
            // retries the retained reclaim; a pin being taken waits for this
            // punch to finish.
            #[cfg(feature = "std")]
            let _unpinned = if let Some(pause) = self.deletion_pause.get() {
                let Some(guard) = pause.begin_mutation(&self.path) else {
                    pause.retain_reclaim(Arc::clone(&self.fs), (*self.path).clone(), reclaimable);
                    return;
                };
                Some(guard)
            } else {
                None
            };

            // Reclaim only what this tree exclusively owns. A completed
            // checkpoint hard-links SST files, and its captured manifest still
            // records this file UNRESTRICTED under its original digest —
            // punching a shared inode would zero those blocks inside the
            // immutable checkpoint too. FAIL CLOSED on a shared link and on a
            // probe that cannot answer — but RETAIN the reclaim rather than
            // discard it: this dropping view holds its only record, the link
            // disappears when the checkpoint is deleted, and only
            // `retry_pending_reclaims` can free the consumed prefix then.
            // Mirrors the delete path's truncate guard and the blob punch.
            #[cfg(feature = "std")]
            match self.fs.hard_link_count(&self.path) {
                Ok(n) if n <= 1 => {}
                Err(e) if e.kind() == crate::io::ErrorKind::NotFound => {
                    // The file is gone (retired): its space is already back.
                    return;
                }
                probe => {
                    log::debug!(
                        "Retaining tight-space punch of table {global_id:?} at {:?} for a retry: \
                         the file is hard-linked (or the link count is unknown: {probe:?})",
                        self.path,
                    );
                    if let Some(pause) = self.deletion_pause.get() {
                        pause.retain_reclaim(
                            Arc::clone(&self.fs),
                            (*self.path).clone(),
                            reclaimable,
                        );
                    }
                    return;
                }
            }

            let mut completed = true;
            for (at, &(extent_off, len)) in reclaimable.iter().enumerate() {
                if let Err(e) = self.fs.punch_hole(&self.path, extent_off, len) {
                    completed = false;
                    log::warn!(
                        "Failed to punch tight-space extent at {extent_off} of table {global_id:?} at {:?}; \
                         stopping the reclaim to keep the hole pattern classifiable: {e:?}",
                        self.path,
                    );
                    // The pass stops here (punching below an unreclaimed extent
                    // would break the top-down hole pattern a sidecar-less
                    // repair reads), but the failure is often transient: the
                    // failed extent and the untried remainder are retained for
                    // the same retry rather than discarded.
                    #[cfg(feature = "std")]
                    if let Some(pause) = self.deletion_pause.get()
                        && let Some(rest) = reclaimable.get(at..).map(<[(u64, u64)]>::to_vec)
                    {
                        pause.retain_reclaim(Arc::clone(&self.fs), (*self.path).clone(), rest);
                    }
                    break;
                }
            }
            if completed && let Some(punched) = self.punched_through.get() {
                punched.fetch_max(off, core::sync::atomic::Ordering::Release);
            }
        }
    }
}

/// The edge a punch above an earlier one starts from: a multiple of every
/// page size a filesystem uses, so the page the earlier punch ended in is
/// covered whole.
pub const PUNCH_ALIGN: u64 = 64 * 1_024;

/// The most one prefix punch call covers, unless a single block is larger: a
/// call is cut only at block edges, so each one that lands zeroes whole blocks.
pub const PUNCH_CHUNK: u64 = 4 * 1_024 * 1_024;

/// The punch calls for `blocks` (ascending, disjoint `[start, end)`), highest
/// first: runs of adjacent blocks, each call as many whole blocks as fit in
/// [`PUNCH_CHUNK`] and at least one.
pub fn punch_extents_top_down(blocks: &[(u64, u64)]) -> alloc::vec::Vec<(u64, u64)> {
    let mut extents = alloc::vec::Vec::new();
    // `[lo, hi)` of the call being built.
    let mut call: Option<(u64, u64)> = None;
    for &(start, end) in blocks.iter().rev() {
        call = match call {
            Some((lo, hi)) if lo == end && hi - start <= PUNCH_CHUNK => Some((start, hi)),
            Some((lo, hi)) => {
                extents.push((lo, hi - lo));
                Some((start, end))
            }
            None => Some((start, end)),
        };
    }
    if let Some((lo, hi)) = call {
        extents.push((lo, hi - lo));
    }
    extents
}