diff --git a/src/Config.zig b/src/Config.zig index b9da8cfa3..88d9f70b1 100644 --- a/src/Config.zig +++ b/src/Config.zig @@ -218,6 +218,8 @@ const CommonOptions = .{ .{ .name = "http_proxy", .type = ?[:0]const u8 }, .{ .name = "http_max_concurrent", .type = ?u8 }, .{ .name = "http_max_host_open", .type = ?u8 }, + .{ .name = "http_nav_delay", .type = ?u32 }, + .{ .name = "http_nav_burst", .type = ?u32 }, .{ .name = "http_timeout", .type = ?u31 }, .{ .name = "http_connect_timeout", .type = ?u31 }, .{ .name = "http_header", .type = HttpHeader, .multiple = true, .validator = httpHeaderValidator }, @@ -574,6 +576,22 @@ pub fn httpMaxHostOpen(self: *const Config) u8 { }; } +pub fn httpNavDelay(self: *const Config) ?u32 { + const ms = switch (self.mode) { + inline .serve, .fetch, .mcp, .agent => |opts| opts.http_nav_delay, + else => unreachable, + } orelse return null; + return if (ms == 0) null else ms; +} + +pub fn httpNavBurst(self: *const Config) u32 { + const burst = switch (self.mode) { + inline .serve, .fetch, .mcp, .agent => |opts| opts.http_nav_burst, + else => unreachable, + } orelse 1; + return @max(burst, 1); +} + pub fn httpConnectTimeout(self: *const Config) u31 { return switch (self.mode) { inline .serve, .fetch, .mcp, .agent => |opts| opts.http_connect_timeout orelse 0, diff --git a/src/Metrics.zig b/src/Metrics.zig index d06fd9b17..be178323b 100644 --- a/src/Metrics.zig +++ b/src/Metrics.zig @@ -73,6 +73,18 @@ http_response_size_bytes: Histogram(&.{ 2 * 1024 * 1024, 4 * 1024 * 1024, }) = .{}, +http_navigation_delay_ms: Histogram(&.{ + 10, + 50, + 100, + 250, + 500, + 1000, + 2500, + 5000, + 10000, + 30000, +}) = .{}, robots_status: CounterEnum("category", @import("network/http.zig").StatusCategory) = .{}, robots_access: CounterEnum("result", enum { allow, deny }) = .{}, @@ -101,6 +113,7 @@ const help = .{ .http_redirects = "HTTP redirect hops followed", .http_duration_ms = "HTTP request wall-clock duration in milliseconds", .http_response_size_bytes = "HTTP response body size in bytes", + .http_navigation_delay_ms = "Time in milliseconds a throttled top-level navigation waited", .robots_status = "robots.txt response status", .robots_access = "robots.txt result", }; diff --git a/src/browser/Frame.zig b/src/browser/Frame.zig index 1ff57df9d..2a3b254e1 100644 --- a/src/browser/Frame.zig +++ b/src/browser/Frame.zig @@ -827,6 +827,7 @@ pub fn navigate(self: *Frame, request_url: [:0]const u8, opts: NavigateOpts) !vo // don't cache top-level pages, most cases won't revisit this, and, if they // do, they probably don't want the cached version. .skip_cache = self.parent == null, + .throttle = self.parent == null, .cookie_jar = &session.cookie_jar, .cookie_origin = opts.initiator_url orelse self.url, .resource_type = .document, diff --git a/src/browser/Runner.zig b/src/browser/Runner.zig index 25ab29217..6f3fbff38 100644 --- a/src/browser/Runner.zig +++ b/src/browser/Runner.zig @@ -583,3 +583,35 @@ test "Runner: idle notifications advance past a resolved condition" { try testing.expectEqual(true, frame._notified_network_idle == .done); try testing.expectEqual(true, frame._notified_network_almost_idle == .done); } + +test "Runner: waits out a throttled navigation" { + const session = testing.test_session; + const http_client = &session.browser.http_client; + + // Enable the per-host navigation throttle for this test only, and spend + // 127.0.0.1's slot so the navigation below has to wait ~300ms. + const network = http_client.network; + network.rate_limiter = @import("../network/RateLimiter.zig").init(testing.allocator, 300, 1); + defer { + network.rate_limiter.?.deinit(); + network.rate_limiter = null; + } + const start = lp.datetime.milliTimestamp(.boot); + _ = try network.rate_limiter.?.reserve("127.0.0.1", start); + + const page = try session.createPage(); + defer page.close(); + try page.navigate("http://127.0.0.1:9582/src/browser/tests/runner/runner1.html", .{}); + try testing.expectEqual(1, http_client.delayed_count); + // A delayed navigation is in-flight work: the wait must not resolve early. + try testing.expectEqual(false, http_client.activity().idle()); + + var runner = session.runner(.{}); + try runner.waitForFrame(page.frame_id, 2000, .{ .until = .done }); + const elapsed = lp.datetime.milliTimestamp(.boot) - start; + try testing.expectEqual(true, elapsed >= 250); + try testing.expectEqual(0, http_client.delayed_count); + + const el = try runner.waitForSelector(page.frame_id, "#sel1", 10); + try testing.expectEqual("selector-1-content", try el.asNode().getTextContentAlloc(testing.arena_allocator)); +} diff --git a/src/help.zon b/src/help.zon index 4675d1318..04623acbf 100644 --- a/src/help.zon +++ b/src/help.zon @@ -377,6 +377,15 @@ \\ Limits the acceptable response size for any request \\ e.g. XHR, fetch, script loading. \\ Defaults to 1 GiB. + \\ --http-nav-burst + \\ Number of top-level navigations to an idle host allowed to + \\ start without waiting for --http-nav-delay. After a burst, + \\ navigations are spaced by --http-nav-delay again. + \\ Defaults to 1. + \\ --http-nav-delay + \\ Minimum time in ms between two top-level navigations to the same + \\ host (see --http-nav-burst). Disable by setting to 0. + \\ Defaults to 0. \\ --http-proxy \\ HTTP proxy for all HTTP requests. \\ username:password may be included for basic auth. diff --git a/src/network/HttpClient.zig b/src/network/HttpClient.zig index ec8808806..74007669c 100644 --- a/src/network/HttpClient.zig +++ b/src/network/HttpClient.zig @@ -89,6 +89,10 @@ in_use: std.DoublyLinkedList = .{}, // Queue for request that are waiting an available connection (aka, easy) pending_queue: std.DoublyLinkedList = .{}, +// Transfers waiting out the per-host navigation throttle, ordered by _run_at. +delayed_queue: std.DoublyLinkedList = .{}, +delayed_count: usize = 0, + // Queue for completed transfers that haven't had their callbacks executed yet dispatch_queue: std.DoublyLinkedList = .{}, @@ -443,6 +447,7 @@ pub fn abort(self: *Client) void { if (comptime lp.IS_DEBUG) { std.debug.assert(self.transfers.size == 0); std.debug.assert(self.pending_queue.first == null); + std.debug.assert(self.delayed_queue.first == null); std.debug.assert(self.dispatch_queue.first == null); std.debug.assert(self.gated_queue.first == null); std.debug.assert(self.in_use.first == null); @@ -526,7 +531,7 @@ pub const Activity = struct { pub fn activity(self: *const Client) Activity { return .{ - .http = self.http_active + self.dispatch_count + self.intercepted, + .http = self.http_active + self.dispatch_count + self.intercepted + self.delayed_count, .ws_events = self.ws_dispatch_count, .ws_conns = self.ws_active, .pending = self.pending_queue.first != null, @@ -658,12 +663,12 @@ pub fn _tick(self: *Client, timeout_ms: u32, mode: DrainMode) !bool { if (dispatched == false and processed == false and self.dispatch_queue.first == null and self.ws_dispatch_queue.first == null) { // Nothing was dispatched, no messages were processed and nothing is // waiting for dispatch. We need to wait for I/O. - if (running > 0 or self.cdp_link_active) { + if (running > 0 or self.cdp_link_active or self.delayed_queue.first != null) { { self.heartbeat.enterWait(); defer self.heartbeat.exitWait(); // The network layer will wake this up if there's acticity. - try self.handles.poll(&.{}, @intCast(timeout_ms)); + try self.handles.poll(&.{}, @intCast(self.clampToDelayed(timeout_ms))); } // poll only waits, so we do the perform -> process dance again _ = try self.handles.perform(); @@ -694,6 +699,7 @@ pub fn _tick(self: *Client, timeout_ms: u32, mode: DrainMode) !bool { // doing some work (e.g. running tasks). Let's assert that we were // right in doing that, else we'll likely introduce latency. std.debug.assert(self.pending_queue.first == null); + std.debug.assert(self.delayed_queue.first == null); std.debug.assert(self.dispatch_queue.first == null); std.debug.assert(self.ws_dispatch_queue.first == null); } @@ -702,6 +708,17 @@ pub fn _tick(self: *Client, timeout_ms: u32, mode: DrainMode) !bool { return waited; } +// Never sleep past the next delayed transfer's start time. +fn clampToDelayed(self: *const Client, timeout_ms: u32) u32 { + const node = self.delayed_queue.first orelse return timeout_ms; + const transfer: *const Transfer = @fieldParentPtr("_node", node); + const now = lp.datetime.milliTimestamp(.boot); + if (transfer._run_at <= now) { + return 0; + } + return @intCast(@min(timeout_ms, transfer._run_at - now)); +} + // Deliver completed response. This is the ONLY place user callbacks run, // so a callback is free to start new requests, abort transfers, or tear down // its frame. The client is never inside libcurl here. @@ -810,6 +827,7 @@ fn isGated(self: *const Client, transfer: *const Transfer) bool { } fn startPending(self: *Client) !void { + try self.startDelayed(); while (self.pending_queue.popFirst()) |queue_node| { const transfer: *Transfer = @fieldParentPtr("_node", queue_node); const conn = self.network.getConnection() orelse { @@ -832,6 +850,57 @@ fn startPending(self: *Client) !void { } } +// Enter the pipeline for every delayed transfer whose time has come. +fn startDelayed(self: *Client) !void { + if (self.delayed_queue.first == null) { + return; + } + const now = lp.datetime.milliTimestamp(.boot); + while (self.delayed_queue.first) |node| { + const transfer: *Transfer = @fieldParentPtr("_node", node); + if (transfer._run_at > now) { + // these are added to the queue in order, so we can exit as soon as + // we hit the first future _run_at. + return; + } + self.delayed_count -= 1; + self.delayed_queue.remove(node); + + transfer.state = .created; + self.pipeline(transfer, .start) catch |err| { + // Same as startPending: this can run from a tick(.sync_wait), and + // error_callback JS must not fire on a blocking request's stack. + if (transfer.state == .created) { + transfer.failAsync(err); + } + return err; + }; + } +} + +// Hold the transfer out of the pipeline until `run_at` (ms, boot clock). +fn delay(self: *Client, transfer: *Transfer, run_at: u64) void { + transfer._run_at = run_at; + transfer.state = .delayed; + self.delayed_count += 1; + + // Ordered insert; reservations for one host are monotonic, so this is + // usually an append. + var node = self.delayed_queue.last; + while (node) |n| { + const other: *const Transfer = @fieldParentPtr("_node", n); + if (other._run_at <= run_at) { + break; + } + node = n.prev; + } + if (node) |n| { + self.delayed_queue.insertAfter(n, &transfer._node); + } else { + self.delayed_queue.prepend(&transfer._node); + } +} + const SubmitFrom = enum { start, after_intercept, network }; // Process a transfer, passing it through our pipeline. A transfer an move off @@ -1721,6 +1790,9 @@ pub const Request = struct { // these do not need to be deferred and do not obey robots.txt. internal: bool = false, + // Whether this request should (possibly) be throttled based on the RateLimiter + throttle: bool = false, + // Set by syncRequest; only used to label the http_requests metric. sync: bool = false, @@ -1973,6 +2045,9 @@ pub const Transfer = struct { start_time: u64, + // Earliest start time (ms, boot clock) while .delayed. + _run_at: u64 = 0, + _notified_fail: bool = false, // Set when conn is temporarily detached from transfer during redirect @@ -1991,9 +2066,10 @@ pub const Transfer = struct { // need to restore (and hence capture) the original headers. _intercept_original_headers: ?[]const RequestHeader = null, - // Linked into client.pending_queue while .queued; reused to link the - // retired transfer into client.graveyard (deinit unlinks it from the - // pending queue first, so the node is always free by then). + // Linked into client.pending_queue while .queued and client.delayed_queue + // while .delayed; reused to link the retired transfer into + // client.graveyard (deinit unlinks it from those queues first, so the + // node is always free by then). _node: std.DoublyLinkedList.Node = .{}, // Buffered response ordered events awaiting dispatch. @@ -2044,6 +2120,11 @@ pub const Transfer = struct { // linked into client.queue. queued, + // On client.delayed_queue, waiting for its per-host navigation slot + // (`_run_at`) before entering the pipeline. `_node` is linked into + // client.delayed_queue. + delayed, + // Response events are buffered on `_events`, waiting for dispatch // to deliver them. `_queue_node` is linked into // client.dispatch_queue. No conn is held. @@ -2144,6 +2225,23 @@ pub const Transfer = struct { return; } + if (self.req.throttle) { + if (self.client.network.rate_limiter) |*rl| { + const now = lp.datetime.milliTimestamp(.boot); + const run_at = rl.reserve(URL.getHostname(self.req.url), now) catch |err| { + self.abortPipelineError(err); + return err; + }; + if (run_at > now) { + const d = run_at - now; + lp.metrics.http_navigation_delay_ms.observe(@intCast(d)); + log.debug(.http, "navigation delayed", .{ .url = self.req.url, .ms = d }); + self.client.delay(self, run_at); + return; + } + } + } + self.client.pipeline(self, .start) catch |err| { self.abortPipelineError(err); return err; @@ -2166,9 +2264,13 @@ pub const Transfer = struct { self._conn = null; } - // Unlink from client.pending_queue if we were waiting for a handle. + // Unlink from client.pending_queue if we were waiting for a handle, + // or from client.delayed_queue if we were waiting for our slot. if (self.state == .queued) { self.client.pending_queue.remove(&self._node); + } else if (self.state == .delayed) { + self.client.delayed_queue.remove(&self._node); + self.client.delayed_count -= 1; } // Same for the dispatch queue: a queued transfer (buffered, or @@ -3463,6 +3565,8 @@ fn initTestClient(client: *Client, pool: *ArenaPool) void { client.transfers = .empty; client.blocking_requests = .empty; client.pending_queue = .{}; + client.delayed_queue = .{}; + client.delayed_count = 0; client.dispatch_queue = .{}; client.gated_queue = .{}; client.ws_dispatch_queue = .{}; @@ -3470,6 +3574,8 @@ fn initTestClient(client: *Client, pool: *ArenaPool) void { client.graveyard = .{}; client.dispatch_count = 0; client.intercepted = 0; + client.http_active = 0; + client.ws_active = 0; client.cache = null; client.serve_mode = false; client.obey_robots = false; @@ -4191,3 +4297,121 @@ test "HttpClient: abort survives an error_callback that tears down the owner" { try testing.expectEqual(null, owner.transfers.first); } } + +test "HttpClient: throttled navigations wait for their per-host slot" { + var pool = ArenaPool.init(testing.allocator, .{}); + defer pool.deinit(); + + var net: Network = undefined; + net.cache = null; + net.adblocker = null; + net.web_bot_auth = null; + // An empty pool makes processTransfer queue a started transfer instead + // of putting it on the wire — .queued IS "entered the pipeline". + net.available = .{}; + net.conn_mutex = .init; + net.rate_limiter = @import("RateLimiter.zig").init(testing.allocator, 60_000, 1); + defer net.rate_limiter.?.deinit(); + + var client: Client = undefined; + initTestClient(&client, &pool); + defer client.processGraveyard(); + client.network = &net; + defer client.transfers.deinit(testing.allocator); + + const Helper = struct { + fn newTransfer(c: *Client, p: *ArenaPool, id: u32, url: [:0]const u8, throttle: bool) !*Transfer { + const arena = try p.acquire(.small, "test"); + const transfer = try arena.create(Transfer); + transfer.* = .{ + .arena = arena, + .owner = null, + .req = .{ + .frame_id = 0, + .loader_id = 0, + .method = .GET, + .url = url, + .cookie_jar = null, + .cookie_origin = "", + .resource_type = .document, + .notification = undefined, + .shutdown_callback = noopShutdown, + .ctx = undefined, + .throttle = throttle, + }, + .client = c, + .id = id, + .start_time = 0, + }; + try c.transfers.putNoClobber(testing.allocator, transfer.id, transfer); + return transfer; + } + }; + + // First navigation to a host goes straight through. + const a1 = try Helper.newTransfer(&client, &pool, 1, "http://a.example.com/1", true); + try a1.submit(); + try testing.expectEqual(true, a1.state == .queued); + try testing.expectEqual(0, client.delayed_count); + + // Later ones to the same host wait, in reservation order. + const a2 = try Helper.newTransfer(&client, &pool, 2, "http://a.example.com/2", true); + try a2.submit(); + try testing.expectEqual(true, a2.state == .delayed); + const a3 = try Helper.newTransfer(&client, &pool, 3, "http://A.EXAMPLE.COM/3", true); + try a3.submit(); + try testing.expectEqual(true, a3.state == .delayed); + try testing.expectEqual(true, a2._run_at < a3._run_at); + + // Other hosts are independent, and non-throttled requests never wait. + const b1 = try Helper.newTransfer(&client, &pool, 4, "http://b.example.com/1", true); + try b1.submit(); + try testing.expectEqual(true, b1.state == .queued); + const b2 = try Helper.newTransfer(&client, &pool, 5, "http://b.example.com/2", true); + try b2.submit(); + try testing.expectEqual(true, b2.state == .delayed); + const a4 = try Helper.newTransfer(&client, &pool, 6, "http://a.example.com/sub", false); + try a4.submit(); + try testing.expectEqual(true, a4.state == .queued); + + // delayed_queue is ordered by _run_at: a2 <= b2 < a3 + try testing.expectEqual(3, client.delayed_count); + try testing.expectEqual(3, client.activity().http); + { + var node = client.delayed_queue.first; + var order: [3]u32 = undefined; + for (&order) |*o| { + const t: *Transfer = @fieldParentPtr("_node", node.?); + o.* = t.id; + node = node.?.next; + } + try testing.expectEqual(null, node); + try testing.expectEqual(.{ 2, 5, 3 }, order); + } + + // The tick's poll never sleeps past the next slot. + const clamped = client.clampToDelayed(200); + try testing.expectEqual(true, clamped <= 200); + // and nothing is due yet + try client.startPending(); + try testing.expectEqual(3, client.delayed_count); + + // Tearing down a delayed transfer unlinks it. + b2.deinit(); + try testing.expectEqual(2, client.delayed_count); + + // Once its time comes, a delayed transfer enters the pipeline. + a2._run_at = 0; + try client.startPending(); + try testing.expectEqual(true, a2.state == .queued); + try testing.expectEqual(true, a3.state == .delayed); + try testing.expectEqual(1, client.delayed_count); + try testing.expectEqual(a3, @as(*Transfer, @fieldParentPtr("_node", client.delayed_queue.first.?))); + + for ([_]*Transfer{ a1, a2, a3, b1, a4 }) |t| { + t.deinit(); + } + try testing.expectEqual(0, client.delayed_count); + try testing.expectEqual(null, client.delayed_queue.first); + try testing.expectEqual(null, client.pending_queue.first); +} diff --git a/src/network/Network.zig b/src/network/Network.zig index ee488392e..4087f2fda 100644 --- a/src/network/Network.zig +++ b/src/network/Network.zig @@ -32,6 +32,7 @@ const http = @import("http.zig"); const IpFilter = @import("IpFilter.zig"); const RobotStore = @import("Robots.zig").RobotStore; const WebBotAuth = @import("WebBotAuth.zig"); +const RateLimiter = @import("RateLimiter.zig"); const CurlDebugAllocator = @import("CurlDebugAllocator.zig"); const Cache = @import("cache/Cache.zig"); @@ -91,6 +92,7 @@ config: *const Config, x509_store: *crypto.X509_STORE, robot_store: RobotStore, web_bot_auth: ?WebBotAuth, +rate_limiter: ?RateLimiter, /// Hostname dictionaries built from `--adblock-lists`. Parsed once here and /// never mutated afterwards, so every HttpClient can share this one copy. adblocker: ?AdBlocker, @@ -276,6 +278,7 @@ pub fn init(allocator: Allocator, app: *App, config: *const Config) !Network { .cache = cache, .robot_store = RobotStore.init(allocator), .web_bot_auth = web_bot_auth, + .rate_limiter = if (config.httpNavDelay()) |ms| RateLimiter.init(allocator, ms, config.httpNavBurst()) else null, .adblocker = adblocker, .ws_pool = .empty, @@ -306,6 +309,9 @@ pub fn deinit(self: *Network) void { self.ws_pool.deinit(self.allocator); self.robot_store.deinit(); + if (self.rate_limiter) |*rl| { + rl.deinit(); + } if (self.web_bot_auth) |wba| { wba.deinit(self.allocator); } @@ -854,3 +860,33 @@ fn loadFromDirectory( } return count; } + +pub fn HostHashMap(comptime V: type) type { + return std.HashMapUnmanaged([]const u8, V, HostContext, 80); +} + +// Case-insensitive host key for host-keyed map +const HostContext = struct { + pub fn hash(_: HostContext, value: []const u8) u64 { + var key = value; + var buf: [128]u8 = undefined; + var h = std.hash.Wyhash.init(value.len); + + while (key.len >= 128) { + const lower = std.ascii.lowerString(buf[0..], key[0..128]); + h.update(lower); + key = key[128..]; + } + + if (key.len > 0) { + const lower = std.ascii.lowerString(buf[0..key.len], key); + h.update(lower); + } + + return h.final(); + } + + pub fn eql(_: HostContext, a: []const u8, b: []const u8) bool { + return std.ascii.eqlIgnoreCase(a, b); + } +}; diff --git a/src/network/RateLimiter.zig b/src/network/RateLimiter.zig new file mode 100644 index 000000000..1d4087d55 --- /dev/null +++ b/src/network/RateLimiter.zig @@ -0,0 +1,187 @@ +// 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 lp = @import("lightpanda"); + +const Network = @import("Network.zig"); + +const Allocator = std.mem.Allocator; + +const RateLimiter = @This(); + +allocator: Allocator, + +// Sustained minimum gap between two navigations to the same host. +interval_ms: u64, + +// Burst tolerance (GCRA tau): how far ahead of "now" a host's theoretical +// arrival time may run before a navigation has to wait. (burst - 1) * +// interval_ms, so burst = 1 means strict spacing. +tau: u64, + +// hostname (no port) -> theoretical arrival time (TAT): the time the host's +// reservations would have reached if every one had been spaced by +// interval_ms. A host is idle once its TAT is in the past. +next: Network.HostHashMap(u64) = .empty, + +mutex: std.Io.Mutex = .init, + +// Idle entries are swept once the map reaches this size +sweep_at: usize = 256, + +pub fn init(allocator: Allocator, interval_ms: u64, burst: u32) RateLimiter { + return .{ + .allocator = allocator, + .interval_ms = interval_ms, + .tau = interval_ms * (@max(burst, 1) - 1), + }; +} + +pub fn deinit(self: *RateLimiter) void { + var it = self.next.keyIterator(); + while (it.next()) |key| { + self.allocator.free(key.*); + } + self.next.deinit(self.allocator); +} + +// Reserve the next navigation slot for the given host. Returns the time the +// request can run at (can be 0, for now). +pub fn reserve(self: *RateLimiter, host: []const u8, now: u64) !u64 { + self.mutex.lockUncancelable(lp.io); + defer self.mutex.unlock(lp.io); + + const gop = try self.next.getOrPut(self.allocator, host); + if (gop.found_existing) { + const tat = gop.value_ptr.*; + const start = @max(now, tat -| self.tau); + gop.value_ptr.* = @max(tat, start) + self.interval_ms; + return start; + } + + gop.key_ptr.* = self.allocator.dupe(u8, host) catch |err| { + self.next.removeByPtr(gop.key_ptr); + return err; + }; + gop.value_ptr.* = now + self.interval_ms; + + if (self.next.count() >= self.sweep_at) { + self.sweep(now); + } + return now; +} + +// Drop every host whose TAT is already in the past. +// Caller holds the mutex. +fn sweep(self: *RateLimiter, now: u64) void { + var it = self.next.iterator(); + while (it.next()) |entry| { + if (entry.value_ptr.* > now) { + continue; + } + const key = entry.key_ptr.*; + self.next.removeByPtr(entry.key_ptr); + self.allocator.free(key); + } + + if (self.next.count() >= self.sweep_at) { + // we don't want reserve() to turn into an O(N), constantly sweeping + // what it can't clean. So we increase the size of what we'll hold + self.sweep_at *= 2; + } +} + +const testing = @import("../testing.zig"); +test "RateLimiter: reserve" { + var rl = RateLimiter.init(testing.allocator, 100, 1); + defer rl.deinit(); + + // idle host starts now, then serializes at the interval + try testing.expectEqual(1000, try rl.reserve("a.test", 1000)); + try testing.expectEqual(1100, try rl.reserve("a.test", 1000)); + try testing.expectEqual(1200, try rl.reserve("a.test", 1050)); + + // other hosts are independent, keys are case-insensitive + try testing.expectEqual(1000, try rl.reserve("b.test", 1000)); + try testing.expectEqual(1100, try rl.reserve("B.TEST", 1000)); + + // once the interval has elapsed, the host is idle again + try testing.expectEqual(5000, try rl.reserve("a.test", 5000)); + try testing.expectEqual(2, rl.next.count()); +} + +test "RateLimiter: burst" { + var rl = RateLimiter.init(testing.allocator, 100, 3); + defer rl.deinit(); + + // an idle host absorbs `burst` navigations at once + try testing.expectEqual(1000, try rl.reserve("a.test", 1000)); + try testing.expectEqual(1000, try rl.reserve("a.test", 1000)); + try testing.expectEqual(1000, try rl.reserve("a.test", 1000)); + // then spaces them by the interval + try testing.expectEqual(1100, try rl.reserve("a.test", 1000)); + try testing.expectEqual(1200, try rl.reserve("a.test", 1000)); + try testing.expectEqual(1300, try rl.reserve("a.test", 1250)); + + // the burst allowance refills one slot per interval: at 1500 the TAT + // (1600) is only one interval ahead, so one extra slot is free + try testing.expectEqual(1500, try rl.reserve("a.test", 1500)); + try testing.expectEqual(1500, try rl.reserve("a.test", 1500)); + try testing.expectEqual(1600, try rl.reserve("a.test", 1500)); + + // fully idle again once the TAT is in the past + try testing.expectEqual(5000, try rl.reserve("a.test", 5000)); + try testing.expectEqual(5000, try rl.reserve("a.test", 5000)); + try testing.expectEqual(5000, try rl.reserve("a.test", 5000)); + try testing.expectEqual(5100, try rl.reserve("a.test", 5000)); + + // burst = 0 is treated as 1 + var strict = RateLimiter.init(testing.allocator, 100, 0); + defer strict.deinit(); + try testing.expectEqual(0, strict.tau); +} + +test "RateLimiter: sweep" { + var rl = RateLimiter.init(testing.allocator, 100, 1); + defer rl.deinit(); + rl.sweep_at = 4; + + var buf: [16]u8 = undefined; + for (0..3) |i| { + _ = try rl.reserve(try std.fmt.bufPrint(&buf, "h{d}.test", .{i}), 1000); + } + try testing.expectEqual(3, rl.next.count()); + + // 4th insert reaches sweep_at; nothing is idle yet, so the threshold grows + _ = try rl.reserve("h3.test", 1000); + try testing.expectEqual(4, rl.next.count()); + try testing.expectEqual(8, rl.sweep_at); + + for (4..7) |i| { + _ = try rl.reserve(try std.fmt.bufPrint(&buf, "h{d}.test", .{i}), 1000); + } + try testing.expectEqual(7, rl.next.count()); + + // hits sweep_at again, far in the future: every earlier host is idle + try testing.expectEqual(9000, try rl.reserve("h7.test", 9000)); + try testing.expectEqual(1, rl.next.count()); + try testing.expectEqual(8, rl.sweep_at); + // and the survivor keeps its reservation + try testing.expectEqual(9100, try rl.reserve("h7.test", 9000)); +} diff --git a/src/network/Robots.zig b/src/network/Robots.zig index 02c72c588..75d5561e0 100644 --- a/src/network/Robots.zig +++ b/src/network/Robots.zig @@ -96,32 +96,7 @@ pub const RobotStore = struct { disallowed, }; - pub const RobotsMap = std.HashMapUnmanaged([]const u8, RobotsEntry, struct { - const Context = @This(); - - pub fn hash(_: Context, value: []const u8) u32 { - var key = value; - var buf: [128]u8 = undefined; - var h = std.hash.Wyhash.init(value.len); - - while (key.len >= 128) { - const lower = std.ascii.lowerString(buf[0..], key[0..128]); - h.update(lower); - key = key[128..]; - } - - if (key.len > 0) { - const lower = std.ascii.lowerString(buf[0..key.len], key); - h.update(lower); - } - - return @truncate(h.final()); - } - - pub fn eql(_: Context, a: []const u8, b: []const u8) bool { - return std.ascii.eqlIgnoreCase(a, b); - } - }, 80); + pub const RobotsMap = @import("Network.zig").HostHashMap(RobotsEntry); allocator: std.mem.Allocator, map: RobotsMap,