From 69088b7f370bf350219fe3ff995d764d2e1c292f Mon Sep 17 00:00:00 2001 From: DualFroz Date: Thu, 10 Sep 2026 17:52:02 +0200 Subject: [PATCH] feat: extend replicate --- src/replicate.zig | 157 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 157 insertions(+) diff --git a/src/replicate.zig b/src/replicate.zig index 9f9e6ad..2ed4870 100644 --- a/src/replicate.zig +++ b/src/replicate.zig @@ -479,3 +479,160 @@ test "replicator applies batched writes from Db.apply as one poll" { const applied = try rep.syncOnce(); try std.testing.expectEqual(@as(usize, 3), applied); + const x = try replica.get(allocator, "x"); + defer if (x) |v| allocator.free(v); + try std.testing.expectEqualStrings("1", x.?); + const z = try replica.get(allocator, "z"); + try std.testing.expectEqual(@as(?[]u8, null), z); +} + +test "replicator poll does not consume a torn tail frame" { + 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("good", "data"); + try std.testing.expectEqual(@as(usize, 1), try rep.syncOnce()); + + // Simulate a writer mid-append: raw bytes appended after the good frame + // that never form a complete, crc-verified frame. + try primary.wal.file.seekFromEnd(0); + try primary.wal.file.writeAll(&[_]u8{ 0xAA, 0xBB, 0xCC }); + try primary.wal.sync(); + + try std.testing.expectEqual(@as(usize, 0), try rep.syncOnce()); + + // The writer "finishes": a real, complete frame lands after the torn + // bytes are naturally never completed (matches wal.zig's own tolerance - + // torn tail bytes are simply never a valid prefix of a later frame, so + // this models a stalled partial write, not a resumed one). + 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, 1), count); +} + +test "replicator surfaces truncation when it falls behind a primary flush" { + 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("k", "v"); + + var rep = try Replicator.init(allocator, primaryDir, &replica); + defer rep.deinit(); + // Advance the tailer's offset without applying, then flush the primary + // (truncates wal.log to 0) so the tailer's remembered offset now points + // past the end of the (now-empty) file. + const staleRecords = try rep.tailer.poll(); + freeAll(allocator, staleRecords); + try primary.flush(); + + // The primary's flush() bumped its WAL generation, so the generation + // check in poll() fires before the (now also true, but weaker) raw + // byte-offset comparison would have. + try std.testing.expectError(ReplicateError.WalGenerationChanged, rep.syncOnce()); + + const recovered = try rep.fullResync(&primary, 500); + try std.testing.expectEqual(@as(usize, 1), recovered); + try std.testing.expectEqual(@as(u64, 0), rep.offset()); + + const v = try replica.get(allocator, "k"); + defer if (v) |val| allocator.free(val); + try std.testing.expectEqualStrings("v", v.?); + + // New writes after the resync are picked up normally through the + // freshly-restarted tail at offset 0. + try primary.put("k2", "v2"); + try std.testing.expectEqual(@as(usize, 1), try rep.syncOnce()); + const v2 = try replica.get(allocator, "k2"); + defer if (v2) |val| allocator.free(val); + try std.testing.expectEqualStrings("v2", v2.?); +} + +test "replicator fullResync bootstraps a brand new replica" { + 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 i: usize = 0; + while (i < 50) : (i += 1) { + var keyBuf: [16]u8 = undefined; + var valBuf: [16]u8 = undefined; + const key = try std.fmt.bufPrint(&keyBuf, "key{d}", .{i}); + const val = try std.fmt.bufPrint(&valBuf, "val{d}", .{i}); + try primary.put(key, val); + } + + var rep = try Replicator.init(allocator, primaryDir, &replica); + defer rep.deinit(); + const applied = try rep.fullResync(&primary, 7); + try std.testing.expectEqual(@as(usize, 50), applied); + + 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, 50), count); + + // Fresh writes after the bootstrap keep tailing correctly from offset 0. + try primary.put("late", "arrival"); + try std.testing.expectEqual(@as(usize, 1), try rep.syncOnce()); + const v = try replica.get(allocator, "late"); + defer if (v) |val| allocator.free(val); + try std.testing.expectEqualStrings("arrival", v.?); +} + +test "walTailer rejects a nonexistent primary directory" { + const allocator = std.testing.allocator; + try std.testing.expectError(error.FileNotFound, WalTailer.open(allocator, "/nonexistent", 0, 0)); +} + +test "loadOffset returns null for a replica directory that never saved state" { + 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 loaded = try loadOffset(allocator, base, "/some/primary"); + try std.testing.expectEqual(@as(?PersistedState, null), loaded); +} +