From dcd1d4c7a8bcc7bd2fed59ff350a9461865dd7ec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Poyraz=20K=C3=BC=C3=A7=C3=BCkarslan?= <83272398+PoyrazK@users.noreply.github.com> Date: Wed, 10 Jun 2026 18:56:34 +0300 Subject: [PATCH 1/3] test(storage-engine): add WAL compaction recovery integration tests Add wal_compaction_recovery_tests.zig with 5 tests covering: - Committed writes surviving full compaction and restart - Uncommitted writes replaying from WAL on restart - Delete/tombestone survival through compaction cycles - Multi-key data surviving compaction and recovery - Overwritten keys recovering correct latest value Add compaction_transaction_tests.zig with 4 tests covering: - Concurrent reads during compaction - Concurrent writes during compaction - Data correctness after multiple compaction cycles - Tombestone isolation during compaction Add cdc_transaction_tests.zig with 4 tests covering: - Multiple puts emitting multiple CDC events - Delete operations emitting DELETE event type - Update operations producing PUT events - CDC events having incrementing sequence numbers --- .../tests/cdc_transaction_tests.zig | 538 ++++++++++++++++++ .../tests/compaction_transaction_tests.zig | 347 +++++++++++ .../tests/wal_compaction_recovery_tests.zig | 325 +++++++++++ 3 files changed, 1210 insertions(+) create mode 100644 storage-engine/tests/cdc_transaction_tests.zig create mode 100644 storage-engine/tests/compaction_transaction_tests.zig create mode 100644 storage-engine/tests/wal_compaction_recovery_tests.zig diff --git a/storage-engine/tests/cdc_transaction_tests.zig b/storage-engine/tests/cdc_transaction_tests.zig new file mode 100644 index 0000000..1861449 --- /dev/null +++ b/storage-engine/tests/cdc_transaction_tests.zig @@ -0,0 +1,538 @@ +const std = @import("std"); +const rpc_message = @import("../src/network/rpc_message.zig"); +const rpc_mod = @import("../src/network/rpc.zig"); +const EngineZig = @import("../src/storage/engine.zig").EngineZig; + +const RpcType = rpc_mod.RpcType; +const RpcServer = rpc_mod.RpcServer; +const RpcClient = rpc_mod.RpcClient; + +const socket_c = @cImport({ + @cInclude("sys/socket.h"); + @cInclude("netinet/in.h"); + @cInclude("arpa/inet.h"); + @cInclude("unistd.h"); +}); + +const CdcEvent = struct { + seq: u64 = 0, + type: []const u8, + table: []const u8, + key: []const u8, + value: []const u8, + + pub fn deserialize(allocator: std.mem.Allocator, bytes: []const u8) !CdcEvent { + if (bytes.len < 8) return error.BufferTooShort; + var offset: usize = 0; + const seq = std.mem.readInt(u64, bytes[0..8][0..8], .little); + offset += 8; + + const type_str = try readStr(allocator, bytes, &offset); + errdefer allocator.free(type_str); + + const table = try readStr(allocator, bytes, &offset); + errdefer allocator.free(table); + + const key = try readStr(allocator, bytes, &offset); + errdefer allocator.free(key); + + const value = try readStr(allocator, bytes, &offset); + + return CdcEvent{ + .seq = seq, + .type = type_str, + .table = table, + .key = key, + .value = value, + }; + } + + fn readStr(allocator: std.mem.Allocator, bytes: []const u8, offset: *usize) ![]const u8 { + if (offset.* + 4 > bytes.len) return error.BufferTooShort; + const len = std.mem.readInt(u32, bytes[offset.* .. offset.* + 4][0..4], .little); + offset.* += 4; + if (offset.* + len > bytes.len) return error.BufferTooShort; + const str = bytes[offset.* .. offset.* + len]; + offset.* += len; + return allocator.dupe(u8, str); + } +}; + +const ServerContext = struct { + allocator: std.mem.Allocator, + engine: *EngineZig, + + fn handler( + ctx: ?*anyopaque, + rpc_type: u8, + flags: u8, + payload_ptr: [*]const u8, + payload_len: usize, + sender_ctx: ?*anyopaque, + sender_cb: *const fn (sender_ctx: ?*anyopaque, reply_ptr: [*]const u8, reply_len: usize) callconv(.c) void, + ) callconv(.c) void { + _ = flags; + const self = @as(*ServerContext, @ptrCast(@alignCast(ctx.?))); + const rtype = @as(RpcType, @enumFromInt(rpc_type)); + + if (rtype == RpcType.PutItem) { + const args = rpc_message.item_args_deserialize(payload_ptr, payload_len).?; + defer rpc_message.item_args_destroy(args); + + var pk_ptr: [*]const u8 = undefined; + var pk_len: usize = 0; + rpc_message.item_args_get_partition_key(args, &pk_ptr, &pk_len); + + var val_ptr: [*]const u8 = undefined; + var val_len: usize = 0; + rpc_message.item_args_get_value(args, &val_ptr, &val_len); + + var lsis = std.ArrayList([]const u8).empty; + defer lsis.deinit(self.allocator); + const lsi_count = rpc_message.item_args_get_lsi_count(args); + var idx: usize = 0; + while (idx < lsi_count) : (idx += 1) { + var lsi_ptr: [*]const u8 = undefined; + var lsi_len: usize = 0; + rpc_message.item_args_get_lsi_key(args, idx, &lsi_ptr, &lsi_len); + lsis.append(self.allocator, lsi_ptr[0..lsi_len]) catch {}; + } + + const success = self.engine.put( + pk_ptr[0..pk_len], + val_ptr[0..val_len], + lsis.items, + "default", + "", + ); + + var reply_buf: [256]u8 = undefined; + var reply_len: usize = 0; + const dummy_ptrs = [_][*]const u8{}; + const dummy_lens = [_]usize{}; + + const ok = rpc_message.rpc_reply_serialize( + success, + "".ptr, 0, + "".ptr, 0, + 0, + "".ptr, 0, + &dummy_ptrs, &dummy_lens, 0, + &dummy_ptrs, &dummy_lens, + &dummy_ptrs, &dummy_lens, + &dummy_ptrs, &dummy_lens, + 0, + &reply_buf, &reply_len + ); + if (ok) { + sender_cb(sender_ctx, &reply_buf, reply_len); + } + } + } +}; + +const SubscriberContext = struct { + allocator: std.mem.Allocator, + received_events: *std.ArrayList(CdcEvent), + running: *std.atomic.Value(bool), + sock_fd: *std.atomic.Value(c_int), + expected_port: u16, +}; + +fn subscriberWorker(ctx: SubscriberContext) void { + const sock = socket_c.socket(socket_c.AF_INET, socket_c.SOCK_STREAM, 0); + if (sock < 0) return; + defer _ = socket_c.close(sock); + + ctx.sock_fd.store(sock, .release); + + var serv_addr: socket_c.struct_sockaddr_in = undefined; + @memset(std.mem.asBytes(&serv_addr), 0); + serv_addr.sin_family = socket_c.AF_INET; + serv_addr.sin_port = socket_c.htons(ctx.expected_port); + _ = socket_c.inet_pton(socket_c.AF_INET, "127.0.0.1", &serv_addr.sin_addr); + + if (socket_c.connect(sock, @ptrCast(&serv_addr), @sizeOf(socket_c.struct_sockaddr_in)) < 0) { + return; + } + + // Send SubscribeCdc header + var header_buf: [12]u8 = undefined; + rpc_message.rpc_header_encode( + 0x44424E53, + 9, // SubscribeCdc + 1, // flags + 0, + 0, + 0, + &header_buf, + ); + _ = socket_c.send(sock, &header_buf, 12, 0); + + // Receive loop + var in_header_buf: [12]u8 = undefined; + while (ctx.running.load(.monotonic)) { + const n = socket_c.recv(sock, &in_header_buf, 12, socket_c.MSG_WAITALL); + if (n != 12) break; + + var magic: u32 = 0; + var rpc_type: u8 = 0; + var flags: u8 = 0; + var res1: u16 = 0; + var res2: u16 = 0; + var payload_len: u16 = 0; + + rpc_message.rpc_header_decode(&in_header_buf, &magic, &rpc_type, &flags, &res1, &res2, &payload_len); + + if (magic == 0x44424E53) { + const payload = ctx.allocator.alloc(u8, payload_len) catch return; + defer ctx.allocator.free(payload); + + const n_pay = socket_c.recv(sock, payload.ptr, payload_len, socket_c.MSG_WAITALL); + if (n_pay != payload_len) break; + + const ev = CdcEvent.deserialize(ctx.allocator, payload) catch continue; + ctx.received_events.append(ctx.allocator, ev) catch { + ctx.allocator.free(ev.type); + ctx.allocator.free(ev.table); + ctx.allocator.free(ev.key); + ctx.allocator.free(ev.value); + }; + } + } +} + +test "CdcTransaction - MultiplePutsEmitMultipleEvents" { + // Test that multiple put operations each emit CDC events + const allocator = std.testing.allocator; + const db_path = "cdc_tx_multi_put_db"; + const port: u16 = 9010; + + var threaded = std.Io.Threaded.init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + const cwd = std.Io.Dir.cwd(); + + cwd.deleteTree(io, db_path) catch {}; + defer cwd.deleteTree(io, db_path) catch {}; + + var engine = try EngineZig.init(allocator, db_path, 4 * 1024 * 1024, false); + defer engine.deinit(); + + var ctx = ServerContext{ + .allocator = allocator, + .engine = engine, + }; + + var server = try RpcServer.init(allocator, port, ServerContext.handler, &ctx); + defer server.deinit(); + + const started = try server.start(); + try std.testing.expect(started); + + try io.sleep(std.Io.Duration.fromMilliseconds(50), .awake); + + var received_events = std.ArrayList(CdcEvent).empty; + defer { + for (received_events.items) |ev| { + allocator.free(ev.type); + allocator.free(ev.table); + allocator.free(ev.key); + allocator.free(ev.value); + } + received_events.deinit(allocator); + } + + var subscriber_running = std.atomic.Value(bool).init(true); + var sock_fd = std.atomic.Value(c_int).init(-1); + + var sub_thread = try std.Thread.spawn(.{}, subscriberWorker, .{SubscriberContext{ + .allocator = allocator, + .received_events = &received_events, + .running = &subscriber_running, + .sock_fd = &sock_fd, + .expected_port = port, + }}); + + try io.sleep(std.Io.Duration.fromMilliseconds(50), .awake); + + // Perform 5 database operations + const lsis = [_][]const u8{}; + try std.testing.expect(engine.put("key_1", "value_1", &lsis, "default", "")); + try std.testing.expect(engine.put("key_2", "value_2", &lsis, "default", "")); + try std.testing.expect(engine.put("key_3", "value_3", &lsis, "default", "")); + try std.testing.expect(engine.put("key_4", "value_4", &lsis, "default", "")); + try std.testing.expect(engine.put("key_5", "value_5", &lsis, "default", "")); + + // Allow events to propagate + try io.sleep(std.Io.Duration.fromMilliseconds(200), .awake); + + // Terminate subscriber thread and stop server + subscriber_running.store(false, .monotonic); + const fd = sock_fd.load(.acquire); + if (fd != -1) { + _ = socket_c.close(fd); + } + sub_thread.join(); + server.stop(); + + // Verify we received 5 CDC events + try std.testing.expectEqual(@as(usize, 5), received_events.items.len); + + // Verify each event has correct data + try std.testing.expectEqualStrings("PUT", received_events.items[0].type); + try std.testing.expectEqualStrings("key_1", received_events.items[0].key); + try std.testing.expectEqualStrings("value_1", received_events.items[0].value); + + try std.testing.expectEqualStrings("PUT", received_events.items[4].type); + try std.testing.expectEqualStrings("key_5", received_events.items[4].key); + try std.testing.expectEqualStrings("value_5", received_events.items[4].value); +} + +test "CdcTransaction - DeleteEventTypeIsDelete" { + // Test that delete operations emit DELETE event type + const allocator = std.testing.allocator; + const db_path = "cdc_tx_delete_db"; + const port: u16 = 9011; + + var threaded = std.Io.Threaded.init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + const cwd = std.Io.Dir.cwd(); + + cwd.deleteTree(io, db_path) catch {}; + defer cwd.deleteTree(io, db_path) catch {}; + + var engine = try EngineZig.init(allocator, db_path, 4 * 1024 * 1024, false); + defer engine.deinit(); + + var ctx = ServerContext{ + .allocator = allocator, + .engine = engine, + }; + + var server = try RpcServer.init(allocator, port, ServerContext.handler, &ctx); + defer server.deinit(); + + const started = try server.start(); + try std.testing.expect(started); + + try io.sleep(std.Io.Duration.fromMilliseconds(50), .awake); + + var received_events = std.ArrayList(CdcEvent).empty; + defer { + for (received_events.items) |ev| { + allocator.free(ev.type); + allocator.free(ev.table); + allocator.free(ev.key); + allocator.free(ev.value); + } + received_events.deinit(allocator); + } + + var subscriber_running = std.atomic.Value(bool).init(true); + var sock_fd = std.atomic.Value(c_int).init(-1); + + var sub_thread = try std.Thread.spawn(.{}, subscriberWorker, .{SubscriberContext{ + .allocator = allocator, + .received_events = &received_events, + .running = &subscriber_running, + .sock_fd = &sock_fd, + .expected_port = port, + }}); + + try io.sleep(std.Io.Duration.fromMilliseconds(50), .awake); + + // First put a key, then delete it + const lsis = [_][]const u8{}; + try std.testing.expect(engine.put("key_to_delete", "some_value", &lsis, "default", "")); + try io.sleep(std.Io.Duration.fromMilliseconds(50), .awake); + try std.testing.expect(engine.delete("key_to_delete", &lsis, "default", "")); + + // Allow events to propagate + try io.sleep(std.Io.Duration.fromMilliseconds(200), .awake); + + // Terminate subscriber thread and stop server + subscriber_running.store(false, .monotonic); + const fd = sock_fd.load(.acquire); + if (fd != -1) { + _ = socket_c.close(fd); + } + sub_thread.join(); + server.stop(); + + // Verify we received 2 CDC events: PUT then DELETE + try std.testing.expectEqual(@as(usize, 2), received_events.items.len); + + try std.testing.expectEqualStrings("PUT", received_events.items[0].type); + try std.testing.expectEqualStrings("key_to_delete", received_events.items[0].key); + + try std.testing.expectEqualStrings("DELETE", received_events.items[1].type); + try std.testing.expectEqualStrings("key_to_delete", received_events.items[1].key); +} + +test "CdcTransaction - UpdateProducesPutEvent" { + // Test that updating an existing key produces a PUT event + const allocator = std.testing.allocator; + const db_path = "cdc_tx_update_db"; + const port: u16 = 9012; + + var threaded = std.Io.Threaded.init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + const cwd = std.Io.Dir.cwd(); + + cwd.deleteTree(io, db_path) catch {}; + defer cwd.deleteTree(io, db_path) catch {}; + + var engine = try EngineZig.init(allocator, db_path, 4 * 1024 * 1024, false); + defer engine.deinit(); + + var ctx = ServerContext{ + .allocator = allocator, + .engine = engine, + }; + + var server = try RpcServer.init(allocator, port, ServerContext.handler, &ctx); + defer server.deinit(); + + const started = try server.start(); + try std.testing.expect(started); + + try io.sleep(std.Io.Duration.fromMilliseconds(50), .awake); + + var received_events = std.ArrayList(CdcEvent).empty; + defer { + for (received_events.items) |ev| { + allocator.free(ev.type); + allocator.free(ev.table); + allocator.free(ev.key); + allocator.free(ev.value); + } + received_events.deinit(allocator); + } + + var subscriber_running = std.atomic.Value(bool).init(true); + var sock_fd = std.atomic.Value(c_int).init(-1); + + var sub_thread = try std.Thread.spawn(.{}, subscriberWorker, .{SubscriberContext{ + .allocator = allocator, + .received_events = &received_events, + .running = &subscriber_running, + .sock_fd = &sock_fd, + .expected_port = port, + }}); + + try io.sleep(std.Io.Duration.fromMilliseconds(50), .awake); + + // Put initial value + const lsis = [_][]const u8{}; + try std.testing.expect(engine.put("key", "value_v1", &lsis, "default", "")); + try io.sleep(std.Io.Duration.fromMilliseconds(50), .awake); + + // Update the same key + try std.testing.expect(engine.put("key", "value_v2", &lsis, "default", "")); + + // Allow events to propagate + try io.sleep(std.Io.Duration.fromMilliseconds(200), .awake); + + // Terminate subscriber thread and stop server + subscriber_running.store(false, .monotonic); + const fd = sock_fd.load(.acquire); + if (fd != -1) { + _ = socket_c.close(fd); + } + sub_thread.join(); + server.stop(); + + // Verify we received 2 CDC events + try std.testing.expectEqual(@as(usize, 2), received_events.items.len); + + // First event: initial put + try std.testing.expectEqualStrings("PUT", received_events.items[0].type); + try std.testing.expectEqualStrings("key", received_events.items[0].key); + try std.testing.expectEqualStrings("value_v1", received_events.items[0].value); + + // Second event: update (also a PUT) + try std.testing.expectEqualStrings("PUT", received_events.items[1].type); + try std.testing.expectEqualStrings("key", received_events.items[1].key); + try std.testing.expectEqualStrings("value_v2", received_events.items[1].value); +} + +test "CdcTransaction - EventsHaveIncrementingSequence" { + // Test that CDC events have incrementing sequence numbers + const allocator = std.testing.allocator; + const db_path = "cdc_tx_seq_db"; + const port: u16 = 9013; + + var threaded = std.Io.Threaded.init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + const cwd = std.Io.Dir.cwd(); + + cwd.deleteTree(io, db_path) catch {}; + defer cwd.deleteTree(io, db_path) catch {}; + + var engine = try EngineZig.init(allocator, db_path, 4 * 1024 * 1024, false); + defer engine.deinit(); + + var ctx = ServerContext{ + .allocator = allocator, + .engine = engine, + }; + + var server = try RpcServer.init(allocator, port, ServerContext.handler, &ctx); + defer server.deinit(); + + const started = try server.start(); + try std.testing.expect(started); + + try io.sleep(std.Io.Duration.fromMilliseconds(50), .awake); + + var received_events = std.ArrayList(CdcEvent).empty; + defer { + for (received_events.items) |ev| { + allocator.free(ev.type); + allocator.free(ev.table); + allocator.free(ev.key); + allocator.free(ev.value); + } + received_events.deinit(allocator); + } + + var subscriber_running = std.atomic.Value(bool).init(true); + var sock_fd = std.atomic.Value(c_int).init(-1); + + var sub_thread = try std.Thread.spawn(.{}, subscriberWorker, .{SubscriberContext{ + .allocator = allocator, + .received_events = &received_events, + .running = &subscriber_running, + .sock_fd = &sock_fd, + .expected_port = port, + }}); + + try io.sleep(std.Io.Duration.fromMilliseconds(50), .awake); + + // Perform 3 operations + const lsis = [_][]const u8{}; + try std.testing.expect(engine.put("key_1", "value_1", &lsis, "default", "")); + try std.testing.expect(engine.put("key_2", "value_2", &lsis, "default", "")); + try std.testing.expect(engine.put("key_3", "value_3", &lsis, "default", "")); + + // Allow events to propagate + try io.sleep(std.Io.Duration.fromMilliseconds(200), .awake); + + // Terminate subscriber thread and stop server + subscriber_running.store(false, .monotonic); + const fd = sock_fd.load(.acquire); + if (fd != -1) { + _ = socket_c.close(fd); + } + sub_thread.join(); + server.stop(); + + // Verify sequence numbers are strictly increasing + try std.testing.expectEqual(@as(usize, 3), received_events.items.len); + try std.testing.expect(received_events.items[0].seq < received_events.items[1].seq); + try std.testing.expect(received_events.items[1].seq < received_events.items[2].seq); +} \ No newline at end of file diff --git a/storage-engine/tests/compaction_transaction_tests.zig b/storage-engine/tests/compaction_transaction_tests.zig new file mode 100644 index 0000000..adbb06d --- /dev/null +++ b/storage-engine/tests/compaction_transaction_tests.zig @@ -0,0 +1,347 @@ +const std = @import("std"); +const EngineZig = @import("../src/storage/engine.zig").EngineZig; +const tm = @import("../src/distributed/transaction_manager.zig"); +const TransactionManager = tm.TransactionManager; +const EngineCallbackTable = tm.EngineCallbackTable; +const TransactionItemC = tm.TransactionItemC; +const c = @cImport({ + @cInclude("unistd.h"); +}); + +fn engineGetHelper(engine: *EngineZig, key: []const u8, out_val: *std.ArrayList(u8), allocator: std.mem.Allocator) bool { + out_val.clearRetainingCapacity(); + const helper = struct { + const Ctx = struct { + list: *std.ArrayList(u8), + allocator: std.mem.Allocator, + }; + fn cb(ctx_ptr: ?*anyopaque, data_ptr: [*]const u8, data_len: usize) callconv(.c) void { + const ctx = @as(*Ctx, @ptrCast(@alignCast(ctx_ptr.?))); + ctx.list.appendSlice(ctx.allocator, data_ptr[0..data_len]) catch {}; + } + }; + var ctx = helper.Ctx{ .list = out_val, .allocator = allocator }; + return engine.get(key, &ctx, helper.cb, "default", ""); +} + +fn putCallback(ctx: ?*anyopaque, pk: [*]const u8, pk_len: usize, val: [*]const u8, val_len: usize, table: [*]const u8, table_len: usize, sk: [*]const u8, sk_len: usize) callconv(.c) bool { + const engine = @as(*EngineZig, @ptrCast(@alignCast(ctx.?))); + return engine.put(pk[0..pk_len], val[0..val_len], &[_][]const u8{}, table[0..table_len], sk[0..sk_len]); +} + +fn getCallback(ctx: ?*anyopaque, pk: [*]const u8, pk_len: usize, table: [*]const u8, table_len: usize, sk: [*]const u8, sk_len: usize, val_cb_obj: ?*anyopaque, val_cb: ?*const fn (obj: ?*anyopaque, [*]const u8, usize) callconv(.c) void) callconv(.c) bool { + const engine = @as(*EngineZig, @ptrCast(@alignCast(ctx.?))); + return engine.get(pk[0..pk_len], val_cb_obj, val_cb, table[0..table_len], sk[0..sk_len]); +} + +fn deleteCallback(ctx: ?*anyopaque, pk: [*]const u8, pk_len: usize, table: [*]const u8, table_len: usize, sk: [*]const u8, sk_len: usize) callconv(.c) void { + const engine = @as(*EngineZig, @ptrCast(@alignCast(ctx.?))); + _ = engine.delete(pk[0..pk_len], &[_][]const u8{}, table[0..table_len], sk[0..sk_len]); +} + +fn queryCallback( + ctx: ?*anyopaque, + table: [*]const u8, table_len: usize, + pk: [*]const u8, pk_len: usize, + op: [*]const u8, op_len: usize, + sk_prefix: [*]const u8, sk_prefix_len: usize, + cb_obj: ?*anyopaque, + cb: ?*const fn (obj: ?*anyopaque, [*]const u8, usize, [*]const u8, usize) callconv(.c) void, +) callconv(.c) void { + const engine = @as(*EngineZig, @ptrCast(@alignCast(ctx.?))); + const Helper = struct { + const WrapperCtx = struct { + real_cb_obj: ?*anyopaque, + real_cb: ?*const fn (obj: ?*anyopaque, [*]const u8, usize, [*]const u8, usize) callconv(.c) void, + }; + fn innerCb(obj: ?*anyopaque, pk_ptr: [*]const u8, inner_pk_len: usize, sk_ptr: [*]const u8, sk_len: usize, val_ptr: [*]const u8, val_len: usize) callconv(.c) void { + _ = pk_ptr; _ = inner_pk_len; + const wrapper = @as(*WrapperCtx, @ptrCast(@alignCast(obj.?))); + if (wrapper.real_cb) |rcb| { + rcb(wrapper.real_cb_obj, sk_ptr, sk_len, val_ptr, val_len); + } + } + }; + var wrapper = Helper.WrapperCtx{ .real_cb_obj = cb_obj, .real_cb = cb }; + engine.query( + table[0..table_len], + pk[0..pk_len], + op[0..op_len], + sk_prefix[0..sk_prefix_len], + &wrapper, + Helper.innerCb, + ) catch {}; +} + +fn getEngineCallbacks() EngineCallbackTable { + return EngineCallbackTable{ + .put = struct { + fn f(ctx: ?*anyopaque, pk: [*]const u8, pk_len: usize, val: [*]const u8, val_len: usize, table: [*]const u8, table_len: usize, sk: [*]const u8, sk_len: usize) callconv(.c) bool { + const eng = @as(*EngineZig, @ptrCast(@alignCast(ctx.?))); + return eng.put(pk[0..pk_len], val[0..val_len], &[_][]const u8{}, table[0..table_len], sk[0..sk_len]); + } + }.f, + .get = struct { + fn f(ctx: ?*anyopaque, pk: [*]const u8, pk_len: usize, table: [*]const u8, table_len: usize, sk: [*]const u8, sk_len: usize, val_cb_obj: ?*anyopaque, val_cb: ?*const fn (obj: ?*anyopaque, [*]const u8, usize) callconv(.c) void) callconv(.c) bool { + const eng = @as(*EngineZig, @ptrCast(@alignCast(ctx.?))); + return eng.get(pk[0..pk_len], val_cb_obj, val_cb, table[0..table_len], sk[0..sk_len]); + } + }.f, + .delete = struct { + fn f(ctx: ?*anyopaque, pk: [*]const u8, pk_len: usize, table: [*]const u8, table_len: usize, sk: [*]const u8, sk_len: usize) callconv(.c) void { + const eng = @as(*EngineZig, @ptrCast(@alignCast(ctx.?))); + _ = eng.delete(pk[0..pk_len], &[_][]const u8{}, table[0..table_len], sk[0..sk_len]); + } + }.f, + .query = struct { + fn f(ctx: ?*anyopaque, table: [*]const u8, table_len: usize, pk: [*]const u8, pk_len: usize, op: [*]const u8, op_len: usize, sk_prefix: [*]const u8, sk_prefix_len: usize, cb_obj: ?*anyopaque, cb: ?*const fn (obj: ?*anyopaque, [*]const u8, usize, [*]const u8, usize) callconv(.c) void) callconv(.c) void { + const eng = @as(*EngineZig, @ptrCast(@alignCast(ctx.?))); + eng.query(table[0..table_len], pk[0..pk_len], op[0..op_len], sk_prefix[0..sk_prefix_len], cb_obj, cb) catch {}; + } + }.f, + }; +} + +fn makeItem(table: []const u8, pk: []const u8, sk: []const u8, value: []const u8) TransactionItemC { + return TransactionItemC{ + .table = table.ptr, + .table_len = table.len, + .pk = pk.ptr, + .pk_len = pk.len, + .sk = sk.ptr, + .sk_len = sk.len, + .value = value.ptr, + .value_len = value.len, + }; +} + +fn readerThreadFunc(engine: *EngineZig, stop: *bool, allocator: std.mem.Allocator) void { + var val = std.ArrayList(u8).empty; + defer val.deinit(allocator); + while (!@atomicLoad(bool, stop, .monotonic)) { + _ = engineGetHelper(engine, "key1", &val, allocator); + _ = c.usleep(100); + } +} + +test "CompactionTransaction - ConcurrentReadsAndCompaction" { + // Test that concurrent reads during compaction don't cause issues + const allocator = std.testing.allocator; + const db_path = "db_compaction_tx_concurrent_reads"; + + var threaded = std.Io.Threaded.init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + const cwd = std.Io.Dir.cwd(); + + cwd.deleteTree(io, db_path) catch {}; + defer cwd.deleteTree(io, db_path) catch {}; + + var lsi_empty = [_][]const u8{}; + + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + var large_val: [2000]u8 = undefined; + @memset(&large_val, 'a'); + + // Write key1 and flush to L0 + try std.testing.expect(engine.put("key1", &large_val, &lsi_empty, "default", "")); + try io.sleep(std.Io.Duration.fromMilliseconds(60), .awake); + + // Start reader thread + var stop: bool = false; + const reader = try std.Thread.spawn(.{}, readerThreadFunc, .{ engine, &stop, allocator }); + + // Compact L0 -> L1 multiple times while reader is active + var i: usize = 0; + while (i < 5) : (i += 1) { + try engine.compactLevel(0); + try io.sleep(std.Io.Duration.fromMilliseconds(10), .awake); + } + + @atomicStore(bool, &stop, true, .monotonic); + reader.join(); + } + + // Verify data is still correct after all compactions + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + var large_val: [2000]u8 = undefined; + @memset(&large_val, 'a'); + + var val = std.ArrayList(u8).empty; + defer val.deinit(allocator); + + try std.testing.expect(engineGetHelper(engine, "key1", &val, allocator)); + try std.testing.expectEqualStrings(&large_val, val.items); + } +} + +test "CompactionTransaction - ConcurrentWritesAndCompaction" { + // Test that concurrent writes while compaction runs are handled correctly + const allocator = std.testing.allocator; + const db_path = "db_compaction_tx_concurrent_writes"; + + var threaded = std.Io.Threaded.init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + const cwd = std.Io.Dir.cwd(); + + cwd.deleteTree(io, db_path) catch {}; + defer cwd.deleteTree(io, db_path) catch {}; + + const lsi_empty = [_][]const u8{}; + var write_count: std.atomic.Value(usize) = std.atomic.Value(usize).init(0); + + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + const count_ptr = &write_count; + const writer = try std.Thread.spawn(.{}, struct { + fn f(ctx: *EngineZig, count: *std.atomic.Value(usize)) void { + var i: usize = 0; + while (i < 50) : (i += 1) { + var key_buf: [32]u8 = undefined; + const key = std.fmt.bufPrint(&key_buf, "key_{d}", .{i}) catch continue; + var val_buf: [32]u8 = undefined; + const val = std.fmt.bufPrint(&val_buf, "value_{d}", .{i}) catch continue; + _ = ctx.put(key, val, &lsi_empty, "default", ""); + count.store(i + 1, .monotonic); + } + } + }.f, .{ engine, count_ptr }); + + // Give writer a head start + try io.sleep(std.Io.Duration.fromMilliseconds(20), .awake); + + // Compact while writes are happening + var i: usize = 0; + while (i < 3) : (i += 1) { + try engine.compactLevel(0); + try io.sleep(std.Io.Duration.fromMilliseconds(20), .awake); + } + + writer.join(); + } + + // Verify all written keys are readable + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + var val = std.ArrayList(u8).empty; + defer val.deinit(allocator); + + const count = write_count.load(.monotonic); + var i: usize = 0; + while (i < count) : (i += 1) { + var key_buf: [32]u8 = undefined; + const key = std.fmt.bufPrint(&key_buf, "key_{d}", .{i}) catch continue; + var expected_buf: [32]u8 = undefined; + const expected = std.fmt.bufPrint(&expected_buf, "value_{d}", .{i}) catch continue; + + try std.testing.expect(engineGetHelper(engine, key, &val, allocator)); + try std.testing.expectEqualStrings(expected, val.items); + } + } +} + +test "CompactionTransaction - DataCorrectnessAfterCompaction" { + // Test that data remains correct after multiple compaction cycles + const allocator = std.testing.allocator; + const db_path = "db_compaction_tx_data_correctness"; + + var threaded = std.Io.Threaded.init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + const cwd = std.Io.Dir.cwd(); + + cwd.deleteTree(io, db_path) catch {}; + defer cwd.deleteTree(io, db_path) catch {}; + + var lsi_empty = [_][]const u8{}; + + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + var large_val: [2000]u8 = undefined; + @memset(&large_val, 'a'); + + // Write multiple keys and compact them down + var i: usize = 0; + while (i < 10) : (i += 1) { + var key_buf: [32]u8 = undefined; + const key = std.fmt.bufPrint(&key_buf, "key_{d}", .{i}) catch @panic("bufPrint failed"); + try std.testing.expect(engine.put(key, &large_val, &lsi_empty, "default", "")); + try io.sleep(std.Io.Duration.fromMilliseconds(60), .awake); + try engine.compactLevel(0); + } + + // Compact all remaining levels + var level: u32 = 1; + while (level < 7) : (level += 1) { + try engine.compactLevel(level); + try io.sleep(std.Io.Duration.fromMilliseconds(20), .awake); + } + + // Verify all keys have correct values + var val = std.ArrayList(u8).empty; + defer val.deinit(allocator); + + i = 0; + while (i < 10) : (i += 1) { + var key_buf: [32]u8 = undefined; + const key = std.fmt.bufPrint(&key_buf, "key_{d}", .{i}) catch @panic("bufPrint failed"); + + try std.testing.expect(engineGetHelper(engine, key, &val, allocator)); + try std.testing.expectEqualStrings(&large_val, val.items); + } +} + +test "CompactionTransaction - TombstoneIsolationDuringCompaction" { + // Test that tombstones properly isolate data during compaction + const allocator = std.testing.allocator; + const db_path = "db_compaction_tx_tombstone"; + + var threaded = std.Io.Threaded.init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + const cwd = std.Io.Dir.cwd(); + + cwd.deleteTree(io, db_path) catch {}; + defer cwd.deleteTree(io, db_path) catch {}; + + var lsi_empty = [_][]const u8{}; + + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + // Write key + try std.testing.expect(engine.put("key1", "original_value", &lsi_empty, "default", "")); + try io.sleep(std.Io.Duration.fromMilliseconds(60), .awake); + try engine.compactLevel(0); + + // Delete key + try std.testing.expect(engine.delete("key1", &lsi_empty, "default", "")); + try io.sleep(std.Io.Duration.fromMilliseconds(60), .awake); + try engine.compactLevel(0); + } + + // Verify key is properly deleted after compaction + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + var val = std.ArrayList(u8).empty; + defer val.deinit(allocator); + + // Key should not be readable + try std.testing.expect(!engineGetHelper(engine, "key1", &val, allocator)); + } +} \ No newline at end of file diff --git a/storage-engine/tests/wal_compaction_recovery_tests.zig b/storage-engine/tests/wal_compaction_recovery_tests.zig new file mode 100644 index 0000000..11a2110 --- /dev/null +++ b/storage-engine/tests/wal_compaction_recovery_tests.zig @@ -0,0 +1,325 @@ +const std = @import("std"); +const EngineZig = @import("../src/storage/engine.zig").EngineZig; + +fn engineGetHelper(engine: *EngineZig, key: []const u8, out_val: *std.ArrayList(u8), allocator: std.mem.Allocator) bool { + out_val.clearRetainingCapacity(); + const helper = struct { + const Ctx = struct { + list: *std.ArrayList(u8), + allocator: std.mem.Allocator, + }; + fn cb(ctx_ptr: ?*anyopaque, data_ptr: [*]const u8, data_len: usize) callconv(.c) void { + const ctx = @as(*Ctx, @ptrCast(@alignCast(ctx_ptr.?))); + ctx.list.appendSlice(ctx.allocator, data_ptr[0..data_len]) catch {}; + } + }; + var ctx = helper.Ctx{ .list = out_val, .allocator = allocator }; + return engine.get(key, &ctx, helper.cb, "default", ""); +} + +test "WalCompactionRecovery - CommittedWritesSurviveCompaction" { + const allocator = std.testing.allocator; + const db_path = "db_wal_compaction_test_1"; + + var threaded = std.Io.Threaded.init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + const cwd = std.Io.Dir.cwd(); + + cwd.deleteTree(io, db_path) catch {}; + defer cwd.deleteTree(io, db_path) catch {}; + + var lsi_empty = [_][]const u8{}; + + // Phase 1: Write 20 keys and commit them + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + var i: usize = 0; + while (i < 20) : (i += 1) { + var key_buf: [32]u8 = undefined; + const key = try std.fmt.bufPrint(&key_buf, "key_{d}", .{i}); + var val_buf: [32]u8 = undefined; + const value = try std.fmt.bufPrint(&val_buf, "value_{d}", .{i}); + try std.testing.expect(engine.put(key, value, &lsi_empty, "default", "")); + } + // Wait for memtable to flush to SSTable + try io.sleep(std.Io.Duration.fromMilliseconds(200), .awake); + } + + // Phase 2: Run full compaction + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + // Compact all levels + var level: u32 = 0; + while (level < 7) : (level += 1) { + try engine.compactLevel(level); + try io.sleep(std.Io.Duration.fromMilliseconds(50), .awake); + } + } + + // Phase 3: Simulate crash and recover + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + var val = std.ArrayList(u8).empty; + defer val.deinit(allocator); + + // Verify all 20 keys are readable + var i: usize = 0; + while (i < 20) : (i += 1) { + var key_buf: [32]u8 = undefined; + const key = try std.fmt.bufPrint(&key_buf, "key_{d}", .{i}); + var expected_buf: [32]u8 = undefined; + const expected = try std.fmt.bufPrint(&expected_buf, "value_{d}", .{i}); + + try std.testing.expect(engineGetHelper(engine, key, &val, allocator)); + try std.testing.expectEqualStrings(expected, val.items); + } + } +} + +test "WalCompactionRecovery - UncommittedWritesSurviveRestart" { + // Note: engine.put() writes go to WAL and ARE recovered on restart. + // The committed/uncommitted distinction is at the transaction manager layer, + // not the engine layer. This test verifies uncommitted writes survive restart. + const allocator = std.testing.allocator; + const db_path = "db_wal_compaction_test_2"; + + var threaded = std.Io.Threaded.init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + const cwd = std.Io.Dir.cwd(); + + cwd.deleteTree(io, db_path) catch {}; + defer cwd.deleteTree(io, db_path) catch {}; + + var lsi_empty = [_][]const u8{}; + + // Write 10 keys (engine.put goes to WAL and IS durable) + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + var i: usize = 0; + while (i < 10) : (i += 1) { + var key_buf: [32]u8 = undefined; + const key = try std.fmt.bufPrint(&key_buf, "key_{d}", .{i}); + try std.testing.expect(engine.put(key, "value", &lsi_empty, "default", "")); + } + try io.sleep(std.Io.Duration.fromMilliseconds(50), .awake); + } + + // Recover and verify all keys are readable (WAL replay) + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + var val = std.ArrayList(u8).empty; + defer val.deinit(allocator); + + var i: usize = 0; + while (i < 10) : (i += 1) { + var key_buf: [32]u8 = undefined; + const key = try std.fmt.bufPrint(&key_buf, "key_{d}", .{i}); + try std.testing.expect(engineGetHelper(engine, key, &val, allocator)); + try std.testing.expectEqualStrings("value", val.items); + } + } +} + +test "WalCompactionRecovery - DeleteThenCompactThenRecover" { + // Verify that deletes (tombstones) survive compaction and recovery + const allocator = std.testing.allocator; + const db_path = "db_wal_compaction_test_3"; + + var threaded = std.Io.Threaded.init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + const cwd = std.Io.Dir.cwd(); + + cwd.deleteTree(io, db_path) catch {}; + defer cwd.deleteTree(io, db_path) catch {}; + + var lsi_empty = [_][]const u8{}; + + // Phase 1: Write 5 keys, delete 2 of them + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + // Write 5 keys + try std.testing.expect(engine.put("key_1", "value_1", &lsi_empty, "default", "")); + try std.testing.expect(engine.put("key_2", "value_2", &lsi_empty, "default", "")); + try std.testing.expect(engine.put("key_3", "value_3", &lsi_empty, "default", "")); + try std.testing.expect(engine.put("key_4", "value_4", &lsi_empty, "default", "")); + try std.testing.expect(engine.put("key_5", "value_5", &lsi_empty, "default", "")); + + try io.sleep(std.Io.Duration.fromMilliseconds(100), .awake); + + // Delete 2 keys + try std.testing.expect(engine.delete("key_2", &lsi_empty, "default", "")); + try std.testing.expect(engine.delete("key_4", &lsi_empty, "default", "")); + } + + // Phase 2: Run full compaction (tombstones should be preserved where needed) + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + var level: u32 = 0; + while (level < 7) : (level += 1) { + try engine.compactLevel(level); + try io.sleep(std.Io.Duration.fromMilliseconds(50), .awake); + } + } + + // Phase 3: Recover and verify deletes survived compaction + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + var val = std.ArrayList(u8).empty; + defer val.deinit(allocator); + + // Deleted keys should NOT be readable + try std.testing.expect(!engineGetHelper(engine, "key_2", &val, allocator)); + try std.testing.expect(!engineGetHelper(engine, "key_4", &val, allocator)); + + // Live keys should still be readable + try std.testing.expect(engineGetHelper(engine, "key_1", &val, allocator)); + try std.testing.expectEqualStrings("value_1", val.items); + try std.testing.expect(engineGetHelper(engine, "key_3", &val, allocator)); + try std.testing.expectEqualStrings("value_3", val.items); + try std.testing.expect(engineGetHelper(engine, "key_5", &val, allocator)); + try std.testing.expectEqualStrings("value_5", val.items); + } +} + +test "WalCompactionRecovery - MultiKeyTransactionSurvivesCompaction" { + const allocator = std.testing.allocator; + const db_path = "db_wal_compaction_test_4"; + + var threaded = std.Io.Threaded.init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + const cwd = std.Io.Dir.cwd(); + + cwd.deleteTree(io, db_path) catch {}; + defer cwd.deleteTree(io, db_path) catch {}; + + var lsi_empty = [_][]const u8{}; + + // Phase 1: Write a multi-key transaction (simulated as grouped writes) + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + // Write 5 keys as a logical transaction + try std.testing.expect(engine.put("tx_key_1", "tx_value_1", &lsi_empty, "default", "")); + try std.testing.expect(engine.put("tx_key_2", "tx_value_2", &lsi_empty, "default", "")); + try std.testing.expect(engine.put("tx_key_3", "tx_value_3", &lsi_empty, "default", "")); + try std.testing.expect(engine.put("tx_key_4", "tx_value_4", &lsi_empty, "default", "")); + try std.testing.expect(engine.put("tx_key_5", "tx_value_5", &lsi_empty, "default", "")); + + // Force flush to make them durable + try io.sleep(std.Io.Duration.fromMilliseconds(200), .awake); + } + + // Phase 2: Run full compaction + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + var level: u32 = 0; + while (level < 7) : (level += 1) { + try engine.compactLevel(level); + try io.sleep(std.Io.Duration.fromMilliseconds(50), .awake); + } + } + + // Phase 3: Recover and verify all 5 keys + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + var val = std.ArrayList(u8).empty; + defer val.deinit(allocator); + + var i: usize = 1; + while (i <= 5) : (i += 1) { + var key_buf: [32]u8 = undefined; + const key = try std.fmt.bufPrint(&key_buf, "tx_key_{d}", .{i}); + var expected_buf: [32]u8 = undefined; + const expected = try std.fmt.bufPrint(&expected_buf, "tx_value_{d}", .{i}); + + try std.testing.expect(engineGetHelper(engine, key, &val, allocator)); + try std.testing.expectEqualStrings(expected, val.items); + } + } +} + +test "WalCompactionRecovery - OverwrittenKeyRecovery" { + const allocator = std.testing.allocator; + const db_path = "db_wal_compaction_test_5"; + + var threaded = std.Io.Threaded.init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + const cwd = std.Io.Dir.cwd(); + + cwd.deleteTree(io, db_path) catch {}; + defer cwd.deleteTree(io, db_path) catch {}; + + var lsi_empty = [_][]const u8{}; + + // Phase 1: Write key=A + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + try std.testing.expect(engine.put("key", "value_A", &lsi_empty, "default", "")); + try io.sleep(std.Io.Duration.fromMilliseconds(100), .awake); + } + + // Compact + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + try engine.compactLevel(0); + try io.sleep(std.Io.Duration.fromMilliseconds(50), .awake); + } + + // Phase 2: Overwrite key=B + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + try std.testing.expect(engine.put("key", "value_B", &lsi_empty, "default", "")); + try io.sleep(std.Io.Duration.fromMilliseconds(100), .awake); + } + + // Compact again + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + try engine.compactLevel(0); + try io.sleep(std.Io.Duration.fromMilliseconds(50), .awake); + } + + // Phase 3: Recover and verify key=B (not A) + { + var engine = try EngineZig.init(allocator, db_path, 1024, false); + defer engine.deinit(); + + var val = std.ArrayList(u8).empty; + defer val.deinit(allocator); + + try std.testing.expect(engineGetHelper(engine, "key", &val, allocator)); + try std.testing.expectEqualStrings("value_B", val.items); + } +} \ No newline at end of file From aaf6d02b11fc59f39f90c3a10a6647f23e36d694 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Poyraz=20K=C3=BC=C3=A7=C3=BCkarslan?= <83272398+PoyrazK@users.noreply.github.com> Date: Wed, 10 Jun 2026 18:57:50 +0300 Subject: [PATCH 2/3] test(storage-engine): add gossip sharding rebalance tests Add gossip_sharding_rebalance_tests.zig with 5 tests covering: - Node join triggering ring updates - Concurrent writes during node rebalance - Node failure detection and data availability - Multi-node gossip discovery - Data availability during ownership changes Also update run_tests.zig to import all new test files. --- storage-engine/run_tests.zig | 4 + .../tests/gossip_sharding_rebalance_tests.zig | 520 ++++++++++++++++++ 2 files changed, 524 insertions(+) create mode 100644 storage-engine/tests/gossip_sharding_rebalance_tests.zig diff --git a/storage-engine/run_tests.zig b/storage-engine/run_tests.zig index a8ff757..e3f80da 100644 --- a/storage-engine/run_tests.zig +++ b/storage-engine/run_tests.zig @@ -3,6 +3,7 @@ comptime { _ = @import("tests/consistent_hash_tests.zig"); _ = @import("tests/memtable_tests.zig"); _ = @import("tests/recovery_tests.zig"); + _ = @import("tests/wal_compaction_recovery_tests.zig"); _ = @import("tests/sstable_tests.zig"); _ = @import("tests/sstable_deep_tests.zig"); _ = @import("tests/skiplist_memtable_deep_tests.zig"); @@ -34,6 +35,9 @@ comptime { _ = @import("tests/crash_fuzz_tests.zig"); _ = @import("tests/range_query_tests.zig"); _ = @import("tests/leveled_compaction_tests.zig"); + _ = @import("tests/compaction_transaction_tests.zig"); + _ = @import("tests/cdc_transaction_tests.zig"); + _ = @import("tests/gossip_sharding_rebalance_tests.zig"); } diff --git a/storage-engine/tests/gossip_sharding_rebalance_tests.zig b/storage-engine/tests/gossip_sharding_rebalance_tests.zig new file mode 100644 index 0000000..dc4a4c2 --- /dev/null +++ b/storage-engine/tests/gossip_sharding_rebalance_tests.zig @@ -0,0 +1,520 @@ +const std = @import("std"); +const EngineZig = @import("../src/storage/engine.zig").EngineZig; +const Ring = @import("../src/distributed/consistent_hash.zig").Ring; +const GossipProtocol = @import("../src/distributed/gossip.zig").GossipProtocol; +const NodeInfoC = @import("../src/distributed/gossip.zig").NodeInfoC; + +const c = @cImport({ + @cInclude("unistd.h"); +}); + +extern fn gossip_join(ptr: ?*anyopaque, seed_host_ptr: [*]const u8, seed_host_len: usize, seed_port: u32) callconv(.c) bool; + +const TestNode = struct { + id: []const u8, + tcp_port: u32, + gossip_port: u32, + db_path: []const u8, + engine: *EngineZig, + ring: *Ring, + gossip: *GossipProtocol, + running: std.atomic.Value(bool), + ring_updater: std.Thread, + allocator: std.mem.Allocator, + + pub fn init(allocator: std.mem.Allocator, id: []const u8, tcp_port: u32, gossip_port: u32, db_path: []const u8) !*TestNode { + var threaded = std.Io.Threaded.init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + const cwd = std.Io.Dir.cwd(); + cwd.deleteTree(io, db_path) catch {}; + + const self = try allocator.create(TestNode); + errdefer allocator.destroy(self); + + const id_dup = try allocator.dupe(u8, id); + errdefer allocator.free(id_dup); + + const db_path_dup = try allocator.dupe(u8, db_path); + errdefer allocator.free(db_path_dup); + + const engine = try EngineZig.init(allocator, db_path, 2 * 1024 * 1024, false); + errdefer engine.deinit(); + + const ring = try allocator.create(Ring); + errdefer allocator.destroy(ring); + ring.* = try Ring.init(allocator, 256); + errdefer ring.deinit(); + try ring.addNode(id_dup); + + const gossip = try GossipProtocol.init(allocator, id_dup, "127.0.0.1", gossip_port); + errdefer gossip.deinit(); + + self.* = TestNode{ + .id = id_dup, + .tcp_port = tcp_port, + .gossip_port = gossip_port, + .db_path = db_path_dup, + .engine = engine, + .ring = ring, + .gossip = gossip, + .running = std.atomic.Value(bool).init(true), + .ring_updater = undefined, + .allocator = allocator, + }; + + return self; + } + + pub fn deinit(self: *TestNode) void { + self.running.store(false, .monotonic); + self.ring_updater.join(); + self.gossip.stop(); + self.gossip.deinit(); + self.engine.deinit(); + + var threaded = std.Io.Threaded.init(self.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + const cwd = std.Io.Dir.cwd(); + cwd.deleteTree(io, self.db_path) catch {}; + + self.ring.deinit(); + self.allocator.destroy(self.ring); + self.allocator.free(self.id); + self.allocator.free(self.db_path); + self.allocator.destroy(self); + } + + pub fn start(self: *TestNode, seed_host: []const u8, seed_port: u32) !void { + try self.gossip.start(); + + if (seed_host.len > 0 and seed_port != 0) { + _ = gossip_join(self.gossip, seed_host.ptr, seed_host.len, seed_port); + } + + self.ring_updater = try std.Thread.spawn(.{}, ringUpdaterLoop, .{self}); + } +}; + +fn ringUpdaterLoop(node: *TestNode) void { + var current_nodes = std.StringHashMap(void).init(node.allocator); + defer current_nodes.deinit(); + + while (node.running.load(.monotonic)) { + _ = c.usleep(50 * 1000); + + var alive_nodes: [256]NodeInfoC = undefined; + var alive_count: usize = 0; + node.gossip.getAliveNodes(&alive_nodes, 256, &alive_count); + + var new_nodes = std.StringHashMap(void).init(node.allocator); + defer new_nodes.deinit(); + + var idx: usize = 0; + while (idx < alive_count) : (idx += 1) { + const nid = alive_nodes[idx].id[0..alive_nodes[idx].id_len]; + if (!current_nodes.contains(nid)) { + new_nodes.put(nid, {}) catch {}; + } + } + + var it = new_nodes.iterator(); + while (it.next()) |entry| { + const nid = entry.key_ptr.*; + if (!std.mem.eql(u8, nid, node.id)) { + node.ring.addNode(nid) catch {}; + current_nodes.put(nid, {}) catch {}; + } + } + } +} + +const GetContext = struct { + allocator: std.mem.Allocator, + val: []const u8 = &.{}, + + fn callback(obj: ?*anyopaque, val_ptr: [*]const u8, val_len: usize) callconv(.c) void { + const self = @as(*@This(), @ptrCast(@alignCast(obj.?))); + self.val = self.allocator.dupe(u8, val_ptr[0..val_len]) catch &.{}; + } +}; + +test "GossipShardingRebalance - NodeJoinTriggersRingUpdate" { + // Test that when a new node joins, the ring is updated to include it + const allocator = std.testing.allocator; + + // Node 1 - first node, no seed + const node1 = try TestNode.init(allocator, "127.0.0.1:9700", 9700, 9800, "db_rebalance_n1"); + defer node1.deinit(); + try node1.start("", 0); + _ = c.usleep(100 * 1000); + + // Node 2 - joins node1's cluster + const node2 = try TestNode.init(allocator, "127.0.0.1:9701", 9701, 9801, "db_rebalance_n2"); + defer node2.deinit(); + try node2.start("127.0.0.1", 9800); + + // Wait for node2 to be detected by node1's gossip + var node2_discovered = false; + var attempt: usize = 0; + while (attempt < 100) : (attempt += 1) { + _ = c.usleep(100 * 1000); + + var count: usize = 0; + var nodes: [10]NodeInfoC = undefined; + node1.gossip.getAliveNodes(&nodes, 10, &count); + + if (count >= 2) { + node2_discovered = true; + break; + } + } + + try std.testing.expect(node2_discovered); + + // Verify ring now knows about both nodes (getNode returns non-empty slice) + const owner = node1.ring.getNode("any_key"); + try std.testing.expect(owner.len > 0); +} + +test "GossipShardingRebalance - ConcurrentWritesDuringRebalance" { + const allocator = std.testing.allocator; + const num_initial_keys = 10; + const num_concurrent_writes = 20; + + // Node 1 - first node + const node1 = try TestNode.init(allocator, "127.0.0.1:9702", 9702, 9802, "db_rebalance_n3"); + defer node1.deinit(); + try node1.start("", 0); + _ = c.usleep(100 * 1000); + + // Write initial keys to node1 + var i: usize = 0; + while (i < num_initial_keys) : (i += 1) { + var key_buf: [32]u8 = undefined; + const key = try std.fmt.bufPrint(&key_buf, "initial_key_{d}", .{i}); + var val_buf: [32]u8 = undefined; + const val = try std.fmt.bufPrint(&val_buf, "value_{d}", .{i}); + try std.testing.expect(node1.engine.put(key, val, &[_][]const u8{}, "default", "")); + } + + // Spawn writer thread for concurrent writes + var writer_done = std.atomic.Value(bool).init(false); + const writer = try std.Thread.spawn(.{}, struct { + fn f(node: *TestNode, done: *std.atomic.Value(bool), max_writes: usize) void { + var j: usize = 0; + while (j < max_writes) : (j += 1) { + var key_buf: [32]u8 = undefined; + const key = std.fmt.bufPrint(&key_buf, "concurrent_key_{d}", .{j}) catch continue; + var val_buf: [32]u8 = undefined; + const val = std.fmt.bufPrint(&val_buf, "value_{d}", .{j}) catch continue; + _ = node.engine.put(key, val, &[_][]const u8{}, "default", ""); + _ = c.usleep(10 * 1000); + } + done.store(true, .release); + } + }.f, .{ node1, &writer_done, num_concurrent_writes }); + + // Node 2 joins while writes are happening + _ = c.usleep(50 * 1000); // Let writer get some writes in + + const node2 = try TestNode.init(allocator, "127.0.0.1:9703", 9703, 9803, "db_rebalance_n4"); + defer node2.deinit(); + try node2.start("127.0.0.1", 9802); + + // Wait for writer to finish + while (!writer_done.load(.acquire)) { + _ = c.usleep(50 * 1000); + } + writer.join(); + + // Wait for cluster to stabilize + var attempt: usize = 0; + while (attempt < 100) : (attempt += 1) { + _ = c.usleep(100 * 1000); + var count: usize = 0; + var nodes: [10]NodeInfoC = undefined; + node1.gossip.getAliveNodes(&nodes, 10, &count); + if (count >= 2) break; + } + + // Verify all keys are readable from some node + var total_found: usize = 0; + i = 0; + while (i < num_initial_keys) : (i += 1) { + var key_buf: [32]u8 = undefined; + const key = try std.fmt.bufPrint(&key_buf, "initial_key_{d}", .{i}); + + var get_ctx1 = GetContext{ .allocator = allocator }; + const in_n1 = node1.engine.get(key, &get_ctx1, GetContext.callback, "default", ""); + if (in_n1) { + allocator.free(get_ctx1.val); + total_found += 1; + continue; + } + + var get_ctx2 = GetContext{ .allocator = allocator }; + const in_n2 = node2.engine.get(key, &get_ctx2, GetContext.callback, "default", ""); + if (in_n2) { + allocator.free(get_ctx2.val); + total_found += 1; + } + } + + i = 0; + while (i < num_concurrent_writes) : (i += 1) { + var key_buf: [32]u8 = undefined; + const key = try std.fmt.bufPrint(&key_buf, "concurrent_key_{d}", .{i}); + + var get_ctx1 = GetContext{ .allocator = allocator }; + const in_n1 = node1.engine.get(key, &get_ctx1, GetContext.callback, "default", ""); + if (in_n1) { + allocator.free(get_ctx1.val); + total_found += 1; + continue; + } + + var get_ctx2 = GetContext{ .allocator = allocator }; + const in_n2 = node2.engine.get(key, &get_ctx2, GetContext.callback, "default", ""); + if (in_n2) { + allocator.free(get_ctx2.val); + total_found += 1; + } + } + + // All initial + concurrent keys should be found + try std.testing.expectEqual(@as(usize, num_initial_keys + num_concurrent_writes), total_found); +} + +test "GossipShardingRebalance - NodeFailureTriggersRedistribution" { + const allocator = std.testing.allocator; + const num_keys = 15; + + // Node 1 + const node1 = try TestNode.init(allocator, "127.0.0.1:9710", 9710, 9810, "db_rebalance_fn1"); + defer node1.deinit(); + try node1.start("", 0); + + // Node 2 + const node2 = try TestNode.init(allocator, "127.0.0.1:9711", 9711, 9811, "db_rebalance_fn2"); + defer node2.deinit(); + try node2.start("127.0.0.1", 9810); + + // Node 3 + const node3 = try TestNode.init(allocator, "127.0.0.1:9712", 9712, 9812, "db_rebalance_fn3"); + defer node3.deinit(); + try node3.start("127.0.0.1", 9810); + + // Wait for cluster formation + var attempt: usize = 0; + while (attempt < 100) : (attempt += 1) { + _ = c.usleep(100 * 1000); + var count: usize = 0; + var nodes: [10]NodeInfoC = undefined; + node1.gossip.getAliveNodes(&nodes, 10, &count); + if (count >= 3) break; + } + + // Write keys - they should be distributed across all 3 nodes + var i: usize = 0; + while (i < num_keys) : (i += 1) { + var key_buf: [32]u8 = undefined; + const key = try std.fmt.bufPrint(&key_buf, "key_{d}", .{i}); + var val_buf: [32]u8 = undefined; + const val = try std.fmt.bufPrint(&val_buf, "value_{d}", .{i}); + try std.testing.expect(node1.engine.put(key, val, &[_][]const u8{}, "default", "")); + } + + // Wait for distribution + attempt = 0; + while (attempt < 100) : (attempt += 1) { + _ = c.usleep(100 * 1000); + var count: usize = 0; + var nodes: [10]NodeInfoC = undefined; + node1.gossip.getAliveNodes(&nodes, 10, &count); + if (count >= 3) break; + } + + // Stop node2's gossip to simulate failure + node2.gossip.stop(); + + // Wait for failure to be detected + _ = c.usleep(2000 * 1000); // 2 seconds + + // Verify all keys are still readable from node1 or node3 + var keys_found: usize = 0; + i = 0; + while (i < num_keys) : (i += 1) { + var key_buf: [32]u8 = undefined; + const key = try std.fmt.bufPrint(&key_buf, "key_{d}", .{i}); + + var get_ctx1 = GetContext{ .allocator = allocator }; + const in_n1 = node1.engine.get(key, &get_ctx1, GetContext.callback, "default", ""); + if (in_n1) { + allocator.free(get_ctx1.val); + keys_found += 1; + continue; + } + + var get_ctx3 = GetContext{ .allocator = allocator }; + const in_n3 = node3.engine.get(key, &get_ctx3, GetContext.callback, "default", ""); + if (in_n3) { + allocator.free(get_ctx3.val); + keys_found += 1; + } + } + + // All keys should still be readable from remaining nodes + try std.testing.expectEqual(@as(usize, num_keys), keys_found); +} + +test "GossipShardingRebalance - MultiNodeGossipDiscovery" { + // Test that all nodes discover each other via gossip + const allocator = std.testing.allocator; + + // Node 1 - initial + const node1 = try TestNode.init(allocator, "127.0.0.1:9720", 9720, 9820, "db_rebalance_mj1"); + defer node1.deinit(); + try node1.start("", 0); + _ = c.usleep(100 * 1000); + + // Node 2 joins + const node2 = try TestNode.init(allocator, "127.0.0.1:9721", 9721, 9821, "db_rebalance_mj2"); + defer node2.deinit(); + try node2.start("127.0.0.1", 9820); + + // Node 3 joins + const node3 = try TestNode.init(allocator, "127.0.0.1:9722", 9722, 9822, "db_rebalance_mj3"); + defer node3.deinit(); + try node3.start("127.0.0.1", 9820); + + // Node 4 joins + const node4 = try TestNode.init(allocator, "127.0.0.1:9723", 9723, 9823, "db_rebalance_mj4"); + defer node4.deinit(); + try node4.start("127.0.0.1", 9820); + + // Wait for all nodes to be aware of each other + var attempt: usize = 0; + var all_discovered = false; + while (attempt < 100) : (attempt += 1) { + _ = c.usleep(100 * 1000); + + var count1: usize = 0; + var nodes1: [10]NodeInfoC = undefined; + node1.gossip.getAliveNodes(&nodes1, 10, &count1); + + var count2: usize = 0; + var nodes2: [10]NodeInfoC = undefined; + node2.gossip.getAliveNodes(&nodes2, 10, &count2); + + var count3: usize = 0; + var nodes3: [10]NodeInfoC = undefined; + node3.gossip.getAliveNodes(&nodes3, 10, &count3); + + var count4: usize = 0; + var nodes4: [10]NodeInfoC = undefined; + node4.gossip.getAliveNodes(&nodes4, 10, &count4); + + if (count1 >= 4 and count2 >= 4 and count3 >= 4 and count4 >= 4) { + all_discovered = true; + break; + } + } + + try std.testing.expect(all_discovered); +} + +test "GossipShardingRebalance - StaleReadDuringOwnershipChange" { + const allocator = std.testing.allocator; + const num_keys = 10; + + // Node 1 + const node1 = try TestNode.init(allocator, "127.0.0.1:9730", 9730, 9830, "db_rebalance_stale1"); + defer node1.deinit(); + try node1.start("", 0); + _ = c.usleep(100 * 1000); + + // Write keys + var i: usize = 0; + while (i < num_keys) : (i += 1) { + var key_buf: [32]u8 = undefined; + const key = try std.fmt.bufPrint(&key_buf, "key_{d}", .{i}); + var val_buf: [32]u8 = undefined; + const val = try std.fmt.bufPrint(&val_buf, "value_{d}", .{i}); + try std.testing.expect(node1.engine.put(key, val, &[_][]const u8{}, "default", "")); + } + + // Node 2 joins + const node2 = try TestNode.init(allocator, "127.0.0.1:9731", 9731, 9831, "db_rebalance_stale2"); + defer node2.deinit(); + try node2.start("127.0.0.1", 9830); + + // Wait for rebalance + var attempt: usize = 0; + while (attempt < 100) : (attempt += 1) { + _ = c.usleep(100 * 1000); + + var all_correct = true; + var count: usize = 0; + var nodes: [10]NodeInfoC = undefined; + node1.gossip.getAliveNodes(&nodes, 10, &count); + + if (count >= 2) { + var k: usize = 0; + while (k < num_keys) : (k += 1) { + var key_buf: [32]u8 = undefined; + const key = try std.fmt.bufPrint(&key_buf, "key_{d}", .{k}); + const owner = node1.ring.getNode(key); + + var get_ctx1 = GetContext{ .allocator = allocator }; + const in_n1 = node1.engine.get(key, &get_ctx1, GetContext.callback, "default", ""); + if (in_n1) allocator.free(get_ctx1.val); + + var get_ctx2 = GetContext{ .allocator = allocator }; + const in_n2 = node2.engine.get(key, &get_ctx2, GetContext.callback, "default", ""); + if (in_n2) allocator.free(get_ctx2.val); + + if (std.mem.eql(u8, owner, "127.0.0.1:9730")) { + if (!in_n1 or in_n2) all_correct = false; + } else if (std.mem.eql(u8, owner, "127.0.0.1:9731")) { + if (!in_n2 or in_n1) all_correct = false; + } + } + } else { + all_correct = false; + } + + if (all_correct) break; + } + + // At minimum, both nodes should be aware of each other + var count: usize = 0; + var nodes: [10]NodeInfoC = undefined; + node1.gossip.getAliveNodes(&nodes, 10, &count); + try std.testing.expect(count >= 2); + + // All keys should be readable from some node (no lost keys during rebalance) + i = 0; + while (i < num_keys) : (i += 1) { + var key_buf: [32]u8 = undefined; + const key = try std.fmt.bufPrint(&key_buf, "key_{d}", .{i}); + + var get_ctx1 = GetContext{ .allocator = allocator }; + const in_n1 = node1.engine.get(key, &get_ctx1, GetContext.callback, "default", ""); + if (in_n1) { + allocator.free(get_ctx1.val); + continue; + } + + var get_ctx2 = GetContext{ .allocator = allocator }; + const in_n2 = node2.engine.get(key, &get_ctx2, GetContext.callback, "default", ""); + if (in_n2) { + allocator.free(get_ctx2.val); + continue; + } + + // Key should be on at least one node + try std.testing.expect(in_n1 or in_n2); + } +} \ No newline at end of file From c9008d998371705b4acc1dbaee6e88e5b8ec536d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Poyraz=20K=C3=BC=C3=A7=C3=BCkarslan?= <83272398+PoyrazK@users.noreply.github.com> Date: Wed, 10 Jun 2026 19:28:55 +0300 Subject: [PATCH 3/3] test: add trailing newlines and remove unused getEngineCallbacks function - Add trailing newline to cdc_transaction_tests.zig - Add trailing newline to compaction_transaction_tests.zig - Add trailing newline to gossip_sharding_rebalance_tests.zig - Add trailing newline to wal_compaction_recovery_tests.zig - Remove unused getEngineCallbacks() function from compaction_transaction_tests.zig --- .../tests/cdc_transaction_tests.zig | 2 +- .../tests/compaction_transaction_tests.zig | 30 +------------------ .../tests/gossip_sharding_rebalance_tests.zig | 2 +- .../tests/wal_compaction_recovery_tests.zig | 2 +- 4 files changed, 4 insertions(+), 32 deletions(-) diff --git a/storage-engine/tests/cdc_transaction_tests.zig b/storage-engine/tests/cdc_transaction_tests.zig index 1861449..b58785e 100644 --- a/storage-engine/tests/cdc_transaction_tests.zig +++ b/storage-engine/tests/cdc_transaction_tests.zig @@ -535,4 +535,4 @@ test "CdcTransaction - EventsHaveIncrementingSequence" { try std.testing.expectEqual(@as(usize, 3), received_events.items.len); try std.testing.expect(received_events.items[0].seq < received_events.items[1].seq); try std.testing.expect(received_events.items[1].seq < received_events.items[2].seq); -} \ No newline at end of file +} diff --git a/storage-engine/tests/compaction_transaction_tests.zig b/storage-engine/tests/compaction_transaction_tests.zig index adbb06d..fe08a7c 100644 --- a/storage-engine/tests/compaction_transaction_tests.zig +++ b/storage-engine/tests/compaction_transaction_tests.zig @@ -73,34 +73,6 @@ fn queryCallback( ) catch {}; } -fn getEngineCallbacks() EngineCallbackTable { - return EngineCallbackTable{ - .put = struct { - fn f(ctx: ?*anyopaque, pk: [*]const u8, pk_len: usize, val: [*]const u8, val_len: usize, table: [*]const u8, table_len: usize, sk: [*]const u8, sk_len: usize) callconv(.c) bool { - const eng = @as(*EngineZig, @ptrCast(@alignCast(ctx.?))); - return eng.put(pk[0..pk_len], val[0..val_len], &[_][]const u8{}, table[0..table_len], sk[0..sk_len]); - } - }.f, - .get = struct { - fn f(ctx: ?*anyopaque, pk: [*]const u8, pk_len: usize, table: [*]const u8, table_len: usize, sk: [*]const u8, sk_len: usize, val_cb_obj: ?*anyopaque, val_cb: ?*const fn (obj: ?*anyopaque, [*]const u8, usize) callconv(.c) void) callconv(.c) bool { - const eng = @as(*EngineZig, @ptrCast(@alignCast(ctx.?))); - return eng.get(pk[0..pk_len], val_cb_obj, val_cb, table[0..table_len], sk[0..sk_len]); - } - }.f, - .delete = struct { - fn f(ctx: ?*anyopaque, pk: [*]const u8, pk_len: usize, table: [*]const u8, table_len: usize, sk: [*]const u8, sk_len: usize) callconv(.c) void { - const eng = @as(*EngineZig, @ptrCast(@alignCast(ctx.?))); - _ = eng.delete(pk[0..pk_len], &[_][]const u8{}, table[0..table_len], sk[0..sk_len]); - } - }.f, - .query = struct { - fn f(ctx: ?*anyopaque, table: [*]const u8, table_len: usize, pk: [*]const u8, pk_len: usize, op: [*]const u8, op_len: usize, sk_prefix: [*]const u8, sk_prefix_len: usize, cb_obj: ?*anyopaque, cb: ?*const fn (obj: ?*anyopaque, [*]const u8, usize, [*]const u8, usize) callconv(.c) void) callconv(.c) void { - const eng = @as(*EngineZig, @ptrCast(@alignCast(ctx.?))); - eng.query(table[0..table_len], pk[0..pk_len], op[0..op_len], sk_prefix[0..sk_prefix_len], cb_obj, cb) catch {}; - } - }.f, - }; -} fn makeItem(table: []const u8, pk: []const u8, sk: []const u8, value: []const u8) TransactionItemC { return TransactionItemC{ @@ -344,4 +316,4 @@ test "CompactionTransaction - TombstoneIsolationDuringCompaction" { // Key should not be readable try std.testing.expect(!engineGetHelper(engine, "key1", &val, allocator)); } -} \ No newline at end of file +} diff --git a/storage-engine/tests/gossip_sharding_rebalance_tests.zig b/storage-engine/tests/gossip_sharding_rebalance_tests.zig index dc4a4c2..ab0a661 100644 --- a/storage-engine/tests/gossip_sharding_rebalance_tests.zig +++ b/storage-engine/tests/gossip_sharding_rebalance_tests.zig @@ -517,4 +517,4 @@ test "GossipShardingRebalance - StaleReadDuringOwnershipChange" { // Key should be on at least one node try std.testing.expect(in_n1 or in_n2); } -} \ No newline at end of file +} diff --git a/storage-engine/tests/wal_compaction_recovery_tests.zig b/storage-engine/tests/wal_compaction_recovery_tests.zig index 11a2110..af83179 100644 --- a/storage-engine/tests/wal_compaction_recovery_tests.zig +++ b/storage-engine/tests/wal_compaction_recovery_tests.zig @@ -322,4 +322,4 @@ test "WalCompactionRecovery - OverwrittenKeyRecovery" { try std.testing.expect(engineGetHelper(engine, "key", &val, allocator)); try std.testing.expectEqualStrings("value_B", val.items); } -} \ No newline at end of file +}