diff --git a/.env.example b/.env.example index 3a642bfad9..1e69c6a0e3 100644 --- a/.env.example +++ b/.env.example @@ -66,12 +66,15 @@ RELAY_URL=ws://localhost:3000 # ----------------------------------------------------------------------------- # Git (NIP-34 bare repositories) # ----------------------------------------------------------------------------- -# Root directory for bare git repos. Repos are stored at -# {path}/{owner_hex}/{repo_id}.git/. Default: ./repos (relative to CWD). -# Set an absolute path to keep repos stable across worktrees. +# Root directory for ephemeral Git workspaces and the disposable pack cache. +# Default: ./repos (relative to CWD). # BUZZ_GIT_REPO_PATH=./repos # BUZZ_GIT_MAX_PACK_BYTES=524288000 # BUZZ_GIT_MAX_REPO_BYTES=1048576000 +# Process-local immutable pack/index cache. Zero disables retention. +# BUZZ_GIT_PACK_CACHE_PATH=./repos/.pack-cache +# BUZZ_GIT_PACK_CACHE_MAX_BYTES=5368709120 +# BUZZ_GIT_PACK_CACHE_MAX_CONCURRENT_POPULATIONS=2 # ----------------------------------------------------------------------------- # Media Upload Admission diff --git a/crates/buzz-relay/src/api/git/hydrate.rs b/crates/buzz-relay/src/api/git/hydrate.rs index 4488e66bb4..69bf9bebda 100644 --- a/crates/buzz-relay/src/api/git/hydrate.rs +++ b/crates/buzz-relay/src/api/git/hydrate.rs @@ -6,11 +6,11 @@ //! //! 1. GET pointer → manifest digest. //! 2. GET manifest (digest-verified) → parsed [`Manifest`]. -//! 3. GET every pack the manifest names (digest-verified, in parallel). -//! 4. **Phase 1** — for each pack: write `pack-.pack`; install and -//! validate a cached `idx/` sidecar when present, otherwise run -//! `git index-pack` to materialize `.idx`. Failure here tears down the -//! tempdir with no refs/HEAD ever written. +//! 3. Resolve every pack through the bounded local content-addressed cache. +//! 4. **Phase 1** — hard-link or copy each verified pack/index pair into the +//! ephemeral repo. A cache miss reads object storage and validates or +//! generates the index once per digest. Failure here tears down the tempdir +//! with no refs/HEAD ever written. //! 5. **Phase 2** — only after all packs are indexed: write loose refs and //! HEAD from the manifest. //! @@ -19,8 +19,8 @@ //! mid-stream. Sami/Max named this explicitly. //! //! The returned [`HydratedRepo`] owns a [`tempfile::TempDir`]; dropping it -//! cleans up. Every read currently re-hydrates from scratch — naive but -//! correct; caching is named follow-up work. +//! cleans up. Cached pack/index pairs are immutable performance state; object +//! storage remains authoritative. // Public surface is consumed by `transport.rs` after Eva's `AppState::git_store` // wires in; the items below are intentionally `pub` to keep the consumer-side @@ -35,6 +35,7 @@ use tokio::process::Command; use super::cas_publish::ParentState; use super::manifest::{is_hex_oid, is_safe_refname, pointer_key, Manifest, ManifestError}; +use super::pack_cache::GitPackCache; use super::store::{ETag, GitStore, StoreError}; use buzz_core::TenantContext; @@ -75,6 +76,19 @@ impl HydratedRepo { } } +/// Process-local resources and limits used to materialize one repository. +#[derive(Clone, Copy)] +pub struct HydrationOptions<'a> { + /// Immutable pack/index cache shared by Git operations in this process. + pub pack_cache: &'a GitPackCache, + /// Parent directory for ephemeral bare repositories. + pub scratch_dir: &'a Path, + /// Maximum bytes accepted for one pack object. + pub max_pack_bytes: u64, + /// Maximum aggregate pack bytes accepted for one repository. + pub max_repo_bytes: u64, +} + /// Hydration errors. /// /// "Repo doesn't exist" is signalled by `Ok(None)` from `hydrate_for_read`, @@ -112,21 +126,10 @@ pub async fn hydrate_for_read( ctx: &TenantContext, owner: &str, repo: &str, - scratch_dir: &Path, - max_pack_bytes: u64, - max_repo_bytes: u64, + options: HydrationOptions<'_>, ) -> Result, HydrateError> { let started_at = std::time::Instant::now(); - let result = hydrate_for_read_inner( - store, - ctx, - owner, - repo, - scratch_dir, - max_pack_bytes, - max_repo_bytes, - ) - .await; + let result = hydrate_for_read_inner(store, ctx, owner, repo, options).await; let outcome = match &result { Ok(Some(repo)) => { metrics::histogram!("buzz_git_hydrate_bytes").record(repo.hydrated_bytes() as f64); @@ -151,23 +154,12 @@ async fn hydrate_for_read_inner( ctx: &TenantContext, owner: &str, repo: &str, - scratch_dir: &Path, - max_pack_bytes: u64, - max_repo_bytes: u64, + options: HydrationOptions<'_>, ) -> Result, HydrateError> { let Some((_etag, _digest, manifest)) = load_pointer(store, ctx, owner, repo).await? else { return Ok(None); }; - Ok(Some( - materialize_manifest( - store, - &manifest, - scratch_dir, - max_pack_bytes, - max_repo_bytes, - ) - .await?, - )) + Ok(Some(materialize_manifest(store, &manifest, options).await?)) } /// Load the current manifest for a read path without materializing pack objects. @@ -212,20 +204,11 @@ pub async fn hydrate_for_write( ctx: &TenantContext, owner: &str, repo: &str, - scratch_dir: &Path, - max_pack_bytes: u64, - max_repo_bytes: u64, + options: HydrationOptions<'_>, ) -> Result<(HydratedRepo, ParentState), HydrateError> { match load_pointer(store, ctx, owner, repo).await? { Some((etag, digest, manifest)) => { - let repo = materialize_manifest( - store, - &manifest, - scratch_dir, - max_pack_bytes, - max_repo_bytes, - ) - .await?; + let repo = materialize_manifest(store, &manifest, options).await?; let parent = ParentState::from_loaded(etag, digest, manifest); Ok((repo, parent)) } @@ -233,8 +216,9 @@ pub async fn hydrate_for_write( // First push: empty bare repo. No packs to fetch, no refs to // install. `receive-pack` will accept whatever the client // sends; `cas_publish` will use `If-None-Match: *`. - let tempdir = TempDir::new_in(scratch_dir) - .map_err(|e| HydrateError::Hydrate(format!("tempdir in {scratch_dir:?}: {e}")))?; + let tempdir = TempDir::new_in(options.scratch_dir).map_err(|e| { + HydrateError::Hydrate(format!("tempdir in {:?}: {e}", options.scratch_dir)) + })?; let path = tempdir.path().to_path_buf(); run_git(&path, &["init", "--bare", "--quiet"]).await?; Ok(( @@ -280,7 +264,7 @@ async fn load_pointer( Ok(Some((etag, digest, manifest))) } -async fn get_verified_limited( +pub(super) async fn get_verified_limited( store: &GitStore, key: &str, digest: &str, @@ -304,13 +288,11 @@ async fn get_verified_limited( async fn materialize_manifest( store: &GitStore, manifest: &Manifest, - scratch_dir: &Path, - max_pack_bytes: u64, - max_repo_bytes: u64, + options: HydrationOptions<'_>, ) -> Result { // Init bare repo. - let tempdir = TempDir::new_in(scratch_dir) - .map_err(|e| HydrateError::Hydrate(format!("tempdir in {scratch_dir:?}: {e}")))?; + let tempdir = TempDir::new_in(options.scratch_dir) + .map_err(|e| HydrateError::Hydrate(format!("tempdir in {:?}: {e}", options.scratch_dir)))?; let path = tempdir.path().to_path_buf(); run_git(&path, &["init", "--bare", "--quiet"]).await?; @@ -324,27 +306,25 @@ async fn materialize_manifest( let digest = key .strip_prefix("packs/") .ok_or_else(|| HydrateError::Hydrate(format!("malformed pack key {key:?}")))?; - let bytes = get_verified_limited(store, key, digest, max_pack_bytes).await?; - let pack_bytes = u64::try_from(bytes.len()).unwrap_or(u64::MAX); - if pack_bytes > max_pack_bytes { + let pack_bytes = options + .pack_cache + .materialize_pack(store, key, digest, &pack_dir, options.max_pack_bytes) + .await?; + if pack_bytes > options.max_pack_bytes { return Err(HydrateError::ResourceLimit(format!( - "pack {digest} is {pack_bytes} bytes (max {max_pack_bytes})" + "pack {digest} is {pack_bytes} bytes (max {})", + options.max_pack_bytes ))); } hydrated_bytes = hydrated_bytes.checked_add(pack_bytes).ok_or_else(|| { HydrateError::ResourceLimit("repo byte count overflowed u64".to_string()) })?; - if hydrated_bytes > max_repo_bytes { + if hydrated_bytes > options.max_repo_bytes { return Err(HydrateError::ResourceLimit(format!( - "repo needs {hydrated_bytes} hydrated bytes (max {max_repo_bytes})" + "repo needs {hydrated_bytes} hydrated bytes (max {})", + options.max_repo_bytes ))); } - - let pack_path = pack_dir.join(format!("pack-{digest}.pack")); - tokio::fs::write(&pack_path, &bytes) - .await - .map_err(|e| HydrateError::Hydrate(format!("write pack {digest}: {e}")))?; - install_or_generate_idx(store, &path, digest, &pack_path).await?; } // Phase 2: install refs and HEAD. After this point, the repo advertises. @@ -393,14 +373,15 @@ async fn materialize_manifest( }) } -async fn install_or_generate_idx( +pub(super) async fn install_or_generate_idx( store: &GitStore, repo_path: &Path, pack_digest: &str, pack_path: &Path, + max_idx_bytes: u64, ) -> Result<(), HydrateError> { let idx_path = pack_path.with_extension("idx"); - match store.get_idx(pack_digest).await { + match store.get_idx(pack_digest, max_idx_bytes).await { Ok(Some(idx_bytes)) => { tokio::fs::write(&idx_path, &idx_bytes) .await @@ -438,7 +419,27 @@ async fn install_or_generate_idx( let pack_path_str = pack_path .to_str() .ok_or_else(|| HydrateError::Hydrate("pack path is not valid utf-8".to_string()))?; - run_git(repo_path, &["index-pack", pack_path_str]).await + run_git(repo_path, &["index-pack", pack_path_str]).await?; + enforce_idx_size(&idx_path, pack_digest, max_idx_bytes).await +} + +async fn enforce_idx_size( + idx_path: &Path, + pack_digest: &str, + max_idx_bytes: u64, +) -> Result<(), HydrateError> { + let idx_bytes = tokio::fs::metadata(&idx_path) + .await + .map_err(|error| { + HydrateError::Hydrate(format!("stat generated idx {pack_digest}: {error}")) + })? + .len(); + if idx_bytes > max_idx_bytes { + return Err(HydrateError::ResourceLimit(format!( + "generated idx {pack_digest} is {idx_bytes} bytes (max {max_idx_bytes})" + ))); + } + Ok(()) } /// Run `git ` in `cwd`, fail on non-zero exit. @@ -499,6 +500,17 @@ mod tests { assert!(!is_hex_oid("")); } + #[tokio::test] + async fn generated_idx_size_is_bounded() { + let scratch = TempDir::new().expect("scratch"); + let idx_path = scratch.path().join("pack-test.idx"); + tokio::fs::write(&idx_path, [0u8; 2]).await.expect("idx"); + + let result = enforce_idx_size(&idx_path, "test", 1).await; + + assert!(matches!(result, Err(HydrateError::ResourceLimit(_)))); + } + // `pointer_key` is tested in `super::manifest::tests` — single source. fn tenant() -> TenantContext { @@ -520,10 +532,20 @@ mod tests { packs: Vec::new(), parent: None, }; + let cache = GitPackCache::new(scratch.path(), u64::MAX, 2).expect("cache"); - let hydrated = materialize_manifest(&store, &manifest, scratch.path(), u64::MAX, u64::MAX) - .await - .expect("materialize empty repo"); + let hydrated = materialize_manifest( + &store, + &manifest, + HydrationOptions { + pack_cache: &cache, + scratch_dir: scratch.path(), + max_pack_bytes: u64::MAX, + max_repo_bytes: u64::MAX, + }, + ) + .await + .expect("materialize empty repo"); assert!(hydrated.path().starts_with(scratch.path())); } @@ -643,11 +665,22 @@ mod tests { // Hydrate. let scratch = TempDir::new().unwrap(); - let hydrated = - hydrate_for_read(&st, &ctx, &owner, repo, scratch.path(), u64::MAX, u64::MAX) - .await - .expect("hydrate") - .expect("hydrate Some"); + let cache = GitPackCache::new(scratch.path(), u64::MAX, 2).expect("cache"); + let hydrated = hydrate_for_read( + &st, + &ctx, + &owner, + repo, + HydrationOptions { + pack_cache: &cache, + scratch_dir: scratch.path(), + max_pack_bytes: u64::MAX, + max_repo_bytes: u64::MAX, + }, + ) + .await + .expect("hydrate") + .expect("hydrate Some"); eprintln!("hydrated to {}", hydrated.path().display()); // The hydrated repo must list the same ref with the same oid. @@ -718,14 +751,18 @@ mod tests { let owner = format!("nope-{}", uuid::Uuid::new_v4()); let ctx = tenant(); let scratch = TempDir::new().unwrap(); + let cache = GitPackCache::new(scratch.path(), u64::MAX, 2).expect("cache"); let result = hydrate_for_read( &st, &ctx, &owner, "ghost", - scratch.path(), - u64::MAX, - u64::MAX, + HydrationOptions { + pack_cache: &cache, + scratch_dir: scratch.path(), + max_pack_bytes: u64::MAX, + max_repo_bytes: u64::MAX, + }, ) .await .expect("ok"); @@ -774,14 +811,18 @@ mod tests { } let scratch = TempDir::new().unwrap(); + let cache = GitPackCache::new(scratch.path(), u64::MAX, 2).expect("cache"); let hydrated = hydrate_for_read( &st, &ctx, &owner, "void", - scratch.path(), - u64::MAX, - u64::MAX, + HydrationOptions { + pack_cache: &cache, + scratch_dir: scratch.path(), + max_pack_bytes: u64::MAX, + max_repo_bytes: u64::MAX, + }, ) .await .expect("hydrate") diff --git a/crates/buzz-relay/src/api/git/mod.rs b/crates/buzz-relay/src/api/git/mod.rs index 730ca225f2..ab0510fbeb 100644 --- a/crates/buzz-relay/src/api/git/mod.rs +++ b/crates/buzz-relay/src/api/git/mod.rs @@ -27,6 +27,7 @@ pub mod hook; pub mod hydrate; pub mod manifest; pub mod manifest_event; +pub mod pack_cache; pub mod policy; pub mod store; pub mod transport; diff --git a/crates/buzz-relay/src/api/git/pack_cache.rs b/crates/buzz-relay/src/api/git/pack_cache.rs new file mode 100644 index 0000000000..fdac4c60a3 --- /dev/null +++ b/crates/buzz-relay/src/api/git/pack_cache.rs @@ -0,0 +1,686 @@ +//! Bounded local cache for immutable Git pack/index pairs. +//! +//! Object storage remains the durable source of truth. Cache entries are +//! content-addressed by the verified pack digest and are published by an +//! atomic directory rename only after both the pack and index are ready. + +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant, SystemTime}; + +use dashmap::mapref::entry::Entry; +use dashmap::DashMap; +use tempfile::Builder; + +use super::hydrate::{get_verified_limited, install_or_generate_idx, HydrateError}; +use super::store::GitStore; + +const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(60); +const STALE_SESSION_AGE: Duration = Duration::from_secs(10 * 60); + +#[derive(Debug)] +struct CachedPack { + dir: PathBuf, + pack_path: PathBuf, + idx_path: PathBuf, + pack_bytes: u64, + total_bytes: u64, + /// Keeps an over-capacity staging directory alive until it is linked into + /// the request workspace. Normal cached entries are atomically renamed and + /// do not need a temporary owner. + _temporary: Option, +} + +impl CachedPack { + fn from_dir(dir: PathBuf, digest: &str) -> Option { + let pack_path = dir.join(format!("pack-{digest}.pack")); + let idx_path = dir.join(format!("pack-{digest}.idx")); + let pack_bytes = regular_file_len(&pack_path)?; + let idx_bytes = regular_file_len(&idx_path)?; + Some(Self { + dir, + pack_path, + idx_path, + pack_bytes, + total_bytes: pack_bytes.saturating_add(idx_bytes), + _temporary: None, + }) + } +} + +struct CacheRecord { + entry: Arc, + last_used: u64, +} + +#[derive(Default)] +struct CacheState { + entries: HashMap, + total_bytes: u64, + tick: u64, +} + +/// Process-local, byte-bounded cache of immutable pack/index pairs. +pub struct GitPackCache { + _session_dir: tempfile::TempDir, + heartbeat_task: Option>, + root: PathBuf, + max_bytes: u64, + population_semaphore: tokio::sync::Semaphore, + state: Mutex, + flights: DashMap>, +} + +struct PopulationFlight { + lock: tokio::sync::Mutex<()>, + participants: AtomicUsize, +} + +struct FlightParticipant<'a> { + cache: &'a GitPackCache, + digest: &'a str, + flight: Arc, +} + +struct PopulationPermit<'a> { + _permit: tokio::sync::SemaphorePermit<'a>, +} + +impl Drop for PopulationPermit<'_> { + fn drop(&mut self) { + metrics::gauge!("buzz_git_pack_cache_populations_active").decrement(1.0); + } +} + +impl Drop for FlightParticipant<'_> { + fn drop(&mut self) { + if self.flight.participants.fetch_sub(1, Ordering::AcqRel) == 1 { + self.cache.remove_flight(self.digest, &self.flight); + } + } +} + +impl GitPackCache { + /// Create an isolated process-lifetime cache beneath `cache_parent`. + pub fn new( + cache_parent: &Path, + max_bytes: u64, + max_concurrent_populations: usize, + ) -> Result { + if max_concurrent_populations == 0 { + return Err("git pack cache population concurrency must be positive".to_string()); + } + std::fs::create_dir_all(cache_parent) + .map_err(|error| format!("create git pack cache {cache_parent:?}: {error}"))?; + if std::fs::symlink_metadata(cache_parent) + .map_err(|error| format!("stat git pack cache {cache_parent:?}: {error}"))? + .file_type() + .is_symlink() + { + return Err(format!( + "git pack cache path must not be a symlink: {cache_parent:?}" + )); + } + cleanup_stale_sessions(cache_parent); + let session_dir = Builder::new() + .prefix("session-") + .tempdir_in(cache_parent) + .map_err(|error| format!("create git pack cache session: {error}"))?; + let root = session_dir.path().to_path_buf(); + let heartbeat_path = root.join(".heartbeat"); + std::fs::write(&heartbeat_path, b"") + .map_err(|error| format!("create git pack cache heartbeat: {error}"))?; + let heartbeat_task = tokio::runtime::Handle::try_current().ok().map(|runtime| { + runtime.spawn(async move { + let mut interval = tokio::time::interval(HEARTBEAT_INTERVAL); + interval.tick().await; + loop { + interval.tick().await; + if tokio::fs::write(&heartbeat_path, b"").await.is_err() { + break; + } + } + }) + }); + let cache = Self { + _session_dir: session_dir, + heartbeat_task, + root, + max_bytes, + population_semaphore: tokio::sync::Semaphore::new(max_concurrent_populations), + state: Mutex::new(CacheState::default()), + flights: DashMap::new(), + }; + emit_size_metrics(&CacheState::default()); + Ok(cache) + } + + /// Materialize one verified pack/index pair into a request workspace. + /// + /// Concurrent misses for the same digest share one population flight. + pub async fn materialize_pack( + &self, + store: &GitStore, + object_key: &str, + digest: &str, + destination: &Path, + max_pack_bytes: u64, + ) -> Result { + validate_digest(digest)?; + if let Some(entry) = self.lookup(digest) { + metrics::counter!("buzz_git_pack_cache_lookups_total", "result" => "hit").increment(1); + let pack_bytes = entry.pack_bytes; + if let Ok(()) = install_entry(&entry, digest, destination).await { + drop(entry); + self.prune(); + return Ok(pack_bytes); + } + self.invalidate(digest); + } + + let (flight, joined) = match self.flights.entry(digest.to_string()) { + Entry::Occupied(entry) => { + entry.get().participants.fetch_add(1, Ordering::Relaxed); + (Arc::clone(entry.get()), true) + } + Entry::Vacant(entry) => { + let flight = Arc::new(PopulationFlight { + lock: tokio::sync::Mutex::new(()), + participants: AtomicUsize::new(1), + }); + entry.insert(Arc::clone(&flight)); + (flight, false) + } + }; + metrics::counter!( + "buzz_git_pack_cache_lookups_total", + "result" => if joined { "coalesced" } else { "miss" } + ) + .increment(1); + + let flight_participant = FlightParticipant { + cache: self, + digest, + flight: Arc::clone(&flight), + }; + let guard = flight.lock.lock().await; + let result = if let Some(entry) = self.lookup(digest) { + install_entry(&entry, digest, destination) + .await + .map(|()| entry.pack_bytes) + } else { + let started_at = Instant::now(); + let populated = self + .populate(store, object_key, digest, max_pack_bytes) + .await; + let outcome = match &populated { + Ok(entry) if entry._temporary.is_some() => "bypass", + Ok(_) => "success", + Err(_) => "error", + }; + metrics::histogram!( + "buzz_git_pack_cache_populate_seconds", + "outcome" => outcome + ) + .record(started_at.elapsed().as_secs_f64()); + match populated { + Ok(entry) => install_entry(&entry, digest, destination) + .await + .map(|()| entry.pack_bytes), + Err(error) => Err(error), + } + }; + drop(guard); + drop(flight_participant); + self.prune(); + result + } + + fn lookup(&self, digest: &str) -> Option> { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + let next_tick = state.tick.saturating_add(1); + state.tick = next_tick; + let record = state.entries.get_mut(digest)?; + if !record.entry.pack_path.is_file() || !record.entry.idx_path.is_file() { + return None; + } + record.last_used = next_tick; + Some(Arc::clone(&record.entry)) + } + + async fn populate( + &self, + store: &GitStore, + object_key: &str, + digest: &str, + max_pack_bytes: u64, + ) -> Result, HydrateError> { + let wait_started_at = Instant::now(); + let permit = self + .population_semaphore + .acquire() + .await + .map_err(|_| HydrateError::Hydrate("pack cache population closed".to_string()))?; + metrics::histogram!("buzz_git_pack_cache_population_wait_seconds") + .record(wait_started_at.elapsed().as_secs_f64()); + metrics::gauge!("buzz_git_pack_cache_populations_active").increment(1.0); + let _population_permit = PopulationPermit { _permit: permit }; + let shard = self.root.join(&digest[..2]); + tokio::fs::create_dir_all(&shard) + .await + .map_err(|error| HydrateError::Hydrate(format!("create pack cache shard: {error}")))?; + let staging = Builder::new() + .prefix(".staging-") + .tempdir_in(&shard) + .map_err(|error| { + HydrateError::Hydrate(format!("create pack cache staging directory: {error}")) + })?; + let pack_path = staging.path().join(format!("pack-{digest}.pack")); + let bytes = get_verified_limited(store, object_key, digest, max_pack_bytes).await?; + let pack_bytes = u64::try_from(bytes.len()).unwrap_or(u64::MAX); + tokio::fs::write(&pack_path, &bytes) + .await + .map_err(|error| HydrateError::Hydrate(format!("write cached pack: {error}")))?; + install_or_generate_idx(store, staging.path(), digest, &pack_path, max_pack_bytes).await?; + let idx_path = pack_path.with_extension("idx"); + let idx_bytes = tokio::fs::metadata(&idx_path) + .await + .map_err(|error| HydrateError::Hydrate(format!("stat cached idx: {error}")))? + .len(); + let total_bytes = pack_bytes.saturating_add(idx_bytes); + + if self.max_bytes == 0 || total_bytes > self.max_bytes { + metrics::counter!("buzz_git_pack_cache_bypasses_total").increment(1); + return Ok(Arc::new(CachedPack { + dir: staging.path().to_path_buf(), + pack_path, + idx_path, + pack_bytes, + total_bytes, + _temporary: Some(staging), + })); + } + + let final_dir = shard.join(digest); + if let Some(entry) = CachedPack::from_dir(final_dir.clone(), digest) { + let entry = Arc::new(entry); + self.insert(digest.to_string(), Arc::clone(&entry)); + return Ok(entry); + } + if final_dir.exists() { + let _ = tokio::fs::remove_dir_all(&final_dir).await; + } + if let Err(error) = tokio::fs::rename(staging.path(), &final_dir).await { + // Another relay process sharing the scratch volume may have won + // the same content-addressed publication race. + if let Some(entry) = CachedPack::from_dir(final_dir.clone(), digest) { + let entry = Arc::new(entry); + self.insert(digest.to_string(), Arc::clone(&entry)); + return Ok(entry); + } + return Err(HydrateError::Hydrate(format!( + "publish cached pack: {error}" + ))); + } + let entry = Arc::new(CachedPack::from_dir(final_dir, digest).ok_or_else(|| { + HydrateError::Hydrate("published pack cache entry is incomplete".to_string()) + })?); + self.insert(digest.to_string(), Arc::clone(&entry)); + Ok(entry) + } + + fn insert(&self, digest: String, entry: Arc) { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + state.tick = state.tick.saturating_add(1); + let last_used = state.tick; + if let Some(previous) = state.entries.remove(&digest) { + state.total_bytes = state.total_bytes.saturating_sub(previous.entry.total_bytes); + } + state.total_bytes = state.total_bytes.saturating_add(entry.total_bytes); + state + .entries + .insert(digest, CacheRecord { entry, last_used }); + emit_size_metrics(&state); + } + + fn invalidate(&self, digest: &str) { + let removed = { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + let removed = state.entries.remove(digest); + if let Some(record) = &removed { + state.total_bytes = state.total_bytes.saturating_sub(record.entry.total_bytes); + } + emit_size_metrics(&state); + removed + }; + if let Some(record) = removed { + if Arc::strong_count(&record.entry) == 1 { + let _ = std::fs::remove_dir_all(&record.entry.dir); + } + } + } + + fn prune(&self) { + let mut removed = Vec::new(); + { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + while state.total_bytes > self.max_bytes { + let candidate = state + .entries + .iter() + .filter(|(_, record)| Arc::strong_count(&record.entry) == 1) + .min_by_key(|(_, record)| record.last_used) + .map(|(digest, _)| digest.clone()); + let Some(digest) = candidate else { + break; + }; + if let Some(record) = state.entries.remove(&digest) { + state.total_bytes = state.total_bytes.saturating_sub(record.entry.total_bytes); + removed.push(record.entry.dir.clone()); + metrics::counter!("buzz_git_pack_cache_evictions_total").increment(1); + } + } + emit_size_metrics(&state); + } + for dir in removed { + let _ = std::fs::remove_dir_all(dir); + } + } + + fn remove_flight(&self, digest: &str, flight: &Arc) { + if let Entry::Occupied(entry) = self.flights.entry(digest.to_string()) { + if Arc::ptr_eq(entry.get(), flight) && flight.participants.load(Ordering::Acquire) == 0 + { + entry.remove(); + } + } + } + + #[cfg(test)] + fn flight(&self, digest: &str) -> (Arc, bool) { + match self.flights.entry(digest.to_string()) { + Entry::Occupied(entry) => { + entry.get().participants.fetch_add(1, Ordering::Relaxed); + (Arc::clone(entry.get()), true) + } + Entry::Vacant(entry) => { + let flight = Arc::new(PopulationFlight { + lock: tokio::sync::Mutex::new(()), + participants: AtomicUsize::new(1), + }); + entry.insert(Arc::clone(&flight)); + (flight, false) + } + } + } +} + +impl Drop for GitPackCache { + fn drop(&mut self) { + if let Some(task) = self.heartbeat_task.take() { + task.abort(); + } + } +} + +async fn install_entry( + entry: &CachedPack, + digest: &str, + destination: &Path, +) -> Result<(), HydrateError> { + let destination_pack = destination.join(format!("pack-{digest}.pack")); + let destination_idx = destination.join(format!("pack-{digest}.idx")); + if tokio::fs::hard_link(&entry.pack_path, &destination_pack) + .await + .is_ok() + && tokio::fs::hard_link(&entry.idx_path, &destination_idx) + .await + .is_ok() + { + return Ok(()); + } + + let _ = tokio::fs::remove_file(&destination_pack).await; + let _ = tokio::fs::remove_file(&destination_idx).await; + tokio::fs::copy(&entry.pack_path, &destination_pack) + .await + .map_err(|error| HydrateError::Hydrate(format!("copy cached pack: {error}")))?; + if let Err(error) = tokio::fs::copy(&entry.idx_path, &destination_idx).await { + let _ = tokio::fs::remove_file(&destination_pack).await; + return Err(HydrateError::Hydrate(format!("copy cached idx: {error}"))); + } + metrics::counter!("buzz_git_pack_cache_copy_fallbacks_total").increment(1); + Ok(()) +} + +fn validate_digest(digest: &str) -> Result<(), HydrateError> { + if digest.len() == 64 + && digest + .chars() + .all(|character| character.is_ascii_hexdigit()) + { + Ok(()) + } else { + Err(HydrateError::Hydrate( + "pack cache digest is malformed".to_string(), + )) + } +} + +fn regular_file_len(path: &Path) -> Option { + let metadata = std::fs::symlink_metadata(path).ok()?; + metadata.file_type().is_file().then_some(metadata.len()) +} + +fn emit_size_metrics(state: &CacheState) { + metrics::gauge!("buzz_git_pack_cache_bytes").set(state.total_bytes as f64); + metrics::gauge!("buzz_git_pack_cache_entries").set(state.entries.len() as f64); +} + +fn cleanup_stale_sessions(cache_parent: &Path) { + cleanup_sessions_older_than(cache_parent, STALE_SESSION_AGE); +} + +fn cleanup_sessions_older_than(cache_parent: &Path, max_age: Duration) { + let Ok(entries) = std::fs::read_dir(cache_parent) else { + return; + }; + for entry in entries.flatten() { + let name = entry.file_name(); + if !name.to_string_lossy().starts_with("session-") + || !entry.file_type().is_ok_and(|file_type| file_type.is_dir()) + { + continue; + } + let heartbeat = entry.path().join(".heartbeat"); + let modified = std::fs::symlink_metadata(&heartbeat) + .or_else(|_| entry.metadata()) + .and_then(|metadata| metadata.modified()); + let stale = modified + .ok() + .and_then(|modified| SystemTime::now().duration_since(modified).ok()) + .is_some_and(|age| age > max_age); + if stale { + let _ = std::fs::remove_dir_all(entry.path()); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn digest(character: char) -> String { + std::iter::repeat_n(character, 64).collect() + } + + fn write_entry(root: &Path, digest: &str, pack_bytes: usize, idx_bytes: usize) -> PathBuf { + let dir = root.join(&digest[..2]).join(digest); + std::fs::create_dir_all(&dir).expect("entry dir"); + std::fs::write(dir.join(format!("pack-{digest}.pack")), vec![1; pack_bytes]).expect("pack"); + std::fs::write(dir.join(format!("pack-{digest}.idx")), vec![2; idx_bytes]).expect("idx"); + dir + } + + #[test] + fn entries_are_bounded_by_total_bytes() { + let scratch = tempfile::TempDir::new().expect("scratch"); + let cache = GitPackCache::new(scratch.path(), 8, 2).expect("cache"); + let first = digest('a'); + let second = digest('b'); + let first_dir = write_entry(&cache.root, &first, 6, 2); + let second_dir = write_entry(&cache.root, &second, 6, 2); + cache.insert( + first.clone(), + Arc::new(CachedPack::from_dir(first_dir, &first).expect("first entry")), + ); + cache.insert( + second.clone(), + Arc::new(CachedPack::from_dir(second_dir, &second).expect("second entry")), + ); + cache.prune(); + + let state = cache.state.lock().expect("state"); + assert_eq!(state.entries.len(), 1); + assert_eq!(state.total_bytes, 8); + } + + #[test] + fn cache_sessions_are_process_isolated() { + let parent = tempfile::TempDir::new().expect("parent"); + let first = GitPackCache::new(parent.path(), 8, 2).expect("first"); + let second = GitPackCache::new(parent.path(), 8, 2).expect("second"); + + assert_ne!(first.root, second.root); + assert!(first.root.starts_with(parent.path())); + assert!(second.root.starts_with(parent.path())); + } + + #[test] + fn abandoned_sessions_are_removed_after_grace_period() { + let parent = tempfile::TempDir::new().expect("parent"); + let abandoned = parent.path().join("session-abandoned"); + std::fs::create_dir(&abandoned).expect("abandoned"); + std::fs::write(abandoned.join(".heartbeat"), b"").expect("heartbeat"); + std::thread::sleep(Duration::from_millis(2)); + + cleanup_sessions_older_than(parent.path(), Duration::ZERO); + + assert!(!abandoned.exists()); + } + + #[cfg(unix)] + #[test] + fn cache_parent_must_not_be_a_symlink() { + let parent = tempfile::TempDir::new().expect("parent"); + let target = parent.path().join("target"); + let link = parent.path().join("link"); + std::fs::create_dir(&target).expect("target"); + std::os::unix::fs::symlink(&target, &link).expect("link"); + + assert!(GitPackCache::new(&link, 8, 2).is_err()); + } + + #[test] + fn cache_digest_cannot_escape_cache_root() { + for digest in ["../pack", "/absolute", "g", &"a".repeat(63)] { + assert!(validate_digest(digest).is_err(), "{digest:?}"); + } + assert!(validate_digest(&"a".repeat(64)).is_ok()); + } + + #[tokio::test] + async fn concurrent_digest_requests_share_one_flight() { + let scratch = tempfile::TempDir::new().expect("scratch"); + let cache = GitPackCache::new(scratch.path(), 1024, 2).expect("cache"); + let digest = digest('c'); + let (leader, joined) = cache.flight(&digest); + assert!(!joined); + let leader_participant = FlightParticipant { + cache: &cache, + digest: &digest, + flight: Arc::clone(&leader), + }; + let leader_guard = leader.lock.lock().await; + + let (waiter, joined) = cache.flight(&digest); + assert!(joined); + assert!(Arc::ptr_eq(&leader, &waiter)); + let waiter_participant = FlightParticipant { + cache: &cache, + digest: &digest, + flight: Arc::clone(&waiter), + }; + assert!( + tokio::time::timeout(std::time::Duration::from_millis(10), waiter.lock.lock()) + .await + .is_err() + ); + + drop(leader_guard); + assert!( + tokio::time::timeout(std::time::Duration::from_secs(1), waiter.lock.lock()) + .await + .is_ok() + ); + drop(waiter_participant); + assert!(cache.flights.contains_key(&digest)); + drop(leader_participant); + assert!(!cache.flights.contains_key(&digest)); + } + + #[tokio::test] + async fn cancelled_waiter_keeps_active_flight_registered() { + let scratch = tempfile::TempDir::new().expect("scratch"); + let cache = GitPackCache::new(scratch.path(), 1024, 2).expect("cache"); + let digest = digest('e'); + let (leader, _) = cache.flight(&digest); + let leader_participant = FlightParticipant { + cache: &cache, + digest: &digest, + flight: Arc::clone(&leader), + }; + let leader_guard = leader.lock.lock().await; + let (waiter, joined) = cache.flight(&digest); + assert!(joined); + let waiter_participant = FlightParticipant { + cache: &cache, + digest: &digest, + flight: waiter, + }; + + drop(waiter_participant); + assert!(cache.flights.contains_key(&digest)); + + drop(leader_guard); + drop(leader_participant); + assert!(!cache.flights.contains_key(&digest)); + } + + #[tokio::test] + async fn cached_pair_is_linked_into_request_workspace() { + let scratch = tempfile::TempDir::new().expect("scratch"); + let cache = GitPackCache::new(scratch.path(), 1024, 2).expect("cache"); + let digest = digest('d'); + let source = write_entry(&cache.root, &digest, 3, 2); + cache.insert( + digest.clone(), + Arc::new(CachedPack::from_dir(source.clone(), &digest).expect("entry")), + ); + let entry = cache.lookup(&digest).expect("entry"); + let destination = tempfile::TempDir::new().expect("destination"); + + install_entry(&entry, &digest, destination.path()) + .await + .expect("install"); + + assert_eq!( + std::fs::read(destination.path().join(format!("pack-{digest}.pack"))) + .expect("installed pack"), + vec![1; 3] + ); + assert!(source.is_dir()); + } +} diff --git a/crates/buzz-relay/src/api/git/store.rs b/crates/buzz-relay/src/api/git/store.rs index 467bd0adf0..43d210e648 100644 --- a/crates/buzz-relay/src/api/git/store.rs +++ b/crates/buzz-relay/src/api/git/store.rs @@ -320,9 +320,13 @@ impl GitStore { /// A missing idx is a cache miss, not a hydrate failure; callers should /// regenerate with `git index-pack`. Other backend failures are surfaced so /// callers can decide whether to fall back or fail. - pub async fn get_idx(&self, pack_digest: &str) -> Result, StoreError> { + pub async fn get_idx( + &self, + pack_digest: &str, + max_bytes: u64, + ) -> Result, StoreError> { let key = Self::idx_key_for_pack_digest(pack_digest)?; - match self.get(&key).await { + match self.get_limited(&key, max_bytes).await { Ok(bytes) => Ok(Some(bytes)), Err(StoreError::NotFound(_)) => Ok(None), Err(e) => Err(e), @@ -384,6 +388,22 @@ impl GitStore { expected_digest: &str, max_bytes: u64, ) -> Result { + let bytes = self.get_limited(key, max_bytes).await?; + let mut hasher = Sha256::new(); + hasher.update(&bytes); + let actual = hex::encode(hasher.finalize()); + if actual != expected_digest { + return Err(StoreError::DigestMismatch { + key: key.into(), + expected: expected_digest.into(), + actual, + }); + } + Ok(bytes) + } + + /// GET an object after rejecting bodies larger than `max_bytes`. + pub async fn get_limited(&self, key: &str, max_bytes: u64) -> Result { let (head, status) = self.bucket.head_object(key).await.map_err(|e| match e { S3Error::HttpFailWithBody(404, _) => StoreError::NotFound(key.into()), other => StoreError::Backend(other), @@ -408,7 +428,7 @@ impl GitStore { } } - let bytes = self.get_verified(key, expected_digest).await?; + let bytes = self.get(key).await?; let size = u64::try_from(bytes.len()).unwrap_or(u64::MAX); if size > max_bytes { return Err(StoreError::ObjectTooLarge { diff --git a/crates/buzz-relay/src/api/git/transport.rs b/crates/buzz-relay/src/api/git/transport.rs index a3622c0f01..fcd86f7bd3 100644 --- a/crates/buzz-relay/src/api/git/transport.rs +++ b/crates/buzz-relay/src/api/git/transport.rs @@ -32,6 +32,7 @@ use super::cas_publish::{cas_publish, CasError, ParentState, PublishLimits}; use super::hook::install_hook; use super::hydrate::{ hydrate_for_read, hydrate_for_write, load_manifest_for_read, HydrateError, HydratedRepo, + HydrationOptions, }; use super::manifest_event::{build_ref_state_event, RefStateInputs}; use crate::state::AppState; @@ -598,9 +599,12 @@ async fn info_refs_subprocess( tenant, ¶ms.owner, ¶ms.repo, - &state.config.git_repo_path, - state.config.git_max_pack_bytes, - state.config.git_max_repo_bytes, + HydrationOptions { + pack_cache: &state.git_pack_cache, + scratch_dir: &state.config.git_repo_path, + max_pack_bytes: state.config.git_max_pack_bytes, + max_repo_bytes: state.config.git_max_repo_bytes, + }, ) .await { @@ -730,9 +734,12 @@ pub async fn upload_pack( &auth.tenant, ¶ms.owner, ¶ms.repo, - &state.config.git_repo_path, - state.config.git_max_pack_bytes, - state.config.git_max_repo_bytes, + HydrationOptions { + pack_cache: &state.git_pack_cache, + scratch_dir: &state.config.git_repo_path, + max_pack_bytes: state.config.git_max_pack_bytes, + max_repo_bytes: state.config.git_max_repo_bytes, + }, ) .await { @@ -816,9 +823,12 @@ pub async fn receive_pack( &auth.tenant, ¶ms.owner, ¶ms.repo, - &state.config.git_repo_path, - state.config.git_max_pack_bytes, - state.config.git_max_repo_bytes, + HydrationOptions { + pack_cache: &state.git_pack_cache, + scratch_dir: &state.config.git_repo_path, + max_pack_bytes: state.config.git_max_pack_bytes, + max_repo_bytes: state.config.git_max_repo_bytes, + }, ) .await .map_err(|e| hydrate_error_to_response(¶ms.owner, ¶ms.repo, e))?; diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index fc47c7a25a..75e70d6179 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -202,13 +202,16 @@ pub struct Config { /// 60 seconds after the last message. pub ephemeral_ttl_override: Option, - /// Root directory for the relay's local git scratch. No per-repo bare repos - /// or persistent git state live here — runtime reads/writes hydrate - /// ephemeral repos from object storage per request, and all temporary Git - /// workspaces and buffered subprocess output are created beneath this path. + /// Root directory for the relay's local git scratch. No authoritative + /// repository state lives here — runtime reads/writes hydrate ephemeral + /// repos from object storage per request. Temporary workspaces, buffered + /// subprocess output, and the disposable immutable pack cache live below + /// this path. /// Repo-name uniqueness lives in Postgres (`git_repo_names`), not on disk, /// so this directory need not be persistent or shared across replicas. pub git_repo_path: std::path::PathBuf, + /// Parent directory for process-isolated immutable pack cache sessions. + pub git_pack_cache_path: std::path::PathBuf, /// Maximum pack file size for git push (bytes). Default: 500 MB. pub git_max_pack_bytes: u64, /// Maximum total bytes materialized for one git repo request. Default: 1 GB. @@ -216,6 +219,11 @@ pub struct Config { /// This bounds clone/fetch hydration work across a repo's historical pack /// set rather than only bounding one incoming push body. pub git_max_repo_bytes: u64, + /// Maximum bytes retained in the process-local immutable pack/index cache. + /// Zero disables retention while preserving request-local hydration. + pub git_pack_cache_max_bytes: u64, + /// Maximum pack digests populated concurrently in one relay process. + pub git_pack_cache_max_concurrent_populations: usize, /// Maximum number of repos per pubkey. Default: 100. pub git_max_repos_per_pubkey: u32, /// Maximum concurrent git subprocess operations. Default: 20. @@ -368,11 +376,18 @@ fn parse_optional_bool(name: &str) -> Result { fn ensure_git_repo_path( raw: impl Into, +) -> Result { + ensure_git_path("BUZZ_GIT_REPO_PATH", raw) +} + +fn ensure_git_path( + setting: &str, + raw: impl Into, ) -> Result { let git_repo_path = raw.into(); if let Err(e) = std::fs::create_dir_all(&git_repo_path) { return Err(ConfigError::InvalidValue(format!( - "BUZZ_GIT_REPO_PATH={} could not be created: {e}", + "{setting}={} could not be created: {e}", git_repo_path.display() ))); } @@ -677,6 +692,12 @@ impl Config { let git_repo_path = ensure_git_repo_path( std::env::var("BUZZ_GIT_REPO_PATH").unwrap_or_else(|_| "./repos".to_string()), )?; + let git_pack_cache_path = ensure_git_path( + "BUZZ_GIT_PACK_CACHE_PATH", + std::env::var("BUZZ_GIT_PACK_CACHE_PATH") + .map(std::path::PathBuf::from) + .unwrap_or_else(|_| git_repo_path.join(".pack-cache")), + )?; let git_max_pack_bytes: u64 = std::env::var("BUZZ_GIT_MAX_PACK_BYTES") .ok() .and_then(|v| v.parse().ok()) @@ -685,6 +706,16 @@ impl Config { .ok() .and_then(|v| v.parse().ok()) .unwrap_or_else(|| git_max_pack_bytes.saturating_mul(2)); // 1 GB at defaults + let git_pack_cache_max_bytes: u64 = std::env::var("BUZZ_GIT_PACK_CACHE_MAX_BYTES") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or_else(|| git_max_repo_bytes.saturating_mul(5)); // 5 GB at defaults + let git_pack_cache_max_concurrent_populations: usize = + std::env::var("BUZZ_GIT_PACK_CACHE_MAX_CONCURRENT_POPULATIONS") + .ok() + .and_then(|v| v.parse().ok()) + .filter(|value| *value > 0) + .unwrap_or(2); let git_max_repos_per_pubkey: u32 = std::env::var("BUZZ_GIT_MAX_REPOS_PER_PUBKEY") .ok() .and_then(|v| v.parse().ok()) @@ -862,8 +893,11 @@ impl Config { audit_enabled, ephemeral_ttl_override, git_repo_path, + git_pack_cache_path, git_max_pack_bytes, git_max_repo_bytes, + git_pack_cache_max_bytes, + git_pack_cache_max_concurrent_populations, git_max_repos_per_pubkey, git_max_concurrent_ops, git_hook_hmac_secret, diff --git a/crates/buzz-relay/src/metrics.rs b/crates/buzz-relay/src/metrics.rs index 718974261d..5ff33e9e45 100644 --- a/crates/buzz-relay/src/metrics.rs +++ b/crates/buzz-relay/src/metrics.rs @@ -87,6 +87,16 @@ pub fn install(port: u16, gauge_idle_timeout_secs: u64) { &GIT_DURATION_BUCKETS_S, ) .expect("valid git stream duration bucket boundaries") + .set_buckets_for_metric( + Matcher::Full("buzz_git_pack_cache_populate_seconds".to_owned()), + &GIT_DURATION_BUCKETS_S, + ) + .expect("valid git cache population duration bucket boundaries") + .set_buckets_for_metric( + Matcher::Full("buzz_git_pack_cache_population_wait_seconds".to_owned()), + &GIT_DURATION_BUCKETS_S, + ) + .expect("valid git cache population wait bucket boundaries") .set_buckets_for_metric( Matcher::Full("buzz_git_hydrate_bytes".to_owned()), &GIT_BYTES_BUCKETS, diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 148e6648b0..b363be3d7e 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -505,6 +505,10 @@ pub struct AppState { /// CAS-guarded manifest pointer). This is the durable git source of truth; /// see `api::git::store` and `docs/git-on-object-storage.md`. pub git_store: crate::api::git::store::GitStore, + /// Process-local, byte-bounded cache of immutable Git pack/index pairs. + /// Object storage remains authoritative; this only avoids repeated reads + /// and index generation for content-addressed packs. + pub git_pack_cache: Arc, /// Audio relay room manager — tracks active huddle audio rooms. pub audio_rooms: Arc, /// Set to `true` on SIGTERM — readiness probe returns 503. @@ -631,6 +635,14 @@ impl AppState { &config.media.s3_region, ) .expect("media storage was already constructed with this S3 config"); + let git_pack_cache = Arc::new( + crate::api::git::pack_cache::GitPackCache::new( + &config.git_pack_cache_path, + config.git_pack_cache_max_bytes, + config.git_pack_cache_max_concurrent_populations, + ) + .expect("git pack cache path must be available"), + ); let nip98_replay: Arc = Arc::new(RedisNip98ReplayGuard::new(redis_pool.clone())); let admission_rate_limiter = Arc::new(RedisRateLimiter::new(redis_pool.clone())); @@ -685,6 +697,7 @@ impl AppState { audit_tx: audit_enabled.then_some(audit_tx), media_storage: Arc::new(media_storage), git_store, + git_pack_cache, audio_rooms: Arc::new(AudioRoomManager::new()), shutting_down: Arc::new(AtomicBool::new(false)), started_at: Instant::now(), diff --git a/deploy/charts/buzz/templates/deployment.yaml b/deploy/charts/buzz/templates/deployment.yaml index 9bb746caa2..f8d67de31d 100644 --- a/deploy/charts/buzz/templates/deployment.yaml +++ b/deploy/charts/buzz/templates/deployment.yaml @@ -145,6 +145,9 @@ spec: # ── Git ────────────────────────────────────────────────── - { name: BUZZ_GIT_REPO_PATH, value: {{ .Values.persistence.git.mountPath | quote }} } - { name: BUZZ_GIT_MAX_PACK_BYTES, value: {{ .Values.git.maxPackBytes | quote }} } + - { name: BUZZ_GIT_PACK_CACHE_PATH, value: {{ .Values.git.packCachePath | quote }} } + - { name: BUZZ_GIT_PACK_CACHE_MAX_BYTES, value: {{ .Values.git.packCacheMaxBytes | quote }} } + - { 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 }} } @@ -221,6 +224,7 @@ spec: volumeMounts: - { name: git-repos, mountPath: {{ .Values.persistence.git.mountPath | quote }} } + - { name: git-pack-cache, mountPath: {{ .Values.git.packCachePath | quote }} } volumes: - name: git-repos @@ -231,3 +235,6 @@ spec: emptyDir: sizeLimit: {{ .Values.persistence.git.size | quote }} {{- end }} + - name: git-pack-cache + emptyDir: + sizeLimit: {{ .Values.git.packCacheVolumeSize | quote }} diff --git a/deploy/charts/buzz/values.schema.json b/deploy/charts/buzz/values.schema.json index 194d679293..203fd9b69b 100644 --- a/deploy/charts/buzz/values.schema.json +++ b/deploy/charts/buzz/values.schema.json @@ -199,6 +199,10 @@ "additionalProperties": false, "properties": { "maxPackBytes": { "type": "integer", "minimum": 1 }, + "packCachePath": { "type": "string", "minLength": 1 }, + "packCacheMaxBytes": { "type": "integer", "minimum": 0 }, + "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 } } diff --git a/deploy/charts/buzz/values.yaml b/deploy/charts/buzz/values.yaml index 2829fd5952..27c6503793 100644 --- a/deploy/charts/buzz/values.yaml +++ b/deploy/charts/buzz/values.yaml @@ -323,6 +323,10 @@ minio: # ── Git server config ──────────────────────────────────────────────────────── git: maxPackBytes: 524288000 # 500 MiB + packCachePath: /var/cache/buzz/git-packs + packCacheMaxBytes: 5368709120 # 5 GiB + packCacheMaxConcurrentPopulations: 2 + packCacheVolumeSize: 7Gi # per-pod emptyDir; includes cold-population staging maxReposPerPubkey: 100 maxConcurrentOps: 20 diff --git a/docs/git-on-object-storage.md b/docs/git-on-object-storage.md index 59599b34c8..105e4ba723 100644 --- a/docs/git-on-object-storage.md +++ b/docs/git-on-object-storage.md @@ -51,13 +51,19 @@ to three stated axioms, each empirically gated per backend" does. ### v1 deployment architecture -The implementation has *no per-repo persistent filesystem state*. Every request -hydrates an ephemeral working tree from the published manifest, runs the -appropriate git subprocess against it, and drops the tree on scope exit: +The implementation has *no authoritative per-repo filesystem state*. Every +request hydrates an ephemeral working tree from the published manifest, runs +the appropriate git subprocess against it, and drops the tree on scope exit: read paths (`info/refs`, `upload-pack`) via `hydrate_for_read`, the write path (`receive-pack`) via `hydrate_for_write`, which also returns the `ParentState` the CAS at §Push step 7 predicates on. The relay is multi-instance-ready by construction: nothing on local disk needs to be coordinated between instances. +Each process may retain a byte-bounded, process-lifetime cache of immutable +pack/index pairs keyed by verified digest. Deployment mounts that cache on a +per-pod ephemeral volume, so accounting and single-flight coordination stay +local. Cache misses, restarts, and evictions only affect performance; object +storage remains the source of truth, and per-request refs/HEAD are still +materialized from the current manifest. The accepted v1 tradeoff: under concurrent same-repo pushes, every contender hydrates and runs receive-pack, and the CAS losers' subprocess work is @@ -475,6 +481,7 @@ symbol search, not line counts.) | `GitStore::{put_pack, put_manifest, put_pointer}` (create-only + CAS) | `store.rs` | | `run_conformance_probe` (A1/A3 fail-closed startup gate) | `store.rs` + `main.rs` | | `hydrate_for_read` / `hydrate_for_write` | `hydrate.rs` | +| Bounded digest-keyed pack/index cache and single-flight population | `pack_cache.rs` | | `ParentState { if_match, parent_digest, parent }` + `from_loaded`/`fresh` | `cas_publish.rs:154` | | `cas_publish(.., &parent_state) -> Result` | `cas_publish.rs:410` | | `CasError::Conflict { winner_manifest, winner_manifest_key }` (typed 412) | `cas_publish.rs:92` |