git ents: feat: add git-reachability crate for WS6 accelerated reachability

Cargo.lock @@ -1680,6 +1680,7 @@ "git-ents-core", "git-member", "git-protocol", + "git-reachability", "git-signed-push", "git-store", "git-toolchain", @@ -1717,6 +1718,7 @@ "git-backend", "git-ents-core", "git-member", + "git-reachability", "git-signed-push", "gix", "gix-actor", @@ -1732,6 +1734,22 @@ "thiserror 2.0.18", ] +[[package]] +name = "git-reachability" +version = "0.0.0" +dependencies = [ + "git-backend", + "gix-actor", + "gix-date", + "gix-hash", + "gix-object", + "odb-files", + "odb-tigris", + "refstore-files", + "tempfile", + "thiserror 2.0.18", +] + [[package]] name = "git-signed-push" version = "0.0.0"

Cargo.toml @@ -11,6 +11,7 @@ "crates/git-ents-server", "crates/git-member", "crates/git-protocol", + "crates/git-reachability", "crates/git-signed-push", "crates/git-store", "crates/git-toolchain", @@ -71,6 +72,7 @@ git-ents-server = { path = "crates/git-ents-server" } git-member = { path = "crates/git-member" } git-protocol = { path = "crates/git-protocol" } +git-reachability = { path = "crates/git-reachability" } git-signed-push = { path = "crates/git-signed-push" } git-store = { path = "crates/git-store" } git-toolchain = { path = "crates/git-toolchain" }

crates/git-ents-server/Cargo.toml @@ -24,6 +24,7 @@ git-effect = { workspace = true } git-member = { workspace = true } git-protocol = { workspace = true } +git-reachability = { workspace = true } git-signed-push = { workspace = true } git-store = { workspace = true } git-toolchain = { workspace = true }

crates/git-protocol/Cargo.toml @@ -9,6 +9,7 @@ git-backend = { workspace = true } git-ents-core = { workspace = true } git-member = { workspace = true } +git-reachability = { workspace = true } git-signed-push = { workspace = true } gix = { workspace = true } gix-actor = { workspace = true }

crates/git-ents-server/src/native_git.rs @@ -74,6 +74,14 @@ objects: Arc::new(objects), authorized_members: git_member::members::without_revoked(members, &revoked), config, + // No reachability artifacts for the disk-hydrated backend yet: + // `git-reachability`'s maintenance effect and artifact storage + // target the cloud stack (`odb-tigris` + its pack registry); + // wiring generation/loading for this resolver is future work, + // and negotiation/ingest degrade to the plain walk in the + // meantime (`docs/scale-out.adoc`: "absence ... degrades + // speed, never answers"). + reachability: git_reachability::ArtifactBundle::empty(), }) } }

crates/git-protocol/src/lib.rs @@ -41,6 +41,12 @@ /// the promoted object store could resolve. #[error("missing object {0}: push rejected, connectivity check failed")] MissingObject(gix_hash::ObjectId), + /// The reachability walk (`git-reachability`, WS6) itself failed — + /// distinct from [`Self::MissingObject`], which is this crate's own + /// mapping of that same failure at the negotiation/ingest call sites + /// that need a typed reason to report back to a client. + #[error(transparent)] + Reachability(#[from] git_reachability::Error), /// Decoding a commit, tree, or tag object failed. #[error("could not decode object: {0}")] Decode(String),

crates/git-protocol/src/walk.rs @@ -1,116 +1,14 @@ -//! A generic reachability walk over whatever [`ObjectSource`] answers -//! `find`, shared by negotiation (`docs/scale-out.adoc`, "Reachability": -//! "Negotiation, push connectivity checking, and GC mark are the same -//! walk") and the ingest connectivity check. +//! Re-exports `git-reachability`'s shared reachability walk (WS6 moved the +//! implementation there — see that crate's `walk` module docs for why: +//! it's where the commit-graph accelerator this walk degrades from now +//! lives). Kept as a `walk` module here too so `negotiate`/`ingest`'s +//! existing `crate::walk::{...}` imports, and this crate's own tests, +//! didn't need to change along with the move. //! -//! This is a naive, one-object-at-a-time walk through -//! [`git_backend::ObjectStore::read`] — correct, not fast. The doc calls out -//! pack generation over ranged reads as its own risk budget (Q6, WS5/WS6); -//! this walk is the thing that eventually needs a commit-graph/bitmap -//! accelerator (WS6) instead of visiting every object. +//! [`crate::native::negotiate`] and [`crate::native::ingest`] call +//! [`git_reachability::engine::accelerated_reachable`] directly rather than +//! [`reachable`] — the accelerated entry point wraps this walk, it isn't +//! re-exported under this name too, so the call sites make plain which one +//! they're using. -use std::collections::BTreeSet; - -use git_backend::ObjectStore; -use gix_hash::ObjectId; -use gix_object::{CommitRef, Kind, TagRef, TreeRefIter}; - -use crate::{Error, Result}; - -/// Where [`reachable`] reads object kind/data from. Lets the same walk run -/// over a repository's promoted object store alone (negotiation, GC mark) -/// or a promoted store combined with a not-yet-promoted incoming pack -/// (ingest connectivity checking). -pub trait ObjectSource { - /// The kind and raw content of `id`, or `None` if this source has never - /// heard of it. - fn find(&self, id: &ObjectId) -> Result<Option<(Kind, Vec<u8>)>>; -} - -/// An [`ObjectSource`] over a repository's promoted [`ObjectStore`] alone. -pub struct StoreSource<'a> { - store: &'a dyn ObjectStore, -} - -impl<'a> StoreSource<'a> { - /// Read only through `store`'s promoted view. - pub fn new(store: &'a dyn ObjectStore) -> Self { - Self { store } - } -} - -impl ObjectSource for StoreSource<'_> { - fn find(&self, id: &ObjectId) -> Result<Option<(Kind, Vec<u8>)>> { - if !self.store.contains(*id)? { - return Ok(None); - } - let object = self.store.read(*id)?; - Ok(Some((object.kind, object.data))) - } -} - -/// Walk every object reachable from `roots` via commit parents, commit/tag -/// targets, and tree entries (skipping gitlink/submodule entries, which name -/// a commit in a different repository's object space). -/// -/// `stop` marks a boundary: when it returns `true` for an id, that id is -/// recorded as seen but never resolved or descended into — the caller -/// already knows it (and everything under it) is accounted for, e.g. -/// negotiation's haves closure, or the ingest connectivity check's existing -/// history. -/// -/// When `lenient` is `false`, an id `stop` did not claim but `source` cannot -/// resolve is a connectivity failure ([`Error::MissingObject`]) — the ingest -/// check's use. When `true`, it is silently dropped instead — appropriate -/// for a client-supplied `have` the server never actually had, which is a -/// stale claim, not a corruption. -pub fn reachable( - roots: impl IntoIterator<Item = ObjectId>, - source: &dyn ObjectSource, - mut stop: impl FnMut(&ObjectId) -> bool, - lenient: bool, -) -> Result<BTreeSet<ObjectId>> { - let mut seen = BTreeSet::new(); - let mut stack: Vec<ObjectId> = roots.into_iter().collect(); - while let Some(id) = stack.pop() { - if !seen.insert(id) { - continue; - } - if stop(&id) { - continue; - } - let found = source.find(&id)?; - let Some((kind, data)) = found else { - if lenient { - continue; - } - return Err(Error::MissingObject(id)); - }; - match kind { - Kind::Commit => { - let commit = CommitRef::from_bytes(&data, gix_hash::Kind::Sha1) - .map_err(|error| Error::Decode(error.to_string()))?; - stack.push(commit.tree()); - stack.extend(commit.parents()); - } - Kind::Tree => { - for entry in TreeRefIter::from_bytes(&data, gix_hash::Kind::Sha1) { - let entry = entry.map_err(|error| Error::Decode(error.to_string()))?; - if entry.mode.kind() == gix_object::tree::EntryKind::Commit { - // A submodule gitlink: an object id in another - // repository's object space, never ours to resolve. - continue; - } - stack.push(entry.oid.to_owned()); - } - } - Kind::Tag => { - let tag = TagRef::from_bytes(&data, gix_hash::Kind::Sha1) - .map_err(|error| Error::Decode(error.to_string()))?; - stack.push(tag.target()); - } - Kind::Blob => {} - } - } - Ok(seen) -} +pub use git_reachability::walk::{ObjectSource, StoreSource, reachable};

crates/odb-tigris/src/registry.rs @@ -61,9 +61,66 @@ pub object_count: Option<u64>, } +/// Which reachability accelerator an [`ArtifactRecord`] holds — see +/// `git-reachability` (`docs/scale-out.adoc`, "Reachability" / WS6) for the +/// binary formats themselves. Named here, rather than in `git-reachability`, +/// because the registry (this trait) is the thing both that crate and its +/// Postgres implementation (`refstore-postgres`) need to agree on. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ArtifactKind { + /// A serialized commit graph: OID -> (tree, parents, generation). + CommitGraph, + /// A reachable-object-set snapshot for one tip-frontier. + ReachableSet, +} + +impl ArtifactKind { + /// A stable string form, used as the on-disk/column discriminator by + /// every [`PackRegistry`] implementation. + #[must_use] + pub fn as_str(&self) -> &'static str { + match self { + Self::CommitGraph => "commit-graph", + Self::ReachableSet => "reachable-set", + } + } + + /// Parse [`Self::as_str`]'s output back, or `None` for anything else — + /// forward-compatible with a future kind an older reader doesn't know. + #[must_use] + pub fn parse(value: &str) -> Option<Self> { + match value { + "commit-graph" => Some(Self::CommitGraph), + "reachable-set" => Some(Self::ReachableSet), + _ => None, + } + } +} + +/// One reachability artifact registered for a repo: enough to fetch its +/// bytes from the bucket. A repo has at most one live artifact per +/// [`ArtifactKind`] — regenerating (`git-reachability`'s maintenance effect) +/// overwrites it, rather than accumulating snapshots, so lookup is by +/// `(repo_id, kind)` alone. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ArtifactRecord { + /// The repo this artifact belongs to. + pub repo_id: String, + /// Which accelerator this artifact holds. + pub kind: ArtifactKind, + /// The bucket key holding the artifact's bytes. + pub key: String, +} + /// Registry of promoted packs: the commit point [`crate::OdbTigris::promote`] /// writes to, and the only thing [`crate::OdbTigris::read`] and /// [`crate::OdbTigris::contains`] consult to learn which packs exist. +/// +/// Also the discovery point for reachability artifacts (`docs/ +/// scale-out.adoc`, "Reachability": "stored beside packs, tracked in the +/// pack registry") — a minimal extension over the pack-only shape WS5 +/// introduced, since both are "what has this repo got, and where" lookups +/// against the same store. pub trait PackRegistry: Send + Sync { /// Record `record` as promoted and live. Called once per pack, after its /// bytes are durably in the bucket at the live keys `record` names. @@ -88,4 +145,30 @@ /// /// Returns an error if the registry cannot be written. fn delete(&self, repo_id: &str, id: &PackId) -> Result<()>; + + /// Record `record` as `repo_id`'s current artifact of its kind, + /// replacing whatever was previously registered for that + /// `(repo_id, kind)` pair. + /// + /// # Errors + /// + /// Returns an error if the record cannot be durably written. + fn record_artifact(&self, record: ArtifactRecord) -> Result<()>; + + /// `repo_id`'s current artifact of `kind`, or `None` if it has never + /// been generated — the "absent artifact" case every consumer must + /// degrade gracefully from (`docs/scale-out.adoc`, "Reachability"). + /// + /// # Errors + /// + /// Returns an error if the registry cannot be read. + fn get_artifact(&self, repo_id: &str, kind: ArtifactKind) -> Result<Option<ArtifactRecord>>; + + /// Remove `repo_id`'s artifact of `kind`, if any. Not an error if + /// already absent. + /// + /// # Errors + /// + /// Returns an error if the registry cannot be written. + fn delete_artifact(&self, repo_id: &str, kind: ArtifactKind) -> Result<()>; }

