git ents: feat: add git-maintenance, the WS9 GC, compaction, and scheduling crate

commit 3ac203c

feat: add git-maintenance, the WS9 GC, compaction, and scheduling crate

Mark from ref tips via git_reachability::gc_mark with reachability artifacts as accelerator; sweep via the pack registry: fully unreachable packs are deleted registry-first, mixed packs are rewritten through the WS5 pack writer honoring lifetime classes, and cache packs only ever die whole by registry delete (rules 4 and 5). Quarantine is structurally unscannable by the sweep. Cache-ref TTL eviction and the rule-4 consolidation effect — the only multi-ref cache writer, one atomic transaction — live here, and every maintenance run serializes per repo under an advisory lock: Postgres session advisory locks in cloud deployments, a file lock locally. schedule_maintenance enqueues the maintenance EffectDefs (including WS6’s deferred reachability regeneration) on ref-update volume thresholds, wired into the server’s native receive-pack endpoint. Real conformance Collectors for the files and tigris backends close the seam WS2 left open, including the staging-timeout boundary.

feat: add per-repo maintenance advisory locks to refstore-postgres feat: implement real conformance `Collector`s for files and tigris feat: schedule maintenance from the server post-ingest path Assisted-by: Claude:claude-sonnet-5

Joseph D. Carpinelli · 1 month ago

tree browse at HEAD · parents: 6cb907d · comment on this commit

Reviews

No reviews of this commit yet — record a verdict below.

Start a review

Cargo.lock @@ -1722,6 +1722,7 @@ "git-effect", "git-ents-core", "git-hydrate", + "git-maintenance", "git-member", "git-protocol", "git-reachability", @@ -1765,6 +1766,29 @@ "uuid", ] +[[package]] +name = "git-maintenance" +version = "0.0.0" +dependencies = [ + "backend-conformance", + "effect-dispatcher", + "git-backend", + "git-protocol", + "git-reachability", + "git-store", + "gix-hash", + "gix-object", + "gix-pack", + "odb-files", + "odb-tigris", + "refstore-files", + "refstore-postgres", + "tempfile", + "tokio", + "tokio-postgres", + "uuid", +] + [[package]] name = "git-member" version = "0.0.0"

Cargo.toml @@ -13,6 +13,7 @@ "crates/git-ents-core", "crates/git-ents-server", "crates/git-hydrate", + "crates/git-maintenance", "crates/git-member", "crates/git-protocol", "crates/git-reachability", @@ -79,6 +80,7 @@ git-ents-core = { path = "crates/git-ents-core" } git-ents-server = { path = "crates/git-ents-server" } git-hydrate = { path = "crates/git-hydrate" } +git-maintenance = { path = "crates/git-maintenance" } git-member = { path = "crates/git-member" } git-protocol = { path = "crates/git-protocol" } git-reachability = { path = "crates/git-reachability" }

crates/git-ents-server/Cargo.toml @@ -23,6 +23,7 @@ git-comment = { workspace = true } git-effect = { workspace = true } git-hydrate = { workspace = true } +git-maintenance = { workspace = true } git-member = { workspace = true } git-protocol = { workspace = true } git-reachability = { workspace = true }

crates/backend-conformance/src/collector.rs @@ -1,11 +1,12 @@ //! [`Collector`]: the seam [`crate::causal_collection_safety`] tests -//! against. Local file backends (`refstore-files`, `odb-files`) have no GC -//! wired up yet (`docs/scale-out.adoc`, WS1), so [`NoopCollector`] lets -//! today's suite exercise what a collection pass must never do — touch a -//! staged/quarantined object — without asserting behavior no collector -//! implements yet. A backend with real GC plugs its own `Collector` -//! (reporting a real [`Collector::staging_grace`] window, if it has one) -//! into the same property function instead. +//! against. [`NoopCollector`] lets a backend with no GC wired up exercise +//! what a collection pass must never do — touch a staged/quarantined +//! object — without asserting behavior no collector implements. A backend +//! with real GC plugs its own `Collector` (reporting a real +//! [`Collector::staging_grace`] window, if it has one) into the same +//! property function instead — `git-maintenance` (WS9) does exactly that +//! for the files and Tigris backends, including the staging-timeout +//! boundary (`crates/git-maintenance/tests/conformance.rs`). use std::time::Duration;

crates/git-ents-server/src/http.rs @@ -718,6 +718,7 @@ web_signing_key: None, live_runs: git_effect::engine::new_live_registry(), hydrate: None, + maintenance: None, } }

crates/git-ents-server/src/lib.rs @@ -112,6 +112,15 @@ /// every served repo from Postgres/blob-store durable state on every /// request, per `crate::http`'s wiring. pub(crate) hydrate: Option<git_hydrate::HydrateConfig>, + /// WS9 maintenance scheduler (`docs/scale-out.adoc`, "WS9 — GC, + /// compaction, maintenance"): accumulates per-repo ref-update volume + /// from accepted pushes (see `native_git::receive_pack`, the wired + /// call site) and enqueues the maintenance effects once a repo + /// crosses the threshold. Present only when hydration (and so the + /// Postgres effect queue the effects are enqueued into) is + /// configured; a direct-disk deployment has no queue to schedule + /// into yet. + pub(crate) maintenance: Option<Arc<git_maintenance::schedule::Scheduler>>, } /// The non-empty value of the environment variable `key`, or `None`. @@ -198,6 +207,18 @@ // spawns must agree on. let hydrate = git_hydrate::HydrateConfig::from_env(); + // WS9: schedule maintenance on ref-update volume (`docs/scale-out.adoc`, + // "Reachability": "triggered by ref-update volume thresholds"), into + // the same Postgres effect queue the WS7 dispatcher drains. + let maintenance = hydrate.as_ref().map(|config| { + Arc::new(git_maintenance::schedule::Scheduler::new( + git_maintenance::schedule::Thresholds::default(), + Arc::new(git_maintenance::schedule::PostgresQueueSink::new( + config.postgres_conninfo.clone(), + )), + )) + }); + let state = AppState { data_dir, init_lock: Arc::new(Mutex::new(())), @@ -209,6 +230,7 @@ web_signing_key, live_runs: git_effect::engine::new_live_registry(), hydrate, + maintenance, }; // Drain queued pushes and run their effects for the life of the server:

crates/git-ents-server/src/native_git.rs @@ -232,6 +232,26 @@ push_cert: parsed.cert.map(PushCertificate::new), }; let outcome = backend(state).receive(push); + + // WS9's wired call site: an accepted push reports its ref-update + // volume to the maintenance scheduler, which enqueues the maintenance + // effects once the repo crosses its threshold (`docs/scale-out.adoc`, + // "Reachability" / WS9). Off the request path (`spawn_blocking` — the + // sink opens a Postgres connection when the threshold trips) and + // never able to fail the push: scheduling errors are logged, the next + // accepted push re-triggers. + if matches!(outcome, Ok(git_protocol::PushOutcome::Accepted { .. })) + && let Some(scheduler) = state.maintenance.clone() + { + let repo_id = repo_rel.to_owned(); + let updates = names.len() as u64; + drop(tokio::task::spawn_blocking(move || { + if let Err(error) = scheduler.note_ref_updates(&repo_id, updates) { + eprintln!("maintenance: could not schedule for {repo_id}: {error}"); + } + })); + } + report_status(&names, &outcome) }

crates/refstore-postgres/src/queue.rs @@ -244,6 +244,55 @@ .map(|_rows_affected| ()) } + /// Try to take this repository's maintenance advisory lock + /// (`docs/scale-out.adoc`, WS9: "Per-repo background effects + /// serialized by advisory lock"): a Postgres *session* advisory lock + /// keyed by `hashtextextended(repo_id)`, held by this store's + /// connection until [`Self::unlock_maintenance`] or the session ends — + /// so a crashed maintenance run releases it automatically. Returns + /// whether the lock was acquired; `false` means another session (a + /// concurrent dispatcher's run) holds it. + /// + /// # Errors + /// + /// Returns [`Error::RefStore`] if the query fails. + pub fn try_maintenance_lock(&self) -> Result<bool> { + self.runtime + .block_on(async { + let client = self.client.lock().await; + client + .query_one( + "SELECT pg_try_advisory_lock(hashtextextended($1, 0))", + &[&self.repo_id], + ) + .await + }) + .map_err(pg_err) + .and_then(|row| row.try_get::<_, bool>(0).map_err(pg_err)) + } + + /// Release the maintenance advisory lock this store's session holds + /// (see [`Self::try_maintenance_lock`]). Returns whether a lock was + /// actually released — `false` means this session did not hold it. + /// + /// # Errors + /// + /// Returns [`Error::RefStore`] if the query fails. + pub fn unlock_maintenance(&self) -> Result<bool> { + self.runtime + .block_on(async { + let client = self.client.lock().await; + client + .query_one( + "SELECT pg_advisory_unlock(hashtextextended($1, 0))", + &[&self.repo_id], + ) + .await + }) + .map_err(pg_err) + .and_then(|row| row.try_get::<_, bool>(0).map_err(pg_err)) + } + /// Record one accepted push's op record OID (`docs/scale-out.adoc`, /// "Attested push": "Push ID = op record OID, uniformly"). ///

crates/refstore-postgres/tests/conformance.rs @@ -200,3 +200,35 @@ assert_eq!(claimed_one.payload, "payload-a"); store.complete_effect(id).expect("complete"); } + +#[test] +fn maintenance_advisory_lock_serializes_per_repo_across_sessions() { + let pg = require_postgres!("maintenance_advisory_lock_serializes_per_repo_across_sessions"); + let repo_id = format!("maintenance-{}", uuid::Uuid::new_v4()); + + // Two sessions (two connections) contending for one repo's + // maintenance lock (`docs/scale-out.adoc`, WS9: "Per-repo background + // effects serialized by advisory lock"): the second skips while the + // first holds it. + let holder = PostgresRefStore::connect(pg.url(), repo_id.clone()).expect("connect holder"); + let contender = + PostgresRefStore::connect(pg.url(), repo_id.clone()).expect("connect contender"); + assert!(holder.try_maintenance_lock().expect("holder acquires")); + assert!( + !contender.try_maintenance_lock().expect("contender tries"), + "a concurrent run must skip while the repo's lock is held" + ); + + // The lock is keyed by repo: a different repository is unaffected. + let other_repo = format!("maintenance-{}", uuid::Uuid::new_v4()); + let other = PostgresRefStore::connect(pg.url(), other_repo).expect("connect other repo"); + assert!(other.try_maintenance_lock().expect("other repo acquires")); + + // Release: the contender proceeds. + assert!(holder.unlock_maintenance().expect("holder releases")); + assert!( + contender + .try_maintenance_lock() + .expect("contender acquires after release") + ); +}

