//! Storage §C: the framed, append-only, content-addressed log over the
//! `StorageFile` capability (storage-track-plan.md §C). 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 — see
//! storage-slice.md's v2 rework note for why a v1 without it was unsound.
//!
//! No RFC section behind this file: native storage-engine design. The
//! oracle is storage-track-plan.md §C plus the four crash-recovery
//! invariants exercised by sim/storage_sweep.zig's DST sweep.
//!
//! 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("storage.zig");
/// 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 (storage-track-plan.md §C).
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;
/// Readers accept any stored version `<= version` (freeze seam, §E).
pub const version: u16 = 1;
/// Smallest possible frame span: hash_len=1, payload=0 ->
/// align8(26+1+0) = 32.
const min_frame: u64 = 32;
const magic = "zitL";
/// Payload encoding. MVP writes/reads `raw` only; `zlib`/`delta` are
/// reserved wire values a reader must fail closed on (deferred to the
/// delta slice).
pub const FrameKind = enum(u8) { raw = 0, zlib = 1, delta = 2, _ };
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
/// (storage-track-plan.md §C: "O(n)-on-open correct at this scale").
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,
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,
payload_len: u32,
/// A phantom was `intern`ed but never `put`: no frame on disk.
phantom: bool,
};
/// Opens (creating a fresh header on an empty file, or re-initializing
/// a torn one) and recovers the in-memory index by scanning frames.
/// `hash_kind` seeds a brand-new header; an existing header's stored
/// value wins on reopen (opaque, round-tripped, never compared).
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);
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;
}
/// 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, .phantom = true });
return rid;
}
/// Stores `bytes` under `hash`, appending one frame. 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(hash.len >= min_hash_len and hash.len <= max_hash_len);
std.debug.assert(bytes.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, bytes);
entry.offset = res.offset;
entry.payload_len = res.len;
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, bytes);
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.
self.file.truncate(res.offset) catch {};
self.end = res.offset;
return e;
};
self.entries.appendAssumeCapacity(.{ .hash = owned, .offset = res.offset, .payload_len = res.len, .phantom = false });
return rid;
}
/// Owned bytes for a rid, or null while it's a phantom. Re-verifies
/// both checksums and the embedded hash on every call (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;
const hl: usize = entry.hash.len;
const fixed_and_crc_len: usize = 22 + hl;
var hbuf: [22 + max_hash_len]u8 = undefined;
const n0 = self.file.pread(entry.offset, hbuf[0..fixed_and_crc_len]) catch return error.Storage;
if (n0 < fixed_and_crc_len) return error.Corrupt;
var hcrc = std.hash.Crc32.init();
hcrc.update(hbuf[0 .. 18 + hl]);
const stored_header_crc = std.mem.readInt(u32, hbuf[18 + hl ..][0..4], .little);
if (hcrc.final() != stored_header_crc) return error.Corrupt;
if (hbuf[4] != @intFromEnum(FrameKind.raw)) return error.Corrupt;
if (!std.mem.eql(u8, hbuf[10..][0..hl], entry.hash)) return error.Corrupt;
const pl: usize = entry.payload_len;
const payload = try gpa.alloc(u8, pl);
errdefer gpa.free(payload);
const payload_offset: u64 = entry.offset + fixed_and_crc_len;
const n1 = self.file.pread(payload_offset, payload) catch return error.Storage;
if (n1 < pl) return error.Corrupt;
var trailer: [4]u8 = undefined;
const n2 = self.file.pread(payload_offset + pl, &trailer) catch return error.Storage;
if (n2 < 4) return error.Corrupt;
var crc = std.hash.Crc32.init();
crc.update(hbuf[0..fixed_and_crc_len]);
crc.update(payload);
if (crc.final() != std.mem.readInt(u32, &trailer, .little)) return error.Corrupt;
return payload;
}
/// 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;
}
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;
}
/// 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.
fn appendFrame(self: *LogStore, hash: []const u8, bytes: []const u8) PutError!AppendResult {
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(FrameKind.raw));
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 |_| {
self.file.truncate(self.end) catch {};
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 },
};
/// 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 from storage-slice.md v2 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;
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;
std.debug.assert(got == take); // bounds check above guarantees `total` covers this
crc.update(chunk[0..take]);
pos += take;
remaining -= take;
}
var trailer: [4]u8 = undefined;
const nt = file.pread(pos, &trailer) catch return error.Storage;
std.debug.assert(nt == 4);
if (crc.final() != std.mem.readInt(u32, &trailer, .little)) return error.Corrupt;
if (flags != @intFromEnum(FrameKind.raw)) 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 } };
}
/// Rebuilds `self.entries`/`self.index` by scanning frames from
/// `header_len`. A torn tail truncates the file to the last valid
/// boundary and stops; a corrupt frame propagates `error.Corrupt`
/// unconditionally (never truncated over).
fn scan(self: *LogStore) 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 => {
self.file.truncate(offset) catch return error.Storage;
self.end = offset;
return;
},
.frame => |f| {
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, .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)).
fn encodeFrame(buf: []u8, hash: []const u8, payload: []const u8, flags: u8) []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], 0, .little); // delta_of = 0 this slice
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;
}
// ---------------------------------------------------------------------------
// Tests (plan section C — storage-track-plan.md / storage-slice.md; no RFC
// section behind this file).
// ---------------------------------------------------------------------------
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 bad magic" {
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, "XXXX", version, 0, false);
try std.testing.expectError(error.Corrupt, LogStore.open(std.testing.allocator, file, 0));
}
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: open re-initializes a torn file header instead of rejecting it" {
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 file.append(&[_]u8{ 'z', 'i', 't' }); // 3 of 64 header bytes: nowhere near complete
var store = try LogStore.open(std.testing.allocator, file, 3);
defer store.deinit(std.testing.allocator);
try std.testing.expectEqual(@as(u64, header_len), store.size());
try std.testing.expectEqual(@as(u8, 3), store.hash_kind);
}
test "logstore §C: put then content round-trips a raw frame" {
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 hash = "h" ** 32;
const rid = try store.put(hash, "hello world");
try std.testing.expectEqualStrings(hash, store.hashOf(rid));
const got = (try store.content(std.testing.allocator, rid)).?;
defer std.testing.allocator.free(got);
try std.testing.expectEqualStrings("hello world", got);
}
test "logstore §C: a written frame occupies an 8-byte-aligned span" {
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);
_ = try store.put("abcde", "xyz"); // unpadded = 26+5+3 = 34, align8 -> 40
try std.testing.expectEqual(@as(u64, 0), store.size() % 8);
try std.testing.expectEqual(header_len + 40, store.size());
}
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));
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: content hash is the uncompressed address; hashOf returns it" {
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 h1 = "1" ** 16;
const h2 = "2" ** 20;
const r1 = try store.put(h1, "aaa");
const r2 = try store.put(h2, "bbbbbb");
try std.testing.expectEqualStrings(h1, store.hashOf(r1));
try std.testing.expectEqualStrings(h2, store.hashOf(r2));
const c1 = (try store.content(std.testing.allocator, r1)).?;
defer std.testing.allocator.free(c1);
try std.testing.expectEqualStrings("aaa", c1);
}
test "logstore §C: put is idempotent for an equal hash and appends nothing" {
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 hash = "dup" ** 8;
const r1 = try store.put(hash, "payload-a");
const size_after_first = store.size();
const r2 = try store.put(hash, "payload-a");
try std.testing.expectEqual(r1, r2);
try std.testing.expectEqual(size_after_first, store.size());
}
test "logstore §C: intern yields a phantom; content is null and no frame is written" {
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 hash = "phantom-hash-example-value";
const size_before = store.size();
const rid = try store.intern(hash);
try std.testing.expectEqual(size_before, store.size());
const got = try store.content(std.testing.allocator, rid);
try std.testing.expectEqual(@as(?[]u8, null), got);
}
test "logstore §C: put upgrades an interned phantom in place, same rid, content now present" {
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);
// intern fixes the rid with no frame; content is null while phantom.
const h = "phantom-upgrade-hash-value-x";
const rid = try store.intern(h);
try std.testing.expectEqual(@as(?[]u8, null), try store.content(std.testing.allocator, rid));
const size0 = store.size();
// put on the same hash upgrades the phantom in place: same rid, and it
// appends the first frame (a phantom had none), so the log grows.
const rid2 = try store.put(h, "content-arrived");
try std.testing.expectEqual(rid, rid2);
try std.testing.expect(store.size() > size0);
{
const got = (try store.content(std.testing.allocator, rid)).?;
defer std.testing.allocator.free(got);
try std.testing.expectEqualStrings("content-arrived", got);
}
try std.testing.expectEqualStrings(h, store.hashOf(rid));
// The upgraded frame is real on disk, not just an in-memory flag: commit,
// reopen, and the hash resolves to content-bearing bytes (§B inv1 for a
// put that began life as a phantom).
try store.commit();
store.deinit(std.testing.allocator);
var reopened = try LogStore.open(std.testing.allocator, file, 0);
defer reopened.deinit(std.testing.allocator);
const rrid = reopened.lookup(h) orelse return error.TestUnexpectedResult;
const got2 = (try reopened.content(std.testing.allocator, rrid)).?;
defer std.testing.allocator.free(got2);
try std.testing.expectEqualStrings("content-arrived", got2);
try std.testing.expectEqualStrings(h, reopened.hashOf(rrid));
}
test "logstore §C(inv1): every committed put is readable and hash-exact after reopen" {
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();
const Item = struct { hash: []const u8, bytes: []const u8 };
const items = [_]Item{
.{ .hash = "h-one" ** 4, .bytes = "one" },
.{ .hash = "h-two" ** 4, .bytes = "two-payload" },
.{ .hash = "h-three" ** 3, .bytes = "three!" },
};
var store = try LogStore.open(std.testing.allocator, file, 0);
for (items) |it| _ = try store.put(it.hash, it.bytes);
try store.commit();
store.deinit(std.testing.allocator);
var reopened = try LogStore.open(std.testing.allocator, file, 0);
defer reopened.deinit(std.testing.allocator);
for (items) |it| {
const rid = reopened.lookup(it.hash) orelse return error.TestUnexpectedResult;
try std.testing.expectEqualStrings(it.hash, reopened.hashOf(rid));
const got = (try reopened.content(std.testing.allocator, rid)).?;
defer std.testing.allocator.free(got);
try std.testing.expectEqualStrings(it.bytes, got);
}
}
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));
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));
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(inv3): the index rebuilt from the log equals the written model" {
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();
const hashes = [_][]const u8{ "r1" ** 8, "r2" ** 8, "r3" ** 8, "r4" ** 8 };
const bodies = [_][]const u8{ "b1", "bb2", "bbb3", "bbbb4" };
var store = try LogStore.open(std.testing.allocator, file, 0);
for (hashes, bodies) |h, b| _ = try store.put(h, b);
try store.commit();
store.deinit(std.testing.allocator);
var reopened = try LogStore.open(std.testing.allocator, file, 0);
defer reopened.deinit(std.testing.allocator);
try std.testing.expectEqual(hashes.len, reopened.entries.items.len);
for (hashes, bodies) |h, b| {
const rid = reopened.lookup(h).?;
const got = (try reopened.content(std.testing.allocator, rid)).?;
defer std.testing.allocator.free(got);
try std.testing.expectEqualStrings(b, got);
}
}
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));
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));
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));
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));
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));
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(inv4): a crc-valid frame with a non-raw flag is rejected" {
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, "zlib-flagged-hash", "some payload bytes", @intFromEnum(FrameKind.zlib));
try file.append(f); // both crcs computed correctly over flags=zlib -- valid, just not raw
try std.testing.expectError(error.Corrupt, LogStore.open(std.testing.allocator, file, 0));
}
test "logstore §C: phantom hash is absent after reopen" {
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);
const hash = "phantom-only-hash-value";
_ = try store.intern(hash);
try store.commit();
store.deinit(std.testing.allocator);
var reopened = try LogStore.open(std.testing.allocator, file, 0);
defer reopened.deinit(std.testing.allocator);
try std.testing.expect(reopened.lookup(hash) == null);
}
fn openNoLeak(gpa: std.mem.Allocator, file: storage.StorageFile) !void {
var store = try LogStore.open(gpa, file, 0);
store.deinit(gpa);
}
test "logstore §C: open propagates allocation failure" {
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);
_ = try seed.put("seed-hash-one-value", "seed payload one");
_ = try seed.put("seed-hash-two-value", "seed payload two, a bit longer");
try seed.commit();
seed.deinit(std.testing.allocator);
try std.testing.checkAllAllocationFailures(std.testing.allocator, openNoLeak, .{file});
}
fn putNoLeak(gpa: std.mem.Allocator, file: storage.StorageFile) !void {
var store = try LogStore.open(gpa, file, 0);
defer store.deinit(gpa);
const size = file.size() catch 0;
var hash_buf: [8]u8 = undefined;
std.mem.writeInt(u64, &hash_buf, size, .little);
_ = try store.put(&hash_buf, "payload bytes exercised under simulated allocation failure");
}
test "logstore §C: put propagates allocation failure" {
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: put() always takes the new-entry path
try std.testing.checkAllAllocationFailures(std.testing.allocator, putNoLeak, .{file});
}
fn putPhantomUpgradeNoLeak(gpa: std.mem.Allocator, file: storage.StorageFile) !void {
var store = try LogStore.open(gpa, file, 0);
defer store.deinit(gpa);
// Hash off the current file size so every run targets a fresh hash: intern
// makes a live phantom this run, then put drives the phantom-upgrade branch
// (a fixed hash would already be content-bearing on disk after the counting
// run's append, sending put down the idempotent path instead).
const size = file.size() catch 0;
var hash_buf: [8]u8 = undefined;
std.mem.writeInt(u64, &hash_buf, size, .little);
_ = try store.intern(&hash_buf);
_ = try store.put(&hash_buf, "payload upgrading a phantom under simulated allocation failure");
}
test "logstore §C: put upgrading a phantom propagates allocation failure" {
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
// On OOM in the upgrade's frame append the entry stays phantom (its fields
// flip only after a successful append), so there is no leak and no
// half-upgraded state at any fail point.
try std.testing.checkAllAllocationFailures(std.testing.allocator, putPhantomUpgradeNoLeak, .{file});
}
fn contentNoLeak(gpa: std.mem.Allocator, file: storage.StorageFile) !void {
var store = try LogStore.open(gpa, file, 0);
defer store.deinit(gpa);
const bytes = try store.content(gpa, 1);
if (bytes) |b| gpa.free(b);
}
test "logstore §C: content propagates allocation failure" {
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);
_ = try seed.put("content-oom-test-hash", "payload bytes for the content allocation-failure test");
try seed.commit();
seed.deinit(std.testing.allocator);
try std.testing.checkAllAllocationFailures(std.testing.allocator, contentNoLeak, .{file});
}
test "logstore fuzz: open never crashes on garbage bytes appended after a valid header" {
var prng = std.Random.DefaultPrng.init(0xC0FFEE);
const random = prng.random();
const max_iters = 200;
var iter: usize = 0;
while (iter < max_iters) : (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 seed = try LogStore.open(std.testing.allocator, file, 0);
seed.deinit(std.testing.allocator);
var garbage: [256]u8 = undefined;
const n = random.uintLessThan(usize, garbage.len + 1);
random.bytes(garbage[0..n]);
try file.append(garbage[0..n]);
if (LogStore.open(std.testing.allocator, file, 0)) |store| {
var opened = store;
opened.deinit(std.testing.allocator);
} else |err| switch (err) {
error.OutOfMemory, error.Storage, error.Corrupt => {},
}
}
}