mirror/browser
1
0 Fork 0
mirror of https://github.com/lightpanda-io/browser.git synced 2026-09-15 23:37:42 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #3401 from lightpanda-io/http-webdriver

Browse Source 
webdriver: HTTP WebDriver session management
This commit is contained in:
Karl Seguin authored and GitHub committed 2026-09-11 05:41:33 +08:00
2 parents 72165ef2a4 2ace9d1548
commit 33c9f3b9fd
14 files changed
+1166 -478
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/Config.zig
+30
View File
@@ -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,15 @@ pub fn maxConnections(self: *const Config) u16 {
};
}
// 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| if (opts.http_session_timeout == 0) null else @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,
@@ -1314,6 +1324,26 @@ test "Config: parseArgs --http-version" {
}
}
test "Config: parseArgs --http-session-timeout" {
// parseArgs allocations live for the process; an arena stands in for main's.
var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
defer arena.deinit();
{
const argv = [_][*:0]const u8{ "lightpanda", "serve" };
const proc_args: std.process.Args = .{ .vector = &argv };
const config = try parseArgs(arena.allocator(), proc_args);
try std.testing.expectEqual(60_000, config.httpSessionTimeout());
}
{
// 0 disables the reaper
const argv = [_][*:0]const u8{ "lightpanda", "serve", "--http-session-timeout", "0" };
const proc_args: std.process.Args = .{ .vector = &argv };
const config = try parseArgs(arena.allocator(), proc_args);
try std.testing.expectEqual(null, config.httpSessionTimeout());
}
}
test "Config: validateUserAgent" {
try validateUserAgent("Lightpanda/1.0");
try std.testing.expectError(error.Reserved, validateUserAgent("mozilla/1.0"));
src/Inbox.zig
+35 -8
View File
@@ -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"),
@@ -118,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
@@ -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");
src/Metrics.zig
+14 -5
View File
@@ -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",
@@ -201,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);
}
};
@@ -226,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)});
}
}
};
src/help.zon
+5
View File
@@ -45,6 +45,11 @@
\\ --host <HOST>
\\ Host of the CDP server.
\\ Defaults to "127.0.0.1".
\\ --http-session-timeout <INT>
\\ Seconds before an idle HTTP session times-out. Only meaningful
\\ when connecting using the WebDriver protocol.
\\ Defaults to 60, disable by setting to 0 (a session then lives
\\ until the driver deletes it).
\\ --port <INT>
\\ Port of the CDP server.
\\ Defaults to 9222.
src/network/HttpClient.zig
+29 -19
View File
@@ -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.
src/server/Driver.zig
+68 -43
View File
@@ -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;
}
src/server/Link.zig
+28 -3
View File
@@ -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,
@@ -64,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 }));
@@ -77,6 +82,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 +94,25 @@ 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 = 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;
}
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 +324,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 +365,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);
src/server/Server.zig
+690 -293
View File
File diff suppressed because it is too large. Load diff
src/server/bidi/BiDi.zig
+122 -23
View File
@@ -1,5 +1,5 @@
// Copyright (C) 2023-2026 Lightpanda (Selecy SAS)
//
// Copyright (C) 2023-2026 Lightpanda (Selecy SAS)
// Francis Bouvier <francis@lightpanda.io>
// Pierre Tachoire <pierre@lightpanda.io>
//
@@ -20,14 +20,18 @@ const std = @import("std");
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 Inbox = @import("../../Inbox.zig");
const Server = @import("../Server.zig");
const script = @import("script.zig");
const remote_value = @import("remote_value.zig");
@@ -38,7 +42,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,32 +91,70 @@ 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,
.browser = undefined,
.user_context = undefined,
.notification = undefined,
.session_id = 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 => {},
}
errdefer if (self.link) |l| l.destroy();
try self.browser.init(app, .{});
errdefer self.browser.deinit();
try self.conn.init(app, socket, .bidi, inbox);
errdefer self.conn.deinit();
self.notification = try Notification.init(allocator);
errdefer self.notification.deinit();
@@ -128,11 +179,55 @@ 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)
if (comptime lp.IS_DEBUG) {
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 {
// 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();
l.destroy();
}
pub fn replaceSession(self: *BiDi, id: []const u8) !void {
self.resetRealm();
try self.newUserContext(id);
@@ -316,6 +411,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, .{});
}
src/server/bidi/browser.zig
+1 -1
View File
@@ -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 };
src/server/bidi/session.zig
+2 -2
View File
@@ -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).
src/server/bidi/testing.zig
+1 -1
View File
@@ -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();
src/server/cdp/CDP.zig
+27 -20
View File
@@ -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");
@@ -59,7 +60,7 @@ pub const InvocationIdGen = Incrementing(u32, "INV");
const CDP = @This();
app: *App,
conn: Link,
link: Link,
browser: Browser,
allocator: Allocator,
@@ -97,24 +98,30 @@ 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,
.conn = 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();
try self.browser.init(app, .{ .env = .{ .with_inspector = true } });
errdefer self.browser.deinit();
try self.conn.init(app, socket, .cdp, inbox);
}
pub fn deinit(self: *CDP) void {
@@ -127,7 +134,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 +169,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 +1151,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 +1177,7 @@ pub const BrowserContext = struct {
std.debug.assert(buf.items.len == message_len);
}
try cdp.conn.sendJSONRaw(buf);
try cdp.link.sendJSONRaw(buf);
}
};
src/server/http.zig
+114 -60
View File
@@ -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,22 @@ 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 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");
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 +486,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 +632,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 +697,39 @@ 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);
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",
.message = "failed to start the session",
.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;
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 +740,55 @@ 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.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 });
}
for (caps.firstMatch orelse &.{}) |first| {
if (first.webSocketUrl == true) {
return true;
}
if (worker.link != null) {
// already joined
return serveHTTPResponse(server, conn, req, .{ .static = session_connected_response });
}
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 +881,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 +914,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;
}