diff --git a/src/Inbox.zig b/src/Inbox.zig index 0320bd916..084cc8be8 100644 --- a/src/Inbox.zig +++ b/src/Inbox.zig @@ -30,6 +30,7 @@ const lp = @import("lightpanda"); const CDP = @import("server/cdp/CDP.zig"); const Link = @import("server/Link.zig"); +const http_command = @import("server/bidi/http_command.zig"); const DoublyLinkedList = std.DoublyLinkedList; @@ -168,10 +169,16 @@ pub const Message = struct { // gets its BiDi connection after the fact). link: *Link, + // An HTTP WebDriver command, parsed on the loop. Its connection is + // parked on the Server.Worker until the consumer responds. + bidi_http: http_command.Command, + pub fn size(self: Payload) usize { return switch (self) { .cdp => |c| c.raw.len, .bidi, .ping => |b| b.len, + // one at a time, it never backs up + .bidi_http => 0, .close, .disconnect, .link, .quit => 0, }; } diff --git a/src/network/HttpClient.zig b/src/network/HttpClient.zig index 59275c0e7..24357b236 100644 --- a/src/network/HttpClient.zig +++ b/src/network/HttpClient.zig @@ -1518,6 +1518,10 @@ fn drainInbox(self: *Client, mode: DrainMode) !void { driver.onLink(link); break :blk false; }, + .bidi_http => |command| blk: { + driver.onHttp(command); + break :blk false; + }, .quit => blk: { driver.onQuit(); break :blk true; // quit always shutsdown @@ -1550,14 +1554,14 @@ fn allowDuringSyncWait(msg: *Inbox.Message) bool { .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. - .bidi => false, + .bidi, .bidi_http => false, }; } fn isTerminal(msg: *Inbox.Message) bool { return switch (msg.payload) { .close, .disconnect, .quit => true, - .ping, .cdp, .bidi, .link => false, + .ping, .cdp, .bidi, .link, .bidi_http => false, }; } @@ -1575,7 +1579,7 @@ fn isFetchInterceptionMethod(method: []const u8) bool { fn isSyncWaitInterrupt(msg: *Inbox.Message) bool { return switch (msg.payload) { .close, .disconnect, .quit => true, - .ping, .link => false, + .ping, .link, .bidi_http => 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 91265569f..d917c759d 100644 --- a/src/server/Driver.zig +++ b/src/server/Driver.zig @@ -27,6 +27,7 @@ const Link = @import("Link.zig"); const CDP = @import("cdp/CDP.zig"); const BiDi = @import("bidi/BiDi.zig"); +const http_command = @import("bidi/http_command.zig"); const log = lp.log; @@ -122,6 +123,18 @@ pub fn onLink(self: *const Driver, l: *Link) void { } } +// Worker Thread. An HTTP WebDriver command; its connection waits on our answer. +pub fn onHttp(self: *const Driver, command: http_command.Command) void { + switch (self.impl) { + .bidi => |bidi| bidi.onHttpCommand(command), + .cdp => { + if (comptime lp.IS_DEBUG) { + lp.assert(false, "Driver.onHttp cdp", .{}); + } + }, + } +} + // 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) diff --git a/src/server/Server.zig b/src/server/Server.zig index fabb212cf..c0b3f6f8a 100644 --- a/src/server/Server.zig +++ b/src/server/Server.zig @@ -31,7 +31,9 @@ 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"); +const http_command = @import("bidi/http_command.zig"); const log = lp.log; const posix = std.posix; @@ -178,7 +180,7 @@ pub fn init(app: *App, address: sys_net.IpAddress) !*Server { .webdriver => protocols.webdriver = true, }; - const request_capacity = 2 * config.maxConnections(); + const request_capacity = 3 * config.maxConnections(); var worker_queue: std.ArrayList(WorkerRequest) = try .initCapacity(allocator, request_capacity); errdefer worker_queue.deinit(allocator); @@ -560,6 +562,43 @@ pub fn quitSession(self: *Server, worker: *Worker) void { } } +// An HTTP command for a session's worker. We need to park the connection, push +// the request to the worker, and park the connection until we get the response +// to send back as the HTTP response. +pub fn parkRequest(self: *Server, worker: *Worker, conn: *Connection, keepalive: bool, arena: *lp.Arena, command: http_command.Command) void { + if (comptime lp.IS_DEBUG) { + lp.assert(worker.http_request == null, "Server.parkRequest busy", .{}); + } + + // stop monitoring the socket + self.detachConnection(conn); + + worker.http_request = .{ .conn = conn, .keepalive = keepalive }; + // a session with a command in flight isn't idle, whatever its link + self.clearIdle(worker); + // the command lives in arena, which the message now owns + worker.inbox.push(arena, .{ .bidi_http = command }); + if (worker.driver) |driver| { + driver.wakeup(); + } +} + +fn deliverResponse(self: *Server, worker: *Worker, response: Connection.Writing.Pooled, now: u64) void { + const parked = worker.http_request orelse { + // the worker answers only what it was sent, once + if (comptime lp.IS_DEBUG) { + lp.assert(false, "Server.deliverResponse unparked", .{}); + } + response.arena.release(); + return; + }; + worker.http_request = null; + http.resumeParked(self, parked.conn, parked.keepalive, .{ .pooled = response }, now); + if (worker.link == null) { + self.markIdle(worker, now); + } +} + // 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) { @@ -589,7 +628,8 @@ fn drainWorkerQueue(self: *Server, now: u64) void { switch (request.op) { .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), + .release => |notify| self.releaseWorker(request.worker, notify, now), + .respond => |response| self.deliverResponse(request.worker, response, now), } } self.worker_drain.clearRetainingCapacity(); @@ -639,6 +679,13 @@ fn markIdle(self: *Server, worker: *Worker, now: u64) void { // ending already (or never a HTTP session) return; } + + if (worker.http_request != null) { + // waiting for a response, not idle, once we [start] to deliver the + // response, then the clock will start ticking again. + return; + } + const timeout = self.session_timeout_ms orelse { // reaping disabled: the session lives until DELETE /session/{id} return; @@ -673,9 +720,14 @@ fn releaseLink(self: *Server, worker: *Worker, notify: *std.Io.Event, now: u64) notify.set(lp.io); } -fn releaseWorker(self: *Server, worker: *Worker, notify: *std.Io.Event) void { +fn releaseWorker(self: *Server, worker: *Worker, notify: *std.Io.Event, now: u64) void { self.unmonitorLink(worker); worker.link = null; + if (worker.http_request) |parked| { + // the worker stopped without answering (DELETE, shutdown, a failed init) + worker.http_request = null; + http.resumeParked(self, parked.conn, false, .{ .static = http.session_ended_response }, now); + } self.releaseWorkerSlot(worker); // The worker is free to deinit its driver and close the fd from here. notify.set(lp.io); @@ -907,7 +959,7 @@ const EPoll = struct { // the low bit set (both are word-aligned, so the bit is free). const WS_TAG: usize = 1; - fn monitorHTTP(self: *const EPoll, conn: *Connection) !void { + pub fn monitorHTTP(self: *const EPoll, conn: *Connection) !void { var event = linux.epoll_event{ .data = .{ .ptr = @intFromPtr(conn) }, .events = READ_EVENTS, @@ -1050,7 +1102,7 @@ const KQueue = struct { return self.change(&.{socketEvent(fd, EVFILT.READ, EV.DELETE, 0)}); } - fn monitorHTTP(self: *const KQueue, conn: *Connection) !void { + pub fn monitorHTTP(self: *const KQueue, conn: *Connection) !void { return self.monitor(conn.socket, EVFILT.READ, @intFromPtr(conn)); } @@ -1214,6 +1266,14 @@ pub const Worker = struct { deadline: ?u64 = null, idle_node: DoublyLinkedList.Node = .{}, + // The HTTP WebDriver command the worker is answering. + http_request: ?ParkedRequest = null, + + const ParkedRequest = struct { + conn: *Connection, + keepalive: bool, + }; + const Pool = struct { slab: []Worker, free: DoublyLinkedList, @@ -1334,6 +1394,12 @@ pub const Worker = struct { notify.waitUncancelable(lp.io); } + // Worker -> loop: the answer to http_request, a complete HTTP response in + // a pooled arena. The loop releases it once it's written. + pub fn respond(self: *Worker, response: Connection.Writing.Pooled) void { + self.notifyLoop(.{ .respond = response }); + } + fn notifyLoop(self: *Worker, op: WorkerRequest.Op) void { const server = self.server; server.worker_mutex.lockUncancelable(lp.io); @@ -1352,6 +1418,7 @@ const WorkerRequest = struct { release: *std.Io.Event, release_link: *std.Io.Event, attach: struct { driver: Driver, link: ?*Link }, + respond: Connection.Writing.Pooled, }; }; @@ -1901,6 +1968,101 @@ test "server: HTTP session bootstrap errors" { try deleteHTTPSession("00000000-0000-4000-8000-000000000000", false); } +test "server: HTTP navigate" { + const session_id = try createHTTPSession("{\"capabilities\":{}}", false); + defer deleteHTTPSession(&session_id, true) catch |err| @panic(@errorName(err)); + + // One keepalive connection throughout: an answered command's connection + // is back on the loop, ready for the next. + var c = try createTestClient(); + defer c.deinit(); + { + const res = try sessionCommand(&c, &session_id, "/url", "not json"); + try testing.expect(std.mem.startsWith(u8, res, "HTTP/1.1 400 Bad Request\r\n")); + try testing.expect(std.mem.endsWith(u8, res, "{\"value\":{\"error\":\"invalid argument\",\"message\":\"invalid body\",\"stacktrace\":\"\"}}")); + } + + const url = "http://127.0.0.1:9582/src/browser/tests/cdp/dom2.html"; + { + const res = try sessionCommand(&c, &session_id, "/url", "{\"url\":\"" ++ url ++ "\"}"); + 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); + } + + // it's the browsing context a websocket on the session sees + var ws = try createTestClient(); + defer ws.deinit(); + var path_buf: [64]u8 = undefined; + try ws.handshake(try std.fmt.bufPrint(&path_buf, "/session/{s}", .{&session_id})); + try ws.bidiCommand("{\"id\":1,\"method\":\"browsingContext.getTree\"}"); + const msg = try ws.readWebsocketMessage() orelse return error.NoMessage; + defer if (msg.cleanup_fragment) ws.reader.cleanup(); + try testing.expect(std.mem.indexOf(u8, msg.data, "\"url\":\"" ++ url ++ "\"") != null); +} + +test "server: HTTP command errors" { + { + var c = try createTestClient(); + defer c.deinit(); + const res = try sessionCommand(&c, "00000000-0000-4000-8000-000000000000", "/url", "{}"); + 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\":\"\"}}")); + } + + const session_id = try createHTTPSession("{\"capabilities\":{}}", false); + + // routing errors are the loop's, in W3C form + { + var c = try createTestClient(); + defer c.deinit(); + var request_buf: [128]u8 = undefined; + const res = try c.httpRequest(try std.fmt.bufPrint(&request_buf, "GET /session/{s}/url HTTP/1.1\r\n\r\n", .{&session_id})); + try testing.expect(std.mem.startsWith(u8, res, "HTTP/1.1 405 Method Not Allowed\r\n")); + try testing.expect(std.mem.endsWith(u8, res, "{\"value\":{\"error\":\"unknown method\",\"message\":\"unknown method\",\"stacktrace\":\"\"}}")); + } + { + var c = try createTestClient(); + defer c.deinit(); + const res = try sessionCommand(&c, &session_id, "/nope", "{}"); + 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\":\"unknown command\",\"message\":\"unknown command\",\"stacktrace\":\"\"}}")); + } + + // a slow page keeps the navigate parked on the worker + var slow = try createTestClient(); + defer slow.deinit(); + try writeSessionCommand(&slow, &session_id, "/url", "{\"url\":\"http://127.0.0.1:9582/src/browser/tests/hi.html?delay_ms=500\"}"); + lp.io.sleep(.fromMilliseconds(50), .awake) catch {}; + + // one command at a time + { + var c = try createTestClient(); + defer c.deinit(); + const res = try sessionCommand(&c, &session_id, "/url", "{\"url\":\"about:blank\"}"); + 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\":\"unknown error\",\"message\":\"a command is already in progress\",\"stacktrace\":\"\"}}")); + } + + // ending the session answers the parked command + try deleteHTTPSession(&session_id, true); + const res = try slow.httpRequest(""); + 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\":\"session ended\",\"stacktrace\":\"\"}}")); +} + +fn sessionCommand(c: *TestClient, session_id: *const [36]u8, command: []const u8, body: []const u8) ![]const u8 { + try writeSessionCommand(c, session_id, command, body); + return c.httpRequest(""); +} + +fn writeSessionCommand(c: *TestClient, session_id: *const [36]u8, command: []const u8, body: []const u8) !void { + var head_buf: [128]u8 = undefined; + try sys_net.writeAll(c.socket, try std.fmt.bufPrint(&head_buf, "POST /session/{s}{s} HTTP/1.1\r\nContent-Length: {d}\r\n\r\n", .{ session_id, command, body.len })); + try sys_net.writeAll(c.socket, body); +} + // 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(); diff --git a/src/server/bidi/BiDi.zig b/src/server/bidi/BiDi.zig index 94d97fa92..44affd688 100644 --- a/src/server/bidi/BiDi.zig +++ b/src/server/bidi/BiDi.zig @@ -30,10 +30,12 @@ const NodeRegistry = @import("../../NodeRegistry.zig"); const Browser = @import("../../browser/Browser.zig"); const Session = @import("../../browser/Session.zig"); +const http = @import("../http.zig"); const Link = @import("../Link.zig"); const Server = @import("../Server.zig"); const script = @import("script.zig"); +const http_command = @import("http_command.zig"); const remote_value = @import("remote_value.zig"); const posix = std.posix; @@ -101,6 +103,11 @@ pub const Mode = union(enum) { http: *Server.Worker, }; +pub const Reply = union(enum) { + bidi: u64, // reply goes to websocket with this id + http: void, // reply is sent as an http response +}; + // 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. @@ -285,26 +292,20 @@ pub fn onMessage(self: *BiDi, data: []const u8) anyerror!void { const id = input.id orelse { return self.sendError(null, "invalid argument", "missing command id"); }; + var cmd: Command = .{ + .bidi = self, + .arena = arena, + .input = .{ .bidi = .{ .id = id, .json = data } }, + }; + const method = input.method orelse { - return self.sendError(id, "invalid argument", "missing command method"); + return cmd.sendError("invalid argument", "missing command method"); }; - lp.metrics.serve_commands.incr(.bidi); - self.dispatch(arena, id, method, data) catch |err| switch (err) { - error.UnknownCommand => { - lp.metrics.serve_unknown_commands.incr(.bidi); - try self.sendError(id, "unknown command", method); - }, - // Command.params already answered the client. - error.InvalidParams => {}, - else => return err, - }; -} -// A BiDi method is always ".". -fn dispatch(self: *BiDi, arena: Allocator, id: u64, method: []const u8, data: []const u8) !void { + // A BiDi method is always ".". const i = std.mem.indexOfScalar(u8, method, '.') orelse { - return error.UnknownCommand; + return unknownCommand(&cmd, method); }; const module = std.meta.stringToEnum(enum { session, @@ -312,53 +313,105 @@ fn dispatch(self: *BiDi, arena: Allocator, id: u64, method: []const u8, data: [] browser, browsingContext, input, - }, method[0..i]) orelse return error.UnknownCommand; + }, method[0..i]) orelse return unknownCommand(&cmd, method); // Only the session module is reachable without a session (it's what // creates one); it gates its own commands. if (self.session_id == null and module != .session) { - return self.sendError(id, "invalid session id", "no active session"); + return cmd.sendError("invalid session id", "no active session"); } - const cmd: Command = .{ - .id = id, - .bidi = self, - .json = data, - .arena = arena, - .action = method[i + 1 ..], + const action = method[i + 1 ..]; + const result = switch (module) { + .session => @import("session.zig").processMessage(&cmd, action), + .script => @import("script.zig").processMessage(&cmd, action), + .browser => @import("browser.zig").processMessage(&cmd, action), + .browsingContext => @import("browsing_context.zig").processMessage(&cmd, action), + .input => @import("input.zig").processMessage(&cmd, action), + }; + result catch |err| { + if (err == error.UnknownCommand and cmd.answered == false) { + return unknownCommand(&cmd, method); + } + cmd.failed(err); }; - - switch (module) { - .session => return @import("session.zig").processMessage(&cmd), - .script => return @import("script.zig").processMessage(&cmd), - .browser => return @import("browser.zig").processMessage(&cmd), - .browsingContext => return @import("browsing_context.zig").processMessage(&cmd), - .input => return @import("input.zig").processMessage(&cmd), - } } -// One command being processed. Handlers answer through it so they don't -// have to thread the id (and the raw message) around. +fn unknownCommand(cmd: *Command, method: []const u8) !void { + lp.metrics.serve_unknown_commands.incr(.bidi); + return cmd.sendError("unknown command", method); +} + +// Dispatch an HTTP WebDriver command from the inbox. Its connection is parked +// until we respond. +pub fn onHttpCommand(self: *BiDi, command: http_command.Command) void { + if (self.mode != .http) { + // the loop only parks commands on an HTTP session's worker + if (comptime lp.IS_DEBUG) { + lp.assert(false, "BiDi.onHttpCommand mode", .{}); + } + return; + } + if (self.browser.env.terminatePending()) { + // the loop answers it once the worker releases + return; + } + + defer _ = self.message_arena.reset(.{ .retain_with_limit = 4096 }); + lp.metrics.serve_commands.incr(.bidi); + + var cmd: Command = .{ + .bidi = self, + .arena = self.message_arena.allocator(), + .input = .{ .http = command }, + }; + http_command.process(&cmd) catch |err| { + cmd.failed(err); + }; +} + +// One command being processed, from either transport. Handlers answer +// through it so they don't have to thread where the answer goes. pub const Command = struct { bidi: *BiDi, + arena: Allocator, // The message_arena; valid for the lifetime of the command. + input: Input, + answered: bool = false, - // The message_arena; valid for the lifetime of the command. - arena: Allocator, + pub const Input = union(enum) { + // A websocket frame: the id is echoed back, and `params` is parsed + // out of the raw json on demand. + bidi: struct { + id: u64, + json: []const u8, + }, - // Echoed back in the response. - id: u64, + // Parsed on the loop, params included. + http: http_command.Command, + }; - // The "" half of ".". - action: []const u8, + pub fn reply(self: *const Command) Reply { + return switch (self.input) { + .bidi => |b| .{ .bidi = b.id }, + .http => .http, + }; + } - // The full raw message; `params` is parsed out of it on demand. - json: []const u8, - - // Parses the command's params object. Answers the client and returns - // error.InvalidParams when the message has no params or they don't - // match T, so callers can just `try`. onMessage swallows that error. - pub fn params(self: *const Command, comptime T: type) !T { - const wrapper = std.json.parseFromSliceLeaky(struct { params: T }, self.arena, self.json, .{ + // Parses the websocket command's params object. Answers the client and + // returns error.InvalidParams when the message has no params or they + // don't match T, so callers can just `try`. `failed` ignores that error. + pub fn params(self: *Command, comptime T: type) !T { + const json = switch (self.input) { + .bidi => |b| b.json, + .http => { + // HTTP handlers get their params from input.http + if (comptime lp.IS_DEBUG) { + lp.assert(false, "Command.params http", .{}); + } + return error.InvalidParams; + }, + }; + const wrapper = std.json.parseFromSliceLeaky(struct { params: T }, self.arena, json, .{ .ignore_unknown_fields = true, }) catch { try self.sendError("invalid argument", "invalid params"); @@ -367,12 +420,22 @@ pub const Command = struct { return wrapper.params; } - pub fn sendResult(self: *const Command, result: anytype) !void { - return self.bidi.sendResult(self.id, result); + // `result` is in the reply's shape: a BiDi result, or HTTP's value. + pub fn sendResult(self: *Command, result: anytype) !void { + self.answered = true; + return self.bidi.replyResult(self.reply(), result); } - pub fn sendError(self: *const Command, code: []const u8, message: []const u8) !void { - return self.bidi.sendError(self.id, code, message); + pub fn sendError(self: *Command, code: []const u8, message: []const u8) !void { + self.answered = true; + return self.bidi.replyError(self.reply(), code, message); + } + + // For an answer sent after the command is gone: whoever holds the reply + // answers it. + pub fn takeReply(self: *Command) Reply { + self.answered = true; + return self.reply(); } // Events aren't tied to the command, but handlers that emit one always @@ -380,6 +443,18 @@ pub const Command = struct { pub fn sendEvent(self: *const Command, method: []const u8, p: anytype) !void { return self.bidi.sendEvent(method, p); } + + fn failed(self: *Command, err: anyerror) void { + if (self.answered) { + if (err != error.InvalidParams) { + // params answers its own error.InvalidParams + lp.log.warn(.bidi, "command failed after answering", .{ .err = err, .reply = self.reply() }); + } + return; + } + lp.log.err(.bidi, "command failed", .{ .err = err, .reply = self.reply() }); + self.sendError("unknown error", @errorName(err)) catch {}; + } }; // An event is delivered when its name is subscribed exactly, or its module @@ -411,6 +486,40 @@ pub fn sendError(self: *BiDi, id: ?u64, code: []const u8, message: []const u8) ! return self.sendJSON(.{ .type = "error", .id = id, .@"error" = code, .message = message }); } +pub fn replyResult(self: *BiDi, reply: Reply, result: anytype) !void { + switch (reply) { + .bidi => |id| return self.sendResult(id, result), + .http => return self.respondHTTP(result), + } +} + +pub fn replyError(self: *BiDi, reply: Reply, code: []const u8, message: []const u8) !void { + switch (reply) { + .bidi => |id| return self.sendError(id, code, message), + .http => return self.respondHTTPStatus(http.webDriverErrorStatus(code), .{ + .@"error" = code, + .message = message, + .stacktrace = "", + }), + } +} + +// {"value": value} to the HTTP request parked on our worker. +pub fn respondHTTP(self: *BiDi, value: anytype) !void { + return self.respondHTTPStatus(.ok, value); +} + +fn respondHTTPStatus(self: *BiDi, status: std.http.Status, value: anytype) !void { + const worker = switch (self.mode) { + .http => |worker| worker, + .bidi_only => return, // an .http reply only comes from onHttpCommand, which checks + }; + const arena = try self.app.arena_pool.acquire(.small, "http response"); + errdefer arena.release(); + const bytes = try http.webDriverResponse(arena, status, value); + worker.respond(.{ .arena = arena, .bytes = bytes }); +} + // 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). diff --git a/src/server/bidi/browser.zig b/src/server/bidi/browser.zig index 8403982ef..ab4f0d715 100644 --- a/src/server/bidi/browser.zig +++ b/src/server/bidi/browser.zig @@ -26,13 +26,13 @@ const browsing_context = @import("browsing_context.zig"); const log = lp.log; -pub fn processMessage(cmd: *const BiDi.Command) !void { +pub fn processMessage(cmd: *BiDi.Command, action: []const u8) !void { const command = std.meta.stringToEnum(enum { close, getUserContexts, createUserContext, removeUserContext, - }, cmd.action) orelse return error.UnknownCommand; + }, action) orelse return error.UnknownCommand; switch (command) { .close => return close(cmd), @@ -42,7 +42,7 @@ pub fn processMessage(cmd: *const BiDi.Command) !void { } } -fn close(cmd: *const BiDi.Command) !void { +fn close(cmd: *BiDi.Command) !void { try cmd.sendResult(struct {}{}); const bidi = cmd.bidi; @@ -52,7 +52,7 @@ fn close(cmd: *const BiDi.Command) !void { const UserContextInfo = struct { userContext: []const u8 }; -fn getUserContexts(cmd: *const BiDi.Command) !void { +fn getUserContexts(cmd: *BiDi.Command) !void { const bidi = cmd.bidi; var infos: [2]UserContextInfo = .{ .{ .userContext = "default" }, undefined }; var len: usize = 1; @@ -63,7 +63,7 @@ fn getUserContexts(cmd: *const BiDi.Command) !void { return cmd.sendResult(.{ .userContexts = infos[0..len] }); } -fn createUserContext(cmd: *const BiDi.Command) !void { +fn createUserContext(cmd: *BiDi.Command) !void { const p = try cmd.params(struct { proxy: ?std.json.Value = null, acceptInsecureCerts: ?bool = null, @@ -92,7 +92,7 @@ fn createUserContext(cmd: *const BiDi.Command) !void { return cmd.sendResult(.{ .userContext = &id }); } -fn removeUserContext(cmd: *const BiDi.Command) !void { +fn removeUserContext(cmd: *BiDi.Command) !void { const p = try cmd.params(struct { userContext: []const u8, }); diff --git a/src/server/bidi/browsing_context.zig b/src/server/bidi/browsing_context.zig index 2bbb0facb..03c0fa847 100644 --- a/src/server/bidi/browsing_context.zig +++ b/src/server/bidi/browsing_context.zig @@ -59,29 +59,31 @@ pub const Context = struct { }; const PendingNavigate = struct { - command_id: u64, - until: enum { interactive, complete }, + until: Until, + reply: BiDi.Reply, + + pub const Until = enum { interactive, complete }; }; -pub fn processMessage(cmd: *const BiDi.Command) !void { +pub fn processMessage(cmd: *BiDi.Command, action: []const u8) !void { const command = std.meta.stringToEnum(enum { getTree, create, navigate, close, locateNodes, - }, cmd.action) orelse return error.UnknownCommand; + }, action) orelse return error.UnknownCommand; switch (command) { .getTree => return getTree(cmd), .create => return create(cmd), - .navigate => return navigate(cmd), + .navigate => return bidiNavigate(cmd), .close => return close(cmd), .locateNodes => return locateNodes(cmd), } } -fn getTree(cmd: *const BiDi.Command) !void { +fn getTree(cmd: *BiDi.Command) !void { const bidi = cmd.bidi; const ctx = &(bidi.browsing_context orelse { return cmd.sendResult(.{ .contexts = &[_]Info{} }); @@ -97,7 +99,7 @@ fn getTree(cmd: *const BiDi.Command) !void { }); } -fn create(cmd: *const BiDi.Command) !void { +fn create(cmd: *BiDi.Command) !void { const p = try cmd.params(struct { type: enum { tab, window }, userContext: ?[]const u8 = null, @@ -116,23 +118,32 @@ fn create(cmd: *const BiDi.Command) !void { return cmd.sendError("no such user context", "unknown user context"); } + const ctx = openContext(bidi) catch |err| switch (err) { + error.CreatePage => return cmd.sendError("unknown error", "failed to create page"), + else => return err, + }; + return cmd.sendResult(.{ .context = &ctx.id }); +} + +// The top-level browsing context, on a fresh about:blank page. +pub fn openContext(bidi: *BiDi) !*Context { const page = bidi.user_context.session.createPage() catch |err| { log.err(.bidi, "create page", .{ .err = err }); - return cmd.sendError("unknown error", "failed to create page"); + return error.CreatePage; }; - var ctx = Context{ + bidi.browsing_context = .{ .id = undefined, .realm_id = undefined, .frame_id = page.frame_id, .navigation_id = undefined, }; + const ctx = &bidi.browsing_context.?; uuidv4(&ctx.id); uuidv4(&ctx.realm_id); uuidv4(&ctx.navigation_id); - bidi.browsing_context = ctx; - try cmd.sendEvent("browsingContext.contextCreated", .{ + try bidi.sendEvent("browsingContext.contextCreated", .{ .context = &ctx.id, .url = "about:blank", .userContext = bidi.user_context.id(), @@ -151,8 +162,7 @@ fn create(cmd: *const BiDi.Command) !void { }; try announceRealm(bidi, frame); } - - return cmd.sendResult(.{ .context = &ctx.id }); + return ctx; } fn announceRealm(bidi: *BiDi, frame: *const Frame) !void { @@ -171,34 +181,44 @@ fn announceRealm(bidi: *BiDi, frame: *const Frame) !void { }); } -fn navigate(cmd: *const BiDi.Command) !void { +fn bidiNavigate(cmd: *BiDi.Command) !void { const p = try cmd.params(struct { url: [:0]const u8, context: []const u8, - wait: enum { none, interactive, complete } = .none, + wait: NavigateOpts.Wait = .none, }); - const bidi = cmd.bidi; const ctx = (try requireContext(cmd, p.context)) orelse return; + return navigate(cmd, ctx, .{ .url = p.url, .wait = p.wait }); +} +pub const NavigateOpts = struct { + url: [:0]const u8, + wait: Wait, + + pub const Wait = enum { none, interactive, complete }; +}; + +pub fn navigate(cmd: *BiDi.Command, ctx: *Context, opts: NavigateOpts) !void { + const bidi = cmd.bidi; const frame = bidi.user_context.session.currentFrame() orelse { return cmd.sendError("unknown error", "no frame"); }; - const encoded_url = URL.resolveNavigation(frame.call_arena, p.url, .{}) catch { + const encoded_url = URL.resolveNavigation(frame.call_arena, opts.url, .{}) catch { return cmd.sendError("invalid argument", "invalid url"); }; // A second navigate supersedes an in-flight one; answer the old command - // so the client isn't left waiting on its id forever. + // so the client isn't left waiting on it forever. try rejectPending(bidi, ctx, "navigation superseded"); // Set before starting: a navigation can reach its wait condition // synchronously (about:blank), which would fire the lifecycle callback // before we got a chance to record the pending command. - switch (p.wait) { + switch (opts.wait) { .none => {}, - .interactive => ctx.pending_navigate = .{ .command_id = cmd.id, .until = .interactive }, - .complete => ctx.pending_navigate = .{ .command_id = cmd.id, .until = .complete }, + .interactive => ctx.pending_navigate = .{ .reply = cmd.takeReply(), .until = .interactive }, + .complete => ctx.pending_navigate = .{ .reply = cmd.takeReply(), .until = .complete }, } // Same fast path as CDP's Page.navigate: a root frame that never @@ -211,16 +231,20 @@ fn navigate(cmd: *const BiDi.Command) !void { nav_result catch |err| { log.warn(.bidi, "navigate", .{ .err = err }); + if (opts.wait != .none and ctx.pending_navigate == null) { + // the lifecycle already answered it + return; + } ctx.pending_navigate = null; return cmd.sendError("unknown error", "navigation failed"); }; - if (p.wait == .none) { + if (opts.wait == .none) { return cmd.sendResult(.{ .navigation = &ctx.navigation_id, .url = encoded_url }); } } -fn close(cmd: *const BiDi.Command) !void { +fn close(cmd: *BiDi.Command) !void { const p = try cmd.params(struct { context: []const u8, }); @@ -231,7 +255,7 @@ fn close(cmd: *const BiDi.Command) !void { } // Closes the page behind `ctx` and reports it gone. -pub fn destroy(cmd: *const BiDi.Command, ctx: *Context) !void { +pub fn destroy(cmd: *BiDi.Command, ctx: *Context) !void { const bidi = cmd.bidi; try rejectPending(bidi, ctx, "browsing context closed"); @@ -259,7 +283,7 @@ pub fn destroy(cmd: *const BiDi.Command, ctx: *Context) !void { }); } -fn locateNodes(cmd: *const BiDi.Command) !void { +fn locateNodes(cmd: *BiDi.Command) !void { const p = try cmd.params(struct { context: []const u8, locator: struct { @@ -375,12 +399,12 @@ fn appendNodes(remotes: *std.ArrayList(remote_value.Remote), arena: std.mem.Allo } } -fn invalidSelector(cmd: *const BiDi.Command, kind: []const u8, selector: []const u8, err: anyerror) !void { +fn invalidSelector(cmd: *BiDi.Command, kind: []const u8, selector: []const u8, err: anyerror) !void { log.debug(.bidi, "locateNodes", .{ .kind = kind, .selector = selector, .err = err }); return cmd.sendError("invalid selector", "invalid selector"); } -pub fn requireContext(cmd: *const BiDi.Command, context: []const u8) !?*Context { +pub fn requireContext(cmd: *BiDi.Command, context: []const u8) !?*Context { if (cmd.bidi.browsing_context) |*ctx| { if (std.mem.eql(u8, &ctx.id, context)) { return ctx; @@ -394,10 +418,14 @@ pub fn requireContext(cmd: *const BiDi.Command, context: []const u8) !?*Context fn answerPending(bidi: *BiDi, ctx: *Context, url: []const u8) !void { const pending = ctx.pending_navigate orelse return; ctx.pending_navigate = null; - return bidi.sendResult(pending.command_id, .{ - .url = url, - .navigation = &ctx.navigation_id, - }); + switch (pending.reply) { + .bidi => |id| return bidi.sendResult(id, .{ + .url = url, + .navigation = &ctx.navigation_id, + }), + // WebDriver's Navigate To answers null + .http => return bidi.respondHTTP(null), + } } // Fails `ctx`'s pending navigate command, if any, and clears it. BiDi has no @@ -406,7 +434,7 @@ fn answerPending(bidi: *BiDi, ctx: *Context, url: []const u8) !void { fn rejectPending(bidi: *BiDi, ctx: *Context, message: []const u8) !void { const pending = ctx.pending_navigate orelse return; ctx.pending_navigate = null; - return bidi.sendError(pending.command_id, "unknown error", message); + return bidi.replyError(pending.reply, "unknown error", message); } pub fn registerNotifications(bidi: *BiDi) !void { @@ -485,7 +513,7 @@ fn onFrameLoaded(ptr: *anyopaque, msg: *const Notification.FrameLoaded) !void { fn frameLifecycleEvent( ptr: *anyopaque, comptime method: []const u8, - reached: @FieldType(PendingNavigate, "until"), + reached: PendingNavigate.Until, frame_id: u32, timestamp: u64, ) !void { diff --git a/src/server/bidi/http_command.zig b/src/server/bidi/http_command.zig new file mode 100644 index 000000000..0f3347b67 --- /dev/null +++ b/src/server/bidi/http_command.zig @@ -0,0 +1,126 @@ +// Copyright (C) 2023-2026 Lightpanda (Selecy SAS) +// +// Francis Bouvier +// Pierre Tachoire +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as +// published by the Free Software Foundation, either version 3 of the +// License, or (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +// The HTTP session's commands, a second entry point onto the BiDi driver. +// The loop parses a request into a Command and parks its connection; the +// worker runs it as a BiDi.Command, which answers the parked request. + +const std = @import("std"); + +const Method = @import("../http.zig").Connection.Method; + +const BiDi = @import("BiDi.zig"); +const browsing_context = @import("browsing_context.zig"); + +const Allocator = std.mem.Allocator; + +pub const Command = union(enum) { + navigate_to: NavigateTo, +}; + +pub const NavigateTo = struct { + url: [:0]const u8, +}; + +const Route = struct { + method: Method, + // what follows /session/{id} + path: []const u8, + command: std.meta.Tag(Command), +}; + +const routes = [_]Route{ + .{ .method = .POST, .path = "/url", .command = .navigate_to }, +}; + +pub const ParseError = error{ + UnknownCommand, + UnknownMethod, + InvalidArgument, + OutOfMemory, +}; + +// Loop. Everything the command references is allocated in `arena`. +pub fn parse(arena: Allocator, method: Method, path: []const u8, body: []const u8) ParseError!Command { + var path_matched = false; + inline for (routes) |route| { + if (std.mem.eql(u8, route.path, path)) { + if (route.method == method) { + const name = @tagName(route.command); + return @unionInit(Command, name, try parseBody(@FieldType(Command, name), arena, body)); + } + path_matched = true; + } + } + if (path_matched) { + return error.UnknownMethod; + } + return error.UnknownCommand; +} + +fn parseBody(comptime T: type, arena: Allocator, body: []const u8) ParseError!T { + return std.json.parseFromSliceLeaky(T, arena, body, .{ + .ignore_unknown_fields = true, + // body is the connection's read buffer, reused once the request is parked + .allocate = .alloc_always, + }) catch |err| switch (err) { + error.OutOfMemory => error.OutOfMemory, + else => error.InvalidArgument, + }; +} + +// Worker. +pub fn process(cmd: *BiDi.Command) !void { + switch (cmd.input.http) { + .navigate_to => |p| return navigateTo(cmd, p), + } +} + +// POST /session/{id}/url. Answers once the page has loaded (the "normal" +// page load strategy). +fn navigateTo(cmd: *BiDi.Command, p: NavigateTo) !void { + const ctx = currentContext(cmd.bidi) catch |err| switch (err) { + error.CreatePage => return cmd.sendError("unknown error", "failed to create page"), + else => return err, + }; + return browsing_context.navigate(cmd, ctx, .{ .url = p.url, .wait = .complete }); +} + +// An HTTP session always has a top-level browsing context; it's opened on +// first use. +fn currentContext(bidi: *BiDi) !*browsing_context.Context { + if (bidi.browsing_context) |*ctx| { + return ctx; + } + return browsing_context.openContext(bidi); +} + +const testing = @import("testing.zig"); +test "bidi.http_command: parse" { + const arena = testing.arena; + + { + const command = try parse(arena, .POST, "/url", "{\"url\":\"about:blank\",\"extra\":1}"); + try testing.expectEqual("about:blank", command.navigate_to.url); + } + + try testing.expectError(error.UnknownMethod, parse(arena, .GET, "/url", "")); + try testing.expectError(error.UnknownCommand, parse(arena, .POST, "/nope", "{}")); + try testing.expectError(error.InvalidArgument, parse(arena, .POST, "/url", "not json")); + try testing.expectError(error.InvalidArgument, parse(arena, .POST, "/url", "{}")); +} diff --git a/src/server/bidi/input.zig b/src/server/bidi/input.zig index b63cfef0a..527beb918 100644 --- a/src/server/bidi/input.zig +++ b/src/server/bidi/input.zig @@ -31,11 +31,11 @@ const log = lp.log; const user_input = Frame.user_input; const Allocator = std.mem.Allocator; -pub fn processMessage(cmd: *const BiDi.Command) !void { +pub fn processMessage(cmd: *BiDi.Command, action: []const u8) !void { const command = std.meta.stringToEnum(enum { performActions, releaseActions, - }, cmd.action) orelse return error.UnknownCommand; + }, action) orelse return error.UnknownCommand; switch (command) { .performActions => return performActions(cmd), @@ -43,7 +43,7 @@ pub fn processMessage(cmd: *const BiDi.Command) !void { } } -fn performActions(cmd: *const BiDi.Command) !void { +fn performActions(cmd: *BiDi.Command) !void { const p = try cmd.params(struct { context: []const u8, actions: []const std.json.Value, @@ -74,14 +74,14 @@ fn performActions(cmd: *const BiDi.Command) !void { .bidi = bidi, .arena = arena, .ticks = ticks, - .command_id = cmd.id, + .reply = cmd.takeReply(), .generation = bidi.input_state.generation, }; return pending.run(); } // Undo everything still held: keys and buttons in reverse press order. -fn releaseActions(cmd: *const BiDi.Command) !void { +fn releaseActions(cmd: *BiDi.Command) !void { const p = try cmd.params(struct { context: []const u8, }); @@ -116,7 +116,7 @@ fn releaseActions(cmd: *const BiDi.Command) !void { return cmd.sendResult(struct {}{}); } -fn dispatchFailed(cmd: *const BiDi.Command, err: DispatchError) !void { +fn dispatchFailed(cmd: *BiDi.Command, err: DispatchError) !void { if (err == error.OutOfMemory) { return err; } @@ -243,7 +243,7 @@ const TickAction = struct { const Pending = struct { bidi: *BiDi, arena: *lp.Arena, - command_id: u64, + reply: BiDi.Reply, generation: u32, ticks: []const []const TickAction, next: usize = 0, @@ -267,11 +267,11 @@ const Pending = struct { const bidi = self.bidi; while (self.next < self.ticks.len) { if (bidi.input_state.generation != self.generation) { - try bidi.sendError(self.command_id, "unknown error", "input state was released while actions were pending"); + try bidi.replyError(self.reply, "unknown error", "input state was released while actions were pending"); return false; } const frame = bidi.user_context.session.currentFrame() orelse { - try bidi.sendError(self.command_id, "no such frame", "no frame"); + try bidi.replyError(self.reply, "no such frame", "no frame"); return false; }; @@ -286,7 +286,7 @@ const Pending = struct { if (err == error.OutOfMemory) { return err; } - try bidi.sendError(self.command_id, errorCode(err), errorMessage(err)); + try bidi.replyError(self.reply, errorCode(err), errorMessage(err)); return false; }; } @@ -297,7 +297,7 @@ const Pending = struct { } } - try bidi.sendResult(self.command_id, struct {}{}); + try bidi.replyResult(self.reply, struct {}{}); return false; } @@ -305,17 +305,17 @@ const Pending = struct { const self: *Pending = @ptrCast(@alignCast(ctx)); // need to capture it here, run can free self - const command_id = self.command_id; + const reply = self.reply; self.run() catch |err| { - log.err(.bidi, "performActions", .{ .err = err, .id = command_id }); + log.err(.bidi, "performActions", .{ .err = err, .reply = reply }); }; return null; } fn cancelled(ctx: *anyopaque) void { const self: *Pending = @ptrCast(@alignCast(ctx)); - self.bidi.sendError(self.command_id, "no such frame", "frame destroyed while actions were pending") catch {}; + self.bidi.replyError(self.reply, "no such frame", "frame destroyed while actions were pending") catch {}; self.deinit(); } }; diff --git a/src/server/bidi/script.zig b/src/server/bidi/script.zig index 69f86c7ab..453816596 100644 --- a/src/server/bidi/script.zig +++ b/src/server/bidi/script.zig @@ -29,12 +29,12 @@ const browsing_context = @import("browsing_context.zig"); const log = lp.log; const Allocator = std.mem.Allocator; -pub fn processMessage(cmd: *const BiDi.Command) !void { +pub fn processMessage(cmd: *BiDi.Command, action: []const u8) !void { const command = std.meta.stringToEnum(enum { evaluate, callFunction, disown, - }, cmd.action) orelse return error.UnknownCommand; + }, action) orelse return error.UnknownCommand; switch (command) { .evaluate => return evaluate(cmd), @@ -53,7 +53,7 @@ const Target = struct { const ResultOwnership = enum { root, none }; -fn evaluate(cmd: *const BiDi.Command) !void { +fn evaluate(cmd: *BiDi.Command) !void { const p = try cmd.params(struct { expression: []const u8, target: Target, @@ -84,7 +84,7 @@ fn evaluate(cmd: *const BiDi.Command) !void { return settle(&reply, frame, &ls.local, value, p.awaitPromise, p.serializationOptions.options(p.resultOwnership == .root)); } -fn callFunction(cmd: *const BiDi.Command) !void { +fn callFunction(cmd: *BiDi.Command) !void { const p = try cmd.params(struct { functionDeclaration: []const u8, target: Target, @@ -139,7 +139,7 @@ fn callFunction(cmd: *const BiDi.Command) !void { return settle(&reply, frame, &ls.local, value, p.awaitPromise, p.serializationOptions.options(p.resultOwnership == .root)); } -fn disown(cmd: *const BiDi.Command) !void { +fn disown(cmd: *BiDi.Command) !void { const p = try cmd.params(struct { handles: []const []const u8, target: Target, @@ -162,14 +162,14 @@ fn disown(cmd: *const BiDi.Command) !void { // reply and one sent from a promise callback after the Command (and its // message_arena) is gone. const Reply = struct { - id: u64, + to: BiDi.Reply, bidi: *BiDi, arena: Allocator, realm_id: [36]u8, - fn init(cmd: *const BiDi.Command, ctx: *const browsing_context.Context) Reply { + fn init(cmd: *BiDi.Command, ctx: *const browsing_context.Context) Reply { return .{ - .id = cmd.id, + .to = cmd.takeReply(), .bidi = cmd.bidi, .arena = cmd.arena, .realm_id = ctx.realm_id, @@ -177,11 +177,11 @@ const Reply = struct { } fn sendResult(self: *const Reply, result: anytype) !void { - return self.bidi.sendResult(self.id, result); + return self.bidi.replyResult(self.to, result); } fn sendError(self: *const Reply, code: []const u8, message: []const u8) !void { - return self.bidi.sendError(self.id, code, message); + return self.bidi.replyError(self.to, code, message); } }; @@ -238,7 +238,7 @@ pub const Pending = struct { .arena = arena, .js_context_id = promise.local.ctx.id, .reply = .{ // clone reply, injecting our Pending's arena - .id = reply.id, + .to = reply.to, .bidi = reply.bidi, .arena = arena.allocator(), .realm_id = reply.realm_id, @@ -290,7 +290,7 @@ pub const Pending = struct { }; result catch |err| { - log.err(.bidi, "await promise", .{ .err = err, .id = reply.id }); + log.err(.bidi, "await promise", .{ .err = err, .reply = reply.to }); }; } @@ -426,7 +426,7 @@ fn serialize( } // returns null when an error response was already sent. -fn toJs(cmd: *const BiDi.Command, local: *const js.Local, value: std.json.Value) !?js.Value { +fn toJs(cmd: *BiDi.Command, local: *const js.Local, value: std.json.Value) !?js.Value { const bidi = cmd.bidi; return remote_value.toJs(local, &bidi.handles, &bidi.node_registry, value) catch |err| switch (err) { error.NoSuchHandle => { @@ -447,7 +447,7 @@ fn toJs(cmd: *const BiDi.Command, local: *const js.Local, value: std.json.Value) // Answers the client and returns null when the target doesn't name our one // context, so callers can `orelse return`. -fn resolveTarget(cmd: *const BiDi.Command, target: Target) !?*browsing_context.Context { +fn resolveTarget(cmd: *BiDi.Command, target: Target) !?*browsing_context.Context { if (target.sandbox != null) { try cmd.sendError("unsupported operation", "sandboxes are not supported"); return null; diff --git a/src/server/bidi/session.zig b/src/server/bidi/session.zig index f9df2b807..3a3f8049c 100644 --- a/src/server/bidi/session.zig +++ b/src/server/bidi/session.zig @@ -24,14 +24,14 @@ const uuidv4 = @import("../../id.zig").uuidv4; const BiDi = @import("BiDi.zig"); -pub fn processMessage(cmd: *const BiDi.Command) !void { +pub fn processMessage(cmd: *BiDi.Command, action: []const u8) !void { const command = std.meta.stringToEnum(enum { status, new, end, subscribe, unsubscribe, - }, cmd.action) orelse return error.UnknownCommand; + }, action) orelse return error.UnknownCommand; // The only two commands that work without a session; `new` is what // creates one. @@ -48,7 +48,7 @@ pub fn processMessage(cmd: *const BiDi.Command) !void { } } -fn status(cmd: *const BiDi.Command) !void { +fn status(cmd: *BiDi.Command) !void { // `ready` reflects whether session.new can succeed on this connection. if (cmd.bidi.session_id == null) { return cmd.sendResult(.{ .ready = true, .message = "" }); @@ -56,7 +56,7 @@ fn status(cmd: *const BiDi.Command) !void { return cmd.sendResult(.{ .ready = false, .message = "session already started" }); } -fn new(cmd: *const BiDi.Command) !void { +fn new(cmd: *BiDi.Command) !void { const bidi = cmd.bidi; if (bidi.session_id != null) { return cmd.sendError("session not created", "session already exists"); @@ -103,7 +103,7 @@ pub const Capabilities = struct { } }; -fn end(cmd: *const BiDi.Command) !void { +fn end(cmd: *BiDi.Command) !void { try cmd.sendResult(struct {}{}); const bidi = cmd.bidi; @@ -113,7 +113,7 @@ fn end(cmd: *const BiDi.Command) !void { // Subscriptions are global (per-context filtering is not supported yet). // Event names aren't validated against a known list. -fn subscribe(cmd: *const BiDi.Command) !void { +fn subscribe(cmd: *BiDi.Command) !void { const p = try cmd.params(struct { events: []const []const u8, }); @@ -135,7 +135,7 @@ fn subscribe(cmd: *const BiDi.Command) !void { return cmd.sendResult(.{ .subscription = &sub_id }); } -fn unsubscribe(cmd: *const BiDi.Command) !void { +fn unsubscribe(cmd: *BiDi.Command) !void { const p = try cmd.params(struct { events: []const []const u8 = &.{}, subscriptions: []const []const u8 = &.{}, diff --git a/src/server/http.zig b/src/server/http.zig index c9fa51d86..51d3447fa 100644 --- a/src/server/http.zig +++ b/src/server/http.zig @@ -27,6 +27,7 @@ const statusCategory = @import("../network/http.zig").statusCategory; const Server = @import("Server.zig"); const Driver = @import("Driver.zig"); const bidi_session = @import("bidi/session.zig"); +const http_command = @import("bidi/http_command.zig"); const uuidv4 = @import("../id.zig").uuidv4; const log = lp.log; @@ -59,11 +60,20 @@ pub const Connection = struct { // lives as long as the server; referenced, never freed static: []const u8, + + // built by a worker in a pooled arena; released once written + pooled: Pooled, + }; + + pub const Pooled = struct { + arena: *lp.Arena, + bytes: []const u8, }; pub fn remaining(self: *const Writing) []const u8 { return switch (self.data) { - inline else => |d| d[self.pos..], + .owned, .static => |d| d[self.pos..], + .pooled => |p| p.bytes[self.pos..], }; } @@ -71,6 +81,7 @@ pub const Connection = struct { switch (self.data) { .static => {}, .owned => |owned| allocator.free(owned), + .pooled => |p| p.arena.release(), } } }; @@ -400,7 +411,8 @@ 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) { + const served = try serveHTTP(server, conn, req); + if (served == .upgraded) { // The fd moved to a WebSocket (and out of server.http); all // that's left of this Connection is to recycle it. // upgradeConnection already took it out of http_connections. @@ -413,6 +425,11 @@ fn processHTTP(server: *Server, conn: *Connection, now: u64) !bool { http.* = .header; conn.buffer.len = 0; + if (served == .parked) { + // off the loop until its worker answers (resumeParked) + return true; + } + if (conn.pending != null) { // We got a WouldBlock and now have a pending write. The // connection stays alive until we flush it. After the write @@ -450,10 +467,18 @@ const empty_json_list_response = staticResponse(.{ .status = "200 OK", .body = " 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" }); +// A parked command whose worker went away before answering. +pub const session_ended_response = staticResponse(.{ + .status = "404 Not Found", + .body = "{\"value\":{\"error\":\"invalid session id\",\"message\":\"session ended\",\"stacktrace\":\"\"}}", + .content_type = "application/json; charset=UTF-8", + .close = true, +}); const Served = enum { responded, upgraded, + parked, }; const Route = struct { @@ -490,8 +515,8 @@ const session_routes = [_]Route{ .{ .method = .DELETE, .path = "", .handler = deleteSession }, }; -// Routes under /session/{id}; path is what follows the id ("" for the -// session itself). The HTTP command surface goes here. +// /session/{id} itself is session_routes; everything under it is a command +// (http_command.parse). const SESSION_PREFIX = "/session/"; const SESSION_ID_LEN = 36; @@ -511,7 +536,10 @@ fn serveHTTP(server: *Server, conn: *Connection, req: *Connection.Request) !Serv return serveNotFound(server, conn, req); } req.session_id = path[SESSION_PREFIX.len..][0..SESSION_ID_LEN]; - return dispatch(server, &session_routes, conn, req, tail); + if (tail.len == 0) { + return dispatch(server, &session_routes, conn, req, tail); + } + return serveSessionCommand(server, conn, req, tail); } return dispatch(server, &routes, conn, req, path); } @@ -588,20 +616,32 @@ fn beginBody(server: *Server) !*std.Io.Writer { return &server.scratch.writer; } -fn serveDynamicHTTPResponse(server: *Server, conn: *Connection, req: *const Connection.Request, comptime status: []const u8, comptime content_type: []const u8) !Served { - const header_format = "HTTP/1.1 " ++ status ++ "\r\n" ++ +fn serveDynamicHTTPResponse(server: *Server, conn: *Connection, req: *const Connection.Request, status: std.http.Status, comptime content_type: []const u8) !Served { + return serveHTTPResponse(server, conn, req, .{ .dynamic = fillHeader(server.scratch.written(), status, content_type) }); +} + +const JSON_CONTENT_TYPE = "application/json; charset=UTF-8"; + +// `buf` is HEADER_RESERVE bytes followed by the body; returns the response, +// its header right-aligned against the body. +fn fillHeader(buf: []u8, status: std.http.Status, comptime content_type: []const u8) []u8 { + const header_format = "HTTP/1.1 {d} {s}\r\n" ++ "Content-Length: {d}\r\n" ++ "Content-Type: " ++ content_type ++ "\r\n\r\n"; - // a usize prints as at most 20 digits - comptime std.debug.assert(header_format.len + 20 <= HEADER_RESERVE); + // 3 status digits, the longest std.http.Status phrase ("Network + // Authentication Required"), a usize's 20 digits + comptime std.debug.assert(header_format.len + 3 + 31 + 20 <= HEADER_RESERVE); - const buf = server.scratch.written(); var header_buf: [HEADER_RESERVE]u8 = undefined; - const header = std.fmt.bufPrint(&header_buf, header_format, .{buf.len - HEADER_RESERVE}) catch unreachable; + const header = std.fmt.bufPrint(&header_buf, header_format, .{ + @intFromEnum(status), + status.phrase() orelse "", + buf.len - HEADER_RESERVE, + }) catch unreachable; const start = HEADER_RESERVE - header.len; @memcpy(buf[start..HEADER_RESERVE], header); - return serveHTTPResponse(server, conn, req, .{ .dynamic = buf[start..] }); + return buf[start..]; } fn errorResponse(comptime status: u16, comptime body: []const u8) []const u8 { @@ -656,7 +696,7 @@ fn serveJSONProtocol(server: *Server, conn: *Connection, req: *Connection.Reques 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"); + return serveDynamicHTTPResponse(server, conn, req, .ok, "text/plain; version=0.0.4; charset=utf-8"); } // GET /status (webdriver) @@ -678,20 +718,12 @@ fn newSession(server: *Server, conn: *Connection, req: *Connection.Request) !Ser firstMatch: ?[]const Capability = null, } = null, }, 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 = "", - }); + return serveWebDriverError(server, conn, req, "invalid argument", "invalid JSON body"); }; 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 = "too many sessions", - .stacktrace = "", - }); + return serveWebDriverError(server, conn, req, "session not created", "too many sessions"); } var session_id: [36]u8 = undefined; @@ -699,11 +731,7 @@ fn newSession(server: *Server, conn: *Connection, req: *Connection.Request) !Ser 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 = "", - }); + return serveWebDriverError(server, conn, req, "session not created", "failed to start the session"); }; // The client never learns the id if we fail to answer (e.g. it hung up), // so nothing would ever DELETE this session. @@ -731,7 +759,7 @@ fn newSession(server: *Server, conn: *Connection, req: *Connection.Request) !Ser break :blk null; }; - return serveWebDriver(server, conn, req, "200 OK", .{ + return serveWebDriver(server, conn, req, .ok, .{ .sessionId = &session_id, .capabilities = bidi_session.Capabilities{ .userAgent = server.app.config.http_headers.user_agent, @@ -763,16 +791,43 @@ fn upgradeSession(server: *Server, conn: *Connection, req: *Connection.Request) // 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 = "", - }); + return serveNoSuchSession(server, conn, req); }; server.quitSession(worker); return serveHTTPResponse(server, conn, req, .{ .static = delete_session_response }); } +// /session/ID/... (webdriver): parsed here, answered by the session's worker. +fn serveSessionCommand(server: *Server, conn: *Connection, req: *Connection.Request, path: []const u8) !Served { + const worker = server.findSession(req.session_id.?) orelse { + return serveNoSuchSession(server, conn, req); + }; + if (worker.http_request != null) { + return serveCommandInProgress(server, conn, req); + } + + const arena = try server.app.arena_pool.acquire(req.body.len, "http command"); + const command = http_command.parse(arena.allocator(), req.method, path, req.body) catch |err| { + arena.release(); + switch (err) { + error.OutOfMemory => return err, + error.UnknownCommand => return serveWebDriverError(server, conn, req, "unknown command", "unknown command"), + error.UnknownMethod => return serveWebDriverError(server, conn, req, "unknown method", "unknown method"), + error.InvalidArgument => return serveWebDriverError(server, conn, req, "invalid argument", "invalid body"), + } + }; + server.parkRequest(worker, conn, req.keepalive, arena, command); + return .parked; +} + +fn serveCommandInProgress(server: *Server, conn: *Connection, req: *const Connection.Request) !Served { + return serveWebDriverError(server, conn, req, "unknown error", "a command is already in progress"); +} + +fn serveNoSuchSession(server: *Server, conn: *Connection, req: *const Connection.Request) !Served { + return serveWebDriverError(server, conn, req, "invalid session id", "no such session"); +} + // 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()) { @@ -783,10 +838,18 @@ fn upgradeSpawn(server: *Server, conn: *Connection, req: *Connection.Request, pr } // 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 { +fn serveWebDriver(server: *Server, conn: *Connection, req: *const Connection.Request, status: std.http.Status, 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"); + return serveDynamicHTTPResponse(server, conn, req, status, JSON_CONTENT_TYPE); +} + +fn serveWebDriverError(server: *Server, conn: *Connection, req: *const Connection.Request, code: []const u8, message: []const u8) !Served { + return serveWebDriver(server, conn, req, webDriverErrorStatus(code), .{ + .@"error" = code, + .message = message, + .stacktrace = "", + }); } fn serveNotFound(server: *Server, conn: *Connection, req: *Connection.Request) !Served { @@ -823,6 +886,81 @@ fn serveHTTPResponse(server: *Server, conn: *Connection, req: *const Connection. return .responded; } +// A parked connection's response is ready, join the loop so that we can start +// writing the response. +pub fn resumeParked(server: *Server, conn: *Connection, req_keepalive: bool, response: Connection.Writing.Data, now: u64) void { + const allocator = server.app.allocator; + // beginShutdown closed the http connections, this one was off the loop then + const keepalive = req_keepalive and server.shutdown_begun == false; + var writing: Connection.Writing = .{ .pos = 0, .data = response, .keepalive = keepalive }; + + server.io_engine.monitorHTTP(conn) catch |err| { + log.err(.serve, "resume monitor", .{ .err = err }); + writing.deinit(allocator); + sys_net.close(conn.socket); + return recycle(server, conn); + }; + conn.deadline = now + IDLE_TIMEOUT_MS; + server.http_connections.append(&conn.node); + + const data = writing.remaining(); + recordResponse(data); + writing.pos = write(conn.socket, data) catch |err| { + log.debug(.serve, "resume write", .{ .err = err }); + writing.deinit(allocator); + return disconnect(server, conn); + }; + + if (writing.pos < data.len) { + // disconnect frees it from here + conn.pending = writing; + server.io_engine.waitWritable(conn) catch |err| { + log.err(.serve, "wait writable", .{ .err = err }); + return disconnect(server, conn); + }; + return; + } + + writing.deinit(allocator); + if (keepalive == false) { + disconnect(server, conn); + } +} + +// A complete {"value": value} response, built by a worker for resumeParked, +// in an arena the loop releases once it's written. +pub fn webDriverResponse(arena: *lp.Arena, status: std.http.Status, value: anytype) ![]const u8 { + var aw: std.Io.Writer.Allocating = try .initCapacity(arena.allocator(), 512); + try aw.writer.splatByteAll(0, HEADER_RESERVE); + try std.json.Stringify.value(.{ .value = value }, .{}, &aw.writer); + return fillHeader(aw.written(), status, JSON_CONTENT_TYPE); +} + +// W3C WebDriver's error table; every code not listed is a 500. +pub fn webDriverErrorStatus(code: []const u8) std.http.Status { + const statuses = std.StaticStringMap(std.http.Status).initComptime(.{ + .{ "detached shadow root", .not_found }, + .{ "element click intercepted", .bad_request }, + .{ "element not interactable", .bad_request }, + .{ "insecure certificate", .bad_request }, + .{ "invalid argument", .bad_request }, + .{ "invalid cookie domain", .bad_request }, + .{ "invalid element state", .bad_request }, + .{ "invalid selector", .bad_request }, + .{ "invalid session id", .not_found }, + .{ "no such alert", .not_found }, + .{ "no such cookie", .not_found }, + .{ "no such element", .not_found }, + .{ "no such frame", .not_found }, + .{ "no such shadow root", .not_found }, + .{ "no such window", .not_found }, + .{ "stale element reference", .not_found }, + .{ "unknown command", .not_found }, + .{ "unknown method", .method_not_allowed }, + }); + return statuses.get(code) orelse .internal_server_error; +} + // HTTP-phase teardown. Websockets tear down via releaseWorker. pub fn disconnect(server: *Server, conn: *Connection) void { server.io_engine.remove(conn.socket);