diff --git a/.env.example b/.env.example index bab92d0..647dc7b 100644 --- a/.env.example +++ b/.env.example @@ -35,6 +35,13 @@ CAIRN_FETCH_TIMEOUT=30s # 256 MiB. Larger than any real package and small enough that a malicious # upstream cannot fill the disk before the read is cut off. CAIRN_MAX_ARTIFACT_BYTES=268435456 +# 32 GiB. The most the blob store may hold: a sweep removes the oldest blobs +# until the store is back within it. Size it against the volume, not against the +# dependency set - the cache refills itself and a miss costs one fetch. +CAIRN_CACHE_MAX_BYTES=34359738368 +# How long a blob is left alone before a sweep may remove it. +CAIRN_CACHE_MIN_AGE=1h +CAIRN_SWEEP_INTERVAL=15m # --- scanner (Node, :8082) ------------------------------------------------- diff --git a/docs/architecture.md b/docs/architecture.md index d875192..6faed44 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -52,6 +52,8 @@ Content-addressing gives three things for free. Two packages shipping identical Writes go to a temporary file and are renamed into place. A fetch killed halfway through used to leave a partial blob that the next request served as a hit, which is the kind of bug that only shows up under load and looks like a corrupted upstream. +The proxy sweeps the store on `CAIRN_SWEEP_INTERVAL`, removing blobs nothing points at and then the oldest blobs until the store is back within `CAIRN_CACHE_MAX_BYTES`. `POST /v1/admin/cache/sweep` runs one now, and `?dry_run=true` reports what one would reclaim without removing anything. + ## The database One Postgres, owned entirely by the registry. Nothing else connects to it except the scanner's queue poll, which reads and claims scan rows with `FOR UPDATE SKIP LOCKED` and touches nothing else. @@ -61,6 +63,5 @@ Schema in [`services/registry/migrations`](../services/registry/migrations), app ## What is deliberately missing - **No object storage.** The blob store is a filesystem path. It should be a volume you can grow, and for a single-node deployment that is enough. Moving to S3-compatible storage is a change to one module. -- **No cache eviction.** The store grows. A cron job deleting blobs whose versions have not been fetched in N days is the obvious next step and has not been needed yet. - **No authentication for the cache route.** The proxy serves anyone who can reach it, and it is expected to be reachable only from inside the network that builds things. Per-organisation cache routes would need a credential in every package manager's configuration, which is a bigger change than it looks. - **No replication, no leader election, no queue broker.** The scan queue is a Postgres table. It has the properties a queue needs at this size, and it is one fewer thing to run. diff --git a/docs/operations.md b/docs/operations.md index f3f5d6b..55df40d 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -42,23 +42,15 @@ Worth alerting on: | `cairn_policy_decisions_total{allowed="false"}` jumping | either a new advisory landed on something popular, or somebody just deployed a policy that refuses more than they meant | | `cairn_scan_reports_total{status="failed"}` rising | the scanner is failing on a class of artifact. Policy will start refusing unscanned versions if any rule has a severity limit | | scans in `pending` growing without bound | the scanner is not claiming work. Its queue poll is the first thing to look at | -| blob store free space | there is no eviction. See below | +| blob store free space | the ceiling is what a sweep enforces, and free space is what tells you the ceiling is wrong. See below | Every request through either service logs a request id and echoes it in `X-Request-Id`, honouring an inbound one if it looks like an identifier. A report about one bad request should quote it; correlating the proxy's line with the registry's is the reason the header is passed through. -## The cache grows and nothing evicts +## Keeping the cache inside its volume -There is no eviction. This is a known gap rather than a design decision, and it is the operational thing most likely to bite you first. +`CAIRN_CACHE_MAX_BYTES` is the most the blob store may hold, and the proxy sweeps every `CAIRN_SWEEP_INTERVAL` to keep it there. A sweep clears the partial writes left by fetches that died, removes blobs no index entry points at, and then removes the oldest blobs until the store is back within the ceiling. `CAIRN_CACHE_MIN_AGE` is the grace period underneath that, so a package one job in a pipeline fetched is still there for the next. Size the ceiling against the volume rather than against the dependency set: the cache refills itself and a miss costs one fetch, so too low is slow builds and too high is a full disk. `POST /v1/admin/cache/sweep` runs one immediately, takes the same bearer token as the purge route, and with `?dry_run=true` reports what a sweep would reclaim without removing anything — which is how to try a new ceiling before setting it. -Watch free space on `CAIRN_BLOB_DIR`. When it becomes a problem, the safe manual intervention is to delete blobs whose versions have not been fetched recently: - -```sql -SELECT v.digest - FROM versions v - WHERE v.cached_at < now() - interval '90 days'; -``` - -Deleting a blob loses nothing permanently. The metadata stays in the database, the digest is still recorded, and the next request for it fetches again and verifies against the same digest. Set `cached_at` to null for the rows you removed so the dashboard stops claiming they are cached. +Deleting a blob loses nothing permanently. The metadata stays in the database, the digest is still recorded, and the next request for it fetches again and verifies against the same digest. Watch `cairn_proxy_sweep_bytes_reclaimed_total`: a sweep reclaiming nothing on a volume that is filling means everything in the store is either referenced or inside its grace period. Do not delete rows from `versions` to reclaim space. The digest recorded there is the evidence that a version's bytes have not changed under us, and removing it means the next fetch records whatever the upstream serves as if it were the first sighting. diff --git a/services/proxy/Cargo.toml b/services/proxy/Cargo.toml index 9ec77cd..7ff0767 100644 --- a/services/proxy/Cargo.toml +++ b/services/proxy/Cargo.toml @@ -36,9 +36,10 @@ serde_json = "1.0.151" sha2 = "0.11.0" thiserror = "2.0.20" # `net` for the listener and `io-util` for the file reads and writes, on top of -# the runtime, the macros and the signal handling. Declared rather than inherited: -# a build that works because another crate happens to enable a tokio feature -# breaks the day that crate stops needing it. +# the runtime, the macros and the signal handling. `time` and `sync` are the +# background sweep's interval and the lock keeping two of them apart. Declared +# rather than inherited: a build that works because another crate happens to +# enable a tokio feature breaks the day that crate stops needing it. tokio = { version = "1.53.1", features = [ "rt-multi-thread", "macros", @@ -46,6 +47,8 @@ tokio = { version = "1.53.1", features = [ "io-util", "net", "signal", + "sync", + "time", ] } # Request tracing and a timeout on the routes that should have one. See # `routes::router` for why the cache route is not one of them. diff --git a/services/proxy/src/config.rs b/services/proxy/src/config.rs index fe2260f..ca21d8a 100644 --- a/services/proxy/src/config.rs +++ b/services/proxy/src/config.rs @@ -60,6 +60,12 @@ pub struct Config { pub upstreams: BTreeMap, pub fetch_timeout: Duration, pub max_artifact_bytes: u64, + /// The most the blob store may hold. A sweep removes the oldest blobs until + /// the store is back within it. + pub cache_max_bytes: u64, + /// How long a blob is left alone before a sweep may remove it. + pub cache_min_age: Duration, + pub sweep_interval: Duration, pub log_level: String, pub log_format: LogFormat, } @@ -71,6 +77,11 @@ const DEFAULT_FETCH_TIMEOUT: Duration = Duration::from_secs(30); /// 256 MiB. Larger than any real package and small enough that a malicious /// upstream cannot fill the disk before the read is cut off. const DEFAULT_MAX_ARTIFACT_BYTES: u64 = 256 * 1024 * 1024; +/// 32 GiB. More than a large monorepo's dependency set, and small enough to fit +/// the volume a proxy is given. +const DEFAULT_CACHE_MAX_BYTES: u64 = 32 * 1024 * 1024 * 1024; +const DEFAULT_CACHE_MIN_AGE: Duration = Duration::from_hours(1); +const DEFAULT_SWEEP_INTERVAL: Duration = Duration::from_mins(15); /// Every problem found in the environment, not just the first. An operator /// fixing three unset variables should learn about all three from one failed @@ -91,6 +102,12 @@ pub fn load() -> Result { .unwrap_or(DEFAULT_FETCH_TIMEOUT); let max_artifact_bytes = parsed("CAIRN_MAX_ARTIFACT_BYTES", parse_bytes, &mut problems) .unwrap_or(DEFAULT_MAX_ARTIFACT_BYTES); + let cache_max_bytes = parsed("CAIRN_CACHE_MAX_BYTES", parse_bytes, &mut problems) + .unwrap_or(DEFAULT_CACHE_MAX_BYTES); + let cache_min_age = parsed("CAIRN_CACHE_MIN_AGE", parse_duration, &mut problems) + .unwrap_or(DEFAULT_CACHE_MIN_AGE); + let sweep_interval = parsed("CAIRN_SWEEP_INTERVAL", parse_duration, &mut problems) + .unwrap_or(DEFAULT_SWEEP_INTERVAL); let log_level = parsed("CAIRN_LOG_LEVEL", parse_level, &mut problems).unwrap_or_else(|| "info".to_owned()); let log_format = @@ -108,6 +125,12 @@ pub fn load() -> Result { if max_artifact_bytes == 0 { problems.push("CAIRN_MAX_ARTIFACT_BYTES: must be at least one byte".to_owned()); } + if cache_max_bytes == 0 { + problems.push("CAIRN_CACHE_MAX_BYTES: must be at least one byte".to_owned()); + } + if sweep_interval.is_zero() { + problems.push("CAIRN_SWEEP_INTERVAL: must be longer than zero".to_owned()); + } let upstreams = upstreams(&mut problems); @@ -122,6 +145,9 @@ pub fn load() -> Result { upstreams, fetch_timeout, max_artifact_bytes, + cache_max_bytes, + cache_min_age, + sweep_interval, log_level, log_format, }) diff --git a/services/proxy/src/main.rs b/services/proxy/src/main.rs index 37be429..452215b 100644 --- a/services/proxy/src/main.rs +++ b/services/proxy/src/main.rs @@ -14,18 +14,20 @@ mod metrics; mod policy; mod routes; mod store; +mod sweep; mod upstream; use std::process::ExitCode; use tokio::net::TcpListener; use tokio::signal::unix::{signal, SignalKind}; +use tokio::time::MissedTickBehavior; use tracing_subscriber::layer::SubscriberExt as _; use tracing_subscriber::util::SubscriberInitExt as _; use tracing_subscriber::EnvFilter; use crate::config::{format_bytes, Config, LogFormat}; -use crate::routes::{App, Startup}; +use crate::routes::{App, Shared, Startup}; #[tokio::main] async fn main() -> ExitCode { @@ -83,6 +85,8 @@ async fn serve(cfg: Config) -> Result<(), Startup> { "proxy starting" ); + spawn_sweeps(app.clone()); + axum::serve(listener, routes::router(app.clone())) .with_graceful_shutdown(shutdown()) .await?; @@ -91,6 +95,31 @@ async fn serve(cfg: Config) -> Result<(), Startup> { Ok(()) } +/// Sweeps the blob store on the configured interval for as long as the process +/// runs. +/// +/// The first tick is one interval away, which is what keeps a proxy that has only +/// just started from sweeping a cache it has not begun to fill. Not part of the +/// graceful shutdown either: a sweep is a sequence of independent unlinks with no +/// half-finished state to protect, so stopping one anywhere leaves the store +/// exactly as consistent as letting it finish would. +fn spawn_sweeps(app: Shared) { + tokio::spawn(async move { + let mut ticks = tokio::time::interval(app.cfg.sweep_interval); + // A sweep that overran its period must not be followed by a burst of + // sweeps catching up on the ticks it missed. The next one starts a full + // interval after this one finished. + ticks.set_missed_tick_behavior(MissedTickBehavior::Delay); + + loop { + ticks.tick().await; + let reclaimed = app.sweeper.run().await; + app.metrics.record_sweep(reclaimed.removed, reclaimed.bytes); + tracing::info!(?reclaimed, "swept the blob store"); + } + }); +} + /// Waits for a terminate or interrupt. /// /// Both, and not just `SIGTERM`: a container runtime sends one and a developer's diff --git a/services/proxy/src/metrics.rs b/services/proxy/src/metrics.rs index 7974978..d0bb742 100644 --- a/services/proxy/src/metrics.rs +++ b/services/proxy/src/metrics.rs @@ -1,4 +1,4 @@ -//! The four counters worth having, and the Prometheus text format by hand. +//! The six counters worth having, and the Prometheus text format by hand. //! //! By hand because that is all this needs. The exposition below is twenty lines //! and the format has been stable for a decade; a client library would bring a @@ -8,7 +8,7 @@ //! Nothing here is labelled. Every label this service could plausibly attach - //! ecosystem, package, version - is unbounded or nearly so, and an artifact name in //! a metric is how a Prometheus instance falls over. Anything finer-grained than -//! these four numbers is a question for the access log. +//! these six numbers is a question for the access log. use std::fmt::Write as _; use std::sync::atomic::{AtomicU64, Ordering}; @@ -23,12 +23,14 @@ pub struct Metrics { cache_misses: AtomicU64, upstream_failures: AtomicU64, policy_refusals: AtomicU64, + sweep_bytes_reclaimed: AtomicU64, + sweep_blobs_removed: AtomicU64, } /// Name and help for each counter, in the order [`Metrics::values`] reads them. /// Keeping the two side by side is what stops a fifth counter from being rendered /// under a fourth one's name. -const COUNTERS: [(&str, &str); 4] = [ +const COUNTERS: [(&str, &str); 6] = [ ( "cairn_proxy_cache_hits_total", "Artifacts served from the local blob store.", @@ -45,10 +47,18 @@ const COUNTERS: [(&str, &str); 4] = [ "cairn_proxy_policy_refusals_total", "Artifacts the registry's policy refused to allow.", ), + ( + "cairn_proxy_sweep_blobs_removed_total", + "Cached blobs a sweep removed to bring the store under its ceiling.", + ), + ( + "cairn_proxy_sweep_bytes_reclaimed_total", + "Bytes those blobs occupied.", + ), ]; impl Metrics { - /// `Relaxed` throughout. These are four independent totals - nothing reads one + /// `Relaxed` throughout. These are six independent totals - nothing reads one /// to decide something about another - so there is no ordering between them for /// a stronger ordering to protect. pub fn record_hit(&self) { @@ -67,18 +77,28 @@ impl Metrics { self.policy_refusals.fetch_add(1, Ordering::Relaxed); } - fn values(&self) -> [u64; 4] { + /// Records what one sweep reclaimed. Two counters, because forty thousand tiny + /// blobs and one large blob look identical in bytes alone. + pub fn record_sweep(&self, blobs: u64, bytes: u64) { + self.sweep_blobs_removed.fetch_add(blobs, Ordering::Relaxed); + self.sweep_bytes_reclaimed + .fetch_add(bytes, Ordering::Relaxed); + } + + fn values(&self) -> [u64; 6] { [ self.cache_hits.load(Ordering::Relaxed), self.cache_misses.load(Ordering::Relaxed), self.upstream_failures.load(Ordering::Relaxed), self.policy_refusals.load(Ordering::Relaxed), + self.sweep_bytes_reclaimed.load(Ordering::Relaxed), + self.sweep_blobs_removed.load(Ordering::Relaxed), ] } /// Renders one scrape. /// - /// The four loads are not a snapshot of a single instant, and do not need to + /// The six loads are not a snapshot of a single instant, and do not need to /// be: a scrape is a sample, the next one is fifteen seconds away, and a /// counter that was one behind for a microsecond is invisible in every query /// anybody writes over it. diff --git a/services/proxy/src/routes/admin.rs b/services/proxy/src/routes/admin.rs index 728026e..dcccd28 100644 --- a/services/proxy/src/routes/admin.rs +++ b/services/proxy/src/routes/admin.rs @@ -1,9 +1,9 @@ -//! The purge route. +//! The purge and sweep routes. //! //! Deleting a blob is safe in the sense that nothing is lost - the store is //! content-addressed and the next request fetches it again - and unsafe in the //! sense that anyone who can do it in a loop can point this proxy's whole fetch -//! volume at an upstream registry. So it is the one route here that asks for a +//! volume at an upstream registry. So these are the two routes here that ask for a //! credential. //! //! The credential is `CAIRN_SERVICE_TOKEN`, which the proxy already holds because @@ -12,9 +12,11 @@ //! worse than one whose credential is shared with the scanner. When the proxy grows //! real inbound authentication, this function is where it goes. -use axum::extract::{Path, State}; +use axum::extract::{Path, Query, State}; use axum::http::{HeaderMap, StatusCode}; use axum::response::{IntoResponse, Response}; +use axum::Json; +use serde::Deserialize; use crate::digest::Digest; use crate::error::Error; @@ -46,6 +48,43 @@ async fn handle(app: &App, headers: &HeaderMap, raw: &str) -> Result, + RequestId(request_id): RequestId, + headers: HeaderMap, + Query(query): Query, +) -> Response { + match reclaim(&app, &headers, query.dry_run) { + Ok(response) => response, + Err(error) => error.into_response_with(request_id.as_deref()), + } +} + +fn reclaim(app: &App, headers: &HeaderMap, dry_run: bool) -> Result { + authorise(app, headers)?; + + let reclaimed = app.sweeper.sweep(dry_run); + // A dry run is not counted. The counters are what an operator alerts on, and a + // total that moved because somebody asked a question would fire the alert on a + // store exactly as full as it was. + if !dry_run { + app.metrics.record_sweep(reclaimed.removed, reclaimed.bytes); + } + + tracing::info!(dry_run, ?reclaimed, "swept the blob store"); + Ok(Json(reclaimed).into_response()) +} + /// Checks the bearer token against the configured service token. fn authorise(app: &App, headers: &HeaderMap) -> Result<(), Error> { let presented = headers diff --git a/services/proxy/src/routes/mod.rs b/services/proxy/src/routes/mod.rs index 411e3bf..c751d33 100644 --- a/services/proxy/src/routes/mod.rs +++ b/services/proxy/src/routes/mod.rs @@ -1,9 +1,9 @@ //! The HTTP surface, and the state the handlers share. //! -//! Five routes. Three of them - the two probes and the metrics endpoint - are +//! Six routes. Three of them - the two probes and the metrics endpoint - are //! unauthenticated, deliberately: a liveness probe that needs a credential reports //! unhealthy the day the credential expires, and none of the three reveals -//! anything. The purge route carries its own check; see `admin`. +//! anything. The purge and sweep routes carry their own check; see `admin`. pub mod admin; pub mod cache; @@ -15,7 +15,7 @@ use axum::extract::{FromRequestParts, State}; use axum::http::request::Parts; use axum::http::{header, HeaderValue, Request, Response, StatusCode}; use axum::response::IntoResponse; -use axum::routing::{delete, get}; +use axum::routing::{delete, get, post}; use axum::{Json, Router}; use serde_json::json; use tower_http::timeout::TimeoutLayer; @@ -26,6 +26,7 @@ use crate::error::Error; use crate::metrics::Metrics; use crate::policy::Registry; use crate::store::BlobStore; +use crate::sweep::Sweeper; use crate::upstream::Upstreams; /// Anything a startup step can fail with. The message goes to the log and the @@ -40,6 +41,7 @@ pub struct App { pub store: BlobStore, pub upstreams: Upstreams, pub registry: Registry, + pub sweeper: Sweeper, pub metrics: Metrics, } @@ -50,12 +52,14 @@ impl App { let store = BlobStore::open(cfg.blob_dir.clone()).await?; let upstreams = Upstreams::build(&cfg)?; let registry = Registry::build(&cfg.registry_url, cfg.service_token.expose())?; + let sweeper = Sweeper::new(&cfg); Ok(Arc::new(Self { cfg, store, upstreams, registry, + sweeper, metrics: Metrics::default(), })) } @@ -102,10 +106,6 @@ fn plausible(id: &str) -> bool { /// Builds the router. pub fn router(app: Shared) -> Router { - // The cache route is deliberately outside the timeout. A request timeout there - // would cut a 200 MiB download that is proceeding perfectly well; the stall it - // is meant to catch is already caught by the idle timeout on the upstream - // client, which is the layer that can tell the difference. let bounded = Router::new() .route("/healthz", get(healthz)) .route("/readyz", get(readyz)) @@ -119,13 +119,19 @@ pub fn router(app: Shared) -> Router { Duration::from_secs(10), )); - let streaming = Router::new().route( - "/v1/cache/{ecosystem}/{name}/{version}/{file}", - get(cache::serve), - ); + // Both of these sit outside the timeout on purpose. A request timeout would + // cut a 200 MiB download that is proceeding perfectly well - the stall it is + // meant to catch is already caught by the idle timeout on the upstream client - + // and a sweep of a large store legitimately takes longer than ten seconds. + let untimed = Router::new() + .route( + "/v1/cache/{ecosystem}/{name}/{version}/{file}", + get(cache::serve), + ) + .route("/v1/admin/cache/sweep", post(admin::sweep)); bounded - .merge(streaming) + .merge(untimed) .layer(axum::middleware::from_fn(echo_request_id)) .layer(TraceLayer::new_for_http()) .with_state(app) diff --git a/services/proxy/src/sweep.rs b/services/proxy/src/sweep.rs new file mode 100644 index 0000000..7a37738 --- /dev/null +++ b/services/proxy/src/sweep.rs @@ -0,0 +1,257 @@ +//! Reclaiming space from the blob store. +//! +//! The store only ever grew: every miss adds a blob and nothing takes one away +//! except an operator purging a digest by hand. A sweep walks what is on disk and +//! removes until the total is back under `CAIRN_CACHE_MAX_BYTES` - first the +//! partial writes left in `incoming` by fetches that did not finish, then blobs no +//! index entry points at, then the oldest blobs. None of it is lost permanently, +//! because the store is content-addressed and the registry keeps the metadata. +//! +//! One sweep at a time. Two of them over one directory would each decide what to +//! remove from a total the other is already changing, and between them they would +//! take the store far below the ceiling either was aiming at. +//! +//! The walk is synchronous. `tokio::fs` hands every operation to the blocking +//! pool, and a store with sixty-five thousand leaf directories under it would be +//! sixty-five thousand round trips through that pool to answer a question that is +//! almost entirely `readdir`. One pass of standard-library calls costs less than +//! the scheduling would. + +use std::collections::BTreeMap; +use std::fs; +use std::path::{Path, PathBuf}; +use std::time::{Duration, SystemTime}; + +use serde::Serialize; +use tokio::sync::Mutex; + +use crate::config::Config; +use crate::digest::Digest; +use crate::store::IndexEntry; + +/// The three directories under `CAIRN_BLOB_DIR`. Named again here rather than +/// reached through `BlobStore`, which exposes one digest at a time on purpose; +/// `store`'s module documentation is the definition of the layout. +const BLOBS: &str = "blobs/sha256"; +const INDEX: &str = "index"; +const INCOMING: &str = "incoming"; + +/// What one sweep did. Serialised to the caller of the on-demand route. +#[derive(Debug, Serialize)] +pub struct Reclaimed { + /// Bytes of blob the store held when the sweep started. + pub held: u64, + /// Blobs the sweep looked at. + pub scanned: u64, + /// Blobs it removed. + pub removed: u64, + /// Bytes those blobs occupied. Blobs only. + pub bytes: u64, + /// Abandoned partial writes it removed. + pub partials: u64, +} + +struct Candidate { + path: PathBuf, + size: u64, + age: Duration, +} + +/// Reclaims space from one blob directory. +pub struct Sweeper { + root: PathBuf, + max_bytes: u64, + min_age: Duration, + /// Held for the whole of a sweep, so this module's one-at-a-time promise is + /// something the type keeps rather than something every caller remembers. + running: Mutex<()>, +} + +impl Sweeper { + #[must_use] + pub fn new(cfg: &Config) -> Self { + Self { + root: cfg.blob_dir.clone(), + max_bytes: cfg.cache_max_bytes, + min_age: cfg.cache_min_age, + running: Mutex::new(()), + } + } + + /// Runs one sweep, waiting for any sweep already under way to finish first. + pub async fn run(&self) -> Reclaimed { + let _running = self.running.lock().await; + self.sweep(false) + } + + /// One pass over the store. `dry_run` answers what a pass would reclaim + /// without removing a blob, so a new ceiling can be tried before it is set. + #[must_use] + pub fn sweep(&self, dry_run: bool) -> Reclaimed { + let partials = clear_partials(&self.root.join(INCOMING), dry_run); + let index = referenced(&self.root.join(INDEX)); + + let mut candidates = Vec::new(); + collect(&self.root.join(BLOBS), &mut candidates); + + let held: u64 = candidates.iter().map(|candidate| candidate.size).sum(); + let scanned = u64::try_from(candidates.len()).unwrap_or(u64::MAX); + + // Oldest first, so the blobs nothing has wanted for longest are the ones + // that go and the package somebody fetched this morning stays. + candidates.sort_by_key(|candidate| candidate.age); + + let mut remaining = held; + let mut removed = 0; + let mut bytes = 0; + let mut gone = Vec::new(); + + for candidate in candidates { + let Some(hex) = candidate.path.file_name().and_then(|name| name.to_str()) else { + continue; + }; + + // Two reasons to remove one. Nothing points at it, or the store is + // over its ceiling and this blob is old enough to be the one that + // goes. Age does not enter into the first: a blob no index entry + // names cannot be reached however new it is, and keeping one for an + // hour only holds bytes nothing is going to ask for. + let remove = !index.contains_key(hex) + || (remaining >= self.max_bytes && candidate.age >= self.min_age); + if !remove { + continue; + } + + if !dry_run { + if let Err(error) = fs::remove_file(&candidate.path) { + // Usually a blob that has already gone - a purge, or another + // proxy over the same directory reached it first - and the + // next request for it simply misses. + tracing::debug!(path = %candidate.path.display(), %error, "could not remove a cached blob"); + } + } + + remaining = remaining.saturating_sub(candidate.size); + bytes += candidate.size; + removed += 1; + gone.push(hex.to_owned()); + } + + forget(&index, &gone); + + Reclaimed { + held, + scanned, + removed, + bytes, + partials, + } + } +} + +/// Every file under `dir`, with what a sweep decides on. +fn collect(dir: &Path, out: &mut Vec) { + let Ok(entries) = fs::read_dir(dir) else { + return; + }; + + for entry in entries.flatten() { + let path = entry.path(); + // A store moved between volumes often has blobs linked into it rather + // than copied. A link is counted at the size of the link and not of + // whatever it points at, so the total stays the number of bytes this + // directory is answerable for. + let Ok(meta) = fs::metadata(&path) else { + continue; + }; + if meta.is_dir() { + collect(&path, out); + continue; + } + out.push(Candidate { + path, + size: meta.len(), + age: age_of(&meta), + }); + } +} + +/// How long ago something was last written. +/// +/// `duration_since` fails when the recorded time is ahead of this machine's +/// clock, which is routine on a shared volume whose server is a few seconds out. +/// The failure flattens rather than propagating: a sweep that gave up because one +/// blob carried an odd timestamp would be a sweep that never ran. +fn age_of(meta: &fs::Metadata) -> Duration { + meta.modified() + .ok() + .and_then(|at| SystemTime::now().duration_since(at).ok()) + .unwrap_or_default() +} + +/// Every digest the index still points at, and the entries naming it. +/// +/// A map rather than a set because one digest is reachable through several sets +/// of coordinates. An entry that will not parse is skipped rather than read as +/// naming nothing: `BlobStore::resolve` treats it as a miss and refetches, so +/// deciding here that it references no blob would delete the bytes that refetch +/// is about to find. +fn referenced(dir: &Path) -> BTreeMap> { + let mut entries = Vec::new(); + collect(dir, &mut entries); + + let mut out: BTreeMap> = BTreeMap::new(); + for entry in entries { + let Some(hex) = fs::read(&entry.path) + .ok() + .and_then(|raw| serde_json::from_slice::(&raw).ok()) + .and_then(|record| Digest::parse(&record.digest).ok()) + .map(|digest| digest.hex().to_owned()) + else { + continue; + }; + out.entry(hex).or_default().push(entry.path); + } + out +} + +/// Removes the partial writes left behind by fetches that did not finish. +/// +/// A partial carries no digest and no name anything can look up - `BlobWriter` +/// gives it one only once every byte has been hashed - so there is nothing to +/// weigh up here the way there is for a blob. A file in `incoming` is a fetch +/// that is not coming back, and its bytes are as good as free. +fn clear_partials(dir: &Path, dry_run: bool) -> u64 { + let Ok(entries) = fs::read_dir(dir) else { + return 0; + }; + + let mut removed = 0; + for entry in entries.flatten() { + if !dry_run { + if let Err(error) = fs::remove_file(entry.path()) { + tracing::debug!(path = %entry.path().display(), %error, "could not remove an abandoned partial blob"); + continue; + } + } + removed += 1; + } + removed +} + +/// Drops the index entries naming digests that are no longer in the store. +/// +/// `BlobStore::remove` leaves these behind on purpose: finding the entries for +/// one digest means walking the whole index, and a stale entry costs a single +/// refetch. That walk has already happened by the time a sweep reaches here, so +/// there is no reason to leave a lookup that can only ever answer with a blob +/// that is gone. +fn forget(index: &BTreeMap>, gone: &[String]) { + for hex in gone { + for path in index.get(hex).into_iter().flatten() { + if let Err(error) = fs::remove_file(path) { + tracing::debug!(path = %path.display(), %error, "could not remove a stale index entry"); + } + } + } +}