//! The DST sweep that drives the real `LogStore` (logstore.zig) through
//! `FaultyFile`, asserting the four crash-recovery invariants
//! (storage-track-plan.md §B). `run` is exercised by `zig build dst`, not
//! a `zig build test` unit — the sweep IS the gate.
//!
//! The model tracks WHAT WAS WRITTEN (hash, bytes, and the store's own
//! `size()` right after each successful append) rather than re-deriving
//! frame-layout byte offsets itself: LogStore owns that arithmetic, and an
//! independent oracle should observe outcomes through the public
//! hash-addressed API, not reimplement wire format math that could share
//! the same bug.
const std = @import("std");
const storage = @import("../storage.zig");
const faulty_file = @import("faulty_file.zig");
const logstore = @import("../logstore.zig");
const delta = @import("zit").delta;
const hash_len = 32; // sha256 digest
const Frame = struct {
hash: [hash_len]u8,
/// Owned UNCOMPRESSED content -- what `content()` must reconstruct for
/// this frame regardless of its on-disk encoding (raw/zlib/delta).
bytes: []u8,
/// `store.size()` immediately before this put's frame was appended.
frame_start: u64,
/// `store.size()` immediately after -- the structural boundary a
/// clean recovery of this frame implies. `frame_end - frame_start`
/// includes trailing zero-pad; the pad is excluded from both crcs
/// (logstore.zig's wire format), so it's inert and never read back --
/// a corrupt offset landing there is harmless, not silent corruption.
frame_end: u64,
/// On-disk payload length (`payloadLenOf`): the container/wire byte count,
/// which differs from `bytes.len` for zlib/delta frames. Bounds the
/// crc-covered span in the silent-corruption check.
disk_payload_len: u32,
/// Delta-chain depth: 0 for raw/zlib, base.depth+1 for a delta. Kept below
/// `max_delta_chain` so every created chain stays resolvable.
depth: u8,
};
/// Ground-truth bookkeeping: every content-bearing put in append order,
/// how many of them are known-durable, and every intern-only (phantom)
/// hash, which must never appear on disk.
const Model = struct {
frames: std.ArrayList(Frame) = .empty,
acked_count: usize = 0,
phantoms: std.ArrayList([hash_len]u8) = .empty,
fn deinit(self: *Model, gpa: std.mem.Allocator) void {
for (self.frames.items) |f| gpa.free(f.bytes);
self.frames.deinit(gpa);
self.phantoms.deinit(gpa);
self.* = undefined;
}
fn recordPut(self: *Model, gpa: std.mem.Allocator, hash: [hash_len]u8, bytes: []const u8, frame_start: u64, frame_end: u64, disk_payload_len: u32, depth: u8) error{OutOfMemory}!void {
const owned = try gpa.dupe(u8, bytes);
errdefer gpa.free(owned);
try self.frames.append(gpa, .{ .hash = hash, .bytes = owned, .frame_start = frame_start, .frame_end = frame_end, .disk_payload_len = disk_payload_len, .depth = depth });
}
fn recordPhantom(self: *Model, gpa: std.mem.Allocator, hash: [hash_len]u8) error{OutOfMemory}!void {
try self.phantoms.append(gpa, hash);
}
/// Drops the phantom at `idx` once it has gained content (order among
/// phantoms is immaterial, so a swap-remove is exact and O(1)).
fn removePhantomAt(self: *Model, idx: usize) void {
std.debug.assert(idx < self.phantoms.items.len);
_ = self.phantoms.swapRemove(idx);
}
/// Acked frames only grow, and only past a frame end the disk has
/// actually promised durable.
fn recomputeAcked(self: *Model, durable_len: u64) void {
while (self.acked_count < self.frames.items.len and
self.frames.items[self.acked_count].frame_end <= durable_len)
{
self.acked_count += 1;
}
}
/// Drops every frame the log didn't actually keep, matching what a
/// fresh recovery just proved survived.
fn truncateTo(self: *Model, gpa: std.mem.Allocator, survived: usize) void {
std.debug.assert(survived <= self.frames.items.len);
for (self.frames.items[survived..]) |f| gpa.free(f.bytes);
self.frames.shrinkRetainingCapacity(survived);
self.acked_count = @min(self.acked_count, survived);
}
};
pub const SweepError = error{
OutOfMemory,
Io,
/// Invariant 1: recovery lost a frame the model had proof was durable.
MissingAcknowledged,
/// Invariant 1/2: a frame beyond the determined survived boundary is
/// nonetheless present -- not a clean prefix -- or a phantom-only
/// hash was found durable (phantoms must never touch disk).
PhantomFrame,
/// Invariant 2: the post-recovery structural size disagrees with the
/// model's own record of where the last surviving frame ended.
IndexMismatch,
/// Invariant 1/4: a surviving frame's content or hash disagrees with
/// what was written.
HashMismatch,
/// Invariant 4: a corrupt offset lies inside the recovered span --
/// validated data that isn't actually trustworthy.
SilentCorruption,
};
fn sha256(payload: []const u8) [hash_len]u8 {
var out: [hash_len]u8 = undefined;
std.crypto.hash.sha2.Sha256.hash(payload, &out, .{});
return out;
}
fn putOnce(gpa: std.mem.Allocator, store: *logstore.LogStore, model: *Model, random: std.Random) SweepError!void {
var buf: [64]u8 = undefined;
const n = 1 + random.uintLessThan(usize, buf.len - 1);
random.bytes(buf[0..n]);
const payload = buf[0..n];
std.debug.assert(payload.len <= logstore.max_payload / 1000); // sweep payloads stay well under the 1 GiB bound
const hash = sha256(payload);
const before = store.size();
if (store.put(&hash, payload)) |rid| {
const after = store.size();
// LogStore.put is idempotent for an already-content-bearing hash:
// it appends nothing and returns the existing rid (self.end
// unchanged). The sweep's small payload space (1..63 random
// bytes) makes that collision real within a single run, so only
// a put that actually landed a frame grows the model -- a
// same-hash repeat is already covered by the frame recorded the
// first time, and recording it again would fabricate a second
// model frame the log never wrote.
if (after != before) {
try model.recordPut(gpa, hash, payload, before, after, store.payloadLenOf(rid), 0);
}
} else |err| switch (err) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => {}, // torn/short write: LogStore already rolled back
}
}
/// A zlib-encoded put: the store compresses `payload` into a fossil zlib
/// container, but `content()` must still round-trip to the uncompressed
/// bytes, so the model records `payload` (uncompressed) and the on-disk
/// container length separately.
fn zlibOnce(gpa: std.mem.Allocator, store: *logstore.LogStore, model: *Model, random: std.Random) SweepError!void {
var buf: [64]u8 = undefined;
const n = 1 + random.uintLessThan(usize, buf.len - 1);
random.bytes(buf[0..n]);
const payload = buf[0..n];
std.debug.assert(payload.len <= logstore.max_payload / 1000);
const hash = sha256(payload);
const before = store.size();
if (store.putZlib(&hash, payload)) |rid| {
const after = store.size();
if (after != before) {
try model.recordPut(gpa, hash, payload, before, after, store.payloadLenOf(rid), 0);
}
} else |err| switch (err) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => {},
}
}
/// A delta-encoded put over an existing content-bearing base whose chain has
/// room (`depth < max_delta_chain`). The target shares a run with the base so
/// the encoder emits a copy; `content()` must resolve the chain back to the
/// target. Falls back to a raw put when no eligible base exists yet.
fn deltaOnce(gpa: std.mem.Allocator, store: *logstore.LogStore, model: *Model, random: std.Random) SweepError!void {
if (model.frames.items.len == 0) return putOnce(gpa, store, model, random);
const start = random.uintLessThan(usize, model.frames.items.len);
var base_idx: ?usize = null;
var k: usize = 0;
while (k < model.frames.items.len) : (k += 1) {
const idx = (start + k) % model.frames.items.len;
if (model.frames.items[idx].depth < logstore.max_delta_chain) {
base_idx = idx;
break;
}
}
const bi = base_idx orelse return putOnce(gpa, store, model, random);
const base = &model.frames.items[bi];
const base_rid = store.lookup(&base.hash) orelse return; // recorded frame must be live
const bn = base.bytes.len;
std.debug.assert(bn >= 1);
var tbuf: [64]u8 = undefined;
@memcpy(tbuf[0..bn], base.bytes);
tbuf[random.uintLessThan(usize, bn)] +%= 1; // one-byte mutation keeps a shared run
const target = tbuf[0..bn];
const raw = delta.create(gpa, base.bytes, target) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.TargetTooLarge => unreachable, // bn <= 63, far below 2^32
};
defer gpa.free(raw);
const hash = sha256(target);
const depth = base.depth + 1;
const before = store.size();
if (store.putDelta(&hash, base_rid, raw)) |rid| {
const after = store.size();
if (after != before) {
try model.recordPut(gpa, hash, target, before, after, store.payloadLenOf(rid), depth);
}
} else |err| switch (err) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => {},
}
}
fn internOnce(gpa: std.mem.Allocator, store: *logstore.LogStore, model: *Model, random: std.Random) SweepError!void {
var seed_bytes: [16]u8 = undefined;
random.bytes(&seed_bytes);
const hash = sha256(&seed_bytes);
_ = try store.intern(&hash);
try model.recordPhantom(gpa, hash);
}
/// Content arrives for a hash the store already knows only as a phantom: the
/// core ArtifactStore lifecycle (intern a hash, put its bytes later). Picks a
/// live phantom so `put` takes the phantom-upgrade branch, then folds the
/// upgraded frame into the durability model exactly like any other put.
fn upgradeOnce(gpa: std.mem.Allocator, store: *logstore.LogStore, model: *Model, random: std.Random) SweepError!void {
std.debug.assert(model.phantoms.items.len > 0); // caller-guarded supply
const idx = random.uintLessThan(usize, model.phantoms.items.len);
const hash = model.phantoms.items[idx]; // live phantom -> hits put's upgrade branch
var buf: [64]u8 = undefined;
const n = 1 + random.uintLessThan(usize, buf.len - 1);
random.bytes(buf[0..n]);
const payload = buf[0..n];
std.debug.assert(payload.len <= logstore.max_payload / 1000);
// Content-addressing is the caller's contract; the sweep only requires that
// content() round-trips whatever bytes were put under this rid, so the
// payload need not hash to `hash`.
const before = store.size();
if (store.put(&hash, payload)) |rid| {
const after = store.size();
std.debug.assert(after != before); // a phantom upgrade always appends its first frame
try model.recordPut(gpa, hash, payload, before, after, store.payloadLenOf(rid), 0);
model.removePhantomAt(idx);
} else |err| switch (err) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => {}, // torn write rolled back: entry stays a phantom in store and model
}
}
/// Determines the survived prefix (recovery only ever drops a suffix),
/// then checks all four invariants against `model` and advances it to
/// match reality.
fn checkInvariants(
gpa: std.mem.Allocator,
faulty: *faulty_file.FaultyFile,
model: *Model,
store: *logstore.LogStore,
guaranteed: usize,
) SweepError!void {
const total = model.frames.items.len;
var survived: usize = 0;
while (survived < total and store.lookup(&model.frames.items[survived].hash) != null) : (survived += 1) {}
if (survived < guaranteed) return error.MissingAcknowledged;
var i: usize = survived;
while (i < total) : (i += 1) {
if (store.lookup(&model.frames.items[i].hash) != null) return error.PhantomFrame;
}
i = 0;
while (i < survived) : (i += 1) {
const f = &model.frames.items[i];
const rid = store.lookup(&f.hash) orelse return error.MissingAcknowledged;
if (!std.mem.eql(u8, store.hashOf(rid), &f.hash)) return error.HashMismatch;
const bytes = store.content(gpa, rid) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => return error.Io,
error.Corrupt => return error.HashMismatch,
} orelse return error.HashMismatch;
defer gpa.free(bytes);
if (!std.mem.eql(u8, bytes, f.bytes)) return error.HashMismatch;
}
for (model.phantoms.items) |h| {
if (store.lookup(&h) != null) return error.PhantomFrame;
}
// A reopen wiped every in-memory phantom (phantoms never touch disk, §C),
// so the live phantom set resets each crash.
model.phantoms.clearRetainingCapacity();
const want_size: u64 = if (survived == 0) logstore.header_len else model.frames.items[survived - 1].frame_end;
if (store.size() != want_size) return error.IndexMismatch;
// No corrupt offset may lie within a surviving frame's MEANINGFUL span
// (its crc-covered bytes, 22+hash_len+payload_len from frame_start) --
// that would mean validated data isn't actually trustworthy. Trailing
// zero-pad up to the 8-byte boundary is explicitly excluded from both
// crcs (logstore.zig) and never read back, so corruption landing only
// in pad is inert, not silent corruption.
var j: usize = 0;
while (j < survived) : (j += 1) {
const f = &model.frames.items[j];
// The crc-covered span is the fixed header plus the ON-DISK payload
// (container length for zlib/delta), not the uncompressed content.
const meaningful_end = f.frame_start + 22 + @as(u64, f.hash.len) + f.disk_payload_len;
for (faulty.corrupt.items) |off| {
if (off >= f.frame_start and off < meaningful_end) return error.SilentCorruption;
}
}
model.truncateTo(gpa, survived);
}
/// A corrupt offset is only ever the payload-derived bitflip fault
/// (storage-track-plan.md §B) -- nothing else in the simulator flips
/// bits. Repair by truncating below the earliest one: a header-region hit
/// (< header_len) makes the whole log untrustworthy (nothing downstream
/// of a bad header can be validated), so that wipes to empty; a
/// frame-region hit keeps every frame that ended before it.
fn repairAfterCorrupt(gpa: std.mem.Allocator, faulty: *faulty_file.FaultyFile, model: *Model) SweepError!void {
std.debug.assert(faulty.corrupt.items.len > 0);
var min_off: u64 = faulty.corrupt.items[0];
for (faulty.corrupt.items) |o| {
if (o < min_off) min_off = o;
}
if (min_off < logstore.header_len) {
faulty.storageFile().truncate(0) catch return error.Io;
model.truncateTo(gpa, 0);
return;
}
var safe: usize = 0;
while (safe < model.frames.items.len and model.frames.items[safe].frame_end <= min_off) : (safe += 1) {}
const cut: u64 = if (safe == 0) logstore.header_len else model.frames.items[safe - 1].frame_end;
faulty.storageFile().truncate(cut) catch return error.Io;
model.truncateTo(gpa, safe);
}
/// Simulates power loss, reopens, and checks the four invariants,
/// repairing (and rechecking) once if the open itself reports corruption.
/// Returns the freshly opened store, which the caller now owns.
fn crashCheckOnce(gpa: std.mem.Allocator, faulty: *faulty_file.FaultyFile, model: *Model) SweepError!logstore.LogStore {
faulty.crashRestart();
model.recomputeAcked(faulty.durable_len);
var opened = logstore.LogStore.open(gpa, faulty.storageFile(), 0);
if (opened) |*store| {
checkInvariants(gpa, faulty, model, store, model.acked_count) catch |e| {
store.deinit(gpa);
return e;
};
return store.*;
} else |err| switch (err) {
error.OutOfMemory => return error.OutOfMemory,
error.Storage => return error.Io,
error.Corrupt => {},
}
std.debug.assert(faulty.corrupt.items.len > 0);
try repairAfterCorrupt(gpa, faulty, model);
var store2 = logstore.LogStore.open(gpa, faulty.storageFile(), 0) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return error.Io,
};
checkInvariants(gpa, faulty, model, &store2, model.acked_count) catch |e| {
store2.deinit(gpa);
return e;
};
return store2;
}
/// The two-phase sweep (mirrors dst.zig's repostate phases): phase 1
/// drives random puts/interns/syncs/crashes with faults live, checking
/// the four recovery invariants after every crash; phase 2 heals the
/// disk and proves one final clean crash+recover converges on the full
/// model.
pub fn run(gpa: std.mem.Allocator, seed: u64, steps: usize) SweepError!void {
var prng = std.Random.DefaultPrng.init(seed);
const random = prng.random();
var faulty = faulty_file.FaultyFile.init(gpa, random);
defer faulty.deinit();
var model: Model = .{};
defer model.deinit(gpa);
// The very first open runs against a clean disk (cfg is still the
// all-zero, fault-free default) -- the four invariants are about
// *recovering* from a crash, not about whether the log can even be
// created under a lossy first write, and LogStore.open has no
// retry/tolerance path for that (nor should it: a fresh empty file
// is never "acknowledged" data to lose). Faults engage only once the
// store exists.
var store = logstore.LogStore.open(gpa, faulty.storageFile(), 0) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return error.Io,
};
faulty.cfg = .{
.short_write_pct = 8,
.lost_tail_sync_pct = 8,
.crash_cut_at_durable_pct = 15,
.crash_rot_pct = 6,
.sector_round = 512,
.sector_round_pct = 25,
};
const crash_every: usize = 41;
var step: usize = 0;
while (step < steps) : (step += 1) {
switch (random.uintLessThan(u8, 6)) {
0, 1 => putOnce(gpa, &store, &model, random) catch |e| {
store.deinit(gpa);
return e;
},
2 => zlibOnce(gpa, &store, &model, random) catch |e| {
store.deinit(gpa);
return e;
},
3 => deltaOnce(gpa, &store, &model, random) catch |e| {
store.deinit(gpa);
return e;
},
4 => internOnce(gpa, &store, &model, random) catch |e| {
store.deinit(gpa);
return e;
},
// Upgrade a live phantom when one exists, else keep the phantom
// supply flowing so the branch has fuel next time.
5 => if (model.phantoms.items.len > 0)
upgradeOnce(gpa, &store, &model, random) catch |e| {
store.deinit(gpa);
return e;
}
else
internOnce(gpa, &store, &model, random) catch |e| {
store.deinit(gpa);
return e;
},
else => unreachable, // uintLessThan(u8, 6) yields 0..5
}
if (random.boolean()) {
store.commit() catch {};
model.recomputeAcked(faulty.durable_len);
}
if (step % crash_every == crash_every - 1) {
store.deinit(gpa);
store = crashCheckOnce(gpa, &faulty, &model) catch |e| return e;
}
}
store.deinit(gpa);
// Phase 2 (liveness): faults healed, everything becomes durable, and a
// harmless crash (nothing left unsynced) must recover the full model.
faulty.cfg = .{};
{
var healed = logstore.LogStore.open(gpa, faulty.storageFile(), 0) catch |e| switch (e) {
error.OutOfMemory => return error.OutOfMemory,
else => return error.Io,
};
healed.commit() catch {};
healed.deinit(gpa);
}
model.recomputeAcked(faulty.durable_len);
std.debug.assert(model.acked_count == model.frames.items.len);
var final_store = try crashCheckOnce(gpa, &faulty, &model);
final_store.deinit(gpa);
std.debug.assert(model.acked_count == model.frames.items.len);
}