Skip to content
zit Artifact [c8caf93bb6]
Login

Artifact [c8caf93bb6]

Check-ins Using Download Hex

Artifact c8caf93bb63c99e0be959eaa9f439c328a4d39920dc93f084ef766306db3ba4a:

//! 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);
    }
};

/// 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,
    /// 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: std.Io.File, initial_size: u64) FileStorage {
        return .{ .io = io, .file = file, .end = initial_size };
    }

    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 {
        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));
        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));
        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 (self.read_cache) |c| return c.pread(self.io, self.file, offset, buf);
        return self.file.readPositionalAll(self.io, buf, 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));
        // 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 §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,
    /// 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);
}