crates/refstore-postgres/migrations/0001_init.sql @@ -111,3 +111,18 @@ CREATE INDEX IF NOT EXISTS git_ents_op_records_repo_idx ON git_ents_op_records (repo_id, created_at DESC); + +-- Reachability artifacts (WS6, `docs/scale-out.adoc`'s "Reachability" +-- section): the commit-graph and reachable-set accelerators +-- `git-reachability`'s maintenance effect generates, tracked here rather +-- than by bucket listing, same as packs above. One row per `(repo_id, +-- kind)` — regenerating overwrites the existing row instead of +-- accumulating snapshots, so `kind` alone (not a generated id) is enough to +-- look one up. +CREATE TABLE IF NOT EXISTS git_ents_reachability_artifacts ( + repo_id TEXT NOT NULL, + kind TEXT NOT NULL, + key TEXT NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (repo_id, kind) +);

crates/refstore-postgres/src/pack_registry.rs @@ -9,7 +9,7 @@ //! already owns the Postgres connection) implements it. use git_backend::{Error, Result}; -use odb_tigris::registry::{PackId, PackRecord, PackRegistry}; +use odb_tigris::registry::{ArtifactKind, ArtifactRecord, PackId, PackRecord, PackRegistry}; use crate::{PostgresRefStore, pg_err}; @@ -96,4 +96,65 @@ .map_err(pg_err) .map(|_rows_affected| ()) } + + fn record_artifact(&self, record: ArtifactRecord) -> Result<()> { + self.runtime + .block_on(async { + let client = self.client.lock().await; + client + .execute( + "INSERT INTO git_ents_reachability_artifacts + (repo_id, kind, key, updated_at) + VALUES ($1, $2, $3, now()) + ON CONFLICT (repo_id, kind) DO UPDATE SET + key = EXCLUDED.key, + updated_at = EXCLUDED.updated_at", + &[&record.repo_id, &record.kind.as_str(), &record.key], + ) + .await + }) + .map_err(pg_err) + .map(|_rows_affected| ()) + } + + fn get_artifact(&self, repo_id: &str, kind: ArtifactKind) -> Result<Option<ArtifactRecord>> { + let row = self + .runtime + .block_on(async { + let client = self.client.lock().await; + client + .query_opt( + "SELECT key FROM git_ents_reachability_artifacts + WHERE repo_id = $1 AND kind = $2", + &[&repo_id, &kind.as_str()], + ) + .await + }) + .map_err(pg_err)?; + row.map(|row| { + let key: String = row.try_get(0).map_err(pg_err)?; + Ok(ArtifactRecord { + repo_id: repo_id.to_owned(), + kind, + key, + }) + }) + .transpose() + } + + fn delete_artifact(&self, repo_id: &str, kind: ArtifactKind) -> Result<()> { + self.runtime + .block_on(async { + let client = self.client.lock().await; + client + .execute( + "DELETE FROM git_ents_reachability_artifacts + WHERE repo_id = $1 AND kind = $2", + &[&repo_id, &kind.as_str()], + ) + .await + }) + .map_err(pg_err) + .map(|_rows_affected| ()) + } }

crates/refstore-postgres/tests/odb_ws5_conformance.rs @@ -191,3 +191,58 @@ let url = pg.url().to_owned(); backend_conformance::object_store_properties(|| WithPostgres::new(&url), &NoopCollector); } + +#[test] +fn reachability_artifact_registry_round_trips_over_postgres() { + use odb_tigris::registry::{ArtifactKind, ArtifactRecord, PackRegistry}; + + let pg = require_postgres!("reachability_artifact_registry_round_trips_over_postgres"); + let repo_id = format!("ws6-artifact-{}", uuid::Uuid::new_v4()); + let registry = PostgresRefStore::connect(pg.url(), repo_id.clone()).expect("connect registry"); + + assert!( + registry + .get_artifact(&repo_id, ArtifactKind::CommitGraph) + .expect("get_artifact") + .is_none() + ); + + registry + .record_artifact(ArtifactRecord { + repo_id: repo_id.clone(), + kind: ArtifactKind::CommitGraph, + key: "some/key.bin".to_owned(), + }) + .expect("record_artifact"); + + let record = registry + .get_artifact(&repo_id, ArtifactKind::CommitGraph) + .expect("get_artifact") + .expect("artifact was just recorded"); + assert_eq!(record.key, "some/key.bin"); + + // Recording again for the same `(repo_id, kind)` replaces it, rather + // than accumulating a second row. + registry + .record_artifact(ArtifactRecord { + repo_id: repo_id.clone(), + kind: ArtifactKind::CommitGraph, + key: "new/key.bin".to_owned(), + }) + .expect("re-record_artifact"); + let record = registry + .get_artifact(&repo_id, ArtifactKind::CommitGraph) + .expect("get_artifact") + .expect("artifact still present"); + assert_eq!(record.key, "new/key.bin"); + + registry + .delete_artifact(&repo_id, ArtifactKind::CommitGraph) + .expect("delete_artifact"); + assert!( + registry + .get_artifact(&repo_id, ArtifactKind::CommitGraph) + .expect("get_artifact") + .is_none() + ); +}

crates/git-protocol/src/native/ingest.rs @@ -26,11 +26,13 @@ use gix_hash::ObjectId; use gix_object::Kind; +use git_reachability::engine::accelerated_reachable; + use super::{BackendResolver, NativeBackend}; use crate::attestation::{self, OP_LOG_REF}; use crate::pack::{PackObject, build_pack}; use crate::types::{AppliedRefEdit, PushOutcome, PushRequest}; -use crate::walk::{self, ObjectSource}; +use crate::walk::ObjectSource; use crate::{Error, IngestPack, Result}; /// An [`ObjectSource`] over the incoming pack's own scratch bundle (staged @@ -44,14 +46,14 @@ } impl ObjectSource for IncomingPackSource<'_> { - fn find(&self, id: &ObjectId) -> Result<Option<(Kind, Vec<u8>)>> { + fn find(&self, id: &ObjectId) -> git_reachability::Result<Option<(Kind, Vec<u8>)>> { if let Some(bundle) = self.bundle { let mut buf = Vec::new(); let mut inflate = gix_features::zlib::Inflate::default(); let mut cache = gix_pack::cache::Never; if let Some((data, _location)) = bundle .find(id, &mut buf, &mut inflate, &mut cache) - .map_err(|error| Error::Pack(error.to_string()))? + .map_err(|error| git_reachability::Error::Decode(error.to_string()))? { return Ok(Some((data.kind, data.data.to_vec()))); } @@ -126,20 +128,21 @@ }; let roots: Vec<ObjectId> = ref_edits.iter().filter_map(|edit| edit.new).collect(); - let connectivity = walk::reachable( + let connectivity = accelerated_reachable( roots, &source, |id| backends.objects.contains(*id).unwrap_or(false), false, + &backends.reachability, ); match connectivity { Ok(_reachable) => {} - Err(Error::MissingObject(id)) => { + Err(git_reachability::Error::MissingObject(id)) => { return Ok(PushOutcome::Rejected { reason: format!("connectivity check failed: missing object {id}"), }); } - Err(other) => return Err(other), + Err(other) => return Err(other.into()), } let quarantine = backends

crates/git-protocol/src/native/mod.rs @@ -45,6 +45,12 @@ pub authorized_members: Vec<Member>, /// The config `authorized_members`' roles are checked against. pub config: git_ents_core::config::Config, + /// This repository's reachability artifacts (WS6), if any have been + /// generated — [`git_reachability::ArtifactBundle::empty`] for a + /// resolver that hasn't wired artifact loading yet, which is exactly + /// the "absent artifact" case negotiation/ingest must (and do) degrade + /// gracefully from. + pub reachability: git_reachability::ArtifactBundle, } /// Resolves a [`RepoId`] to the backends that serve it.

crates/git-protocol/src/native/negotiate.rs @@ -2,15 +2,17 @@ //! same reachability walk ([`crate::walk`]) negotiation, push connectivity //! checking, and GC mark all share (`docs/scale-out.adoc`, "Reachability"). //! -//! This walks every object one at a time through -//! [`git_backend::ObjectStore::read`] — correct, not fast. A commit-graph -//! accelerator (WS6) is what turns this into the ranged, sublinear -//! negotiation the doc's Q6 calls out; this is the correctness-first -//! baseline it replaces. +//! Routed through [`git_reachability::engine::accelerated_reachable`] +//! (WS6): a commit-graph and, whenever a client's `haves` happen to equal a +//! server-known tip-frontier, a cached reachable-set snapshot both +//! accelerate this — absent either artifact, it is exactly the +//! correctness-first, one-object-at-a-time walk it always was. + +use git_reachability::engine::accelerated_reachable; use super::{BackendResolver, NativeBackend}; use crate::types::{NegotiationState, PackPlan}; -use crate::walk::{self, StoreSource}; +use crate::walk::StoreSource; use crate::{Negotiate, Result}; impl<R: BackendResolver> Negotiate for NativeBackend<R> { @@ -22,18 +24,24 @@ // not resend anything behind — tolerate a have the server never // actually had (a stale or misremembered claim) rather than fail // the whole negotiation over it. - let haves_closure = - walk::reachable(session.haves.iter().copied(), &source, |_id| false, true)?; + let haves_closure = accelerated_reachable( + session.haves.iter().copied(), + &source, + |_id| false, + true, + &backends.reachability, + )?; // Everything reachable from `wants`, not descending past the haves // boundary. A want neither the haves boundary nor the store itself // can resolve is a real negotiation failure, so this walk is // strict. - let wants_seen = walk::reachable( + let wants_seen = accelerated_reachable( session.wants.iter().copied(), &source, |id| haves_closure.contains(id), false, + &backends.reachability, )?; let objects = wants_seen.difference(&haves_closure).copied().collect();

crates/git-protocol/src/native/test_support.rs @@ -79,6 +79,9 @@ pub authorized_members: Vec<Member>, /// The config `authorized_members`' roles are checked against. pub config: git_ents_core::config::Config, + /// Reachability artifacts every test sees — empty by default, so tests + /// exercise the plain-walk fallback unless a test explicitly sets this. + pub reachability: git_reachability::ArtifactBundle, } impl FixedResolver { @@ -90,6 +93,7 @@ objects: Arc::new(odb_files::OdbFiles::open(path).unwrap()), authorized_members: Vec::new(), config: git_ents_core::config::Config::default(), + reachability: git_reachability::ArtifactBundle::empty(), } } } @@ -101,6 +105,7 @@ objects: self.objects.clone(), authorized_members: self.authorized_members.clone(), config: self.config.clone(), + reachability: self.reachability.clone(), }) } }

