Reduce allocations made during request interception

Stream (to json) the Transfer as a request and response object in the various
network interception-related events (e.g. Network.responseReceived).

Add a page.request_intercepted boolean flag for CDP to signal the page that
requests have been intercepted, allowing Page.wait to prioritize intercept
handling (or, at least, not block it).

Author: karlseguin

src/browser/page.zig

@@ -94,6 +94,16 @@ pub const Page = struct {
 
     load_state: LoadState = .parsing,
 
+    // Page.wait balances waiting for resources / tasks and producing an output.
+    // Up until a timeout, Page.wait will always wait for inflight or pending
+    // HTTP requests, via the Http.Client.active counter. However, intercepted
+    // requests (via CDP, but it could be anything), aren't considered "active"
+    // connection. So it's possible that we have intercepted requests (which are
+    // pending on some driver to continue/abort) while Http.Client.active == 0.
+    // This boolean exists to supplment Http.Client.active and inform Page.wait
+    // of pending connections.
+    request_intercepted: bool = false,
+
     const Mode = union(enum) {
         pre: void,
         err: anyerror,
@@ -275,16 +285,26 @@ pub const Page = struct {
         while (true) {
             SW: switch (self.mode) {
                 .pre, .raw => {
+                    if (self.request_intercepted) {
+                        // the page request was intercepted.
+
+                        // there shouldn't be any active requests;
+                        std.debug.assert(http_client.active == 0);
+
+                        // nothing we can do for this, need to kick the can up
+                        // the chain and wait for activity (e.g. a CDP message)
+                        // to unblock this.
+                        return;
+                    }
+
                     // The main page hasn't started/finished navigating.
                     // There's no JS to run, and no reason to run the scheduler.
-
                     if (http_client.active == 0) {
                         // haven't started navigating, I guess.
                         return;
                     }
 
                     // There should only be 1 active http transfer, the main page
-                    std.debug.assert(http_client.active == 1);
                     try http_client.tick(ms_remaining);
                 },
                 .html, .parsed => {
@@ -330,17 +350,29 @@ pub const Page = struct {
 
                     _ = try scheduler.runLowPriority();
 
-                    // We'll block here, waiting for network IO. We know
-                    // when the next timeout is scheduled, and we know how long
-                    // the caller wants to wait for, so we can pick a good wait
-                    // duration
-                    const ms_to_wait = @min(ms_remaining, ms_to_next_task orelse 1000);
+                    const request_intercepted = self.request_intercepted;
+
+                    // We want to prioritize processing intercepted requests
+                    // because, the sooner they get unblocked, the sooner we
+                    // can start the HTTP request. But we still want to advanced
+                    // existing HTTP requests, if possible. So, if we have
+                    // intercepted requests, we'll still look at existing HTTP
+                    // requests, but we won't block waiting for more data.
+                    const ms_to_wait =
+                        if (request_intercepted) 0
+
+                        // But if we have no intercepted requests, we'll wait
+                        // for as long as we can for data to our existing
+                        // inflight requests
+                        else @min(ms_remaining, ms_to_next_task orelse 1000);
+
                     try http_client.tick(ms_to_wait);
 
-                    if (try_catch.hasCaught()) {
-                        const msg = (try try_catch.err(self.arena)) orelse "unknown";
-                        log.warn(.user_script, "page wait", .{ .err = msg, .src = "data" });
-                        return error.JsError;
+                    if (request_intercepted) {
+                        // Again, proritizing intercepted requests. Exit this
+                        // loop so that our caller can hopefully resolve them
+                        // (i.e. continue or abort them);
+                        return;
                     }
                 },
                 .err => |err| return err,

src/cdp/cdp.zig

@@ -495,7 +495,7 @@ pub fn BrowserContext(comptime CDP_T: type) type {
         pub fn onHttpRequestIntercept(ctx: *anyopaque, data: *const Notification.RequestIntercept) !void {
             const self: *Self = @alignCast(@ptrCast(ctx));
             defer self.resetNotificationArena();
-            try @import("domains/fetch.zig").requestPaused(self.notification_arena, self, data);
+            try @import("domains/fetch.zig").requestIntercept(self.notification_arena, self, data);
         }
 
         pub fn onHttpRequestFail(ctx: *anyopaque, data: *const Notification.RequestFail) !void {

src/cdp/domains/fetch.zig

@@ -18,10 +18,13 @@
 
 const std = @import("std");
 const Allocator = std.mem.Allocator;
-const Notification = @import("../../notification.zig").Notification;
+
 const log = @import("../../log.zig");
+const network = @import("network.zig");
+
 const Method = @import("../../http/Client.zig").Method;
 const Transfer = @import("../../http/Client.zig").Transfer;
+const Notification = @import("../../notification.zig").Notification;
 
 pub fn processMessage(cmd: anytype) !void {
     const action = std.meta.stringToEnum(enum {
@@ -41,22 +44,38 @@ pub fn processMessage(cmd: anytype) !void {
 
 // Stored in CDP
 pub const InterceptState = struct {
-    const Self = @This();
-    waiting: std.AutoArrayHashMap(u64, *Transfer),
+    allocator: Allocator,
+    waiting: std.AutoArrayHashMapUnmanaged(u64, *Transfer),
 
     pub fn init(allocator: Allocator) !InterceptState {
         return .{
-            .waiting = std.AutoArrayHashMap(u64, *Transfer).init(allocator),
+            .waiting = .empty,
+            .allocator = allocator,
         };
     }
 
-    pub fn deinit(self: *Self) void {
-        self.waiting.deinit();
+    pub fn empty(self: *const InterceptState) bool {
+        return self.waiting.count() == 0;
+    }
+
+    pub fn put(self: *InterceptState, transfer: *Transfer) !void {
+        return self.waiting.put(self.allocator, transfer.id, transfer);
+    }
+
+    pub fn remove(self: *InterceptState, id: u64) ?*Transfer {
+        const entry = self.waiting.fetchSwapRemove(id) orelse return null;
+        return entry.value;
+    }
+
+    pub fn deinit(self: *InterceptState) void {
+        self.waiting.deinit(self.allocator);
     }
 };
 
 const RequestPattern = struct {
-    urlPattern: []const u8 = "*", // Wildcards ('*' -> zero or more, '?' -> exactly one) are allowed. Escape character is backslash. Omitting is equivalent to "*".
+    // Wildcards ('*' -> zero or more, '?' -> exactly one) are allowed.
+    // Escape character is backslash. Omitting is equivalent to "*".
+    urlPattern: []const u8 = "*",
     resourceType: ?ResourceType = null,
     requestStage: RequestStage = .Request,
 };
@@ -115,60 +134,46 @@ fn disable(cmd: anytype) !void {
 
 fn enable(cmd: anytype) !void {
     const params = (try cmd.params(EnableParam)) orelse EnableParam{};
-    if (params.patterns.len != 0) log.warn(.cdp, "Fetch.enable No patterns yet", .{});
-    if (params.handleAuthRequests) log.warn(.cdp, "Fetch.enable No auth yet", .{});
+    if (params.patterns.len != 0) {
+        log.warn(.cdp, "not implemented", .{ .feature = "Fetch.enable No patterns yet" });
+    }
+    if (params.handleAuthRequests) {
+        log.warn(.cdp, "not implemented", .{ .feature = "Fetch.enable No auth yet" });
+    }
 
     const bc = cmd.browser_context orelse return error.BrowserContextNotLoaded;
     try bc.fetchEnable();
 
     return cmd.sendResult(null, .{});
 }
 
-pub fn requestPaused(arena: Allocator, bc: anytype, intercept: *const Notification.RequestIntercept) !void {
+pub fn requestIntercept(arena: Allocator, bc: anytype, intercept: *const Notification.RequestIntercept) !void {
     var cdp = bc.cdp;
 
     // unreachable because we _have_ to have a page.
     const session_id = bc.session_id orelse unreachable;
     const target_id = bc.target_id orelse unreachable;
+    const page = bc.session.currentPage() orelse unreachable;
 
     // We keep it around to wait for modifications to the request.
     // NOTE: we assume whomever created the request created it with a lifetime of the Page.
     // TODO: What to do when receiving replies for a previous page's requests?
 
     const transfer = intercept.transfer;
-    try cdp.intercept_state.waiting.put(transfer.id, transfer);
-
-    // NOTE: .request data preparation is duped from network.zig
-    const full_request_url = transfer.uri;
-    const request_url = try @import("network.zig").urlToString(arena, &full_request_url, .{
-        .scheme = true,
-        .authentication = true,
-        .authority = true,
-        .path = true,
-        .query = true,
-    });
-    const request_fragment = try @import("network.zig").urlToString(arena, &full_request_url, .{
-        .fragment = true,
-    });
-    const headers = try transfer.req.headers.asHashMap(arena);
-    // End of duped code
+    try cdp.intercept_state.put(transfer);
 
     try cdp.sendEvent("Fetch.requestPaused", .{
         .requestId = try std.fmt.allocPrint(arena, "INTERCEPT-{d}", .{transfer.id}),
-        .request = .{
-            .url = request_url,
-            .urlFragment = request_fragment,
-            .method = @tagName(transfer.req.method),
-            .hasPostData = transfer.req.body != null,
-            .headers = std.json.ArrayHashMap([]const u8){ .map = headers },
-        },
+        .request = network.TransferAsRequestWriter.init(transfer),
         .frameId = target_id,
         .resourceType = ResourceType.Document, //  TODO!
         .networkId = try std.fmt.allocPrint(arena, "REQ-{d}", .{transfer.id}),
     }, .{ .session_id = session_id });
 
     // Await either continueRequest, failRequest or fulfillRequest
+
     intercept.wait_for_interception.* = true;
+    page.request_intercepted = true;
 }
 
 const HeaderEntry = struct {
@@ -186,46 +191,67 @@ fn continueRequest(cmd: anytype) !void {
         headers: ?[]const HeaderEntry = null,
         interceptResponse: bool = false,
     })) orelse return error.InvalidParams;
-    if (params.postData != null or params.headers != null or params.interceptResponse) return error.NotYetImplementedParams;
 
+    if (params.postData != null or params.headers != null or params.interceptResponse) {
+        return error.NotYetImplementedParams;
+    }
+
+    const page = bc.session.currentPage() orelse return error.PageNotLoaded;
+
+    var intercept_state = &bc.cdp.intercept_state;
     const request_id = try idFromRequestId(params.requestId);
-    const entry = bc.cdp.intercept_state.waiting.fetchSwapRemove(request_id) orelse return error.RequestNotFound;
-    const transfer = entry.value;
+    const transfer = intercept_state.remove(request_id) orelse return error.RequestNotFound;
 
     // Update the request with the new parameters
     if (params.url) |url| {
-        // The request url must be modified in a way that's not observable by page. So page.url is not updated.
-        try transfer.updateURL(try bc.cdp.browser.page_arena.allocator().dupeZ(u8, url));
+        // The request url must be modified in a way that's not observable by page.
+        // So page.url is not updated.
+        try transfer.updateURL(try page.arena.dupeZ(u8, url));
     }
     if (params.method) |method| {
         transfer.req.method = std.meta.stringToEnum(Method, method) orelse return error.InvalidParams;
     }
 
-    log.info(.cdp, "Request continued by intercept", .{ .id = params.requestId });
+    log.info(.cdp, "Request continued by intercept", .{
+        .id = params.requestId,
+        .url = transfer.uri,
+    });
     try bc.cdp.browser.http_client.process(transfer);
 
+    if (intercept_state.empty()) {
+        page.request_intercepted = false;
+    }
+
     return cmd.sendResult(null, .{});
 }
 
 fn failRequest(cmd: anytype) !void {
     const bc = cmd.browser_context orelse return error.BrowserContextNotLoaded;
-    var state = &bc.cdp.intercept_state;
     const params = (try cmd.params(struct {
         requestId: []const u8, // "INTERCEPT-{d}"
         errorReason: ErrorReason,
     })) orelse return error.InvalidParams;
 
+    const page = bc.session.currentPage() orelse return error.PageNotLoaded;
+
+    var intercept_state = &bc.cdp.intercept_state;
     const request_id = try idFromRequestId(params.requestId);
-    const entry = state.waiting.fetchSwapRemove(request_id) orelse return error.RequestNotFound;
-    // entry.value is the transfer
-    entry.value.abort();
+
+    const transfer = intercept_state.remove(request_id) orelse return error.RequestNotFound;
+    transfer.abort();
+
+    if (intercept_state.empty()) {
+        page.request_intercepted = false;
+    }
 
     log.info(.cdp, "Request aborted by intercept", .{ .reason = params.errorReason });
     return cmd.sendResult(null, .{});
 }
 
 // Get u64 from requestId which is formatted as: "INTERCEPT-{d}"
 fn idFromRequestId(request_id: []const u8) !u64 {
-    if (!std.mem.startsWith(u8, request_id, "INTERCEPT-")) return error.InvalidParams;
+    if (!std.mem.startsWith(u8, request_id, "INTERCEPT-")) {
+        return error.InvalidParams;
+    }
     return std.fmt.parseInt(u64, request_id[10..], 10) catch return error.InvalidParams;
 }