git-remote-object-store 0.2.4

//! Two-phase mark-and-sweep garbage collection for orphan packs
//! (issue #66, Phase 5 of #52).
//!
//! Orphan packs are pack files in `<prefix>/packs/` that no
//! `chain.json` references. They accumulate from:
//!
//! - **Force push**: replaces a chain's segments; old packs become orphan.
//! - **Lost-race push**: a pre-lock pack upload by the loser of a
//!   concurrent push (Phase 2 design — packs upload pre-lock to keep
//!   the lock window short, and the loser's pack is left orphan).
//! - **Aborted push**: a crash between pack upload and chain.json
//!   commit leaves orphans the next push doesn't reach.
//! - **Branch deletion**: `delete-branch` removes `chain.json` and
//!   `path-index.json` but does not touch `<prefix>/packs/`. The
//!   issue umbrella's "exclusively owned by that branch" claim is
//!   wrong under content-hash dedup; pack keys can be shared across
//!   branches that ever pushed identical object sets. The baseline
//!   bundle (`<prefix>/<ref>/<full_at>.bundle`) is tombstoned rather
//!   than deleted synchronously (issue #143), so an in-flight fetcher
//!   that already read the prior `chain.json` can still complete its
//!   range GET; the bundle is reclaimed by [`sweep`] after the grace
//!   window.
//! - **Compaction** (when implemented): a chain rewrite leaves the
//!   superseded segment packs orphan.
//! - **Missing `.idx`** (rare): a `.pack` whose sibling `.idx` was
//!   manually deleted is treated as orphan and tombstoned.
//!
//! ## Two-phase mark-and-sweep
//!
//! Naive deletion ("delete every pack older than 24 h") races a
//! concurrent fetch on a freshly-orphaned pack: the pack's
//! `last_modified` reflects upload time, not orphan time. The
//! mark/sweep split fixes this by tombstoning at orphan time and
//! deferring deletion until after a configurable grace window.
//!
//! ### Phase 1 (mark)
//!
//! 1. List `<prefix>/packs/` to snapshot the packs currently on the
//!    bucket. Packs-first is deliberate (issue #135): see "Concurrency"
//!    below.
//! 2. List `<prefix>/refs/**/chain.json` across every ref namespace
//!    (`refs/heads/`, `refs/tags/`, `refs/notes/`, etc.), parse each,
//!    collect referenced pack content-shas.
//! 3. **Fail closed** on parse error: abort, log the bad key, do not
//!    write tombstones. A corrupt chain could under-report the
//!    referenced set and tombstone live packs.
//! 4. Derive the orphan set (`on_bucket - referenced`) and write
//!    `<prefix>/gc/tombstones-<run_id>-<rfc3339>.json`.
//!
//! ### Phase 2 (sweep)
//!
//! 1. List `<prefix>/gc/tombstones-*.json`.
//! 2. For each tombstone past the grace age:
//!    - Re-derive the orphan set from the *current* chain state.
//!      Repeated **per tombstone**, not cached across the sweep: a
//!      concurrent push committing `chain.json` mid-sweep would let
//!      a cached snapshot delete a pack the new chain references,
//!      permanently dangling the reference (issue #140). Force-revert
//!      is the canonical trigger — deterministic gix pack emission
//!      lets the new push reuse the tombstoned pack key without
//!      re-uploading. The cost is one `list("refs/")` per eligible
//!      tombstone vs one per sweep; correctness wins over the linear
//!      overhead for the O(1)-eligible-tombstones common case.
//!    - For each pack still orphan, delete `.pack` + `.idx`
//!      idempotently (a prior partial sweep is fine).
//!    - Delete the tombstone itself.
//! 3. Younger tombstones survive for the next sweep.
//!
//! ### Baseline-bundle tombstones (issues #134, #143)
//!
//! Baseline bundles at `<prefix>/<ref>/<full_at>.bundle` are NOT
//! reapable by the mark/sweep flow above — they live outside
//! `<prefix>/packs/`, so [`list_pack_shas`] never sees them. The
//! compact, force-push, and `delete-branch` code paths instead enqueue
//! a baseline tombstone at `<prefix>/gc/baseline-tomb-<uuid>.json`
//! whenever they supersede or remove a baseline. Sweep processes those alongside pack
//! tombstones: after the grace window expires it re-checks the
//! current `chain.json` for the ref (skipping the delete if a later
//! push re-baselined to the same SHA), then deletes the bundle and
//! the tombstone. The bundle stays in place for the entire grace
//! window, so a concurrent fetch that read the prior `chain.json`
//! before the compact/force-push committed can still download it.
//!
//! ### `--force`
//!
//! Skips ONLY the grace window. The live-pack re-check still runs:
//! a tombstone whose SHA appears in the current chain set is left
//! alone. This closes the race where `mark()` snapshots packs after a
//! concurrent push has uploaded `packs/<sha>.{pack,idx}` but has not
//! yet committed `chain.json` — by sweep time the chain has landed
//! and the pack is live, so the stale tombstone must not delete it.
//! A `tracing::warn!` line records the operator's choice.
//!
//! ## Concurrency
//!
//! Two operators running `gc` simultaneously each get a `UUIDv4` run id
//! → distinct tombstone files, no clobber. Concurrent sweeps tolerate
//! `NotFound` on already-deleted packs.
//!
//! Mark lists packs first, then chains (issue #135). With this order,
//! a push landing during mark either:
//!
//! - uploaded its pack *after* [`list_pack_shas`] — the pack is not in
//!   the on-bucket snapshot, so it cannot enter the orphan set
//!   regardless of when its chain commits; or
//! - uploaded its pack *before* [`list_pack_shas`] AND committed
//!   `chain.json` before [`list_referenced_packs`] — the pack is in the
//!   referenced set, so it is filtered out of orphans; or
//! - uploaded its pack *before* [`list_pack_shas`] and has not yet
//!   committed `chain.json` by the time [`list_referenced_packs`] runs
//!   — the pack is tombstoned, but the grace window leaves it readable
//!   long enough for the push to complete (the genuine-orphan case for
//!   an aborted push is exactly what the GC is designed to reap).
//!
//! The reverse order (chains-first) is the bug fixed by #135: a chain
//! commit landing between the chain list and the pack list would let a
//! freshly-uploaded pack appear in [`list_pack_shas`] without appearing
//! in [`list_referenced_packs`], producing a false-positive tombstone.
//! Sweep's per-tombstone re-derive (issue #140) would usually catch
//! that at sweep time, but a `--force` sweep run in the same session as
//! mark (e.g. `compact --with-gc`) could still delete the live pack
//! before the push's chain commit lands.
//!
//! The grace window separately covers a fetch reading an old chain
//! whose packs are about to be swept.

use std::collections::HashSet;

use bytes::Bytes;
use futures::stream::{StreamExt, TryStreamExt};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;
use tracing::{debug, error, info, warn};
use uuid::Uuid;

use crate::git::RefName;
use crate::keys;
use crate::object_store::{ObjectStore, ObjectStoreError, PutOpts};
use crate::protocol::fetch::MAX_FETCH_CONCURRENCY;

use super::PackchainError;
use super::manifest::load_chain;
use super::schema::{ChainManifest, Sha40};

/// Default grace window between mark and sweep (24 hours). A pack
/// tombstoned during mark is only deletable after this duration has
/// elapsed since `marked_at`.
pub const DEFAULT_GRACE_HOURS: u64 = 24;

/// Decision returned by [`check_grace_window`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum GraceDecision {
    /// `marked_at` is recent enough that the tombstone must remain.
    Within,
    /// `marked_at` is older than `grace_hours`; the sweep may proceed.
    Past,
}

/// Format `now` as an RFC 3339 string and wrap the underlying
/// formatter error in a [`PackchainError::Io`]. Centralises the
/// `OffsetDateTime::format` + `map_err` shape previously inlined at
/// the two tombstone-write paths in [`write_baseline_tombstone`] and
/// [`mark`].
fn rfc3339_now() -> Result<String, PackchainError> {
    OffsetDateTime::now_utc().format(&Rfc3339).map_err(|e| {
        PackchainError::Io(std::io::Error::other(format!("rfc3339 format failed: {e}")))
    })
}

/// Parse `marked_at` as RFC 3339 and decide whether `now - marked_at`
/// has crossed `grace_hours`. A negative age (tombstone marked in the
/// future under operator clock skew) is treated as
/// [`GraceDecision::Within`] so a sweep does not run prematurely.
///
/// `kind` is interpolated into both the parse-error message and the
/// `debug!` log line so the two sweep call paths (pack tombstones vs.
/// baseline tombstones) stay distinguishable in operator logs.
fn check_grace_window(
    marked_at: &str,
    grace_hours: u64,
    kind: &'static str,
) -> Result<GraceDecision, PackchainError> {
    let marked_at_ts = OffsetDateTime::parse(marked_at, &Rfc3339).map_err(|e| {
        PackchainError::Io(std::io::Error::other(format!(
            "{kind} marked_at parse failed: {e}"
        )))
    })?;
    let age_hours = (OffsetDateTime::now_utc() - marked_at_ts).whole_hours();
    // Negative age = a tombstone marked in the future (operator clock
    // skew). Treat as "still within grace" rather than sweeping
    // prematurely. The `try_into` is the canonical way to compare an
    // `i64` against an unsigned grace window without a sign-loss cast.
    let within = age_hours
        .try_into()
        .map_or(true, |hours: u64| hours < grace_hours);
    Ok(if within {
        GraceDecision::Within
    } else {
        GraceDecision::Past
    })
}

/// Best-effort version of [`write_baseline_tombstone`]: writes the
/// tombstone and, on error, logs at `warn` with the orphan key and
/// `source` discriminator (`"force-push"` / `"compact"`). Used by
/// callers that run AFTER `chain.json` is durable, where a tombstone
/// failure must NOT propagate as a push/compact failure — retrying the
/// caller would short-circuit through `AlreadyMinimal` (compact) or
/// the no-op same-SHA branch (push) and never re-attempt the cleanup,
/// leaving the orphaned bundle without a tombstone.
///
/// Returns `true` iff a tombstone was successfully written. Callers
/// that emit a success-only debug trace check the return value;
/// `false` covers both the same-SHA short-circuit and a warned-on
/// write error.
pub(crate) async fn write_baseline_tombstone_best_effort(
    store: &dyn ObjectStore,
    prefix: Option<&str>,
    ref_name: &RefName,
    prior_full_sha: &Sha40,
    current_full_sha: &Sha40,
    source: &'static str,
) -> bool {
    match write_baseline_tombstone(store, prefix, ref_name, prior_full_sha, current_full_sha).await
    {
        Ok(()) => prior_full_sha != current_full_sha,
        Err(e) => {
            let orphan_key = keys::bundle_key(prefix, ref_name.as_str(), prior_full_sha.as_str());
            warn!(
                source,
                ref_path = %ref_name.as_str(),
                key = %orphan_key,
                error = %e,
                "baseline tombstone write failed (chain.json already committed); \
                 orphan bundle left for manual cleanup",
            );
            false
        }
    }
}

/// Environment variable that overrides [`DEFAULT_GRACE_HOURS`] when
/// set to a positive integer. Mirrors the shape of
/// `GIT_REMOTE_OBJECT_STORE_LOCK_TTL_SECONDS` used by the protocol REPL.
pub(crate) const ENV_GC_GRACE_HOURS: &str = "GIT_REMOTE_OBJECT_STORE_GC_GRACE_HOURS";

/// On-bucket schema version this build reads and writes.
pub const TOMBSTONE_SCHEMA_VERSION: u32 = 1;

/// Reject a parsed schema version that does not match
/// [`TOMBSTONE_SCHEMA_VERSION`]. Shared by the two tombstone parsers
/// ([`Tombstone::from_json_bytes`] and [`BaselineTombstone::from_json_bytes`])
/// so both report the same `UnsupportedSchemaVersion` shape against
/// the same constant.
fn check_tombstone_schema_version(found: u32) -> Result<(), PackchainError> {
    if found == TOMBSTONE_SCHEMA_VERSION {
        Ok(())
    } else {
        Err(PackchainError::UnsupportedSchemaVersion {
            found,
            expected: TOMBSTONE_SCHEMA_VERSION,
        })
    }
}

/// On-bucket tombstone — a record of one mark phase's orphan set.
///
/// Lives at `<prefix>/gc/tombstones-<run_id>-<rfc3339>.json`. The
/// timestamp in the filename is for human inspection; the
/// authoritative `marked_at` is the field inside the JSON body.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct Tombstone {
    /// Schema version. Always [`TOMBSTONE_SCHEMA_VERSION`] when written.
    pub(crate) v: u32,
    /// `UUIDv4` run identifier. Two concurrent `gc` runs each get a
    /// distinct id, so their tombstone keys don't clobber.
    pub(crate) run_id: String,
    /// RFC 3339 timestamp at which the mark phase produced this set.
    /// Sweep compares this against the grace window.
    pub(crate) marked_at: String,
    /// Content-shas of orphan packs at mark time. Sweep re-checks
    /// each against the current chain state before deleting.
    pub(crate) orphan_packs: Vec<Sha40>,
}

impl Tombstone {
    /// Parse `bytes` as a tombstone JSON, validating the schema
    /// version before returning.
    ///
    /// # Errors
    ///
    /// - [`PackchainError::ParseJson`] for malformed JSON / missing
    ///   fields / `Sha40` validation failures.
    /// - [`PackchainError::UnsupportedSchemaVersion`] when `v` is not
    ///   [`TOMBSTONE_SCHEMA_VERSION`].
    pub(crate) fn from_json_bytes(bytes: &[u8]) -> Result<Self, PackchainError> {
        let parsed: Self = serde_json::from_slice(bytes)?;
        check_tombstone_schema_version(parsed.v)?;
        Ok(parsed)
    }

    /// Render to pretty-printed JSON bytes.
    ///
    /// # Errors
    ///
    /// `serde_json::to_vec_pretty` is infallible for this schema
    /// today, but the function returns `Result` for forward
    /// compatibility with future fields.
    pub(crate) fn to_json_pretty(&self) -> Result<Vec<u8>, PackchainError> {
        Ok(serde_json::to_vec_pretty(self)?)
    }
}

/// On-bucket tombstone for a superseded baseline bundle (issues #134, #143).
///
/// Lives at `<prefix>/gc/baseline-tomb-<uuid>.json`. Written by
/// [`super::compact`], [`super::push`], and the management
/// `delete-branch` flow whenever a chain rewrite or ref removal makes
/// a `<prefix>/<ref>/<sha>.bundle` unreachable. Unlike pack
/// tombstones the body names a specific (ref, sha) — there is exactly
/// one bundle key per record — so [`sweep`] does not need to re-derive
/// an orphan set from listings.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct BaselineTombstone {
    /// Schema version. Always [`TOMBSTONE_SCHEMA_VERSION`] when written.
    pub(crate) v: u32,
    /// RFC 3339 timestamp at which the tombstone was written. Sweep
    /// compares this against the grace window.
    pub(crate) marked_at: String,
    /// Ref the orphaned bundle belonged to (e.g. `refs/heads/main`).
    /// Stored as a raw string for forward compatibility with whatever
    /// `RefName` accepts at sweep time.
    pub(crate) ref_name: String,
    /// Content-SHA of the bundle (matches the `<sha>.bundle` filename).
    /// Sweep skips the delete when the ref's current `chain.full_at`
    /// equals this SHA (a later push re-baselined to the same tip).
    pub(crate) sha: Sha40,
}