crates/odb-tigris/src/registry/memory.rs @@ -6,12 +6,13 @@ use git_backend::Result; -use super::{PackId, PackRecord, PackRegistry}; +use super::{ArtifactKind, ArtifactRecord, PackId, PackRecord, PackRegistry}; /// A [`PackRegistry`] held entirely in memory, scoped to one process. #[derive(Default)] pub struct InMemoryRegistry { records: Mutex<Vec<PackRecord>>, + artifacts: Mutex<Vec<ArtifactRecord>>, } impl InMemoryRegistry { @@ -40,6 +41,27 @@ lock(&self.records).retain(|record| !(record.repo_id == repo_id && &record.id == id)); Ok(()) } + + fn record_artifact(&self, record: ArtifactRecord) -> Result<()> { + let mut artifacts = lock(&self.artifacts); + artifacts.retain(|existing| { + !(existing.repo_id == record.repo_id && existing.kind == record.kind) + }); + artifacts.push(record); + Ok(()) + } + + fn get_artifact(&self, repo_id: &str, kind: ArtifactKind) -> Result<Option<ArtifactRecord>> { + Ok(lock(&self.artifacts) + .iter() + .find(|record| record.repo_id == repo_id && record.kind == kind) + .cloned()) + } + + fn delete_artifact(&self, repo_id: &str, kind: ArtifactKind) -> Result<()> { + lock(&self.artifacts).retain(|record| !(record.repo_id == repo_id && record.kind == kind)); + Ok(()) + } } /// Lock `mutex`, recovering the guard from a poisoned lock rather than

crates/git-reachability/Cargo.toml @@ -1,0 +1,23 @@ +[package] +name = "git-reachability" +version = "0.0.0" +edition.workspace = true +publish.workspace = true +license.workspace = true + +[dependencies] +git-backend = { workspace = true } +gix-hash = { workspace = true } +gix-object = { workspace = true } +odb-tigris = { workspace = true } +thiserror = { workspace = true } + +[dev-dependencies] +gix-actor = { workspace = true } +gix-date = { workspace = true } +odb-files = { workspace = true } +refstore-files = { workspace = true } +tempfile = { workspace = true } + +[lints] +workspace = true

crates/git-reachability/src/codec.rs @@ -1,0 +1,194 @@ +//! Hand-rolled binary (de)serialization shared by [`crate::commitgraph`] and +//! [`crate::reachable_set`] — no serde, no bincode (dependency policy): +//! length-prefixed where variable, fixed-width where not, every format +//! opening with a magic tag and a version byte so a future incompatible +//! change fails loudly (a wrong magic/version) rather than silently +//! misreading bytes. + +use gix_hash::ObjectId; + +use crate::{Error, Result}; + +/// An OID is always 20 bytes: every backend in this workspace pins SHA-1 +/// (`docs/scale-out.adoc` backends all say so explicitly), so these formats +/// do too rather than carry a hash-kind byte for a case that never occurs. +const OID_LEN: usize = 20; + +/// Append-only binary writer: a thin `Vec<u8>` wrapper naming the encoding +/// this module's readers expect. +#[derive(Default)] +pub struct Writer(Vec<u8>); + +impl Writer { + /// A fresh, empty writer. + #[must_use] + pub fn new() -> Self { + Self::default() + } + + /// Append `byte`. + pub fn u8(&mut self, byte: u8) { + self.0.push(byte); + } + + /// Append `value` as 4 little-endian bytes. + pub fn u32(&mut self, value: u32) { + self.0.extend_from_slice(&value.to_le_bytes()); + } + + /// Append `id`'s raw 20 bytes. + pub fn oid(&mut self, id: &ObjectId) { + self.0.extend_from_slice(id.as_slice()); + } + + /// Append `magic` verbatim, then `version` — every format's header. + pub fn header(&mut self, magic: &[u8; 4], version: u8) { + self.0.extend_from_slice(magic); + self.u8(version); + } + + /// Consume the writer, returning the bytes written so far. + #[must_use] + pub fn into_bytes(self) -> Vec<u8> { + self.0 + } +} + +/// A cursor over a byte slice, reading the primitives [`Writer`] writes and +/// erroring — never panicking or slicing out of bounds — on truncation. +pub struct Reader<'a> { + data: &'a [u8], + pos: usize, +} + +impl<'a> Reader<'a> { + /// A reader starting at the beginning of `data`. + #[must_use] + pub fn new(data: &'a [u8]) -> Self { + Self { data, pos: 0 } + } + + /// The next `len` bytes, advancing past them. + /// + /// # Errors + /// + /// Returns [`Error::Format`] if fewer than `len` bytes remain. + pub fn take(&mut self, len: usize) -> Result<&'a [u8]> { + let end = self + .pos + .checked_add(len) + .ok_or_else(|| Error::Format("length overflow while reading artifact".to_owned()))?; + let slice = self + .data + .get(self.pos..end) + .ok_or_else(|| Error::Format("artifact truncated".to_owned()))?; + self.pos = end; + Ok(slice) + } + + /// The next byte. + /// + /// # Errors + /// + /// Returns [`Error::Format`] if the reader is at the end of the data. + pub fn u8(&mut self) -> Result<u8> { + let byte = self + .take(1)? + .first() + .copied() + .ok_or_else(|| Error::Format("artifact truncated reading a byte".to_owned()))?; + Ok(byte) + } + + /// The next 4 bytes as a little-endian `u32`. + /// + /// # Errors + /// + /// Returns [`Error::Format`] if fewer than 4 bytes remain. + pub fn u32(&mut self) -> Result<u32> { + let bytes: [u8; 4] = self + .take(4)? + .try_into() + .map_err(|_error| Error::Format("artifact truncated reading a u32".to_owned()))?; + Ok(u32::from_le_bytes(bytes)) + } + + /// The next 20 bytes as an [`ObjectId`]. + /// + /// # Errors + /// + /// Returns [`Error::Format`] if fewer than 20 bytes remain. + pub fn oid(&mut self) -> Result<ObjectId> { + let bytes = self.take(OID_LEN)?; + ObjectId::try_from(bytes) + .map_err(|_error| Error::Format("artifact carried a malformed object id".to_owned())) + } + + /// Check and consume a header written by [`Writer::header`]. + /// + /// # Errors + /// + /// Returns [`Error::Format`] if the magic does not match or the version + /// is not exactly `expected_version` — this crate does not (yet) carry + /// more than one format version, so any mismatch is unreadable rather + /// than an upgrade to shim. + pub fn header(&mut self, magic: &[u8; 4], expected_version: u8) -> Result<()> { + let got_magic = self.take(4)?; + if got_magic != magic { + return Err(Error::Format(format!( + "unrecognized artifact magic {got_magic:02x?}" + ))); + } + let version = self.u8()?; + if version != expected_version { + return Err(Error::Format(format!( + "unsupported artifact version {version} (expected {expected_version})" + ))); + } + Ok(()) + } + + /// Whether every byte has been consumed. + #[must_use] + pub fn at_end(&self) -> bool { + self.pos >= self.data.len() + } +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, reason = "unit test")] + + use super::*; + + #[test] + fn round_trips_primitives() { + let mut writer = Writer::new(); + writer.header(b"TEST", 1); + writer.u32(42); + let id = ObjectId::from_hex(b"0123456789abcdef0123456789abcdef01234567").unwrap(); + writer.oid(&id); + let bytes = writer.into_bytes(); + + let mut reader = Reader::new(&bytes); + reader.header(b"TEST", 1).unwrap(); + assert_eq!(reader.u32().unwrap(), 42); + assert_eq!(reader.oid().unwrap(), id); + assert!(reader.at_end()); + } + + #[test] + fn rejects_a_truncated_buffer() { + let mut reader = Reader::new(&[1, 2, 3]); + let _error = reader.u32().unwrap_err(); + } + + #[test] + fn rejects_a_version_mismatch() { + let mut writer = Writer::new(); + writer.header(b"TEST", 2); + let bytes = writer.into_bytes(); + let mut reader = Reader::new(&bytes); + let _error = reader.header(b"TEST", 1).unwrap_err(); + } +}

