//! The pre-publication verifier for an imported store.
//!
//! `zit import-git` writes its published store directly and withholds from it
//! the candidate-only drop set, which lives only in the mirror store until the
//! run ends. This is the proof that what it is about to publish is exactly
//! what the import wrote, minus exactly that set, with every byte intact. It
//! is the safety net under an operation that deletes mirror metadata from a
//! history, so every clause below returns its OWN error and prints its own
//! name: an operator who is told only "verification failed" cannot tell a
//! dropped artifact from a tampered provenance line, and those two need
//! different answers.
//!
//! Ordered, and the order is load-bearing. Storage-level facts come first
//! because every semantic clause reads content and a clause that read corrupt
//! bytes would report the wrong thing about them. Within the semantic half,
//! commits are walked in stream order, and within one commit the tree comes
//! before the parents and the parents before the attribution -- so the most
//! direct explanation of a divergence is always the one that fires.
//!
//! The name accounting is a count plus an order-independent fingerprint
//! (`logstore.NameTally`): the writer tallies every name it puts into the
//! published store, and the verifier tallies the closed file's index. Equal
//! tallies are the fast answer. Unequal ones fall back to a slow pass that
//! derives what the store should hold from the witness by reachability and
//! names the artifacts that differ, so the refusal still says what moved.
//!
//! One refinement is documented at `deferDifference`: a check-in's NAME is a
//! function of its content, so a check-in appearing or disappearing is never
//! an independent fact. It is the shadow of a content change, and the content
//! clauses are what explain content. Such a difference is therefore deferred,
//! not dropped -- if no later clause explains it, it is reported at the end
//! under its own name.
//!
//! SCOPE, stated so the operator-facing promise is not read wider than it is.
//! What this file proves is a relation between the published store, the
//! mirror store and the witness the import took of its own decisions: the
//! published store holds exactly the names the import wrote into it, none of
//! the drop set, every artifact reconstructs and hashes to its name, every
//! published check-in flattens/parents/attributes to what the import
//! recorded, and the three staging files agree with all of it. What it does
//! NOT prove is that the import was a faithful translation of the Git stream
//! -- that is the translator's own contract, established by the e2e lane
//! against real `git`. An artifact a translation bug wrote, that is in no drop
//! set, satisfies every clause here. "Nothing else moved" is the claim; "the
//! source was right" is not.
const std = @import("std");
const zit = @import("zit");
const artifact = zit.artifact;
const git_import = zit.git_import;
const logstore = zit.logstore;
const repoview = zit.repoview;
const repostate = zit.repostate;
const Repository = zit.Repository;
const store_workers = zit.store_workers;
const Meter = zit.progress.Meter;
/// One named invariant. The token each renders is the operator-facing contract:
/// `import-git: verifier: <token>`.
pub const Clause = enum {
mirror_store_unreadable,
mirror_rehash_mismatch,
published_store_unreadable,
artifact_rehash_mismatch,
delta_base_mismatch,
retained_omitted_artifact,
dropped_noncandidate_artifact,
dropped_required_artifact,
added_artifact,
written_artifact_mismatch,
checkin_unresolvable,
checkin_name_mismatch,
normalized_tree_digest_mismatch,
parent_order_mismatch,
attribution_mismatch,
ref_target_mismatch,
tag_mismatch,
provenance_record_mismatch,
provenance_digest_mismatch,
repository_view_mismatch,
checkout_invalid,
map_unreadable,
witness_unreadable,
pub fn token(self: Clause) []const u8 {
return switch (self) {
.mirror_store_unreadable => "mirror-store-unreadable",
.mirror_rehash_mismatch => "mirror-rehash-mismatch",
.published_store_unreadable => "published-store-unreadable",
.artifact_rehash_mismatch => "artifact-rehash-mismatch",
.delta_base_mismatch => "delta-base-mismatch",
.retained_omitted_artifact => "retained-omitted-artifact",
.dropped_noncandidate_artifact => "dropped-noncandidate-artifact",
.dropped_required_artifact => "dropped-required-artifact",
.added_artifact => "added-artifact",
.written_artifact_mismatch => "written-artifact-mismatch",
.checkin_unresolvable => "checkin-unresolvable",
.checkin_name_mismatch => "checkin-name-mismatch",
.normalized_tree_digest_mismatch => "normalized-tree-digest-mismatch",
.parent_order_mismatch => "parent-order-mismatch",
.attribution_mismatch => "attribution-mismatch",
.ref_target_mismatch => "ref-target-mismatch",
.tag_mismatch => "tag-mismatch",
.provenance_record_mismatch => "provenance-record-mismatch",
.provenance_digest_mismatch => "provenance-digest-mismatch",
.repository_view_mismatch => "repository-view-mismatch",
.checkout_invalid => "checkout-invalid",
.map_unreadable => "map-unreadable",
.witness_unreadable => "witness-unreadable",
};
}
/// The error a clause refuses with. One per clause, so a caller can switch
/// on the reason rather than on a printed string.
pub fn err(self: Clause) Error {
return switch (self) {
.mirror_store_unreadable => error.VerifierMirrorStoreUnreadable,
.mirror_rehash_mismatch => error.VerifierMirrorRehashMismatch,
.published_store_unreadable => error.VerifierPublishedStoreUnreadable,
.artifact_rehash_mismatch => error.VerifierArtifactRehashMismatch,
.delta_base_mismatch => error.VerifierDeltaBaseMismatch,
.retained_omitted_artifact => error.VerifierRetainedOmittedArtifact,
.dropped_noncandidate_artifact => error.VerifierDroppedNoncandidateArtifact,
.dropped_required_artifact => error.VerifierDroppedRequiredArtifact,
.added_artifact => error.VerifierAddedArtifact,
.written_artifact_mismatch => error.VerifierWrittenArtifactMismatch,
.checkin_unresolvable => error.VerifierCheckinUnresolvable,
.checkin_name_mismatch => error.VerifierCheckinNameMismatch,
.normalized_tree_digest_mismatch => error.VerifierNormalizedTreeDigestMismatch,
.parent_order_mismatch => error.VerifierParentOrderMismatch,
.attribution_mismatch => error.VerifierAttributionMismatch,
.ref_target_mismatch => error.VerifierRefTargetMismatch,
.tag_mismatch => error.VerifierTagMismatch,
.provenance_record_mismatch => error.VerifierProvenanceRecordMismatch,
.provenance_digest_mismatch => error.VerifierProvenanceDigestMismatch,
.repository_view_mismatch => error.VerifierRepositoryViewMismatch,
.checkout_invalid => error.VerifierCheckoutInvalid,
.map_unreadable => error.VerifierMapUnreadable,
.witness_unreadable => error.VerifierWitnessUnreadable,
};
}
};
pub const Error = error{
VerifierMirrorStoreUnreadable,
VerifierMirrorRehashMismatch,
VerifierPublishedStoreUnreadable,
VerifierArtifactRehashMismatch,
VerifierDeltaBaseMismatch,
VerifierRetainedOmittedArtifact,
VerifierDroppedNoncandidateArtifact,
VerifierDroppedRequiredArtifact,
VerifierAddedArtifact,
VerifierWrittenArtifactMismatch,
VerifierCheckinUnresolvable,
VerifierCheckinNameMismatch,
VerifierNormalizedTreeDigestMismatch,
VerifierParentOrderMismatch,
VerifierAttributionMismatch,
VerifierRefTargetMismatch,
VerifierTagMismatch,
VerifierProvenanceRecordMismatch,
VerifierProvenanceDigestMismatch,
VerifierRepositoryViewMismatch,
VerifierCheckoutInvalid,
VerifierMapUnreadable,
VerifierWitnessUnreadable,
OutOfMemory,
};
const hex_name_len = 2 * logstore.max_hash_len;
pub const default_min_run: usize = 1024;
/// A store's artifact names, in persisted order and as a set.
const Inventory = struct {
order: std.ArrayList([]const u8) = .empty,
set: std.StringHashMapUnmanaged(void) = .empty,
fn add(self: *Inventory, gpa: std.mem.Allocator, name: []const u8) !void {
try self.order.append(gpa, name);
try self.set.put(gpa, name, {});
}
};
/// Every witnessed OID by value, to the stream index of the LAST commit that
/// carried it -- what a table keyed by OID string would answer. The OIDs are
/// packed into one buffer and the table holds only indexes into it, since an
/// entry per commit is paid at every commit of the history.
const OidIndex = struct {
bytes: std.ArrayList(u8) = .empty,
/// Commit `i`'s OID is `bytes[starts[i]..starts[i + 1]]`; empty where it
/// carried none.
starts: std.ArrayList(usize) = .empty,
table: std.HashMapUnmanaged(u32, void, Context, std.hash_map.default_max_load_percentage) = .empty,
const Context = struct {
index: *const OidIndex,
pub fn hash(self: Context, key: u32) u64 {
return std.hash.Wyhash.hash(0, self.index.oidAt(key));
}
pub fn eql(self: Context, a: u32, b: u32) bool {
return std.mem.eql(u8, self.index.oidAt(a), self.index.oidAt(b));
}
};
const ByOid = struct {
index: *const OidIndex,
pub fn hash(_: ByOid, oid: []const u8) u64 {
return std.hash.Wyhash.hash(0, oid);
}
pub fn eql(self: ByOid, oid: []const u8, key: u32) bool {
return std.mem.eql(u8, oid, self.index.oidAt(key));
}
};
fn deinit(self: *OidIndex, gpa: std.mem.Allocator) void {
self.bytes.deinit(gpa);
self.starts.deinit(gpa);
self.table.deinit(gpa);
}
fn oidAt(self: *const OidIndex, key: u32) []const u8 {
return self.bytes.items[self.starts.items[key]..self.starts.items[key + 1]];
}
/// The next commit in stream order.
fn add(self: *OidIndex, gpa: std.mem.Allocator, oid: ?[]const u8) error{OutOfMemory}!void {
if (self.starts.items.len == 0) try self.starts.append(gpa, 0);
const key = std.math.cast(u32, self.starts.items.len - 1) orelse return error.OutOfMemory;
if (oid) |text| try self.bytes.appendSlice(gpa, text);
try self.starts.append(gpa, self.bytes.items.len);
const text = oid orelse return;
const slot = try self.table.getOrPutContextAdapted(gpa, text, ByOid{ .index = self }, .{ .index = self });
slot.key_ptr.* = key;
}
fn find(self: *const OidIndex, oid: []const u8) ?u32 {
const key = self.table.getKeyPtrAdapted(oid, ByOid{ .index = self }) orelse return null;
return key.*;
}
};
/// A name difference whose explanation the content clauses may still own.
const Deferred = struct {
clause: Clause,
name: []const u8,
};
/// One artifact's content identity, memoized across the whole tree sweep. A
/// wide repository names the same blob from many commits, and without this the
/// normalized-tree recompute would be quadratic in the history rather than
/// linear in the store.
const ContentFacts = struct {
length: u64,
digest: git_import.Digest,
};
/// One `kind:"commit"` record of the finalized JSONL.
const MapCommit = struct {
oid: []const u8,
zit: []const u8,
author: MapIdentity,
committer: MapIdentity,
tree_digest: []const u8,
provenance_digest: []const u8,
};
const MapIdentity = struct {
name: []const u8,
email: []const u8,
timestamp: i64,
timezone_offset_minutes: i64,
};
/// One `kind:"ref"` or `kind:"tag"` record, reduced to the mapping it asserts.
const MapMapping = struct {
canonical: []const u8,
target: []const u8,
};
const RefRow = struct {
canonical: []const u8,
target: []const u8,
};
pub const Input = struct {
/// The mirror store, frozen: the drop set's only copy.
mirror: *logstore.LogStore,
/// The published store, frozen -- the one that will be renamed.
published: *logstore.LogStore,
/// What the import decided, captured before the freeze destroyed it.
witness: *const git_import.ImportWitness,
/// The witness's commits, spilled at capture: `witness.commit_count`
/// records under `witness.commit_key`, read a window at a time.
witness_commits: *std.Io.File.Reader,
/// The classifier's candidate-only drop set, as hex artifact names.
drop_names: []const []const u8,
/// The finalized provenance map, read one record at a time and read
/// again, by position, for the window of commits a clause is on. The
/// reader's buffer is the longest record it admits.
map: *std.Io.File.Reader,
/// Called each time a window's map records have been read again. A test
/// seam only.
map_read: ?zit.publication.PublicationHooks.MapReadObserver = null,
/// The `refs` interop file, verbatim.
refs: []const u8,
/// The header `writeRefs` opens `refs` with.
refs_header: []const u8,
/// The branch `.zit-checkout` names.
checkout_branch: []const u8,
/// Suppress the sub-manifest memo report (`--quiet`). The report is a
/// measurement, not a result, so it is the first thing a quiet run drops.
quiet: bool = false,
/// The import's worker threads, for the reads behind clauses 2, 4 to 6
/// and 9. Null, or a pool of one thread, reads everything in one pass on
/// this thread. More threads change how long a clause takes, never what it
/// reports: threads read, parse and rehash, each clause consumes their
/// results in its own order on this thread, and the first failure in
/// that order is the one reported, exactly as a single pass would stop.
workers: ?*store_workers.Pool = null,
/// The shortest run worth splitting the store into for the sweeps: a
/// split costs a header read per delta, which a small store does not
/// repay. Lowered only by tests, so a fixture splits the way a large
/// store does.
min_run: usize = default_min_run,
/// Where the long clauses narrate themselves; null under `--quiet`.
progress: ?Progress = null,
/// Commits per window of the witness, and so of pooled check-in reads.
/// Lowered only by tests, so a fixture crosses windows the way a large
/// history does.
checkin_window: usize = default_checkin_window,
};
pub const default_checkin_window: usize = 4096;
/// The verifier's progress lines, `import-git: verifying N/M <unit> (P%), R/s`.
pub const Progress = struct {
out: *std.Io.Writer,
io: std.Io,
/// Lowered only by tests, so a fixture's few frames are narrated.
interval_ns: u64 = Meter.default_interval_ns,
stride: u64 = Meter.default_stride,
};
/// Run the ordered pipeline. Returns on the FIRST clause that refuses, having
/// already printed its name and enough detail to act on.
pub fn verify(gpa: std.mem.Allocator, input: Input, err: *std.Io.Writer) Error!void {
var arena_state = std.heap.ArenaAllocator.init(gpa);
defer arena_state.deinit();
var self: Verifier = .{
.gpa = gpa,
.arena = arena_state.allocator(),
.window_arena = std.heap.ArenaAllocator.init(gpa),
.err = err,
.input = input,
};
defer self.deinit();
try self.run();
self.reportSubManifestMemo();
}
const Verifier = struct {
gpa: std.mem.Allocator,
/// Names, parsed records and memo keys. Freed in one act by `verify`.
arena: std.mem.Allocator,
/// The window of witnessed commits a clause is on, and their map
/// records. Reset by every `Walk.next`, so nothing held past a window
/// may point into it.
window_arena: std.heap.ArenaAllocator,
err: *std.Io.Writer,
input: Input,
/// The mirror store's names. Small: mirror metadata only.
mirror_names: Inventory = .{},
/// The drop set, each name to its position in `drop_names`.
drops: std.StringHashMapUnmanaged(usize) = .empty,
deferred: std.ArrayList(Deferred) = .empty,
facts: std.StringHashMapUnmanaged(ContentFacts) = .empty,
/// Witnessed check-in name -> the name the published store carries for
/// it, held only where the two differ: identity for every commit no
/// content clause had to explain, and that is nearly all of them.
resolved: std.StringHashMapUnmanaged([]const u8) = .empty,
/// Shared across EVERY commit's tree enumeration, which is the point:
/// adjacent check-ins share nearly all of their sub-manifest nodes, so a
/// per-call memo would throw away the sharing the format exists to create
/// and re-inflate a root-adjacent node once per descendant check-in.
/// Keyed by artifact name, so a hit is the same artifact by content
/// address. Bounded by the memo's own retention budget, and it degrades to
/// the naive behaviour rather than failing when that budget is met.
sub_memo: ?repostate.SubManifestMemo = null,
/// Where in the map each witnessed commit's record starts, and a tag of
/// its bytes, by stream index: the LAST commit record carrying that commit's OID, as an index
/// by OID would answer. `no_record` where there is none. The records
/// themselves are read again for the window a clause is on, because held
/// whole they grow with the history.
map_record_at: []MapRecordAt = &.{},
/// The first commit record, in map order, whose OID no commit carried.
map_unwitnessed: ?[]const u8 = null,
map_refs: std.ArrayList(MapMapping) = .empty,
map_tags: std.ArrayList(MapMapping) = .empty,
map_summary_digest: ?[]const u8 = null,
// Indexes, not conveniences. Every lookup below sits inside a loop over
// the whole import, so a linear scan would make verification quadratic in
// the history -- unnoticeable on a fixture and ruinous on a real one.
map_ref_index: std.StringHashMapUnmanaged(usize) = .empty,
map_tag_index: std.StringHashMapUnmanaged(usize) = .empty,
witness_ref_index: std.StringHashMapUnmanaged(usize) = .empty,
witness_oids: OidIndex = .{},
/// One memo for the PUBLISHED store, shared by every semantic clause.
/// A delta chain runs to `logstore.max_delta_chain` frames, and every
/// content read below sits inside a loop over the whole import -- reading
/// uncached would multiply the store by the chain depth, once per clause.
published_cache: logstore.PlaintextCache = .{},
/// The published store's persisted order split into runs of whole delta
/// trees, made by clause 2's sweep and reused by clause 9's derive. It
/// decides only which thread reads which frame, never what is read.
chain_split: ?zit.chain_runs.Runs = null,
fn deinit(self: *Verifier) void {
self.published_cache.deinit(self.gpa);
if (self.chain_split) |*s| s.deinit(self.gpa);
self.window_arena.deinit();
// Every allocation these hold came from the arena; only the maps' own
// spines are the general allocator's.
self.mirror_names.order.deinit(self.gpa);
self.mirror_names.set.deinit(self.gpa);
self.drops.deinit(self.gpa);
self.deferred.deinit(self.gpa);
self.facts.deinit(self.gpa);
self.resolved.deinit(self.gpa);
// Owns an arena holding every cached node view; the walk itself never
// frees it, because the whole point is that the views outlive the
// check-in that first resolved them.
if (self.sub_memo) |*memo| memo.deinit();
self.gpa.free(self.map_record_at);
self.map_refs.deinit(self.gpa);
self.map_tags.deinit(self.gpa);
self.map_ref_index.deinit(self.gpa);
self.map_tag_index.deinit(self.gpa);
self.witness_ref_index.deinit(self.gpa);
self.witness_oids.deinit(self.gpa);
}
/// Report what the sub-manifest memo actually achieved on this run.
///
/// The memo exists to collapse repeated parses of nodes that adjacent
/// check-ins share (submanifests §7.4), and its value is entirely a
/// function of how much sharing a repository has — which no benchmark can
/// stand in for, because the sharing ratio IS the variable. A corpus with
/// narrow per-commit churn shares nearly every node; a wide, heavily-touched
/// tree shares few, and the memo's bookkeeping is then paid on every miss
/// and recovered on none. Those two regimes are indistinguishable from
/// outside the process, which is why a real import that got slower could not
/// be told apart from one that got faster without re-running it for hours.
///
/// `evicted` is the load-bearing field. Zero means the whole working set
/// fit and the hit rate is the best this corpus can give. Nonzero means the
/// budget is the binding constraint: entries were dropped and later
/// re-parsed, so raising `repostate.submanifest_memo_budget` would buy hit
/// rate. That is a smooth dial now — the memo evicts by recency and keeps
/// what is hot, rather than the earlier behaviour of filling once and then
/// freezing on whatever it happened to see first.
///
/// `degraded` is separate and much rarer: an allocation failed, so an answer
/// was returned uncached. It says nothing about the budget.
/// One in-flight memo line per `memo_report_stride` commits. The
/// stride is commits rather than time so re-runs of one corpus emit an
/// identical, comparable series.
const memo_report_stride: usize = 25_000;
fn reportSubManifestMemoProgress(self: *Verifier, processed: usize) void {
if (self.input.quiet) return;
const memo = if (self.sub_memo) |*m| m else return;
const total = memo.hits + memo.misses;
if (total == 0) return;
self.err.print(
"import-git: {d} commits verified; sub-manifest memo {d}/{d} hits ({d}%), {d} KiB retained ({d} KiB budget), {d} evicted\n",
.{ processed, memo.hits, total, memo.hits * 100 / total, memo.retained / 1024, memo.budget / 1024, memo.evicted },
) catch {};
}
fn reportSubManifestMemo(self: *Verifier) void {
if (self.input.quiet) return;
const memo = if (self.sub_memo) |*m| m else return; // no tree walk ran
const total = memo.hits + memo.misses;
if (total == 0) return;
self.err.print(
"import-git: sub-manifest memo {d}/{d} hits ({d}%), {d} KiB retained, {d} evicted{s}\n",
.{
memo.hits,
total,
memo.hits * 100 / total,
memo.retained / 1024,
memo.evicted,
if (memo.degraded) " (an allocation failed; some answers uncached)" else "",
},
) catch {};
}
/// A meter for one long clause, or null when the run is quiet.
fn meter(self: *const Verifier, what: []const u8, unit: []const u8, total: u64) ?Meter {
const p = self.input.progress orelse return null;
var m: Meter = .{ .out = p.out, .io = p.io, .what = what, .unit = unit, .total = total, .interval_ns = p.interval_ns, .stride = p.stride };
m.start();
return m;
}
fn fail(self: *Verifier, clause: Clause, comptime fmt: []const u8, args: anytype) Error {
self.err.print("import-git: verifier: {s}: " ++ fmt ++ "\n", .{clause.token()} ++ args) catch {};
return clause.err();
}
fn run(self: *Verifier) Error!void {
for (self.input.drop_names, 0..) |name, index| try self.drops.put(self.gpa, name, index);
for (self.input.witness.refs, 0..) |mapping, index| {
try self.witness_ref_index.put(self.gpa, mapping.canonical, index);
}
// One pass over the witnessed commits: every OID, by value, and which
// of the drop set some commit carried as a mirror candidate. The drop
// set is only allowed to name those.
var candidate_drops = try std.DynamicBitSetUnmanaged.initEmpty(self.gpa, self.input.drop_names.len);
defer candidate_drops.deinit(self.gpa);
{
var windows = try self.walk(false);
while (try windows.next()) for (windows.commits) |commit| {
try self.witness_oids.add(self.gpa, commit.oid);
for (commit.candidates) |slot| {
const candidate = slot orelse continue;
if (self.drops.get(candidate.name)) |at| candidate_drops.set(at);
}
};
}
try self.readMap();
// 0. The drop set is CANDIDATE-only. Every later clause proves the
// published store is what the import wrote minus this set; that
// only means what it should if the set itself holds nothing but
// mirror metadata. True by construction -- the classifier builds it
// out of `mirror_candidate_names` alone -- and checked anyway,
// because "the equation holds against whatever names it was
// handed" is a materially weaker statement than the one this file
// promises.
for (self.input.drop_names) |name| {
if (candidate_drops.isSet(self.drops.get(name).?)) continue;
return self.fail(.dropped_noncandidate_artifact, "{s} is in the drop set but is no commit's mirror candidate", .{name});
}
// 1. The mirror store holds every artifact of the drop set, and each
// reconstructs and hashes to its name. The artifacts about to be
// left out of the repository for good are the only ones nothing
// else looks at, and "these bytes really were what the classifier
// decided about" is the last chance to establish that. The mirror
// store's other artifacts were copied into the published store and
// are proven there.
try self.sweepMirror();
// 2. Every artifact of the store that will actually be published
// reconstructs and hashes to its name -- the import's one full
// content check, over the closed file, the copy that gets renamed.
// The same pass tallies the names the file's index holds.
const held = try self.sweepPublished();
// 3. The published store holds exactly what the import wrote into it,
// and none of the drop set.
try self.accountNames(held);
// 4-6. Every witnessed commit, resolved out of the published store and
// compared against what the import decided it was.
try self.verifyCommits();
// 7. Every branch, lightweight tag and annotated tag.
try self.verifyRefs();
// 8. Every provenance record, recomputed.
try self.verifyProvenance();
// 9. The repository view, derived from scratch.
try self.verifyRepositoryView();
// Anything clauses 4-9 declined to explain.
try self.reportDeferred();
}
// -- 1 and 2: reconstruct and rehash -----------------------------------
const Store = enum {
mirror,
published,
fn label(self: Store) []const u8 {
return switch (self) {
.mirror => "mirror",
.published => "published",
};
}
};
fn sweepMirror(self: *Verifier) Error!void {
const log = self.input.mirror;
const rids = try log.ridsInPersistedOrder(self.gpa);
defer self.gpa.free(rids);
var cache = logstore.PlaintextCache.init();
defer cache.deinit(self.gpa);
for (rids) |rid| {
var buffer: [hex_name_len]u8 = undefined;
const transient = hexName(log.hashOf(rid), &buffer);
// A name is listed AFTER its content passed, never instead of it.
if (self.drops.contains(transient)) try self.rehash(log, rid, transient, .mirror, &cache);
try self.mirror_names.add(self.gpa, try self.arena.dupe(u8, transient));
}
for (self.input.drop_names) |name| {
if (self.mirror_names.set.contains(name)) continue;
return self.fail(.mirror_rehash_mismatch, "mirror store: {s} is in the drop set and has no content", .{name});
}
}
/// Names are tallied, not listed: the store can hold millions, and the
/// tally is all clause 3 needs unless it disagrees.
fn sweepPublished(self: *Verifier) Error!logstore.NameTally {
const written = self.input.witness.written orelse {
return self.fail(.written_artifact_mismatch, "the import recorded no tally of what it wrote", .{});
};
const log = self.input.published;
const rids = try log.ridsInPersistedOrder(self.gpa);
defer self.gpa.free(rids);
var artifacts = self.meter("import-git: verifying", "artifacts", rids.len);
var sweep: Sweep = .{ .gpa = self.gpa, .log = log, .rids = rids, .min_run = self.input.min_run, .progress = if (artifacts) |*m| m else null };
defer sweep.deinit();
try sweep.run(self.input.workers);
self.chain_split = sweep.split;
sweep.split = null;
if (sweep.firstFault()) |found| {
const rid = rids[found.at];
var buffer: [hex_name_len]u8 = undefined;
return self.reportRehash(log, rid, hexName(log.hashOf(rid), &buffer), .published, found.fault);
}
var held = logstore.NameTally.init(written.key);
for (rids) |rid| held.add(log.hashOf(rid));
return held;
}
/// Clause 2's reconstruct-and-rehash pass. Bases persist strictly before
/// their deltas, so walking persisted order keeps each base hot and
/// collapses a delta chain's read amplification to about one pass over
/// the store. With threads, the store is split into runs of whole delta
/// trees (`chain_runs`), each walked in persisted order through its own
/// memo, so every run keeps that property over its share. A run reads the
/// frozen file through positional reads and its own memo and nothing
/// else, which is what lets runs share the store.
const Sweep = struct {
gpa: std.mem.Allocator,
log: *logstore.LogStore,
rids: []const logstore.Rid,
/// Null for the single pass.
split: ?zit.chain_runs.Runs = null,
/// Each run's first fault, by persisted index.
faults: []?Found = &.{},
min_run: usize,
/// Threads sharing the single pass's memo budget; set by `run`.
threads: usize = 1,
/// Advanced by every frame checked, from whichever thread checked it.
progress: ?*Meter = null,
/// Runs per pool thread. More than one so a run of large artifacts
/// does not leave the other threads idle at the end.
const runs_per_thread: usize = 4;
const Found = struct { at: usize, fault: RehashFault };
fn run(self: *Sweep, workers: ?*store_workers.Pool) Error!void {
const threads = if (workers) |pool| pool.threadCount() else 1;
const count = if (threads == 1) 1 else std.math.clamp(self.rids.len / self.min_run, 1, threads * runs_per_thread);
self.faults = try self.gpa.alloc(?Found, count);
@memset(self.faults, null);
// One run is the single pass, on this thread, through the same
// code: a pool of one thread verifies exactly as before.
if (count == 1) return runOne(self, 0);
self.split = zit.chain_runs.split(self.gpa, self.log, self.rids, count, workers.?) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
// Too many frames to index: the single pass, as above.
error.TooManyFrames => return runOne(self, 0),
};
self.threads = threads;
try workers.?.runEach(self.gpa, count, self, runOne);
}
fn deinit(self: *Sweep) void {
if (self.split) |*s| s.deinit(self.gpa);
self.gpa.free(self.faults);
}
/// Each run stops at its own first fault, and a fault is a fact about
/// one frame's bytes, not about which run read it -- so the lowest
/// among the runs' first faults is the one a single pass stops at.
fn firstFault(self: *const Sweep) ?Found {
var first: ?Found = null;
for (self.faults) |slot| {
const found = slot orelse continue;
if (first == null or found.at < first.?.at) first = found;
}
return first;
}
fn runOne(ctx: *anyopaque, index: usize) void {
const self: *Sweep = @ptrCast(@alignCast(ctx));
var cache = logstore.PlaintextCache.init();
// The single pass's memo budget, split among the runs alive at
// once: threads shorten the sweep, they do not multiply its footprint.
cache.max_bytes = logstore.PlaintextCache.default_max_bytes / self.threads;
cache.max_entries = logstore.PlaintextCache.default_max_entries / self.threads;
defer cache.deinit(self.gpa);
if (self.split) |*s| {
for (s.of(index)) |at| if (!self.check(index, at, &cache)) return;
} else {
for (0..self.rids.len) |at| if (!self.check(index, at, &cache)) return;
}
}
/// Rehash frame `at`; false, with the fault recorded, if it fails.
fn check(self: *Sweep, index: usize, at: usize, cache: *logstore.PlaintextCache) bool {
const rid = self.rids[at];
var buffer: [hex_name_len]u8 = undefined;
const name = hexName(self.log.hashOf(rid), &buffer);
if (self.progress) |m| m.add(1);
const fault = rehashFault(self.gpa, self.log, rid, name, cache) orelse return true;
self.faults[index] = .{ .at = at, .fault = fault };
return false;
}
};
/// Why an artifact failed to reconstruct and rehash. Recorded by whichever
/// thread found it and rendered by `reportRehash` on the verifier's own.
const RehashFault = union(enum) {
unreconstructable: logstore.ContentError,
no_content,
wrong_hash,
};
fn rehashFault(
gpa: std.mem.Allocator,
log: *logstore.LogStore,
rid: logstore.Rid,
name: []const u8,
cache: *logstore.PlaintextCache,
) ?RehashFault {
const content = log.contentUnmasked(gpa, rid, cache) catch |e| {
return .{ .unreconstructable = e };
} orelse return .no_content;
defer gpa.free(content);
if (!artifact.verify(name, content)) return .wrong_hash;
return null;
}
fn rehash(
self: *Verifier,
log: *logstore.LogStore,
rid: logstore.Rid,
name: []const u8,
store: Store,
cache: *logstore.PlaintextCache,
) Error!void {
const fault = rehashFault(self.gpa, log, rid, name, cache) orelse return;
return self.reportRehash(log, rid, name, store, fault);
}
fn reportRehash(
self: *Verifier,
log: *logstore.LogStore,
rid: logstore.Rid,
name: []const u8,
store: Store,
fault: RehashFault,
) Error {
const mismatch: Clause = switch (store) {
.mirror => .mirror_rehash_mismatch,
.published => .artifact_rehash_mismatch,
};
switch (fault) {
.unreconstructable => |e| {
// Reconstruction FAILED rather than produced wrong bytes. On a
// delta frame that is a statement about the chain -- a base
// that is not the one the payload was encoded against -- and
// it is a different repair from a frame whose bytes rotted.
const clause: Clause = if (log.kindOf(rid) == .delta) .delta_base_mismatch else mismatch;
return self.fail(clause, "{s} store: {s} cannot be reconstructed: {s}", .{
store.label(),
name,
@errorName(e),
});
},
// Unreachable today: `ridsInPersistedOrder` filters phantoms and
// `contentUnmasked` returns null for nothing else. Refused rather
// than skipped all the same -- the clause standing guard over an
// irreversible delete would otherwise report success for an
// artifact physically present in the store, which is the wrong
// direction to degrade in.
.no_content => return self.fail(mismatch, "{s} store: {s} has no content", .{ store.label(), name }),
.wrong_hash => return self.fail(mismatch, "{s} store: {s} does not hash to its name", .{
store.label(),
name,
}),
}
}
// -- 3: name accounting -------------------------------------------------
/// A check-in's name is derived from its bytes, so a check-in that appears
/// or disappears is never an independent fact: it is the shadow of a
/// content change, and clauses 4 to 8 are what explain content changes.
/// Reporting the shadow first would mean an artifact-set-preserving tree
/// edit and a reordered parent run both surfaced as "an artifact was
/// added", which tells an operator nothing about what actually moved.
///
/// So a difference among CHECK-INS is deferred -- never discarded. If no
/// later clause explains it, `reportDeferred` states it under its own name,
/// and the rename still blocks publication either way.
fn deferDifference(self: *Verifier, clause: Clause, log: *logstore.LogStore, name: []const u8) Error!void {
if (!self.namesCheckin(log, name)) return self.fail(clause, "{s}", .{name});
try self.deferred.append(self.gpa, .{ .clause = clause, .name = name });
}
fn accountNames(self: *Verifier, held: logstore.NameTally) Error!void {
// Upward: nothing the classifier authorized dropping reached the
// published store. This is the direction that would silently make the
// whole omission pointless, and it is checked first because it can
// never be a shadow of anything.
for (self.input.drop_names) |name| {
if (self.publishedHolds(name)) {
return self.fail(.retained_omitted_artifact, "{s} is in the drop set and in the published store", .{name});
}
}
// Downward, for the mirror store: everything outside the drop set was
// carried into the published store.
for (self.mirror_names.order.items) |name| {
if (self.drops.contains(name)) continue;
if (self.publishedHolds(name)) continue;
try self.deferDifference(.dropped_required_artifact, self.input.mirror, name);
}
// Everything else: the store holds exactly the names its writer
// recorded. Equal tallies are the whole answer; unequal ones are
// explained by name.
const written = self.input.witness.written.?;
if (held.eql(written)) return;
try self.diagnoseWritten(held, written);
}
/// The slow half of clause 3, reached only when the tallies disagree.
///
/// The tally says THAT the published store and the writer's record differ;
/// this says WHICH artifacts. What the import wrote is re-derived from the
/// witness: every check-in, every node and blob its tree reaches in the
/// published store, every control artifact, and every blob no commit
/// named. A written name the store lacks, and a held name nothing the
/// import recorded reaches, are each reported under their own clause, with
/// check-ins deferred like any other name difference. When the
/// difference is invisible to reachability the tally clause itself
/// refuses, so a disagreement is never admitted for lack of a name.
fn diagnoseWritten(self: *Verifier, held: logstore.NameTally, written: logstore.NameTally) Error!void {
var expected: Expected = .{};
defer expected.deinit(self.gpa);
var checkins: std.StringHashMapUnmanaged(void) = .empty;
defer checkins.deinit(self.gpa);
var windows = try self.walk(true);
while (try windows.next()) for (windows.commits, windows.records) |*commit, record| {
try expected.add(self, commit.checkin);
try checkins.put(self.gpa, try self.arena.dupe(u8, commit.checkin), {});
// Walked from the name the published store carries for it, so a
// renamed check-in still accounts for the tree it reaches.
try self.expectTree(&expected, publishedNameOf(commit, record), 0);
};
for (self.input.witness.controls) |name| try expected.add(self, name);
for (self.input.witness.loose_blobs) |name| try expected.add(self, name);
const before = self.deferred.items.len;
for (expected.order.items) |name| {
if (self.publishedHolds(name)) continue;
if (checkins.contains(name)) {
try self.deferred.append(self.gpa, .{ .clause = .dropped_required_artifact, .name = name });
continue;
}
return self.fail(.dropped_required_artifact, "{s} was written by the import and is not in the published store", .{name});
}
const log = self.input.published;
const rids = try log.ridsInPersistedOrder(self.gpa);
defer self.gpa.free(rids);
for (rids) |rid| {
var buffer: [hex_name_len]u8 = undefined;
const transient = hexName(log.hashOf(rid), &buffer);
if (expected.set.contains(transient)) continue;
try self.deferDifference(.added_artifact, log, try self.arena.dupe(u8, transient));
}
if (self.deferred.items.len != before) return;
return self.fail(.written_artifact_mismatch, "the published store holds {d} artifacts and the import wrote {d}, under different name fingerprints, and no single artifact accounts for it", .{
held.count,
written.count,
});
}
/// Names the import is expected to have written, in discovery order.
const Expected = struct {
order: std.ArrayList([]const u8) = .empty,
set: std.StringHashMapUnmanaged(void) = .empty,
fn add(self: *Expected, verifier: *Verifier, name: []const u8) Error!void {
if (self.set.contains(name)) return;
const owned = try verifier.arena.dupe(u8, name);
try self.set.put(verifier.gpa, owned, {});
try self.order.append(verifier.gpa, owned);
}
fn deinit(self: *Expected, gpa: std.mem.Allocator) void {
self.order.deinit(gpa);
self.set.deinit(gpa);
}
};
/// Every blob and node `name` reaches in the published store. A node
/// already expected was walked by whoever expected it first, so each is
/// read once however many check-ins share it.
fn expectTree(self: *Verifier, expected: *Expected, name: []const u8, depth: usize) Error!void {
const bytes = (self.contentByName(self.input.published, name) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return,
}) orelse return;
defer self.gpa.free(bytes);
var manifest = artifact.parse(self.gpa, bytes) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return,
};
defer manifest.deinit();
for (manifest.f) |card| if (card.hash) |hash| try expected.add(self, hash);
if (depth >= artifact.max_tree_depth) return;
for (manifest.s) |card| {
if (expected.set.contains(card.hash)) continue;
try expected.add(self, card.hash);
try self.expectTree(expected, card.hash, depth + 1);
}
}
fn publishedHolds(self: *Verifier, name: []const u8) bool {
var digest: [logstore.max_hash_len]u8 = undefined;
const raw = rawName(name, &digest) orelse return false;
return self.input.published.lookup(raw) != null;
}
/// First deferral wins, for the same reason the pipeline stops at its first
/// refusing clause: one difference is an explanation, a list of them is a
/// diff. If any survived to here, none of clauses 4 to 9 accounted for it.
fn reportDeferred(self: *Verifier) Error!void {
if (self.deferred.items.len == 0) return;
const item = self.deferred.items[0];
return self.fail(item.clause, "{s} (no content clause accounts for it)", .{item.name});
}
// -- 4 to 6: every commit ----------------------------------------------
fn verifyCommits(self: *Verifier) Error!void {
const threads = if (self.input.workers) |pool| pool.threadCount() else 1;
// With threads, each window of check-ins is read and parsed on the
// pool first, and this loop then consumes the results strictly in
// stream order -- the order every clause below depends on. Without,
// each is read here, through the published store's memo, as before.
const window_len: usize = @intCast(@min(self.input.witness.commit_count, self.windowLen()));
var prefetch: ?CheckinPrefetch = null;
defer if (prefetch) |*p| {
self.gpa.free(p.names);
self.gpa.free(p.reads);
};
if (threads > 1) {
const names = try self.gpa.alloc([]const u8, window_len);
errdefer self.gpa.free(names);
prefetch = .{
.gpa = self.gpa,
.log = self.input.published,
.names = names,
.reads = try self.gpa.alloc(CheckinRead, window_len),
.threads = threads,
};
}
// Stream order, and that is what makes commit-major iteration safe: a
// parent always precedes its children, so a check-in whose content
// changed trips its OWN tree or parent clause before any child's parent
// clause can trip on the resulting rename.
var processed: usize = 0;
var commits = self.meter("import-git: verifying", "commits", self.input.witness.commit_count);
var windows = try self.walk(true);
while (try windows.next()) {
if (prefetch) |*p| try p.fill(self, windows.commits, windows.records);
defer if (prefetch) |*p| p.release(self);
for (windows.commits, windows.records, 0..) |*commit, record, at| {
processed += 1;
// The memo line every `memo_report_stride` commits, not only
// at the end: a budget-starved run (evictions climbing against a
// wide tree) is visible in minutes instead of discovered after
// hours. Same fields as the final report, so the two read as one
// series.
if (processed % memo_report_stride == 0) self.reportSubManifestMemoProgress(processed);
if (commits) |*m| m.add(1);
var unread: CheckinRead = .pending;
const read: *CheckinRead = if (prefetch) |*p| &p.reads[at] else &unread;
try self.verifyCommit(commit, record, read);
}
}
}
fn verifyCommit(self: *Verifier, commit: *const git_import.WitnessCommit, record: ?MapCommit, read: *CheckinRead) Error!void {
const name = publishedNameOf(commit, record);
try self.resolve(commit.checkin, name);
// The one assertion the map exists to make -- "this Git commit
// became THIS Zit check-in" -- and the map is a file under test,
// not an authority. Without this a tampered map redirects the
// verifier instead of being caught by it: exchange the `zit` values
// of two commits that share a tree, a parent run and a selected
// identity and clauses 4 to 8 all read the wrong-but-consistent
// check-in and pass.
//
// Deferred rather than immediate, for `deferDifference`'s reason:
// the two store faults that permute a check-in's content rename it
// and carry the rename into the map, so the tree or parent clause
// must be what fires. Nothing escapes -- `reportDeferred` states it
// when nothing else did.
if (!std.mem.eql(u8, name, commit.checkin)) {
try self.deferred.append(self.gpa, .{ .clause = .checkin_name_mismatch, .name = try self.arena.dupe(u8, name) });
}
// Not prefetched (no pool), or too large to hold: read it here. A
// read made here is freed here; the pool frees the ones it made.
const here = read.* == .pending or read.* == .too_large;
if (here) read.* = readCheckin(self.gpa, self.input.published, name, &self.published_cache, null);
const parsed = switch (read.*) {
.pending, .too_large => unreachable,
.out_of_memory => return error.OutOfMemory,
.unreconstructable => return self.fail(.checkin_unresolvable, "{s}: {s} cannot be reconstructed", .{ self.describe(commit), name }),
.absent => return self.fail(.checkin_unresolvable, "{s}: {s} is absent from the published store", .{ self.describe(commit), name }),
.unparsable => |e| return self.fail(.checkin_unresolvable, "{s}: {s} does not parse: {s}", .{ self.describe(commit), name, @errorName(e) }),
.parsed => |*p| p,
};
defer if (here) read.release(self.gpa);
const manifest = &parsed.manifest;
if (manifest.kind != .manifest) {
return self.fail(.checkin_unresolvable, "{s}: {s} is a {s}, not a check-in", .{ self.describe(commit), name, @tagName(manifest.kind) });
}
// A tree root's digest is structural: the published manifest's own
// S-run, whose hashes are content addresses that commit to the
// whole file set transitively. No flatten -- the whole-store
// clause already reconstructs and re-hashes every node these
// hashes name, so enumerating 100K files per commit here bought
// nothing that check does not. A flat root keeps the per-file
// digest, computed against the PUBLISHED store -- the import's
// own digest covered a commit-local staging store that no longer
// exists.
if (manifest.s.len != 0) {
try self.verifyStructuralTreeDigest(commit, name, manifest.s);
} else {
try self.verifyTreeDigest(commit, name, manifest.f);
}
try self.verifyParents(commit, name, manifest);
try self.verifyAttribution(commit, record, name, manifest);
}
/// Record what `resolved` answers for `checkin`: the last commit that
/// carried it decides, as a full table's `put` would.
fn resolve(self: *Verifier, checkin: []const u8, name: []const u8) Error!void {
if (std.mem.eql(u8, checkin, name)) {
_ = self.resolved.remove(checkin);
return;
}
const slot = try self.resolved.getOrPut(self.gpa, checkin);
if (!slot.found_existing) slot.key_ptr.* = try self.arena.dupe(u8, checkin);
slot.value_ptr.* = try self.arena.dupe(u8, name);
}
/// One check-in as read for clauses 4 to 6, by whichever thread read it.
const CheckinRead = union(enum) {
/// Not read yet.
pending,
/// Present, and larger than its run could still hold; read again in turn.
too_large,
out_of_memory,
unreconstructable,
absent,
unparsable: artifact.ParseError,
parsed: struct { bytes: []u8, manifest: artifact.Manifest },
fn release(self: *CheckinRead, gpa: std.mem.Allocator) void {
switch (self.*) {
.parsed => |*p| {
p.manifest.deinit();
gpa.free(p.bytes);
},
else => {},
}
self.* = .pending;
}
};
/// `contentByName` and the strict parse, touching nothing but `log` and
/// `cache`, so a pool thread can run it. A check-in larger than
/// `hold_max` is left for its turn.
fn readCheckin(gpa: std.mem.Allocator, log: *logstore.LogStore, name: []const u8, cache: *logstore.PlaintextCache, hold_max: ?usize) CheckinRead {
var digest: [logstore.max_hash_len]u8 = undefined;
const raw = rawName(name, &digest) orelse return .absent;
const rid = log.lookup(raw) orelse return .absent;
const bytes = (log.contentUnmasked(gpa, rid, cache) catch |e| switch (e) {
error.OutOfMemory => return .out_of_memory,
else => return .unreconstructable,
}) orelse return .absent;
if (hold_max) |max| if (bytes.len > max) {
gpa.free(bytes);
return .too_large;
};
const manifest = artifact.parse(gpa, bytes) catch |e| {
gpa.free(bytes);
return .{ .unparsable = e };
};
return .{ .parsed = .{ .bytes = bytes, .manifest = manifest } };
}
/// Clauses 4 to 6's reads, a window of commits at a time, split into
/// contiguous runs of stream order. A check-in deltas against its parent's
/// as a rule, so a run walking its slice keeps that chain hot in its own
/// memo and rebuilds it only once, at its start.
const CheckinPrefetch = struct {
gpa: std.mem.Allocator,
log: *logstore.LogStore,
names: [][]const u8,
reads: []CheckinRead,
threads: usize,
len: usize = 0,
const runs_per_thread: usize = 2;
/// A flat tree's check-in names every file, and a window of those
/// would hold the tree once per commit.
const hold_max: usize = 256 << 10;
/// The most check-in bytes one window holds, shared evenly by its
/// runs: a window of check-ins each just under `hold_max` would
/// otherwise hold a gigabyte. Past its share a run leaves the rest
/// for their turn, read on the verifier's thread.
const held_budget: usize = 64 << 20;
fn fill(self: *CheckinPrefetch, verifier: *Verifier, commits: []const git_import.WitnessCommit, records: []const ?MapCommit) Error!void {
self.len = commits.len;
for (commits, records, 0..) |*commit, record, at| {
self.names[at] = publishedNameOf(commit, record);
self.reads[at] = .pending;
}
const runs = @max(1, @min(commits.len, self.threads * runs_per_thread));
try verifier.input.workers.?.runEach(self.gpa, runs, self, runOne);
}
/// Free the window's reads on the pool that made them: freed here, on
/// the verifier's thread, each would strand on a free list the pool
/// never draws from (`store_workers.in_task`).
fn release(self: *CheckinPrefetch, verifier: *Verifier) void {
const runs = @max(1, @min(self.len, self.threads * runs_per_thread));
verifier.input.workers.?.runEach(self.gpa, runs, self, releaseOne) catch {
for (self.reads[0..self.len]) |*read| read.release(self.gpa);
};
}
fn releaseOne(ctx: *anyopaque, index: usize) void {
const self: *CheckinPrefetch = @ptrCast(@alignCast(ctx));
const runs = @max(1, @min(self.len, self.threads * runs_per_thread));
for (self.reads[self.len * index / runs .. self.len * (index + 1) / runs]) |*read| read.release(self.gpa);
}
fn runOne(ctx: *anyopaque, index: usize) void {
const self: *CheckinPrefetch = @ptrCast(@alignCast(ctx));
const runs = @max(1, @min(self.len, self.threads * runs_per_thread));
var cache = logstore.PlaintextCache.init();
// The single pass's memo budget, split among the runs alive at once.
cache.max_bytes = logstore.PlaintextCache.default_max_bytes / self.threads;
cache.max_entries = logstore.PlaintextCache.default_max_entries / self.threads;
defer cache.deinit(self.gpa);
const share = held_budget / runs;
var held: usize = 0;
for (self.len * index / runs..self.len * (index + 1) / runs) |at| {
const read = &self.reads[at];
read.* = readCheckin(self.gpa, self.log, self.names[at], &cache, @min(hold_max, share - held));
if (read.* == .parsed) held += read.parsed.bytes.len;
}
}
};
/// The name the published store carries for `commit`, given its map
/// record. The finalized map is the authority when it names one, because
/// the map ships WITH the store and an operator reads them together; the
/// witness is the fallback for a commit the stream carried no OID for,
/// which therefore has no map line at all.
fn publishedNameOf(commit: *const git_import.WitnessCommit, record: ?MapCommit) []const u8 {
const found = record orelse return commit.checkin;
return found.zit;
}
fn describe(self: *Verifier, commit: *const git_import.WitnessCommit) []const u8 {
_ = self;
return commit.oid orelse "(no original-oid)";
}
fn verifyStructuralTreeDigest(
self: *Verifier,
commit: *const git_import.WitnessCommit,
name: []const u8,
subs: []const artifact.SCard,
) Error!void {
const recomputed = git_import.structuralTreeDigest(subs);
if (!std.mem.eql(u8, &recomputed, &commit.tree_digest)) {
var want: [64]u8 = undefined;
var got: [64]u8 = undefined;
return self.fail(.normalized_tree_digest_mismatch, "{s}: {s} carries S-run digest {s}, imported as {s}", .{
self.describe(commit),
name,
git_import.formatDigest(recomputed, &got),
git_import.formatDigest(commit.tree_digest, &want),
});
}
}
fn verifyTreeDigest(
self: *Verifier,
commit: *const git_import.WitnessCommit,
name: []const u8,
files: []const artifact.FCard,
) Error!void {
var builder = git_import.TreeDigestBuilder.init();
var previous: ?[]const u8 = null;
for (files) |card| {
// Checked rather than asserted: the builder asserts canonical order
// because its producers guarantee it, and this one is reading a
// store it is not allowed to trust.
if (previous) |last| {
if (artifact.compareBytes(last, card.filename) != .lt) {
return self.fail(.checkin_unresolvable, "{s}: {s} enumerates '{s}' out of canonical order", .{ self.describe(commit), name, card.filename });
}
}
previous = card.filename;
const hash = card.hash orelse {
return self.fail(.checkin_unresolvable, "{s}: {s} has no content for '{s}'", .{ self.describe(commit), name, card.filename });
};
const facts = (try self.contentFacts(hash)) orelse {
return self.fail(.normalized_tree_digest_mismatch, "{s}: {s} names absent content {s} at '{s}'", .{ self.describe(commit), name, hash, card.filename });
};
builder.entry(card.filename, card.mode, facts.length, facts.digest);
}
const recomputed = builder.final();
if (!std.mem.eql(u8, &recomputed, &commit.tree_digest)) {
var want: [64]u8 = undefined;
var got: [64]u8 = undefined;
return self.fail(.normalized_tree_digest_mismatch, "{s}: {s} flattens to {s}, imported as {s}", .{
self.describe(commit),
name,
git_import.formatDigest(recomputed, &got),
git_import.formatDigest(commit.tree_digest, &want),
});
}
}
fn verifyParents(
self: *Verifier,
commit: *const git_import.WitnessCommit,
name: []const u8,
manifest: *const artifact.Manifest,
) Error!void {
if (manifest.p.len > 1) {
return self.fail(.parent_order_mismatch, "{s}: {s} carries {d} P cards", .{ self.describe(commit), name, manifest.p.len });
}
const text: []const u8 = if (manifest.p.len == 0) "" else (manifest.p[0].parents orelse "");
var seen: usize = 0;
var parts = std.mem.splitScalar(u8, text, ' ');
while (parts.next()) |stored| {
if (stored.len == 0) continue;
if (seen >= commit.parents.len) {
return self.fail(.parent_order_mismatch, "{s}: {s} names more than the {d} imported parents", .{ self.describe(commit), name, commit.parents.len });
}
// Through the same substitution the check-in itself went through,
// so a renamed parent is accounted for once -- as its own deferred
// difference -- rather than a second time here.
const expected = self.resolved.get(commit.parents[seen]) orelse commit.parents[seen];
if (!std.mem.eql(u8, stored, expected)) {
return self.fail(.parent_order_mismatch, "{s}: {s} names {s} as parent {d}, imported with {s}", .{ self.describe(commit), name, stored, seen, expected });
}
seen += 1;
}
if (seen != commit.parents.len) {
return self.fail(.parent_order_mismatch, "{s}: {s} names {d} parents, imported with {d}", .{ self.describe(commit), name, seen, commit.parents.len });
}
}
fn verifyAttribution(
self: *Verifier,
commit: *const git_import.WitnessCommit,
record: ?MapCommit,
name: []const u8,
manifest: *const artifact.Manifest,
) Error!void {
const selected = self.input.witness.selected(commit);
const card = manifest.u orelse {
return self.fail(.attribution_mismatch, "{s}: {s} carries no U card", .{ self.describe(commit), name });
};
const login = git_import.loginFor(selected.name, selected.email);
if (!std.mem.eql(u8, card.login, login)) {
return self.fail(.attribution_mismatch, "{s}: {s} is attributed to '{s}', imported as '{s}'", .{ self.describe(commit), name, card.login, login });
}
// The record half. The store carries only the SELECTED identity, so a
// map that exchanged the two would leave the store untouched and every
// artifact hashing correctly -- this is the only clause that can see it.
const found = record orelse return;
try self.expectIdentity(commit, "author", found.author, commit.author);
try self.expectIdentity(commit, "committer", found.committer, commit.committer);
}
fn expectIdentity(
self: *Verifier,
commit: *const git_import.WitnessCommit,
role: []const u8,
recorded: MapIdentity,
imported: git_import.WitnessIdentity,
) Error!void {
if (std.mem.eql(u8, recorded.name, imported.name) and
std.mem.eql(u8, recorded.email, imported.email) and
recorded.timestamp == imported.timestamp and
recorded.timezone_offset_minutes == imported.timezone_offset_minutes) return;
return self.fail(.attribution_mismatch, "{s}: recorded {s} '{s} <{s}> {d}' but imported '{s} <{s}> {d}'", .{
self.describe(commit),
role,
recorded.name,
recorded.email,
recorded.timestamp,
imported.name,
imported.email,
imported.timestamp,
});
}
// -- 7: refs, lightweight tags and annotated tags -----------------------
fn verifyRefs(self: *Verifier) Error!void {
var rows: std.ArrayList(RefRow) = .empty;
defer rows.deinit(self.gpa);
var row_index: std.StringHashMapUnmanaged(usize) = .empty;
defer row_index.deinit(self.gpa);
try self.parseRefs(&rows);
// Indexed like every other lookup in this file. A mirror can carry tens
// of thousands of refs, and this lookup sits inside the loop over all
// of them.
for (rows.items, 0..) |row, index| try row_index.put(self.gpa, row.canonical, index);
for (self.input.witness.refs) |mapping| {
const clause = clauseForRef(mapping.canonical);
const expected = self.resolved.get(mapping.target) orelse mapping.target;
const row = findRow(&row_index, rows.items, mapping.canonical) orelse {
return self.fail(clause, "'{s}' is missing from refs", .{mapping.canonical});
};
if (!std.mem.eql(u8, row.target, expected)) {
return self.fail(clause, "'{s}' points at {s}, imported at {s}", .{ mapping.canonical, row.target, expected });
}
// A ref naming something the published store does not hold is a
// repository nobody can check out, however well its rows compare.
if (!self.publishedHolds(row.target)) {
return self.fail(clause, "'{s}' points at {s}, which the published store does not hold", .{ mapping.canonical, row.target });
}
// The map's own copy of the same mapping. Annotated tags are
// deliberately absent from its `ref` records and appear as `tag`
// records instead, so each is looked for where it belongs.
const recorded = if (mapping.annotated)
self.findMapping(&self.map_tag_index, self.map_tags.items, mapping.canonical)
else
self.findMapping(&self.map_ref_index, self.map_refs.items, mapping.canonical);
const found = recorded orelse {
return self.fail(clause, "'{s}' has no record in the import map", .{mapping.canonical});
};
if (!std.mem.eql(u8, found.target, expected)) {
return self.fail(clause, "'{s}' is recorded at {s}, imported at {s}", .{ mapping.canonical, found.target, expected });
}
}
for (rows.items) |row| {
if (self.witnessedRef(row.canonical) != null) continue;
return self.fail(clauseForRef(row.canonical), "'{s}' is in refs but was never imported", .{row.canonical});
}
// The same "nothing the import did not produce" sweep the commit
// records get, over the map's own ref and tag records. Without it a map
// could carry a mapping for a tag that was never imported and every
// clause above would still pass -- each one only ever reads the map
// FROM the witness side.
for (self.map_refs.items) |record| {
if (self.witnessedRef(record.canonical) != null) continue;
return self.fail(.ref_target_mismatch, "the import map records ref '{s}', which was never imported", .{record.canonical});
}
for (self.map_tags.items) |record| {
if (self.witnessedRef(record.canonical) != null) continue;
return self.fail(.tag_mismatch, "the import map records tag '{s}', which was never imported", .{record.canonical});
}
}
fn parseRefs(self: *Verifier, rows: *std.ArrayList(RefRow)) Error!void {
const body = self.input.refs;
if (!std.mem.startsWith(u8, body, self.input.refs_header)) {
return self.fail(.ref_target_mismatch, "refs does not open with its version header", .{});
}
var lines = std.mem.splitScalar(u8, body[self.input.refs_header.len..], '\n');
while (lines.next()) |line| {
if (line.len == 0) continue;
const space = std.mem.lastIndexOfScalar(u8, line, ' ') orelse {
return self.fail(.ref_target_mismatch, "refs row '{s}' has no target", .{line});
};
try rows.append(self.gpa, .{ .canonical = line[0..space], .target = line[space + 1 ..] });
}
}
fn witnessedRef(self: *Verifier, canonical: []const u8) ?*const git_import.WitnessRef {
const index = self.witness_ref_index.get(canonical) orelse return null;
return &self.input.witness.refs[index];
}
/// Which clause owns a ref row. A tag and a branch are different kinds of
/// mapping and a change that turns one into the other is a tag change, not
/// a retarget -- so the clause follows the canonical namespace.
fn clauseForRef(canonical: []const u8) Clause {
return if (std.mem.startsWith(u8, canonical, "sym-tags/")) .tag_mismatch else .ref_target_mismatch;
}
fn findRow(index: *const std.StringHashMapUnmanaged(usize), rows: []const RefRow, canonical: []const u8) ?RefRow {
const at = index.get(canonical) orelse return null;
return rows[at];
}
fn findMapping(
self: *Verifier,
index: *const std.StringHashMapUnmanaged(usize),
items: []const MapMapping,
canonical: []const u8,
) ?MapMapping {
_ = self;
const at = index.get(canonical) orelse return null;
return items[at];
}
// -- 8: provenance ------------------------------------------------------
fn verifyProvenance(self: *Verifier) Error!void {
var windows = try self.walk(true);
while (try windows.next()) for (windows.commits, windows.records) |*commit, found| {
const oid = commit.oid orelse continue;
const record = found orelse {
return self.fail(.provenance_record_mismatch, "{s} has no commit record in the import map", .{oid});
};
var buffer: [64]u8 = undefined;
const tree = git_import.formatDigest(commit.tree_digest, &buffer);
if (!std.mem.eql(u8, record.tree_digest, tree)) {
return self.fail(.provenance_record_mismatch, "{s} records tree digest {s}, imported as {s}", .{ oid, record.tree_digest, tree });
}
// Two comparisons, and they are worth different amounts. The
// recompute runs `provenanceDigestOf` over the witnessed inputs and
// compares it to the digest the import recorded from the SAME
// inputs -- so it can only fail if `captureWitness` dropped a digest
// input, which is a narrow but real class of bug and the only one
// reachable from inside a verifier. The comparison that follows it,
// against the value the JSONL actually carries, is the one that
// sees a tampered line.
const recomputed = self.provenanceDigestOf(commit);
var want: [64]u8 = undefined;
const rendered = git_import.formatDigest(recomputed, &want);
if (!std.mem.eql(u8, &recomputed, &commit.provenance_digest)) {
var imported: [64]u8 = undefined;
return self.fail(.provenance_digest_mismatch, "{s} recomputes to {s}, imported as {s}", .{
oid,
rendered,
git_import.formatDigest(commit.provenance_digest, &imported),
});
}
if (!std.mem.eql(u8, record.provenance_digest, rendered)) {
return self.fail(.provenance_digest_mismatch, "{s} records {s}, recomputes to {s}", .{ oid, record.provenance_digest, rendered });
}
};
// Nothing in the map that the import did not produce.
if (self.map_unwitnessed) |oid| {
return self.fail(.provenance_record_mismatch, "the import map records {s}, which was never imported", .{oid});
}
// The whole-import digest. Its inputs -- annotated-tag-origin refs and
// every tagger identity -- are deliberately not all present in the
// JSONL, so the summary's value is checked against the witnessed one
// rather than re-derived from the file it sits in.
const summary = self.map_summary_digest orelse {
return self.fail(.provenance_record_mismatch, "the import map carries no summary record", .{});
};
var buffer: [64]u8 = undefined;
const expected = git_import.formatDigest(self.input.witness.provenance_digest, &buffer);
if (!std.mem.eql(u8, summary, expected)) {
return self.fail(.provenance_digest_mismatch, "the summary records {s}, imported as {s}", .{ summary, expected });
}
}
fn provenanceDigestOf(self: *Verifier, commit: *const git_import.WitnessCommit) git_import.Digest {
const selected = self.input.witness.selected(commit);
var candidates: [git_import.mirror_candidate_names.len]git_import.ProvenanceCandidate = undefined;
var present: usize = 0;
for (commit.candidates) |slot| {
const candidate = slot orelse continue;
candidates[present] = .{
.path = candidate.path,
.name = candidate.name,
.oid = candidate.oid orelse "",
};
present += 1;
}
return git_import.provenanceDigestOf(.{
.oid = commit.oid orelse "",
.parent_oids = commit.parent_oids,
.attribution = @tagName(self.input.witness.attribution),
.identity_name = selected.name,
.identity_email = selected.email,
.identity_timestamp = selected.timestamp,
.identity_zone_minutes = selected.timezone_offset_minutes,
.tree_digest = commit.tree_digest,
.decision = @tagName(commit.decision),
.reason = if (commit.decision == .retained) @tagName(commit.reason) else "",
.candidates = candidates[0..present],
.checkin = commit.checkin,
});
}
// -- 9: the repository, derived from scratch ----------------------------
fn verifyRepositoryView(self: *Verifier) Error!void {
var view: repoview.RepoView = undefined;
// `load` is transactional -- it tears itself down on failure -- so the
// `defer` belongs strictly after the `catch`, and a `deinit` reached on
// the failure path would be a double free rather than the leak it looks
// like.
//
// Its own reconstruction and parse, never the sweep's: the point of
// the clause is the repository's derive reading the store from
// scratch, so a decode the sweep got wrong cannot be handed to it.
// The pool only spreads that derive's reads, over the sweep's split of
// the store into delta trees -- a choice of thread per frame, nothing
// read -- and the crosslink stays in persisted order on this thread.
var frames = self.meter("import-git: deriving", "frames", 0);
const loaded = if (self.input.workers) |pool|
view.loadPooled(self.gpa, self.input.published, pool, if (self.chain_split) |*s| s else null, if (frames) |*m| m else null)
else
view.load(self.gpa, self.input.published);
loaded catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return self.fail(.repository_view_mismatch, "the published store does not crosslink: {s}", .{@errorName(e)}),
};
defer view.deinit(self.gpa);
if (!std.mem.eql(u8, view.main_branch, self.input.witness.primary_branch)) {
return self.fail(.repository_view_mismatch, "derives default branch '{s}', imported as '{s}'", .{ view.main_branch, self.input.witness.primary_branch });
}
// Every published check-in is a check-in to the DERIVED view too, not
// merely a well-formed artifact that parses as one. This is what makes
// the crosslink pass a statement rather than a smoke test: a manifest
// the repository declines to crosslink -- no D card the reader accepts,
// no event row -- is a check-in `zit timeline` will never show and
// `zit update` will never reach, however cleanly it hashes.
//
// A name two commits share is checked twice, with the same answer
// both times: cheaper than a table of every name beside the view.
var windows = try self.walk(false);
while (try windows.next()) for (windows.commits) |*commit| {
const name = self.resolved.get(commit.checkin) orelse commit.checkin;
const rid = view.lookup(name) orelse {
return self.fail(.repository_view_mismatch, "{s}: {s} is not in the derived repository", .{ self.describe(commit), name });
};
if (view.state.mtimeOf(rid) == null) {
return self.fail(.repository_view_mismatch, "{s}: {s} crosslinks no check-in event", .{ self.describe(commit), name });
}
};
if (self.input.checkout_branch.len == 0) {
return self.fail(.checkout_invalid, "the published checkout names no branch", .{});
}
if (!std.mem.eql(u8, self.input.checkout_branch, self.input.witness.primary_branch)) {
return self.fail(.checkout_invalid, "the published checkout names '{s}', imported as '{s}'", .{ self.input.checkout_branch, self.input.witness.primary_branch });
}
}
// -- the finalized map --------------------------------------------------
fn readMap(self: *Verifier) Error!void {
self.map_record_at = try self.gpa.alloc(MapRecordAt, @intCast(self.input.witness.commit_count));
@memset(self.map_record_at, .{ .offset = no_record, .tag = 0 });
const map = self.input.map;
while (true) {
const offset = map.logicalPos();
const line = try self.takeMapLine() orelse break;
if (line.len == 0) continue;
var parsed = try self.parseMapLine(line);
defer parsed.deinit();
const object = parsed.value.object;
const kind = stringOf(object.get("kind")) orelse {
return self.fail(.map_unreadable, "a record carries no kind", .{});
};
if (std.mem.eql(u8, kind, "commit")) {
// Parsed whole, so a malformed record is refused here as it
// always was; only where it sits is kept.
const record = try self.mapCommit(parsed.arena.allocator(), object);
if (self.witness_oids.find(record.oid)) |index| {
self.map_record_at[index] = .{ .offset = offset, .tag = self.recordTag(line) };
} else if (self.map_unwitnessed == null) {
self.map_unwitnessed = try self.arena.dupe(u8, record.oid);
}
} else if (std.mem.eql(u8, kind, "ref")) {
const record: MapMapping = .{
.canonical = try self.mapString(self.arena, object, "canonical"),
.target = try self.mapString(self.arena, object, "zit"),
};
try self.map_ref_index.put(self.gpa, record.canonical, self.map_refs.items.len);
try self.map_refs.append(self.gpa, record);
} else if (std.mem.eql(u8, kind, "tag")) {
const record: MapMapping = .{
.canonical = try self.mapString(self.arena, object, "canonical"),
.target = try self.mapString(self.arena, object, "target_zit"),
};
try self.map_tag_index.put(self.gpa, record.canonical, self.map_tags.items.len);
try self.map_tags.append(self.gpa, record);
} else if (std.mem.eql(u8, kind, "summary")) {
self.map_summary_digest = try self.mapString(self.arena, object, "provenance_digest");
}
}
}
const no_record = std.math.maxInt(u64);
const MapRecordAt = struct {
offset: u64,
/// Of the record's bytes as `readMap` admitted them, keyed by the run's
/// witness key, so no rewrite can be chosen to keep it.
tag: u64,
};
fn recordTag(self: *const Verifier, line: []const u8) u64 {
const Mac = std.crypto.auth.siphash.SipHash64(2, 4);
var mac = Mac.init(&self.input.witness.commit_key);
mac.update("map record");
mac.update(line);
return mac.finalInt();
}
/// The commit record `oid` resolves to, read again from where `readMap`
/// found it, with every string in `arena`.
fn mapCommitOf(self: *Verifier, arena: std.mem.Allocator, oid: []const u8) Error!?MapCommit {
const index = self.witness_oids.find(oid) orelse return null;
const at = self.map_record_at[index];
if (at.offset == no_record) return null;
self.input.map.seekTo(at.offset) catch return self.fail(.map_unreadable, "the map does not read", .{});
// The same bytes `readMap` admitted, or every clause after it would
// be judging a different map from the one the index was built over.
const line = try self.takeMapLine() orelse return self.mapMoved();
if (self.recordTag(line) != at.tag) return self.mapMoved();
var parsed = try self.parseMapLine(line);
defer parsed.deinit();
return try self.mapCommit(arena, parsed.value.object);
}
fn mapMoved(self: *Verifier) Error {
return self.fail(.map_unreadable, "the map changed while it was read", .{});
}
fn takeMapLine(self: *Verifier) Error!?[]u8 {
const map = &self.input.map.interface;
return map.takeDelimiter('\n') catch |e| switch (e) {
error.ReadFailed => return self.fail(.map_unreadable, "the map does not read", .{}),
error.StreamTooLong => return self.fail(.map_unreadable, "a record does not fit in {d} bytes", .{map.buffer.len}),
};
}
/// One record, as an object.
fn parseMapLine(self: *Verifier, line: []const u8) Error!std.json.Parsed(std.json.Value) {
var parsed = std.json.parseFromSlice(std.json.Value, self.gpa, line, .{}) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return self.fail(.map_unreadable, "a record does not parse: {s}", .{@errorName(e)}),
};
if (parsed.value != .object) {
parsed.deinit();
return self.fail(.map_unreadable, "a record is not an object", .{});
}
return parsed;
}
fn mapCommit(self: *Verifier, arena: std.mem.Allocator, object: std.json.ObjectMap) Error!MapCommit {
return .{
.oid = try self.mapString(arena, object, "oid"),
.zit = try self.mapString(arena, object, "zit"),
.author = try self.mapIdentity(arena, object, "author"),
.committer = try self.mapIdentity(arena, object, "committer"),
.tree_digest = try self.mapString(arena, object, "tree_digest"),
.provenance_digest = try self.mapString(arena, object, "provenance_digest"),
};
}
fn mapString(self: *Verifier, arena: std.mem.Allocator, object: std.json.ObjectMap, field: []const u8) Error![]const u8 {
const value = object.get(field) orelse {
return self.fail(.map_unreadable, "a record carries no '{s}'", .{field});
};
switch (value) {
// Duped: the parse arena dies with the line.
.string => |text| return try arena.dupe(u8, text),
// Bytes that are not UTF-8, written as hex.
.object => |inner| {
const hex = stringOf(inner.get("bytes")) orelse return self.fail(.map_unreadable, "'{s}' is neither a string nor bytes", .{field});
if (hex.len % 2 != 0) return self.fail(.map_unreadable, "'{s}' bytes are not hex", .{field});
const bytes = try arena.alloc(u8, hex.len / 2);
_ = std.fmt.hexToBytes(bytes, hex) catch return self.fail(.map_unreadable, "'{s}' bytes are not hex", .{field});
return bytes;
},
else => return self.fail(.map_unreadable, "a record carries no '{s}'", .{field}),
}
}
fn mapIdentity(self: *Verifier, arena: std.mem.Allocator, object: std.json.ObjectMap, field: []const u8) Error!MapIdentity {
const value = object.get(field) orelse {
return self.fail(.map_unreadable, "a record carries no '{s}'", .{field});
};
const inner = switch (value) {
.object => |o| o,
else => return self.fail(.map_unreadable, "'{s}' is not an identity", .{field}),
};
return .{
.name = try self.mapString(arena, inner, "name"),
.email = try self.mapString(arena, inner, "email"),
.timestamp = integerOf(inner.get("timestamp")) orelse {
return self.fail(.map_unreadable, "'{s}' carries no timestamp", .{field});
},
.timezone_offset_minutes = integerOf(inner.get("timezone_offset_minutes")) orelse {
return self.fail(.map_unreadable, "'{s}' carries no timezone offset", .{field});
},
};
}
// -- the witnessed commits ----------------------------------------------
fn windowLen(self: *const Verifier) usize {
return @max(self.input.checkin_window, 1);
}
/// A pass over the witnessed commits from the first, in stream order.
fn walk(self: *Verifier, with_records: bool) Error!Walk {
self.input.witness_commits.seekTo(0) catch return self.fail(.witness_unreadable, "the witness does not read", .{});
return .{
.verifier = self,
.spill = .{
.in = &self.input.witness_commits.interface,
.key = &self.input.witness.commit_key,
.count = self.input.witness.commit_count,
},
.with_records = with_records,
};
}
/// The witnessed commits a window at a time, each with the map record
/// its OID resolves to when asked for. A window lives until the next
/// call, in `window_arena`; only one walk runs at once.
const Walk = struct {
verifier: *Verifier,
spill: git_import.witness_spill.Reader,
with_records: bool,
commits: []git_import.WitnessCommit = &.{},
records: []?MapCommit = &.{},
fn next(self: *Walk) Error!bool {
const verifier = self.verifier;
_ = verifier.window_arena.reset(.retain_capacity);
const arena = verifier.window_arena.allocator();
const remaining = self.spill.count - self.spill.index;
const len: usize = @intCast(@min(remaining, verifier.windowLen()));
self.commits = try arena.alloc(git_import.WitnessCommit, len);
for (self.commits) |*commit| {
// Never null: `len` stops short of `count`, and a file that
// ends early is `error.Corrupt`.
commit.* = (try self.take(arena)).?;
}
if (len == 0) {
// After the last record, the file must end too.
if (try self.take(arena)) |_| unreachable;
return false;
}
if (!self.with_records) return true;
// From the file, not from what an earlier window left buffered.
verifier.input.map.interface.tossBuffered();
self.records = try arena.alloc(?MapCommit, len);
for (self.records, self.commits) |*record, commit| {
record.* = if (commit.oid) |oid| try verifier.mapCommitOf(arena, oid) else null;
}
if (verifier.input.map_read) |seam| seam.read(seam.context);
return true;
}
fn take(self: *Walk, arena: std.mem.Allocator) Error!?git_import.WitnessCommit {
const at = self.spill.index;
return self.spill.next(arena) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.ReadFailed => return self.verifier.fail(.witness_unreadable, "the witness does not read", .{}),
error.Corrupt => return self.verifier.fail(.witness_unreadable, "commit record {d} of {d} does not read back as the import wrote it", .{ at, self.spill.count }),
};
}
};
// -- content helpers ----------------------------------------------------
/// The content identity of one artifact of the published store,
/// memoized by name.
fn contentFacts(self: *Verifier, name: []const u8) Error!?ContentFacts {
if (self.facts.get(name)) |found| return found;
const bytes = (self.contentByName(self.input.published, name) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
// Every artifact in this store already reconstructed cleanly in
// clause 2, so this is unreachable in practice; treating it as
// absent keeps the caller's one failure path honest.
else => return null,
}) orelse return null;
defer self.gpa.free(bytes);
var facts: ContentFacts = .{ .length = bytes.len, .digest = undefined };
std.crypto.hash.sha2.Sha256.hash(bytes, &facts.digest, .{});
try self.facts.put(self.gpa, try self.arena.dupe(u8, name), facts);
return facts;
}
fn contentByName(self: *Verifier, log: *logstore.LogStore, name: []const u8) !?[]u8 {
var digest: [logstore.max_hash_len]u8 = undefined;
const raw = rawName(name, &digest) orelse return null;
const rid = log.lookup(raw) orelse return null;
// The published store is read by every semantic clause; the mirror
// store is read here only on the rare deferred-difference path, so it
// gets no memo.
const cache: ?*logstore.PlaintextCache =
if (log == self.input.published) &self.published_cache else null;
return log.contentUnmasked(self.gpa, rid, cache);
}
/// Strict `artifact.parse`, deliberately -- not the lenient
/// `parseForImport` the read path uses for fossil-origin history. What is
/// being read here is an artifact THIS import wrote seconds ago through the
/// canonical builder, so anything the strict reader rejects is a writer
/// defect, and admitting it would publish a repository whose own writer
/// cannot round-trip it.
fn namesCheckin(self: *Verifier, log: *logstore.LogStore, name: []const u8) bool {
const bytes = (self.contentByName(log, name) catch return false) orelse return false;
defer self.gpa.free(bytes);
var manifest = artifact.parse(self.gpa, bytes) catch return false;
defer manifest.deinit();
return manifest.kind == .manifest;
}
};
fn hexName(digest: []const u8, buffer: *[hex_name_len]u8) []const u8 {
return std.fmt.bufPrint(buffer, "{x}", .{digest}) catch unreachable;
}
fn rawName(name: []const u8, buffer: *[logstore.max_hash_len]u8) ?[]const u8 {
if (name.len != 40 and name.len != 64) return null;
const raw = buffer[0 .. name.len / 2];
_ = std.fmt.hexToBytes(raw, name) catch return null;
return raw;
}
fn stringOf(value: ?std.json.Value) ?[]const u8 {
return switch (value orelse return null) {
.string => |text| text,
else => null,
};
}
fn integerOf(value: ?std.json.Value) ?i64 {
return switch (value orelse return null) {
.integer => |number| number,
else => null,
};
}
test "the whole-store sweep reports the same first fault at every thread count" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const raw = try tmp.dir.createFile(io, "store", .{ .read = true });
defer raw.close(io);
var file_storage = zit.storage.FileStorage.init(io, raw, 0);
var log = try logstore.LogStore.open(gpa, file_storage.storageFile(), 3);
defer log.deinit(gpa);
// Twelve families of eight versions, each version offered its
// predecessor as a base, so the store holds delta chains that cross the
// boundaries of the small runs below.
const families = 12;
const versions = 8;
var latest: [families]?logstore.Rid = @splat(null);
for (0..versions) |v| for (0..families) |f| {
var content: std.ArrayList(u8) = .empty;
defer content.deinit(gpa);
for (0..64) |line| try content.print(gpa, "family {d} line {d}\n", .{ f, line });
try content.print(gpa, "version {d}\n", .{v});
var digest: [32]u8 = undefined;
std.crypto.hash.sha3.Sha3_256.hash(content.items, &digest, .{});
latest[f] = try log.putArtifact(&digest, content.items, latest[f]);
};
const order = try log.ridsInPersistedOrder(gpa);
defer gpa.free(order);
try std.testing.expectEqual(@as(usize, families * versions), order.len);
// Two faults, the earlier in persisted order first in the file too. A
// byte flipped mid-frame fails the frame's checksum without moving any
// other frame.
const first = 40;
const second = 70;
for ([_]usize{ first, second }) |at| {
const mid = (log.offsetOf(order[at]) + log.offsetOf(order[at + 1])) / 2;
var byte: [1]u8 = undefined;
_ = try raw.readPositionalAll(io, &byte, mid);
byte[0] ^= 0xff;
try raw.writePositionalAll(io, &byte, mid);
}
var threaded = std.Io.Threaded.init(gpa, .{});
defer threaded.deinit();
for ([_]usize{ 1, 2, 4 }) |threads| {
var pool: store_workers.Pool = .{ .io = threaded.io() };
pool.start(threads);
defer pool.deinit();
var sweep: Verifier.Sweep = .{ .gpa = gpa, .log = &log, .rids = order, .min_run = 4 };
defer sweep.deinit();
try sweep.run(&pool);
var faulted_runs: usize = 0;
for (sweep.faults) |slot| {
if (slot != null) faulted_runs += 1;
}
const found = sweep.firstFault() orelse return error.TestExpectedFault;
try std.testing.expect(found.fault == .unreconstructable);
const reported: ?usize = found.at;
try std.testing.expectEqual(@as(?usize, first), reported);
// With more than one thread the second fault sits in a later run and
// is found there too: what picks the report is persisted order, not
// which run finished first.
if (threads > 1) try std.testing.expect(faulted_runs >= 2);
}
}