impl BaselineTombstone {
    /// Parse `bytes` as a baseline tombstone JSON, validating the
    /// schema version before returning.
    ///
    /// # Errors
    ///
    /// - [`PackchainError::ParseJson`] for malformed JSON / missing
    ///   fields / `Sha40` validation failures.
    /// - [`PackchainError::UnsupportedSchemaVersion`] when `v` is not
    ///   [`TOMBSTONE_SCHEMA_VERSION`].
    pub(crate) fn from_json_bytes(bytes: &[u8]) -> Result<Self, PackchainError> {
        let parsed: Self = serde_json::from_slice(bytes)?;
        check_tombstone_schema_version(parsed.v)?;
        Ok(parsed)
    }

    /// Render to pretty-printed JSON bytes.
    pub(crate) fn to_json_pretty(&self) -> Result<Vec<u8>, PackchainError> {
        Ok(serde_json::to_vec_pretty(self)?)
    }
}

/// Write a baseline tombstone for the bundle at
/// `<prefix>/<ref_name>/<sha>.bundle` (issue #134).
///
/// Called from [`super::compact`] and [`super::push`] after the new
/// `chain.json` is durable — at that point the bundle has no chain
/// reference and is eligible for deletion, but a fetch that loaded
/// the prior chain may still be about to GET it. The tombstone
/// defers the delete to the next `gc sweep` past the grace window.
///
/// `prior_full_sha` is the SHA of the superseded baseline; `current_full_sha`
/// is the new chain's `full_at`. When they are equal the function
/// returns without writing a tombstone — the keys alias the same live
/// bundle (compact left `full_at` unchanged, or force-push targeted
/// the same tip).
///
/// # Errors
///
/// Returns [`PackchainError::Store`] on a PUT failure. Callers run
/// this AFTER `chain.json` is committed and must treat the failure as
/// best-effort: log a warning and report success, since retrying
/// would short-circuit through `AlreadyMinimal` and never re-attempt
/// the cleanup.
pub(crate) async fn write_baseline_tombstone(
    store: &dyn ObjectStore,
    prefix: Option<&str>,
    ref_name: &RefName,
    prior_full_sha: &Sha40,
    current_full_sha: &Sha40,
) -> Result<(), PackchainError> {
    if prior_full_sha == current_full_sha {
        return Ok(());
    }
    write_baseline_tombstone_unconditional(store, prefix, ref_name, prior_full_sha).await
}

/// Write a baseline tombstone naming `orphan_sha` regardless of any
/// successor SHA — used by `delete-branch` (issue #143), which
/// removes the chain entirely and has no replacement baseline to
/// compare against. Otherwise identical in shape to
/// [`write_baseline_tombstone`]: the body is parsed by
/// [`sweep_one_baseline_tombstone`], which sees a `chain.json`-less
/// ref (the synchronous sweep deletes it) and proceeds with the
/// deferred bundle delete after the grace window.
///
/// # Errors
///
/// Returns [`PackchainError::Store`] on a PUT failure. Callers run
/// this BEFORE the synchronous sweep that removes the rest of the
/// ref's objects; a failure here should fall back to immediate
/// bundle deletion so the operator's "ref is gone" intent is still
/// satisfied, rather than leaving a half-tombstoned half-deleted
/// state behind.
pub(crate) async fn write_baseline_tombstone_for_orphan(
    store: &dyn ObjectStore,
    prefix: Option<&str>,
    ref_name: &RefName,
    orphan_sha: &Sha40,
) -> Result<(), PackchainError> {
    write_baseline_tombstone_unconditional(store, prefix, ref_name, orphan_sha).await
}

/// Attempt to tombstone the baseline bundle for a delete so the
/// synchronous sweep loop can skip it (issue #143 / #203).
/// Returns the bundle key that was deferred, or `None` when deferral
/// is not actionable (bundle-engine ref with no `chain.json`,
/// unparseable `chain.json`, no `<full_at>.bundle` in the listing, or
/// a tombstone PUT failure).
///
/// The `None` fall-through is the correct behaviour for every
/// "deferral is not actionable" case: with no tombstone, `gc sweep`
/// has nothing to reclaim, so the sweep loop must remove the bundle
/// synchronously instead. A logged warning surfaces the rarer
/// load/parse/PUT failures for operator review without blocking the
/// delete.
///
/// Runs UNDER the per-ref lock (#158): a concurrent push that landed
/// between the tombstone and the chain.json delete would otherwise
/// leave the bucket with a tombstone referencing a SHA no longer in
/// the chain, and `gc sweep` would reclaim a live bundle.
///
/// `log_context` discriminates the warn-event source for log
/// scraping (`"packchain delete"`, `"delete-branch"`, etc.).
///
/// Replaces the previous per-call-site clones in
/// `manage::branch::ManageBranch::try_tombstone_baseline` and
/// `packchain::push::try_tombstone_baseline_for_delete` (#221).
pub(crate) async fn try_write_baseline_tombstone(
    store: &dyn ObjectStore,
    prefix: Option<&str>,
    remote_ref: &RefName,
    fresh: &[crate::object_store::ObjectMeta],
    log_context: &'static str,
) -> Option<String> {
    let chain = match load_chain(store, prefix, remote_ref).await {
        Ok(Some(chain)) => chain,
        Ok(None) => return None,
        Err(err) => {
            warn!(
                source = log_context,
                ref_path = %remote_ref.as_str(),
                error = %err,
                "chain.json read/parse failed; falling back to synchronous bundle delete",
            );
            return None;
        }
    };
    let bundle_key = keys::bundle_key(prefix, remote_ref.as_str(), chain.full_at.as_str());
    // The baseline bundle must actually be in the under-lock listing
    // — otherwise the deferred delete has nothing to defer (it was
    // already gone, or the chain points outside our prefix). A
    // mismatched `full_at` against listing reality is the canonical
    // "chain.json points at a missing bundle" doctor case; immediate
    // sweep is the right fallback there too.
    if !fresh.iter().any(|m| m.key == bundle_key) {
        return None;
    }
    match write_baseline_tombstone_for_orphan(store, prefix, remote_ref, &chain.full_at).await {
        Ok(()) => Some(bundle_key),
        Err(err) => {
            warn!(
                source = log_context,
                ref_path = %remote_ref.as_str(),
                key = %bundle_key,
                error = %err,
                "baseline tombstone write failed; falling back to synchronous bundle delete",
            );
            None
        }
    }
}

/// Return the set of bundle keys (full `<prefix>/<ref>/<sha>.bundle`
/// paths) currently named by any baseline tombstone under
/// `<prefix>/gc/baseline-tomb-*.json`.
///
/// Issue #157: the bundle engine derives `list`'s per-ref `<sha>`
/// from the bundle keys themselves (unlike packchain, which reads
/// `chain.tip`). Once a force-push tombstones the prior bundle, the
/// listing must hide that bundle so:
///
/// 1. The `list` wire output does not advertise two SHAs for the
///    same ref (which would emit two `<sha> <ref>\n` lines and
///    confuse git), and
/// 2. The under-lock multi-bundle guard in
///    `crate::protocol::push::perform_push_under_lock` does not
///    refuse the next push with the "multiple bundles" wire error.
///
/// The bundle keys themselves remain readable at their original
/// paths — fetchers that already advertised the tombstoned SHA
/// (issue #157's race) complete normally. `gc sweep` reclaims them
/// after the grace window.
///
/// Returns full bucket keys, not stems, so callers can compare
/// directly against `ObjectMeta::key` without re-deriving the prefix.
/// A tombstone whose `ref_name` no longer parses as a `RefName` is
/// skipped with a warn (mirrors `sweep_one_baseline_tombstone`'s
/// invalid-ref handling) — the bundle key is unknowable but the
/// tombstone itself remains for operator review.
///
/// # Errors
///
/// Propagates [`ObjectStoreError`] from the listing call. Per-tombstone
/// parse failures are skipped with a warn — a single malformed
/// tombstone must not block every listing.
pub(crate) async fn tombstoned_bundle_keys(
    store: &dyn ObjectStore,
    prefix: Option<&str>,
) -> Result<HashSet<String>, ObjectStoreError> {
    let prefix_str = prefix.unwrap_or("");
    let gc_listing = gc_listing_prefix(prefix_str);
    let metas = match store.list(&gc_listing).await {
        Ok(m) => m,
        // NotFound on an unsupported listing prefix on a fresh bucket
        // is normal — no tombstones to hide.
        Err(ObjectStoreError::NotFound(_)) => return Ok(HashSet::new()),
        Err(e) => return Err(e),
    };
    let mut keys = HashSet::new();
    for meta in metas {
        if !is_baseline_tombstone_key(&meta.key, prefix_str) {
            continue;
        }
        let body = match store.get_bytes(&meta.key).await {
            Ok(b) => b,
            Err(ObjectStoreError::NotFound(_)) => continue, // raced delete
            Err(e) => return Err(e),
        };
        let tombstone = match BaselineTombstone::from_json_bytes(&body) {
            Ok(t) => t,
            Err(e) => {
                warn!(
                    key = %meta.key,
                    error = %e,
                    "tombstoned_bundle_keys: skipping unparseable baseline tombstone",
                );
                continue;
            }
        };
        let Ok(ref_name) = RefName::new(tombstone.ref_name.clone()) else {
            warn!(
                key = %meta.key,
                ref_name = %tombstone.ref_name,
                "tombstoned_bundle_keys: skipping tombstone with invalid ref_name",
            );
            continue;
        };
        keys.insert(keys::bundle_key(prefix, &ref_name, tombstone.sha.as_str()));
    }
    Ok(keys)
}

/// Shared body of [`write_baseline_tombstone`] and
/// [`write_baseline_tombstone_for_orphan`]: emit a
/// `<prefix>/gc/baseline-tomb-<uuid>.json` record naming
/// `(ref_name, orphan_sha)`. Centralised so the two call shapes
/// (with-successor and unconditional) cannot drift on the JSON body
/// shape or the key namespace.
async fn write_baseline_tombstone_unconditional(
    store: &dyn ObjectStore,
    prefix: Option<&str>,
    ref_name: &RefName,
    orphan_sha: &Sha40,
) -> Result<(), PackchainError> {
    let marked_at = rfc3339_now()?;
    let tombstone = BaselineTombstone {
        v: TOMBSTONE_SCHEMA_VERSION,
        marked_at,
        ref_name: ref_name.as_str().to_owned(),
        sha: orphan_sha.clone(),
    };
    let key = baseline_tombstone_key(prefix.unwrap_or(""), &Uuid::new_v4().to_string());
    let body = Bytes::from(tombstone.to_json_pretty()?);
    store.put_bytes(&key, body, PutOpts::default()).await?;
    debug!(
        key = %key,
        ref_path = %ref_name.as_str(),
        sha = %orphan_sha.as_str(),
        "gc: baseline tombstone written",
    );
    Ok(())
}

/// Outcome of [`mark`].
#[derive(Debug, Clone)]
pub struct MarkOutcome {
    /// `UUIDv4` run id assigned to this mark pass. Embedded in the
    /// tombstone filename and body.
    pub run_id: String,
    /// Number of orphan packs identified.
    pub orphan_count: usize,
    /// Bucket key the tombstone was written to.
    pub tombstone_key: String,
}

/// Outcome of [`sweep`].
#[derive(Debug, Clone, Default)]
pub struct SweepOutcome {
    /// Tombstones whose packs were deleted (and which were themselves
    /// deleted as a result).
    pub swept_tombstones: usize,
    /// Tombstones still inside the grace window — left for the next
    /// sweep.
    pub deferred_tombstones: usize,
    /// Pack file deletions executed (counts both `.pack` and `.idx`
    /// deletions, so two per orphan in the typical case).
    pub deleted_objects: usize,
    /// Tombstoned packs that were no longer orphan at sweep time
    /// (re-referenced between mark and sweep, or deleted by an
    /// earlier sweep). Skipped without error.
    pub skipped_repointed_packs: usize,
}

/// Knobs for [`mark`].
#[derive(Debug, Clone, Copy, Default)]
pub struct MarkOpts {
    /// When `true`, list and report but do not write a tombstone file
    /// or modify the bucket. Used by `doctor` to surface orphan stats.
    pub dry_run: bool,
}

/// Knobs for [`sweep`].
#[derive(Debug, Clone, Copy)]
pub struct SweepOpts {
    /// Grace duration in hours. Tombstones with `marked_at` younger
    /// than this stay deferred. Ignored when `force` is `true`.
    pub grace_hours: u64,
    /// When `true`, skip the grace check. The live-pack re-derive
    /// still runs — a tombstone whose SHA is now referenced by a
    /// committed chain is left alone (closes the mark/commit race
    /// from #117). The grace window is the only safety check this
    /// flag suppresses; concurrent fetches that still hold a SHA in
    /// flight are NOT protected by either path.
    pub force: bool,
}

impl Default for SweepOpts {
    fn default() -> Self {
        Self {
            grace_hours: DEFAULT_GRACE_HOURS,
            force: false,
        }
    }
}

/// Read [`DEFAULT_GRACE_HOURS`] subject to the
/// [`ENV_GC_GRACE_HOURS`] override. Returns the default for unset
/// vars, non-numeric values, or zero (a zero grace would defeat the
/// mark/sweep design's point).
#[must_use]
pub(crate) fn grace_hours_from_env() -> u64 {
    std::env::var(ENV_GC_GRACE_HOURS)
        .ok()
        .and_then(|v| v.parse::<u64>().ok())
        .filter(|h| *h > 0)
        .unwrap_or(DEFAULT_GRACE_HOURS)
}

/// Resolve a caller-supplied `Option<u64>` grace-hours value to a
/// concrete count, deferring to [`grace_hours_from_env`] when the
/// caller passes `None` (#221).
///
/// Unlike [`crate::protocol::push::resolve_lock_ttl_seconds`], this
/// resolver does **not** clamp `Some(0)`. A zero grace window is a
/// legitimate operator intent — "sweep without a grace window, e.g.
/// in force mode" — and is the canonical value used by force-sweep
/// tests (`SweepOpts { grace_hours: 0, force: true }`). The lock-TTL
/// clamp protects against `acquire_lock` treating every held lock as
/// instantly stale (#208); grace-hours has no analogous foot-gun.
#[must_use]
pub(crate) fn resolve_grace_hours(opt: Option<u64>) -> u64 {
    opt.unwrap_or_else(grace_hours_from_env)
}

