perf(relay): cache Git pack hydration (#2169)

This commit is contained in:
thomaspblock
2026-07-20 16:50:11 +02:00
committed by GitHub
parent 984645ecaa
commit a4d82ec722
13 changed files with 944 additions and 104 deletions
+6 -3
View File
@@ -66,12 +66,15 @@ RELAY_URL=ws://localhost:3000
# ----------------------------------------------------------------------------- # -----------------------------------------------------------------------------
# Git (NIP-34 bare repositories) # Git (NIP-34 bare repositories)
# ----------------------------------------------------------------------------- # -----------------------------------------------------------------------------
# Root directory for bare git repos. Repos are stored at # Root directory for ephemeral Git workspaces and the disposable pack cache.
# {path}/{owner_hex}/{repo_id}.git/. Default: ./repos (relative to CWD). # Default: ./repos (relative to CWD).
# Set an absolute path to keep repos stable across worktrees.
# BUZZ_GIT_REPO_PATH=./repos # BUZZ_GIT_REPO_PATH=./repos
# BUZZ_GIT_MAX_PACK_BYTES=524288000 # BUZZ_GIT_MAX_PACK_BYTES=524288000
# BUZZ_GIT_MAX_REPO_BYTES=1048576000 # 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 # Media Upload Admission
+122 -81
View File
@@ -6,11 +6,11 @@
//! //!
//! 1. GET pointer → manifest digest. //! 1. GET pointer → manifest digest.
//! 2. GET manifest (digest-verified) → parsed [`Manifest`]. //! 2. GET manifest (digest-verified) → parsed [`Manifest`].
//! 3. GET every pack the manifest names (digest-verified, in parallel). //! 3. Resolve every pack through the bounded local content-addressed cache.
//! 4. **Phase 1** — for each pack: write `pack-<hex>.pack`; install and //! 4. **Phase 1** — hard-link or copy each verified pack/index pair into the
//! validate a cached `idx/<hex>` sidecar when present, otherwise run //! ephemeral repo. A cache miss reads object storage and validates or
//! `git index-pack` to materialize `.idx`. Failure here tears down the //! generates the index once per digest. Failure here tears down the tempdir
//! tempdir with no refs/HEAD ever written. //! with no refs/HEAD ever written.
//! 5. **Phase 2** — only after all packs are indexed: write loose refs and //! 5. **Phase 2** — only after all packs are indexed: write loose refs and
//! HEAD from the manifest. //! HEAD from the manifest.
//! //!
@@ -19,8 +19,8 @@
//! mid-stream. Sami/Max named this explicitly. //! mid-stream. Sami/Max named this explicitly.
//! //!
//! The returned [`HydratedRepo`] owns a [`tempfile::TempDir`]; dropping it //! The returned [`HydratedRepo`] owns a [`tempfile::TempDir`]; dropping it
//! cleans up. Every read currently re-hydrates from scratch — naive but //! cleans up. Cached pack/index pairs are immutable performance state; object
//! correct; caching is named follow-up work. //! storage remains authoritative.
// Public surface is consumed by `transport.rs` after Eva's `AppState::git_store` // 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 // 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::cas_publish::ParentState;
use super::manifest::{is_hex_oid, is_safe_refname, pointer_key, Manifest, ManifestError}; 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 super::store::{ETag, GitStore, StoreError};
use buzz_core::TenantContext; 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. /// Hydration errors.
/// ///
/// "Repo doesn't exist" is signalled by `Ok(None)` from `hydrate_for_read`, /// "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, ctx: &TenantContext,
owner: &str, owner: &str,
repo: &str, repo: &str,
scratch_dir: &Path, options: HydrationOptions<'_>,
max_pack_bytes: u64,
max_repo_bytes: u64,
) -> Result<Option<HydratedRepo>, HydrateError> { ) -> Result<Option<HydratedRepo>, HydrateError> {
let started_at = std::time::Instant::now(); let started_at = std::time::Instant::now();
let result = hydrate_for_read_inner( let result = hydrate_for_read_inner(store, ctx, owner, repo, options).await;
store,
ctx,
owner,
repo,
scratch_dir,
max_pack_bytes,
max_repo_bytes,
)
.await;
let outcome = match &result { let outcome = match &result {
Ok(Some(repo)) => { Ok(Some(repo)) => {
metrics::histogram!("buzz_git_hydrate_bytes").record(repo.hydrated_bytes() as f64); metrics::histogram!("buzz_git_hydrate_bytes").record(repo.hydrated_bytes() as f64);
@@ -151,23 +154,12 @@ async fn hydrate_for_read_inner(
ctx: &TenantContext, ctx: &TenantContext,
owner: &str, owner: &str,
repo: &str, repo: &str,
scratch_dir: &Path, options: HydrationOptions<'_>,
max_pack_bytes: u64,
max_repo_bytes: u64,
) -> Result<Option<HydratedRepo>, HydrateError> { ) -> Result<Option<HydratedRepo>, HydrateError> {
let Some((_etag, _digest, manifest)) = load_pointer(store, ctx, owner, repo).await? else { let Some((_etag, _digest, manifest)) = load_pointer(store, ctx, owner, repo).await? else {
return Ok(None); return Ok(None);
}; };
Ok(Some( Ok(Some(materialize_manifest(store, &manifest, options).await?))
materialize_manifest(
store,
&manifest,
scratch_dir,
max_pack_bytes,
max_repo_bytes,
)
.await?,
))
} }
/// Load the current manifest for a read path without materializing pack objects. /// Load the current manifest for a read path without materializing pack objects.
@@ -212,20 +204,11 @@ pub async fn hydrate_for_write(
ctx: &TenantContext, ctx: &TenantContext,
owner: &str, owner: &str,
repo: &str, repo: &str,
scratch_dir: &Path, options: HydrationOptions<'_>,
max_pack_bytes: u64,
max_repo_bytes: u64,
) -> Result<(HydratedRepo, ParentState), HydrateError> { ) -> Result<(HydratedRepo, ParentState), HydrateError> {
match load_pointer(store, ctx, owner, repo).await? { match load_pointer(store, ctx, owner, repo).await? {
Some((etag, digest, manifest)) => { Some((etag, digest, manifest)) => {
let repo = materialize_manifest( let repo = materialize_manifest(store, &manifest, options).await?;
store,
&manifest,
scratch_dir,
max_pack_bytes,
max_repo_bytes,
)
.await?;
let parent = ParentState::from_loaded(etag, digest, manifest); let parent = ParentState::from_loaded(etag, digest, manifest);
Ok((repo, parent)) 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 // First push: empty bare repo. No packs to fetch, no refs to
// install. `receive-pack` will accept whatever the client // install. `receive-pack` will accept whatever the client
// sends; `cas_publish` will use `If-None-Match: *`. // sends; `cas_publish` will use `If-None-Match: *`.
let tempdir = TempDir::new_in(scratch_dir) let tempdir = TempDir::new_in(options.scratch_dir).map_err(|e| {
.map_err(|e| HydrateError::Hydrate(format!("tempdir in {scratch_dir:?}: {e}")))?; HydrateError::Hydrate(format!("tempdir in {:?}: {e}", options.scratch_dir))
})?;
let path = tempdir.path().to_path_buf(); let path = tempdir.path().to_path_buf();
run_git(&path, &["init", "--bare", "--quiet"]).await?; run_git(&path, &["init", "--bare", "--quiet"]).await?;
Ok(( Ok((
@@ -280,7 +264,7 @@ async fn load_pointer(
Ok(Some((etag, digest, manifest))) Ok(Some((etag, digest, manifest)))
} }
async fn get_verified_limited( pub(super) async fn get_verified_limited(
store: &GitStore, store: &GitStore,
key: &str, key: &str,
digest: &str, digest: &str,
@@ -304,13 +288,11 @@ async fn get_verified_limited(
async fn materialize_manifest( async fn materialize_manifest(
store: &GitStore, store: &GitStore,
manifest: &Manifest, manifest: &Manifest,
scratch_dir: &Path, options: HydrationOptions<'_>,
max_pack_bytes: u64,
max_repo_bytes: u64,
) -> Result<HydratedRepo, HydrateError> { ) -> Result<HydratedRepo, HydrateError> {
// Init bare repo. // Init bare repo.
let tempdir = TempDir::new_in(scratch_dir) let tempdir = TempDir::new_in(options.scratch_dir)
.map_err(|e| HydrateError::Hydrate(format!("tempdir in {scratch_dir:?}: {e}")))?; .map_err(|e| HydrateError::Hydrate(format!("tempdir in {:?}: {e}", options.scratch_dir)))?;
let path = tempdir.path().to_path_buf(); let path = tempdir.path().to_path_buf();
run_git(&path, &["init", "--bare", "--quiet"]).await?; run_git(&path, &["init", "--bare", "--quiet"]).await?;
@@ -324,27 +306,25 @@ async fn materialize_manifest(
let digest = key let digest = key
.strip_prefix("packs/") .strip_prefix("packs/")
.ok_or_else(|| HydrateError::Hydrate(format!("malformed pack key {key:?}")))?; .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 = options
let pack_bytes = u64::try_from(bytes.len()).unwrap_or(u64::MAX); .pack_cache
if pack_bytes > max_pack_bytes { .materialize_pack(store, key, digest, &pack_dir, options.max_pack_bytes)
.await?;
if pack_bytes > options.max_pack_bytes {
return Err(HydrateError::ResourceLimit(format!( 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(|| { hydrated_bytes = hydrated_bytes.checked_add(pack_bytes).ok_or_else(|| {
HydrateError::ResourceLimit("repo byte count overflowed u64".to_string()) 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!( 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. // 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, store: &GitStore,
repo_path: &Path, repo_path: &Path,
pack_digest: &str, pack_digest: &str,
pack_path: &Path, pack_path: &Path,
max_idx_bytes: u64,
) -> Result<(), HydrateError> { ) -> Result<(), HydrateError> {
let idx_path = pack_path.with_extension("idx"); 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)) => { Ok(Some(idx_bytes)) => {
tokio::fs::write(&idx_path, &idx_bytes) tokio::fs::write(&idx_path, &idx_bytes)
.await .await
@@ -438,7 +419,27 @@ async fn install_or_generate_idx(
let pack_path_str = pack_path let pack_path_str = pack_path
.to_str() .to_str()
.ok_or_else(|| HydrateError::Hydrate("pack path is not valid utf-8".to_string()))?; .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 <args>` in `cwd`, fail on non-zero exit. /// Run `git <args>` in `cwd`, fail on non-zero exit.
@@ -499,6 +500,17 @@ mod tests {
assert!(!is_hex_oid("")); 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. // `pointer_key` is tested in `super::manifest::tests` — single source.
fn tenant() -> TenantContext { fn tenant() -> TenantContext {
@@ -520,10 +532,20 @@ mod tests {
packs: Vec::new(), packs: Vec::new(),
parent: None, 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) let hydrated = materialize_manifest(
.await &store,
.expect("materialize empty repo"); &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())); assert!(hydrated.path().starts_with(scratch.path()));
} }
@@ -643,11 +665,22 @@ mod tests {
// Hydrate. // Hydrate.
let scratch = TempDir::new().unwrap(); let scratch = TempDir::new().unwrap();
let hydrated = let cache = GitPackCache::new(scratch.path(), u64::MAX, 2).expect("cache");
hydrate_for_read(&st, &ctx, &owner, repo, scratch.path(), u64::MAX, u64::MAX) let hydrated = hydrate_for_read(
.await &st,
.expect("hydrate") &ctx,
.expect("hydrate Some"); &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()); eprintln!("hydrated to {}", hydrated.path().display());
// The hydrated repo must list the same ref with the same oid. // 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 owner = format!("nope-{}", uuid::Uuid::new_v4());
let ctx = tenant(); let ctx = tenant();
let scratch = TempDir::new().unwrap(); let scratch = TempDir::new().unwrap();
let cache = GitPackCache::new(scratch.path(), u64::MAX, 2).expect("cache");
let result = hydrate_for_read( let result = hydrate_for_read(
&st, &st,
&ctx, &ctx,
&owner, &owner,
"ghost", "ghost",
scratch.path(), HydrationOptions {
u64::MAX, pack_cache: &cache,
u64::MAX, scratch_dir: scratch.path(),
max_pack_bytes: u64::MAX,
max_repo_bytes: u64::MAX,
},
) )
.await .await
.expect("ok"); .expect("ok");
@@ -774,14 +811,18 @@ mod tests {
} }
let scratch = TempDir::new().unwrap(); let scratch = TempDir::new().unwrap();
let cache = GitPackCache::new(scratch.path(), u64::MAX, 2).expect("cache");
let hydrated = hydrate_for_read( let hydrated = hydrate_for_read(
&st, &st,
&ctx, &ctx,
&owner, &owner,
"void", "void",
scratch.path(), HydrationOptions {
u64::MAX, pack_cache: &cache,
u64::MAX, scratch_dir: scratch.path(),
max_pack_bytes: u64::MAX,
max_repo_bytes: u64::MAX,
},
) )
.await .await
.expect("hydrate") .expect("hydrate")
+1
View File
@@ -27,6 +27,7 @@ pub mod hook;
pub mod hydrate; pub mod hydrate;
pub mod manifest; pub mod manifest;
pub mod manifest_event; pub mod manifest_event;
pub mod pack_cache;
pub mod policy; pub mod policy;
pub mod store; pub mod store;
pub mod transport; pub mod transport;
+686
View File
@@ -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<tempfile::TempDir>,
}
impl CachedPack {
fn from_dir(dir: PathBuf, digest: &str) -> Option<Self> {
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<CachedPack>,
last_used: u64,
}
#[derive(Default)]
struct CacheState {
entries: HashMap<String, CacheRecord>,
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<tokio::task::JoinHandle<()>>,
root: PathBuf,
max_bytes: u64,
population_semaphore: tokio::sync::Semaphore,
state: Mutex<CacheState>,
flights: DashMap<String, Arc<PopulationFlight>>,
}
struct PopulationFlight {
lock: tokio::sync::Mutex<()>,
participants: AtomicUsize,
}
struct FlightParticipant<'a> {
cache: &'a GitPackCache,
digest: &'a str,
flight: Arc<PopulationFlight>,
}
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<Self, String> {
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<u64, HydrateError> {
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<Arc<CachedPack>> {
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<Arc<CachedPack>, 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<CachedPack>) {
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<PopulationFlight>) {
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<PopulationFlight>, 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<u64> {
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());
}
}
+23 -3
View File
@@ -320,9 +320,13 @@ impl GitStore {
/// A missing idx is a cache miss, not a hydrate failure; callers should /// A missing idx is a cache miss, not a hydrate failure; callers should
/// regenerate with `git index-pack`. Other backend failures are surfaced so /// regenerate with `git index-pack`. Other backend failures are surfaced so
/// callers can decide whether to fall back or fail. /// callers can decide whether to fall back or fail.
pub async fn get_idx(&self, pack_digest: &str) -> Result<Option<Bytes>, StoreError> { pub async fn get_idx(
&self,
pack_digest: &str,
max_bytes: u64,
) -> Result<Option<Bytes>, StoreError> {
let key = Self::idx_key_for_pack_digest(pack_digest)?; 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)), Ok(bytes) => Ok(Some(bytes)),
Err(StoreError::NotFound(_)) => Ok(None), Err(StoreError::NotFound(_)) => Ok(None),
Err(e) => Err(e), Err(e) => Err(e),
@@ -384,6 +388,22 @@ impl GitStore {
expected_digest: &str, expected_digest: &str,
max_bytes: u64, max_bytes: u64,
) -> Result<Bytes, StoreError> { ) -> Result<Bytes, StoreError> {
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<Bytes, StoreError> {
let (head, status) = self.bucket.head_object(key).await.map_err(|e| match e { let (head, status) = self.bucket.head_object(key).await.map_err(|e| match e {
S3Error::HttpFailWithBody(404, _) => StoreError::NotFound(key.into()), S3Error::HttpFailWithBody(404, _) => StoreError::NotFound(key.into()),
other => StoreError::Backend(other), 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); let size = u64::try_from(bytes.len()).unwrap_or(u64::MAX);
if size > max_bytes { if size > max_bytes {
return Err(StoreError::ObjectTooLarge { return Err(StoreError::ObjectTooLarge {
+19 -9
View File
@@ -32,6 +32,7 @@ use super::cas_publish::{cas_publish, CasError, ParentState, PublishLimits};
use super::hook::install_hook; use super::hook::install_hook;
use super::hydrate::{ use super::hydrate::{
hydrate_for_read, hydrate_for_write, load_manifest_for_read, HydrateError, HydratedRepo, hydrate_for_read, hydrate_for_write, load_manifest_for_read, HydrateError, HydratedRepo,
HydrationOptions,
}; };
use super::manifest_event::{build_ref_state_event, RefStateInputs}; use super::manifest_event::{build_ref_state_event, RefStateInputs};
use crate::state::AppState; use crate::state::AppState;
@@ -598,9 +599,12 @@ async fn info_refs_subprocess(
tenant, tenant,
&params.owner, &params.owner,
&params.repo, &params.repo,
&state.config.git_repo_path, HydrationOptions {
state.config.git_max_pack_bytes, pack_cache: &state.git_pack_cache,
state.config.git_max_repo_bytes, 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 .await
{ {
@@ -730,9 +734,12 @@ pub async fn upload_pack(
&auth.tenant, &auth.tenant,
&params.owner, &params.owner,
&params.repo, &params.repo,
&state.config.git_repo_path, HydrationOptions {
state.config.git_max_pack_bytes, pack_cache: &state.git_pack_cache,
state.config.git_max_repo_bytes, 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 .await
{ {
@@ -816,9 +823,12 @@ pub async fn receive_pack(
&auth.tenant, &auth.tenant,
&params.owner, &params.owner,
&params.repo, &params.repo,
&state.config.git_repo_path, HydrationOptions {
state.config.git_max_pack_bytes, pack_cache: &state.git_pack_cache,
state.config.git_max_repo_bytes, 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 .await
.map_err(|e| hydrate_error_to_response(&params.owner, &params.repo, e))?; .map_err(|e| hydrate_error_to_response(&params.owner, &params.repo, e))?;
+39 -5
View File
@@ -202,13 +202,16 @@ pub struct Config {
/// 60 seconds after the last message. /// 60 seconds after the last message.
pub ephemeral_ttl_override: Option<i32>, pub ephemeral_ttl_override: Option<i32>,
/// Root directory for the relay's local git scratch. No per-repo bare repos /// Root directory for the relay's local git scratch. No authoritative
/// or persistent git state live here — runtime reads/writes hydrate /// repository state lives here — runtime reads/writes hydrate ephemeral
/// ephemeral repos from object storage per request, and all temporary Git /// repos from object storage per request. Temporary workspaces, buffered
/// workspaces and buffered subprocess output are created beneath this path. /// subprocess output, and the disposable immutable pack cache live below
/// this path.
/// Repo-name uniqueness lives in Postgres (`git_repo_names`), not on disk, /// Repo-name uniqueness lives in Postgres (`git_repo_names`), not on disk,
/// so this directory need not be persistent or shared across replicas. /// so this directory need not be persistent or shared across replicas.
pub git_repo_path: std::path::PathBuf, 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. /// Maximum pack file size for git push (bytes). Default: 500 MB.
pub git_max_pack_bytes: u64, pub git_max_pack_bytes: u64,
/// Maximum total bytes materialized for one git repo request. Default: 1 GB. /// 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 /// This bounds clone/fetch hydration work across a repo's historical pack
/// set rather than only bounding one incoming push body. /// set rather than only bounding one incoming push body.
pub git_max_repo_bytes: u64, 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. /// Maximum number of repos per pubkey. Default: 100.
pub git_max_repos_per_pubkey: u32, pub git_max_repos_per_pubkey: u32,
/// Maximum concurrent git subprocess operations. Default: 20. /// Maximum concurrent git subprocess operations. Default: 20.
@@ -368,11 +376,18 @@ fn parse_optional_bool(name: &str) -> Result<bool, ConfigError> {
fn ensure_git_repo_path( fn ensure_git_repo_path(
raw: impl Into<std::path::PathBuf>, raw: impl Into<std::path::PathBuf>,
) -> Result<std::path::PathBuf, ConfigError> {
ensure_git_path("BUZZ_GIT_REPO_PATH", raw)
}
fn ensure_git_path(
setting: &str,
raw: impl Into<std::path::PathBuf>,
) -> Result<std::path::PathBuf, ConfigError> { ) -> Result<std::path::PathBuf, ConfigError> {
let git_repo_path = raw.into(); let git_repo_path = raw.into();
if let Err(e) = std::fs::create_dir_all(&git_repo_path) { if let Err(e) = std::fs::create_dir_all(&git_repo_path) {
return Err(ConfigError::InvalidValue(format!( return Err(ConfigError::InvalidValue(format!(
"BUZZ_GIT_REPO_PATH={} could not be created: {e}", "{setting}={} could not be created: {e}",
git_repo_path.display() git_repo_path.display()
))); )));
} }
@@ -677,6 +692,12 @@ impl Config {
let git_repo_path = ensure_git_repo_path( let git_repo_path = ensure_git_repo_path(
std::env::var("BUZZ_GIT_REPO_PATH").unwrap_or_else(|_| "./repos".to_string()), 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") let git_max_pack_bytes: u64 = std::env::var("BUZZ_GIT_MAX_PACK_BYTES")
.ok() .ok()
.and_then(|v| v.parse().ok()) .and_then(|v| v.parse().ok())
@@ -685,6 +706,16 @@ impl Config {
.ok() .ok()
.and_then(|v| v.parse().ok()) .and_then(|v| v.parse().ok())
.unwrap_or_else(|| git_max_pack_bytes.saturating_mul(2)); // 1 GB at defaults .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") let git_max_repos_per_pubkey: u32 = std::env::var("BUZZ_GIT_MAX_REPOS_PER_PUBKEY")
.ok() .ok()
.and_then(|v| v.parse().ok()) .and_then(|v| v.parse().ok())
@@ -862,8 +893,11 @@ impl Config {
audit_enabled, audit_enabled,
ephemeral_ttl_override, ephemeral_ttl_override,
git_repo_path, git_repo_path,
git_pack_cache_path,
git_max_pack_bytes, git_max_pack_bytes,
git_max_repo_bytes, git_max_repo_bytes,
git_pack_cache_max_bytes,
git_pack_cache_max_concurrent_populations,
git_max_repos_per_pubkey, git_max_repos_per_pubkey,
git_max_concurrent_ops, git_max_concurrent_ops,
git_hook_hmac_secret, git_hook_hmac_secret,
+10
View File
@@ -87,6 +87,16 @@ pub fn install(port: u16, gauge_idle_timeout_secs: u64) {
&GIT_DURATION_BUCKETS_S, &GIT_DURATION_BUCKETS_S,
) )
.expect("valid git stream duration bucket boundaries") .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( .set_buckets_for_metric(
Matcher::Full("buzz_git_hydrate_bytes".to_owned()), Matcher::Full("buzz_git_hydrate_bytes".to_owned()),
&GIT_BYTES_BUCKETS, &GIT_BYTES_BUCKETS,
+13
View File
@@ -505,6 +505,10 @@ pub struct AppState {
/// CAS-guarded manifest pointer). This is the durable git source of truth; /// CAS-guarded manifest pointer). This is the durable git source of truth;
/// see `api::git::store` and `docs/git-on-object-storage.md`. /// see `api::git::store` and `docs/git-on-object-storage.md`.
pub git_store: crate::api::git::store::GitStore, 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<crate::api::git::pack_cache::GitPackCache>,
/// Audio relay room manager — tracks active huddle audio rooms. /// Audio relay room manager — tracks active huddle audio rooms.
pub audio_rooms: Arc<AudioRoomManager>, pub audio_rooms: Arc<AudioRoomManager>,
/// Set to `true` on SIGTERM — readiness probe returns 503. /// Set to `true` on SIGTERM — readiness probe returns 503.
@@ -631,6 +635,14 @@ impl AppState {
&config.media.s3_region, &config.media.s3_region,
) )
.expect("media storage was already constructed with this S3 config"); .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<dyn Nip98ReplayGuard> = let nip98_replay: Arc<dyn Nip98ReplayGuard> =
Arc::new(RedisNip98ReplayGuard::new(redis_pool.clone())); Arc::new(RedisNip98ReplayGuard::new(redis_pool.clone()));
let admission_rate_limiter = Arc::new(RedisRateLimiter::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), audit_tx: audit_enabled.then_some(audit_tx),
media_storage: Arc::new(media_storage), media_storage: Arc::new(media_storage),
git_store, git_store,
git_pack_cache,
audio_rooms: Arc::new(AudioRoomManager::new()), audio_rooms: Arc::new(AudioRoomManager::new()),
shutting_down: Arc::new(AtomicBool::new(false)), shutting_down: Arc::new(AtomicBool::new(false)),
started_at: Instant::now(), started_at: Instant::now(),
@@ -145,6 +145,9 @@ spec:
# ── Git ────────────────────────────────────────────────── # ── Git ──────────────────────────────────────────────────
- { name: BUZZ_GIT_REPO_PATH, value: {{ .Values.persistence.git.mountPath | quote }} } - { 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_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_REPOS_PER_PUBKEY, value: {{ .Values.git.maxReposPerPubkey | quote }} }
- { name: BUZZ_GIT_MAX_CONCURRENT_OPS, value: {{ .Values.git.maxConcurrentOps | quote }} } - { name: BUZZ_GIT_MAX_CONCURRENT_OPS, value: {{ .Values.git.maxConcurrentOps | quote }} }
@@ -221,6 +224,7 @@ spec:
volumeMounts: volumeMounts:
- { name: git-repos, mountPath: {{ .Values.persistence.git.mountPath | quote }} } - { name: git-repos, mountPath: {{ .Values.persistence.git.mountPath | quote }} }
- { name: git-pack-cache, mountPath: {{ .Values.git.packCachePath | quote }} }
volumes: volumes:
- name: git-repos - name: git-repos
@@ -231,3 +235,6 @@ spec:
emptyDir: emptyDir:
sizeLimit: {{ .Values.persistence.git.size | quote }} sizeLimit: {{ .Values.persistence.git.size | quote }}
{{- end }} {{- end }}
- name: git-pack-cache
emptyDir:
sizeLimit: {{ .Values.git.packCacheVolumeSize | quote }}
+4
View File
@@ -199,6 +199,10 @@
"additionalProperties": false, "additionalProperties": false,
"properties": { "properties": {
"maxPackBytes": { "type": "integer", "minimum": 1 }, "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 }, "maxReposPerPubkey": { "type": "integer", "minimum": 1 },
"maxConcurrentOps": { "type": "integer", "minimum": 1 } "maxConcurrentOps": { "type": "integer", "minimum": 1 }
} }
+4
View File
@@ -323,6 +323,10 @@ minio:
# ── Git server config ──────────────────────────────────────────────────────── # ── Git server config ────────────────────────────────────────────────────────
git: git:
maxPackBytes: 524288000 # 500 MiB 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 maxReposPerPubkey: 100
maxConcurrentOps: 20 maxConcurrentOps: 20
+10 -3
View File
@@ -51,13 +51,19 @@ to three stated axioms, each empirically gated per backend" does.
### v1 deployment architecture ### v1 deployment architecture
The implementation has *no per-repo persistent filesystem state*. Every request The implementation has *no authoritative per-repo filesystem state*. Every
hydrates an ephemeral working tree from the published manifest, runs the request hydrates an ephemeral working tree from the published manifest, runs
appropriate git subprocess against it, and drops the tree on scope exit: 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 read paths (`info/refs`, `upload-pack`) via `hydrate_for_read`, the write path
(`receive-pack`) via `hydrate_for_write`, which also returns the `ParentState` (`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 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. 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 The accepted v1 tradeoff: under concurrent same-repo pushes, every contender
hydrates and runs receive-pack, and the CAS losers' subprocess work is 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` | | `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` | | `run_conformance_probe` (A1/A3 fail-closed startup gate) | `store.rs` + `main.rs` |
| `hydrate_for_read` / `hydrate_for_write` | `hydrate.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` | | `ParentState { if_match, parent_digest, parent }` + `from_loaded`/`fresh` | `cas_publish.rs:154` |
| `cas_publish(.., &parent_state) -> Result<CasSuccess, CasError>` | `cas_publish.rs:410` | | `cas_publish(.., &parent_state) -> Result<CasSuccess, CasError>` | `cas_publish.rs:410` |
| `CasError::Conflict { winner_manifest, winner_manifest_key }` (typed 412) | `cas_publish.rs:92` | | `CasError::Conflict { winner_manifest, winner_manifest_key }` (typed 412) | `cas_publish.rs:92` |