milestone 18: collapse duplicated infrastructure into shared listener core, crud list helper, resource shells, transport race, name and line helpers, ui modules

src/upstream/transport.zig

@@ -229,6 +229,97 @@ pub fn mapLocal(err: anyerror) ?ExchangeError {
     };
 }
 
+/// Names the peer fault of the phase the call site is in, unless the error is
+/// one `mapLocal` claims for this process. Every transport classifies its
+/// failures through this one function.
+pub fn mapPhase(err: anyerror, phase: PeerFault) ExchangeError {
+    return mapLocal(err) orelse phase;
+}
+
+/// Closes `target` with cancellation blocked.
+///
+/// A transport's close runs from a `defer` chain that a lost timeout race is
+/// unwinding. The next cancelable `Io` call in that chain returns
+/// `error.Canceled` and skips the close, leaking the descriptor, so the close
+/// swaps cancellation protection for the duration.
+///
+/// `net.Stream` and `net.Socket` close through an `Io`; `tls_client.TlsStream`
+/// owns the one it was built with and takes none. Both shapes are accepted so
+/// that one helper covers every close in the transports.
+pub fn closeBlocked(io: std.Io, target: anytype) void {
+    const prev = io.swapCancelProtection(.blocked);
+    defer _ = io.swapCancelProtection(prev);
+    const Target = @typeInfo(@TypeOf(target)).pointer.child;
+    if (@typeInfo(@TypeOf(Target.close)).@"fn".params.len == 2) {
+        target.close(io);
+    } else {
+        target.close();
+    }
+}
+
+/// The payload of `f`'s return type, which `raceWithin` requires to be
+/// `ExchangeError!T`. A raced function with any other error set would let a
+/// failure reach the pool without passing through `group`.
+fn RacedPayload(comptime f: anytype) type {
+    const info = @typeInfo(@TypeOf(f));
+    if (info != .@"fn") @compileError("raceWithin needs a function, found " ++ @typeName(@TypeOf(f)));
+    const Return = info.@"fn".return_type orelse
+        @compileError("raceWithin needs a function with a concrete return type");
+    const union_info = switch (@typeInfo(Return)) {
+        .error_union => |u| u,
+        else => @compileError("raceWithin needs `ExchangeError!T`, found " ++ @typeName(Return)),
+    };
+    if (union_info.error_set != ExchangeError)
+        @compileError("raceWithin needs `ExchangeError!T`, found " ++ @typeName(Return));
+    return union_info.payload;
+}
+
+/// Runs `f(args...)` raced against `budget`, and cancels the loser.
+///
+/// No stream read or write in 0.16.0 takes a timeout, so a deadline is a second
+/// task rather than a socket option. This is the one copy of that harness: the
+/// pool races an attempt and its whole failover loop through it, and the
+/// forward client races its TCP exchange.
+///
+/// `error.Timeout` means the budget won. A canceled sleep means the whole task
+/// is being torn down rather than the budget running out, so it stays
+/// `error.Canceled`. A backend that cannot start a second task is
+/// `error.SystemResources`, which `group` keeps off the peer's health.
+pub fn raceWithin(
+    io: std.Io,
+    budget: std.Io.Clock.Duration,
+    comptime f: anytype,
+    args: anytype,
+) ExchangeError!RacedPayload(f) {
+    const Outcome = union(enum) {
+        raced: ExchangeError!RacedPayload(f),
+        expiry: std.Io.Cancelable!void,
+    };
+
+    var outcomes: [2]Outcome = undefined;
+    var race: std.Io.Select(Outcome) = .init(io, &outcomes);
+    defer race.cancelDiscard();
+
+    race.concurrent(.raced, f, args) catch |err| switch (err) {
+        error.ConcurrencyUnavailable => return error.SystemResources,
+    };
+    race.concurrent(.expiry, expire, .{ io, budget }) catch |err| switch (err) {
+        error.ConcurrencyUnavailable => return error.SystemResources,
+    };
+
+    switch (try race.await()) {
+        .raced => |result| return result,
+        .expiry => |result| {
+            try result;
+            return error.Timeout;
+        },
+    }
+}
+
+fn expire(io: std.Io, budget: std.Io.Clock.Duration) std.Io.Cancelable!void {
+    return budget.sleep(io);
+}
+
 /// A thing that sends one DNS message and returns one validated DNS message.
 /// Implemented by DohClient, DotClient, Pool, and test fakes.
 pub const Client = struct {
@@ -430,6 +521,117 @@ test "mapLocal folds only local and cancellation errors" {
     try testing.expectEqual(@as(?ExchangeError, null), mapLocal(error.TlsInitializationFailed));
 }
 
+test "mapPhase names the phase unless the error is this process's own" {
+    try testing.expectEqual(ExchangeError.Canceled, mapPhase(error.Canceled, error.ReceiveFailed));
+    try testing.expectEqual(
+        ExchangeError.SystemResources,
+        mapPhase(error.SystemResources, error.ConnectFailed),
+    );
+    try testing.expectEqual(
+        ExchangeError.ConnectFailed,
+        mapPhase(error.ConnectionRefused, error.ConnectFailed),
+    );
+    try testing.expectEqual(
+        Group.local_resource,
+        group(mapPhase(error.OutOfMemory, error.SendFailed)),
+    );
+}
+
+/// Stands in for whatever the pool or the forward client races. The variants
+/// are the three ways such a task ends: in time, too late, or with a failure of
+/// its own.
+fn racedReply(io: std.Io, delay_ms: i64, result: ExchangeError!usize) ExchangeError!usize {
+    if (delay_ms != 0) {
+        const duration: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(delay_ms), .clock = .awake };
+        try duration.sleep(io);
+    }
+    return result;
+}
+
+test "raceWithin returns the raced value when it finishes inside the budget" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    const budget: std.Io.Clock.Duration = .{ .raw = .fromSeconds(30), .clock = .awake };
+    const len = try raceWithin(io, budget, racedReply, .{ io, 0, @as(ExchangeError!usize, 7) });
+    try testing.expectEqual(@as(usize, 7), len);
+}
+
+test "raceWithin returns Timeout when the budget wins" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    const budget: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(20), .clock = .awake };
+    const started = std.Io.Clock.awake.now(io);
+    try testing.expectError(
+        error.Timeout,
+        raceWithin(io, budget, racedReply, .{ io, 30_000, @as(ExchangeError!usize, 7) }),
+    );
+    const elapsed_ns = std.Io.Clock.awake.now(io).nanoseconds - started.nanoseconds;
+
+    // Far under the raced task's own sleep, so the budget is provably what
+    // ended the call rather than the task finishing on its own.
+    try testing.expect(elapsed_ns < @as(i96, 5) * std.time.ns_per_s);
+}
+
+test "raceWithin passes the raced task's own failure through" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    const budget: std.Io.Clock.Duration = .{ .raw = .fromSeconds(30), .clock = .awake };
+    // A local resource error keeps its group: the race must not turn it into
+    // the peer fault the budget would have produced.
+    const failed = raceWithin(io, budget, racedReply, .{
+        io, 0, @as(ExchangeError!usize, error.OutOfMemory),
+    });
+    try testing.expectError(error.OutOfMemory, failed);
+    try testing.expectError(
+        error.ConnectFailed,
+        raceWithin(io, budget, racedReply, .{ io, 0, @as(ExchangeError!usize, error.ConnectFailed) }),
+    );
+}
+
+test "closeBlocked closes a target of either close shape" {
+    // The two shapes the transports use: a socket or a plain stream, which
+    // closes through the `Io`, and a `TlsStream`, which owns the one it was
+    // built with. Dispatching on the wrong one is a compile error, so
+    // instantiating both is the check.
+    //
+    // That the close runs with cancellation blocked is not asserted here: the
+    // Threaded backend's `swapCancelProtection` is a no-op off one of its own
+    // task threads, so a unit test cannot observe the state it sets.
+    const WithIo = struct {
+        closed: bool = false,
+
+        fn close(self: *@This(), io: std.Io) void {
+            _ = io;
+            self.closed = true;
+        }
+    };
+    const WithoutIo = struct {
+        closed: bool = false,
+
+        fn close(self: *@This()) void {
+            self.closed = true;
+        }
+    };
+
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var with_io: WithIo = .{};
+    closeBlocked(io, &with_io);
+    try testing.expect(with_io.closed);
+
+    var without_io: WithoutIo = .{};
+    closeBlocked(io, &without_io);
+    try testing.expect(without_io.closed);
+}
+
 test "a fake client satisfies the Client interface" {
     const Fake = struct {
         calls: usize = 0,