From b33701a6ca2ba77218b51cdd0c25f68457877cdf Mon Sep 17 00:00:00 2001 From: DualFroz Date: Thu, 10 Sep 2026 17:27:54 +0200 Subject: [PATCH] feat: add loadOffset to replicate --- src/replicate.zig | 157 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 157 insertions(+) diff --git a/src/replicate.zig b/src/replicate.zig index adc1074..9f9e6ad 100644 --- a/src/replicate.zig +++ b/src/replicate.zig @@ -322,3 +322,160 @@ pub fn saveState(allocator: Allocator, replicaDirPath: []const u8, primaryDirPat var dir = try std.fs.cwd().openDir(replicaDirPath, .{}); defer dir.close(); + var body = std.ArrayList(u8).init(allocator); + defer body.deinit(); + const writer = encode.ByteWriter.init(&body); + try writer.slice(stateMagic); + try writer.putU32(stateFormatVersion); + try writer.putU64(offsetValue); + try writer.putU64(generationValue); + try writer.putU32(@intCast(primaryDirPath.len)); + try writer.slice(primaryDirPath); + try writer.putU32(encode.crc(body.items)); + + { + const file = try dir.createFile(stateTmpName, .{ .truncate = true }); + defer file.close(); + try file.writeAll(body.items); + try file.sync(); + } + try dir.rename(stateTmpName, stateFileName); +} + +/// Loads a previously persisted tail offset for a replica directory. Returns +/// null (not an error) when no state file has ever been written there yet - +/// callers treat that as "fresh replica, bootstrap via fullResync()" exactly +/// like before this existed. Returns StateError.PrimaryMismatch if the file +/// was written while tailing a different primary directory, so a replica +/// can never silently resume against the wrong primary from a stale offset. +pub fn loadOffset(allocator: Allocator, replicaDirPath: []const u8, primaryDirPath: []const u8) !?PersistedState { + var dir = try std.fs.cwd().openDir(replicaDirPath, .{}); + defer dir.close(); + + const data = dir.readFileAlloc(allocator, stateFileName, maxStateRead) catch |err| switch (err) { + error.FileNotFound => return null, + else => return err, + }; + defer allocator.free(data); + + const fixedLen = 4 + 4 + 8 + 8 + 4; + if (data.len < fixedLen + 4) return StateError.Truncated; + if (!std.mem.eql(u8, data[0..4], stateMagic)) return StateError.BadMagic; + const version = encode.readU32(data, 4); + if (version != stateFormatVersion) return StateError.UnsupportedVersion; + const offsetValue = encode.readU64(data, 8); + const generationValue = encode.readU64(data, 16); + const pathLen = encode.readU32(data, 24); + const pathStart = fixedLen; + if (data.len != pathStart + pathLen + 4) return StateError.Truncated; + const storedPath = data[pathStart .. pathStart + pathLen]; + const storedCrc = encode.readU32(data, pathStart + pathLen); + if (encode.crc(data[0 .. pathStart + pathLen]) != storedCrc) return StateError.Corrupt; + if (!std.mem.eql(u8, storedPath, primaryDirPath)) return StateError.PrimaryMismatch; + + return .{ .offset = offsetValue, .generation = generationValue }; +} + +test "replicator mirrors puts and deletes from a live primary" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + + const base = try tmp.dir.realpathAlloc(allocator, "."); + defer allocator.free(base); + const primaryDir = try std.fs.path.join(allocator, &.{ base, "primary" }); + defer allocator.free(primaryDir); + const replicaDir = try std.fs.path.join(allocator, &.{ base, "replica" }); + defer allocator.free(replicaDir); + + var primary = try Db.open(allocator, primaryDir, .{}); + defer primary.close(); + var replica = try Db.open(allocator, replicaDir, .{}); + defer replica.close(); + + try primary.put("alpha", "one"); + try primary.put("bravo", "two"); + + var rep = try Replicator.init(allocator, primaryDir, &replica); + defer rep.deinit(); + + const applied = try rep.syncOnce(); + try std.testing.expectEqual(@as(usize, 2), applied); + + const a = try replica.get(allocator, "alpha"); + defer if (a) |v| allocator.free(v); + try std.testing.expectEqualStrings("one", a.?); + const b = try replica.get(allocator, "bravo"); + defer if (b) |v| allocator.free(v); + try std.testing.expectEqualStrings("two", b.?); + + try primary.delete("alpha"); + const applied2 = try rep.syncOnce(); + try std.testing.expectEqual(@as(usize, 1), applied2); + + const a2 = try replica.get(allocator, "alpha"); + try std.testing.expectEqual(@as(?[]u8, null), a2); +} + +test "replicator syncOnce is idempotent between polls" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + const base = try tmp.dir.realpathAlloc(allocator, "."); + defer allocator.free(base); + const primaryDir = try std.fs.path.join(allocator, &.{ base, "primary" }); + defer allocator.free(primaryDir); + const replicaDir = try std.fs.path.join(allocator, &.{ base, "replica" }); + defer allocator.free(replicaDir); + + var primary = try Db.open(allocator, primaryDir, .{}); + defer primary.close(); + var replica = try Db.open(allocator, replicaDir, .{}); + defer replica.close(); + + var rep = try Replicator.init(allocator, primaryDir, &replica); + defer rep.deinit(); + + try primary.put("k1", "v1"); + try std.testing.expectEqual(@as(usize, 1), try rep.syncOnce()); + try std.testing.expectEqual(@as(usize, 0), try rep.syncOnce()); + + try primary.put("k2", "v2"); + try std.testing.expectEqual(@as(usize, 1), try rep.syncOnce()); + try std.testing.expectEqual(@as(usize, 0), try rep.syncOnce()); + + var count: usize = 0; + var iter = try replica.scan(allocator, null, null); + defer iter.deinit(); + while (try iter.next()) |_| count += 1; + try std.testing.expectEqual(@as(usize, 2), count); +} + +test "replicator applies batched writes from Db.apply as one poll" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + const base = try tmp.dir.realpathAlloc(allocator, "."); + defer allocator.free(base); + const primaryDir = try std.fs.path.join(allocator, &.{ base, "primary" }); + defer allocator.free(primaryDir); + const replicaDir = try std.fs.path.join(allocator, &.{ base, "replica" }); + defer allocator.free(replicaDir); + + var primary = try Db.open(allocator, primaryDir, .{}); + defer primary.close(); + var replica = try Db.open(allocator, replicaDir, .{}); + defer replica.close(); + + var writeBatch = WriteBatch.init(allocator); + defer writeBatch.deinit(); + try writeBatch.put("x", "1"); + try writeBatch.put("y", "2"); + try writeBatch.delete("z"); + try primary.apply(&writeBatch); + + var rep = try Replicator.init(allocator, primaryDir, &replica); + defer rep.deinit(); + const applied = try rep.syncOnce(); + try std.testing.expectEqual(@as(usize, 3), applied); +