crates/git-maintenance/Cargo.toml @@ -1,0 +1,30 @@ +[package] +name = "git-maintenance" +version = "0.0.0" +edition.workspace = true +publish.workspace = true +license.workspace = true + +[dependencies] +backend-conformance = { workspace = true } +effect-dispatcher = { workspace = true } +git-backend = { workspace = true } +git-protocol = { workspace = true } +git-reachability = { workspace = true } +gix-hash = { workspace = true } +gix-object = { workspace = true } +gix-pack = { workspace = true } +odb-files = { workspace = true } +odb-tigris = { workspace = true } +refstore-files = { workspace = true } +refstore-postgres = { workspace = true } +uuid = { workspace = true } + +[dev-dependencies] +git-store = { workspace = true, features = ["test-support"] } +tempfile = { workspace = true } +tokio = { workspace = true } +tokio-postgres = { workspace = true } + +[lints] +workspace = true

crates/git-maintenance/src/cache.rs @@ -1,0 +1,303 @@ +//! Cache-namespace maintenance (`docs/scale-out.adoc`, rule 4 and WS9): +//! TTL eviction of cache refs, and the consolidation effect — the only +//! multi-ref cache writer, and load-bearing rather than hygiene: without +//! it the per-key writer discipline (one ref per key, no CAS contention +//! between concurrent workers) would grow the ref namespace without bound. +//! +//! # Eviction +//! +//! Rule 4: "Eviction = ref deletion + registry delete." [`evict_expired`] +//! is the ref-deletion half; the registry delete follows structurally from +//! the pack-lifetime rule (rule 5): cache objects live in their own packs, +//! so once their refs are gone the next [`crate::gc::collect`] finds those +//! packs fully unreachable and deletes them whole — registry delete plus +//! blob delete, never repack surgery. +//! +//! # Consolidation +//! +//! [`consolidate`] compacts `refs/cache/<namespace>/<key>` per-key refs +//! into one tree under [`git_backend::cache_ns::consolidated_ref`] and +//! deletes the per-key refs, all in **one atomic multi-ref transaction** +//! (the `RefStore` contract makes that contractual, not best-effort). The +//! new tree objects are staged and promoted *before* the transaction — +//! promoted-but-unreferenced objects are invisible to reachability until +//! the ref commit (rule 2), and this ordering means a failure at any point +//! leaves every key resolvable: before the transaction the per-key refs +//! still stand; after it the consolidated tree answers. There is no state +//! in between (see [`git_backend::cache_ns::resolve`], the shared read +//! path). + +use std::collections::BTreeMap; +use std::time::Duration; + +use git_backend::cache_ns; +use git_backend::{Expected, ObjectStore, PackStream, RefEdit, RefName, RefStore, TxOutcome}; +use gix_hash::ObjectId; +use gix_object::WriteTo as _; + +use crate::{Error, Result}; + +/// Delete every cache ref (both [`cache_ns::CACHE_PREFIXES`] namespaces) +/// whose latest reflog entry is older than `ttl` as of `now_secs` (seconds +/// since the epoch — injected rather than read from a clock so callers and +/// tests share one notion of "now"). Returns the refs evicted. +/// +/// Each eviction is its own single-ref compare-and-swap transaction: a ref +/// a concurrent writer moves between read and delete is simply skipped +/// (the write refreshed it). A ref with no reflog entry is kept — with no +/// timestamp there is no expiry to assert, and keeping a cache entry is +/// always safe (rule 4: reconstructible, evictable *later*). +/// +/// Plain `RefStore` transactions, not attested pushes: cache namespaces +/// are exempt from provenance (rule 4). +/// +/// # Errors +/// +/// Returns an error if the ref store fails; never because a CAS lost a +/// race. +pub fn evict_expired(refs: &dyn RefStore, ttl: Duration, now_secs: u64) -> Result<Vec<RefName>> { + let mut evicted = Vec::new(); + for prefix in cache_ns::CACHE_PREFIXES { + let entries: Vec<(RefName, ObjectId)> = refs + .iter_prefix(&RefName::new(prefix))? + .collect::<git_backend::Result<_>>()?; + for (name, oid) in entries { + let Some(written_secs) = latest_write_secs(refs, &name)? else { + continue; + }; + if now_secs.saturating_sub(written_secs) <= ttl.as_secs() { + continue; + } + let outcome = refs.transaction(&[RefEdit { + name: name.clone(), + expected: Expected::MustExistAndMatch(oid), + new: None, + }])?; + if matches!(outcome, TxOutcome::Applied) { + evicted.push(name); + } + } + } + Ok(evicted) +} + +/// The epoch seconds of `name`'s most recent reflog entry, or `None` if it +/// has no reflog. +fn latest_write_secs(refs: &dyn RefStore, name: &RefName) -> Result<Option<u64>> { + match refs.log(name)?.next() { + Some(entry) => Ok(Some(entry?.seconds)), + None => Ok(None), + } +} + +/// A prepared consolidation: the tree already staged and promoted, and the +/// one multi-ref transaction that publishes it. Split from +/// [`consolidate`] so the atomicity boundary is testable: a failure after +/// [`prepare_consolidation`] but before [`commit_consolidation`] must +/// leave every key resolvable through its per-key ref. +#[derive(Debug)] +pub struct ConsolidationPlan { + /// The consolidated tree's root, already promoted (visible to reads, + /// unreachable until the transaction commits — rule 2). + pub tree: ObjectId, + /// The atomic edit batch: publish the consolidated ref, delete every + /// per-key ref, all-or-nothing. + pub edits: Vec<RefEdit>, + /// How many keys this plan consolidates. + pub keys: usize, +} + +/// Build `namespace`'s consolidation: read every per-key ref, merge with +/// the existing consolidated tree (per-key wins — it is always current, +/// see [`cache_ns::resolve`]), write the merged tree's objects through the +/// staged-pack path (no `write_loose` exists — rule 2's staging applies to +/// maintenance too), promote them, and return the transaction that +/// publishes the result. `None` when there are no per-key refs to compact. +/// +/// # Errors +/// +/// Returns an error if any store operation fails, or if two keys collide +/// as blob-vs-directory in the tree (e.g. keys `a` and `a/b`) — a +/// namespace whose writer permits that cannot be consolidated into a tree. +pub fn prepare_consolidation( + namespace: &str, + refs: &dyn RefStore, + objects: &dyn ObjectStore, +) -> Result<Option<ConsolidationPlan>> { + let prefix = cache_ns::per_key_prefix(namespace); + let per_key: Vec<(RefName, ObjectId)> = refs + .iter_prefix(&prefix)? + .collect::<git_backend::Result<_>>()?; + if per_key.is_empty() { + return Ok(None); + } + + let consolidated = cache_ns::consolidated_ref(namespace); + let existing = refs.get(&consolidated)?; + + let mut root = match existing { + Some(tree) => read_tree_node(objects, tree)?, + None => Node::Dir(BTreeMap::new()), + }; + for (name, oid) in &per_key { + let key = name + .as_str() + .strip_prefix(prefix.as_str()) + .ok_or_else(|| Error::RefStore(format!("{name} is not under {prefix}")))?; + insert_key(&mut root, key, *oid)?; + } + + let mut new_objects = Vec::new(); + let tree = write_node(&root, &mut new_objects)?; + let pack = git_protocol::pack::build_pack(&new_objects) + .map_err(|error| Error::ObjectStore(error.to_string()))?; + let quarantine = objects.stage_pack(PackStream::new(std::io::Cursor::new(pack)))?; + objects.promote(quarantine)?; + + let mut edits = vec![RefEdit { + name: consolidated, + expected: match existing { + Some(old) => Expected::MustExistAndMatch(old), + None => Expected::MustNotExist, + }, + new: Some(tree), + }]; + let keys = per_key.len(); + edits.extend(per_key.into_iter().map(|(name, oid)| RefEdit { + name, + expected: Expected::MustExistAndMatch(oid), + new: None, + })); + + Ok(Some(ConsolidationPlan { tree, edits, keys })) +} + +/// Apply a [`ConsolidationPlan`]'s transaction. `Ok(true)` when it +/// applied; `Ok(false)` when a concurrent writer moved any touched ref and +/// the whole batch was rejected — nothing changed (all-or-nothing), the +/// next maintenance run re-prepares against the new state. +/// +/// # Errors +/// +/// Returns an error only if the ref store itself fails. +pub fn commit_consolidation(refs: &dyn RefStore, plan: &ConsolidationPlan) -> Result<bool> { + Ok(matches!(refs.transaction(&plan.edits)?, TxOutcome::Applied)) +} + +/// The consolidation effect (`docs/scale-out.adoc`, rule 4): compact +/// `namespace`'s per-key cache refs into its consolidated tree ref in one +/// atomic multi-ref transaction. Returns how many keys were consolidated — +/// `0` when there was nothing to do or a concurrent writer won the race. +/// +/// # Errors +/// +/// See [`prepare_consolidation`] and [`commit_consolidation`]. +pub fn consolidate( + namespace: &str, + refs: &dyn RefStore, + objects: &dyn ObjectStore, +) -> Result<usize> { + let Some(plan) = prepare_consolidation(namespace, refs, objects)? else { + return Ok(0); + }; + if commit_consolidation(refs, &plan)? { + Ok(plan.keys) + } else { + Ok(0) + } +} + +/// An in-memory consolidated tree under construction: cache blobs at the +/// leaves, directories per key path segment. +enum Node { + Leaf(ObjectId), + Dir(BTreeMap<String, Node>), +} + +/// Insert `key` (slash-separated path) pointing at `oid` into `root`, +/// failing on a blob-vs-directory collision rather than silently dropping +/// either side. +fn insert_key(root: &mut Node, key: &str, oid: ObjectId) -> Result<()> { + let mut node = root; + let mut segments = key.split('/').peekable(); + while let Some(segment) = segments.next() { + let Node::Dir(children) = node else { + return Err(Error::ObjectStore(format!( + "cache key {key} collides with another key at segment {segment}" + ))); + }; + if segments.peek().is_none() { + children.insert(segment.to_owned(), Node::Leaf(oid)); + return Ok(()); + } + node = children + .entry(segment.to_owned()) + .or_insert_with(|| Node::Dir(BTreeMap::new())); + } + Ok(()) +} + +/// Read an existing consolidated tree back into a [`Node`]: tree entries +/// recurse, everything else is a leaf. +fn read_tree_node(objects: &dyn ObjectStore, tree: ObjectId) -> Result<Node> { + let object = objects.read(tree)?; + if object.kind != gix_object::Kind::Tree { + return Err(Error::ObjectStore(format!( + "consolidated ref points at a {:?}, not a tree", + object.kind + ))); + } + let parsed = gix_object::TreeRef::from_bytes(&object.data, gix_hash::Kind::Sha1) + .map_err(|error| Error::ObjectStore(format!("malformed consolidated tree: {error}")))?; + let mut children = BTreeMap::new(); + for entry in parsed.entries { + let name = std::str::from_utf8(entry.filename) + .map_err(|_error| { + Error::ObjectStore("consolidated tree entry name is not UTF-8".to_owned()) + })? + .to_owned(); + let child = if entry.mode.is_tree() { + read_tree_node(objects, entry.oid.to_owned())? + } else { + Node::Leaf(entry.oid.to_owned()) + }; + children.insert(name, child); + } + Ok(Node::Dir(children)) +} + +/// Write `node` (and every subtree) as tree objects, appending each new +/// tree to `out` for packing, returning `node`'s id. Leaves are recorded +/// as plain blobs — cache values are content blobs, their bytes already in +/// the store. +fn write_node(node: &Node, out: &mut Vec<git_protocol::pack::PackObject>) -> Result<ObjectId> { + match node { + Node::Leaf(oid) => Ok(*oid), + Node::Dir(children) => { + let mut entries = Vec::with_capacity(children.len()); + for (name, child) in children { + let oid = write_node(child, out)?; + entries.push(gix_object::tree::Entry { + mode: match child { + Node::Leaf(_oid) => gix_object::tree::EntryKind::Blob.into(), + Node::Dir(_children) => gix_object::tree::EntryKind::Tree.into(), + }, + filename: name.as_str().into(), + oid, + }); + } + entries.sort(); + let tree = gix_object::Tree { entries }; + let mut data = Vec::new(); + tree.write_to(&mut data)?; + let oid = gix_object::compute_hash(gix_hash::Kind::Sha1, gix_object::Kind::Tree, &data) + .map_err(|error| Error::ObjectStore(error.to_string()))?; + out.push(git_protocol::pack::PackObject { + id: oid, + kind: gix_object::Kind::Tree, + data, + }); + Ok(oid) + } + } +}

