From 7759ee858ec472f547c9a0d2662402328e6cdb90 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Thu, 10 Sep 2026 18:17:46 +0200 Subject: [PATCH] feat: expand replicate --- src/replicate.zig | 170 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 170 insertions(+) diff --git a/src/replicate.zig b/src/replicate.zig index 2ed4870..4221db5 100644 --- a/src/replicate.zig +++ b/src/replicate.zig @@ -636,3 +636,173 @@ test "loadOffset returns null for a replica directory that never saved state" { try std.testing.expectEqual(@as(?PersistedState, null), loaded); } +test "saveState then loadOffset round-trips the offset, generation, and primary path" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + const base = try tmp.dir.realpathAlloc(allocator, "."); + defer allocator.free(base); + + try saveState(allocator, base, "/data/primary", 4096, 7); + const loaded = try loadOffset(allocator, base, "/data/primary"); + try std.testing.expectEqual(@as(u64, 4096), loaded.?.offset); + try std.testing.expectEqual(@as(u64, 7), loaded.?.generation); +} + +test "saveState overwrites a prior state file with the newest offset and generation" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + const base = try tmp.dir.realpathAlloc(allocator, "."); + defer allocator.free(base); + + try saveState(allocator, base, "/data/primary", 10, 0); + try saveState(allocator, base, "/data/primary", 20, 1); + try saveState(allocator, base, "/data/primary", 30, 2); + const loaded = try loadOffset(allocator, base, "/data/primary"); + try std.testing.expectEqual(@as(u64, 30), loaded.?.offset); + try std.testing.expectEqual(@as(u64, 2), loaded.?.generation); +} + +test "loadOffset rejects a state file saved for a different 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); + + try saveState(allocator, base, "/data/primary-a", 123, 0); + try std.testing.expectError(StateError.PrimaryMismatch, loadOffset(allocator, base, "/data/primary-b")); +} + +test "loadOffset rejects a corrupted state file without crashing" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + const base = try tmp.dir.realpathAlloc(allocator, "."); + defer allocator.free(base); + + try saveState(allocator, base, "/data/primary", 55, 3); + + { + const file = try tmp.dir.openFile(stateFileName, .{ .mode = .read_write }); + defer file.close(); + try file.seekTo(9); + try file.writeAll(&[_]u8{0xFF}); + } + + try std.testing.expectError(StateError.Corrupt, loadOffset(allocator, base, "/data/primary")); +} + +test "loadOffset rejects a truncated state file" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + const base = try tmp.dir.realpathAlloc(allocator, "."); + defer allocator.free(base); + + try saveState(allocator, base, "/data/primary", 55, 3); + + { + const file = try tmp.dir.openFile(stateFileName, .{ .mode = .read_write }); + defer file.close(); + try file.setEndPos(6); + } + + try std.testing.expectError(StateError.Truncated, loadOffset(allocator, base, "/data/primary")); +} + +test "loadOffset rejects a bad magic header" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + const base = try tmp.dir.realpathAlloc(allocator, "."); + defer allocator.free(base); + + try saveState(allocator, base, "/data/primary", 55, 3); + + { + const file = try tmp.dir.openFile(stateFileName, .{ .mode = .read_write }); + defer file.close(); + try file.seekTo(0); + try file.writeAll("XXXX"); + } + + try std.testing.expectError(StateError.BadMagic, loadOffset(allocator, base, "/data/primary")); +} + +test "loadOffset rejects a state file written in the pre-generation v1 format" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + const base = try tmp.dir.realpathAlloc(allocator, "."); + defer allocator.free(base); + + var body = std.ArrayList(u8).init(allocator); + defer body.deinit(); + const writer = encode.ByteWriter.init(&body); + try writer.slice("CRPS"); + try writer.putU32(1); + try writer.putU64(999); + const primaryPath = "/data/primary"; + try writer.putU32(@intCast(primaryPath.len)); + try writer.slice(primaryPath); + try writer.putU32(encode.crc(body.items)); + + const file = try tmp.dir.createFile(stateFileName, .{ .truncate = true }); + defer file.close(); + try file.writeAll(body.items); + + try std.testing.expectError(StateError.UnsupportedVersion, loadOffset(allocator, base, "/data/primary")); +} + +test "saveState and loadOffset round-trip through a real replicate directory end to end" { + 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("a", "1"); + try primary.put("b", "2"); + + var rep = try Replicator.init(allocator, primaryDir, &replica); + defer rep.deinit(); + _ = try rep.syncOnce(); + try saveState(allocator, replicaDir, primaryDir, rep.offset(), rep.generation()); + const savedOffset = rep.offset(); + try std.testing.expect(savedOffset > 0); + + // A brand new process would reopen the replica Db and reconstruct a + // Replicator from the persisted offset instead of offset 0. + var replica2 = try Db.open(allocator, replicaDir, .{}); + defer replica2.close(); + const resumed = (try loadOffset(allocator, replicaDir, primaryDir)).?; + try std.testing.expectEqual(savedOffset, resumed.offset); + try std.testing.expectEqual(rep.generation(), resumed.generation); + + var rep2 = try Replicator.initFromOffset(allocator, primaryDir, &replica2, resumed.offset, resumed.generation); + defer rep2.deinit(); + try primary.put("c", "3"); + const applied = try rep2.syncOnce(); + try std.testing.expectEqual(@as(usize, 1), applied); + + const c = try replica2.get(allocator, "c"); + defer if (c) |v| allocator.free(v); + try std.testing.expectEqualStrings("3", c.?); + // The earlier keys were never re-applied through rep2 (it resumed past + // them), yet they are still visible because replica2 is the same + // on-disk replica Db that rep already wrote them into. + const a = try replica2.get(allocator, "a"); + defer if (a) |v| allocator.free(v); + try std.testing.expectEqualStrings("1", a.?); +}