Move multi to Network: async submission, slot-based completions

Author: mrdimidium

src/browser/HttpClient.zig

@@ -62,8 +62,7 @@ pub const InterceptionLayer = @import("../network/layer/InterceptionLayer.zig");
 // impacting those other http requests.
 pub const Client = @This();
 
-// Our curl multi handle. Owns in_use/ready_queue/http_active/ws_active/
-// performing — see Network.Handle.
+// Our curl multi handle. Owns in_use/http_active/ws_active — see Network.Handle.
 handle: Network.Handle,
 
 // WebSockets with queued events to be drained from the worker thread.
@@ -206,6 +205,27 @@ pub fn init(self: *Client, allocator: Allocator, network: *Network, cdp_client:
 
 pub fn deinit(self: *Client) void {
     self.abort();
+
+    // Cancellations submitted by abort flow through the network thread
+    // and come back as canceled completions. If the network thread has
+    // already stopped, drive its queues ourselves; otherwise spin until
+    // they all arrive and we drained them.
+    var spins: usize = 0;
+    while (self.handle.in_use.first != null and spins < 1000) : (spins += 1) {
+        if (self.network.shutdown.load(.acquire)) {
+            self.network.drainPendingForShutdown();
+        }
+        _ = self.processMessages() catch {};
+        self.drainReadyWs();
+        if (self.handle.in_use.first == null) break;
+        std.Thread.sleep(std.time.ns_per_ms);
+    }
+    if (self.handle.in_use.first != null) {
+        log.warn(.http, "deinit with active conns", .{
+            .count = self.handle.http_active + self.handle.ws_active,
+        });
+    }
+
     self.handle.deinit();
 
     self.ws_ready.deinit(self.allocator);
@@ -260,14 +280,8 @@ pub fn setTlsVerify(self: *Client, verify: bool) !void {
 
     var it = self.handle.in_use.first;
     while (it) |node| : (it = node.next) {
-        const conn: *http.Connection = @fieldParentPtr("node", node);
-        try self.handle.submitTlsVerify(conn, verify, self.use_proxy);
-    }
-
-    it = self.handle.ready_queue.first;
-    while (it) |node| : (it = node.next) {
-        const conn: *http.Connection = @fieldParentPtr("node", node);
-        try self.handle.submitTlsVerify(conn, verify, self.use_proxy);
+        const conn: *http.Connection = @fieldParentPtr("_worker_node", node);
+        self.handle.submitTlsVerify(conn, verify, self.use_proxy);
     }
 
     self.tls_verify = verify;
@@ -315,16 +329,13 @@ pub fn abort(self: *Client) void {
     // each transfer's deinit:
     //   - self.transfers : transfers.remove(self.id)
     //   - self.queue     : unlinked if _queued is set
-    //   - handle.in_use / handle.ready_queue : via submitRemove
-    //   - handle.dirty  : drained at end of each perform
+    //   - handle.in_use : cancellations complete through Handle completions
     // Any non-empty list means a transfer escaped cleanup — assert so we
     // catch the regression rather than silently leaking on next use.
     if (comptime IS_DEBUG) {
         std.debug.assert(self.transfers.size == 0);
         std.debug.assert(self.queue.first == null);
         std.debug.assert(self.handle.in_use.first == null);
-        std.debug.assert(self.handle.ready_queue.first == null);
-        std.debug.assert(self.handle.dirty.first == null);
     }
 }
 
@@ -538,14 +549,9 @@ pub fn syncRequest(self: *Client, allocator: Allocator, req: Request) !SyncRespo
 // cases, the interceptor is expected to call resume to continue the transfer
 // or transfer.abort() to abort it.
 fn process(self: *Client, transfer: *Transfer) !void {
-    // libcurl doesn't allow recursive calls, if we're in a `perform()` operation
-    // then we _have_ to queue this.
-    if (self.handle.performing == false) {
-        if (self.handle.getConnection()) |conn| {
-            return self.makeRequest(conn, transfer);
-        }
+    if (self.handle.getConnection()) |conn| {
+        return self.makeRequest(conn, transfer);
     }
-
     self.queue.append(&transfer._node);
     transfer._queued = true;
     transfer.loop_owned = true;
@@ -623,17 +629,12 @@ pub const PerformStatus = enum {
 };
 
 fn perform(self: *Client, timeout_ms: c_int) anyerror!PerformStatus {
-    // Handle.perform manages its own performing flag and drains
-    // ready_queue after curl_multi_perform returns.
-    const running = try self.handle.perform();
-
-    // Drain queued WebSocket events. ws callbacks (called from libcurl during
-    // perform above) only buffer/queue — actual JS dispatch happens here, on
-    // the worker thread.
+    // The network thread drives the multi; this just drains whatever
+    // it's already pushed (WS events queued by callbacks, completions
+    // delivered through Handle's wake pipe).
     self.drainReadyWs();
 
-    // We're potentially going to block for a while until we get data. Process
-    // whatever messages we have waiting ahead of time.
+    // Process completions we already have before deciding to block.
     if (try self.processMessages()) {
         return .normal;
     }
@@ -649,7 +650,9 @@ fn perform(self: *Client, timeout_ms: c_int) anyerror!PerformStatus {
         if (pollfds[0].revents != 0) {
             status = .cdp_socket;
         }
-    } else if (running > 0) {
+    } else if (self.handle.in_use.first != null) {
+        // Block until the network thread pushes a completion (or
+        // timeout). With nothing in flight there's no reason to wait.
         try self.handle.poll(&.{}, timeout_ms);
     }
 
@@ -1045,14 +1048,14 @@ pub const Transfer = struct {
 
     fn releaseConn(self: *Transfer) void {
         if (self._conn) |conn| {
-            self.client.handle.submitRemove(conn);
+            self.client.handle.finishConn(conn);
             self._conn = null;
         }
     }
 
     pub fn deinit(self: *Transfer) void {
         if (self._conn) |conn| {
-            self.client.handle.submitRemove(conn);
+            self.client.handle.finishConn(conn);
             self._conn = null;
         }
 
@@ -1104,7 +1107,7 @@ pub const Transfer = struct {
     }
 
     // Decide whether to tear down now or defer until processOneMessage
-    // eventually drains the in-flight curl handle.
+    // eventually receives the canceled/completed connection.
     //
     // Two cases force deferral:
     //   * `_performing` — processOneMessage is currently processing THIS
@@ -1113,18 +1116,19 @@ pub const Transfer = struct {
     //     here would double-free. Note that `_conn` is cleared partway
     //     through this window (the "release conn ASAP" step before
     //     done_callback fires), so we cannot rely on `_conn != null`.
-    //   * `client.performing` + we have a libcurl handle — libcurl could
-    //     still fire callbacks for us. Releasing the arena now would UAF
-    //     from inside curl.
+    //   * `_conn != null` — the network thread owns the easy. Submit a
+    //     cancellation and let the completion path call deinit.
     //
     // Otherwise (parked / queued / never-trackConn'd / fully drained),
     // there is nothing left referencing this transfer and we can safely
     // deinit inline even from inside a perform callback.
     fn detachOrDeinit(self: *Transfer) void {
-        const must_defer = self._performing or
-            (self.client.handle.performing and self._conn != null);
-        if (must_defer) {
+        if (self.aborted) return;
+        if (self._performing) {
+            self.detachInPerform();
+        } else if (self._conn) |conn| {
             self.detachInPerform();
+            self.client.handle.submitRemove(conn);
         } else {
             self.deinit();
         }

src/browser/webapi/net/WebSocket.zig

@@ -59,6 +59,7 @@ _http_client: *HttpClient,
 _req_headers: http.Headers,
 
 _owner_node: std.DoublyLinkedList.Node = .{},
+_owner_linked: bool = false,
 
 // buffered outgoing messages
 _send_queue: std.ArrayList(Message) = .empty,
@@ -88,6 +89,11 @@ _pending_close: ?PendingClose = null,
 // WebSocket stays alive between queueing and drain.
 _in_ready_list: bool = false,
 
+// Set while a cancel is in flight. We hold an extra ref so the WS
+// can't be freed before the canceled completion arrives via
+// drainCompletions → disconnected.
+_cancel_pending: bool = false,
+
 // close info for event dispatch
 _close_code: u16 = 1000,
 _close_reason: []const u8 = "",
@@ -185,6 +191,7 @@ pub fn init(url: []const u8, protocols: [][]const u8, frame: *Frame) !*WebSocket
     conn.transport = .{ .websocket = self };
     try http_client.handle.submitRequest(conn);
     frame._http_owner.addWS(self);
+    self._owner_linked = true;
 
     if (comptime IS_DEBUG) {
         log.info(.websocket, "connecting", .{ .url = url });
@@ -199,7 +206,10 @@ pub fn init(url: []const u8, protocols: [][]const u8, frame: *Frame) !*WebSocket
 }
 
 pub fn deinit(self: *WebSocket, page: *Page) void {
-    self.cleanup();
+    // By the time we reach deinit (RC=0) the canceled completion has
+    // already routed through cleanup(true). If conn is still set
+    // here it's an upstream invariant break — defensively finish.
+    self.cleanup(true);
 
     if (self._on_open) |func| {
         func.release();
@@ -235,15 +245,18 @@ fn asEventTarget(self: *WebSocket) *EventTarget {
 
 // we're being aborted internally (e.g. frame shutting down)
 pub fn kill(self: *WebSocket) void {
-    self.cleanup();
+    self._ready_state = .closed;
+    self.cleanup(false);
 }
 
 pub fn disconnected(self: *WebSocket, err_: ?anyerror) void {
-    if (self._ready_state == .closed) {
-        // already disconnected (e.g. close-handshake disconnected us, then
-        // libcurl reports the same connection completion).
-        return;
-    }
+    // Always run terminal cleanup — even on the already-closed early
+    // out, since the canceled completion delivery still needs to run
+    // finishConn and drop the cancel-pending ref.
+    defer self.cleanup(true);
+
+    if (self._ready_state == .closed) return;
+
     const was_clean = self._ready_state == .closing and err_ == null;
     self._ready_state = .closed;
 
@@ -253,29 +266,44 @@ pub fn disconnected(self: *WebSocket, err_: ?anyerror) void {
         log.info(.websocket, "disconnected", .{ .url = self._url, .reason = "closed" });
     }
 
-    // Queue events first (markReady acquires a "pending" ref), then cleanup
-    // (which releases the create-time ref). The pending ref keeps us alive
-    // until drainPending dispatches and releases it.
     self._pending_close = .{
         .code = if (was_clean) self._close_code else 1006,
         .reason = if (was_clean) self._close_reason else "",
         .was_clean = was_clean,
         .with_error = !was_clean,
     };
     self.markReady();
-
-    self.cleanup();
 }
 
-fn cleanup(self: *WebSocket) void {
-    if (self._conn) |conn| {
+// completed=true: terminal — conn already detached (delivered via
+// drainCompletions). Decrements counters via finishConn, releases
+// create-time ref (and cancel-pending ref if held).
+//
+// completed=false: initiate cancel — submitRemove queues the network
+// remove; we hold an extra ref until the canceled completion routes
+// back through disconnected → cleanup(true).
+fn cleanup(self: *WebSocket, completed: bool) void {
+    const conn = self._conn orelse return;
+    if (self._owner_linked) {
         self._frame._http_owner.removeWS(self);
+        self._owner_linked = false;
+    }
+    if (!completed) {
+        if (self._cancel_pending) return;
+        self._cancel_pending = true;
+        self.acquireRef();
         self._http_client.handle.submitRemove(conn);
-        self._req_headers.deinit();
-        self._conn = null;
-        self.releaseRef(self._frame._page);
-        self._send_queue.clearRetainingCapacity();
+        return;
+    }
+    self._http_client.handle.finishConn(conn);
+    self._req_headers.deinit();
+    self._conn = null;
+    self.releaseRef(self._frame._page); // create-time
+    if (self._cancel_pending) {
+        self._cancel_pending = false;
+        self.releaseRef(self._frame._page); // pending-cancel
     }
+    self._send_queue.clearRetainingCapacity();
 }
 
 fn queueMessage(self: *WebSocket, msg: Message) !void {
@@ -285,7 +313,7 @@ fn queueMessage(self: *WebSocket, msg: Message) !void {
     if (was_empty) {
         // Unpause the send callback so libcurl will request data
         if (self._conn) |conn| {
-            try self._http_client.handle.submitUnpause(conn);
+            self._http_client.handle.submitUnpause(conn);
         }
     }
 }
@@ -396,7 +424,7 @@ pub fn close(self: *WebSocket, code_: ?u16, reason_: ?[]const u8) !void {
             .with_error = false,
         };
         self.markReady();
-        self.cleanup();
+        self.cleanup(false);
         return;
     }
 
@@ -742,9 +770,11 @@ fn _receivedDataCallback(conn: *http.Connection, data: []const u8) !void {
                 1005; // No status code received
 
             if (self._ready_state == .closing) {
-                // Client-initiated close: this is the server's response.
-                // Close handshake complete - disconnect.
-                self.disconnected(null);
+                // Client-initiated close — server's response. Don't
+                // disconnect inline: we're inside a libcurl callback
+                // and tearing the conn down here would UAF the easy
+                // handle. Curl will deliver normal completion when the
+                // server closes the socket per RFC 6455 §5.5.1.
             } else {
                 // Server-initiated close: send reciprocal close frame per RFC 6455 §5.5.1
                 self._close_code = received_code;