crates/git-maintenance/src/collector.rs @@ -1,0 +1,110 @@ +//! Real [`backend_conformance::Collector`]s — the seam WS2 left open, now +//! closed: the causal-collection-safety property (`docs/scale-out.adoc`, +//! correctness rule 1) runs against collection passes that actually +//! collect, instead of [`backend_conformance::NoopCollector`]. +//! +//! Both collectors panic if a collection pass errors: they exist to drive +//! a conformance property, and a collector that swallowed its own failure +//! would let the property pass vacuously ("no fake assertions"). + +use std::time::Duration; + +use backend_conformance::Collector; +use git_backend::RefStore; +use odb_tigris::OdbTigris; +use odb_tigris::registry::PackRegistry; +use odb_tigris::transport::BlobTransport; + +/// A [`Collector`] over one [`OdbTigris`] store: one collection pass = +/// expire staging sessions past their grace window (the grace-based cruft +/// arm of rule 1, a no-op for a store without one), then a full +/// mark-and-sweep ([`crate::gc::collect`]). +pub struct TigrisCollector<'a, T, R> { + repo_id: &'a str, + refs: &'a dyn RefStore, + store: &'a OdbTigris<T, R>, + transport: &'a dyn BlobTransport, + registry: &'a dyn PackRegistry, +} + +impl<'a, T, R> TigrisCollector<'a, T, R> +where + T: BlobTransport, + R: PackRegistry, +{ + /// A collector over `store`, marking from `refs` and sweeping via + /// `registry`/`transport` — the same transport and registry `store` + /// itself was built over. + #[must_use] + pub fn new( + repo_id: &'a str, + refs: &'a dyn RefStore, + store: &'a OdbTigris<T, R>, + transport: &'a dyn BlobTransport, + registry: &'a dyn PackRegistry, + ) -> Self { + Self { + repo_id, + refs, + store, + transport, + registry, + } + } +} + +impl<T, R> Collector for TigrisCollector<'_, T, R> +where + T: BlobTransport, + R: PackRegistry, +{ + #[expect( + clippy::expect_used, + reason = "conformance driver: a failed collection pass must fail the \ + property loudly, never let it pass vacuously" + )] + fn collect(&self) { + self.store + .expire_stale_quarantines() + .expect("expire stale quarantines"); + crate::gc::collect( + self.repo_id, + self.refs, + self.store, + self.transport, + self.registry, + ) + .expect("mark-and-sweep collection pass"); + } + + fn staging_grace(&self) -> Option<Duration> { + self.store.staging_grace() + } +} + +/// A [`Collector`] over a local bare repository (`refstore-files` + +/// `odb-files`): one collection pass = [`crate::gc::collect_files`]. No +/// grace window ([`Collector::staging_grace`] stays `None`) — the local +/// backend bounds staging by promotion alone, never by a clock. +pub struct FilesCollector { + repo: std::path::PathBuf, +} + +impl FilesCollector { + /// A collector over the bare repository at `repo`. + #[must_use] + pub fn new(repo: impl Into<std::path::PathBuf>) -> Self { + Self { repo: repo.into() } + } +} + +impl Collector for FilesCollector { + #[expect( + clippy::expect_used, + reason = "conformance driver: a failed collection pass must fail the \ + property loudly, never let it pass vacuously" + )] + fn collect(&self) { + crate::gc::collect_files(&self.repo).expect("files mark-and-sweep collection pass"); + } +}

