//! A trivial reference append-log (NOT the eventual LogStore framing) plus
//! the DST sweep that drives it through FaultyFile. Just enough framing —
//! `len | payload | crc32(len||payload)` — to exercise torn-tail vs.
//! corruption recovery (storage-track-plan.md §B).
//!
//! Native design, no RFC section. Oracle: storage-track-plan.md §B and the
//! four crash-recovery invariants it enumerates. `run` is exercised by
//! `zig build dst`, not a `zig build test` unit — the sweep IS the gate.
const std = @import("std");
const storage = @import("../storage.zig");
const faulty_file = @import("faulty_file.zig");
pub const Frame = struct {
pub const header_len = 4; // len:u32 LE
pub const crc_len = 4; // crc32(len||payload), LE
pub const min_len = header_len + crc_len; // 8-byte floor, empty payload
pub const max_payload = 4096; // explicit bound (storage-track-plan.md §B)
/// The most bytes a single append() call can ever contribute, hence the
/// most a genuine torn write can ever leave dangling at the true tail.
pub const max_len = header_len + max_payload + crc_len;
};
fn encodeInto(buf: []u8, payload: []const u8) []const u8 {
std.debug.assert(payload.len <= Frame.max_payload);
std.debug.assert(buf.len >= Frame.header_len + payload.len + Frame.crc_len);
std.mem.writeInt(u32, buf[0..4], @intCast(payload.len), .little);
@memcpy(buf[4..][0..payload.len], payload);
var crc = std.hash.Crc32.init();
crc.update(buf[0 .. 4 + payload.len]);
std.mem.writeInt(u32, buf[4 + payload.len ..][0..4], crc.final(), .little);
return buf[0 .. 4 + payload.len + 4];
}
pub fn append(file: storage.StorageFile, payload: []const u8) storage.StorageError!void {
std.debug.assert(payload.len <= Frame.max_payload);
var buf: [Frame.header_len + Frame.max_payload + Frame.crc_len]u8 = undefined;
try file.append(encodeInto(&buf, payload));
}
pub fn commit(file: storage.StorageFile) storage.StorageError!void {
return file.sync();
}
pub const IndexEntry = struct { offset: u64, len: u32 };
pub const Index = struct {
entries: std.ArrayList(IndexEntry) = .empty,
pub fn deinit(self: *Index, gpa: std.mem.Allocator) void {
self.entries.deinit(gpa);
self.* = undefined;
}
};
pub const RecoverError = error{ Io, Corrupt, OutOfMemory };
/// Scans from offset 0, rebuilding the index. An incomplete final frame
/// (short read, or a declared length that runs past the file's end) is a
/// torn tail: truncate to the prior boundary and stop, cleanly. A COMPLETE
/// frame whose crc mismatches is corruption: never truncate over it, never
/// return its bytes — surface `error.Corrupt` instead (storage-track-
/// plan.md §B, the torn-vs-corrupt split).
///
/// The two are only ambiguous when the length field itself doesn't decode
/// to something that fits: that happens both for a genuine torn write (the
/// length field landed but payload/crc didn't) AND for a bitflip on an
/// otherwise-complete, non-final frame's length field. They're told apart
/// by how much file remains: append() only ever writes one frame per call,
/// so a genuine torn write can leave at most one frame's worth of trailing
/// garbage. If more than that remains past this point, this can't be a
/// torn write — some earlier, complete frame's header was corrupted.
pub fn recover(gpa: std.mem.Allocator, file: storage.StorageFile) RecoverError!Index {
const total = try file.size();
var entries: std.ArrayList(IndexEntry) = .empty;
errdefer entries.deinit(gpa);
const max_frames = total / Frame.min_len + 1; // explicit loop bound
var frames: u64 = 0;
var offset: u64 = 0;
while (offset < total) {
frames += 1;
std.debug.assert(frames <= max_frames);
var header: [Frame.header_len]u8 = undefined;
if (try file.pread(offset, &header) < Frame.header_len) {
// Not even a full length field landed — this position can only
// be the tail of a torn write.
try file.truncate(offset);
break;
}
const payload_len = std.mem.readInt(u32, &header, .little);
const frame_len: u64 = Frame.header_len + @as(u64, payload_len) + Frame.crc_len;
if (payload_len > Frame.max_payload or offset + frame_len > total) {
if (total - offset < Frame.max_len) {
try file.truncate(offset);
break;
}
return error.Corrupt;
}
var payload_buf: [Frame.max_payload]u8 = undefined;
const payload = payload_buf[0..payload_len];
// The bounds check above guarantees `total` covers the rest of this
// frame, so a short read here would mean `file.size()` lied —
// an internal StorageFile inconsistency, not a recovery scenario.
const payload_got = try file.pread(offset + Frame.header_len, payload);
std.debug.assert(payload_got == payload_len);
var trailer: [Frame.crc_len]u8 = undefined;
const trailer_got = try file.pread(offset + Frame.header_len + payload_len, &trailer);
std.debug.assert(trailer_got == Frame.crc_len);
var crc = std.hash.Crc32.init();
crc.update(&header);
crc.update(payload);
if (crc.final() != std.mem.readInt(u32, &trailer, .little)) return error.Corrupt;
try entries.append(gpa, .{ .offset = offset, .len = payload_len });
const next_offset = offset + frame_len;
std.debug.assert(next_offset > offset); // strictly increasing: provable termination
offset = next_offset;
}
return .{ .entries = entries };
}
/// Ground-truth bookkeeping for the sweep: every payload ever appended, in
/// order, plus how many of them are known-durable ("acknowledged" — the
/// end of the frame passed under `durable_len` after a truthful sync).
/// `next_offset` always equals the storage file's true structural length
/// between steps (nothing torn, nothing pending outside a frame boundary).
const Model = struct {
all_payloads: std.ArrayList([]u8) = .empty,
frame_ends: std.ArrayList(u64) = .empty,
acked_count: usize = 0,
next_offset: u64 = 0,
fn deinit(self: *Model, gpa: std.mem.Allocator) void {
for (self.all_payloads.items) |p| gpa.free(p);
self.all_payloads.deinit(gpa);
self.frame_ends.deinit(gpa);
self.* = undefined;
}
fn recordAppend(self: *Model, gpa: std.mem.Allocator, payload: []const u8) error{OutOfMemory}!void {
const owned = try gpa.dupe(u8, payload);
errdefer gpa.free(owned);
try self.all_payloads.append(gpa, owned);
errdefer _ = self.all_payloads.pop();
self.next_offset += Frame.header_len + payload.len + Frame.crc_len;
try self.frame_ends.append(gpa, self.next_offset);
}
/// Invariant (a): acked_count only grows, and only past a frame whose
/// end offset the disk has actually promised durable.
fn recomputeAcked(self: *Model, durable_len: u64) void {
while (self.acked_count < self.frame_ends.items.len and
self.frame_ends.items[self.acked_count] <= durable_len)
{
self.acked_count += 1;
}
}
/// Drop everything the log didn't actually keep: `survived` frames
/// remain, matching what a fresh recover() just proved is on disk.
fn truncateTo(self: *Model, gpa: std.mem.Allocator, survived: usize) void {
std.debug.assert(survived <= self.all_payloads.items.len);
for (self.all_payloads.items[survived..]) |p| gpa.free(p);
self.all_payloads.shrinkRetainingCapacity(survived);
self.frame_ends.shrinkRetainingCapacity(survived);
self.next_offset = if (survived == 0) 0 else self.frame_ends.items[survived - 1];
self.acked_count = @min(self.acked_count, survived);
}
};
pub const SweepError = error{
OutOfMemory,
Io,
/// Invariant 1: recover() lost a frame the model had proof was durable.
MissingAcknowledged,
/// Invariant 1/2 upper bound: recover() returned a frame past anything
/// ever actually appended.
PhantomFrame,
/// Invariant 2/3: a surviving frame's offset/length, or the post-
/// recover file size, disagrees with the model's own bookkeeping.
IndexMismatch,
/// A surviving frame's declared length exceeded what pread returned.
ShortRead,
/// Invariant 1/4: a surviving frame's bytes don't match what was
/// written.
HashMismatch,
};
fn checkInvariants(
gpa: std.mem.Allocator,
faulty: *faulty_file.FaultyFile,
model: *Model,
idx: Index,
guaranteed: usize,
) SweepError!void {
const survived = idx.entries.items.len;
if (survived < guaranteed) return error.MissingAcknowledged;
if (survived > model.all_payloads.items.len) return error.PhantomFrame;
const file = faulty.storageFile();
var buf: [Frame.max_payload]u8 = undefined;
var i: usize = 0;
while (i < survived) : (i += 1) {
const entry = idx.entries.items[i];
const want_offset: u64 = if (i == 0) 0 else model.frame_ends.items[i - 1];
const want_payload = model.all_payloads.items[i];
if (entry.offset != want_offset or entry.len != want_payload.len) return error.IndexMismatch;
const got = buf[0..entry.len];
const n = file.pread(entry.offset + Frame.header_len, got) catch return error.Io;
if (n != entry.len) return error.ShortRead;
if (!std.mem.eql(u8, got, want_payload)) return error.HashMismatch;
}
const want_size: u64 = if (survived == 0) 0 else model.frame_ends.items[survived - 1];
const got_size = file.size() catch return error.Io;
if (got_size != want_size) return error.IndexMismatch;
model.truncateTo(gpa, survived);
}
/// A CRC mismatch is only ever the payload-derived bitflip fault
/// (storage-track-plan.md §B) — nothing else in this simulator flips bits.
/// Repair by truncating below the earliest corrupt offset, mirroring what
/// an operator does with a naive single-file WAL: accept loss from the
/// break point on, keep everything provably clean 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;
}
var safe: usize = 0;
while (safe < model.frame_ends.items.len and model.frame_ends.items[safe] <= min_off) : (safe += 1) {}
model.truncateTo(gpa, safe);
faulty.storageFile().truncate(model.next_offset) catch return error.Io;
}
fn recoverAndCheck(gpa: std.mem.Allocator, faulty: *faulty_file.FaultyFile, model: *Model, guaranteed: usize) SweepError!void {
const result = recover(gpa, faulty.storageFile());
if (result) |idx_val| {
var idx = idx_val;
defer idx.deinit(gpa);
try checkInvariants(gpa, faulty, model, idx, guaranteed);
} else |err| switch (err) {
error.Corrupt => try repairAfterCorrupt(gpa, faulty, model),
error.Io => return error.Io,
error.OutOfMemory => return error.OutOfMemory,
}
}
/// The largest prefix of `acked_count` that recover() is unconditionally
/// obligated to return. A frame within `Frame.max_len` bytes of the current
/// tail can't be told apart, on a corrupted header alone, from a genuine
/// torn write of that same size (storage-track-plan.md §B has no separate
/// header checksum) — real WAL implementations resolve that ambiguity in
/// favor of a clean, available recovery over refusing to start, so
/// recover() may safely (never returning wrong bytes) drop an ambiguous
/// tail frame instead of flagging it corrupt. Frames further back than
/// that are unambiguous: more complete data necessarily follows them, so a
/// broken header there can only be corruption, always caught.
fn guaranteedFloor(model: *const Model, post_crash_total: u64) usize {
var guaranteed = model.acked_count;
while (guaranteed > 0) {
const frame_end = model.frame_ends.items[guaranteed - 1];
if (frame_end <= post_crash_total and post_crash_total - frame_end >= Frame.max_len) break;
guaranteed -= 1;
}
return guaranteed;
}
fn crashCheckOnce(gpa: std.mem.Allocator, faulty: *faulty_file.FaultyFile, model: *Model) SweepError!void {
faulty.crashRestart();
model.recomputeAcked(faulty.durable_len);
const guaranteed = guaranteedFloor(model, faulty.bytes.items.len);
try recoverAndCheck(gpa, faulty, model, guaranteed);
}
fn appendOnce(gpa: std.mem.Allocator, faulty: *faulty_file.FaultyFile, 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];
const file = faulty.storageFile();
if (append(file, payload)) |_| {
try model.recordAppend(gpa, payload);
} else |_| {
// A short write left a partial frame fragment on disk; strip it
// back to the last clean boundary so the next append starts
// aligned again (torn-tail discipline applies to the writer too).
file.truncate(model.next_offset) catch {};
}
}
/// The two-phase sweep (mirrors dst.zig's repostate phases): phase 1 drives
/// random appends/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();
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,
};
var model: Model = .{};
defer model.deinit(gpa);
const crash_every: usize = 41;
var step: usize = 0;
while (step < steps) : (step += 1) {
try appendOnce(gpa, &faulty, &model, random);
if (random.boolean()) {
faulty.storageFile().sync() catch {};
model.recomputeAcked(faulty.durable_len);
}
if (step % crash_every == crash_every - 1) {
try crashCheckOnce(gpa, &faulty, &model);
}
}
// Phase 2 (liveness): faults healed, everything becomes durable, and a
// harmless crash (nothing left unsynced) must recover the full model.
faulty.cfg = .{};
faulty.storageFile().sync() catch {};
model.recomputeAcked(faulty.durable_len);
std.debug.assert(model.acked_count == model.all_payloads.items.len);
try crashCheckOnce(gpa, &faulty, &model);
std.debug.assert(model.acked_count == model.all_payloads.items.len);
}
test "storage_sweep §B(1): every acknowledged put is readable and hash-exact after crash+recover" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var raw = try tmp.dir.createFile(io, "reflog.bin", .{ .read = true });
defer raw.close(io);
var fs = storage.FileStorage.init(io, raw, 0);
const file = fs.storageFile();
const payloads = [_][]const u8{ "alpha", "beta", "gamma" };
for (payloads) |p| try append(file, p);
try commit(file);
var idx = try recover(std.testing.allocator, file);
defer idx.deinit(std.testing.allocator);
try std.testing.expectEqual(payloads.len, idx.entries.items.len);
var buf: [Frame.max_payload]u8 = undefined;
for (idx.entries.items, payloads) |entry, want| {
const got = buf[0..entry.len];
_ = try file.pread(entry.offset + Frame.header_len, got);
try std.testing.expectEqualStrings(want, got);
}
}
test "storage_sweep §B(2): a torn tail truncates to exactly the last valid frame boundary" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var raw = try tmp.dir.createFile(io, "reflog.bin", .{ .read = true });
defer raw.close(io);
var fs = storage.FileStorage.init(io, raw, 0);
const file = fs.storageFile();
try append(file, "one");
try append(file, "two");
const boundary = try file.size();
var frame_buf: [Frame.header_len + Frame.max_payload + Frame.crc_len]u8 = undefined;
const frame3 = encodeInto(&frame_buf, "three");
try file.append(frame3[0 .. frame3.len - 2]); // torn: crc trailer never fully landed
var idx = try recover(std.testing.allocator, file);
defer idx.deinit(std.testing.allocator);
try std.testing.expectEqual(@as(usize, 2), idx.entries.items.len);
try std.testing.expectEqual(boundary, try file.size());
}
test "storage_sweep §B(2): a garbage-length torn final frame is treated as torn tail, not corruption" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var raw = try tmp.dir.createFile(io, "reflog.bin", .{ .read = true });
defer raw.close(io);
var fs = storage.FileStorage.init(io, raw, 0);
const file = fs.storageFile();
try append(file, "one");
const boundary = try file.size();
try file.append(&[_]u8{ 0xff, 0xee }); // only 2 of the 4 length-field bytes landed
var idx = try recover(std.testing.allocator, file);
defer idx.deinit(std.testing.allocator);
try std.testing.expectEqual(@as(usize, 1), idx.entries.items.len);
try std.testing.expectEqual(boundary, try file.size());
}
test "storage_sweep §B(3): the index rebuilt from the log equals the model index" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var raw = try tmp.dir.createFile(io, "reflog.bin", .{ .read = true });
defer raw.close(io);
var fs = storage.FileStorage.init(io, raw, 0);
const file = fs.storageFile();
const payloads = [_][]const u8{ "a", "bb", "ccc", "dddd" };
var want: [payloads.len]IndexEntry = undefined;
var offset: u64 = 0;
for (payloads, 0..) |p, i| {
try append(file, p);
want[i] = .{ .offset = offset, .len = @intCast(p.len) };
offset += Frame.header_len + p.len + Frame.crc_len;
}
var idx = try recover(std.testing.allocator, file);
defer idx.deinit(std.testing.allocator);
try std.testing.expectEqual(want.len, idx.entries.items.len);
for (idx.entries.items, want) |got, w| {
try std.testing.expectEqual(w.offset, got.offset);
try std.testing.expectEqual(w.len, got.len);
}
}
test "storage_sweep §B(4): a corrupted acknowledged frame surfaces as error.Corrupt, never wrong bytes" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var raw = try tmp.dir.createFile(io, "reflog.bin", .{ .read = true });
defer raw.close(io);
var fs = storage.FileStorage.init(io, raw, 0);
const file = fs.storageFile();
var frame_buf: [Frame.header_len + Frame.max_payload + Frame.crc_len]u8 = undefined;
const good_frame1 = encodeInto(&frame_buf, "first");
var corrupted: [64]u8 = undefined;
@memcpy(corrupted[0..good_frame1.len], good_frame1);
corrupted[Frame.header_len] ^= 0xff; // flip a byte inside frame 1's payload
try file.append(corrupted[0..good_frame1.len]);
try append(file, "second");
try std.testing.expectError(error.Corrupt, recover(std.testing.allocator, file));
}
test "storage_sweep §B: a corrupted length field on the tail frame is an ambiguous torn tail, not corruption" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var raw = try tmp.dir.createFile(io, "reflog.bin", .{ .read = true });
defer raw.close(io);
var fs = storage.FileStorage.init(io, raw, 0);
const file = fs.storageFile();
try append(file, "one");
const boundary = try file.size();
var frame_buf: [Frame.header_len + Frame.max_payload + Frame.crc_len]u8 = undefined;
const good_frame = encodeInto(&frame_buf, "two");
var corrupted: [64]u8 = undefined;
@memcpy(corrupted[0..good_frame.len], good_frame);
corrupted[0] ^= 0xff; // flip a byte in the length field itself -- this is the tail frame, nothing follows it
try file.append(corrupted[0..good_frame.len]);
// A corrupted header this close to the true end is indistinguishable
// from a genuine torn write of the same size: recover() must resolve
// it as torn (available, no wrong bytes ever returned), never Corrupt.
var idx = try recover(std.testing.allocator, file);
defer idx.deinit(std.testing.allocator);
try std.testing.expectEqual(@as(usize, 1), idx.entries.items.len);
try std.testing.expectEqual(boundary, try file.size());
}
fn recoverNoLeak(gpa: std.mem.Allocator, file: storage.StorageFile) !void {
var idx = try recover(gpa, file);
idx.deinit(gpa);
}
test "storage_sweep §B(3): recover propagates allocation failure" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var raw = try tmp.dir.createFile(io, "reflog.bin", .{ .read = true });
defer raw.close(io);
var fs = storage.FileStorage.init(io, raw, 0);
const file = fs.storageFile();
try append(file, "a");
try append(file, "bb");
try std.testing.checkAllAllocationFailures(std.testing.allocator, recoverNoLeak, .{file});
}