address feedback

tighten socket ownership (on error paths)

allow reaper to be disabled

Handle window where link is being destroyed, worker is still alive, and client
attempts to re-link.

src/Config.zig

@@ -880,9 +880,10 @@ pub fn maxConnections(self: *const Config) u16 {
     };
 }
 
-pub fn httpSessionTimeout(self: *const Config) u64 {
+// Null disables the reaper: sessions then only end on DELETE /session/{id}.
+pub fn httpSessionTimeout(self: *const Config) ?u64 {
     return switch (self.mode) {
-        .serve => |opts| @as(u64, opts.http_session_timeout) * 1000,
+        .serve => |opts| if (opts.http_session_timeout == 0) null else @as(u64, opts.http_session_timeout) * 1000,
         .mcp => 60_000, // 1 minute
         else => unreachable,
     };
@@ -1323,6 +1324,26 @@ test "Config: parseArgs --http-version" {
     }
 }
 
+test "Config: parseArgs --http-session-timeout" {
+    // parseArgs allocations live for the process; an arena stands in for main's.
+    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
+    defer arena.deinit();
+
+    {
+        const argv = [_][*:0]const u8{ "lightpanda", "serve" };
+        const proc_args: std.process.Args = .{ .vector = &argv };
+        const config = try parseArgs(arena.allocator(), proc_args);
+        try std.testing.expectEqual(60_000, config.httpSessionTimeout());
+    }
+    {
+        // 0 disables the reaper
+        const argv = [_][*:0]const u8{ "lightpanda", "serve", "--http-session-timeout", "0" };
+        const proc_args: std.process.Args = .{ .vector = &argv };
+        const config = try parseArgs(arena.allocator(), proc_args);
+        try std.testing.expectEqual(null, config.httpSessionTimeout());
+    }
+}
+
 test "Config: validateUserAgent" {
     try validateUserAgent("Lightpanda/1.0");
     try std.testing.expectError(error.Reserved, validateUserAgent("mozilla/1.0"));

src/help.zon

@@ -48,7 +48,8 @@
     \\  --http-session-timeout <INT>
     \\          Seconds before an idle HTTP session times-out. Only meaningful
     \\          when connecting using the WebDriver protocol.
-    \\          Defaults to 60.
+    \\          Defaults to 60, disable by setting to 0 (a session then lives
+    \\          until the driver deletes it).
     \\  --port <INT>
     \\          Port of the CDP server.
     \\          Defaults to 9222.

src/server/Link.zig

@@ -66,6 +66,9 @@ pub fn init(
     protocol: Driver.Protocol,
     inbox: *Inbox,
 ) !void {
+    // The Link owns the socket from here on
+    errdefer sys_net.close(socket);
+
     if (lp.IS_TEST == false) {
         const socket_flags = try sys_net.fcntl(socket, posix.F.GETFL, 0);
         const nonblocking = @as(u32, @bitCast(posix.O{ .NONBLOCK = true }));
@@ -95,8 +98,13 @@ pub fn deinit(self: *Link) void {
 }
 
 pub fn create(app: *App, socket: posix.socket_t, protocol: Driver.Protocol, inbox: *Inbox) !*Link {
-    const link = try app.allocator.create(Link);
+    const link = app.allocator.create(Link) catch |err| {
+        sys_net.close(socket);
+        return err;
+    };
     errdefer app.allocator.destroy(link);
+
+    // init immediately takes ownership of the socket
     try link.init(app, socket, protocol, inbox);
     return link;
 }

src/server/Server.zig

@@ -87,8 +87,8 @@ sessions: std.AutoHashMapUnmanaged([36]u8, *Worker),
 // head read.
 idle_sessions: DoublyLinkedList,
 
-// --session-timeout, see Worker.deadline
-session_timeout_ms: u64,
+// --http-session-timeout, see Worker.deadline. Null disables the reaper.
+session_timeout_ms: ?u64,
 
 // Worker communicates with the main loop through this queue, protected by the
 // mutex.
@@ -484,7 +484,6 @@ pub fn attachConnection(self: *Server, worker: *Worker, conn: *Connection) void
     lp.assert(worker.link == null, "Server.deliverLink held", .{});
     const link = Link.create(self.app, conn.socket, worker.protocol, &worker.inbox) catch |err| {
         log.err(.serve, "link create", .{ .err = err });
-        sys_net.close(conn.socket);
         return;
     };
 
@@ -640,8 +639,13 @@ fn markIdle(self: *Server, worker: *Worker, now: u64) void {
         // ending already (or never a HTTP session)
         return;
     }
+    const timeout = self.session_timeout_ms orelse {
+        // reaping disabled: the session lives until DELETE /session/{id}
+        return;
+    };
+
     lp.assert(worker.deadline == null, "Server.markIdle idle", .{});
-    worker.deadline = now + self.session_timeout_ms;
+    worker.deadline = now + timeout;
     self.idle_sessions.append(&worker.idle_node);
 }
 
@@ -1253,6 +1257,13 @@ pub const Worker = struct {
         session: [36]u8,
     };
 
+    pub fn linkDropping(self: *const Worker) bool {
+        // There's a window where the loop has stopped reading the link but the
+        // worker hasn't handed it back yet. A driver that tries to re-establish
+        // the link will get an error and will have to retry.
+        return self.monitored == false and self.link != null;
+    }
+
     // -- Worker thread from here down --
 
     // The origin travels as an argument: the loop owns session_id and may
@@ -1789,7 +1800,8 @@ test "server: HTTP session outlives its websocket" {
     }
 
     // The worker lets go of its link right after replying; the loop learns
-    // of it a moment later, and refuses a new one until then.
+    // of it a moment later, and asks for a retry (429, not the 409 of a
+    // session someone else is actually connected to) until then.
     var c = try createTestClient();
     defer c.deinit();
     var attempts: usize = 0;
@@ -1798,7 +1810,7 @@ test "server: HTTP session outlives its websocket" {
         if (std.mem.startsWith(u8, res, "HTTP/1.1 101 ")) {
             break;
         }
-        try testing.expect(std.mem.startsWith(u8, res, "HTTP/1.1 409 "));
+        try testing.expect(std.mem.startsWith(u8, res, "HTTP/1.1 429 "));
         try testing.expect(attempts < 100);
         c.deinit();
         c = try createTestClient();
@@ -1838,6 +1850,20 @@ test "server: HTTP session idle timeout" {
     }
 }
 
+test "server: HTTP session idle timeout disabled" {
+    const server = testing.test_cdp_server.?;
+    const original = server.session_timeout_ms;
+    defer server.session_timeout_ms = original;
+    // what --http-session-timeout 0 gives us
+    server.session_timeout_ms = null;
+
+    const session_id = try createHTTPSession("{\"capabilities\":{}}", false);
+
+    // never idle-listed, so nothing reaps it: it's still there to DELETE
+    lp.io.sleep(.fromMilliseconds(50), .awake) catch {};
+    try deleteHTTPSession(&session_id, true);
+}
+
 test "server: HTTP session ended before its worker attached" {
     // The mailbox is alive from spawn: a DELETE that lands while the worker
     // is still starting up is a plain push, drained on its first tick.

src/server/bidi/BiDi.zig

@@ -134,15 +134,15 @@ pub fn init(self: *BiDi, app: *App, inbox: *Inbox, origin: Origin) !void {
         .session_arena = std.heap.ArenaAllocator.init(allocator),
     };
 
-    try self.browser.init(app, .{});
-    errdefer self.browser.deinit();
-
     switch (origin) {
         .socket => |socket| self.link = try Link.create(app, socket, .bidi, inbox),
         .session => {},
     }
     errdefer if (self.link) |l| l.destroy();
 
+    try self.browser.init(app, .{});
+    errdefer self.browser.deinit();
+
     self.notification = try Notification.init(allocator);
     errdefer self.notification.deinit();
 
@@ -180,7 +180,9 @@ pub fn adoptLink(self: *BiDi, l: *Link) void {
     if (self.link != null) {
         // the loop only hands one over once it has seen the previous one
         // released (Server.Worker.link is null)
-        lp.assert(false, "BiDi.adoptLink held", .{});
+        if (comptime lp.IS_DEBUG) {
+            lp.assert(false, "BiDi.adoptLink held", .{});
+        }
         l.destroy();
         return;
     }
@@ -201,7 +203,13 @@ pub fn onLinkGone(self: *BiDi) bool {
 }
 
 fn releaseLink(self: *BiDi, worker: *Server.Worker) void {
-    const l = self.link orelse return;
+    const l = self.link orelse {
+        // the loop only tells us the link is gone while we hold it
+        if (comptime lp.IS_DEBUG) {
+            lp.assert(false, "BiDi.releaseLink empty", .{});
+        }
+        return;
+    };
     self.link = null;
     // blocks until the loop has stopped reading from it
     worker.releaseLink();

src/server/cdp/CDP.zig

@@ -111,10 +111,10 @@ pub fn init(self: *CDP, app: *App, socket: posix.socket_t, inbox: *Inbox) !void
         .streams = .{ .allocator = allocator },
     };
 
-    try self.browser.init(app, .{ .env = .{ .with_inspector = true } });
-    errdefer self.browser.deinit();
-
     try self.link.init(app, socket, .cdp, inbox);
+    errdefer self.link.deinit();
+
+    try self.browser.init(app, .{ .env = .{ .with_inspector = true } });
 }
 
 pub fn deinit(self: *CDP) void {

src/server/http.zig

@@ -440,6 +440,7 @@ const forbidden_host_response = errorResponse(403, "Host not allowed");
 const request_too_large_response = errorResponse(413, "Request too large");
 const not_found_response = errorResponse(404, "Not found");
 const session_connected_response = errorResponse(409, "Session already connected");
+const session_busy_response = errorResponse(429, "Session is releasing its previous connection");
 const method_not_allowed_response = errorResponse(405, "Method not allowed");
 const service_unavailable_response = errorResponse(503, "Too many connections");
 const internal_error_response = errorResponse(500, "Internal server error");
@@ -696,7 +697,7 @@ fn newSession(server: *Server, conn: *Connection, req: *Connection.Request) !Ser
     var session_id: [36]u8 = undefined;
     uuidv4(&session_id);
 
-    _ = server.spawnWorker(.bidi, .{ .session = session_id }) catch |err| {
+    const worker = server.spawnWorker(.bidi, .{ .session = session_id }) catch |err| {
         log.err(.serve, "worker spawn", .{ .err = err });
         return serveWebDriver(server, conn, req, "500 Internal Server Error", .{
             .@"error" = "session not created",
@@ -704,6 +705,9 @@ fn newSession(server: *Server, conn: *Connection, req: *Connection.Request) !Ser
             .stacktrace = "",
         });
     };
+    // The client never learns the id if we fail to answer (e.g. it hung up),
+    // so nothing would ever DELETE this session.
+    errdefer server.quitSession(worker);
 
     const is_requesting_websocket_url = blk: {
         const caps = parsed.capabilities orelse break :blk false;
@@ -742,6 +746,12 @@ fn upgradeSession(server: *Server, conn: *Connection, req: *Connection.Request)
         return serveNotFound(server, conn, req);
     };
 
+    if (worker.linkDropping()) {
+        // The previous connection is gone but the worker hasn't given the
+        // link back yet. Dirver can retry.
+        return serveHTTPResponse(server, conn, req, .{ .static = session_busy_response });
+    }
+
     if (worker.link != null) {
         // already joined
         return serveHTTPResponse(server, conn, req, .{ .static = session_connected_response });