crates/git-maintenance/src/gc.rs @@ -1,0 +1,313 @@ +//! Mark-and-sweep GC (`docs/scale-out.adoc`, WS9: "Mark from RefStore via +//! reachability artifacts; sweep via pack registry"). +//! +//! # Mark +//! +//! [`collect`] marks with [`git_reachability::gc_mark`] — every object +//! reachable from every current ref tip, accelerated by whatever +//! reachability artifacts the repo has (absence degrades speed, never +//! answers). A second, durable-tips-only walk splits the marked set into +//! lifetime classes ([`odb_tigris::pack_writer::LifetimeClass`]) so the +//! sweep's repack path can honor the pack-lifetime rule (rule 5). +//! +//! # Sweep — and why quarantine is structurally safe +//! +//! The sweep enumerates [`odb_tigris::registry::PackRegistry::list`] and +//! nothing else. Quarantined (staged) packs are *not in the registry* — +//! [`odb_tigris::OdbTigris::promote`] is what records a pack, and it is +//! only called after the ref transaction the pack was staged for commits — +//! and [`odb_tigris::transport::BlobTransport`] exposes no listing call at +//! all, so there is no API through which this module *could* scan +//! quarantine (correctness rules 1 and 2: "GC never scans quarantine"). +//! That safety is structural, not a filter this code must remember to +//! apply. +//! +//! Grace-based staging (rule 1's time-bounded arm) lives on the store +//! itself: [`odb_tigris::OdbTigris::with_staging_grace`] bounds staging +//! sessions (a session past its window aborts at `promote` rather than +//! becoming collectible mid-flight), and +//! [`odb_tigris::OdbTigris::expire_stale_quarantines`] is the cruft pass a +//! grace-based collector runs — see [`crate::collector::TigrisCollector`]. +//! +//! # Sweep outcomes per pack +//! +//! - every object unreachable → **delete**: registry delete (the commit +//! point), then best-effort blob deletes. +//! - every object reachable → keep. +//! - mixed, with at least one durable reachable object → **rewrite**: +//! reachable objects are repacked through the WS5 pack writer +//! ([`odb_tigris::pack_writer::partition_and_pack`], which partitions by +//! lifetime class so cache and durable objects never share the new +//! pack), the new pack(s) are recorded, and only then is the old pack +//! deleted — no window where a live object is unregistered. +//! - mixed, all reachable objects cache-lifetime → left whole: cache packs +//! die by registry delete when their refs are evicted, never repack +//! surgery (rule 5). + +use std::collections::BTreeSet; + +use git_backend::cache_ns; +use git_backend::{ObjectStore, RefName, RefStore}; +use git_reachability::walk::StoreSource; +use gix_hash::ObjectId; +use odb_tigris::pack_writer::{ClassifiedObject, LifetimeClass, index_pack, partition_and_pack}; +use odb_tigris::registry::{PackId, PackRecord, PackRegistry}; +use odb_tigris::transport::BlobTransport; + +use crate::{Error, Result}; + +/// What one [`collect`] pass did. +#[derive(Debug, Default, Clone, PartialEq, Eq)] +pub struct GcOutcome { + /// Packs whose objects were all unreachable, deleted whole. + pub deleted_packs: usize, + /// Mixed packs rewritten to contain only their reachable objects. + pub rewritten_packs: usize, + /// The size of the marked (reachable) set. + pub live_objects: usize, +} + +/// One full mark-and-sweep pass for `repo_id` (see the module docs for the +/// mark/sweep design and the structural quarantine-safety argument). +/// +/// # Errors +/// +/// Returns an error if the mark walk fails (a ref tip whose history is +/// incomplete is corruption, never grounds to collect), or if a registry +/// or transport operation the sweep depends on fails. +pub fn collect( + repo_id: &str, + refs: &dyn RefStore, + objects: &dyn ObjectStore, + transport: &dyn BlobTransport, + registry: &dyn PackRegistry, +) -> Result<GcOutcome> { + let artifacts = git_reachability::store::load_bundle(transport, registry, repo_id) + .map_err(|error| Error::ObjectStore(error.to_string()))?; + let marked = git_reachability::gc_mark(refs, objects, &artifacts) + .map_err(|error| Error::ObjectStore(error.to_string()))?; + let durable = durable_reachable(refs, objects, &artifacts)?; + + let mut outcome = GcOutcome { + live_objects: marked.len(), + ..GcOutcome::default() + }; + + for record in registry.list(repo_id)? { + let ids = pack_object_ids(transport, &record)?; + let live: Vec<ObjectId> = ids + .iter() + .filter(|id| marked.contains(*id)) + .copied() + .collect(); + + if live.is_empty() { + // Registry delete first — it is the commit point; blob deletes + // after it are best-effort cleanup (a leaked key wastes space, + // never correctness), mirroring `OdbTigris::promote`. + registry.delete(repo_id, &record.id)?; + let _ignored = transport.delete(&record.pack_key); + let _ignored = transport.delete(&record.idx_key); + outcome.deleted_packs = outcome.deleted_packs.saturating_add(1); + continue; + } + if live.len() == ids.len() { + continue; + } + + // Mixed pack. A pack whose reachable objects are all + // cache-lifetime is a cache pack: never repack surgery (rule 5) — + // it dies whole once its cache refs are evicted. + if live.iter().all(|id| !durable.contains(id)) { + continue; + } + + rewrite_pack( + repo_id, objects, transport, registry, &record, &live, &durable, + )?; + outcome.rewritten_packs = outcome.rewritten_packs.saturating_add(1); + } + Ok(outcome) +} + +/// The objects reachable from durable (non-cache) ref tips alone — the +/// lifetime-class oracle for repack partitioning: marked objects in this +/// set are [`LifetimeClass::Durable`], marked objects outside it are +/// reachable only through cache refs and so [`LifetimeClass::Cache`]. +fn durable_reachable( + refs: &dyn RefStore, + objects: &dyn ObjectStore, + artifacts: &git_reachability::ArtifactBundle, +) -> Result<BTreeSet<ObjectId>> { + let tips = refs + .iter_prefix(&RefName::new("refs/"))? + .filter(|entry| match entry { + Ok((name, _oid)) => !cache_ns::is_cache_ref(name), + Err(_error) => true, + }) + .map(|entry| entry.map(|(_name, oid)| oid)) + .collect::<git_backend::Result<Vec<ObjectId>>>()?; + let source = StoreSource::new(objects); + git_reachability::accelerated_reachable(tips, &source, |_id| false, false, artifacts) + .map_err(|error| Error::ObjectStore(error.to_string())) +} + +/// Every object id in `record`'s pack, read from its `.idx` — never from a +/// bucket listing (the transport has none to offer). +fn pack_object_ids(transport: &dyn BlobTransport, record: &PackRecord) -> Result<Vec<ObjectId>> { + let bytes = transport.get(&record.idx_key)?; + let idx = gix_pack::index::File::from_data( + bytes, + std::path::PathBuf::from(&record.idx_key), + gix_hash::Kind::Sha1, + ) + .map_err(|error| Error::ObjectStore(format!("parsing index {}: {error}", record.idx_key)))?; + Ok(idx.iter().map(|entry| entry.oid).collect()) +} + +/// Repack `live` (the reachable objects of a mixed pack) into fresh +/// pack(s) through the WS5 pack writer — partitioned by lifetime class, so +/// the pack-lifetime rule survives the rewrite — record them, and only +/// then delete the old pack. +fn rewrite_pack( + repo_id: &str, + objects: &dyn ObjectStore, + transport: &dyn BlobTransport, + registry: &dyn PackRegistry, + record: &PackRecord, + live: &[ObjectId], + durable: &BTreeSet<ObjectId>, +) -> Result<()> { + let classified: Vec<ClassifiedObject> = live + .iter() + .map(|id| { + let object = objects.read(*id)?; + Ok(ClassifiedObject { + id: *id, + kind: object.kind, + data: object.data, + lifetime: if durable.contains(id) { + LifetimeClass::Durable + } else { + LifetimeClass::Cache + }, + }) + }) + .collect::<Result<_>>()?; + + let packs = partition_and_pack(classified)?; + for pack_bytes in [packs.durable, packs.cache].into_iter().flatten() { + record_new_pack(repo_id, transport, registry, pack_bytes)?; + } + + registry.delete(repo_id, &record.id)?; + let _ignored = transport.delete(&record.pack_key); + let _ignored = transport.delete(&record.idx_key); + Ok(()) +} + +/// Index freshly written pack bytes, upload them at new live keys, and +/// record them — the same key layout `OdbTigris` promotes into. +fn record_new_pack( + repo_id: &str, + transport: &dyn BlobTransport, + registry: &dyn PackRegistry, + pack_bytes: Vec<u8>, +) -> Result<()> { + let (pack, idx) = index_pack(pack_bytes)?; + let object_count = count_pack_objects(&idx); + let id = uuid::Uuid::new_v4().to_string(); + let pack_key = format!("{repo_id}/live/{id}.pack"); + let idx_key = format!("{repo_id}/live/{id}.idx"); + transport.put(&pack_key, pack)?; + transport.put(&idx_key, idx)?; + registry.record(PackRecord { + id: PackId::new(id), + repo_id: repo_id.to_owned(), + pack_key, + idx_key, + object_count, + }) +} + +/// The object count out of freshly written `.idx` bytes, or `None` if they +/// fail to parse — the count is informational only ([`PackRecord`]'s field +/// docs), so an unparsable count is not worth failing a rewrite over. +fn count_pack_objects(idx_bytes: &[u8]) -> Option<u64> { + gix_pack::index::File::from_data( + idx_bytes.to_vec(), + std::path::PathBuf::from("rewrite.idx"), + gix_hash::Kind::Sha1, + ) + .ok() + .map(|idx| u64::from(idx.num_objects())) +} + +/// What one [`collect_files`] pass did. +#[derive(Debug, Default, Clone, PartialEq, Eq)] +pub struct FilesGcOutcome { + /// Packs under `objects/pack/` whose objects were all unreachable, + /// deleted whole. + pub deleted_packs: usize, + /// The size of the marked (reachable) set. + pub live_objects: usize, +} + +/// Mark-and-sweep for the local files backend (`refstore-files` + +/// `odb-files`): mark from ref tips, then delete every pack under +/// `objects/pack/` whose objects are all unreachable. Whole-pack reaping +/// only — the local backend has no pack registry to rewrite through, and +/// mixed packs are simply kept (correct, just less compact; the +/// registry-backed sweep is where rewriting lives). +/// +/// Structurally quarantine-safe for the same reason `odb-files` itself is: +/// this sweep scans `objects/pack/` and nothing else, and staged packs +/// live under `objects/quarantine/<id>/` until promoted. +/// +/// # Errors +/// +/// Returns an error if the repository cannot be opened, the mark walk +/// fails, or a doomed pack cannot be deleted. +pub fn collect_files(repo: &std::path::Path) -> Result<FilesGcOutcome> { + let (marked, doomed) = { + let refs = refstore_files::FilesRefStore::open(repo)?; + let objects = odb_files::OdbFiles::open(repo)?; + let marked = + git_reachability::gc_mark(&refs, &objects, &git_reachability::ArtifactBundle::empty()) + .map_err(|error| Error::ObjectStore(error.to_string()))?; + + let pack_dir = repo.join("objects").join("pack"); + let mut doomed = Vec::new(); + if pack_dir.is_dir() { + for entry in std::fs::read_dir(&pack_dir)? { + let path = entry?.path(); + if path.extension().is_some_and(|ext| ext == "idx") { + let idx = gix_pack::index::File::at(&path, gix_hash::Kind::Sha1).map_err( + |error| { + Error::ObjectStore(format!("parsing index {}: {error}", path.display())) + }, + )?; + if idx.iter().all(|entry| !marked.contains(&entry.oid)) { + doomed.push(path); + } + } + } + } + (marked, doomed) + // `objects` (and its pack mmaps) drop here, before any deletion. + }; + + let mut outcome = FilesGcOutcome { + live_objects: marked.len(), + ..FilesGcOutcome::default() + }; + for idx_path in doomed { + std::fs::remove_file(&idx_path)?; + let pack_path = idx_path.with_extension("pack"); + if pack_path.exists() { + std::fs::remove_file(&pack_path)?; + } + outcome.deleted_packs = outcome.deleted_packs.saturating_add(1); + } + Ok(outcome) +}

crates/git-maintenance/src/lib.rs @@ -1,0 +1,38 @@ +//! WS9: GC, compaction, and maintenance (`docs/scale-out.adoc`, "WS9 — +//! GC, compaction, maintenance"). +//! +//! > Per-repo background effects serialized by advisory lock. Mark from +//! > RefStore via reachability artifacts; sweep via pack registry; cruft +//! > semantics where grace-based. Cache-ref TTL deletion; the +//! > consolidation effect from rule 4 lives here and is load-bearing. +//! > Reachability-artifact regeneration scheduled here. +//! +//! The pieces, one module each: +//! +//! - [`gc`] — mark ([`git_reachability::gc_mark`], artifacts as +//! accelerator) and sweep (over the pack registry, never a bucket +//! listing and *structurally* never quarantine — see the module doc). +//! - [`cache`] — TTL eviction of cache refs and the consolidation effect, +//! the only multi-ref cache writer (`docs/scale-out.adoc`, rule 4). +//! - [`lock`] — the per-repo advisory lock every maintenance run holds for +//! its whole duration, so concurrent dispatchers can't double-run a +//! repo: a Postgres advisory lock in cloud deployments, a file lock +//! locally. +//! - [`schedule`] — the maintenance [`git_backend::EffectDef`]s and the +//! [`schedule::Scheduler`] a server calls post-ingest to enqueue them on +//! ref-update volume thresholds (including reachability regeneration via +//! [`git_reachability::maintenance::should_regenerate`], the trigger WS6 +//! left for this crate to schedule). +//! - [`collector`] — real [`backend_conformance::Collector`]s over the +//! files and Tigris backends, closing the seam WS2 left open: the +//! causal-collection-safety property now runs against a collection pass +//! that actually collects, including the staging-timeout boundary +//! (`docs/scale-out.adoc`, correctness rule 1). + +pub mod cache; +pub mod collector; +pub mod gc; +pub mod lock; +pub mod schedule; + +pub use git_backend::{Error, Result};

