feat: add readGenerationByPath and OwnedRecord

src/replicate.zig

@@ -0,0 +1,162 @@
+const std = @import("std");
+const encode = @import("encode.zig");
+const dbmod = @import("db.zig");
+const walmod = @import("wal.zig");
+
+const Allocator = std.mem.Allocator;
+const Kind = encode.Kind;
+const Db = dbmod.Db;
+const WriteBatch = dbmod.WriteBatch;
+
+// Physical WAL-tailing replication: reads the primary's on-disk wal.log file
+// directly (same binary frame format as wal.zig: [crc32 u32][seq u64][kind u8]
+// [keyLen u32][valLen u32][key][value], or a batch frame marked by kind byte
+// 0xB0 holding a count then per-op [kind][seq][keyLen][valLen][key][value]),
+// independently of wal.zig's private replay-into-memtable path, and applies
+// newly appended, crc-verified, complete frames onto a separate replica Db.
+//
+// This mirrors real physical log shipping (e.g. Postgres WAL streaming): it
+// only sees what is still in the WAL file. Db.flush() truncates the primary's
+// wal.log to empty once its memtable is durable in an SSTable (see db.zig
+// flush()), so a tailer that has fallen behind a flush can no longer recover
+// the truncated records from the WAL alone - poll() detects that condition
+// (current file size smaller than the tailer's already-consumed offset) and
+// fails loudly with error.WalTruncatedAheadOfTailer instead of silently
+// resuming from zero and dropping data. The caller is expected to recover via
+// fullResync(), which re-bases the replica from a live Snapshot (paying the
+// same "logical snapshot, not a physical replica of on-disk levels/history"
+// cost already documented for backup.zig) and restarts tailing from the
+// (now known-empty) WAL, exactly mirroring a real base-backup-plus-streaming
+// recovery: bulk-copy current state, then resume incremental log shipping
+// from the offset the bulk copy was taken at.
+
+const headerLen = 8 + 1 + 4 + 4;
+const batchMarker: u8 = 0xB0;
+const batchHeader = 8 + 1 + 4;
+const opHeader = 1 + 8 + 4 + 4;
+const walName = "wal.log";
+const maxTailRead: usize = 256 * 1024 * 1024;
+
+pub const ReplicateError = error{ WalTruncatedAheadOfTailer, WalGenerationChanged };
+
+/// Reads the WAL generation stamp (see wal.zig's currentGeneration) for a
+/// primary directory given only its path, for callers (WalTailer.poll,
+/// Replicator construction from a persisted offset) that do not hold a live
+/// *Db handle to the primary and so cannot reuse its already-open Dir.
+fn readGenerationByPath(primaryDirPath: []const u8) !u64 {
+    var dir = try std.fs.cwd().openDir(primaryDirPath, .{});
+    defer dir.close();
+    return walmod.currentGeneration(dir);
+}
+
+pub const OwnedRecord = struct {
+    kind: Kind,
+    seq: u64,
+    key: []u8,
+    value: []u8,
+
+    pub fn deinit(self: OwnedRecord, allocator: Allocator) void {
+        allocator.free(self.key);
+        allocator.free(self.value);
+    }
+};
+
+fn freeAll(allocator: Allocator, records: []OwnedRecord) void {
+    for (records) |record| record.deinit(allocator);
+    allocator.free(records);
+}
+
+/// Tails a single WAL file from a byte offset, decoding whole, crc-verified
+/// frames appended since the last poll. Never advances past an incomplete or
+/// corrupt tail frame (mirrors wal.zig's own torn-write tolerance), so a
+/// concurrent writer mid-append is simply not observed until its frame is
+/// fully flushed and synced.
+pub const WalTailer = struct {
+    allocator: Allocator,
+    file: std.fs.File,
+    offset: u64,
+    primaryDir: []u8,
+    generation: u64,
+
+    pub fn open(allocator: Allocator, primaryDirPath: []const u8, startOffset: u64, startGeneration: u64) !WalTailer {
+        const path = try std.fs.path.join(allocator, &.{ primaryDirPath, walName });
+        defer allocator.free(path);
+        const file = try std.fs.cwd().openFile(path, .{});
+        errdefer file.close();
+        const dirCopy = try allocator.dupe(u8, primaryDirPath);
+        return .{ .allocator = allocator, .file = file, .offset = startOffset, .primaryDir = dirCopy, .generation = startGeneration };
+    }
+
+    pub fn deinit(self: *WalTailer) void {
+        self.file.close();
+        self.allocator.free(self.primaryDir);
+    }
+
+    /// Returns newly observed records in append order. Caller owns the
+    /// returned slice and each record's key/value (free via freeAll or by
+    /// calling deinit() on each record then freeing the slice). Checks the
+    /// primary's WAL generation stamp before anything else: a changed
+    /// generation means the primary truncated (flushed) since the last poll,
+    /// which the caller must recover from via fullResync() regardless of how
+    /// the raw byte offsets happen to compare (see wal.zig's bumpGeneration
+    /// doc comment for why the byte-size comparison alone is not enough).
+    pub fn poll(self: *WalTailer) ![]OwnedRecord {
+        const nowGeneration = try readGenerationByPath(self.primaryDir);
+        if (nowGeneration != self.generation) return ReplicateError.WalGenerationChanged;
+
+        const endPos = try self.file.getEndPos();
+        if (endPos < self.offset) return ReplicateError.WalTruncatedAheadOfTailer;
+        if (endPos == self.offset) return &.{};
+
+        try self.file.seekTo(self.offset);
+        const data = try self.file.readToEndAlloc(self.allocator, maxTailRead);
+        defer self.allocator.free(data);
+
+        var out = std.ArrayList(OwnedRecord).init(self.allocator);
+        errdefer {
+            for (out.items) |record| record.deinit(self.allocator);
+            out.deinit();
+        }
+
+        var pos: usize = 0;
+        while (pos + 4 + batchHeader <= data.len) {
+            const storedCrc = encode.readU32(data, pos);
+            const bodyStart = pos + 4;
+            const bodyLen = frameBodyLen(data, bodyStart) orelse break;
+            if (bodyStart + bodyLen > data.len) break;
+            const body = data[bodyStart .. bodyStart + bodyLen];
+            if (encode.crc(body) != storedCrc) break;
+
+            try decodeFrame(self.allocator, body, &out);
+
+            pos = bodyStart + bodyLen;
+        }
+
+        self.offset += pos;
+        return try out.toOwnedSlice();
+    }
+};
+
+fn frameBodyLen(data: []const u8, bodyStart: usize) ?usize {
+    const marker = data[bodyStart + 8];
+    if (marker == batchMarker) {
+        if (bodyStart + batchHeader > data.len) return null;
+        const opCount = encode.readU32(data, bodyStart + 9);
+        var p = bodyStart + batchHeader;
+        var n: u32 = 0;
+        while (n < opCount) : (n += 1) {
+            if (p + opHeader > data.len) return null;
+            const keyLen = encode.readU32(data, p + 9);
+            const valLen = encode.readU32(data, p + 13);
+            const end = p + opHeader + keyLen + valLen;
+            if (end > data.len) return null;
+            p = end;
+        }
+        return p - bodyStart;
+    }
+    if (bodyStart + headerLen > data.len) return null;
+    const keyLen = encode.readU32(data, bodyStart + 9);
+    const valLen = encode.readU32(data, bodyStart + 13);
+    return headerLen + keyLen + valLen;
+}
+