From 3e3825108ddd367643f66aca50cd8e63fe4a2914 Mon Sep 17 00:00:00 2001 From: Muki Kiboigo Date: Sat, 18 Jul 2026 11:33:22 -0700 Subject: [PATCH] seperate SingleFlight from RobotsGate --- src/network/HttpClient.zig | 19 +- src/network/RobotsGate.zig | 155 +++++++--------- src/network/SingleFlight.zig | 346 +++++++++++++++++++++++++++++++++++ 3 files changed, 424 insertions(+), 96 deletions(-) create mode 100644 src/network/SingleFlight.zig diff --git a/src/network/HttpClient.zig b/src/network/HttpClient.zig index 543aa891b..501979085 100644 --- a/src/network/HttpClient.zig +++ b/src/network/HttpClient.zig @@ -226,7 +226,7 @@ pub fn init(self: *Client, allocator: Allocator, network: *Network, cdp: ?*CDP) .serve_mode = network.config.mode == .serve, .obey_robots = network.config.obeyRobots(), - .robots = .{ .allocator = allocator, .network = network }, + .robots = .{ .allocator = allocator, .network = network, .single_flight = .{ .allocator = allocator } }, .url_blocklist = url_blocklist, .arena_pool = &network.app.arena_pool, }; @@ -431,7 +431,7 @@ pub fn abort(self: *Client) void { std.debug.assert(self.ws_dispatch_queue.first == null); // - self.robots.pending : each robots fetch's shutdown_callback // drops its entry; parked waiters unlink in their own deinit. - std.debug.assert(self.robots.pending.count() == 0); + std.debug.assert(self.robots.single_flight.count() == 0); } } @@ -3204,7 +3204,7 @@ fn initTestClient(client: *Client, pool: *ArenaPool) void { client.cache = null; client.serve_mode = false; client.obey_robots = false; - client.robots = .{ .allocator = testing.allocator, .network = undefined }; + client.robots = .{ .allocator = testing.allocator, .network = undefined, .single_flight = .{ .allocator = testing.allocator }}; client.url_blocklist = null; } @@ -3336,6 +3336,7 @@ test "HttpClient: aborting a robots-parked transfer unlinks it from the gate" { defer client.transfers.deinit(testing.allocator); defer client.robots.deinit(); + const pending = &client.robots.single_flight.pending; const robots_url = "http://example.com/robots.txt"; var waiting: std.ArrayList(*Transfer) = .empty; @@ -3364,18 +3365,18 @@ test "HttpClient: aborting a robots-parked transfer unlinks it from the gate" { try waiting.append(testing.allocator, transfer); transfer.park(.robots); } - try client.robots.pending.putNoClobber(testing.allocator, robots_url, waiting); + try pending.putNoClobber(testing.allocator, robots_url, waiting); - const t1 = client.robots.pending.get(robots_url).?.items[0]; - const t2 = client.robots.pending.get(robots_url).?.items[1]; + const t1 = pending.get(robots_url).?.items[0]; + const t2 = pending.get(robots_url).?.items[1]; t1.abort(error.Abort); - try testing.expectEqual(1, client.robots.pending.get(robots_url).?.items.len); - try testing.expect(client.robots.pending.get(robots_url).?.items[0] == t2); + try testing.expectEqual(1, pending.get(robots_url).?.items.len); + try testing.expect(pending.get(robots_url).?.items[0] == t2); try testing.expectEqual(1, client.transfers.count()); t2.abort(error.Abort); - try testing.expectEqual(0, client.robots.pending.get(robots_url).?.items.len); + try testing.expectEqual(0, pending.get(robots_url).?.items.len); try testing.expectEqual(0, client.transfers.count()); } diff --git a/src/network/RobotsGate.zig b/src/network/RobotsGate.zig index fea5fae90..d15aba1af 100644 --- a/src/network/RobotsGate.zig +++ b/src/network/RobotsGate.zig @@ -22,33 +22,29 @@ // fetch, and resumes (or fails) the parked transfers when it resolves. const std = @import("std"); -const lp = @import("lightpanda"); - -const URL = @import("../browser/URL.zig"); -const ArenaPool = @import("../ArenaPool.zig"); - -const http = @import("http.zig"); -const Robots = @import("Robots.zig"); -const Network = @import("Network.zig"); -const Transfer = @import("HttpClient.zig").Transfer; - -const log = lp.log; const Allocator = std.mem.Allocator; +const lp = @import("lightpanda"); +const log = lp.log; + +const ArenaPool = @import("../ArenaPool.zig"); +const URL = @import("../browser/URL.zig"); +const http = @import("http.zig"); +const Network = @import("Network.zig"); +const Robots = @import("Robots.zig"); +const SingleFlight = @import("SingleFlight.zig"); +const Transfer = @import("HttpClient.zig").Transfer; + const RobotsGate = @This(); network: *Network, allocator: Allocator, -pending: std.StringHashMapUnmanaged(std.ArrayList(*Transfer)) = .empty, +single_flight: SingleFlight, pub const Result = enum { allowed, blocked, pending }; pub fn deinit(self: *RobotsGate) void { - var it = self.pending.iterator(); - while (it.next()) |entry| { - entry.value_ptr.deinit(self.allocator); - } - self.pending.deinit(self.allocator); + self.single_flight.deinit(); } pub fn check(self: *RobotsGate, transfer: *Transfer) !Result { @@ -77,94 +73,80 @@ pub fn check(self: *RobotsGate, transfer: *Transfer) !Result { // stays: the in-flight fetch owns it (the key lives on the fetch's context // arena) and still resolves the remaining waiters. pub fn remove(self: *RobotsGate, transfer: *Transfer) void { - var it = self.pending.valueIterator(); - while (it.next()) |waiting| { - for (waiting.items, 0..) |t, i| { - if (t == transfer) { - _ = waiting.swapRemove(i); - return; - } - } - } + self.single_flight.remove(transfer); } fn fetchThenResume(self: *RobotsGate, robots_url: [:0]const u8, transfer: *Transfer) !void { - if (self.pending.getPtr(robots_url)) |waiting| { - // A fetch for this robots.txt is already in flight, queue behind it. - try waiting.append(self.allocator, transfer); - transfer.park(.robots); - return; - } - const client = transfer.client; - - // The context, the response buffer and the pending-map key live on - // their own pooled arena, NOT on transfer.arena — any waiter (this one - // included) can be aborted while the fetch is still in flight, and the - // fetch's callbacks must survive that. The arena is released by - // whichever terminal callback fires (done / error / shutdown). const arena = try client.arena_pool.acquire(.small, "RobotsGate.RobotsContext"); errdefer client.arena_pool.release(arena); const owned_url = try arena.dupeZ(u8, robots_url); - const robots_ctx = try arena.create(RobotsContext); - robots_ctx.* = .{ - .gate = self, - .buffer = .empty, - .arena = arena, - .arena_pool = client.arena_pool, - .robots_url = owned_url, - }; + const res = try self.single_flight.enter(robots_url, transfer, .robots); + switch (res) { + .queued => { + // joined inflight fetch so release it. + client.arena_pool.release(arena); + return; + }, + .initial => { + errdefer self.single_flight.abort(robots_url); - var waiting: std.ArrayList(*Transfer) = .empty; - try waiting.append(self.allocator, transfer); - errdefer waiting.deinit(self.allocator); + // The context, the response buffer and the pending-map key live on + // their own pooled arena, NOT on transfer.arena — any waiter (this one + // included) can be aborted while the fetch is still in flight, and the + // fetch's callbacks must survive that. The arena is released by + // whichever terminal callback fires (done / error / shutdown). + const robots_ctx = try arena.create(RobotsContext); + robots_ctx.* = .{ + .gate = self, + .buffer = .empty, + .arena = arena, + .arena_pool = client.arena_pool, + .robots_url = owned_url, + }; - try self.pending.putNoClobber(self.allocator, owned_url, waiting); - errdefer _ = self.pending.remove(owned_url); + log.debug(.browser, "fetching robots.txt", .{ .robots_url = owned_url }); - transfer.park(.robots); - errdefer transfer.unpark(); + // Only the parent's frame/loader ids (CDP correlation) and notification + // carry over — no cookies, credentials, headers, or timeout. + const fetch_transfer = try client.newRequest(.{ + .url = owned_url, + .method = .GET, + .internal = true, + .resource_type = .fetch, + .frame_id = transfer.req.frame_id, + .loader_id = transfer.req.loader_id, + .notification = transfer.req.notification, + .cookie_jar = null, + .cookie_origin = owned_url, + .ctx = robots_ctx, + .header_callback = RobotsContext.headerCallback, + .data_callback = RobotsContext.dataCallback, + .done_callback = RobotsContext.doneCallback, + .error_callback = RobotsContext.errorCallback, + .shutdown_callback = RobotsContext.shutdownCallback, + }, null); - log.debug(.browser, "fetching robots.txt", .{ .robots_url = owned_url }); - - // Only the parent's frame/loader ids (CDP correlation) and notification - // carry over — no cookies, credentials, headers, or timeout. - const fetch_transfer = try client.newRequest(.{ - .url = owned_url, - .method = .GET, - .internal = true, - .resource_type = .fetch, - .frame_id = transfer.req.frame_id, - .loader_id = transfer.req.loader_id, - .notification = transfer.req.notification, - .cookie_jar = null, - .cookie_origin = owned_url, - .ctx = robots_ctx, - .header_callback = RobotsContext.headerCallback, - .data_callback = RobotsContext.dataCallback, - .done_callback = RobotsContext.doneCallback, - .error_callback = RobotsContext.errorCallback, - .shutdown_callback = RobotsContext.shutdownCallback, - }, null); - - // From here the fetch owns the pending entry and the context arena. If - // submit fails it fires error_callback — possibly synchronously, right - // here — which resolves the waiters (fail-open, may already have resumed - // `transfer`) and releases the arena. So there is nothing to unwind - // locally and the errdefers above must not run: swallow the error. - fetch_transfer.submit() catch {}; + // From here the fetch owns the pending entry and the context arena. If + // submit fails it fires error_callback — possibly synchronously, right + // here — which resolves the waiters (fail-open, may already have resumed + // `transfer`) and releases the arena. So there is nothing to unwind + // locally and the errdefers above must not run: swallow the error. + fetch_transfer.submit() catch {}; + }, + } } // The robots.txt fetch resolved: hand every waiter back to the pipeline, // each judged against its own path. No store entry (fetch failed, or a 200 // whose body never got parsed) fails open. fn flushPending(self: *RobotsGate, robots_url: []const u8) void { - var queued = self.pending.fetchRemove(robots_url) orelse return; - defer queued.value.deinit(self.allocator); + var queued = self.single_flight.take(robots_url) orelse return; + defer queued.deinit(self.allocator); const robot_entry = self.network.robot_store.get(robots_url); - for (queued.value.items) |transfer| { + for (queued.items) |transfer| { transfer.unpark(); const allowed = if (robot_entry) |entry| switch (entry) { @@ -193,8 +175,7 @@ fn flushPending(self: *RobotsGate, robots_url: []const u8) void { // where every waiter is being kill()'d by the same loop; their deinit // finds no gate entry left (or unlinks itself first) and no-ops. fn flushPendingShutdown(self: *RobotsGate, robots_url: []const u8) void { - var pending = self.pending.fetchRemove(robots_url) orelse return; - pending.value.deinit(self.allocator); + self.single_flight.discard(robots_url); } const RobotsContext = struct { diff --git a/src/network/SingleFlight.zig b/src/network/SingleFlight.zig new file mode 100644 index 000000000..a13311950 --- /dev/null +++ b/src/network/SingleFlight.zig @@ -0,0 +1,346 @@ +// 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 . + +const std = @import("std"); +const Transfer = @import("HttpClient.zig").Transfer; +const Allocator = std.mem.Allocator; + +const SingleFlight = @This(); + +allocator: Allocator, +pending: std.StringHashMapUnmanaged(std.ArrayList(*Transfer)) = .empty, + +pub fn deinit(self: *SingleFlight) void { + var it = self.pending.iterator(); + while (it.next()) |entry| { + entry.value_ptr.deinit(self.allocator); + } + self.pending.deinit(self.allocator); +} + +pub const EnterResult = enum { initial, queued }; + +pub fn enter(self: *SingleFlight, key: []const u8, transfer: *Transfer, reason: Transfer.ParkedBy) !EnterResult { + const gop = try self.pending.getOrPut(self.allocator, key); + var waiting = gop.value_ptr; + + if (gop.found_existing) { + try waiting.append(self.allocator, transfer); + transfer.park(reason); + return .queued; + } + + waiting.* = .empty; + try waiting.append(self.allocator, transfer); + errdefer waiting.deinit(self.allocator); + transfer.park(reason); + + return .initial; +} + +pub fn abort(self: *SingleFlight, key: []const u8) void { + var entry = self.pending.fetchRemove(key) orelse return; + entry.value.deinit(self.allocator); +} + +pub fn remove(self: *SingleFlight, transfer: *Transfer) void { + var it = self.pending.valueIterator(); + while (it.next()) |waiting| { + for (waiting.items, 0..) |t, i| { + if (t == transfer) { + _ = waiting.swapRemove(i); + return; + } + } + } +} + +pub fn take(self: *SingleFlight, key: []const u8) ?std.ArrayList(*Transfer) { + const entry = self.pending.fetchRemove(key) orelse return null; + return entry.value; +} + +pub fn discard(self: *SingleFlight, key: []const u8) void { + var entry = self.pending.fetchRemove(key) orelse return; + entry.value.deinit(self.allocator); +} + +pub fn count(self: *SingleFlight) u32 { + return self.pending.count(); +} + +const testing = @import("../testing.zig"); +const ArenaPool = @import("../ArenaPool.zig"); +const HttpClient = @import("HttpClient.zig"); + +fn makeTestTransfer(arena: Allocator, client: *HttpClient, id: u32) !*Transfer { + const t = try arena.create(Transfer); + t.* = .{ + .arena = arena, + .owner = null, + .req = .{ + .frame_id = 0, + .loader_id = 0, + .method = .GET, + .url = "http://example.com/", + .cookie_jar = null, + .cookie_origin = "", + .resource_type = .document, + .notification = undefined, + .shutdown_callback = HttpClient.noopShutdown, + }, + .client = client, + .id = id, + .start_time = 0, + }; + return t; +} + +test "SingleFlight: enter returns initial for the first waiter" { + var pool = ArenaPool.init(testing.allocator, .{}); + defer pool.deinit(); + + var client: HttpClient = undefined; + // Only transfers.remove/pending_queue.remove/etc. touched by deinit + // matter here; a minimal zeroed client is enough since these tests + // never call transfer.deinit(), only single_flight directly. + client = undefined; + client.transfers = .empty; + client.intercepted = 0; + + var sf = SingleFlight{ .allocator = testing.allocator }; + defer sf.deinit(); + + const arena = try pool.acquire(.small, "test"); + defer pool.release(arena); + + const t1 = try makeTestTransfer(arena, &client, 1); + const t2 = try makeTestTransfer(arena, &client, 2); + const t3 = try makeTestTransfer(arena, &client, 3); + + try testing.expectEqual(.initial, try sf.enter("key", t1, .robots)); + try testing.expectEqual(.queued, try sf.enter("key", t2, .robots)); + try testing.expectEqual(.queued, try sf.enter("key", t3, .robots)); + + try testing.expectEqual(1, sf.count()); + try testing.expectEqual(Transfer.State{ .parked = .robots }, t1.state); + try testing.expectEqual(Transfer.State{ .parked = .robots }, t2.state); + try testing.expectEqual(Transfer.State{ .parked = .robots }, t3.state); +} + +test "SingleFlight: different keys get independent entries" { + var pool = ArenaPool.init(testing.allocator, .{}); + defer pool.deinit(); + + var client: HttpClient = undefined; + client.transfers = .empty; + client.intercepted = 0; + + var sf = SingleFlight{ .allocator = testing.allocator }; + defer sf.deinit(); + + const arena = try pool.acquire(.small, "test"); + defer pool.release(arena); + + const t1 = try makeTestTransfer(arena, &client, 1); + const t2 = try makeTestTransfer(arena, &client, 2); + + try testing.expectEqual(.initial, try sf.enter("key-a", t1, .robots)); + try testing.expectEqual(.initial, try sf.enter("key-b", t2, .robots)); + + try testing.expectEqual(2, sf.count()); +} + +test "SingleFlight: take removes and returns the waiter list" { + var pool = ArenaPool.init(testing.allocator, .{}); + defer pool.deinit(); + + var client: HttpClient = undefined; + client.transfers = .empty; + client.intercepted = 0; + + var sf = SingleFlight{ .allocator = testing.allocator }; + defer sf.deinit(); + + const arena = try pool.acquire(.small, "test"); + defer pool.release(arena); + + const t1 = try makeTestTransfer(arena, &client, 1); + const t2 = try makeTestTransfer(arena, &client, 2); + + _ = try sf.enter("key", t1, .robots); + _ = try sf.enter("key", t2, .robots); + + var waiting = sf.take("key") orelse return error.TestUnexpectedResult; + defer waiting.deinit(testing.allocator); + + try testing.expectEqual(2, waiting.items.len); + try testing.expect(waiting.items[0] == t1); + try testing.expect(waiting.items[1] == t2); + + // Entry is gone: a second take on the same key finds nothing. + try testing.expectEqual(null, sf.take("key")); + try testing.expectEqual(0, sf.count()); +} + +test "SingleFlight: take on an unknown key returns null" { + var sf = SingleFlight{ .allocator = testing.allocator }; + defer sf.deinit(); + + try testing.expectEqual(null, sf.take("missing")); +} + +test "SingleFlight: abort drops the pending entry without resolving waiters" { + var pool = ArenaPool.init(testing.allocator, .{}); + defer pool.deinit(); + + var client: HttpClient = undefined; + client.transfers = .empty; + client.intercepted = 0; + + var sf = SingleFlight{ .allocator = testing.allocator }; + defer sf.deinit(); + + const arena = try pool.acquire(.small, "test"); + defer pool.release(arena); + + const t1 = try makeTestTransfer(arena, &client, 1); + _ = try sf.enter("key", t1, .robots); + + sf.abort("key"); + + try testing.expectEqual(0, sf.count()); + try testing.expectEqual(null, sf.take("key")); +} + +test "SingleFlight: discard is equivalent to abort for the shutdown path" { + var pool = ArenaPool.init(testing.allocator, .{}); + defer pool.deinit(); + + var client: HttpClient = undefined; + client.transfers = .empty; + client.intercepted = 0; + + var sf = SingleFlight{ .allocator = testing.allocator }; + defer sf.deinit(); + + const arena = try pool.acquire(.small, "test"); + defer pool.release(arena); + + const t1 = try makeTestTransfer(arena, &client, 1); + const t2 = try makeTestTransfer(arena, &client, 2); + _ = try sf.enter("key", t1, .robots); + _ = try sf.enter("key", t2, .robots); + + sf.discard("key"); + + try testing.expectEqual(0, sf.count()); +} + +test "SingleFlight: remove unlinks a single waiter from its key's list" { + // Regression-style, mirrors "aborting a robots-parked transfer unlinks + // it from the gate" but exercised directly against SingleFlight rather + // than through RobotsGate. + var pool = ArenaPool.init(testing.allocator, .{}); + defer pool.deinit(); + + var client: HttpClient = undefined; + client.transfers = .empty; + client.intercepted = 0; + + var sf = SingleFlight{ .allocator = testing.allocator }; + defer sf.deinit(); + + const arena = try pool.acquire(.small, "test"); + defer pool.release(arena); + + const t1 = try makeTestTransfer(arena, &client, 1); + const t2 = try makeTestTransfer(arena, &client, 2); + const t3 = try makeTestTransfer(arena, &client, 3); + + _ = try sf.enter("key", t1, .robots); + _ = try sf.enter("key", t2, .robots); + _ = try sf.enter("key", t3, .robots); + + sf.remove(t2); + + var waiting = sf.take("key") orelse return error.TestUnexpectedResult; + defer waiting.deinit(testing.allocator); + + try testing.expectEqual(2, waiting.items.len); + try testing.expect(waiting.items[0] == t1); + try testing.expect(waiting.items[1] == t3); +} + +test "SingleFlight: remove on a transfer not in any list is a no-op" { + var pool = ArenaPool.init(testing.allocator, .{}); + defer pool.deinit(); + + var client: HttpClient = undefined; + client.transfers = .empty; + client.intercepted = 0; + + var sf = SingleFlight{ .allocator = testing.allocator }; + defer sf.deinit(); + + const arena = try pool.acquire(.small, "test"); + defer pool.release(arena); + + const t1 = try makeTestTransfer(arena, &client, 1); + const stray = try makeTestTransfer(arena, &client, 2); + + _ = try sf.enter("key", t1, .robots); + + // stray was never entered anywhere; remove must not touch t1's entry. + sf.remove(stray); + + try testing.expectEqual(1, sf.count()); + var waiting = sf.take("key") orelse return error.TestUnexpectedResult; + defer waiting.deinit(testing.allocator); + try testing.expectEqual(1, waiting.items.len); +} + +test "SingleFlight: removing every waiter for a key leaves an empty (but present) list" { + // remove() only swapRemoves from the waiter list; it does not delete the + // map entry even if the list becomes empty. take()/abort()/discard() are + // the only ways the entry itself disappears. This documents that + // asymmetry so a future change doesn't accidentally break RobotsGate's + // "entry stays, in-flight fetch still owns it" invariant. + var pool = ArenaPool.init(testing.allocator, .{}); + defer pool.deinit(); + + var client: HttpClient = undefined; + client.transfers = .empty; + client.intercepted = 0; + + var sf = SingleFlight{ .allocator = testing.allocator }; + defer sf.deinit(); + + const arena = try pool.acquire(.small, "test"); + defer pool.release(arena); + + const t1 = try makeTestTransfer(arena, &client, 1); + _ = try sf.enter("key", t1, .robots); + + sf.remove(t1); + + try testing.expectEqual(1, sf.count()); + var waiting = sf.take("key") orelse return error.TestUnexpectedResult; + defer waiting.deinit(testing.allocator); + try testing.expectEqual(0, waiting.items.len); +}