feat: add op-replay corpus type and replay harness
commit
0a5942bfeat: add op-replay corpus type and replay harness
Every accepted push through a WS0-style write path can now log a git_protocol::CorpusEntry (push-cert OID, applied ref edits, pack bytes) durably in Postgres, and backend_conformance::replay_corpus replays a logged corpus against any RefStore+ObjectStore pair, asserting an identical outcome. This is the conformance seed corpus docs/scale-out.adoc asks WS0 to feed WS2.
feat: add git_protocol::corpus::CorpusEntry feat: add PostgresRefStore::log_corpus_entry/corpus_log over a new git_ents_corpus_log table feat: add backend_conformance::replay_corpus and reachable_object_set test: add a synthesized-corpus replay test in backend-conformance Assisted-by: Claude:claude-sonnet-5
Reviews
No reviews of this commit yet — record a verdict below.
Start a review
Cargo.lock
@@ -571,9 +571,13 @@
version = "0.0.0"
dependencies = [
"git-backend",
+ "git-protocol",
+ "git-reachability",
"git-store",
"gix-hash",
"gix-object",
+ "odb-files",
+ "refstore-files",
"tempfile",
]
@@ -4091,6 +4095,7 @@
dependencies = [
"backend-conformance",
"git-backend",
+ "git-protocol",
"gix-hash",
"gix-object",
"odb-tiered",
crates/backend-conformance/Cargo.toml
@@ -7,10 +7,16 @@
[dependencies]
git-backend = { workspace = true }
+git-protocol = { workspace = true }
+git-reachability = { workspace = true }
git-store = { workspace = true, features = ["test-support"] }
gix-hash = { workspace = true }
gix-object = { workspace = true }
tempfile = { workspace = true }
+[dev-dependencies]
+odb-files = { workspace = true }
+refstore-files = { workspace = true }
+
[lints]
workspace = true
crates/refstore-postgres/Cargo.toml
@@ -7,6 +7,7 @@
[dependencies]
git-backend = { workspace = true }
+git-protocol = { workspace = true }
gix-hash = { workspace = true }
gix-object = { workspace = true }
odb-tiered = { workspace = true }
crates/backend-conformance/src/lib.rs
@@ -25,6 +25,7 @@
//! to use it.
mod collector;
+mod corpus;
mod fixture_oids;
mod object_store;
mod ref_store;
@@ -32,6 +33,7 @@
mod support;
pub use collector::{Collector, NoopCollector};
+pub use corpus::{reachable_object_set, replay_corpus};
pub use fixture_oids::FixtureOids;
pub use object_store::{
causal_collection_safety, object_store_properties, quarantine_invisibility,
crates/git-protocol/src/lib.rs
@@ -19,12 +19,14 @@
//! server-signed op record every accepted push emits.
pub mod attestation;
+pub mod corpus;
pub mod native;
pub mod pack;
mod traits;
pub mod types;
pub mod walk;
+pub use corpus::CorpusEntry;
pub use traits::{Advertise, GeneratePack, IngestPack, Negotiate};
pub use types::{
AdSpec, AppliedRefEdit, NegotiationState, PackPlan, PushCertificate, PushOutcome, PushRequest,
crates/refstore-postgres/migrations/0001_init.sql
@@ -112,6 +112,25 @@
CREATE INDEX IF NOT EXISTS git_ents_op_records_repo_idx
ON git_ents_op_records (repo_id, created_at DESC);
+-- Op-replay corpus (WS0, `docs/scale-out.adoc`'s "op replay corpus" /
+-- WS2's conformance seed corpus): one row per accepted push through the
+-- hydration write path, carrying enough to replay it against any
+-- RefStore+ObjectStore pair — see `git_protocol::corpus::CorpusEntry`,
+-- which this table's rows serialize. `ref_edits` is one `name\told\tnew`
+-- line per edit (`-` for a missing old/new); `pack` is the exact bytes
+-- staged for the push (may be empty, e.g. a pure ref deletion).
+CREATE TABLE IF NOT EXISTS git_ents_corpus_log (
+ id BIGSERIAL PRIMARY KEY,
+ repo_id TEXT NOT NULL,
+ push_cert_oid TEXT,
+ ref_edits TEXT NOT NULL,
+ pack BYTEA NOT NULL,
+ recorded_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS git_ents_corpus_log_repo_idx
+ ON git_ents_corpus_log (repo_id, id);
+
-- 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
crates/refstore-postgres/src/lib.rs
@@ -31,6 +31,7 @@
//! (`fly-replay`, replica topology), not this crate's job — there is no
//! fencing code here, deliberately.
+mod corpus;
mod notify;
mod pack_registry;
mod queue;
crates/backend-conformance/src/corpus.rs
@@ -1,0 +1,186 @@
+//! Replay harness for [`git_protocol::CorpusEntry`] (`docs/scale-out.adoc`,
+//! WS0's "op replay corpus" — the conformance seed corpus WS2 replays):
+//! [`replay_corpus`] applies a logged corpus, in order, against any
+//! `RefStore`+`ObjectStore` pair; [`reachable_object_set`] is the
+//! "identical reachable object sets" half of the assertion a replay test
+//! makes (the "identical final refs" half is an ordinary
+//! `RefStore::iter_prefix` comparison, needing no helper here).
+
+use std::collections::BTreeSet;
+
+use git_backend::{Expected, ObjectStore, PackStream, RefEdit, RefStore, TxOutcome};
+use git_protocol::CorpusEntry;
+use gix_hash::ObjectId;
+
+/// Replay `entries`, in order, against `refs`/`objects`: stage each
+/// entry's pack, apply its ref edits as one atomic transaction (the same
+/// shape the entry was originally accepted with), then promote once that
+/// transaction applies — never before, so a replayed-but-rejected entry's
+/// pack stays quarantined rather than becoming visible garbage.
+///
+/// A corpus is expected to replay cleanly against a backend starting from
+/// the same state (typically empty) it was recorded from; an entry whose
+/// recorded `old`/`new` no longer apply against `refs`'s current state is
+/// reported as an error rather than silently skipped, since that means the
+/// corpus and the target have already diverged — exactly what this
+/// harness exists to catch.
+///
+/// # Errors
+///
+/// Returns an error if staging, transacting, or promoting any entry
+/// fails, including a rejected compare-and-swap.
+pub fn replay_corpus(
+ entries: &[CorpusEntry],
+ refs: &dyn RefStore,
+ objects: &dyn ObjectStore,
+) -> git_backend::Result<()> {
+ for entry in entries {
+ let quarantine =
+ objects.stage_pack(PackStream::new(std::io::Cursor::new(entry.pack.clone())))?;
+ let edits: Vec<RefEdit> = entry
+ .ref_edits
+ .iter()
+ .map(|edit| RefEdit {
+ name: edit.name.clone(),
+ expected: match edit.old {
+ Some(oid) => Expected::MustExistAndMatch(oid),
+ None => Expected::MustNotExist,
+ },
+ new: edit.new,
+ })
+ .collect();
+ match refs.transaction(&edits)? {
+ TxOutcome::Applied => objects.promote(quarantine)?,
+ TxOutcome::Rejected { name } => {
+ return Err(git_backend::Error::RefStore(format!(
+ "corpus replay: ref {name} did not match its recorded expected value"
+ )));
+ }
+ }
+ }
+ Ok(())
+}
+
+/// The set of every object reachable from `refs`' current tips over
+/// `objects` — a thin wrapper over [`git_reachability::gc_mark`] (no
+/// reachability artifacts, so a plain walk) for a replay test to compare
+/// between the original backend and the one it replayed a corpus into.
+///
+/// # Errors
+///
+/// Returns an error if the ref or object store cannot be read, or the walk
+/// finds a ref tip whose history is incomplete.
+pub fn reachable_object_set(
+ refs: &dyn RefStore,
+ objects: &dyn ObjectStore,
+) -> git_reachability::Result<BTreeSet<ObjectId>> {
+ git_reachability::gc_mark(refs, objects, &git_reachability::ArtifactBundle::empty())
+}
+
+#[cfg(test)]
+mod tests {
+ #![allow(
+ clippy::unwrap_used,
+ clippy::expect_used,
+ reason = "test fixture, not application code"
+ )]
+
+ use std::process::{Command, Stdio};
+
+ use git_backend::RefName;
+ use git_protocol::types::AppliedRefEdit;
+ use git_store::test_support::{commit_all, head, repo};
+
+ use super::*;
+
+ /// Pack every object reachable from `commit` and not from `boundary`
+ /// (or the commit's whole history, when `boundary` is `None`) — the
+ /// same shape `git_hydrate::pre_receive::build_pack` produces for a
+ /// real push, so this synthesized corpus exercises `replay_corpus`
+ /// against realistic incremental packs, not just whole-history ones.
+ fn pack_for(dir: &std::path::Path, commit: &str, boundary: Option<&str>) -> Vec<u8> {
+ let mut rev_list_args = vec!["rev-list", "--objects", commit];
+ if let Some(boundary) = boundary {
+ rev_list_args.push("--not");
+ rev_list_args.push(boundary);
+ }
+ let mut rev_list = Command::new("git")
+ .arg("-C")
+ .arg(dir)
+ .args(&rev_list_args)
+ .stdout(Stdio::piped())
+ .spawn()
+ .expect("spawn git rev-list");
+ let pack_objects = Command::new("git")
+ .arg("-C")
+ .arg(dir)
+ .args(["pack-objects", "--stdout", "-q"])
+ .stdin(rev_list.stdout.take().expect("rev-list stdout"))
+ .stdout(Stdio::piped())
+ .spawn()
+ .expect("spawn git pack-objects");
+ let output = pack_objects
+ .wait_with_output()
+ .expect("wait for pack-objects");
+ assert!(rev_list.wait().expect("wait for rev-list").success());
+ assert!(output.status.success());
+ output.stdout
+ }
+
+ // @relation(role=Verifies)
+ #[test]
+ fn replays_a_synthesized_corpus_identically() {
+ let dir = repo();
+ std::fs::write(dir.path().join("file"), "one").expect("write fixture file");
+ commit_all(dir.path(), "first");
+ let commit1 = head(dir.path());
+ std::fs::write(dir.path().join("file"), "two").expect("write fixture file");
+ commit_all(dir.path(), "second");
+ let commit2 = head(dir.path());
+
+ let commit1_oid = ObjectId::from_hex(commit1.as_bytes()).expect("valid oid");
+ let commit2_oid = ObjectId::from_hex(commit2.as_bytes()).expect("valid oid");
+
+ let entries = vec![
+ CorpusEntry::new(
+ None,
+ vec![AppliedRefEdit {
+ name: RefName::new("refs/heads/main"),
+ old: None,
+ new: Some(commit1_oid),
+ }],
+ pack_for(dir.path(), &commit1, None),
+ ),
+ CorpusEntry::new(
+ None,
+ vec![AppliedRefEdit {
+ name: RefName::new("refs/heads/main"),
+ old: Some(commit1_oid),
+ new: Some(commit2_oid),
+ }],
+ pack_for(dir.path(), &commit2, Some(&commit1)),
+ ),
+ ];
+
+ let target = tempfile::tempdir().expect("tempdir");
+ let status = Command::new("git")
+ .arg("init")
+ .arg("-q")
+ .arg("--bare")
+ .arg(target.path())
+ .status()
+ .expect("git init --bare");
+ assert!(status.success());
+ let refs = refstore_files::FilesRefStore::open(target.path()).expect("open refs");
+ let objects = odb_files::OdbFiles::open(target.path()).expect("open objects");
+
+ replay_corpus(&entries, &refs, &objects).expect("replay_corpus");
+
+ let main = RefName::new("refs/heads/main");
+ assert_eq!(refs.get(&main).expect("get"), Some(commit2_oid));
+
+ let reachable = reachable_object_set(&refs, &objects).expect("reachable_object_set");
+ assert!(reachable.contains(&commit1_oid));
+ assert!(reachable.contains(&commit2_oid));
+ }
+}
crates/git-protocol/src/corpus.rs
@@ -1,0 +1,56 @@
+//! The op-replay corpus (`docs/scale-out.adoc`, WS0's "op replay corpus",
+//! feeding WS2's conformance suite): one [`CorpusEntry`] per accepted push,
+//! carrying enough to replay it against any `RefStore`/`ObjectStore` pair and
+//! assert an identical outcome — same final refs, same reachable object set.
+//!
+//! A corpus entry deliberately does *not* carry the op record itself (the
+//! server-signed commit chained under [`crate::attestation::OP_LOG_REF`]):
+//! that record embeds a wall-clock timestamp and a fresh signature, so
+//! replaying it would never hash-match the original, and the audit trail it
+//! represents is a server-internal concern orthogonal to "did this push
+//! reproduce the same content." What a replay must reproduce is the
+//! client-visible outcome: the ref edits the push asked for, applied in the
+//! same order, over the same pack.
+
+use gix_hash::ObjectId;
+
+use crate::types::AppliedRefEdit;
+
+/// One accepted push, durably logged so it can be replayed later against a
+/// different `RefStore`/`ObjectStore` pair as a conformance fixture
+/// (`docs/scale-out.adoc`, WS0: "every push logs (push-cert OID, ref edits
+/// old/new, pack OIDs)").
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CorpusEntry {
+ /// The client's push certificate, embedded by OID — `None` only during
+ /// the bootstrap window (no members enrolled yet, so no certificate was
+ /// required). Informational (the push's *intent*): not needed to
+ /// replay the ref/object outcome, only to audit it.
+ pub push_cert_oid: Option<ObjectId>,
+ /// The ref edits this push applied, old and new — the push's *outcome*,
+ /// and what a replay must reproduce exactly.
+ pub ref_edits: Vec<AppliedRefEdit>,
+ /// The pack introducing every object this push's ref edits need that
+ /// the repository didn't already have. May be an empty pack (e.g. a
+ /// pure ref deletion).
+ pub pack: Vec<u8>,
+}
+
+impl CorpusEntry {
+ /// Build an entry from a push's certificate bytes (already hashed the
+ /// same way [`crate::native::ingest`] hashes an incoming certificate
+ /// blob), its applied ref edits, and the pack that carried its new
+ /// objects.
+ #[must_use]
+ pub fn new(
+ push_cert_oid: Option<ObjectId>,
+ ref_edits: Vec<AppliedRefEdit>,
+ pack: Vec<u8>,
+ ) -> Self {
+ Self {
+ push_cert_oid,
+ ref_edits,
+ pack,
+ }
+ }
+}
crates/refstore-postgres/src/corpus.rs
@@ -1,0 +1,130 @@
+//! [`PostgresRefStore::log_corpus_entry`]/[`PostgresRefStore::corpus_log`]:
+//! durable storage for `git_protocol::CorpusEntry` (`docs/scale-out.adoc`,
+//! WS0's "op replay corpus"), over `git_ents_corpus_log`. Kept in this crate
+//! rather than `git-protocol` itself for the same reason `pack_registry.rs`
+//! and `queue.rs` are here: `git-protocol` never needs a `tokio-postgres`
+//! dependency of its own.
+
+use git_backend::{Error, Result};
+use git_protocol::CorpusEntry;
+use git_protocol::types::AppliedRefEdit;
+use gix_hash::ObjectId;
+
+use crate::{PostgresRefStore, pg_err};
+
+/// Serialize `edits` as one `name\told\tnew` line each (`-` standing in for
+/// a missing old/new oid), so the whole batch fits in a single `TEXT`
+/// column without pulling in a serialization dependency this crate doesn't
+/// already have.
+fn encode_ref_edits(edits: &[AppliedRefEdit]) -> String {
+ edits
+ .iter()
+ .map(|edit| {
+ let old = edit
+ .old
+ .map_or_else(|| "-".to_owned(), |oid| oid.to_hex().to_string());
+ let new = edit
+ .new
+ .map_or_else(|| "-".to_owned(), |oid| oid.to_hex().to_string());
+ format!("{}\t{old}\t{new}", edit.name)
+ })
+ .collect::<Vec<_>>()
+ .join("\n")
+}
+
+/// Parse [`encode_ref_edits`]'s format back into [`AppliedRefEdit`]s. A line
+/// that cannot be parsed (malformed hex, wrong field count) is skipped
+/// rather than failing the whole read — a corpus reader degrades to a
+/// shorter replay rather than an unusable one.
+fn decode_ref_edits(text: &str) -> Vec<AppliedRefEdit> {
+ text.lines()
+ .filter_map(|line| {
+ let mut fields = line.splitn(3, '\t');
+ let name = fields.next()?;
+ let old = fields.next()?;
+ let new = fields.next()?;
+ Some(AppliedRefEdit {
+ name: git_backend::RefName::new(name),
+ old: parse_optional_oid(old),
+ new: parse_optional_oid(new),
+ })
+ })
+ .collect()
+}
+
+fn parse_optional_oid(hex: &str) -> Option<ObjectId> {
+ if hex == "-" {
+ None
+ } else {
+ ObjectId::from_hex(hex.as_bytes()).ok()
+ }
+}
+
+impl PostgresRefStore {
+ /// Durably append one accepted push's [`CorpusEntry`] to this store's
+ /// repo-scoped corpus log.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`Error::RefStore`] if the insert fails.
+ pub fn log_corpus_entry(&self, entry: &CorpusEntry) -> Result<()> {
+ let push_cert_oid = entry.push_cert_oid.map(|oid| oid.to_hex().to_string());
+ let ref_edits = encode_ref_edits(&entry.ref_edits);
+ self.runtime
+ .block_on(async {
+ let client = self.client.lock().await;
+ client
+ .execute(
+ "INSERT INTO git_ents_corpus_log
+ (repo_id, push_cert_oid, ref_edits, pack)
+ VALUES ($1, $2, $3, $4)",
+ &[&self.repo_id, &push_cert_oid, &ref_edits, &entry.pack],
+ )
+ .await
+ })
+ .map_err(pg_err)
+ .map(|_rows_affected| ())
+ }
+
+ /// This store's logged corpus entries, oldest first — the order a
+ /// replay must apply them in.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`Error::RefStore`] if the query fails, or a stored
+ /// `push_cert_oid` is not valid hex (a corrupted row, never written by
+ /// [`Self::log_corpus_entry`]).
+ pub fn corpus_log(&self) -> Result<Vec<CorpusEntry>> {
+ let rows = self
+ .runtime
+ .block_on(async {
+ let client = self.client.lock().await;
+ client
+ .query(
+ "SELECT push_cert_oid, ref_edits, pack
+ FROM git_ents_corpus_log
+ WHERE repo_id = $1 ORDER BY id ASC",
+ &[&self.repo_id],
+ )
+ .await
+ })
+ .map_err(pg_err)?;
+
+ rows.into_iter()
+ .map(|row| {
+ let push_cert_oid: Option<String> = row.try_get(0).map_err(pg_err)?;
+ let ref_edits: String = row.try_get(1).map_err(pg_err)?;
+ let pack: Vec<u8> = row.try_get(2).map_err(pg_err)?;
+ let push_cert_oid = push_cert_oid
+ .map(|hex| ObjectId::from_hex(hex.as_bytes()))
+ .transpose()
+ .map_err(|error| Error::RefStore(error.to_string()))?;
+ Ok(CorpusEntry {
+ push_cert_oid,
+ ref_edits: decode_ref_edits(&ref_edits),
+ pack,
+ })
+ })
+ .collect()
+ }
+}