/// Run the mark phase: snapshot every pack on the bucket, then every
/// chain, then write a tombstone naming the orphans.
///
/// `prefix` is the repository prefix without leading or trailing
/// slashes — pass an empty string for bucket-root repositories.
///
/// # Ordering
///
/// Listings run packs-first, chains-second (issue #135). The reverse
/// order races a concurrent push that uploads a new pack between the
/// two listings and commits its `chain.json` *between the chain list
/// and the pack list*: the new pack would appear in the on-bucket set
/// but not in the referenced set, producing a false-positive tombstone.
/// Packs-first inverts the staleness: the referenced set is always at
/// least as fresh as the on-bucket set, so a pack appearing in the
/// snapshot is either also in the chain set (saved) or genuinely
/// orphan at some point during the mark (correctly tombstoned, with
/// the grace window covering in-flight pushes).
///
/// # Errors
///
/// - Any chain.json that fails to parse aborts the mark with
///   [`PackchainError::ParseJson`] / [`PackchainError::InvalidSha`] /
///   [`PackchainError::UnsupportedSchemaVersion`]. The tombstone is
///   not written. Operators must repair the bad chain (or remove it)
///   before re-running.
/// - [`PackchainError::Store`] / [`PackchainError::Io`] for transport
///   or local-I/O failures.
///
/// # Example
///
/// ```no_run
/// # #[tokio::main] async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// use git_remote_object_store::Remote;
/// use git_remote_object_store::packchain::gc::{MarkOpts, mark};
///
/// let remote = Remote::connect("s3+https://bucket/repo?engine=packchain").await?;
/// let outcome = mark(remote.store(), remote.prefix(), MarkOpts::default()).await?;
/// println!(
///     "{} orphan pack(s) tombstoned (run id {})",
///     outcome.orphan_count, outcome.run_id,
/// );
/// # Ok(())
/// # }
/// ```
pub async fn mark(
    store: &dyn ObjectStore,
    prefix: &str,
    opts: MarkOpts,
) -> Result<MarkOutcome, PackchainError> {
    // Packs-first ordering (issue #135): the on-bucket snapshot must
    // be at least as stale as the referenced set. A pack uploaded by a
    // concurrent push between these two listings is harmless — it
    // either is not in `on_bucket` (uploaded after the pack list) or
    // is in `referenced` (uploaded before the pack list AND its
    // chain.json committed before the chain list). The reverse order
    // produces false-positive tombstones; see the module docstring.
    let on_bucket = list_pack_shas(store, prefix).await?;
    let referenced = list_referenced_packs(store, prefix).await?;
    let orphans: Vec<Sha40> = on_bucket
        .into_iter()
        .filter(|sha| !referenced.contains(sha))
        .collect();

    let run_id = Uuid::new_v4().to_string();
    let marked_at = rfc3339_now()?;
    let tombstone_key = tombstone_key(prefix, &run_id, &marked_at);
    let orphan_count = orphans.len();
    let tombstone = Tombstone {
        v: TOMBSTONE_SCHEMA_VERSION,
        run_id: run_id.clone(),
        marked_at,
        orphan_packs: orphans,
    };
    let outcome = MarkOutcome {
        run_id,
        orphan_count,
        tombstone_key,
    };

    if opts.dry_run {
        debug!(
            run_id = %outcome.run_id,
            orphans = outcome.orphan_count,
            "gc mark: dry-run, not writing tombstone",
        );
        return Ok(outcome);
    }

    if outcome.orphan_count == 0 {
        info!(run_id = %outcome.run_id, "gc mark: no orphans; skipping tombstone");
        return Ok(outcome);
    }

    let body = Bytes::from(tombstone.to_json_pretty()?);
    store
        .put_bytes(&outcome.tombstone_key, body, PutOpts::default())
        .await?;
    info!(
        run_id = %outcome.run_id,
        orphans = outcome.orphan_count,
        key = %outcome.tombstone_key,
        "gc mark: tombstone written",
    );
    Ok(outcome)
}

/// Run the sweep phase: walk tombstones, delete eligible orphans.
///
/// `prefix` and the threading semantics match [`mark`].
///
/// # Errors
///
/// Sweep is best-effort: a single tombstone failure does not abort
/// the run (errors are logged and the next tombstone is tried).
/// Returns [`PackchainError::Store`] only when the initial
/// tombstone-list call fails.
///
/// # Example
///
/// ```no_run
/// # #[tokio::main] async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// use git_remote_object_store::Remote;
/// use git_remote_object_store::packchain::gc::{SweepOpts, sweep};
///
/// let remote = Remote::connect("s3+https://bucket/repo?engine=packchain").await?;
/// let outcome = sweep(
///     remote.store(),
///     remote.prefix(),
///     SweepOpts::default(),
/// )
/// .await?;
/// println!(
///     "swept {} tombstone(s), deleted {} object(s), deferred {}",
///     outcome.swept_tombstones,
///     outcome.deleted_objects,
///     outcome.deferred_tombstones,
/// );
/// # Ok(())
/// # }
/// ```
pub async fn sweep(
    store: &dyn ObjectStore,
    prefix: &str,
    opts: SweepOpts,
) -> Result<SweepOutcome, PackchainError> {
    let tombstones_prefix = gc_listing_prefix(prefix);
    let metas = store.list(&tombstones_prefix).await?;
    let mut outcome = SweepOutcome::default();

    if opts.force {
        warn!("gc sweep: --force in effect; skipping grace window");
    }

    for meta in metas {
        if !meta.key.as_bytes().ends_with(b".json") {
            continue;
        }
        let step = if is_tombstone_key(&meta.key, prefix) {
            sweep_one_tombstone(store, prefix, &meta.key, opts).await
        } else if is_baseline_tombstone_key(&meta.key, prefix) {
            sweep_one_baseline_tombstone(store, prefix, &meta.key, opts).await
        } else {
            continue;
        };
        match step {
            Ok(SweepStep::Deferred) => outcome.deferred_tombstones += 1,
            Ok(SweepStep::Swept {
                deleted_objects,
                skipped_repointed_packs,
            }) => {
                outcome.swept_tombstones += 1;
                outcome.deleted_objects += deleted_objects;
                outcome.skipped_repointed_packs += skipped_repointed_packs;
            }
            Err(e) => {
                warn!(key = %meta.key, error = %e, "gc sweep: tombstone failed");
            }
        }
    }
    Ok(outcome)
}

#[derive(Debug)]
enum SweepStep {
    Deferred,
    Swept {
        deleted_objects: usize,
        skipped_repointed_packs: usize,
    },
}

async fn sweep_one_tombstone(
    store: &dyn ObjectStore,
    prefix: &str,
    tombstone_key: &str,
    opts: SweepOpts,
) -> Result<SweepStep, PackchainError> {
    let body = match store.get_bytes(tombstone_key).await {
        Ok(b) => b,
        Err(ObjectStoreError::NotFound(_)) => {
            // Concurrent sweep already cleaned this up.
            return Ok(SweepStep::Swept {
                deleted_objects: 0,
                skipped_repointed_packs: 0,
            });
        }
        Err(e) => return Err(PackchainError::Store(e)),
    };
    let tombstone = Tombstone::from_json_bytes(&body)?;

    if !opts.force
        && check_grace_window(&tombstone.marked_at, opts.grace_hours, "tombstone")?
            == GraceDecision::Within
    {
        debug!(
            key = %tombstone_key,
            marked_at = %tombstone.marked_at,
            "gc sweep: tombstone within grace window",
        );
        return Ok(SweepStep::Deferred);
    }

    // Re-derive the live referenced set per pack delete, AFTER the
    // grace check passes — never cache across iterations (issue #140
    // for the cross-tombstone race, issue #152 for the cross-pack
    // race inside a single tombstone). A concurrent push committing
    // chain.json after a per-tombstone snapshot still loses a live
    // pack named later in the same tombstone's `orphan_packs` vector,
    // because `mark()` packs all orphans for a run into one tombstone
    // body. Force-revert is the canonical trigger: gix pack emission
    // is deterministic for the same object set, so the new pack key
    // aliases the tombstoned one and the push skips upload, only
    // touching chain.json. Per-pack re-listing costs one extra
    // `list("refs/")` + bounded-parallel chain GETs per orphan pack;
    // for the rare GC maintenance path this is acceptable overhead in
    // exchange for closing the cross-pack window. The recompute also
    // runs under --force: that flag suppresses the grace window only,
    // NOT this guard (issue #117). A residual TOCTOU window remains
    // between this listing and the delete that follows; the fix
    // shrinks the window from "one snapshot per tombstone" to "one
    // snapshot per pack", which is the bound the issue asks for.
    let mut deleted_objects = 0usize;
    let mut skipped_repointed_packs = 0usize;
    for sha in &tombstone.orphan_packs {
        // Always honour the live-pack guard, including under --force.
        // See the recompute comment above and issue #117 for why.
        let referenced = list_referenced_packs(store, prefix).await?;
        if referenced.contains(sha) {
            skipped_repointed_packs += 1;
            debug!(
                sha = %sha.as_str(),
                "gc sweep: tombstoned pack re-referenced; skipping",
            );
            continue;
        }
        let pack_key = super::keys::pack_key(Some(prefix), sha);
        let idx_key = super::keys::pack_idx_key(Some(prefix), sha);
        if delete_idempotent(store, &pack_key).await? {
            deleted_objects += 1;
        }
        if delete_idempotent(store, &idx_key).await? {
            deleted_objects += 1;
        }
    }
    // Drop the tombstone last so a sweep crash mid-deletion leaves a
    // tombstone the next sweep can finish.
    delete_idempotent(store, tombstone_key).await?;
    info!(
        key = %tombstone_key,
        deleted = deleted_objects,
        skipped = skipped_repointed_packs,
        "gc sweep: tombstone applied",
    );
    Ok(SweepStep::Swept {
        deleted_objects,
        skipped_repointed_packs,
    })
}

/// Sweep one baseline tombstone (issue #134). Parses the tombstone,
/// honours the grace window, re-checks the ref's current `chain.full_at`
/// to skip a re-baselined-to-same-SHA case, and then idempotently
/// deletes both the bundle and the tombstone.
///
/// The live-state recheck mirrors the pack sweep's
/// `referenced.contains` guard: a tombstone written by a force-push
/// can be invalidated by a subsequent force-push that lands on the
/// same SHA, in which case the bundle is once again live.
async fn sweep_one_baseline_tombstone(
    store: &dyn ObjectStore,
    prefix: &str,
    tombstone_key: &str,
    opts: SweepOpts,
) -> Result<SweepStep, PackchainError> {
    let body = match store.get_bytes(tombstone_key).await {
        Ok(b) => b,
        Err(ObjectStoreError::NotFound(_)) => {
            return Ok(SweepStep::Swept {
                deleted_objects: 0,
                skipped_repointed_packs: 0,
            });
        }
        Err(e) => return Err(PackchainError::Store(e)),
    };
    let tombstone = BaselineTombstone::from_json_bytes(&body)?;

    if !opts.force
        && check_grace_window(&tombstone.marked_at, opts.grace_hours, "baseline tombstone")?
            == GraceDecision::Within
    {
        debug!(
            key = %tombstone_key,
            marked_at = %tombstone.marked_at,
            "gc sweep: baseline tombstone within grace window",
        );
        return Ok(SweepStep::Deferred);
    }

    // Re-check the live chain. A subsequent push that re-baselined to
    // the same SHA (force-push at the same tip, or compact short-cut)
    // makes this bundle live again — leave it alone, drop the now-stale
    // tombstone. A missing ref (chain deleted) means the bundle is also
    // unreachable; proceed with the delete.
    let ref_name = match RefName::new(tombstone.ref_name.clone()) {
        Ok(r) => r,
        Err(e) => {
            // Issue #146: a tombstone whose recorded ref_name no longer
            // parses as a RefName cannot be turned into a bundle key, so
            // deleting the tombstone would orphan the bundle with no
            // record on the bucket. Preserve both records and surface
            // the ref_name + tombstone key at error! so an operator can
            // locate and reconcile the corruption manually. Reachable
            // today only via manual bucket tampering or a future schema
            // tweak that loosens what BaselineTombstone::ref_name
            // accepts at write time relative to RefName::new at read
            // time.
            error!(
                key = %tombstone_key,
                ref_name = %tombstone.ref_name,
                sha = %tombstone.sha.as_str(),
                error = %e,
                "gc sweep: baseline tombstone names invalid ref; preserving \
                 tombstone and bundle for operator review",
            );
            return Ok(SweepStep::Deferred);
        }
    };
    let prefix_opt = (!prefix.is_empty()).then_some(prefix);
    let chain = load_chain(store, prefix_opt, &ref_name).await?;
    let mut skipped_repointed_packs = 0usize;
    let mut deleted_objects = 0usize;
    let still_live = chain.as_ref().is_some_and(|c| c.full_at == tombstone.sha);
    if still_live {
        skipped_repointed_packs += 1;
        debug!(
            key = %tombstone_key,
            ref_path = %ref_name.as_str(),
            sha = %tombstone.sha.as_str(),
            "gc sweep: baseline re-referenced; skipping delete",
        );
    } else {
        // Issue #153: re-read the chain IMMEDIATELY before deleting the
        // bundle. The earlier `load_chain` above closes the common
        // stale-tombstone path, but a concurrent force-push or compact
        // can re-baseline the ref to the tombstoned SHA between that
        // read and the delete that follows; without this recheck, sweep
        // would erase a now-live bundle and then drop the tombstone.
        // The window between this recheck and the bundle delete is
        // bounded by a single network round-trip — the same bound the
        // pack-tombstone path uses (issues #140, #152). When the
        // recheck catches the race, return `Deferred` so the tombstone
        // is preserved for a future sweep (mirrors #146's
        // operator-review pattern) — the chain may flip back to a
        // different SHA before then, in which case the tombstone
        // becomes actionable again.
        let recheck = load_chain(store, prefix_opt, &ref_name).await?;
        if recheck.as_ref().is_some_and(|c| c.full_at == tombstone.sha) {
            debug!(
                key = %tombstone_key,
                ref_path = %ref_name.as_str(),
                sha = %tombstone.sha.as_str(),
                "gc sweep: baseline re-referenced between checks; deferring",
            );
            return Ok(SweepStep::Deferred);
        }
        let bundle_key = keys::bundle_key(prefix_opt, &ref_name, tombstone.sha.as_str());
        if delete_idempotent(store, &bundle_key).await? {
            deleted_objects += 1;
        }
    }
    // Drop the tombstone last so a crash mid-delete leaves it for the
    // next sweep to finish.
    delete_idempotent(store, tombstone_key).await?;
    info!(
        key = %tombstone_key,
        deleted = deleted_objects,
        skipped = skipped_repointed_packs,
        "gc sweep: baseline tombstone applied",
    );
    Ok(SweepStep::Swept {
        deleted_objects,
        skipped_repointed_packs,
    })
}

/// `<prefix>/gc/` prefix for [`ObjectStore::list`]. Empty `prefix`
/// drops the leading slash (matches the project's bucket-root rule).
fn gc_listing_prefix(prefix: &str) -> String {
    keys::join(Some(prefix), "gc/")
}

/// Build a tombstone key. The `marked_at` segment may contain `:`
/// characters; S3 / Azure both accept colons in keys.
fn tombstone_key(prefix: &str, run_id: &str, marked_at: &str) -> String {
    keys::join(
        Some(prefix),
        &format!("gc/tombstones-{run_id}-{marked_at}.json"),
    )
}

/// Key-namespace fragment for baseline tombstones (issue #134).
/// Composed with a bucket prefix via [`keys::join`] / [`baseline_tombstone_listing_prefix`]
/// to form the full listable prefix; the UUID-suffixed body filename
/// is appended by [`baseline_tombstone_key`].
///
/// Single source of truth for the on-bucket key shape — production
/// builders and test assertions both compose against this constant
/// rather than embedding the literal string (#221).
pub(crate) const BASELINE_TOMBSTONE_KEY_FRAGMENT: &str = "gc/baseline-tomb-";

