feat: expand replicate
This commit is contained in:
dualfroz committed
1 parent
69088b7f37
commit
7759ee858e
1 file changed
+170
@@ -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.?);
|
||||
}
|
||||
Reference in new issue
Block a user