feat: add loadOffset to replicate

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);
+