crates/git-maintenance/src/lock.rs @@ -1,0 +1,140 @@ +//! Per-repo maintenance serialization (`docs/scale-out.adoc`, WS9: +//! "Per-repo background effects serialized by advisory lock"). +//! +//! [`run_exclusive`] wraps a *whole* maintenance run in one +//! [`MaintenanceLock`] acquisition, so two dispatchers (or a dispatcher +//! and an operator's manual run) can never double-run one repository: the +//! second acquirer skips — maintenance is periodic and idempotent, so +//! "skip and let the next trigger retry" beats blocking a dispatcher +//! thread on another machine's run. +//! +//! Two implementations, one per deployment shape: +//! - [`FileMaintenanceLock`] — an OS advisory file lock (`flock`-style, +//! via `std::fs::File::try_lock`) beside the local bare repository. +//! - [`PgMaintenanceLock`] — a Postgres session advisory lock keyed by +//! repo id, for Postgres-backed deployments where the contending +//! dispatchers are on different machines (see +//! `refstore_postgres::PostgresRefStore::try_maintenance_lock`). + +use std::path::{Path, PathBuf}; + +use crate::Result; + +/// Holds a per-repo maintenance lock; released on drop. The release action +/// is captured as a closure so file and Postgres guards share one type. +pub struct MaintenanceGuard<'a> { + release: Option<Box<dyn FnOnce() + 'a>>, +} + +impl Drop for MaintenanceGuard<'_> { + fn drop(&mut self) { + if let Some(release) = self.release.take() { + release(); + } + } +} + +/// A per-repo advisory lock a maintenance run holds for its whole +/// duration. +pub trait MaintenanceLock { + /// Try to take the lock: `Some(guard)` when this caller now holds it, + /// `None` when another maintenance run does (the caller should skip). + /// + /// # Errors + /// + /// Returns an error if the locking mechanism itself fails — never for + /// mere contention, which is the `None` case. + fn try_acquire(&self) -> Result<Option<MaintenanceGuard<'_>>>; +} + +/// Run `work` under `lock`, holding it for the whole run. `Ok(None)` means +/// another run holds the lock and this one was skipped. +/// +/// # Errors +/// +/// Returns an error if acquiring fails or `work` fails. +pub fn run_exclusive<T>( + lock: &dyn MaintenanceLock, + work: impl FnOnce() -> Result<T>, +) -> Result<Option<T>> { + let Some(guard) = lock.try_acquire()? else { + return Ok(None); + }; + let outcome = work()?; + drop(guard); + Ok(Some(outcome)) +} + +/// [`MaintenanceLock`] over an OS advisory file lock — the local +/// deployment's serializer, correct across processes on one machine. +pub struct FileMaintenanceLock { + path: PathBuf, +} + +impl FileMaintenanceLock { + /// A lock at `path` (created if absent; its content is never read). + #[must_use] + pub fn new(path: impl Into<PathBuf>) -> Self { + Self { path: path.into() } + } + + /// The conventional lock for the bare repository at `repo`: + /// `<repo>/maintenance.lock`. + #[must_use] + pub fn for_repo(repo: &Path) -> Self { + Self::new(repo.join("maintenance.lock")) + } +} + +impl MaintenanceLock for FileMaintenanceLock { + fn try_acquire(&self) -> Result<Option<MaintenanceGuard<'_>>> { + let file = std::fs::OpenOptions::new() + .create(true) + .truncate(false) + .write(true) + .open(&self.path)?; + match file.try_lock() { + // Dropping the file both unlocks and closes it. + Ok(()) => Ok(Some(MaintenanceGuard { + release: Some(Box::new(move || drop(file))), + })), + Err(std::fs::TryLockError::WouldBlock) => Ok(None), + Err(std::fs::TryLockError::Error(error)) => Err(error.into()), + } + } +} + +/// [`MaintenanceLock`] over a Postgres session advisory lock keyed by the +/// store's repo id — the cloud deployment's serializer, correct across +/// machines because the lock lives in the one Postgres primary every +/// dispatcher already talks to. +pub struct PgMaintenanceLock<'a> { + store: &'a refstore_postgres::PostgresRefStore, +} + +impl<'a> PgMaintenanceLock<'a> { + /// Lock through `store`'s connection (session advisory locks are held + /// by the session — this store's connection — and released on + /// [`MaintenanceGuard`] drop or session death, so a crashed + /// maintenance run never wedges the repo). + #[must_use] + pub fn new(store: &'a refstore_postgres::PostgresRefStore) -> Self { + Self { store } + } +} + +impl MaintenanceLock for PgMaintenanceLock<'_> { + fn try_acquire(&self) -> Result<Option<MaintenanceGuard<'_>>> { + if !self.store.try_maintenance_lock()? { + return Ok(None); + } + let store = self.store; + Ok(Some(MaintenanceGuard { + release: Some(Box::new(move || { + // Session death releases the lock anyway; a failed explicit + // unlock is not worth panicking a Drop over. + let _ignored = store.unlock_maintenance(); + })), + })) + } +}

crates/git-maintenance/src/schedule.rs @@ -1,0 +1,231 @@ +//! Scheduling maintenance (`docs/scale-out.adoc`, WS9 and "Reachability": +//! "Regeneration is scheduled with repack (WS9) and triggered by +//! ref-update volume thresholds"). +//! +//! The maintenance effects — GC, cache TTL, consolidation, and +//! reachability regeneration ([`git_reachability::maintenance`], whose +//! scheduling WS6 explicitly deferred here) — are defined as +//! [`EffectDef`]s and enqueued by [`schedule_maintenance`] whenever a +//! repo's accumulated ref-update volume crosses its threshold. The +//! [`Scheduler`] is the piece a server holds: it does the per-repo +//! accumulation so the ingest path only has to report "this push applied +//! N ref edits" (see `git-ents-server`'s native receive-pack endpoint, the +//! wired call site). +//! +//! Like `reachability-maintenance`, these effects run as in-process +//! maintenance code, so their [`EffectDef::command`] is `None`; the queue +//! rows are the schedule, and the runner executes the bodies +//! ([`crate::gc::collect`], [`crate::cache::evict_expired`], +//! [`crate::cache::consolidate`], +//! [`git_reachability::maintenance::regenerate`]) under the per-repo +//! advisory lock ([`crate::lock`]). + +use std::collections::{BTreeMap, HashMap}; +use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; + +use git_backend::{EffectDef, MaterializedInputs}; + +use crate::Result; + +/// The GC effect's name (mark-and-sweep, [`crate::gc::collect`]). +pub const GC_EFFECT: &str = "maintenance-gc"; + +/// The cache TTL eviction effect's name ([`crate::cache::evict_expired`]). +pub const CACHE_TTL_EFFECT: &str = "maintenance-cache-ttl"; + +/// The cache consolidation effect's name ([`crate::cache::consolidate`]). +pub const CONSOLIDATION_EFFECT: &str = "maintenance-cache-consolidation"; + +/// The static [`EffectDef`] for one in-process maintenance effect — +/// `command`/`image` `None`, mirroring +/// [`git_reachability::maintenance::definition`]. +fn definition(name: &str) -> EffectDef { + EffectDef { + name: name.to_owned(), + command: None, + image: None, + } +} + +/// The ref-update volume thresholds that trigger maintenance. Two knobs +/// because reachability regeneration has its own trigger predicate +/// ([`git_reachability::maintenance::should_regenerate`]) and may +/// reasonably fire less often than repack. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Thresholds { + /// Ref updates before a maintenance run (GC, TTL, consolidation) is + /// enqueued. + pub maintenance: u64, + /// Ref updates before reachability regeneration rides along + /// ([`git_reachability::maintenance::should_regenerate`]). + pub reachability: u64, +} + +impl Default for Thresholds { + fn default() -> Self { + Self { + maintenance: 64, + reachability: 64, + } + } +} + +/// Per-repo maintenance-relevant activity since the last scheduled run. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Stats { + /// Applied ref edits since maintenance was last enqueued for the repo. + pub ref_updates_since_last: u64, +} + +/// Where scheduled maintenance effects land — the effect queue in a +/// Postgres deployment ([`PostgresQueueSink`]), anything else a test or a +/// future local runner supplies. +pub trait MaintenanceSink: Send + Sync { + /// Enqueue `effects` for `repo_id`, in order. + /// + /// # Errors + /// + /// Returns an error if the underlying queue cannot be written. + fn enqueue(&self, repo_id: &str, effects: &[EffectDef]) -> Result<()>; +} + +/// Enqueue `repo_id`'s maintenance effects into `sink` if `stats` crosses +/// `thresholds.maintenance` — GC, cache TTL, consolidation, plus +/// reachability regeneration when +/// [`git_reachability::maintenance::should_regenerate`] says the volume +/// also warrants that. Returns what was enqueued (empty below threshold). +/// +/// # Errors +/// +/// Returns an error if the sink fails; nothing is retried here — the +/// caller's accumulated count survives (see [`Scheduler`]) so the next +/// update re-triggers. +pub fn schedule_maintenance( + repo_id: &str, + stats: &Stats, + thresholds: &Thresholds, + sink: &dyn MaintenanceSink, +) -> Result<Vec<EffectDef>> { + if stats.ref_updates_since_last < thresholds.maintenance { + return Ok(Vec::new()); + } + let mut effects = vec![ + definition(GC_EFFECT), + definition(CACHE_TTL_EFFECT), + definition(CONSOLIDATION_EFFECT), + ]; + if git_reachability::maintenance::should_regenerate( + stats.ref_updates_since_last, + thresholds.reachability, + ) { + effects.push(git_reachability::maintenance::definition()); + } + sink.enqueue(repo_id, &effects)?; + Ok(effects) +} + +/// The per-repo accumulator a server holds: ingest reports applied ref +/// edits through [`Scheduler::note_ref_updates`], and once a repo's count +/// crosses the threshold its maintenance effects are enqueued and the +/// count reset. On a sink failure the count is restored, so a transient +/// queue outage delays maintenance rather than losing the trigger. +pub struct Scheduler { + thresholds: Thresholds, + sink: Arc<dyn MaintenanceSink>, + counts: Mutex<HashMap<String, u64>>, +} + +impl Scheduler { + /// A scheduler enqueuing into `sink` at `thresholds`. + #[must_use] + pub fn new(thresholds: Thresholds, sink: Arc<dyn MaintenanceSink>) -> Self { + Self { + thresholds, + sink, + counts: Mutex::new(HashMap::new()), + } + } + + /// Record that a push applied `updates` ref edits to `repo_id`, + /// enqueuing the repo's maintenance effects if that crosses the + /// threshold. Returns what was enqueued (usually nothing). + /// + /// # Errors + /// + /// Returns an error if the sink fails — the accumulated count is + /// restored first, so the trigger is delayed, not lost. + pub fn note_ref_updates(&self, repo_id: &str, updates: u64) -> Result<Vec<EffectDef>> { + let due = { + let mut counts = lock(&self.counts); + let count = counts.entry(repo_id.to_owned()).or_insert(0); + *count = count.saturating_add(updates); + if *count >= self.thresholds.maintenance { + let accumulated = *count; + *count = 0; + Some(accumulated) + } else { + None + } + }; + let Some(accumulated) = due else { + return Ok(Vec::new()); + }; + let stats = Stats { + ref_updates_since_last: accumulated, + }; + match schedule_maintenance(repo_id, &stats, &self.thresholds, &*self.sink) { + Ok(effects) => Ok(effects), + Err(error) => { + let mut counts = lock(&self.counts); + let count = counts.entry(repo_id.to_owned()).or_insert(0); + *count = count.saturating_add(accumulated); + Err(error) + } + } + } +} + +fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> { + mutex.lock().unwrap_or_else(PoisonError::into_inner) +} + +/// [`MaintenanceSink`] over the Postgres effect queue: each effect is +/// encoded with [`effect_dispatcher::job::encode`] — the payload shape the +/// WS7 dispatcher drains — with a null tree, since maintenance effects run +/// against repository state, not a materialized input tree. +/// +/// Connects per enqueue call: enqueues happen once per threshold crossing, +/// not per push, so a short-lived connection is the simple correct choice +/// over holding one open on the ingest path. +pub struct PostgresQueueSink { + conninfo: String, +} + +impl PostgresQueueSink { + /// A sink enqueuing into the queue at `conninfo` (a libpq connection + /// string). + #[must_use] + pub fn new(conninfo: impl Into<String>) -> Self { + Self { + conninfo: conninfo.into(), + } + } +} + +impl MaintenanceSink for PostgresQueueSink { + fn enqueue(&self, repo_id: &str, effects: &[EffectDef]) -> Result<()> { + let store = refstore_postgres::PostgresRefStore::connect(&self.conninfo, repo_id)?; + for effect in effects { + let payload = effect_dispatcher::job::encode(&effect_dispatcher::job::Job { + effect: effect.clone(), + inputs: MaterializedInputs { + tree: gix_hash::ObjectId::null(gix_hash::Kind::Sha1), + toolchain_paths: BTreeMap::new(), + cache: None, + }, + }); + store.enqueue_effect(&payload)?; + } + Ok(()) + } +}