crates/git-reachability/src/commitgraph.rs @@ -1,0 +1,488 @@ +//! [`CommitGraph`]: OID -> (tree, parents, generation number), the +//! commit-parent accelerator `docs/scale-out.adoc`'s "Reachability" section +//! calls for ("Maintenance effects generate commit-graph and reachability +//! bitmaps"). `gix-commitgraph` (survey, WS6/Q4) reads git's own +//! `commit-graph` file format but offers no writer, and the format itself is +//! a stock-git interop surface this crate doesn't need to match — artifacts +//! here are accelerators for the native backends, not stock-git interop +//! (`docs/scale-out.adoc`'s reachability artifacts are explicitly allowed to +//! be workspace-private), so this module defines its own minimal, versioned +//! binary format instead of a git-compatible one. +//! +//! # Format (version 1) +//! +//! ```text +//! magic "RGCG" (4 bytes) +//! version 1 (1 byte) +//! count u32 LE +//! oids count * 20 bytes, sorted ascending — the index table +//! `parents` below refers into +//! trees count * 20 bytes, `trees[i]` is `oids[i]`'s root tree +//! generations count * u32 LE, `generations[i]` is `oids[i]`'s generation +//! number (1 + max(parent generations), or 1 for a root commit) +//! parents count entries, each: parent_count (1 byte) then +//! parent_count * u32 LE indices into `oids` +//! ``` +//! +//! Parents are stored as indices into the sorted OID table (git's own +//! commit-graph format does the same) rather than as OIDs again: a lookup +//! by OID is one `binary_search` over `oids`, and every parent reference +//! costs 4 bytes instead of 20. + +use std::collections::BTreeMap; + +use gix_hash::ObjectId; + +use crate::codec::{Reader, Writer}; +use crate::walk::ObjectSource; +use crate::{Error, Result}; + +const MAGIC: &[u8; 4] = b"RGCG"; +const VERSION: u8 = 1; + +/// One commit's data as read back from a [`CommitGraph`]: its tree, +/// parents, and generation number, exactly what [`crate::walk::reachable`] +/// needs to descend a commit without ever reading it from the object store. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct GraphEntry { + /// The commit's root tree. + pub tree: ObjectId, + /// The commit's parents, in the order the commit object listed them. + pub parents: Vec<ObjectId>, + /// `1 + max(parent generations)`, or `1` for a commit with no parents. + pub generation: u32, +} + +/// A serialized commit graph: every commit reachable (via parent edges +/// alone) from the tips it was [`build`](CommitGraph::build) against, +/// mapped to its tree, parents, and generation number. +/// +/// Coverage is partial by construction whenever new commits have landed +/// since the graph was built — [`entry`](CommitGraph::entry) simply returns +/// `None` for anything it doesn't have, so a caller degrades to reading +/// that one commit from the object store instead of failing +/// (`docs/scale-out.adoc`: "absence or staleness degrades speed, never +/// answers"). +#[derive(Debug, Clone, Default)] +pub struct CommitGraph { + /// Sorted ascending; `oids.binary_search` is how every other field is + /// looked up by OID. + oids: Vec<ObjectId>, + trees: Vec<ObjectId>, + generations: Vec<u32>, + /// `parents[i]` holds indices into `oids` for `oids[i]`'s parents. + parents: Vec<Vec<u32>>, +} + +impl CommitGraph { + /// Build a commit graph covering every commit reachable from `tips` via + /// parent edges alone (trees/blobs are never visited — this is a + /// commit-only structure). A tip or ancestor `source` cannot resolve, or + /// that is not itself a commit, is simply not included rather than + /// treated as an error: this is a best-effort maintenance artifact, not + /// a correctness gate (`crate::maintenance::regenerate` is free to + /// re-run and does not block anything on this being complete). + /// + /// # Errors + /// + /// Returns an error if `source` fails outright (not: "doesn't have it"), + /// or if a resolved commit fails to decode. + pub fn build( + tips: impl IntoIterator<Item = ObjectId>, + source: &dyn ObjectSource, + ) -> Result<Self> { + let mut commits: BTreeMap<ObjectId, (ObjectId, Vec<ObjectId>)> = BTreeMap::new(); + let mut seen: std::collections::BTreeSet<ObjectId> = std::collections::BTreeSet::new(); + let mut stack: Vec<ObjectId> = tips.into_iter().collect(); + + while let Some(id) = stack.pop() { + if !seen.insert(id) { + continue; + } + let Some((kind, data)) = source.find(&id)? else { + continue; + }; + if kind != gix_object::Kind::Commit { + continue; + } + let commit = gix_object::CommitRef::from_bytes(&data, gix_hash::Kind::Sha1) + .map_err(|error| Error::Decode(error.to_string()))?; + let tree = commit.tree(); + let parent_oids: Vec<ObjectId> = commit.parents().collect(); + stack.extend(parent_oids.iter().copied()); + commits.insert(id, (tree, parent_oids)); + } + + let oids: Vec<ObjectId> = commits.keys().copied().collect(); + let mut trees = Vec::with_capacity(oids.len()); + let mut parent_oid_lists = Vec::with_capacity(oids.len()); + for id in &oids { + let (tree, parent_oids) = commits + .get(id) + .ok_or_else(|| Error::Format("commit graph build lost an entry".to_owned()))?; + trees.push(*tree); + parent_oid_lists.push(parent_oids.clone()); + } + + let parents: Vec<Vec<u32>> = parent_oid_lists + .iter() + .map(|parent_oids| { + parent_oids + .iter() + .filter_map(|parent| { + oids.binary_search(parent) + .ok() + .and_then(|index| u32::try_from(index).ok()) + }) + .collect() + }) + .collect(); + + let generations = compute_generations(oids.len(), &parents)?; + + Ok(Self { + oids, + trees, + generations, + parents, + }) + } + + /// `id`'s tree, parents, and generation number, or `None` if this graph + /// does not cover `id`. + #[must_use] + pub fn entry(&self, id: &ObjectId) -> Option<GraphEntry> { + let index = self.oids.binary_search(id).ok()?; + let tree = *self.trees.get(index)?; + let generation = *self.generations.get(index)?; + let parent_indices = self.parents.get(index)?; + let parents = parent_indices + .iter() + .filter_map(|&parent_index| self.oids.get(usize::try_from(parent_index).ok()?).copied()) + .collect(); + Some(GraphEntry { + tree, + parents, + generation, + }) + } + + /// `id`'s generation number alone, or `None` if this graph does not + /// cover `id`. + #[must_use] + pub fn generation(&self, id: &ObjectId) -> Option<u32> { + let index = self.oids.binary_search(id).ok()?; + self.generations.get(index).copied() + } + + /// How many commits this graph covers. + #[must_use] + pub fn len(&self) -> usize { + self.oids.len() + } + + /// Whether this graph covers no commits at all. + #[must_use] + pub fn is_empty(&self) -> bool { + self.oids.is_empty() + } + + /// Serialize to this module's binary format (version 1). + #[must_use] + pub fn serialize(&self) -> Vec<u8> { + let mut writer = Writer::new(); + writer.header(MAGIC, VERSION); + let count = u32::try_from(self.oids.len()).unwrap_or(u32::MAX); + writer.u32(count); + for oid in &self.oids { + writer.oid(oid); + } + for tree in &self.trees { + writer.oid(tree); + } + for generation in &self.generations { + writer.u32(*generation); + } + for parent_indices in &self.parents { + let parent_count = u8::try_from(parent_indices.len()).unwrap_or(u8::MAX); + writer.u8(parent_count); + for &parent_index in parent_indices.iter().take(usize::from(parent_count)) { + writer.u32(parent_index); + } + } + writer.into_bytes() + } + + /// Parse this module's binary format back. + /// + /// # Errors + /// + /// Returns [`Error::Format`] if the header, length, or any entry is + /// malformed or truncated. + pub fn deserialize(bytes: &[u8]) -> Result<Self> { + let mut reader = Reader::new(bytes); + reader.header(MAGIC, VERSION)?; + let count = usize::try_from(reader.u32()?) + .map_err(|_error| Error::Format("commit graph count overflowed usize".to_owned()))?; + + let mut oids = Vec::with_capacity(count); + for _ in 0..count { + oids.push(reader.oid()?); + } + let mut trees = Vec::with_capacity(count); + for _ in 0..count { + trees.push(reader.oid()?); + } + let mut generations = Vec::with_capacity(count); + for _ in 0..count { + generations.push(reader.u32()?); + } + let mut parents = Vec::with_capacity(count); + for _ in 0..count { + let parent_count = reader.u8()?; + let mut indices = Vec::with_capacity(usize::from(parent_count)); + for _ in 0..parent_count { + let index = reader.u32()?; + let in_range = usize::try_from(index).is_ok_and(|index| index < count); + if !in_range { + return Err(Error::Format( + "commit graph parent index out of range".to_owned(), + )); + } + indices.push(index); + } + parents.push(indices); + } + + if !reader.at_end() { + return Err(Error::Format( + "trailing bytes after commit graph artifact".to_owned(), + )); + } + + Ok(Self { + oids, + trees, + generations, + parents, + }) + } +} + +/// Compute every commit's generation number: `1 + max(parent generations)`, +/// or `1` for a commit with no parents. `parents[i]` (indices into a virtual +/// `0..n` id space) must reference only indices `< n` — [`CommitGraph:: +/// build`] guarantees this by construction (every index came from a +/// successful `binary_search` into the same table). +/// +/// Iterative post-order DFS rather than recursion: a commit history can be +/// far deeper than Rust's default stack tolerates, and this workspace's +/// lints forbid the indexing/unwrapping a naive recursive version would +/// otherwise reach for just as much as this iterative one avoids. +fn compute_generations(n: usize, parents: &[Vec<u32>]) -> Result<Vec<u32>> { + let mut generation: Vec<u32> = vec![0; n]; + let mut done: Vec<bool> = vec![false; n]; + + for start in 0..n { + if *done + .get(start) + .ok_or_else(|| Error::Format("generation computation index out of range".to_owned()))? + { + continue; + } + // Each stack frame is (node index, how many of its parents this + // frame has already pushed for processing). + let mut stack: Vec<(usize, usize)> = vec![(start, 0)]; + while let Some(&(index, next_parent)) = stack.last() { + let node_parents = parents.get(index).ok_or_else(|| { + Error::Format("generation computation index out of range".to_owned()) + })?; + + if let Some(&parent_index) = node_parents.get(next_parent) { + if let Some(frame) = stack.last_mut() { + frame.1 = next_parent.saturating_add(1); + } + let parent_index = usize::try_from(parent_index).map_err(|_error| { + Error::Format("generation computation index overflowed usize".to_owned()) + })?; + if !*done.get(parent_index).ok_or_else(|| { + Error::Format("generation computation index out of range".to_owned()) + })? { + stack.push((parent_index, 0)); + } + continue; + } + + // Every parent has already been assigned a generation. + let mut max_parent_generation = 0u32; + for &parent_index in node_parents { + let parent_index = usize::try_from(parent_index).map_err(|_error| { + Error::Format("generation computation index overflowed usize".to_owned()) + })?; + let parent_generation = *generation.get(parent_index).ok_or_else(|| { + Error::Format("generation computation index out of range".to_owned()) + })?; + max_parent_generation = max_parent_generation.max(parent_generation); + } + let this_generation = if node_parents.is_empty() { + 1 + } else { + max_parent_generation + .checked_add(1) + .ok_or_else(|| Error::Format("generation number overflowed u32".to_owned()))? + }; + let generation_slot = generation.get_mut(index).ok_or_else(|| { + Error::Format("generation computation index out of range".to_owned()) + })?; + *generation_slot = this_generation; + let done_slot = done.get_mut(index).ok_or_else(|| { + Error::Format("generation computation index out of range".to_owned()) + })?; + *done_slot = true; + stack.pop(); + } + } + + Ok(generation) +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, reason = "unit test")] + + use std::collections::HashMap; + + use gix_object::Kind; + use gix_object::WriteTo as _; + + use super::*; + + /// An in-memory [`ObjectSource`] over a fixed set of commits, keyed by + /// id, for building [`CommitGraph`]s in tests without a real repo. + struct FakeCommits(HashMap<ObjectId, Vec<u8>>); + + impl ObjectSource for FakeCommits { + fn find(&self, id: &ObjectId) -> Result<Option<(Kind, Vec<u8>)>> { + Ok(self.0.get(id).map(|data| (Kind::Commit, data.clone()))) + } + } + + fn commit_bytes(tree: ObjectId, parents: &[ObjectId]) -> Vec<u8> { + let identity = gix_actor::Signature { + name: "test".into(), + email: "test@example.com".into(), + time: gix_date::Time::default(), + }; + let commit = gix_object::Commit { + tree, + parents: parents.iter().copied().collect(), + author: identity.clone(), + committer: identity, + encoding: None, + message: "message".into(), + extra_headers: Vec::new(), + }; + let mut buf = Vec::new(); + commit.write_to(&mut buf).unwrap(); + buf + } + + fn oid(byte: u8) -> ObjectId { + let mut bytes = [0_u8; 20]; + if let Some(last) = bytes.last_mut() { + *last = byte; + } + ObjectId::from(bytes) + } + + #[test] + fn generation_is_one_for_a_root_commit() { + let tree = oid(0xAA); + let root = oid(1); + let mut commits = HashMap::new(); + commits.insert(root, commit_bytes(tree, &[])); + let source = FakeCommits(commits); + + let graph = CommitGraph::build([root], &source).unwrap(); + assert_eq!(graph.generation(&root), Some(1)); + let entry = graph.entry(&root).unwrap(); + assert_eq!(entry.tree, tree); + assert!(entry.parents.is_empty()); + } + + #[test] + fn generation_increases_along_a_chain() { + let tree = oid(0xAA); + let root = oid(1); + let child = oid(2); + let grandchild = oid(3); + let mut commits = HashMap::new(); + commits.insert(root, commit_bytes(tree, &[])); + commits.insert(child, commit_bytes(tree, &[root])); + commits.insert(grandchild, commit_bytes(tree, &[child])); + let source = FakeCommits(commits); + + let graph = CommitGraph::build([grandchild], &source).unwrap(); + assert_eq!(graph.generation(&root), Some(1)); + assert_eq!(graph.generation(&child), Some(2)); + assert_eq!(graph.generation(&grandchild), Some(3)); + } + + #[test] + fn generation_at_a_merge_is_one_plus_the_max_parent() { + let tree = oid(0xAA); + let root = oid(1); + let left = oid(2); + let right_chain_a = oid(3); + let right_chain_b = oid(4); + let merge = oid(5); + let mut commits = HashMap::new(); + commits.insert(root, commit_bytes(tree, &[])); + commits.insert(left, commit_bytes(tree, &[root])); + commits.insert(right_chain_a, commit_bytes(tree, &[root])); + commits.insert(right_chain_b, commit_bytes(tree, &[right_chain_a])); + commits.insert(merge, commit_bytes(tree, &[left, right_chain_b])); + let source = FakeCommits(commits); + + let graph = CommitGraph::build([merge], &source).unwrap(); + assert_eq!(graph.generation(&left), Some(2)); + assert_eq!(graph.generation(&right_chain_b), Some(3)); + assert_eq!(graph.generation(&merge), Some(4)); + } + + #[test] + fn entry_is_none_for_an_uncovered_commit() { + let tree = oid(0xAA); + let root = oid(1); + let mut commits = HashMap::new(); + commits.insert(root, commit_bytes(tree, &[])); + let source = FakeCommits(commits); + + let graph = CommitGraph::build([root], &source).unwrap(); + assert_eq!(graph.entry(&oid(99)), None); + } + + #[test] + fn round_trips_through_serialize_and_deserialize() { + let tree = oid(0xAA); + let root = oid(1); + let child = oid(2); + let mut commits = HashMap::new(); + commits.insert(root, commit_bytes(tree, &[])); + commits.insert(child, commit_bytes(tree, &[root])); + let source = FakeCommits(commits); + + let graph = CommitGraph::build([child], &source).unwrap(); + let bytes = graph.serialize(); + let read_back = CommitGraph::deserialize(&bytes).unwrap(); + + assert_eq!(read_back.entry(&root), graph.entry(&root)); + assert_eq!(read_back.entry(&child), graph.entry(&child)); + assert_eq!(read_back.len(), graph.len()); + } + + #[test] + fn deserialize_rejects_garbage() { + let _error = CommitGraph::deserialize(b"not a commit graph").unwrap_err(); + } +}

