From 8ad9eaf48d5e8992ca6acc217b0016713c4a6d17 Mon Sep 17 00:00:00 2001 From: Karl Seguin Date: Fri, 4 Sep 2026 16:08:50 +0800 Subject: [PATCH 1/5] webdriver: HTTP WebDriver session management This is a small step towards WebDriver supports (non-bidi). It allows creating and deleting a BiDi "Session" (e.g. a worker). It also allows attaching a BiDi driver to an HTTP-created BiDi session (the typical selenium startup flow). This change unblocks the most basic setup/teardown of Selenium, so it still isn't enough to actually use a Selenium script as-is. But it's significant because it models a worker (thread) that isn't tied to a WebSocket, something we haven't had before. A consequence of a pure HTTP Session is that we don't have a clear cleanup signal. There is no "the socket is disconnected". There's a new HTTP reaper which kills HTTP Sessions after --http-session-timeout. It's expected that drivers properly DELETE /session/:id. I imagine we're going to run into --cdp-max-connections limits and need to tweak this code. BUT, this entire flow is only enabled with --protocol webdriver, so it won't impact exiting CDP users. --- src/Config.zig | 9 + src/Inbox.zig | 41 +- src/Metrics.zig | 6 +- src/help.zon | 4 + src/network/HttpClient.zig | 48 +- src/server/Driver.zig | 111 +++-- src/server/Link.zig | 23 +- src/server/Server.zig | 952 +++++++++++++++++++++++++----------- src/server/bidi/BiDi.zig | 94 +++- src/server/bidi/browser.zig | 2 +- src/server/bidi/session.zig | 4 +- src/server/bidi/testing.zig | 2 +- src/server/cdp/CDP.zig | 14 +- src/server/http.zig | 166 ++++--- 14 files changed, 1031 insertions(+), 445 deletions(-) diff --git a/src/Config.zig b/src/Config.zig index fc37bcbae..1994b2289 100644 --- a/src/Config.zig +++ b/src/Config.zig @@ -406,6 +406,7 @@ const Commands = cli.Builder(.{ .{ .name = "cdp_max_message_size", .type = u32, .default = 1024 * 1024 }, // Don't widen this without growing the reader buffer in the HTTP path. .{ .name = "cdp_max_http_message_size", .type = u14, .default = 4096 }, + .{ .name = "http_session_timeout", .type = u32, .default = 60 }, .{ .name = "disable_metrics", .type = bool }, }, .shared_options = CommonOptions, @@ -879,6 +880,14 @@ pub fn maxConnections(self: *const Config) u16 { }; } +pub fn httpSessionTimeout(self: *const Config) u64 { + return switch (self.mode) { + .serve => |opts| @as(u64, opts.http_session_timeout) * 1000, + .mcp => 60_000, // 1 minute + else => unreachable, + }; +} + pub fn maxPendingConnections(self: *const Config) u31 { return switch (self.mode) { .serve => |opts| opts.cdp_max_pending_connections, diff --git a/src/Inbox.zig b/src/Inbox.zig index 94c589eae..699e721c6 100644 --- a/src/Inbox.zig +++ b/src/Inbox.zig @@ -29,6 +29,7 @@ const std = @import("std"); const lp = @import("lightpanda"); const CDP = @import("server/cdp/CDP.zig"); +const Link = @import("server/Link.zig"); const DoublyLinkedList = std.DoublyLinkedList; @@ -46,7 +47,7 @@ pub fn deinit(self: *Inbox) void { defer self.mutex.unlock(lp.io); while (self.queue.popFirst()) |node| { const msg: *Message = @fieldParentPtr("node", node); - msg.deinit(); + msg.discard(); } self.queued_bytes = 0; } @@ -57,6 +58,12 @@ pub fn queuedBytes(self: *Inbox) usize { return self.queued_bytes; } +pub fn isEmpty(self: *Inbox) bool { + self.mutex.lockUncancelable(lp.io); + defer self.mutex.unlock(lp.io); + return self.queue.first == null; +} + pub fn push(self: *Inbox, arena: *lp.Arena, payload: Message.Payload) void { const msg = arena.create(Message) catch |err| switch (err) { error.OutOfMemory => @panic("OOM"), @@ -137,24 +144,35 @@ pub const Message = struct { // expected to echo via pong on its thread. ping: []u8, - // A close frame was received from the peer, or the worker decided - // to close (BiDi's session.end). Consumer is expected to send the - // close frame and tear the connection down. A peer's close body is - // dropped — we always send CLOSE_NORMAL (status 1000) regardless of - // what the peer sent. + // A close frame was received from the peer. Consumer is expected to + // send the close frame and tear the connection down. This may or may + // not kill the worker (up to the driver, CDP: always yes, WebDriver: + // depends) close: void, + // The Session is over. Currently WebDriver only. Always kills the worker. + // This is because for WebDriver, the Worker isn't necessarily tied to + // a WebSocket connection, so only an explicit DELETE /session/:id (or + // the HTTP reaper) can kill it. tl;dr an explicit "close" needed for + // WebDriver since the implicit socket-is-gone (aka .close) is ambiguous + // for WebDriver. + quit: void, + // No allocation; conveys "no more messages will arrive on // this inbox" plus an optional reason. The Network thread // pushes this on peer EOF, fatal WS framing error, or // (now) JSON parse failure. disconnect: ?anyerror, + // A websocket for the consumer to adopt (an HTTP WebDriver session + // gets its BiDi connection after the fact). + link: *Link, + pub fn size(self: Payload) usize { return switch (self) { .cdp => |c| c.raw.len, .bidi, .ping => |b| b.len, - .close, .disconnect => 0, + .close, .disconnect, .link, .quit => 0, }; } }; @@ -167,6 +185,15 @@ pub const Message = struct { pub fn deinit(self: *const Message) void { self.arena.release(); } + + // For messages that never reached the consumer (Inbox.deinit). + fn discard(self: *const Message) void { + switch (self.payload) { + .link => |link| link.destroy(), + else => {}, + } + self.deinit(); + } }; const testing = @import("testing.zig"); diff --git a/src/Metrics.zig b/src/Metrics.zig index f469e3640..55f0942d3 100644 --- a/src/Metrics.zig +++ b/src/Metrics.zig @@ -24,6 +24,7 @@ const Driver = @import("server/Driver.zig").Protocol; serve_http_requests: CounterEnum("status", @import("network/http.zig").StatusCategory) = .{}, serve_http_evictions: Counter = .{}, +serve_session_timeouts: Counter = .{}, serve_inbox_backlog: Counter = .{}, serve_connections: CounterEnum("driver", Driver) = .{}, serve_connection_limit: Counter = .{}, @@ -102,10 +103,11 @@ adblock_rules: GaugeEnum("state", enum { loaded, skipped, cosmetic }) = .{}, const help = .{ .serve_http_requests = "HTTP responses sent, by status category (includes the pre-parse 400/413 rejections)", .serve_http_evictions = "HTTP connections closed for sitting past their deadline without completing a request", + .serve_session_timeouts = "WebDriver sessions timed out", .serve_inbox_backlog = "Websocket connections closed for queueing more unprocessed messages than the worker could drain", - .serve_connections = "Websocket connections accepted, by driver protocol", + .serve_connections = "Drivers started, by protocol", .serve_connection_limit = "Accepts deferred because the connection budget was full: the listener pauses until a slot frees (counted before any handshake, so no driver label)", - .serve_active_connections = "Currently connected clients, by driver protocol", + .serve_active_connections = "Drivers currently running, by protocol", .serve_commands = "Commands dispatched, by driver protocol", .serve_unknown_commands = "Commands rejected for an unknown domain, module or method, by driver protocol", .js_heap_limits = "Pages terminated for reaching the V8 heap limit", diff --git a/src/help.zon b/src/help.zon index 48a43b012..97a20c6c2 100644 --- a/src/help.zon +++ b/src/help.zon @@ -45,6 +45,10 @@ \\ --host \\ Host of the CDP server. \\ Defaults to "127.0.0.1". + \\ --http-session-timeout + \\ Seconds before an idle HTTP session times-out. Only meaningful + \\ when connecting using the WebDriver protocol. + \\ Defaults to 60. \\ --port \\ Port of the CDP server. \\ Defaults to 9222. diff --git a/src/network/HttpClient.zig b/src/network/HttpClient.zig index a9533dd8a..fb61c5bc8 100644 --- a/src/network/HttpClient.zig +++ b/src/network/HttpClient.zig @@ -1499,23 +1499,33 @@ fn drainInbox(self: *Client, mode: DrainMode) !void { defer msg.deinit(); - switch (msg.payload) { - .cdp, .bidi => driver.onMessage(msg) catch |err| { - // A single malformed/failed dispatch shouldn't poison - // the rest of the batch — log and continue. - log.err(.app, "client dispatch", .{ .err = err }); + const done = switch (msg.payload) { + .cdp, .bidi => blk: { + driver.onMessage(msg) catch |err| { + // A single malformed/failed dispatch shouldn't poison + // the rest of the batch — log and continue. + log.err(.app, "client dispatch", .{ .err = err }); + }; + break :blk false; }, - .ping => |body| driver.onPing(body), - .close => { - driver.onClose(); - self.disconnected = true; - return error.ClientDisconnected; + .ping => |body| blk: { + driver.onPing(body); + break :blk false; }, - .disconnect => |err| { - driver.onDisconnect(err); - self.disconnected = true; - return error.ClientDisconnected; + .link => |link| blk: { + driver.onLink(link); + break :blk false; }, + .quit => blk: { + driver.onQuit(); + break :blk true; // quit always shutsdown + }, + .close => driver.onClose(), // close is up to the driver if it shutsdown + .disconnect => |err| driver.onDisconnect(err), // same with disconnect + }; + if (done) { + self.disconnected = true; + return error.ClientDisconnected; } } } @@ -1534,7 +1544,7 @@ fn drainInbox(self: *Client, mode: DrainMode) !void { // eval frame above us will dereference. fn allowDuringSyncWait(msg: *Inbox.Message) bool { return switch (msg.payload) { - .ping, .close, .disconnect => true, + .ping, .close, .disconnect, .quit, .link => true, .cdp => |c| isFetchInterceptionMethod(c.input.method), // BiDi has no request interception yet, so nothing it can send is // safe to dispatch from inside a JS callback. @@ -1544,8 +1554,8 @@ fn allowDuringSyncWait(msg: *Inbox.Message) bool { fn isTerminal(msg: *Inbox.Message) bool { return switch (msg.payload) { - .close, .disconnect => true, - .ping, .cdp, .bidi => false, + .close, .disconnect, .quit => true, + .ping, .cdp, .bidi, .link => false, }; } @@ -1562,8 +1572,8 @@ fn isFetchInterceptionMethod(method: []const u8) bool { // teardown command sits undispatched behind the sync_wait allowlist. fn isSyncWaitInterrupt(msg: *Inbox.Message) bool { return switch (msg.payload) { - .close, .disconnect => true, - .ping => false, + .close, .disconnect, .quit => true, + .ping, .link => false, .cdp => |c| isTeardownMethod(c.input.method), // Frames aren't parsed on the Network thread for BiDi, so we // can't spot a teardown command without re-parsing here. diff --git a/src/server/Driver.zig b/src/server/Driver.zig index 37d8b8fbc..91265569f 100644 --- a/src/server/Driver.zig +++ b/src/server/Driver.zig @@ -45,8 +45,6 @@ const Impl = union(Protocol) { impl: Impl, -// every implementation has this -conn: *Link, browser: *Browser, // The worker's mailbox, owned by the loop's connection slot (it outlives @@ -60,51 +58,42 @@ pub fn init(impl: Impl, inbox: *Inbox) Driver { return switch (impl) { inline else => |d, tag| .{ .impl = impl, - .conn = &d.conn, - .browser = &d.browser, .inbox = inbox, + .browser = &d.browser, // browser will still be undefined at this point, but its address is known .scope = @field(log.Scope, @tagName(tag)), // The tag names line up with the log scopes of the same name. }, }; } -// server loop. The socket is readable, drain up to budget bytes -pub fn onReadable(self: *const Driver, budget: usize) anyerror!bool { - const read = try self.conn.readAvailable(budget); - if (read.pushed) { - self.wakeup(); - } - return read.keep; -} - -// server loop. Called when it drops the link unsolicited (peer EOF, ...) -pub fn onLinkDisconnect(self: *const Driver, err: ?anyerror) void { - const arena = self.browser.arena_pool.acquire(.tiny, "driver disconnect") catch |e| switch (e) { - error.OutOfMemory => @panic("OOM"), +// server loop. Whether losing the link ends the worker. +pub fn connectionScoped(self: *const Driver) bool { + return switch (self.impl) { + .cdp => true, // always true for CDP; CDP is WebSocket only + .bidi => |bidi| bidi.mode == .bidi_only, // depends if this is BiDi-only WebDriver session }; - // order matters, this ensures that the disconnect message is in the inbox - // when tick() discovers the terminatePending flag is set. - self.inbox.push(arena, .{ .disconnect = err }); - self.browser.env.requestTerminate(); - self.wakeup(); } -// server loop. We used to send a nice WS close frame here but (a) it isn't strictly -// required and (b) we'd have to protect against an interleaved write from -// the worker thread. +// server loop. The loop shuts the link's read side itself (Server.Worker +// owns that pointer); this only stops the JS. pub fn shutdown(self: *const Driver) void { self.browser.env.terminate(); - self.conn.shutdown(); } -// a server-processed call (onReadable, onLinkDisconnect) wants to signal the -// worker that there's data in its inbox waiting to be processed. -fn wakeup(self: *const Driver) void { +// server loop. Something was pushed to the inbox; wake the worker from its poll. +pub fn wakeup(self: *const Driver) void { self.browser.http_client.handles.wakeup() catch |err| { log.err(self.scope, "wakeup", .{ .err = err }); }; } +// Worker thread. Note that (for bidi at least) the link can come and go +fn link(self: *const Driver) ?*Link { + return switch (self.impl) { + .cdp => |cdp| &cdp.link, + .bidi => |bidi| bidi.link, + }; +} + // Worker thread. We're processing messages from the inbox. pub fn onMessage(self: *const Driver, msg: *Inbox.Message) anyerror!void { return switch (self.impl) { @@ -115,27 +104,61 @@ pub fn onMessage(self: *const Driver, msg: *Inbox.Message) anyerror!void { // Worker Thread. We're processing messages from the inbox. pub fn onPing(self: *const Driver, body: []const u8) void { - self.conn.sendPong(body) catch |err| { + const l = self.link() orelse return; + l.sendPong(body) catch |err| { log.warn(self.scope, "pong", .{ .err = err }); }; } -// Worker Thread. We're processing messages from the inbox. -pub fn onClose(self: *const Driver) void { - self.conn.send(&WS.CLOSE_NORMAL) catch |err| { - log.warn(self.scope, "close reply", .{ .err = err }); - }; - self.onDisconnect(null); +// Worker Thread. The worker is being given a link +pub fn onLink(self: *const Driver, l: *Link) void { + switch (self.impl) { + .bidi => |bidi| bidi.adoptLink(l), + .cdp => { + // a CDP worker is born with its link and never offered another + log.err(self.scope, "unexpected link", .{}); + l.destroy(); + }, + } } -// Worker Thread. We're processing messages from the inbox. -pub fn onDisconnect(self: *const Driver, err: ?anyerror) void { +// Worker Thread. The websocket is closing. Should we kill the worker? That's +// up to the implementation (hint: for CDP, it's always "yes" and for WebDriver +// it's "yes" for a BiDi-only session) +pub fn onClose(self: *const Driver) bool { + if (self.link()) |l| { + l.send(&WS.CLOSE_NORMAL) catch |err| { + log.warn(self.scope, "close reply", .{ .err = err }); + }; + } + return self.onDisconnect(null); +} + +// Worker Thread. Unlike onClose, this is an unconditional termination. +// (Currently only comes from WebDriver endpoints (HTTP or WS)) +pub fn onQuit(self: *const Driver) void { + if (self.link()) |l| { + l.send(&WS.CLOSE_NORMAL) catch |err| { + log.warn(self.scope, "quit close", .{ .err = err }); + }; + } + log.info(self.scope, "session ended", .{}); +} + +// Worker Thread. Returns true when the worker is done. +pub fn onDisconnect(self: *const Driver, err: ?anyerror) bool { if (err) |e| { if (WS.errorReply(e)) |close_frame| { - self.conn.send(close_frame) catch {}; + if (self.link()) |l| { + l.send(close_frame) catch {}; + } } } log.info(self.scope, "disconnect", .{ .err = err }); + return switch (self.impl) { + .cdp => true, + .bidi => |bidi| bidi.onLinkGone(), + }; } // Worker thread. @@ -166,7 +189,7 @@ pub fn detach(self: *const Driver) void { // One iteration of the worker loop. Returns false to disconnect. fn tick(self: *const Driver) !bool { if (self.browser.env.terminatePending()) { - // Our own requestTerminate from onLinkDisconnect: the peer is gone or + // Our own requestTerminate from Server.dropWebSocket: the peer is gone or // sent garbage. Report it with its own close code, nothing to warn // about. Pops close/disconnect only: nothing else may be dispatched // in a shutting-down state. @@ -179,9 +202,11 @@ fn tick(self: *const Driver) !bool { log.warn(self.scope, "closing connection", .{ .reason = "pending terminate" }); // The worker thread is the sole writer of this socket, so sending // the close frame here can't interleave with another write. - self.conn.send(&WS.CLOSE_GOING_AWAY) catch |err| { - log.warn(self.scope, "terminate close", .{ .err = err }); - }; + if (self.link()) |l| { + l.send(&WS.CLOSE_GOING_AWAY) catch |err| { + log.warn(self.scope, "terminate close", .{ .err = err }); + }; + } return false; } diff --git a/src/server/Link.zig b/src/server/Link.zig index 014591f81..0df00b4a1 100644 --- a/src/server/Link.zig +++ b/src/server/Link.zig @@ -29,9 +29,10 @@ const CDP = @import("cdp/CDP.zig"); const Driver = @import("Driver.zig"); const posix = std.posix; +const Allocator = std.mem.Allocator; const ArenaAllocator = std.heap.ArenaAllocator; -// The worker's end of an upgraded connection (the loop's is Server.WebSocket). +// The worker's end of an upgraded connection (the loop's is Server.Worker). // Reads/framing happen on the server run loop (readAvailable → inbox); the worker // thread is the sole writer (send*). The two sides touch disjoint state // (reader+inbox vs send_arena+socket write) so no lock is needed beyond the @@ -49,6 +50,7 @@ const SEND_TIMEOUT_MS = 5_000; const INBOX_BACKLOG_MESSAGES = 32; inbox: *Inbox, +allocator: Allocator, arena_pool: *ArenaPool, socket: posix.socket_t, protocol: Driver.Protocol, @@ -77,6 +79,7 @@ pub fn init( .inbox = inbox, .socket = socket, .protocol = protocol, + .allocator = allocator, .arena_pool = &app.arena_pool, .reader = try .init(allocator, config.cdpMaxMessageSize()), .send_arena = ArenaAllocator.init(allocator), @@ -88,6 +91,20 @@ pub fn init( pub fn deinit(self: *Link) void { self.reader.deinit(); self.send_arena.deinit(); + sys_net.close(self.socket); +} + +pub fn create(app: *App, socket: posix.socket_t, protocol: Driver.Protocol, inbox: *Inbox) !*Link { + const link = try app.allocator.create(Link); + errdefer app.allocator.destroy(link); + try link.init(app, socket, protocol, inbox); + return link; +} + +pub fn destroy(self: *Link) void { + const allocator = self.allocator; + self.deinit(); + allocator.destroy(self); } pub fn send(self: *Link, data: []const u8) !void { @@ -299,8 +316,8 @@ test "link: send gives up when the peer stops reading" { if (std.c.socketpair(posix.AF.LOCAL, posix.SOCK.STREAM, 0, &pair) != 0) { return error.SocketPairFailed; } + // pair[1] is the link's, closed by its deinit defer sys_net.close(pair[0]); - defer sys_net.close(pair[1]); const small = std.mem.toBytes(@as(c_int, 4096)); try posix.setsockopt(pair[0], posix.SOL.SOCKET, posix.SO.RCVBUF, &small); @@ -340,8 +357,8 @@ test "link: stops reading once the worker's inbox backs up" { if (std.c.socketpair(posix.AF.LOCAL, posix.SOCK.STREAM, 0, &pair) != 0) { return error.SocketPairFailed; } + // pair[1] is the link's, closed by its deinit defer sys_net.close(pair[0]); - defer sys_net.close(pair[1]); const nonblocking = @as(u32, @bitCast(posix.O{ .NONBLOCK = true })); const flags = try sys_net.fcntl(pair[1], posix.F.GETFL, 0); diff --git a/src/server/Server.zig b/src/server/Server.zig index 6b0a7b967..506178a7e 100644 --- a/src/server/Server.zig +++ b/src/server/Server.zig @@ -29,6 +29,7 @@ const BiDi = @import("bidi/BiDi.zig"); const WS = @import("WS.zig"); const http = @import("http.zig"); +const Link = @import("Link.zig"); const Driver = @import("Driver.zig"); const Inbox = @import("../Inbox.zig"); @@ -54,66 +55,6 @@ const FD_HEADROOM = 128; // rather than dozens, without starving the other connections. const WS_READ_BUDGET = 256 * 1024; -// A websocket connection, this loop's side of Link.zig (the worker's side) -const WebSocket = struct { - socket: posix.socket_t, - address: sys_net.IpAddress, - // threads `websockets` while live, the pool's free list otherwise - node: DoublyLinkedList.Node, - protocol: Driver.Protocol, - // The worker's mailbox, which is how the main loop communicates with the worker. - inbox: Inbox = .{}, - // null until the worker attaches - driver: ?Driver = null, - // whether or not socket is in the poll set. Makes sure we don't double-remove - monitored: bool = false, - - const Pool = struct { - slab: []WebSocket, - free: DoublyLinkedList, - live: usize, // acquired and not yet released - - fn init(allocator: Allocator, capacity: usize) !Pool { - const slab = try allocator.alloc(WebSocket, capacity); - var free: DoublyLinkedList = .{}; - for (slab) |*ws| { - ws.node = .{}; - free.append(&ws.node); - } - return .{ .slab = slab, .free = free, .live = 0 }; - } - - fn deinit(self: *Pool, allocator: Allocator) void { - allocator.free(self.slab); - } - - fn acquire(self: *Pool) !*WebSocket { - const node = self.free.popFirst() orelse return error.NoWebSocketSlot; - self.live += 1; - return @fieldParentPtr("node", node); - } - - pub fn isFull(self: *const Pool) bool { - return self.live == self.slab.len; - } - - fn release(self: *Pool, ws: *WebSocket) void { - self.live -= 1; - ws.node = .{}; - self.free.append(&ws.node); - } - }; -}; - -// Worker -> loop request, see worker_queue. -const WorkerRequest = struct { - ws: *WebSocket, - op: union(enum) { - attach: Driver, - release: *std.Io.Event, - }, -}; - app: *App, io_engine: IOEngine, listener: posix.socket_t, @@ -132,9 +73,22 @@ protocols: Protocols, http_connections: DoublyLinkedList, http_connection_pool: Connection.Pool, -// Websocket connections, attached or not -websockets: DoublyLinkedList, -websocket_pool: WebSocket.Pool, +// Live workers, attached or not +workers: DoublyLinkedList, +worker_pool: Worker.Pool, + +// HTTP sessions by id, what /session/{id} resolves. Sized to the pool at +// init, so no allocation per request. +sessions: std.AutoHashMapUnmanaged([36]u8, *Worker), + +// HTTP sessions with no link, ordered by deadline: every deadline is +// now + session_timeout_ms, so appending keeps the order (the same trick +// as http_connections), reaping pops the head and the wait timeout is a +// head read. +idle_sessions: DoublyLinkedList, + +// --session-timeout, see Worker.deadline +session_timeout_ms: u64, // Worker communicates with the main loop through this queue, protected by the // mutex. @@ -148,16 +102,21 @@ worker_drain: std.ArrayList(WorkerRequest), shutdown_begun: bool, // Will block on this until all workers are shutdown -workers: lp.WaitGroup, +worker_wg: lp.WaitGroup, // Dynamic responses (/metrics, ...) are built here. If they can't be written // immediately, it will be copied to the Connection's pending scratch: std.Io.Writer.Allocating, -json_version_response: []const u8, +// Request-scoped allocations (parsed bodies, ...), handed to handlers as +// Request.arena and reset once the request is answered. +request_arena: std.heap.ArenaAllocator, + // ws://host:port/session/ — what POST /session advertises, the id goes on the end bidi_session_url: []const u8, +json_version_response: []const u8, + pub fn init(app: *App, address: sys_net.IpAddress) !*Server { const config = app.config; const allocator = app.allocator; @@ -226,8 +185,15 @@ pub fn init(app: *App, address: sys_net.IpAddress) !*Server { var worker_drain: std.ArrayList(WorkerRequest) = try .initCapacity(allocator, request_capacity); errdefer worker_drain.deinit(allocator); - var websocket_pool = try WebSocket.Pool.init(allocator, config.maxConnections()); - errdefer websocket_pool.deinit(allocator); + var worker_pool = try Worker.Pool.init(allocator, config.maxConnections()); + errdefer worker_pool.deinit(allocator); + + var sessions: std.AutoHashMapUnmanaged([36]u8, *Worker) = .empty; + if ((comptime lp.IS_TEST) or protocols.webdriver) { + // only used for http sessions + try sessions.ensureTotalCapacity(allocator, config.maxConnections()); + } + errdefer sessions.deinit(allocator); const self = try allocator.create(Server); errdefer allocator.destroy(self); @@ -238,19 +204,23 @@ pub fn init(app: *App, address: sys_net.IpAddress) !*Server { .listener = listener, .listener_paused = false, .scratch = scratch, + .request_arena = std.heap.ArenaAllocator.init(allocator), .protocols = protocols, .http_connections = .{}, .http_connection_pool = http_connection_pool, .json_version_response = json_version_response, .bidi_session_url = bidi_session_url, .max_connections = max_connections, - .websockets = .{}, - .websocket_pool = websocket_pool, + .workers = .{}, + .worker_pool = worker_pool, + .sessions = sessions, + .idle_sessions = .{}, + .session_timeout_ms = config.httpSessionTimeout(), .worker_mutex = .init, .worker_queue = worker_queue, .worker_drain = worker_drain, .shutdown_begun = false, - .workers = .{}, + .worker_wg = .{}, }; return self; } @@ -258,14 +228,16 @@ pub fn init(app: *App, address: sys_net.IpAddress) !*Server { pub fn deinit(self: *Server) void { const allocator = self.app.allocator; - self.workers.wait(); - lp.assert(self.websockets.first == null, "Server.deinit websockets", .{}); + self.worker_wg.wait(); + lp.assert(self.workers.first == null, "Server.deinit workers", .{}); while (self.http_connections.first) |node| { http.disconnect(self, @fieldParentPtr("node", node)); } self.scratch.deinit(); - self.websocket_pool.deinit(allocator); + self.request_arena.deinit(); + self.worker_pool.deinit(allocator); + self.sessions.deinit(allocator); self.worker_queue.deinit(allocator); self.worker_drain.deinit(allocator); self.http_connection_pool.deinit(); @@ -292,14 +264,9 @@ pub fn run(self: *Server) void { } fn runOnce(self: *Server) bool { - const deadline = blk: { - // self.http_connections is ordered by deadline - const node = self.http_connections.first orelse break :blk null; - const conn: *Connection = @fieldParentPtr("node", node); - break :blk conn.deadline -| lp.datetime.milliTimestamp(.boot); - }; + const timeout: ?u64 = if (self.nextDeadline()) |deadline| deadline -| lp.datetime.milliTimestamp(.boot) else null; - var events = self.io_engine.wait(deadline); + var events = self.io_engine.wait(timeout); const now = lp.datetime.milliTimestamp(.boot); var pending_accept = false; @@ -310,7 +277,7 @@ fn runOnce(self: *Server) bool { .accept => pending_accept = true, .read_write => |rw| switch (rw.target) { .http => |conn| http.processEvent(self, conn, rw, now), - .ws => |ws| self.processWebSocketEvent(ws, rw), + .worker => |worker| self.processWebSocketEvent(worker, rw), }, .signal => pending_signal = true, .shutdown => pending_shutdown = true, @@ -319,7 +286,7 @@ fn runOnce(self: *Server) bool { // signal first: a worker that released frees a slot the accept can use. if (pending_signal) { - self.drainWorkerQueue(); + self.drainWorkerQueue(now); } if (pending_accept) { self.accept(now) catch |err| log.err(.serve, "accept", .{ .err = err }); @@ -341,8 +308,9 @@ fn runOnce(self: *Server) bool { lp.metrics.serve_http_evictions.incr(); http.disconnect(self, conn); } + self.reapSessions(now); - if (self.shutdown_begun and self.websocket_pool.live == 0) { + if (self.shutdown_begun and self.worker_pool.live == 0) { return false; } return true; @@ -424,7 +392,7 @@ fn setSocketOption(socket: posix.socket_t, level: i32, option: u32, value: anyty } fn liveConnections(self: *const Server) usize { - return self.http_connection_pool.live + self.websocket_pool.live; + return self.http_connection_pool.live + self.worker_pool.live; } // We want to accept a connection, but have reached the connection limit. See @@ -451,32 +419,37 @@ fn saturated(self: *Server) !void { self.listener_paused = true; } -fn processWebSocketEvent(self: *Server, ws: *WebSocket, rw: IOEvent.ReadWrite) void { - if (ws.monitored == false) { - // only attachWorker puts a websocket in the poll set, and only once - // the driver is set; an unmonitored slot has no business here. +fn processWebSocketEvent(self: *Server, worker: *Worker, rw: IOEvent.ReadWrite) void { + if (worker.monitored == false) { + // only monitorLink puts a websocket in the poll set; an unmonitored + // slot has no business here. return; } - const driver = ws.driver orelse { - // the socket is only monitered after an attach, which sets the driver - lp.assert(false, "Server.processWebSocketEvent driver", .{}); + const link = worker.link orelse { + lp.assert(false, "Server.processWebSocketEvent link", .{}); unreachable; }; if (rw.readable) { - const keep = driver.onReadable(WS_READ_BUDGET) catch |err| switch (err) { - error.Closed => return self.dropWebSocket(ws, null, true), // peer EOF + const read = link.readAvailable(WS_READ_BUDGET) catch |err| switch (err) { + error.Closed => return self.dropWebSocket(worker, null, true), // peer EOF // read error or fatal framing error: the worker doesn't know, so notify - else => return self.dropWebSocket(ws, err, true), + else => return self.dropWebSocket(worker, err, true), }; - if (keep == false) { + if (read.pushed) { + // before the attach there's nobody to wake; the first tick drains + if (worker.driver) |driver| { + driver.wakeup(); + } + } + if (read.keep == false) { // Close frame consumed: the framer already pushed .close, the - // worker will reply and disconnect itself. - return self.dropWebSocket(ws, null, false); + // worker will reply and let go of the link itself. + return self.dropWebSocket(worker, null, false); } } else if (rw.hangup) { - return self.dropWebSocket(ws, null, true); + return self.dropWebSocket(worker, null, true); } } @@ -493,58 +466,131 @@ pub fn slotFreed(self: *Server) void { } } -// The 101 has been written: take the fd off the http Connection (the http -// side recycles it) into a websocket slot/ -pub fn upgradeConnection(self: *Server, conn: *Connection, protocol: Driver.Protocol, session_id: ?[36]u8) void { - // it'll get added back once the Worker is started and able to process messages - self.io_engine.remove(conn.socket); - self.http_connections.remove(&conn.node); +// An HTTP connection is being upgraded to a WebSocket connection for use with +// a new Worker. CDP goes this route as do some WebDriver libraries. +pub fn upgradeConnection(self: *Server, conn: *Connection, protocol: Driver.Protocol) void { + self.detachConnection(conn); // removes from the loop, will get re-added + _ = self.spawnWorker(protocol, .{ .socket = conn.socket }) catch |err| { + log.err(.serve, "worker spawn", .{ .err = err }); + sys_net.close(conn.socket); + }; +} - const ws = self.websocket_pool.acquire() catch |err| { - if (comptime lp.IS_DEBUG) { - // should not be reachable. In the HTTP upgrade processing, we - // checked isFull() - unreachable; - } +// An HTTP connection is being upgraded to a WebSocket connection for use with +// a *EXISTING* Worker. Some WebDrivers (e.g. Selenium) go this route. +pub fn attachConnection(self: *Server, worker: *Worker, conn: *Connection) void { + self.detachConnection(conn); // removes from the loop, will get re-added - // but, let's be safe.. - log.err(.serve, "websocket slot", .{ .err = err }); + 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; }; - ws.* = .{ - .node = .{}, - .socket = conn.socket, - .address = conn.address, - .protocol = protocol, + // precedes anything read off the socket + self.push(worker, .{ .link = link }); + worker.link = link; + self.clearIdle(worker); + self.monitorLink(worker); +} + +fn detachConnection(self: *Server, conn: *Connection) void { + self.io_engine.remove(conn.socket); + self.http_connections.remove(&conn.node); +} + +// Takes a slot and starts the thread. +pub fn spawnWorker(self: *Server, protocol: Driver.Protocol, origin: Worker.Origin) !*Worker { + const worker = self.worker_pool.acquire() catch |err| { + if (comptime lp.IS_DEBUG) { + // should not be reachable, callers check isFull() + unreachable; + } + // but, let's be safe.. + return err; }; - self.websockets.append(&ws.node); + + worker.* = .{ + .node = .{}, + .server = self, + .protocol = protocol, + .session_id = switch (origin) { + .socket => null, + .session => |id| id, + }, + }; + self.workers.append(&worker.node); + if (origin == .session) { + lp.assert(self.protocols.webdriver, "spawnWorker session without webdriver", .{}); + // because we sized self.sessions to config.maxConnections() + self.sessions.putAssumeCapacityNoClobber(origin.session, worker); + } lp.metrics.serve_connections.incr(protocol); lp.metrics.serve_active_connections.incr(protocol); - self.workers.start(); - const thread = std.Thread.spawn(.{}, Worker.start, .{ self, ws, session_id }) catch |err| { + self.worker_wg.start(); + const thread = std.Thread.spawn(.{}, Worker.start, .{ worker, origin }) catch |err| { // cleanup what we just did prior to spawning. - log.err(.serve, "worker spawn", .{ .err = err }); - self.workers.finish(); - sys_net.close(ws.socket); - self.releaseWebSocket(ws); - return; + self.worker_wg.finish(); + self.releaseWorkerSlot(worker); + return err; }; thread.detach(); + return worker; } -fn drainWorkerQueue(self: *Server) void { +// Find an HTTP session by ID +pub fn findSession(self: *Server, id: *const [36]u8) ?*Worker { + return self.sessions.get(id.*); +} + +// Ends an HTTP session: DELETE /session/{id}, or the idle reaper. +pub fn quitSession(self: *Server, worker: *Worker) void { + self.forgetSession(worker); + self.push(worker, .quit); + if (worker.driver) |driver| { + // the message is in the inbox before the flag is observed + driver.browser.env.requestTerminate(); + } else { + // What if there's no driver? It means it's still starting up or that + // we haven't processed it's attach message yet. Either way, we've + // queued the .quit message. Attach will send a .wakeup() so that it + // gets picked up promptly. + } +} + +// Into the worker's mailbox. +fn push(self: *Server, worker: *Worker, payload: Inbox.Message.Payload) void { + const arena = self.app.arena_pool.acquire(.tiny, "worker push") catch |err| switch (err) { + error.OutOfMemory => @panic("OOM"), + }; + worker.inbox.push(arena, payload); + if (worker.driver) |driver| { + driver.wakeup(); + } +} + +fn monitorLink(self: *Server, worker: *Worker) void { + self.io_engine.monitorWebSocket(worker) catch |err| { + log.err(.serve, "ws monitor", .{ .err = err }); + // never monitored, so this only tells the worker + return self.dropWebSocket(worker, err, true); + }; + worker.monitored = true; +} + +fn drainWorkerQueue(self: *Server, now: u64) void { self.worker_mutex.lockUncancelable(lp.io); std.mem.swap(std.ArrayList(WorkerRequest), &self.worker_queue, &self.worker_drain); self.worker_mutex.unlock(lp.io); for (self.worker_drain.items) |request| { switch (request.op) { - .attach => |driver| self.attachWorker(request.ws, driver), - .release => |notify| self.releaseWorker(request.ws, notify), + .attach => |attach| self.attachWorker(request.worker, attach.driver, attach.link, now), + .release_link => |notify| self.releaseLink(request.worker, notify, now), + .release => |notify| self.releaseWorker(request.worker, notify), } } self.worker_drain.clearRetainingCapacity(); @@ -553,61 +599,125 @@ fn drainWorkerQueue(self: *Server) void { // The Worker is spawned, the Driver is setup. It has signaled us that it's // ready to receive messages and given us the driver to associate to the // connection. -fn attachWorker(self: *Server, ws: *WebSocket, driver: Driver) void { +fn attachWorker(self: *Server, worker: *Worker, driver: Driver, link: ?*Link, now: u64) void { if (comptime lp.IS_DEBUG) { // a worker attaches exactly once - lp.assert(ws.driver == null, "Server.attachWorker attached", .{}); + lp.assert(worker.driver == null, "Server.attachWorker attached", .{}); } - ws.driver = driver; + worker.driver = driver; + + if (worker.inbox.isEmpty() == false) { + // we might have pushed message before we had the driver, so we weren't + // able to wakeup the worker. + driver.wakeup(); + } + + // The worker's own link (cdp, bidi-only); an HTTP session's may already + // be here, delivered while the worker was starting up. + if (link) |l| { + lp.assert(worker.link == null, "Server.attachWorker link", .{}); + worker.link = l; + self.monitorLink(worker); + } + if (self.shutdown_begun) { driver.shutdown(); + if (worker.link) |l| { + l.shutdown(); + } + } + + if (worker.link == null) { + // This is an HTTP session, we need to watch it and reap it if it + // stays idle too long. + self.markIdle(worker, now); } - self.io_engine.monitorWebSocket(ws) catch |err| { - log.err(.serve, "ws monitor", .{ .err = err }); - // never monitored, so this only tells the worker - return self.dropWebSocket(ws, err, true); - }; - ws.monitored = true; } -fn releaseWorker(self: *Server, ws: *WebSocket, notify: *std.Io.Event) void { - if (ws.monitored) { - ws.monitored = false; - self.io_engine.remove(ws.socket); +// A HTTP session without a link: the reaper's clock starts. +fn markIdle(self: *Server, worker: *Worker, now: u64) void { + if (worker.session_id == null) { + // ending already (or never a HTTP session) + return; } - self.releaseWebSocket(ws); + lp.assert(worker.deadline == null, "Server.markIdle idle", .{}); + worker.deadline = now + self.session_timeout_ms; + self.idle_sessions.append(&worker.idle_node); +} + +fn clearIdle(self: *Server, worker: *Worker) void { + if (worker.deadline != null) { + worker.deadline = null; + self.idle_sessions.remove(&worker.idle_node); + } +} + +// The session id stops resolving; the worker may still be running. +fn forgetSession(self: *Server, worker: *Worker) void { + self.clearIdle(worker); + if (worker.session_id) |id| { + worker.session_id = null; + _ = self.sessions.remove(id); + } +} + +fn releaseLink(self: *Server, worker: *Worker, notify: *std.Io.Event, now: u64) void { + self.unmonitorLink(worker); + worker.link = null; + self.markIdle(worker, now); + // The worker is free to destroy the link from here. + notify.set(lp.io); +} + +fn releaseWorker(self: *Server, worker: *Worker, notify: *std.Io.Event) void { + self.unmonitorLink(worker); + worker.link = null; + self.releaseWorkerSlot(worker); // The worker is free to deinit its driver and close the fd from here. notify.set(lp.io); } +fn unmonitorLink(self: *Server, worker: *Worker) void { + if (worker.monitored) { + worker.monitored = false; + self.io_engine.remove(worker.link.?.socket); + } +} + // Frees the slot once the loop is done with the fd. The fd itself is closed // by whoever owns the end of its life: the worker after its driver's deinit, -// or upgradeConnection when there never was a worker. -fn releaseWebSocket(self: *Server, ws: *WebSocket) void { - self.websockets.remove(&ws.node); - ws.driver = null; - ws.inbox.deinit(); - lp.metrics.serve_active_connections.decr(ws.protocol); - self.websocket_pool.release(ws); +// or the inbox's deinit for a link the worker never adopted. +fn releaseWorkerSlot(self: *Server, worker: *Worker) void { + // still registered when the worker ended on its own (session.end, a + // failed init) + self.forgetSession(worker); + self.workers.remove(&worker.node); + worker.driver = null; + worker.inbox.deinit(); + lp.metrics.serve_active_connections.decr(worker.protocol); + self.worker_pool.release(worker); self.slotFreed(); } // unlike close above, this stops the polling on the socket and, optionally, -// informs the Worker that it should shut down. Ultimately, when it does shutdown -// releaseWebSocket above will be called. -fn dropWebSocket(self: *Server, ws: *WebSocket, err: ?anyerror, notify: bool) void { - if (ws.monitored) { - // only turned on in attachWorker, so it'll never be turned on again - ws.monitored = false; - self.io_engine.remove(ws.socket); - } +// informs the Worker that it should let go of the link. Ultimately, when it +// does, releaseLink or releaseWorker above will be called. +fn dropWebSocket(self: *Server, worker: *Worker, err: ?anyerror, notify: bool) void { + // only turned on in monitorLink, so it'll never be turned on again + // until the worker has released this link + self.unmonitorLink(worker); if (notify) { // Some closes the drivers knows about, some it doesn't. But the driver // is always the final authority on cleanup, so we always inform it of // the close. - if (ws.driver) |driver| { - driver.onLinkDisconnect(err); + self.push(worker, .{ .disconnect = err }); + if (worker.driver) |driver| { + if (driver.connectionScoped()) { + // nobody is left to hear the result of whatever is running; + // the message is in the inbox before the flag is observed + driver.browser.env.requestTerminate(); + } } } } @@ -626,16 +736,49 @@ fn beginShutdown(self: *Server) void { http.disconnect(self, @fieldParentPtr("node", node)); } - var node = self.websockets.first; + var node = self.workers.first; while (node) |n| : (node = n.next) { - const ws: *WebSocket = @fieldParentPtr("node", n); + const worker: *Worker = @fieldParentPtr("node", n); // not attached yet: attachWorker terminates it on arrival - if (ws.driver) |driver| { - driver.shutdown(); + const driver = worker.driver orelse continue; + driver.shutdown(); + if (worker.link) |link| { + link.shutdown(); } } } +// HTTP sessions nobody has talked to for --session-timeout. +fn reapSessions(self: *Server, now: u64) void { + while (self.idle_sessions.first) |node| { + const worker: *Worker = @fieldParentPtr("idle_node", node); + if (worker.deadline.? > now) { + // ordered by deadline: none after this one is due either + return; + } + log.info(.serve, "session timeout", .{ .id = &worker.session_id.? }); + lp.metrics.serve_session_timeouts.incr(); + self.quitSession(worker); + } +} + +// The earliest of the http connections' idle deadline and the idle +// sessions' reap deadline; null when there's nothing to wait for. Both +// lists are ordered, so this is two head reads. +fn nextDeadline(self: *const Server) ?u64 { + var next: ?u64 = null; + if (self.http_connections.first) |node| { + const conn: *Connection = @fieldParentPtr("node", node); + next = conn.deadline; + } + if (self.idle_sessions.first) |node| { + const worker: *Worker = @fieldParentPtr("idle_node", node); + const deadline = worker.deadline.?; + next = @min(next orelse deadline, deadline); + } + return next; +} + fn fdBudget(config: *const Config) usize { const reserve: usize = @as(usize, config.httpMaxConcurrent()) + config.wsMaxConcurrent() + FD_HEADROOM; const soft: u64 = blk: { @@ -651,74 +794,6 @@ fn fdBudget(config: *const Config) usize { return @intCast(@max(budget, 8)); } -fn signal(self: *Server) void { - self.io_engine.signal(); -} - -// Stateless, but helps to group things that run on the Worker thread -const Worker = struct { - fn start(server: *Server, ws: *WebSocket, session_id: ?[36]u8) void { - defer server.workers.finish(); - Worker._start(server, ws, session_id) catch |err| { - log.err(.serve, "worker init", .{ .err = err }); - Worker.releaseConnection(server, ws); - }; - } - - fn _start(server: *Server, ws: *WebSocket, session_id: ?[36]u8) !void { - const allocator = server.app.allocator; - // The socket outlives the slot: the driver's deinit below still - // writes to it (inspector detach notifications), so it closes last, - // after the loop has released us. `ws` itself must not be touched - // after releaseConnection returns. - const socket = ws.socket; - defer sys_net.close(socket); - switch (ws.protocol) { - .cdp => { - const cdp = try allocator.create(CDP); - defer allocator.destroy(cdp); - try cdp.init(server.app, ws.socket, &ws.inbox); - defer cdp.deinit(); - Worker.run(server, ws, .init(.{ .cdp = cdp }, &ws.inbox)); - }, - .bidi => { - const bidi = try allocator.create(BiDi); - defer allocator.destroy(bidi); - try bidi.init(server.app, ws.socket, &ws.inbox, session_id); - defer bidi.deinit(); - Worker.run(server, ws, .init(.{ .bidi = bidi }, &ws.inbox)); - }, - } - } - - fn run(server: *Server, ws: *WebSocket, driver: Driver) void { - Worker.notifyLoopOfChange(server, .{ .ws = ws, .op = .{ .attach = driver } }); - driver.run(); - // Release first: until the loop has let go of this websocket it can - // still drop the link, and onLinkDisconnect requests a terminate. - Worker.releaseConnection(server, ws); - - // CDP/BiDi close the session before Browser.deinit, so disarm now. - driver.browser.prepareForTeardown(); - } - - // Worker -> loop: synchronous release. Blocks until the loop has dropped the - // fd and won't call feed() again, so the caller can safely deinit the driver - // (which frees the reader). - fn releaseConnection(server: *Server, ws: *WebSocket) void { - var notify: std.Io.Event = .unset; - Worker.notifyLoopOfChange(server, .{ .ws = ws, .op = .{ .release = ¬ify } }); - notify.waitUncancelable(lp.io); - } - - fn notifyLoopOfChange(server: *Server, request: WorkerRequest) void { - server.worker_mutex.lockUncancelable(lp.io); - server.worker_queue.appendAssumeCapacity(request); - server.worker_mutex.unlock(lp.io); - server.io_engine.signal(); - } -}; - const IOEngine = switch (builtin.os.tag) { .linux => EPoll, .macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => KQueue, @@ -734,7 +809,7 @@ pub const IOEvent = union(enum) { pub const ReadWrite = struct { target: union(enum) { - ws: *WebSocket, + worker: *Worker, http: *Connection, }, hangup: bool, @@ -824,7 +899,7 @@ const EPoll = struct { // surfaces as a write error instead. const WRITE_EVENTS = linux.EPOLL.OUT; - // Poll data carries the owner: an http Connection as-is, an WebSocket with + // Poll data carries the owner: an http Connection as-is, a Worker with // the low bit set (both are word-aligned, so the bit is free). const WS_TAG: usize = 1; @@ -836,12 +911,12 @@ const EPoll = struct { return sys_net.epoll_ctl(self.fd, linux.EPOLL.CTL_ADD, conn.socket, &event); } - fn monitorWebSocket(self: *const EPoll, ws: *WebSocket) !void { + fn monitorWebSocket(self: *const EPoll, worker: *Worker) !void { var event = linux.epoll_event{ - .data = .{ .ptr = @intFromPtr(ws) | WS_TAG }, + .data = .{ .ptr = @intFromPtr(worker) | WS_TAG }, .events = READ_EVENTS, }; - return sys_net.epoll_ctl(self.fd, linux.EPOLL.CTL_ADD, ws.socket, &event); + return sys_net.epoll_ctl(self.fd, linux.EPOLL.CTL_ADD, worker.link.?.socket, &event); } pub fn waitWritable(self: *const EPoll, conn: *Connection) !void { @@ -899,7 +974,7 @@ const EPoll = struct { .target = if (nptr & WS_TAG == 0) .{ .http = @ptrFromInt(nptr) } else - .{ .ws = @ptrFromInt(nptr & ~WS_TAG) }, + .{ .worker = @ptrFromInt(nptr & ~WS_TAG) }, .readable = flags & linux.EPOLL.IN != 0, .writable = flags & linux.EPOLL.OUT != 0, .hangup = flags & (linux.EPOLL.RDHUP | linux.EPOLL.HUP | linux.EPOLL.ERR) != 0, @@ -919,7 +994,7 @@ const KQueue = struct { const EVFILT = std.c.EVFILT; const Kevent = std.c.Kevent; - // Poll data carries the owner: an http Connection as-is, an WebSocket with + // Poll data carries the owner: an http Connection as-is, a Worker with // the low bit set (both are word-aligned, so the bit is free). const WS_TAG: usize = 1; @@ -975,8 +1050,8 @@ const KQueue = struct { return self.monitor(conn.socket, EVFILT.READ, @intFromPtr(conn)); } - fn monitorWebSocket(self: *const KQueue, ws: *WebSocket) !void { - return self.monitor(ws.socket, EVFILT.READ, @intFromPtr(ws) | WS_TAG); + fn monitorWebSocket(self: *const KQueue, worker: *Worker) !void { + return self.monitor(worker.link.?.socket, EVFILT.READ, @intFromPtr(worker) | WS_TAG); } // A socket only ever has one of the two filters registered, so flipping is @@ -1086,7 +1161,7 @@ const KQueue = struct { .target = if (nptr & WS_TAG == 0) .{ .http = @ptrFromInt(nptr) } else - .{ .ws = @ptrFromInt(nptr & ~WS_TAG) }, + .{ .worker = @ptrFromInt(nptr & ~WS_TAG) }, .readable = event.filter == EVFILT.READ, .writable = event.filter == EVFILT.WRITE, // EV_EOF on a read filter can still come with buffered @@ -1100,6 +1175,175 @@ const KQueue = struct { }; }; +// One thread driving a Browser. Once the thread spawns, this will get +// associated with a Driver (CDP or WebDriver) and optionally a Link which is +// the Worker's side of the WebSocket connection. A worker can be Linkless, in +// the case of an HTTP Session, though that Link could be attached at some point +// in the future. +pub const Worker = struct { + server: *Server, + // threads `workers` while live, the pool's free list otherwise + node: DoublyLinkedList.Node, + protocol: Driver.Protocol, + + // The worker's mailbox, how the loop talks to it. Alive from spawn, so + // the loop can push before the worker has attached (it drains on its + // first tick); deinit'd with the slot, after the worker released it. + inbox: Inbox = .{}, + + // null until the worker thread attaches + driver: ?Driver = null, + + // The WebSocket connection, null for HTTP sessions (which can be upgraded + // in which case the link is set) + link: ?*Link = null, + + // whether link's socket is in the poll set. Makes sure we don't double-remove + monitored: bool = false, + + // HTTP WebDriver session id, null for cdp and bidi-only workers. + session_id: ?[36]u8 = null, + + // A HTTP session without a link is reaped at this time, and threads + // `idle_sessions` until then. Null while a link is attached (TCP + // keepalive covers a dead peer then). + deadline: ?u64 = null, + idle_node: DoublyLinkedList.Node = .{}, + + const Pool = struct { + slab: []Worker, + free: DoublyLinkedList, + live: usize, // acquired and not yet released + + fn init(allocator: Allocator, capacity: usize) !Pool { + const slab = try allocator.alloc(Worker, capacity); + var free: DoublyLinkedList = .{}; + for (slab) |*worker| { + worker.node = .{}; + free.append(&worker.node); + } + return .{ .slab = slab, .free = free, .live = 0 }; + } + + fn deinit(self: *Pool, allocator: Allocator) void { + allocator.free(self.slab); + } + + fn acquire(self: *Pool) !*Worker { + const node = self.free.popFirst() orelse return error.NoWorkerSlot; + self.live += 1; + return @fieldParentPtr("node", node); + } + + pub fn isFull(self: *const Pool) bool { + return self.live == self.slab.len; + } + + fn release(self: *Pool, worker: *Worker) void { + self.live -= 1; + worker.node = .{}; + self.free.append(&worker.node); + } + }; + + // What a worker is born from: the upgraded socket, or an HTTP session + // whose websocket, if any, comes later (see attachConnection). + pub const Origin = union(enum) { + socket: posix.socket_t, + session: [36]u8, + }; + + // -- Worker thread from here down -- + + // The origin travels as an argument: the loop owns session_id and may + // clear it while the thread starts up. + fn start(self: *Worker, origin: Origin) void { + defer self.server.worker_wg.finish(); + self._start(origin) catch |err| { + log.err(.serve, "worker init", .{ .err = err }); + self.releaseConnection(); + }; + } + + fn _start(self: *Worker, origin: Origin) !void { + const server = self.server; + const allocator = server.app.allocator; + switch (self.protocol) { + .cdp => { + // only WebDriver has HTTP sessions + const socket = switch (origin) { + .socket => |socket| socket, + .session => return error.NoSocket, + }; + const cdp = try allocator.create(CDP); + defer allocator.destroy(cdp); + try cdp.init(server.app, socket, &self.inbox); + defer cdp.deinit(); + self.run(.init(.{ .cdp = cdp }, &self.inbox), &cdp.link); + }, + .bidi => { + const bidi = try allocator.create(BiDi); + defer allocator.destroy(bidi); + try bidi.init(server.app, &self.inbox, switch (origin) { + .socket => |socket| .{ .socket = socket }, + .session => |id| .{ .session = .{ .id = id, .worker = self } }, + }); + defer bidi.deinit(); + self.run(.init(.{ .bidi = bidi }, &self.inbox), bidi.link); + }, + } + } + + fn run(self: *Worker, driver: Driver, link: ?*Link) void { + self.notifyLoop(.{ .attach = .{ .driver = driver, .link = link } }); + driver.run(); + // Release first: until the loop has let go of this worker it can + // still drop the link, and dropWebSocket requests a terminate. + self.releaseConnection(); + + // CDP/BiDi close the session before Browser.deinit, so disarm now. + driver.browser.prepareForTeardown(); + } + + // Worker -> loop: synchronous release. Blocks until the loop has dropped the + // fd and won't read it again, so the caller can safely deinit the driver + // (which frees the link). + fn releaseConnection(self: *Worker) void { + var notify: std.Io.Event = .unset; + self.notifyLoop(.{ .release = ¬ify }); + notify.waitUncancelable(lp.io); + } + + // Worker -> loop: the worker is dropping its link but carrying on + // (a HTTP session whose BiDi client went away). Blocks until the + // loop has stopped reading from it, so the caller can destroy it. + pub fn releaseLink(self: *Worker) void { + var notify: std.Io.Event = .unset; + self.notifyLoop(.{ .release_link = ¬ify }); + notify.waitUncancelable(lp.io); + } + + fn notifyLoop(self: *Worker, op: WorkerRequest.Op) void { + const server = self.server; + server.worker_mutex.lockUncancelable(lp.io); + server.worker_queue.appendAssumeCapacity(.{ .worker = self, .op = op }); + server.worker_mutex.unlock(lp.io); + server.io_engine.signal(); + } +}; + +// Worker -> loop request, see worker_queue. +const WorkerRequest = struct { + op: Op, + worker: *Worker, + + const Op = union(enum) { + release: *std.Io.Event, + release_link: *std.Io.Event, + attach: struct { driver: Driver, link: ?*Link }, + }; +}; + const testing = @import("../testing.zig"); test "server: buildJSONVersionResponse" { const res = try http.buildJSONVersionResponse(testing.test_app, testing.test_app.config.port()); @@ -1468,43 +1712,16 @@ test "server: bidi browsingContext" { try assertBidiMessage(&c, .{ .type = "success", .id = 9, .result = .{ .contexts = .{} } }); } -test "server: classic session bootstrap" { - // What Selenium does before it speaks BiDi: a classic POST /session +test "server: HTTP session bootstrap" { + // What Selenium does before it speaks BiDi: a POST /session // that hands back the websocket URL, then a DELETE on quit. - const session_id = blk: { - var c = try createTestClient(); - defer c.deinit(); - - const body = "{\"capabilities\":{\"firstMatch\":[{}],\"alwaysMatch\":{\"browserName\":\"firefox\",\"webSocketUrl\":true}}}"; - const res = try c.httpRequest(std.fmt.comptimePrint("POST /session HTTP/1.1\r\n" ++ - "Content-Type: application/json;charset=UTF-8\r\n" ++ - "Content-Length: {d}\r\n\r\n" ++ - "{s}", .{ body.len, body })); - try testing.expect(std.mem.startsWith(u8, res, "HTTP/1.1 200 OK\r\n")); - - const json = res[std.mem.indexOf(u8, res, "\r\n\r\n").? + 4 ..]; - const parsed = try std.json.parseFromSlice(std.json.Value, testing.allocator, json, .{}); - defer parsed.deinit(); - - const value = parsed.value.object.get("value").?.object; - const id = value.get("sessionId").?.string; - try testing.expectEqual(36, id.len); - - const capabilities = value.get("capabilities").?.object; - try testing.expectEqual("Lightpanda", capabilities.get("browserName").?.string); - try testing.expectEqual(false, capabilities.get("acceptInsecureCerts").?.bool); - const ws_url = capabilities.get("webSocketUrl").?.string; - try testing.expectEqual("ws://127.0.0.1:9583/session/", ws_url[0 .. ws_url.len - 36]); - try testing.expectEqual(id, ws_url[ws_url.len - 36 ..]); - - break :blk id[0..36].*; - }; + const session_id = try createHTTPSession("{\"capabilities\":{\"firstMatch\":[{}],\"alwaysMatch\":{\"browserName\":\"firefox\",\"webSocketUrl\":true}}}", true); + var c = try createTestClient(); + defer c.deinit(); { // The session already exists on the advertised URL: no session.new // needed (or possible), everything else works as usual. - var c = try createTestClient(); - defer c.deinit(); var path_buf: [64]u8 = undefined; try c.handshake(try std.fmt.bufPrint(&path_buf, "/session/{s}", .{&session_id})); @@ -1518,31 +1735,127 @@ test "server: classic session bootstrap" { try assertBidiMessage(&c, .{ .type = "success", .id = 3, .result = .{ .contexts = .{} } }); } + // one websocket per session { + var c2 = try createTestClient(); + defer c2.deinit(); + var path_buf: [64]u8 = undefined; + const res = try c2.upgradeRequest(try std.fmt.bufPrint(&path_buf, "/session/{s}", .{&session_id})); + try testing.expectEqual("HTTP/1.1 409 \r\nConnection: Close\r\nContent-Length: 25\r\n\r\nSession already connected", res); + } + + try deleteHTTPSession(&session_id, true); + + // ending the session closes the websocket + { + const msg = try c.readWebsocketMessage() orelse return error.NoMessage; + defer if (msg.cleanup_fragment) c.reader.cleanup(); + try testing.expectEqual(.close, msg.type); + } + + // and it's gone + try deleteHTTPSession(&session_id, false); + { + var c2 = try createTestClient(); + defer c2.deinit(); + var path_buf: [64]u8 = undefined; + const res = try c2.upgradeRequest(try std.fmt.bufPrint(&path_buf, "/session/{s}", .{&session_id})); + try testing.expectEqual("HTTP/1.1 404 \r\nConnection: Close\r\nContent-Length: 9\r\n\r\nNot found", res); + } +} + +test "server: HTTP session outlives its websocket" { + // Without webSocketUrl the session is driven over HTTP alone; asking for + // it later still works, and closing the websocket doesn't end the session. + const session_id = try createHTTPSession("{\"capabilities\":{\"alwaysMatch\":{\"browserName\":\"chrome\"}}}", false); + defer deleteHTTPSession(&session_id, true) catch |err| @panic(@errorName(err)); + + var path_buf: [64]u8 = undefined; + const path = try std.fmt.bufPrint(&path_buf, "/session/{s}", .{&session_id}); + + { + var c = try createTestClient(); + defer c.deinit(); + try c.handshake(path); + + try c.bidiCommand("{\"id\":1,\"method\":\"browsingContext.create\",\"params\":{\"type\":\"tab\"}}"); + try discardBidiMessage(&c); + + // a client-initiated close is answered and the session carries on + try sys_net.writeAll(c.socket, &[_]u8{ 136, 128, 0, 0, 0, 0 }); + const msg = try c.readWebsocketMessage() orelse return error.NoMessage; + defer if (msg.cleanup_fragment) c.reader.cleanup(); + try testing.expectEqual(.close, msg.type); + } + + // 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. + var c = try createTestClient(); + defer c.deinit(); + var attempts: usize = 0; + while (true) : (attempts += 1) { + const res = try c.upgradeRequest(path); + 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(attempts < 100); + c.deinit(); + c = try createTestClient(); + lp.io.sleep(.fromMilliseconds(10), .awake) catch {}; + } + + // same worker: the context created over the first websocket is there + try c.bidiCommand("{\"id\":2,\"method\":\"browsingContext.getTree\"}"); + const res = try c.readWebsocketMessage() orelse return error.NoMessage; + defer if (res.cleanup_fragment) c.reader.cleanup(); + try testing.expect(std.mem.indexOf(u8, res.data, "\"url\":\"about:blank\"") != null); +} + +test "server: HTTP session idle timeout" { + const server = testing.test_cdp_server.?; + const original = server.session_timeout_ms; + defer server.session_timeout_ms = original; + server.session_timeout_ms = 50; + + const session_id = try createHTTPSession("{\"capabilities\":{}}", false); + + // reaped: the DELETE has nothing to find + var attempts: usize = 0; + while (true) : (attempts += 1) { var c = try createTestClient(); defer c.deinit(); var request_buf: [128]u8 = undefined; const res = try c.httpRequest(try std.fmt.bufPrint(&request_buf, "DELETE /session/{s} HTTP/1.1\r\nContent-Length: 0\r\n\r\n", .{&session_id})); - try testing.expectEqual("HTTP/1.1 200 OK\r\n" ++ - "Content-Length: 14\r\n" ++ - "Content-Type: application/json; charset=UTF-8\r\n\r\n" ++ - "{\"value\":null}", res); + if (std.mem.startsWith(u8, res, "HTTP/1.1 404 ")) { + break; + } + // DELETE on a live session ends it, which is what the reaper was + // about to do: only a still-running session can answer 200 here + try testing.expect(std.mem.startsWith(u8, res, "HTTP/1.1 200 ")); + try testing.expect(attempts == 0); + lp.io.sleep(.fromMilliseconds(20), .awake) catch {}; } } -test "server: classic session bootstrap errors" { - { - // the body can arrive after the headers - var c = try createTestClient(); - defer c.deinit(); - const body = "{\"capabilities\":{\"alwaysMatch\":{\"browserName\":\"firefox\"}}}"; - try sys_net.writeAll(c.socket, std.fmt.comptimePrint("POST /session HTTP/1.1\r\nContent-Length: {d}\r\n\r\n", .{body.len})); - lp.io.sleep(.fromMilliseconds(20), .awake) catch {}; - const res = try c.httpRequest(body); - try testing.expect(std.mem.startsWith(u8, res, "HTTP/1.1 500 Internal Server Error\r\n")); - try testing.expect(std.mem.endsWith(u8, res, "{\"value\":{\"error\":\"session not created\",\"message\":\"only WebDriver BiDi sessions are supported; request the webSocketUrl capability\",\"stacktrace\":\"\"}}")); - } +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. + const server = testing.test_cdp_server.?; + const live = server.worker_pool.live; + const session_id = try createHTTPSession("{\"capabilities\":{}}", false); + try deleteHTTPSession(&session_id, true); + + // the worker exited and gave its slot back + var attempts: usize = 0; + while (server.worker_pool.live != live) : (attempts += 1) { + try testing.expect(attempts < 200); + lp.io.sleep(.fromMilliseconds(10), .awake) catch {}; + } +} + +test "server: HTTP session bootstrap errors" { { var c = try createTestClient(); defer c.deinit(); @@ -1556,6 +1869,60 @@ test "server: classic session bootstrap errors" { try assertHTTPError(405, "Method not allowed", "DELETE /session HTTP/1.1\r\nContent-Length: 0\r\n\r\n"); // a websocket upgrade on /session/ needs a real session id try assertHTTPError(404, "Not found", "GET /session/abc HTTP/1.1\r\n\r\n"); + try assertHTTPError(404, "Not found", "GET /session/00000000-0000-4000-8000-000000000000 HTTP/1.1\r\n" ++ + "Connection: upgrade\r\nUpgrade: websocket\r\nsec-websocket-version:13\r\nsec-websocket-key: k\r\n\r\n"); + try deleteHTTPSession("00000000-0000-4000-8000-000000000000", false); +} + +// POST /session; asserts the response and whether it advertised a websocket +fn createHTTPSession(body: []const u8, expect_ws_url: bool) ![36]u8 { + var c = try createTestClient(); + defer c.deinit(); + + // the body can arrive after the headers + var head_buf: [128]u8 = undefined; + try sys_net.writeAll(c.socket, try std.fmt.bufPrint(&head_buf, "POST /session HTTP/1.1\r\n" ++ + "Content-Type: application/json;charset=UTF-8\r\n" ++ + "Content-Length: {d}\r\n\r\n", .{body.len})); + lp.io.sleep(.fromMilliseconds(20), .awake) catch {}; + const res = try c.httpRequest(body); + try testing.expect(std.mem.startsWith(u8, res, "HTTP/1.1 200 OK\r\n")); + + const json = res[std.mem.indexOf(u8, res, "\r\n\r\n").? + 4 ..]; + const parsed = try std.json.parseFromSlice(std.json.Value, testing.allocator, json, .{}); + defer parsed.deinit(); + + const value = parsed.value.object.get("value").?.object; + const id = value.get("sessionId").?.string; + try testing.expectEqual(36, id.len); + + const capabilities = value.get("capabilities").?.object; + try testing.expectEqual("Lightpanda", capabilities.get("browserName").?.string); + try testing.expectEqual(false, capabilities.get("acceptInsecureCerts").?.bool); + if (expect_ws_url) { + const ws_url = capabilities.get("webSocketUrl").?.string; + try testing.expectEqual("ws://127.0.0.1:9583/session/", ws_url[0 .. ws_url.len - 36]); + try testing.expectEqual(id, ws_url[ws_url.len - 36 ..]); + } else { + try testing.expectEqual(null, capabilities.get("webSocketUrl")); + } + return id[0..36].*; +} + +fn deleteHTTPSession(session_id: *const [36]u8, expect_live: bool) !void { + var c = try createTestClient(); + defer c.deinit(); + var request_buf: [128]u8 = undefined; + const res = try c.httpRequest(try std.fmt.bufPrint(&request_buf, "DELETE /session/{s} HTTP/1.1\r\nContent-Length: 0\r\n\r\n", .{session_id})); + if (expect_live) { + try testing.expectEqual("HTTP/1.1 200 OK\r\n" ++ + "Content-Length: 14\r\n" ++ + "Content-Type: application/json; charset=UTF-8\r\n\r\n" ++ + "{\"value\":null}", res); + } else { + try testing.expect(std.mem.startsWith(u8, res, "HTTP/1.1 404 Not Found\r\n")); + try testing.expect(std.mem.endsWith(u8, res, "{\"value\":{\"error\":\"invalid session id\",\"message\":\"no such session\",\"stacktrace\":\"\"}}")); + } } test "server: protocol gate" { @@ -2043,7 +2410,7 @@ const TestClient = struct { return cl + header.len; } - fn handshake(self: *TestClient, path: []const u8) !void { + fn upgradeRequest(self: *TestClient, path: []const u8) ![]const u8 { var request_buf: [256]u8 = undefined; const request = try std.fmt.bufPrint(&request_buf, "GET {s} HTTP/1.1\r\n" ++ "Connection: upgrade\r\n" ++ @@ -2051,8 +2418,11 @@ const TestClient = struct { "sec-websocket-version:13\r\n" ++ "sec-websocket-key: this is my key\r\n" ++ "Custom: Header-Value\r\n\r\n", .{path}); + return self.httpRequest(request); + } - const res = try self.httpRequest(request); + fn handshake(self: *TestClient, path: []const u8) !void { + const res = try self.upgradeRequest(path); try testing.expectEqual("HTTP/1.1 101 Switching Protocols\r\n" ++ "Upgrade: websocket\r\n" ++ "Connection: upgrade\r\n" ++ diff --git a/src/server/bidi/BiDi.zig b/src/server/bidi/BiDi.zig index 0e6b2c300..896f73c6e 100644 --- a/src/server/bidi/BiDi.zig +++ b/src/server/bidi/BiDi.zig @@ -38,7 +38,16 @@ const Allocator = std.mem.Allocator; const BiDi = @This(); app: *App, -conn: Link, + +// The websocket, when a client is connected. Null for an HTTP WebDriver +// session (until it optionally connects via WebSocket) +link: ?*Link, + +// The worker's mailbox, owned by the Worker and thus outliving the link. +inbox: *Inbox, + +// WebDriver can be BiDi only, or HTTP WebDriver + BiDi or HTTP WebDriver only. +mode: Mode, // Re-used arena for processing a message. Works because we strictly process // one message at a time. @@ -78,20 +87,46 @@ const Subscription = struct { event: []const u8, }; +pub const Mode = union(enum) { + // Directly created via websocket upgrade, tied to the websocket's lifetime + bidi_only: void, + + // created via HTTP, a websocket may or may not web associated with it (it + // can come and go), but the lifetime is explicit: either removed via HTTP + // (DELETE /session/:id) or by the HTTP reaper + http: *Server.Worker, +}; + +// What a worker is born from: a websocket upgrade (the session comes later +// via session.new) or an HTTP session (a websocket may come later via +// GET /session/{id}); never both. +pub const Origin = union(enum) { + socket: posix.socket_t, + session: struct { id: [36]u8, worker: *Server.Worker }, +}; + const InputMessage = struct { id: ?u64 = null, method: ?[]const u8 = null, }; -pub fn init(self: *BiDi, app: *App, socket: posix.socket_t, inbox: *Inbox, session_id: ?[36]u8) !void { +pub fn init(self: *BiDi, app: *App, inbox: *Inbox, origin: Origin) !void { const allocator = app.allocator; self.* = .{ .app = app, - .conn = undefined, + .link = null, + .inbox = inbox, + .mode = switch (origin) { + .socket => .bidi_only, + .session => |session| .{ .http = session.worker }, + }, .browser = undefined, .user_context = undefined, .notification = undefined, - .session_id = session_id, + .session_id = switch (origin) { + .socket => null, + .session => |session| session.id, + }, .node_registry = .init(allocator), .handles = .{ .allocator = allocator }, .message_arena = std.heap.ArenaAllocator.init(allocator), @@ -101,8 +136,11 @@ pub fn init(self: *BiDi, app: *App, socket: posix.socket_t, inbox: *Inbox, sessi try self.browser.init(app, .{}); errdefer self.browser.deinit(); - try self.conn.init(app, socket, .bidi, inbox); - errdefer self.conn.deinit(); + switch (origin) { + .socket => |socket| self.link = try Link.create(app, socket, .bidi, inbox), + .session => {}, + } + errdefer if (self.link) |l| l.destroy(); self.notification = try Notification.init(allocator); errdefer self.notification.deinit(); @@ -128,11 +166,47 @@ pub fn deinit(self: *BiDi) void { self.node_registry.deinit(); self.notification.deinit(); self.browser.deinit(); - self.conn.deinit(); + // The loop let go of the link before we got here (Server.Worker.run) + if (self.link) |l| { + l.destroy(); + } self.message_arena.deinit(); self.session_arena.deinit(); } +// Worker thread, from the inbox: the loop is already reading from it. +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", .{}); + l.destroy(); + return; + } + self.link = l; +} + +// Worker thread. The link is gone (peer closed, or the loop dropped it). +// Returns true when the worker is done with it: a bidi-only session dies +// with its connection, an HTTP session just drops the link and waits +// for the next one, or for DELETE / the idle reaper. +pub fn onLinkGone(self: *BiDi) bool { + const worker = switch (self.mode) { + .bidi_only => return true, + .http => |worker| worker, + }; + self.releaseLink(worker); + return false; +} + +fn releaseLink(self: *BiDi, worker: *Server.Worker) void { + const l = self.link orelse return; + self.link = null; + // blocks until the loop has stopped reading from it + worker.releaseLink(); + l.destroy(); +} + pub fn replaceSession(self: *BiDi, id: []const u8) !void { self.resetRealm(); try self.newUserContext(id); @@ -316,6 +390,10 @@ pub fn sendError(self: *BiDi, id: ?u64, code: []const u8, message: []const u8) ! return self.sendJSON(.{ .type = "error", .id = id, .@"error" = code, .message = message }); } +// Without a link there's nobody to tell: an HTTP session between +// connections drops events and late results (a navigate that completes +// after the client went away). fn sendJSON(self: *BiDi, message: anytype) !void { - return self.conn.sendJSON(message, .{}); + const l = self.link orelse return; + return l.sendJSON(message, .{}); } diff --git a/src/server/bidi/browser.zig b/src/server/bidi/browser.zig index b4049047d..8403982ef 100644 --- a/src/server/bidi/browser.zig +++ b/src/server/bidi/browser.zig @@ -47,7 +47,7 @@ fn close(cmd: *const BiDi.Command) !void { const bidi = cmd.bidi; const arena = try bidi.browser.arena_pool.acquire(.tiny, "bidi browser close"); - bidi.conn.inbox.push(arena, .close); + bidi.inbox.push(arena, .quit); } const UserContextInfo = struct { userContext: []const u8 }; diff --git a/src/server/bidi/session.zig b/src/server/bidi/session.zig index e11e4d167..f9df2b807 100644 --- a/src/server/bidi/session.zig +++ b/src/server/bidi/session.zig @@ -81,7 +81,7 @@ pub const Capabilities = struct { setWindowRect: bool = false, userAgent: []const u8, proxy: struct {} = .{}, - webSocketUrl: ?[]const u8 = null, // only reported for the classic handshake + webSocketUrl: ?[]const u8 = null, // only reported for the HTTP handshake // ugh, are you kidding me? All this so we don't emit the webSocketUrl // when it's null. @@ -108,7 +108,7 @@ fn end(cmd: *const BiDi.Command) !void { const bidi = cmd.bidi; const arena = try bidi.browser.arena_pool.acquire(.tiny, "bidi session end"); - bidi.conn.inbox.push(arena, .close); + bidi.inbox.push(arena, .quit); } // Subscriptions are global (per-context filtering is not supported yet). diff --git a/src/server/bidi/testing.zig b/src/server/bidi/testing.zig index fc5c08700..1c7d1aea4 100644 --- a/src/server/bidi/testing.zig +++ b/src/server/bidi/testing.zig @@ -66,7 +66,7 @@ pub const TestContext = struct { pub fn bidi(self: *TestContext) *BiDi { if (!self.bidi_initialized) { - self.bidi_.init(base.test_app, self.bidi_socket, &self.inbox, null) catch |err| @panic(@errorName(err)); + self.bidi_.init(base.test_app, &self.inbox, .{ .socket = self.bidi_socket }) catch |err| @panic(@errorName(err)); self.bidi_initialized = true; self.driver = .init(.{ .bidi = &self.bidi_ }, &self.inbox); self.driver.attach(); diff --git a/src/server/cdp/CDP.zig b/src/server/cdp/CDP.zig index 2a79c6956..25911bb8e 100644 --- a/src/server/cdp/CDP.zig +++ b/src/server/cdp/CDP.zig @@ -59,7 +59,7 @@ pub const InvocationIdGen = Incrementing(u32, "INV"); const CDP = @This(); app: *App, -conn: Link, +link: Link, browser: Browser, allocator: Allocator, @@ -100,7 +100,7 @@ pub fn init(self: *CDP, app: *App, socket: posix.socket_t, inbox: *Inbox) !void self.* = .{ .app = app, - .conn = undefined, + .link = undefined, .browser = undefined, .allocator = allocator, .browser_context = null, @@ -114,7 +114,7 @@ pub fn init(self: *CDP, app: *App, socket: posix.socket_t, inbox: *Inbox) !void try self.browser.init(app, .{ .env = .{ .with_inspector = true } }); errdefer self.browser.deinit(); - try self.conn.init(app, socket, .cdp, inbox); + try self.link.init(app, socket, .cdp, inbox); } pub fn deinit(self: *CDP) void { @@ -127,7 +127,7 @@ pub fn deinit(self: *CDP) void { self.notification_arena.deinit(); self.browser_context_arena.deinit(); self.streams.deinit(); - self.conn.deinit(); + self.link.deinit(); } // Called by the Server run loop when readable bytes arrive on the CDP // socket. Feeds them through the WS framer and pushes each parsed frame @@ -162,7 +162,7 @@ pub fn processMessage(self: *CDP, msg: []const u8) !void { } pub fn sendJSON(self: *CDP, message: anytype) !void { - try self.conn.sendJSON(message, .{ .emit_null_optional_fields = false }); + try self.link.sendJSON(message, .{ .emit_null_optional_fields = false }); } // Parse-then-dispatch entry point. Used by: @@ -1144,7 +1144,7 @@ pub const BrowserContext = struct { }; const cdp = self.cdp; - const allocator = cdp.conn.send_arena.allocator(); + const allocator = cdp.link.send_arena.allocator(); const field = ",\"sessionId\":\""; @@ -1170,7 +1170,7 @@ pub const BrowserContext = struct { std.debug.assert(buf.items.len == message_len); } - try cdp.conn.sendJSONRaw(buf); + try cdp.link.sendJSONRaw(buf); } }; diff --git a/src/server/http.zig b/src/server/http.zig index a4b8efa3c..5777148ea 100644 --- a/src/server/http.zig +++ b/src/server/http.zig @@ -104,6 +104,9 @@ pub const Connection = struct { // Filled in by the router for /session/{id}[/...] routes; points // into the read buffer like path does. session_id: ?*const [36]u8 = null, + + // valid for handling a single request up to sending the response + arena: Allocator, }; pub const Method = enum { @@ -117,7 +120,7 @@ pub const Connection = struct { header: void, // still parsing the header request: Request, - fn parseHeader(self: *State, data: []u8) !bool { + fn parseHeader(self: *State, arena: Allocator, data: []u8) !bool { const header_index = std.mem.indexOf(u8, data, "\r\n\r\n") orelse { return false; }; @@ -147,12 +150,13 @@ pub const Connection = struct { .keepalive = keepalive, .body = data[body_start..total], .head = data[0..body_start], + .arena = arena, } }; return true; } - // The classic WebDriver bootstrap (POST /session) is the only thing + // The HTTP WebDriver bootstrap (POST /session) is the only thing // that sends a body; everything else is 0. fn contentLength(header: []const u8) !usize { const key = "\r\ncontent-length:"; @@ -320,6 +324,8 @@ pub const Connection = struct { // How long a connection may sit without completing a request before we close it. pub const IDLE_TIMEOUT_MS = 10_000; +const REQUEST_ARENA_RETAIN = 8192; + pub fn processEvent(server: *Server, conn: *Connection, rw: Server.IOEvent.ReadWrite, now: u64) void { if (conn.pending != null) { // registered for OUT only; a hangup shows up as a write error @@ -377,11 +383,12 @@ fn flush(server: *Server, conn: *Connection, now: u64) void { fn processHTTP(server: *Server, conn: *Connection, now: u64) !bool { const http = &conn.state; + const arena = server.request_arena.allocator(); while (true) { switch (http.*) { .header => { const data = try conn.buffer.read(conn.socket); - if (try http.parseHeader(data) == false) { + if (try http.parseHeader(arena, data) == false) { // don't have a complete header yet return true; } @@ -392,6 +399,7 @@ fn processHTTP(server: *Server, conn: *Connection, now: u64) !bool { } }, .request => |*req| { + defer _ = server.request_arena.reset(.{ .retain_with_limit = REQUEST_ARENA_RETAIN }); if (try serveHTTP(server, conn, req) == .upgraded) { // The fd moved to a WebSocket (and out of server.http); all // that's left of this Connection is to recycle it. @@ -425,33 +433,21 @@ fn processHTTP(server: *Server, conn: *Connection, now: u64) !bool { // Error responses use a minimal, uniform shape: no reason phrase, an explicit // Connection: Close, and no Content-Type. errorResponse builds it at comptime. const invalid_request_response = errorResponse(400, "Invalid request"); - const invalid_protocol_response = errorResponse(400, "Invalid HTTP protocol"); - const missing_header_response = errorResponse(400, "Missing required header"); - const forbidden_origin_response = errorResponse(403, "Origin not allowed"); - 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 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"); - const empty_json_list_response = staticResponse(.{ .status = "200 OK", .body = "[]", .content_type = "application/json; charset=UTF-8" }); - // WebDriver's discovery endpoint; `ready` is whether a new session can be // created, which the bootstrap never refuses. const status_response = staticResponse(.{ .status = "200 OK", .body = "{\"value\":{\"ready\":true,\"message\":\"\"}}", .content_type = "application/json; charset=UTF-8" }); - const delete_session_response = staticResponse(.{ .status = "200 OK", .body = "{\"value\":null}", .content_type = "application/json; charset=UTF-8" }); - const protocol_response = staticResponse(.{ .status = "200 OK", .body = @embedFile("../data/protocol.json"), .content_type = "application/json; charset=UTF-8" }); const Served = enum { @@ -489,12 +485,12 @@ const routes = [_]Route{ }; const session_routes = [_]Route{ - .{ .method = .GET, .path = "", .handler = upgradeBiDi }, + .{ .method = .GET, .path = "", .handler = upgradeSession }, .{ .method = .DELETE, .path = "", .handler = deleteSession }, }; // Routes under /session/{id}; path is what follows the id ("" for the -// session itself). The classic command surface goes here. +// session itself). The HTTP command surface goes here. const SESSION_PREFIX = "/session/"; const SESSION_ID_LEN = 36; @@ -635,65 +631,64 @@ fn gateOpen(server: *const Server, gate: Route.Gate) bool { }; } +// GET / (cdp) fn upgradeCDP(server: *Server, conn: *Connection, req: *Connection.Request) !Served { - return upgrade(server, conn, req, .cdp, null); + return upgradeSpawn(server, conn, req, .cdp); } +// GET /json/version (cdp) fn serveJSONVersion(server: *Server, conn: *Connection, req: *Connection.Request) !Served { return serveHTTPResponse(server, conn, req, .{ .static = server.json_version_response }); } +// GET /json/list or GET /json (cdp) fn serveJSONList(server: *Server, conn: *Connection, req: *Connection.Request) !Served { return serveHTTPResponse(server, conn, req, .{ .static = empty_json_list_response }); } +// GET /json/protocol (cdp) fn serveJSONProtocol(server: *Server, conn: *Connection, req: *Connection.Request) !Served { return serveHTTPResponse(server, conn, req, .{ .static = protocol_response }); } +// GET /metrics (internal) fn serveMetrics(server: *Server, conn: *Connection, req: *Connection.Request) !Served { const writer = try beginBody(server); lp.metrics.write(writer); return serveDynamicHTTPResponse(server, conn, req, "200 OK", "text/plain; version=0.0.4; charset=utf-8"); } +// GET /status (webdriver) fn serveStatus(server: *Server, conn: *Connection, req: *Connection.Request) !Served { return serveHTTPResponse(server, conn, req, .{ .static = status_response }); } -// req.session_id is null for GET /session, set for GET /session/{id} +// GET /session (webdriver (direct bidi)) fn upgradeBiDi(server: *Server, conn: *Connection, req: *Connection.Request) !Served { - const session_id: ?[36]u8 = if (req.session_id) |s| s.* else null; - return upgrade(server, conn, req, .bidi, session_id); + return upgradeSpawn(server, conn, req, .bidi); } -// What Selenium does before it speaks BiDi: a classic POST /session that -// hands back the websocket URL of a session that already exists. +// POST /session (webdriver) fn newSession(server: *Server, conn: *Connection, req: *Connection.Request) !Served { - const allocator = server.app.allocator; - const Capability = struct { webSocketUrl: ?bool = null }; - const parsed = std.json.parseFromSlice(struct { + const parsed = std.json.parseFromSliceLeaky(struct { capabilities: ?struct { alwaysMatch: ?Capability = null, firstMatch: ?[]const Capability = null, } = null, - }, allocator, req.body, .{ .ignore_unknown_fields = true }) catch { + }, req.arena, req.body, .{ .ignore_unknown_fields = true }) catch { return serveWebDriver(server, conn, req, "400 Bad Request", .{ .@"error" = "invalid argument", .message = "invalid JSON body", .stacktrace = "", }); }; - defer parsed.deinit(); - // Without the capability the client intends to drive the session over - // HTTP, which this server doesn't serve: tell it now rather than 404 - // its first real command. - if (!requestsWebSocketUrl(parsed.value.capabilities)) { + if (server.worker_pool.isFull()) { + lp.metrics.serve_connection_limit.incr(); return serveWebDriver(server, conn, req, "500 Internal Server Error", .{ .@"error" = "session not created", - .message = "only WebDriver BiDi sessions are supported; request the webSocketUrl capability", + .message = "too many sessions", .stacktrace = "", }); } @@ -701,8 +696,36 @@ fn newSession(server: *Server, conn: *Connection, req: *Connection.Request) !Ser var session_id: [36]u8 = undefined; uuidv4(&session_id); - const url = try std.fmt.allocPrint(allocator, "{s}{s}", .{ server.bidi_session_url, &session_id }); - defer allocator.free(url); + _ = 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", + .message = "failed to start the session", + .stacktrace = "", + }); + }; + + const is_requesting_websocket_url = blk: { + const caps = parsed.capabilities orelse break :blk false; + if (caps.alwaysMatch) |always| { + if (always.webSocketUrl == true) { + break :blk true; + } + } + for (caps.firstMatch orelse &.{}) |first| { + if (first.webSocketUrl == true) { + break :blk true; + } + } + break :blk false; + }; + + const url: ?[]const u8 = blk: { + if (is_requesting_websocket_url) { + break :blk try std.fmt.allocPrint(req.arena, "{s}{s}", .{ server.bidi_session_url, &session_id }); + } + break :blk null; + }; return serveWebDriver(server, conn, req, "200 OK", .{ .sessionId = &session_id, @@ -713,32 +736,49 @@ fn newSession(server: *Server, conn: *Connection, req: *Connection.Request) !Ser }); } -fn requestsWebSocketUrl(capabilities: anytype) bool { - const caps = capabilities orelse return false; - if (caps.alwaysMatch) |always| { - if (always.webSocketUrl == true) { - return true; - } +// GET /session/ID (webdriver (upgrade to bidi)) +fn upgradeSession(server: *Server, conn: *Connection, req: *Connection.Request) !Served { + const worker = server.findSession(req.session_id.?) orelse { + return serveNotFound(server, conn, req); + }; + + if (worker.link != null) { + // already joined + return serveHTTPResponse(server, conn, req, .{ .static = session_connected_response }); } - for (caps.firstMatch orelse &.{}) |first| { - if (first.webSocketUrl == true) { - return true; - } - } - return false; + + return upgrade(server, conn, req, .{ .attach = worker }); } -// Answers a classic WebDriver request with {"value": value}. +// DELETE /session/ID (webdriver) +fn deleteSession(server: *Server, conn: *Connection, req: *Connection.Request) !Served { + const worker = server.findSession(req.session_id.?) orelse { + return serveWebDriver(server, conn, req, "404 Not Found", .{ + .@"error" = "invalid session id", + .message = "no such session", + .stacktrace = "", + }); + }; + server.quitSession(worker); + return serveHTTPResponse(server, conn, req, .{ .static = delete_session_response }); +} + +// CDP or Bidi directly creating a Worker from an websocket upgrade +fn upgradeSpawn(server: *Server, conn: *Connection, req: *Connection.Request, protocol: Driver.Protocol) !Served { + if (server.worker_pool.isFull()) { + lp.metrics.serve_connection_limit.incr(); + return serveHTTPResponse(server, conn, req, .{ .static = service_unavailable_response }); + } + return upgrade(server, conn, req, .{ .spawn = protocol }); +} + +// Answers a HTTP WebDriver request with {"value": value}. fn serveWebDriver(server: *Server, conn: *Connection, req: *const Connection.Request, comptime status: []const u8, value: anytype) !Served { const writer = try beginBody(server); try std.json.Stringify.value(.{ .value = value }, .{}, writer); return serveDynamicHTTPResponse(server, conn, req, status, "application/json; charset=UTF-8"); } -fn deleteSession(server: *Server, conn: *Connection, req: *Connection.Request) !Served { - return serveHTTPResponse(server, conn, req, .{ .static = delete_session_response }); -} - fn serveNotFound(server: *Server, conn: *Connection, req: *Connection.Request) !Served { return serveHTTPResponse(server, conn, req, .{ .static = not_found_response }); } @@ -831,14 +871,15 @@ pub fn buildJSONVersionResponse(app: *const App, port: u16) ![]const u8 { return try std.fmt.allocPrint(app.allocator, response_format, .{ body_len, host, port }); } -// Shared upgrade path: validate the WebSocket headers, write the 101, park the -// fd, and spawn the worker that will build the driver and attach it. -fn upgrade(server: *Server, conn: *Connection, req: *Connection.Request, protocol: Driver.Protocol, session_id: ?[36]u8) !Served { - if (server.websocket_pool.isFull()) { - lp.metrics.serve_connection_limit.incr(); - return serveHTTPResponse(server, conn, req, .{ .static = service_unavailable_response }); - } +// Where the upgraded socket goes: a new worker, or an existing session's. +const Upgrade = union(enum) { + spawn: Driver.Protocol, + attach: *Server.Worker, +}; +// Shared upgrade path: validate the WebSocket headers, write the 101, and +// hand the fd to its worker (spawning one for a new connection). +fn upgrade(server: *Server, conn: *Connection, req: *Connection.Request, target: Upgrade) !Served { var accept_buf: [28]u8 = undefined; const accept_key = webSocketAccept(req.head, &accept_buf) catch |err| { const response: []const u8 = switch (err) { @@ -863,7 +904,10 @@ fn upgrade(server: *Server, conn: *Connection, req: *Connection.Request, protoco return error.ConnectionClosed; } - server.upgradeConnection(conn, protocol, session_id); + switch (target) { + .spawn => |protocol| server.upgradeConnection(conn, protocol), + .attach => |worker| server.attachConnection(worker, conn), + } return .upgraded; } From 63924548927f632612df22dd7a8e083f2fbca464 Mon Sep 17 00:00:00 2001 From: Karl Seguin Date: Fri, 4 Sep 2026 17:16:18 +0800 Subject: [PATCH 2/5] fix test-only tsan issue --- src/Metrics.zig | 13 ++++++++++--- src/server/Server.zig | 7 ++++--- 2 files changed, 14 insertions(+), 6 deletions(-) diff --git a/src/Metrics.zig b/src/Metrics.zig index 55f0942d3..38f77323f 100644 --- a/src/Metrics.zig +++ b/src/Metrics.zig @@ -203,7 +203,11 @@ const Gauge = struct { fn write(self: *const Gauge, comptime name: []const u8, comptime help_text: []const u8, writer: *std.Io.Writer) !void { try writer.writeAll("# HELP " ++ name ++ " " ++ help_text ++ "\n" ++ "# TYPE " ++ name ++ " gauge\n"); - try writer.print(name ++ " {d}\n", .{@atomicLoad(isize, &self.value, .monotonic)}); + try writer.print(name ++ " {d}\n", .{self.get()}); + } + + fn get(self: *const Gauge) isize { + return @atomicLoad(isize, &self.value, .monotonic); } }; @@ -228,11 +232,14 @@ fn GaugeEnum(comptime label: []const u8, comptime T: type) type { self.values.getPtr(tag).add(n); } + pub fn get(self: *const Self, tag: T) isize { + return self.values.getPtrConst(tag).get(); + } + fn write(self: *const Self, comptime name: []const u8, comptime help_text: []const u8, writer: *std.Io.Writer) !void { try writer.writeAll("# HELP " ++ name ++ " " ++ help_text ++ "\n" ++ "# TYPE " ++ name ++ " gauge\n"); inline for (comptime std.enums.values(Tag)) |tag| { - const value = @atomicLoad(isize, &self.values.getPtrConst(tag).value, .monotonic); - try writer.print(name ++ "{{" ++ label ++ "=\"" ++ @tagName(tag) ++ "\"}} {d}\n", .{value}); + try writer.print(name ++ "{{" ++ label ++ "=\"" ++ @tagName(tag) ++ "\"}} {d}\n", .{self.get(tag)}); } } }; diff --git a/src/server/Server.zig b/src/server/Server.zig index 506178a7e..7ebeabc88 100644 --- a/src/server/Server.zig +++ b/src/server/Server.zig @@ -1841,15 +1841,16 @@ test "server: HTTP session idle timeout" { 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. - const server = testing.test_cdp_server.?; - const live = server.worker_pool.live; + // worker_pool.live is the loop's; the gauge is the cross-thread view of it + const gauge = &lp.metrics.serve_active_connections; + const live = gauge.get(.bidi); const session_id = try createHTTPSession("{\"capabilities\":{}}", false); try deleteHTTPSession(&session_id, true); // the worker exited and gave its slot back var attempts: usize = 0; - while (server.worker_pool.live != live) : (attempts += 1) { + while (gauge.get(.bidi) != live) : (attempts += 1) { try testing.expect(attempts < 200); lp.io.sleep(.fromMilliseconds(10), .awake) catch {}; } From 3fb4539b87e16e4f98e9eefdc59ad56b6a3a1c6f Mon Sep 17 00:00:00 2001 From: Karl Seguin Date: Mon, 7 Sep 2026 15:15:22 +0800 Subject: [PATCH 3/5] fix import --- src/server/bidi/BiDi.zig | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/server/bidi/BiDi.zig b/src/server/bidi/BiDi.zig index 896f73c6e..7f72ff2c3 100644 --- a/src/server/bidi/BiDi.zig +++ b/src/server/bidi/BiDi.zig @@ -1,5 +1,5 @@ -// Copyright (C) 2023-2026 Lightpanda (Selecy SAS) // +// Copyright (C) 2023-2026 Lightpanda (Selecy SAS) // Francis Bouvier // Pierre Tachoire // @@ -20,6 +20,7 @@ const std = @import("std"); const lp = @import("lightpanda"); const App = @import("../../App.zig"); +const Inbox = @import("../../Inbox.zig"); const uuidv4 = @import("../../id.zig").uuidv4; const Browser = @import("../../browser/Browser.zig"); const Session = @import("../../browser/Session.zig"); @@ -27,7 +28,7 @@ const Notification = @import("../../Notification.zig"); const NodeRegistry = @import("../../NodeRegistry.zig"); const Link = @import("../Link.zig"); -const Inbox = @import("../../Inbox.zig"); +const Server = @import("../Server.zig"); const script = @import("script.zig"); const remote_value = @import("remote_value.zig"); From bcf69ced9c9568dbc458fe90de06efed8b881d89 Mon Sep 17 00:00:00 2001 From: Karl Seguin Date: Thu, 10 Sep 2026 09:04:20 +0800 Subject: [PATCH 4/5] 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 | 25 +++++++++++++++++++++++-- src/help.zon | 3 ++- src/server/Link.zig | 10 +++++++++- src/server/Server.zig | 38 ++++++++++++++++++++++++++++++++------ src/server/bidi/BiDi.zig | 18 +++++++++++++----- src/server/cdp/CDP.zig | 6 +++--- src/server/http.zig | 12 +++++++++++- 7 files changed, 93 insertions(+), 19 deletions(-) diff --git a/src/Config.zig b/src/Config.zig index 1994b2289..e00dac697 100644 --- a/src/Config.zig +++ b/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")); diff --git a/src/help.zon b/src/help.zon index 97a20c6c2..df0afe47c 100644 --- a/src/help.zon +++ b/src/help.zon @@ -48,7 +48,8 @@ \\ --http-session-timeout \\ 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 \\ Port of the CDP server. \\ Defaults to 9222. diff --git a/src/server/Link.zig b/src/server/Link.zig index 0df00b4a1..ffe97fa38 100644 --- a/src/server/Link.zig +++ b/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; } diff --git a/src/server/Server.zig b/src/server/Server.zig index 7ebeabc88..fabb212cf 100644 --- a/src/server/Server.zig +++ b/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. diff --git a/src/server/bidi/BiDi.zig b/src/server/bidi/BiDi.zig index 7f72ff2c3..e75ef157f 100644 --- a/src/server/bidi/BiDi.zig +++ b/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(); diff --git a/src/server/cdp/CDP.zig b/src/server/cdp/CDP.zig index 25911bb8e..560e923b6 100644 --- a/src/server/cdp/CDP.zig +++ b/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 { diff --git a/src/server/http.zig b/src/server/http.zig index 5777148ea..c9fa51d86 100644 --- a/src/server/http.zig +++ b/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 }); From 2ace9d15480429ac3ac573c1084e3a4163836943 Mon Sep 17 00:00:00 2001 From: Karl Seguin Date: Thu, 10 Sep 2026 09:20:06 +0800 Subject: [PATCH 5/5] make socket ownership more explicit and future-proof --- src/Inbox.zig | 2 +- src/server/bidi/BiDi.zig | 56 ++++++++++++++++++++++++---------------- src/server/cdp/CDP.zig | 31 +++++++++++++--------- 3 files changed, 54 insertions(+), 35 deletions(-) diff --git a/src/Inbox.zig b/src/Inbox.zig index 699e721c6..0320bd916 100644 --- a/src/Inbox.zig +++ b/src/Inbox.zig @@ -125,7 +125,7 @@ pub const Message = struct { payload: Payload, node: DoublyLinkedList.Node = .{}, - const Payload = union(enum) { + pub const Payload = union(enum) { // A CDP text/binary frame, parsed on the Network thread. `raw` // is the original JSON bytes (owned). `arena` holds any // auxiliary allocations from parseFromSliceLeaky (typically diff --git a/src/server/bidi/BiDi.zig b/src/server/bidi/BiDi.zig index e75ef157f..94d97fa92 100644 --- a/src/server/bidi/BiDi.zig +++ b/src/server/bidi/BiDi.zig @@ -21,12 +21,15 @@ const lp = @import("lightpanda"); const App = @import("../../App.zig"); const Inbox = @import("../../Inbox.zig"); + +const sys_net = @import("../../sys/net.zig"); const uuidv4 = @import("../../id.zig").uuidv4; -const Browser = @import("../../browser/Browser.zig"); -const Session = @import("../../browser/Session.zig"); const Notification = @import("../../Notification.zig"); const NodeRegistry = @import("../../NodeRegistry.zig"); +const Browser = @import("../../browser/Browser.zig"); +const Session = @import("../../browser/Session.zig"); + const Link = @import("../Link.zig"); const Server = @import("../Server.zig"); @@ -113,27 +116,36 @@ const InputMessage = struct { pub fn init(self: *BiDi, app: *App, inbox: *Inbox, origin: Origin) !void { const allocator = app.allocator; - self.* = .{ - .app = app, - .link = null, - .inbox = inbox, - .mode = switch (origin) { - .socket => .bidi_only, - .session => |session| .{ .http = session.worker }, - }, - .browser = undefined, - .user_context = undefined, - .notification = undefined, - .session_id = switch (origin) { - .socket => null, - .session => |session| session.id, - }, - .node_registry = .init(allocator), - .handles = .{ .allocator = allocator }, - .message_arena = std.heap.ArenaAllocator.init(allocator), - .session_arena = std.heap.ArenaAllocator.init(allocator), - }; + { + // this is documentation, and future-proofing, to show exactly where + // the socket's ownership is + errdefer if (origin == .socket) { + sys_net.close(origin.socket); + }; + self.* = .{ + .app = app, + .link = null, + .inbox = inbox, + .mode = switch (origin) { + .socket => .bidi_only, + .session => |session| .{ .http = session.worker }, + }, + .browser = undefined, + .user_context = undefined, + .notification = undefined, + .session_id = switch (origin) { + .socket => null, + .session => |session| session.id, + }, + .node_registry = .init(allocator), + .handles = .{ .allocator = allocator }, + .message_arena = std.heap.ArenaAllocator.init(allocator), + .session_arena = std.heap.ArenaAllocator.init(allocator), + }; + } + + // Link.create takes ownership of the socket switch (origin) { .socket => |socket| self.link = try Link.create(app, socket, .bidi, inbox), .session => {}, diff --git a/src/server/cdp/CDP.zig b/src/server/cdp/CDP.zig index 560e923b6..228fd7aff 100644 --- a/src/server/cdp/CDP.zig +++ b/src/server/cdp/CDP.zig @@ -23,6 +23,7 @@ const App = @import("../../App.zig"); const Inbox = @import("../../Inbox.zig"); const Notification = @import("../../Notification.zig"); +const sys_net = @import("../../sys/net.zig"); const http = @import("../../network/http.zig"); const HttpClient = @import("../../network/HttpClient.zig"); @@ -97,20 +98,26 @@ streams: @import("domains/io.zig").Streams, pub fn init(self: *CDP, app: *App, socket: posix.socket_t, inbox: *Inbox) !void { const allocator = app.allocator; + { + // this is documentation, and future-proofing, to show exactly where + // the socket's ownership is + errdefer sys_net.close(socket); - self.* = .{ - .app = app, - .link = undefined, - .browser = undefined, - .allocator = allocator, - .browser_context = null, - .frame_arena = std.heap.ArenaAllocator.init(allocator), - .message_arena = std.heap.ArenaAllocator.init(allocator), - .notification_arena = std.heap.ArenaAllocator.init(allocator), - .browser_context_arena = std.heap.ArenaAllocator.init(allocator), - .streams = .{ .allocator = allocator }, - }; + self.* = .{ + .app = app, + .link = undefined, + .browser = undefined, + .allocator = allocator, + .browser_context = null, + .frame_arena = std.heap.ArenaAllocator.init(allocator), + .message_arena = std.heap.ArenaAllocator.init(allocator), + .notification_arena = std.heap.ArenaAllocator.init(allocator), + .browser_context_arena = std.heap.ArenaAllocator.init(allocator), + .streams = .{ .allocator = allocator }, + }; + } + // takes ownership of the socket try self.link.init(app, socket, .cdp, inbox); errdefer self.link.deinit();