diff --git a/.env.example b/.env.example index 3dccd6a174..1e69c6a0e3 100644 --- a/.env.example +++ b/.env.example @@ -75,14 +75,6 @@ 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 diff --git a/bin/.helm-3.16.4.pkg b/bin/.helm-3.16.4.pkg deleted file mode 120000 index 383f4511d4..0000000000 --- a/bin/.helm-3.16.4.pkg +++ /dev/null @@ -1 +0,0 @@ -hermit \ No newline at end of file diff --git a/bin/helm b/bin/helm deleted file mode 120000 index 840f4f2a13..0000000000 --- a/bin/helm +++ /dev/null @@ -1 +0,0 @@ -.helm-3.16.4.pkg \ No newline at end of file diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 3eeff5dee7..fdd72c3c32 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -198,16 +198,16 @@ pub struct DbPoolStats { pub max: u32, } -/// Owns a detached Postgres session holding a relay advisory leader lock. +/// Owns the detached Postgres session holding the relay usage-metrics advisory 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 AdvisoryLockLeader { +pub struct UsageMetricsLeader { connection: PgConnection, } -impl AdvisoryLockLeader { +impl UsageMetricsLeader { /// Returns whether the lock-owning session is still reachable. /// /// Bounded to 5 seconds — a blackholed connection (no RST) would otherwise @@ -508,20 +508,23 @@ impl Db { }) } - /// Try to acquire a detached session advisory lock for singleton relay work. + /// Try to acquire the detached session advisory lock for relay usage metrics. /// /// 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_advisory_lock(&self, lock_key: i64) -> Result> { + pub async fn try_lock_usage_metrics( + &self, + lock_key: i64, + ) -> Result> { 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(AdvisoryLockLeader { + Ok(Some(UsageMetricsLeader { connection: connection.detach(), })) } else { @@ -4660,14 +4663,14 @@ mod tests { let key = 0x4255_5A5A_4D45_5452; let mut leader = first - .try_advisory_lock(key) + .try_lock_usage_metrics(key) .await .expect("first lock attempt") .expect("first database handle becomes leader"); assert!(leader.is_live().await, "lock owner remains reachable"); assert!( second - .try_advisory_lock(key) + .try_lock_usage_metrics(key) .await .expect("second lock attempt") .is_none(), @@ -4677,7 +4680,7 @@ mod tests { drop(leader); assert!( second - .try_advisory_lock(key) + .try_lock_usage_metrics(key) .await .expect("lock attempt after leader drop") .is_some(), diff --git a/crates/buzz-relay/src/api/git/gc.rs b/crates/buzz-relay/src/api/git/gc.rs deleted file mode 100644 index 992b8c0220..0000000000 --- a/crates/buzz-relay/src/api/git/gc.rs +++ /dev/null @@ -1,529 +0,0 @@ -//! 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 { - 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, config: GitGcWorkerConfig) { - if !config.enabled { - return; - } - - let jitter_bound = config.interval.as_secs().max(1); - tokio::time::sleep(Duration::from_secs(rand::random::() % jitter_bound)).await; - let mut interval = tokio::time::interval(config.interval); - interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); - let mut leader: Option = 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 { - 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 { - 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, - remaining_manifest_bytes: u64, -) -> Result { - 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) -> (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 { - 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 { - match std::env::var(name) { - Ok(value) => value - .parse::() - .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 { - match std::env::var(name) { - Ok(value) => value - .parse::() - .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, ®ion) - .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; - } -} diff --git a/crates/buzz-relay/src/api/git/mod.rs b/crates/buzz-relay/src/api/git/mod.rs index c6c4370274..ab0510fbeb 100644 --- a/crates/buzz-relay/src/api/git/mod.rs +++ b/crates/buzz-relay/src/api/git/mod.rs @@ -23,7 +23,6 @@ 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; diff --git a/crates/buzz-relay/src/api/git/store.rs b/crates/buzz-relay/src/api/git/store.rs index 8712509d0c..43d210e648 100644 --- a/crates/buzz-relay/src/api/git/store.rs +++ b/crates/buzz-relay/src/api/git/store.rs @@ -145,26 +145,6 @@ 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, - /// 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}")] @@ -260,70 +240,6 @@ 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 { - 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 diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 5f04b894d1..eed4a53f8e 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -133,14 +133,7 @@ async fn main() -> anyhow::Result<()> { ); let usage_interval_secs = usage_metrics_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); + let usage_idle_timeout_secs = usage_metrics_idle_timeout_secs(usage_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!( @@ -505,21 +498,6 @@ 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 @@ -1393,7 +1371,7 @@ fn dropped_in_memory_keys( async fn run_usage_metrics_tick( state: &AppState, emission_scope: &EmissionScope, - leader: &mut Option, + leader: &mut Option, emitted_in_memory: &mut HashSet, ) -> anyhow::Result<()> { let host_map: HashMap = match state.db.usage_community_hosts().await { @@ -1421,7 +1399,10 @@ async fn run_usage_metrics_tick( } } if leader.is_none() && !demoted { - *leader = state.db.try_advisory_lock(USAGE_METRICS_LOCK_KEY).await?; + *leader = state + .db + .try_lock_usage_metrics(USAGE_METRICS_LOCK_KEY) + .await?; if leader.is_some() { info!("Acquired usage metrics leader lock"); } diff --git a/crates/buzz-relay/src/metrics.rs b/crates/buzz-relay/src/metrics.rs index 7ff27fb5f8..16e521a44e 100644 --- a/crates/buzz-relay/src/metrics.rs +++ b/crates/buzz-relay/src/metrics.rs @@ -102,11 +102,6 @@ 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, diff --git a/deploy/charts/buzz/templates/deployment.yaml b/deploy/charts/buzz/templates/deployment.yaml index 5b22566dd4..f8d67de31d 100644 --- a/deploy/charts/buzz/templates/deployment.yaml +++ b/deploy/charts/buzz/templates/deployment.yaml @@ -150,12 +150,6 @@ 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" . }} diff --git a/deploy/charts/buzz/values.schema.json b/deploy/charts/buzz/values.schema.json index 16d10a67e2..203fd9b69b 100644 --- a/deploy/charts/buzz/values.schema.json +++ b/deploy/charts/buzz/values.schema.json @@ -204,19 +204,7 @@ "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 }, - "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 } - } - } + "maxConcurrentOps": { "type": "integer", "minimum": 1 } } }, "migrate": { diff --git a/deploy/charts/buzz/values.yaml b/deploy/charts/buzz/values.yaml index add4c78f1a..27c6503793 100644 --- a/deploy/charts/buzz/values.yaml +++ b/deploy/charts/buzz/values.yaml @@ -329,13 +329,6 @@ 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. diff --git a/docs/git-on-object-storage.md b/docs/git-on-object-storage.md index a84c349c2f..1d87720edc 100644 --- a/docs/git-on-object-storage.md +++ b/docs/git-on-object-storage.md @@ -158,17 +158,6 @@ 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.)