//! zit sync / network / store-transfer verbs (serve, clone, open, pack, pull,
//! push, sync, remote) and the helpers and tests private to them.
//!
//! Extracted verbatim from src/main.zig (release step S2). Every declaration
//! below is byte-identical to its main.zig source; only this import prologue is
//! new. Helpers still shared with other verbs stay in main.zig and are aliased
//! back here so the moved bodies call them unqualified. main.zig's verb-dispatch
//! switch calls into this file (`cmd_sync.runServe` ...); its test block imports
//! it so these inline tests keep running under `zig build test`.
const std = @import("std");
const zit = @import("zit");
const cli = @import("../cli.zig");
const extras = @import("../extras.zig");
const logstore = @import("zit").logstore;
const storage = @import("zit").storage;
const update = @import("zit").update;
const checkout = @import("zit").checkout;
const Repository = @import("zit").Repository;
const serve_mod = @import("../serve.zig");
const transport = @import("../transport.zig");
const sync_driver = @import("../sync_driver.zig");
const Io = std.Io;
const File = Io.File;
const repostate = zit.repostate;
const managed = zit.managed;
const artifact = zit.artifact;
// Shared CLI helpers and test fixtures that stay in main.zig (other verbs use
// them too); aliased back so the moved code and tests reference them by their
// original unqualified names.
const main = @import("../main.zig");
const runInit = main.runInit;
const repo_filename = main.repo_filename;
const renderStorage = main.renderStorage;
const reasonOf = main.reasonOf;
const stateOom = main.stateOom;
const not_a_repo_msg = main.not_a_repo_msg;
const repoReadFail = main.repoReadFail;
const repoOpenFail = main.repoOpenFail;
const openRepo = main.openRepo;
const updateApplyFail = main.updateApplyFail;
const digest_len = main.digest_len;
const hexToDigest = main.hexToDigest;
const digestToWireName = main.digestToWireName;
const decodeWireName = main.decodeWireName;
const open_help = main.open_help;
const pack_help = main.pack_help;
const serve_help = main.serve_help;
const clone_help = main.clone_help;
const pull_help = main.pull_help;
const push_help = main.push_help;
const sync_help = main.sync_help;
const remote_help = main.remote_help;
const packSeedBlob = main.packSeedBlob;
const packSeedCheckin = main.packSeedCheckin;
const packEvolvingBlob = main.packEvolvingBlob;
const packOpenWritable = main.packOpenWritable;
const packOpenReadOnly = main.packOpenReadOnly;
/// Clone phase edges let benchmark code attribute the real command without
/// coupling clocks or counters into production behavior.
pub const ClonePhase = enum { transport, initialize, exchange, persist, config, warm_cache };
pub const ClonePhaseEvent = enum { begin, end };
pub const CloneObserver = struct {
context: *anyopaque,
callback: *const fn (context: *anyopaque, phase: ClonePhase, event: ClonePhaseEvent) void,
};
fn observeClonePhase(observer: ?*const CloneObserver, phase: ClonePhase, event: ClonePhaseEvent) void {
if (observer) |value| value.callback(value.context, phase, event);
}
/// `zit serve [--port N]` — put an HTTP face on the stateless sync responder.
/// Opens `.zit` read-only and keeps it open for the server's lifetime; every
/// request reads a fresh content view of it and is answered independently
/// (sync §3.8). `std.http`/socket I/O lives entirely in `serve.zig`; this only
/// wires the repository and the listen address.
pub fn runServe(
gpa: std.mem.Allocator,
io: std.Io,
arena: std.mem.Allocator,
verb_args: []const []const u8,
streams: cli.Streams,
) cli.Exit {
const parsed = switch (cli.parseServeArgs(verb_args)) {
.ok => |p| p,
.unknown_option => |tok| return cli.usageError(streams, "serve", .unknown_option, tok),
.extra_argument => |tok| return cli.usageError(streams, "serve", .extra_argument, tok),
.bad_port => |tok| {
streams.err.print("zit serve: invalid --port value '{s}'\n", .{tok}) catch {};
return .usage;
},
};
if (parsed.help) {
streams.out.writeAll(serve_help) catch {};
return .ok;
}
const root = (cli.discoverRoot(arena, io) catch return stateOom(streams, "serve")) orelse {
streams.err.writeAll(not_a_repo_msg) catch {};
return .failure;
};
var root_dir = root.dir;
var file = root_dir.openFile(io, repo_filename, .{}) catch {
root_dir.close(io);
return repoReadFail(streams, "serve");
};
defer file.close(io);
const log_st = file.stat(io) catch {
root_dir.close(io);
return repoReadFail(streams, "serve");
};
// Bring-up order is load-bearing for #13 (rationale in `serve_mod.listen`'s
// doc): open the listener FIRST — before ANY O(store) work, which here means
// both the config read below and the prime, each a full-log scan. A client
// that connects during either queues in the accept backlog instead of being
// refused, so the readiness window starts at the bind rather than after the
// scans. Announce "listening" only once the bind has actually succeeded (so
// it never lies about a listener that failed to come up); the accept loop
// then arms each connection's deadline on accept. The backlog property this
// rests on is pinned by serve.zig's "listen opens the accept backlog before
// serve" test.
const address = std.Io.net.IpAddress.parse("127.0.0.1", parsed.port) catch {
streams.err.print("zit serve: invalid --port value '{d}'\n", .{parsed.port}) catch {};
return .usage;
};
var listener = serve_mod.listen(io, address) catch {
streams.err.print("zit serve: cannot bind 127.0.0.1:{d}\n", .{parsed.port}) catch {};
return .failure;
};
defer listener.deinit(io);
streams.err.print("zit serve: listening on http://127.0.0.1:{d}/xfer (Ctrl-C to stop)\n", .{parsed.port}) catch {};
streams.err.flush() catch {};
var fs = storage.FileStorage.init(io, file, log_st.size);
var log = logstore.LogStore.openReadOnly(gpa, fs.storageFile()) catch |e| {
root_dir.close(io);
switch (e) {
error.OutOfMemory => streams.err.writeAll("zit serve: out of memory\n") catch {},
else => streams.err.print("zit serve: cannot read repository: {s}\n", .{reasonOf(e)}) catch {},
}
return .failure;
};
defer log.deinit(gpa);
defer root_dir.close(io);
// §9.1/§8.2 — emit the repository's stored project code in the clone
// reply's `push` card, so a clone learns the shared project identity. A
// repo predating the stored code, or an unreadable config, falls back to
// the fixed placeholder (informational; servercode never gates a clone).
var maybe_cfg: ?logstore.OwnedConfig = log.readConfig(gpa) catch null;
defer if (maybe_cfg) |*oc| oc.deinit(gpa);
const project_code: []const u8 = if (maybe_cfg) |oc|
(if (oc.config.project_code.len > 0) oc.config.project_code else serve_mod.server_projectcode)
else
serve_mod.server_projectcode;
const server = zit.sync.Server{
.version = serve_mod.server_version,
.credential = null,
.self_servercode = serve_mod.server_servercode,
.project_code = project_code,
};
var ctx = ServeCtx{ .io = io, .root_dir = root_dir, .file = file, .server = &server };
defer ctx.dropSnapshot();
const responder = serve_mod.Responder{ .ctx = &ctx, .answerFn = serveAnswer, .maintainFn = ServeCtx.maintain };
if (ctx.baseSnapshot(gpa)) |_| {} else |_| return repoReadFail(streams, "serve");
serve_mod.serve(io, gpa, &listener, responder, .{}) catch |e| switch (e) {
error.Accept => {
streams.err.writeAll("zit serve: connection accept failed\n") catch {};
return .failure;
},
};
return .ok;
}
/// The per-request context threaded into `serve.Responder`: the open log to read
/// a fresh store from, the server identity to answer with, and the injected IO +
/// repository directory used to make pushed content durable (a push request
/// ingests new artifacts into the per-request store, which must be persisted into
/// the served repository's on-disk log — sync §3.3).
const ServeCtx = struct {
io: std.Io,
root_dir: std.Io.Dir,
/// The open read handle on the served `.zit`. `respond` re-reads the CURRENT
/// on-disk log from it (§3.8): a fresh stat fixes the committed end-of-log,
/// so a check-in appended by a separate `zit commit` while `serve` runs is
/// advertised on the next request — no restart.
///
/// The handle pins an INODE, not a path. `zit pack` replaces `.zit` by
/// rename, so a pack performed under a running server leaves that server
/// reading the replaced inode until it is restarted. That is a property of
/// holding the handle open, unchanged by whether the store read from it is
/// cached: before caching, every request re-read the same dead inode. It is
/// recorded here because "a changed size is a changed log" below is a
/// statement about THIS inode.
file: File,
server: *const zit.sync.Server,
/// The most recent snapshot of the on-disk log, reused across requests. A
/// log-backed INDEX (not a full-container copy): it reads the containers it
/// serves on demand and holds only the index between rounds (#14). Built from
/// the long-lived allocator `serve` threads into every answer, so it outlives
/// the request that built it, and owned here — heap-allocated because it owns
/// the `FileStorage` its log borrows; `dropSnapshot` frees it. Never handed to
/// `respond` directly — see `serveAnswer`.
snapshot: ?*LogIndexStore = null,
/// The log size `snapshot` was built from. The log is append-only, so a
/// changed size is a changed log and an unchanged size is an unchanged log:
/// one `stat` per request decides whether the snapshot still describes it.
snapshot_log_size: u64 = 0,
/// How many snapshots this server has built. Not used by the server; it is
/// the observable that makes "snapshot builds do not grow with the round
/// count" a testable property rather than a wall-clock claim.
loads: u64 = 0,
/// Free the cached snapshot, leaving the context usable: the next
/// `baseSnapshot` rebuilds. Both a caller finishing with the context and the
/// rebuild path itself want exactly this.
fn dropSnapshot(self: *ServeCtx) void {
if (self.snapshot) |s| s.destroy();
self.snapshot = null;
self.snapshot_log_size = 0;
}
/// Drop plaintext the cached snapshot memoised while answering, keeping the
/// retained containers it serves from. Called once per round, after every
/// borrow of the snapshot has ended.
///
/// Inflating is how `content()` answers a wire-backed entry, and the
/// responder's own §7.2 size proofs call `content()` on every artifact they
/// emit. Against a throwaway per-request store that plaintext died with the
/// request; against a store that outlives the request it would accumulate
/// until the server held the whole repository's uncompressed history. The
/// containers stay, so the next round answers identically.
///
/// The memo is given up rather than capped because in an optimized build
/// there is almost nothing to give up: the size proofs are `runtime_safety`
/// only, so a clone round inflates nothing and the memo would stay empty
/// regardless. Measured over four clones of a 3001-artifact repository,
/// ReleaseFast: 12 MB resident and 0.09 s of server CPU with this release,
/// against 26 MB and 0.15 s without it. A safety-checked build does re-prove
/// every emitted artifact each round and so pays to re-inflate; that cost
/// belongs to the proofs, not to the cache.
fn releaseSnapshotPlaintext(self: *ServeCtx) void {
if (self.snapshot) |s| s.releaseTransient();
}
/// §3.8 — the store describing the CURRENT on-disk log, rebuilt only when the
/// log has actually grown. The §3.8 guarantee is that a check-in appended by
/// a separate `zit commit` is advertised on the NEXT request with no restart;
/// that needs a fresh `stat` per request, not a fresh rebuild per request.
/// Rebuilding regardless made every round cost the whole repository, which is
/// what turned a clone of a large store into O(rounds x store) server work.
///
/// A torn tail from an append caught mid-write is rejected by the read-only
/// scanner exactly as before; the previous snapshot is dropped only once a
/// replacement is fully built, so a failed rebuild leaves the cache coherent.
///
/// The rebuild itself is O(store) and must therefore not fall inside a
/// request whose connection carries a deadline. `runServe` primes the cache
/// before it accepts anything. A round that appends to the log does NOT
/// rebuild before it returns — it deliberately leaves the snapshot stale
/// (describing the pre-append log) rather than pay the O(store) rebuild inside
/// its own reply deadline (#13). The append-only stat gate above makes that
/// stale snapshot self-correcting: the size no longer matches, so the next
/// `baseSnapshot` rebuilds. That next call is the accept loop's `maintain`
/// hook (see `ServeCtx.maintain`), run between connections with no connection
/// open — so the deferred rebuild still lands outside every connection
/// deadline, never on an innocent later request. Nothing here can bound the
/// startup build itself; that is bounded by the store, and always was.
fn baseSnapshot(self: *ServeCtx, gpa: std.mem.Allocator) serve_mod.AnswerError!repostate.ArtifactStore {
const st = self.file.stat(self.io) catch return error.ServerStorage;
if (self.snapshot) |s| {
if (st.size == self.snapshot_log_size) return s.store();
}
// Build the fresh index BEFORE dropping the old one: a failed rebuild
// leaves the previous snapshot (and its size) intact, so the stat gate
// keeps mismatching until a rebuild succeeds and stale content is never
// served in the meantime. `openFromFile` fully frees itself on any error.
const fresh = LogIndexStore.openFromFile(gpa, self.io, self.file, st.size) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage, error.Corrupt => return error.ServerStorage,
};
self.dropSnapshot();
self.snapshot = fresh;
self.snapshot_log_size = st.size;
self.loads += 1;
return fresh.store();
}
/// The accept loop's between-connections maintenance hook (see
/// `serve.Responder.maintainFn`). Runs after a connection has closed and
/// before the next accept, so no connection is open to charge its O(store)
/// work to. It re-primes the cache via the same stat-gated `baseSnapshot`:
/// after a pure read round the stat is unchanged and this is one `stat`; after
/// a round that appended (a push) the log grew, so this rebuilds the snapshot
/// the pushing round deliberately left stale rather than pay inside its
/// deadline (#13). A failed rebuild is swallowed for the same reason
/// `baseSnapshot` leaves the cache coherent on failure: `baseSnapshot` drops
/// the old snapshot only after a fresh one is fully built and updates
/// `snapshot_log_size` only on success, so a failed rebuild RETAINS the
/// previous (stale) snapshot with its old size — the stat gate keeps
/// mismatching until a rebuild succeeds, and stale content is never served in
/// the meantime. The next round (or the next `maintain`) retries, still
/// outside any earlier round's budget. Type-erased to match the `maintainFn`
/// seam.
fn maintain(ctx_ptr: *anyopaque, gpa: std.mem.Allocator) void {
const self: *ServeCtx = @ptrCast(@alignCast(ctx_ptr));
if (self.baseSnapshot(gpa)) |_| {} else |_| {}
}
};
/// The `serve.Responder` seam: snapshot the repository into an in-memory store,
/// answer the framed request over it, and — when the request pushed new content —
/// make that content durable in the served repository's on-disk log (sync §3.3).
///
/// `respond` stays sans-IO: it only `put`s pushed artifacts into the snapshot
/// store it is given. The durable write is this serve-layer's job and reuses the
/// idempotent `persistStore` (a re-push of an already-stored hash appends nothing
/// — content is addressed by hash), the very path clone/pull persist through. A
/// pure gimme/pull request makes nothing new servable, so the overlay stays
/// empty and the log is never reopened; only a genuine push touches disk.
///
/// `respond` is answered over a per-request WRITE OVERLAY of the cached snapshot
/// (`repostate.OverlayStore`), never over the snapshot itself: the responder
/// interns phantoms and upgrades entries as it works, and letting those survive
/// into the next round changes what that round serves. Reads fall through to the
/// snapshot, so reuse costs nothing; writes are confined to the overlay, and the
/// persist pass — which is idempotent over everything already in the log — can
/// only append what the overlay holds.
///
/// The pushed bytes were size-validated by `respond`'s request scan before they
/// entered the store, so `persistStore` only ever writes well-formed content, and
/// a mid-write fault surfaces before `commit`, never as a partial log append.
fn serveAnswer(ctx_ptr: *anyopaque, gpa: std.mem.Allocator, request_container: []const u8) serve_mod.AnswerError!serve_mod.XferReply {
const ctx: *ServeCtx = @ptrCast(@alignCast(ctx_ptr));
const reply = try serveRound(ctx, gpa, request_container);
// Past this point nothing borrows the snapshot: the overlay is gone with
// `serveRound`'s frame and the reply owns its own bytes. Dropping the
// plaintext memo is bounded (it frees only what this round inflated), so it
// is safe inside the deadline.
//
// A round that appended (a push) has left the snapshot stale — it still
// describes the pre-push log. It is NOT rebuilt here: an O(store) rebuild
// inside the reply's connection deadline is exactly #13's failure. The
// rebuild is the accept loop's job, via `ServeCtx.maintain`, run after this
// connection closes and before the next accept — outside every deadline. The
// append-only stat gate in `baseSnapshot` makes the stale snapshot self-
// correcting: the next `baseSnapshot` (in `maintain` or the next round) sees
// the grown size and rebuilds, so no `appended` signal needs to be threaded
// out of the round.
//
// A post-scan read fault this round mapped a held artifact to null. The §8.2
// clone pager skips such a rid and still reports the sequence complete
// (NEXT 0), silently finishing a clone MISSING an artifact — a failure the
// copy-based prime could not produce (a fault surfaced at prime, and a
// retained-but-uninflatable container was still served verbatim). Read the
// flag BEFORE releasing the round's memos (which clears it), then surface it
// as a storage error so the client retries; the next round re-reads.
const faulted = if (ctx.snapshot) |s| s.read_fault else false;
ctx.releaseSnapshotPlaintext();
if (faulted) {
gpa.free(reply.body);
return error.ServerStorage;
}
return reply;
}
/// One round over a per-request write overlay. Split out of `serveAnswer` so the
/// overlay — which borrows the cached snapshot — is provably destroyed before
/// `serveAnswer` touches that snapshot's storage. A round that pushes new content
/// persists it durably here; the snapshot it read from is left describing the
/// pre-push log, and `baseSnapshot`'s append-only stat gate rebuilds it on the
/// next call (in the accept loop's `maintain`, never inside this round's
/// deadline — #13), so this round signals nothing back to its caller.
fn serveRound(
ctx: *ServeCtx,
gpa: std.mem.Allocator,
request_container: []const u8,
) serve_mod.AnswerError!serve_mod.XferReply {
var overlay = repostate.OverlayStore.init(gpa, try ctx.baseSnapshot(gpa));
defer overlay.deinit();
// submanifests §8.1/§7.2 — the per-round arming of the send gate. The
// long-lived `ctx.server` is a build identity (version, servercode, project
// code) and outlives every snapshot; the seen-submanifest indicator is a
// fact about the CURRENT log and is re-read whenever the snapshot is. Copying
// the server per round is what keeps the two from drifting: a push that
// stores this repository's first tree arms the gate for the next round rather
// than for the next process.
var round_server = ctx.server.*;
round_server.seen_submanifest = if (ctx.snapshot) |snap| snap.seen_submanifest else false;
// submanifests §7.2 — the same per-round freshness for the §8.1 setting: it
// decides whether this reply carries `pragma avoid-submanifests`, and an
// operator who runs `zit submanifest enable` against a served repository
// must be obeyed by the next round, not by the next process.
round_server.submanifests_enabled = if (ctx.snapshot) |snap| snap.submanifests_enabled else zit.sync.submanifest_emission_default;
// sync §9.3 / submanifests §6.3 — the two settings this server only ever
// REPORTS, on the same per-round freshness as the two above: a `zit
// submanifest retune` against a served repository must be what the next
// clone adopts.
round_server.submanifest_k = if (ctx.snapshot) |snap| snap.submanifest_k else 0;
round_server.submanifest_auto_retune = if (ctx.snapshot) |snap| snap.submanifest_auto_retune else false;
const reply = try serve_mod.handleXfer(gpa, &round_server, overlay.store(), request_container);
errdefer gpa.free(reply.body);
// sync §10.3 r33 — an AUTHORIZED pushed config card's settings, applied to
// the served repository here because this is where the config lives; the
// responder is sans-IO and only reports the decision. Unreachable today by
// construction: `runServe` configures no credential, so `Server.hasAdmin` is
// false for every requester and every push is rejected (the e2e pins that).
// It is wired anyway because the alternative — a server that can AUTHORIZE a
// push and then silently drop it — is the failure mode worth ruling out.
if (reply.config_apply.any()) {
updateConfig(gpa, ctx.io, ctx.root_dir, .{
.enable_submanifests = reply.config_apply.enable,
.submanifest_k = reply.config_apply.k,
.submanifest_auto_retune = reply.config_apply.auto_retune,
}) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage, error.Corrupt => return error.ServerStorage,
};
}
if (overlay.writtenCount() > 0) {
// submanifests §7.1 — the durable half of the push-side duty; the
// reply-visible half is `sync.ingest`, which phantomizes into the very
// store the §3.2 walk reads. The gap COUNT is discarded on purpose: a
// stateless server (§3.8/§8.9) has no operator to report to, and §7.1
// asks for a phantom, not a diagnostic.
_ = persistStore(gpa, ctx.io, ctx.root_dir, overlay.store()) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage, error.Corrupt => return error.ServerStorage,
};
}
return reply;
}
/// A serve snapshot that reads container bytes from the log ON DEMAND instead of
/// copying the whole repository into RAM at prime (#14). It owns a read-only
/// `LogStore` — whose open scan already whole-frame-CRC'd every byte, so the
/// single read stays; the second read `loadContentStore` performed does not — and
/// a resident INDEX: the adapter-rid→log-rid map plus the lowercase-hex wire names
/// the responder walks. `held()` answers from that index with zero I/O; `wire()`
/// and `content()` read the one frame they are asked for and memoise it for the
/// round; `releaseTransient` drops those memos between rounds, so a round's
/// resident bytes are bounded by what it emitted, not by the store.
///
/// Heap-allocated because it owns a `storage.FileStorage` whose `storageFile()`
/// vtable points back into the struct — the box must not move once the log borrows
/// it. `openFromFile` takes that ownership (the serve path); `open` borrows a
/// caller-owned `StorageFile` (a test instrumenting read traffic).
///
/// Adapter rids `1..=rids.len` name the log's non-config artifact frames in
/// `ridsInPersistedOrder` — the SAME enumeration `loadContentStore` assigns, so
/// adapter rid r is byte-for-byte the rid the old MemStore snapshot gave, which is
/// what keeps the clone pager's persisted-order cursor (sync §8.2) stable. Rids
/// above `rids.len` are an appendix `MemStore` holding any hash a caller `intern`s
/// or `put`s that the log does not carry. The serve path never writes the base
/// (every write lands in the per-request `OverlayStore`), so the appendix stays
/// empty in production; it exists so this store honours the whole `ArtifactStore`
/// interface — composed over the existing `MemStore` rather than hand-rolled —
/// instead of trapping on a write it will never see.
const LogIndexStore = struct {
gpa: std.mem.Allocator,
/// Owned storage binding on the `openFromFile` path (the log borrows
/// `&self.fs`); `undefined` on the `open` path, where the caller owns the
/// storage and keeps it alive for this store's life.
fs: storage.FileStorage,
log: logstore.LogStore,
/// Adapter rid a (1-based) → `rids[a-1]` = the log rid. Persisted order, config
/// frames removed.
rids: []logstore.Rid,
/// Adapter rid a → `names[a-1]` = its lowercase-hex wire name (sync §3.3),
/// backed by `name_arena`. The single site the binary-log digest becomes the
/// wire name — the boundary `loadContentStore` held, kept to one place.
names: [][]const u8,
name_arena: std.heap.ArenaAllocator,
by_name: std.StringHashMapUnmanaged(repostate.Rid) = .empty,
/// Per-round transient memos over the persisted rids: the owned plaintext /
/// wire container that `content()`/`wire()` read on demand, released each round
/// by `releaseTransient`.
content_memo: []?[]u8,
wire_memo: []?WireHold,
/// Hashes not in the log, interned/put by a caller. Empty on the serve path.
appendix: repostate.MemStore,
/// Total plaintext this store has read on demand from the log, monotonic
/// across `releaseTransient` — the work counter that makes "answering a round
/// materialises only what it emits" testable (the log-backed twin of
/// `MemStore.inflated_bytes`). Container reads are excluded, as they were
/// there: this measures inflation, not verbatim retransmission.
materialized_bytes: usize = 0,
/// Set true when an on-demand `content()`/`wire()` read this round mapped a
/// real error to null (a post-scan I/O fault or an OOM building the wire
/// form) — NOT when a frame legitimately has no plaintext or no wire form.
/// The vtable cannot signal an error, so the §8.2 clone pager would skip the
/// null rid and still report the sequence complete, shipping a clone missing
/// an artifact; `serveAnswer` reads this after the round and converts a set
/// flag to `error.ServerStorage` so the client retries instead. Cleared with
/// the round's memos by `releaseTransient`, so every round starts clean.
read_fault: bool = false,
/// draft-fossil-submanifests-00 §8.1 — this log's durable seen-submanifest
/// indicator, read from the config frame of the SAME log generation this
/// index describes. It arms the §7.2 send gate (`serveRound`), and it lives
/// here rather than beside the long-lived `sync.Server` so it is re-read
/// whenever the snapshot is: a push that lands the repository's first tree
/// grows the log, the stat gate rebuilds the index, and the very next round
/// serves that tree through an ARMED gate. Cached on a stale snapshot it
/// would stay false for the life of the process.
seen_submanifest: bool = false,
/// draft-fossil-submanifests-00 §8.1 — this log's `enable-submanifests`
/// setting, read from the same config frame and for the same reason: §7.2
/// makes the reply's `avoid-submanifests` pragma a function of it, so a
/// stale copy would advertise the wrong policy for the life of the process.
submanifests_enabled: bool = zit.sync.submanifest_emission_default,
/// draft-fossil-submanifests-00 §6.3 / §6.4 erratum — this log's pinned
/// partition parameter and its auto-retune switch, read off the same config
/// frame as the two above and for the same freshness reason: sync §9.3
/// serves them to a cloning peer, so an operator who re-pins k must be
/// answered by the next round, not by the next process.
submanifest_k: u8 = 0,
submanifest_auto_retune: bool = false,
/// A memoised wire form for one persisted rid: the owned §4.1 container and,
/// for the delta form, the basis' owned hex name (DELTASRC) and the
/// reconstructed target size (delta §3.2). Freed by `releaseTransient`.
const WireHold = struct {
container: []u8,
delta: ?Delta = null,
const Delta = struct { src_hex: []u8, target_size: u64 };
};
/// Build over a caller-owned `StorageFile` (the log borrows it; `fs` unused).
fn open(gpa: std.mem.Allocator, storage_file: storage.StorageFile) logstore.OpenError!*LogIndexStore {
const self = try gpa.create(LogIndexStore);
errdefer gpa.destroy(self);
self.gpa = gpa;
self.fs = undefined;
try self.build(storage_file);
return self;
}
/// Build over an owned `FileStorage` bound to `file` (the serve path). `size`
/// must equal `file`'s current length (`FileStorage.init`'s precondition).
fn openFromFile(gpa: std.mem.Allocator, io: std.Io, file: File, size: u64) logstore.OpenError!*LogIndexStore {
const self = try gpa.create(LogIndexStore);
errdefer gpa.destroy(self);
self.gpa = gpa;
self.fs = storage.FileStorage.init(io, file, size);
try self.build(self.fs.storageFile());
return self;
}
fn build(self: *LogIndexStore, storage_file: storage.StorageFile) logstore.OpenError!void {
self.materialized_bytes = 0;
// `self` came from `gpa.create` (undefined memory) and is filled field by
// field, so the struct-declaration default `read_fault = false` never runs.
// `serveAnswer` reads this flag on the FIRST round after a prime/rebuild,
// before any `releaseTransient` clears it — leave it uninitialised and a
// garbage-true value spuriously fails a healthy round. Zero it here.
self.read_fault = false;
self.name_arena = std.heap.ArenaAllocator.init(self.gpa);
errdefer self.name_arena.deinit();
self.by_name = .empty;
errdefer self.by_name.deinit(self.gpa);
self.appendix = repostate.MemStore.init(self.gpa);
errdefer self.appendix.deinit();
self.log = try logstore.LogStore.openReadOnly(self.gpa, storage_file);
errdefer self.log.deinit(self.gpa);
// submanifests §8.1 — one config read per index build, never per round;
// both settings come off the same frame.
self.seen_submanifest = false;
self.submanifests_enabled = zit.sync.submanifest_emission_default;
self.submanifest_k = 0;
self.submanifest_auto_retune = false;
if (try self.log.readConfig(self.gpa)) |*oc_const| {
var oc = oc_const.*;
defer oc.deinit(self.gpa);
self.seen_submanifest = oc.config.seen_submanifest;
self.submanifests_enabled = oc.config.enable_submanifests;
self.submanifest_k = oc.config.submanifest_k;
self.submanifest_auto_retune = oc.config.submanifest_auto_retune;
}
const persisted = try self.log.ridsInPersistedOrder(self.gpa);
defer self.gpa.free(persisted);
var n: usize = 0;
for (persisted) |lr| {
if (self.log.kindOf(lr) != .config) n += 1;
}
self.rids = try self.gpa.alloc(logstore.Rid, n);
errdefer self.gpa.free(self.rids);
self.names = try self.gpa.alloc([]const u8, n);
errdefer self.gpa.free(self.names);
self.content_memo = try self.gpa.alloc(?[]u8, n);
errdefer self.gpa.free(self.content_memo);
self.wire_memo = try self.gpa.alloc(?WireHold, n);
errdefer self.gpa.free(self.wire_memo);
try self.by_name.ensureTotalCapacity(self.gpa, @intCast(n));
var a: usize = 0;
for (persisted) |lr| {
if (self.log.kindOf(lr) == .config) continue;
// The one site the binary-log digest becomes the wire name (sync §3.3),
// exactly as `loadContentStore` did — held to one place.
var name_buf: [2 * digest_len]u8 = undefined;
const wire_name = digestToWireName(self.log.hashOf(lr), &name_buf);
const owned = try self.name_arena.allocator().dupe(u8, wire_name);
self.rids[a] = lr;
self.names[a] = owned;
self.content_memo[a] = null;
self.wire_memo[a] = null;
// Hashes are unique per log entry (content-addressed intern), so no key
// collides; adapter rid is a+1.
self.by_name.putAssumeCapacity(owned, @intCast(a + 1));
a += 1;
}
std.debug.assert(a == n);
}
/// Free every per-round transient memo, restoring the store to "index only,
/// no materialised container bytes" — the shape between rounds. Every borrowed
/// `content()`/`wire()` slice handed out this round is invalidated here, so it
/// runs only once every borrow of the snapshot has ended (see
/// `ServeCtx.releaseSnapshotPlaintext`).
fn releaseTransient(self: *LogIndexStore) void {
for (self.content_memo) |*slot| {
if (slot.*) |b| self.gpa.free(b);
slot.* = null;
}
for (self.wire_memo) |*slot| {
if (slot.*) |wh| {
self.gpa.free(wh.container);
if (wh.delta) |d| self.gpa.free(d.src_hex);
}
slot.* = null;
}
// A round's fault is a property of that round's reads; clearing it here
// (once every borrow of the round's memos has ended) lets a retry start
// clean over the same snapshot.
self.read_fault = false;
}
/// Total container/plaintext bytes currently memoised — a test/accounting
/// probe. ~0 right after a prime (the whole point: no whole-store copy) and
/// after `releaseTransient`; rises only as a round materialises what it emits.
fn residentBytes(self: *const LogIndexStore) usize {
var total: usize = 0;
for (self.content_memo) |slot| {
if (slot) |b| total += b.len;
}
for (self.wire_memo) |slot| {
if (slot) |wh| total += wh.container.len;
}
return total;
}
fn deinit(self: *LogIndexStore) void {
self.releaseTransient();
self.gpa.free(self.content_memo);
self.gpa.free(self.wire_memo);
self.gpa.free(self.rids);
self.gpa.free(self.names);
self.by_name.deinit(self.gpa);
self.name_arena.deinit();
self.appendix.deinit();
self.log.deinit(self.gpa);
}
/// Free the store's internals and the heap box. `self` is invalid after.
fn destroy(self: *LogIndexStore) void {
const gpa = self.gpa;
self.deinit();
gpa.destroy(self);
}
fn store(self: *LogIndexStore) repostate.ArtifactStore {
return .{ .context = self, .vtable = &vtable };
}
const vtable = repostate.ArtifactStore.VTable{
.intern = internImpl,
.put = putImpl,
.content = contentImpl,
.hashOf = hashOfImpl,
.lookup = lookupImpl,
.count = countImpl,
.wire = wireImpl,
.retain = retainImpl,
.held = heldImpl,
};
fn wireFrameOf(wh: *const WireHold) repostate.WireFrame {
if (wh.delta) |d| return .{ .container = wh.container, .delta = .{ .src_hash = d.src_hex, .target_size = d.target_size } };
return .{ .container = wh.container };
}
fn countImpl(context: *anyopaque) usize {
const self: *LogIndexStore = @ptrCast(@alignCast(context));
return self.rids.len + self.appendix.store().count();
}
fn internImpl(context: *anyopaque, hash: []const u8) error{OutOfMemory}!repostate.Rid {
const self: *LogIndexStore = @ptrCast(@alignCast(context));
if (self.by_name.get(hash)) |rid| return rid;
return self.rids.len + try self.appendix.store().intern(hash);
}
fn lookupImpl(context: *anyopaque, hash: []const u8) ?repostate.Rid {
const self: *LogIndexStore = @ptrCast(@alignCast(context));
if (self.by_name.get(hash)) |rid| return rid;
if (self.appendix.store().lookup(hash)) |l| return self.rids.len + l;
return null;
}
fn putImpl(context: *anyopaque, hash: []const u8, content_bytes: []const u8) error{OutOfMemory}!repostate.Rid {
const self: *LogIndexStore = @ptrCast(@alignCast(context));
// A persisted hash is content-addressed and already carries these bytes,
// so `put` is idempotent and `content(rid)` still reads them from the log
// — the serve path never reaches this arm (writes go to the overlay).
if (self.by_name.get(hash)) |rid| {
// Parity with `MemStore.putImpl`: a repeat put must carry identical
// bytes (a differing answer is a hash collision). Verified only under
// runtime safety, and only when the log frame actually inflates — a
// read fault leaves `content` null and the assert is skipped rather
// than firing spuriously. Inflate through a THROWAWAY read (not
// `contentImpl`), so this assertion path never memoises a container or
// bumps `materialized_bytes`/`residentBytes` — the probes the round
// tests assert on stay untouched by an assert that never runs on serve.
if (std.debug.runtime_safety) {
if (self.log.content(self.gpa, self.rids[@intCast(rid - 1)]) catch null) |existing| {
defer self.gpa.free(existing);
std.debug.assert(std.mem.eql(u8, existing, content_bytes));
}
}
return rid;
}
return self.rids.len + try self.appendix.store().put(hash, content_bytes);
}
fn hashOfImpl(context: *anyopaque, rid: repostate.Rid) []const u8 {
const self: *LogIndexStore = @ptrCast(@alignCast(context));
if (rid <= self.rids.len) return self.names[@intCast(rid - 1)];
return self.appendix.store().hashOf(rid - self.rids.len);
}
fn heldImpl(context: *anyopaque, rid: repostate.Rid) bool {
const self: *LogIndexStore = @ptrCast(@alignCast(context));
// Every persisted (non-phantom, non-config) frame carries a form to serve
// verbatim, so it is held — answered from the index with no frame read.
if (rid <= self.rids.len) return true;
return self.appendix.store().held(rid - self.rids.len);
}
fn retainImpl(context: *anyopaque, rid: repostate.Rid, ingest: repostate.WireIngest) error{OutOfMemory}!void {
const self: *LogIndexStore = @ptrCast(@alignCast(context));
// Advisory (sync §7.2): a persisted rid already carries a stored form, so
// first-form-wins makes this a no-op — the log is the retained form.
if (rid <= self.rids.len) return;
return self.appendix.store().retain(rid - self.rids.len, ingest);
}
fn contentImpl(context: *anyopaque, rid: repostate.Rid) ?[]const u8 {
const self: *LogIndexStore = @ptrCast(@alignCast(context));
if (rid > self.rids.len) return self.appendix.store().content(rid - self.rids.len);
const i: usize = @intCast(rid - 1);
if (self.content_memo[i]) |b| return b;
// Read the frame on demand and memoise for the round. The vtable cannot
// signal an error, so a post-scan I/O fault (frames were CRC-verified at
// open, so this is transient EIO or bitrot after the scan) is recorded on
// `read_fault` and returned as null; `serveAnswer` turns a set flag into
// `error.ServerStorage`. A frame that legitimately has no plaintext is a
// plain null and does NOT set the flag.
const maybe = self.log.content(self.gpa, self.rids[i]) catch {
self.read_fault = true;
return null;
};
const bytes = maybe orelse return null;
self.content_memo[i] = bytes;
self.materialized_bytes += bytes.len;
return bytes;
}
fn wireImpl(context: *anyopaque, rid: repostate.Rid) ?repostate.WireFrame {
const self: *LogIndexStore = @ptrCast(@alignCast(context));
if (rid > self.rids.len) return self.appendix.store().wire(rid - self.rids.len);
const i: usize = @intCast(rid - 1);
if (self.wire_memo[i]) |*wh| return wireFrameOf(wh);
// A frame read fault is a real error (post-scan EIO/bitrot) recorded on
// `read_fault` so `serveAnswer` can fail the round; a `.raw`/phantom frame
// legitimately has no wire form (the responder compresses `content()`
// instead) and yields a plain null that does NOT set the flag.
var w = (self.log.wire(self.gpa, self.rids[i]) catch {
self.read_fault = true;
return null;
}) orelse return null;
switch (w) {
.full => |c| {
self.wire_memo[i] = .{ .container = c };
},
.delta => |d| {
// The basis precedes this frame in the log, so its wire name is the
// hex of the base digest the wire read carried (delta §3.2 target
// size alongside).
var src_buf: [2 * digest_len]u8 = undefined;
const src_hex = digestToWireName(d.baseHash(), &src_buf);
const owned_src = self.gpa.dupe(u8, src_hex) catch {
// Failing to produce the wire form here would silently drop a
// held delta from the clone; record it so the round errors.
w.deinit(self.gpa);
self.read_fault = true;
return null;
};
self.wire_memo[i] = .{ .container = d.container, .delta = .{ .src_hex = owned_src, .target_size = d.target_size } };
},
}
return wireFrameOf(&self.wire_memo[i].?);
}
};
/// Snapshot every stored artifact from the persisted log into `ms`, skipping
/// config frames (not artifacts). A content-bearing frame becomes a held entry;
/// a content-absent frame becomes a phantom — exactly the held/phantom split
/// `respond` walks (sync §3.2/§3.3).
///
/// The log keys frames by the RAW binary digest, but the sync wire names an
/// artifact by that digest's lowercase hex (sync §3.3 — a `file`/`igot`/`gimme`
/// HASH is 40/64 hex characters). The store the responder walks must therefore
/// hold the hex name, or `respond` would emit non-hex bytes that break the
/// peer's card tokenizer. Every hash crossing this boundary is hex-encoded once,
/// here — the single place binary-log identity becomes wire identity.
fn loadContentStore(
gpa: std.mem.Allocator,
log: *logstore.LogStore,
ms: *repostate.MemStore,
) serve_mod.AnswerError!void {
const rids = log.ridsInPersistedOrder(gpa) catch return error.OutOfMemory;
defer gpa.free(rids);
for (rids) |rid| {
if (log.kindOf(rid) == .config) continue;
var name_buf: [2 * digest_len]u8 = undefined;
const name = digestToWireName(log.hashOf(rid), &name_buf);
// sync §4.1/§7.2 — copy the stored container into the snapshot verbatim
// so the clone serves it without inflating here and re-compressing at
// serve. `wire()` is null only for a `.raw` frame or a phantom; those
// fall back to the plaintext path below. A `.zlib` frame carries the
// full form; a `.delta` frame carries the stored delta container plus
// its basis name and reconstructed size (the §7.2 USIZE, delta §3.2).
var maybe_wire = log.wire(gpa, rid) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage, error.Corrupt => return error.ServerStorage,
};
if (maybe_wire) |*w| {
defer w.deinit(gpa);
switch (w.*) {
.full => |container| {
_ = ms.putWire(name, container) catch return error.OutOfMemory;
},
.delta => |d| {
// The basis frame precedes this one in the log (delta_of <
// offset), so it was already inserted at an earlier rid; its
// wire name is the hex of the base's binary digest.
var src_name_buf: [2 * digest_len]u8 = undefined;
const src_name = digestToWireName(d.baseHash(), &src_name_buf);
_ = ms.putWireDelta(name, d.container, src_name, d.target_size) catch return error.OutOfMemory;
},
}
continue;
}
const maybe = log.content(gpa, rid) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage, error.Corrupt => return error.ServerStorage,
};
if (maybe) |bytes| {
defer gpa.free(bytes);
_ = ms.store().put(name, bytes) catch return error.OutOfMemory;
} else {
_ = ms.store().intern(name) catch return error.OutOfMemory;
}
}
}
// ---------------------------------------------------------------------------
// `zit clone URL DIR` — the interop milestone: a real client converging against
// a real `zit serve` over HTTP. Create a fresh repo at DIR, open an HTTP
// transport to URL, pump the sans-IO clone client (`sync.Client`) to §3.7
// convergence through `sync_driver.runSync`, then persist the received
// artifacts into DIR's on-disk log so the clone is durable. Anonymous only.
// ---------------------------------------------------------------------------
/// §9.1 — a fixed placeholder repository identity emitted in the client's
/// pull/push operation cards. Neither a fresh clone nor a pulled repository
/// carries a persistent SERVERCODE/PROJECTCODE of its own yet (real per-repo
/// sync identity is future work); the stateless server echoes the peer's codes
/// and never gates on a configured project code (sync §9.1), so any well-formed
/// 16-hex pair carries a content-only exchange. Shared by clone and pull.
const sync_servercode = "0000000000000000";
const sync_projectcode = "0000000000000000";
/// §10.1 — the release integer the sync client announces as `pragma
/// client-version`. A fixed build identity, matching `serve`'s own version
/// constant style. (`announcedVersion` raises what actually goes on the wire —
/// submanifests §7.2.)
const sync_client_version: u32 = 22700;
/// §10.1 r11/r12 — the DATED client-version pragma every zit run announces.
///
/// The DATE and TIME are not decoration: §10.1 r11/r12 make the server's
/// `server-version` reply conditional on them ("at least five tokens total"),
/// and submanifests §7.2 gates every upload on "the remote peer's announced
/// version". An undated pragma therefore leaves `Client.peer.version` at 0 for
/// the whole session — below every threshold — so an armed client would refuse
/// its own trees to a server that would have announced support if only it had
/// been asked. Soliciting the reply is what makes the gate a peer-version gate
/// rather than a permanent refusal. Fossil sends the dated form too (§A.4).
const sync_client_version_info: zit.sync.VersionInfo = .{
.version = sync_client_version,
.datetime = .{ .date = 20260715, .time = 0 },
};
/// An explicit upper bound on sync rounds (TigerStyle: every loop is bounded).
/// A single-hop reconcile converges in a handful of rounds; this is deep slack,
/// so a misbehaving peer yields `error.RoundLimitExceeded` instead of hanging.
const max_sync_rounds: usize = 1024;
/// `zit clone URL DIR` — thin wiring over three deep modules: `transport`
/// (HTTP/IO), `sync.Client` + `sync_driver.runSync` (sans-IO §3.7 loop), and
/// `LogStore` (persistence). The URL is used verbatim (sync §4 r1 — the client
/// never appends /xfer). All-or-nothing failures surface an honest exit-1
/// diagnostic; a bad URL scheme is rejected before any repo is created.
pub fn runClone(
gpa: std.mem.Allocator,
io: std.Io,
arena: std.mem.Allocator,
verb_args: []const []const u8,
streams: cli.Streams,
) cli.Exit {
return runCloneObserved(gpa, io, arena, verb_args, streams, null);
}
/// `runClone` with benchmark-owned phase attribution. Sync §8.2's project-code
/// adoption remains in the observed config phase; a null observer is the
/// ordinary CLI path, which reads no clock and owns no counters.
pub fn runCloneObserved(
gpa: std.mem.Allocator,
io: std.Io,
arena: std.mem.Allocator,
verb_args: []const []const u8,
streams: cli.Streams,
observer: ?*const CloneObserver,
) cli.Exit {
const parsed = switch (cli.parseCloneArgs(verb_args)) {
.ok => |p| p,
.unknown_option => |tok| return cli.usageError(streams, "clone", .unknown_option, tok),
.extra_argument => |tok| return cli.usageError(streams, "clone", .extra_argument, tok),
.no_url => return cli.usageError(streams, "clone", .no_url, ""),
.no_dir => return cli.usageError(streams, "clone", .no_dir, ""),
};
if (parsed.help) {
streams.out.writeAll(clone_help) catch {};
return .ok;
}
// Open the transport first: a malformed URL fails before any repo lands on
// disk (nothing half-made to roll back). `open` performs no network I/O.
observeClonePhase(observer, .transport, .begin);
var tr = transport.Transport.open(io, gpa, parsed.url) catch |e| {
observeClonePhase(observer, .transport, .end);
return syncOpenFail(streams, "clone", parsed.url, e);
};
observeClonePhase(observer, .transport, .end);
defer tr.close();
// Create a fresh `.zit` at DIR by reusing the init sequence, quietly (its
// own success line is suppressed; clone prints its own). A pre-existing repo
// there is init's "already exists" error, propagated verbatim.
observeClonePhase(observer, .initialize, .begin);
const init_rc = runInit(gpa, io, arena, &.{ "-q", "--", parsed.dir }, streams);
observeClonePhase(observer, .initialize, .end);
if (init_rc != .ok) return init_rc;
// Converge an in-memory store against the peer, then persist it into DIR.
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
observeClonePhase(observer, .exchange, .begin);
const outcome = cloneRun(gpa, &tr, &ms) catch |e| {
observeClonePhase(observer, .exchange, .end);
return syncRunFail(streams, "clone", e);
};
observeClonePhase(observer, .exchange, .end);
observeClonePhase(observer, .persist, .begin);
var target = std.Io.Dir.cwd().openDir(io, parsed.dir, .{}) catch
return renderStorage(streams, parsed.dir, "cannot reopen the clone target directory");
defer target.close(io);
var file = target.openFile(io, repo_filename, .{ .mode = .read_write }) catch
return renderStorage(streams, parsed.dir, "cannot write the cloned repository");
defer file.close(io);
const st = file.stat(io) catch
return renderStorage(streams, parsed.dir, "cannot write the cloned repository");
var fs = storage.FileStorage.init(io, file, st.size);
var log = logstore.LogStore.open(gpa, fs.storageFile(), 0) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "clone"),
error.Storage => return renderStorage(streams, parsed.dir, "cannot write the cloned repository"),
error.Corrupt => return renderStorage(streams, parsed.dir, "the peer sent a malformed artifact name"),
};
defer log.deinit(gpa);
const persisted = persistStoreOpen(gpa, &log, ms.store()) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "clone"),
error.Storage => return renderStorage(streams, parsed.dir, "cannot write the cloned repository"),
error.Corrupt => return renderStorage(streams, parsed.dir, "the peer sent a malformed artifact name"),
};
const written = persisted.written;
observeClonePhase(observer, .persist, .end);
// §8.2 and submanifests §6.3/§8.1 — adopt every origin-owned setting in
// one field-preserving overlay. Missing fields leave `init`'s values
// untouched, and one combined frame avoids intermediate clone config state.
observeClonePhase(observer, .config, .begin);
const project_code = outcome.projectCode();
const cloned_patch: ConfigPatch = .{
.project_code = if (project_code.len > 0) project_code else null,
.enable_submanifests = outcome.submanifest.enable,
.submanifest_k = outcome.submanifest.k,
.submanifest_auto_retune = outcome.submanifest.auto_retune,
};
if (cloned_patch.any()) {
updateConfigOpen(gpa, &log, cloned_patch) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "clone"),
error.Storage => return renderStorage(streams, parsed.dir, "cannot write the cloned repository"),
error.Corrupt => return renderStorage(streams, parsed.dir, "the cloned repository config is unreadable"),
};
}
// Every persisted artifact and the optional combined config frame must be
// durable before the cache observes this log or clone reports success.
log.commit() catch return renderStorage(streams, parsed.dir, "cannot write the cloned repository");
observeClonePhase(observer, .config, .end);
// The committed writable log is still open, so reuse its derivation path
// directly instead of reopening `.zit` solely to prime the disposable cache.
observeClonePhase(observer, .warm_cache, .begin);
Repository.warmCacheFromLog(gpa, io, target, &log);
observeClonePhase(observer, .warm_cache, .end);
streams.out.print("cloned {d} artifacts into {s} in {d} rounds\n", .{ written, parsed.dir, outcome.rounds }) catch {};
// submanifests §7.1/§7.3 — a clone converges on the cfile cursor, so it
// copies the peer's holdings faithfully, INCLUDING an incomplete tree
// closure the peer itself was carrying. The phantoms the persist hook minted
// say so; report them, because a check-in whose file set cannot be produced
// (§5.2) is not a converged clone, and "exit 0 with nothing said" is exactly
// how a repository stays quietly broken. Reported AFTER the success line and
// AFTER the repository is fully written: a partial clone is still progress,
// and this is the same shape `pull` already uses for artifacts it did not
// obtain.
if (persisted.submanifest_gaps > 0) {
streams.err.print(
"zit clone: {d} sub-manifest references are missing; the cloned repository cannot enumerate every check-in\n",
.{persisted.submanifest_gaps},
) catch {};
return .failure;
}
return .ok;
}
/// Render an `open` diagnostic to stderr and fail. Best-effort on the write: a
/// diagnostic that cannot be printed must not change the exit code.
fn openDiag(s: cli.Streams, msg: []const u8) cli.Exit {
s.err.print("zit open: {s}\n", .{msg}) catch {};
return .failure;
}
/// Map a materialization fault to its `open` diagnostic, one arm per
/// `update.ApplyError` variant and NO `else`: a new variant must fail this
/// compile and force a decision here, exactly as it does in `updateApplyFail`.
///
/// Only `InvalidCheckinId` and `CheckinNotFound` are genuinely unreachable —
/// `open` hands the engine a hash it just resolved out of this repository's own
/// derived state, never a caller-supplied name — and even those get honest
/// prose rather than a corruption claim. Everything else is reachable on an
/// intact repository: the content/manifest arms describe a repository that
/// really is damaged, while `SymlinkUnsupported` / `InvalidPath` /
/// `ReservedPath` are raised while planning the TARGET manifest, so a clone of
/// any repository containing a symlink or a path this platform cannot represent
/// reaches them with nothing corrupt at all. Reporting those as corruption
/// would be a false accusation against a healthy repository; the prose mirrors
/// `updateApplyFail`'s.
fn openApplyFail(s: cli.Streams, err: update.ApplyError) cli.Exit {
return switch (err) {
error.OutOfMemory => openDiag(s, "out of memory"),
error.InvalidCheckinId => openDiag(s, "cannot read repository: the resolved check-in ID is not a valid artifact name"),
error.CheckinNotFound => openDiag(s, "cannot read repository: the resolved check-in is not present"),
error.CheckinContentMissing => openDiag(s, "cannot read repository: check-in content is missing"),
error.CheckinNameMismatch => openDiag(s, "cannot read repository: check-in content does not match its name"),
error.TargetNotManifest => openDiag(s, "cannot read repository: the check-in is not a valid manifest"),
error.BaselineNotFound => openDiag(s, "cannot read repository: the check-in baseline is not found"),
error.BaselineContentMissing => openDiag(s, "cannot read repository: the check-in baseline content is missing"),
error.BaselineNameMismatch => openDiag(s, "cannot read repository: the check-in baseline content does not match its name"),
error.BaselineNotManifest => openDiag(s, "cannot read repository: the check-in baseline is not a valid manifest"),
error.BaselineIsDelta => openDiag(s, "cannot read repository: the check-in baseline is not fully resolvable"),
error.StateRecordInvalid => openDiag(s, "cannot read repository: " ++ update.state_record_invalid_reason),
error.Corrupt => openDiag(s, "cannot read repository: repository is corrupt"),
error.SymlinkUnsupported => openDiag(s, "the check-in contains an unsupported symlink entry"),
error.InvalidPath => openDiag(s, "the check-in names an invalid file path"),
error.ReservedPath => openDiag(s, "the check-in names a reserved file path"),
error.WorktreeNotClean => openDiag(s, "working tree is not clean"),
error.UnmanagedCollision => openDiag(s, "an untracked file or directory is in the way"),
error.CurrentStateInvalid => openDiag(s, "current working-tree state is invalid"),
error.Storage => openDiag(s, "cannot update working tree: input/output error"),
error.TransactionUnavailable => openDiag(s, "cannot create a transaction directory"),
error.RollbackFailed => openDiag(s, "open failed and could not be rolled back; the previous working-tree state is retained for recovery"),
error.CleanupFailed => openDiag(s, "open committed, but the transaction directory could not be removed"),
};
}
/// `zit open` — the second half of `zit clone URL DIR`: turn a repository whose
/// checkout is still UNBORN into a materialized working tree. Fossil's own UX is
/// the same two steps (`fossil clone` then `fossil open`), and splitting them
/// keeps `runClone` strictly repo-only.
///
/// Thin wiring over the shipped machinery, deliberately: probe the checkout
/// record, resolve the tip of the main branch out of the derived repo state, and
/// hand materialization to the SAME `update.applyCheckin` engine every other
/// working-tree move rides (journal-armed, transactional, rollback-safe). Only
/// the tip resolution is open's own work, and it is load-bearing: `update`
/// accepts nothing but a literal 40/64-hex check-in ID, so a freshly cloned tree
/// has no way to name its own tip.
///
/// It writes NO checkout record and NO managed set of its own. `init` — which
/// `clone` delegates repo creation to — already wrote the unborn record; an
/// absent managed set already reads as the empty set; and the update engine is
/// the single writer of the born record and the managed set. Adding a second
/// writer of that bootstrap state would only create two things to keep in sync.
pub fn runOpen(
gpa: std.mem.Allocator,
io: std.Io,
arena: std.mem.Allocator,
verb_args: []const []const u8,
streams: cli.Streams,
) cli.Exit {
const parsed = switch (cli.parseOpenArgs(verb_args)) {
.ok => |p| p,
.unknown_option => |tok| return cli.usageError(streams, "open", .unknown_option, tok),
.extra_argument => |tok| return cli.usageError(streams, "open", .extra_argument, tok),
.repeated_option => |tok| return cli.usageError(streams, "open", .repeated_option, tok),
.empty_value => |tok| return cli.usageError(streams, "open", .empty_value, tok),
.bad_manifest => |tok| return cli.usageError(streams, "open", .bad_manifest, tok),
};
if (parsed.help) {
streams.out.writeAll(open_help) catch {};
return .ok;
}
// The full derive: resolving the main branch reads the tag tables, which the
// `.fast_read` image path does not carry.
var repo: Repository = undefined;
var root_dir: std.Io.Dir = undefined;
const rc = openRepo(gpa, io, arena, streams, "open", null, false, false, &repo, &root_dir);
if (rc != .ok) return rc;
defer root_dir.close(io);
defer repo.deinit(gpa);
// BORN-ness, not presence, is the "already open" signal. Every fresh
// repository carries a checkout record — `init` writes an unborn one
// (`version == ""`, branch `trunk`) as its step 8, and `clone` delegates to
// `init` — so a presence-based gate would reject `zit clone && zit open` on
// every clone, the primary path this verb exists to serve. Only a non-empty
// version means a tree has actually been materialized against a check-in.
var ck_buf: [checkout.max_ck_payload + 1]u8 = undefined;
const current = checkout.read(io, root_dir, &ck_buf) catch |e| switch (e) {
error.Storage => return openDiag(streams, "cannot read the checkout state: input/output error"),
error.Corrupt => return openDiag(streams, "the checkout state is corrupt"),
};
if (current) |state| if (state.version.len != 0) {
streams.err.print(
"zit open: this working tree is already open at check-in '{s}'; use 'zit update' to change check-ins\n",
.{state.version},
) catch {};
return .failure;
};
// The tip of the main branch (repo-state §9 current-trunk).
const main_branch = repo.mainBranch();
// Switched, not an unbound `catch`: the success message below is an exit-0
// claim that the tree is already correct, and it is only honest for
// `NotFound`. Naming the arm means a later Storage/OOM arm on `currentTrunk`
// fails this compile instead of silently reporting a genuine failure as
// success on a materialization verb.
const tip = repo.state().currentTrunk(main_branch) catch |e| switch (e) {
// A repository with no check-in on its main branch is ALREADY in its
// correct end state — the unborn record and an empty tree. There is
// nothing to materialize and nothing wrong, so this is reported, not
// failed, and the record is left exactly as it was found.
error.NotFound => {
streams.out.print(
"open: no check-in on '{s}' yet; the working tree is already in its correct (empty) state\n",
.{main_branch},
) catch {};
return .ok;
},
// Settling an equal-mtime tie by ancestry allocates (repo-state
// `MtimeMaxPick`). A failure here is a genuine failure — reporting it as
// the exit-0 "already correct" message would be the lie the switch
// above exists to prevent.
error.OutOfMemory => return openDiag(streams, "out of memory"),
};
// Both borrow the derived view, which outlives the update below.
const target = repo.hashOf(tip);
const branch = repo.branchOf(tip);
// An assertion, not a check: unlike `update`, whose target is user-supplied
// and therefore validated, open's target comes out of this repository's own
// derived state. A malformed name here is a defect in the derivation, not
// bad input, so it belongs in the negative space rather than in a
// diagnostic path no correct run can reach.
std.debug.assert(artifact.validateName(target));
const log = repo.logStore(gpa) catch |e| return repoOpenFail(streams, "open", e);
// `open` is the ONE verb that SETS the manifest opt-in; absent flag records
// "none", which is what an untouched checkout already carried.
const manifest: update.ManifestOpt = .{ .set = parsed.manifest orelse "" };
update.applyCheckin(gpa, io, root_dir, log, target, manifest) catch |e| return openApplyFail(streams, e);
streams.out.print("opened {s} on {s}\n", .{ target, branch }) catch {};
return .ok;
}
/// The fields a `updateConfig` overlay may change; a null field is left exactly
/// as the stored config has it. Naming a single change site keeps the "copy the
/// whole config, overlay one field" discipline in one place — a caller that
/// means to change only the project code can never silently drop the credential
/// or the saved remote (the class of bug the pre-`updateConfig` re-puts had).
const ConfigPatch = struct {
project_code: ?[]const u8 = null,
login: ?zit.sync.Credential = null,
remote_url: ?[]const u8 = null,
/// draft-fossil-submanifests-00 §6.3/§8.1 and the §6.4 erratum — the three
/// project-scope settings sync §9.3's scalar channel carries. Optional for
/// the same reason every field here is: a peer that named only one of them
/// must leave the other two exactly as stored.
enable_submanifests: ?bool = null,
submanifest_k: ?u8 = null,
submanifest_auto_retune: ?bool = null,
fn any(self: ConfigPatch) bool {
return self.project_code != null or
self.login != null or
self.remote_url != null or
self.enable_submanifests != null or
self.submanifest_k != null or
self.submanifest_auto_retune != null;
}
};
/// Read the live config from an already-open log, overlay `patch`, and append
/// the result while preserving every unnamed field. The caller owns durability,
/// so a clone can combine its origin settings in one config frame and commit
/// its complete result once.
fn updateConfigOpen(
gpa: std.mem.Allocator,
log: *logstore.LogStore,
patch: ConfigPatch,
) PersistError!void {
var oc = (log.readConfig(gpa) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => return error.Storage,
error.Corrupt => return error.Corrupt,
}) orelse return error.Corrupt; // init always writes a config frame
defer oc.deinit(gpa);
// Copy every field, then overlay only the named ones. `next`'s string slices
// borrow `oc.backing` / `patch`; both outlive the `putConfig` memcpy below.
var next = oc.config;
if (patch.project_code) |c| next.project_code = c;
if (patch.login) |cred| {
next.login_user = cred.user;
// Point at the function-scoped `patch` storage, not the block-local
// capture `cred`: `next` is read by `putConfig` after this block closes.
next.login_secret_hex = &patch.login.?.secret_hex;
}
if (patch.remote_url) |u| next.remote_url = u;
if (patch.enable_submanifests) |v| next.enable_submanifests = v;
if (patch.submanifest_k) |v| next.submanifest_k = v;
if (patch.submanifest_auto_retune) |v| next.submanifest_auto_retune = v;
log.putConfig(next) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => return error.Storage,
};
}
/// Open `dir/.zit` for a field-preserving config update, retaining the legacy
/// wrapper for pull, sync, serve, and remote callers that do not already own a
/// writable log.
fn updateConfig(
gpa: std.mem.Allocator,
io: std.Io,
dir: std.Io.Dir,
patch: ConfigPatch,
) PersistError!void {
var file = dir.openFile(io, repo_filename, .{ .mode = .read_write }) catch return error.Storage;
defer file.close(io);
const st = file.stat(io) catch return error.Storage;
var fs = storage.FileStorage.init(io, file, st.size);
var log = logstore.LogStore.open(gpa, fs.storageFile(), 0) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => return error.Storage,
error.Corrupt => return error.Corrupt,
};
defer log.deinit(gpa);
try updateConfigOpen(gpa, &log, patch);
log.commit() catch return error.Storage;
}
/// §8.2 — persist the origin's project code into the freshly cloned repo at
/// `dir/.zit`, preserving every other field `init` wrote. A no-op when the peer
/// sent no code. Thin wrapper over `updateConfig` (the field-preserving RMW).
fn storeClonedProjectCode(
gpa: std.mem.Allocator,
io: std.Io,
dir: std.Io.Dir,
project_code: []const u8,
) PersistError!void {
if (project_code.len == 0) return; // peer sent none; keep the local init code
return updateConfig(gpa, io, dir, .{ .project_code = project_code });
}
/// draft-fossil-submanifests-00 §6.3/§8.1 (and the §6.4 erratum) over sync §9.3
/// — persist the origin's three sub-manifest settings into the freshly cloned
/// repo at `dir/.zit`, preserving every other field `init` wrote. A no-op when
/// the peer named none. Thin wrapper over `updateConfig`, the same shape
/// `storeClonedProjectCode` has and for the same reason.
///
/// One-way and clone-only, deliberately. A fresh clone has nothing of its own
/// to conflict with, so adoption needs no merge; an EXISTING repository would,
/// and §9.3's channel resolves conflicts by MTIME, which zit's config frame does
/// not store (see `sync.submanifest_config_mtime`). Refreshing settings on a
/// later pull is therefore a real conflict-resolution feature with a config
/// schema change behind it, not plumbing this wrapper can be widened into —
/// `runPull`/`runPush` deliberately have no seam here.
fn storeClonedSubmanifestSettings(
gpa: std.mem.Allocator,
io: std.Io,
dir: std.Io.Dir,
cfg: zit.sync.SubmanifestConfig,
) PersistError!void {
if (!cfg.any()) return; // peer sent none; keep the local init defaults
return updateConfig(gpa, io, dir, .{
.enable_submanifests = cfg.enable,
.submanifest_k = cfg.k,
.submanifest_auto_retune = cfg.auto_retune,
});
}
/// Drives a modern-clone client (§3.8 CLONE_ROUND0 → EXCHANGE) to convergence
/// over `tr`, filling `ms` with the peer's artifacts; returns the round count.
/// Sans-IO on this side of `tr.exchange`: the state machine only moves
/// `[]const u8` bytes. Its closed error set is the driver's own causes unioned
/// with the transport's, so `runClone` switches the whole surface exhaustively.
/// The result of a clone run: the round count plus the origin's project code
/// (§8.2), copied out of the sans-IO client so the caller can persist the
/// shared identity after the client is torn down.
const CloneOutcome = struct {
rounds: usize,
project_code_buf: [zit.sync.max_projectcode_len]u8,
project_code_len: usize,
/// submanifests §6.3/§8.1 over sync §9.3 — the origin's three project-scope
/// settings, learned from the cleanup round's config card. Copied out of the
/// sans-IO client for the same reason the project code is: the client is
/// torn down before the caller persists anything. Owns no memory (three
/// optional scalars), so no buffer twin is needed.
submanifest: zit.sync.SubmanifestConfig,
fn projectCode(self: *const CloneOutcome) []const u8 {
return self.project_code_buf[0..self.project_code_len];
}
};
fn cloneRun(
gpa: std.mem.Allocator,
tr: *transport.Transport,
ms: *repostate.MemStore,
) sync_driver.RunError(*transport.Transport)!CloneOutcome {
var client = zit.sync.Client{
.state = zit.sync.initialState(.clone),
.operation = .clone,
.self_servercode = sync_servercode,
.project_code = sync_projectcode,
.version = sync_client_version_info,
};
defer client.deinit(gpa);
const stats = try sync_driver.runSync(gpa, &client, ms.store(), tr, max_sync_rounds);
// §8.2 — the peer's project code, learned from the clone reply's `push`
// card, copied out before the client is deinitialized.
var outcome = CloneOutcome{
.rounds = stats.rounds,
.project_code_buf = undefined,
.project_code_len = 0,
// §9.3 — the origin's sub-manifest settings, latched from the config
// card the §8.4 cleanup round asked for.
.submanifest = client.peerSubmanifestConfig(),
};
const pc = client.peerProjectCode();
@memcpy(outcome.project_code_buf[0..pc.len], pc);
outcome.project_code_len = pc.len;
return outcome;
}
/// The closed error set of `persistStore`: memory, a storage write fault, or a
/// corrupt/oversized peer artifact name that will not decode to a log key.
const PersistError = error{ OutOfMemory, Storage, Corrupt };
/// The disposition of one artifact in `persistStore`'s ordered pass.
const PersistOutcome = union(enum) {
/// A non-phantom frame for this artifact now exists in the log — written
/// here, or already present from a prior cascade / a pull re-persist.
persisted,
/// A reverse-delta whose basis is a content-bearing store entry not YET
/// written to the log. It is deferred on the basis wire-hash and re-driven
/// once that basis lands, so its `delta_of` offset points strictly backward.
deferred: []const u8,
};
/// What one `persistStore` sweep did: the artifacts it made durable, and the
/// sub-manifest references (draft-fossil-submanifests-00 §7.1) it phantomized
/// and could not resolve from this repository's own holdings.
const PersistSummary = struct {
/// Content-bearing artifacts in the store — the count every caller reports.
written: usize,
/// §7.1 — distinct S references still standing as phantoms when the sweep
/// ended: this repository holds an artifact whose tree closure is open.
/// Zero for every repository that carries no S card, which is every
/// repository predating this draft.
submanifest_gaps: usize,
};
/// draft-fossil-submanifests-00 §7.1 — the S references one persist sweep saw.
///
/// Rids, not names: `LogStore` entries are append-only, so an interned rid stays
/// valid for the sweep, and whether a reference is still a GAP can only be
/// decided once the sweep has ended — a sub-manifest may be persisted after the
/// tree manifest that names it, and `putZlib` upgrades the phantom minted for it
/// in place. Deciding per S card instead would report a gap for every reference
/// whose artifact happens to sit later in rid order.
const SubManifestRefs = struct {
rids: std.AutoArrayHashMapUnmanaged(logstore.Rid, void) = .empty,
/// submanifests §8.1 — set the first time this sweep stores an artifact
/// inside §7.2's gated set: "When the first S-bearing artifact is stored, an
/// implementation SHOULD record a local seen-submanifest indicator". Both
/// §7.2 shapes count, leaf sub-manifests included — see
/// `phantomizeSubManifestRefs` for why the literal S-card-only reading would
/// leave the gate blind to exactly the artifacts that carry a tree's files.
/// It is also the durable gate on the §7.1 phantom REBUILD (see
/// `persistStore`), which is why it tracks artifacts seen, not gaps left.
seen: bool = false,
fn deinit(self: *SubManifestRefs, gpa: std.mem.Allocator) void {
self.rids.deinit(gpa);
self.* = undefined;
}
/// Distinct references that never became a content-bearing frame.
fn unresolved(self: *const SubManifestRefs, log: *const logstore.LogStore) usize {
var n: usize = 0;
for (self.rids.keys()) |rid| {
if (log.isPhantom(rid)) n += 1;
}
std.debug.assert(n <= self.rids.count());
return n;
}
};
/// draft-fossil-submanifests-00 §7.1 — "Storing a structural artifact that
/// carries S cards whose referenced sub-manifests are absent MUST create a
/// phantom for each missing reference, with the referencing artifact's privacy
/// classification, exactly as an absent delta source creates a source phantom."
///
/// This is §7.1's SECOND storing boundary. The receive path
/// (`sync.Client.absorbSubManifestRefs`) covers artifacts a pull absorbs; it
/// cannot cover a clone, which stores every artifact wire-backed and never
/// inflates one (so it has no plaintext to scan), nor a server-side push
/// ingest, which absorbs nothing at all. Both of those reach the durable store
/// only HERE — and here the plaintext is already in hand, reconstructed for the
/// §7.7 verify, so the whole duty costs one `sBearingCandidate` scan of bytes
/// this function was handed anyway. The clone's O(wire) receive bound is
/// untouched: no container is inflated that §7.7 was not already inflating.
///
/// A clone from a peer whose OWN closure is incomplete is exactly the case the
/// receive path cannot see: the cursor delivers every artifact the peer holds,
/// and the peer does not hold the sub-manifests. Without this, such a clone
/// records its gap nowhere and never asks for the leaves.
///
/// Privacy: zit's phantom is an interned entry with no content and carries no
/// per-entry privacy at either boundary, so an S-reference phantom is minted
/// exactly as an absent delta source's is — which is what §7.1 asks for.
fn phantomizeSubManifestRefs(
gpa: std.mem.Allocator,
log: *logstore.LogStore,
content: []const u8,
seen: *SubManifestRefs,
) PersistError!void {
// §8.1's indicator, over BOTH §7.2 shapes rather than over S cards alone.
//
// §8.1 says "when the first S-bearing artifact is stored"; a LEAF
// sub-manifest carries F cards and no S card (§4.4), so the literal reading
// records nothing for a repository holding leaves. That reading cannot be
// right: this indicator is what arms the §7.2 transfer gate, §7.2 gates
// leaves by TYPE, and §8's whole reason for the transfer layer is that a
// pre-S reader can only hold a leaf as "unreadable ... opaque content". §8.2
// reads the same way — "a repository containing an S-bearing artifact" is a
// repository holding any part of a tree, and a repository that holds a tree's
// leaves is exactly that. So the indicator answers `transferGatedArtifact`.
//
// Cheap where it runs: the predicate rejects any byte string whose first two
// bytes are not a card letter followed by SP or LF — every file blob, in two
// comparisons. A structural artifact pays one allocation-free walk of its
// card lines, beside the full-artifact rehash §7.7 already charges here.
if (!seen.seen and repostate.transferGatedArtifact(content)) seen.seen = true;
var refs: std.ArrayListUnmanaged([]const u8) = .empty;
defer refs.deinit(gpa);
// Validate-then-apply lives in the shared extractor: a malformed card run
// mints nothing, so a half-read artifact never leaves a half-minted phantom
// set. `content` outlives `refs`, which borrows its hash tokens.
if (!try repostate.collectSubManifestRefs(gpa, content, &refs)) return;
for (refs.items) |name| {
var buf: [digest_len]u8 = undefined;
// §3.5 admits a 40- or 64-hex name and `decodeWireName` decodes both; a
// name too wide for this repository's digest cannot key a frame here, so
// it is a corrupt reference rather than a gap.
const digest = decodeWireName(name, &buf) orelse return error.Corrupt;
const rid = log.intern(digest) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
};
try seen.rids.put(gpa, rid, {});
}
}
/// Persist one artifact `rid` into `log`, choosing its stored form, and verify
/// §7.7 integrity (the reconstructed content MUST hash to its wire name) BEFORE
/// any durable write. A retained delta whose basis is already a held log frame
/// under the read depth bound is stored VERBATIM as a `.delta`; a retained delta
/// whose basis is a not-yet-written content-bearing store entry is DEFERRED
/// (when `allow_defer`) so the basis frame is written first (reverse-delta
/// ordering); every other case — no wire form, a full container, or an
/// absent / phantom / over-deep basis — lands as the full form, never a
/// phantom-based or unreadable delta. Idempotent: an artifact already holding a
/// non-phantom frame returns `.persisted` without a re-write or re-verification
/// (§7.7 was proven when it was first persisted).
fn persistArtifact(
gpa: std.mem.Allocator,
store: repostate.ArtifactStore,
log: *logstore.LogStore,
rid: repostate.Rid,
allow_defer: bool,
/// A bounded plaintext memo shared across `persistStore`'s whole sweep. Each
/// artifact reconstructs its own plaintext exactly once here — for the §7.7
/// verify — and primes it under its just-written frame's offset, so a later
/// reverse-delta naming this frame as basis reuses that plaintext instead of
/// re-walking (re-inflating, re-applying) the chain. Bounded (LRU, fixed byte
/// + entry caps): a cache miss just costs a fresh reconstruction, never a
/// wrong answer, so the stored bytes are identical with or without it.
cache: *logstore.PlaintextCache,
/// submanifests §7.1 — the sweep-wide S-reference record. Every plaintext
/// this function reconstructs for the §7.7 verify is offered to it, so the
/// phantom duty attaches to STORING and not to any one arrival path.
srefs: *SubManifestRefs,
/// submanifests §7.1/§8.1 — REBUILD the S-reference phantom set for an
/// artifact that is ALREADY a durable frame. A phantom is an interned entry
/// with no frame (`LogStore.intern`: "Phantoms never reach disk"), so the
/// §7.1 phantoms this repository minted on a previous run are gone the
/// moment the process exits; the gap has to be re-derived from the durable
/// artifacts that name it. Without this an open tree closes itself on the
/// next run: the tree manifest is already persisted, the early return skips
/// its scan, and the repository reports converged while it still cannot
/// enumerate that check-in.
///
/// Passed as a parameter rather than read here because the decision is the
/// repository's, not this artifact's: §8.1's durable seen-submanifest
/// indicator is false for every repository predating this draft, so the
/// rescan — and its one inflate per already-persisted artifact — is
/// unreachable for all of them and the persist sweep is what it always was.
rescan_persisted: bool,
) PersistError!PersistOutcome {
const name = store.hashOf(rid);
var digest_buf: [digest_len]u8 = undefined;
const digest = decodeWireName(name, &digest_buf) orelse return error.Corrupt;
if (log.lookup(digest)) |ex| {
if (!log.isPhantom(ex)) {
if (rescan_persisted) {
// Read through the sweep's own memo, so an artifact this sweep
// already reconstructed costs nothing to rescan.
const plain = (log.contentCached(gpa, ex, cache) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => return error.Storage,
error.Corrupt => return error.Corrupt,
}) orelse return .persisted;
defer gpa.free(plain);
try phantomizeSubManifestRefs(gpa, log, plain, srefs);
}
return .persisted;
}
}
// A genuine phantom carries nothing durable — skip it (never a drop of held
// content). `held` classifies without inflating a wire-backed entry.
if (!store.held(rid)) return .persisted;
// §7.7 L1332/L1336-1338 — the repository transaction-commit hook. Recompute
// the artifact hash over the reconstructed bytes and abort BEFORE any frame is
// written on a mismatch (a lying container inflates but does not name its
// content). The verify plaintext is reconstructed against the LOG basis (a
// read bounded by `max_delta_chain`) — NOT the in-memory wire chain, which,
// now a clone stores every artifact wire-backed, is the full reverse-delta
// history and would exceed the read depth bound. `persistStore` writes basis
// frames before the deltas that name them (reverse-delta order), so a delta's
// basis is already a log frame here.
if (store.wire(rid)) |wf| {
if (wf.delta) |d| {
var base_buf: [digest_len]u8 = undefined;
const base_digest = decodeWireName(d.src_hash, &base_buf) orelse return error.Corrupt;
const maybe_base = log.lookup(base_digest);
if (maybe_base == null or log.isPhantom(maybe_base.?)) {
// The basis is not yet a log frame. Defer on it (reverse-delta
// order: the newer basis frame must be written first) when it is a
// held store entry that WILL be persisted.
if (allow_defer) {
if (store.lookup(d.src_hash)) |bs_rid| {
if (store.held(bs_rid)) return .{ .deferred = d.src_hash };
}
}
// Basis absent and not deferrable (§3.3): fall back to FULL form
// from the entry's OWN plaintext — a put+retain entry carries it
// directly (never a phantom-based delta). In a converged clone this
// is unreachable (every delta's basis is a written frame, handled
// below); a wire-delta entry whose basis truly never arrived cannot
// reconstruct — a hard error, never a fabricated artifact.
const content = store.content(rid) orelse return error.Corrupt;
if (!zit.artifact.verify(name, content)) return error.Corrupt;
// submanifests §7.1 — the plaintext §7.7 just verified is the same
// bytes the phantom duty is stated over; scan it here rather than
// forcing a second inflate anywhere else.
try phantomizeSubManifestRefs(gpa, log, content, srefs);
const new_rid = log.putZlib(digest, content) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => return error.Storage,
};
try cache.put(gpa, log.offsetOf(new_rid), content);
return .persisted;
}
const base_rid = maybe_base.?;
const base_off = log.offsetOf(base_rid);
// Reuse the basis plaintext memoised when the basis frame was
// persisted — reverse-delta order writes every basis before the
// delta that names it, so the memo is normally warm. Only a miss
// (first use, or an LRU eviction) reconstructs it by walking the
// chain; either way `base_plain` is owned and freed below.
const base_plain = if (cache.get(base_off)) |hit| blk: {
// §7.7 positive space: a memo entry is the exact plaintext that
// named its wire hash at the basis's own persist, so the reused
// bytes still hash to the basis name (debug-only assertion; the
// hash is never recomputed on the hot path).
std.debug.assert(zit.artifact.verify(d.src_hash, hit));
break :blk gpa.dupe(u8, hit) catch return error.OutOfMemory;
} else blk: {
const p = (log.contentCached(gpa, base_rid, cache) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => return error.Storage,
error.Corrupt => return error.Corrupt,
}) orelse return error.Corrupt;
// `contentCached` does not memoise a top-level full-form basis;
// prime it here so sibling reverse-deltas on the same basis reuse
// this reconstruction instead of repeating it.
try cache.put(gpa, base_off, p);
break :blk p;
};
defer gpa.free(base_plain);
const raw = zit.container.decode(gpa, wf.container, @intCast(logstore.max_payload)) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return error.Corrupt,
};
defer gpa.free(raw);
const content = zit.delta.apply(gpa, base_plain, raw) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return error.Corrupt,
};
defer gpa.free(content);
if (!zit.artifact.verify(name, content)) return error.Corrupt;
// submanifests §7.1 — the plaintext §7.7 just verified is the same
// bytes the phantom duty is stated over; scan it here rather than
// forcing a second inflate anywhere else.
try phantomizeSubManifestRefs(gpa, log, content, srefs);
// Store the delta VERBATIM iff its basis is a held frame under the READ
// depth bound; an over-deep basis falls back to full form (never an
// unreadable chain). `delta_of` is the basis's backward offset —
// strictly lower than this new frame, the invariant `putDeltaContainer`
// asserts.
if (log.depthOf(base_rid) < logstore.max_delta_chain) {
const new_rid = log.putDeltaContainer(digest, base_rid, wf.container) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => return error.Storage,
};
// Prime the just-verified plaintext so an older reverse-delta
// naming this frame as basis reuses it (no chain re-walk).
try cache.put(gpa, log.offsetOf(new_rid), content);
} else {
const new_rid = log.putZlib(digest, content) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => return error.Storage,
};
try cache.put(gpa, log.offsetOf(new_rid), content);
}
return .persisted;
}
// §7.2 full form — decode to verify, then store the container verbatim.
const content = zit.container.decode(gpa, wf.container, @intCast(logstore.max_payload)) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return error.Corrupt,
};
defer gpa.free(content);
if (!zit.artifact.verify(name, content)) return error.Corrupt;
// submanifests §7.1 — the plaintext §7.7 just verified is the same
// bytes the phantom duty is stated over; scan it here rather than
// forcing a second inflate anywhere else.
try phantomizeSubManifestRefs(gpa, log, content, srefs);
const new_rid = log.putContainer(digest, wf.container) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => return error.Storage,
};
// Prime the just-verified full-form plaintext so a reverse-delta naming
// this frame as basis reuses it instead of re-inflating the container.
try cache.put(gpa, log.offsetOf(new_rid), content);
return .persisted;
}
// §4.1 — no retained wire (a raw pull `put`): the plaintext is memoised in the
// store, no chain to walk. Verify and compress it; the hash names the
// UNCOMPRESSED bytes, so compression is address-transparent.
const content = store.content(rid) orelse return error.Corrupt;
if (!zit.artifact.verify(name, content)) return error.Corrupt;
// submanifests §7.1 — the plaintext §7.7 just verified is the same
// bytes the phantom duty is stated over; scan it here rather than
// forcing a second inflate anywhere else.
try phantomizeSubManifestRefs(gpa, log, content, srefs);
const new_rid = log.putZlib(digest, content) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => return error.Storage,
};
// Prime the plaintext so a reverse-delta naming this frame as basis reuses it.
try cache.put(gpa, log.offsetOf(new_rid), content);
return .persisted;
}
/// Persists every content-bearing artifact in `store` into an already-open
/// writable log. The caller owns durability, allowing clone to append artifacts
/// and its combined config overlay before one final commit.
fn persistStoreOpen(
gpa: std.mem.Allocator,
log: *logstore.LogStore,
store: repostate.ArtifactStore,
) PersistError!PersistSummary {
var written: usize = 0;
const n = store.count();
// Reverse-delta persist ordering (§7.2/§8.2). Fossil stores an OLDER
// artifact as a delta of a NEWER one, so a delta cfile can land in the store
// at a LOWER rid than the basis it points at, while the log's `delta_of` is a
// BACKWARD offset that REQUIRES the basis frame to precede the delta. Persist
// therefore drains a worklist: a delta whose basis is not yet a frame is
// deferred (keyed by the basis wire-hash) and re-driven the instant that
// basis lands. Each content-bearing artifact seeds the drain once in rid
// order; each deferred delta is re-driven exactly once, so the pass is O(n)
// and terminates — a chain bottoms out at fossil's full newest artifact, and
// content addressing forbids a cycle.
var waiting: std.StringHashMapUnmanaged(std.ArrayListUnmanaged(repostate.Rid)) = .empty;
defer {
var it = waiting.valueIterator();
while (it.next()) |list| list.deinit(gpa);
waiting.deinit(gpa);
}
var ready: std.ArrayListUnmanaged(repostate.Rid) = .empty;
defer ready.deinit(gpa);
// One bounded plaintext memo for the whole sweep: a basis reconstructed for
// its own §7.7 verify is reused by every reverse-delta that names it, so a
// delta family reconstructs each basis ~once rather than once per referencing
// delta. Bounded (LRU byte + entry caps) — it never retains the whole repo.
var cache = logstore.PlaintextCache.init();
defer cache.deinit(gpa);
// submanifests §7.1 — one record for the whole sweep, so a reference is
// judged a gap only after every artifact this store carries has landed.
var srefs: SubManifestRefs = .{};
defer srefs.deinit(gpa);
// submanifests §8.1/§7.1 — THE REBUILD-ON-OPEN INVARIANT, stated once here.
//
// §7.1's phantoms are not durable and are not made durable: a phantom is an
// interned entry with no frame, exactly as an absent delta source's is
// ("exactly as an absent delta source creates a source phantom"), and
// `LogStore` writes none of either kind ("Phantoms never reach disk"). The
// `.zit` on-disk format version is frozen (`logstore.version`), so minting a
// durable phantom record would reject every repository already written.
//
// What makes the in-memory phantoms CORRECT across a restart is this
// invariant: the S-reference gap set is a pure function of the durable
// content-bearing frames, and this sweep re-derives it in full on every run.
// A repository whose §8.1 indicator says it has ever stored an S reference
// rescans its already-persisted artifacts (`rescan_persisted`); one that has
// not — every repository predating this draft — rescans nothing and pays
// nothing. The indicator itself IS durable, which is the whole trick: one
// config byte survives the restart instead of a phantom table.
const rescan = blk: {
var oc = log.readConfig(gpa) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => return error.Storage,
error.Corrupt => return error.Corrupt,
};
if (oc) |*c| {
defer c.deinit(gpa);
break :blk c.config.seen_submanifest;
}
break :blk false;
};
var seed: repostate.Rid = 1;
while (seed <= n) : (seed += 1) {
if (!store.held(seed)) continue; // residual phantom: not durable
written += 1; // count each content-bearing artifact exactly once, at its rid
try ready.append(gpa, seed);
while (ready.pop()) |rid| {
switch (try persistArtifact(gpa, store, log, rid, true, &cache, &srefs, rescan)) {
.persisted => {
// `rid`'s frame now exists: re-drive any reverse-deltas that
// deferred on it as their (now-written) basis.
if (waiting.fetchRemove(store.hashOf(rid))) |kv| {
var list = kv.value;
defer list.deinit(gpa);
try ready.appendSlice(gpa, list.items);
}
},
.deferred => |base_hash| {
const gop = try waiting.getOrPut(gpa, base_hash);
if (!gop.found_existing) gop.value_ptr.* = .empty;
try gop.value_ptr.append(gpa, rid);
},
}
}
}
// Defensive drain: a deferred basis is itself content-bearing and therefore
// persisted above (draining its waiters), so `waiting` is normally empty
// here. Should a basis nonetheless be missing, its waiters land as full form
// rather than be dropped — every content-bearing artifact stays durable.
var leftover = waiting.iterator();
while (leftover.next()) |e| {
for (e.value_ptr.items) |rid| {
switch (try persistArtifact(gpa, store, log, rid, false, &cache, &srefs, rescan)) {
.persisted => {},
.deferred => unreachable, // allow_defer=false never defers
}
}
}
// submanifests §7.1 — decided once, at the end: a reference is a gap only
// if nothing in this whole store turned out to name it.
const gaps = srefs.unresolved(log);
// submanifests §8.1 — "When the first S-bearing artifact is stored, an
// implementation SHOULD record a local seen-submanifest indicator so
// operational surfaces can warn before the setting is enabled or a pre-S
// client is granted access." Written once, on the false->true edge, so a
// steady-state sync appends no config frame; it is also what arms the
// rebuild above on every subsequent run of this repository.
if (srefs.seen and !rescan) {
var oc = log.readConfig(gpa) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => return error.Storage,
error.Corrupt => return error.Corrupt,
};
var cfg: logstore.Config = if (oc) |c| c.config else .{ .hash_policy = 0, .default_branch = "", .project_id = "" };
defer if (oc) |*c| c.deinit(gpa);
cfg.seen_submanifest = true;
log.putConfig(cfg) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => return error.Storage,
};
}
return .{ .written = written, .submanifest_gaps = gaps };
}
/// Opens `dir/.zit` for the legacy persistence callers, preserving their
/// artifact result and commit boundary while routing the write through the
/// already-open-log implementation.
fn persistStore(
gpa: std.mem.Allocator,
io: std.Io,
dir: std.Io.Dir,
store: repostate.ArtifactStore,
) PersistError!PersistSummary {
var file = dir.openFile(io, repo_filename, .{ .mode = .read_write }) catch return error.Storage;
defer file.close(io);
const st = file.stat(io) catch return error.Storage;
var fs = storage.FileStorage.init(io, file, st.size);
var log = logstore.LogStore.open(gpa, fs.storageFile(), 0) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => return error.Storage,
error.Corrupt => return error.Corrupt,
};
defer log.deinit(gpa);
const summary = try persistStoreOpen(gpa, &log, store);
log.commit() catch return error.Storage;
return summary;
}
// ---------------------------------------------------------------------------
// `zit pack` — repack the store into a compact new generation.
//
// On-disk storage is UNSPECIFIED by the RFCs (they fix the wire format and
// content-addressed naming, §4.1 L1206-1208), so pack is free to reorganize
// the log as long as every artifact's exact content and name survive and the
// §7.2 verbatim wire bytes are still servable. It re-ingests every artifact,
// in the source's offset order, into a FRESH log through `putArtifact` with a
// lineage base hint — a check-in against its parent manifest, a file blob
// against the same path's blob in the parent check-in (the commit path's own
// §4.1/§7.2 compaction heuristic, applied offline). Offset order plus the
// strictly-backward `delta_of` invariant means every base is already written
// before the delta that names it, so no worklist is needed (unlike
// `persistStore`, whose input is rid-ordered).
//
// The new log is written to a temporary file beside `.zit`, fsynced, and then
// atomically renamed over `.zit`: a crash anywhere before the rename leaves the
// original byte-identical (the temp is discarded next run). The `.zit-cache`
// invalidates itself on the tail change and rebuilds on the next read verb.
// ---------------------------------------------------------------------------
/// pack's core store-to-store transform lives in the `zit` module (pack.zig)
/// so the DST/fuzz harnesses drive the REAL pass in-process; the CLI wiring
/// below (temp file, atomic rename, stats rendering) stays here. These aliases
/// keep every call site and test reading `packLog`/`PackError`/`PackStats`.
const PackError = zit.pack.PackError;
const PackStats = zit.pack.PackStats;
const packLog = zit.pack.packLog;
/// The name of pack's transient new-generation file, beside `.zit`. A fixed
/// name (never a random one) so a temp left by a crashed pack is simply
/// overwritten by the next run rather than accumulating. Sourced from
/// `extras.zig`, which owns the literal for the single never-tracked predicate.
const pack_tmp_filename = extras.pack_tmp_filename;
/// `zit pack` — repack the current repository's store into a compact new
/// generation and atomically swap it in. Thin CLI wiring over `packLog`: parse
/// (no positional), discover the checkout root, open `.zit` read-only (which
/// whole-frame-CRC-verifies every frame — the integrity pre-flight) and capture
/// its tail key, stream the packed log into a temp file, fsync, re-check the
/// source tail (the no-repo-lock concurrency guard), then rename the temp over
/// `.zit`. On any failure the temp is deleted and the original store is intact.
pub fn runPack(
gpa: std.mem.Allocator,
io: std.Io,
arena: std.mem.Allocator,
verb_args: []const []const u8,
streams: cli.Streams,
) cli.Exit {
const help = switch (cli.parseNoPositionalArgs(verb_args)) {
.ok => |a| a.help,
.unknown_option => |tok| return cli.usageError(streams, "pack", .unknown_option, tok),
.extra_argument => |tok| return cli.usageError(streams, "pack", .extra_argument, tok),
};
if (help) {
streams.out.writeAll(pack_help) catch {};
return .ok;
}
const root = (cli.discoverRoot(arena, io) catch return stateOom(streams, "pack")) orelse {
streams.err.writeAll(not_a_repo_msg) catch {};
return .failure;
};
var root_dir = root.dir;
defer root_dir.close(io);
// Open the source read-only: `openReadOnly` scans and whole-frame-CRC-checks
// every frame, so a corrupt source aborts here (no temp written). The tail
// key pins the log we packed for the pre-rename concurrency re-check.
var src_file = root_dir.openFile(io, repo_filename, .{}) catch return repoReadFail(streams, "pack");
defer src_file.close(io);
const src_st = src_file.stat(io) catch return repoReadFail(streams, "pack");
var src_fs = storage.FileStorage.init(io, src_file, src_st.size);
var src = logstore.LogStore.openReadOnly(gpa, src_fs.storageFile()) catch |e| {
switch (e) {
error.OutOfMemory => streams.err.writeAll("zit pack: out of memory\n") catch {},
else => streams.err.print("zit pack: cannot read repository: {s}\n", .{reasonOf(e)}) catch {},
}
return .failure;
};
defer src.deinit(gpa);
const src_key = src.logKey() catch return packFail(streams, error.Corrupt);
// New-generation file: a readable temp beside `.zit` (createFileAtomic opens
// write-only, but pack reads its own delta bases back, so we roll our own
// readable temp and rename it). `truncate` overwrites a temp left by a prior
// crashed pack. Cleaned up on every path except the successful rename.
var tmp_file = root_dir.createFile(io, pack_tmp_filename, .{ .read = true, .truncate = true }) catch
return packFail(streams, error.Storage);
var renamed = false;
defer {
tmp_file.close(io);
if (!renamed) root_dir.deleteFile(io, pack_tmp_filename) catch {};
}
var dst_fs = storage.FileStorage.init(io, tmp_file, 0);
var dst = logstore.LogStore.open(gpa, dst_fs.storageFile(), src.hash_kind) catch |e| switch (e) {
error.OutOfMemory => return packFail(streams, error.OutOfMemory),
error.Storage => return packFail(streams, error.Storage),
error.Corrupt => return packFail(streams, error.Corrupt),
};
const stats = packLog(gpa, &src, &dst) catch |e| {
dst.deinit(gpa);
return packFail(streams, e);
};
dst.deinit(gpa); // `packLog` already fsynced; deinit only frees memory
// No repo lock exists: if the source log changed under us since we scanned
// it, abort rather than clobber a concurrent write with our stale snapshot.
// O(1) (one stat + one header read); the original stays intact.
if (!logstore.LogStore.validateTail(src_fs.storageFile(), src_key)) {
streams.err.writeAll(
"zit pack: repository changed during pack; aborted (original is intact)\n",
) catch {};
return .failure;
}
// The single durable transition: atomically replace `.zit` with the packed
// log. A crash before this leaves the original untouched; after it, the
// complete new log. There is no window where `.zit` is partial.
root_dir.rename(pack_tmp_filename, root_dir, repo_filename, io) catch
return packFail(streams, error.Storage);
renamed = true;
renderPackStats(streams.out, stats);
return .ok;
}
/// Render pack's failure `err` as `zit pack: <reason>` on stderr, exit 1. The
/// original store is always intact when this runs (every write went to the
/// temp, which the caller's `defer` deletes).
fn packFail(s: cli.Streams, err: PackError) cli.Exit {
s.err.print("zit pack: {s}\n", .{reasonOf(err)}) catch {};
return .failure;
}
/// One human line summarizing a repack. "already packed" when the new log did
/// not shrink (the store was already compact — pack still verified every frame,
/// its integrity-scrub value). Split from `runPack` so the exact wording is unit
/// tested without a filesystem.
fn renderPackStats(out: *std.Io.Writer, stats: PackStats) void {
const verb = if (stats.bytes_after < stats.bytes_before) "packed" else "already packed";
out.print("{s}: {d} artifacts, {d} re-deltaed, {d} -> {d} bytes\n", .{
verb, stats.artifacts, stats.deltas, stats.bytes_before, stats.bytes_after,
}) catch {};
}
/// Maps a transport-open failure (bad URL) to the `verb` diagnostic + exit 1.
/// Shared by clone and pull — the only difference is the verb label.
fn syncOpenFail(s: cli.Streams, verb: []const u8, url: []const u8, err: transport.TransportError) cli.Exit {
switch (err) {
error.InvalidUrl => s.err.print("zit {s}: '{s}' is not a valid URL\n", .{ verb, url }) catch {},
error.UnsupportedScheme => s.err.print("zit {s}: unsupported URL scheme in '{s}' (use http or https)\n", .{ verb, url }) catch {},
error.OutOfMemory => s.err.print("zit {s}: out of memory\n", .{verb}) catch {},
else => s.err.print("zit {s}: cannot open '{s}'\n", .{ verb, url }) catch {},
}
return .failure;
}
/// Maps a sync run (transport + driver) failure to the `verb` diagnostic + exit
/// 1. Exhaustive over the driver's own causes and the transport's, so the whole
/// closed error set is accounted for (no `else`). Shared by clone and pull.
fn syncRunFail(s: cli.Streams, verb: []const u8, err: sync_driver.RunError(*transport.Transport)) cli.Exit {
const reason: []const u8 = switch (err) {
error.OutOfMemory => "out of memory",
error.RoundLimitExceeded => "the peer did not converge within the round limit",
error.FatalProtocolError => "the peer sent a fatal protocol error",
error.NotImplemented => "the peer used an unsupported protocol feature",
error.RequestTooLarge => "the sync request exceeded the transport limit",
error.ReplyTruncated, error.ReplyMalformed => "the peer sent a malformed reply",
error.ReplyTooLarge => "the peer's reply exceeded the size limit",
error.RedirectLimit => "too many HTTP redirects",
error.InsecureRedirect => "the peer redirected HTTPS to plaintext HTTP",
error.RedirectNotAllowed => "the peer redirected on a transport that forbids it",
error.MissingLocation => "the peer sent a redirect with no location",
error.InvalidUrl => "a redirect target was not a valid URL",
error.UnsupportedScheme => "a redirect named an unsupported URL scheme",
error.ProtocolError => "the peer answered with an unexpected HTTP status",
error.ConnectionFailed => "cannot connect to the peer",
};
s.err.print("zit {s}: {s}\n", .{ verb, reason }) catch {};
return .failure;
}
/// The causes `resolveRemoteUrl` can surface: a store read failure, OOM, or the
/// repository having no URL to use (`NoRemote` — neither an explicit arg nor a
/// saved remote). Closed set so the caller switches exhaustively.
const RemoteResolveError = error{ OutOfMemory, Storage, NoRemote };
/// sync §4/§6 — resolve the peer URL for a sync-family verb: an explicit `arg`
/// (non-empty) wins; otherwise the saved remote (`config.remote_url`) is used;
/// a repository with neither is a usage error (`NoRemote`). The saved URL is
/// copied into `buf` so it outlives the read-only config frame. `buf` need only
/// live until `transport.Transport.open` (which dupes the URL). An out-of-spec
/// stored URL longer than the config-field cap is treated as none.
fn resolveRemoteUrl(
gpa: std.mem.Allocator,
io: std.Io,
dir: std.Io.Dir,
arg: []const u8,
buf: *[logstore.max_config_str]u8,
) RemoteResolveError![]const u8 {
if (arg.len > 0) return arg;
var file = dir.openFile(io, repo_filename, .{}) catch return error.Storage;
defer file.close(io);
const st = file.stat(io) catch return error.Storage;
var fs = storage.FileStorage.init(io, file, st.size);
var log = logstore.LogStore.openReadOnly(gpa, fs.storageFile()) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return error.Storage,
};
defer log.deinit(gpa);
var oc = (log.readConfig(gpa) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return error.Storage,
}) orelse return error.NoRemote;
defer oc.deinit(gpa);
const saved = oc.config.remote_url;
if (saved.len == 0 or saved.len > buf.len) return error.NoRemote;
@memcpy(buf[0..saved.len], saved);
return buf[0..saved.len];
}
/// sync §4/§6 — the usage error (exit 2) when a sync-family `verb` is given no
/// URL and the repository has no saved remote to fall back to. Names the fix.
/// Shared by pull, push, and sync so the wording lives in one place.
fn noRemoteError(s: cli.Streams, verb: []const u8) cli.Exit {
s.err.print("zit {s}: no repository URL given and no saved remote\n", .{verb}) catch {};
s.err.writeAll("set one with 'zit remote URL', or pass the URL explicitly\n") catch {};
return .usage;
}
// ---------------------------------------------------------------------------
// `zit pull URL` — incremental sync into an EXISTING repository. The sibling of
// clone: it reuses the same transport + `runSync` + persist machinery, but opens
// the cwd's `.zit` (never creating one), announces what it already holds so the
// peer sends only the missing artifacts (sync §3.2/§3.3 igot/gimme), and drives
// to §3.7 termination. Receive-only (§3.3 r24: pull obtains but never uploads).
// ---------------------------------------------------------------------------
/// `zit pull URL` — open the current repository, converge it toward the peer at
/// URL, and persist the newly received artifacts. Thin wiring over the same deep
/// modules clone uses: `transport` (HTTP/IO), `sync.Client` + `runSync` (sans-IO
/// §3.7 loop), and `LogStore` (persistence). With URL omitted the saved remote
/// (`zit remote`) is used; a repo with neither is a usage error. Anonymous only.
pub fn runPull(
gpa: std.mem.Allocator,
io: std.Io,
arena: std.mem.Allocator,
verb_args: []const []const u8,
streams: cli.Streams,
) cli.Exit {
const parsed = switch (cli.parsePullArgs(verb_args)) {
.ok => |p| p,
.unknown_option => |tok| return cli.usageError(streams, "pull", .unknown_option, tok),
.extra_argument => |tok| return cli.usageError(streams, "pull", .extra_argument, tok),
// An omitted URL is not an error: it falls back to the saved remote below.
.no_url => cli.PullArgs{ .url = "" },
};
if (parsed.help) {
streams.out.writeAll(pull_help) catch {};
return .ok;
}
// Locate the current repository before any network I/O: a pull outside a repo
// is a usage failure, not a wasted connection.
const root = (cli.discoverRoot(arena, io) catch return stateOom(streams, "pull")) orelse {
streams.err.writeAll(not_a_repo_msg) catch {};
return .failure;
};
var root_dir = root.dir;
defer root_dir.close(io);
// sync §4/§6 — the explicit URL wins; an omitted one falls back to the saved
// remote; a repo with neither is a usage error. `url_buf` outlives `open`.
var url_buf: [logstore.max_config_str]u8 = undefined;
const url = resolveRemoteUrl(gpa, io, root_dir, parsed.url, &url_buf) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "pull"),
error.Storage => return repoReadFail(streams, "pull"),
error.NoRemote => return noRemoteError(streams, "pull"),
};
var tr = transport.Transport.open(io, gpa, url) catch |e|
return syncOpenFail(streams, "pull", url, e);
defer tr.close();
// Snapshot what this repository already holds into the in-memory store: those
// entries become the client's `igot` announcement, so the peer sends only the
// artifacts this side lacks (sync §3.3 — the incremental of the exchange).
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
loadRepoHoldings(gpa, io, root_dir, &ms) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "pull"),
error.Storage => return repoReadFail(streams, "pull"),
};
const held_before = contentCount(ms.store());
// §9.1 — announce this repository's adopted project code in the pull card;
// a real peer refuses a PROJECTCODE mismatch, so the placeholder cannot be
// sent against fossil. The buffer outlives the run (the client borrows it).
var pc_buf: [zit.sync.max_projectcode_len]u8 = undefined;
const identity = repoSyncIdentity(gpa, io, root_dir, &pc_buf) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "pull"),
error.Storage => return repoReadFail(streams, "pull"),
};
const outcome = pullRun(gpa, &tr, &ms, identity) catch |e|
return syncRunFail(streams, "pull", e);
const persisted = persistStore(gpa, io, root_dir, ms.store()) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "pull"),
error.Storage => return repoReadFail(streams, "pull"),
error.Corrupt => return renderStorage(streams, root.abs, "the peer sent a malformed artifact name"),
};
const written = persisted.written;
// A pull only adds content to what was already held (§3.3 r24: it never drops
// an artifact), so the persisted count can never fall below the starting set.
std.debug.assert(written >= held_before);
const received = written - held_before;
// The peer named artifacts this run never obtained. Everything is persisted
// — a partial pull is still progress — but the exit status says so, because
// "exit 0 with nothing said" is indistinguishable from a converged sync and
// leaves a repository quietly out of date. This holds whether `received` is
// zero or not: the invariant is convergence, not motion.
if (outcome.unresolved > 0) {
streams.err.print(
"zit pull: the peer holds {d} artifacts this pull did not obtain\n",
.{outcome.unresolved},
) catch {};
return .failure;
}
// submanifests §7.1/§7.3 — the same reasoning one level down. A pull can
// converge against the peer (nothing left to gimme) while the tree closure
// this repository now holds is still open, because the peer's own closure
// was open: §7.1's "at most d+1 pull rounds" bounds a tree whose levels are
// ARRIVING, not one whose leaves nobody has. The phantoms the persist hook
// minted are that distinction, so report them rather than call it converged.
//
// §7.3/§9.4 — two counts, because they answer different questions and the
// draft names both: `submanifest_gaps` is the durable store's missing-hash
// count, `deferred_trees` is the session's DEPENDENT-keyed record ("enough
// to rediscover that manifest"). A run reporting the first without the
// second tells the operator something is missing but not what it breaks.
if (persisted.submanifest_gaps > 0 or outcome.deferred_trees > 0) {
// Two sentences, not one, because the two counts have different
// LIFETIMES and a single sentence reads as one fact. `submanifest_gaps`
// is a property of the durable store and survives a restart;
// `deferred_trees` is this SESSION's §9.4 record and is zero for a run
// that received no S card — including every re-run against an already
// open closure. Printed together they said "N references are missing; 0
// check-ins cannot be enumerated", which is exactly backwards.
if (persisted.submanifest_gaps > 0) {
streams.err.print(
"zit pull: {d} sub-manifest references are missing; the check-ins that reference them cannot be enumerated\n",
.{persisted.submanifest_gaps},
) catch {};
}
if (outcome.deferred_trees > 0) {
streams.err.print(
"zit pull: {d} check-ins received this session are deferred until they arrive\n",
.{outcome.deferred_trees},
) catch {};
}
return .failure;
}
if (received == 0) {
streams.out.writeAll("already up to date\n") catch {};
} else {
streams.out.print("received {d} artifacts in {d} rounds\n", .{ received, outcome.rounds }) catch {};
}
return .ok;
}
/// Drives a pull client (§3.2/§3.3) to convergence over `tr`, filling `ms` with
/// the artifacts the peer holds and this side lacked; returns the round count.
/// The client announces `ms`'s holdings via igot and requests the rest via
/// gimme (receive-only — §3.3 r24: pull obtains content but never uploads).
/// Sans-IO on this side of `tr.exchange`. Its closed error set is the driver's
/// own causes unioned with the transport's, so `runPull` switches it exhaustively.
fn pullRun(
gpa: std.mem.Allocator,
tr: *transport.Transport,
ms: *repostate.MemStore,
identity: RepoSyncIdentity,
) sync_driver.RunError(*transport.Transport)!PullOutcome {
var client = zit.sync.Client{
.state = zit.sync.initialState(.pull),
.operation = .pull,
.self_servercode = sync_servercode,
.project_code = identity.project_code,
.version = sync_client_version_info,
// submanifests §8.1/§7.2 — the durable indicator arms this run's send
// gate. A pull uploads nothing, so it is inert here; it is set anyway so
// no construction site has to remember which operations upload.
.seen_submanifest = identity.seen_submanifest,
.submanifests_enabled = identity.enable_submanifests,
};
defer client.deinit(gpa);
const stats = try sync_driver.runSync(gpa, &client, ms.store(), tr, max_sync_rounds);
return .{
.rounds = stats.rounds,
.unresolved = client.unresolvedPhantoms(ms.store()),
// submanifests §7.3/§9.4 — the DEPENDENT-keyed half of the verdict: how
// many received check-ins still have a recorded S-reference gap.
.deferred_trees = client.deferredTrees(),
};
}
/// What one pull run reports: the round count, and how many artifacts the peer
/// announced (§3.3 r24 — by igot, or as a member of a cluster it sent) that this
/// side still lacks when §3.7 reached its clean terminal. A nonzero `unresolved`
/// is a pull that stopped short of what it was told existed; the caller must not
/// report success for it, however well-formed every card on the wire was.
/// submanifests §7.3/§9.4 — `deferred_trees` is the §9.4 record read back: the
/// count of DEPENDENTS (check-ins) whose processing this session deferred on a
/// sub-manifest that never arrived. `unresolved` counts missing HASHES; a
/// dependent-keyed count is what §9.4 requires be recorded and is what an
/// operator can act on ("which check-in is broken", not "how many blobs").
const PullOutcome = struct { rounds: usize, unresolved: usize, deferred_trees: usize = 0 };
/// Opens `dir/.zit` read-only and snapshots its stored artifacts into `ms` — the
/// held/phantom split the pull client announces via `igot` (sync §3.2/§3.3).
/// Reuses `loadContentStore` (the very snapshot `serve` builds per request), so
/// the binary-log→wire-hex name encoding lives in exactly one place. The
/// read-only handle is closed before `persistStore` reopens the log read-write.
fn loadRepoHoldings(
gpa: std.mem.Allocator,
io: std.Io,
dir: std.Io.Dir,
ms: *repostate.MemStore,
) error{ OutOfMemory, Storage }!void {
var file = dir.openFile(io, repo_filename, .{}) catch return error.Storage;
defer file.close(io);
const st = file.stat(io) catch return error.Storage;
var fs = storage.FileStorage.init(io, file, st.size);
var log = logstore.LogStore.openReadOnly(gpa, fs.storageFile()) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return error.Storage,
};
defer log.deinit(gpa);
loadContentStore(gpa, &log, ms) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return error.Storage,
};
}
/// sync §9.1 — the repository's stored project code, copied into `buf`, or the
/// zero placeholder when the repo carries none (a repo predating the stored
/// code) or an out-of-spec-length one. `init` derives a project code and a
/// clone adopts the origin's (`storeClonedProjectCode`), so a real repo
/// announces its adopted code here; the pull/push client emits it verbatim as
/// the operation card's PROJECTCODE, and a real peer (fossil §9.1, ABNF
/// `pull-card = "pull" SP servercode SP projectcode`) refuses a mismatch — so a
/// zero placeholder would be rejected. Bounded by `max_projectcode_len`: a
/// longer stored field is out of spec and never truncated into the wire.
fn repoProjectCode(
gpa: std.mem.Allocator,
io: std.Io,
dir: std.Io.Dir,
buf: *[zit.sync.max_projectcode_len]u8,
) error{ OutOfMemory, Storage }![]const u8 {
return (try repoSyncIdentity(gpa, io, dir, buf)).project_code;
}
/// The durable config facts every sync-family run needs before it can build
/// a peer: §9.1's project code and draft-fossil-submanifests-00 §8.1's
/// seen-submanifest indicator and emission setting. Read together because they
/// come out of the same config frame — a second open to fetch another field
/// would be one more way for them to disagree about which frame they came from.
const RepoSyncIdentity = struct {
/// §9.1 — the adopted project code, or the zero placeholder.
project_code: []const u8,
/// submanifests §8.1 — has this repository ever stored a §7.2-gated
/// artifact? False for every repository predating this draft, and the
/// second thing that arms the §7.2 send gate (see `Server.seen_submanifest`).
seen_submanifest: bool,
/// submanifests §8.1 — the emission setting. The client twin of
/// `serveContext`'s server-side refresh: it is the OTHER half of the §7.2
/// arming disjunction (`Client.mayHoldSubmanifests`). Today it is
/// belt-and-braces, because any repository actually holding a gated artifact
/// also has `seen_submanifest` set — the two are written in one `log.commit`
/// with the first tree. It is threaded anyway so the field is never read
/// stale: leaving it at its `false` default would silently disarm the gate
/// for any future surface that commits mid-session, where the setting is on
/// but the indicator has not been written yet.
enable_submanifests: bool,
};
fn repoSyncIdentity(
gpa: std.mem.Allocator,
io: std.Io,
dir: std.Io.Dir,
buf: *[zit.sync.max_projectcode_len]u8,
) error{ OutOfMemory, Storage }!RepoSyncIdentity {
var file = dir.openFile(io, repo_filename, .{}) catch return error.Storage;
defer file.close(io);
const st = file.stat(io) catch return error.Storage;
var fs = storage.FileStorage.init(io, file, st.size);
var log = logstore.LogStore.openReadOnly(gpa, fs.storageFile()) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return error.Storage,
};
defer log.deinit(gpa);
var oc = (log.readConfig(gpa) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return error.Storage,
}) orelse return .{
.project_code = sync_projectcode,
.seen_submanifest = false,
.enable_submanifests = zit.sync.submanifest_emission_default,
};
defer oc.deinit(gpa);
const seen = oc.config.seen_submanifest;
const enabled = oc.config.enable_submanifests;
const code = oc.config.project_code;
if (code.len == 0 or code.len > buf.len) return .{
.project_code = sync_projectcode,
.seen_submanifest = seen,
.enable_submanifests = enabled,
};
@memcpy(buf[0..code.len], code);
return .{
.project_code = buf[0..code.len],
.seen_submanifest = seen,
.enable_submanifests = enabled,
};
}
/// The number of content-bearing (non-phantom) artifacts in `store` — the held
/// set (§3.3 r24). A pull measures its yield as the held count after convergence
/// minus the count before, so "already up to date" is an exact zero, never a guess.
fn contentCount(store: repostate.ArtifactStore) usize {
var n: usize = 0;
const total = store.count();
var rid: repostate.Rid = 1;
while (rid <= total) : (rid += 1) {
if (store.content(rid) != null) n += 1;
}
return n;
}
// ---------------------------------------------------------------------------
// `zit push URL` — the mirror of pull: UPLOAD this repository's artifacts into an
// EXISTING served repo. It opens the cwd's `.zit`, announces what it holds so the
// peer requests only what it lacks (sync §3.2/§3.3 igot/gimme), and uploads the
// missing artifacts as `file` cards, driving to §3.7 termination. The peer's
// `zit serve` persists what it receives; nothing changes on this side (§3.3 r24).
// Anonymous only.
// ---------------------------------------------------------------------------
/// What one push run reports: the number of artifacts uploaded and the round
/// count it took. `sent` reads the client's own upload counter, so it is the
/// exact count of `file` cards emitted — never an estimate.
/// submanifests §7.2 — `refused` counts artifacts the §7.2 gate declined to
/// upload because the peer's announced version predates sub-manifest support.
/// Carried alongside `sent` so a run that withheld an artifact can never be
/// reported as a success: the draft rules out silent omission.
const PushOutcome = struct { rounds: usize, sent: usize, refused: usize = 0 };
/// `zit push URL` — open the current repository, announce its holdings to the
/// peer at URL, and upload every artifact the peer lacks until §3.7 termination.
/// Thin wiring over the same deep modules pull uses: `transport` (HTTP/IO),
/// `sync.Client` + `runSync` (sans-IO §3.7 loop). With URL omitted the saved
/// remote (`zit remote`) is used; a repo with neither is a usage error.
pub fn runPush(
gpa: std.mem.Allocator,
io: std.Io,
arena: std.mem.Allocator,
verb_args: []const []const u8,
streams: cli.Streams,
amb: cli.Ambient,
) cli.Exit {
const parsed = switch (cli.parsePushArgs(verb_args)) {
.ok => |p| p,
.unknown_option => |tok| return cli.usageError(streams, "push", .unknown_option, tok),
.extra_argument => |tok| return cli.usageError(streams, "push", .extra_argument, tok),
// An omitted URL is not an error: it falls back to the saved remote below.
.no_url => cli.PushArgs{ .url = "" },
};
if (parsed.help) {
streams.out.writeAll(push_help) catch {};
return .ok;
}
// sync §6.3 — a login name is a whitespace-free token (the shared-secret
// derivation embeds it between "/" separators). Reject a malformed --user
// as a usage failure rather than reaching the derivation's assertion.
if (parsed.user) |u| {
if (u.len > 0 and zit.sync.containsFossilWhitespace(u)) {
streams.err.writeAll("zit push: --user must not contain whitespace\n") catch {};
return .failure;
}
}
// Locate the current repository before any network I/O: a push outside a repo
// is a usage failure, not a wasted connection.
const root = (cli.discoverRoot(arena, io) catch return stateOom(streams, "push")) orelse {
streams.err.writeAll(not_a_repo_msg) catch {};
return .failure;
};
var root_dir = root.dir;
defer root_dir.close(io);
// sync §4/§6 — the explicit URL wins; an omitted one falls back to the saved
// remote; a repo with neither is a usage error. `url_buf` outlives `open`.
var url_buf: [logstore.max_config_str]u8 = undefined;
const url = resolveRemoteUrl(gpa, io, root_dir, parsed.url, &url_buf) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "push"),
error.Storage => return repoReadFail(streams, "push"),
error.NoRemote => return noRemoteError(streams, "push"),
};
var tr = transport.Transport.open(io, gpa, url) catch |e|
return syncOpenFail(streams, "push", url, e);
defer tr.close();
// Snapshot what this repository holds: those entries become the client's igot
// announcement, so the peer requests (and this side uploads) only the artifacts
// the peer lacks (sync §3.2/§3.3 — the incremental of the exchange).
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
loadRepoHoldings(gpa, io, root_dir, &ms) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "push"),
error.Storage => return repoReadFail(streams, "push"),
};
// §9.1 — announce this repository's adopted project code in the push card
// (same identity gate as pull; a real peer refuses a mismatch). The buffer
// outlives the run — the client borrows it.
var pc_buf: [zit.sync.max_projectcode_len]u8 = undefined;
const identity = repoSyncIdentity(gpa, io, root_dir, &pc_buf) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "push"),
error.Storage => return repoReadFail(streams, "push"),
};
// sync §6 — resolve the login credential (null for an anonymous push). The
// password plaintext is confined to `pw_buf`/argv/env; only the derived
// 40-hex secret (§6.3) is signed with and — on success — persisted.
var pw_buf: [max_password_len]u8 = undefined;
// Defense-in-depth: once the 40-hex secret is derived the plaintext in
// pw_buf is spent, so scrub it on every exit from this scope — success and
// error alike — rather than leave it resident until the stack is reused.
defer std.crypto.secureZero(u8, &pw_buf);
const resolved = resolvePushCredential(gpa, io, root_dir, streams, identity.project_code, parsed, amb, &pw_buf) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "push"),
error.Storage => return repoReadFail(streams, "push"),
error.NoPassword => {
streams.err.writeAll("zit push: no password supplied for --user\n") catch {};
return .failure;
},
};
const outcome = pushRun(gpa, &tr, &ms, identity, if (resolved) |r| r.credential else null) catch |e|
return syncRunFail(streams, "push", e);
// DECISION-3 — persist the derived secret (never the plaintext) at 0600 only
// after the peer accepted the authenticated push, and only when it was
// freshly derived: a reused stored secret is already durable.
if (resolved) |r| {
if (r.persist) {
storePushCredential(gpa, io, root_dir, r.credential) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "push"),
else => return repoReadFail(streams, "push"),
};
}
}
// submanifests §7.2 — "the sender MUST refuse to transmit that artifact,
// with an explicit error -- never by converting, downgrading, or silently
// omitting it." A client cannot put an `error` card in a REQUEST (it is
// outside the §3.9 strict-accept request vocabulary), so the explicit error
// is this: the operator is told exactly what was withheld and why, and the
// run fails. Reported before the success line, because there was none.
if (outcome.refused > 0) {
streams.err.print(
"zit push: {d} artifacts were withheld: {s}\n",
.{ outcome.refused, zit.sync.submanifest_peer_version_error_text },
) catch {};
return .failure;
}
if (outcome.sent == 0) {
streams.out.writeAll("nothing to push\n") catch {};
} else {
streams.out.print("sent {d} artifacts in {d} rounds\n", .{ outcome.sent, outcome.rounds }) catch {};
}
return .ok;
}
/// The upper bound on an interactively-typed or `$ZIT_PASSWORD` password. A
/// login password far shorter than this in practice; the fixed bound keeps the
/// no-echo read on the stack with no unbounded allocation (TigerStyle).
const max_password_len: usize = 1024;
/// sync §6.3 — a resolved push credential and whether its secret was freshly
/// derived (so it should be persisted after a successful push).
const ResolvedCredential = struct {
credential: zit.sync.Credential,
persist: bool,
};
/// The causes `resolvePushCredential` can surface: store read failure, OOM, or
/// a `--user` with no obtainable password (`NoPassword`). Closed set so the
/// caller switches exhaustively.
const ResolveError = error{ OutOfMemory, Storage, NoPassword };
/// sync §6.3/DECISION-3 — turn `push`'s `--user`/`--password` into a signing
/// credential. A null (or empty) `--user` is an anonymous push (null result).
/// Otherwise the 40-hex secret is resolved, in order, from: a stored credential
/// for this exact login (reused verbatim, already durable), `--password`,
/// `$ZIT_PASSWORD`, or a no-echo terminal prompt. A freshly derived secret is
/// flagged `persist`; the plaintext lives only in `pw_buf`/argv/env and is
/// never returned or stored.
fn resolvePushCredential(
gpa: std.mem.Allocator,
io: std.Io,
dir: std.Io.Dir,
streams: cli.Streams,
project_code: []const u8,
parsed: cli.PushArgs,
amb: cli.Ambient,
pw_buf: *[max_password_len]u8,
) ResolveError!?ResolvedCredential {
const user = parsed.user orelse return null;
if (user.len == 0) return null; // an empty --user is an anonymous push
std.debug.assert(!zit.sync.containsFossilWhitespace(user)); // pre-validated in runPush
// A stored secret for this exact login is reused as-is: no re-derive, no
// prompt, no re-persist. Only consulted when no explicit password is given.
if (parsed.password == null and amb.push_password == null) {
if (try storedSecretFor(gpa, io, dir, user)) |sec| {
return .{ .credential = .{ .user = user, .secret_hex = sec }, .persist = false };
}
}
// Otherwise derive from a plaintext password. Precedence: --password, then
// $ZIT_PASSWORD, then an interactive no-echo prompt.
const password: []const u8 = parsed.password orelse amb.push_password orelse
(readPasswordNoEcho(streams, user, pw_buf) orelse return error.NoPassword);
const secret = zit.sync.deriveSharedSecret(project_code, user, password);
return .{ .credential = .{ .user = user, .secret_hex = secret }, .persist = true };
}
/// DECISION-3 — the 40-hex secret stored for `user` in this repo's config, or
/// null when none is stored or it belongs to a different login. Read-only; the
/// returned value is an owned copy (the config frame is freed here).
fn storedSecretFor(
gpa: std.mem.Allocator,
io: std.Io,
dir: std.Io.Dir,
user: []const u8,
) error{ OutOfMemory, Storage }!?[40]u8 {
var file = dir.openFile(io, repo_filename, .{}) catch return error.Storage;
defer file.close(io);
const st = file.stat(io) catch return error.Storage;
var fs = storage.FileStorage.init(io, file, st.size);
var log = logstore.LogStore.openReadOnly(gpa, fs.storageFile()) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return error.Storage,
};
defer log.deinit(gpa);
var oc = (log.readConfig(gpa) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return error.Storage,
}) orelse return null;
defer oc.deinit(gpa);
if (oc.config.login_secret_hex.len != 40) return null;
if (!std.mem.eql(u8, oc.config.login_user, user)) return null;
var out: [40]u8 = undefined;
@memcpy(&out, oc.config.login_secret_hex[0..40]);
return out;
}
/// DECISION-3 — persist the derived login credential into the repo config,
/// preserving every other field, then tighten the repository file to 0600 so
/// the stored secret is owner-only at rest. The permission tightening is
/// best-effort: the secret is already durable, so a filesystem that refuses
/// chmod does not fail the push (perms + FDE are the boundary, not the sole
/// guarantee — per DECISION-3). NEVER writes the plaintext password.
fn storePushCredential(
gpa: std.mem.Allocator,
io: std.Io,
dir: std.Io.Dir,
credential: zit.sync.Credential,
) PersistError!void {
std.debug.assert(credential.secret_hex.len == 40);
// The field-preserving RMW writes the credential without disturbing the
// project code or a saved remote (a bare `putConfig` here would drop them).
try updateConfig(gpa, io, dir, .{ .login = credential });
// Owner-only now that the file carries a credential (best-effort; see above).
var file = dir.openFile(io, repo_filename, .{ .mode = .read_write }) catch return;
defer file.close(io);
file.setPermissions(io, File.Permissions.fromMode(0o600)) catch {};
}
/// CLI/IO seam — prompt on stderr and read one password line from the terminal
/// with echo disabled, returning the trimmed line in `buf` (never its newline),
/// or null on a read failure or an empty line. Echo is restored on every exit
/// path; when stdin is not a terminal (`tcgetattr` fails) the line is read with
/// no termios toggle. Not exercised by the automated gates (they pass
/// `--password`/`$ZIT_PASSWORD`) — this is the interactive path.
fn readPasswordNoEcho(
streams: cli.Streams,
user: []const u8,
buf: *[max_password_len]u8,
) ?[]const u8 {
streams.err.print("Password for {s}: ", .{user}) catch {};
streams.err.flush() catch {};
const fd = std.posix.STDIN_FILENO;
const restore: ?std.posix.termios = blk: {
var t = std.posix.tcgetattr(fd) catch break :blk null;
const saved = t;
t.lflag.ECHO = false;
std.posix.tcsetattr(fd, .FLUSH, t) catch break :blk null;
break :blk saved;
};
defer if (restore) |saved| {
std.posix.tcsetattr(fd, .FLUSH, saved) catch {};
streams.err.writeAll("\n") catch {};
streams.err.flush() catch {};
};
const n = std.posix.read(fd, buf) catch return null;
if (n == 0) return null;
var line = buf[0..n];
// Trim a single trailing CRLF/LF; interior bytes (including spaces) are kept.
if (line.len > 0 and line[line.len - 1] == '\n') line = line[0 .. line.len - 1];
if (line.len > 0 and line[line.len - 1] == '\r') line = line[0 .. line.len - 1];
if (line.len == 0) return null;
return line;
}
/// Drives a push client (§3.2/§3.3) to convergence over `tr`, uploading the
/// artifacts the peer lacks; returns the upload count and round count. The client
/// announces `ms`'s holdings via igot and uploads what the peer gimme's (§3.3
/// r24: push sends content but never obtains it). Sans-IO on this side of
/// `tr.exchange`. Its closed error set is the driver's own causes unioned with
/// the transport's, so `runPush` switches it exhaustively.
fn pushRun(
gpa: std.mem.Allocator,
tr: *transport.Transport,
ms: *repostate.MemStore,
identity: RepoSyncIdentity,
credential: ?zit.sync.Credential,
) sync_driver.RunError(*transport.Transport)!PushOutcome {
var client = zit.sync.Client{
.state = zit.sync.initialState(.push),
.operation = .push,
.self_servercode = sync_servercode,
.project_code = identity.project_code,
.version = sync_client_version_info,
// submanifests §8.1/§7.2 — the durable indicator arms the §7.2 upload
// gate for a repository that HOLDS gated artifacts, whatever §8.1's
// emission setting says. Without it every zit push ships its trees to
// whatever is at the other end.
.seen_submanifest = identity.seen_submanifest,
.submanifests_enabled = identity.enable_submanifests,
// sync §6 — a named credential makes `buildRequest` emit the §6.1 login
// card; null leaves the push anonymous (unchanged behavior).
.credential = credential,
};
defer client.deinit(gpa);
const stats = try sync_driver.runSync(gpa, &client, ms.store(), tr, max_sync_rounds);
// Read the upload counter before `deinit` runs (the return expression is
// evaluated first): the exact number of `file` cards this session emitted.
return .{ .rounds = stats.rounds, .sent = client.content_sent, .refused = client.submanifests_refused };
}
// ---------------------------------------------------------------------------
// `zit sync [URL]` — BIDIRECTIONAL convergence with a peer in ONE session.
// Unlike pull-then-push, a single `operation=.sync` client announces igot,
// requests gimme, AND uploads on one §3.7 exchange (§3.2 round structure: a sync
// request carries both push and pull each round) — not a third copy of the driver
// loop over two transport sessions.
// Opens the cwd's `.zit`, resolves the URL (arg → saved remote), may
// authenticate (sync uploads), persists what it receives like pull, and reports
// both directions. A saved `zit remote` lets the URL be omitted.
// ---------------------------------------------------------------------------
/// What one sync run reports: uploads (`sent`) and the round count. Received is
/// measured by the caller as the persisted-count growth (as pull does), so it is
/// not carried here.
/// submanifests §7.2 — see `PushOutcome.refused`.
const SyncOutcome = struct { rounds: usize, sent: usize, refused: usize = 0 };
pub fn runSync(
gpa: std.mem.Allocator,
io: std.Io,
arena: std.mem.Allocator,
verb_args: []const []const u8,
streams: cli.Streams,
amb: cli.Ambient,
) cli.Exit {
const parsed = switch (cli.parseSyncArgs(verb_args)) {
.ok => |p| p,
.unknown_option => |tok| return cli.usageError(streams, "sync", .unknown_option, tok),
.extra_argument => |tok| return cli.usageError(streams, "sync", .extra_argument, tok),
};
if (parsed.help) {
streams.out.writeAll(sync_help) catch {};
return .ok;
}
// sync §6.3 — a login name is whitespace-free (mirrors push). Reject a
// malformed --user as usage before the secret derivation's assertion.
if (parsed.user) |u| {
if (u.len > 0 and zit.sync.containsFossilWhitespace(u)) {
streams.err.writeAll("zit sync: --user must not contain whitespace\n") catch {};
return .failure;
}
}
// Locate the current repository before any network I/O.
const root = (cli.discoverRoot(arena, io) catch return stateOom(streams, "sync")) orelse {
streams.err.writeAll(not_a_repo_msg) catch {};
return .failure;
};
var root_dir = root.dir;
defer root_dir.close(io);
// sync §4/§6 — the explicit URL wins; an omitted one falls back to the saved
// remote; a repo with neither is a usage error. `url_buf` outlives `open`.
var url_buf: [logstore.max_config_str]u8 = undefined;
const url = resolveRemoteUrl(gpa, io, root_dir, parsed.url, &url_buf) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "sync"),
error.Storage => return repoReadFail(streams, "sync"),
error.NoRemote => return noRemoteError(streams, "sync"),
};
var tr = transport.Transport.open(io, gpa, url) catch |e|
return syncOpenFail(streams, "sync", url, e);
defer tr.close();
// Snapshot holdings: the igot announcement and the baseline for the received
// count. A sync both uploads what the peer lacks and receives what this side
// lacks — the held set can only grow (§3.3 storage only adds content).
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
loadRepoHoldings(gpa, io, root_dir, &ms) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "sync"),
error.Storage => return repoReadFail(streams, "sync"),
};
const held_before = contentCount(ms.store());
// §9.1 — announce this repository's adopted project code (a real peer refuses
// a mismatch). The buffer outlives the run — the client borrows it.
var pc_buf: [zit.sync.max_projectcode_len]u8 = undefined;
const identity = repoSyncIdentity(gpa, io, root_dir, &pc_buf) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "sync"),
error.Storage => return repoReadFail(streams, "sync"),
};
// sync §6 — resolve the login credential (null for anonymous). A sync
// uploads, so it may authenticate exactly as push does; the plaintext lives
// only in `pw_buf`/argv/env and is scrubbed on every exit from this scope.
var pw_buf: [max_password_len]u8 = undefined;
defer std.crypto.secureZero(u8, &pw_buf);
const resolved = resolvePushCredential(gpa, io, root_dir, streams, identity.project_code, .{ .url = url, .user = parsed.user, .password = parsed.password }, amb, &pw_buf) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "sync"),
error.Storage => return repoReadFail(streams, "sync"),
error.NoPassword => {
streams.err.writeAll("zit sync: no password supplied for --user\n") catch {};
return .failure;
},
};
const outcome = syncRun(gpa, &tr, &ms, identity, if (resolved) |r| r.credential else null) catch |e|
return syncRunFail(streams, "sync", e);
const persisted = persistStore(gpa, io, root_dir, ms.store()) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "sync"),
error.Storage => return repoReadFail(streams, "sync"),
error.Corrupt => return renderStorage(streams, root.abs, "the peer sent a malformed artifact name"),
};
const written = persisted.written;
std.debug.assert(written >= held_before);
const received = written - held_before;
// DECISION-3 — persist a freshly derived secret only after the peer accepted
// the authenticated session, and only when it was freshly derived (mirrors push).
if (resolved) |r| {
if (r.persist) {
storePushCredential(gpa, io, root_dir, r.credential) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "sync"),
else => return repoReadFail(streams, "sync"),
};
}
}
// submanifests §7.2 — "the sender MUST refuse to transmit that artifact,
// with an explicit error -- never by converting, downgrading, or silently
// omitting it." A client cannot put an `error` card in a REQUEST (it is
// outside the §3.9 strict-accept request vocabulary), so the explicit error
// is this: the operator is told exactly what was withheld and why, and the
// run fails. Reported before the success line, because there was none.
if (outcome.refused > 0) {
streams.err.print(
"zit sync: {d} artifacts were withheld: {s}\n",
.{ outcome.refused, zit.sync.submanifest_peer_version_error_text },
) catch {};
return .failure;
}
// submanifests §7.1/§7.3 — see `runPull`: convergence against the peer is
// not the same fact as a complete tree closure.
if (persisted.submanifest_gaps > 0) {
streams.err.print(
"zit sync: {d} sub-manifest references are missing; some check-ins cannot be enumerated\n",
.{persisted.submanifest_gaps},
) catch {};
return .failure;
}
if (outcome.sent == 0 and received == 0) {
streams.out.writeAll("already in sync\n") catch {};
} else {
streams.out.print("sent {d}, received {d} in {d} rounds\n", .{ outcome.sent, received, outcome.rounds }) catch {};
}
return .ok;
}
/// Drives a bidirectional sync client (§3.2 round structure — both directions on
/// ONE §3.7 session) to convergence over `tr`; returns the upload count and round
/// count.
/// The received count is the caller's persisted-count growth. Sans-IO on this
/// side of `tr.exchange`; its closed error set is switched exhaustively by
/// `runSync`. The sibling of `pullRun`/`pushRun`, differing only in the
/// operation tag (`.sync`, which drives igot + gimme + upload on one session).
fn syncRun(
gpa: std.mem.Allocator,
tr: *transport.Transport,
ms: *repostate.MemStore,
identity: RepoSyncIdentity,
credential: ?zit.sync.Credential,
) sync_driver.RunError(*transport.Transport)!SyncOutcome {
var client = zit.sync.Client{
.state = zit.sync.initialState(.sync),
.operation = .sync,
.self_servercode = sync_servercode,
.project_code = identity.project_code,
.version = sync_client_version_info,
// submanifests §8.1/§7.2 — see `pushRun`: a sync uploads too.
.seen_submanifest = identity.seen_submanifest,
.submanifests_enabled = identity.enable_submanifests,
// sync §6 — a named credential emits the §6.1 login card; null is anonymous.
.credential = credential,
};
defer client.deinit(gpa);
const stats = try sync_driver.runSync(gpa, &client, ms.store(), tr, max_sync_rounds);
return .{ .rounds = stats.rounds, .sent = client.content_sent, .refused = client.submanifests_refused };
}
// ---------------------------------------------------------------------------
// `zit remote [URL | --unset]` — read, set, or clear the single default peer URL
// (sync §4/§6). Bare prints the saved remote; a URL saves it; --unset clears it.
// A local, non-synced config value (`config.remote_url`) — anonymous only, no
// credential in the URL — that lets pull/push/sync omit their URL argument. Not
// a git-style named-remote registry: one value, matching the single-value config.
// ---------------------------------------------------------------------------
pub fn runRemote(
gpa: std.mem.Allocator,
io: std.Io,
arena: std.mem.Allocator,
verb_args: []const []const u8,
streams: cli.Streams,
) cli.Exit {
const parsed = switch (cli.parseRemoteArgs(verb_args)) {
.ok => |p| p,
.unknown_option => |tok| return cli.usageError(streams, "remote", .unknown_option, tok),
.extra_argument => |tok| return cli.usageError(streams, "remote", .extra_argument, tok),
};
if (parsed.help) {
streams.out.writeAll(remote_help) catch {};
return .ok;
}
const root = (cli.discoverRoot(arena, io) catch return stateOom(streams, "remote")) orelse {
streams.err.writeAll(not_a_repo_msg) catch {};
return .failure;
};
var root_dir = root.dir;
defer root_dir.close(io);
if (parsed.unset) {
// Overlay remote_url with empty — clears the saved value, every other
// config field preserved by the field-preserving RMW.
updateConfig(gpa, io, root_dir, .{ .remote_url = "" }) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "remote"),
else => return repoReadFail(streams, "remote"),
};
streams.out.writeAll("remote cleared\n") catch {};
return .ok;
}
if (parsed.url.len > 0) {
// A URL over the config-field cap cannot be stored (the putConfig assert);
// reject it as usage rather than trip the assertion.
if (parsed.url.len > logstore.max_config_str) {
streams.err.print("zit remote: URL too long (max {d} bytes)\n", .{logstore.max_config_str}) catch {};
return .failure;
}
updateConfig(gpa, io, root_dir, .{ .remote_url = parsed.url }) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "remote"),
else => return repoReadFail(streams, "remote"),
};
streams.out.print("remote set to {s}\n", .{parsed.url}) catch {};
return .ok;
}
// Bare `zit remote` — print the saved remote (reusing resolveRemoteUrl's read
// with an empty arg; NoRemote means none is saved).
var url_buf: [logstore.max_config_str]u8 = undefined;
const url = resolveRemoteUrl(gpa, io, root_dir, "", &url_buf) catch |e| switch (e) {
error.OutOfMemory => return stateOom(streams, "remote"),
error.Storage => return repoReadFail(streams, "remote"),
error.NoRemote => {
streams.out.writeAll("no remote set\n") catch {};
return .ok;
},
};
streams.out.print("{s}\n", .{url}) catch {};
return .ok;
}
test "serve §3.3: a pushed file card is persisted durably into the served repository's log" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
// A fresh, empty on-disk repository (header only) — the "server" a peer pushes
// new content into. Built through the real LogStore, not seeded from fossil.
const hash_kind: u8 = @intFromEnum(zit.artifact.defaultPolicy(false, false));
{
var f = try tmp.dir.createFile(io, repo_filename, .{ .read = true, .exclusive = true });
defer f.close(io);
var fs = storage.FileStorage.init(io, f, 0);
var seed = try logstore.LogStore.open(gpa, fs.storageFile(), hash_kind);
defer seed.deinit(gpa);
try seed.commit();
}
// A pushed full-content file card for a 64-hex (SHA3-256 width) artifact the
// server lacks (sync §7.1: `file HASH SIZE\n` + payload; §4.1 zlib container).
// The name is the artifact's real SHA3-256 (computed by our own hasher, not
// fossil) so the server's §7.7 commit-hook recompute accepts it.
const push_body = "pushed artifact bytes\n"; // 22 bytes
var name_buf: [64]u8 = undefined;
const push_hash = zit.artifact.artifactName(.sha3_256, push_body, &name_buf);
const request_plain = try std.fmt.allocPrint(gpa, "pragma client-version 20000\nfile {s} {d}\n{s}", .{ push_hash, push_body.len, push_body });
defer gpa.free(request_plain);
const request = try zit.container.encode(gpa, request_plain, .default);
defer gpa.free(request);
var digest_buf: [digest_len]u8 = undefined;
const push_digest = decodeWireName(push_hash, &digest_buf).?;
const server = zit.sync.Server{ .version = serve_mod.server_version };
// Answer the push twice through the real `serve.Responder` seam over a SINGLE
// open handle (§3.8): the second answer re-reads the current on-disk log from
// the same handle and sees the first push's frame, so it is an idempotent
// re-push (content is addressed by hash — no dup frame). A stale startup
// snapshot would instead re-append the artifact; this handle is never reopened.
{
var file = try tmp.dir.openFile(io, repo_filename, .{});
defer file.close(io);
var ctx = ServeCtx{ .io = io, .root_dir = tmp.dir, .file = file, .server = &server };
defer ctx.dropSnapshot();
for (0..2) |_| {
const reply = try serveAnswer(&ctx, gpa, request);
gpa.free(reply.body);
}
}
// THE PROOF: re-open the log from disk in a fresh handle; the pushed artifact
// is present with its exact content — durably stored, not discarded — and
// appears exactly once (the idempotent re-push added no second frame).
var file = try tmp.dir.openFile(io, repo_filename, .{});
defer file.close(io);
const st = try file.stat(io);
var fs = storage.FileStorage.init(io, file, st.size);
var log = try logstore.LogStore.openReadOnly(gpa, fs.storageFile());
defer log.deinit(gpa);
const rid = log.lookup(push_digest) orelse return error.PushedArtifactMissing;
const stored = (try log.content(gpa, rid)).?;
defer gpa.free(stored);
try std.testing.expectEqualStrings(push_body, stored);
const rids = try log.ridsInPersistedOrder(gpa);
defer gpa.free(rids);
var artifacts: usize = 0;
for (rids) |r| {
if (log.kindOf(r) != .config) artifacts += 1;
}
try std.testing.expectEqual(@as(usize, 1), artifacts);
}
// --- D4: client verbatim retention + the §7.7 persist hook ------------------
/// Create a fresh, empty (header-only, committed) `.zit` in `dir` so a
/// `persistStore` test can persist into it, exactly like a just-cloned repo.
fn d4SeedEmptyRepo(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir) !void {
const hash_kind: u8 = @intFromEnum(zit.artifact.defaultPolicy(false, false));
var f = try dir.createFile(io, repo_filename, .{ .read = true, .exclusive = true });
defer f.close(io);
var fs = storage.FileStorage.init(io, f, 0);
var seed = try logstore.LogStore.open(gpa, fs.storageFile(), hash_kind);
defer seed.deinit(gpa);
try seed.commit();
}
/// Reopen the persisted `.zit` in `dir` for reading. `fs_out` is a caller-owned
/// `FileStorage` that MUST outlive the returned `LogStore` (its `storageFile()`
/// vtable borrows `fs_out`); the caller closes `fs_out.file` and deinits `log`.
fn d4OpenLog(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir, fs_out: *storage.FileStorage) !logstore.LogStore {
const file = try dir.openFile(io, repo_filename, .{ .mode = .read_write });
const st = try file.stat(io);
fs_out.* = storage.FileStorage.init(io, file, st.size);
return try logstore.LogStore.open(gpa, fs_out.storageFile(), 0);
}
fn d4SeedConfiguredRepo(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir) !void {
const hash_kind: u8 = @intFromEnum(zit.artifact.defaultPolicy(false, false));
var f = try dir.createFile(io, repo_filename, .{ .read = true, .exclusive = true });
defer f.close(io);
var fs = storage.FileStorage.init(io, f, 0);
var seed = try logstore.LogStore.open(gpa, fs.storageFile(), hash_kind);
defer seed.deinit(gpa);
try seed.putConfig(.{
.hash_policy = 9,
.default_branch = "stable",
.project_id = "project",
.project_code = "init-code",
.login_user = "operator",
.login_secret_hex = "0" ** 40,
.remote_url = "https://example.test/remote",
.seen_submanifest = true,
.enable_submanifests = false,
.submanifest_k = 3,
.submanifest_auto_retune = false,
.submanifest_settled_cost_shape = [_]u8{0x42} ** 32,
});
try seed.commit();
}
test "sync clone persistence: open-log artifacts and one combined config overlay preserve the wrapper result" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var wrapped_dir = std.testing.tmpDir(.{});
defer wrapped_dir.cleanup();
var open_dir = std.testing.tmpDir(.{});
defer open_dir.cleanup();
try d4SeedConfiguredRepo(gpa, io, wrapped_dir.dir);
try d4SeedConfiguredRepo(gpa, io, open_dir.dir);
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
const content = "one content-bearing clone artifact";
var name_buf: [64]u8 = undefined;
const name = zit.artifact.artifactName(.sha3_256, content, &name_buf);
const container = try zit.container.encode(gpa, content, .default);
defer gpa.free(container);
const rid = try ms.store().put(name, content);
try ms.store().retain(rid, .{ .full = container });
const patch: ConfigPatch = .{
.project_code = "origin-code",
.enable_submanifests = true,
.submanifest_k = 5,
.submanifest_auto_retune = true,
};
const wrapped = try persistStore(gpa, io, wrapped_dir.dir, ms.store());
try updateConfig(gpa, io, wrapped_dir.dir, patch);
var open_fs: storage.FileStorage = undefined;
var open_log = try d4OpenLog(gpa, io, open_dir.dir, &open_fs);
defer open_fs.file.close(io);
defer open_log.deinit(gpa);
const opened = try persistStoreOpen(gpa, &open_log, ms.store());
try updateConfigOpen(gpa, &open_log, patch);
try open_log.commit();
try std.testing.expectEqual(wrapped.written, opened.written);
try std.testing.expectEqual(wrapped.submanifest_gaps, opened.submanifest_gaps);
try std.testing.expectEqual(@as(u64, 3), (try open_log.logKey()).frame_count);
var wrapped_fs: storage.FileStorage = undefined;
var wrapped_log = try d4OpenLog(gpa, io, wrapped_dir.dir, &wrapped_fs);
defer wrapped_fs.file.close(io);
defer wrapped_log.deinit(gpa);
var wrapped_cfg = (try wrapped_log.readConfig(gpa)).?;
defer wrapped_cfg.deinit(gpa);
var opened_cfg = (try open_log.readConfig(gpa)).?;
defer opened_cfg.deinit(gpa);
try std.testing.expectEqualDeep(wrapped_cfg.config, opened_cfg.config);
try std.testing.expectEqualStrings("origin-code", opened_cfg.config.project_code);
try std.testing.expectEqualStrings("operator", opened_cfg.config.login_user);
try std.testing.expectEqualStrings("https://example.test/remote", opened_cfg.config.remote_url);
try std.testing.expect(opened_cfg.config.seen_submanifest);
try std.testing.expect(opened_cfg.config.enable_submanifests);
try std.testing.expectEqual(@as(u8, 5), opened_cfg.config.submanifest_k);
try std.testing.expect(opened_cfg.config.submanifest_auto_retune);
try std.testing.expectEqualDeep([_]u8{0x42} ** 32, opened_cfg.config.submanifest_settled_cost_shape.?);
var digest_buf: [digest_len]u8 = undefined;
const stored = (try open_log.content(gpa, open_log.lookup(decodeWireName(name, &digest_buf).?).?)).?;
defer gpa.free(stored);
try std.testing.expectEqualStrings(content, stored);
}
test "sync clone config: absent origin settings append no frame and preserve init values" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
try d4SeedConfiguredRepo(gpa, io, tmp.dir);
var fs: storage.FileStorage = undefined;
var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
defer fs.file.close(io);
defer log.deinit(gpa);
const patch: ConfigPatch = .{};
try std.testing.expect(!patch.any());
if (patch.any()) try updateConfigOpen(gpa, &log, patch);
try log.commit();
try std.testing.expectEqual(@as(u64, 1), (try log.logKey()).frame_count);
var cfg = (try log.readConfig(gpa)).?;
defer cfg.deinit(gpa);
try std.testing.expectEqual(@as(u8, 9), cfg.config.hash_policy);
try std.testing.expectEqualStrings("stable", cfg.config.default_branch);
try std.testing.expectEqualStrings("project", cfg.config.project_id);
try std.testing.expectEqualStrings("init-code", cfg.config.project_code);
try std.testing.expectEqualStrings("operator", cfg.config.login_user);
try std.testing.expectEqualStrings("https://example.test/remote", cfg.config.remote_url);
try std.testing.expect(cfg.config.seen_submanifest);
try std.testing.expect(!cfg.config.enable_submanifests);
try std.testing.expectEqual(@as(u8, 3), cfg.config.submanifest_k);
try std.testing.expect(!cfg.config.submanifest_auto_retune);
try std.testing.expectEqualDeep([_]u8{0x42} ** 32, cfg.config.submanifest_settled_cost_shape.?);
}
test "sync persistStoreOpen: malformed artifact retains the wrapper's corrupt error and writes no frame" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
try d4SeedEmptyRepo(gpa, io, tmp.dir);
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
const content = "content that the wire name does not actually name\n";
const lying_name = "0" ** 64;
const rid = try ms.store().put(lying_name, content);
const container = try zit.container.encode(gpa, content, .default);
defer gpa.free(container);
try ms.store().retain(rid, .{ .full = container });
var fs: storage.FileStorage = undefined;
var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
defer fs.file.close(io);
defer log.deinit(gpa);
try std.testing.expectError(error.Corrupt, persistStoreOpen(gpa, &log, ms.store()));
try log.commit();
var digest_buf: [digest_len]u8 = undefined;
try std.testing.expect(log.lookup(decodeWireName(lying_name, &digest_buf).?) == null);
}
test "sync §7.7: persistStore recomputes the artifact hash and aborts on a lying container, writing no frame" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
try d4SeedEmptyRepo(gpa, io, tmp.dir);
// A received full cfile whose wire NAME does not name its content (§7.7 —
// a clone skips the transfer-layer pre-storage check, so a lying container
// reaches the persist commit hook).
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
const s = ms.store();
const content = "content that the wire name does not actually name\n";
const lying_name = "0" ** 64; // not SHA3-256(content)
const rid = try s.put(lying_name, content);
const container = try zit.container.encode(gpa, content, .default);
defer gpa.free(container);
try s.retain(rid, .{ .full = container });
// §7.7 L1336-1338 — recompute and abort before ANY frame is written.
try std.testing.expectError(error.Corrupt, persistStore(gpa, io, tmp.dir, ms.store()));
var fs: storage.FileStorage = undefined;
var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
defer fs.file.close(io);
defer log.deinit(gpa);
var digest_buf: [digest_len]u8 = undefined;
const digest = decodeWireName(lying_name, &digest_buf).?;
try std.testing.expect(log.lookup(digest) == null); // no frame landed
}
test "sync §7.2/§4.1: persistStore stores received full and delta wire forms verbatim; a re-serve preserves the delta container byte-for-byte" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
try d4SeedEmptyRepo(gpa, io, tmp.dir);
// Model a clone reply in §8.2 RID order: a full base cfile then a delta
// cfile against it. Names are the real SHA3-256 so the §7.7 hook accepts.
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
const s = ms.store();
const base = "the base artifact with a long shared run so the delta stays a compact copy payload distinct in size from the reconstructed target";
const target = base ++ " plus a small trailing mutation";
var bn: [64]u8 = undefined;
var tn: [64]u8 = undefined;
const base_name = zit.artifact.artifactName(.sha3_256, base, &bn);
const target_name = zit.artifact.artifactName(.sha3_256, target, &tn);
const base_container = try zit.container.encode(gpa, base, .default);
defer gpa.free(base_container);
const base_rid = try s.put(base_name, base);
try s.retain(base_rid, .{ .full = base_container });
const raw = try zit.delta.create(gpa, base, target);
defer gpa.free(raw);
const delta_container = try zit.container.encode(gpa, raw, .default);
defer gpa.free(delta_container);
const ts = try zit.delta.targetSize(raw);
const delta_rid = try s.put(target_name, target);
try s.retain(delta_rid, .{ .delta = .{ .src = base_name, .container = delta_container, .target_size = ts } });
const written = try persistStore(gpa, io, tmp.dir, ms.store());
try std.testing.expectEqual(@as(usize, 2), written.written);
var fs: storage.FileStorage = undefined;
var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
defer fs.file.close(io);
defer log.deinit(gpa);
// The target persisted as a `.delta` frame that reconstructs to the original.
var tdig: [digest_len]u8 = undefined;
const target_digest = decodeWireName(target_name, &tdig).?;
const trid = log.lookup(target_digest).?;
try std.testing.expectEqual(logstore.FrameKind.delta, log.kindOf(trid));
const got = (try log.content(gpa, trid)).?;
defer gpa.free(got);
try std.testing.expectEqualStrings(target, got);
// Clone-of-clone: a re-serve snapshot emits the SAME delta container bytes
// (§7.2 retransmitted verbatim) — the compactness survives the second hop.
var serve_ms = repostate.MemStore.init(gpa);
defer serve_ms.deinit();
try loadContentStore(gpa, &log, &serve_ms);
const serve_rid = serve_ms.store().lookup(target_name).?;
const wf = serve_ms.store().wire(serve_rid).?;
try std.testing.expect(wf.delta != null);
try std.testing.expectEqualSlices(u8, delta_container, wf.container);
try std.testing.expectEqual(@as(u64, target.len), wf.delta.?.target_size);
}
/// A tree manifest naming `subs` (submanifests §3.6), plus the §6.2 Z card, so
/// the fixture is a well-formed card sequence rather than a bare S run.
fn treeManifestBytes(gpa: std.mem.Allocator, subs: []const []const u8) ![]u8 {
var body: std.ArrayListUnmanaged(u8) = .empty;
defer body.deinit(gpa);
try body.appendSlice(gpa, "C tree\nD 2026-07-27T12:00:00.000\n");
// §3.4 — the S run ascends by filename; `a`, `b`, ... keeps it legal.
for (subs, 0..) |h, i| {
try body.print(gpa, "S {c}.txt {s}\n", .{ @as(u8, 'a') + @as(u8, @intCast(i)), h });
}
try body.appendSlice(gpa, "U u\n");
var digest: [16]u8 = undefined;
std.crypto.hash.Md5.hash(body.items, &digest, .{});
return std.fmt.allocPrint(gpa, "{s}Z {x}\n", .{ body.items, &digest });
}
test "submanifests §7.1: the persist commit hook phantomizes every absent S reference" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
try d4SeedEmptyRepo(gpa, io, tmp.dir);
// Two leaves that were never received — the state a clone from a peer whose
// OWN closure is incomplete lands in. Nothing on the receive path can see
// this: a clone stores wire-backed and inflates nothing.
const leaf_a = "F a.txt " ++ ("1" ** 64) ++ "\n";
const leaf_b = "F b.txt " ++ ("2" ** 64) ++ "\n";
var an: [64]u8 = undefined;
var bn: [64]u8 = undefined;
const name_a = zit.artifact.artifactName(.sha3_256, leaf_a, &an);
const name_b = zit.artifact.artifactName(.sha3_256, leaf_b, &bn);
const tree = try treeManifestBytes(gpa, &.{ name_a, name_b });
defer gpa.free(tree);
var tn: [64]u8 = undefined;
const tree_name = zit.artifact.artifactName(.sha3_256, tree, &tn);
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
const s = ms.store();
const rid = try s.put(tree_name, tree);
const container = try zit.container.encode(gpa, tree, .default);
defer gpa.free(container);
try s.retain(rid, .{ .full = container });
// §7.1: "Storing a structural artifact that carries S cards whose referenced
// sub-manifests are absent MUST create a phantom for each missing reference."
const summary = try persistStore(gpa, io, tmp.dir, ms.store());
try std.testing.expectEqual(@as(usize, 1), summary.written);
try std.testing.expectEqual(@as(usize, 2), summary.submanifest_gaps);
}
test "submanifests §7.1/§8.1: the S-reference gap survives a RESTART, rebuilt from the durable indicator" {
// §7.1's phantom is an interned entry with no frame, "exactly as an absent
// delta source creates a source phantom" — and `LogStore.intern` states the
// consequence outright: "Phantoms never reach disk." So a restart loses
// them, and the only thing that can make them correct across one is a
// REBUILD from the durable content-bearing frames.
//
// This test is that invariant: run 1 mints the gap, the process ends, run 2
// re-derives the identical gap over a store loaded from `.zit` alone. Remove
// the rebuild (`rescan_persisted`) and run 2 reports ZERO gaps — a
// repository that quietly declares an unenumerable check-in converged.
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
try d4SeedEmptyRepo(gpa, io, tmp.dir);
const leaf_a = "F a.txt " ++ ("1" ** 64) ++ "\n";
const leaf_b = "F b.txt " ++ ("2" ** 64) ++ "\n";
var an: [64]u8 = undefined;
var bn: [64]u8 = undefined;
const name_a = zit.artifact.artifactName(.sha3_256, leaf_a, &an);
const name_b = zit.artifact.artifactName(.sha3_256, leaf_b, &bn);
const tree = try treeManifestBytes(gpa, &.{ name_a, name_b });
defer gpa.free(tree);
var tn: [64]u8 = undefined;
const tree_name = zit.artifact.artifactName(.sha3_256, tree, &tn);
// RUN 1 — the tree lands; neither sub-manifest is held.
{
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
_ = try ms.store().put(tree_name, tree);
const first = try persistStore(gpa, io, tmp.dir, ms.store());
try std.testing.expectEqual(@as(usize, 1), first.written);
try std.testing.expectEqual(@as(usize, 2), first.submanifest_gaps);
}
// RESTART — nothing survives but the bytes in `.zit`.
var ms2 = repostate.MemStore.init(gpa);
defer ms2.deinit();
{
var fs: storage.FileStorage = undefined;
var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
defer fs.file.close(io);
defer log.deinit(gpa);
// §8.1 — the indicator IS durable: "record a local seen-submanifest
// indicator". It is the one byte the rebuild rests on.
var oc = (try log.readConfig(gpa)) orelse return error.NoConfig;
defer oc.deinit(gpa);
try std.testing.expect(oc.config.seen_submanifest);
// …and the phantoms are NOT durable, asserted rather than assumed —
// this is the premise the whole rebuild exists to answer.
var adig: [digest_len]u8 = undefined;
try std.testing.expect(log.lookup(decodeWireName(name_a, &adig).?) == null);
// The snapshot a fresh process starts every verb from.
try loadContentStore(gpa, &log, &ms2);
}
// Only the tree came back: the store a second run works from holds no trace
// of the gap.
try std.testing.expectEqual(@as(usize, 1), ms2.store().count());
// RUN 2 — the gap is re-derived, identically, from the durable frames.
const second = try persistStore(gpa, io, tmp.dir, ms2.store());
try std.testing.expectEqual(@as(usize, 1), second.written);
try std.testing.expectEqual(@as(usize, 2), second.submanifest_gaps);
}
test "submanifests §8.1: a repository predating this draft records no indicator and rescans nothing" {
// The compatibility half of the rebuild: the indicator is false, so the
// rescan is unreachable and a second persist over the same store is exactly
// the sweep it always was — no config frame appended, no gap invented.
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
try d4SeedEmptyRepo(gpa, io, tmp.dir);
const flat = "C msg\nD 2026-07-27T12:00:00.000\nF a.txt " ++ ("1" ** 64) ++ "\nU u\n";
var fn_buf: [64]u8 = undefined;
const flat_name = zit.artifact.artifactName(.sha3_256, flat, &fn_buf);
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
_ = try ms.store().put(flat_name, flat);
_ = try persistStore(gpa, io, tmp.dir, ms.store());
const again = try persistStore(gpa, io, tmp.dir, ms.store());
try std.testing.expectEqual(@as(usize, 0), again.submanifest_gaps);
var fs: storage.FileStorage = undefined;
var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
defer fs.file.close(io);
defer log.deinit(gpa);
// No indicator was written at all: `d4SeedEmptyRepo` stores no config, and
// nothing above gave it a reason to start one.
try std.testing.expect((try log.readConfig(gpa)) == null);
}
test "submanifests §7.1: a persisted tree whose sub-manifests are held reports no gap" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
try d4SeedEmptyRepo(gpa, io, tmp.dir);
const leaf_a = "F a.txt " ++ ("1" ** 64) ++ "\n";
var an: [64]u8 = undefined;
const name_a = zit.artifact.artifactName(.sha3_256, leaf_a, &an);
const tree = try treeManifestBytes(gpa, &.{name_a});
defer gpa.free(tree);
var tn: [64]u8 = undefined;
const tree_name = zit.artifact.artifactName(.sha3_256, tree, &tn);
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
const s = ms.store();
// The sub-manifest is stored at a LOWER rid than the tree in one direction
// and a higher one in the other; the gap verdict must not depend on which,
// which is why it is taken once, after the whole sweep.
_ = try s.put(tree_name, tree);
_ = try s.put(name_a, leaf_a);
const summary = try persistStore(gpa, io, tmp.dir, ms.store());
try std.testing.expectEqual(@as(usize, 2), summary.written);
try std.testing.expectEqual(@as(usize, 0), summary.submanifest_gaps);
// The §7.1 phantom was minted for the leaf BEFORE the leaf's own frame was
// written (the tree persists first), so this asserts the phantom-upgrade
// path storage actually takes: the frame must read back byte-for-byte, not
// merely exist. A §7.1 duty that corrupted a frame would be worse than an
// unmet one.
var fs: storage.FileStorage = undefined;
var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
defer fs.file.close(io);
defer log.deinit(gpa);
var ldig: [digest_len]u8 = undefined;
const leaf_digest = decodeWireName(name_a, &ldig).?;
const lrid = log.lookup(leaf_digest).?;
try std.testing.expect(!log.isPhantom(lrid));
const got = (try log.content(gpa, lrid)).?;
defer gpa.free(got);
try std.testing.expectEqualStrings(leaf_a, got);
}
test "submanifests §7.1: an artifact predating this draft costs the persist hook no gap and no phantom" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
try d4SeedEmptyRepo(gpa, io, tmp.dir);
// The backwards-compatibility guarantee as a test: a flat manifest carries
// no S card, so §7.1's duty over it is empty and the persisted bytes and
// frame count are exactly what they were before this rule existed.
const flat = "C msg\nD 2026-07-27T12:00:00.000\nF a.txt " ++ ("1" ** 64) ++ "\nU u\n";
var fn_buf: [64]u8 = undefined;
const flat_name = zit.artifact.artifactName(.sha3_256, flat, &fn_buf);
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
_ = try ms.store().put(flat_name, flat);
const summary = try persistStore(gpa, io, tmp.dir, ms.store());
try std.testing.expectEqual(@as(usize, 1), summary.written);
try std.testing.expectEqual(@as(usize, 0), summary.submanifest_gaps);
var fs: storage.FileStorage = undefined;
var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
defer fs.file.close(io);
defer log.deinit(gpa);
const rids = try log.ridsInPersistedOrder(gpa);
defer gpa.free(rids);
var artifacts: usize = 0;
for (rids) |r| {
if (log.kindOf(r) != .config) artifacts += 1;
}
try std.testing.expectEqual(@as(usize, 1), artifacts);
}
test "submanifests §7.1: a malformed S run mints no phantom at the persist hook" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
try d4SeedEmptyRepo(gpa, io, tmp.dir);
// §3.2: a third token on S is rejected. Validate-then-apply means the FIRST
// (well-formed) S card mints nothing either — a half-read artifact never
// leaves a half-minted phantom set behind.
const bad = "C tree\nS a.txt " ++ ("1" ** 64) ++ "\nS b.txt " ++ ("2" ** 64) ++ " extra\nU u\n";
var bn: [64]u8 = undefined;
const bad_name = zit.artifact.artifactName(.sha3_256, bad, &bn);
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
_ = try ms.store().put(bad_name, bad);
const summary = try persistStore(gpa, io, tmp.dir, ms.store());
try std.testing.expectEqual(@as(usize, 1), summary.written);
try std.testing.expectEqual(@as(usize, 0), summary.submanifest_gaps);
}
test "sync §3.3: persistStore falls back to full form when a retained delta's basis is absent, never a phantom-based delta" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
try d4SeedEmptyRepo(gpa, io, tmp.dir);
// Only the delta is present; its basis was never received/stored (a legal
// §3.3 outcome that must not persist a delta over a phantom base).
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
const s = ms.store();
const base = "an absent basis whose long shared run would make a compact delta if it were here at all";
const target = base ++ " mutated once";
var bn: [64]u8 = undefined;
var tn: [64]u8 = undefined;
const base_name = zit.artifact.artifactName(.sha3_256, base, &bn);
const target_name = zit.artifact.artifactName(.sha3_256, target, &tn);
const raw = try zit.delta.create(gpa, base, target);
defer gpa.free(raw);
const dc = try zit.container.encode(gpa, raw, .default);
defer gpa.free(dc);
const ts = try zit.delta.targetSize(raw);
const rid = try s.put(target_name, target);
try s.retain(rid, .{ .delta = .{ .src = base_name, .container = dc, .target_size = ts } });
const written = try persistStore(gpa, io, tmp.dir, ms.store());
try std.testing.expectEqual(@as(usize, 1), written.written);
var fs: storage.FileStorage = undefined;
var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
defer fs.file.close(io);
defer log.deinit(gpa);
// The target persisted as a FULL (.zlib) frame — the fallback, not a delta.
var tdig: [digest_len]u8 = undefined;
const target_digest = decodeWireName(target_name, &tdig).?;
const trid = log.lookup(target_digest).?;
try std.testing.expectEqual(logstore.FrameKind.zlib, log.kindOf(trid));
const got = (try log.content(gpa, trid)).?;
defer gpa.free(got);
try std.testing.expectEqualStrings(target, got);
// No phantom basis frame was minted for the absent DELTASRC.
var bdig: [digest_len]u8 = undefined;
const base_digest = decodeWireName(base_name, &bdig).?;
try std.testing.expect(log.lookup(base_digest) == null);
}
test "sync §7.7/§3.3: persistStore stores a retained delta chain up to the read bound and falls back to full form beyond it" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
try d4SeedEmptyRepo(gpa, io, tmp.dir);
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
const s = ms.store();
const cap: usize = logstore.max_delta_chain; // the READ bound (32)
const n: usize = cap + 2; // links 0..cap+1: link cap+1's basis sits at depth cap
// A chain grown one byte per link so each delta is tiny. Link i's basis is
// link i-1 (§8.2 RID order); the persist path stores a `.delta` iff the base
// depth is under `max_delta_chain`, so link cap+1 falls back to full form.
var content: std.ArrayList(u8) = .empty;
defer content.deinit(gpa);
try content.appendSlice(gpa, "chain-root-artifact-with-a-long-shared-run-for-compact-forward-deltas");
var prev_name: [64]u8 = undefined;
var prev_name_len: usize = 0;
var deep_name: [64]u8 = undefined; // link cap (deepest stored delta)
var deep_len: usize = 0;
var last_name: [64]u8 = undefined; // link cap+1 (the fallback)
var last_len: usize = 0;
var i: usize = 0;
while (i < n) : (i += 1) {
try content.append(gpa, 'a' + @as(u8, @intCast(i % 26)));
const cur = content.items;
const prev = cur[0 .. cur.len - 1];
var nbuf: [64]u8 = undefined;
const name = zit.artifact.artifactName(.sha3_256, cur, &nbuf);
const rid = try s.put(name, cur);
if (i == 0) {
const c = try zit.container.encode(gpa, cur, .default);
defer gpa.free(c);
try s.retain(rid, .{ .full = c });
} else {
const raw = try zit.delta.create(gpa, prev, cur);
defer gpa.free(raw);
const c = try zit.container.encode(gpa, raw, .default);
defer gpa.free(c);
const tsz = try zit.delta.targetSize(raw);
try s.retain(rid, .{ .delta = .{ .src = prev_name[0..prev_name_len], .container = c, .target_size = tsz } });
}
@memcpy(prev_name[0..name.len], name);
prev_name_len = name.len;
if (i == cap) {
@memcpy(deep_name[0..name.len], name);
deep_len = name.len;
}
if (i == n - 1) {
@memcpy(last_name[0..name.len], name);
last_len = name.len;
}
}
_ = try persistStore(gpa, io, tmp.dir, ms.store());
var fs: storage.FileStorage = undefined;
var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
defer fs.file.close(io);
defer log.deinit(gpa);
// Link cap is the deepest `.delta` the write bound allows; link cap+1 fell
// back to a full `.zlib` frame yet still reconstructs (never a Corrupt read).
var ddig: [digest_len]u8 = undefined;
const deep_rid = log.lookup(decodeWireName(deep_name[0..deep_len], &ddig).?).?;
try std.testing.expectEqual(logstore.FrameKind.delta, log.kindOf(deep_rid));
var ldig: [digest_len]u8 = undefined;
const last_rid = log.lookup(decodeWireName(last_name[0..last_len], &ldig).?).?;
try std.testing.expectEqual(logstore.FrameKind.zlib, log.kindOf(last_rid));
const got = (try log.content(gpa, last_rid)).?;
defer gpa.free(got);
try std.testing.expectEqualStrings(content.items, got);
}
test "sync §7.2/§7.4/§3.3: a dangling reverse-delta cfile persists as a verbatim .delta frame that re-serves byte-identically" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
try d4SeedEmptyRepo(gpa, io, tmp.dir);
// Fossil's clone stream: a reverse delta (the OLDER revision stored as a
// delta of the NEWER one) arrives BEFORE its basis, so it is dangling. The
// client retains the §4.1 container through the pending queue and, when the
// basis lands, persists it as a compact `.delta` — not a full re-compression.
const newer = "artifact NEWER: a long shared body kept verbatim so the reverse delta reconstructing the older revision stays a compact copy op";
const older = "artifact NEWER: a long shared body kept verbatim so the reverse delta reconstructing the older revision stays a compact copy op minus tail";
var nn: [64]u8 = undefined;
var on: [64]u8 = undefined;
const newer_name = zit.artifact.artifactName(.sha3_256, newer, &nn);
const older_name = zit.artifact.artifactName(.sha3_256, older, &on);
const newer_container = try zit.container.encode(gpa, newer, .default);
defer gpa.free(newer_container);
const raw = try zit.delta.create(gpa, newer, older); // OLDER = delta ∘ NEWER
defer gpa.free(raw);
const delta_container = try zit.container.encode(gpa, raw, .default);
defer gpa.free(delta_container);
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
var client = zit.sync.Client{ .operation = .clone, .state = .clone_round0 };
defer client.deinit(gpa);
var body: std.ArrayList(u8) = .empty;
defer body.deinit(gpa);
try body.appendSlice(gpa, "clone_seqno 0\npush 1111111111111111 aa87ea41b0bb05ba528f7a3bdba700d7e23d6318\n");
// Reverse (dangling) order: the delta cfile precedes its basis cfile.
const dh = try std.fmt.allocPrint(gpa, "cfile {s} {s} {d} {d}\n", .{ older_name, newer_name, older.len, delta_container.len });
defer gpa.free(dh);
try body.appendSlice(gpa, dh);
try body.appendSlice(gpa, delta_container);
try body.append(gpa, '\n');
const bh = try std.fmt.allocPrint(gpa, "cfile {s} {d} {d}\n", .{ newer_name, newer.len, newer_container.len });
defer gpa.free(bh);
try body.appendSlice(gpa, bh);
try body.appendSlice(gpa, newer_container);
try body.append(gpa, '\n');
_ = try client.consumeReply(gpa, ms.store(), body.items);
const written = try persistStore(gpa, io, tmp.dir, ms.store());
try std.testing.expectEqual(@as(usize, 2), written.written);
var fs: storage.FileStorage = undefined;
var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
defer fs.file.close(io);
defer log.deinit(gpa);
// §3.3 — the older revision persisted as a `.delta`, not a full re-compression,
// and reconstructs the original bytes (delta ∘ basis).
var od: [digest_len]u8 = undefined;
const older_rid = log.lookup(decodeWireName(older_name, &od).?).?;
try std.testing.expectEqual(logstore.FrameKind.delta, log.kindOf(older_rid));
const got = (try log.content(gpa, older_rid)).?;
defer gpa.free(got);
try std.testing.expectEqualStrings(older, got);
// §7.2 — a clone-of-clone re-serve re-emits the SAME delta container byte-for-byte.
var serve_ms = repostate.MemStore.init(gpa);
defer serve_ms.deinit();
try loadContentStore(gpa, &log, &serve_ms);
const srid = serve_ms.store().lookup(older_name).?;
const wf = serve_ms.store().wire(srid).?;
try std.testing.expect(wf.delta != null);
try std.testing.expectEqualSlices(u8, delta_container, wf.container);
try std.testing.expectEqualStrings(newer_name, wf.delta.?.src_hash);
}
test "sync §7.2/§3.3: persistStore stores a whole reverse-delta chain, ordering every basis frame before the delta that points at it" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
try d4SeedEmptyRepo(gpa, io, tmp.dir);
// A reverse-delta chain in fossil's storage shape: revision i is stored as a
// delta of the NEWER revision i+1, the newest stored full. The store rids run
// oldest..newest (put order), so EVERY delta's basis sits at a HIGHER rid than
// the delta — the worst case for physical ordering. persist's worklist must
// still write each basis frame before its delta so `delta_of` points backward;
// naive rid-order would degrade all but the oldest link to full form.
const links = 6;
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
const s = ms.store();
var bodies: [links][]u8 = undefined;
var made: usize = 0;
defer for (bodies[0..made]) |b| gpa.free(b);
var names: [links][64]u8 = undefined;
var nlen: [links]usize = undefined;
for (0..links) |i| {
bodies[i] = try std.fmt.allocPrint(gpa, "chain body: one long shared prefix kept across every revision for compact reverse deltas -- revision {d}", .{i});
made += 1;
nlen[i] = zit.artifact.artifactName(.sha3_256, bodies[i], &names[i]).len;
}
// put oldest..newest (store rids 1..links); older i is a reverse delta of newer i+1.
for (0..links) |i| {
const rid = try s.put(names[i][0..nlen[i]], bodies[i]);
if (i == links - 1) {
const c = try zit.container.encode(gpa, bodies[i], .default);
defer gpa.free(c);
try s.retain(rid, .{ .full = c });
} else {
const dr = try zit.delta.create(gpa, bodies[i + 1], bodies[i]); // older from newer
defer gpa.free(dr);
const c = try zit.container.encode(gpa, dr, .default);
defer gpa.free(c);
try s.retain(rid, .{ .delta = .{ .src = names[i + 1][0..nlen[i + 1]], .container = c, .target_size = bodies[i].len } });
}
}
_ = try persistStore(gpa, io, tmp.dir, ms.store());
var fs: storage.FileStorage = undefined;
var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
defer fs.file.close(io);
defer log.deinit(gpa);
// The newest link is full; every older link is a `.delta` whose basis frame is
// physically earlier (a LOWER log rid = a lower offset), and all reconstruct.
for (0..links) |i| {
var d: [digest_len]u8 = undefined;
const rid = log.lookup(decodeWireName(names[i][0..nlen[i]], &d).?).?;
const rgot = (try log.content(gpa, rid)).?;
defer gpa.free(rgot);
try std.testing.expectEqualStrings(bodies[i], rgot);
if (i == links - 1) {
try std.testing.expectEqual(logstore.FrameKind.zlib, log.kindOf(rid));
} else {
try std.testing.expectEqual(logstore.FrameKind.delta, log.kindOf(rid));
var bd: [digest_len]u8 = undefined;
const brid = log.lookup(decodeWireName(names[i + 1][0..nlen[i + 1]], &bd).?).?;
try std.testing.expect(brid < rid); // basis frame precedes the reverse-delta
}
}
}
test "sync §7.7: persistStore verifies a retained delta's reconstruction and refuses a lying delta container, writing no frame" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
try d4SeedEmptyRepo(gpa, io, tmp.dir);
// A retained delta whose stored plaintext does NOT hash to its wire name (a
// clone skips the transfer-layer pre-storage check, §7.7 r6, so a lying
// delta reaches the persist commit hook). The §7.7 recompute runs BEFORE the
// wire-form branch, so a delta-form entry is rejected exactly like a full one.
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
const s = ms.store();
const content = "content that the wire name does not actually name, delivered as a delta\n";
const lying_name = "0" ** 64; // not SHA3-256(content)
const some_basis = "1" ** 64; // an absent basis; verify fails before it is even consulted
const raw = try zit.delta.create(gpa, "a plausible basis blob", content);
defer gpa.free(raw);
const dc = try zit.container.encode(gpa, raw, .default);
defer gpa.free(dc);
const rid = try s.put(lying_name, content);
try s.retain(rid, .{ .delta = .{ .src = some_basis, .container = dc, .target_size = content.len } });
try std.testing.expectError(error.Corrupt, persistStore(gpa, io, tmp.dir, ms.store()));
var fs: storage.FileStorage = undefined;
var log = try d4OpenLog(gpa, io, tmp.dir, &fs);
defer fs.file.close(io);
defer log.deinit(gpa);
var digest_buf: [digest_len]u8 = undefined;
try std.testing.expect(log.lookup(decodeWireName(lying_name, &digest_buf).?) == null); // no frame landed
}
fn persistReverseChainNoLeak(allocator: std.mem.Allocator) !void {
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
try d4SeedEmptyRepo(allocator, io, tmp.dir);
// A small reverse-delta chain built on the (failing) allocator so injected
// OOM is exercised in both the store build and the persist worklist.
var ms = repostate.MemStore.init(allocator);
defer ms.deinit();
const s = ms.store();
const links = 4;
var bodies: [links][]u8 = undefined;
var made: usize = 0;
defer for (bodies[0..made]) |b| allocator.free(b);
var names: [links][64]u8 = undefined;
var nlen: [links]usize = undefined;
for (0..links) |i| {
bodies[i] = try std.fmt.allocPrint(allocator, "reverse chain oom: one shared body across revisions -- revision {d}", .{i});
made += 1;
nlen[i] = zit.artifact.artifactName(.sha3_256, bodies[i], &names[i]).len;
}
for (0..links) |i| {
const rid = try s.put(names[i][0..nlen[i]], bodies[i]);
if (i == links - 1) {
const c = try zit.container.encode(allocator, bodies[i], .default);
defer allocator.free(c);
try s.retain(rid, .{ .full = c });
} else {
const dr = try zit.delta.create(allocator, bodies[i + 1], bodies[i]);
defer allocator.free(dr);
const c = try zit.container.encode(allocator, dr, .default);
defer allocator.free(c);
try s.retain(rid, .{ .delta = .{ .src = names[i + 1][0..nlen[i + 1]], .container = c, .target_size = bodies[i].len } });
}
}
_ = try persistStore(allocator, io, tmp.dir, ms.store());
}
test "sync §7.2/§3.3: persistStore drains the reverse-delta worklist without leaking under allocation failure" {
try std.testing.checkAllAllocationFailures(std.testing.allocator, persistReverseChainNoLeak, .{});
}
/// Drives `persistArtifact` over a FORWARD delta chain (each link a delta of the
/// immediately older, already-persisted lower-rid link) in rid order, so a test
/// owns the shared basis memo and can read its hit counter. Every basis precedes
/// its delta, so `allow_defer=false` never defers and no worklist is needed —
/// mirrors `persistStore`'s open/commit around a plain rid-order drive.
fn persistForwardChain(
gpa: std.mem.Allocator,
io: std.Io,
dir: std.Io.Dir,
ms: *repostate.MemStore,
cache: *logstore.PlaintextCache,
) !void {
var file = try dir.openFile(io, repo_filename, .{ .mode = .read_write });
defer file.close(io);
const st = try file.stat(io);
var fs = storage.FileStorage.init(io, file, st.size);
var log = try logstore.LogStore.open(gpa, fs.storageFile(), 0);
defer log.deinit(gpa);
const store = ms.store();
const n = store.count();
var srefs: SubManifestRefs = .{};
defer srefs.deinit(gpa);
var rid: repostate.Rid = 1;
while (rid <= n) : (rid += 1) {
switch (try persistArtifact(gpa, ms.store(), &log, rid, false, cache, &srefs, false)) {
.persisted => {},
.deferred => unreachable, // forward chain: no basis is ever absent
}
}
try log.commit();
}
test "sync §7.2/§7.4: the basis memo reuses each basis (hits) and reconstructs a byte-identical store to the uncached path" {
const gpa = std.testing.allocator;
const io = std.testing.io;
const links = 8;
// A forward delta chain: link 0 is full, every later link is a delta that
// reconstructs its body FROM the older body (its basis a lower, already-
// persisted rid) — the case where the memo turns each basis reconstruction
// into a lookup for the delta built on it. Store owns its copies (retain
// dupes), but the bodies/names back the reconstructed-content assertions.
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
const s = ms.store();
var bodies: [links][]u8 = undefined;
var made: usize = 0;
defer for (bodies[0..made]) |b| gpa.free(b);
var names: [links][64]u8 = undefined;
var nlen: [links]usize = undefined;
for (0..links) |i| {
bodies[i] = try std.fmt.allocPrint(gpa, "forward chain shared prefix across every revision -- revision {d}", .{i});
made += 1;
nlen[i] = zit.artifact.artifactName(.sha3_256, bodies[i], &names[i]).len;
}
for (0..links) |i| {
const rid = try s.put(names[i][0..nlen[i]], bodies[i]);
if (i == 0) {
const c = try zit.container.encode(gpa, bodies[i], .default);
defer gpa.free(c);
try s.retain(rid, .{ .full = c });
} else {
const dr = try zit.delta.create(gpa, bodies[i - 1], bodies[i]); // newer body FROM its older basis
defer gpa.free(dr);
const c = try zit.container.encode(gpa, dr, .default);
defer gpa.free(c);
try s.retain(rid, .{ .delta = .{ .src = names[i - 1][0..nlen[i - 1]], .container = c, .target_size = bodies[i].len } });
}
}
// (A) a normal memo; (B) a zero-budget memo that can hold nothing — the exact
// uncached reconstruction. Same store, two freshly-seeded repos.
var tmp_a = std.testing.tmpDir(.{});
defer tmp_a.cleanup();
try d4SeedEmptyRepo(gpa, io, tmp_a.dir);
var cache_a = logstore.PlaintextCache.init();
defer cache_a.deinit(gpa);
try persistForwardChain(gpa, io, tmp_a.dir, &ms, &cache_a);
var tmp_b = std.testing.tmpDir(.{});
defer tmp_b.cleanup();
try d4SeedEmptyRepo(gpa, io, tmp_b.dir);
var cache_b = logstore.PlaintextCache.init();
cache_b.max_bytes = 0; // never memoises: forces a fresh reconstruction each time
defer cache_b.deinit(gpa);
try persistForwardChain(gpa, io, tmp_b.dir, &ms, &cache_b);
// The memo reused every basis exactly once (one hit per delta link); the
// zero-budget run reused nothing — the two ways the same bytes get built.
try std.testing.expectEqual(@as(u64, links - 1), cache_a.hits);
try std.testing.expectEqual(@as(u64, 0), cache_b.hits);
// Byte-identical store: the memo changes only the work, never the bytes.
const bytes_a = try tmp_a.dir.readFileAlloc(io, repo_filename, gpa, .limited(1 << 24));
defer gpa.free(bytes_a);
const bytes_b = try tmp_b.dir.readFileAlloc(io, repo_filename, gpa, .limited(1 << 24));
defer gpa.free(bytes_b);
try std.testing.expectEqualSlices(u8, bytes_b, bytes_a);
// Every link reconstructs to its exact body (§7.7 chain correctness).
var fs: storage.FileStorage = undefined;
var log = try d4OpenLog(gpa, io, tmp_a.dir, &fs);
defer fs.file.close(io);
defer log.deinit(gpa);
for (0..links) |i| {
var d: [digest_len]u8 = undefined;
const rid = log.lookup(decodeWireName(names[i][0..nlen[i]], &d).?).?;
const got = (try log.content(gpa, rid)).?;
defer gpa.free(got);
try std.testing.expectEqualStrings(bodies[i], got);
}
}
test "sync §9.1: repoProjectCode returns the stored code, else the zero placeholder" {
const gpa = std.testing.allocator;
const io = std.testing.io;
const hash_kind: u8 = @intFromEnum(zit.artifact.defaultPolicy(false, false));
const seed = struct {
fn withCode(g: std.mem.Allocator, i: std.Io, dir: std.Io.Dir, hk: u8, code: []const u8) !void {
var f = try dir.createFile(i, repo_filename, .{ .read = true, .exclusive = true });
defer f.close(i);
var fs = storage.FileStorage.init(i, f, 0);
var log = try logstore.LogStore.open(g, fs.storageFile(), hk);
defer log.deinit(g);
try log.putConfig(.{ .hash_policy = hk, .default_branch = "trunk", .project_id = "", .project_code = code });
try log.commit();
}
};
// (a) a repo whose config stores a real 40-hex project code: it is returned
// verbatim — the code the pull/push card announces (§9.1), never the zeros
// fossil would refuse.
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const code = "a1b2c3d4" ** 5; // 40 hex chars
try seed.withCode(gpa, io, tmp.dir, hash_kind, code);
var buf: [zit.sync.max_projectcode_len]u8 = undefined;
const got = try repoProjectCode(gpa, io, tmp.dir, &buf);
try std.testing.expectEqualStrings(code, got);
try std.testing.expect(!std.mem.eql(u8, got, sync_projectcode));
// (b) a repo predating the stored code (empty project_code): the placeholder
// is the honest fallback (no adopted identity to announce yet).
var tmp2 = std.testing.tmpDir(.{});
defer tmp2.cleanup();
try seed.withCode(gpa, io, tmp2.dir, hash_kind, "");
var buf2: [zit.sync.max_projectcode_len]u8 = undefined;
const got2 = try repoProjectCode(gpa, io, tmp2.dir, &buf2);
try std.testing.expectEqualStrings(sync_projectcode, got2);
}
test "sync §9.1: the pull request card carries the adopted project code, not the placeholder" {
const gpa = std.testing.allocator;
// The pull client, constructed exactly as `pullRun` does, with a stored code.
const code = "a1b2c3d4" ** 5; // a real (non-zero) 40-hex project code
var client = zit.sync.Client{
.state = zit.sync.initialState(.pull),
.operation = .pull,
.self_servercode = sync_servercode,
.project_code = code,
.version = .{ .version = sync_client_version },
};
defer client.deinit(gpa);
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
var buf: std.ArrayListUnmanaged(u8) = .empty;
defer buf.deinit(gpa);
const request = try client.buildRequest(gpa, ms.store(), &buf);
// §9.1 ABNF `pull-card = "pull" SP servercode SP projectcode`: the adopted
// code MUST be the card's PROJECTCODE (a real peer refuses a mismatch).
const expected = "pull " ++ sync_servercode ++ " " ++ ("a1b2c3d4" ** 5) ++ "\n";
try std.testing.expect(std.mem.indexOf(u8, request, expected) != null);
// And the all-zero placeholder is NOT the project code the card announces.
const rejected = "pull " ++ sync_servercode ++ " " ++ sync_projectcode ++ "\n";
try std.testing.expect(std.mem.indexOf(u8, request, rejected) == null);
}
test "serve §3.8: a per-request store-load on one serve handle sees a later append (no restart)" {
const gpa = std.testing.allocator;
const io = std.testing.io;
const hash_kind: u8 = @intFromEnum(zit.artifact.defaultPolicy(false, false));
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const append = struct {
fn artifactFrame(g: std.mem.Allocator, i: std.Io, dir: std.Io.Dir, hk: u8, digest: []const u8, bytes: []const u8) !void {
var f = dir.openFile(i, repo_filename, .{ .mode = .read_write }) catch
try dir.createFile(i, repo_filename, .{ .read = true });
defer f.close(i);
const st = try f.stat(i);
var fs = storage.FileStorage.init(i, f, st.size);
var log = try logstore.LogStore.open(g, fs.storageFile(), hk);
defer log.deinit(g);
_ = try log.put(digest, bytes);
try log.commit();
}
};
// Seed the repo holding exactly one content artifact (a 32-byte raw digest).
const digest_a = "a" ** digest_len;
try append.artifactFrame(gpa, io, tmp.dir, hash_kind, digest_a, "content a\n");
// Open ONE read handle — the handle `serve` holds for its whole lifetime.
var handle = try tmp.dir.openFile(io, repo_filename, .{});
defer handle.close(io);
// First per-request read: one content artifact.
var index1 = try LogIndexStore.openFromFile(gpa, io, handle, (try handle.stat(io)).size);
defer index1.destroy();
const n1 = contentCount(index1.store());
try std.testing.expectEqual(@as(usize, 1), n1);
// A SEPARATE writer appends a second artifact while the handle stays open —
// exactly a `zit commit` running alongside a live `zit serve`.
const digest_b = "b" ** digest_len;
try append.artifactFrame(gpa, io, tmp.dir, hash_kind, digest_b, "content b\n");
// Second per-request read on the SAME handle sees the new artifact: proof the
// server re-reads the current on-disk log per request (§3.8). A startup
// snapshot would still report n1 — no restart is needed here.
var index2 = try LogIndexStore.openFromFile(gpa, io, handle, (try handle.stat(io)).size);
defer index2.destroy();
try std.testing.expectEqual(n1 + 1, contentCount(index2.store()));
}
/// A test-only `storage.StorageFile` that wraps another and tallies the bytes
/// every `pread` actually returned — the instrument for "how many times does a
/// prime read the log?". Every other vtable method delegates verbatim, so a log
/// opened over it behaves identically while its read traffic is measured.
const CountingStorage = struct {
inner: storage.StorageFile,
read_bytes: usize = 0,
fn storageFile(self: *CountingStorage) storage.StorageFile {
return .{ .context = self, .vtable = &vtable };
}
const vtable = storage.StorageFile.VTable{
.append = appendImpl,
.pread = preadImpl,
.sync = syncImpl,
.size = sizeImpl,
.truncate = truncateImpl,
};
fn appendImpl(ctx: *anyopaque, bytes: []const u8) storage.StorageError!void {
const self: *CountingStorage = @ptrCast(@alignCast(ctx));
return self.inner.append(bytes);
}
fn preadImpl(ctx: *anyopaque, offset: u64, buf: []u8) storage.StorageError!usize {
const self: *CountingStorage = @ptrCast(@alignCast(ctx));
const n = try self.inner.pread(offset, buf);
self.read_bytes += n;
return n;
}
fn syncImpl(ctx: *anyopaque) storage.StorageError!void {
const self: *CountingStorage = @ptrCast(@alignCast(ctx));
return self.inner.sync();
}
fn sizeImpl(ctx: *anyopaque) storage.StorageError!u64 {
const self: *CountingStorage = @ptrCast(@alignCast(ctx));
return self.inner.size();
}
fn truncateImpl(ctx: *anyopaque, len: u64) storage.StorageError!void {
const self: *CountingStorage = @ptrCast(@alignCast(ctx));
return self.inner.truncate(len);
}
};
/// Seed a `.zit` under `dir` holding a full+delta+manifest+config mix (via
/// `packSeedDeltaStore`) and return its committed byte size — the fixture the
/// serve-prime read-traffic and index-equivalence tests both build over.
fn seedServeRepo(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir) !u64 {
const hash_kind: u8 = @intFromEnum(zit.artifact.defaultPolicy(false, false));
var f = try dir.createFile(io, repo_filename, .{ .read = true });
defer f.close(io);
var fs = storage.FileStorage.init(io, f, 0);
var log = try logstore.LogStore.open(gpa, fs.storageFile(), hash_kind);
defer log.deinit(gpa);
var tip: [64]u8 = undefined;
try packSeedDeltaStore(gpa, &log, &tip);
return log.size();
}
test "#14 serve prime reads the served log at most once (no second full-container copy)" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const file_size = try seedServeRepo(gpa, io, tmp.dir);
var handle = try tmp.dir.openFile(io, repo_filename, .{});
defer handle.close(io);
var fs = storage.FileStorage.init(io, handle, file_size);
var counting = CountingStorage{ .inner = fs.storageFile() };
// Build the serve base snapshot over the counting storage and measure how
// many bytes priming read. `openReadOnly`'s scan already whole-frame-CRC'd
// every byte once (#15's concern); an index prime adds NOTHING on top, so
// total prime reads stay at roughly one pass over the log. The copy prime
// (`loadContentStore`) read every container a SECOND time — this bound fails
// for it, which is exactly the cost #14 removes.
var index = try LogIndexStore.open(gpa, counting.storageFile());
defer index.destroy();
try std.testing.expect(counting.read_bytes <= file_size + file_size / 2);
}
test "#14 log-backed index answers count/hashOf/held/content/wire identically to the MemStore snapshot" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const file_size = try seedServeRepo(gpa, io, tmp.dir);
// Reference: the whole-container MemStore snapshot `loadContentStore` builds.
var ref_handle = try tmp.dir.openFile(io, repo_filename, .{});
defer ref_handle.close(io);
var ref_fs = storage.FileStorage.init(io, ref_handle, file_size);
var ref_log = try logstore.LogStore.openReadOnly(gpa, ref_fs.storageFile());
defer ref_log.deinit(gpa);
var ref = repostate.MemStore.init(gpa);
defer ref.deinit();
try loadContentStore(gpa, &ref_log, &ref);
const rs = ref.store();
// Subject: the log-backed index over the same file.
var idx_handle = try tmp.dir.openFile(io, repo_filename, .{});
defer idx_handle.close(io);
var index = try LogIndexStore.openFromFile(gpa, io, idx_handle, file_size);
defer index.destroy();
const is = index.store();
try std.testing.expectEqual(rs.count(), is.count());
try std.testing.expect(rs.count() > 0);
var rid: repostate.Rid = 1;
while (rid <= rs.count()) : (rid += 1) {
// Identity, held-ness, and lookup round-trip.
try std.testing.expectEqualStrings(rs.hashOf(rid), is.hashOf(rid));
try std.testing.expectEqual(rs.held(rid), is.held(rid));
try std.testing.expectEqual(@as(?repostate.Rid, rid), is.lookup(rs.hashOf(rid)));
// Plaintext byte-for-byte.
const rc = rs.content(rid);
const ic = is.content(rid);
try std.testing.expectEqual(rc == null, ic == null);
if (rc) |rb| try std.testing.expectEqualSlices(u8, rb, ic.?);
// Wire form: container bytes, and the delta arm's DELTASRC + target size.
const rw = rs.wire(rid);
const iw = is.wire(rid);
try std.testing.expectEqual(rw == null, iw == null);
if (rw) |rwf| {
try std.testing.expectEqualSlices(u8, rwf.container, iw.?.container);
try std.testing.expectEqual(rwf.delta == null, iw.?.delta == null);
if (rwf.delta) |rd| {
try std.testing.expectEqualStrings(rd.src_hash, iw.?.delta.?.src_hash);
try std.testing.expectEqual(rd.target_size, iw.?.delta.?.target_size);
}
}
}
}
test "#14 appendix honours intern/put/retain for hashes the log does not carry" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const file_size = try seedServeRepo(gpa, io, tmp.dir);
var handle = try tmp.dir.openFile(io, repo_filename, .{});
defer handle.close(io);
var index = try LogIndexStore.openFromFile(gpa, io, handle, file_size);
defer index.destroy();
const s = index.store();
// The serve base is contractually read-only — every write lands in the
// per-request OverlayStore (ticket constraint 3), so these appendix arms never
// fire in production. They exist so the store honours the whole ArtifactStore
// interface instead of trapping on a write; exercise them directly here so that
// completeness is not dead-untested surface.
const persisted = s.count();
try std.testing.expect(persisted > 0);
// A hash the log cannot hold (all-0xAB digest, hex-named as the wire does).
var d1: [digest_len]u8 = undefined;
@memset(&d1, 0xAB);
var h1_buf: [2 * digest_len]u8 = undefined;
const h1 = digestToWireName(&d1, &h1_buf);
try std.testing.expectEqual(@as(?repostate.Rid, null), s.lookup(h1));
// intern mints an appendix phantom ABOVE the persisted rids: not held (no
// content yet), count grows by one, identity round-trips, and it is idempotent.
const r1 = try s.intern(h1);
try std.testing.expect(r1 > persisted);
try std.testing.expectEqual(r1, try s.intern(h1));
try std.testing.expectEqual(@as(?repostate.Rid, r1), s.lookup(h1));
try std.testing.expectEqualStrings(h1, s.hashOf(r1));
try std.testing.expect(!s.held(r1));
try std.testing.expectEqual(@as(?[]const u8, null), s.content(r1));
try std.testing.expectEqual(persisted + 1, s.count());
// put upgrades that phantom in place: held flips true, content reads back, and
// the persisted count is unchanged (an off-log hash never touches the index).
const body = "off-log body bytes";
try std.testing.expectEqual(r1, try s.put(h1, body));
try std.testing.expect(s.held(r1));
try std.testing.expectEqualSlices(u8, body, s.content(r1).?);
try std.testing.expectEqual(persisted + 1, s.count());
// retain on a PERSISTED rid is the advisory no-op (sync §7.2 first-form-wins;
// the log IS the retained form): it returns without error and ignores ingest.
try s.retain(1, .{ .full = "ignored — a persisted rid keeps its log form" });
// None of the appendix traffic left persisted-round memos resident.
try std.testing.expectEqual(@as(usize, 0), index.residentBytes());
}
test "#14 index materialises only the container a round reads and releases it between rounds" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const file_size = try seedServeRepo(gpa, io, tmp.dir);
var handle = try tmp.dir.openFile(io, repo_filename, .{});
defer handle.close(io);
var index = try LogIndexStore.openFromFile(gpa, io, handle, file_size);
defer index.destroy();
const s = index.store();
// Right after the prime the index holds ZERO container/plaintext bytes — the
// whole point: it did not copy the store into RAM.
try std.testing.expectEqual(@as(usize, 0), index.residentBytes());
// Find one wire-bearing rid (a zlib/delta frame).
var target: ?repostate.Rid = null;
var rid: repostate.Rid = 1;
while (rid <= s.count()) : (rid += 1) {
if (s.wire(rid) != null) {
target = rid;
break;
}
}
try std.testing.expect(target != null);
// Reading it materialised only its container; drop the round's memos.
try std.testing.expect(index.residentBytes() > 0);
index.releaseTransient();
try std.testing.expectEqual(@as(usize, 0), index.residentBytes());
// A single round's residency is EXACTLY the one container it emitted — not
// the whole store's containers (which the old snapshot held resident).
const one = s.wire(target.?).?;
try std.testing.expectEqual(one.container.len, index.residentBytes());
}
/// A test-only `storage.StorageFile` that reads faithfully until `fail_reads` is
/// flipped, then fails every `pread` — the instrument for a fault that strikes
/// AFTER `openReadOnly` has scanned and whole-frame-CRC'd the log, exactly the
/// post-scan bitrot / transient EIO the on-demand serve reads must not swallow
/// silently. Every other method delegates verbatim.
const FaultAfterScanStorage = struct {
inner: storage.StorageFile,
fail_reads: bool = false,
fn storageFile(self: *FaultAfterScanStorage) storage.StorageFile {
return .{ .context = self, .vtable = &vtable };
}
const vtable = storage.StorageFile.VTable{
.append = appendImpl,
.pread = preadImpl,
.sync = syncImpl,
.size = sizeImpl,
.truncate = truncateImpl,
};
fn appendImpl(ctx: *anyopaque, bytes: []const u8) storage.StorageError!void {
const self: *FaultAfterScanStorage = @ptrCast(@alignCast(ctx));
return self.inner.append(bytes);
}
fn preadImpl(ctx: *anyopaque, offset: u64, buf: []u8) storage.StorageError!usize {
const self: *FaultAfterScanStorage = @ptrCast(@alignCast(ctx));
if (self.fail_reads) return error.Io;
return self.inner.pread(offset, buf);
}
fn syncImpl(ctx: *anyopaque) storage.StorageError!void {
const self: *FaultAfterScanStorage = @ptrCast(@alignCast(ctx));
return self.inner.sync();
}
fn sizeImpl(ctx: *anyopaque) storage.StorageError!u64 {
const self: *FaultAfterScanStorage = @ptrCast(@alignCast(ctx));
return self.inner.size();
}
fn truncateImpl(ctx: *anyopaque, len: u64) storage.StorageError!void {
const self: *FaultAfterScanStorage = @ptrCast(@alignCast(ctx));
return self.inner.truncate(len);
}
};
test "#14 a post-scan read fault on a held rid is recorded, not silently swallowed" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const file_size = try seedServeRepo(gpa, io, tmp.dir);
var handle = try tmp.dir.openFile(io, repo_filename, .{});
defer handle.close(io);
var fs = storage.FileStorage.init(io, handle, file_size);
var faulting = FaultAfterScanStorage{ .inner = fs.storageFile() };
// Open succeeds: the scan CRCs the whole log BEFORE any fault is armed.
var index = try LogIndexStore.open(gpa, faulting.storageFile());
defer index.destroy();
const s = index.store();
// From here every on-demand container/plaintext read faults (post-scan EIO).
faulting.fail_reads = true;
// Walk the store as the §8.2 clone pager does: a held rid whose wire() AND
// content() both return null is one the pager silently skips while still
// reporting the sequence complete — a clone finishing MISSING an artifact.
var silently_omitted = false;
var rid: repostate.Rid = 1;
while (rid <= s.count()) : (rid += 1) {
if (!s.held(rid)) continue;
if (s.wire(rid) == null and s.content(rid) == null) silently_omitted = true;
}
// The fault genuinely makes a held rid unemittable (the bug's precondition)...
try std.testing.expect(silently_omitted);
// ...and the store RECORDED it, so the serve layer can convert the round to an
// error instead of completing a broken clone. A copy-based prime could not
// reach this state: a read fault surfaced at prime, not silently at serve.
try std.testing.expect(index.read_fault);
}
test "#14 serveAnswer converts a recorded read fault into a storage error (no silent short clone)" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
var hex_buf: [2 * digest_len]u8 = undefined;
try i9SeedRepo(gpa, io, tmp.dir, 4, &hex_buf);
const server = zit.sync.Server{
.version = serve_mod.server_version,
.self_servercode = serve_mod.server_servercode,
.project_code = serve_mod.server_projectcode,
};
var file = try tmp.dir.openFile(io, repo_filename, .{});
defer file.close(io);
var ctx = ServeCtx{ .io = io, .root_dir = tmp.dir, .file = file, .server = &server };
defer ctx.dropSnapshot();
const clone_plain = try std.fmt.allocPrint(gpa, "pragma client-version 20000\nclone 3 1\n", .{});
defer gpa.free(clone_plain);
const clone_req = try zit.container.encode(gpa, clone_plain, .default);
defer gpa.free(clone_req);
// A healthy round builds the cached snapshot and completes normally.
const ok = try serveAnswer(&ctx, gpa, clone_req);
gpa.free(ok.body);
// Simulate a post-scan read fault having struck THIS round — the test above
// proves the store records exactly this. serveAnswer MUST refuse to hand back
// a reply built while an artifact read faulted: otherwise the §8.2 pager's
// silent skip ships a clone missing an artifact and reports it complete.
ctx.snapshot.?.read_fault = true;
try std.testing.expectError(error.ServerStorage, serveAnswer(&ctx, gpa, clone_req));
// The flag is cleared with the round's memos, so an immediately retried round
// over the now-healthy snapshot succeeds again (a transient fault self-heals).
const recovered = try serveAnswer(&ctx, gpa, clone_req);
gpa.free(recovered.body);
}
/// Build a full-form 3-check-in chain over an evolving file into `log`, plus a
/// live config frame; returns the tip check-in's 64-hex name. Every blob and
/// manifest lands as a FULL frame — pack's re-delta target.
fn packSeedFullStore(gpa: std.mem.Allocator, log: *logstore.LogStore, tip_out: *[64]u8) !void {
try log.putConfig(.{ .hash_policy = 2, .default_branch = "trunk", .project_id = "pid", .project_code = "code-abc" });
var root_name: [64]u8 = undefined;
var mid_name: [64]u8 = undefined;
const v1 = try packEvolvingBlob(gpa, 1);
defer gpa.free(v1);
const b1 = try packSeedBlob(log, v1);
try packSeedCheckin(gpa, log, "2020-01-01T00:00:00", null, "f.txt", &b1, &root_name);
const v2 = try packEvolvingBlob(gpa, 2);
defer gpa.free(v2);
const b2 = try packSeedBlob(log, v2);
try packSeedCheckin(gpa, log, "2020-01-02T00:00:00", &root_name, "f.txt", &b2, &mid_name);
const v3 = try packEvolvingBlob(gpa, 3);
defer gpa.free(v3);
const b3 = try packSeedBlob(log, v3);
try packSeedCheckin(gpa, log, "2020-01-03T00:00:00", &mid_name, "f.txt", &b3, tip_out);
try log.commit();
}
/// Store `content` under its sha3-256 name, delta-encoded against `base` when
/// one is given (the evolving blobs delta to a fraction, so the frame lands as a
/// `.delta`), and return its rid. Models a fossil-clone store that RETAINED
/// compact wire deltas rather than full frames.
fn packSeedDeltaBlob(log: *logstore.LogStore, content: []const u8, base: ?logstore.Rid) !logstore.Rid {
var digest: [digest_len]u8 = undefined;
std.crypto.hash.sha3.Sha3_256.hash(content, &digest, .{});
return log.putArtifact(&digest, content, base);
}
/// Build a clone-shaped store whose successive blobs are RETAINED as `.delta`
/// frames (the compact wire form), under DISTINCT filenames so pack's
/// same-path-blob lineage hint finds NO base for them — exactly the case where
/// the old pack reconstructed every delta to a full frame and inflated the
/// store. `.zit` therefore holds real delta frames whose bases pack cannot
/// re-derive, so a correct pack must preserve them verbatim. Returns the tip
/// check-in's 64-hex name.
fn packSeedDeltaStore(gpa: std.mem.Allocator, log: *logstore.LogStore, tip_out: *[64]u8) !void {
try log.putConfig(.{ .hash_policy = 2, .default_branch = "trunk", .project_id = "pid", .project_code = "code-abc" });
var root_name: [64]u8 = undefined;
var mid_name: [64]u8 = undefined;
const v1 = try packEvolvingBlob(gpa, 1);
defer gpa.free(v1);
const r1 = try packSeedDeltaBlob(log, v1, null); // first version: full frame (base of the chain)
var b1: [64]u8 = undefined;
_ = zit.artifact.artifactName(.sha3_256, v1, &b1);
try packSeedCheckin(gpa, log, "2020-01-01T00:00:00", null, "f1.txt", &b1, &root_name);
const v2 = try packEvolvingBlob(gpa, 2);
defer gpa.free(v2);
const r2 = try packSeedDeltaBlob(log, v2, r1); // delta of v1
var b2: [64]u8 = undefined;
_ = zit.artifact.artifactName(.sha3_256, v2, &b2);
try packSeedCheckin(gpa, log, "2020-01-02T00:00:00", &root_name, "f2.txt", &b2, &mid_name);
const v3 = try packEvolvingBlob(gpa, 3);
defer gpa.free(v3);
_ = try packSeedDeltaBlob(log, v3, r2); // delta of v2
var b3: [64]u8 = undefined;
_ = zit.artifact.artifactName(.sha3_256, v3, &b3);
try packSeedCheckin(gpa, log, "2020-01-03T00:00:00", &mid_name, "f3.txt", &b3, tip_out);
try log.commit();
}
/// Seed a store holding a wire delta that pack CANNOT beat by re-deltification,
/// so ONLY the §6.1 monotonic-shrink guard keeps the store from regressing. `B`
/// is a one-line edit of `A`, stored as a COMPACT zit delta against `A` — the
/// smallest form pack could ever reproduce. But `max_base_candidates` unrelated
/// filler blobs sit between `A` and `B`, so pack's bounded backward candidate
/// window (§5.2) saturates on the fillers and never reaches `A`; with no lineage
/// hint either (bare blobs, no F-card path), pack's best re-encoding of `B` is a
/// full zlib frame STRICTLY LARGER than the stored A-based delta. The pre-guard
/// pack replaced the delta with that larger full frame (the +33% clone
/// regression in miniature); the guard keeps the existing smaller frame. Fills
/// A's and B's 64-hex names.
fn packSeedUnbeatableWireDelta(
gpa: std.mem.Allocator,
log: *logstore.LogStore,
a_name: *[64]u8,
b_name: *[64]u8,
) !void {
try log.putConfig(.{ .hash_policy = 2, .default_branch = "trunk", .project_id = "pid", .project_code = "code-abc" });
// A: B's ideal base, stored full (zlib).
const a = try packEvolvingBlob(gpa, 1);
defer gpa.free(a);
var a_digest: [digest_len]u8 = undefined;
std.crypto.hash.sha3.Sha3_256.hash(a, &a_digest, .{});
const a_rid = try log.putZlib(&a_digest, a);
_ = zit.artifact.artifactName(.sha3_256, a, a_name);
// `max_base_candidates` unrelated fillers between A and B: they saturate B's
// backward candidate window so A is out of reach when pack re-encodes B.
var f: usize = 0;
while (f < logstore.max_base_candidates) : (f += 1) {
var fbuf: std.ArrayList(u8) = .empty;
defer fbuf.deinit(gpa);
var i: usize = 0;
while (i < 200) : (i += 1) try fbuf.print(gpa, "filler {d} unrelated row {d}\n", .{ f, i });
var fdig: [digest_len]u8 = undefined;
std.crypto.hash.sha3.Sha3_256.hash(fbuf.items, &fdig, .{});
_ = try log.putZlib(&fdig, fbuf.items);
}
// B: a one-line edit of A, stored as a COMPACT zit delta against A (the
// smallest form pack could reproduce — and only against A, which pack's
// window cannot reach here).
const b = try packEvolvingBlob(gpa, 2);
defer gpa.free(b);
var b_digest: [digest_len]u8 = undefined;
std.crypto.hash.sha3.Sha3_256.hash(b, &b_digest, .{});
_ = zit.artifact.artifactName(.sha3_256, b, b_name);
const raw = try zit.delta.create(gpa, a, b);
defer gpa.free(raw);
const container = try zit.container.encode(gpa, raw, .default);
defer gpa.free(container);
_ = try log.putDeltaContainer(&b_digest, a_rid, container);
try log.commit();
}
/// Assert the §6.1 per-frame monotonicity: NO artifact's packed frame is larger
/// on disk than its source frame. Pack keeps the smaller of {existing frame, new
/// encoding} for every artifact, so a repack is non-worsening frame by frame.
/// This holds universally over THESE fixtures because their chains stay below the
/// read cap; the one path that can enlarge a frame — `copySrcFrame`'s §3.3
/// deep-chain fallback re-storing a kept delta as full zlib — needs a base at
/// `max_delta_chain` depth, which the seeds never build.
fn packExpectNoFrameLarger(gpa: std.mem.Allocator, src: *logstore.LogStore, dst: *logstore.LogStore) !void {
const src_rids = try src.ridsInPersistedOrder(gpa);
defer gpa.free(src_rids);
for (src_rids) |sr| {
const dr = dst.lookup(src.hashOf(sr)) orelse return error.PackedArtifactMissing;
try std.testing.expect(dst.frameSpanOf(dr) <= src.frameSpanOf(sr));
}
}
/// Count the `.delta` frames among a store's persisted artifacts.
fn packDeltaFrameCount(gpa: std.mem.Allocator, log: *logstore.LogStore) !usize {
const rids = try log.ridsInPersistedOrder(gpa);
defer gpa.free(rids);
var n: usize = 0;
for (rids) |r| {
if (log.kindOf(r) == .delta) n += 1;
}
return n;
}
/// Assert that every artifact in `src` reconstructs byte-identically in `dst`
/// and the two artifact SETS are equal (no drop, no addition).
fn packExpectContentEqual(gpa: std.mem.Allocator, src: *logstore.LogStore, dst: *logstore.LogStore) !void {
const src_rids = try src.ridsInPersistedOrder(gpa);
defer gpa.free(src_rids);
const dst_rids = try dst.ridsInPersistedOrder(gpa);
defer gpa.free(dst_rids);
try std.testing.expectEqual(src_rids.len, dst_rids.len);
for (src_rids) |sr| {
const digest = src.hashOf(sr);
const dr = dst.lookup(digest) orelse return error.PackedArtifactMissing;
const a = (try src.content(gpa, sr)).?;
defer gpa.free(a);
const b = (try dst.content(gpa, dr)).?;
defer gpa.free(b);
try std.testing.expectEqualSlices(u8, a, b);
}
}
test "pack: the packed store's artifact set + every reconstructed content byte-equal the original (config preserved)" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
var tip: [64]u8 = undefined;
{
var wfs: storage.FileStorage = undefined;
var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
defer wfs.file.close(io);
defer wlog.deinit(gpa);
try packSeedFullStore(gpa, &wlog, &tip);
}
var sfs: storage.FileStorage = undefined;
var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
defer sfs.file.close(io);
defer src.deinit(gpa);
var dfile = try tmp.dir.createFile(io, "packed", .{ .read = true, .truncate = true });
defer dfile.close(io);
var dfs = storage.FileStorage.init(io, dfile, 0);
var dst = try logstore.LogStore.open(gpa, dfs.storageFile(), src.hash_kind);
defer dst.deinit(gpa);
const stats = try packLog(gpa, &src, &dst);
// No drop: the packed content-frame count equals the source's, and every
// artifact reconstructs byte-for-byte.
try std.testing.expectEqual(src.entries.items.len, stats.artifacts);
try packExpectContentEqual(gpa, &src, &dst);
// The live config survives as exactly one frame with identical fields.
var soc = (try src.readConfig(gpa)).?;
defer soc.deinit(gpa);
var doc = (try dst.readConfig(gpa)).?;
defer doc.deinit(gpa);
try std.testing.expectEqual(soc.config.hash_policy, doc.config.hash_policy);
try std.testing.expectEqualStrings(soc.config.default_branch, doc.config.default_branch);
try std.testing.expectEqualStrings(soc.config.project_id, doc.config.project_id);
try std.testing.expectEqualStrings(soc.config.project_code, doc.config.project_code);
}
test "pack re-deltas full-form frames (delta count rises, .zit shrinks) with byte-identical content" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
var tip: [64]u8 = undefined;
{
var wfs: storage.FileStorage = undefined;
var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
defer wfs.file.close(io);
defer wlog.deinit(gpa);
try packSeedFullStore(gpa, &wlog, &tip);
}
var sfs: storage.FileStorage = undefined;
var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
defer sfs.file.close(io);
defer src.deinit(gpa);
// The source is all full-form: not one delta frame.
{
const rids = try src.ridsInPersistedOrder(gpa);
defer gpa.free(rids);
for (rids) |r| try std.testing.expect(src.kindOf(r) != .delta);
}
var dfile = try tmp.dir.createFile(io, "packed", .{ .read = true, .truncate = true });
defer dfile.close(io);
var dfs = storage.FileStorage.init(io, dfile, 0);
var dst = try logstore.LogStore.open(gpa, dfs.storageFile(), src.hash_kind);
defer dst.deinit(gpa);
const stats = try packLog(gpa, &src, &dst);
// The two child blobs (v2 against v1, v3 against v2) re-delta against the
// same path's parent blob, so the delta count rises and the store shrinks —
// with content preserved exactly.
try std.testing.expect(stats.deltas >= 2);
try std.testing.expect(stats.bytes_after < stats.bytes_before);
try packExpectContentEqual(gpa, &src, &dst);
}
test "pack preserves wire deltas — a store with delta frames packs to <= source size (regression for the 8.65x inflation bug)" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
var tip: [64]u8 = undefined;
{
var wfs: storage.FileStorage = undefined;
var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
defer wfs.file.close(io);
defer wlog.deinit(gpa);
try packSeedDeltaStore(gpa, &wlog, &tip);
}
var sfs: storage.FileStorage = undefined;
var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
defer sfs.file.close(io);
defer src.deinit(gpa);
// The source genuinely holds retained wire deltas (the compact form pack
// must not discard) whose bases pack's lineage hint cannot re-derive.
const src_deltas = try packDeltaFrameCount(gpa, &src);
try std.testing.expect(src_deltas >= 2);
var dfile = try tmp.dir.createFile(io, "packed", .{ .read = true, .truncate = true });
defer dfile.close(io);
var dfs = storage.FileStorage.init(io, dfile, 0);
var dst = try logstore.LogStore.open(gpa, dfs.storageFile(), src.hash_kind);
defer dst.deinit(gpa);
const stats = try packLog(gpa, &src, &dst);
// MONOTONIC: the packed store is never larger than the source. The old pack
// reconstructed each delta to a full zlib frame and inflated the store; a
// correct pack preserves the compact deltas verbatim.
try std.testing.expect(stats.bytes_after <= stats.bytes_before);
// Every source delta survives AS a delta frame — none reconstructed to full
// (the optional re-delta may add further deltas, so the total only grows).
{
const rids = try src.ridsInPersistedOrder(gpa);
defer gpa.free(rids);
for (rids) |sr| {
if (src.kindOf(sr) != .delta) continue;
const dr = dst.lookup(src.hashOf(sr)).?;
try std.testing.expectEqual(logstore.FrameKind.delta, dst.kindOf(dr));
}
}
try std.testing.expect(try packDeltaFrameCount(gpa, &dst) >= src_deltas);
// No drop, every reconstruction byte-exact.
try packExpectContentEqual(gpa, &src, &dst);
}
test "pack copies a delta frame verbatim re-linked to its base's new rid, content byte-identical" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
var tip: [64]u8 = undefined;
{
var wfs: storage.FileStorage = undefined;
var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
defer wfs.file.close(io);
defer wlog.deinit(gpa);
try packSeedDeltaStore(gpa, &wlog, &tip);
}
var sfs: storage.FileStorage = undefined;
var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
defer sfs.file.close(io);
defer src.deinit(gpa);
var dfile = try tmp.dir.createFile(io, "packed", .{ .read = true, .truncate = true });
defer dfile.close(io);
var dfs = storage.FileStorage.init(io, dfile, 0);
var dst = try logstore.LogStore.open(gpa, dfs.storageFile(), src.hash_kind);
defer dst.deinit(gpa);
_ = try packLog(gpa, &src, &dst);
// For every delta frame in the source, the packed store holds it as a delta
// whose stored container is byte-identical (verbatim re-serve) and whose base
// names the SAME artifact (re-linked to the base's new rid). Content matches.
const rids = try src.ridsInPersistedOrder(gpa);
defer gpa.free(rids);
var checked: usize = 0;
for (rids) |sr| {
if (src.kindOf(sr) != .delta) continue;
const digest = src.hashOf(sr);
const dr = dst.lookup(digest).?;
try std.testing.expectEqual(logstore.FrameKind.delta, dst.kindOf(dr));
var sw = (try src.wire(gpa, sr)).?;
defer sw.deinit(gpa);
var dw = (try dst.wire(gpa, dr)).?;
defer dw.deinit(gpa);
// Verbatim container bytes and the same base hash (re-link preserves the
// logical base while its rid/offset differs in the packed generation).
try std.testing.expectEqualSlices(u8, sw.delta.container, dw.delta.container);
try std.testing.expectEqualSlices(u8, sw.delta.baseHash(), dw.delta.baseHash());
const a = (try src.content(gpa, sr)).?;
defer gpa.free(a);
const b = (try dst.content(gpa, dr)).?;
defer gpa.free(b);
try std.testing.expectEqualSlices(u8, a, b);
checked += 1;
}
try std.testing.expect(checked >= 2);
}
/// The rid naming the 64-hex `name` in `log`, or null when absent (test-only
/// name→digest→rid convenience over `lookup`).
fn packRidByName(log: *logstore.LogStore, name: *const [64]u8) ?logstore.Rid {
const d = hexToDigest(name) orelse return null;
return log.lookup(&d);
}
/// Seed a store modeling a peer-supplied (verbatim wire) delta whose sender
/// chose a POOR base. `A` (a large file) and `C` (an unrelated large blob) are
/// stored full; `B` (a one-line edit of `A`) is stored as a hand-crafted §4.1
/// delta container against `C` — a large, suboptimal delta the sender picked.
/// `A` precedes `B` chronologically and is B's ideal base, so pack's best-of-N
/// re-deltification (§6.1) must re-base `B` onto `A` rather than trust the
/// sender's `C`. Fills the 64-hex names of A/B/C.
fn packSeedSuboptimalWireDelta(
gpa: std.mem.Allocator,
log: *logstore.LogStore,
a_name: *[64]u8,
b_name: *[64]u8,
c_name: *[64]u8,
) !void {
try log.putConfig(.{ .hash_policy = 2, .default_branch = "trunk", .project_id = "pid", .project_code = "code-abc" });
// A: the ideal base for B (near-identical predecessor), stored full.
const a = try packEvolvingBlob(gpa, 1);
defer gpa.free(a);
a_name.* = try packSeedBlob(log, a);
// C: an unrelated large blob (a poor base for B), stored full.
var cbuf: std.ArrayList(u8) = .empty;
defer cbuf.deinit(gpa);
var i: usize = 0;
while (i < 200) : (i += 1) try cbuf.print(gpa, "unrelated content row {d}\n", .{i});
const c = cbuf.items;
c_name.* = try packSeedBlob(log, c);
const c_rid = packRidByName(log, c_name).?;
// B: a one-line edit of A. delta(A,B) is tiny; delta(C,B) is large.
const b = try packEvolvingBlob(gpa, 2);
defer gpa.free(b);
_ = zit.artifact.artifactName(.sha3_256, b, b_name);
var b_digest: [digest_len]u8 = undefined;
std.crypto.hash.sha3.Sha3_256.hash(b, &b_digest, .{});
// Hand-craft the verbatim wire delta against the POOR base C — models a peer
// that chose C as base, retained at rest by `putDeltaContainer`.
const raw = try zit.delta.create(gpa, c, b);
defer gpa.free(raw);
const container = try zit.container.encode(gpa, raw, .default);
defer gpa.free(container);
_ = try log.putDeltaContainer(&b_digest, c_rid, container);
try log.commit();
}
test "pack §6.1: a verbatim wire delta is re-deltified against the size-minimizing base, not the sender's" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
var a_name: [64]u8 = undefined;
var b_name: [64]u8 = undefined;
var c_name: [64]u8 = undefined;
{
var wfs: storage.FileStorage = undefined;
var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
defer wfs.file.close(io);
defer wlog.deinit(gpa);
try packSeedSuboptimalWireDelta(gpa, &wlog, &a_name, &b_name, &c_name);
}
var sfs: storage.FileStorage = undefined;
var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
defer sfs.file.close(io);
defer src.deinit(gpa);
// Precondition: the source really holds B as a delta whose base is C (the
// sender's suboptimal choice) — the exact frame the old pack copied verbatim.
{
const sb = packRidByName(&src, &b_name).?;
try std.testing.expectEqual(logstore.FrameKind.delta, src.kindOf(sb));
var sw = (try src.wire(gpa, sb)).?;
defer sw.deinit(gpa);
var cbin: [digest_len]u8 = undefined;
_ = try std.fmt.hexToBytes(&cbin, &c_name);
try std.testing.expectEqualSlices(u8, &cbin, sw.delta.baseHash());
}
var dfile = try tmp.dir.createFile(io, "packed", .{ .read = true, .truncate = true });
defer dfile.close(io);
var dfs = storage.FileStorage.init(io, dfile, 0);
var dst = try logstore.LogStore.open(gpa, dfs.storageFile(), src.hash_kind);
defer dst.deinit(gpa);
const stats = try packLog(gpa, &src, &dst);
// After pack, B is still a delta, but its base is now A (the size-minimizing
// candidate) — the sender's base C is no longer trusted at rest (§6.1/§3).
const db = packRidByName(&dst, &b_name).?;
try std.testing.expectEqual(logstore.FrameKind.delta, dst.kindOf(db));
var dw = (try dst.wire(gpa, db)).?;
defer dw.deinit(gpa);
var abin: [digest_len]u8 = undefined;
_ = try std.fmt.hexToBytes(&abin, &a_name);
try std.testing.expectEqualSlices(u8, &abin, dw.delta.baseHash());
// Every artifact still reconstructs byte-for-byte and the store shrank (a
// tiny A-based delta replaced the large C-based one).
try packExpectContentEqual(gpa, &src, &dst);
try std.testing.expect(stats.bytes_after < stats.bytes_before);
try std.testing.expectEqual(dst.size(), stats.bytes_after);
try std.testing.expectEqual(src.size(), stats.bytes_before);
}
/// Seed the §6.2 same-path scenario: two versions of ONE tracked file whose ideal
/// base (the earlier version) sits OUTSIDE pack's bounded chronological window.
/// `v1` of "tracked.txt" is stored full, then `max_base_candidates` unrelated
/// filler blobs saturate the backward candidate window, then `v2` (a one-line
/// edit of `v1`) is stored full. With ONLY the chronological window (step §6.1)
/// pack cannot reach `v1` from `v2` — the fillers fill every slot — so `v2` stays
/// a full frame and the store does not shrink. The mlink-analog same-path walk
/// (§6.2) reads both check-in manifests, learns tracked.txt: [v1, v2], and offers
/// `v1` as `v2`'s priority candidate, so pack re-deltifies `v2` onto `v1`. Both
/// check-ins are FULL frames (pack's re-delta target); fills v1/v2's 64-hex names.
fn packSeedSamePathOutOfWindow(
gpa: std.mem.Allocator,
log: *logstore.LogStore,
v1_name: *[64]u8,
v2_name: *[64]u8,
) !void {
try log.putConfig(.{ .hash_policy = 2, .default_branch = "trunk", .project_id = "pid", .project_code = "code-abc" });
// v1 of tracked.txt (full), and the root check-in that names it.
const v1 = try packEvolvingBlob(gpa, 1);
defer gpa.free(v1);
v1_name.* = try packSeedBlob(log, v1);
var root_name: [64]u8 = undefined;
try packSeedCheckin(gpa, log, "2020-01-01T00:00:00", null, "tracked.txt", v1_name, &root_name);
// `max_base_candidates` unrelated fillers between v1 and v2: they saturate
// v2's bounded backward window so v1 is out of chronological reach.
var f: usize = 0;
while (f < logstore.max_base_candidates) : (f += 1) {
var fbuf: std.ArrayList(u8) = .empty;
defer fbuf.deinit(gpa);
var i: usize = 0;
while (i < 200) : (i += 1) try fbuf.print(gpa, "samepath filler {d} unrelated row {d}\n", .{ f, i });
_ = try packSeedBlob(log, fbuf.items);
}
// v2 of tracked.txt (a one-line edit of v1), stored FULL, and its check-in.
const v2 = try packEvolvingBlob(gpa, 2);
defer gpa.free(v2);
v2_name.* = try packSeedBlob(log, v2);
var tip_name: [64]u8 = undefined;
try packSeedCheckin(gpa, log, "2020-01-02T00:00:00", &root_name, "tracked.txt", v2_name, &tip_name);
try log.commit();
}
test "pack §6.2: a same-path predecessor beyond the chronological window becomes v2's base (store shrinks)" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
var v1_name: [64]u8 = undefined;
var v2_name: [64]u8 = undefined;
{
var wfs: storage.FileStorage = undefined;
var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
defer wfs.file.close(io);
defer wlog.deinit(gpa);
try packSeedSamePathOutOfWindow(gpa, &wlog, &v1_name, &v2_name);
}
var sfs: storage.FileStorage = undefined;
var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
defer sfs.file.close(io);
defer src.deinit(gpa);
// Precondition: the source holds v2 as a NON-delta (full) frame — the case
// step §6.1's chronological window alone cannot improve (v1 is out of reach).
{
const sv2 = packRidByName(&src, &v2_name).?;
try std.testing.expect(src.kindOf(sv2) != .delta);
}
var dfile = try tmp.dir.createFile(io, "packed", .{ .read = true, .truncate = true });
defer dfile.close(io);
var dfs = storage.FileStorage.init(io, dfile, 0);
var dst = try logstore.LogStore.open(gpa, dfs.storageFile(), src.hash_kind);
defer dst.deinit(gpa);
const stats = try packLog(gpa, &src, &dst);
// After pack, v2 is a delta whose base is the same-path predecessor v1 — the
// §6.2 mlink analog reached a base the chronological window could not.
const dv2 = packRidByName(&dst, &v2_name).?;
try std.testing.expectEqual(logstore.FrameKind.delta, dst.kindOf(dv2));
var dw = (try dst.wire(gpa, dv2)).?;
defer dw.deinit(gpa);
var v1bin: [digest_len]u8 = undefined;
_ = try std.fmt.hexToBytes(&v1bin, &v1_name);
try std.testing.expectEqualSlices(u8, &v1bin, dw.delta.baseHash());
// The store shrank (a tiny v1-based delta replaced the full v2 frame), every
// artifact still reconstructs byte-for-byte, and no frame grew.
try std.testing.expect(stats.bytes_after < stats.bytes_before);
try packExpectContentEqual(gpa, &src, &dst);
try packExpectNoFrameLarger(gpa, &src, &dst);
}
/// Seed the §6.2 same-path REVERT scenario: one tracked file whose content goes
/// A→B→A across three check-ins — the third restores a byte-identical earlier
/// blob. Because the store is content-addressed, the reverted version resolves to
/// the SAME blob rid as v1, so the pack manifest walk records that rid twice,
/// non-consecutively, for one path. Exercises `deriveCandidateHints`'s
/// distinct-rid compaction: a blob must never be offered as its own same-path
/// predecessor. Fills v1/v2's 64-hex names.
fn packSeedSamePathRevert(
gpa: std.mem.Allocator,
log: *logstore.LogStore,
v1_name: *[64]u8,
v2_name: *[64]u8,
) !void {
try log.putConfig(.{ .hash_policy = 2, .default_branch = "trunk", .project_id = "pid", .project_code = "code-abc" });
const v1 = try packEvolvingBlob(gpa, 1);
defer gpa.free(v1);
const v2 = try packEvolvingBlob(gpa, 2);
defer gpa.free(v2);
// v1 of tracked.txt (full) + its root check-in.
v1_name.* = try packSeedBlob(log, v1);
var c1: [64]u8 = undefined;
try packSeedCheckin(gpa, log, "2020-01-01T00:00:00", null, "tracked.txt", v1_name, &c1);
// v2 (a one-line edit) + its check-in.
v2_name.* = try packSeedBlob(log, v2);
var c2: [64]u8 = undefined;
try packSeedCheckin(gpa, log, "2020-01-02T00:00:00", &c1, "tracked.txt", v2_name, &c2);
// Revert: tracked.txt returns to v1's byte-identical content. The blob is
// already stored (content-addressed), so this check-in re-references v1's rid
// — the non-consecutive duplicate that recordPathBlob's consecutive-only dedup
// lets through.
var c3: [64]u8 = undefined;
try packSeedCheckin(gpa, log, "2020-01-03T00:00:00", &c2, "tracked.txt", v1_name, &c3);
try log.commit();
}
test "pack §6.2: a same-path revert (A→B→A) re-references one blob rid twice and packs cleanly (no self-predecessor panic)" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
var v1_name: [64]u8 = undefined;
var v2_name: [64]u8 = undefined;
{
var wfs: storage.FileStorage = undefined;
var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
defer wfs.file.close(io);
defer wlog.deinit(gpa);
try packSeedSamePathRevert(gpa, &wlog, &v1_name, &v2_name);
}
var sfs: storage.FileStorage = undefined;
var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
defer sfs.file.close(io);
defer src.deinit(gpa);
var dfile = try tmp.dir.createFile(io, "packed", .{ .read = true, .truncate = true });
defer dfile.close(io);
var dfs = storage.FileStorage.init(io, dfile, 0);
var dst = try logstore.LogStore.open(gpa, dfs.storageFile(), src.hash_kind);
defer dst.deinit(gpa);
// Before the distinct-rid compaction in deriveCandidateHints this panicked on
// a self-predecessor (v1 sorted adjacent to its own duplicate ⇒ equal offsets
// fail the strict offset-below assert). Now the pack completes and every
// artifact reconstructs byte-for-byte with no frame enlarged.
const stats = try packLog(gpa, &src, &dst);
try packExpectContentEqual(gpa, &src, &dst);
try packExpectNoFrameLarger(gpa, &src, &dst);
try std.testing.expectEqual(dst.size(), stats.bytes_after);
try std.testing.expectEqual(src.size(), stats.bytes_before);
}
test "pack §6.1 monotonic guard: a wire delta pack cannot beat is KEPT, not replaced by a larger frame (no regression)" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
var a_name: [64]u8 = undefined;
var b_name: [64]u8 = undefined;
{
var wfs: storage.FileStorage = undefined;
var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
defer wfs.file.close(io);
defer wlog.deinit(gpa);
try packSeedUnbeatableWireDelta(gpa, &wlog, &a_name, &b_name);
}
var sfs: storage.FileStorage = undefined;
var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
defer sfs.file.close(io);
defer src.deinit(gpa);
// Precondition: the source holds B as a compact delta whose base is A.
{
const sb = packRidByName(&src, &b_name).?;
try std.testing.expectEqual(logstore.FrameKind.delta, src.kindOf(sb));
var sw = (try src.wire(gpa, sb)).?;
defer sw.deinit(gpa);
var abin: [digest_len]u8 = undefined;
_ = try std.fmt.hexToBytes(&abin, &a_name);
try std.testing.expectEqualSlices(u8, &abin, sw.delta.baseHash());
}
var dfile = try tmp.dir.createFile(io, "packed", .{ .read = true, .truncate = true });
defer dfile.close(io);
var dfs = storage.FileStorage.init(io, dfile, 0);
var dst = try logstore.LogStore.open(gpa, dfs.storageFile(), src.hash_kind);
defer dst.deinit(gpa);
const stats = try packLog(gpa, &src, &dst);
// (a) decode-all byte-identical pre/post.
try packExpectContentEqual(gpa, &src, &dst);
// (b) NO frame is larger post-pack than its pre-pack encoding.
try packExpectNoFrameLarger(gpa, &src, &dst);
// (c) the whole store did not regress on this wire-delta-heavy fixture.
try std.testing.expect(stats.bytes_after <= stats.bytes_before);
// Accounting exact: reported sizes are the real store sizes.
try std.testing.expectEqual(src.size(), stats.bytes_before);
try std.testing.expectEqual(dst.size(), stats.bytes_after);
// B stayed a delta (the compact A-based frame was kept, re-linked to A's new
// rid) — the guard refused to replace it with the larger full zlib pack's
// window-only re-encoding would have produced.
const db = packRidByName(&dst, &b_name).?;
try std.testing.expectEqual(logstore.FrameKind.delta, dst.kindOf(db));
var dw = (try dst.wire(gpa, db)).?;
defer dw.deinit(gpa);
var abin: [digest_len]u8 = undefined;
_ = try std.fmt.hexToBytes(&abin, &a_name);
try std.testing.expectEqualSlices(u8, &abin, dw.delta.baseHash());
// The kept frame is no larger than the source's (equal — copied verbatim).
try std.testing.expect(dst.frameSpanOf(db) <= src.frameSpanOf(packRidByName(&src, &b_name).?));
}
/// Seed a store exercising EVERY frame class pack must handle in one place: a
/// raw frame, a zlib frame, a locally-created delta, a verbatim wire delta, an
/// interned phantom (no frame on disk), and the config singleton. Fills the
/// content-bearing artifacts' 64-hex names plus the phantom's binary digest.
fn packSeedMixedStore(
gpa: std.mem.Allocator,
log: *logstore.LogStore,
raw_name: *[64]u8,
zlib_name: *[64]u8,
delta_name: *[64]u8,
wire_name: *[64]u8,
phantom_digest: *[digest_len]u8,
) !void {
try log.putConfig(.{ .hash_policy = 2, .default_branch = "trunk", .project_id = "pid", .project_code = "code-abc" });
// raw: a unique blob large enough to be a real (if unrelated) base.
var rbuf: std.ArrayList(u8) = .empty;
defer rbuf.deinit(gpa);
var i: usize = 0;
while (i < 120) : (i += 1) try rbuf.print(gpa, "raw-only content row {d}\n", .{i});
const rbytes = rbuf.items;
raw_name.* = try packSeedBlob(log, rbytes); // `put` => .raw
// zlib: a large evolving blob stored as a full zlib container.
const z = try packEvolvingBlob(gpa, 1);
defer gpa.free(z);
var z_digest: [digest_len]u8 = undefined;
std.crypto.hash.sha3.Sha3_256.hash(z, &z_digest, .{});
const z_rid = try log.putZlib(&z_digest, z);
_ = zit.artifact.artifactName(.sha3_256, z, zlib_name);
// local delta: an edit of the zlib blob, delta-encoded against it.
const d = try packEvolvingBlob(gpa, 2);
defer gpa.free(d);
var d_digest: [digest_len]u8 = undefined;
std.crypto.hash.sha3.Sha3_256.hash(d, &d_digest, .{});
_ = try log.putArtifact(&d_digest, d, z_rid);
_ = zit.artifact.artifactName(.sha3_256, d, delta_name);
// wire delta: a hand-crafted §4.1 delta container against the raw blob (a
// poor base) — the verbatim-retained peer form.
const w = try packEvolvingBlob(gpa, 3);
defer gpa.free(w);
var w_digest: [digest_len]u8 = undefined;
std.crypto.hash.sha3.Sha3_256.hash(w, &w_digest, .{});
_ = zit.artifact.artifactName(.sha3_256, w, wire_name);
const r_rid = packRidByName(log, raw_name).?;
const raw_delta = try zit.delta.create(gpa, rbytes, w);
defer gpa.free(raw_delta);
const container = try zit.container.encode(gpa, raw_delta, .default);
defer gpa.free(container);
_ = try log.putDeltaContainer(&w_digest, r_rid, container);
// phantom: an interned forward reference that never reaches disk.
std.crypto.hash.sha3.Sha3_256.hash("absent forward-referenced artifact", phantom_digest, .{});
_ = try log.intern(phantom_digest);
try log.commit();
}
test "pack §6.1/§3: a mixed raw/zlib/delta/wire-delta/phantom/config store round-trips, config+phantom preserved, accounting exact" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
var raw_name: [64]u8 = undefined;
var zlib_name: [64]u8 = undefined;
var delta_name: [64]u8 = undefined;
var wire_name: [64]u8 = undefined;
var phantom_digest: [digest_len]u8 = undefined;
// A phantom never reaches disk, so we pack the in-memory writable store
// directly (a reopen would drop it) — the case pack must tolerate without
// choking or emitting a garbage frame.
var wfs: storage.FileStorage = undefined;
var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
defer wfs.file.close(io);
defer wlog.deinit(gpa);
try packSeedMixedStore(gpa, &wlog, &raw_name, &zlib_name, &delta_name, &wire_name, &phantom_digest);
// The seeded store really holds one frame of each class plus the phantom.
try std.testing.expectEqual(logstore.FrameKind.raw, wlog.kindOf(packRidByName(&wlog, &raw_name).?));
try std.testing.expectEqual(logstore.FrameKind.zlib, wlog.kindOf(packRidByName(&wlog, &zlib_name).?));
try std.testing.expectEqual(logstore.FrameKind.delta, wlog.kindOf(packRidByName(&wlog, &delta_name).?));
try std.testing.expectEqual(logstore.FrameKind.delta, wlog.kindOf(packRidByName(&wlog, &wire_name).?));
try std.testing.expect(wlog.isPhantom(wlog.lookup(&phantom_digest).?));
var dfile = try tmp.dir.createFile(io, "packed", .{ .read = true, .truncate = true });
defer dfile.close(io);
var dfs = storage.FileStorage.init(io, dfile, 0);
var dst = try logstore.LogStore.open(gpa, dfs.storageFile(), wlog.hash_kind);
defer dst.deinit(gpa);
const stats = try packLog(gpa, &wlog, &dst);
// Decode-all byte-identical pre/post, and the artifact SETS match (the
// phantom is excluded from both, so counts still agree).
try packExpectContentEqual(gpa, &wlog, &dst);
// §6.1 per-frame monotonicity: no frame grew in the repack.
try packExpectNoFrameLarger(gpa, &wlog, &dst);
// Config preserved verbatim in fields.
var soc = (try wlog.readConfig(gpa)).?;
defer soc.deinit(gpa);
var doc = (try dst.readConfig(gpa)).?;
defer doc.deinit(gpa);
try std.testing.expectEqual(soc.config.hash_policy, doc.config.hash_policy);
try std.testing.expectEqualStrings(soc.config.default_branch, doc.config.default_branch);
try std.testing.expectEqualStrings(soc.config.project_id, doc.config.project_id);
try std.testing.expectEqualStrings(soc.config.project_code, doc.config.project_code);
// Phantom preserved as a contentless forward reference: it carries no bytes,
// so pack writes no frame for it (a phantom never reaches disk) and it is not
// resurrected as a garbage artifact in the packed store.
try std.testing.expect(dst.lookup(&phantom_digest) == null);
// Accounting exact: the reported sizes are the actual store sizes and the
// artifact tally is the source's real (non-phantom) frame count.
try std.testing.expectEqual(wlog.size(), stats.bytes_before);
try std.testing.expectEqual(dst.size(), stats.bytes_after);
const src_rids = try wlog.ridsInPersistedOrder(gpa);
defer gpa.free(src_rids);
try std.testing.expectEqual(src_rids.len, stats.artifacts);
}
test "pack §7.7: packLog aborts on a source frame whose content does not name it (no packed frame written)" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
// A frame keyed by a name that is NOT sha3-256 of its bytes: `put` stores
// bytes verbatim under any key, so this models a corrupt/lying source frame
// that survives `openReadOnly` (its CRC + embedded-hash label are intact).
const lying_name = [_]u8{0xab} ** digest_len;
{
var wfs: storage.FileStorage = undefined;
var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
defer wfs.file.close(io);
defer wlog.deinit(gpa);
_ = try wlog.put(&lying_name, "content these bytes do not name");
try wlog.commit();
}
var sfs: storage.FileStorage = undefined;
var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
defer sfs.file.close(io);
defer src.deinit(gpa);
var dfile = try tmp.dir.createFile(io, "packed", .{ .read = true, .truncate = true });
defer dfile.close(io);
var dfs = storage.FileStorage.init(io, dfile, 0);
var dst = try logstore.LogStore.open(gpa, dfs.storageFile(), src.hash_kind);
defer dst.deinit(gpa);
try std.testing.expectError(error.Corrupt, packLog(gpa, &src, &dst));
try std.testing.expect(dst.lookup(&lying_name) == null); // no frame landed on the abort
}
test "pack is idempotent: pack∘pack is byte-identical (0 additional re-deltas)" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
var tip: [64]u8 = undefined;
{
var wfs: storage.FileStorage = undefined;
var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
defer wfs.file.close(io);
defer wlog.deinit(gpa);
try packSeedFullStore(gpa, &wlog, &tip);
}
// First pack: source (full) → L1.
{
var sfs: storage.FileStorage = undefined;
var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
defer sfs.file.close(io);
defer src.deinit(gpa);
var f1 = try tmp.dir.createFile(io, "pack1", .{ .read = true, .truncate = true });
defer f1.close(io);
var f1fs = storage.FileStorage.init(io, f1, 0);
var dst = try logstore.LogStore.open(gpa, f1fs.storageFile(), src.hash_kind);
defer dst.deinit(gpa);
_ = try packLog(gpa, &src, &dst);
}
// Second pack: L1 → L2, byte-for-byte equal, and no further re-deltaing.
var l1fs: storage.FileStorage = undefined;
var l1 = try packOpenReadOnly2(gpa, io, tmp.dir, "pack1", &l1fs);
defer l1fs.file.close(io);
defer l1.deinit(gpa);
var f2 = try tmp.dir.createFile(io, "pack2", .{ .read = true, .truncate = true });
defer f2.close(io);
var f2fs = storage.FileStorage.init(io, f2, 0);
var l2 = try logstore.LogStore.open(gpa, f2fs.storageFile(), l1.hash_kind);
defer l2.deinit(gpa);
const stats2 = try packLog(gpa, &l1, &l2);
try std.testing.expectEqual(l1.size(), l2.size());
const a = try tmp.dir.readFileAlloc(io, "pack1", gpa, .limited(1 << 24));
defer gpa.free(a);
const b = try tmp.dir.readFileAlloc(io, "pack2", gpa, .limited(1 << 24));
defer gpa.free(b);
try std.testing.expectEqualSlices(u8, a, b); // byte-identical: pack∘pack == pack
// An already-packed store still reports its deltas (they were re-created
// identically), and packing it does not shrink further.
try std.testing.expectEqual(l1.size(), stats2.bytes_after);
}
/// Reopen a named log file in `dir` read-only (a sibling to `packOpenReadOnly`
/// that takes an arbitrary name, for the multi-generation idempotence test).
fn packOpenReadOnly2(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir, name: []const u8, fs_out: *storage.FileStorage) !logstore.LogStore {
const f = try dir.openFile(io, name, .{});
const st = try f.stat(io);
fs_out.* = storage.FileStorage.init(io, f, st.size);
return logstore.LogStore.openReadOnly(gpa, fs_out.storageFile());
}
test "pack: packLog never writes the source (a crash before the rename leaves the original byte-identical)" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
var tip: [64]u8 = undefined;
{
var wfs: storage.FileStorage = undefined;
var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
defer wfs.file.close(io);
defer wlog.deinit(gpa);
try packSeedFullStore(gpa, &wlog, &tip);
}
const before = try tmp.dir.readFileAlloc(io, repo_filename, gpa, .limited(1 << 24));
defer gpa.free(before);
// Pack into a separate temp (the new generation). packLog only ever reads
// the source; the atomic swap in `runPack` is the sole mutation of `.zit`.
{
var sfs: storage.FileStorage = undefined;
var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
defer sfs.file.close(io);
defer src.deinit(gpa);
var f = try tmp.dir.createFile(io, pack_tmp_filename, .{ .read = true, .truncate = true });
defer f.close(io);
var ffs = storage.FileStorage.init(io, f, 0);
var dst = try logstore.LogStore.open(gpa, ffs.storageFile(), src.hash_kind);
defer dst.deinit(gpa);
_ = try packLog(gpa, &src, &dst);
}
// No rename happened (the "crash"): the original `.zit` is byte-for-byte
// unchanged and still opens.
const after = try tmp.dir.readFileAlloc(io, repo_filename, gpa, .limited(1 << 24));
defer gpa.free(after);
try std.testing.expectEqualSlices(u8, before, after);
}
/// The allocation-failure probe: pack a prebuilt source `.zit` into a fresh temp
/// destination, using the failing allocator throughout. Every OOM point must
/// unwind cleanly (no leak) and a fully-funded run must succeed.
fn packOomProbe(alloc: std.mem.Allocator, io: std.Io, dir: std.Io.Dir) !void {
var sfs: storage.FileStorage = undefined;
var src = try packOpenReadOnly(alloc, io, dir, &sfs);
defer sfs.file.close(io);
defer src.deinit(alloc);
var f = try dir.createFile(io, "oomdst", .{ .read = true, .truncate = true });
defer f.close(io);
var ffs = storage.FileStorage.init(io, f, 0);
var dst = try logstore.LogStore.open(alloc, ffs.storageFile(), src.hash_kind);
defer dst.deinit(alloc);
_ = try packLog(alloc, &src, &dst);
}
test "pack: checkAllAllocationFailures on packLog (every OOM path unwinds cleanly)" {
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
var tip: [64]u8 = undefined;
{
var wfs: storage.FileStorage = undefined;
var wlog = try packOpenWritable(std.testing.allocator, io, tmp.dir, &wfs);
defer wfs.file.close(io);
defer wlog.deinit(std.testing.allocator);
try packSeedFullStore(std.testing.allocator, &wlog, &tip);
}
try std.testing.checkAllAllocationFailures(std.testing.allocator, packOomProbe, .{ io, tmp.dir });
}
test "pack∘derive == from-scratch derive: the packed store derives the identical repo view" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
var tip: [64]u8 = undefined;
{
var wfs: storage.FileStorage = undefined;
var wlog = try packOpenWritable(gpa, io, tmp.dir, &wfs);
defer wfs.file.close(io);
defer wlog.deinit(gpa);
try packSeedFullStore(gpa, &wlog, &tip);
}
var sfs: storage.FileStorage = undefined;
var src = try packOpenReadOnly(gpa, io, tmp.dir, &sfs);
defer sfs.file.close(io);
defer src.deinit(gpa);
var src_view: zit.repoview.RepoView = undefined;
try src_view.load(gpa, &src);
defer src_view.deinit(gpa);
var pfile = try tmp.dir.createFile(io, "packed", .{ .read = true, .truncate = true });
defer pfile.close(io);
var pfs = storage.FileStorage.init(io, pfile, 0);
var dst = try logstore.LogStore.open(gpa, pfs.storageFile(), src.hash_kind);
_ = try packLog(gpa, &src, &dst);
dst.deinit(gpa);
var rfs: storage.FileStorage = undefined;
var packed_log = try packOpenReadOnly2(gpa, io, tmp.dir, "packed", &rfs);
defer rfs.file.close(io);
defer packed_log.deinit(gpa);
var packed_view: zit.repoview.RepoView = undefined;
try packed_view.load(gpa, &packed_log);
defer packed_view.deinit(gpa);
// Same derived view: main branch, artifact count, and the leaf set + order
// (§7.5) are identical — pack changed only the storage layout.
try std.testing.expectEqualStrings(src_view.main_branch, packed_view.main_branch);
try std.testing.expectEqual(src_view.store.store().count(), packed_view.store.store().count());
var la = src_view.state.leaves();
var lb = packed_view.state.leaves();
while (la.next()) |ra| {
const rb = lb.next() orelse return error.LeafCountMismatch;
try std.testing.expectEqualStrings(src_view.hashOf(ra), packed_view.hashOf(rb));
}
try std.testing.expect(lb.next() == null);
}
test "renderPackStats: 'packed' when the store shrank, 'already packed' otherwise" {
var buf: [128]u8 = undefined;
{
var w: std.Io.Writer = .fixed(&buf);
renderPackStats(&w, .{ .artifacts = 1695, .deltas = 1373, .bytes_before = 9_680_000, .bytes_after = 920_000 });
try std.testing.expectEqualStrings(
"packed: 1695 artifacts, 1373 re-deltaed, 9680000 -> 920000 bytes\n",
w.buffered(),
);
}
{
var w: std.Io.Writer = .fixed(&buf);
renderPackStats(&w, .{ .artifacts = 3, .deltas = 2, .bytes_before = 4096, .bytes_after = 4096 });
try std.testing.expectEqualStrings(
"already packed: 3 artifacts, 2 re-deltaed, 4096 -> 4096 bytes\n",
w.buffered(),
);
}
}
/// Render one `ApplyError` through `openApplyFail` into `buf`.
fn openFailText(buf: []u8, err: update.ApplyError) []const u8 {
var w = Io.Writer.fixed(buf);
var sink: [1]u8 = undefined;
var wnull = Io.Writer.fixed(&sink);
const s: cli.Streams = .{ .out = &wnull, .err = &w };
_ = openApplyFail(s, err);
return w.buffered();
}
// Same guard as the `zit update`, `zit merge` and `zit bisect` mappers.
// `Corrupt` alleges the stored bytes do not match the name they are stored
// under; `StateRecordInvalid` says only that an artifact was unusable as a
// state record. A clone of an intact fossil repository reaches the second
// without anything being damaged, so it must never be answered with a
// corruption claim.
test "open: the two read-path failures do not share a diagnostic" {
var a: [256]u8 = undefined;
var b: [256]u8 = undefined;
const corrupt = openFailText(&a, error.Corrupt);
const state = openFailText(&b, error.StateRecordInvalid);
try std.testing.expect(!std.mem.eql(u8, corrupt, state));
try std.testing.expect(std.mem.indexOf(u8, corrupt, "corrupt") != null);
try std.testing.expect(std.mem.indexOf(u8, state, "corrupt") == null);
}
// ---------------------------------------------------------------------------
// Issue #9 — `zit serve` must answer one sync round in time bounded by the
// REQUEST, not by the served store. Two per-request costs made a clone
// O(rounds x store): counting held artifacts by forcing a full inflate of every
// artifact, and rebuilding the entire in-memory snapshot from the log on every
// request. The guards below pin both as SCALING properties measured with work
// counters (plaintext bytes materialised, snapshot builds performed) — never
// wall-clock thresholds, which flake and prove nothing about the mechanism.
// ---------------------------------------------------------------------------
/// Seed `ms` with `n` distinct wire-backed artifacts exactly as
/// `loadContentStore` does for a served repository: one §4.1 full container per
/// artifact, with NO plaintext materialised. `ms.plaintextBytesHeld()` is 0 on
/// return, so any later non-zero reading is inflation this store did not ask for.
fn i9SeedWireStore(gpa: std.mem.Allocator, ms: *repostate.MemStore, n: usize) !void {
var i: usize = 0;
while (i < n) : (i += 1) {
var body_buf: [64]u8 = undefined;
const body = try std.fmt.bufPrint(&body_buf, "issue nine artifact {d}\n", .{i});
var name_buf: [64]u8 = undefined;
const name = zit.artifact.artifactName(.sha3_256, body, &name_buf);
const container = try zit.container.encode(gpa, body, .default);
defer gpa.free(container);
_ = try ms.putWire(name, container);
}
std.debug.assert(ms.plaintextBytesHeld() == 0);
}
test "serve #9: the held-artifact count serve uses never materialises plaintext (constant in store size)" {
const gpa = std.testing.allocator;
// Two store sizes: the work this count performs must not grow with n. The
// count VALUE alone would pass before and after the fix — it is the
// materialised-plaintext reading that observes the defect, so that is what
// is asserted, at both sizes, as the scaling statement.
for ([_]usize{ 8, 64 }) |n| {
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
try i9SeedWireStore(gpa, &ms, n);
try std.testing.expectEqual(n, ms.store().heldCount());
try std.testing.expectEqual(@as(usize, 0), ms.plaintextBytesHeld());
}
}
test "serve #9: a held artifact whose container does not inflate is still counted as servable" {
const gpa = std.testing.allocator;
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
// §4.1 shape (4-byte length prefix) with a payload that is not a valid zlib
// stream: `held()==true`, `content()==null` — the case `held()`'s contract at
// repostate.zig names. `respond` retransmits it verbatim and `persistStore`
// counts it, so serve's own count must agree with them.
const bogus = "\x00\x00\x00\x0anot-zlib-bytes";
_ = try ms.putWire("f" ** 64, bogus);
try std.testing.expectEqual(@as(usize, 1), ms.store().heldCount());
try std.testing.expect(ms.store().held(1));
try std.testing.expect(ms.store().content(1) == null);
}
/// Append one content frame (raw `digest` naming `bytes`) to the `.zit` in
/// `dir`, creating the repository on first call — a separate writer's committed
/// append, exactly what a `zit commit` alongside a live `zit serve` performs.
fn i9AppendFrame(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir, digest: []const u8, bytes: []const u8) !void {
const hash_kind: u8 = @intFromEnum(zit.artifact.defaultPolicy(false, false));
var f = dir.openFile(io, repo_filename, .{ .mode = .read_write }) catch
try dir.createFile(io, repo_filename, .{ .read = true });
defer f.close(io);
const st = try f.stat(io);
var fs = storage.FileStorage.init(io, f, st.size);
var log = try logstore.LogStore.open(gpa, fs.storageFile(), hash_kind);
defer log.deinit(gpa);
_ = try log.put(digest, bytes);
try log.commit();
}
/// A §4.1-framed read-only request naming `hash` in a `gimme` card.
fn i9GimmeRequest(gpa: std.mem.Allocator, card: []const u8, hash: []const u8) ![]u8 {
const plain = try std.fmt.allocPrint(gpa, "pragma client-version 20000\n{s} {s}\n", .{ card, hash });
defer gpa.free(plain);
return zit.container.encode(gpa, plain, .default);
}
test "serve #9: repeated rounds over an unchanged log build the snapshot once (constant in round count)" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const digest_a = "a" ** digest_len;
try i9AppendFrame(gpa, io, tmp.dir, digest_a, "content a\n");
var hex_buf: [2 * digest_len]u8 = undefined;
const hex_a = digestToWireName(digest_a, &hex_buf);
const request = try i9GimmeRequest(gpa, "gimme", hex_a);
defer gpa.free(request);
const server = zit.sync.Server{ .version = serve_mod.server_version };
var file = try tmp.dir.openFile(io, repo_filename, .{});
defer file.close(io);
var ctx = ServeCtx{ .io = io, .root_dir = tmp.dir, .file = file, .server = &server };
defer ctx.dropSnapshot();
// Eight rounds, log unchanged: the snapshot is a function of the log, not of
// the request, so the number of builds must not grow with the round count.
for (0..8) |_| {
const reply = try serveAnswer(&ctx, gpa, request);
gpa.free(reply.body);
}
try std.testing.expectEqual(@as(u64, 1), ctx.loads);
}
test "serve #9: a responder's phantom interning cannot leak into the next round" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const digest_a = "a" ** digest_len;
try i9AppendFrame(gpa, io, tmp.dir, digest_a, "content a\n");
var hex_buf: [2 * digest_len]u8 = undefined;
const hex_a = digestToWireName(digest_a, &hex_buf);
// §3.3 — an `igot` for a hash the server lacks makes `respond` intern a
// phantom into whatever store it answers over. That mutation must not
// survive into the reused snapshot.
const unknown_hex = "c" ** (2 * digest_len);
const igot_req = try i9GimmeRequest(gpa, "igot", unknown_hex);
defer gpa.free(igot_req);
const gimme_req = try i9GimmeRequest(gpa, "gimme", hex_a);
defer gpa.free(gimme_req);
const server = zit.sync.Server{ .version = serve_mod.server_version };
var file = try tmp.dir.openFile(io, repo_filename, .{});
defer file.close(io);
// NOTE ON THIS GUARD'S ROLE: it is green against the pre-fix code (which
// rebuilds per request and so cannot leak) — the round-count guard above is
// the red one. This guard exists because the obvious WRONG way to satisfy
// that guard — answer directly over one shared mutable snapshot — passes it
// and truncates real clones, since the phantoms and memoisations `respond`
// writes change what later rounds serve. Without this assertion that
// implementation ships.
var ctx = ServeCtx{ .io = io, .root_dir = tmp.dir, .file = file, .server = &server };
defer ctx.dropSnapshot();
const r1 = try serveAnswer(&ctx, gpa, igot_req);
gpa.free(r1.body);
const snap = ctx.snapshot.?.store();
try std.testing.expectEqual(@as(usize, 1), snap.count()); // the one on-disk artifact, nothing more
try std.testing.expect(snap.lookup(unknown_hex) == null); // the phantom did not land here
const r2 = try serveAnswer(&ctx, gpa, gimme_req);
defer gpa.free(r2.body);
try std.testing.expectEqual(@as(u64, 1), ctx.loads); // still one build: the log never changed
// The reply is byte-identical to the one a server with no prior round produces.
var fresh_file = try tmp.dir.openFile(io, repo_filename, .{});
defer fresh_file.close(io);
var fresh_ctx = ServeCtx{ .io = io, .root_dir = tmp.dir, .file = fresh_file, .server = &server };
defer fresh_ctx.dropSnapshot();
const expected = try serveAnswer(&fresh_ctx, gpa, gimme_req);
defer gpa.free(expected.body);
try std.testing.expectEqualSlices(u8, expected.body, r2.body);
}
test "serve #9/§3.8: a reused snapshot still sees a concurrent commit's append" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const digest_a = "a" ** digest_len;
const digest_b = "b" ** digest_len;
try i9AppendFrame(gpa, io, tmp.dir, digest_a, "content a\n");
var hex_a_buf: [2 * digest_len]u8 = undefined;
var hex_b_buf: [2 * digest_len]u8 = undefined;
const hex_a = digestToWireName(digest_a, &hex_a_buf);
const hex_b = digestToWireName(digest_b, &hex_b_buf);
const req_a = try i9GimmeRequest(gpa, "gimme", hex_a);
defer gpa.free(req_a);
const req_b = try i9GimmeRequest(gpa, "gimme", hex_b);
defer gpa.free(req_b);
const server = zit.sync.Server{ .version = serve_mod.server_version };
var file = try tmp.dir.openFile(io, repo_filename, .{});
defer file.close(io);
var ctx = ServeCtx{ .io = io, .root_dir = tmp.dir, .file = file, .server = &server };
defer ctx.dropSnapshot();
const r1 = try serveAnswer(&ctx, gpa, req_a);
gpa.free(r1.body);
try std.testing.expectEqual(@as(usize, 1), ctx.snapshot.?.store().count());
// A SEPARATE writer appends while the serve handle stays open.
try i9AppendFrame(gpa, io, tmp.dir, digest_b, "content b\n");
const r2 = try serveAnswer(&ctx, gpa, req_b);
gpa.free(r2.body);
// The log grew, so the snapshot was rebuilt and the new artifact is servable:
// a cache that never invalidates would still report one artifact and one load.
try std.testing.expectEqual(@as(u64, 2), ctx.loads);
try std.testing.expectEqual(@as(usize, 2), ctx.snapshot.?.store().count());
try std.testing.expect(ctx.snapshot.?.store().lookup(hex_b) != null);
}
/// Seed `dir`'s `.zit` with `n` distinct compressible artifacts, returning the
/// wire (hex) name of the first. Compressible so the log retains a §4.1
/// container per frame — the shape a served repository actually has, and the
/// only shape in which "answering a round does not inflate the store" has
/// anything to say.
fn i9SeedRepo(gpa: std.mem.Allocator, io: std.Io, dir: std.Io.Dir, n: usize, first_hex: *[2 * digest_len]u8) !void {
const hash_kind: u8 = @intFromEnum(zit.artifact.defaultPolicy(false, false));
var f = dir.openFile(io, repo_filename, .{ .mode = .read_write }) catch
try dir.createFile(io, repo_filename, .{ .read = true });
defer f.close(io);
const st = try f.stat(io);
var fs = storage.FileStorage.init(io, f, st.size);
var log = try logstore.LogStore.open(gpa, fs.storageFile(), hash_kind);
defer log.deinit(gpa);
var i: usize = 0;
while (i < n) : (i += 1) {
var body: [512]u8 = undefined;
@memset(&body, 'a' + @as(u8, @intCast(i % 26)));
var digest: [digest_len]u8 = undefined;
@memset(&digest, 0);
std.mem.writeInt(u64, digest[0..8], i + 1, .big);
_ = try log.putZlib(&digest, &body);
if (i == 0) _ = digestToWireName(&digest, first_hex);
}
try log.commit();
}
test "serve #9: answering a round inflates only what the round emits (constant in store size)" {
const gpa = std.testing.allocator;
const io = std.testing.io;
// The same read round against two store sizes. The reply is identical, so
// the plaintext the server inflates to produce it must be identical too: any
// dependence on n is the server paying for artifacts the round never names.
// This is the assertion the whole change exists to make true — the pre-fix
// held-count walked `content()` over every rid, and so did inflate the store.
var inflated: [2]usize = undefined;
for ([_]usize{ 4, 40 }, 0..) |n, k| {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
var hex_buf: [2 * digest_len]u8 = undefined;
try i9SeedRepo(gpa, io, tmp.dir, n, &hex_buf);
const request = try i9GimmeRequest(gpa, "gimme", &hex_buf);
defer gpa.free(request);
const server = zit.sync.Server{
.version = serve_mod.server_version,
.self_servercode = serve_mod.server_servercode,
.project_code = serve_mod.server_projectcode,
};
var file = try tmp.dir.openFile(io, repo_filename, .{});
defer file.close(io);
var ctx = ServeCtx{ .io = io, .root_dir = tmp.dir, .file = file, .server = &server };
defer ctx.dropSnapshot();
for (0..3) |_| {
const reply = try serveAnswer(&ctx, gpa, request);
gpa.free(reply.body);
// Nothing the round materialised is still resident: the cached index
// reads containers on demand and frees them between rounds, never
// holding the repository's uncompressed history (or its containers).
try std.testing.expectEqual(@as(usize, 0), ctx.snapshot.?.residentBytes());
}
const emitted = ctx.snapshot.?.materialized_bytes;
// A CLONE round is the worst case for residency: it emits every artifact
// and proves each one's declared size against the plaintext, so it
// legitimately inflates the whole store. What must not survive it is the
// plaintext — a snapshot that outlives the round would otherwise end up
// holding the entire repository uncompressed, for the process's life.
const clone_plain = try std.fmt.allocPrint(gpa, "pragma client-version 20000\nclone 3 1\n", .{});
defer gpa.free(clone_plain);
const clone_req = try zit.container.encode(gpa, clone_plain, .default);
defer gpa.free(clone_req);
const clone_reply = try serveAnswer(&ctx, gpa, clone_req);
gpa.free(clone_reply.body);
try std.testing.expectEqual(@as(usize, 0), ctx.snapshot.?.residentBytes());
try std.testing.expect(ctx.snapshot.?.materialized_bytes >= emitted);
try std.testing.expectEqual(@as(usize, n), ctx.snapshot.?.store().count());
inflated[k] = emitted;
}
try std.testing.expectEqual(inflated[0], inflated[1]);
// ... and the constant is the round's own emission, not the store's: a
// 40-artifact store holds 20480 bytes of plaintext, so an equality that
// happened to hold at "everything" would still be caught here.
try std.testing.expect(inflated[1] < 4 * 512);
}
test "serve #13: the pushing round does not rebuild inside its deadline; maintain rebuilds between connections" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const hash_kind: u8 = @intFromEnum(zit.artifact.defaultPolicy(false, false));
{
var f = try tmp.dir.createFile(io, repo_filename, .{ .read = true, .exclusive = true });
defer f.close(io);
var fs = storage.FileStorage.init(io, f, 0);
var seed = try logstore.LogStore.open(gpa, fs.storageFile(), hash_kind);
defer seed.deinit(gpa);
try seed.commit();
}
const push_body = "pushed artifact bytes\n";
var name_buf: [64]u8 = undefined;
const push_hash = zit.artifact.artifactName(.sha3_256, push_body, &name_buf);
const plain = try std.fmt.allocPrint(gpa, "pragma client-version 20000\nfile {s} {d}\n{s}", .{ push_hash, push_body.len, push_body });
defer gpa.free(plain);
const push_req = try zit.container.encode(gpa, plain, .default);
defer gpa.free(push_req);
const read_req = try i9GimmeRequest(gpa, "gimme", push_hash);
defer gpa.free(read_req);
const server = zit.sync.Server{ .version = serve_mod.server_version };
var file = try tmp.dir.openFile(io, repo_filename, .{});
defer file.close(io);
var ctx = ServeCtx{ .io = io, .root_dir = tmp.dir, .file = file, .server = &server };
defer ctx.dropSnapshot();
const r0 = try serveAnswer(&ctx, gpa, read_req);
gpa.free(r0.body);
try std.testing.expectEqual(@as(u64, 1), ctx.loads);
// The push makes the snapshot stale, but answering it must NOT pay the
// O(store) rebuild: the reply is written inside a connection deadline, and
// #13's failure is exactly the whole-repository rebuild landing inside that
// budget. The rebuild is deferred to `maintain`, which the accept loop runs
// after the connection closes — outside every deadline. So the pushing round
// leaves `loads` untouched.
const r1 = try serveAnswer(&ctx, gpa, push_req);
gpa.free(r1.body);
try std.testing.expectEqual(@as(u64, 1), ctx.loads); // no rebuild inside the round
// The snapshot is momentarily stale — it predates the push — yet the push is
// already durable on disk: `persistStore` ran inside the round.
try std.testing.expect(ctx.snapshot != null);
try std.testing.expect(ctx.snapshot.?.store().lookup(push_hash) == null);
// The accept loop's post-connection maintenance hook does the O(store)
// rebuild, between connections, with no connection open to charge it to.
ServeCtx.maintain(&ctx, gpa);
try std.testing.expectEqual(@as(u64, 2), ctx.loads);
try std.testing.expect(ctx.snapshot.?.store().lookup(push_hash) != null);
// The rebuilt snapshot describes the grown log exactly (the append-only stat
// gate that `baseSnapshot` already carries).
const st = try file.stat(io);
try std.testing.expectEqual(st.size, ctx.snapshot_log_size);
// A later read round is stat-gated to no rebuild, and still serves the push.
const r2 = try serveAnswer(&ctx, gpa, read_req);
defer gpa.free(r2.body);
try std.testing.expectEqual(@as(u64, 2), ctx.loads);
}
test "submanifests §8.1/§7.2: a repository that HOLDS a leaf arms its serve-side gate against a pre-S peer" {
// The PRODUCTION wiring of the §7.2 gate, end to end through the durable
// store: `persistStore` records §8.1's seen-submanifest indicator for a
// stored SUBMANIFEST (which carries no S card at all, §4.4), the serve
// snapshot re-reads it, and `serveRound` arms the gate with it. Every link
// is load-bearing:
//
// * drop the `transferGatedArtifact` line in `phantomizeSubManifestRefs`
// and the indicator stays false — a leaf is not "S-bearing" literally;
// * drop `round_server.seen_submanifest = …` in `serveRound` and the gate
// is unarmed, because §8.1's emission setting is false on every zit;
// * drop `or self.seen_submanifest` from `Server.submanifestGateArmed`
// and it is unarmed for the same reason.
//
// Each of the three leaves the leaf in the reply to a fossil-2.28 peer,
// which §8 says can only hold it as "unreadable ... opaque content".
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
var hex_buf: [2 * digest_len]u8 = undefined;
try i9SeedRepo(gpa, io, tmp.dir, 1, &hex_buf);
// §4.3: an F run with no D card is type SUBMANIFEST. The flat check-in
// manifest beside it is everything that predates this draft.
const leaf = "F a.txt " ++ ("3" ** 64) ++ "\n";
const flat = "C m\nD 2026-07-27T12:00:00.000\nF a.txt " ++ ("2" ** 64) ++ "\nU u\n";
var leaf_name: [2 * digest_len]u8 = undefined;
var flat_name: [2 * digest_len]u8 = undefined;
_ = zit.artifact.artifactName(.sha3_256, leaf, &leaf_name);
_ = zit.artifact.artifactName(.sha3_256, flat, &flat_name);
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
_ = try ms.store().put(&leaf_name, leaf);
_ = try ms.store().put(&flat_name, flat);
_ = try persistStore(gpa, io, tmp.dir, ms.store());
// §8.1 — the indicator is DURABLE and covers §7.2's type shape, not only
// its S-card shape. Read back through the same function the CLI uses.
var pc_buf: [zit.sync.max_projectcode_len]u8 = undefined;
const identity = try repoSyncIdentity(gpa, io, tmp.dir, &pc_buf);
try std.testing.expect(identity.seen_submanifest);
// …and the emission setting is the OTHER, independent half of the arming
// disjunction: this repository holds a gated artifact it received, and has
// never been told to emit one. A single flag could not tell the two apart.
try std.testing.expect(!identity.enable_submanifests);
// …and the served reply refuses the leaf to a peer announcing fossil 2.28,
// while still handing it the artifact it can read.
const server = zit.sync.Server{
.version = serve_mod.server_version,
.self_servercode = serve_mod.server_servercode,
.project_code = serve_mod.server_projectcode,
};
try std.testing.expect(!server.submanifests_enabled); // §8.1's default, stated
var file = try tmp.dir.openFile(io, repo_filename, .{});
defer file.close(io);
var ctx = ServeCtx{ .io = io, .root_dir = tmp.dir, .file = file, .server = &server };
defer ctx.dropSnapshot();
const plain = try std.fmt.allocPrint(
gpa,
"pragma client-version 22800\ngimme {s}\ngimme {s}\n",
.{ &leaf_name, &flat_name },
);
defer gpa.free(plain);
const request = try zit.container.encode(gpa, plain, .default);
defer gpa.free(request);
const reply = try serveAnswer(&ctx, gpa, request);
defer gpa.free(reply.body);
const body = try zit.container.decode(gpa, reply.body, zit.sync.reply_body_cap);
defer gpa.free(body);
// Built through the same writer the server uses (§11.1's escape encoding
// applies to the text), so the expectation cannot drift from the emission.
var card_buf: [2 * zit.sync.submanifest_peer_version_error_text.len + 16]u8 = undefined;
const refusal = zit.sync.writeSignalCard(.err, zit.sync.submanifest_peer_version_error_text, &card_buf);
try std.testing.expect(std.mem.indexOf(u8, body, refusal) != null);
try std.testing.expect(std.mem.indexOf(u8, body, &leaf_name) == null);
try std.testing.expect(std.mem.indexOf(u8, body, &flat_name) != null);
}
test "submanifests §7.2/§10.1: the CLI announces a DATED client-version so the peer's version can reach the gate" {
// §10.1 r11/r12 — the server replies `pragma server-version` exactly when
// the received client-version carried DATE and TIME, and §7.2's gate reads
// "the remote peer's announced version". An undated pragma therefore leaves
// `Client.peer.version` at 0 for the whole session and an armed client
// refuses its own trees to an opted-in server: pullable, never pushable.
// (`sync_driver_test` drives that consequence end to end; this pins the one
// CLI constant the whole chain hangs off.)
const gpa = std.testing.allocator;
try std.testing.expect(sync_client_version_info.datetime != null);
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
var client = zit.sync.Client{
.state = zit.sync.initialState(.pull),
.operation = .pull,
.self_servercode = sync_servercode,
.project_code = sync_projectcode,
.version = sync_client_version_info,
};
defer client.deinit(gpa);
var buf: std.ArrayListUnmanaged(u8) = .empty;
defer buf.deinit(gpa);
const request = try client.buildRequest(gpa, ms.store(), &buf);
// The emitted pragma carries at least §10.1 r11's five tokens.
const eol = std.mem.indexOfScalar(u8, request, '\n').?;
var tokens: usize = 0;
var it = std.mem.tokenizeScalar(u8, request[0..eol], ' ');
while (it.next()) |_| tokens += 1;
try std.testing.expect(std.mem.startsWith(u8, request, "pragma client-version "));
try std.testing.expect(tokens >= zit.sync.min_datetime_tokens);
try std.testing.expect(zit.sync.serverEmitsServerVersion(.{
.version = sync_client_version,
.datetime = sync_client_version_info.datetime,
}));
}
test "clone observer forwards phase edges without owning measurement state" {
const Probe = struct {
count: usize = 0,
phase: ClonePhase = .transport,
event: ClonePhaseEvent = .begin,
fn callback(context: *anyopaque, phase: ClonePhase, event: ClonePhaseEvent) void {
const self: *@This() = @ptrCast(@alignCast(context));
self.count += 1;
self.phase = phase;
self.event = event;
}
};
var probe = Probe{};
const observer = CloneObserver{ .context = &probe, .callback = Probe.callback };
observeClonePhase(&observer, .persist, .end);
try std.testing.expectEqual(@as(usize, 1), probe.count);
try std.testing.expectEqual(ClonePhase.persist, probe.phase);
try std.testing.expectEqual(ClonePhaseEvent.end, probe.event);
}