crates/git-reachability/src/engine.rs @@ -1,0 +1,164 @@ +//! [`accelerated_reachable`]: the entry point negotiation, ingest +//! connectivity, and GC mark all call instead of [`crate::walk::reachable`] +//! directly (`docs/scale-out.adoc`, "Reachability"). +//! +//! Two independent accelerations, either or both possibly absent: +//! +//! - An exact tip-frontier match against a cached +//! [`crate::reachable_set::ReachableSetArtifact`] answers instantly, no +//! walk at all — see that module's docs for why exact match is the right +//! bar rather than a more general (and more expensive to verify) ancestor +//! check. +//! - Otherwise, [`crate::walk::reachable_with_graph`] still benefits from a +//! [`crate::commitgraph::CommitGraph`] wherever it covers the walk's +//! commits, and degrades to a plain [`crate::walk::reachable`] wherever it +//! doesn't. + +use std::collections::BTreeSet; + +use gix_hash::ObjectId; + +use crate::Result; +use crate::commitgraph::CommitGraph; +use crate::reachable_set::ReachableSetArtifact; +use crate::walk::{self, ObjectSource}; + +/// The reachability artifacts currently available for one repository — +/// possibly neither, in which case [`accelerated_reachable`] is exactly +/// [`crate::walk::reachable`] (`docs/scale-out.adoc`'s "absence ... +/// degrades speed, never answers"). +#[derive(Debug, Clone, Default)] +pub struct ArtifactBundle { + /// The commit-parent accelerator, if generated. + pub commit_graph: Option<CommitGraph>, + /// The tip-frontier reachable-set snapshot, if generated. + pub reachable_set: Option<ReachableSetArtifact>, +} + +impl ArtifactBundle { + /// No artifacts at all — every consumer using this degrades fully to + /// the slow walk. The default for a repo whose maintenance effect has + /// never run, and for any backend that hasn't wired artifact loading + /// yet. + #[must_use] + pub fn empty() -> Self { + Self::default() + } +} + +/// [`crate::walk::reachable`], accelerated by whatever `artifacts` holds. +/// +/// # Errors +/// +/// Returns an error under the same conditions as +/// [`crate::walk::reachable_with_graph`]. +pub fn accelerated_reachable( + roots: impl IntoIterator<Item = ObjectId>, + source: &dyn ObjectSource, + stop: impl FnMut(&ObjectId) -> bool, + lenient: bool, + artifacts: &ArtifactBundle, +) -> Result<BTreeSet<ObjectId>> { + let roots: Vec<ObjectId> = roots.into_iter().collect(); + + if let Some(set) = &artifacts.reachable_set { + let root_set: BTreeSet<ObjectId> = roots.iter().copied().collect(); + if set.frontier == root_set { + return Ok(set.objects.clone()); + } + } + + walk::reachable_with_graph( + roots, + source, + stop, + lenient, + artifacts.commit_graph.as_ref(), + ) +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, reason = "unit test")] + + use std::collections::HashMap; + + use gix_object::Kind; + + use super::*; + + struct FakeBlobs(HashMap<ObjectId, (Kind, Vec<u8>)>); + + impl ObjectSource for FakeBlobs { + fn find(&self, id: &ObjectId) -> Result<Option<(Kind, Vec<u8>)>> { + Ok(self.0.get(id).cloned()) + } + } + + fn oid(byte: u8) -> ObjectId { + let mut bytes = [0_u8; 20]; + if let Some(last) = bytes.last_mut() { + *last = byte; + } + ObjectId::from(bytes) + } + + #[test] + fn exact_frontier_match_short_circuits_the_walk() { + let blob = oid(1); + // No entry for `blob` in the source at all: if the fast path didn't + // fire, this would error rather than return the cached set. + let source = FakeBlobs(HashMap::new()); + let cached = ReachableSetArtifact { + frontier: BTreeSet::from([blob]), + objects: BTreeSet::from([blob]), + }; + let artifacts = ArtifactBundle { + commit_graph: None, + reachable_set: Some(cached), + }; + + let result = + accelerated_reachable([blob], &source, |_id| false, false, &artifacts).unwrap(); + assert_eq!(result, BTreeSet::from([blob])); + } + + #[test] + fn a_frontier_mismatch_falls_back_to_a_real_walk() { + let blob = oid(1); + let other = oid(2); + let mut objects = HashMap::new(); + objects.insert(other, (Kind::Blob, b"content".to_vec())); + let source = FakeBlobs(objects); + let cached = ReachableSetArtifact { + frontier: BTreeSet::from([blob]), + objects: BTreeSet::from([blob]), + }; + let artifacts = ArtifactBundle { + commit_graph: None, + reachable_set: Some(cached), + }; + + let result = + accelerated_reachable([other], &source, |_id| false, false, &artifacts).unwrap(); + assert_eq!(result, BTreeSet::from([other])); + } + + #[test] + fn no_artifacts_at_all_behaves_like_the_plain_walk() { + let blob = oid(1); + let mut objects = HashMap::new(); + objects.insert(blob, (Kind::Blob, b"content".to_vec())); + let source = FakeBlobs(objects); + + let result = accelerated_reachable( + [blob], + &source, + |_id| false, + false, + &ArtifactBundle::empty(), + ) + .unwrap(); + assert_eq!(result, BTreeSet::from([blob])); + } +}

crates/git-reachability/src/lib.rs @@ -1,0 +1,121 @@ +//! WS6: the reachability subsystem (`docs/scale-out.adoc`, "Reachability"). +//! +//! > Negotiation, push connectivity checking, and GC mark are the same +//! > walk, and over a remote ODB that walk is the scaling wall: nothing may +//! > traverse Tigris object-by-object. +//! +//! This crate owns that shared walk ([`walk`], moved here from +//! `git-protocol`) plus the two accelerators that make it cheap at scale: +//! +//! - [`commitgraph::CommitGraph`] — commit OID -> (tree, parents, +//! generation number), so a commit-parent traversal never has to round +//! trip through [`git_backend::ObjectStore::read`] for a commit the graph +//! already covers. This is the accelerator every walk benefits from +//! regardless of which tips it starts from. +//! - [`reachable_set::ReachableSetArtifact`] — the full transitive object +//! closure (commits, trees, blobs, tags) from one exact tip-frontier +//! snapshot. Cheap to produce (one walk, at maintenance time) and, when a +//! query's roots exactly match the frontier it was built from, cheap to +//! consume: the cached closure *is* the answer, no walk at all. This is +//! GC mark's steady-state case (roots = current ref tips, nothing moved +//! since the last regeneration) and, whenever a client's `haves` happen to +//! equal a server-known frontier, negotiation's. +//! +//! [`engine`] wires both into [`engine::accelerated_reachable`], the single +//! entry point [`native::negotiate`](../git_protocol/native/negotiate)-style +//! consumers call instead of the raw [`walk::reachable`]. [`store`] persists +//! artifacts beside packs, tracked in the pack registry +//! (`odb_tigris::registry::PackRegistry`). [`maintenance`] is the effect +//! that (re)generates them. +//! +//! # Correctness property +//! +//! > absence [or staleness] degrades speed, never answers +//! +//! [`commitgraph::CommitGraph::entry`] returns `None` for any commit it +//! doesn't cover — the walk falls back to an ordinary `ObjectStore` read for +//! exactly that commit, nothing more. [`engine::accelerated_reachable`]'s +//! whole-frontier fast path only ever fires on an *exact* set match between +//! the cached frontier and the query's roots; anything else — including a +//! frontier that is a strict ancestor of the current roots after new +//! commits landed — falls through to a full walk (itself still +//! commit-graph-accelerated wherever the graph covers it). Neither path can +//! ever produce a smaller-than-correct answer: the exact-match path returns +//! exactly what a from-scratch walk from those same roots would have +//! produced (that is what building the artifact ran), and the graph path +//! only ever substitutes a stored decode for an identical live one. +//! +//! # gc_mark +//! +//! [`gc_mark`] is WS9's entry point: the reachable set from every current +//! ref tip. GC itself (mark-and-sweep scheduling, cruft handling) is WS9's +//! job; this crate only proves out and tests the "mark" half. + +mod codec; +pub mod commitgraph; +pub mod engine; +pub mod maintenance; +pub mod reachable_set; +pub mod store; +pub mod walk; + +use std::collections::BTreeSet; + +use git_backend::{RefName, RefStore}; +use gix_hash::ObjectId; + +pub use engine::{ArtifactBundle, accelerated_reachable}; + +/// A failure in this crate's artifact formats or the reachability walk +/// itself. +#[derive(Debug, thiserror::Error)] +pub enum Error { + /// The underlying storage traits reported a failure. + #[error(transparent)] + Backend(#[from] git_backend::Error), + /// A reachability walk found an object neither the graph nor `source` + /// could resolve, and the walk was not marked lenient. + #[error("missing object {0}")] + MissingObject(ObjectId), + /// Decoding a commit, tree, or tag object failed. + #[error("could not decode object: {0}")] + Decode(String), + /// A serialized artifact was truncated, carried an unsupported version, + /// or was otherwise malformed. + #[error("malformed reachability artifact: {0}")] + Format(String), +} + +/// This crate's `Result` alias. +pub type Result<T> = std::result::Result<T, Error>; + +/// Every ref's current tip in `refs` — the tip-frontier +/// [`maintenance::regenerate`] and [`gc_mark`] both walk from. +/// +/// # Errors +/// +/// Returns an error if the ref store cannot be read. +pub fn ref_tips(refs: &dyn RefStore) -> Result<Vec<ObjectId>> { + refs.iter_prefix(&RefName::new("refs/"))? + .map(|entry| entry.map(|(_name, oid)| oid).map_err(Error::from)) + .collect() +} + +/// WS9's entry point: the set of every object reachable from `refs`' +/// current tips over `objects`, accelerated by whatever `artifacts` this +/// repo currently has (possibly none — see the module docs' correctness +/// property). +/// +/// # Errors +/// +/// Returns an error if the ref or object store cannot be read, or if the +/// walk finds a ref tip whose history is incomplete in `objects`. +pub fn gc_mark( + refs: &dyn RefStore, + objects: &dyn git_backend::ObjectStore, + artifacts: &ArtifactBundle, +) -> Result<BTreeSet<ObjectId>> { + let tips = ref_tips(refs)?; + let source = walk::StoreSource::new(objects); + engine::accelerated_reachable(tips, &source, |_id| false, false, artifacts) +}