crates/git-maintenance/tests/cache.rs @@ -1,0 +1,233 @@ +//! Cache maintenance (`docs/scale-out.adoc`, rule 4): TTL eviction +//! (expired evicted, fresh kept, registry row gone) and consolidation +//! atomicity (per-key refs deleted + consolidated ref written in one +//! transaction; reads resolve every key before, during simulated failure, +//! and after). + +#![allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + reason = "test assertions, not application code" +)] + +mod util; + +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use git_backend::{ObjectStore as _, RefName, RefStore as _, cache_ns}; +use git_maintenance::{cache, gc}; +use git_store::test_support::{commit_all, head, repo}; +use odb_tigris::OdbTigris; +use odb_tigris::registry::PackRegistry as _; +use odb_tigris::registry::memory::InMemoryRegistry; +use odb_tigris::transport::fs::FsTransport; + +fn now_secs() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs() +} + +/// TTL eviction end to end, including rule 4's "eviction = ref deletion + +/// registry delete": the expired ref goes, the fresh one stays, and the +/// expired entry's own cache pack (rule 5: cache objects get their own +/// packs) leaves the registry on the following collect — a registry +/// delete, never repack surgery. +#[test] +fn ttl_evicts_expired_cache_refs_and_their_registry_rows() { + let work = repo(); + util::use_main_branch(work.path()); + std::fs::write(work.path().join("file"), "durable").unwrap(); + commit_all(work.path(), "durable"); + let durable = head(work.path()); + + let bucket = tempfile::tempdir().unwrap(); + let transport = FsTransport::open(bucket.path()).unwrap(); + let registry = InMemoryRegistry::new(); + let store = OdbTigris::new(&transport, &registry, "repo"); + util::stage_and_promote(&store, util::pack_for(work.path(), &durable)); + + // Two cache entries, each in its own pack (rule 5), each behind its + // own per-key ref. + let old_blob = util::hash_blob(work.path(), "old-entry"); + let fresh_blob = util::hash_blob(work.path(), "fresh-entry"); + util::stage_and_promote(&store, util::pack_of(work.path(), &[&old_blob])); + util::stage_and_promote(&store, util::pack_of(work.path(), &[&fresh_blob])); + + let refs = refstore_files::FilesRefStore::open(work.path()).unwrap(); + + // Reflog timestamps have one-second granularity, so age the entries + // apart with a real gap wider than the TTL: `old` is written, the TTL + // elapses, then `fresh` is written just before eviction runs. + let ttl = Duration::from_secs(1); + util::set_ref(&refs, "refs/cache/sccache/old", util::oid(&old_blob)); + std::thread::sleep(Duration::from_millis(2200)); + // A fresh store handle: gitoxide snapshots the committer signature + // (and its timestamp) per repository open, so the second write needs + // its own open to get a current reflog time. + let refs = refstore_files::FilesRefStore::open(work.path()).unwrap(); + util::set_ref(&refs, "refs/cache/sccache/fresh", util::oid(&fresh_blob)); + + let evicted = cache::evict_expired(&refs, ttl, now_secs()).unwrap(); + assert_eq!( + evicted, + vec![RefName::new("refs/cache/sccache/old")], + "exactly the expired ref is evicted" + ); + assert!( + refs.get(&RefName::new("refs/cache/sccache/old")) + .unwrap() + .is_none(), + "expired ref deleted" + ); + assert!( + refs.get(&RefName::new("refs/cache/sccache/fresh")) + .unwrap() + .is_some(), + "fresh ref kept" + ); + + // The registry-delete half: after eviction, collect finds the old + // entry's cache pack fully unreachable and deletes it whole. + let before = registry.list("repo").unwrap().len(); + let outcome = gc::collect("repo", &refs, &store, &transport, &registry).unwrap(); + assert_eq!(outcome.deleted_packs, 1, "the evicted entry's own pack"); + assert_eq!(outcome.rewritten_packs, 0, "never repack surgery for cache"); + assert_eq!(registry.list("repo").unwrap().len(), before - 1); + assert!(!store.contains(util::oid(&old_blob)).unwrap()); + assert!(store.contains(util::oid(&fresh_blob)).unwrap()); +} + +/// Consolidation atomicity over the files backend: reads resolve every +/// key before the transaction, after a simulated failure between +/// staging/promotion and the ref commit, and after the commit — and the +/// commit itself deletes every per-key ref and publishes the consolidated +/// tree in one all-or-nothing transaction. +#[test] +fn consolidation_is_atomic_and_reads_resolve_every_key_throughout() { + let work = repo(); + util::use_main_branch(work.path()); + std::fs::write(work.path().join("file"), "seed").unwrap(); + commit_all(work.path(), "seed"); + + let refs = refstore_files::FilesRefStore::open(work.path()).unwrap(); + let objects = odb_files::OdbFiles::open(work.path()).unwrap(); + + let entries = [ + ("aa/bb", "value-1"), + ("cc", "value-2"), + ("dd/ee/ff", "value-3"), + ]; + for (key, value) in entries { + let blob = util::hash_blob(work.path(), value); + util::set_ref( + &refs, + &format!("refs/cache/sccache/{key}"), + util::oid(&blob), + ); + } + let resolve_all = |label: &str| { + for (key, value) in entries { + let oid = cache_ns::resolve(&refs, &objects, "sccache", key) + .unwrap() + .unwrap_or_else(|| panic!("{label}: key {key} must resolve")); + assert_eq!( + objects.read(oid).unwrap().data, + value.as_bytes(), + "{label}: {key}" + ); + } + }; + + // Before. + resolve_all("before consolidation"); + + // During simulated failure: the plan is prepared — tree objects + // staged and promoted — but the process "dies" before the ref + // transaction. Every key still resolves through its per-key ref. + let plan = cache::prepare_consolidation("sccache", &refs, &objects) + .unwrap() + .expect("three per-key refs to consolidate"); + assert_eq!(plan.keys, 3); + resolve_all("after prepare, before commit (simulated failure window)"); + + // Commit: one atomic multi-ref transaction. + assert!(cache::commit_consolidation(&refs, &plan).unwrap()); + + // After: the consolidated ref exists, every per-key ref is gone, and + // every key still resolves — now through the tree. + let consolidated = refs + .get(&cache_ns::consolidated_ref("sccache")) + .unwrap() + .expect("consolidated ref"); + assert_eq!(consolidated, plan.tree); + let leftover: Vec<_> = refs + .iter_prefix(&cache_ns::per_key_prefix("sccache")) + .unwrap() + .collect::<Result<Vec<_>, _>>() + .unwrap(); + assert!( + leftover.is_empty(), + "every per-key ref deleted: {leftover:?}" + ); + resolve_all("after consolidation"); + + // A later write plus a second consolidation merges into the existing + // tree without losing already-consolidated keys. + let blob = util::hash_blob(work.path(), "value-4"); + util::set_ref(&refs, "refs/cache/sccache/gg", util::oid(&blob)); + assert_eq!(cache::consolidate("sccache", &refs, &objects).unwrap(), 1); + resolve_all("after second consolidation"); + let gg = cache_ns::resolve(&refs, &objects, "sccache", "gg") + .unwrap() + .expect("gg resolves via the merged tree"); + assert_eq!(objects.read(gg).unwrap().data, b"value-4"); +} + +/// All-or-nothing under a racing writer: if any per-key ref moves between +/// prepare and commit, the whole transaction is rejected — the +/// consolidated ref is not written and no per-key ref is deleted. +#[test] +fn a_racing_cache_write_rejects_the_whole_consolidation_transaction() { + let work = repo(); + util::use_main_branch(work.path()); + std::fs::write(work.path().join("file"), "seed").unwrap(); + commit_all(work.path(), "seed"); + + let refs = refstore_files::FilesRefStore::open(work.path()).unwrap(); + let objects = odb_files::OdbFiles::open(work.path()).unwrap(); + + let blob_a = util::hash_blob(work.path(), "a"); + let blob_b = util::hash_blob(work.path(), "b"); + util::set_ref(&refs, "refs/cache/sccache/aa", util::oid(&blob_a)); + util::set_ref(&refs, "refs/cache/sccache/bb", util::oid(&blob_b)); + + let plan = cache::prepare_consolidation("sccache", &refs, &objects) + .unwrap() + .expect("two per-key refs to consolidate"); + + // Race: another writer replaces one per-key ref before the commit. + let racer = util::hash_blob(work.path(), "racer"); + util::set_ref(&refs, "refs/cache/sccache/aa", util::oid(&racer)); + + assert!( + !cache::commit_consolidation(&refs, &plan).unwrap(), + "a moved per-key ref must reject the batch" + ); + // Nothing changed: no consolidated ref, both per-key refs intact. + assert!( + refs.get(&cache_ns::consolidated_ref("sccache")) + .unwrap() + .is_none() + ); + assert_eq!( + refs.get(&RefName::new("refs/cache/sccache/aa")).unwrap(), + Some(util::oid(&racer)) + ); + assert_eq!( + refs.get(&RefName::new("refs/cache/sccache/bb")).unwrap(), + Some(util::oid(&blob_b)) + ); +}

