feat: extend replicate
This commit is contained in:
dualfroz committed
1 parent
b33701a6ca
commit
69088b7f37
1 file changed
+157
@@ -479,3 +479,160 @@ test "replicator applies batched writes from Db.apply as one poll" {
|
|||||||
const applied = try rep.syncOnce();
|
const applied = try rep.syncOnce();
|
||||||
try std.testing.expectEqual(@as(usize, 3), applied);
|
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);
|
||||||
|
}
|
||||||
|
|
||||||
Reference in new issue
Block a user