From b82a534bc8c90779d3138cd4564b15c3def90670 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Thu, 10 Sep 2026 17:21:14 +0200 Subject: [PATCH] feat: add decodeFrame and Replicator to replicate --- src/replicate.zig | 162 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 162 insertions(+) diff --git a/src/replicate.zig b/src/replicate.zig index 19ba0d5..adc1074 100644 --- a/src/replicate.zig +++ b/src/replicate.zig @@ -160,3 +160,165 @@ fn frameBodyLen(data: []const u8, bodyStart: usize) ?usize { return headerLen + keyLen + valLen; } +fn decodeFrame(allocator: Allocator, body: []const u8, out: *std.ArrayList(OwnedRecord)) !void { + if (body[8] == batchMarker) { + const opCount = encode.readU32(body, 9); + var p: usize = batchHeader; + var n: u32 = 0; + while (n < opCount) : (n += 1) { + const kind: Kind = @enumFromInt(body[p]); + const seq = encode.readU64(body, p + 1); + const keyLen = encode.readU32(body, p + 9); + const valLen = encode.readU32(body, p + 13); + const keyStart = p + opHeader; + const key = try allocator.dupe(u8, body[keyStart .. keyStart + keyLen]); + errdefer allocator.free(key); + const value = try allocator.dupe(u8, body[keyStart + keyLen .. keyStart + keyLen + valLen]); + try out.append(.{ .kind = kind, .seq = seq, .key = key, .value = value }); + p = keyStart + keyLen + valLen; + } + return; + } + const seq = encode.readU64(body, 0); + const kind: Kind = @enumFromInt(body[8]); + const keyLen = encode.readU32(body, 9); + const valLen = encode.readU32(body, 13); + const key = try allocator.dupe(u8, body[headerLen .. headerLen + keyLen]); + errdefer allocator.free(key); + const value = try allocator.dupe(u8, body[headerLen + keyLen .. headerLen + keyLen + valLen]); + try out.append(.{ .kind = kind, .seq = seq, .key = key, .value = value }); +} + +/// Ties a WalTailer to a target Db, applying newly tailed records as a single +/// WriteBatch per syncOnce(). Replicated writes get FRESH sequence numbers on +/// the target (Db.apply always assigns from its own nextSeq), so the replica +/// mirrors primary key/value state but not primary's original seq history - +/// same documented tradeoff as backup.zig's restore. +pub const Replicator = struct { + allocator: Allocator, + tailer: WalTailer, + target: *Db, + + pub fn init(allocator: Allocator, primaryDirPath: []const u8, target: *Db) !Replicator { + const gen = try readGenerationByPath(primaryDirPath); + return initFromOffset(allocator, primaryDirPath, target, 0, gen); + } + + pub fn initFromOffset(allocator: Allocator, primaryDirPath: []const u8, target: *Db, startOffset: u64, startGeneration: u64) !Replicator { + const tailer = try WalTailer.open(allocator, primaryDirPath, startOffset, startGeneration); + return .{ .allocator = allocator, .tailer = tailer, .target = target }; + } + + pub fn deinit(self: *Replicator) void { + self.tailer.deinit(); + } + + pub fn offset(self: *const Replicator) u64 { + return self.tailer.offset; + } + + pub fn generation(self: *const Replicator) u64 { + return self.tailer.generation; + } + + /// Applies every record newly visible in the primary's WAL since the last + /// call. Returns the number of records applied (0 if nothing new). + /// Propagates error.WalTruncatedAheadOfTailer if the primary flushed + /// (truncating its WAL) before this replicator consumed everything - the + /// caller must recover via fullResync(). + pub fn syncOnce(self: *Replicator) !usize { + const records = try self.tailer.poll(); + defer freeAll(self.allocator, records); + if (records.len == 0) return 0; + + var batch = WriteBatch.init(self.allocator); + defer batch.deinit(); + for (records) |record| { + switch (record.kind) { + .value => try batch.put(record.key, record.value), + .tombstone => try batch.delete(record.key), + } + } + try self.target.apply(&batch); + return records.len; + } + + /// Recovers from a fallen-behind tailer (or bootstraps a brand new + /// replica) by copying every live key/value from a fresh primary + /// Snapshot into the target in chunks, then re-bases the tailer at the + /// primary's current WAL offset. Snapshot() flushes the primary first + /// (see db.zig), so immediately after this call the primary's wal.log is + /// known to be empty - offset 0 is therefore always the correct restart + /// point. Requires no concurrent writer on the primary during the call + /// (same single-writer assumption as the rest of this project - Db has + /// no internal cross-thread synchronization). + pub fn fullResync(self: *Replicator, primary: *Db, chunkSize: usize) !usize { + const snap = try primary.snapshot(); + var iter = try snap.scan(self.allocator, null, null); + defer iter.deinit(); + + var batch = WriteBatch.init(self.allocator); + defer batch.deinit(); + var applied: usize = 0; + var staged: usize = 0; + while (try iter.next()) |entry| { + try batch.put(entry.key, entry.value); + staged += 1; + applied += 1; + if (staged >= chunkSize) { + try self.target.apply(&batch); + batch.deinit(); + batch = WriteBatch.init(self.allocator); + staged = 0; + } + } + if (staged > 0) try self.target.apply(&batch); + + self.tailer.offset = 0; + // snapshot() above flushes the primary if it had unflushed writes, + // which bumps its WAL generation (see wal.zig) - re-baseline here so + // the very next poll() does not immediately see a "changed" + // generation caused by this resync itself. + self.tailer.generation = try walmod.currentGeneration(primary.dir); + return applied; + } +}; + +const stateMagic = "CRPS"; +const stateFormatVersion: u32 = 2; + +/// A tail offset paired with the WAL generation it was captured against. +/// Persisting both (not just the offset) is what lets a brand-new process - +/// which never saw the in-memory generation check in WalTailer.poll() - tell +/// whether the primary truncated its WAL while this process was not running +/// at all, the same blind spot the in-memory check closes for a live one. +pub const PersistedState = struct { + offset: u64, + generation: u64, +}; +const stateFileName = "replicate.state"; +const stateTmpName = "replicate.state.tmp"; +const maxStateRead: usize = 64 * 1024; + +pub const StateError = error{ + BadMagic, + UnsupportedVersion, + Truncated, + Corrupt, + PrimaryMismatch, +}; + +/// Atomically persists a replicator's tail offset next to the replica's own +/// data (in replicaDirPath), so a later, brand-new `cobalt replicate --once` +/// process - which has no in-memory Replicator to resume from - can pick up +/// tailing exactly where the previous invocation left off instead of paying +/// for a full snapshot fullResync() every single time it starts. Also stores +/// primaryDirPath so a mismatched later call (pointed at a different primary +/// by mistake) is caught by loadOffset() rather than silently tailing the +/// wrong WAL from a stale offset. Written via tmp-file + rename (same pattern +/// as version.zig's writeCurrent), so a crash mid-write can never leave a +/// half-written state file where a good one used to be. +pub fn saveState(allocator: Allocator, replicaDirPath: []const u8, primaryDirPath: []const u8, offsetValue: u64, generationValue: u64) !void { + var dir = try std.fs.cwd().openDir(replicaDirPath, .{}); + defer dir.close(); +