feat: add undefinedServerForDrainTest and Ctx to server
This commit is contained in:
1 parent
6e8c2bae39
commit
44c5e9b9cb
1 file changed
+160
+160
@@ -2011,3 +2011,163 @@ fn testDb(allocator: Allocator, dir: []const u8) !Db {
|
|||||||
return Db.open(allocator, dir, .{});
|
return Db.open(allocator, dir, .{});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// `drainConnections` only ever touches `self.active`/`self.stopping`, so it
|
||||||
|
// can be exercised on a `Server` value that was never `init()`ed with a
|
||||||
|
// real `Db`/socket - every other field stays `undefined` and is never
|
||||||
|
// read. This isolates the drain/timeout contract itself from the (much
|
||||||
|
// harder to force deterministically) real-socket race it exists to close.
|
||||||
|
fn undefinedServerForDrainTest() Server {
|
||||||
|
var srv: Server = undefined;
|
||||||
|
srv.active = std.atomic.Value(usize).init(0);
|
||||||
|
srv.stopping = std.atomic.Value(bool).init(false);
|
||||||
|
return srv;
|
||||||
|
}
|
||||||
|
|
||||||
|
test "drainConnections returns immediately when there is nothing to drain" {
|
||||||
|
var srv = undefinedServerForDrainTest();
|
||||||
|
const start = std.time.nanoTimestamp();
|
||||||
|
srv.drainConnections(2 * std.time.ns_per_s);
|
||||||
|
const elapsedNanos = std.time.nanoTimestamp() - start;
|
||||||
|
try std.testing.expect(elapsedNanos < 200 * std.time.ns_per_ms);
|
||||||
|
}
|
||||||
|
|
||||||
|
test "drainConnections waits for a connection that finishes before the timeout" {
|
||||||
|
var srv = undefinedServerForDrainTest();
|
||||||
|
srv.active.store(1, .release);
|
||||||
|
|
||||||
|
const Ctx = struct {
|
||||||
|
srv: *Server,
|
||||||
|
fn run(ctx: *@This()) void {
|
||||||
|
std.time.sleep(30 * std.time.ns_per_ms);
|
||||||
|
_ = ctx.srv.active.fetchSub(1, .acq_rel);
|
||||||
|
}
|
||||||
|
};
|
||||||
|
var ctx = Ctx{ .srv = &srv };
|
||||||
|
const t = try std.Thread.spawn(.{}, Ctx.run, .{&ctx});
|
||||||
|
defer t.join();
|
||||||
|
|
||||||
|
const start = std.time.nanoTimestamp();
|
||||||
|
srv.drainConnections(2 * std.time.ns_per_s);
|
||||||
|
const elapsedNanos = std.time.nanoTimestamp() - start;
|
||||||
|
|
||||||
|
try std.testing.expectEqual(@as(usize, 0), srv.activeConnections());
|
||||||
|
// Returned once the worker actually finished, not after burning the
|
||||||
|
// full 2-second timeout - proves this is a real wait-for-drain, not a
|
||||||
|
// fixed sleep in disguise.
|
||||||
|
try std.testing.expect(elapsedNanos < 1 * std.time.ns_per_s);
|
||||||
|
}
|
||||||
|
|
||||||
|
test "drainConnections gives up after the timeout instead of hanging forever" {
|
||||||
|
var srv = undefinedServerForDrainTest();
|
||||||
|
srv.active.store(1, .release); // never decremented by anyone
|
||||||
|
|
||||||
|
const start = std.time.nanoTimestamp();
|
||||||
|
srv.drainConnections(20 * std.time.ns_per_ms);
|
||||||
|
const elapsedNanos = std.time.nanoTimestamp() - start;
|
||||||
|
|
||||||
|
try std.testing.expect(elapsedNanos >= 20 * std.time.ns_per_ms);
|
||||||
|
// Still stuck at 1 - drainConnections gave up, it did not (and cannot)
|
||||||
|
// force the phantom connection closed itself.
|
||||||
|
try std.testing.expectEqual(@as(usize, 1), srv.activeConnections());
|
||||||
|
}
|
||||||
|
|
||||||
|
test "server responds to ping, set, get, and delete over loopback" {
|
||||||
|
const allocator = std.testing.allocator;
|
||||||
|
var tmp = std.testing.tmpDir(.{});
|
||||||
|
defer tmp.cleanup();
|
||||||
|
const path = try tmp.dir.realpathAlloc(allocator, ".");
|
||||||
|
defer allocator.free(path);
|
||||||
|
|
||||||
|
var db = try testDb(allocator, path);
|
||||||
|
defer db.close();
|
||||||
|
|
||||||
|
const address = try std.net.Address.parseIp("127.0.0.1", 0);
|
||||||
|
var server = try Server.init(allocator, &db, address);
|
||||||
|
try server.start();
|
||||||
|
defer server.stop();
|
||||||
|
|
||||||
|
const client_mod = @import("client.zig");
|
||||||
|
var client = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||||
|
defer client.close();
|
||||||
|
|
||||||
|
try client.ping();
|
||||||
|
try client.set("hello", "world");
|
||||||
|
const got = try client.get(allocator, "hello");
|
||||||
|
defer if (got) |g| allocator.free(g);
|
||||||
|
try std.testing.expectEqualStrings("world", got.?);
|
||||||
|
|
||||||
|
try client.delete("hello");
|
||||||
|
const afterDelete = try client.get(allocator, "hello");
|
||||||
|
try std.testing.expect(afterDelete == null);
|
||||||
|
}
|
||||||
|
|
||||||
|
test "server scan, prefix, and stats round trip real data" {
|
||||||
|
const allocator = std.testing.allocator;
|
||||||
|
var tmp = std.testing.tmpDir(.{});
|
||||||
|
defer tmp.cleanup();
|
||||||
|
const path = try tmp.dir.realpathAlloc(allocator, ".");
|
||||||
|
defer allocator.free(path);
|
||||||
|
|
||||||
|
var db = try testDb(allocator, path);
|
||||||
|
defer db.close();
|
||||||
|
|
||||||
|
const address = try std.net.Address.parseIp("127.0.0.1", 0);
|
||||||
|
var server = try Server.init(allocator, &db, address);
|
||||||
|
try server.start();
|
||||||
|
defer server.stop();
|
||||||
|
|
||||||
|
const client_mod = @import("client.zig");
|
||||||
|
var client = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||||
|
defer client.close();
|
||||||
|
|
||||||
|
try client.set("user:1", "alice");
|
||||||
|
try client.set("user:2", "bob");
|
||||||
|
try client.set("order:1", "widget");
|
||||||
|
|
||||||
|
const prefixed = try client.prefix(allocator, "user:", 0);
|
||||||
|
defer protocol.freeOwnedEntries(allocator, prefixed);
|
||||||
|
try std.testing.expectEqual(@as(usize, 2), prefixed.len);
|
||||||
|
|
||||||
|
const scanned = try client.scan(allocator, .{});
|
||||||
|
defer protocol.freeOwnedEntries(allocator, scanned);
|
||||||
|
try std.testing.expectEqual(@as(usize, 3), scanned.len);
|
||||||
|
|
||||||
|
const s = try client.stats();
|
||||||
|
try std.testing.expectEqual(@as(usize, 3), s.memtableEntries);
|
||||||
|
}
|
||||||
|
|
||||||
|
test "server batch applies puts and deletes atomically" {
|
||||||
|
const allocator = std.testing.allocator;
|
||||||
|
var tmp = std.testing.tmpDir(.{});
|
||||||
|
defer tmp.cleanup();
|
||||||
|
const path = try tmp.dir.realpathAlloc(allocator, ".");
|
||||||
|
defer allocator.free(path);
|
||||||
|
|
||||||
|
var db = try testDb(allocator, path);
|
||||||
|
defer db.close();
|
||||||
|
|
||||||
|
const address = try std.net.Address.parseIp("127.0.0.1", 0);
|
||||||
|
var server = try Server.init(allocator, &db, address);
|
||||||
|
try server.start();
|
||||||
|
defer server.stop();
|
||||||
|
|
||||||
|
const client_mod = @import("client.zig");
|
||||||
|
var client = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||||
|
defer client.close();
|
||||||
|
|
||||||
|
try client.set("keep", "1");
|
||||||
|
const ops = [_]protocol.BatchOp{
|
||||||
|
.{ .put = .{ .key = "a", .value = "1" } },
|
||||||
|
.{ .put = .{ .key = "b", .value = "2" } },
|
||||||
|
.{ .del = "keep" },
|
||||||
|
};
|
||||||
|
try client.batch(&ops);
|
||||||
|
|
||||||
|
const a = try client.get(allocator, "a");
|
||||||
|
defer if (a) |v| allocator.free(v);
|
||||||
|
try std.testing.expectEqualStrings("1", a.?);
|
||||||
|
|
||||||
|
const kept = try client.get(allocator, "keep");
|
||||||
|
try std.testing.expect(kept == null);
|
||||||
|
}
|
||||||
|
|
||||||
Reference in new issue
Block a user