feat(relay): inventory unreachable Git objects (#2264)

.env.example

@@ -75,6 +75,14 @@ RELAY_URL=ws://localhost:3000
 # BUZZ_GIT_PACK_CACHE_PATH=./repos/.pack-cache
 # BUZZ_GIT_PACK_CACHE_MAX_BYTES=5368709120
 # BUZZ_GIT_PACK_CACHE_MAX_CONCURRENT_POPULATIONS=2
+# Deployment-global dry-run inventory for future physical object-store GC.
+# No objects are deleted. Disabled by default.
+# BUZZ_GIT_GC_ENABLED=false
+# BUZZ_GIT_GC_INTERVAL_SECS=3600
+# BUZZ_GIT_GC_MAX_POINTERS=10000
+# BUZZ_GIT_GC_MAX_OBJECTS_PER_PREFIX=10000
+# BUZZ_GIT_GC_MAX_MANIFEST_BYTES_PER_SCAN=536870912
+# BUZZ_GIT_GC_SCAN_TIMEOUT_SECS=300
 
 # -----------------------------------------------------------------------------
 # Media Upload Admission

bin/.helm-3.16.4.pkg

@@ -0,0 +1 @@
+hermit

bin/helm

@@ -0,0 +1 @@
+.helm-3.16.4.pkg

crates/buzz-db/src/lib.rs

@@ -198,16 +198,16 @@ pub struct DbPoolStats {
     pub max: u32,
 }
 
-/// Owns the detached Postgres session holding the relay usage-metrics advisory lock.
+/// Owns a detached Postgres session holding a relay advisory leader lock.
 ///
 /// The connection deliberately does not return to the main pool: session advisory
 /// locks must remain bound to this exact physical connection, and the poller
 /// pings it before each leader-only collection tick.
-pub struct UsageMetricsLeader {
+pub struct AdvisoryLockLeader {
     connection: PgConnection,
 }
 
-impl UsageMetricsLeader {
+impl AdvisoryLockLeader {
     /// Returns whether the lock-owning session is still reachable.
     ///
     /// Bounded to 5 seconds — a blackholed connection (no RST) would otherwise
@@ -508,23 +508,20 @@ impl Db {
         })
     }
 
-    /// Try to acquire the detached session advisory lock for relay usage metrics.
+    /// Try to acquire a detached session advisory lock for singleton relay work.
     ///
     /// The returned guard owns the exact connection that acquired the lock. It is
     /// detached from the shared pool so a stable leader neither returns a locked
     /// session to other callers nor permanently consumes a pool slot. Dropping the
     /// guard closes the connection and releases the session-scoped lock.
-    pub async fn try_lock_usage_metrics(
-        &self,
-        lock_key: i64,
-    ) -> Result<Option<UsageMetricsLeader>> {
+    pub async fn try_advisory_lock(&self, lock_key: i64) -> Result<Option<AdvisoryLockLeader>> {
         let mut connection = self.pool.acquire().await?;
         let acquired = sqlx::query_scalar::<_, bool>("SELECT pg_try_advisory_lock($1)")
             .bind(lock_key)
             .fetch_one(&mut *connection)
             .await?;
         if acquired {
-            Ok(Some(UsageMetricsLeader {
+            Ok(Some(AdvisoryLockLeader {
                 connection: connection.detach(),
             }))
         } else {
@@ -4663,14 +4660,14 @@ mod tests {
         let key = 0x4255_5A5A_4D45_5452;
 
         let mut leader = first
-            .try_lock_usage_metrics(key)
+            .try_advisory_lock(key)
             .await
             .expect("first lock attempt")
             .expect("first database handle becomes leader");
         assert!(leader.is_live().await, "lock owner remains reachable");
         assert!(
             second
-                .try_lock_usage_metrics(key)
+                .try_advisory_lock(key)
                 .await
                 .expect("second lock attempt")
                 .is_none(),
@@ -4680,7 +4677,7 @@ mod tests {
         drop(leader);
         assert!(
             second
-                .try_lock_usage_metrics(key)
+                .try_advisory_lock(key)
                 .await
                 .expect("lock attempt after leader drop")
                 .is_some(),

crates/buzz-relay/src/api/git/gc.rs

@@ -0,0 +1,529 @@
+//! Dry-run inventory for future physical Git object-store garbage collection.
+//!
+//! This module deliberately does not delete objects. It computes a
+//! deployment-global reachability snapshot and reports immutable objects that
+//! were not referenced by any current repository pointer. Deletion needs a
+//! durable, continuous-unreachability grace period plus coordination with
+//! concurrent publishers; neither safety condition should be approximated.
+
+use std::collections::HashSet;
+use std::sync::Arc;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use super::manifest::Manifest;
+use super::store::{GitStore, ObjectList, StoreError, StoredObject};
+use crate::state::AppState;
+
+const MAX_GC_MANIFEST_BYTES: u64 = 4 * 1024 * 1024;
+const IMMUTABLE_PREFIXES: [&str; 3] = ["manifests/", "packs/", "idx/"];
+const GIT_GC_LOCK_KEY: i64 = 0x4255_5A5A_4749_5447;
+
+/// Configuration for the singleton dry-run inventory worker.
+#[derive(Debug, Clone, Copy)]
+pub struct GitGcWorkerConfig {
+    /// Whether the worker is enabled.
+    pub enabled: bool,
+    /// Delay between inventory scans.
+    pub interval: Duration,
+    /// Per-scan inventory limits.
+    pub limits: GitGcScanLimits,
+}
+
+impl GitGcWorkerConfig {
+    /// Read worker configuration from `BUZZ_GIT_GC_*` environment variables.
+    pub fn from_env() -> Result<Self, String> {
+        Ok(Self {
+            enabled: parse_bool_env("BUZZ_GIT_GC_ENABLED", false)?,
+            interval: Duration::from_secs(parse_positive_u64_env(
+                "BUZZ_GIT_GC_INTERVAL_SECS",
+                3_600,
+            )?),
+            limits: GitGcScanLimits {
+                max_pointers: parse_positive_usize_env("BUZZ_GIT_GC_MAX_POINTERS", 10_000)?,
+                max_objects_per_prefix: parse_positive_usize_env(
+                    "BUZZ_GIT_GC_MAX_OBJECTS_PER_PREFIX",
+                    10_000,
+                )?,
+                max_manifest_bytes: parse_positive_u64_env(
+                    "BUZZ_GIT_GC_MAX_MANIFEST_BYTES_PER_SCAN",
+                    512 * 1024 * 1024,
+                )?,
+                timeout: Duration::from_secs(parse_positive_u64_env(
+                    "BUZZ_GIT_GC_SCAN_TIMEOUT_SECS",
+                    300,
+                )?),
+            },
+        })
+    }
+}
+
+/// Limits for one dry-run inventory scan.
+#[derive(Debug, Clone, Copy)]
+pub struct GitGcScanLimits {
+    /// Maximum repository pointers that may be considered.
+    pub max_pointers: usize,
+    /// Maximum objects listed from each immutable Git prefix.
+    pub max_objects_per_prefix: usize,
+    /// Maximum manifest bytes downloaded while marking live pointers.
+    pub max_manifest_bytes: u64,
+    /// Hard deadline for one complete inventory attempt.
+    pub timeout: Duration,
+}
+
+impl Default for GitGcScanLimits {
+    fn default() -> Self {
+        Self {
+            max_pointers: 10_000,
+            max_objects_per_prefix: 10_000,
+            max_manifest_bytes: 512 * 1024 * 1024,
+            timeout: Duration::from_secs(300),
+        }
+    }
+}
+
+/// Run the leader-elected dry-run inventory worker until process shutdown.
+pub async fn run_git_gc_worker(state: Arc<AppState>, config: GitGcWorkerConfig) {
+    if !config.enabled {
+        return;
+    }
+
+    let jitter_bound = config.interval.as_secs().max(1);
+    tokio::time::sleep(Duration::from_secs(rand::random::<u64>() % jitter_bound)).await;
+    let mut interval = tokio::time::interval(config.interval);
+    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    let mut leader: Option<buzz_db::AdvisoryLockLeader> = None;
+
+    loop {
+        interval.tick().await;
+        let mut demoted = false;
+        if let Some(leader_guard) = leader.as_mut() {
+            if !leader_guard.is_live().await {
+                tracing::warn!("Git object-store GC inventory leader demoting");
+                leader = None;
+                demoted = true;
+            }
+        }
+        if leader.is_none() && !demoted {
+            match state.db.try_advisory_lock(GIT_GC_LOCK_KEY).await {
+                Ok(acquired) => {
+                    leader = acquired;
+                    if leader.is_some() {
+                        tracing::info!("Acquired Git object-store GC inventory leader lock");
+                    }
+                }
+                Err(error) => {
+                    tracing::warn!(%error, "Git object-store GC leader election failed");
+                }
+            }
+        }
+        metrics::gauge!("buzz_git_object_store_gc_is_leader").set(if leader.is_some() {
+            1.0
+        } else {
+            0.0
+        });
+
+        if leader.is_none() {
+            continue;
+        }
+        match scan_git_object_store(&state.git_store, config.limits).await {
+            Ok(report) => {
+                tracing::info!(
+                    pointers = report.pointers,
+                    reachable_objects = report.reachable_objects,
+                    observed_candidate_objects = report.observed_candidate_objects,
+                    observed_candidate_bytes = report.observed_candidate_bytes,
+                    pagination_complete = report.pagination_complete,
+                    "Git object-store GC dry-run inventory completed"
+                );
+            }
+            Err(error) => {
+                tracing::warn!(%error, "Git object-store GC dry-run inventory failed");
+            }
+        }
+    }
+}
+
+/// Result of one dry-run inventory scan.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct GitGcScanReport {
+    /// Number of current repository pointers inspected.
+    pub pointers: usize,
+    /// Number of immutable Git objects marked reachable.
+    pub reachable_objects: usize,
+    /// Number of listed immutable objects not in the reachability set.
+    pub observed_candidate_objects: usize,
+    /// Total bytes represented by the candidates.
+    pub observed_candidate_bytes: u64,
+    /// False when any bounded prefix listing was truncated.
+    pub pagination_complete: bool,
+}
+
+/// Errors from a dry-run inventory scan.
+#[derive(Debug, thiserror::Error)]
+pub enum GitGcError {
+    /// Object-store operation failed.
+    #[error(transparent)]
+    Store(#[from] StoreError),
+    /// A live pointer or manifest was malformed, so classification failed closed.
+    #[error("invalid live Git object {key}: {reason}")]
+    InvalidLiveObject {
+        /// Pointer or manifest key.
+        key: String,
+        /// Validation failure.
+        reason: String,
+    },
+    /// Pointer inventory exceeded its safety bound.
+    #[error("repository pointer inventory exceeded the configured limit")]
+    PointerInventoryTruncated,
+    /// Live-manifest reads exceeded the configured transfer budget.
+    #[error("live manifest inventory exceeded the configured byte budget")]
+    ManifestBudgetExceeded,
+    /// The scan did not finish before its configured deadline.
+    #[error("Git object-store inventory scan timed out")]
+    TimedOut,
+}
+
+/// Scan all current Git pointers and classify unreferenced immutable objects.
+///
+/// This is observability only. A candidate is not necessarily safe to delete:
+/// the scan does not prove continuous unreachability or exclude a concurrent
+/// publisher from making an existing content-addressed object reachable.
+pub async fn scan_git_object_store(
+    store: &GitStore,
+    limits: GitGcScanLimits,
+) -> Result<GitGcScanReport, GitGcError> {
+    let started = Instant::now();
+    let result = tokio::time::timeout(limits.timeout, scan_git_object_store_inner(store, limits))
+        .await
+        .map_err(|_| GitGcError::TimedOut)
+        .and_then(|result| result);
+    match &result {
+        Ok(report) => {
+            metrics::counter!("buzz_git_object_store_gc_scans_total", "result" => "success")
+                .increment(1);
+            metrics::gauge!("buzz_git_object_store_gc_observed_candidate_objects")
+                .set(report.observed_candidate_objects as f64);
+            metrics::gauge!("buzz_git_object_store_gc_observed_candidate_bytes")
+                .set(report.observed_candidate_bytes as f64);
+            metrics::gauge!("buzz_git_object_store_gc_scan_complete")
+                .set(if report.pagination_complete { 1.0 } else { 0.0 });
+            metrics::gauge!("buzz_git_object_store_gc_last_success_timestamp_seconds")
+                .set(unix_timestamp_seconds());
+        }
+        Err(_) => {
+            metrics::counter!("buzz_git_object_store_gc_scans_total", "result" => "error")
+                .increment(1);
+            metrics::gauge!("buzz_git_object_store_gc_scan_complete").set(0.0);
+        }
+    }
+    metrics::histogram!("buzz_git_object_store_gc_scan_seconds")
+        .record(started.elapsed().as_secs_f64());
+    result
+}
+
+async fn scan_git_object_store_inner(
+    store: &GitStore,
+    limits: GitGcScanLimits,
+) -> Result<GitGcScanReport, GitGcError> {
+    let pointer_list = store.list_prefix("repos/", limits.max_pointers).await?;
+    if pointer_list.truncated {
+        return Err(GitGcError::PointerInventoryTruncated);
+    }
+
+    let pointers: Vec<_> = pointer_list
+        .objects
+        .into_iter()
+        .filter(|object| object.key.ends_with("/pointer"))
+        .collect();
+    let mut reachable = HashSet::new();
+    let mut manifest_bytes = 0u64;
+    for pointer in &pointers {
+        let remaining = limits.max_manifest_bytes.saturating_sub(manifest_bytes);
+        if remaining == 0 {
+            return Err(GitGcError::ManifestBudgetExceeded);
+        }
+        manifest_bytes = manifest_bytes.saturating_add(
+            mark_current_manifest(store, &pointer.key, &mut reachable, remaining).await?,
+        );
+    }
+
+    let mut observed_candidate_objects = 0usize;
+    let mut observed_candidate_bytes = 0u64;
+    let mut pagination_complete = true;
+    for prefix in IMMUTABLE_PREFIXES {
+        let listed = store
+            .list_prefix(prefix, limits.max_objects_per_prefix)
+            .await?;
+        pagination_complete &= !listed.truncated;
+        let (count, bytes) = classify_candidates(&listed, &reachable);
+        observed_candidate_objects = observed_candidate_objects.saturating_add(count);
+        observed_candidate_bytes = observed_candidate_bytes.saturating_add(bytes);
+    }
+
+    Ok(GitGcScanReport {
+        pointers: pointers.len(),
+        reachable_objects: reachable.len(),
+        observed_candidate_objects,
+        observed_candidate_bytes,
+        pagination_complete,
+    })
+}
+
+async fn mark_current_manifest(
+    store: &GitStore,
+    pointer_key: &str,
+    reachable: &mut HashSet<String>,
+    remaining_manifest_bytes: u64,
+) -> Result<u64, GitGcError> {
+    let Some((_etag, pointer_body)) = store.get_pointer(pointer_key).await? else {
+        return Ok(0);
+    };
+    let digest = std::str::from_utf8(&pointer_body)
+        .map(str::trim)
+        .map_err(|error| GitGcError::InvalidLiveObject {
+            key: pointer_key.to_string(),
+            reason: error.to_string(),
+        })?;
+    if !is_digest(digest) {
+        return Err(GitGcError::InvalidLiveObject {
+            key: pointer_key.to_string(),
+            reason: "pointer body is not a SHA-256 digest".to_string(),
+        });
+    }
+
+    let manifest_key = format!("manifests/{digest}");
+    let read_limit = remaining_manifest_bytes.min(MAX_GC_MANIFEST_BYTES);
+    let bytes = match store
+        .get_verified_limited(&manifest_key, digest, read_limit)
+        .await
+    {
+        Err(StoreError::ObjectTooLarge { .. })
+            if remaining_manifest_bytes < MAX_GC_MANIFEST_BYTES =>
+        {
+            return Err(GitGcError::ManifestBudgetExceeded);
+        }
+        result => result?,
+    };
+    let manifest = Manifest::from_bytes(&bytes).map_err(|error| GitGcError::InvalidLiveObject {
+        key: manifest_key.clone(),
+        reason: error.to_string(),
+    })?;
+    manifest
+        .validate()
+        .map_err(|error| GitGcError::InvalidLiveObject {
+            key: manifest_key.clone(),
+            reason: error.to_string(),
+        })?;
+
+    reachable.insert(manifest_key);
+    for pack_key in manifest.packs {
+        if let Some(digest) = pack_key.strip_prefix("packs/") {
+            reachable.insert(format!("idx/{digest}"));
+        }
+        reachable.insert(pack_key);
+    }
+    Ok(bytes.len() as u64)
+}
+
+fn classify_candidates(listed: &ObjectList, reachable: &HashSet<String>) -> (usize, u64) {
+    listed
+        .objects
+        .iter()
+        .filter(|object| is_gc_object(object) && !reachable.contains(&object.key))
+        .fold((0usize, 0u64), |(count, bytes), object| {
+            (count.saturating_add(1), bytes.saturating_add(object.size))
+        })
+}
+
+fn is_gc_object(object: &StoredObject) -> bool {
+    IMMUTABLE_PREFIXES
+        .iter()
+        .any(|prefix| object.key.strip_prefix(prefix).is_some_and(is_digest))
+}
+
+fn is_digest(value: &str) -> bool {
+    value.len() == 64 && value.chars().all(|character| character.is_ascii_hexdigit())
+}
+
+fn unix_timestamp_seconds() -> f64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_secs_f64()
+}
+
+fn parse_bool_env(name: &str, default: bool) -> Result<bool, String> {
+    match std::env::var(name) {
+        Ok(value) => match value.trim().to_ascii_lowercase().as_str() {
+            "true" | "1" => Ok(true),
+            "false" | "0" => Ok(false),
+            _ => Err(format!("{name} must be true or false")),
+        },
+        Err(std::env::VarError::NotPresent) => Ok(default),
+        Err(std::env::VarError::NotUnicode(_)) => Err(format!("{name} must be valid Unicode")),
+    }
+}
+
+fn parse_positive_u64_env(name: &str, default: u64) -> Result<u64, String> {
+    match std::env::var(name) {
+        Ok(value) => value
+            .parse::<u64>()
+            .ok()
+            .filter(|value| *value > 0)
+            .ok_or_else(|| format!("{name} must be a positive integer")),
+        Err(std::env::VarError::NotPresent) => Ok(default),
+        Err(std::env::VarError::NotUnicode(_)) => Err(format!("{name} must be valid Unicode")),
+    }
+}
+
+fn parse_positive_usize_env(name: &str, default: usize) -> Result<usize, String> {
+    match std::env::var(name) {
+        Ok(value) => value
+            .parse::<usize>()
+            .ok()
+            .filter(|value| *value > 0)
+            .ok_or_else(|| format!("{name} must be a positive integer")),
+        Err(std::env::VarError::NotPresent) => Ok(default),
+        Err(std::env::VarError::NotUnicode(_)) => Err(format!("{name} must be valid Unicode")),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::{BTreeMap, HashSet};
+    use std::time::Duration;
+
+    use super::{
+        classify_candidates, is_digest, parse_bool_env, parse_positive_u64_env,
+        parse_positive_usize_env, scan_git_object_store, GitGcScanLimits,
+    };
+    use crate::api::git::manifest::{Manifest, MANIFEST_VERSION};
+    use crate::api::git::store::{GitStore, ObjectList, Precond, StoredObject};
+
+    fn object(key: &str, size: u64) -> StoredObject {
+        StoredObject {
+            key: key.to_string(),
+            size,
+            last_modified: "2026-07-21T00:00:00Z".to_string(),
+        }
+    }
+
+    #[test]
+    fn candidate_classification_is_prefix_scoped_and_reachability_aware() {
+        let live_pack = format!("packs/{}", "a".repeat(64));
+        let orphan_pack = format!("packs/{}", "b".repeat(64));
+        let listed = ObjectList {
+            objects: vec![
+                object(&live_pack, 10),
+                object(&orphan_pack, 20),
+                object("packs/not-a-digest", 30),
+                object(&format!("media/{}", "c".repeat(64)), 40),
+            ],
+            truncated: false,
+        };
+        let reachable = HashSet::from([live_pack]);
+
+        assert_eq!(classify_candidates(&listed, &reachable), (1, 20));
+    }
+
+    #[test]
+    fn digest_validation_is_exact() {
+        assert!(is_digest(&"a".repeat(64)));
+        assert!(!is_digest(&"a".repeat(63)));
+        assert!(!is_digest(&"g".repeat(64)));
+    }
+
+    #[test]
+    fn gc_environment_values_are_strict() {
+        assert_eq!(
+            parse_bool_env("BUZZ_TEST_GC_MISSING_BOOL", false),
+            Ok(false)
+        );
+        assert_eq!(
+            parse_positive_u64_env("BUZZ_TEST_GC_MISSING_U64", 10),
+            Ok(10)
+        );
+        assert_eq!(
+            parse_positive_usize_env("BUZZ_TEST_GC_MISSING_USIZE", 10),
+            Ok(10)
+        );
+    }
+
+    fn live_store() -> GitStore {
+        let endpoint = std::env::var("BUZZ_GIT_S3_ENDPOINT")
+            .or_else(|_| std::env::var("BUZZ_S3_ENDPOINT"))
+            .unwrap_or_else(|_| "http://localhost:9000".into());
+        let access_key = std::env::var("BUZZ_GIT_S3_ACCESS_KEY")
+            .or_else(|_| std::env::var("BUZZ_S3_ACCESS_KEY"))
+            .unwrap_or_else(|_| "buzz_dev".into());
+        let secret_key = std::env::var("BUZZ_GIT_S3_SECRET_KEY")
+            .or_else(|_| std::env::var("BUZZ_S3_SECRET_KEY"))
+            .unwrap_or_else(|_| "buzz_dev_secret".into());
+        let bucket = std::env::var("BUZZ_GIT_S3_BUCKET")
+            .or_else(|_| std::env::var("BUZZ_S3_BUCKET"))
+            .unwrap_or_else(|_| "buzz-media".into());
+        let region = std::env::var("BUZZ_GIT_S3_REGION")
+            .or_else(|_| std::env::var("BUZZ_S3_REGION"))
+            .unwrap_or_else(|_| "us-east-1".into());
+        GitStore::new(&endpoint, &access_key, &secret_key, &bucket, &region)
+            .expect("connect to live object store")
+    }
+
+    #[tokio::test]
+    async fn live_scan_marks_current_objects_and_observes_orphans() {
+        if std::env::var("BUZZ_GIT_S3_PROBE").as_deref() != Ok("1") {
+            return;
+        }
+
+        let store = live_store();
+        let live_pack_key = store
+            .put_pack(b"live-gc-pack")
+            .await
+            .expect("put live pack");
+        let orphan_pack_key = store
+            .put_pack(b"orphan-gc-pack")
+            .await
+            .expect("put orphan pack");
+        let commit = "1".repeat(40);
+        let manifest = Manifest {
+            version: MANIFEST_VERSION,
+            head: "refs/heads/main".to_string(),
+            refs: BTreeMap::from([("refs/heads/main".to_string(), commit)]),
+            packs: vec![live_pack_key.clone()],
+            parent: None,
+        };
+        let manifest_key = store
+            .put_manifest(&manifest.canonical_bytes().expect("manifest bytes"))
+            .await
+            .expect("put manifest");
+        let digest = manifest_key
+            .strip_prefix("manifests/")
+            .expect("manifest digest");
+        let pointer_key = format!("repos/{}/gc-test/repo/pointer", uuid::Uuid::new_v4());
+        store
+            .put_pointer(&pointer_key, digest.as_bytes(), Precond::IfNoneMatchStar)
+            .await
+            .expect("put pointer");
+
+        let report = scan_git_object_store(
+            &store,
+            GitGcScanLimits {
+                max_pointers: 10_000,
+                max_objects_per_prefix: 10_000,
+                max_manifest_bytes: 64 * 1024 * 1024,
+                timeout: Duration::from_secs(30),
+            },
+        )
+        .await
+        .expect("scan object store");
+        assert!(report.pointers >= 1);
+        assert!(report.reachable_objects >= 3);
+        assert!(report.observed_candidate_objects >= 1);
+        assert!(report.observed_candidate_bytes >= b"orphan-gc-pack".len() as u64);
+
+        store.delete_for_test(&pointer_key).await;
+        store.delete_for_test(&manifest_key).await;
+        store.delete_for_test(&live_pack_key).await;
+        store.delete_for_test(&orphan_pack_key).await;
+    }
+}

crates/buzz-relay/src/api/git/mod.rs

@@ -23,6 +23,7 @@ use tower_http::limit::RequestBodyLimitLayer;
 use crate::state::AppState;
 
 pub mod cas_publish;
+pub mod gc;
 pub mod hook;
 pub mod hydrate;
 pub mod manifest;

crates/buzz-relay/src/api/git/store.rs

@@ -145,6 +145,26 @@ pub struct ProbeReport {
     pub transport_drops: usize,
 }
 
+/// One object returned by a bounded prefix listing.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct StoredObject {
+    /// Full object key.
+    pub key: String,
+    /// Object size reported by the backend.
+    pub size: u64,
+    /// Backend-provided last-modified timestamp.
+    pub last_modified: String,
+}
+
+/// Bounded result from listing one object-store prefix.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ObjectList {
+    /// Objects returned, up to the caller's limit.
+    pub objects: Vec<StoredObject>,
+    /// Whether more objects exist beyond this result.
+    pub truncated: bool,
+}
+
 /// Failure carrying the phase that failed plus enough context to diagnose.
 #[derive(Debug, thiserror::Error)]
 #[error("conformance probe failed in phase '{phase}' (round {round}, key {key}): {reason}")]
@@ -240,6 +260,70 @@ impl GitStore {
         Ok(format!("idx/{pack_digest}"))
     }
 
+    /// List at most `max_objects` objects below `prefix`.
+    ///
+    /// Pagination is handled internally, but the caller-provided bound keeps
+    /// background inventory work from growing without limit. A truncated
+    /// result must never be treated as a complete reachability snapshot.
+    pub async fn list_prefix(
+        &self,
+        prefix: &str,
+        max_objects: usize,
+    ) -> Result<ObjectList, StoreError> {
+        if max_objects == 0 {
+            return Ok(ObjectList {
+                objects: Vec::new(),
+                truncated: true,
+            });
+        }
+
+        let mut objects = Vec::new();
+        let mut continuation_token = None;
+        loop {
+            let remaining = max_objects.saturating_sub(objects.len());
+            if remaining == 0 {
+                return Ok(ObjectList {
+                    objects,
+                    truncated: continuation_token.is_some(),
+                });
+            }
+            let page_size = remaining.min(1_000);
+            let (page, status) = self
+                .bucket
+                .list_page(
+                    prefix.to_string(),
+                    None,
+                    continuation_token,
+                    None,
+                    Some(page_size),
+                )
+                .await?;
+            if !(200..300).contains(&status) {
+                return Err(StoreError::Backend(S3Error::HttpFailWithBody(
+                    status,
+                    "unexpected list status".to_string(),
+                )));
+            }
+            objects.extend(page.contents.into_iter().map(|object| StoredObject {
+                key: object.key,
+                size: object.size,
+                last_modified: object.last_modified,
+            }));
+            continuation_token = page.next_continuation_token;
+            if continuation_token.is_none() {
+                return Ok(ObjectList {
+                    objects,
+                    truncated: false,
+                });
+            }
+        }
+    }
+
+    #[cfg(test)]
+    pub(crate) async fn delete_for_test(&self, key: &str) {
+        let _ = self.bucket.delete_object(key).await;
+    }
+
     /// Create-only write of a content-addressed object (pack or manifest).
     ///
     /// **The caller does not choose the key.** It is derived as

crates/buzz-relay/src/main.rs

@@ -133,7 +133,14 @@ async fn main() -> anyhow::Result<()> {
     );
 
     let usage_interval_secs = usage_metrics_interval_secs();
-    let usage_idle_timeout_secs = usage_metrics_idle_timeout_secs(usage_interval_secs);
+    let git_gc_config =
+        buzz_relay::api::git::gc::GitGcWorkerConfig::from_env().map_err(anyhow::Error::msg)?;
+    let metrics_refresh_interval_secs = if git_gc_config.enabled {
+        usage_interval_secs.max(git_gc_config.interval.as_secs())
+    } else {
+        usage_interval_secs
+    };
+    let usage_idle_timeout_secs = usage_metrics_idle_timeout_secs(metrics_refresh_interval_secs);
     relay_metrics::install(config.metrics_port, usage_idle_timeout_secs);
     metrics::gauge!("buzz_audit_enabled").set(if config.audit_enabled { 1.0 } else { 0.0 });
     info!(
@@ -498,6 +505,21 @@ async fn main() -> anyhow::Result<()> {
         );
     }
 
+    if git_gc_config.enabled {
+        info!(
+            interval_seconds = git_gc_config.interval.as_secs(),
+            max_pointers = git_gc_config.limits.max_pointers,
+            max_objects_per_prefix = git_gc_config.limits.max_objects_per_prefix,
+            max_manifest_bytes = git_gc_config.limits.max_manifest_bytes,
+            scan_timeout_seconds = git_gc_config.limits.timeout.as_secs(),
+            "Git object-store GC dry-run inventory enabled"
+        );
+        tokio::spawn(buzz_relay::api::git::gc::run_git_gc_worker(
+            Arc::clone(&state),
+            git_gc_config,
+        ));
+    }
+
     // NIP-43: reconcile the event-backed roster for every provisioned
     // community before opening the listener. `relay_members` is canonical;
     // this repairs pre-snapshot communities and any publication that failed
@@ -1371,7 +1393,7 @@ fn dropped_in_memory_keys(
 async fn run_usage_metrics_tick(
     state: &AppState,
     emission_scope: &EmissionScope,
-    leader: &mut Option<buzz_db::UsageMetricsLeader>,
+    leader: &mut Option<buzz_db::AdvisoryLockLeader>,
     emitted_in_memory: &mut HashSet<InMemoryMetricKey>,
 ) -> anyhow::Result<()> {
     let host_map: HashMap<Uuid, String> = match state.db.usage_community_hosts().await {
@@ -1399,10 +1421,7 @@ async fn run_usage_metrics_tick(
         }
     }
     if leader.is_none() && !demoted {
-        *leader = state
-            .db
-            .try_lock_usage_metrics(USAGE_METRICS_LOCK_KEY)
-            .await?;
+        *leader = state.db.try_advisory_lock(USAGE_METRICS_LOCK_KEY).await?;
         if leader.is_some() {
             info!("Acquired usage metrics leader lock");
         }

crates/buzz-relay/src/metrics.rs

@@ -102,6 +102,11 @@ pub fn install(port: u16, gauge_idle_timeout_secs: u64) {
             &GIT_DURATION_BUCKETS_S,
         )
         .expect("valid git compaction duration bucket boundaries")
+        .set_buckets_for_metric(
+            Matcher::Full("buzz_git_object_store_gc_scan_seconds".to_owned()),
+            &GIT_DURATION_BUCKETS_S,
+        )
+        .expect("valid git GC inventory duration bucket boundaries")
         .set_buckets_for_metric(
             Matcher::Full("buzz_git_hydrate_bytes".to_owned()),
             &GIT_BYTES_BUCKETS,

deploy/charts/buzz/templates/deployment.yaml

@@ -150,6 +150,12 @@ spec:
             - { name: BUZZ_GIT_PACK_CACHE_MAX_CONCURRENT_POPULATIONS, value: {{ .Values.git.packCacheMaxConcurrentPopulations | quote }} }
             - { name: BUZZ_GIT_MAX_REPOS_PER_PUBKEY, value: {{ .Values.git.maxReposPerPubkey | quote }} }
             - { name: BUZZ_GIT_MAX_CONCURRENT_OPS, value: {{ .Values.git.maxConcurrentOps | quote }} }
+            - { name: BUZZ_GIT_GC_ENABLED, value: {{ .Values.git.gc.enabled | quote }} }
+            - { name: BUZZ_GIT_GC_INTERVAL_SECS, value: {{ .Values.git.gc.intervalSeconds | quote }} }
+            - { name: BUZZ_GIT_GC_MAX_POINTERS, value: {{ .Values.git.gc.maxPointers | quote }} }
+            - { name: BUZZ_GIT_GC_MAX_OBJECTS_PER_PREFIX, value: {{ .Values.git.gc.maxObjectsPerPrefix | quote }} }
+            - { name: BUZZ_GIT_GC_MAX_MANIFEST_BYTES_PER_SCAN, value: {{ .Values.git.gc.maxManifestBytesPerScan | quote }} }
+            - { name: BUZZ_GIT_GC_SCAN_TIMEOUT_SECS, value: {{ .Values.git.gc.scanTimeoutSeconds | quote }} }
 
             # ── S3 (non-secret) ──────────────────────────────────────
             {{- $s3Endpoint := include "buzz.s3Endpoint" . }}

deploy/charts/buzz/values.schema.json

@@ -204,7 +204,19 @@
         "packCacheMaxConcurrentPopulations": { "type": "integer", "minimum": 1 },
         "packCacheVolumeSize": { "type": "string", "pattern": "^[0-9]+(\\.[0-9]+)?(E|P|T|G|M|K|Ei|Pi|Ti|Gi|Mi|Ki)?$" },
         "maxReposPerPubkey": { "type": "integer", "minimum": 1 },
-        "maxConcurrentOps": { "type": "integer", "minimum": 1 }
+        "maxConcurrentOps": { "type": "integer", "minimum": 1 },
+        "gc": {
+          "type": "object",
+          "additionalProperties": false,
+          "properties": {
+            "enabled": { "type": "boolean" },
+            "intervalSeconds": { "type": "integer", "minimum": 1 },
+            "maxPointers": { "type": "integer", "minimum": 1 },
+            "maxObjectsPerPrefix": { "type": "integer", "minimum": 1 },
+            "maxManifestBytesPerScan": { "type": "integer", "minimum": 1 },
+            "scanTimeoutSeconds": { "type": "integer", "minimum": 1 }
+          }
+        }
       }
     },
     "migrate": {

deploy/charts/buzz/values.yaml

@@ -329,6 +329,13 @@ git:
   packCacheVolumeSize: 7Gi       # per-pod emptyDir; includes cold-population staging
   maxReposPerPubkey: 100
   maxConcurrentOps: 20
+  gc:
+    enabled: false               # dry-run inventory only; never deletes objects
+    intervalSeconds: 3600
+    maxPointers: 10000
+    maxObjectsPerPrefix: 10000
+    maxManifestBytesPerScan: 536870912
+    scanTimeoutSeconds: 300
 
 # ── Migrations ───────────────────────────────────────────────────────────────
 # Relay runs sqlx migrations at startup via BUZZ_AUTO_MIGRATE=true.

docs/git-on-object-storage.md

@@ -158,6 +158,17 @@ assumption for any S3-compatible backend.
   orphaned could 404 a concurrent reader mid-hydrate — see Theorem 2's reliance
   on every named pack being GETtable.)
 
+  The relay includes an opt-in, leader-elected **dry-run inventory** for this
+  future GC (`BUZZ_GIT_GC_ENABLED=true`). It lists only `manifests/`, `packs/`,
+  and `idx/`, computes reachability from every current `repos/*/pointer`, and
+  exports observed candidate counts and bytes. Pointer count, listed objects,
+  manifest bytes, and scan duration are bounded; truncated prefix inventories
+  are explicitly reported as incomplete. It never deletes. A candidate report
+  is not a deletion proof: production reclamation additionally requires
+  durable evidence that an object stayed unreachable longer than the maximum
+  read lifetime and coordination preventing a concurrent publisher from
+  making an old content-addressed object reachable during sweep.
+
 - **(A2) Strong read-after-write.** A read issued after a successful `PUT`
   observes that write. (AWS S3 provides this for all regions and all
   PUT/DELETE.)