crates/git-maintenance/tests/conformance.rs @@ -1,0 +1,136 @@ +//! Closing WS2's collector seam: the causal-collection-safety property +//! (`docs/scale-out.adoc`, correctness rule 1) instantiated against +//! collectors that actually collect — [`TigrisCollector`] (grace-based, +//! including the staging-timeout boundary the suite's own docs assign to +//! the backend's instantiation) and [`FilesCollector`]. + +#![allow( + clippy::unwrap_used, + clippy::expect_used, + reason = "test assertions, not application code" +)] + +mod util; + +use std::time::Duration; + +use backend_conformance::Collector as _; +use git_backend::{ObjectStore as _, PackStream}; +use git_maintenance::collector::{FilesCollector, TigrisCollector}; +use git_store::test_support::{commit_all, head, repo}; +use odb_tigris::OdbTigris; +use odb_tigris::registry::memory::InMemoryRegistry; +use odb_tigris::transport::fs::FsTransport; + +/// The causal-collection-safety property against a Tigris store whose +/// collector really collects: `collector.collect()` runs a full expire + +/// mark-and-sweep pass while the fixture pack is staged — a collector +/// that reaped or unregistered a staged object fails the property's +/// promote/read assertions. +#[test] +fn causal_collection_safety_holds_under_a_real_tigris_collector() { + let scratch = repo(); + let bucket = tempfile::tempdir().unwrap(); + let transport = FsTransport::open(bucket.path()).unwrap(); + let registry = InMemoryRegistry::new(); + // A generous grace window: the property itself never sleeps, so a + // staging session inside it is always in-window — the boundary is + // exercised separately below. + let store = + OdbTigris::new(&transport, &registry, "repo").with_staging_grace(Duration::from_secs(300)); + let refs = refstore_files::FilesRefStore::open(scratch.path()).unwrap(); + let collector = TigrisCollector::new("repo", &refs, &store, &transport, &registry); + assert_eq!(collector.staging_grace(), Some(Duration::from_secs(300))); + + backend_conformance::causal_collection_safety(&store, &collector); +} + +/// The staging-timeout boundary (rule 1: "a staging session that cannot +/// complete within the grace window aborts rather than becoming +/// collectible mid-flight; the suite tests the boundary"): a session held +/// past its grace window is *aborted* — its promote fails and its objects +/// stay invisible — never half-collected under a committed ref. +#[test] +fn a_staging_session_past_its_grace_window_aborts_rather_than_promotes() { + let scratch = repo(); + let work = repo(); + let bucket = tempfile::tempdir().unwrap(); + let transport = FsTransport::open(bucket.path()).unwrap(); + let registry = InMemoryRegistry::new(); + let grace = Duration::from_millis(200); + let store = OdbTigris::new(&transport, &registry, "repo").with_staging_grace(grace); + let refs = refstore_files::FilesRefStore::open(scratch.path()).unwrap(); + let collector = TigrisCollector::new("repo", &refs, &store, &transport, &registry); + assert_eq!(collector.staging_grace(), Some(grace)); + + let blob = util::hash_blob(work.path(), "too-slow"); + let quarantine = store + .stage_pack(PackStream::new(std::io::Cursor::new(util::pack_of( + work.path(), + &[&blob], + )))) + .unwrap(); + + // Hold the session open past its deadline, then run a collection + // pass — the grace-based cruft arm reaps the expired quarantine. + std::thread::sleep(grace + Duration::from_millis(300)); + collector.collect(); + + // The boundary: the session aborts. Promote must fail — succeeding + // here would mean a session became collectible mid-flight and then + // committed anyway — and the staged object must remain invisible. + assert!( + store.promote(quarantine).is_err(), + "a staging session past its grace window must abort, not promote" + ); + assert!(!store.contains(util::oid(&blob)).unwrap()); +} + +/// The same boundary without a collection pass: expiry is a property of +/// the session's own deadline, not of whether a collector happened to run +/// first — promote-after-deadline aborts either way. +#[test] +fn promote_past_the_grace_window_aborts_even_without_a_collection_pass() { + let work = repo(); + let bucket = tempfile::tempdir().unwrap(); + let transport = FsTransport::open(bucket.path()).unwrap(); + let registry = InMemoryRegistry::new(); + let grace = Duration::from_millis(200); + let store = OdbTigris::new(&transport, &registry, "repo").with_staging_grace(grace); + + let blob = util::hash_blob(work.path(), "also-too-slow"); + let quarantine = store + .stage_pack(PackStream::new(std::io::Cursor::new(util::pack_of( + work.path(), + &[&blob], + )))) + .unwrap(); + std::thread::sleep(grace + Duration::from_millis(300)); + + assert!(store.promote(quarantine).is_err()); + assert!(!store.contains(util::oid(&blob)).unwrap()); +} + +/// The causal-collection-safety property against the files backend with a +/// collector that really collects ([`git_maintenance::gc::collect_files`]). +/// No grace window: the local backend bounds staging by promotion alone. +#[test] +fn causal_collection_safety_holds_under_a_real_files_collector() { + let dest = repo(); + // Give the collector's mark something real: a committed, reachable + // history in the same repository the property stages into. + util::use_main_branch(dest.path()); + std::fs::write(dest.path().join("file"), "reachable").unwrap(); + commit_all(dest.path(), "reachable"); + let reachable = head(dest.path()); + + let store = odb_files::OdbFiles::open(dest.path()).unwrap(); + let collector = FilesCollector::new(dest.path()); + assert_eq!(collector.staging_grace(), None); + + backend_conformance::causal_collection_safety(&store, &collector); + + // And the pass was a real one: the repo's reachable history survived. + let object = store.read(util::oid(&reachable)).unwrap(); + assert_eq!(object.kind, gix_object::Kind::Commit); +}

crates/git-maintenance/tests/gc.rs @@ -1,0 +1,150 @@ +//! Mark-and-sweep correctness (`docs/scale-out.adoc`, WS9): fully +//! unreachable packs are deleted, mixed packs are rewritten preserving +//! every reachable object, and staged/quarantined objects are untouched +//! (correctness rules 1 and 2). + +#![allow( + clippy::unwrap_used, + clippy::expect_used, + reason = "test assertions, not application code" +)] + +mod util; + +use git_backend::{ObjectStore as _, PackStream}; +use git_maintenance::gc; +use git_store::test_support::{commit_all, head, repo}; +use odb_tigris::OdbTigris; +use odb_tigris::registry::PackRegistry as _; +use odb_tigris::registry::memory::InMemoryRegistry; +use odb_tigris::transport::{BlobTransport as _, fs::FsTransport}; + +#[test] +fn a_fully_unreachable_pack_is_deleted_registry_first() { + let work = repo(); + util::use_main_branch(work.path()); + std::fs::write(work.path().join("file"), "one").unwrap(); + commit_all(work.path(), "a"); + let a = head(work.path()); + + // An independent root the refs will never point at once its branch is + // gone: its pack becomes fully unreachable. + util::git(work.path(), &["checkout", "-q", "--orphan", "gone"]); + std::fs::write(work.path().join("file"), "doomed").unwrap(); + commit_all(work.path(), "c"); + let c = head(work.path()); + util::git(work.path(), &["checkout", "-q", "main"]); + util::git(work.path(), &["branch", "-q", "-D", "gone"]); + + let bucket = tempfile::tempdir().unwrap(); + let transport = FsTransport::open(bucket.path()).unwrap(); + let registry = InMemoryRegistry::new(); + let store = OdbTigris::new(&transport, &registry, "repo"); + util::stage_and_promote(&store, util::pack_for(work.path(), &a)); + util::stage_and_promote(&store, util::pack_for(work.path(), &c)); + assert_eq!(registry.list("repo").unwrap().len(), 2); + + let refs = refstore_files::FilesRefStore::open(work.path()).unwrap(); + let outcome = gc::collect("repo", &refs, &store, &transport, &registry).unwrap(); + + assert_eq!(outcome.deleted_packs, 1); + assert_eq!(outcome.rewritten_packs, 0); + let remaining = registry.list("repo").unwrap(); + assert_eq!(remaining.len(), 1, "only the reachable pack remains"); + assert!(store.contains(util::oid(&a)).unwrap()); + assert!( + !store.contains(util::oid(&c)).unwrap(), + "the unreachable commit's pack must be gone from the registry" + ); + // The blobs are gone from the bucket too, not just unregistered. + for record in &remaining { + assert!(transport.exists(&record.pack_key).unwrap()); + } +} + +#[test] +fn a_mixed_pack_is_rewritten_preserving_reachable_objects() { + let work = repo(); + util::use_main_branch(work.path()); + std::fs::write(work.path().join("file"), "one").unwrap(); + commit_all(work.path(), "a"); + let a = head(work.path()); + std::fs::write(work.path().join("file2"), "two").unwrap(); + commit_all(work.path(), "b"); + let b = head(work.path()); + + let bucket = tempfile::tempdir().unwrap(); + let transport = FsTransport::open(bucket.path()).unwrap(); + let registry = InMemoryRegistry::new(); + let store = OdbTigris::new(&transport, &registry, "repo"); + // One pack holding both commits' closures. + util::stage_and_promote(&store, util::pack_for(work.path(), &b)); + let original = registry.list("repo").unwrap(); + assert_eq!(original.len(), 1); + let original = original.into_iter().next().unwrap(); + + // Rewind main to `a`: `b`'s commit/tree/blob become unreachable, `a`'s + // closure stays live — a mixed pack. + util::git(work.path(), &["update-ref", "refs/heads/main", &a]); + util::git(work.path(), &["reset", "-q", "--hard", &a]); + + let refs = refstore_files::FilesRefStore::open(work.path()).unwrap(); + let outcome = gc::collect("repo", &refs, &store, &transport, &registry).unwrap(); + + assert_eq!(outcome.deleted_packs, 0); + assert_eq!(outcome.rewritten_packs, 1); + let rewritten = registry.list("repo").unwrap(); + assert_eq!(rewritten.len(), 1); + let rewritten = rewritten.into_iter().next().unwrap(); + assert_ne!(rewritten.id, original.id, "the old pack was replaced"); + assert!( + !transport.exists(&original.pack_key).unwrap(), + "the old mixed pack's bytes are gone" + ); + + // Every reachable object survived the rewrite; the unreachable commit + // did not. + let commit_a = store.read(util::oid(&a)).unwrap(); + assert_eq!(commit_a.kind, gix_object::Kind::Commit); + assert!(!store.contains(util::oid(&b)).unwrap()); + // a's closure is commit + tree + one blob. + assert_eq!(rewritten.object_count, Some(3)); +} + +#[test] +fn staged_quarantined_objects_are_untouched_by_a_collection_pass() { + let work = repo(); + util::use_main_branch(work.path()); + std::fs::write(work.path().join("file"), "committed").unwrap(); + commit_all(work.path(), "committed"); + let committed = head(work.path()); + + let bucket = tempfile::tempdir().unwrap(); + let transport = FsTransport::open(bucket.path()).unwrap(); + let registry = InMemoryRegistry::new(); + let store = OdbTigris::new(&transport, &registry, "repo"); + util::stage_and_promote(&store, util::pack_for(work.path(), &committed)); + + // Staged for an in-flight transaction, never promoted. The sweep + // enumerates the registry only — structurally, a quarantine cannot + // even be named by it (rule 2: "GC never scans quarantine"). + let staged_blob = util::hash_blob(work.path(), "in-flight"); + let quarantine = store + .stage_pack(PackStream::new(std::io::Cursor::new(util::pack_of( + work.path(), + &[&staged_blob], + )))) + .unwrap(); + + let refs = refstore_files::FilesRefStore::open(work.path()).unwrap(); + let outcome = gc::collect("repo", &refs, &store, &transport, &registry).unwrap(); + assert_eq!(outcome.deleted_packs, 0); + + // The in-flight transaction can still commit: promote succeeds and + // the objects are correct — a collector that touched quarantine would + // fail here. + store.promote(quarantine).unwrap(); + let object = store.read(util::oid(&staged_blob)).unwrap(); + assert_eq!(object.kind, gix_object::Kind::Blob); + assert_eq!(object.data, b"in-flight"); +}

