upstream: one absolute per-query budget across queueing and failover

waiting for a slot now spends the query budget; truncated attempts that
expire fault the budget, not the upstream, and are never attributed.
admission sweeps in priority order before blocking. forward zones spend
read_timeout_ms once across udp, truncation and tcp. adds
nxdns_upstream_budget_exhausted_total and a 64-upstream validation limit.

src/local/forward_client.zig

@@ -109,6 +109,11 @@ pub const ForwardClient = struct {
 
     /// `.udp` resolvers send one datagram and fall back to TCP when the answer
     /// comes back with TC=1. `.tcp` resolvers skip straight to the TCP path.
+    ///
+    /// `read_timeout` bounds the WHOLE exchange, truncation fallback included:
+    /// the instant is computed once here and every blocking step inside runs
+    /// against it, so a truncated UDP answer followed by a stalled TCP retry
+    /// costs one budget rather than two.
     pub fn exchange(
         self: *ForwardClient,
         io: std.Io,
@@ -120,10 +125,22 @@ pub const ForwardClient = struct {
         if (response_buf.len == 0) return error.BufferTooSmall;
 
         self.stats.queries += 1;
-        return self.route(io, query, response_buf) catch |err| {
+        const expiry_at: std.Io.Clock.Timestamp = .fromNow(io, self.read_timeout);
+        // The outcome is deliberately discarded. A forward zone has exactly one
+        // configured resolver, so an expiry here is still that resolver failing
+        // to answer in time: `error.Timeout` is peer evidence, and the
+        // budget/peer distinction is upstream-pool policy.
+        var outcome: transport.RaceOutcome = .completed;
+        return transport.raceUntilTagged(io, expiry_at, &outcome, route, .{
+            self,
+            io,
+            query,
+            response_buf,
+            expiry_at,
+        }) catch |err| {
             switch (transport.group(err)) {
                 .peer_fault, .local_resource => self.stats.failures += 1,
-                .cancellation => {},
+                .cancellation, .budget_exhausted => {},
             }
             return err;
         };
@@ -134,11 +151,12 @@ pub const ForwardClient = struct {
         io: std.Io,
         query: []const u8,
         response_buf: []u8,
+        expiry_at: std.Io.Clock.Timestamp,
     ) transport.ExchangeError![]u8 {
         if (self.resolver.scheme == .udp) {
-            if (try self.exchangeUdp(io, query, response_buf)) |reply| return reply;
+            if (try self.exchangeUdp(io, query, response_buf, expiry_at)) |reply| return reply;
         }
-        return self.exchangeTcp(io, query, response_buf);
+        return self.tcpOnce(io, query, response_buf);
     }
 
     /// `null` means the resolver set TC=1 and the caller must retry over TCP.
@@ -151,6 +169,7 @@ pub const ForwardClient = struct {
         io: std.Io,
         query: []const u8,
         response_buf: []u8,
+        expiry_at: std.Io.Clock.Timestamp,
     ) transport.ExchangeError!?[]u8 {
         const dest = self.destination();
         const local = wildcardFor(dest);
@@ -166,9 +185,10 @@ pub const ForwardClient = struct {
             return transport.mapPhase(err, error.SendFailed);
         };
 
-        // A deadline, not a duration: a discarded foreign datagram restarts the
-        // receive, and a duration would hand each retry the full budget again.
-        const deadline = (std.Io.Timeout{ .duration = self.read_timeout }).toDeadline(io);
+        // The exchange-wide instant, not a fresh duration: a discarded foreign
+        // datagram restarts the receive, and a duration would hand each retry
+        // the full budget again.
+        const deadline: std.Io.Timeout = .{ .deadline = expiry_at };
 
         while (true) {
             const msg = socket.receiveTimeout(io, response_buf, deadline) catch |err| switch (err) {
@@ -205,18 +225,11 @@ pub const ForwardClient = struct {
         }
     }
 
-    /// The read budget bounds the whole TCP exchange through
-    /// `transport.raceWithin`. `ConnectOptions.timeout` is never set: the
-    /// Threaded backend panics on it (Threaded.zig:12076).
-    fn exchangeTcp(
-        self: *ForwardClient,
-        io: std.Io,
-        query: []const u8,
-        response_buf: []u8,
-    ) transport.ExchangeError![]u8 {
-        return transport.raceWithin(io, self.read_timeout, tcpOnce, .{ self, io, query, response_buf });
-    }
-
+    /// Unbounded on its own: `exchange` runs it inside the exchange-wide race,
+    /// which is what cancels a stalled connect or read. It takes no budget of
+    /// its own, so a truncation fallback does not start a second one.
+    /// `ConnectOptions.timeout` is never set: the Threaded backend panics on it
+    /// (Threaded.zig:12076).
     fn tcpOnce(
         self: *ForwardClient,
         io: std.Io,
@@ -307,6 +320,11 @@ fn receiveFailure(stream_reader: *const net.Stream.Reader, err: anyerror) transp
 }
 
 const testing = std.testing;
+const build_options = @import("build_options");
+const name_mod = @import("../dns/name.zig");
+const packet = @import("../dns/packet.zig");
+const question = @import("../dns/question.zig");
+const types = @import("../dns/types.zig");
 
 fn testBuf() [min_frame_buf]u8 {
     return undefined;
@@ -467,3 +485,113 @@ test "a stashed stream error is preferred over the collapsed one" {
         receiveFailure(&stream_reader, error.EndOfStream),
     );
 }
+
+const one_budget_ms = 400;
+/// Late enough in the budget that a second, fresh budget for the TCP leg would
+/// be unmistakable in the elapsed time.
+const truncate_after_ms = 300;
+
+/// Answers the first datagram late in the budget with TC=1, which sends the
+/// client to TCP — where a listener that never accepts leaves it stalled.
+fn truncatingThenStallingResolver(io: std.Io, socket: *const net.Socket) void {
+    var buf: [2048]u8 = undefined;
+    const msg = socket.receive(io, &buf) catch return;
+    const request = packet.parse(msg.data) catch return;
+
+    (std.Io.Clock.Duration{
+        .raw = .fromMilliseconds(truncate_after_ms),
+        .clock = .awake,
+    }).sleep(io) catch return;
+
+    // The question has to be echoed: `transport.validateResponse` runs before
+    // the client reads the TC bit, so a bare header would come back as
+    // `BadResponse` and never reach the TCP fallback this case is about.
+    const q = packet.firstQuestion(request) orelse return;
+    var reply_buf: [512]u8 = undefined;
+    var b = packet.ResponseBuilder.init(&reply_buf, request.header, q) catch return;
+    const reply = b.finish();
+
+    var parsed = dns_header.parse(reply) catch return;
+    parsed.flags.tc = true;
+    dns_header.encode(parsed, reply[0..types.header_len]);
+    socket.send(io, &msg.from, reply) catch return;
+}
+
+fn testQuery(io: std.Io, buf: []u8) ![]const u8 {
+    var id_bytes: [2]u8 = undefined;
+    io.random(&id_bytes);
+    dns_header.encode(.{
+        .id = std.mem.readInt(u16, &id_bytes, .big),
+        .flags = .{
+            .rcode = .no_error,
+            .z = 0,
+            .ra = false,
+            .rd = true,
+            .tc = false,
+            .aa = false,
+            .opcode = .query,
+            .qr = false,
+        },
+        .qdcount = 1,
+        .ancount = 0,
+        .nscount = 0,
+        .arcount = 0,
+    }, buf[0..types.header_len]);
+
+    var w: std.Io.Writer = .fixed(buf[types.header_len..]);
+    try question.encode(.{
+        .name = try name_mod.fromText("nas.lan"),
+        .qtype = .a,
+        .qclass = .in,
+    }, &w);
+    return buf[0 .. types.header_len + w.buffered().len];
+}
+
+test "one budget covers the udp leg, the TC=1 fallback and the tcp leg" {
+    if (!build_options.integration) return error.SkipZigTest;
+
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    const bind_address: net.IpAddress = try .parse("127.0.0.1", 0);
+    const socket = try bind_address.bind(io, .{ .mode = .dgram });
+    defer socket.close(io);
+    const port = socket.address.ip4.port;
+
+    // Bound but never accepted: the connect completes out of the kernel's
+    // backlog and the read then waits forever, which is the stall a second
+    // budget would be spent on.
+    const tcp_address: net.IpAddress = try .parse("127.0.0.1", port);
+    var tcp_listener = try tcp_address.listen(io, .{ .reuse_address = true });
+    defer tcp_listener.deinit(io);
+
+    var group: std.Io.Group = .init;
+    defer group.cancel(io);
+    try group.concurrent(io, truncatingThenStallingResolver, .{ io, &socket });
+
+    var url_buf: [64]u8 = undefined;
+    const url = try std.fmt.bufPrint(&url_buf, "udp://127.0.0.1:{d}", .{port});
+
+    var frame_buf: [min_frame_buf]u8 = undefined;
+    var fc: ForwardClient = .init(
+        try validate.parseResolver(url),
+        &frame_buf,
+        .{ .raw = .fromMilliseconds(one_budget_ms), .clock = .awake },
+    );
+
+    var query_buf: [types.header_len + types.max_name_len + 4]u8 = undefined;
+    const query = try testQuery(io, &query_buf);
+    var response_buf: [2048]u8 = undefined;
+
+    const started = std.Io.Clock.awake.now(io);
+    try testing.expectError(error.Timeout, fc.exchange(io, query, &response_buf));
+    const elapsed = started.durationTo(std.Io.Clock.awake.now(io)).nanoseconds;
+
+    // Two budgets would spend 300 ms on the UDP leg and then a fresh 400 ms on
+    // the stalled TCP leg. The bound sits between one budget and that sum, so
+    // the double-budget shape cannot pass.
+    try testing.expect(elapsed < @as(i96, one_budget_ms + truncate_after_ms / 2) * std.time.ns_per_ms);
+    try testing.expectEqual(@as(u64, 1), fc.stats.udp_truncated);
+    try testing.expectEqual(@as(u64, 1), fc.stats.failures);
+}