crates/git-reachability/src/maintenance.rs @@ -1,0 +1,188 @@ +//! `reachability-maintenance`: the effect that (re)generates a repo's +//! commit-graph and reachable-set artifacts (`docs/scale-out.adoc`, +//! "Reachability": "Maintenance effects generate commit-graph and +//! reachability bitmaps", "Regeneration is scheduled with repack (WS9) and +//! triggered by ref-update volume thresholds"). +//! +//! Follows `git-effect`'s definition/execution split at the seam that +//! already exists for it: [`git_backend::EffectDef`] is the same static +//! "what to spawn" shape `git_effect::Effect` mirrors for user-configured +//! push effects (`refs/meta/effects/*`); [`definition`] returns one for this +//! effect. Unlike those, `reachability-maintenance` is not user-configured +//! or pushable — there is no `refs/meta/effects/reachability-maintenance` +//! ref, and [`regenerate`] is plain in-process maintenance code, not a +//! sandboxed shell command, so [`git_backend::EffectDef::command`] is +//! `None`. +//! +//! Scheduling — deciding *when* an `EffectExecutor` (WS7) actually spawns +//! this effect, e.g. alongside repack or on a timer — is WS9's job. This +//! module only supplies the trigger predicate ([`should_regenerate`]) and +//! the effect body ([`regenerate`]) a future scheduler calls. + +use git_backend::{EffectDef, ObjectStore, RefStore}; +use odb_tigris::registry::{ArtifactKind, PackRegistry}; +use odb_tigris::transport::BlobTransport; + +use crate::commitgraph::CommitGraph; +use crate::reachable_set::ReachableSetArtifact; +use crate::walk::StoreSource; +use crate::{Result, ref_tips, store}; + +/// The name this effect is identified by wherever an [`EffectDef`] needs +/// one — distinct from any `refs/meta/effects/*` name, which names a +/// user-configured push effect instead. +pub const EFFECT_NAME: &str = "reachability-maintenance"; + +/// The static [`EffectDef`] an `EffectExecutor` (WS7) spawns to run +/// [`regenerate`]. `command` and `image` are `None`: this effect runs as +/// in-process maintenance code, never a sandboxed shell command — those +/// fields exist on [`EffectDef`] for the general case, not because this +/// effect needs them. +#[must_use] +pub fn definition() -> EffectDef { + EffectDef { + name: EFFECT_NAME.to_owned(), + command: None, + image: None, + } +} + +/// Whether enough ref-update volume has accumulated since the last +/// regeneration to warrant running this effect again — the trigger +/// `docs/scale-out.adoc` calls for ("triggered by ref-update volume +/// thresholds"). Pure and total: a scheduler (WS9) is responsible for +/// tracking `ref_updates_since_last` and deciding when to actually spawn +/// the effect; this function only answers the yes/no question. +#[must_use] +pub fn should_regenerate(ref_updates_since_last: u64, threshold: u64) -> bool { + ref_updates_since_last >= threshold +} + +/// Regenerate `repo_id`'s commit-graph and reachable-set artifacts from +/// `refs`'s current tips over `objects`, storing both via `transport`/ +/// `registry` (`docs/scale-out.adoc`, "Reachability"). Replaces whatever was +/// previously registered for each kind (`PackRegistry::record_artifact`) — +/// regeneration is a full recompute, not an incremental update. +/// +/// # Errors +/// +/// Returns an error if the ref store or object store cannot be read, if +/// building either artifact fails, or if storing either fails. +pub fn regenerate( + repo_id: &str, + refs: &dyn RefStore, + objects: &dyn ObjectStore, + transport: &dyn BlobTransport, + registry: &dyn PackRegistry, +) -> Result<()> { + let tips = ref_tips(refs)?; + let source = StoreSource::new(objects); + + let graph = CommitGraph::build(tips.iter().copied(), &source)?; + store::store_artifact( + transport, + registry, + repo_id, + ArtifactKind::CommitGraph, + graph.serialize(), + )?; + + let reachable = ReachableSetArtifact::build(tips, &source)?; + store::store_artifact( + transport, + registry, + repo_id, + ArtifactKind::ReachableSet, + reachable.serialize(), + )?; + + Ok(()) +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, reason = "unit test")] + + use odb_tigris::registry::memory::InMemoryRegistry; + use odb_tigris::transport::fs::FsTransport; + + use super::*; + + #[test] + fn should_regenerate_trips_at_the_threshold() { + assert!(!should_regenerate(4, 5)); + assert!(should_regenerate(5, 5)); + assert!(should_regenerate(6, 5)); + } + + #[test] + fn definition_names_the_effect_with_no_shell_command() { + let def = definition(); + assert_eq!(def.name, EFFECT_NAME); + assert!(def.command.is_none()); + assert!(def.image.is_none()); + } + + #[test] + fn regenerate_stores_both_artifacts_from_a_real_repo() { + let bare = tempfile::tempdir().unwrap(); + let status = std::process::Command::new("git") + .args(["init", "-q", "--bare", "-b", "main"]) + .arg(bare.path()) + .status() + .unwrap(); + assert!(status.success()); + + let work = tempfile::tempdir().unwrap(); + for args in [ + &["init", "-q", "-b", "main"][..], + &["config", "user.email", "test@example.com"], + &["config", "user.name", "test"], + ] { + assert!( + std::process::Command::new("git") + .arg("-C") + .arg(work.path()) + .args(args) + .status() + .unwrap() + .success() + ); + } + std::fs::write(work.path().join("file"), "content").unwrap(); + for args in [ + &["add", "-A"][..], + &["commit", "-q", "-m", "commit"], + &[ + "push", + bare.path().to_str().unwrap(), + "main:refs/heads/main", + ], + ] { + assert!( + std::process::Command::new("git") + .arg("-C") + .arg(work.path()) + .args(args) + .status() + .unwrap() + .success() + ); + } + + let refs = refstore_files::FilesRefStore::open(bare.path()).unwrap(); + let objects = odb_files::OdbFiles::open(bare.path()).unwrap(); + let artifact_dir = tempfile::tempdir().unwrap(); + let transport = FsTransport::open(artifact_dir.path()).unwrap(); + let registry = InMemoryRegistry::new(); + + regenerate("repo", &refs, &objects, &transport, &registry).unwrap(); + + let bundle = store::load_bundle(&transport, &registry, "repo").unwrap(); + assert!(bundle.commit_graph.is_some()); + assert!(bundle.reachable_set.is_some()); + let reachable = bundle.reachable_set.unwrap(); + // commit + tree + blob. + assert_eq!(reachable.objects.len(), 3); + } +}

crates/git-reachability/src/reachable_set.rs @@ -1,0 +1,184 @@ +//! [`ReachableSetArtifact`]: a full reachable-object-set snapshot for one +//! exact tip-frontier (`docs/scale-out.adoc`, "Reachability": "reachability +//! bitmaps" — this crate's equivalent, in its own hand-rolled format rather +//! than git's bitmap-index format, per this crate's module docs on why). +//! +//! One snapshot per `(repo_id, kind)` is kept (see +//! `odb_tigris::registry::PackRegistry`) — regenerating replaces it rather +//! than accumulating a history of frontiers. This is deliberately the +//! simplest artifact that helps: GC mark's roots are *always* "every current +//! ref tip", so between two maintenance runs with no intervening ref update +//! the frontier this snapshot was built from and GC mark's query roots are +//! identical, and [`crate::engine::accelerated_reachable`]'s exact-match +//! fast path returns the cached set with no walk at all. The same applies to +//! negotiation whenever a client's `haves` happen to equal a server-known +//! frontier (e.g. the tips as of its last fetch). Any other roots — a +//! frontier from before the most recent push, say — simply miss the fast +//! path and fall through to a full (still commit-graph-accelerated where +//! covered) walk: never a wrong answer, only a slower one. +//! +//! # Format (version 1) +//! +//! ```text +//! magic "RGRS" (4 bytes) +//! version 1 (1 byte) +//! frontier_count u32 LE +//! frontier frontier_count * 20 bytes, sorted ascending +//! object_count u32 LE +//! objects object_count * 20 bytes, sorted ascending +//! ``` + +use std::collections::BTreeSet; + +use gix_hash::ObjectId; + +use crate::Result; +use crate::codec::{Reader, Writer}; +use crate::walk::{self, ObjectSource}; + +const MAGIC: &[u8; 4] = b"RGRS"; +const VERSION: u8 = 1; + +/// A snapshot of every object reachable from one exact tip-frontier. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ReachableSetArtifact { + /// The exact set of tips this snapshot was computed from — the + /// [`crate::engine::accelerated_reachable`] fast path only applies when + /// a query's roots equal this set exactly. + pub frontier: BTreeSet<ObjectId>, + /// Every object (commits, trees, blobs, tags) reachable from + /// `frontier`. + pub objects: BTreeSet<ObjectId>, +} + +impl ReachableSetArtifact { + /// Compute the full reachable-object-set snapshot for `tips` over + /// `source` — a plain, unaccelerated walk (this *is* how the + /// accelerator gets built) with no `stop` boundary, since the whole + /// point is a complete closure to cache. + /// + /// # Errors + /// + /// Returns an error if the walk finds a tip or ancestor `source` cannot + /// resolve — a real inconsistency, not tolerated here the way a + /// client's possibly-stale `have` is elsewhere. + pub fn build( + tips: impl IntoIterator<Item = ObjectId>, + source: &dyn ObjectSource, + ) -> Result<Self> { + let frontier: BTreeSet<ObjectId> = tips.into_iter().collect(); + let objects = walk::reachable(frontier.iter().copied(), source, |_id| false, false)?; + Ok(Self { frontier, objects }) + } + + /// Serialize to this module's binary format (version 1). + #[must_use] + pub fn serialize(&self) -> Vec<u8> { + let mut writer = Writer::new(); + writer.header(MAGIC, VERSION); + write_oid_set(&mut writer, &self.frontier); + write_oid_set(&mut writer, &self.objects); + writer.into_bytes() + } + + /// Parse this module's binary format back. + /// + /// # Errors + /// + /// Returns [`Error::Format`] if the header, length, or any entry is + /// malformed or truncated. + pub fn deserialize(bytes: &[u8]) -> Result<Self> { + let mut reader = Reader::new(bytes); + reader.header(MAGIC, VERSION)?; + let frontier = read_oid_set(&mut reader)?; + let objects = read_oid_set(&mut reader)?; + if !reader.at_end() { + return Err(crate::Error::Format( + "trailing bytes after reachable-set artifact".to_owned(), + )); + } + Ok(Self { frontier, objects }) + } +} + +fn write_oid_set(writer: &mut Writer, set: &BTreeSet<ObjectId>) { + let count = u32::try_from(set.len()).unwrap_or(u32::MAX); + writer.u32(count); + for id in set { + writer.oid(id); + } +} + +fn read_oid_set(reader: &mut Reader<'_>) -> Result<BTreeSet<ObjectId>> { + let count = reader.u32()?; + let mut set = BTreeSet::new(); + for _ in 0..count { + set.insert(reader.oid()?); + } + Ok(set) +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, reason = "unit test")] + + use std::collections::HashMap; + + use gix_object::Kind; + + use super::*; + + struct FakeBlobs(HashMap<ObjectId, (Kind, Vec<u8>)>); + + impl ObjectSource for FakeBlobs { + fn find(&self, id: &ObjectId) -> Result<Option<(Kind, Vec<u8>)>> { + Ok(self.0.get(id).cloned()) + } + } + + fn oid(byte: u8) -> ObjectId { + let mut bytes = [0_u8; 20]; + if let Some(last) = bytes.last_mut() { + *last = byte; + } + ObjectId::from(bytes) + } + + #[test] + fn build_walks_a_single_blob_tip() { + let blob = oid(1); + let mut objects = HashMap::new(); + objects.insert(blob, (Kind::Blob, b"content".to_vec())); + let source = FakeBlobs(objects); + + let artifact = ReachableSetArtifact::build([blob], &source).unwrap(); + assert_eq!(artifact.frontier, BTreeSet::from([blob])); + assert_eq!(artifact.objects, BTreeSet::from([blob])); + } + + #[test] + fn round_trips_through_serialize_and_deserialize() { + let a = oid(1); + let b = oid(2); + let mut objects = HashMap::new(); + objects.insert(a, (Kind::Blob, b"a".to_vec())); + objects.insert(b, (Kind::Blob, b"b".to_vec())); + let source = FakeBlobs(objects); + + let artifact = ReachableSetArtifact::build([a, b], &source).unwrap(); + let bytes = artifact.serialize(); + let read_back = ReachableSetArtifact::deserialize(&bytes).unwrap(); + assert_eq!(read_back, artifact); + } + + #[test] + fn deserialize_rejects_garbage() { + let _error = ReachableSetArtifact::deserialize(b"nope").unwrap_err(); + } + + #[test] + fn build_fails_on_a_missing_object() { + let source = FakeBlobs(HashMap::new()); + let _error = ReachableSetArtifact::build([oid(1)], &source).unwrap_err(); + } +}