crates/git-maintenance/tests/lock_and_schedule.rs @@ -1,0 +1,194 @@ +//! Advisory-lock serialization (two concurrent maintenance runs — one +//! runs, one skips) and threshold-driven scheduling +//! (`docs/scale-out.adoc`, WS9; the Postgres advisory-lock counterpart is +//! exercised in `refstore-postgres`'s docker-gated suite). + +#![allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::unwrap_in_result, + reason = "test assertions, not application code" +)] + +use std::sync::Mutex; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use git_backend::EffectDef; +use git_maintenance::lock::{FileMaintenanceLock, MaintenanceLock as _, run_exclusive}; +use git_maintenance::schedule::{ + MaintenanceSink, Scheduler, Stats, Thresholds, schedule_maintenance, +}; + +#[test] +fn two_concurrent_maintenance_runs_serialize_one_skips() { + let dir = tempfile::tempdir().unwrap(); + let lock_a = FileMaintenanceLock::for_repo(dir.path()); + let lock_b = FileMaintenanceLock::for_repo(dir.path()); + + // While one run holds the lock, a concurrent run skips whole — + // `run_exclusive` returns `None` without executing its work. + let guard = lock_a.try_acquire().unwrap().expect("first acquisition"); + let ran = std::sync::atomic::AtomicBool::new(false); + let outcome = run_exclusive(&lock_b, || { + ran.store(true, Ordering::SeqCst); + Ok(()) + }) + .unwrap(); + assert!(outcome.is_none(), "the second run must skip"); + assert!(!ran.load(Ordering::SeqCst), "skipped means never executed"); + + // Release: the next run proceeds. + drop(guard); + let outcome = run_exclusive(&lock_b, || Ok(42)).unwrap(); + assert_eq!(outcome, Some(42)); +} + +#[test] +fn the_lock_wraps_the_whole_run() { + let dir = tempfile::tempdir().unwrap(); + let lock_a = FileMaintenanceLock::for_repo(dir.path()); + let lock_b = FileMaintenanceLock::for_repo(dir.path()); + + // From inside a run, a concurrent acquisition fails — the lock is + // held for the run's full duration, not per phase. + let outcome = run_exclusive(&lock_a, || { + assert!(lock_b.try_acquire().unwrap().is_none()); + Ok(()) + }) + .unwrap(); + assert!(outcome.is_some()); + // And it is released once the run returns. + assert!(lock_b.try_acquire().unwrap().is_some()); +} + +/// A sink recording every enqueue. +#[derive(Default)] +struct RecordingSink { + enqueued: Mutex<Vec<(String, Vec<EffectDef>)>>, +} + +impl MaintenanceSink for RecordingSink { + fn enqueue(&self, repo_id: &str, effects: &[EffectDef]) -> git_backend::Result<()> { + self.enqueued + .lock() + .unwrap() + .push((repo_id.to_owned(), effects.to_vec())); + Ok(()) + } +} + +#[test] +fn schedule_maintenance_enqueues_all_four_effects_at_threshold() { + let sink = RecordingSink::default(); + let thresholds = Thresholds { + maintenance: 10, + reachability: 10, + }; + + // Below threshold: nothing. + let effects = schedule_maintenance( + "repo", + &Stats { + ref_updates_since_last: 9, + }, + &thresholds, + &sink, + ) + .unwrap(); + assert!(effects.is_empty()); + assert!(sink.enqueued.lock().unwrap().is_empty()); + + // At threshold: GC, cache TTL, consolidation, and reachability + // regeneration (WS6's `should_regenerate` trigger, scheduled here). + let effects = schedule_maintenance( + "repo", + &Stats { + ref_updates_since_last: 10, + }, + &thresholds, + &sink, + ) + .unwrap(); + let names: Vec<&str> = effects.iter().map(|effect| effect.name.as_str()).collect(); + assert_eq!( + names, + vec![ + git_maintenance::schedule::GC_EFFECT, + git_maintenance::schedule::CACHE_TTL_EFFECT, + git_maintenance::schedule::CONSOLIDATION_EFFECT, + git_reachability::maintenance::EFFECT_NAME, + ] + ); + let enqueued = sink.enqueued.lock().unwrap(); + assert_eq!(enqueued.len(), 1); + let (repo, batch) = enqueued.first().unwrap(); + assert_eq!(repo, "repo"); + assert_eq!(batch.len(), 4); +} + +#[test] +fn the_scheduler_accumulates_per_repo_and_resets_on_trigger() { + let sink = std::sync::Arc::new(RecordingSink::default()); + let scheduler = Scheduler::new( + Thresholds { + maintenance: 5, + reachability: 5, + }, + sink.clone(), + ); + + // Accumulate across pushes; trigger only at the threshold. + assert!(scheduler.note_ref_updates("a", 2).unwrap().is_empty()); + assert!(scheduler.note_ref_updates("a", 2).unwrap().is_empty()); + // A different repo's count is independent. + assert!(scheduler.note_ref_updates("b", 4).unwrap().is_empty()); + let effects = scheduler.note_ref_updates("a", 1).unwrap(); + assert_eq!(effects.len(), 4, "threshold crossed for repo a"); + + // The count reset: the next small update does not re-trigger. + assert!(scheduler.note_ref_updates("a", 1).unwrap().is_empty()); + assert_eq!(sink.enqueued.lock().unwrap().len(), 1); +} + +/// A sink that fails a configurable number of times. +struct FlakySink { + failures_left: AtomicUsize, + inner: RecordingSink, +} + +impl MaintenanceSink for FlakySink { + fn enqueue(&self, repo_id: &str, effects: &[EffectDef]) -> git_backend::Result<()> { + if self + .failures_left + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |left| { + left.checked_sub(1) + }) + .is_ok() + { + return Err(git_backend::Error::RefStore("queue outage".to_owned())); + } + self.inner.enqueue(repo_id, effects) + } +} + +#[test] +fn a_sink_failure_delays_the_trigger_rather_than_losing_it() { + let sink = std::sync::Arc::new(FlakySink { + failures_left: AtomicUsize::new(1), + inner: RecordingSink::default(), + }); + let scheduler = Scheduler::new( + Thresholds { + maintenance: 3, + reachability: 3, + }, + sink.clone(), + ); + + // Crossing the threshold during the outage errors — but restores the + // count, so the very next update re-triggers. + let _outage = scheduler.note_ref_updates("a", 3).unwrap_err(); + let effects = scheduler.note_ref_updates("a", 1).unwrap(); + assert_eq!(effects.len(), 4); + assert_eq!(sink.inner.enqueued.lock().unwrap().len(), 1); +}

crates/git-maintenance/tests/util/mod.rs @@ -1,0 +1,137 @@ +//! Shared fixtures for the WS9 maintenance tests: real git objects and +//! packs, built the same way `odb-files`' and `backend-conformance`'s own +//! fixtures build them. + +#![allow( + clippy::unwrap_used, + clippy::expect_used, + reason = "test fixtures, not application code" +)] +#![allow( + dead_code, + reason = "shared by several test binaries, each using a different subset" +)] + +use std::path::Path; +use std::process::{Command, Stdio}; + +use git_backend::{Expected, ObjectStore, PackStream, RefEdit, RefName, RefStore, TxOutcome}; +use gix_hash::ObjectId; + +/// Run `git` in `dir`, asserting success. +pub fn git(dir: &Path, args: &[&str]) { + let status = Command::new("git") + .arg("-C") + .arg(dir) + .args(args) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .unwrap(); + assert!(status.success(), "git {args:?} failed in {dir:?}"); +} + +/// Pin a fresh repository's unborn `HEAD` to `refs/heads/main`, so tests +/// can name the default branch regardless of the host's +/// `init.defaultBranch`. +pub fn use_main_branch(dir: &Path) { + git(dir, &["symbolic-ref", "HEAD", "refs/heads/main"]); +} + +/// A pack of `revspec` and everything it reaches, via +/// `git rev-list | git pack-objects` — the same bytes a push transmits. +pub fn pack_for(dir: &Path, revspec: &str) -> Vec<u8> { + let mut rev_list = Command::new("git") + .arg("-C") + .arg(dir) + .args(["rev-list", "--objects", revspec]) + .stdout(Stdio::piped()) + .spawn() + .unwrap(); + let pack_objects = Command::new("git") + .arg("-C") + .arg(dir) + // `--delta-base-offset`: emit OFS deltas (what a push negotiates), + // which gitoxide's indexer resolves in-pack; the REF deltas + // pack-objects emits by default assume a lookup no staged pack has. + .args(["pack-objects", "--stdout", "-q", "--delta-base-offset"]) + .stdin(rev_list.stdout.take().unwrap()) + .stdout(Stdio::piped()) + .spawn() + .unwrap(); + let output = pack_objects.wait_with_output().unwrap(); + assert!(rev_list.wait().unwrap().success()); + assert!(output.status.success()); + output.stdout +} + +/// A pack containing exactly the objects named by `oids` (hex), fed to +/// `git pack-objects` directly — for packing dangling blobs a rev walk +/// would never reach. +pub fn pack_of(dir: &Path, oids: &[&str]) -> Vec<u8> { + let mut child = Command::new("git") + .arg("-C") + .arg(dir) + .args(["pack-objects", "--stdout", "-q"]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .spawn() + .unwrap(); + { + use std::io::Write as _; + let mut stdin = child.stdin.take().unwrap(); + for oid in oids { + writeln!(stdin, "{oid}").unwrap(); + } + } + let output = child.wait_with_output().unwrap(); + assert!(output.status.success()); + output.stdout +} + +/// Write `content` as a blob into `dir`'s object database, returning its +/// hex id. +pub fn hash_blob(dir: &Path, content: &str) -> String { + let mut child = Command::new("git") + .arg("-C") + .arg(dir) + .args(["hash-object", "-w", "--stdin"]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .spawn() + .unwrap(); + { + use std::io::Write as _; + let mut stdin = child.stdin.take().unwrap(); + stdin.write_all(content.as_bytes()).unwrap(); + } + let output = child.wait_with_output().unwrap(); + assert!(output.status.success()); + String::from_utf8(output.stdout).unwrap().trim().to_owned() +} + +/// Parse a hex object id. +pub fn oid(hex: &str) -> ObjectId { + ObjectId::from_hex(hex.as_bytes()).unwrap() +} + +/// Stage `pack` into `store` and promote it immediately. +pub fn stage_and_promote(store: &dyn ObjectStore, pack: Vec<u8>) { + let quarantine = store + .stage_pack(PackStream::new(std::io::Cursor::new(pack))) + .unwrap(); + store.promote(quarantine).unwrap(); +} + +/// Set `name` to `target` through a `RefStore` transaction (creating or +/// clobbering), asserting it applied. +pub fn set_ref(refs: &dyn RefStore, name: &str, target: ObjectId) { + let outcome = refs + .transaction(&[RefEdit { + name: RefName::new(name), + expected: Expected::Any, + new: Some(target), + }]) + .unwrap(); + assert!(matches!(outcome, TxOutcome::Applied)); +}