diff --git a/src/server.zig b/src/server.zig index 935201d..4067cf8 100644 --- a/src/server.zig +++ b/src/server.zig @@ -1851,3 +1851,163 @@ pub const Server = struct { try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), out); } + /// Executes a `user_admin` request (create/set_password/set_role/delete/ + /// list) against a fresh `authmod.Store` over the same `*Db` the rest of + /// the server serves. Deliberately does NOT consult `self.authStore` - + /// that field only controls whether *login enforcement* is turned on + /// for this server, not whether the user table exists, so remote user + /// management works the same way whether or not `--auth` is active. + /// `requiredRole` already gates this opcode behind `admin` whenever + /// auth enforcement IS enabled; a server with auth disabled enforces + /// nothing on any opcode, this one included, matching every other op's + /// documented behavior. + fn handleUserAdmin(self: *Server, payload: []const u8, writer: anytype) !void { + const args = protocol.decodeUserAdmin(payload) catch { + return writeErr(writer, .bad_request, "malformed user_admin payload"); + }; + const store = authmod.Store.init(self.db); + + switch (args.action) { + .create => { + const role = authmod.Role.fromByte(args.roleByte) catch { + return writeErr(writer, .bad_request, "invalid role"); + }; + const rounds = if (args.rounds != 0) args.rounds else authmod.recommendedRounds; + store.createUser(self.allocator, args.username, args.password, role, rounds) catch |err| { + return switch (err) { + error.UserExists => writeErr(writer, .conflict, "user already exists"), + else => writeErr(writer, .internal, @errorName(err)), + }; + }; + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), ""); + }, + + .set_password => { + const rounds = if (args.rounds != 0) args.rounds else authmod.recommendedRounds; + store.setPassword(self.allocator, args.username, args.password, rounds) catch |err| { + return switch (err) { + error.UserNotFound => writeErr(writer, .not_found, "user not found"), + else => writeErr(writer, .internal, @errorName(err)), + }; + }; + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), ""); + }, + + .set_role => { + const role = authmod.Role.fromByte(args.roleByte) catch { + return writeErr(writer, .bad_request, "invalid role"); + }; + store.setRole(self.allocator, args.username, role) catch |err| { + return switch (err) { + error.UserNotFound => writeErr(writer, .not_found, "user not found"), + else => writeErr(writer, .internal, @errorName(err)), + }; + }; + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), ""); + }, + + .delete => { + store.deleteUser(self.allocator, args.username) catch |err| { + return switch (err) { + error.UserNotFound => writeErr(writer, .not_found, "user not found"), + else => writeErr(writer, .internal, @errorName(err)), + }; + }; + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), ""); + }, + + .list => { + const users = store.listUsers(self.allocator) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + defer authmod.freeUserList(self.allocator, users); + const out = protocol.encodeUserList(self.allocator, users) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + defer self.allocator.free(out); + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), out); + }, + } + } + + /// Stops accepting new connections and waits for the accept loop to + /// exit. Closing the listening socket while another thread is + /// blocked inside `accept()` on that same fd is a race the standard + /// library treats as a bug (it asserts on EBADF), so instead we flip + /// `stopping` and connect a throwaway client to our own listen + /// address: that wakes the pending `accept()` with a real, briefly + /// alive connection, which the loop then discards and exits on. Only + /// once the accept thread has rejoined is it safe to close the + /// listening socket itself. + /// + /// Already-accepted connections are `.detach()`ed worker threads this + /// struct holds no `std.Thread` handle for, so they cannot be `join`ed + /// directly - `drainConnections` below waits on the `active` counter + /// instead. This used to be entirely the caller's problem ("leave them + /// to finish on their own, call `activeConnections()` first if you need + /// to wait"), but every real caller in this codebase - every test in + /// this file included - called `stop()` and then immediately tore down + /// the `Db`/allocator the detached worker's `dispatch()` still reads + /// from, with nothing enforcing that the worker had actually noticed + /// its connection closed and returned first. Most of the time the + /// worker's blocking read already sees EOF and exits before the caller + /// gets around to teardown, so the race stayed invisible for a long + /// time; occasionally it does not, and a worker thread touches a freed + /// `Db`/`Allocator.VTable`, corrupting the heap in a way that only + /// surfaces later as an unrelated test's leak/double-free abort - this + /// is exactly the flaky `zig build test` failure the gate re-run in + /// this session caught (roughly 1 run in 4-6, always attributed to + /// whichever unrelated test happened to run next). Waiting here by + /// default, bounded so a genuinely long-lived idle connection (still + /// blocked in `stream.reader()` with nothing sent and no EOF) cannot + /// hang shutdown forever, closes that gap for every caller instead of + /// relying on each one to remember `activeConnections()`. + pub fn stop(self: *Server) void { + if (self.acceptThread == null) return; + self.stopping.store(true, .release); + const listenAddr = self.net.listen_address; + if (std.net.tcpConnectToAddress(listenAddr)) |s| { + s.close(); + } else |_| {} + self.acceptThread.?.join(); + self.acceptThread = null; + if (self.sweeperThread) |t| { + t.join(); + self.sweeperThread = null; + } + if (self.ttlSweeperThread) |t| { + t.join(); + self.ttlSweeperThread = null; + } + if (self.collTtlSweeperThread) |t| { + t.join(); + self.collTtlSweeperThread = null; + } + self.drainConnections(500 * std.time.ns_per_ms); + self.net.deinit(); + // Best-effort, same as `drainConnections` above: a subscriber + // connection that never disconnects (client vanished without + // closing its socket) is still `active` past the drain timeout, so + // its `Subscriber` is never `destroySubscriber`'d and leaks for the + // remaining life of the process - this `deinit` only frees `Hub`'s + // own channel map, which by then holds no dangling entries for any + // subscriber that WAS properly torn down. + self.pubsub.deinit(); + } + + /// Polls `activeConnections()` until it reaches zero or `timeoutNanos` + /// elapses, whichever comes first. See the doc comment on `stop()` for + /// why this exists and why it is bounded rather than unconditional. + fn drainConnections(self: *Server, timeoutNanos: u64) void { + const startedAt = std.time.nanoTimestamp(); + while (self.activeConnections() > 0) { + if (std.time.nanoTimestamp() - startedAt >= timeoutNanos) return; + std.time.sleep(1 * std.time.ns_per_ms); + } + } +}; + +fn testDb(allocator: Allocator, dir: []const u8) !Db { + return Db.open(allocator, dir, .{}); +} +