From 035f769df45e53336dbbdc33f05c8917ecf0b987 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Thu, 13 Aug 2026 15:30:53 -0500 Subject: [PATCH] transport: bound the response-head phase with a deadline StallGuard only watches body bytes, so connect, TLS, request send, and receiveHead were unbounded: a peer that accepts and never answers (the 2026-08-13 zlay authority stall) held a caller until the kernel gave up (~2min for an unanswered SYN) or forever on an established socket. head_deadline_ns (default 10s, 0 disables) runs each fetch attempt as a task and cancels it if no response head arrives in time, surfacing error.ResponseHeadTimeout. Once the head is seen, the body remains StallGuard's problem. Mirrors indigo's identity directory client, which deadlines every phase (10s client timeout, 3s dial). Co-Authored-By: Claude Fable 5 --- src/internal/xrpc/transport.zig | 115 ++++++++++++++++++++++++++++++-- 1 file changed, 111 insertions(+), 4 deletions(-) diff --git a/src/internal/xrpc/transport.zig b/src/internal/xrpc/transport.zig index b0fdb24..d2380fb 100644 --- a/src/internal/xrpc/transport.zig +++ b/src/internal/xrpc/transport.zig @@ -23,6 +23,12 @@ pub const HttpTransport = struct { /// the caller's own deadline allows, and a pool of workers can sit "busy" /// while moving almost no bytes. Set any field to 0 to disable that check. stall: StallGuard = .{}, + /// Bounds connect, TLS, request send, and the wait for the response head — + /// the phases StallGuard cannot see, because no body bytes exist yet. A + /// peer that accepts and never answers, or a SYN that goes unanswered, + /// otherwise blocks the caller until the kernel gives up (~2 minutes) or + /// forever on an established socket. 0 disables. + head_deadline_ns: u64 = 10 * std.time.ns_per_s, /// A pooled connection the origin closed while it sat idle looks identical /// to a request that failed, but nothing was ever delivered — so replaying /// an idempotent request on a fresh connection is safe, and skipping the @@ -89,10 +95,7 @@ pub const HttpTransport = struct { var attempt: usize = 0; while (true) { attempt += 1; - const result = if (options.resolved_connection) |resolved| - self.fetchResolved(options, resolved, headers, extra) - else - self.fetchUrl(options, headers, extra); + const result = self.fetchDeadlined(options, headers, extra); return result catch |err| { if (attempt < self.dead_connection_attempts and @@ -103,11 +106,67 @@ pub const HttpTransport = struct { } } + const FetchProgress = struct { + head_seen: std.atomic.Value(bool) = .init(false), + done: std.atomic.Value(bool) = .init(false), + }; + + /// Run one fetch attempt under `head_deadline_ns`. The attempt runs as a + /// task; the watcher cancels it if the response head has not arrived by + /// the deadline. Once the head is seen the body is StallGuard's problem + /// and the watcher steps aside. + fn fetchDeadlined( + self: *HttpTransport, + options: FetchOptions, + headers: std.http.Client.Request.Headers, + extra_headers: []const std.http.Header, + ) !FetchResult { + var progress: FetchProgress = .{}; + if (self.head_deadline_ns == 0) + return self.fetchOnce(options, headers, extra_headers, &progress); + + var future = self.io.async(fetchOnce, .{ self, options, headers, extra_headers, &progress }); + + const tick_ns: u64 = 50 * std.time.ns_per_ms; + var elapsed_ns: u64 = 0; + while (!progress.done.load(.acquire)) { + if (progress.head_seen.load(.acquire)) break; + if (elapsed_ns >= self.head_deadline_ns) { + if (future.cancel(self.io)) |completed| { + var result = completed; + result.deinit(self.allocator); + } else |_| {} + return error.ResponseHeadTimeout; + } + std.Io.Clock.Duration.sleep( + .{ .raw = .fromNanoseconds(tick_ns), .clock = .awake }, + self.io, + ) catch break; + elapsed_ns += tick_ns; + } + + return future.await(self.io); + } + + fn fetchOnce( + self: *HttpTransport, + options: FetchOptions, + headers: std.http.Client.Request.Headers, + extra_headers: []const std.http.Header, + progress: *FetchProgress, + ) !FetchResult { + defer progress.done.store(true, .release); + if (options.resolved_connection) |resolved| + return self.fetchResolved(options, resolved, headers, extra_headers, progress); + return self.fetchUrl(options, headers, extra_headers, progress); + } + fn fetchUrl( self: *HttpTransport, options: FetchOptions, headers: std.http.Client.Request.Headers, extra_headers: []const std.http.Header, + progress: *FetchProgress, ) !FetchResult { const uri = try std.Uri.parse(options.url); const redirect_behavior = redirectBehavior(options); @@ -133,6 +192,7 @@ pub const HttpTransport = struct { defer self.freeRedirectBuffer(redirect_buffer, redirect_behavior); var response = try request.receiveHead(redirect_buffer); + progress.head_seen.store(true, .release); return try self.readFetchResult(options, &response); } @@ -142,6 +202,7 @@ pub const HttpTransport = struct { resolved: ResolvedConnection, headers: std.http.Client.Request.Headers, extra_headers: []const std.http.Header, + progress: *FetchProgress, ) !FetchResult { const uri = try std.Uri.parse(options.url); const protocol = std.http.Client.Protocol.fromUri(uri) orelse return error.UnsupportedUriScheme; @@ -188,6 +249,7 @@ pub const HttpTransport = struct { defer self.freeRedirectBuffer(redirect_buffer, redirect_behavior); var response = try request.receiveHead(redirect_buffer); + progress.head_seen.store(true, .release); return try self.readFetchResult(options, &response); } @@ -610,6 +672,24 @@ const TestOrigin = struct { _ = reader.interface.readVec(&tail) catch {}; } + /// Accepts, reads the request, then never answers — the shape of the + /// wedged authority checks in the 2026-08-13 zlay stall: connection + /// established, response head never arrives, nothing for StallGuard to + /// measure. + fn serveSilent(self: *TestOrigin) !void { + var conn = try self.listener.accept(self.io); + defer conn.close(self.io); + var rbuf: [4096]u8 = undefined; + var reader = conn.reader(self.io, &rbuf); + var scratch: [2048]u8 = undefined; + var slices = [_][]u8{&scratch}; + _ = reader.interface.readVec(&slices) catch {}; + // Say nothing and block until the client hangs up. + var sink: [1]u8 = undefined; + var tail = [_][]u8{&sink}; + _ = reader.interface.readVec(&tail) catch {}; + } + /// Accepts, reads the request, then hangs up without answering — what a /// pooled connection reaped by the origin looks like on next reuse. fn serveDeadThenOk(self: *TestOrigin, response: []const u8) !void { @@ -756,6 +836,33 @@ test "a gzip response decodes: the identity override was covering an empty decom try std.testing.expectEqualStrings(payload, res.body); } +test "a peer that accepts and never answers is abandoned at the head deadline" { + const allocator = std.testing.allocator; + var threaded: std.Io.Threaded = .init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var origin = try TestOrigin.init(io); + defer origin.deinit(); + const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/silent", .{origin.port()}); + defer allocator.free(url); + + var server = try std.Thread.spawn(.{}, struct { + fn run(o: *TestOrigin) void { + o.serveSilent() catch {}; + } + }.run, .{&origin}); + defer server.join(); + + var transport = HttpTransport.init(io, allocator); + defer transport.deinit(); + transport.head_deadline_ns = 300 * std.time.ns_per_ms; + + // Unguarded, this call sits in receiveHead until the peer or kernel gives + // up — for a wedged keep-alive socket, never. + try std.testing.expectError(error.ResponseHeadTimeout, transport.fetch(.{ .url = url })); +} + test "a connection closed before it answers is replayed, not surfaced" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); -- 2.51.2