/// Build a baseline tombstone key. UUID-keyed so concurrent compacts
/// / force-pushes across different refs never clobber, and the
/// timestamp lives in the body rather than the filename to keep the
/// `is_baseline_tombstone_key` predicate cheap.
fn baseline_tombstone_key(prefix: &str, run_id: &str) -> String {
    keys::join(
        Some(prefix),
        &format!("{BASELINE_TOMBSTONE_KEY_FRAGMENT}{run_id}.json"),
    )
}

/// Listable prefix for every baseline tombstone under `prefix`
/// (e.g. `"repo/gc/baseline-tomb-"`). Composes
/// [`BASELINE_TOMBSTONE_KEY_FRAGMENT`] with the bucket prefix via
/// [`keys::join`] so callers don't open-code the literal.
pub(crate) fn baseline_tombstone_listing_prefix(prefix: Option<&str>) -> String {
    keys::join(prefix, BASELINE_TOMBSTONE_KEY_FRAGMENT)
}

/// Robust check that `key` is a tombstone under our prefix. Guards
/// against unrelated `.json` files in `<prefix>/gc/` and against a
/// regression where a future schema rev moves the prefix.
///
/// Root-prefix (`prefix == ""`) case: `expected_prefix` is just
/// `"gc/tombstones-"`, so every `gc/tombstones-*.json` key at the
/// bucket root matches. That is the intended behaviour — a root
/// repo owns the entire `gc/` namespace.
fn is_tombstone_key(key: &str, prefix: &str) -> bool {
    let expected_prefix = keys::join(Some(prefix), "gc/tombstones-");
    key.starts_with(&expected_prefix)
}

/// Robust check that `key` is a baseline tombstone under our prefix
/// (issue #134). Mirrors [`is_tombstone_key`] for the
/// [`BASELINE_TOMBSTONE_KEY_FRAGMENT`] namespace.
fn is_baseline_tombstone_key(key: &str, prefix: &str) -> bool {
    key.starts_with(&baseline_tombstone_listing_prefix(Some(prefix)))
}

/// List every `<prefix>/refs/**/chain.json` (across every ref
/// namespace — `refs/heads/`, `refs/tags/`, `refs/notes/`, etc.) and
/// union the pack content-shas they reference. Fail closed on parse
/// error.
async fn list_referenced_packs(
    store: &dyn ObjectStore,
    prefix: &str,
) -> Result<HashSet<Sha40>, PackchainError> {
    let refs_prefix = keys::join(Some(prefix), "refs/");
    let metas = store.list(&refs_prefix).await?;

    // Bounded-parallel `get_bytes` per chain.json, parse-as-fetched.
    // Mirrors `list::list_refs` (#89 widened the listing prefix to
    // all `refs/` namespaces, so candidate count scales with branches
    // + tags + notes). `MAX_FETCH_CONCURRENCY` (= 8) is the same bound
    // Phase 3 fetch uses for chain pack downloads. `try_fold` folds
    // each body into the set as soon as `buffer_unordered` yields it,
    // so parse overlaps the next batch's fetch latency and no
    // intermediate `Vec<Bytes>` is held.
    //
    // Fail-closed semantics: a transport failure on any GET, or a
    // parse failure on any chain, aborts the run — the mark phase
    // cannot tombstone live packs because of an under-reporting
    // corrupt chain.
    futures::stream::iter(
        metas
            .into_iter()
            .filter(|m| super::keys::is_chain_json_key(&m.key))
            .map(|m| m.key),
    )
    .map(|key| async move { store.get_bytes(&key).await.map_err(PackchainError::Store) })
    .buffer_unordered(MAX_FETCH_CONCURRENCY)
    .try_fold(HashSet::<Sha40>::new(), |mut acc, body| async move {
        let chain = ChainManifest::from_json_bytes(&body)?;
        for segment in chain.segments {
            // gc fails closed on a malformed pack key — the chain is
            // corrupt and tombstoning live packs based on it would be
            // unsafe. Uses the same `MalformedPackEntry` variant as
            // every other consumer (read, fetch, compact) so error
            // wording stays aligned across the engine.
            let sha = super::keys::segment_pack_sha(&segment)?;
            acc.insert(sha);
        }
        Ok(acc)
    })
    .await
}

/// List every `<prefix>/packs/*.pack` and `*.idx` and return the union
/// of their content-shas. The set is keyed by sha so a pack with a
/// missing-but-tombstoneable idx still counts (and vice versa).
async fn list_pack_shas(
    store: &dyn ObjectStore,
    prefix: &str,
) -> Result<HashSet<Sha40>, PackchainError> {
    let packs_prefix = keys::join(Some(prefix), "packs/");
    let metas = store.list(&packs_prefix).await?;
    let mut shas: HashSet<Sha40> = HashSet::new();
    for meta in metas {
        let basename = meta
            .key
            .rsplit('/')
            .next()
            .expect("rsplit('/') on a non-empty key yields at least one element");
        let candidate = basename
            .strip_suffix(".pack")
            .or_else(|| basename.strip_suffix(".idx"));
        if let Some(sha) = candidate
            && let Ok(parsed) = Sha40::try_new(sha)
        {
            shas.insert(parsed);
        }
    }
    Ok(shas)
}

