//! The narrow file capability the storage engine reads and writes
//! through. Five operations, one error, nothing else — not a wrapper over
//! all of `std.Io`. Kept internal (not re-exported from root.zig) until a
//! real consumer (LogStore) needs it; its unit tests reach `zig build
//! test` through the dst module, which imports this file directly.
//!
//! There is no RFC section behind this file: it is native storage-engine
//! design, not a port. The oracle is this capability's own crash-recovery
//! invariants, exercised by the DST sweep in sim/storage_sweep.zig: every
//! acknowledged put stays readable and hash-verifies; a torn tail
//! truncates to exactly the last complete frame; the rebuilt index
//! matches the write-time model; and corruption always surfaces as an
//! error, never as silently wrong bytes.
const std = @import("std");
/// The device failed. Corruption is deliberately NOT a `StorageError`: a
/// corrupt read returns wrong bytes successfully (short of hardware I/O
/// failure), and the frame layer above is what turns "wrong bytes" into a
/// decision — the crash-recovery invariant that corruption is always an
/// error, never silent data. One variant is the whole set — every other
/// outcome (short read, no-op truncate, EOF) is defined as ordinary data,
/// not an error.
pub const StorageError = error{Io};
/// A capability, not a resource: `context` is caller-owned and outlives
/// every call made through the vtable.
pub const StorageFile = struct {
context: *anyopaque,
vtable: *const VTable,
pub const VTable = struct {
append: *const fn (ctx: *anyopaque, bytes: []const u8) StorageError!void,
/// Never errors on short read — returns the count actually read,
/// `0..buf.len`, with 0 meaning "at or past EOF": the error is
/// defined out of existence for EOF rather than threaded through
/// every caller.
pread: *const fn (ctx: *anyopaque, offset: u64, buf: []u8) StorageError!usize,
sync: *const fn (ctx: *anyopaque) StorageError!void,
size: *const fn (ctx: *anyopaque) StorageError!u64,
/// A no-op (same length) or a repeat truncate is legal, never an
/// error.
truncate: *const fn (ctx: *anyopaque, len: u64) StorageError!void,
/// Re-read the file's true length as the append watermark. Called
/// by `LogStore.open` once the caller holds the store's writer lock
/// (`cli.openStoreAppend`), which is what makes the length final: an
/// append lands at the file's real end only if that end was read
/// after every other writer was excluded. A length recorded earlier (a
/// `stat` taken to build this binding) can be stale, and appending at
/// it writes over another writer's frames with valid ones no checksum
/// can tell from a clean write. A binding with a single writer by
/// construction (an in-memory store) makes this a no-op.
refresh_end: *const fn (ctx: *anyopaque) StorageError!void,
};
pub fn append(self: StorageFile, bytes: []const u8) StorageError!void {
return self.vtable.append(self.context, bytes);
}
pub fn pread(self: StorageFile, offset: u64, buf: []u8) StorageError!usize {
return self.vtable.pread(self.context, offset, buf);
}
pub fn sync(self: StorageFile) StorageError!void {
return self.vtable.sync(self.context);
}
pub fn size(self: StorageFile) StorageError!u64 {
return self.vtable.size(self.context);
}
pub fn truncate(self: StorageFile, len: u64) StorageError!void {
return self.vtable.truncate(self.context, len);
}
pub fn refreshEnd(self: StorageFile) StorageError!void {
return self.vtable.refresh_end(self.context);
}
};
/// Make the directory's entries durable: a create, rename or delete in it
/// survives a power loss only once the directory itself is fsynced.
///
/// Through the Io seam, not `std.c.fsync`: reaching for libc here made every
/// Linux build demand explicit libc linkage for one syscall. The reopen is
/// load-bearing on Linux too: a plain openDir hands back an O_PATH descriptor
/// there, which fsync rejects outright (BADF); asking for iterate rights
/// forces a real file descriptor that can sync.
pub fn syncDirectory(io: std.Io, directory: std.Io.Dir) error{DirectorySyncFailed}!void {
var real = directory.openDir(io, ".", .{ .iterate = true }) catch return error.DirectorySyncFailed;
defer real.close(io);
const as_file: std.Io.File = .{ .handle = real.handle, .flags = .{ .nonblocking = false } };
as_file.sync(io) catch return error.DirectorySyncFailed;
}
/// Production binding: a real file reached only through the injected
/// `std.Io` (design-seams rule — no ambient filesystem access). `end`
/// tracks the append watermark so `append` never needs a `size` round trip.
pub const FileStorage = struct {
io: std.Io,
file: std.Io.File,
end: u64,
/// A read-only binding keeps its accepted descriptor length forever.
/// Refresh cannot promote later appends into this snapshot.
read_only: bool = false,
/// Optional block cache in front of `pread`, attached by `useReadCache`.
/// Null by default, so every construction site that does not opt in behaves
/// exactly as it did before this existed — which is why it is a field rather
/// than an `init` parameter (there are ~200 of those sites, and almost none
/// of them are on a read-heavy path).
read_cache: ?*ReadCache = null,
/// Precondition: `initial_size` equals `file`'s current length. Not
/// verified here — doing so would mean a stat syscall inside a plain
/// constructor, and TigerStyle assertions must not carry I/O side
/// effects. A caller that gets this wrong sees the mismatch surface
/// downstream, as a wrong `size()`/`pread` result.
pub fn init(io: std.Io, file: anytype, initial_size: u64) FileStorage {
// A captured read handle must not lose its bound at a generic caller.
if (@TypeOf(file) != std.Io.File) {
std.debug.assert(initial_size <= file.size);
return initReadOnly(io, file.file, initial_size);
}
return .{ .io = io, .file = file, .end = initial_size };
}
/// The caller retains `file` until every reader of this binding finishes.
pub fn initReadOnly(io: std.Io, file: std.Io.File, accepted_size: u64) FileStorage {
return .{ .io = io, .file = file, .end = accepted_size, .read_only = true };
}
pub fn storageFile(self: *FileStorage) StorageFile {
return .{ .context = self, .vtable = &vtable };
}
/// Serve reads through `cache`. The cache is caller-owned and must outlive
/// this storage — the same ownership shape `logstore.PlaintextCache` uses,
/// and for the same reason: whoever can prove the keys still mean what they
/// meant owns it.
pub fn useReadCache(self: *FileStorage, cache: *ReadCache) void {
// Even a reused binding address can now name a different descriptor.
// Invalidate at attachment as well as when readers switch bindings.
cache.invalidate();
cache.binding = self;
self.read_cache = cache;
}
const vtable: StorageFile.VTable = .{
.append = appendImpl,
.pread = preadImpl,
.sync = syncImpl,
.size = sizeImpl,
.truncate = truncateImpl,
.refresh_end = refreshEndImpl,
};
/// The file's length now, which the caller's writer lock holds still.
fn refreshEndImpl(ctx: *anyopaque) StorageError!void {
const self: *FileStorage = @ptrCast(@alignCast(ctx));
if (self.read_only) return;
const st = self.file.stat(self.io) catch return error.Io;
// A block read before the lock may have been the file's short last
// block, since grown by another writer: start the cache clean.
if (self.read_cache) |c| c.invalidate();
self.end = st.size;
}
fn appendImpl(ctx: *anyopaque, bytes: []const u8) StorageError!void {
const self: *FileStorage = @ptrCast(@alignCast(ctx));
if (self.read_only) return error.Io;
self.file.writePositionalAll(self.io, bytes, self.end) catch return error.Io;
self.end += bytes.len;
}
fn preadImpl(ctx: *anyopaque, offset: u64, buf: []u8) StorageError!usize {
const self: *FileStorage = @ptrCast(@alignCast(ctx));
if (offset >= self.end) return 0;
const bounded = buf[0..@min(buf.len, self.end - offset)];
if (self.read_cache) |c| {
if (c.binding != self) {
c.invalidate();
c.binding = self;
}
return c.pread(self.io, self.file, offset, bounded);
}
return self.file.readPositionalAll(self.io, bounded, offset) catch error.Io;
}
fn syncImpl(ctx: *anyopaque) StorageError!void {
const self: *FileStorage = @ptrCast(@alignCast(ctx));
self.file.sync(self.io) catch return error.Io;
}
fn sizeImpl(ctx: *anyopaque) StorageError!u64 {
const self: *FileStorage = @ptrCast(@alignCast(ctx));
return self.end;
}
fn truncateImpl(ctx: *anyopaque, len: u64) StorageError!void {
const self: *FileStorage = @ptrCast(@alignCast(ctx));
if (self.read_only) return error.Io;
// Truncation is the ONE event that can make a cached block stale: it is
// the only way a byte offset below the append watermark comes to name
// different content. `logstore.clearBaseMemo` exists for exactly this
// hazard at a higher layer; this is the same invalidation at the bottom.
if (self.read_cache) |c| c.invalidate();
self.file.setLength(self.io, len) catch return error.Io;
// setLength makes the file exactly `len` bytes — it both shrinks and
// extends — so the append watermark is `len`, not `@min(end, len)`,
// which would leave `end` stale (and the next append overwriting the
// extended tail) whenever `len` grows the file.
self.end = len;
}
};
test "storage: reads cannot cross the binding watermark, including cache hits" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
const len = ReadCache.block_len * 2;
var raw = try seedPattern(io, tmp.dir, "bounded.bin", len);
defer raw.close(io);
var fs = FileStorage.init(io, raw, 7);
var rc = try ReadCache.init(std.testing.allocator, 1);
defer rc.deinit();
fs.useReadCache(&rc);
var bytes: [32]u8 = undefined;
try std.testing.expectEqual(@as(usize, 4), try fs.storageFile().pread(3, &bytes));
try std.testing.expectEqual(@as(usize, 0), try fs.storageFile().pread(7, &bytes));
try std.testing.expectEqual(@as(usize, 4), try fs.storageFile().pread(3, &bytes));
}
test "storage: sharing a read cache cannot mix descriptors at the same offsets" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
const len = ReadCache.block_len;
var a = try seedPattern(io, tmp.dir, "a.bin", len);
defer a.close(io);
var b = try seedPattern(io, tmp.dir, "b.bin", len);
defer b.close(io);
try b.writePositionalAll(io, "DIFFERENT", 0);
var fa = FileStorage.init(io, a, len);
var fb = FileStorage.init(io, b, len);
var rc = try ReadCache.init(std.testing.allocator, 1);
defer rc.deinit();
fa.useReadCache(&rc);
fb.useReadCache(&rc);
var bytes: [9]u8 = undefined;
_ = try fa.storageFile().pread(0, &bytes);
_ = try fb.storageFile().pread(0, &bytes);
try std.testing.expectEqualStrings("DIFFERENT", &bytes);
_ = try fa.storageFile().pread(0, &bytes);
try std.testing.expectEqual(@as(u8, 0), bytes[0]);
}
test "storage: read-only snapshots neither grow on refresh nor permit mutation" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var raw = try tmp.dir.createFile(io, "snapshot.bin", .{ .read = true });
defer raw.close(io);
try raw.writePositionalAll(io, "old", 0);
var fs = FileStorage.initReadOnly(io, raw, 3);
const snapshot = fs.storageFile();
try raw.writePositionalAll(io, "new", 3);
try snapshot.refreshEnd();
try std.testing.expectEqual(@as(u64, 3), try snapshot.size());
var bytes: [6]u8 = undefined;
try std.testing.expectEqual(@as(usize, 3), try snapshot.pread(0, &bytes));
try std.testing.expectEqualStrings("old", bytes[0..3]);
try std.testing.expectError(error.Io, snapshot.append("x"));
try std.testing.expectError(error.Io, snapshot.truncate(0));
try std.testing.expectEqual(@as(u64, 6), (try raw.stat(io)).size);
}
test "storage §B: append then pread round-trips through the prod binding" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var raw = try tmp.dir.createFile(io, "storage-test.bin", .{ .read = true });
defer raw.close(io);
var fs = FileStorage.init(io, raw, 0);
const file = fs.storageFile();
try file.append("hello");
var buf: [5]u8 = undefined;
const n = try file.pread(0, &buf);
try std.testing.expectEqual(@as(usize, 5), n);
try std.testing.expectEqualStrings("hello", &buf);
}
test "storage §B: size tracks appended length and truncate cuts the tail exactly" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var raw = try tmp.dir.createFile(io, "storage-test.bin", .{ .read = true });
defer raw.close(io);
var fs = FileStorage.init(io, raw, 0);
const file = fs.storageFile();
try file.append("0123456789");
try std.testing.expectEqual(@as(u64, 10), try file.size());
try file.truncate(4);
try std.testing.expectEqual(@as(u64, 4), try file.size());
var buf: [4]u8 = undefined;
const n = try file.pread(0, &buf);
try std.testing.expectEqual(@as(usize, 4), n);
try std.testing.expectEqualStrings("0123", &buf);
}
test "storage §B: sync on a healthy file returns without error" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var raw = try tmp.dir.createFile(io, "storage-test.bin", .{ .read = true });
defer raw.close(io);
var fs = FileStorage.init(io, raw, 0);
const file = fs.storageFile();
try file.append("payload");
try file.sync();
}
test "storage §B: pread at/after EOF returns a short count, not an error" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
var raw = try tmp.dir.createFile(io, "storage-test.bin", .{ .read = true });
defer raw.close(io);
var fs = FileStorage.init(io, raw, 0);
const file = fs.storageFile();
try file.append("ab");
var buf: [4]u8 = undefined;
const n = try file.pread(2, &buf); // exactly at EOF
try std.testing.expectEqual(@as(usize, 0), n);
const n2 = try file.pread(10, &buf); // well past EOF
try std.testing.expectEqual(@as(usize, 0), n2);
}
/// A bounded block cache in front of `FileStorage.pread`.
///
/// The store is APPEND-ONLY, and that is what makes this sound. A block is
/// retained only when the read that filled it returned a FULL block — which
/// proves the file already held every one of those bytes. Appends only ever
/// write at or beyond the current end, so they cannot alter a byte a full block
/// already covers, and therefore need no invalidation at all. Only `truncate`
/// can make an offset name different content, and it invalidates everything.
///
/// A short read is never cached. That is the same rule stated from the other
/// side: a partial block is one that reached end-of-file, so the bytes past it
/// are exactly the ones a later append will write.
///
/// Deleting this cache can only change speed, never any answer.
pub const ReadCache = struct {
/// Block granularity. Large enough that one miss amortises a syscall across
/// many small nearby reads (a frame header, a payload, its CRC trailer),
/// small enough that a miss does not read far more than it will use.
pub const block_len: usize = 64 << 10;
/// Default resident blocks: 4 MiB. The store's reads are offset-clustered —
/// a delta chain walks frames written near each other — so a small window
/// captures most of the locality there is.
pub const default_blocks: usize = 64;
const Block = struct {
base: u64 = 0,
valid: bool = false,
last_used: u64 = 0,
data: []u8,
};
gpa: std.mem.Allocator,
blocks: []Block,
tick: u64 = 0,
/// A block offset is meaningful only within one retained binding.
binding: ?*FileStorage = null,
/// Observability only; never gates a decision.
hits: u64 = 0,
misses: u64 = 0,
/// Reads served straight from the file because they were at least a block
/// wide. Caching those would evict many small entries to hold bytes the
/// caller already has, and a large read amortises its own syscall.
bypassed: u64 = 0,
pub fn init(gpa: std.mem.Allocator, n_blocks: usize) error{OutOfMemory}!ReadCache {
std.debug.assert(n_blocks >= 1);
const blocks = try gpa.alloc(Block, n_blocks);
errdefer gpa.free(blocks);
var made: usize = 0;
errdefer for (blocks[0..made]) |b| gpa.free(b.data);
while (made < n_blocks) : (made += 1) {
blocks[made] = .{ .data = try gpa.alloc(u8, block_len) };
}
return .{ .gpa = gpa, .blocks = blocks };
}
pub fn deinit(self: *ReadCache) void {
for (self.blocks) |b| self.gpa.free(b.data);
self.gpa.free(self.blocks);
self.* = undefined;
}
/// Drop every retained block. Only `truncate` needs this.
pub fn invalidate(self: *ReadCache) void {
for (self.blocks) |*b| b.valid = false;
}
fn find(self: *ReadCache, base: u64) ?*Block {
for (self.blocks) |*b| if (b.valid and b.base == base) return b;
return null;
}
/// The least-recently-used block, or any still-unused one.
fn victim(self: *ReadCache) *Block {
var pick: *Block = &self.blocks[0];
for (self.blocks) |*b| {
if (!b.valid) return b;
if (b.last_used < pick.last_used) pick = b;
}
return pick;
}
fn touch(self: *ReadCache, b: *Block) void {
self.tick += 1;
b.last_used = self.tick;
}
/// `readPositionalAll` semantics: returns the number of bytes read, which is
/// short only at end of file.
pub fn pread(self: *ReadCache, io: std.Io, file: std.Io.File, offset: u64, buf: []u8) StorageError!usize {
if (buf.len >= block_len) {
self.bypassed += 1;
return file.readPositionalAll(io, buf, offset) catch error.Io;
}
var done: usize = 0;
while (done < buf.len) {
const want = offset + done;
const base = want - (want % block_len);
const off_in: usize = @intCast(want - base);
if (self.find(base)) |b| {
self.hits += 1;
self.touch(b);
const n = @min(buf.len - done, block_len - off_in);
@memcpy(buf[done..][0..n], b.data[off_in..][0..n]);
done += n;
continue;
}
self.misses += 1;
const slot = self.victim();
slot.valid = false; // not retained until proven whole
const got = file.readPositionalAll(io, slot.data, base) catch return error.Io;
if (got < block_len) {
// End of file inside this block: serve what exists and stop.
// Never retained — these are the bytes a later append rewrites.
if (got <= off_in) return done;
const n = @min(buf.len - done, got - off_in);
@memcpy(buf[done..][0..n], slot.data[off_in..][0..n]);
return done + n;
}
slot.base = base;
slot.valid = true;
self.touch(slot);
const n = @min(buf.len - done, block_len - off_in);
@memcpy(buf[done..][0..n], slot.data[off_in..][0..n]);
done += n;
}
return done;
}
};
// --- ReadCache -------------------------------------------------------------
// The cache may only ever change SPEED. Each test below pins one of the three
// facts that make that true, so a regression shows up as a wrong ANSWER in a
// test rather than as corrupt content in a repository.
/// A file of `len` bytes whose byte at index i is `i % 251` — distinct enough
/// that a block served from the wrong base is caught by value, not just length.
fn seedPattern(io: std.Io, dir: std.Io.Dir, name: []const u8, len: usize) !std.Io.File {
var f = try dir.createFile(io, name, .{ .read = true });
const bytes = try std.testing.allocator.alloc(u8, len);
defer std.testing.allocator.free(bytes);
for (bytes, 0..) |*b, i| b.* = @intCast(i % 251);
try f.writePositionalAll(io, bytes, 0);
return f;
}
test "storage: a cached read returns the same bytes as an uncached one, across block seams" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
const len = ReadCache.block_len * 3 + 777;
var raw = try seedPattern(io, tmp.dir, "rc.bin", len);
defer raw.close(io);
var plain = FileStorage.init(io, raw, len);
var cached = FileStorage.init(io, raw, len);
var rc = try ReadCache.init(std.testing.allocator, 4);
defer rc.deinit();
cached.useReadCache(&rc);
// Offsets chosen to straddle block boundaries in both directions, which is
// where an off-by-one in the base arithmetic would show up.
const offsets = [_]u64{ 0, 1, ReadCache.block_len - 1, ReadCache.block_len, ReadCache.block_len + 5, 2 * ReadCache.block_len - 3, len - 10 };
for (offsets) |off| {
var a: [64]u8 = undefined;
var b: [64]u8 = undefined;
const na = try plain.storageFile().pread(off, &a);
const nb = try cached.storageFile().pread(off, &b);
try std.testing.expectEqual(na, nb);
try std.testing.expectEqualSlices(u8, a[0..na], b[0..nb]);
}
// Repeating them must still agree — this is the pass that actually hits.
try std.testing.expect(rc.hits > 0);
for (offsets) |off| {
var a: [64]u8 = undefined;
var b: [64]u8 = undefined;
const na = try plain.storageFile().pread(off, &a);
const nb = try cached.storageFile().pread(off, &b);
try std.testing.expectEqualSlices(u8, a[0..na], b[0..nb]);
try std.testing.expectEqual(na, nb);
}
}
test "storage: appending after a cached read does not serve stale bytes" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
// Shorter than one block, so the first read is a SHORT read — precisely the
// case that must not be retained, because the bytes just past it are the
// ones the append is about to write.
var raw = try seedPattern(io, tmp.dir, "rc.bin", 100);
defer raw.close(io);
var fs = FileStorage.init(io, raw, 100);
var rc = try ReadCache.init(std.testing.allocator, 4);
defer rc.deinit();
fs.useReadCache(&rc);
const file = fs.storageFile();
var before: [100]u8 = undefined;
try std.testing.expectEqual(@as(usize, 100), try file.pread(0, &before));
try file.append("APPENDED");
// The appended bytes must be visible, not masked by a retained partial block.
var after: [8]u8 = undefined;
try std.testing.expectEqual(@as(usize, 8), try file.pread(100, &after));
try std.testing.expectEqualStrings("APPENDED", &after);
}
test "storage: truncation invalidates the cache rather than renaming old bytes" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
const len = ReadCache.block_len * 2;
var raw = try seedPattern(io, tmp.dir, "rc.bin", len);
defer raw.close(io);
var fs = FileStorage.init(io, raw, len);
var rc = try ReadCache.init(std.testing.allocator, 4);
defer rc.deinit();
fs.useReadCache(&rc);
const file = fs.storageFile();
// Warm a full block, so offset 0 is genuinely resident.
var warm: [32]u8 = undefined;
_ = try file.pread(0, &warm);
// Truncate away everything and write DIFFERENT content at the same offsets:
// the hazard `logstore.clearBaseMemo` guards at a higher layer, where an old
// offset comes to name a new frame.
try file.truncate(0);
try file.append("XXXXXXXXXXXXXXXX");
var got: [16]u8 = undefined;
try std.testing.expectEqual(@as(usize, 16), try file.pread(0, &got));
try std.testing.expectEqualStrings("XXXXXXXXXXXXXXXX", &got);
}
test "storage: a read at least one block wide bypasses the cache" {
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const io = std.testing.io;
const len = ReadCache.block_len * 2;
var raw = try seedPattern(io, tmp.dir, "rc.bin", len);
defer raw.close(io);
var fs = FileStorage.init(io, raw, len);
var rc = try ReadCache.init(std.testing.allocator, 4);
defer rc.deinit();
fs.useReadCache(&rc);
const big = try std.testing.allocator.alloc(u8, ReadCache.block_len);
defer std.testing.allocator.free(big);
_ = try fs.storageFile().pread(0, big);
// Retaining it would evict many small entries to hold bytes the caller
// already has, and a read this size amortises its own syscall.
try std.testing.expectEqual(@as(u64, 1), rc.bypassed);
try std.testing.expectEqual(@as(u64, 0), rc.misses);
for (big, 0..) |b, i| try std.testing.expectEqual(@as(u8, @intCast(i % 251)), b);
}