feat: add decodeFrame and Replicator to replicate
This commit is contained in:
dualfroz committed
1 parent
203be45379
commit
b82a534bc8
1 file changed
+162
@@ -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();
|
||||
|
||||
Reference in new issue
Block a user