/// Best-effort delete: returns `Ok(true)` on a real delete, `Ok(false)`
/// when the object was already absent (concurrent sweep raced ahead,
/// or a partial sweep ran earlier).
async fn delete_idempotent(store: &dyn ObjectStore, key: &str) -> Result<bool, PackchainError> {
    match store.delete(key).await {
        Ok(()) => Ok(true),
        Err(ObjectStoreError::NotFound(_)) => Ok(false),
        Err(e) => Err(PackchainError::Store(e)),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::git::RefName;
    use crate::object_store::mock::MockStore;
    use crate::packchain::manifest::write_chain;
    use crate::packchain::schema::ChainSegment;

    const SHA_TIP: &str = "0000000000000000000000000000000000000001";
    const SHA_FULL: &str = "0000000000000000000000000000000000000002";
    const SHA_PACK_LIVE: &str = "1111111111111111111111111111111111111111";
    const SHA_PACK_ORPHAN: &str = "2222222222222222222222222222222222222222";
    const SHA_PACK_ORPHAN_2: &str = "3333333333333333333333333333333333333333";

    fn sha40(s: &str) -> Sha40 {
        Sha40::try_new(s).unwrap()
    }

    fn ref_main() -> RefName {
        RefName::new("refs/heads/main").unwrap()
    }

    fn segment(pack_sha: &str, parent: Option<&str>) -> ChainSegment {
        ChainSegment {
            sha: sha40(SHA_TIP),
            parent_sha: parent.map(sha40),
            pack: format!("packs/{pack_sha}.pack"),
            bytes: 1_024,
        }
    }

    async fn seed_live_chain(store: &MockStore, prefix: Option<&str>) {
        let chain = ChainManifest {
            v: 1,
            tip: sha40(SHA_TIP),
            full_at: sha40(SHA_FULL),
            segments: vec![segment(SHA_PACK_LIVE, None)],
        };
        write_chain(store, prefix, &ref_main(), &chain)
            .await
            .unwrap();
    }

    fn insert_pack_pair(store: &MockStore, prefix: Option<&str>, sha: &str) {
        let pack_key = super::super::keys::pack_key(prefix, &sha40(sha));
        let idx_key = super::super::keys::pack_idx_key(prefix, &sha40(sha));
        store.insert(pack_key, Bytes::from_static(b"PACKDATA"));
        store.insert(idx_key, Bytes::from_static(b"IDXDATA"));
    }

    // --- mark -----------------------------------------------------------

    #[tokio::test]
    async fn mark_with_no_chains_treats_all_packs_as_orphan() {
        let store = MockStore::new();
        insert_pack_pair(&store, Some("repo"), SHA_PACK_ORPHAN);
        let outcome = mark(&store, "repo", MarkOpts::default()).await.unwrap();
        assert_eq!(outcome.orphan_count, 1);
        // Tombstone written to the correct prefix.
        let body = store.get_bytes(&outcome.tombstone_key).await.unwrap();
        let parsed = Tombstone::from_json_bytes(&body).unwrap();
        assert_eq!(parsed.orphan_packs, vec![sha40(SHA_PACK_ORPHAN)]);
    }

    #[tokio::test]
    async fn mark_skips_chain_referenced_packs() {
        let store = MockStore::new();
        seed_live_chain(&store, Some("repo")).await;
        insert_pack_pair(&store, Some("repo"), SHA_PACK_LIVE);
        insert_pack_pair(&store, Some("repo"), SHA_PACK_ORPHAN);
        let outcome = mark(&store, "repo", MarkOpts::default()).await.unwrap();
        assert_eq!(outcome.orphan_count, 1);
        let body = store.get_bytes(&outcome.tombstone_key).await.unwrap();
        let parsed = Tombstone::from_json_bytes(&body).unwrap();
        assert_eq!(parsed.orphan_packs, vec![sha40(SHA_PACK_ORPHAN)]);
    }

    #[tokio::test]
    async fn mark_no_orphans_skips_tombstone_write() {
        let store = MockStore::new();
        seed_live_chain(&store, Some("repo")).await;
        insert_pack_pair(&store, Some("repo"), SHA_PACK_LIVE);
        let outcome = mark(&store, "repo", MarkOpts::default()).await.unwrap();
        assert_eq!(outcome.orphan_count, 0);
        // No tombstone listed.
        let metas = store.list("repo/gc/").await.unwrap();
        assert!(
            metas.is_empty(),
            "tombstone must not exist for empty orphan set"
        );
    }

    #[tokio::test]
    async fn mark_dry_run_does_not_write_tombstone() {
        let store = MockStore::new();
        insert_pack_pair(&store, Some("repo"), SHA_PACK_ORPHAN);
        let outcome = mark(&store, "repo", MarkOpts { dry_run: true })
            .await
            .unwrap();
        assert_eq!(outcome.orphan_count, 1);
        let metas = store.list("repo/gc/").await.unwrap();
        assert!(metas.is_empty(), "dry-run must not write tombstone");
    }

    #[tokio::test]
    async fn mark_treats_tag_chain_referenced_packs_as_live() {
        // A pack referenced only from a chain under refs/tags/ must
        // not be tombstoned. (Regression for issue #89.)
        let store = MockStore::new();
        let chain = ChainManifest {
            v: 1,
            tip: sha40(SHA_TIP),
            full_at: sha40(SHA_FULL),
            segments: vec![segment(SHA_PACK_LIVE, None)],
        };
        let tag_ref = RefName::new("refs/tags/v1").unwrap();
        write_chain(&store, Some("repo"), &tag_ref, &chain)
            .await
            .unwrap();
        insert_pack_pair(&store, Some("repo"), SHA_PACK_LIVE);
        insert_pack_pair(&store, Some("repo"), SHA_PACK_ORPHAN);

        let referenced = list_referenced_packs(&store, "repo").await.unwrap();
        assert!(
            referenced.contains(&sha40(SHA_PACK_LIVE)),
            "pack referenced from refs/tags/ chain must be in the live set",
        );

        let outcome = mark(&store, "repo", MarkOpts::default()).await.unwrap();
        assert_eq!(outcome.orphan_count, 1);
        let body = store.get_bytes(&outcome.tombstone_key).await.unwrap();
        let parsed = Tombstone::from_json_bytes(&body).unwrap();
        assert_eq!(parsed.orphan_packs, vec![sha40(SHA_PACK_ORPHAN)]);
    }

    #[tokio::test]
    async fn mark_treats_notes_chain_referenced_packs_as_live() {
        // refs/notes/commits is the standard git notes ref. A pack
        // referenced only from a notes chain must not be tombstoned.
        let store = MockStore::new();
        let chain = ChainManifest {
            v: 1,
            tip: sha40(SHA_TIP),
            full_at: sha40(SHA_FULL),
            segments: vec![segment(SHA_PACK_LIVE, None)],
        };
        let notes_ref = RefName::new("refs/notes/commits").unwrap();
        write_chain(&store, Some("repo"), &notes_ref, &chain)
            .await
            .unwrap();
        insert_pack_pair(&store, Some("repo"), SHA_PACK_LIVE);

        let referenced = list_referenced_packs(&store, "repo").await.unwrap();
        assert!(
            referenced.contains(&sha40(SHA_PACK_LIVE)),
            "pack referenced from refs/notes/ chain must be in the live set",
        );

        let outcome = mark(&store, "repo", MarkOpts::default()).await.unwrap();
        assert_eq!(outcome.orphan_count, 0);
    }

    #[tokio::test]
    async fn list_referenced_packs_unions_across_namespaces() {
        // A live chain in refs/heads/ AND in refs/tags/ both
        // contribute to the referenced set.
        let store = MockStore::new();
        let head_chain = ChainManifest {
            v: 1,
            tip: sha40(SHA_TIP),
            full_at: sha40(SHA_FULL),
            segments: vec![segment(SHA_PACK_LIVE, None)],
        };
        write_chain(&store, Some("repo"), &ref_main(), &head_chain)
            .await
            .unwrap();
        let tag_chain = ChainManifest {
            v: 1,
            tip: sha40(SHA_TIP),
            full_at: sha40(SHA_FULL),
            segments: vec![segment(SHA_PACK_ORPHAN_2, None)],
        };
        let tag_ref = RefName::new("refs/tags/v1").unwrap();
        write_chain(&store, Some("repo"), &tag_ref, &tag_chain)
            .await
            .unwrap();

        let referenced = list_referenced_packs(&store, "repo").await.unwrap();
        assert!(referenced.contains(&sha40(SHA_PACK_LIVE)));
        assert!(referenced.contains(&sha40(SHA_PACK_ORPHAN_2)));
        assert_eq!(referenced.len(), 2);
    }

    #[tokio::test]
    async fn list_referenced_packs_ignores_sibling_artefacts() {
        // path-index.json, .bundle baselines, and other artefacts
        // under refs/<namespace>/<name>/ must not be parsed as
        // chain.json.
        let store = MockStore::new();
        seed_live_chain(&store, Some("repo")).await;
        // Add sibling artefacts that share the ref directory.
        store.insert(
            "repo/refs/heads/main/path-index.json",
            Bytes::from_static(b"{}"),
        );
        store.insert(
            format!("repo/refs/heads/main/{SHA_TIP}.bundle"),
            Bytes::from_static(b"BUNDLE"),
        );
        // And a tombstone-style key under refs/ that must be filtered.
        store.insert(
            "repo/refs/tags/v1/path-index.json",
            Bytes::from_static(b"{}"),
        );

        let referenced = list_referenced_packs(&store, "repo").await.unwrap();
        assert_eq!(referenced.len(), 1);
        assert!(referenced.contains(&sha40(SHA_PACK_LIVE)));
    }

    #[tokio::test]
    async fn list_referenced_packs_empty_for_no_chains() {
        let store = MockStore::new();
        let referenced = list_referenced_packs(&store, "repo").await.unwrap();
        assert!(referenced.is_empty());
    }

    #[tokio::test]
    async fn list_referenced_packs_unions_many_chains_with_bounded_parallel_fetch() {
        // Regression guard for the buffer_unordered fetch path:
        // exercise more chain.json bodies than MAX_FETCH_CONCURRENCY
        // (= 8) so multiple batches must complete and union without
        // dropping any pack sha. Spans heads, tags, and notes so the
        // listing prefix widening from #89 stays exercised.
        let store = MockStore::new();
        let chain_count = MAX_FETCH_CONCURRENCY * 3 + 1;
        let namespaces = ["refs/heads", "refs/tags", "refs/notes"];
        let mut expected: HashSet<Sha40> = HashSet::new();
        for i in 0..chain_count {
            let pack_sha = format!("{:040x}", 0x1000 + i);
            let pack_sha40 = sha40(&pack_sha);
            let namespace = namespaces[i % namespaces.len()];
            let ref_name = RefName::new(format!("{namespace}/r{i}")).unwrap();
            let chain = ChainManifest {
                v: 1,
                tip: sha40(SHA_TIP),
                full_at: sha40(SHA_FULL),
                segments: vec![ChainSegment {
                    sha: sha40(SHA_TIP),
                    parent_sha: None,
                    pack: format!("packs/{pack_sha}.pack"),
                    bytes: 1_024,
                }],
            };
            write_chain(&store, Some("repo"), &ref_name, &chain)
                .await
                .unwrap();
            expected.insert(pack_sha40);
        }

        let referenced = list_referenced_packs(&store, "repo").await.unwrap();
        assert_eq!(referenced, expected);
    }

    #[tokio::test]
    async fn mark_fails_closed_on_corrupt_chain() {
        let store = MockStore::new();
        // chain.json with malformed JSON.
        store.insert(
            "repo/refs/heads/main/chain.json",
            Bytes::from_static(b"{not valid json"),
        );
        let err = mark(&store, "repo", MarkOpts::default()).await.unwrap_err();
        assert!(matches!(err, PackchainError::ParseJson(_)));
        // No tombstone written.
        let metas = store.list("repo/gc/").await.unwrap();
        assert!(metas.is_empty());
    }

    #[tokio::test]
    async fn mark_fails_closed_on_unsupported_schema_version() {
        let store = MockStore::new();
        store.insert(
            "repo/refs/heads/main/chain.json",
            Bytes::from_static(
                br#"{"v":2,"tip":"0000000000000000000000000000000000000001","full_at":"0000000000000000000000000000000000000002","segments":[]}"#,
            ),
        );
        let err = mark(&store, "repo", MarkOpts::default()).await.unwrap_err();
        assert!(matches!(
            err,
            PackchainError::UnsupportedSchemaVersion { .. }
        ));
    }

    // --- sweep ----------------------------------------------------------

    fn sha_set<I: IntoIterator<Item = &'static str>>(shas: I) -> Vec<Sha40> {
        shas.into_iter().map(sha40).collect()
    }

    fn write_tombstone(
        store: &MockStore,
        prefix: &str,
        marked_at: &str,
        shas: Vec<Sha40>,
    ) -> String {
        let run_id = Uuid::new_v4().to_string();
        let key = tombstone_key(prefix, &run_id, marked_at);
        let body = Tombstone {
            v: 1,
            run_id,
            marked_at: marked_at.to_string(),
            orphan_packs: shas,
        }
        .to_json_pretty()
        .unwrap();
        store.insert(&key, Bytes::from(body));
        key
    }

    #[tokio::test]
    async fn sweep_inside_grace_defers_tombstone() {
        let store = MockStore::new();
        let now = OffsetDateTime::now_utc().format(&Rfc3339).unwrap();
        let tombstone = write_tombstone(&store, "repo", &now, sha_set([SHA_PACK_ORPHAN]));
        insert_pack_pair(&store, Some("repo"), SHA_PACK_ORPHAN);

        let outcome = sweep(
            &store,
            "repo",
            SweepOpts {
                grace_hours: 24,
                force: false,
            },
        )
        .await
        .unwrap();
        assert_eq!(outcome.deferred_tombstones, 1);
        assert_eq!(outcome.swept_tombstones, 0);
        // Tombstone and packs survive.
        store.get_bytes(&tombstone).await.unwrap();
        store
            .get_bytes(&format!("repo/packs/{SHA_PACK_ORPHAN}.pack"))
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn sweep_after_grace_deletes_orphan_packs_and_tombstone() {
        let store = MockStore::new();
        let stale = (OffsetDateTime::now_utc() - time::Duration::hours(48))
            .format(&Rfc3339)
            .unwrap();
        let tombstone = write_tombstone(&store, "repo", &stale, sha_set([SHA_PACK_ORPHAN]));
        insert_pack_pair(&store, Some("repo"), SHA_PACK_ORPHAN);

        let outcome = sweep(&store, "repo", SweepOpts::default()).await.unwrap();
        assert_eq!(outcome.swept_tombstones, 1);
        assert_eq!(outcome.deleted_objects, 2, "pack + idx");
        // Tombstone and packs gone.
        let pack_err = store
            .get_bytes(&format!("repo/packs/{SHA_PACK_ORPHAN}.pack"))
            .await
            .unwrap_err();
        assert!(matches!(pack_err, ObjectStoreError::NotFound(_)));
        let tomb_err = store.get_bytes(&tombstone).await.unwrap_err();
        assert!(matches!(tomb_err, ObjectStoreError::NotFound(_)));
    }

    #[tokio::test]
    async fn sweep_skips_repointed_packs() {
        // A tombstoned pack got re-referenced by a chain rewrite
        // before the grace expired. Sweep must NOT delete it.
        let store = MockStore::new();
        let stale = (OffsetDateTime::now_utc() - time::Duration::hours(48))
            .format(&Rfc3339)
            .unwrap();
        // The tombstone names SHA_PACK_LIVE — but a chain now references it.
        write_tombstone(&store, "repo", &stale, sha_set([SHA_PACK_LIVE]));
        let chain = ChainManifest {
            v: 1,
            tip: sha40(SHA_TIP),
            full_at: sha40(SHA_FULL),
            segments: vec![segment(SHA_PACK_LIVE, None)],
        };
        write_chain(&store, Some("repo"), &ref_main(), &chain)
            .await
            .unwrap();
        insert_pack_pair(&store, Some("repo"), SHA_PACK_LIVE);

        let outcome = sweep(&store, "repo", SweepOpts::default()).await.unwrap();
        assert_eq!(outcome.swept_tombstones, 1);
        assert_eq!(outcome.skipped_repointed_packs, 1);
        assert_eq!(outcome.deleted_objects, 0);
        // Pack still present.
        store
            .get_bytes(&format!("repo/packs/{SHA_PACK_LIVE}.pack"))
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn sweep_force_bypasses_grace_only_not_live_recheck() {
        // Regression for #117: --force must skip ONLY the grace window,
        // not the live-pack re-check. A fresh tombstone names a pack
        // that has since been referenced by a committed chain — the
        // classic outcome of mark() snapshotting between a concurrent
        // push's pack upload and its chain.json commit. Sweep with
        // --force must NOT delete that pack.
        let store = MockStore::new();
        let now = OffsetDateTime::now_utc().format(&Rfc3339).unwrap();
        write_tombstone(&store, "repo", &now, sha_set([SHA_PACK_LIVE]));
        let chain = ChainManifest {
            v: 1,
            tip: sha40(SHA_TIP),
            full_at: sha40(SHA_FULL),
            segments: vec![segment(SHA_PACK_LIVE, None)],
        };
        write_chain(&store, Some("repo"), &ref_main(), &chain)
            .await
            .unwrap();
        insert_pack_pair(&store, Some("repo"), SHA_PACK_LIVE);

        let outcome = sweep(
            &store,
            "repo",
            SweepOpts {
                grace_hours: 24,
                force: true,
            },
        )
        .await
        .unwrap();
        // Grace was bypassed (fresh tombstone got processed instead of
        // deferred), but the live-pack guard fired and the pack stayed.
        assert_eq!(outcome.swept_tombstones, 1);
        assert_eq!(outcome.deferred_tombstones, 0);
        assert_eq!(outcome.skipped_repointed_packs, 1);
        assert_eq!(outcome.deleted_objects, 0);
        store
            .get_bytes(&format!("repo/packs/{SHA_PACK_LIVE}.pack"))
            .await
            .expect("live pack must survive --force sweep");
        store
            .get_bytes(&format!("repo/packs/{SHA_PACK_LIVE}.idx"))
            .await
            .expect("live idx must survive --force sweep");
    }

    #[tokio::test]
    async fn sweep_force_deletes_truly_orphan_pack_inside_grace() {
        // The happy path for --force: a fresh tombstone naming a pack
        // that is NOT in any chain. Grace is bypassed, the live-pack
        // re-check finds the SHA absent, the pack is deleted.
        let store = MockStore::new();
        let now = OffsetDateTime::now_utc().format(&Rfc3339).unwrap();
        write_tombstone(&store, "repo", &now, sha_set([SHA_PACK_ORPHAN]));
        insert_pack_pair(&store, Some("repo"), SHA_PACK_ORPHAN);

        let outcome = sweep(
            &store,
            "repo",
            SweepOpts {
                grace_hours: 24,
                force: true,
            },
        )
        .await
        .unwrap();
        assert_eq!(outcome.swept_tombstones, 1);
        assert_eq!(outcome.deferred_tombstones, 0);
        assert_eq!(outcome.skipped_repointed_packs, 0);
        assert_eq!(outcome.deleted_objects, 2);
        let err = store
            .get_bytes(&format!("repo/packs/{SHA_PACK_ORPHAN}.pack"))
            .await
            .unwrap_err();
        assert!(matches!(err, ObjectStoreError::NotFound(_)));
    }

    #[tokio::test]
    async fn sweep_tolerates_already_deleted_pack() {
        // Tombstone names a pack that no longer exists on the bucket
        // (e.g. a previous partial sweep deleted the .pack but
        // crashed before deleting the .idx). Sweep must complete
        // without error.
        let store = MockStore::new();
        let stale = (OffsetDateTime::now_utc() - time::Duration::hours(48))
            .format(&Rfc3339)
            .unwrap();
        write_tombstone(&store, "repo", &stale, sha_set([SHA_PACK_ORPHAN]));
        // No pack inserted.
        let outcome = sweep(&store, "repo", SweepOpts::default()).await.unwrap();
        assert_eq!(outcome.swept_tombstones, 1);
        assert_eq!(outcome.deleted_objects, 0);
    }

    #[tokio::test]
    async fn sweep_handles_multiple_tombstones_independently() {
        let store = MockStore::new();
        let stale = (OffsetDateTime::now_utc() - time::Duration::hours(48))
            .format(&Rfc3339)
            .unwrap();
        let now = OffsetDateTime::now_utc().format(&Rfc3339).unwrap();
        // One stale tombstone (must sweep) + one fresh (must defer).
        write_tombstone(&store, "repo", &stale, sha_set([SHA_PACK_ORPHAN]));
        write_tombstone(&store, "repo", &now, sha_set([SHA_PACK_ORPHAN_2]));
        insert_pack_pair(&store, Some("repo"), SHA_PACK_ORPHAN);
        insert_pack_pair(&store, Some("repo"), SHA_PACK_ORPHAN_2);

        let outcome = sweep(&store, "repo", SweepOpts::default()).await.unwrap();
        assert_eq!(outcome.swept_tombstones, 1);
        assert_eq!(outcome.deferred_tombstones, 1);
        assert_eq!(outcome.deleted_objects, 2);
    }

    // --- end-to-end ---------------------------------------------------

    #[tokio::test]
    async fn mark_then_force_sweep_round_trips() {
        let store = MockStore::new();
        seed_live_chain(&store, Some("repo")).await;
        insert_pack_pair(&store, Some("repo"), SHA_PACK_LIVE);
        insert_pack_pair(&store, Some("repo"), SHA_PACK_ORPHAN);

        let mark_out = mark(&store, "repo", MarkOpts::default()).await.unwrap();
        assert_eq!(mark_out.orphan_count, 1);

        // Force sweep — bypass grace.
        let sweep_out = sweep(
            &store,
            "repo",
            SweepOpts {
                grace_hours: 24,
                force: true,
            },
        )
        .await
        .unwrap();
        assert_eq!(sweep_out.swept_tombstones, 1);
        assert_eq!(sweep_out.deleted_objects, 2);

        // Live pack survives, orphan pack is gone.
        store
            .get_bytes(&format!("repo/packs/{SHA_PACK_LIVE}.pack"))
            .await
            .unwrap();
        let err = store
            .get_bytes(&format!("repo/packs/{SHA_PACK_ORPHAN}.pack"))
            .await
            .unwrap_err();
        assert!(matches!(err, ObjectStoreError::NotFound(_)));
    }

    // --- baseline tombstones (issue #134) -----------------------------

    fn insert_baseline_bundle(store: &MockStore, prefix: Option<&str>, sha: &str) -> String {
        let key = keys::bundle_key(prefix, ref_main(), sha);
        store.insert(&key, Bytes::from_static(b"BUNDLE"));
        key
    }

    fn write_baseline_tombstone_at(
        store: &MockStore,
        prefix: &str,
        marked_at: &str,
        sha: &str,
    ) -> String {
        let key = baseline_tombstone_key(prefix, &Uuid::new_v4().to_string());
        let body = BaselineTombstone {
            v: TOMBSTONE_SCHEMA_VERSION,
            marked_at: marked_at.to_owned(),
            ref_name: ref_main().as_str().to_owned(),
            sha: sha40(sha),
        }
        .to_json_pretty()
        .unwrap();
        store.insert(&key, Bytes::from(body));
        key
    }

    /// Issue #157: [`tombstoned_bundle_keys`] enumerates every bundle
    /// key currently named by a baseline tombstone, regardless of
    /// which engine wrote the tombstone. The bundle engine relies on
    /// this set to hide tombstoned bundles from `list` and from the
    /// under-lock multi-bundle guard.
    #[tokio::test]
    async fn tombstoned_bundle_keys_returns_bundle_paths_for_each_tombstone() {
        let store = MockStore::new();
        // Two tombstones for distinct (ref, sha) pairs.
        write_baseline_tombstone_for_orphan(&store, Some("repo"), &ref_main(), &sha40(SHA_FULL))
            .await
            .unwrap();
        let other_ref = RefName::new("refs/heads/feature").unwrap();
        write_baseline_tombstone_for_orphan(&store, Some("repo"), &other_ref, &sha40(SHA_TIP))
            .await
            .unwrap();

        let keys = tombstoned_bundle_keys(&store, Some("repo")).await.unwrap();
        assert_eq!(keys.len(), 2, "one bundle key per tombstone (got {keys:?})");
        assert!(keys.contains(&format!("repo/refs/heads/main/{SHA_FULL}.bundle")));
        assert!(keys.contains(&format!("repo/refs/heads/feature/{SHA_TIP}.bundle")));
    }

    /// A fresh bucket with no `gc/` directory must not error — empty
    /// set is the right answer.
    #[tokio::test]
    async fn tombstoned_bundle_keys_empty_when_no_tombstones() {
        let store = MockStore::new();
        let keys = tombstoned_bundle_keys(&store, Some("repo")).await.unwrap();
        assert!(keys.is_empty(), "empty bucket yields no tombstoned keys");
    }

    /// Root-prefix repos (no `<prefix>/` segment, keys collapse to
    /// `gc/baseline-tomb-*.json` and `refs/heads/<ref>/<sha>.bundle`)
    /// must produce the same tombstone → bundle-key mapping. All other
    /// tests cover `Some("repo")`; this is the negative-control for
    /// the `prefix.unwrap_or("")` path in `tombstoned_bundle_keys`.
    #[tokio::test]
    async fn tombstoned_bundle_keys_handles_root_prefix() {
        let store = MockStore::new();
        write_baseline_tombstone_for_orphan(&store, None, &ref_main(), &sha40(SHA_FULL))
            .await
            .unwrap();

        let keys = tombstoned_bundle_keys(&store, None).await.unwrap();
        assert_eq!(keys.len(), 1, "got {keys:?}");
        assert!(
            keys.contains(&format!("refs/heads/main/{SHA_FULL}.bundle")),
            "root-prefix bundle key (no leading repo/) must be produced; got {keys:?}",
        );
    }

    /// An unparseable tombstone must not block the rest. Mirrors
    /// `sweep_one_baseline_tombstone`'s tolerance for bad records:
    /// the sweep loop logs a warn and continues.
    #[tokio::test]
    async fn tombstoned_bundle_keys_skips_unparseable_tombstones() {
        let store = MockStore::new();
        // One good tombstone + one garbage tombstone-keyed file.
        write_baseline_tombstone_for_orphan(&store, Some("repo"), &ref_main(), &sha40(SHA_FULL))
            .await
            .unwrap();
        store.insert(
            "repo/gc/baseline-tomb-garbage.json",
            Bytes::from_static(b"not json"),
        );

        let keys = tombstoned_bundle_keys(&store, Some("repo")).await.unwrap();
        assert_eq!(
            keys.len(),
            1,
            "good tombstone must still be returned despite garbage sibling",
        );
        assert!(keys.contains(&format!("repo/refs/heads/main/{SHA_FULL}.bundle")));
    }

    #[tokio::test]
    async fn write_baseline_tombstone_round_trips() {
        // Writer + parser agree on the on-bucket shape. Regression
        // guard: a future serde tweak that broke the JSON layout would
        // make sweep silently skip every baseline tombstone.
        let store = MockStore::new();
        let prior = sha40(SHA_FULL);
        let current = sha40(SHA_TIP);
        write_baseline_tombstone(&store, Some("repo"), &ref_main(), &prior, &current)
            .await
            .unwrap();
        let metas = store.list("repo/gc/").await.unwrap();
        let tomb_key = metas
            .iter()
            .find(|m| {
                m.key
                    .starts_with(&baseline_tombstone_listing_prefix(Some("repo")))
            })
            .map(|m| m.key.clone())
            .expect("baseline tombstone written");
        let body = store.get_bytes(&tomb_key).await.unwrap();
        let parsed = BaselineTombstone::from_json_bytes(&body).unwrap();
        assert_eq!(parsed.v, TOMBSTONE_SCHEMA_VERSION);
        assert_eq!(parsed.ref_name, "refs/heads/main");
        assert_eq!(parsed.sha, prior);
    }

    #[tokio::test]
    async fn write_baseline_tombstone_skips_when_prior_equals_current() {
        // No-op when the keys alias: a tombstone in this case would
        // later cause sweep to delete the live baseline bundle.
        let store = MockStore::new();
        let sha = sha40(SHA_FULL);
        write_baseline_tombstone(&store, Some("repo"), &ref_main(), &sha, &sha)
            .await
            .unwrap();
        let metas = store.list("repo/gc/").await.unwrap();
        assert!(
            metas.is_empty(),
            "aliasing prior/current must not write a tombstone",
        );
    }

    #[tokio::test]
    async fn sweep_defers_baseline_tombstone_within_grace_window() {
        // Issue #134: a fetch that started before compact must be able
        // to read the prior baseline within the grace window. Concrete
        // manifestation: a baseline tombstone marked "now" is left
        // alone, and the bundle it names stays on the bucket.
        let store = MockStore::new();
        let bundle_key = insert_baseline_bundle(&store, Some("repo"), SHA_FULL);
        let now = OffsetDateTime::now_utc().format(&Rfc3339).unwrap();
        let tomb_key = write_baseline_tombstone_at(&store, "repo", &now, SHA_FULL);

        let outcome = sweep(
            &store,
            "repo",
            SweepOpts {
                grace_hours: 24,
                force: false,
            },
        )
        .await
        .unwrap();
        assert_eq!(outcome.deferred_tombstones, 1);
        assert_eq!(outcome.swept_tombstones, 0);
        assert_eq!(outcome.deleted_objects, 0);
        store
            .get_bytes(&bundle_key)
            .await
            .expect("bundle must survive sweep within grace");
        store
            .get_bytes(&tomb_key)
            .await
            .expect("tombstone must survive sweep within grace");
    }

    #[tokio::test]
    async fn sweep_reclaims_baseline_tombstone_after_grace_window() {
        // Issue #134: past the grace window, sweep deletes the bundle
        // and the tombstone. This is the path that reclaims the
        // orphan baseline left in place by compact / force-push.
        let store = MockStore::new();
        let bundle_key = insert_baseline_bundle(&store, Some("repo"), SHA_FULL);
        let stale = (OffsetDateTime::now_utc() - time::Duration::hours(48))
            .format(&Rfc3339)
            .unwrap();
        let tomb_key = write_baseline_tombstone_at(&store, "repo", &stale, SHA_FULL);

        let outcome = sweep(&store, "repo", SweepOpts::default()).await.unwrap();
        assert_eq!(outcome.swept_tombstones, 1);
        assert_eq!(outcome.deferred_tombstones, 0);
        assert_eq!(outcome.deleted_objects, 1, "bundle delete");
        let bundle_err = store.get_bytes(&bundle_key).await.unwrap_err();
        assert!(matches!(bundle_err, ObjectStoreError::NotFound(_)));
        let tomb_err = store.get_bytes(&tomb_key).await.unwrap_err();
        assert!(matches!(tomb_err, ObjectStoreError::NotFound(_)));
    }

    #[tokio::test]
    async fn sweep_skips_re_baselined_bundle_after_grace() {
        // A later push re-baselined to the SAME SHA the tombstone names
        // (force-push at the same tip, or compact short-cut). Sweep
        // must NOT delete the bundle — it is live again. The
        // now-stale tombstone is dropped.
        let store = MockStore::new();
        let bundle_key = insert_baseline_bundle(&store, Some("repo"), SHA_FULL);
        // Live chain points at the same SHA the tombstone names.
        let chain = ChainManifest {
            v: 1,
            tip: sha40(SHA_TIP),
            full_at: sha40(SHA_FULL),
            segments: vec![segment(SHA_PACK_LIVE, None)],
        };
        write_chain(&store, Some("repo"), &ref_main(), &chain)
            .await
            .unwrap();
        let stale = (OffsetDateTime::now_utc() - time::Duration::hours(48))
            .format(&Rfc3339)
            .unwrap();
        let tomb_key = write_baseline_tombstone_at(&store, "repo", &stale, SHA_FULL);

        let outcome = sweep(&store, "repo", SweepOpts::default()).await.unwrap();
        assert_eq!(outcome.swept_tombstones, 1);
        assert_eq!(outcome.skipped_repointed_packs, 1);
        assert_eq!(outcome.deleted_objects, 0);
        store
            .get_bytes(&bundle_key)
            .await
            .expect("re-baselined bundle must survive");
        let tomb_err = store.get_bytes(&tomb_key).await.unwrap_err();
        assert!(matches!(tomb_err, ObjectStoreError::NotFound(_)));
    }

    #[tokio::test]
    async fn sweep_baseline_tolerates_already_deleted_bundle() {
        // The bundle was deleted out of band (operator cleanup, or a
        // ref deletion that happened to sweep it). Sweep must finish
        // cleanly.
        let store = MockStore::new();
        let stale = (OffsetDateTime::now_utc() - time::Duration::hours(48))
            .format(&Rfc3339)
            .unwrap();
        let tomb_key = write_baseline_tombstone_at(&store, "repo", &stale, SHA_FULL);
        // No bundle inserted.

        let outcome = sweep(&store, "repo", SweepOpts::default()).await.unwrap();
        assert_eq!(outcome.swept_tombstones, 1);
        assert_eq!(outcome.deleted_objects, 0);
        let tomb_err = store.get_bytes(&tomb_key).await.unwrap_err();
        assert!(matches!(tomb_err, ObjectStoreError::NotFound(_)));
    }

    #[tokio::test]
    async fn sweep_baseline_force_bypasses_grace_only_not_live_recheck() {
        // --force on a fresh baseline tombstone whose SHA is now live
        // (re-baselined). Grace is bypassed (tombstone is processed),
        // but the live-state guard fires and the bundle stays.
        let store = MockStore::new();
        let bundle_key = insert_baseline_bundle(&store, Some("repo"), SHA_FULL);
        let chain = ChainManifest {
            v: 1,
            tip: sha40(SHA_TIP),
            full_at: sha40(SHA_FULL),
            segments: vec![segment(SHA_PACK_LIVE, None)],
        };
        write_chain(&store, Some("repo"), &ref_main(), &chain)
            .await
            .unwrap();
        let now = OffsetDateTime::now_utc().format(&Rfc3339).unwrap();
        write_baseline_tombstone_at(&store, "repo", &now, SHA_FULL);

        let outcome = sweep(
            &store,
            "repo",
            SweepOpts {
                grace_hours: 24,
                force: true,
            },
        )
        .await
        .unwrap();
        assert_eq!(outcome.swept_tombstones, 1);
        assert_eq!(outcome.deferred_tombstones, 0);
        assert_eq!(outcome.skipped_repointed_packs, 1);
        assert_eq!(outcome.deleted_objects, 0);
        store
            .get_bytes(&bundle_key)
            .await
            .expect("live bundle must survive --force sweep");
    }

    #[tokio::test]
    async fn sweep_processes_pack_and_baseline_tombstones_in_one_pass() {
        // Mixed tombstone types under `<prefix>/gc/`. Sweep must
        // dispatch each to the right handler without mis-counting or
        // skipping.
        let store = MockStore::new();
        let bundle_key = insert_baseline_bundle(&store, Some("repo"), SHA_FULL);
        insert_pack_pair(&store, Some("repo"), SHA_PACK_ORPHAN);
        let stale = (OffsetDateTime::now_utc() - time::Duration::hours(48))
            .format(&Rfc3339)
            .unwrap();
        write_tombstone(&store, "repo", &stale, sha_set([SHA_PACK_ORPHAN]));
        write_baseline_tombstone_at(&store, "repo", &stale, SHA_FULL);

        let outcome = sweep(&store, "repo", SweepOpts::default()).await.unwrap();
        assert_eq!(outcome.swept_tombstones, 2);
        // pack + idx + bundle = 3 deletions
        assert_eq!(outcome.deleted_objects, 3);
        let bundle_err = store.get_bytes(&bundle_key).await.unwrap_err();
        assert!(matches!(bundle_err, ObjectStoreError::NotFound(_)));
        let pack_err = store
            .get_bytes(&format!("repo/packs/{SHA_PACK_ORPHAN}.pack"))
            .await
            .unwrap_err();
        assert!(matches!(pack_err, ObjectStoreError::NotFound(_)));
    }

    #[tokio::test]
    async fn compact_to_sweep_round_trip_simulates_concurrent_fetch_then_gc() {
        // End-to-end issue #134 scenario: compact writes a tombstone
        // (we simulate by hand to avoid pulling in the full compact
        // fixture), an in-flight fetch reads the prior bundle within
        // grace and succeeds, and a later sweep past the grace
        // reclaims it.
        let store = MockStore::new();
        let bundle_key = insert_baseline_bundle(&store, Some("repo"), SHA_FULL);
        // Compact moved the baseline to a new SHA — simulate by
        // writing a chain pointing to SHA_TIP as full_at.
        let chain = ChainManifest {
            v: 1,
            tip: sha40(SHA_TIP),
            full_at: sha40(SHA_TIP),
            segments: vec![segment(SHA_PACK_LIVE, None)],
        };
        write_chain(&store, Some("repo"), &ref_main(), &chain)
            .await
            .unwrap();
        let prior = sha40(SHA_FULL);
        let current = sha40(SHA_TIP);
        write_baseline_tombstone(&store, Some("repo"), &ref_main(), &prior, &current)
            .await
            .unwrap();

        // In-flight fetch: bundle GET within grace MUST succeed.
        let body = store.get_bytes(&bundle_key).await.unwrap();
        assert_eq!(&body[..], b"BUNDLE");
        let in_grace = sweep(
            &store,
            "repo",
            SweepOpts {
                grace_hours: 24,
                force: false,
            },
        )
        .await
        .unwrap();
        assert_eq!(in_grace.deferred_tombstones, 1);
        store
            .get_bytes(&bundle_key)
            .await
            .expect("bundle must survive in-grace sweep");

        // Backdate the tombstone past the grace and re-sweep —
        // bundle is reaped.
        let metas = store.list("repo/gc/").await.unwrap();
        let tomb_key = metas
            .iter()
            .find(|m| {
                m.key
                    .starts_with(&baseline_tombstone_listing_prefix(Some("repo")))
            })
            .map(|m| m.key.clone())
            .unwrap();
        let stale = (OffsetDateTime::now_utc() - time::Duration::hours(48))
            .format(&Rfc3339)
            .unwrap();
        let body = store.get_bytes(&tomb_key).await.unwrap();
        let mut tomb: BaselineTombstone = serde_json::from_slice(&body).unwrap();
        tomb.marked_at = stale;
        let new_body = serde_json::to_vec_pretty(&tomb).unwrap();
        store.insert(&tomb_key, Bytes::from(new_body));

        let post_grace = sweep(&store, "repo", SweepOpts::default()).await.unwrap();
        assert_eq!(post_grace.swept_tombstones, 1);
        assert_eq!(post_grace.deleted_objects, 1);
        let err = store.get_bytes(&bundle_key).await.unwrap_err();
        assert!(matches!(err, ObjectStoreError::NotFound(_)));
    }

    #[tokio::test]
    async fn sweep_preserves_corrupt_baseline_tombstone_for_diagnosis() {
        // Issue #146: a baseline tombstone whose `ref_name` no longer
        // passes `RefName::new` cannot be turned into a bundle key, so
        // deleting the tombstone would orphan the bundle on the bucket
        // with no record. Sweep must preserve BOTH records (tombstone
        // and any bundle it would have named under the raw string), and
        // signal `Deferred` so an operator can reconcile manually.
        let store = MockStore::new();
        // Seed a "bundle" at the raw-string path the tombstone names,
        // so a regression that reconstructs a key from the raw ref_name
        // and deletes it would also be caught here.
        let bad_ref = "refs/heads/[bad]";
        assert!(
            RefName::new(bad_ref).is_err(),
            "fixture relies on this ref_name failing RefName::new",
        );
        let bundle_key = format!("repo/{bad_ref}/{SHA_FULL}.bundle");
        store.insert(&bundle_key, Bytes::from_static(b"BUNDLE"));

        let stale = (OffsetDateTime::now_utc() - time::Duration::hours(48))
            .format(&Rfc3339)
            .unwrap();
        let tomb_key = baseline_tombstone_key("repo", &Uuid::new_v4().to_string());
        let body = BaselineTombstone {
            v: TOMBSTONE_SCHEMA_VERSION,
            marked_at: stale,
            ref_name: bad_ref.to_owned(),
            sha: sha40(SHA_FULL),
        }
        .to_json_pretty()
        .unwrap();
        store.insert(&tomb_key, Bytes::from(body));

        let outcome = sweep(&store, "repo", SweepOpts::default()).await.unwrap();
        assert_eq!(
            outcome.deferred_tombstones, 1,
            "corrupt tombstone counts as deferred, not swept",
        );
        assert_eq!(outcome.swept_tombstones, 0);
        assert_eq!(outcome.deleted_objects, 0);
        // Tombstone survives (operator must inspect).
        let surviving = store
            .get_bytes(&tomb_key)
            .await
            .expect("corrupt tombstone must survive sweep");
        let parsed = BaselineTombstone::from_json_bytes(&surviving).unwrap();
        assert_eq!(parsed.ref_name, bad_ref);
        // Bundle at the would-be key survives (the key was unreachable
        // through the normal RefName path, but sweep must not have
        // reconstructed it from the raw string either).
        store
            .get_bytes(&bundle_key)
            .await
            .expect("orphan bundle must survive corrupt-tombstone sweep");
    }

    // --- per-tombstone live-pack recompute (issue #140) --------------

    /// One-shot post-delete hook used by [`PostDeleteHookStore`].
    type PostDeleteHook = Box<dyn FnOnce(&MockStore) + Send>;

    /// Test-only [`ObjectStore`] decorator that runs a one-shot
    /// callback the first time `delete()` succeeds on a key matching
    /// `trigger_prefix`, *after* the inner delete completes. Used to
    /// inject a concurrent push (writing a fresh `chain.json`) between
    /// successive `sweep_one_tombstone` iterations and verify that the
    /// per-tombstone live-pack recompute picks it up.
    ///
    /// Every other trait method forwards to the inner store unchanged.
    struct PostDeleteHookStore {
        inner: MockStore,
        hook: std::sync::Mutex<Option<PostDeleteHook>>,
        /// Key-prefix the hook fires on. The pack-tombstone case
        /// uses `<prefix>/gc/tombstones-`; the test never deletes
        /// other keys before the intended trigger so this stays
        /// unambiguous.
        trigger_prefix: String,
    }

    impl PostDeleteHookStore {
        fn new(
            inner: MockStore,
            trigger_prefix: impl Into<String>,
            hook: impl FnOnce(&MockStore) + Send + 'static,
        ) -> Self {
            Self {
                inner,
                hook: std::sync::Mutex::new(Some(Box::new(hook))),
                trigger_prefix: trigger_prefix.into(),
            }
        }
    }

    crate::delegate_to_inner_impl! {
        impl ObjectStore for PostDeleteHookStore {
            forward: list, get_to_file, get_bytes, get_bytes_range,
                     put_bytes, put_path, put_if_absent,
                     head, copy;

            async fn delete(&self, key: &str) -> Result<(), ObjectStoreError> {
                let result = self.inner.delete(key).await;
                if result.is_ok()
                    && key.starts_with(&self.trigger_prefix)
                    && let Some(hook) = self.hook.lock().unwrap().take()
                {
                    hook(&self.inner);
                }
                result
            }
        }
    }

    #[tokio::test]
    async fn sweep_re_derives_referenced_set_per_tombstone() {
        // Issue #140 regression: a concurrent push committing
        // chain.json between two `sweep_one_tombstone` iterations
        // must not let sweep delete a pack the new chain references.
        //
        // Layout: two stale tombstones, each naming a distinct pack
        // on its own ref. After the FIRST tombstone is fully
        // processed and deleted, the post-delete hook fires and
        // writes BOTH refs' `chain.json` files — simulating a
        // concurrent push that committed chain.json for the second
        // ref between sweep's two iterations. The second iteration
        // must re-derive the live set and skip the delete.
        //
        // Pre-fix: the once-per-sweep snapshot is empty for both
        // iterations and BOTH packs are deleted (`deleted_objects = 4`).
        // Post-fix: the second iteration's recompute picks up the new
        // chain and the second pack survives
        // (`deleted_objects = 2`, `skipped_repointed_packs = 1`).
        //
        // The hook writes chains for both refs (rather than guessing
        // which tombstone runs first) so the assertions are independent
        // of MockStore iteration order. Writing the first ref's chain
        // is a no-op for that pack — its delete already happened
        // before the hook fired — and the second ref's chain is what
        // protects the still-pending pack.
        let inner = MockStore::new();
        let stale_a = (OffsetDateTime::now_utc() - time::Duration::hours(49))
            .format(&Rfc3339)
            .unwrap();
        let stale_b = (OffsetDateTime::now_utc() - time::Duration::hours(48))
            .format(&Rfc3339)
            .unwrap();
        write_tombstone(&inner, "repo", &stale_a, sha_set([SHA_PACK_ORPHAN]));
        write_tombstone(&inner, "repo", &stale_b, sha_set([SHA_PACK_ORPHAN_2]));
        insert_pack_pair(&inner, Some("repo"), SHA_PACK_ORPHAN);
        insert_pack_pair(&inner, Some("repo"), SHA_PACK_ORPHAN_2);

        // After the FIRST tombstone delete completes, simulate the
        // concurrent push by committing chain.json files for both
        // refs at once.
        let store = PostDeleteHookStore::new(inner, "repo/gc/tombstones-", |inner| {
            for (ref_path, pack_sha) in [
                ("repo/refs/heads/branch_a/chain.json", SHA_PACK_ORPHAN),
                ("repo/refs/heads/branch_b/chain.json", SHA_PACK_ORPHAN_2),
            ] {
                let chain = ChainManifest {
                    v: 1,
                    tip: sha40(SHA_TIP),
                    full_at: sha40(SHA_FULL),
                    segments: vec![segment(pack_sha, None)],
                };
                let body =
                    serde_json::to_vec_pretty(&chain).expect("chain.json serializes for the test");
                inner.insert(ref_path, Bytes::from(body));
            }
        });

        let outcome = sweep(&store, "repo", SweepOpts::default()).await.unwrap();
        // Both tombstones processed.
        assert_eq!(outcome.swept_tombstones, 2);
        // Whichever tombstone ran first deleted its pack pair (2
        // objects). The second iteration's recompute saw the
        // freshly-committed chain and skipped the delete.
        assert_eq!(outcome.deleted_objects, 2);
        assert_eq!(outcome.skipped_repointed_packs, 1);

        // Exactly one of the two packs survives — the one whose
        // tombstone was processed second.
        let first_survives = store
            .inner
            .get_bytes(&format!("repo/packs/{SHA_PACK_ORPHAN}.pack"))
            .await
            .is_ok();
        let second_survives = store
            .inner
            .get_bytes(&format!("repo/packs/{SHA_PACK_ORPHAN_2}.pack"))
            .await
            .is_ok();
        assert!(
            first_survives ^ second_survives,
            "exactly one pack must survive: \
             first_survives={first_survives}, second_survives={second_survives}",
        );
    }

    #[tokio::test]
    async fn sweep_re_derives_referenced_set_per_pack_within_tombstone() {
        // Issue #152 regression: a concurrent push committing
        // chain.json AFTER `sweep_one_tombstone`'s referenced-set
        // snapshot but BEFORE a later pack in the SAME tombstone is
        // reached must not let sweep delete the now-live pack.
        //
        // Layout: one stale tombstone naming TWO orphan packs (the
        // shape `mark()` produces — every orphan SHA for a run goes
        // into one tombstone body). The post-delete hook fires on
        // the FIRST `packs/` delete (the first pack's `.pack` key,
        // mid-iter-1) and writes a chain.json that references the
        // SECOND pack — simulating a concurrent push that landed
        // after the per-tombstone snapshot.
        //
        // Pre-fix (single snapshot per tombstone, taken before the
        // loop): `referenced` is empty for both iterations and both
        // packs are deleted (`deleted_objects = 4`,
        // `skipped_repointed_packs = 0`). Post-fix (per-pack
        // recompute): the second iteration's fresh recompute picks
        // up the new chain and skips the delete
        // (`deleted_objects = 2`, `skipped_repointed_packs = 1`).
        //
        // The pack order inside `orphan_packs` is the Vec insertion
        // order — deterministic — so the assertion is on the EXACT
        // surviving pack key (`SHA_PACK_ORPHAN_2`), not a generic
        // "one of two survives". This catches a regression that
        // dropped the recompute entirely (both deleted) AND a
        // regression that kept the recompute but forgot to skip on
        // re-reference (would also delete the second pack).
        let inner = MockStore::new();
        let stale = (OffsetDateTime::now_utc() - time::Duration::hours(48))
            .format(&Rfc3339)
            .unwrap();
        write_tombstone(
            &inner,
            "repo",
            &stale,
            sha_set([SHA_PACK_ORPHAN, SHA_PACK_ORPHAN_2]),
        );
        insert_pack_pair(&inner, Some("repo"), SHA_PACK_ORPHAN);
        insert_pack_pair(&inner, Some("repo"), SHA_PACK_ORPHAN_2);

        // Fire on the first `packs/` delete (the first pack's `.pack`
        // key) — strictly between the two `orphan_packs` iterations
        // from the caller's perspective: iter-1's deletes complete,
        // iter-2 has not yet run its `list_referenced_packs`. The
        // hook writes a chain.json that re-references the SECOND
        // pack, which the per-pack recompute must observe.
        let store = PostDeleteHookStore::new(inner, "repo/packs/", |inner| {
            let chain = ChainManifest {
                v: 1,
                tip: sha40(SHA_TIP),
                full_at: sha40(SHA_FULL),
                segments: vec![segment(SHA_PACK_ORPHAN_2, None)],
            };
            let body =
                serde_json::to_vec_pretty(&chain).expect("chain.json serializes for the test");
            inner.insert("repo/refs/heads/concurrent/chain.json", Bytes::from(body));
        });

        let outcome = sweep(&store, "repo", SweepOpts::default()).await.unwrap();
        assert_eq!(outcome.swept_tombstones, 1);
        // First pack: deleted (.pack + .idx = 2). Second pack: skipped
        // because the per-pack recompute saw the freshly-committed
        // chain referencing it.
        assert_eq!(
            outcome.deleted_objects, 2,
            "only the first pack's pair deleted; the second was re-referenced",
        );
        assert_eq!(
            outcome.skipped_repointed_packs, 1,
            "second iteration's per-pack recompute must skip the re-referenced pack",
        );
        // Exact surviving pack: the second one (the one the
        // concurrent push re-referenced). Asserting on the specific
        // key — not a broad "one survives" — catches a regression
        // that flipped the iteration order or skipped the wrong pack.
        store
            .inner
            .get_bytes(&format!("repo/packs/{SHA_PACK_ORPHAN_2}.pack"))
            .await
            .expect("re-referenced pack must survive");
        store
            .inner
            .get_bytes(&format!("repo/packs/{SHA_PACK_ORPHAN_2}.idx"))
            .await
            .expect("re-referenced pack idx must survive");
        // First pack is gone.
        let first_err = store
            .inner
            .get_bytes(&format!("repo/packs/{SHA_PACK_ORPHAN}.pack"))
            .await
            .unwrap_err();
        assert!(matches!(first_err, ObjectStoreError::NotFound(_)));
    }

    #[tokio::test]
    async fn sweep_reclaims_genuinely_orphan_pack_with_per_tombstone_recompute() {
        // Sanity: the per-tombstone recompute does NOT regress the
        // normal sweep path. A stale tombstone naming a pack with no
        // chain reference is reclaimed exactly as before.
        let store = MockStore::new();
        let stale = (OffsetDateTime::now_utc() - time::Duration::hours(48))
            .format(&Rfc3339)
            .unwrap();
        write_tombstone(&store, "repo", &stale, sha_set([SHA_PACK_ORPHAN]));
        insert_pack_pair(&store, Some("repo"), SHA_PACK_ORPHAN);
        // No chain.json at all: referenced set is empty for every
        // recompute pass.

        let outcome = sweep(&store, "repo", SweepOpts::default()).await.unwrap();
        assert_eq!(outcome.swept_tombstones, 1);
        assert_eq!(outcome.deleted_objects, 2);
        assert_eq!(outcome.skipped_repointed_packs, 0);
    }

    #[tokio::test]
    async fn sweep_one_baseline_tombstone_re_reads_chain_per_tombstone() {
        // Companion to `sweep_re_derives_referenced_set_per_tombstone`
        // for the baseline tombstone path. `sweep_one_baseline_tombstone`
        // calls `load_chain` inside the function, so each tombstone in
        // a sweep pass sees a freshly-loaded chain. A regression that
        // hoisted the chain load out of the per-tombstone loop would
        // let a concurrent push between iterations slip past the
        // re-baselined-to-same-SHA guard, deleting a bundle that is
        // once again live.
        //
        // Layout: two stale baseline tombstones for the same ref both
        // naming SHA_FULL, plus a baseline bundle at SHA_FULL. With
        // no live chain initially, the first iteration sees
        // `chain.is_none()` → `still_live = false` → deletes the
        // bundle. The post-delete hook (firing after the tombstone
        // delete that follows the bundle delete) writes a chain whose
        // `full_at == SHA_FULL`, simulating a force-push that
        // re-baselined to the same SHA. The second iteration must
        // re-read the chain, observe `full_at == tombstone.sha`, set
        // `still_live = true`, and refuse to re-delete (the bundle is
        // already gone anyway, but the assertion is on the counter:
        // `skipped_repointed_packs == 1`).
        let inner = MockStore::new();
        let bundle_key = insert_baseline_bundle(&inner, Some("repo"), SHA_FULL);
        let stale_a = (OffsetDateTime::now_utc() - time::Duration::hours(49))
            .format(&Rfc3339)
            .unwrap();
        let stale_b = (OffsetDateTime::now_utc() - time::Duration::hours(48))
            .format(&Rfc3339)
            .unwrap();
        let tomb_a = write_baseline_tombstone_at(&inner, "repo", &stale_a, SHA_FULL);
        let tomb_b = write_baseline_tombstone_at(&inner, "repo", &stale_b, SHA_FULL);

        // Trigger on the baseline-tomb prefix: the hook fires AFTER
        // the FIRST iteration's tombstone delete completes (the
        // tombstone delete is the last delete in `sweep_one_baseline_tombstone`),
        // which is precisely the window in which a concurrent
        // force-push could land before the second iteration's chain
        // re-read.
        let tomb_listing = baseline_tombstone_listing_prefix(Some("repo"));
        let store = PostDeleteHookStore::new(inner, &tomb_listing, |inner| {
            let chain = ChainManifest {
                v: 1,
                tip: sha40(SHA_TIP),
                full_at: sha40(SHA_FULL),
                segments: vec![segment(SHA_PACK_LIVE, None)],
            };
            let body =
                serde_json::to_vec_pretty(&chain).expect("chain.json serializes for the test");
            inner.insert("repo/refs/heads/main/chain.json", Bytes::from(body));
        });

        let outcome = sweep(&store, "repo", SweepOpts::default()).await.unwrap();
        // Both baseline tombstones processed.
        assert_eq!(outcome.swept_tombstones, 2);
        // First iteration deleted the bundle (1 object). Second
        // iteration's fresh chain read showed full_at == tombstone.sha
        // and skipped the delete — the recompute is per-tombstone.
        assert_eq!(outcome.deleted_objects, 1, "only one bundle delete");
        assert_eq!(
            outcome.skipped_repointed_packs, 1,
            "second iteration must see the re-baselined chain and skip",
        );
        // Both tombstones are gone.
        for key in [&tomb_a, &tomb_b] {
            let err = store.inner.get_bytes(key).await.unwrap_err();
            assert!(matches!(err, ObjectStoreError::NotFound(_)));
        }
        // The bundle was deleted by the first iteration. Asserting on
        // the counter (not the bundle's presence) is what proves the
        // chain is re-read per tombstone — the survival is necessarily
        // about the COUNTER because the first iteration already removed
        // the bundle before the hook fired.
        let bundle_err = store.inner.get_bytes(&bundle_key).await.unwrap_err();
        assert!(matches!(bundle_err, ObjectStoreError::NotFound(_)));
    }

    #[tokio::test]
    async fn sweep_protects_pack_when_concurrent_push_aliases_existing_key() {
        // Issue #140's canonical scenario, framed as the issue
        // describes it: a force-revert republishes a pack with the
        // SAME content SHA as the tombstoned pack (deterministic gix
        // pack emission). The concurrent push only updates
        // chain.json; the pack key is reused. Sweep must observe
        // the new chain reference and leave the pack alone.
        //
        // Modelled at the post-fix invariant level: the chain
        // referencing the tombstoned SHA exists when
        // `sweep_one_tombstone` runs its recompute, and the pack is
        // preserved with `skipped_repointed_packs += 1`.
        let store = MockStore::new();
        let stale = (OffsetDateTime::now_utc() - time::Duration::hours(48))
            .format(&Rfc3339)
            .unwrap();
        write_tombstone(&store, "repo", &stale, sha_set([SHA_PACK_LIVE]));
        // Insert the pack, then commit chain.json referencing it —
        // identical-content SHA path through the engine ends here.
        insert_pack_pair(&store, Some("repo"), SHA_PACK_LIVE);
        let chain = ChainManifest {
            v: 1,
            tip: sha40(SHA_TIP),
            full_at: sha40(SHA_FULL),
            segments: vec![segment(SHA_PACK_LIVE, None)],
        };
        write_chain(&store, Some("repo"), &ref_main(), &chain)
            .await
            .unwrap();

        let outcome = sweep(&store, "repo", SweepOpts::default()).await.unwrap();
        assert_eq!(outcome.swept_tombstones, 1);
        assert_eq!(outcome.skipped_repointed_packs, 1);
        assert_eq!(outcome.deleted_objects, 0);
        store
            .get_bytes(&format!("repo/packs/{SHA_PACK_LIVE}.pack"))
            .await
            .expect("aliased pack must survive sweep");
    }

    #[tokio::test]
    async fn grace_hours_env_override_falls_back_for_unset_or_invalid() {
        // `EnvGuard` holds the per-key lock for the whole test and
        // restores the prior value on drop, including on panic.
        let env = crate::test_util::EnvGuard::take(ENV_GC_GRACE_HOURS);
        // Unset returns default.
        env.clear();
        assert_eq!(grace_hours_from_env(), DEFAULT_GRACE_HOURS);
        // Non-numeric falls back.
        env.set_to("not-a-number");
        assert_eq!(grace_hours_from_env(), DEFAULT_GRACE_HOURS);
        // Zero falls back (would defeat the design).
        env.set_to("0");
        assert_eq!(grace_hours_from_env(), DEFAULT_GRACE_HOURS);
        // Positive integer wins.
        env.set_to("72");
        assert_eq!(grace_hours_from_env(), 72);
    }

    #[test]
    fn resolve_grace_hours_honours_some_zero() {
        // The key semantic divergence from `resolve_lock_ttl_seconds`:
        // `Some(0)` is a legitimate "no grace window" operator intent
        // (force-mode tests like `delete_tombstone_is_reaped_by_gc_sweep`
        // depend on this), so the resolver must NOT clamp it. A
        // regression that copy-pasted the lock-TTL filter would silently
        // turn `--grace-hours 0` into `--grace-hours <env-default>`.
        assert_eq!(resolve_grace_hours(Some(0)), 0);
    }

    #[test]
    fn resolve_grace_hours_returns_explicit_value() {
        assert_eq!(resolve_grace_hours(Some(7)), 7);
    }

    #[tokio::test]
    async fn resolve_grace_hours_falls_back_to_env_for_none() {
        let env = crate::test_util::EnvGuard::take(ENV_GC_GRACE_HOURS);
        env.set_to("72");
        assert_eq!(resolve_grace_hours(None), 72);
    }

    // --- mark list-order race (issue #135) ---------------------------

    /// One-shot post-`list` hook used by [`PostListHookStore`].
    type PostListHook = Box<dyn FnOnce(&MockStore) + Send>;

    /// Test-only [`ObjectStore`] decorator that runs a one-shot
    /// callback the first time `list()` returns successfully, *after*
    /// the inner list completes. Used to simulate a concurrent push
    /// that uploads a new pack AND commits its `chain.json` between
    /// `mark`'s two listings — the regression scenario for issue #135.
    /// Firing on the first list (regardless of prefix) means the hook
    /// runs between `list_pack_shas` and `list_referenced_packs` under
    /// either ordering, so the test exercises the race against both
    /// the buggy chains-first order and the fixed packs-first order.
    struct PostListHookStore {
        inner: MockStore,
        hook: std::sync::Mutex<Option<PostListHook>>,
    }

    impl PostListHookStore {
        fn new(inner: MockStore, hook: impl FnOnce(&MockStore) + Send + 'static) -> Self {
            Self {
                inner,
                hook: std::sync::Mutex::new(Some(Box::new(hook))),
            }
        }
    }

    crate::delegate_to_inner_impl! {
        impl ObjectStore for PostListHookStore {
            forward: get_to_file, get_bytes, get_bytes_range,
                     put_bytes, put_path, put_if_absent,
                     head, copy, delete;

            async fn list(
                &self,
                prefix: &str,
            ) -> Result<Vec<crate::object_store::ObjectMeta>, ObjectStoreError> {
                let result = self.inner.list(prefix).await;
                if result.is_ok() {
                    let hook = self.hook.lock().unwrap().take();
                    if let Some(hook) = hook {
                        hook(&self.inner);
                    }
                }
                result
            }
        }
    }

    #[tokio::test]
    async fn mark_packs_first_ordering_avoids_false_positive_under_concurrent_push() {
        // Issue #135 regression: a concurrent push that uploads a new
        // pack AND commits its chain.json between mark's two listings
        // must not be tombstoned as orphan.
        //
        // The hook fires after the FIRST list call against
        // `<prefix>/packs/` and inserts a new pack pair plus a
        // chain.json referencing it. With the fixed packs-first
        // ordering, the new pack is absent from the on-bucket snapshot
        // (the snapshot was already taken) and present in the
        // referenced set (the chain.json is committed before the
        // chain list runs). Either way, the new pack is NOT in the
        // orphan set.
        //
        // Pre-fix (chains-first ordering): the hook would fire after
        // the chain list, the chain commit would miss the chain
        // listing, and the pack would appear in `list_pack_shas` →
        // tombstoned as a false positive. The fix flips the ordering
        // so this test asserts orphan_count == 0.
        let inner = MockStore::new();
        // Seed an existing live chain + its pack so the test exercises
        // a realistic non-empty state.
        seed_live_chain(&inner, Some("repo")).await;
        insert_pack_pair(&inner, Some("repo"), SHA_PACK_LIVE);

        let store = PostListHookStore::new(inner, |inner| {
            // Simulate the concurrent push landing mid-mark: upload a
            // fresh pack AND commit its chain.json BEFORE mark's
            // second listing runs.
            insert_pack_pair(inner, Some("repo"), SHA_PACK_ORPHAN);
            let new_chain = ChainManifest {
                v: 1,
                tip: sha40(SHA_TIP),
                full_at: sha40(SHA_FULL),
                segments: vec![segment(SHA_PACK_ORPHAN, None)],
            };
            // `write_chain` is async; the hook is sync, so insert
            // chain.json directly at the canonical key.
            let body =
                serde_json::to_vec_pretty(&new_chain).expect("chain.json serializes for the test");
            inner.insert("repo/refs/heads/concurrent/chain.json", Bytes::from(body));
        });

        let outcome = mark(&store, "repo", MarkOpts::default()).await.unwrap();
        // The fresh pack must NOT be tombstoned: under packs-first
        // ordering it is either absent from `on_bucket` or present in
        // `referenced`.
        assert_eq!(
            outcome.orphan_count, 0,
            "packs-first ordering must not tombstone packs uploaded \
             during mark whose chain commits before the chain listing"
        );
        // No tombstone object emitted for an empty orphan set.
        let gc_metas = store.inner.list("repo/gc/").await.unwrap();
        assert!(gc_metas.is_empty(), "no tombstone for empty orphan set");
    }

    // --- baseline-tombstone post-recheck race (issue #153) -----------

    /// One-shot post-`get_bytes` hook used by [`PostGetHookStore`].
    type PostGetHook = Box<dyn FnOnce(&MockStore) + Send>;

    /// Test-only [`ObjectStore`] decorator that runs a one-shot
    /// callback the first time `get_bytes()` succeeds on `trigger_key`,
    /// *after* the inner read completes. Used to deterministically
    /// simulate a concurrent force-push landing between the initial
    /// `load_chain` in `sweep_one_baseline_tombstone` and the
    /// immediate-pre-delete recheck added by issue #153.
    ///
    /// The `trigger_key` filter is exact-match so the hook fires only
    /// on the targeted chain.json read, not on the tombstone-body
    /// `get_bytes` that runs earlier in the same sweep.
    struct PostGetHookStore {
        inner: MockStore,
        hook: std::sync::Mutex<Option<PostGetHook>>,
        trigger_key: String,
    }

    impl PostGetHookStore {
        fn new(
            inner: MockStore,
            trigger_key: impl Into<String>,
            hook: impl FnOnce(&MockStore) + Send + 'static,
        ) -> Self {
            Self {
                inner,
                hook: std::sync::Mutex::new(Some(Box::new(hook))),
                trigger_key: trigger_key.into(),
            }
        }

        /// `true` once the hook has been consumed — used to witness
        /// that the production code reached the targeted read rather
        /// than skipping past it on a different branch.
        fn hook_fired(&self) -> bool {
            self.hook.lock().unwrap().is_none()
        }
    }

    crate::delegate_to_inner_impl! {
        impl ObjectStore for PostGetHookStore {
            forward: list, get_to_file, get_bytes_range,
                     put_bytes, put_path, put_if_absent,
                     head, copy, delete;

            async fn get_bytes(&self, key: &str) -> Result<Bytes, ObjectStoreError> {
                let result = self.inner.get_bytes(key).await;
                if result.is_ok()
                    && key == self.trigger_key
                    && let Some(hook) = self.hook.lock().unwrap().take()
                {
                    hook(&self.inner);
                }
                result
            }
        }
    }

    #[tokio::test]
    async fn sweep_baseline_defers_when_recheck_observes_re_baseline() {
        // Issue #153 regression: a concurrent force-push or compact may
        // re-baseline the ref to the tombstoned SHA between the initial
        // `load_chain` and the bundle delete that follows. Without the
        // immediate-pre-delete recheck, sweep would erase a now-live
        // bundle and then drop its tombstone.
        //
        // Layout: a baseline bundle at SHA_FULL, a stale baseline
        // tombstone naming SHA_FULL, and an initial chain.json with
        // `full_at = SHA_TIP` (different from SHA_FULL) so the initial
        // check sees `still_live = false` and proceeds toward the
        // delete. The PostGetHookStore fires AFTER the first read of
        // `chain.json` (the initial `load_chain`) and overwrites it
        // with a chain whose `full_at = SHA_FULL` — modelling the
        // concurrent force-push landing in the gap.
        //
        // With the fix in place: the immediate-pre-delete recheck
        // observes the new state, `still_live` flips to true, and
        // sweep returns `Deferred` — preserving BOTH the bundle and
        // the tombstone so a future sweep can retry.
        let inner = MockStore::new();
        let bundle_key = insert_baseline_bundle(&inner, Some("repo"), SHA_FULL);
        // Initial chain points at a different SHA, so the first
        // `still_live` check is false and the code falls into the
        // delete branch where the recheck now lives.
        let initial_chain = ChainManifest {
            v: 1,
            tip: sha40(SHA_TIP),
            full_at: sha40(SHA_TIP),
            segments: vec![segment(SHA_PACK_LIVE, None)],
        };
        write_chain(&inner, Some("repo"), &ref_main(), &initial_chain)
            .await
            .unwrap();
        let stale = (OffsetDateTime::now_utc() - time::Duration::hours(48))
            .format(&Rfc3339)
            .unwrap();
        let tomb_key = write_baseline_tombstone_at(&inner, "repo", &stale, SHA_FULL);

        // Hook fires AFTER the FIRST get_bytes of chain.json (the
        // initial load_chain). Before the fix, that was the only
        // chain read; the bundle delete that followed would erase a
        // live bundle. After the fix, a second get_bytes (the
        // immediate-pre-delete recheck) sees this updated state.
        let chain_key = "repo/refs/heads/main/chain.json";
        let store = PostGetHookStore::new(inner, chain_key, move |inner| {
            let re_baselined = ChainManifest {
                v: 1,
                tip: sha40(SHA_TIP),
                full_at: sha40(SHA_FULL),
                segments: vec![segment(SHA_PACK_LIVE, None)],
            };
            let body = serde_json::to_vec_pretty(&re_baselined)
                .expect("chain.json serializes for the test");
            inner.insert(chain_key, Bytes::from(body));
        });

        let outcome = sweep(&store, "repo", SweepOpts::default()).await.unwrap();
        // The recheck caught the race: tombstone deferred for a
        // future sweep.
        assert_eq!(
            outcome.deferred_tombstones, 1,
            "recheck must defer when the chain re-baselined to the \
             tombstoned SHA between checks",
        );
        assert_eq!(outcome.swept_tombstones, 0);
        assert_eq!(outcome.deleted_objects, 0);
        assert_eq!(outcome.skipped_repointed_packs, 0);

        // Witness that the production code actually executed the
        // recheck branch (otherwise the hook would still be armed and
        // the test would be vacuously passing).
        assert!(
            store.hook_fired(),
            "production code must have read chain.json so the hook \
             could inject the concurrent re-baseline",
        );

        // Bundle MUST survive — the whole point of the fix.
        store
            .inner
            .get_bytes(&bundle_key)
            .await
            .expect("re-baselined bundle must survive sweep");
        // Tombstone MUST survive — preserved for a future sweep to
        // retry once the race settles.
        store
            .inner
            .get_bytes(&tomb_key)
            .await
            .expect("tombstone must survive deferred path");
    }
}