crates/git-reachability/src/store.rs @@ -1,0 +1,177 @@ +//! Persisting reachability artifacts beside packs, tracked in the pack +//! registry (`docs/scale-out.adoc`, "Reachability": "stored beside packs, +//! tracked in the pack registry") — the same `BlobTransport` + +//! `PackRegistry` seam `odb-tigris` uses for packs themselves, reused here +//! rather than inventing a parallel storage path. + +use odb_tigris::registry::{ArtifactKind, ArtifactRecord, PackRegistry}; +use odb_tigris::transport::BlobTransport; + +use crate::commitgraph::CommitGraph; +use crate::engine::ArtifactBundle; +use crate::reachable_set::ReachableSetArtifact; +use crate::{Error, Result}; + +fn artifact_key(repo_id: &str, kind: ArtifactKind) -> String { + format!("{repo_id}/reachability/{}.bin", kind.as_str()) +} + +/// Write `bytes` as `repo_id`'s current artifact of `kind`, replacing +/// whatever was previously registered. +/// +/// # Errors +/// +/// Returns an error if the transport write or the registry record fails. +pub fn store_artifact( + transport: &dyn BlobTransport, + registry: &dyn PackRegistry, + repo_id: &str, + kind: ArtifactKind, + bytes: Vec<u8>, +) -> Result<()> { + let key = artifact_key(repo_id, kind); + transport.put(&key, bytes).map_err(Error::Backend)?; + registry + .record_artifact(ArtifactRecord { + repo_id: repo_id.to_owned(), + kind, + key, + }) + .map_err(Error::Backend) +} + +/// Load `repo_id`'s current artifact bytes of `kind`, or `None` if it has +/// never been generated. +/// +/// # Errors +/// +/// Returns an error if the registry or transport read fails. +pub fn load_artifact( + transport: &dyn BlobTransport, + registry: &dyn PackRegistry, + repo_id: &str, + kind: ArtifactKind, +) -> Result<Option<Vec<u8>>> { + let Some(record) = registry + .get_artifact(repo_id, kind) + .map_err(Error::Backend)? + else { + return Ok(None); + }; + let bytes = transport.get(&record.key).map_err(Error::Backend)?; + Ok(Some(bytes)) +} + +/// Remove `repo_id`'s artifact of `kind`, both its bucket bytes and its +/// registry record. Not an error if already absent. +/// +/// # Errors +/// +/// Returns an error if the transport or registry delete fails. +pub fn delete_artifact( + transport: &dyn BlobTransport, + registry: &dyn PackRegistry, + repo_id: &str, + kind: ArtifactKind, +) -> Result<()> { + if let Some(record) = registry + .get_artifact(repo_id, kind) + .map_err(Error::Backend)? + { + transport.delete(&record.key).map_err(Error::Backend)?; + } + registry + .delete_artifact(repo_id, kind) + .map_err(Error::Backend) +} + +/// Load and parse `repo_id`'s full [`ArtifactBundle`], degrading each +/// artifact independently to `None` when absent — never an error just +/// because one or both artifacts don't exist yet. +/// +/// # Errors +/// +/// Returns an error only if a *present* artifact fails to parse (corrupt, +/// or from an incompatible format version) — never for a missing one. +pub fn load_bundle( + transport: &dyn BlobTransport, + registry: &dyn PackRegistry, + repo_id: &str, +) -> Result<ArtifactBundle> { + let commit_graph = load_artifact(transport, registry, repo_id, ArtifactKind::CommitGraph)? + .map(|bytes| CommitGraph::deserialize(&bytes)) + .transpose()?; + let reachable_set = load_artifact(transport, registry, repo_id, ArtifactKind::ReachableSet)? + .map(|bytes| ReachableSetArtifact::deserialize(&bytes)) + .transpose()?; + Ok(ArtifactBundle { + commit_graph, + reachable_set, + }) +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, reason = "unit test")] + + use odb_tigris::registry::memory::InMemoryRegistry; + use odb_tigris::transport::fs::FsTransport; + + use super::*; + + #[test] + fn store_then_load_bundle_round_trips_both_artifacts() { + let dir = tempfile::tempdir().unwrap(); + let transport = FsTransport::open(dir.path()).unwrap(); + let registry = InMemoryRegistry::new(); + + let graph = CommitGraph::default(); + store_artifact( + &transport, + &registry, + "repo", + ArtifactKind::CommitGraph, + graph.serialize(), + ) + .unwrap(); + + let bundle = load_bundle(&transport, &registry, "repo").unwrap(); + assert!(bundle.commit_graph.is_some()); + assert!(bundle.reachable_set.is_none()); + } + + #[test] + fn load_bundle_is_empty_when_nothing_was_ever_generated() { + let dir = tempfile::tempdir().unwrap(); + let transport = FsTransport::open(dir.path()).unwrap(); + let registry = InMemoryRegistry::new(); + + let bundle = load_bundle(&transport, &registry, "repo").unwrap(); + assert!(bundle.commit_graph.is_none()); + assert!(bundle.reachable_set.is_none()); + } + + #[test] + fn delete_artifact_removes_it_from_both_transport_and_registry() { + let dir = tempfile::tempdir().unwrap(); + let transport = FsTransport::open(dir.path()).unwrap(); + let registry = InMemoryRegistry::new(); + + let graph = CommitGraph::default(); + store_artifact( + &transport, + &registry, + "repo", + ArtifactKind::CommitGraph, + graph.serialize(), + ) + .unwrap(); + delete_artifact(&transport, &registry, "repo", ArtifactKind::CommitGraph).unwrap(); + + assert!( + load_artifact(&transport, &registry, "repo", ArtifactKind::CommitGraph) + .unwrap() + .is_none() + ); + } +}

crates/git-reachability/src/walk.rs @@ -1,0 +1,157 @@ +//! A generic reachability walk over whatever [`ObjectSource`] answers +//! `find`, shared by negotiation, push connectivity checking, and GC mark +//! (`docs/scale-out.adoc`, "Reachability": "Negotiation, push connectivity +//! checking, and GC mark are the same walk"). +//! +//! Moved here from `git-protocol` (WS6): this crate is where the +//! accelerator lives ([`crate::commitgraph`], [`crate::reachable_set`]), so +//! the walk they accelerate belongs beside them rather than in the protocol +//! crate that merely calls it. `git-protocol` still gets `ObjectSource` +//! and `reachable` through its own `walk` module, now a thin re-export. +//! +//! [`reachable`] is the plain, one-object-at-a-time walk — correct, not +//! fast. [`reachable_with_graph`] is the same walk with one addition: for +//! any id a [`crate::commitgraph::CommitGraph`] covers, its tree and +//! parents come from the graph instead of an [`ObjectSource::find`] round +//! trip. `reachable` is exactly `reachable_with_graph` with `graph: None`, +//! so there is one walk implementation, not two kept in sync by hand. + +use std::collections::BTreeSet; + +use git_backend::ObjectStore; +use gix_hash::ObjectId; +use gix_object::{CommitRef, Kind, TagRef, TreeRefIter}; + +use crate::commitgraph::CommitGraph; +use crate::{Error, Result}; + +/// Where a walk reads object kind/data from. Lets the same walk run over a +/// repository's promoted object store alone (negotiation, GC mark) or a +/// promoted store combined with a not-yet-promoted incoming pack (ingest +/// connectivity checking). +pub trait ObjectSource { + /// The kind and raw content of `id`, or `None` if this source has never + /// heard of it. + fn find(&self, id: &ObjectId) -> Result<Option<(Kind, Vec<u8>)>>; +} + +/// An [`ObjectSource`] over a repository's promoted [`ObjectStore`] alone. +pub struct StoreSource<'a> { + store: &'a dyn ObjectStore, +} + +impl<'a> StoreSource<'a> { + /// Read only through `store`'s promoted view. + pub fn new(store: &'a dyn ObjectStore) -> Self { + Self { store } + } +} + +impl ObjectSource for StoreSource<'_> { + fn find(&self, id: &ObjectId) -> Result<Option<(Kind, Vec<u8>)>> { + if !self.store.contains(*id)? { + return Ok(None); + } + let object = self.store.read(*id)?; + Ok(Some((object.kind, object.data))) + } +} + +/// Walk every object reachable from `roots` via commit parents, commit/tag +/// targets, and tree entries (skipping gitlink/submodule entries, which name +/// a commit in a different repository's object space). +/// +/// `stop` marks a boundary: when it returns `true` for an id, that id is +/// recorded as seen but never resolved or descended into — the caller +/// already knows it (and everything under it) is accounted for, e.g. +/// negotiation's haves closure, or the ingest connectivity check's existing +/// history. +/// +/// When `lenient` is `false`, an id `stop` did not claim but `source` cannot +/// resolve is a connectivity failure ([`Error::MissingObject`]) — the ingest +/// check's use. When `true`, it is silently dropped instead — appropriate +/// for a client-supplied `have` the server never actually had, which is a +/// stale claim, not a corruption. +/// +/// Plain, correct, not fast — see [`reachable_with_graph`] for the +/// commit-graph-accelerated form this degrades from. +pub fn reachable( + roots: impl IntoIterator<Item = ObjectId>, + source: &dyn ObjectSource, + stop: impl FnMut(&ObjectId) -> bool, + lenient: bool, +) -> Result<BTreeSet<ObjectId>> { + reachable_with_graph(roots, source, stop, lenient, None) +} + +/// [`reachable`], accelerated by `graph` when given: for any id `graph` +/// covers ([`CommitGraph::entry`] returns `Some`), its tree and parents come +/// from the graph — no [`ObjectSource::find`] call, and so (for a remote +/// object store) no read against it at all — instead of decoding the +/// commit. Anything the graph doesn't cover (trees, blobs, tags, and any +/// commit a stale or partial graph is missing) resolves through `source` +/// exactly as [`reachable`] would. +/// +/// # Errors +/// +/// Returns an error if decoding a resolved commit, tree, or tag fails, or +/// (when `lenient` is `false`) if an id neither `stop` nor `graph` nor +/// `source` accounts for. +pub fn reachable_with_graph( + roots: impl IntoIterator<Item = ObjectId>, + source: &dyn ObjectSource, + mut stop: impl FnMut(&ObjectId) -> bool, + lenient: bool, + graph: Option<&CommitGraph>, +) -> Result<BTreeSet<ObjectId>> { + let mut seen = BTreeSet::new(); + let mut stack: Vec<ObjectId> = roots.into_iter().collect(); + while let Some(id) = stack.pop() { + if !seen.insert(id) { + continue; + } + if stop(&id) { + continue; + } + + if let Some(entry) = graph.and_then(|graph| graph.entry(&id)) { + stack.push(entry.tree); + stack.extend(entry.parents); + continue; + } + + let found = source.find(&id)?; + let Some((kind, data)) = found else { + if lenient { + continue; + } + return Err(Error::MissingObject(id)); + }; + match kind { + Kind::Commit => { + let commit = CommitRef::from_bytes(&data, gix_hash::Kind::Sha1) + .map_err(|error| Error::Decode(error.to_string()))?; + stack.push(commit.tree()); + stack.extend(commit.parents()); + } + Kind::Tree => { + for entry in TreeRefIter::from_bytes(&data, gix_hash::Kind::Sha1) { + let entry = entry.map_err(|error| Error::Decode(error.to_string()))?; + if entry.mode.kind() == gix_object::tree::EntryKind::Commit { + // A submodule gitlink: an object id in another + // repository's object space, never ours to resolve. + continue; + } + stack.push(entry.oid.to_owned()); + } + } + Kind::Tag => { + let tag = TagRef::from_bytes(&data, gix_hash::Kind::Sha1) + .map_err(|error| Error::Decode(error.to_string()))?; + stack.push(tag.target()); + } + Kind::Blob => {} + } + } + Ok(seen) +}

