//! The framed, append-only, content-addressed log over the `StorageFile`
//! capability. One file: a 64-byte header followed by 8-byte-aligned
//! frames, each self-verifying via a header checksum (over the fixed
//! fields) plus a whole-frame checksum (over header+payload). The header
//! checksum is what makes torn-vs-corrupt discrimination exact and
//! size-independent: without it, a torn write (a clean prefix, safe to
//! truncate) and a corrupted length field (unsafe to trust) both read as
//! "not enough of a frame," and nothing short of decoding payload bytes
//! that may not exist could tell them apart.
//!
//! No RFC section behind this file: native storage-engine design. The
//! oracle is this file's own framing invariants, checked by
//! sim/storage_sweep.zig's DST sweep: every acknowledged put stays
//! readable and hash-verifies; a torn tail truncates to exactly the last
//! complete frame; the index rebuilt on open matches the write-time
//! model; and corruption always surfaces as an error, never as silently
//! wrong bytes.
//!
//! A second, narrower checksum guards the frame's fixed 6-byte prefix
//! (`len|flags|hash_len`) on its own, in a 10-byte window whose size
//! never depends on `hash_len`. Without it, scanning a frame whose
//! `hash_len` byte is itself the bitflipped one has no way to know how
//! many bytes the rest of the header needs before it can check the
//! header checksum -- a corrupted `hash_len` that happens to demand more
//! bytes than the file has left reads exactly like a torn tail. The
//! prefix checksum removes that ambiguity: `hash_len` is proven correct
//! (or rejected outright) from a fixed-size read, before it is ever used
//! to size anything else.
const std = @import("std");
const storage = @import("zit_storage");
const delta = @import("zit").delta;
const flate = std.compress.flate;
/// Local artifact id: dense, 1-based, assigned in intern/put order. 0 is
/// never issued (mirrors repostate.zig's MemStore).
pub const Rid = u64;
/// FILE header size: magic(4) | version u16(2) | hash_kind u8(1) |
/// reserved(53) | crc32(4).
pub const header_len: u64 = 64;
/// Explicit payload bound: keeps a single frame's read/decode work
/// bounded no matter how large a check-in gets.
pub const max_payload: u64 = 1 << 30;
pub const max_hash_len: u8 = 32;
pub const min_hash_len: u8 = 1;
/// Every frame's on-disk span is a multiple of this.
pub const frame_align: u64 = 8;
/// Frozen on-disk format version — the `.zit` ship-format compatibility
/// promise. A reader accepts any stored version `<=` this and rejects a
/// newer one (`validateHeader`); an unknown frame flag-kind is a clean
/// reject, never UB (`scanOneFrame`). Bumped only on a
/// backward-compatible layout change — a copied `.zit` file is a complete,
/// self-verifying repository, so this field is what absorbs any future
/// layout evolution.
pub const version: u16 = 1;
/// Smallest possible frame span: hash_len=1, payload=0 ->
/// align8(26+1+0) = 32.
const min_frame: u64 = 32;
/// Explicit upper bound on delta-chain resolution: a `content()` on a
/// delta frame walks at most this many `delta_of` hops before rejecting
/// the chain as corrupt. Caps read recursion and read-amplification;
/// generous versus the short chains a real store produces.
pub const max_delta_chain: u8 = 32;
const magic = "zitL";
/// Payload encoding. `raw` = uncompressed bytes; `zlib` = a fossil zlib
/// container (4-byte BE uncompressed length + RFC1950 stream, sync §4.1);
/// `delta` = a zlib-contained delta whose `delta_of` is the byte offset of
/// the base frame, so resolving a chain link costs one seek and never a
/// second index lookup; `config` = a distinguished singleton repo-config
/// record, NOT content-addressed and never interned. Any other flag byte
/// is an unknown kind and rejected on open. The
/// content hash always names the UNCOMPRESSED, fully-reconstructed bytes, so
/// encoding never affects naming (sync §4.1 L1206-1208).
pub const FrameKind = enum(u8) { raw = 0, zlib = 1, delta = 2, config = 3, _ };
/// Repo-local config: default-branch name, project id, and an opaque
/// hash-policy byte the storage layer round-trips but never interprets
/// (exactly like the header's `hash_kind`). It has a home IN the file as
/// one distinguished `config` frame; "last config frame wins" is derived
/// by scan order, reusing the log's own append-prefix property.
pub const max_config_str: usize = 255;
/// Fixed config record bound: `[hash_policy u8][branch_len u16][branch]
/// [pid_len u16][pid]`.
pub const max_config_payload: usize = 1 + 2 * (2 + max_config_str); // = 515
/// A config frame is never keyed by hash — it is found only via
/// `config_offset`, never through `index`/`lookup` — so its 1-byte embedded
/// hash is inert bookkeeping (`appendFrame` requires some hash). It is never
/// an index key and never compared.
const config_sentinel: []const u8 = "\x00";
/// One repo config. Both slices borrow a backing buffer (`OwnedConfig`);
/// `hash_policy` is opaque to storage.
pub const Config = struct {
hash_policy: u8,
default_branch: []const u8,
project_id: []const u8,
};
/// A `readConfig` result: the owned frame payload plus a `Config` whose
/// slices point into it. Mirrors `StoredDelta`'s ownership shape.
pub const OwnedConfig = struct {
backing: []u8,
config: Config,
pub fn deinit(self: *OwnedConfig, gpa: std.mem.Allocator) void {
gpa.free(self.backing);
self.* = undefined;
}
};
pub const OpenError = error{ OutOfMemory, Storage, Corrupt };
/// No read happens on the write path, so `Corrupt` can't occur here —
/// advertising it would be dead mechanism.
pub const PutError = error{ OutOfMemory, Storage };
/// A phantom is purely in-memory bookkeeping: no I/O, so no `Storage`.
pub const InternError = error{OutOfMemory};
pub const ContentError = error{ OutOfMemory, Storage, Corrupt };
pub const CommitError = error{Storage};
/// The framed, content-addressed log. `open` rebuilds `entries`/`index`
/// from the file each time — there is no persistent side index yet.
/// O(n) on open is correct at this scale; a persistent, regenerable
/// sidecar is a noted future extension, built only if a benchmark shows
/// open time missing its target at large artifact counts.
pub const LogStore = struct {
gpa: std.mem.Allocator,
file: storage.StorageFile,
hash_kind: u8,
/// Structural end of the log: header + every complete frame written so
/// far. Always a multiple of `frame_align`.
end: u64,
entries: std.ArrayList(Entry) = .empty,
index: std.StringHashMapUnmanaged(Rid) = .empty,
/// Byte offset of the live (latest in scan order) config frame, or null
/// when the log carries none. Config is found ONLY through this — never
/// via `index`/`entries` (§E).
config_offset: ?u64 = null,
/// Set when a rollback truncate fails and `end` can no longer be proven
/// equal to the file's physical end. Every append refuses to run while it
/// is set, so a later frame never lands at a diverged offset (the storage
/// invariant: offset bookkeeping equals the file end). One-way: a store
/// only ever goes healthy -> poisoned, never back.
poisoned: bool = false,
const Entry = struct {
/// Owned duplicate; also the index's key (stable across `entries`
/// reallocation since it's a separate heap allocation).
hash: []u8,
/// Frame start offset. Meaningless while `phantom`.
offset: u64,
/// On-disk (container/wire) payload length: raw bytes for `raw`,
/// the zlib-container length for `zlib`/`delta`.
payload_len: u32,
/// Payload encoding of this frame. `.raw` while `phantom` (no frame).
kind: FrameKind,
/// A phantom was `intern`ed but never `put`: no frame on disk.
phantom: bool,
};
/// Opens for append, creating or repairing the file when necessary, and
/// rebuilds the in-memory index by scanning every complete frame.
pub fn open(gpa: std.mem.Allocator, file: storage.StorageFile, hash_kind: u8) OpenError!LogStore {
const file_size = file.size() catch return error.Storage;
var self: LogStore = .{ .gpa = gpa, .file = file, .hash_kind = hash_kind, .end = header_len };
errdefer self.deinit(gpa);
if (file_size == 0) {
try writeHeader(file, hash_kind);
std.debug.assert(self.end % frame_align == 0);
return self;
}
if (file_size < header_len) {
// Torn header: nothing before byte `header_len` can be trusted
// as a frame boundary either, so start clean.
file.truncate(0) catch return error.Storage;
try writeHeader(file, hash_kind);
std.debug.assert(self.end % frame_align == 0);
return self;
}
self.hash_kind = try validateHeader(file);
try scan(&self, true);
std.debug.assert(self.end % frame_align == 0);
return self;
}
/// Opens an existing log strictly for reads. Unlike `open`, this path never
/// creates, rewrites, or truncates storage: an empty file, torn header, or
/// torn frame tail is corruption. Complete-frame checksum and encoding
/// faults are rejected by the same scanner used by the append path.
pub fn openReadOnly(gpa: std.mem.Allocator, file: storage.StorageFile) OpenError!LogStore {
const file_size = file.size() catch return error.Storage;
if (file_size < header_len) return error.Corrupt;
var self: LogStore = .{ .gpa = gpa, .file = file, .hash_kind = 0, .end = header_len };
errdefer self.deinit(gpa);
self.hash_kind = try validateHeader(file);
try scan(&self, false);
std.debug.assert(self.end == file_size);
std.debug.assert(self.end % frame_align == 0);
return self;
}
pub fn deinit(self: *LogStore, gpa: std.mem.Allocator) void {
for (self.entries.items) |e| gpa.free(e.hash);
self.entries.deinit(gpa);
self.index.deinit(gpa);
self.* = undefined;
}
/// Clone = copy + derive: byte-copies `source` into `dest`, then opens
/// `dest` — rebuilding the index and recovering
/// config with no network (no network capability is even in scope here).
/// The result is an independent, equal repository: same artifact set, same
/// config, byte-identical log. A damaged copy needs no clone-specific
/// error — `open`'s torn-tail truncation drops exactly the incomplete
/// tail, so the copy degrades gracefully through the shared derive path.
/// Precondition (documented, not asserted — assertions carry no I/O):
/// `dest` is empty. `hash_kind` is a don't-care: `dest` is non-empty after
/// the copy, so `open` reads the copied header's stored kind.
pub fn cloneInto(
gpa: std.mem.Allocator,
source: storage.StorageFile,
dest: storage.StorageFile,
) OpenError!LogStore {
const src_size = source.size() catch return error.Storage;
var chunk: [64 * 1024]u8 = undefined;
var copied: u64 = 0;
// Bounded: `copied` rises by `n > 0` every pass, so the loop runs at
// most ceil(src_size / chunk.len) times before `copied == src_size`.
while (copied < src_size) {
const n = source.pread(copied, &chunk) catch return error.Storage;
// A 0-count before EOF on a stable file is a device failure, not a
// boundary — return rather than spin (no unbounded loop).
if (n == 0) return error.Storage;
dest.append(chunk[0..n]) catch return error.Storage;
copied += n;
}
std.debug.assert(copied == src_size);
return open(gpa, dest, 0);
}
/// Resolves `hash` to its rid, creating a phantom (no I/O) when
/// unknown. Phantoms never reach disk.
pub fn intern(self: *LogStore, hash: []const u8) InternError!Rid {
std.debug.assert(hash.len >= min_hash_len and hash.len <= max_hash_len);
if (self.index.get(hash)) |rid| return rid;
try self.entries.ensureUnusedCapacity(self.gpa, 1);
const owned = try self.gpa.dupe(u8, hash);
errdefer self.gpa.free(owned);
const rid: Rid = self.entries.items.len + 1;
try self.index.put(self.gpa, owned, rid);
self.entries.appendAssumeCapacity(.{ .hash = owned, .offset = 0, .payload_len = 0, .kind = .raw, .phantom = true });
return rid;
}
/// Stores `bytes` verbatim (a `raw` frame) under `hash`. Idempotent for
/// an already-content-bearing hash (returns the existing rid, appends
/// nothing); upgrades a phantom in place.
pub fn put(self: *LogStore, hash: []const u8, bytes: []const u8) PutError!Rid {
std.debug.assert(bytes.len <= max_payload);
return self.insert(hash, bytes, .raw, 0);
}
/// Stores `bytes` compressed into a fossil zlib container (§4.1, default
/// level) under `hash`. The hash still names the UNCOMPRESSED bytes, so
/// storage compression never changes an artifact's address (§4.1
/// L1206-1208). Idempotent + phantom-upgrade like `put`.
pub fn putZlib(self: *LogStore, hash: []const u8, bytes: []const u8) PutError!Rid {
std.debug.assert(bytes.len <= max_payload);
const container = try containerEncode(self.gpa, bytes);
defer self.gpa.free(container);
return self.insert(hash, container, .zlib, 0);
}
/// Stores `raw_delta` (a delta reconstructing this artifact from the frame
/// at `base_rid`) as a zlib-contained `delta` frame; `delta_of` records
/// the base frame's byte offset. The stored container IS the wire payload
/// a `cfile` serves verbatim (§7.2). Idempotent + phantom-upgrade like
/// `put`.
pub fn putDelta(self: *LogStore, hash: []const u8, base_rid: Rid, raw_delta: []const u8) PutError!Rid {
std.debug.assert(base_rid >= 1 and base_rid <= self.entries.items.len);
const base = &self.entries.items[@intCast(base_rid - 1)];
std.debug.assert(!base.phantom);
std.debug.assert(base.offset >= header_len and base.offset < self.end);
std.debug.assert(raw_delta.len >= 1 and raw_delta.len <= max_payload);
const container = try containerEncode(self.gpa, raw_delta);
defer self.gpa.free(container);
return self.insert(hash, container, .delta, base.offset);
}
/// Shared insert core: append one frame carrying `payload` under `hash`
/// with encoding `kind` (and `delta_of` for `.delta`), then update the
/// index. Idempotent for an already-content-bearing hash; upgrades a
/// phantom in place. On a bookkeeping failure after the frame landed, the
/// file is truncated back so disk and memory stay in lockstep; a rollback
/// truncate that itself fails poisons the store.
fn insert(self: *LogStore, hash: []const u8, payload: []const u8, kind: FrameKind, delta_of: u64) PutError!Rid {
std.debug.assert(hash.len >= min_hash_len and hash.len <= max_hash_len);
std.debug.assert(payload.len <= max_payload);
if (self.index.get(hash)) |rid| {
const entry = &self.entries.items[@intCast(rid - 1)];
if (!entry.phantom) return rid;
const res = try self.appendFrame(hash, payload, kind, delta_of);
entry.offset = res.offset;
entry.payload_len = res.len;
entry.kind = kind;
entry.phantom = false;
return rid;
}
try self.entries.ensureUnusedCapacity(self.gpa, 1);
const owned = try self.gpa.dupe(u8, hash);
errdefer self.gpa.free(owned);
const res = try self.appendFrame(hash, payload, kind, delta_of);
const rid: Rid = self.entries.items.len + 1;
self.index.put(self.gpa, owned, rid) catch |e| {
// Frame already landed on disk but bookkeeping failed: undo the
// write so the file matches in-memory state exactly. If the
// rollback truncate itself fails, `end` and the file end diverge
// and no later append can be trusted — poison the store.
self.file.truncate(res.offset) catch {
self.poisoned = true;
};
self.end = res.offset;
return e;
};
self.entries.appendAssumeCapacity(.{ .hash = owned, .offset = res.offset, .payload_len = res.len, .kind = kind, .phantom = false });
return rid;
}
/// The payload encoding of `rid` (`.raw` for a phantom). Lets a serve
/// loop pick a full-content versus delta cfile without a frame read.
pub fn kindOf(self: *const LogStore, rid: Rid) FrameKind {
std.debug.assert(rid >= 1 and rid <= self.entries.items.len);
return self.entries.items[@intCast(rid - 1)].kind;
}
/// The on-disk (container/wire) payload length of `rid`; 0 for a phantom.
/// Symmetric with `hashOf`; the DST sweep uses it to bound the crc-covered
/// span of a frame whose uncompressed content differs in length.
pub fn payloadLenOf(self: *const LogStore, rid: Rid) u32 {
std.debug.assert(rid >= 1 and rid <= self.entries.items.len);
return self.entries.items[@intCast(rid - 1)].payload_len;
}
/// Owned uncompressed bytes for a rid, or null while it's a phantom.
/// Re-reads and re-verifies the frame (both checksums, embedded hash) on
/// every call, then decodes by kind: `raw` copied out, `zlib` inflated,
/// `delta` resolved by walking `delta_of` and applying via `delta.apply`.
/// Never trusts a stale scan result.
pub fn content(self: *LogStore, gpa: std.mem.Allocator, rid: Rid) ContentError!?[]u8 {
std.debug.assert(rid >= 1 and rid <= self.entries.items.len);
const entry = &self.entries.items[@intCast(rid - 1)];
if (entry.phantom) return null;
return try self.contentAtOffset(gpa, entry.offset, max_delta_chain, entry.hash);
}
/// Reconstructs the uncompressed content of the frame at `offset`.
/// `depth_remaining` bounds delta recursion: each `delta` hop consumes
/// one, and a delta reached at zero is `error.Corrupt` (never unbounded
/// recursion). `expect_hash`, when non-null, requires the frame's embedded
/// hash to match (the top-level integrity check); recursion into a base
/// passes null since a base carries its own name.
fn contentAtOffset(self: *LogStore, gpa: std.mem.Allocator, offset: u64, depth_remaining: u8, expect_hash: ?[]const u8) ContentError![]u8 {
std.debug.assert(depth_remaining <= max_delta_chain);
const fv = try self.readFrameAt(gpa, offset);
if (expect_hash) |eh| {
if (!std.mem.eql(u8, fv.hash[0..fv.hash_len], eh)) {
gpa.free(fv.payload);
return error.Corrupt;
}
}
switch (fv.kind) {
.raw => return fv.payload, // ownership transfers to the caller
.zlib => {
defer gpa.free(fv.payload);
return containerDecode(gpa, fv.payload, max_payload);
},
.delta => {
defer gpa.free(fv.payload);
if (depth_remaining == 0) return error.Corrupt;
// A base must be an earlier, in-range frame: a forward or
// self reference is corruption, and a mid-frame/garbage
// offset fails `readFrameAt`'s checksums below.
if (fv.delta_of < header_len or fv.delta_of >= offset) return error.Corrupt;
const raw_delta = try containerDecode(gpa, fv.payload, max_payload);
defer gpa.free(raw_delta);
const base = try self.contentAtOffset(gpa, fv.delta_of, depth_remaining - 1, null);
defer gpa.free(base);
return delta.apply(gpa, base, raw_delta) catch |e| switch (e) {
error.OutOfMemory => error.OutOfMemory,
error.MalformedDelta, error.ChecksumMismatch => error.Corrupt,
};
},
.config => {
// A delta base (or content lookup) resolving to a config
// frame is corruption: config is not content-addressed and
// never a base (§E).
gpa.free(fv.payload);
return error.Corrupt;
},
_ => unreachable, // readFrameAt rejects any flag byte outside {raw,zlib,delta,config}
}
}
/// One fully validated frame read from `offset`: kind, `delta_of`, the
/// embedded hash, and the owned on-disk payload (raw bytes, or the zlib
/// container for `zlib`/`delta`). Caller owns `payload`.
const FrameView = struct {
kind: FrameKind,
delta_of: u64,
hash: [max_hash_len]u8,
hash_len: u8,
payload: []u8,
};
/// Reads and checksum-verifies the frame at `offset`, returning its
/// decoded fields and owned payload. Unlike scan, a short read here is
/// `error.Corrupt` (the frame was known to exist): the file lied or lost
/// bytes. Any flag byte outside {raw,zlib,delta,config} is rejected.
fn readFrameAt(self: *LogStore, gpa: std.mem.Allocator, offset: u64) ContentError!FrameView {
var prefix: [prefix_len]u8 = undefined;
const np = self.file.pread(offset, &prefix) catch return error.Storage;
if (np < prefix_len) return error.Corrupt;
const payload_len = std.mem.readInt(u32, prefix[0..4], .little);
const flags = prefix[4];
const hash_len = prefix[5];
if (hash_len < min_hash_len or hash_len > max_hash_len) return error.Corrupt;
var pcrc = std.hash.Crc32.init();
pcrc.update(prefix[0..6]);
if (pcrc.final() != std.mem.readInt(u32, prefix[6..][0..4], .little)) return error.Corrupt;
if (flags > @intFromEnum(FrameKind.config)) return error.Corrupt;
const hl: usize = hash_len;
const fixed_and_crc_len: usize = 22 + hl;
var hbuf: [22 + max_hash_len]u8 = undefined;
const nh = self.file.pread(offset, hbuf[0..fixed_and_crc_len]) catch return error.Storage;
if (nh < fixed_and_crc_len) return error.Corrupt;
var hcrc = std.hash.Crc32.init();
hcrc.update(hbuf[0 .. 18 + hl]);
if (hcrc.final() != std.mem.readInt(u32, hbuf[18 + hl ..][0..4], .little)) return error.Corrupt;
const pl: usize = payload_len;
if (pl > max_payload) return error.Corrupt; // wire-derived bound before allocation
const payload = try gpa.alloc(u8, pl);
errdefer gpa.free(payload);
const payload_offset: u64 = offset + fixed_and_crc_len;
const npay = self.file.pread(payload_offset, payload) catch return error.Storage;
if (npay < pl) return error.Corrupt;
var trailer: [4]u8 = undefined;
const nt = self.file.pread(payload_offset + pl, &trailer) catch return error.Storage;
if (nt < 4) return error.Corrupt;
var fcrc = std.hash.Crc32.init();
fcrc.update(hbuf[0..fixed_and_crc_len]);
fcrc.update(payload);
if (fcrc.final() != std.mem.readInt(u32, &trailer, .little)) return error.Corrupt;
var fv: FrameView = .{
.kind = @enumFromInt(flags),
.delta_of = std.mem.readInt(u64, hbuf[10 + hl ..][0..8], .little),
.hash = undefined,
.hash_len = hash_len,
.payload = payload,
};
@memcpy(fv.hash[0..hl], hbuf[10..][0..hl]);
return fv;
}
/// A stored delta served byte-for-byte (§7.2 "the sender does not
/// re-compress the payload; it writes the stored bytes"): the on-disk
/// zlib-container payload plus the base artifact's hash. Caller frees
/// `wire`.
pub const StoredDelta = struct {
wire: []u8,
base_hash_buf: [max_hash_len]u8,
base_hash_len: u8,
pub fn baseHash(self: *const StoredDelta) []const u8 {
return self.base_hash_buf[0..self.base_hash_len];
}
};
/// Serves the `delta` frame at `rid` verbatim for a cfile (§7.2): its
/// stored container bytes, uncompressed and unmodified, plus the base
/// hash read from the base frame. Precondition: `kindOf(rid) == .delta`.
pub fn storedDelta(self: *LogStore, gpa: std.mem.Allocator, rid: Rid) ContentError!StoredDelta {
std.debug.assert(rid >= 1 and rid <= self.entries.items.len);
const entry = &self.entries.items[@intCast(rid - 1)];
std.debug.assert(!entry.phantom and entry.kind == .delta);
const fv = try self.readFrameAt(gpa, entry.offset);
errdefer gpa.free(fv.payload);
std.debug.assert(fv.kind == .delta);
// The base frame's own header carries its hash; reading it also
// checksum-validates that `delta_of` still points at a real frame.
const base = try self.readFrameAt(gpa, fv.delta_of);
gpa.free(base.payload);
var out: StoredDelta = .{ .wire = fv.payload, .base_hash_buf = undefined, .base_hash_len = base.hash_len };
@memcpy(out.base_hash_buf[0..base.hash_len], base.hash[0..base.hash_len]);
return out;
}
/// Borrows the hash naming `rid`. Every rid, phantom or not, has one.
pub fn hashOf(self: *const LogStore, rid: Rid) []const u8 {
std.debug.assert(rid >= 1 and rid <= self.entries.items.len);
return self.entries.items[@intCast(rid - 1)].hash;
}
/// Returns caller-owned non-phantom artifact RIDs in the order their
/// frames were persisted. Dense RID order is deliberately not used:
/// `intern` may create a phantom before a later `put` writes its frame.
pub fn ridsInPersistedOrder(self: *const LogStore, gpa: std.mem.Allocator) error{OutOfMemory}![]Rid {
var count: usize = 0;
for (self.entries.items) |entry| {
if (!entry.phantom) count += 1;
}
const rids = try gpa.alloc(Rid, count);
count = 0;
for (self.entries.items, 0..) |entry, i| {
if (entry.phantom) continue;
rids[count] = @intCast(i + 1);
count += 1;
}
std.mem.sort(Rid, rids, self, ridOffsetLessThan);
return rids;
}
fn ridOffsetLessThan(self: *const LogStore, a: Rid, b: Rid) bool {
return self.entries.items[@intCast(a - 1)].offset < self.entries.items[@intCast(b - 1)].offset;
}
pub fn lookup(self: *const LogStore, hash: []const u8) ?Rid {
return self.index.get(hash);
}
pub fn commit(self: *LogStore) CommitError!void {
self.file.sync() catch return error.Storage;
}
/// Appends a new config frame and makes it the live config (§E). Latest
/// wins by scan order, so an edit is one append — no rewrite, no mutable
/// pointer. On a torn append `appendFrame` rolls the file back and
/// `config_offset` is left untouched, so failure leaves the prior config
/// live.
pub fn putConfig(self: *LogStore, cfg: Config) PutError!void {
std.debug.assert(cfg.default_branch.len <= max_config_str);
std.debug.assert(cfg.project_id.len <= max_config_str);
var buf: [max_config_payload]u8 = undefined;
const payload = encodeConfig(&buf, cfg);
const res = try self.appendFrame(config_sentinel, payload, .config, 0);
std.debug.assert(res.offset + payload.len < self.end); // frame landed before the new end
self.config_offset = res.offset;
}
/// The live config, or null when the log carries none (§E: absent config
/// means the caller — CLI/repostate — owns the default; storage bakes
/// none). Re-reads and re-verifies the frame on every call. Caller owns
/// the returned `OwnedConfig`.
pub fn readConfig(self: *LogStore, gpa: std.mem.Allocator) ContentError!?OwnedConfig {
const off = self.config_offset orelse return null;
const fv = try self.readFrameAt(gpa, off);
errdefer gpa.free(fv.payload);
if (fv.kind != .config) return error.Corrupt;
const parsed = try parseConfig(fv.payload);
return .{ .backing = fv.payload, .config = parsed };
}
/// Structural end of the log (header + every complete frame written so
/// far), for callers that track write progress without duplicating
/// frame-layout arithmetic (e.g. the DST sweep's crash-recovery model).
pub fn size(self: *const LogStore) u64 {
return self.end;
}
const AppendResult = struct { offset: u64, len: u32 };
/// Encodes one frame and appends it in a single `StorageFile.append`
/// call — one call is one atomic torn-write draw point, matching the
/// crash simulator's fault granularity. On failure, truncates back to
/// the prior structural end so a partial write never lingers; if that
/// rollback truncate itself fails the store is poisoned and refuses every
/// later append.
fn appendFrame(self: *LogStore, hash: []const u8, bytes: []const u8, kind: FrameKind, delta_of: u64) PutError!AppendResult {
if (self.poisoned) return error.Storage;
const hl: u64 = hash.len;
const pl: u64 = bytes.len;
const frame_len = alignUp8(u64, 26 + hl + pl);
std.debug.assert(frame_len <= max_payload + 64);
const frame = try self.gpa.alloc(u8, @intCast(frame_len));
defer self.gpa.free(frame);
_ = encodeFrame(frame, hash, bytes, @intFromEnum(kind), delta_of);
const offset = self.end;
if (self.file.append(frame)) |_| {
self.end += frame_len;
std.debug.assert(self.end % frame_align == 0);
return .{ .offset = offset, .len = @intCast(pl) };
} else |_| {
// A torn append may have left a prefix; if the rollback truncate
// can't remove it, `end` no longer matches the file end — poison.
self.file.truncate(self.end) catch {
self.poisoned = true;
};
return error.Storage;
}
}
};
fn writeHeader(file: storage.StorageFile, hash_kind: u8) OpenError!void {
var buf: [header_len]u8 = [_]u8{0} ** header_len;
@memcpy(buf[0..4], magic);
std.mem.writeInt(u16, buf[4..6], version, .little);
buf[6] = hash_kind;
var crc = std.hash.Crc32.init();
crc.update(buf[0..60]);
std.mem.writeInt(u32, buf[60..64], crc.final(), .little);
file.append(&buf) catch return error.Storage;
}
/// Returns the stored `hash_kind` on success. Rejects: bad magic, a stored
/// version newer than this reader, or a header crc mismatch.
fn validateHeader(file: storage.StorageFile) OpenError!u8 {
var buf: [header_len]u8 = undefined;
const n = file.pread(0, &buf) catch return error.Storage;
if (n < header_len) return error.Corrupt;
if (!std.mem.eql(u8, buf[0..4], magic)) return error.Corrupt;
const stored_version = std.mem.readInt(u16, buf[4..6], .little);
if (stored_version > version) return error.Corrupt;
const stored_crc = std.mem.readInt(u32, buf[60..64], .little);
var crc = std.hash.Crc32.init();
crc.update(buf[0..60]);
if (crc.final() != stored_crc) return error.Corrupt;
return buf[6];
}
/// Outcome of reading one frame at a scan offset: either a torn tail (the
/// file ends mid-frame — truncate and stop) or a validated frame.
const ScanOutcome = union(enum) {
torn,
frame: struct { hash: [max_hash_len]u8, hash_len: u8, frame_len: u64, payload_len: u32, kind: FrameKind },
};
/// Fixed prefix window: len(4)|flags(1)|hash_len(1)|prefix_crc32(4). Its
/// size does not depend on `hash_len` — that is what lets `hash_len` be
/// proven correct (or rejected) before it is used to size anything else.
const prefix_len: usize = 10;
/// Reads and validates exactly one frame at `offset`. Two-stage: first the
/// fixed-size prefix window (hash_len-independent, so a short read there
/// is unambiguously a torn write — nothing about its size can be thrown
/// off by corruption), which proves `hash_len` trustworthy via its own
/// crc; only then is `hash_len` used to size the rest of the header read,
/// at which point the same "short read = torn, checksum fails = corrupt"
/// reasoning applies validly (append() only ever writes one frame per
/// call and a torn write is a clean prefix — never damaged bytes; a full
/// window whose checksum fails can only be a bitflip on already-landed
/// bytes).
fn scanOneFrame(file: storage.StorageFile, offset: u64, total: u64) OpenError!ScanOutcome {
var prefix: [prefix_len]u8 = undefined;
const n0 = file.pread(offset, &prefix) catch return error.Storage;
if (n0 < prefix_len) return .torn;
const payload_len = std.mem.readInt(u32, prefix[0..4], .little);
const flags = prefix[4];
const hash_len = prefix[5];
if (hash_len < min_hash_len or hash_len > max_hash_len) return error.Corrupt;
var pcrc = std.hash.Crc32.init();
pcrc.update(prefix[0..6]);
if (pcrc.final() != std.mem.readInt(u32, prefix[6..][0..4], .little)) return error.Corrupt;
// A prefix-verified flag byte is trustworthy (a torn write is a clean
// truncation, never damaged bytes), so any kind outside
// {raw,zlib,delta,config} is a genuine unknown-kind frame, rejected — not
// a torn tail.
if (flags > @intFromEnum(FrameKind.config)) return error.Corrupt;
const fixed_and_crc_len: usize = 22 + @as(usize, hash_len);
var hbuf: [22 + max_hash_len]u8 = undefined;
const n1 = file.pread(offset, hbuf[0..fixed_and_crc_len]) catch return error.Storage;
if (n1 < fixed_and_crc_len) return .torn;
var hcrc = std.hash.Crc32.init();
hcrc.update(hbuf[0 .. 18 + @as(usize, hash_len)]);
const stored_header_crc = std.mem.readInt(u32, hbuf[18 + @as(usize, hash_len) ..][0..4], .little);
if (hcrc.final() != stored_header_crc) return error.Corrupt;
const frame_len = alignUp8(u64, 26 + @as(u64, hash_len) + payload_len);
if (offset + frame_len > total) return .torn;
var crc = std.hash.Crc32.init();
crc.update(hbuf[0..fixed_and_crc_len]);
var remaining: u64 = payload_len;
var pos = offset + @as(u64, fixed_and_crc_len);
var chunk: [4096]u8 = undefined;
while (remaining > 0) {
const take: usize = @intCast(@min(remaining, @as(u64, chunk.len)));
const got = file.pread(pos, chunk[0..take]) catch return error.Storage;
if (got < take) return .torn;
crc.update(chunk[0..take]);
pos += take;
remaining -= take;
}
var trailer: [4]u8 = undefined;
const nt = file.pread(pos, &trailer) catch return error.Storage;
if (nt < trailer.len) return .torn;
if (crc.final() != std.mem.readInt(u32, &trailer, .little)) return error.Corrupt;
var hash: [max_hash_len]u8 = undefined;
@memcpy(hash[0..hash_len], hbuf[10..][0..hash_len]);
return .{ .frame = .{ .hash = hash, .hash_len = hash_len, .frame_len = frame_len, .payload_len = payload_len, .kind = @enumFromInt(flags) } };
}
/// Rebuilds `self.entries`/`self.index` by scanning frames from
/// `header_len`. With `recover_torn`, a torn tail truncates to the last valid
/// boundary; otherwise it is reported as corruption without mutating storage.
/// A corrupt complete frame always propagates `error.Corrupt`.
fn scan(self: *LogStore, recover_torn: bool) OpenError!void {
const total = self.file.size() catch return error.Storage;
const max_frames = (total - header_len) / min_frame + 1;
var offset: u64 = header_len;
var frames: u64 = 0;
while (offset < total) {
frames += 1;
std.debug.assert(frames <= max_frames);
switch (try scanOneFrame(self.file, offset, total)) {
.torn => {
if (!recover_torn) return error.Corrupt;
self.file.truncate(offset) catch return error.Storage;
self.end = offset;
return;
},
.frame => |f| {
if (f.kind == .config) {
// Config is not interned: it never enters entries/index,
// and "latest wins" is just the last one scan reaches (§E).
if (self.config_offset) |prev| std.debug.assert(offset > prev);
self.config_offset = offset;
} else {
const owned = try self.gpa.dupe(u8, f.hash[0..f.hash_len]);
errdefer self.gpa.free(owned);
try self.entries.append(self.gpa, .{ .hash = owned, .offset = offset, .payload_len = f.payload_len, .kind = f.kind, .phantom = false });
errdefer _ = self.entries.pop();
const rid: Rid = self.entries.items.len;
try self.index.put(self.gpa, owned, rid);
}
const next = offset + f.frame_len;
std.debug.assert(next > offset);
offset = next;
},
}
}
self.end = offset;
}
/// Encodes one frame into `buf` (which must be at least the padded frame
/// span) and returns the written slice. Shared by the production write
/// path and by tests that hand-craft torn/corrupt fixtures.
///
/// Layout (all little-endian): len(4) | flags(1) | hash_len(1) |
/// prefix_crc32(4) = crc32(bytes[0..6)) | hash(hash_len) | delta_of(8) |
/// header_crc32(4) = crc32(bytes[0..18+hash_len)) | payload(len) |
/// frame_crc32(4) = crc32(bytes[0..22+hash_len+len)). `delta_of` is the
/// base frame's byte offset for a `delta` frame, and 0 otherwise.
fn encodeFrame(buf: []u8, hash: []const u8, payload: []const u8, flags: u8, delta_of: u64) []const u8 {
std.debug.assert(hash.len >= min_hash_len and hash.len <= max_hash_len);
std.debug.assert(payload.len <= max_payload);
const hl: usize = hash.len;
const pl: usize = payload.len;
const hash_len: u8 = @intCast(hl);
const payload_len: u32 = @intCast(pl);
const unpadded: usize = 26 + hl + pl;
const frame_len: usize = @intCast(alignUp8(u64, unpadded));
std.debug.assert(buf.len >= frame_len);
std.mem.writeInt(u32, buf[0..4], payload_len, .little);
buf[4] = flags;
buf[5] = hash_len;
var pcrc = std.hash.Crc32.init();
pcrc.update(buf[0..6]);
std.mem.writeInt(u32, buf[6..][0..4], pcrc.final(), .little);
@memcpy(buf[10..][0..hl], hash);
std.mem.writeInt(u64, buf[10 + hl ..][0..8], delta_of, .little);
var hcrc = std.hash.Crc32.init();
hcrc.update(buf[0 .. 18 + hl]);
std.mem.writeInt(u32, buf[18 + hl ..][0..4], hcrc.final(), .little);
@memcpy(buf[22 + hl ..][0..pl], payload);
var fcrc = std.hash.Crc32.init();
fcrc.update(buf[0 .. 22 + hl + pl]);
std.mem.writeInt(u32, buf[22 + hl + pl ..][0..4], fcrc.final(), .little);
if (frame_len > unpadded) @memset(buf[unpadded..frame_len], 0);
return buf[0..frame_len];
}
fn alignUp8(comptime T: type, n: T) T {
const a: T = frame_align;
return (n + (a - 1)) / a * a;
}
/// Encodes `cfg` into `buf` as `[hash_policy u8][branch_len u16 LE][branch]
/// [pid_len u16 LE][pid]` and returns the written slice (§E fixed field
/// order — deterministic, not a hash-map dump). Fixed bounds, no unbounded
/// copy.
fn encodeConfig(buf: []u8, cfg: Config) []const u8 {
std.debug.assert(buf.len >= max_config_payload);
std.debug.assert(cfg.default_branch.len <= max_config_str);
std.debug.assert(cfg.project_id.len <= max_config_str);
const bl: usize = cfg.default_branch.len;
const pl: usize = cfg.project_id.len;
buf[0] = cfg.hash_policy;
std.mem.writeInt(u16, buf[1..3], @intCast(bl), .little);
@memcpy(buf[3..][0..bl], cfg.default_branch);
std.mem.writeInt(u16, buf[3 + bl ..][0..2], @intCast(pl), .little);
@memcpy(buf[5 + bl ..][0..pl], cfg.project_id);
return buf[0 .. 5 + bl + pl];
}
/// Parses an `encodeConfig` payload. Every LE length read is bounds-checked
/// against the remaining slice before use, each length is capped at
/// `max_config_str`, and total consumption must be exact (trailing bytes are
/// corruption). Borrows `payload`. All wire-derived, validated before use.
fn parseConfig(payload: []const u8) error{Corrupt}!Config {
if (payload.len < 1) return error.Corrupt; // no hash_policy byte
const hash_policy = payload[0];
var pos: usize = 1;
if (payload.len - pos < 2) return error.Corrupt;
const blen: usize = std.mem.readInt(u16, payload[pos..][0..2], .little);
pos += 2;
if (blen > max_config_str) return error.Corrupt;
if (payload.len - pos < blen) return error.Corrupt;
const branch = payload[pos..][0..blen];
pos += blen;
if (payload.len - pos < 2) return error.Corrupt;
const plen: usize = std.mem.readInt(u16, payload[pos..][0..2], .little);
pos += 2;
if (plen > max_config_str) return error.Corrupt;
if (payload.len - pos < plen) return error.Corrupt;
const pid = payload[pos..][0..plen];
pos += plen;
if (pos != payload.len) return error.Corrupt; // no trailing bytes
return .{ .hash_policy = hash_policy, .default_branch = branch, .project_id = pid };
}
/// `Compress.init` asserts its output writer already holds more than 8 bytes
/// of buffer; `Writer.Allocating` starts empty, so pre-size before init.
/// This headroom covers the 4-byte length prefix and 2-byte zlib header
/// written first.
const compress_out_min: usize = 64;
/// sync §4.1 L1192-1197: the fossil zlib container — a 4-byte big-endian
/// uncompressed length, then an RFC1950 zlib stream (2-byte header, DEFLATE
/// body, 4-byte Adler-32 trailer) at the default level. The only compression
/// in the store. Kept private to the storage layer (no layer-inverting import
/// of the protocol module); sync carries its own cited copy.
fn containerEncode(gpa: std.mem.Allocator, bytes: []const u8) error{OutOfMemory}![]u8 {
std.debug.assert(bytes.len <= max_payload); // max_payload < 2^32, fits the length field
var aw = std.Io.Writer.Allocating.init(gpa);
defer aw.deinit();
aw.ensureUnusedCapacity(compress_out_min) catch return error.OutOfMemory;
aw.writer.writeInt(u32, @intCast(bytes.len), .big) catch return error.OutOfMemory;
const window = gpa.alloc(u8, flate.max_window_len) catch return error.OutOfMemory;
defer gpa.free(window);
var c = flate.Compress.init(&aw.writer, window, .zlib, .level_6) catch return error.OutOfMemory;
c.writer.writeAll(bytes) catch return error.OutOfMemory;
c.finish() catch return error.OutOfMemory;
const out = aw.toOwnedSlice() catch return error.OutOfMemory;
std.debug.assert(std.mem.readInt(u32, out[0..4], .big) == bytes.len);
return out;
}
/// Decodes a `containerEncode` container. `max_output` bounds the declared
/// length (a lying header is rejected before any inflate work) and the stream
/// is pumped no further than `declared + 1` bytes, so a length lie cannot
/// drive unbounded allocation and an overrun past `declared` is caught.
/// Malformed input (short header, length mismatch, bad stream) is
/// `error.Corrupt`; decode is compression-level-independent (§4.1 L1204-1206).
fn containerDecode(gpa: std.mem.Allocator, wire: []const u8, max_output: usize) ContentError![]u8 {
if (wire.len < 4) return error.Corrupt;
const declared = std.mem.readInt(u32, wire[0..4], .big);
if (declared > max_output) return error.Corrupt;
var fr = std.Io.Reader.fixed(wire[4..]);
const window = gpa.alloc(u8, flate.max_window_len) catch return error.OutOfMemory;
defer gpa.free(window);
var dc = flate.Decompress.init(&fr, .zlib, window);
var aw = std.Io.Writer.Allocating.init(gpa);
defer aw.deinit();
const cap: usize = @as(usize, declared) + 1;
var total: usize = 0;
// stream() may make progress without emitting bytes (parsing a block
// header), so 0 is not completion — only EndOfStream or reaching `cap`
// ends the pump. The iteration count is still explicitly bounded so a
// stalling stdlib regression surfaces as Corrupt rather than a hang.
const max_iters: usize = wire.len + cap + 64;
var iters: usize = 0;
while (total < cap and iters < max_iters) : (iters += 1) {
const n = dc.reader.stream(&aw.writer, .limited(cap - total)) catch |err| switch (err) {
error.EndOfStream => break,
error.WriteFailed => return error.OutOfMemory,
error.ReadFailed => return error.Corrupt,
};
total += n;
}
if (iters >= max_iters) return error.Corrupt;
if (total != declared) return error.Corrupt; // inflated length must equal the declared length
const out = aw.toOwnedSlice() catch return error.OutOfMemory;
std.debug.assert(out.len == declared);
return out;
}
// ---------------------------------------------------------------------------
// Framing tests (no RFC section behind this file; the oracle is the module
// doc's crash-recovery invariants above).
// ---------------------------------------------------------------------------
fn openTmp(tmp: anytype, io: std.Io, name: []const u8) !storage.FileStorage {
const raw = try tmp.dir.createFile(io, name, .{ .read = true });
return storage.FileStorage.init(io, raw, 0);
}
fn writeRawHeader(file: storage.StorageFile, magic_bytes: []const u8, ver: u16, hk: u8, bad_crc: bool) !void {
std.debug.assert(magic_bytes.len == 4);
var buf: [header_len]u8 = [_]u8{0} ** header_len;
@memcpy(buf[0..4], magic_bytes);
std.mem.writeInt(u16, buf[4..6], ver, .little);
buf[6] = hk;
var crc = std.hash.Crc32.init();
crc.update(buf[0..60]);
var crc_val = crc.final();
if (bad_crc) crc_val ^= 1;
std.mem.writeInt(u32, buf[60..64], crc_val, .little);
try file.append(&buf);
}
test "logstore §C: header round-trips magic/version/hash_kind through open" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
var store1 = try LogStore.open(std.testing.allocator, file, 7);
store1.deinit(std.testing.allocator);
var raw: [header_len]u8 = undefined;
const n = try file.pread(0, &raw);
try std.testing.expectEqual(@as(usize, header_len), n);
try std.testing.expectEqualStrings(magic, raw[0..4]);
try std.testing.expectEqual(@as(u16, version), std.mem.readInt(u16, raw[4..6], .little));
try std.testing.expectEqual(@as(u8, 7), raw[6]);
for (raw[7..60]) |b| try std.testing.expectEqual(@as(u8, 0), b);
var crc = std.hash.Crc32.init();
crc.update(raw[0..60]);
try std.testing.expectEqual(crc.final(), std.mem.readInt(u32, raw[60..64], .little));
var store2 = try LogStore.open(std.testing.allocator, file, 0); // hint ignored on reopen
defer store2.deinit(std.testing.allocator);
try std.testing.expectEqual(@as(u8, 7), store2.hash_kind);
}
test "logstore §C: open rejects a corrupt file-header crc" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
try writeRawHeader(file, magic, version, 0, true);
try std.testing.expectError(error.Corrupt, LogStore.open(std.testing.allocator, file, 0));
}
test "logstore §C: open rejects a version newer than this reader" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
try writeRawHeader(file, magic, version + 1, 0, false);
try std.testing.expectError(error.Corrupt, LogStore.open(std.testing.allocator, file, 0));
}
test "logstore §C: the frame header crc covers len/flags/hash_len/hash/delta_of" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
var seed = try LogStore.open(std.testing.allocator, file, 0);
seed.deinit(std.testing.allocator); // header only
var buf: [64]u8 = undefined;
const frame = encodeFrame(&buf, "aaaa", "payload", @intFromEnum(FrameKind.raw), 0);
var corrupted: [64]u8 = undefined;
@memcpy(corrupted[0..frame.len], frame);
corrupted[10] ^= 0xff; // inside the hash bytes, part of the fixed header
try file.append(corrupted[0..frame.len]);
try std.testing.expectError(error.Corrupt, LogStore.open(std.testing.allocator, file, 0));
}
test "logstore §C(inv2): a genuinely torn (prefix-truncated) final frame truncates to exactly the last valid boundary" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
var store = try LogStore.open(std.testing.allocator, file, 0);
_ = try store.put("frame-one-hash", "payload-one");
try store.commit();
const boundary = store.size();
store.deinit(std.testing.allocator);
var buf: [96]u8 = undefined;
const frame2 = encodeFrame(&buf, "frame-two-hash", "payload-two", @intFromEnum(FrameKind.raw), 0);
try file.append(frame2[0..10]); // short of even the fixed header+crc window
var reopened = try LogStore.open(std.testing.allocator, file, 0);
defer reopened.deinit(std.testing.allocator);
try std.testing.expectEqual(boundary, reopened.size());
try std.testing.expect(reopened.lookup("frame-two-hash") == null);
}
test "logstore §C(inv2): a torn payload/crc tail truncates exactly" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
var store = try LogStore.open(std.testing.allocator, file, 0);
_ = try store.put("frame-a-hash-here", "hello-payload");
try store.commit();
const boundary = store.size();
store.deinit(std.testing.allocator);
var buf: [128]u8 = undefined;
const frame2 = encodeFrame(&buf, "frame-b-hash-value", "a longer second payload for this test", @intFromEnum(FrameKind.raw), 0);
const header_window = 22 + "frame-b-hash-value".len;
try file.append(frame2[0 .. header_window + 3]); // full header+crc landed; payload/crc badly short
var reopened = try LogStore.open(std.testing.allocator, file, 0);
defer reopened.deinit(std.testing.allocator);
try std.testing.expectEqual(boundary, reopened.size());
}
test "logstore §C(inv4): a bitflipped header on a durable non-final frame surfaces error.Corrupt" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
var seed = try LogStore.open(std.testing.allocator, file, 0);
seed.deinit(std.testing.allocator);
var buf1: [64]u8 = undefined;
const f1 = encodeFrame(&buf1, "frame-one-abc", "payload-one", @intFromEnum(FrameKind.raw), 0);
var corrupted: [64]u8 = undefined;
@memcpy(corrupted[0..f1.len], f1);
corrupted[10] ^= 0xff;
try file.append(corrupted[0..f1.len]);
var buf2: [64]u8 = undefined;
const f2 = encodeFrame(&buf2, "frame-two-def", "payload-two", @intFromEnum(FrameKind.raw), 0);
try file.append(f2);
try std.testing.expectError(error.Corrupt, LogStore.open(std.testing.allocator, file, 0));
}
test "logstore §C(inv4): a bitflipped complete final frame is error.Corrupt" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
var seed = try LogStore.open(std.testing.allocator, file, 0);
seed.deinit(std.testing.allocator);
var buf: [64]u8 = undefined;
const f = encodeFrame(&buf, "final-frame-hash", "payload-for-final", @intFromEnum(FrameKind.raw), 0);
var corrupted: [64]u8 = undefined;
@memcpy(corrupted[0..f.len], f);
corrupted[6] ^= 0xff;
try file.append(corrupted[0..f.len]); // complete, fully landed, nothing follows
try std.testing.expectError(error.Corrupt, LogStore.open(std.testing.allocator, file, 0));
}
test "logstore §C(inv4): a corrupted payload on a complete frame surfaces error.Corrupt" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
var seed = try LogStore.open(std.testing.allocator, file, 0);
seed.deinit(std.testing.allocator);
var buf: [96]u8 = undefined;
const f = encodeFrame(&buf, "payload-hash-value", "some payload bytes here", @intFromEnum(FrameKind.raw), 0);
var corrupted: [96]u8 = undefined;
@memcpy(corrupted[0..f.len], f);
const header_window = 22 + "payload-hash-value".len;
corrupted[header_window] ^= 0xff; // inside the payload, not the fixed header
try file.append(corrupted[0..f.len]);
try std.testing.expectError(error.Corrupt, LogStore.open(std.testing.allocator, file, 0));
}
test "logstore §C(inv4): a bitflipped hash_len that would need more bytes than the file has surfaces error.Corrupt, not a torn tail" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
var seed = try LogStore.open(std.testing.allocator, file, 0);
seed.deinit(std.testing.allocator);
// hash_len=1, payload="" -> frame_len = align8(26+1+0) = 32, fully
// landed, nothing follows. Corrupting hash_len to an in-range value
// (20) makes the naive "read 18+hash_len bytes, short = torn" reading
// demand 38 bytes from a 32-byte frame -- exactly the false-torn gap
// the prefix checksum closes: it fails on this corrupted byte before
// any window-size decision is made from it.
var buf: [64]u8 = undefined;
const f = encodeFrame(&buf, "h", "", @intFromEnum(FrameKind.raw), 0);
var corrupted: [64]u8 = undefined;
@memcpy(corrupted[0..f.len], f);
corrupted[5] = 20;
try file.append(corrupted[0..f.len]);
try std.testing.expectError(error.Corrupt, LogStore.open(std.testing.allocator, file, 0));
}
test "logstore §C: a crc-valid frame with an unknown flag kind (not raw/zlib/delta) is rejected on open" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
var seed = try LogStore.open(std.testing.allocator, file, 0);
seed.deinit(std.testing.allocator);
// flags=4 is one past `config`: both crcs are computed correctly over it,
// so the frame is byte-intact -- it is simply an encoding this reader does
// not know, which must fail closed rather than be interpreted.
var buf: [64]u8 = undefined;
const f = encodeFrame(&buf, "unknown-flag-hash", "some payload bytes", 4, 0);
try file.append(f);
try std.testing.expectError(error.Corrupt, LogStore.open(std.testing.allocator, file, 0));
}
// ---------------------------------------------------------------------------
// zlib and delta frames (container sync §4.1; verbatim serve sync §7.2;
// delta apply delta §4).
// ---------------------------------------------------------------------------
/// Appends one hand-crafted frame (valid crcs) to `file`, for fixtures that
/// need a payload/flag/delta_of a normal put would never produce.
fn appendCrafted(gpa: std.mem.Allocator, file: storage.StorageFile, hash: []const u8, payload: []const u8, kind: FrameKind, delta_of: u64) !void {
const frame_len: usize = @intCast(alignUp8(u64, 26 + hash.len + payload.len));
const buf = try gpa.alloc(u8, frame_len);
defer gpa.free(buf);
const f = encodeFrame(buf, hash, payload, @intFromEnum(kind), delta_of);
try file.append(f);
}
test "logstore §C: content on a malformed zlib container is error.Corrupt" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
var seed = try LogStore.open(std.testing.allocator, file, 0);
seed.deinit(std.testing.allocator);
// A crc-valid zlib frame whose payload is not a valid container: two bytes
// is short of even the 4-byte length header, so containerDecode rejects it.
const hash = "malformed-zlib-container-hash-01";
try appendCrafted(std.testing.allocator, file, hash, "\x00\x00", .zlib, 0);
var store = try LogStore.open(std.testing.allocator, file, 0);
defer store.deinit(std.testing.allocator);
const rid = store.lookup(hash).?;
try std.testing.expectError(error.Corrupt, store.content(std.testing.allocator, rid));
}
test "logstore §C/sync §4.1: a zlib container whose declared length exceeds max_payload is rejected without a huge allocation" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
var seed = try LogStore.open(std.testing.allocator, file, 0);
seed.deinit(std.testing.allocator);
// Declared length 0xFFFFFFFF is far past max_payload; containerDecode must
// reject on the declared value before inflating anything.
const hash = "oversize-declared-length-hash-01";
try appendCrafted(std.testing.allocator, file, hash, "\xff\xff\xff\xff\x78\x9c", .zlib, 0);
var store = try LogStore.open(std.testing.allocator, file, 0);
defer store.deinit(std.testing.allocator);
const rid = store.lookup(hash).?;
try std.testing.expectError(error.Corrupt, store.content(std.testing.allocator, rid));
}
test "logstore §C/delta §4: content on a delta whose inflated payload is not a valid delta is error.Corrupt" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
var seed = try LogStore.open(std.testing.allocator, file, 0);
seed.deinit(std.testing.allocator);
// A valid raw base, then a crc-valid delta frame whose container inflates
// to bytes that are not a conforming delta -> delta.apply rejects -> Corrupt.
try appendCrafted(std.testing.allocator, file, "garbage-delta-base-hash-00000001", "base content here", .raw, 0);
const container = try containerEncode(std.testing.allocator, "this is not a delta at all");
defer std.testing.allocator.free(container);
try appendCrafted(std.testing.allocator, file, "garbage-delta-hash-value-0000001", container, .delta, header_len);
var store = try LogStore.open(std.testing.allocator, file, 0);
defer store.deinit(std.testing.allocator);
const rid = store.lookup("garbage-delta-hash-value-0000001").?;
try std.testing.expectError(error.Corrupt, store.content(std.testing.allocator, rid));
}
test "logstore §C: content on a delta whose delta_of is out of range or mid-frame is error.Corrupt" {
const cases = [_]struct { name: []const u8, delta_of: u64 }{
.{ .name = "below-header", .delta_of = 0 },
.{ .name = "forward-ref", .delta_of = 1 << 20 },
.{ .name = "mid-base-frame", .delta_of = header_len + 3 },
};
for (cases) |c| {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
var seed = try LogStore.open(std.testing.allocator, file, 0);
seed.deinit(std.testing.allocator);
try appendCrafted(std.testing.allocator, file, "bad-delta-of-base-hash-000000001", "base content bytes", .raw, 0);
const container = try containerEncode(std.testing.allocator, "1\ndelta-ish payload; only delta_of matters here");
defer std.testing.allocator.free(container);
try appendCrafted(std.testing.allocator, file, "bad-delta-of-hash-value-00000001", container, .delta, c.delta_of);
var store = try LogStore.open(std.testing.allocator, file, 0);
defer store.deinit(std.testing.allocator);
const rid = store.lookup("bad-delta-of-hash-value-00000001").?;
try std.testing.expectError(error.Corrupt, store.content(std.testing.allocator, rid));
}
}
// Seeded deterministic fuzz: decode/apply on adversarial wire bytes must
// yield a value or a declared error, never a crash or a hang.
test "logstore §C/sync §4.1 fuzz: a garbage zlib container never crashes content" {
var prng = std.Random.DefaultPrng.init(0x2172B);
const random = prng.random();
var iter: usize = 0;
while (iter < 150) : (iter += 1) {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
var store = try LogStore.open(std.testing.allocator, file, 0);
store.deinit(std.testing.allocator); // header only; craft the frame directly
// A crc-valid zlib frame with a random payload must never crash content.
var garbage: [64]u8 = undefined;
const gn = random.uintLessThan(usize, garbage.len + 1);
random.bytes(garbage[0..gn]);
var ghash: [9]u8 = undefined;
ghash[0] = 'g';
std.mem.writeInt(u64, ghash[1..9], iter, .little);
try appendCrafted(std.testing.allocator, file, &ghash, garbage[0..gn], .zlib, 0);
var reopened = try LogStore.open(std.testing.allocator, file, 0);
defer reopened.deinit(std.testing.allocator);
const grid = reopened.lookup(&ghash).?;
if (reopened.content(std.testing.allocator, grid)) |maybe| {
if (maybe) |b| std.testing.allocator.free(b);
} else |err| switch (err) {
error.OutOfMemory, error.Storage, error.Corrupt => {},
}
}
}
test "logstore §C/delta §4 fuzz: a garbage delta payload or delta_of never crashes content" {
var prng = std.Random.DefaultPrng.init(0xD317A);
const random = prng.random();
var iter: usize = 0;
while (iter < 150) : (iter += 1) {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
var store = try LogStore.open(std.testing.allocator, file, 0);
store.deinit(std.testing.allocator); // header only; craft the frame directly
// A crc-valid delta frame with a random container payload and a random
// delta_of must never crash content (Corrupt/OOM/value only).
var garbage: [64]u8 = undefined;
const gn = random.uintLessThan(usize, garbage.len + 1);
random.bytes(garbage[0..gn]);
var ghash: [9]u8 = undefined;
ghash[0] = 'x';
std.mem.writeInt(u64, ghash[1..9], iter, .little);
try appendCrafted(std.testing.allocator, file, &ghash, garbage[0..gn], .delta, random.int(u64));
var reopened = try LogStore.open(std.testing.allocator, file, 0);
defer reopened.deinit(std.testing.allocator);
const grid = reopened.lookup(&ghash).?;
if (reopened.content(std.testing.allocator, grid)) |maybe| {
if (maybe) |b| std.testing.allocator.free(b);
} else |err| switch (err) {
error.OutOfMemory, error.Storage, error.Corrupt => {},
}
}
}
test "logstore §C: a config frame is not interned — rids stay content-only and contiguous" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
var store = try LogStore.open(std.testing.allocator, file, 0);
defer store.deinit(std.testing.allocator);
const r1 = try store.put("config-interleave-hash-one-0001", "one");
try store.putConfig(.{ .hash_policy = 0, .default_branch = "trunk", .project_id = "pid" });
const r2 = try store.put("config-interleave-hash-two-0001", "two");
// The config frame consumed a byte range but no rid: content rids stay
// dense and contiguous, and the sentinel hash is never an index key.
try std.testing.expectEqual(r1 + 1, r2);
try std.testing.expect(store.lookup(config_sentinel) == null);
}
test "logstore §C: a delta base offset pointing at a config frame is rejected as corrupt" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
var seed = try LogStore.open(std.testing.allocator, file, 0);
seed.deinit(std.testing.allocator); // header only
// A config frame at header_len, then a delta whose (valid) container base
// offset points back at it: resolving the base hits the .config arm.
var cbuf: [max_config_payload]u8 = undefined;
const cfg_payload = encodeConfig(&cbuf, .{ .hash_policy = 4, .default_branch = "trunk", .project_id = "pid" });
try appendCrafted(std.testing.allocator, file, config_sentinel, cfg_payload, .config, 0);
const container = try containerEncode(std.testing.allocator, "delta-ish bytes; only the base offset matters here");
defer std.testing.allocator.free(container);
try appendCrafted(std.testing.allocator, file, "delta-over-config-base-hash-0001", container, .delta, header_len);
var store = try LogStore.open(std.testing.allocator, file, 0);
defer store.deinit(std.testing.allocator);
const rid = store.lookup("delta-over-config-base-hash-0001").?;
try std.testing.expectError(error.Corrupt, store.content(std.testing.allocator, rid));
}
test "logstore §E: parseConfig rejects malformed config payloads" {
var over: [3]u8 = undefined;
over[0] = 0; // hash_policy
std.mem.writeInt(u16, over[1..3], @intCast(max_config_str + 1), .little); // branch_len > cap
const cases = [_][]const u8{
&[_]u8{}, // (a) empty: no hash_policy byte
&[_]u8{ 0x07, 0x05 }, // (b) truncated branch_len (1 of 2 bytes)
&over, // (c) branch_len exceeds max_config_str
&[_]u8{ 0x00, 0x05, 0x00, 'a', 'b' }, // (d) branch_len 5 exceeds remaining
&[_]u8{ 0x00, 0x01, 0x00, 'b', 0x02 }, // (e) truncated pid_len after branch
&[_]u8{ 0x00, 0x01, 0x00, 'b', 0x00, 0x00, 0x09 }, // (f) trailing byte after pid
};
for (cases) |c| try std.testing.expectError(error.Corrupt, parseConfig(c));
}
test "logstore §B: a torn config-frame write reverts to the prior committed config" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
var store = try LogStore.open(std.testing.allocator, file, 0);
try store.putConfig(.{ .hash_policy = 9, .default_branch = "trunk", .project_id = "keep" });
try store.commit();
store.deinit(std.testing.allocator);
// Append a torn config B frame (prefix only): scan truncates it away, so
// the surviving latest config is still A.
var pbuf: [max_config_payload]u8 = undefined;
const cfgB = encodeConfig(&pbuf, .{ .hash_policy = 1, .default_branch = "x", .project_id = "y" });
var fbuf: [64]u8 = undefined;
const frameB = encodeFrame(&fbuf, config_sentinel, cfgB, @intFromEnum(FrameKind.config), 0);
try file.append(frameB[0..10]); // torn: short of the fixed header+crc window
var reopened = try LogStore.open(std.testing.allocator, file, 0);
defer reopened.deinit(std.testing.allocator);
var oc = (try reopened.readConfig(std.testing.allocator)).?;
defer oc.deinit(std.testing.allocator);
try std.testing.expectEqual(@as(u8, 9), oc.config.hash_policy);
try std.testing.expectEqualStrings("trunk", oc.config.default_branch);
try std.testing.expectEqualStrings("keep", oc.config.project_id);
}
// Seeded deterministic fuzz: parseConfig on adversarial wire bytes yields a
// value or error.Corrupt, never a crash.
test "logstore §E fuzz: parseConfig never crashes on adversarial payloads" {
var prng = std.Random.DefaultPrng.init(0xC04F16);
const random = prng.random();
var iter: usize = 0;
while (iter < 400) : (iter += 1) {
var garbage: [64]u8 = undefined;
const gn = random.uintAtMost(usize, garbage.len);
random.bytes(garbage[0..gn]);
if (parseConfig(garbage[0..gn])) |_| {} else |err| switch (err) {
error.Corrupt => {},
}
}
}
test "logstore §E: open accepts a stored version older than this reader" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "log.bin");
defer fs.file.close(io);
const file = fs.storageFile();
// Header-only file stamped v0 with a valid crc. §E freeze: a reader
// accepts any stored version <= its own, and 0 <= version, so open
// succeeds and round-trips the opaque hash_kind.
try writeRawHeader(file, magic, 0, 5, false);
var store = try LogStore.open(std.testing.allocator, file, 0);
defer store.deinit(std.testing.allocator);
try std.testing.expectEqual(@as(u8, 5), store.hash_kind);
try std.testing.expectEqual(@as(u64, header_len), store.size());
}
test "logstore §C: persisted RID order follows frame order after phantom upgrade" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var fs = try openTmp(&tmp, io, "order.bin");
defer fs.file.close(io);
var store = try LogStore.open(std.testing.allocator, fs.storageFile(), 0);
defer store.deinit(std.testing.allocator);
_ = try store.intern("A");
_ = try store.put("B", "second rid, first persisted frame");
_ = try store.put("A", "first rid, second persisted frame");
const rids = try store.ridsInPersistedOrder(std.testing.allocator);
defer std.testing.allocator.free(rids);
try std.testing.expectEqual(@as(usize, 2), rids.len);
try std.testing.expectEqualStrings("B", store.hashOf(rids[0]));
try std.testing.expectEqualStrings("A", store.hashOf(rids[1]));
}