//! The horizon-clone equivalence sweep (draft-fossil-horizon-00). One seeded
//! random history is built at an origin by the real CLI verbs, then cloned
//! three ways through the real sync code -- dense, at a content horizon, and
//! at a structural horizon of the same check-in C and rule -- and every derived
//! view a user reads is compared, verb output against verb output, for the
//! check-ins the horizon holds.
//!
//! Driving the CLI rather than the sans-IO client is the point: every horizon
//! decision a user meets (the persist filter, the seed sidecar, the sync
//! cursor, structural closure, the deepen an `update` below C performs) lives
//! in the verbs, so a sweep under them would prove a layer no user runs. The
//! origin is answered in this process through `transport.in_process_peer`, by
//! the same `ServeCtx` `zit serve` answers HTTP with, so the wire between them
//! is a seeded simulated link that can drop a request or a reply.
//!
//! Oracles, all against the dense clone of the same origin:
//! (a) every artifact a horizon clone holds hashes to its own name and is
//! held by the dense clone; every check-in of the draft's held set K
//! (section 3.1, computed here from the dense graph's P cards) is held;
//! (b) for every held check-in: its timeline line, its tags, its file list,
//! every file's bytes; for sampled held pairs: their diff and merge
//! base; the open leaves and every branch whose tip is held;
//! (c) after the origin moves on (new check-ins, forks below C, tags below
//! C that propagate to C, cancels) an ordinary pull re-establishes (b);
//! (d) deepening to the root through `zit update` makes the horizon clone
//! dense: its artifact set equals the dense clone's and verify is clean.
//! Part 2 cuts the same runs with crashes (sim/faulty_dir.zig under the verb's
//! io) and lost rounds, and requires a verifying store after each cut and the
//! fault-free state after the next sync.
//!
//! The seed is the whole universe: one PRNG draws the history, C, the rule,
//! and every fault, so `reproduce with seed=N` replays a failure exactly.
const std = @import("std");
const zit = @import("zit");
const cli = zit.cli;
const logstore = zit.logstore;
const storage = zit.storage;
const artifact = zit.artifact;
const main_cli = @import("../main.zig");
const sync_cmd = @import("../cmd/sync.zig");
const transport = @import("../transport.zig");
const serve_mod = @import("../serve.zig");
const faulty_dir = @import("faulty_dir.zig");
const Dir = std.Io.Dir;
const Io = std.Io;
const assert = std.debug.assert;
pub const Error = error{
OutOfMemory,
/// The harness itself could not set up or tear down a workspace.
Workspace,
/// A verb the generator or the oracle relies on failed where it must not.
VerbFailed,
/// A held artifact does not hash to its name.
ArtifactNameMismatch,
/// A horizon clone holds an artifact the dense clone does not.
ArtifactNotInDense,
/// A check-in of the draft's held set K is not held.
HeldSetMissing,
/// A derived view of a held check-in differs from the dense clone's.
ViewMismatch,
/// Verification of a store reported damage.
VerifyFailed,
/// A deepened clone's artifact set is not the dense clone's.
NotDenseAfterDeepen,
/// After a fault and a fault-free sync, a store differs from the
/// fault-free run's.
FaultDiverged,
/// A horizon clone streamed from the filtering origin stored other
/// bytes than the batched local filter, or did not stream at all.
StreamedCloneDiverged,
};
/// The URL every clone saves as its remote. Never dialled: the in-process
/// peer answers before the HTTP client is reached.
const origin_url = "http://127.0.0.1:1/xfer";
/// One CLI invocation's result: the exit code and what it wrote.
const Ran = struct {
exit: cli.Exit,
out: []const u8,
err: []const u8,
};
/// The CLI, run in process. Owns the capture buffers and the simulated clock.
const Shell = struct {
gpa: std.mem.Allocator,
io: Io,
arena: std.heap.ArenaAllocator,
out: std.Io.Writer.Allocating,
err: std.Io.Writer.Allocating,
/// Unix seconds handed to every verb as `now_epoch`; advanced by the
/// generator so check-in times are distinct and increasing.
clock: i64 = 1_700_000_000,
fn init(gpa: std.mem.Allocator, io: Io) Shell {
return .{
.gpa = gpa,
.io = io,
.arena = .init(gpa),
.out = .init(gpa),
.err = .init(gpa),
};
}
fn deinit(self: *Shell) void {
self.out.deinit();
self.err.deinit();
self.arena.deinit();
}
/// Run `zit ARGS...` with `io` (the healthy one unless a fault run hands
/// in a decorated one). The result borrows the shell's buffers until the
/// next call.
fn runIo(self: *Shell, io: Io, args: []const []const u8) Error!Ran {
_ = self.arena.reset(.retain_capacity);
self.out.clearRetainingCapacity();
self.err.clearRetainingCapacity();
const a = self.arena.allocator();
const argv = a.alloc([]const u8, args.len + 1) catch return error.OutOfMemory;
argv[0] = "zit";
@memcpy(argv[1..], args);
const streams: cli.Streams = .{ .out = &self.out.writer, .err = &self.err.writer };
const amb: cli.Ambient = .{ .now_epoch = self.clock, .user_env = "sim" };
const code = main_cli.run(self.gpa, io, a, argv, streams, amb);
streams.out.flush() catch {};
streams.err.flush() catch {};
return .{ .exit = code, .out = self.out.written(), .err = self.err.written() };
}
fn run(self: *Shell, args: []const []const u8) Error!Ran {
return self.runIo(self.io, args);
}
/// Run a verb that must succeed; a failure is a harness or product bug
/// and is reported with the verb's own words.
fn must(self: *Shell, args: []const []const u8) Error!Ran {
const r = try self.run(args);
if (r.exit != .ok) {
std.debug.print("horizon-dst: zit", .{});
for (args) |x| std.debug.print(" {s}", .{x});
std.debug.print(" -> {s}\n{s}{s}", .{ @tagName(r.exit), r.out, r.err });
return error.VerbFailed;
}
return r;
}
};
/// Wire faults for one exchange, drawn per round while armed.
const WireFaults = struct {
/// Percent chance a request is lost before the origin sees it.
drop_request_pct: u8 = 0,
/// Percent chance the origin answers and the reply is lost.
drop_reply_pct: u8 = 0,
};
/// The origin, answered in process by the serve path `zit serve` uses.
const Origin = struct {
gpa: std.mem.Allocator,
io: Io,
dir: Dir,
file: std.Io.File,
server: zit.sync.Server,
serve_code: zit.sync.HashHex,
project_code_buf: [zit.sync.max_projectcode_len]u8 = undefined,
ctx: sync_cmd.ServeCtx,
random: std.Random,
faults: WireFaults = .{},
/// Rounds answered and rounds lost, for the report and the fault oracle.
rounds: u64 = 0,
lost: u64 = 0,
/// Bind to the repository at `dir` (a `zit init` checkout holding `.zit`).
/// Mirrors `runServe`'s identity setup: the stored project code, and a
/// servercode derived from the instance's path.
fn open(self: *Origin, gpa: std.mem.Allocator, io: Io, dir: Dir, random: std.Random) Error!void {
self.* = .{
.gpa = gpa,
.io = io,
.dir = dir,
.file = dir.openFile(io, zit.repo_filename, .{}) catch return error.Workspace,
.server = undefined,
.serve_code = zit.sync.deriveServerCode(&.{"sim-origin"}),
.ctx = undefined,
.random = random,
};
const project_code = self.readProjectCode() catch serve_mod.server_projectcode;
self.server = .{
.version = serve_mod.server_version,
.credential = null,
.self_servercode = &self.serve_code,
.project_code = project_code,
// Small batches, so a clone spans many rounds and a streamed
// persist meets deltas whose basis lands rounds later.
.clone_batch_max = 1 + random.uintLessThan(usize, 16),
};
self.ctx = .{ .io = io, .root_dir = dir, .file = &self.file, .server = &self.server };
}
fn readProjectCode(self: *Origin) ![]const u8 {
const st = try self.file.stat(self.io);
var fs = storage.FileStorage.initReadOnly(self.io, self.file, st.size);
var log = try logstore.LogStore.openReadOnly(self.gpa, fs.storageFile());
defer log.deinit(self.gpa);
var cfg = (try log.readConfig(self.gpa)) orelse return error.NoConfig;
defer cfg.deinit(self.gpa);
const pc = cfg.config.project_code;
if (pc.len == 0 or pc.len > self.project_code_buf.len) return error.NoConfig;
@memcpy(self.project_code_buf[0..pc.len], pc);
return self.project_code_buf[0..pc.len];
}
fn close(self: *Origin) void {
self.ctx.deinit();
self.file.close(self.io);
}
fn peer(self: *Origin) transport.InProcessPeer {
return .{ .ctx = self, .exchangeFn = exchange };
}
fn exchange(ctx_ptr: *anyopaque, gpa: std.mem.Allocator, framed: []const u8) transport.TransportError!transport.InProcessReply {
const self: *Origin = @ptrCast(@alignCast(ctx_ptr));
self.rounds += 1;
if (self.faults.drop_request_pct != 0 and self.random.uintLessThan(u8, 100) < self.faults.drop_request_pct) {
self.lost += 1;
return error.ConnectionFailed;
}
const reply = sync_cmd.serveAnswer(&self.ctx, gpa, framed, null, .{}, true) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return error.ProtocolError,
};
if (reply.diverted) |d| gpa.free(d);
if (self.faults.drop_reply_pct != 0 and self.random.uintLessThan(u8, 100) < self.faults.drop_reply_pct) {
gpa.free(reply.body);
self.lost += 1;
return error.ConnectionFailed;
}
return .{ .body = reply.body, .content_type = reply.content_type };
}
};
/// Enter `dir` as the process working directory: the checkout verbs find
/// their repository from the cwd, exactly as a user's shell does.
fn enter(io: Io, dir: Dir) Error!void {
std.process.setCurrentDir(io, dir) catch return error.Workspace;
}
/// A check-in's full name. Every check-in this sweep makes is SHA3-256.
const Hash = [64]u8;
/// Bounds on one history: operations before the clones, and after them.
const first_ops_min = 12;
const first_ops_max = 28;
const later_ops_min = 3;
const later_ops_max = 10;
const Branch = struct { name: []const u8, tip: usize, open: bool };
const TagRec = struct { name: []const u8, raw: bool, target: usize, active: bool };
/// The origin's history, made by the CLI in the origin checkout. Files are
/// owned by the branch whose name prefixes their path and only that branch
/// edits, renames or deletes them, so every merge is clean and the history
/// is all the generator has to get right.
const Gen = struct {
sh: *Shell,
io: Io,
dir: Dir,
random: std.Random,
/// Owns every name and list below; lives for the whole run.
a: std.mem.Allocator,
checkins: std.ArrayList(Hash) = .empty,
branches: std.ArrayList(Branch) = .empty,
tags: std.ArrayList(TagRec) = .empty,
/// The branch whose tip the origin checkout is on.
cur: usize = 0,
seq: u32 = 0,
submanifests: bool = false,
fn tick(self: *Gen) void {
self.sh.clock += 1 + self.random.uintLessThan(u16, 300);
}
fn hashOf(self: *const Gen, i: usize) []const u8 {
return &self.checkins.items[i];
}
fn indexOf(self: *const Gen, hash: []const u8) ?usize {
for (self.checkins.items, 0..) |*h, i| {
if (std.mem.startsWith(u8, h, hash)) return i;
}
return null;
}
/// Record the check-in a successful commit printed.
fn recordCommit(self: *Gen, out: []const u8) Error!usize {
const prefix = "check-in ";
const at = std.mem.indexOf(u8, out, prefix) orelse return error.VerbFailed;
const start = at + prefix.len;
if (out.len < start + 64) return error.VerbFailed;
var h: Hash = undefined;
@memcpy(&h, out[start .. start + 64]);
try self.checkins.append(self.a, h);
return self.checkins.items.len - 1;
}
fn newPath(self: *Gen, owner: []const u8) Error![]const u8 {
self.seq += 1;
const dir_n = self.random.uintLessThan(u8, 3);
return std.fmt.allocPrint(self.a, "{s}/d{d}/f{d}.txt", .{ owner, dir_n, self.seq }) catch error.OutOfMemory;
}
fn writeLines(self: *Gen, path: []const u8, extra: []const u8, append: bool) Error!void {
var body: std.ArrayList(u8) = .empty;
defer body.deinit(self.a);
if (append) {
const old = self.dir.readFileAlloc(self.io, path, self.a, .limited(1 << 20)) catch return error.Workspace;
body.appendSlice(self.a, old) catch return error.OutOfMemory;
}
body.appendSlice(self.a, extra) catch return error.OutOfMemory;
if (std.fs.path.dirname(path)) |d| self.dir.createDirPath(self.io, d) catch return error.Workspace;
self.dir.writeFile(self.io, .{ .sub_path = path, .data = body.items }) catch return error.Workspace;
}
fn lineFor(self: *Gen) Error![]const u8 {
self.seq += 1;
return std.fmt.allocPrint(self.a, "line {d} {x}\n", .{ self.seq, self.random.int(u32) }) catch error.OutOfMemory;
}
/// The managed files of the current checkout that `owner` owns.
fn ownFiles(self: *Gen, owner: []const u8) Error![][]const u8 {
const r = try self.sh.must(&.{"ls"});
var list: std.ArrayList([]const u8) = .empty;
var it = std.mem.tokenizeScalar(u8, r.out, '\n');
while (it.next()) |line| {
if (line.len > owner.len and std.mem.startsWith(u8, line, owner) and line[owner.len] == '/') {
list.append(self.a, self.a.dupe(u8, line) catch return error.OutOfMemory) catch return error.OutOfMemory;
}
}
return list.items;
}
fn root(self: *Gen) Error!void {
try self.branches.append(self.a, .{ .name = "trunk", .tip = 0, .open = true });
var n: usize = 0;
while (n < 3) : (n += 1) {
const p = try self.newPath("trunk");
try self.writeLines(p, try self.lineFor(), false);
_ = try self.sh.must(&.{ "add", p });
}
self.tick();
const r = try self.sh.must(&.{ "commit", "--no-submanifest-advice", "-m", "root" });
self.branches.items[0].tip = try self.recordCommit(r.out);
self.cur = 0;
}
/// A check-in on the current branch: one to three changes to its own files.
fn commit(self: *Gen) Error!void {
const br = self.branches.items[self.cur];
const own = try self.ownFiles(br.name);
var used = std.DynamicBitSetUnmanaged.initEmpty(self.a, own.len) catch return error.OutOfMemory;
const changes = 1 + self.random.uintLessThan(usize, 3);
var c: usize = 0;
while (c < changes) : (c += 1) {
const roll = self.random.uintLessThan(u8, 100);
const pick: ?usize = if (own.len == 0) null else self.random.uintLessThan(usize, own.len);
if (pick == null or roll < 35 or used.isSet(pick.?)) {
const p = try self.newPath(br.name);
try self.writeLines(p, try self.lineFor(), false);
_ = try self.sh.must(&.{ "add", p });
continue;
}
const f = own[pick.?];
used.set(pick.?);
if (roll < 75) {
try self.writeLines(f, try self.lineFor(), true);
} else if (roll < 87) {
_ = try self.sh.must(&.{ "rm", f });
// Gone from disk too, so no later update meets a stray file.
self.dir.deleteFile(self.io, f) catch {};
} else {
const to = try self.newPath(br.name);
if (std.fs.path.dirname(to)) |d| self.dir.createDirPath(self.io, d) catch return error.Workspace;
_ = try self.sh.must(&.{ "mv", f, to });
}
}
self.tick();
const msg = std.fmt.allocPrint(self.a, "{s} work {d}", .{ br.name, self.seq }) catch return error.OutOfMemory;
const r = try self.sh.must(&.{ "commit", "--no-submanifest-advice", "-m", msg });
self.branches.items[self.cur].tip = try self.recordCommit(r.out);
}
fn update(self: *Gen, target: usize) Error!void {
_ = try self.sh.must(&.{ "update", self.hashOf(target) });
}
fn switchTo(self: *Gen, b: usize) Error!void {
try self.update(self.branches.items[b].tip);
self.cur = b;
}
/// A new branch forked from `from`, any check-in of any age.
fn fork(self: *Gen, from: usize) Error!void {
try self.update(from);
const name = std.fmt.allocPrint(self.a, "b{d}", .{self.branches.items.len}) catch return error.OutOfMemory;
const p = try self.newPath(name);
try self.writeLines(p, try self.lineFor(), false);
_ = try self.sh.must(&.{ "add", p });
self.tick();
const msg = std.fmt.allocPrint(self.a, "start {s}", .{name}) catch return error.OutOfMemory;
const r = try self.sh.must(&.{ "commit", "--no-submanifest-advice", "--branch", name, "-m", msg });
const tip = try self.recordCommit(r.out);
try self.branches.append(self.a, .{ .name = name, .tip = tip, .open = true });
self.cur = self.branches.items.len - 1;
}
/// Merge another branch's tip into the current branch.
fn merge(self: *Gen) Error!void {
if (self.branches.items.len < 2) return self.commit();
var other = self.random.uintLessThan(usize, self.branches.items.len);
if (other == self.cur) other = (other + 1) % self.branches.items.len;
return self.mergeFrom(other);
}
fn mergeFrom(self: *Gen, other: usize) Error!void {
assert(other != self.cur);
const from = self.hashOf(self.branches.items[other].tip);
self.tick();
const m = try self.sh.run(&.{ "merge", from });
// The other tip is already an ancestor: nothing to record.
if (m.exit != .ok and std.mem.indexOf(u8, m.err, "nothing to merge") != null) return self.commit();
if (m.exit != .ok) {
std.debug.print("horizon-dst: merge refused: {s}{s}", .{ m.out, m.err });
return error.VerbFailed;
}
const msg = std.fmt.allocPrint(self.a, "merge {s} into {s}", .{ self.branches.items[other].name, self.branches.items[self.cur].name }) catch return error.OutOfMemory;
const r = try self.sh.must(&.{ "commit", "--no-submanifest-advice", "--allow-empty", "-m", msg });
self.branches.items[self.cur].tip = try self.recordCommit(r.out);
}
fn tag(self: *Gen, target: usize) Error!void {
return self.tagWith(target, self.random.boolean());
}
/// A symbolic name, or (`raw`) a propagating property with a value.
fn tagWith(self: *Gen, target: usize, raw: bool) Error!void {
self.tick();
self.seq += 1;
const name = std.fmt.allocPrint(self.a, "{s}{d}", .{ if (raw) "p" else "t", self.seq }) catch return error.OutOfMemory;
if (raw) {
const value = std.fmt.allocPrint(self.a, "v{d}", .{self.seq}) catch return error.OutOfMemory;
_ = try self.sh.must(&.{ "tag", "add", "--raw", "--propagate", name, self.hashOf(target), value });
} else {
_ = try self.sh.must(&.{ "tag", "add", name, self.hashOf(target) });
}
try self.tags.append(self.a, .{ .name = name, .raw = raw, .target = target, .active = true });
}
fn cancelTag(self: *Gen) Error!void {
var live: usize = 0;
for (self.tags.items) |t| live += @intFromBool(t.active);
if (live == 0) return;
var k = self.random.uintLessThan(usize, live);
for (self.tags.items) |*t| {
if (!t.active) continue;
if (k == 0) {
self.tick();
if (t.raw) {
_ = try self.sh.must(&.{ "tag", "cancel", "--raw", t.name, self.hashOf(t.target) });
} else {
_ = try self.sh.must(&.{ "tag", "cancel", t.name, self.hashOf(t.target) });
}
t.active = false;
return;
}
k -= 1;
}
}
/// Close an open branch other than trunk, or reopen a closed one.
fn closeOrReopen(self: *Gen) Error!void {
if (self.branches.items.len < 2) return;
const b = 1 + self.random.uintLessThan(usize, self.branches.items.len - 1);
const br = &self.branches.items[b];
self.tick();
if (br.open) {
_ = try self.sh.must(&.{ "branch", "close", br.name });
br.open = false;
// Work continues somewhere open.
if (self.cur == b) try self.switchTo(0);
} else {
_ = try self.sh.must(&.{ "branch", "reopen", br.name });
br.open = true;
}
}
fn openBranch(self: *Gen) usize {
var open: usize = 0;
for (self.branches.items) |b| open += @intFromBool(b.open);
var k = self.random.uintLessThan(usize, open);
for (self.branches.items, 0..) |b, i| {
if (!b.open) continue;
if (k == 0) return i;
k -= 1;
}
unreachable; // trunk is never closed
}
/// One random operation. `below` is where a "below the horizon" op
/// aims, once a horizon exists: forks and tags there are what re-seeding
/// and boundary labelling exist for.
fn step(self: *Gen, below: []const usize) Error!void {
const roll = self.random.uintLessThan(u8, 100);
const any = self.random.uintLessThan(usize, self.checkins.items.len);
const aim = if (below.len != 0 and self.random.boolean()) below[self.random.uintLessThan(usize, below.len)] else any;
if (roll < 40) {
try self.commit();
} else if (roll < 52) {
try self.switchTo(self.openBranch());
} else if (roll < 62) {
try self.fork(aim);
} else if (roll < 72) {
try self.merge();
} else if (roll < 84) {
try self.tag(aim);
} else if (roll < 89) {
try self.cancelTag();
} else if (roll < 95) {
try self.closeOrReopen();
} else if (!self.submanifests) {
_ = try self.sh.must(&.{ "submanifest", "enable", "--k", "2" });
self.submanifests = true;
} else {
try self.commit();
}
}
};
/// The check-in graph as the dense clone's manifests spell it: P cards read
/// from each manifest's own bytes, never from a derived view, so the held
/// set below is the draft's definition evaluated independently of the code
/// under test.
const Graph = struct {
/// parents[i]: indices of the check-ins check-in i names in its P card,
/// primary first.
parents: []const []const usize,
fn build(a: std.mem.Allocator, sh: *Shell, gen: *const Gen) Error!Graph {
const out = a.alloc([]const usize, gen.checkins.items.len) catch return error.OutOfMemory;
for (gen.checkins.items, 0..) |*h, i| {
const r = try sh.must(&.{ "artifact", h });
var list: std.ArrayList(usize) = .empty;
var rows = std.mem.splitScalar(u8, r.out, '\n');
while (rows.next()) |line| {
if (!std.mem.startsWith(u8, line, "P ")) continue;
var toks = std.mem.tokenizeScalar(u8, line[2..], ' ');
while (toks.next()) |t| {
const p = gen.indexOf(t) orelse return error.ViewMismatch;
list.append(a, p) catch return error.OutOfMemory;
}
}
out[i] = list.items;
}
return .{ .parents = out };
}
/// Every ancestor of `c`, `c` included.
fn ancestors(self: Graph, a: std.mem.Allocator, c: usize) Error!std.DynamicBitSetUnmanaged {
var set = std.DynamicBitSetUnmanaged.initEmpty(a, self.parents.len) catch return error.OutOfMemory;
var work: std.ArrayList(usize) = .empty;
set.set(c);
work.append(a, c) catch return error.OutOfMemory;
while (work.pop()) |x| for (self.parents[x]) |p| {
if (set.isSet(p)) continue;
set.set(p);
work.append(a, p) catch return error.OutOfMemory;
};
return set;
}
/// Draft section 3.1: K for check-in `c` under `rule`.
fn heldSet(self: Graph, a: std.mem.Allocator, c: usize, rule: zit.cone.HorizonRule) Error!std.DynamicBitSetUnmanaged {
const n = self.parents.len;
var k = std.DynamicBitSetUnmanaged.initEmpty(a, n) catch return error.OutOfMemory;
// descendants: C and everything that reaches C through P cards.
k.set(c);
var changed = true;
while (changed) {
changed = false;
for (self.parents, 0..) |ps, i| {
if (k.isSet(i)) continue;
for (ps) |p| if (k.isSet(p)) {
k.set(i);
changed = true;
break;
};
}
}
if (rule == .leaves) {
// The descendants of every nearest common ancestor of C and a
// check-in with no children (open or closed) outside them.
const desc = try k.clone(a);
const anc_c = try self.ancestors(a, c);
const anc = a.alloc(std.DynamicBitSetUnmanaged, n) catch return error.OutOfMemory;
for (anc, 0..) |*set, i| set.* = try self.ancestors(a, i);
var childless = std.DynamicBitSetUnmanaged.initFull(a, n) catch return error.OutOfMemory;
for (self.parents) |ps| for (ps) |p| childless.unset(p);
for (0..n) |leaf| {
if (!childless.isSet(leaf) or desc.isSet(leaf)) continue;
var common = try anc[leaf].clone(a);
common.setIntersection(anc_c);
for (0..n) |m| {
if (!common.isSet(m)) continue;
// Nearest: no other common ancestor descends from it.
var nearest = true;
for (0..n) |y| if (y != m and common.isSet(y) and anc[y].isSet(m)) {
nearest = false;
break;
};
if (!nearest) continue;
for (0..n) |x| if (anc[x].isSet(m)) k.set(x);
}
}
}
return k;
}
};
/// What a clone is, for the oracle's expectations.
const Kind = enum { dense, content, structural };
/// The views the oracle reads, taken in one clone's directory.
const Views = struct {
sh: *Shell,
io: Io,
dir: Dir,
a: std.mem.Allocator,
fn get(self: Views, args: []const []const u8) Error![]const u8 {
try enter(self.io, self.dir);
const r = try self.sh.must(args);
return self.a.dupe(u8, r.out) catch error.OutOfMemory;
}
/// A view that may fail closed: null when the verb refused.
fn tryGet(self: Views, args: []const []const u8) Error!?[]const u8 {
try enter(self.io, self.dir);
const r = try self.sh.run(args);
if (r.exit != .ok) return null;
return self.a.dupe(u8, r.out) catch error.OutOfMemory;
}
};
/// Drop what is a property of one checkout rather than of the repository:
/// the `*CURRENT*` marker, the leading `*` of the current branch, and the
/// `(N)` ordinal a listing renumbers when it shows fewer lines.
fn normalize(a: std.mem.Allocator, line: []const u8) Error![]const u8 {
var out: std.ArrayList(u8) = .empty;
var rest = line;
if (std.mem.startsWith(u8, rest, "* ")) rest = rest[2..] else if (std.mem.startsWith(u8, rest, " ")) rest = rest[2..];
if (rest.len > 0 and rest[0] == '(') {
if (std.mem.indexOfScalar(u8, rest, ')')) |close| rest = rest[close + 1 ..];
}
var it = std.mem.splitSequence(u8, rest, "*CURRENT* ");
while (it.next()) |part| out.appendSlice(a, part) catch return error.OutOfMemory;
return out.items;
}
fn lines(a: std.mem.Allocator, text: []const u8) Error![][]const u8 {
var list: std.ArrayList([]const u8) = .empty;
var it = std.mem.tokenizeScalar(u8, text, '\n');
while (it.next()) |l| list.append(a, try normalize(a, l)) catch return error.OutOfMemory;
return list.items;
}
fn contains(haystack: []const []const u8, needle: []const u8) bool {
for (haystack) |h| if (std.mem.eql(u8, h, needle)) return true;
return false;
}
/// The check-in a listing line names by its 12-digit short hash, if any.
fn lineCheckin(gen: *const Gen, line: []const u8) ?usize {
var it = std.mem.tokenizeAny(u8, line, " []");
while (it.next()) |tok| {
if (tok.len != 12) continue;
for (tok) |ch| {
if (!std.ascii.isHex(ch)) break;
} else return gen.indexOf(tok);
}
return null;
}
/// Two `ls --long` listings name the same files with the same blob hashes,
/// whatever each says about a blob's size and type.
fn sameListing(a: std.mem.Allocator, dense: []const u8, other: []const u8) bool {
const d = lines(a, dense) catch return false;
const o = lines(a, other) catch return false;
if (d.len != o.len) return false;
for (d, o) |dl, ol| {
const dh = std.mem.indexOfScalar(u8, dl, ' ') orelse return false;
const oh = std.mem.indexOfScalar(u8, ol, ' ') orelse return false;
if (!std.mem.eql(u8, dl[0..dh], ol[0..oh])) return false;
const dn = std.mem.lastIndexOfScalar(u8, dl, ' ') orelse return false;
const on = std.mem.lastIndexOfScalar(u8, ol, ' ') orelse return false;
if (!std.mem.eql(u8, dl[dn..], ol[on..])) return false;
}
return true;
}
/// Where the run is, for the report a failing seed prints. Single-run state,
/// like the peer install.
var stage: []const u8 = "";
var stage_phase: []const u8 = "";
/// Report a mismatch with both sides, so a failing seed explains itself.
fn mismatch(what: []const u8, detail: []const u8, dense: []const u8, other: []const u8) Error {
std.debug.print("horizon-dst: [{s}] {s} differs ({s})\n--- dense\n{s}\n--- horizon\n{s}\n", .{ stage, what, detail, dense, other });
return error.ViewMismatch;
}
const Horizon = struct {
c: usize,
rule: zit.cone.HorizonRule,
/// A path cone composed with the horizon (draft section 3.2): blobs
/// outside it are elided at any age.
cone: ?[]const u8 = null,
fn covers(self: Horizon, path: []const u8) bool {
const c = self.cone orelse return true;
return path.len > c.len and std.mem.startsWith(u8, path, c) and path[c.len] == '/';
}
};
/// `sameListing`, and every line of a path the cone covers identical.
fn sameCoveredListing(a: std.mem.Allocator, hz: Horizon, dense: []const u8, other: []const u8) bool {
if (!sameListing(a, dense, other)) return false;
const d = lines(a, dense) catch return false;
const o = lines(a, other) catch return false;
for (d, o) |dl, ol| {
const at = std.mem.lastIndexOfScalar(u8, dl, ' ') orelse return false;
if (hz.covers(dl[at + 1 ..]) and !std.mem.eql(u8, dl, ol)) return false;
}
return true;
}
/// Compare one horizon clone's views with the dense clone's, for every
/// check-in it holds, and its holdings with the draft's held set.
fn compare(
a: std.mem.Allocator,
random: std.Random,
gen: *const Gen,
graph: Graph,
hz: Horizon,
dense: Views,
other: Views,
kind: Kind,
) Error!void {
assert(kind != .dense);
const n = gen.checkins.items.len;
stage = std.fmt.allocPrint(a, "{s} {s} clone, C={s} rule={s}", .{ stage_phase, @tagName(kind), gen.hashOf(hz.c)[0..12], hz.rule.name() }) catch stage;
const d_leaves = try lines(a, try dense.get(&.{"leaves"}));
const k = try graph.heldSet(a, hz.c, hz.rule);
// The check-ins the clone holds, by its own timeline.
const d_tl = try lines(a, try dense.get(&.{ "timeline", "-F", "oneline", "-n", "0" }));
const o_tl = try lines(a, try other.get(&.{ "timeline", "-F", "oneline", "-n", "0" }));
var held = std.DynamicBitSetUnmanaged.initEmpty(a, n) catch return error.OutOfMemory;
for (o_tl) |l| {
if (!contains(d_tl, l)) return mismatch("timeline line", l, "(absent)", l);
if (lineCheckin(gen, l)) |i| held.set(i);
}
if (kind == .content and o_tl.len != d_tl.len) {
var absent: std.ArrayList([]const u8) = .empty;
for (d_tl) |l| if (!contains(o_tl, l)) absent.append(a, l) catch return error.OutOfMemory;
return mismatch("timeline", "a content horizon holds every manifest", std.mem.join(a, "\n", absent.items) catch return error.OutOfMemory, "(those absent)");
}
for (0..n) |i| if (k.isSet(i) and !held.isSet(i)) {
std.debug.print("horizon-dst: [{s}] {s} clone at {s} ({s}) lacks K member {s}\n", .{ stage, @tagName(kind), gen.hashOf(hz.c)[0..12], @tagName(hz.rule), gen.hashOf(i)[0..12] });
// The graph, so the miss can be read against the rule.
for (graph.parents, 0..) |ps, x| {
std.debug.print(" {d} {s} K={} held={} <-", .{ x, gen.hashOf(x)[0..8], k.isSet(x), held.isSet(x) });
for (ps) |p| std.debug.print(" {d}", .{p});
std.debug.print("\n", .{});
}
return error.HeldSetMissing;
};
// Per held check-in: what it is, what it carries, and its tree.
for (0..n) |i| {
if (!held.isSet(i)) continue;
const h = gen.hashOf(i);
for ([_][]const []const u8{
&.{ "info", h },
&.{ "tag", "list", "--raw", h },
&.{ "ls", "--long", h },
}) |args| {
const dv = try dense.get(args);
const ov = try other.tryGet(args) orelse return mismatch(args[0], h[0..12], dv, "(refused)");
if (std.mem.eql(u8, dv, ov)) continue;
// Outside K a blob may be elided, and `ls --long` then reports
// its size and type as absent; the name and blob hash still are
// the manifest's own and must agree.
const ls = std.mem.eql(u8, args[0], "ls");
const fine = ls and if (k.isSet(i)) hz.cone != null and sameCoveredListing(a, hz, dv, ov) else sameListing(a, dv, ov);
if (!fine) return mismatch(args[0], std.fmt.allocPrint(a, "{s}, {s} K", .{ h[0..12], if (k.isSet(i)) "in" else "outside" }) catch h, dv, ov);
}
const files = try dense.get(&.{ "ls", h });
var it = std.mem.tokenizeScalar(u8, files, '\n');
while (it.next()) |f| {
const dv = try dense.get(&.{ "cat", f, h });
const ov = try other.tryGet(&.{ "cat", f, h });
if (ov) |bytes| {
if (!std.mem.eql(u8, dv, bytes)) return mismatch("cat", f, dv, bytes);
} else if (k.isSet(i) and hz.covers(f)) {
return mismatch("cat", f, dv, "(refused inside K)");
}
}
}
// Pairs inside K: diff and merge base.
var members: std.ArrayList(usize) = .empty;
for (0..n) |i| if (k.isSet(i)) members.append(a, i) catch return error.OutOfMemory;
var pairs: usize = 0;
while (pairs < 8 and members.items.len > 1) : (pairs += 1) {
const x = gen.hashOf(members.items[random.uintLessThan(usize, members.items.len)]);
const y = gen.hashOf(members.items[random.uintLessThan(usize, members.items.len)]);
const narrow: []const []const u8 = if (hz.cone) |c| &.{c} else &.{};
const dargs = std.mem.concat(a, []const u8, &.{ &.{ "diff", "--from", x, "--to", y }, narrow }) catch return error.OutOfMemory;
const dd = try dense.get(dargs);
const od = try other.tryGet(dargs) orelse return mismatch("diff", x[0..12], dd, "(refused)");
if (!std.mem.eql(u8, dd, od)) return mismatch("diff", x[0..12], dd, od);
const dm = try dense.tryGet(&.{ "merge-base", x, y });
const om = try other.tryGet(&.{ "merge-base", x, y });
if (om) |ob| {
// A horizon clone may decline; it must never name another base.
if (dm == null or !std.mem.eql(u8, dm.?, ob)) return mismatch("merge-base", x[0..12], dm orelse "(none)", ob);
} else if (dm) |db| {
const base = gen.indexOf(std.mem.trim(u8, db, " \n")) orelse return error.ViewMismatch;
// A structural clone may also refuse when both ancestries cross
// the boundary (a better base could lie past it): fail closed.
_ = base;
if (kind == .content) return mismatch("merge-base", x[0..12], db, "(refused by a content horizon)");
}
}
// Open leaves: every held one, and nothing else.
const o_leaves = try lines(a, try other.get(&.{"leaves"}));
var want: std.ArrayList([]const u8) = .empty;
for (d_leaves) |l| {
const i = lineCheckin(gen, l) orelse continue;
if (kind == .content or held.isSet(i)) want.append(a, l) catch return error.OutOfMemory;
}
if (want.items.len != o_leaves.len) return mismatch("leaves", "count", try std.mem.join(a, "\n", want.items), try std.mem.join(a, "\n", o_leaves));
for (want.items, o_leaves) |w, o| if (!std.mem.eql(u8, w, o)) return mismatch("leaves", "line", w, o);
// Branches: any it lists is the dense one exactly; every one whose tip
// is held is listed.
const d_br = try lines(a, try dense.get(&.{ "branch", "list", "-v" }));
const o_br = try lines(a, try other.get(&.{ "branch", "list", "-v" }));
for (o_br) |l| if (!contains(d_br, l)) return mismatch("branch list", l, try std.mem.join(a, "\n", d_br), try std.mem.join(a, "\n", o_br));
for (d_br) |l| {
const tip = lineCheckin(gen, l) orelse continue;
if ((kind == .content or held.isSet(tip)) and !contains(o_br, l)) return mismatch("branch list", l, try std.mem.join(a, "\n", d_br), try std.mem.join(a, "\n", o_br));
}
// Symbolic names: one held at its target resolves there; one whose
// target is not held resolves to the same name or not at all.
const d_tags = try lines(a, try dense.get(&.{ "tag", "list" }));
for (d_tags) |name| {
const dv = try dense.get(&.{ "info", name });
const target = gen.indexOf(dv[0..@min(dv.len, 64)]) orelse continue;
const ov = try other.tryGet(&.{ "info", name });
if (ov) |o| {
if (!std.mem.eql(u8, dv[0..64], o[0..@min(o.len, 64)])) return mismatch("tag resolution", name, dv, o);
} else if (kind == .content or held.isSet(target)) {
return mismatch("tag resolution", name, dv, "(unresolved, target held)");
}
}
_ = try other.get(&.{"verify"});
}
/// The names of every artifact a store holds content for, each checked
/// against its own bytes: a held artifact that does not hash to its name is
/// damage whatever else is true.
fn heldArtifacts(a: std.mem.Allocator, io: Io, dir: Dir) Error!std.StringHashMapUnmanaged(void) {
var file = dir.openFile(io, zit.repo_filename, .{}) catch return error.Workspace;
defer file.close(io);
const st = file.stat(io) catch return error.Workspace;
var fs = storage.FileStorage.initReadOnly(io, file, st.size);
var log = logstore.LogStore.openReadOnly(a, fs.storageFile()) catch return error.VerifyFailed;
defer log.deinit(a);
var set: std.StringHashMapUnmanaged(void) = .empty;
var rid: logstore.Rid = 1;
while (rid <= log.entryCount()) : (rid += 1) {
if (log.isPhantom(rid)) continue;
const digest = log.hashOf(rid);
const hex = a.alloc(u8, digest.len * 2) catch return error.OutOfMemory;
for (digest, 0..) |b, j| {
hex[2 * j] = "0123456789abcdef"[b >> 4];
hex[2 * j + 1] = "0123456789abcdef"[b & 15];
}
const bytes = (log.content(a, rid) catch return error.VerifyFailed) orelse return error.VerifyFailed;
var name_buf: [64]u8 = undefined;
const name = artifact.artifactName(if (hex.len == 64) .sha3_256 else .sha1, bytes, &name_buf);
if (!std.mem.eql(u8, name, hex)) {
std.debug.print("horizon-dst: held artifact {s} hashes to {s}\n", .{ hex, name });
return error.ArtifactNameMismatch;
}
// Draft section 4.5: a horizon clone holds no cluster at any depth,
// so "dense" is the dense set less its clusters.
if (artifact.parse(a, bytes)) |parsed| {
var m = parsed;
defer m.deinit();
if (m.kind == .cluster) continue;
} else |_| {}
set.put(a, hex, {}) catch return error.OutOfMemory;
}
return set;
}
/// Oracle (a): what a horizon clone holds is the dense clone's, by name; and
/// after deepening to the root, exactly the dense clone's.
fn checkHoldings(a: std.mem.Allocator, io: Io, dense: Dir, other: Dir, exact: bool) Error!void {
const d = try heldArtifacts(a, io, dense);
const o = try heldArtifacts(a, io, other);
var it = o.keyIterator();
while (it.next()) |name| if (!d.contains(name.*)) {
std.debug.print("horizon-dst: horizon clone holds {s}, the dense clone does not\n", .{name.*});
return error.ArtifactNotInDense;
};
if (exact and d.count() != o.count()) {
var dit = d.keyIterator();
while (dit.next()) |name| if (!o.contains(name.*)) std.debug.print("horizon-dst: deepened clone lacks {s}\n", .{name.*});
return error.NotDenseAfterDeepen;
}
}
const Clone = struct {
name: []const u8,
kind: Kind,
rule: zit.cone.HorizonRule = .descendants,
/// A path cone beside the horizon, for the composition of draft 3.2.
cone: ?[]const u8 = null,
dir: Dir = undefined,
fn horizon(self: *const Clone, c: usize) Horizon {
return .{ .c = c, .rule = self.rule, .cone = self.cone };
}
};
/// One run's world: the origin and its history, the three clones, and the
/// shell that drives them. Every phase below works on it in place.
const World = struct {
a: std.mem.Allocator,
io: Io,
random: std.Random,
sh: *Shell,
ws: Dir,
origin_dir: Dir,
gen: *Gen,
c: usize,
clones: [3]Clone,
graph: Graph = undefined,
fn dense(self: *World) Views {
return .{ .sh = self.sh, .io = self.io, .dir = self.clones[0].dir, .a = self.a };
}
fn views(self: *World, c: *const Clone) Views {
return .{ .sh = self.sh, .io = self.io, .dir = c.dir, .a = self.a };
}
/// Re-read the graph from the dense clone, after it has pulled.
fn regraph(self: *World) Error!void {
try enter(self.io, self.clones[0].dir);
self.graph = try Graph.build(self.a, self.sh, self.gen);
}
/// Clone `c` from the origin and open its working tree, at C or at the
/// default tip (which, outside K, is a deepen). `io` is the verb's: a
/// fault run hands in a decorated one.
fn clone(self: *World, c: *Clone, io: Io) Error!Ran {
try enter(self.io, self.ws);
const horizon_flag = std.fmt.allocPrint(self.a, "--horizon={s}", .{self.gen.hashOf(self.c)}) catch return error.OutOfMemory;
const rule_flag = std.fmt.allocPrint(self.a, "--horizon-rule={s}", .{c.rule.name()}) catch return error.OutOfMemory;
return switch (c.kind) {
.dense => self.sh.runIo(io, &.{ "clone", "--no-open", origin_url, c.name }),
.content => self.sh.runIo(io, try withCone(self.a, c, &.{ "clone", "--no-open", horizon_flag, rule_flag, origin_url, c.name })),
.structural => self.sh.runIo(io, try withCone(self.a, c, &.{ "clone", "--no-open", horizon_flag, rule_flag, "--structural", origin_url, c.name })),
};
}
/// `clone`, for a horizon clone, made twice at `c`'s path (`init`
/// derives the project code its config frames carry from it): through
/// the batched local filter, then streamed, as this origin's filtering
/// lets it. The two stores must be the same bytes; the streamed one stays.
fn cloneStreamedEqualsBatched(self: *World, c: *Clone, io: Io) Error!Ran {
assert(c.kind != .dense);
sync_cmd.clone_batched_persist = true;
const batched_run = self.clone(c, io);
sync_cmd.clone_batched_persist = false;
const b = try batched_run;
if (b.exit != .ok) return b;
try enter(self.io, self.ws);
const store_path = std.fmt.allocPrint(self.a, "{s}/{s}", .{ c.name, zit.repo_filename }) catch return error.OutOfMemory;
const batched = self.ws.readFileAlloc(self.io, store_path, self.a, .limited(1 << 28)) catch return error.Workspace;
self.ws.deleteTree(self.io, c.name) catch return error.Workspace;
const r = try self.clone(c, io);
if (r.exit != .ok) return r;
const streamed = self.ws.readFileAlloc(self.io, store_path, self.a, .limited(1 << 28)) catch return error.Workspace;
if (std.mem.indexOf(u8, r.out, "nothing elided locally") == null) {
std.debug.print("horizon-dst: the {s} clone did not stream from a filtering origin:\n{s}", .{ c.name, r.out });
return error.StreamedCloneDiverged;
}
if (!std.mem.eql(u8, batched, streamed)) {
std.debug.print("horizon-dst: the streamed {s} clone stored {d} bytes, the batched one {d}, not the same\n", .{ c.name, streamed.len, batched.len });
return error.StreamedCloneDiverged;
}
return r;
}
fn openTree(self: *World, c: *Clone) Error!void {
return self.openTreeAt(c, c.kind != .dense and self.random.boolean());
}
fn openTreeAt(self: *World, c: *Clone, at_c: bool) Error!void {
c.dir = self.ws.openDir(self.io, c.name, .{ .access_sub_paths = true }) catch return error.Workspace;
try enter(self.io, c.dir);
if (at_c) {
_ = try self.sh.must(try withCone(self.a, c, &.{ "open", self.gen.hashOf(self.c) }));
} else {
_ = try self.sh.must(try withCone(self.a, c, &.{"open"}));
}
}
/// Oracles (a) and (b) for every horizon clone.
fn check(self: *World, phase: []const u8) Error!void {
stage_phase = phase;
for (self.clones[1..]) |*c| {
try compare(self.a, self.random, self.gen, self.graph, c.horizon(self.c), self.dense(), self.views(c), c.kind);
try checkHoldings(self.a, self.io, self.clones[0].dir, c.dir, false);
}
}
/// The origin moves on, partly below the horizon: forks from C's
/// ancestry, tags there that may propagate to C, cancels.
fn moveOrigin(self: *World) Error!void {
const below_set = try self.graph.ancestors(self.a, self.c);
var below: std.ArrayList(usize) = .empty;
for (0..self.gen.checkins.items.len) |i| if (below_set.isSet(i) and i != self.c) below.append(self.a, i) catch return error.OutOfMemory;
const later_ops = later_ops_min + self.random.uintLessThan(usize, later_ops_max - later_ops_min + 1);
for (0..later_ops) |_| {
try enter(self.io, self.origin_dir);
try self.gen.step(below.items);
}
}
fn pull(self: *World, c: *const Clone, io: Io) Error!Ran {
try enter(self.io, c.dir);
return self.sh.runIo(io, &.{"pull"});
}
/// The oldest check-in `c` cannot check out from what it holds: its
/// manifest or one of its blobs is absent. Null when it lacks none.
fn firstMissing(self: *World, c: *const Clone) Error!?usize {
try enter(self.io, c.dir);
for (self.gen.checkins.items, 0..) |*h, i| {
const r = try self.sh.run(&.{ "ls", "--long", h });
if (r.exit != .ok) return i;
var it = std.mem.tokenizeScalar(u8, r.out, '\n');
const hz = c.horizon(0);
while (it.next()) |line| {
const at = std.mem.lastIndexOfScalar(u8, line, ' ') orelse continue;
if (std.mem.indexOf(u8, line, " absent ") != null and hz.covers(line[at + 1 ..])) return i;
}
}
return null;
}
/// One deepen of `c`, through `zit update` to the oldest check-in it
/// cannot check out, the way a user reaches history below C. Null when
/// there is nothing left below its horizon.
fn deepen(self: *World, c: *const Clone, io: Io) Error!?Ran {
const target = try self.firstMissing(c) orelse return null;
try enter(self.io, c.dir);
return try self.sh.runIo(io, &.{ "update", self.gen.hashOf(target) });
}
/// Deepen until nothing is below the horizon, bounded by the history.
fn deepenFully(self: *World, c: *const Clone) Error!void {
for (0..self.gen.checkins.items.len + 1) |_| {
const r = try self.deepen(c, self.io) orelse return;
if (r.exit != .ok) return verbFailed(c.name, r);
}
std.debug.print("horizon-dst: [{s}] {s} still lacks history after deepening once per check-in\n", .{ stage_phase, c.name });
return error.NotDenseAfterDeepen;
}
/// Oracle (d): deepened to the root, a horizon clone is the dense clone.
fn checkDense(self: *World, c: *const Clone) Error!void {
stage_phase = "after deepen";
try compare(self.a, self.random, self.gen, self.graph, c.horizon(0), self.dense(), self.views(c), c.kind);
// Under a path cone the blobs outside it stay elided at any depth.
try checkHoldings(self.a, self.io, self.clones[0].dir, c.dir, c.cone == null);
}
};
/// Build the origin's history and the world around it, up to (not
/// including) the clones.
fn makeWorld(a: std.mem.Allocator, io: Io, random: std.Random, sh: *Shell, ws: Dir, gen: *Gen) Error!World {
try enter(io, ws);
_ = try sh.must(&.{ "init", "-q", "origin" });
const origin_dir = ws.openDir(io, "origin", .{ .access_sub_paths = true }) catch return error.Workspace;
try enter(io, origin_dir);
gen.* = .{ .sh = sh, .io = io, .dir = origin_dir, .random = random, .a = a };
try gen.root();
const first_ops = first_ops_min + random.uintLessThan(usize, first_ops_max - first_ops_min + 1);
for (0..first_ops) |_| try gen.step(&.{});
// Half the seeds end with trunk's tip forked onto a new branch and C on
// it: trunk's open leaf then has its only child on another branch, and
// C is not the common ancestor of the open leaves.
var c = random.uintLessThan(usize, gen.checkins.items.len);
if (random.boolean()) {
try gen.fork(gen.branches.items[0].tip);
if (random.boolean()) try gen.commit();
c = gen.checkins.items.len - 1 - random.uintLessThan(usize, 2);
}
const rule: zit.cone.HorizonRule = if (random.boolean()) .descendants else .leaves;
// A third of the seeds compose a path cone with the horizon.
const cone: ?[]const u8 = if (random.uintLessThan(u8, 3) == 0) "trunk" else null;
return .{
.a = a,
.io = io,
.random = random,
.sh = sh,
.ws = ws,
.origin_dir = origin_dir,
.gen = gen,
.c = c,
.clones = .{
.{ .name = "dense", .kind = .dense },
.{ .name = "content", .kind = .content, .rule = rule, .cone = cone },
// Every seed runs the leaves rule: its K must be as exact as the
// descendants rule's wherever C sits.
.{ .name = "structural", .kind = .structural, .rule = .leaves, .cone = cone },
},
};
}
/// `args`, with the clone's path cone appended as the CLI takes it.
fn withCone(a: std.mem.Allocator, c: *const Clone, args: []const []const u8) Error![]const []const u8 {
const cone = c.cone orelse return args;
const flag = std.fmt.allocPrint(a, "--cone={s}", .{cone}) catch return error.OutOfMemory;
return std.mem.concat(a, []const u8, &.{ args, &.{flag} }) catch error.OutOfMemory;
}
/// A fresh workspace under `root` for one run, removed when the run ends.
const Workspace = struct {
root: Dir,
name_buf: [48]u8 = undefined,
name: []const u8 = "",
dir: Dir = undefined,
home: Dir = undefined,
fn open(self: *Workspace, io: Io, label: []const u8, seed: u64) Error!void {
self.name = std.fmt.bufPrint(&self.name_buf, "hz-{s}-{d}", .{ label, seed }) catch unreachable;
self.root.deleteTree(io, self.name) catch {};
self.root.createDir(io, self.name, .default_dir) catch return error.Workspace;
self.dir = self.root.openDir(io, self.name, .{ .access_sub_paths = true }) catch return error.Workspace;
// The verbs resolve from the cwd; it is restored however the run ends.
self.home = Dir.cwd().openDir(io, ".", .{}) catch return error.Workspace;
}
fn close(self: *Workspace, io: Io) void {
std.process.setCurrentDir(io, self.home) catch {};
self.home.close(io);
self.dir.close(io);
self.root.deleteTree(io, self.name) catch {};
}
};
/// Part 1: the fault-free comparison, from clone through pull to deepen.
pub fn run(gpa: std.mem.Allocator, io: Io, root: Dir, seed: u64) Error!void {
var prng = std.Random.DefaultPrng.init(seed ^ 0x686f72697a6f6e);
const random = prng.random();
var arena_state = std.heap.ArenaAllocator.init(gpa);
defer arena_state.deinit();
const a = arena_state.allocator();
var space: Workspace = .{ .root = root };
try space.open(io, "run", seed);
defer space.close(io);
var sh = Shell.init(gpa, io);
defer sh.deinit();
var gen: Gen = undefined;
var w = try makeWorld(a, io, random, &sh, space.dir, &gen);
defer w.origin_dir.close(io);
var origin: Origin = undefined;
try origin.open(gpa, io, w.origin_dir, random);
defer origin.close();
transport.in_process_peer = origin.peer();
defer transport.in_process_peer = null;
for (&w.clones) |*c| {
const r = if (c.kind == .dense) try w.clone(c, io) else try w.cloneStreamedEqualsBatched(c, io);
if (r.exit != .ok) return verbFailed("clone", r);
try w.openTree(c);
}
defer for (&w.clones) |*c| c.dir.close(io);
try w.regraph();
try w.check("after clone");
try w.moveOrigin();
for (&w.clones) |*c| {
const r = try w.pull(c, io);
if (r.exit != .ok) return verbFailed(c.name, r);
}
try w.regraph();
try w.check("after pull");
for (w.clones[1..]) |*c| {
try w.deepenFully(c);
try w.checkDense(c);
}
}
fn verbFailed(verb: []const u8, r: Ran) Error {
std.debug.print("horizon-dst: [{s}] zit verb in {s} -> {s}\n{s}{s}", .{ stage_phase, verb, @tagName(r.exit), r.out, r.err });
return error.VerbFailed;
}
// ---------------------------------------------------------------------------
// Part 2: the same runs under crashes and lost rounds.
// ---------------------------------------------------------------------------
/// The verb a fault run cuts.
const FaultVerb = enum { clone, pull, deepen };
/// How it is cut: a power loss at one mutating operation of the verb's own
/// io (sim/faulty_dir.zig), or a wire that loses requests and replies.
const FaultKind = enum { crash, wire };
/// Bound on the retries a lossy wire gets before it is healed: a verb that
/// cannot finish over a lossy wire must still finish over a clean one.
const wire_tries = 6;
/// Lock files a crashed in-process verb leaves naming this, still-live,
/// process; rewritten to name a dead one, as a real crash leaves them.
const lock_names = [_][]const u8{ ".zit.lock", ".zit.lock.break" };
fn orphanCrashedLocks(io: Io, dir: Dir) Error!void {
var host_buf: [std.posix.HOST_NAME_MAX]u8 = undefined;
const host = std.posix.gethostname(&host_buf) catch return error.Workspace;
var rec_buf: [256]u8 = undefined;
const rec = std.fmt.bufPrint(&rec_buf, "pid={d} host={s} time=0\n", .{ std.math.maxInt(std.posix.pid_t), host }) catch unreachable;
for (lock_names) |name| {
dir.access(io, name, .{}) catch continue;
dir.writeFile(io, .{ .sub_path = name, .data = rec }) catch return error.Workspace;
}
}
fn exists(io: Io, dir: Dir, name: []const u8) bool {
dir.access(io, name, .{}) catch return false;
return true;
}
/// The same change committed in the twin and then the victim, at one clock:
/// a line appended to the first file the checkout holds. Returns the twin's
/// check-in. The victim's commit may refuse (its store is cut); a refusal
/// must leave no check-in behind, and it is retried after the next sync.
fn commitBoth(w: *World, twin: *Clone, victim: *Clone) Error!?Hash {
try enter(w.io, twin.dir);
const listed = try w.sh.run(&.{"ls"});
if (listed.exit != .ok) return null;
var it = std.mem.tokenizeScalar(u8, listed.out, '\n');
const first = w.a.dupe(u8, it.next() orelse return null) catch return error.OutOfMemory;
w.sh.clock += 7;
var h: Hash = undefined;
for ([_]*Clone{ twin, victim }, 0..) |c, n| {
const old = c.dir.readFileAlloc(w.io, first, w.a, .limited(1 << 20)) catch {
if (n == 0) return null;
return h;
};
const body = std.mem.concat(w.a, u8, &.{ old, "after a cut\n" }) catch return error.OutOfMemory;
c.dir.writeFile(w.io, .{ .sub_path = first, .data = body }) catch return error.Workspace;
try enter(w.io, c.dir);
const r = try w.sh.run(&.{ "commit", "--no-sync", "--allow-fork", "--no-submanifest-advice", "-m", "after a cut" });
if (n == 0) {
if (r.exit != .ok) return verbFailed("commit", r);
const at = std.mem.indexOf(u8, r.out, "check-in ") orelse return error.VerbFailed;
@memcpy(&h, r.out[at + 9 .. at + 9 + 64]);
} else if (r.exit == .ok and std.mem.indexOf(u8, r.out, &h) == null) {
std.debug.print("horizon-dst: [{s}] the victim's commit is not the twin's: {s}\n", .{ stage_phase, r.out });
return error.FaultDiverged;
}
}
return h;
}
/// Run `verb` on `c` with `io`.
fn runVerb(w: *World, c: *Clone, verb: FaultVerb, io: Io) Error!Ran {
return switch (verb) {
.clone => w.clone(c, io),
.pull => w.pull(c, io),
.deepen => (try w.deepen(c, io)) orelse .{ .exit = .ok, .out = "", .err = "" },
};
}
/// Part 2: one seeded fault per seed, on one horizon clone, against a twin
/// that runs the same verbs fault-free. After the fault the store must
/// verify, and after the next sync it must hold exactly what the twin holds
/// and agree with the dense clone view for view.
pub fn runFaults(gpa: std.mem.Allocator, io: Io, root: Dir, seed: u64) Error!void {
// The history is Part 1's for the same seed; the faults draw their own.
var prng = std.Random.DefaultPrng.init(seed ^ 0x686f72697a6f6e);
const random = prng.random();
var fprng = std.Random.DefaultPrng.init(seed ^ 0x6661756c7473);
const fr = fprng.random();
var arena_state = std.heap.ArenaAllocator.init(gpa);
defer arena_state.deinit();
const a = arena_state.allocator();
var space: Workspace = .{ .root = root };
try space.open(io, "fault", seed);
defer space.close(io);
var sh = Shell.init(gpa, io);
defer sh.deinit();
var gen: Gen = undefined;
var w = try makeWorld(a, io, random, &sh, space.dir, &gen);
defer w.origin_dir.close(io);
var origin: Origin = undefined;
try origin.open(gpa, io, w.origin_dir, fr);
defer origin.close();
transport.in_process_peer = origin.peer();
defer transport.in_process_peer = null;
const target: Kind = if (fr.boolean()) .content else .structural;
const verb = fr.enumValue(FaultVerb);
const fault: FaultKind = if (fr.uintLessThan(u8, 100) < 70) .crash else .wire;
const base = w.clones[if (target == .content) 1 else 2];
w.clones[1] = .{ .name = "twin", .kind = target, .rule = base.rule, .cone = base.cone };
w.clones[2] = .{ .name = "victim", .kind = target, .rule = base.rule, .cone = base.cone };
const twin = &w.clones[1];
const victim = &w.clones[2];
// Twin and victim open the same way, so they hold the same horizon.
const open_at_c = fr.boolean();
stage_phase = std.fmt.allocPrint(a, "fault {s}/{s} on a {s} clone", .{ @tagName(fault), @tagName(verb), @tagName(target) }) catch "fault";
// Everything before the cut, on all three clones alike; a cut clone's
// twin is made by the counted run below.
for (w.clones[0..if (verb == .clone) 1 else 2]) |*c| {
const r = try w.clone(c, io);
if (r.exit != .ok) return verbFailed("clone", r);
try w.openTreeAt(c, c.kind != .dense and open_at_c);
}
defer for (w.clones[0..2]) |*c| c.dir.close(io);
var victim_open = false;
defer if (victim_open) victim.dir.close(io);
if (verb != .clone) {
const r = try w.clone(victim, io);
if (r.exit != .ok) return verbFailed("clone", r);
try w.openTreeAt(victim, open_at_c);
victim_open = true;
try w.regraph();
try w.moveOrigin();
for ([_]*Clone{ &w.clones[0], twin, victim }) |c| {
if (verb == .pull and c == victim) continue;
if (verb == .pull and c == twin) continue;
const p = try w.pull(c, io);
if (p.exit != .ok) return verbFailed("pull", p);
}
}
// The twin runs the verb fault-free, counted.
var counter = faulty_dir.FaultyDir.init(io, fr, .{});
counter.install();
const tr = runVerb(&w, twin, verb, counter.io());
faulty_dir.FaultyDir.uninstall();
const twin_run = try tr;
if (twin_run.exit != .ok) return verbFailed(@tagName(verb), twin_run);
const total = counter.op_index;
if (verb == .clone) try w.openTreeAt(twin, open_at_c);
const store_path = std.fmt.allocPrint(a, "{s}/{s}", .{ victim.name, zit.repo_filename }) catch return error.OutOfMemory;
// The victim, cut.
switch (fault) {
.crash => {
const k = 1 + fr.uintLessThan(u64, @max(total, 1));
var cut = faulty_dir.FaultyDir.init(io, fr, .{ .crash_at_op = k, .torn_write = fr.boolean() });
cut.install();
const cr = runVerb(&w, victim, verb, cut.io());
faulty_dir.FaultyDir.uninstall();
_ = try cr; // the verb is expected to fail: the disk is the oracle
if (exists(io, space.dir, victim.name)) {
var vd = space.dir.openDir(io, victim.name, .{}) catch return error.Workspace;
defer vd.close(io);
try orphanCrashedLocks(io, vd);
}
},
.wire => {
origin.faults = .{ .drop_request_pct = 10 + fr.uintLessThan(u8, 30), .drop_reply_pct = 10 + fr.uintLessThan(u8, 30) };
defer origin.faults = .{};
var tries: usize = 0;
while (tries < wire_tries) : (tries += 1) {
const r = try runVerb(&w, victim, verb, io);
if (r.exit == .ok) break;
// A clone that failed leaves no store to come back to; the
// operator's remedy is to clone again.
if (verb == .clone and !exists(io, space.dir, store_path)) space.dir.deleteTree(io, victim.name) catch {};
}
},
}
// After the cut: a store that exists verifies.
if (exists(io, space.dir, store_path)) {
var vd = space.dir.openDir(io, victim.name, .{}) catch return error.Workspace;
defer vd.close(io);
try enter(io, vd);
// -R: a clone cut before it opened a working tree is a bare store.
const v = try sh.run(&.{ "verify", "-R", "." });
if (v.exit != .ok) {
std.debug.print("horizon-dst: [{s}] verify after the fault -> {s}\n{s}{s}", .{ stage_phase, @tagName(v.exit), v.out, v.err });
return error.VerifyFailed;
}
}
// A commit straight after a cut pull, before any sync: where the cut
// left a tree's closure open, the commit's own dependency fetch (the
// structural-closure pull) is what must recover it. The twin commits the
// same change at the same time, so the two check-ins are one artifact.
const committed = target == .structural and verb == .pull and fault == .crash;
var commit_hash: ?Hash = null;
if (committed) commit_hash = try commitBoth(&w, twin, victim);
// The next sync. A clone cut before it made a store is made again; any
// other store syncs where it stands.
const has_cone = exists(io, space.dir, std.fmt.allocPrint(a, "{s}/.zit-cache/store-cone", .{victim.name}) catch return error.OutOfMemory);
// A clone cut before `init` recorded the checkout is half-made: no verb
// can find its store, and the operator's remedy is to clone again.
const has_checkout = exists(io, space.dir, std.fmt.allocPrint(a, "{s}/.zit-checkout", .{victim.name}) catch return error.OutOfMemory);
if (!exists(io, space.dir, store_path) or (verb == .clone and !has_checkout)) {
space.dir.deleteTree(io, victim.name) catch {};
const r = try w.clone(victim, io);
if (r.exit != .ok) return verbFailed("clone", r);
}
var opened_here = false;
if (!victim_open) {
victim.dir = space.dir.openDir(io, victim.name, .{ .access_sub_paths = true }) catch return error.Workspace;
victim_open = true;
opened_here = true;
}
try enter(io, victim.dir);
const p = try sh.run(&.{ "pull", origin_url });
if (p.exit != .ok) return verbFailed("pull", p);
if (commit_hash) |h| {
// A commit the cut refused is made again on the healed store; it
// must be the twin's check-in, byte for byte.
try enter(io, victim.dir);
const have = try sh.run(&.{ "artifact", &h });
if (have.exit != .ok) {
const r = try sh.run(&.{ "commit", "--no-sync", "--allow-fork", "--no-submanifest-advice", "-m", "after a cut" });
if (r.exit != .ok) return verbFailed("commit", r);
if (std.mem.indexOf(u8, r.out, &h) == null) {
std.debug.print("horizon-dst: [{s}] the victim's commit is not the twin's: {s}\n", .{ stage_phase, r.out });
return error.FaultDiverged;
}
}
// The twin pushes it, so the dense clone and the oracle know it.
try enter(io, twin.dir);
const pushed = try sh.run(&.{"push"});
if (pushed.exit != .ok) return verbFailed("push", pushed);
w.gen.checkins.append(a, h) catch return error.OutOfMemory;
try enter(io, w.clones[0].dir);
const dp = try sh.run(&.{"pull"});
if (dp.exit != .ok) return verbFailed("pull", dp);
// And the victim learns what the push changed at the origin.
try enter(io, victim.dir);
const vp = try sh.run(&.{"pull"});
if (vp.exit != .ok) return verbFailed("pull", vp);
try enter(io, twin.dir);
const tp = try sh.run(&.{"pull"});
if (tp.exit != .ok) return verbFailed("pull", tp);
}
// A cut clone never reached its `open`; after the sync that heals it, it
// is opened as its twin was.
if (opened_here) {
victim.dir.close(io);
try w.openTreeAt(victim, open_at_c);
}
// A deepen cut before it moved C is rerun as the operator would; one cut
// after completed through the pull. Its twin deepens all the way too.
if (verb == .deepen) try w.deepenFully(victim);
try w.regraph();
// A clone cut before it recorded its horizon has none, and healed
// toward dense, as the draft's policy-loss rule says it must.
const healed_dense = verb == .clone and !has_cone and has_checkout and exists(io, space.dir, store_path);
const dense_dir = w.clones[0].dir;
if (healed_dense) {
try checkHoldings(a, io, dense_dir, victim.dir, true);
return;
}
if (verb == .deepen) try w.deepenFully(twin);
// What the syncs left, taken before the views are read: a whole-tree
// listing on a path cone's clone fetches the nodes its origin withheld
// (sparse-cone RFC section 5.3), and the twin's views are not read here.
const t = try heldArtifacts(a, io, twin.dir);
const v = try heldArtifacts(a, io, victim.dir);
const hz = victim.horizon(if (verb == .deepen) 0 else w.c);
try compare(a, fr, w.gen, w.graph, hz, w.dense(), w.views(victim), target);
var it = t.keyIterator();
var differ = t.count() != v.count();
while (it.next()) |name| if (!v.contains(name.*)) {
differ = true;
std.debug.print("horizon-dst: [{s}] victim lacks {s}\n", .{ stage_phase, name.* });
};
if (differ) return error.FaultDiverged;
}
// ---------------------------------------------------------------------------
// The fixed regression history: one scripted origin that holds every shape a
// random seed has failed on, run on every invocation of the sweep.
// ---------------------------------------------------------------------------
/// Each step names the defect class it keeps covered:
/// - a branch forked below C that merges C in: a check-in of K whose
/// primary parent is below the horizon, so its branch enters through
/// the seed, not from C;
/// - the default open target outside K: the deepen moves C to a common
/// ancestor, never to the target, so C's own descendants stay held;
/// - a check-in arriving after the clone that names blobs the clone
/// elided: the pull that brings it must bring them;
/// - a propagating tag below C that reaches a K check-in: re-seeding;
/// - deepening to the root past a sync cursor, the root arriving in the
/// same session it is classified in: the deepened store is dense.
pub fn runRegression(gpa: std.mem.Allocator, io: Io, root: Dir) Error!void {
var prng = std.Random.DefaultPrng.init(0x7265677265737369);
const random = prng.random();
var arena_state = std.heap.ArenaAllocator.init(gpa);
defer arena_state.deinit();
const a = arena_state.allocator();
var space: Workspace = .{ .root = root };
try space.open(io, "regression", 0);
defer space.close(io);
var sh = Shell.init(gpa, io);
defer sh.deinit();
try enter(io, space.dir);
_ = try sh.must(&.{ "init", "-q", "origin" });
const origin_dir = space.dir.openDir(io, "origin", .{ .access_sub_paths = true }) catch return error.Workspace;
defer origin_dir.close(io);
try enter(io, origin_dir);
var gen: Gen = .{ .sh = &sh, .io = io, .dir = origin_dir, .random = random, .a = a };
try gen.root(); // 0, trunk
try gen.commit(); // 1, trunk
try gen.fork(0); // 2, b1: forked below C
try gen.switchTo(0);
try gen.commit(); // 3, trunk
try gen.fork(3); // 4, b2
try gen.commit(); // 5, b2: C
const c: usize = 5;
try gen.switchTo(1);
try gen.mergeFrom(2); // 6, b1 merges C: primary parent below the horizon
try gen.switchTo(0);
try gen.commit(); // 7, trunk: the default open target, outside K
var origin: Origin = undefined;
try origin.open(gpa, io, origin_dir, random);
defer origin.close();
transport.in_process_peer = origin.peer();
defer transport.in_process_peer = null;
var w: World = .{
.a = a,
.io = io,
.random = random,
.sh = &sh,
.ws = space.dir,
.origin_dir = origin_dir,
.gen = &gen,
.c = c,
.clones = .{
.{ .name = "dense", .kind = .dense },
.{ .name = "content", .kind = .content },
.{ .name = "structural", .kind = .structural, .rule = .leaves },
},
};
for (&w.clones) |*cl| {
const r = try w.clone(cl, io);
if (r.exit != .ok) return verbFailed("clone", r);
cl.dir = space.dir.openDir(io, cl.name, .{ .access_sub_paths = true }) catch return error.Workspace;
try enter(io, cl.dir);
// The content clone opens the default tip, outside K; the structural
// one opens C, so its own horizon stays where the scenario put it.
if (cl.kind == .structural) {
_ = try sh.must(&.{ "open", gen.hashOf(c) });
} else {
_ = try sh.must(&.{"open"});
}
}
defer for (&w.clones) |*cl| cl.dir.close(io);
try w.regraph();
try w.check("regression, after clone");
try enter(io, origin_dir);
try gen.switchTo(1);
try gen.commit(); // 8, b1, in K through 6
try gen.tagWith(2, true); // below C, propagating along 6's primary chain
try gen.fork(0); // 9, b3, below C
try gen.switchTo(2);
try gen.mergeFrom(3); // 10, b2 merges b3: elided blobs become above
try gen.mergeFrom(0); // 11, b2 merges trunk
for (&w.clones) |*cl| {
const r = try w.pull(cl, io);
if (r.exit != .ok) return verbFailed(cl.name, r);
}
try w.regraph();
try w.check("regression, after pull");
for (w.clones[1..]) |*cl| {
try w.deepenFully(cl);
try w.checkDense(cl);
}
}