crates/git-reachability/tests/equivalence.rs @@ -1,0 +1,263 @@ +//! Conformance-style equivalence for the accelerated walk +//! (`docs/scale-out.adoc`, "Reachability": "absence or staleness degrades +//! speed, never answers"): over several generated DAG shapes, the +//! commit-graph-accelerated walk must return exactly the same reachable set +//! as the plain one, and a stale cached [`git_reachability::reachable_set:: +//! ReachableSetArtifact`] must still yield a correct answer once new commits +//! have landed past the frontier it was built from. +//! +//! Runs against real `odb-files`/`refstore-files` repositories built with +//! the `git` CLI (mirroring `git-protocol`'s own test fixtures) rather than +//! synthetic in-memory commits, so the DAG shapes exercise real tree/blob +//! objects too, not just the commit-parent skeleton. + +#![allow(clippy::unwrap_used, reason = "test fixture")] + +use std::path::Path; +use std::process::Command; + +use git_reachability::commitgraph::CommitGraph; +use git_reachability::engine::{self, ArtifactBundle}; +use git_reachability::reachable_set::ReachableSetArtifact; +use git_reachability::walk::{self, StoreSource}; +use gix_hash::ObjectId; + +fn bare_repo() -> tempfile::TempDir { + let dir = tempfile::tempdir().unwrap(); + let status = Command::new("git") + .args(["init", "-q", "--bare", "-b", "main"]) + .arg(dir.path()) + .status() + .unwrap(); + assert!(status.success()); + dir +} + +fn run(dir: &Path, args: &[&str]) { + let status = Command::new("git") + .arg("-C") + .arg(dir) + .args(args) + .status() + .unwrap(); + assert!(status.success()); +} + +fn ref_exists(bare: &Path, branch: &str) -> bool { + Command::new("git") + .arg("-C") + .arg(bare) + .args([ + "rev-parse", + "--verify", + "-q", + &format!("refs/heads/{branch}"), + ]) + .output() + .map(|output| output.status.success()) + .unwrap_or(false) +} + +fn rev_parse_head(work: &Path) -> ObjectId { + let hex = String::from_utf8( + Command::new("git") + .arg("-C") + .arg(work) + .args(["rev-parse", "HEAD"]) + .output() + .unwrap() + .stdout, + ) + .unwrap(); + ObjectId::from_hex(hex.trim().as_bytes()).unwrap() +} + +/// Commit `content` in `file_name` onto `branch` in `bare`, basing the new +/// commit on `branch`'s (or, if `branch` doesn't exist yet, `main`'s) +/// current tip so a series of calls with the same `branch` builds a linear +/// chain. Returns the new commit's id. +fn commit_onto(bare: &Path, branch: &str, file_name: &str, content: &str) -> ObjectId { + let work = tempfile::tempdir().unwrap(); + run(work.path(), &["init", "-q", "-b", branch]); + run(work.path(), &["config", "user.email", "test@example.com"]); + run(work.path(), &["config", "user.name", "test"]); + + let base = if ref_exists(bare, branch) { + Some(branch) + } else if ref_exists(bare, "main") { + Some("main") + } else { + None + }; + if let Some(base) = base { + run(work.path(), &["fetch", "-q", bare.to_str().unwrap(), base]); + run(work.path(), &["reset", "-q", "--hard", "FETCH_HEAD"]); + } + + std::fs::write(work.path().join(file_name), content).unwrap(); + run(work.path(), &["add", "-A"]); + run(work.path(), &["commit", "-q", "-m", "commit"]); + let commit = rev_parse_head(work.path()); + run( + work.path(), + &[ + "push", + bare.to_str().unwrap(), + &format!("HEAD:refs/heads/{branch}"), + ], + ); + commit +} + +/// Merge `from` into `into` in `bare` with a real merge commit (two +/// parents), and push the result back onto `into`. Returns the merge +/// commit's id. +fn merge(bare: &Path, into: &str, from: &str) -> ObjectId { + let work = tempfile::tempdir().unwrap(); + run(work.path(), &["init", "-q", "-b", into]); + run(work.path(), &["config", "user.email", "test@example.com"]); + run(work.path(), &["config", "user.name", "test"]); + run(work.path(), &["fetch", "-q", bare.to_str().unwrap(), into]); + run(work.path(), &["reset", "-q", "--hard", "FETCH_HEAD"]); + run( + work.path(), + &[ + "fetch", + "-q", + bare.to_str().unwrap(), + &format!("{from}:{from}"), + ], + ); + run( + work.path(), + &["merge", "-q", "--no-ff", "-m", "merge", from], + ); + let commit = rev_parse_head(work.path()); + run( + work.path(), + &[ + "push", + bare.to_str().unwrap(), + &format!("HEAD:refs/heads/{into}"), + ], + ); + commit +} + +/// Assert the commit-graph-accelerated walk agrees with the plain walk from +/// `tips`, both directly ([`walk::reachable_with_graph`]) and through +/// [`engine::accelerated_reachable`] with no cached reachable-set (so only +/// the commit-graph acceleration is in play). +fn assert_accelerated_matches_slow(objects: &odb_files::OdbFiles, tips: &[ObjectId]) { + let source = StoreSource::new(objects); + + let slow = walk::reachable(tips.iter().copied(), &source, |_id| false, false).unwrap(); + + let graph = CommitGraph::build(tips.iter().copied(), &source).unwrap(); + let accelerated = walk::reachable_with_graph( + tips.iter().copied(), + &source, + |_id| false, + false, + Some(&graph), + ) + .unwrap(); + assert_eq!( + accelerated, slow, + "graph-accelerated walk disagreed with the slow one" + ); + + let bundle = ArtifactBundle { + commit_graph: Some(graph), + reachable_set: None, + }; + let via_engine = + engine::accelerated_reachable(tips.iter().copied(), &source, |_id| false, false, &bundle) + .unwrap(); + assert_eq!( + via_engine, slow, + "engine entry point disagreed with the slow walk" + ); +} + +#[test] +fn linear_chain() { + let bare = bare_repo(); + commit_onto(bare.path(), "main", "a", "1"); + commit_onto(bare.path(), "main", "a", "2"); + let tip = commit_onto(bare.path(), "main", "a", "3"); + + let objects = odb_files::OdbFiles::open(bare.path()).unwrap(); + assert_accelerated_matches_slow(&objects, &[tip]); +} + +#[test] +fn diamond_merge() { + let bare = bare_repo(); + commit_onto(bare.path(), "main", "base", "0"); + commit_onto(bare.path(), "left", "left", "l"); + commit_onto(bare.path(), "right", "right", "r"); + merge(bare.path(), "main", "left"); + let tip = merge(bare.path(), "main", "right"); + + let objects = odb_files::OdbFiles::open(bare.path()).unwrap(); + assert_accelerated_matches_slow(&objects, &[tip]); +} + +#[test] +fn disconnected_roots_as_separate_tips() { + let bare = bare_repo(); + let tip_a = commit_onto(bare.path(), "a", "a", "1"); + let tip_b = commit_onto(bare.path(), "b", "b", "1"); + + let objects = odb_files::OdbFiles::open(bare.path()).unwrap(); + assert_accelerated_matches_slow(&objects, &[tip_a, tip_b]); +} + +#[test] +fn a_stale_reachable_set_snapshot_still_yields_a_correct_answer_after_new_commits_land() { + let bare = bare_repo(); + let old_tip = commit_onto(bare.path(), "main", "a", "1"); + + let objects = odb_files::OdbFiles::open(bare.path()).unwrap(); + let source = StoreSource::new(&objects); + let stale = ReachableSetArtifact::build([old_tip], &source).unwrap(); + + let new_tip = commit_onto(bare.path(), "main", "a", "2"); + let bundle = ArtifactBundle { + commit_graph: None, + reachable_set: Some(stale), + }; + + let accelerated = + engine::accelerated_reachable([new_tip], &source, |_id| false, false, &bundle).unwrap(); + let slow = walk::reachable([new_tip], &source, |_id| false, false).unwrap(); + + assert_eq!( + accelerated, slow, + "a stale frontier must fall back to a full walk, not a wrong answer" + ); + // The new tip's own commit must be present — a bug that just returned + // the stale cached set unconditionally would have missed it. + assert!(accelerated.contains(&new_tip)); +} + +#[test] +fn commit_graph_missing_a_new_commit_still_degrades_correctly() { + let bare = bare_repo(); + let old_tip = commit_onto(bare.path(), "main", "a", "1"); + + let objects = odb_files::OdbFiles::open(bare.path()).unwrap(); + let source = StoreSource::new(&objects); + // A graph built before `new_tip` exists: `entry()` will return `None` + // for it, so the walk must fall back to an ordinary object-store read + // for exactly that commit. + let stale_graph = CommitGraph::build([old_tip], &source).unwrap(); + + let new_tip = commit_onto(bare.path(), "main", "a", "2"); + let accelerated = + walk::reachable_with_graph([new_tip], &source, |_id| false, false, Some(&stale_graph)) + .unwrap(); + let slow = walk::reachable([new_tip], &source, |_id| false, false).unwrap(); + assert_eq!(accelerated, slow); +}

feat: add git-reachability crate for WS6 accelerated reachability

Reviews

Start a review