From 7dde0d59edea89ac7e22654839735bc216a581e0 Mon Sep 17 00:00:00 2001 From: Roland Groza Date: Sun, 16 Aug 2026 18:40:54 +0900 Subject: [PATCH] feat: add LRU disk eviction of idle mirrors Bound the on-disk cache with an optional byte cap (--cache-max-mb, default 0 = unlimited). A clone/fetch flags the mirror; a background task measures it and evicts least-recently-used idle mirrors until under the cap, off the request path. Evicted mirrors re-clone transparently on the next request. Assisted-by: Claude:claude-opus-4-8 --- Cargo.lock | 7 + Cargo.toml | 3 + README.md | 7 +- src/config.rs | 8 + src/evict.rs | 519 +++++++++++++++++++++++++++++++++++++++++++++++++ src/git.rs | 57 +++++- src/lib.rs | 1 + src/main.rs | 36 +++- src/metrics.rs | 54 ++++- src/repo.rs | 20 +- tests/e2e.rs | 17 +- tests/http.rs | 17 +- 12 files changed, 726 insertions(+), 20 deletions(-) create mode 100644 src/evict.rs diff --git a/Cargo.lock b/Cargo.lock index b4c3f63..2822689 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -325,6 +325,7 @@ dependencies = [ "bytes", "clap", "flate2", + "lru", "prometheus", "subtle", "tempfile", @@ -467,6 +468,12 @@ version = "0.4.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" +[[package]] +name = "lru" +version = "0.18.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d2f2f9b4ba7e6b24d95e7e899329d35be83bcded72c8540cdd5368932d1d90a" + [[package]] name = "matchers" version = "0.2.0" diff --git a/Cargo.toml b/Cargo.toml index 37c9437..7048013 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,6 +19,9 @@ axum = "0.8" bytes = "1" clap = { version = "4.5", features = ["derive", "env"] } flate2 = "1" +# default-features off drops the `hashbrown` backend (and its allocator-api2 / +# foldhash / equivalent deps); the std `HashMap` backend is plenty for this. +lru = { version = "0.18", default-features = false } prometheus = { version = "0.14", default-features = false } subtle = "2" tokio = { version = "1", features = [ diff --git a/README.md b/README.md index ff4f32a..998c63a 100644 --- a/README.md +++ b/README.md @@ -147,6 +147,7 @@ Every flag has an environment-variable equivalent. | `--fetch-ttl-seconds` | `GITCACHEPROXY_FETCH_TTL_SECONDS` | `10` | Skip upstream fetch if refreshed within this window (`0` = always fetch) | | `--max-concurrent-requests` | `GITCACHEPROXY_MAX_CONCURRENT_REQUESTS` | `64` | Max concurrent in-flight requests; excess queue (`0` = unlimited) | | `--max-decoded-body-mb` | `GITCACHEPROXY_MAX_DECODED_BODY_MB` | `512` | Cap on a decoded upload-pack request body, in MiB (bounds memory / gzip bombs) | +| `--cache-max-mb` | `GITCACHEPROXY_CACHE_MAX_MB` | `0` | Cap on total on-disk mirror cache, in MiB; evicts least-recently-used idle mirrors when exceeded (`0` = unlimited, no eviction) | | `--git-binary` | `GITCACHEPROXY_GIT_BINARY` | `git` | Path to git | Endpoints: `/healthz`, `/readyz`, `/metrics` (Prometheus). @@ -206,8 +207,9 @@ explicit before you expose it: not place it on an untrusted one without a token and TLS. - **DoS knobs.** `--max-concurrent-requests` caps concurrent upstream clone/fetch work and `--max-decoded-body-mb` bounds request-body memory - (defusing a decompression bomb). The on-disk cache still grows unbounded (no - eviction yet - see the roadmap), so isolate and monitor the cache volume. + (defusing a decompression bomb). `--cache-max-mb` bounds on-disk growth by + evicting least-recently-used idle mirrors; it defaults to `0` (unlimited), so + set it - or isolate and monitor the cache volume - on an untrusted network. - **Read-only.** Only `git-upload-pack` (clone/fetch) is served; `git-receive-pack` (push) is refused and upstream is only ever pulled from, never written. @@ -233,7 +235,6 @@ rely on it. Not yet implemented, in rough priority order: -- LRU disk eviction of idle mirrors (the cache currently grows unbounded). - Per-repo latency histograms (fetch/serve durations); per-repo counters exist. - A background/scheduled refresh option (today every `info/refs` triggers an on-demand, TTL-coalesced fetch). diff --git a/src/config.rs b/src/config.rs index 1f9a3b4..f0fe357 100644 --- a/src/config.rs +++ b/src/config.rs @@ -81,6 +81,14 @@ pub struct Config { #[arg(long, env = "GITCACHEPROXY_MAX_DECODED_BODY_MB", default_value_t = 512)] pub max_decoded_body_mb: u64, + /// Maximum total size, in MiB, of the on-disk mirror cache. When a clone or + /// fetch pushes the total over this, least-recently-used idle mirrors are + /// evicted in the background until it is back under; an evicted mirror is + /// transparently re-cloned on its next request. `0` = unlimited: no eviction + /// and no accounting, so the cache grows without bound (the default). + #[arg(long, env = "GITCACHEPROXY_CACHE_MAX_MB", default_value_t = 0)] + pub cache_max_mb: u64, + /// Path to the git binary. #[arg(long, env = "GITCACHEPROXY_GIT_BINARY", default_value = "git")] pub git_binary: String, diff --git a/src/evict.rs b/src/evict.rs new file mode 100644 index 0000000..6688e8c --- /dev/null +++ b/src/evict.rs @@ -0,0 +1,519 @@ +// SPDX-License-Identifier: Apache-2.0 +//! Bounding the on-disk cache with LRU eviction of idle mirrors. +//! +//! When a byte cap is configured (`--cache-max-mb`), a [`CacheIndex`] tracks every +//! mirror's size and access order in memory. The request path only ever does O(1) +//! bookkeeping against it - never disk IO: +//! - `touch` on every request (a served cache hit counts as use), +//! - `mark_changed` after a clone/fetch, flagging the mirror for (re)measurement. +//! +//! All disk work - measuring a changed mirror's size, and evicting mirrors - runs +//! in the background [`run`] task, off the critical path, so a client's clone/fetch +//! is never blocked by cache maintenance. Access order lives in an `lru::LruCache` +//! (a hashmap plus an intrusive list), so `touch` promotes in O(1) and eviction +//! pops the least-recently-used tail until back under the cap - no re-sorting. An +//! evicted mirror is transparently re-cloned on its next request, so eviction is a +//! cache-management concern only, never a correctness one. With no cap set, no index +//! is built and the default path keeps its current zero-overhead unbounded-growth +//! behaviour. + +use std::collections::HashSet; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::SystemTime; + +use lru::LruCache; +use tokio::sync::{Notify, watch}; + +use crate::git::GitCache; +use crate::metrics::Metrics; + +struct Inner { + /// Mirror name -> size, kept in access order (front = most-recently-used, back = + /// least). The `lru` crate does the O(1) promote-on-use and tail eviction. + cache: LruCache, + /// Mirrors whose size changed and needs (re)measuring by the background task. + dirty: HashSet, + /// Sum of the sizes in `cache`, maintained incrementally so the cap check is + /// O(1). The byte total, not `LruCache`'s item count, is what bounds the cache. + total: u64, +} + +/// In-memory record of the on-disk mirror cache, plus the eviction trigger. Shared +/// (via `Arc`) between the `GitCache` that mutates it on clone/fetch/serve and the +/// background task in [`run`] that measures and evicts. +pub struct CacheIndex { + cache_root: PathBuf, + max_bytes: u64, + metrics: Arc, + /// Woken when a mirror changes or the cache may be over cap. `notify_one` + /// coalesces a burst into a single maintenance pass and stores a permit if the + /// task is mid-pass, so no wakeup is lost. + work: Notify, + state: Mutex, +} + +impl CacheIndex { + /// Build the index by scanning `cache_root` once. Seeds access order from each + /// mirror's newest file mtime (oldest first, so the oldest lands at the LRU + /// tail), so recency roughly survives a restart. Blocking, but runs at startup + /// before the server binds. + pub fn new(cache_root: PathBuf, max_bytes: u64, metrics: Arc) -> Arc { + let mut mirrors = find_mirrors(&cache_root); + mirrors.sort_by_key(|m| m.mtime); // oldest first -> pushed to the LRU tail first + // Unbounded: the cap is enforced by byte total, not `LruCache`'s item count. + let mut cache: LruCache = LruCache::unbounded(); + let mut total = 0u64; + for m in mirrors { + total += m.size; + cache.put(m.name, m.size); + } + let inner = Inner { + cache, + dirty: HashSet::new(), + total, + }; + metrics.set_cache_size(total, inner.cache.len()); + let over = total > max_bytes; + let idx = Arc::new(Self { + cache_root, + max_bytes, + metrics, + work: Notify::new(), + state: Mutex::new(inner), + }); + if over { + idx.work.notify_one(); // a previous run may have left the cache over-cap + } + idx + } + + /// Mark a repo used. Serving does not grow the cache, so this never wakes the + /// evictor; it only keeps the access order honest. No-op for a repo not yet + /// tracked (its first clone tracks it via `mark_changed`). + pub fn touch(&self, name: &str) { + // `get` promotes to most-recently-used; the value itself is unused. + let _ = self.lock().cache.get(name); + } + + /// Flag a mirror as changed after a clone/fetch: promote it (it was just used), + /// schedule it for measurement, and wake the background task. O(1) bookkeeping + /// only - the size walk happens off the request path. + pub fn mark_changed(&self, name: &str) { + { + let mut inner = self.lock(); + // `get` promotes an existing entry; otherwise track it with a placeholder + // size until the background pass measures it. + if inner.cache.get(name).is_none() { + inner.cache.put(name.to_string(), 0); + } + inner.dirty.insert(name.to_string()); + self.set_gauges(&inner); + } + self.work.notify_one(); + } + + /// On-disk path of a tracked mirror. + pub fn cache_dir(&self, name: &str) -> PathBuf { + self.cache_root.join(name) + } + + /// Current `(total_bytes, mirror_count)` - the values mirrored to the gauges. + pub fn totals(&self) -> (u64, usize) { + let inner = self.lock(); + (inner.total, inner.cache.len()) + } + + /// Take the set of mirrors needing (re)measurement, clearing it. + fn take_dirty(&self) -> Vec { + self.lock().dirty.drain().collect() + } + + /// Set a mirror's measured size, adjusting the running total. Called by the + /// background task after walking the mirror. Uses `peek`/`peek_mut`, which leave + /// recency untouched - a background measurement is not an access. + fn set_size(&self, name: &str, size: u64) { + let mut inner = self.lock(); + let Some(old) = inner.cache.peek(name).copied() else { + return; // evicted between mark and measure + }; + inner.total = inner.total - old + size; + if let Some(v) = inner.cache.peek_mut(name) { + *v = size; + } + self.set_gauges(&inner); + } + + /// Pop least-recently-used mirrors off the tail until the total would be back + /// under the cap, removing them from the index. Returns `(name, dir)` for each + /// so the caller can delete it from disk. Empty when already under cap. + fn take_victims(&self) -> Vec<(String, PathBuf)> { + let mut inner = self.lock(); + let mut victims = Vec::new(); + while inner.total > self.max_bytes { + let Some((name, size)) = inner.cache.pop_lru() else { + break; + }; + inner.total -= size; + let dir = self.cache_root.join(&name); + victims.push((name, dir)); + } + self.set_gauges(&inner); + victims + } + + fn lock(&self) -> MutexGuard<'_, Inner> { + self.state.lock().expect("cache index lock") + } + + fn set_gauges(&self, inner: &Inner) { + self.metrics.set_cache_size(inner.total, inner.cache.len()); + } +} + +/// Background maintenance task. Measures changed mirrors and evicts the LRU tail +/// whenever the index signals work, and exits cleanly when `shutdown` fires (or its +/// sender drops). +pub async fn run( + cache: Arc, + index: Arc, + mut shutdown: watch::Receiver, +) { + loop { + // Run first: covers the startup-over-cap permit and any signal that arrived + // while the previous pass ran (a stored `notify_one` permit makes the next + // wait return immediately, so no work is missed). + maintain(&cache, &index).await; + tokio::select! { + biased; // prefer shutdown over another pass when both are ready + _ = shutdown.changed() => break, + _ = index.work.notified() => {} + } + } + tracing::debug!("cache evictor stopped"); +} + +/// One maintenance pass: (re)measure changed mirrors, then evict the LRU tail until +/// under the cap. All disk IO lives here, off the request path. +async fn maintain(cache: &GitCache, index: &CacheIndex) { + let dirty = index.take_dirty(); + if !dirty.is_empty() { + let dirs: Vec<(String, PathBuf)> = dirty + .into_iter() + .map(|n| { + let dir = index.cache_dir(&n); + (n, dir) + }) + .collect(); + // The walk is blocking; keep it off the runtime. + let measured = tokio::task::spawn_blocking(move || { + dirs.into_iter() + .map(|(name, dir)| (name, measure(&dir).0)) + .collect::>() + }) + .await + .unwrap_or_default(); + for (name, size) in measured { + index.set_size(&name, size); + } + } + + for (name, dir) in index.take_victims() { + match cache.evict(&name, &dir).await { + Ok(()) => { + index.metrics.record_eviction(); + tracing::info!(repo = %name, "evicted idle mirror"); + } + // The entry is already out of the index; a failed unlink just leaves an + // untracked dir on disk, which the next startup scan picks back up. + Err(e) => tracing::warn!(repo = %name, error = %e, "evict failed"), + } + } +} + +/// A mirror found on disk during the startup scan. +struct Scanned { + name: String, + size: u64, + mtime: SystemTime, +} + +/// Discover every mirror under the cache root. Mirrors live at arbitrary depth +/// (`resolve` maps `group/team/foo.git` onto nested dirs), so this descends the +/// namespace dirs - via an explicit stack, not recursion - and stops at each mirror +/// root (a dir with a top-level `HEAD` file, the same "initialised mirror" marker +/// `ensure_fresh` uses). Reserved staging/trash dirs are skipped so they never +/// count toward the budget. Startup-only; steady state is the in-memory index. It +/// reads only the namespace dirs here; `measure` reads inside each mirror, so the +/// two never traverse the same directory twice. +fn find_mirrors(cache_root: &Path) -> Vec { + let mut out = Vec::new(); + let mut stack = vec![cache_root.to_path_buf()]; + while let Some(dir) = stack.pop() { + if dir.join("HEAD").is_file() { + let name = rel_name(cache_root, &dir); + if name.is_empty() { + continue; // the cache root itself is not a mirror + } + let (size, mtime) = measure(&dir); + out.push(Scanned { name, size, mtime }); + continue; // a mirror's subdirs are not themselves mirrors + } + let Ok(entries) = std::fs::read_dir(&dir) else { + continue; + }; + for entry in entries.flatten() { + let Ok(ft) = entry.file_type() else { continue }; + if !ft.is_dir() { + continue; + } + let fname = entry.file_name(); + let fname = fname.to_string_lossy(); + if fname.ends_with(crate::repo::INCOMING_SUFFIX) + || fname.ends_with(crate::repo::EVICTING_SUFFIX) + { + continue; + } + stack.push(entry.path()); + } + } + out +} + +/// A mirror's cache-key name: its path relative to the root, `/`-joined so it +/// matches the key `resolve` produces regardless of the platform separator. +fn rel_name(root: &Path, dir: &Path) -> String { + dir.strip_prefix(root) + .unwrap_or(dir) + .components() + .map(|c| c.as_os_str().to_string_lossy()) + .collect::>() + .join("/") +} + +/// Total byte size of a mirror and the newest mtime among its files. Reused for the +/// startup scan and the background size refresh. A bare mirror is a handful of +/// (mostly packed) files, so the walk cost tracks file count, not bytes. +pub(crate) fn measure(dir: &Path) -> (u64, SystemTime) { + let mut size = 0u64; + let mut mtime = SystemTime::UNIX_EPOCH; + let mut stack = vec![dir.to_path_buf()]; + while let Some(d) = stack.pop() { + let Ok(entries) = std::fs::read_dir(&d) else { + continue; + }; + for entry in entries.flatten() { + let Ok(md) = entry.metadata() else { continue }; + if md.is_dir() { + stack.push(entry.path()); + } else { + size += md.len(); + if let Ok(mt) = md.modified() + && mt > mtime + { + mtime = mt; + } + } + } + } + (size, mtime) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Write; + use std::time::Duration; + + use tokio::sync::watch; + + use crate::git::{GitCache, GitConfig}; + + #[test] + fn size_accounting_tracks_the_total() { + let tmp = tempfile::tempdir().unwrap(); + let idx = CacheIndex::new(tmp.path().to_path_buf(), u64::MAX, Arc::new(Metrics::new())); + idx.mark_changed("a"); // placeholder, size 0 + idx.set_size("a", 100); + idx.mark_changed("b"); + idx.set_size("b", 50); + assert_eq!(idx.totals(), (150, 2)); + idx.set_size("a", 200); // re-measure in place, not a new entry + assert_eq!(idx.totals(), (250, 2)); + } + + #[test] + fn victims_pop_oldest_first_until_under_cap() { + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path(); + let now = SystemTime::now(); + make_mirror( + &root.join("old.git"), + 4096, + Some(now - Duration::from_secs(120)), + ); + make_mirror( + &root.join("mid.git"), + 4096, + Some(now - Duration::from_secs(60)), + ); + make_mirror(&root.join("new.git"), 4096, Some(now)); + + // Each mirror is ~4 KiB; a 6000-byte cap leaves room for one, so the two + // oldest are popped, oldest first. + let idx = CacheIndex::new(root.to_path_buf(), 6000, Arc::new(Metrics::new())); + let names: Vec = idx.take_victims().into_iter().map(|(n, _)| n).collect(); + assert_eq!(names, vec!["old.git".to_string(), "mid.git".to_string()]); + assert_eq!(idx.totals().1, 1); // one mirror left in the index + } + + #[test] + fn touch_promotes_and_spares_from_eviction() { + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path(); + let now = SystemTime::now(); + make_mirror( + &root.join("old.git"), + 4096, + Some(now - Duration::from_secs(120)), + ); + make_mirror( + &root.join("mid.git"), + 4096, + Some(now - Duration::from_secs(60)), + ); + make_mirror(&root.join("new.git"), 4096, Some(now)); + + let idx = CacheIndex::new(root.to_path_buf(), 6000, Arc::new(Metrics::new())); + idx.touch("old.git"); // now the most-recently-used, must be spared + let names: Vec = idx.take_victims().into_iter().map(|(n, _)| n).collect(); + assert_eq!(names, vec!["mid.git".to_string(), "new.git".to_string()]); + } + + #[tokio::test] + async fn maintain_measures_then_evicts_on_disk() { + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path(); + // Empty index, tiny cap; a single mirror appears on disk and is flagged. + make_mirror(&root.join("big.git"), 8192, None); + + let metrics = Arc::new(Metrics::new()); + let idx = CacheIndex::new(root.to_path_buf(), 4096, metrics.clone()); + let cache = GitCache::new(dummy_cfg(), metrics.clone(), Some(idx.clone())); + idx.mark_changed("big.git"); // request path would do this after a clone + + maintain(&cache, &idx).await; // measures big.git (>cap) then evicts it + + assert!( + !root.join("big.git").exists(), + "over-cap mirror should be evicted" + ); + assert_eq!(idx.totals(), (0, 0)); + assert!(metrics.gather().contains("gitcacheproxy_evictions_total 1")); + } + + #[tokio::test] + async fn run_evicts_over_cap_then_stops_on_shutdown() { + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path(); + let now = SystemTime::now(); + make_mirror( + &root.join("old.git"), + 4096, + Some(now - Duration::from_secs(120)), + ); + make_mirror(&root.join("new.git"), 4096, None); + + let metrics = Arc::new(Metrics::new()); + let idx = CacheIndex::new(root.to_path_buf(), 6000, metrics.clone()); // over cap + let cache = Arc::new(GitCache::new( + dummy_cfg(), + metrics.clone(), + Some(idx.clone()), + )); + + let (shutdown_tx, shutdown_rx) = watch::channel(false); + let handle = tokio::spawn(run(cache, idx.clone(), shutdown_rx)); + // `run` drains once before its first `select`, so the eviction completes + // before the task can observe shutdown; awaiting the handle after signalling + // guarantees the drain ran and the loop exited cleanly. + shutdown_tx.send(true).unwrap(); + handle.await.unwrap(); + + assert!(!root.join("old.git").exists(), "oldest mirror evicted"); + assert!(root.join("new.git").exists(), "newest mirror kept"); + assert!(metrics.gather().contains("gitcacheproxy_evictions_total 1")); + } + + #[test] + fn set_size_ignores_an_untracked_mirror() { + let tmp = tempfile::tempdir().unwrap(); + let idx = CacheIndex::new(tmp.path().to_path_buf(), u64::MAX, Arc::new(Metrics::new())); + // No entry for this name (e.g. evicted between mark and measure): no-op. + idx.set_size("never-tracked", 999); + assert_eq!(idx.totals(), (0, 0)); + } + + #[test] + fn scan_skips_stray_files_and_reserved_dirs() { + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path(); + make_mirror(&root.join("good.git"), 1024, None); + // A non-directory entry at the root must be ignored by the walk. + std::fs::write(root.join("stray.txt"), b"x").unwrap(); + // Crashed clone/eviction leftovers must not be scanned as mirrors. + make_mirror( + &root.join(format!("wip.git{}", crate::repo::INCOMING_SUFFIX)), + 1024, + None, + ); + make_mirror( + &root.join(format!("gone.git{}", crate::repo::EVICTING_SUFFIX)), + 1024, + None, + ); + + let idx = CacheIndex::new(root.to_path_buf(), u64::MAX, Arc::new(Metrics::new())); + assert_eq!(idx.totals().1, 1, "only the real mirror is tracked"); + } + + #[tokio::test] + async fn evict_is_a_noop_when_the_mirror_is_already_gone() { + let tmp = tempfile::tempdir().unwrap(); + let cache = GitCache::new(dummy_cfg(), Arc::new(Metrics::new()), None); + // No `HEAD` at this path, so `evict` returns early without touching disk. + let dir = tmp.path().join("absent.git"); + cache.evict("absent.git", &dir).await.unwrap(); + assert!(!dir.exists()); + } + + fn dummy_cfg() -> GitConfig { + // Eviction never shells out to git, so the binary is irrelevant. + GitConfig { + git_binary: "git".into(), + upstream_auth_header: None, + fetch_ttl: Duration::from_secs(10), + } + } + + /// A fake bare mirror: `HEAD` plus a data file summing to at least `data_bytes`. + /// When `mtime` is set, both files are stamped with it so the mirror's last-used + /// signal is deterministic. + fn make_mirror(dir: &Path, data_bytes: usize, mtime: Option) { + std::fs::create_dir_all(dir.join("objects")).unwrap(); + write_file(&dir.join("HEAD"), b"ref: refs/heads/main\n", mtime); + write_file( + &dir.join("objects/pack.data"), + &vec![b'x'; data_bytes], + mtime, + ); + } + + fn write_file(path: &Path, bytes: &[u8], mtime: Option) { + let mut f = std::fs::File::create(path).unwrap(); + f.write_all(bytes).unwrap(); + if let Some(t) = mtime { + f.set_modified(t).unwrap(); + } + } +} diff --git a/src/git.rs b/src/git.rs index 64ba1ea..b9ffc48 100644 --- a/src/git.rs +++ b/src/git.rs @@ -14,6 +14,7 @@ //! stale for the ref the client actually asked for. use std::collections::HashMap; +use std::path::Path; use std::process::Stdio; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -67,14 +68,23 @@ pub struct GitCache { /// anyway, and the critical section is a single O(1) map operation - far too /// short for reader/writer separation to pay off. slots: Mutex>>, + /// Cache-size index for LRU eviction, or `None` when no cap is configured. + /// When present it is `touch`ed on every request and `record`ed after each + /// clone/fetch; when absent the eviction machinery is entirely inert. + index: Option>, } impl GitCache { - pub fn new(cfg: GitConfig, metrics: Arc) -> Self { + pub fn new( + cfg: GitConfig, + metrics: Arc, + index: Option>, + ) -> Self { Self { cfg, metrics, slots: Mutex::new(HashMap::new()), + index, } } @@ -91,11 +101,54 @@ impl GitCache { .clone() } + /// Flag a mirror as changed after a clone/fetch so the background evictor + /// (re)measures it and rebalances the cache. O(1) bookkeeping only - the size + /// walk and any eviction run off this request path. No-op when eviction is + /// disabled. + fn mark_changed(&self, repo: &RepoRef) { + if let Some(idx) = &self.index { + idx.mark_changed(&repo.name); + } + } + + /// Evict a mirror: rename it out of the way, then remove it. Serialized against + /// clone/fetch for the same repo via its slot lock, so it never races the work + /// that populates the mirror. The rename is atomic and fast; the (possibly slow) + /// removal runs after the lock is released. `ensure_fresh` keys off + /// `HEAD.exists()`, so the next request for an evicted repo transparently + /// re-clones. An `upload-pack` already streaming from the old directory keeps + /// its open file descriptors and drains cleanly (POSIX unlink semantics). + pub async fn evict(&self, name: &str, cache_dir: &Path) -> Result<()> { + let slot = self.slot(name).await; + let guard = slot.fetch_lock.lock().await; + if !cache_dir.join("HEAD").exists() { + return Ok(()); // already gone (raced a prior eviction or manual removal) + } + // Rename to a reserved sibling (rejected as a client path by `repo::resolve`) + // and remove any leftover from a crashed prior eviction first. Appending the + // suffix to the full path mirrors `clone_mirror`'s staging discipline. + let mut trash = cache_dir.as_os_str().to_owned(); + trash.push(crate::repo::EVICTING_SUFFIX); + let trash = std::path::PathBuf::from(trash); + let _ = tokio::fs::remove_dir_all(&trash).await; + tokio::fs::rename(cache_dir, &trash) + .await + .with_context(|| format!("rename mirror for eviction: {name}"))?; + drop(guard); // the mirror is gone from its path; free the slot before the slow delete + let _ = tokio::fs::remove_dir_all(&trash).await; + Ok(()) + } + /// Ensure the mirror exists and (when `want_fetch`) is fresh. Concurrent /// callers for the same repo are serialized; the first does the work, the /// rest see it already fresh. pub async fn ensure_fresh(&self, repo: &RepoRef, want_fetch: bool) -> Result { let slot = self.slot(&repo.name).await; + // Mark the repo used on every request - a served cache hit counts as much as + // a fetch - so the eviction index keeps a truthful last-access ordering. + if let Some(idx) = &self.index { + idx.touch(&repo.name); + } let mut last = slot.fetch_lock.lock().await; if !repo.cache_dir.join("HEAD").exists() { @@ -240,6 +293,7 @@ impl GitCache { .await .context("rename mirror into place")?; self.metrics.record_upstream("clone", "ok", &repo.name); + self.mark_changed(repo); Ok(()) } @@ -264,6 +318,7 @@ impl GitCache { bail!("git fetch failed for {}", repo.name); } self.metrics.record_upstream("fetch", "ok", &repo.name); + self.mark_changed(repo); Ok(()) } diff --git a/src/lib.rs b/src/lib.rs index 6758efc..cd1b7d7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -6,6 +6,7 @@ //! the git cache in-process. See the binary crate for the runnable entry point. pub mod config; +pub mod evict; pub mod git; pub mod metrics; pub mod repo; diff --git a/src/main.rs b/src/main.rs index b964fe7..9f4ad80 100644 --- a/src/main.rs +++ b/src/main.rs @@ -15,7 +15,7 @@ use clap::Parser; use tracing_subscriber::EnvFilter; use git_cache_proxy::config::{Config, LogFormat}; -use git_cache_proxy::{git, metrics, server}; +use git_cache_proxy::{evict, git, metrics, server}; #[tokio::main(flavor = "multi_thread")] async fn main() -> Result<()> { @@ -54,16 +54,38 @@ async fn main() -> Result<()> { }; let metrics = Arc::new(metrics::Metrics::new()); + + // Bound the cache on disk only when a cap is set; with `0` no index is built + // and the eviction machinery stays inert, preserving the zero-overhead + // unbounded-growth default. The startup scan runs here, before binding. + let index = (cfg.cache_max_mb > 0).then(|| { + tracing::info!("cache eviction enabled: cap {} MiB", cfg.cache_max_mb); + evict::CacheIndex::new( + cfg.cache_root.clone(), + cfg.cache_max_mb.saturating_mul(1024 * 1024), + metrics.clone(), + ) + }); + + let cache = Arc::new(git::GitCache::new(git_cfg, metrics.clone(), index.clone())); let state = server::AppState { - cache: Arc::new(git::GitCache::new(git_cfg, metrics.clone())), + cache: cache.clone(), upstream_base: cfg.upstream.trim_end_matches('/').to_string(), cache_root: cfg.cache_root.clone(), serve_token: cfg.serve_token.clone(), max_decoded_body: (cfg.max_decoded_body_mb as usize).saturating_mul(1024 * 1024), max_concurrent: cfg.max_concurrent_requests, - metrics, + metrics: metrics.clone(), }; + // Spawn the event-driven evictor and keep its handle so shutdown can stop it + // cleanly rather than leaving it dangling. `watch` carries the shutdown signal. + let evictor = index.map(|idx| { + let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false); + let handle = tokio::spawn(evict::run(cache.clone(), idx, shutdown_rx)); + (shutdown_tx, handle) + }); + let listener = tokio::net::TcpListener::bind(&cfg.bind) .await .with_context(|| format!("bind {}", cfg.bind))?; @@ -78,6 +100,14 @@ async fn main() -> Result<()> { .with_graceful_shutdown(shutdown_signal()) .await .context("http server")?; + + // The server has drained; stop the evictor and wait for it to finish any + // in-progress eviction before exiting. + if let Some((shutdown_tx, handle)) = evictor { + let _ = shutdown_tx.send(true); + let _ = handle.await; + } + tracing::info!("shutdown complete"); Ok(()) } diff --git a/src/metrics.rs b/src/metrics.rs index 309449d..845b4f4 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -11,7 +11,7 @@ //! clones), so a flood of distinct but doomed repo paths cannot inflate the //! series count. -use prometheus::{Encoder, IntCounterVec, Opts, Registry, TextEncoder}; +use prometheus::{Encoder, IntCounter, IntCounterVec, IntGauge, Opts, Registry, TextEncoder}; pub struct Metrics { pub registry: Registry, @@ -22,6 +22,14 @@ pub struct Metrics { /// `upstream_ops_total{op, result, repo}` - op = clone | fetch; result = ok | /// error; repo = the repo path when result = ok, else `-`. upstream: IntCounterVec, + /// `cache_bytes` - total size of the on-disk mirror cache, maintained + /// incrementally as mirrors are added, refreshed, and evicted. Populated only + /// when a cap is configured (`--cache-max-mb`); with no cap it stays `0`. + cache_bytes: IntGauge, + /// `cache_mirrors` - number of cached mirrors (same caveat as `cache_bytes`). + cache_mirrors: IntGauge, + /// `evictions_total` - idle mirrors evicted to keep the cache under the cap. + evictions: IntCounter, } impl Metrics { @@ -40,16 +48,43 @@ impl Metrics { &["op", "result", "repo"], ) .expect("valid metric"); + let cache_bytes = IntGauge::new( + "gitcacheproxy_cache_bytes", + "Total size of the on-disk mirror cache in bytes", + ) + .expect("valid metric"); + let cache_mirrors = IntGauge::new( + "gitcacheproxy_cache_mirrors", + "Number of cached mirrors on disk", + ) + .expect("valid metric"); + let evictions = IntCounter::new( + "gitcacheproxy_evictions_total", + "Idle mirrors evicted to keep the cache under the configured cap", + ) + .expect("valid metric"); registry .register(Box::new(requests.clone())) .expect("register requests"); registry .register(Box::new(upstream.clone())) .expect("register upstream"); + registry + .register(Box::new(cache_bytes.clone())) + .expect("register cache_bytes"); + registry + .register(Box::new(cache_mirrors.clone())) + .expect("register cache_mirrors"); + registry + .register(Box::new(evictions.clone())) + .expect("register evictions"); Self { registry, requests, upstream, + cache_bytes, + cache_mirrors, + evictions, } } @@ -69,6 +104,17 @@ impl Metrics { self.upstream.with_label_values(&[op, result, repo]).inc(); } + /// Refresh the cache-size gauges from the eviction index. + pub fn set_cache_size(&self, bytes: u64, mirrors: usize) { + self.cache_bytes.set(bytes as i64); + self.cache_mirrors.set(mirrors as i64); + } + + /// Record one evicted mirror. + pub fn record_eviction(&self) { + self.evictions.inc(); + } + pub fn gather(&self) -> String { let mut buf = Vec::new(); let enc = TextEncoder::new(); @@ -95,8 +141,14 @@ mod tests { m.record_request("upload_pack", "error", "group/bar.git"); m.record_upstream("fetch", "ok", "group/foo.git"); m.record_upstream("clone", "error", "group/bar.git"); + m.set_cache_size(2048, 3); + m.record_eviction(); + m.record_eviction(); let out = m.gather(); + assert!(out.contains("gitcacheproxy_cache_bytes 2048")); + assert!(out.contains("gitcacheproxy_cache_mirrors 3")); + assert!(out.contains("gitcacheproxy_evictions_total 2")); assert!(out.contains( r#"gitcacheproxy_requests_total{kind="info_refs",repo="group/foo.git",result="ok"} 1"# )); diff --git a/src/repo.rs b/src/repo.rs index f0992df..898e0d4 100644 --- a/src/repo.rs +++ b/src/repo.rs @@ -33,6 +33,12 @@ pub fn repo_name_from_path(path: &str, suffix: &str) -> Option { /// in-flight clone directory. pub const INCOMING_SUFFIX: &str = ".__incoming__"; +/// Reserved suffix for the trash directory a mirror is renamed to during eviction +/// before its (possibly slow) removal (see `git::GitCache::evict`). Reserved for +/// the same reason as `INCOMING_SUFFIX`: a client path must never alias a mirror +/// mid-eviction. +pub const EVICTING_SUFFIX: &str = ".__evicting__"; + /// Validate a repo path (no traversal / absolute / NUL) and resolve it against /// the upstream base and cache root. pub fn resolve(name: &str, upstream_base: &str, cache_root: &Path) -> Result { @@ -44,9 +50,10 @@ pub fn resolve(name: &str, upstream_base: &str, cache_root: &Path) -> Result.__incoming__`). + // (`.__incoming__`) or eviction trash (`.__evicting__`) + // of another. assert!(resolve(&format!("foo{INCOMING_SUFFIX}"), "https://up", root).is_err()); assert!(resolve(&format!("a/b{INCOMING_SUFFIX}"), "https://up", root).is_err()); + assert!(resolve(&format!("foo{EVICTING_SUFFIX}"), "https://up", root).is_err()); + assert!(resolve(&format!("a/b{EVICTING_SUFFIX}"), "https://up", root).is_err()); } } diff --git a/tests/e2e.rs b/tests/e2e.rs index db7a3f1..a206328 100644 --- a/tests/e2e.rs +++ b/tests/e2e.rs @@ -15,6 +15,7 @@ use std::time::Duration; use axum::body::Body; use axum::http::{Request, StatusCode}; +use git_cache_proxy::evict::CacheIndex; use git_cache_proxy::git::{GitCache, GitConfig}; use git_cache_proxy::metrics::Metrics; use git_cache_proxy::server::{AppState, router}; @@ -87,7 +88,7 @@ async fn upload_pack_decodes_gzip_encoded_request() { fetch_ttl: Duration::from_secs(0), }; let state = AppState { - cache: Arc::new(GitCache::new(cfg, metrics.clone())), + cache: Arc::new(GitCache::new(cfg, metrics.clone(), None)), upstream_base: format!("file://{}", up.path().display()), cache_root: cache.path().to_path_buf(), serve_token: None, @@ -167,8 +168,12 @@ async fn clones_through_proxy_serves_all_refs_and_rejects_push() { upstream_auth_header: None, fetch_ttl: Duration::from_secs(0), }; + // Eviction enabled with an effectively unbounded cap: no mirror is ever + // evicted, but the on-request index bookkeeping (touch on serve, mark-changed + // on clone/fetch) runs, which the assertion below checks. + let idx = CacheIndex::new(cache.path().to_path_buf(), u64::MAX, metrics.clone()); let state = AppState { - cache: Arc::new(GitCache::new(cfg, metrics.clone())), + cache: Arc::new(GitCache::new(cfg, metrics.clone(), Some(idx.clone()))), upstream_base: format!("file://{}", up.path().display()), cache_root: cache.path().to_path_buf(), serve_token: None, @@ -260,6 +265,14 @@ async fn clones_through_proxy_serves_all_refs_and_rejects_push() { "missing per-repo fetch metric after second clone:\n{scraped}" ); + // The clone and fetch both flowed through the cache index: it tracks the one + // mirror the proxy created. + assert_eq!( + idx.totals().1, + 1, + "cache index should track the cloned repo" + ); + // --- A push attempt is rejected over the wire (403). --- let mut stream = tokio::net::TcpStream::connect(addr).await.unwrap(); let req = format!( diff --git a/tests/http.rs b/tests/http.rs index 157b394..23129d8 100644 --- a/tests/http.rs +++ b/tests/http.rs @@ -26,7 +26,7 @@ fn state(serve_token: Option) -> AppState { fetch_ttl: Duration::from_secs(10), }; AppState { - cache: Arc::new(GitCache::new(cfg, metrics.clone())), + cache: Arc::new(GitCache::new(cfg, metrics.clone(), None)), upstream_base: "https://upstream.invalid".into(), cache_root, serve_token, @@ -147,9 +147,9 @@ async fn valid_token_passes_auth() { #[tokio::test] async fn reserved_suffix_repo_path_is_rejected() { - // A client path carrying the internal staging suffix could otherwise resolve - // onto a mirror's in-flight clone dir; resolve rejects it before any upstream - // work, over both endpoints. + // A client path carrying an internal suffix could otherwise resolve onto a + // mirror's in-flight clone dir (`.__incoming__`) or a mirror mid-eviction + // (`.__evicting__`); resolve rejects both before any upstream work. let resp = get( state(None), "/repo.__incoming__.git/info/refs?service=git-upload-pack", @@ -157,6 +157,13 @@ async fn reserved_suffix_repo_path_is_rejected() { .await; assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + let resp = get( + state(None), + "/repo.__evicting__.git/info/refs?service=git-upload-pack", + ) + .await; + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + let resp = router(state(None)) .oneshot( Request::post("/repo.__incoming__.git/git-upload-pack") @@ -196,7 +203,7 @@ async fn upstream_failure_returns_bad_gateway_and_records_error() { fetch_ttl: Duration::from_secs(10), }; let st = AppState { - cache: Arc::new(GitCache::new(cfg, metrics.clone())), + cache: Arc::new(GitCache::new(cfg, metrics.clone(), None)), upstream_base: "file:///nonexistent/git-cache-proxy-upstream".into(), cache_root: cache.path().to_path_buf(), serve_token: None,