mokhtar/nxdns
1
0
Fork 0
Code Pull Requests Actions Packages Releases 21 Activity

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

Browse Source 
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.
This commit is contained in:
mokhtar
2026-08-27 21:10:43 +02:00
parent 6170571d23
commit 72cdbbd113
16 changed files with 1598 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cli.zig
+1 -1
View File
@@ -1012,7 +1012,7 @@ fn probeUpstreams(r: Runner, cfg: model.Config) !usize {
.priority = server.priority,
.enabled = true,
.health = .init,
.sem = .{ .permits = slots.len },
.admission = .{ .permits = slots.len },
.reuse_recoveries = &recoveries,
}};
var single: pool.Pool = .init(&entries, .{}, timeouts, seed);
src/config/model.zig
+5 -3
View File
@@ -57,9 +57,11 @@ pub const Config = struct {
pub const Upstream = struct {
/// Bounds one attempt against one upstream inside the pool's failover loop.
attempt_timeout_ms: u32 = 2500,
/// The forward-zone client's read deadline, and nothing else. It bounds a
/// different subsystem from the two above (`src/local/forward_client.zig`),
/// so no cross-check relates it to them.
/// The forward-zone client's whole-exchange budget, and nothing else: one
/// bound covers the UDP attempt, a TC=1 fallback and the TCP retry
/// together, not each of them. It bounds a different subsystem from the two
/// above (`src/local/forward_client.zig`), so no cross-check relates it to
/// them.
read_timeout_ms: u32 = 3000,
/// The whole-exchange budget: every failover attempt together, not one of
/// them. The pool races the entire loop against it.
src/config/validate.zig
+66 -2
View File
@@ -52,6 +52,7 @@ const limits = @import("limits.zig");
const logger = @import("../storage/logger.zig");
const regex = @import("../filter/regex.zig");
const safe_url = @import("../safe_url.zig");
const pool = @import("../upstream/pool.zig");
const transport = @import("../upstream/transport.zig");
const Config = model.Config;
@@ -69,6 +70,7 @@ const Prefix = address.Prefix;
/// like every other resource failure.
pub const ValidateError = error{
NoUpstreams,
TooManyUpstreams,
BadUpstreamUrl,
UpstreamHostNotIpLiteral,
DuplicateUpstreamUrl,
@@ -357,8 +359,9 @@ fn checkScalars(cfg: Config, diags: *Diagnostics) error{OutOfMemory}!void {
// The only cross-check that relates two knobs of one subsystem: the pool
// races one attempt against `attempt` and the whole failover loop against
// `total`, so an attempt budget above the total one can never be reached.
// `read_timeout_ms` belongs to the forward-zone client and is deliberately
// unrelated to both.
// `read_timeout_ms` bounds the forward-zone client's whole exchange —
// UDP attempt, TC=1 fallback and TCP retry under one budget — and is
// deliberately unrelated to both.
if (up.attempt_timeout_ms > up.total_timeout_ms) {
try diags.add(
error.BadTimeout,
@@ -823,6 +826,21 @@ fn checkCollections(cfg: Config, diags: *Diagnostics, scratch: Allocator) error{
.{},
);
}
// Each enabled upstream becomes one pool entry, and the failover loop
// tracks the entries it has spent in a fixed bitset of `Pool.max_entries`
// bits. Without this check a config past that bound reaches an assert and
// panics at startup, which is the wrong way to tell an operator that a
// number is too large. Disabled upstreams are not counted: they never
// become entries.
if (enabled_upstreams > pool.Pool.max_entries) {
try diags.add(
error.TooManyUpstreams,
"upstreams",
.{},
"{d} upstreams are enabled; nxdns is built for at most {d}",
.{ enabled_upstreams, pool.Pool.max_entries },
);
}
var client_ips: IndexSet = .empty;
for (cfg.clients, 0..) |client, i| {
@@ -1323,6 +1341,52 @@ test "error.NoUpstreams when nothing is enabled" {
try expectProblem(cfg, error.NoUpstreams, "upstreams");
}
/// `count` distinct enabled upstreams. Generated rather than written out
/// because the bound this exercises is 64, and a hand-written list that long
/// would say less than the loop does.
fn ManyUpstreams(comptime count: usize) type {
return struct {
const list: [count]model.UpstreamServer = blk: {
var built: [count]model.UpstreamServer = undefined;
for (&built, 0..) |*server, i| {
server.* = .{ .url = std.fmt.comptimePrint("https://u{d}.example/dns-query", .{i}) };
}
break :blk built;
};
};
}
fn manyUpstreams(comptime count: usize) []const model.UpstreamServer {
return &ManyUpstreams(count).list;
}
test "as many enabled upstreams as the pool holds validates cleanly" {
var cfg = baseConfig();
cfg.upstreams = manyUpstreams(pool.Pool.max_entries);
try expectClean(cfg);
}
test "error.TooManyUpstreams one enabled upstream past the pool's bound" {
// The pool asserts this bound, so without the check here a valid-looking
// config panics at startup instead of being reported.
var cfg = baseConfig();
cfg.upstreams = manyUpstreams(pool.Pool.max_entries + 1);
try expectProblem(cfg, error.TooManyUpstreams, "upstreams");
}
test "upstreams past the pool's bound are fine while they are disabled" {
// Only enabled upstreams become pool entries, so a long list with a small
// enabled subset is not near the bound at all.
var cfg = baseConfig();
cfg.upstreams = comptime blk: {
var list = manyUpstreams(pool.Pool.max_entries + 1)[0 .. pool.Pool.max_entries + 1].*;
for (list[1..]) |*server| server.enabled = false;
const frozen = list;
break :blk &frozen;
};
try expectClean(cfg);
}
test "error.BadUpstreamUrl on an unsupported scheme" {
var cfg = baseConfig();
cfg.upstreams = &.{.{ .url = "ftp://dns.example/" }};
src/local/forward_client.zig
+147 -19
View File
@@ -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);
}
src/server/handler.zig
+11 -2
View File
@@ -676,7 +676,10 @@ const Context = struct {
const answer = client.exchange(ctx.io, ctx.query, ctx.response_buf) catch |err| {
return switch (transport.group(err)) {
.cancellation => .drop,
.peer_fault, .local_resource => ctx.servFail(),
// A budget that ran out is SERVFAIL like any other failure: no
// rcode says "I gave up in time". It is not logged per query —
// `nxdns_upstream_budget_exhausted_total` is the record.
.peer_fault, .local_resource, .budget_exhausted => ctx.servFail(),
};
};
@@ -737,7 +740,13 @@ const Context = struct {
// the client is about to lose the socket anyway; the listener's
// own counters record the abandoned datagram.
.cancellation => return .drop,
.peer_fault, .local_resource => return ctx.servFail(),
// A budget that ran out is SERVFAIL like any other failure: no
// rcode says "I gave up in time". It is not logged per query —
// `nxdns_upstream_budget_exhausted_total` is the record — and
// the pool leaves `selected` unchanged, so the row still names
// the last attributable endpoint if there was one, and names
// none only when no attributable attempt happened.
.peer_fault, .local_resource, .budget_exhausted => return ctx.servFail(),
};
};
bump(&ctx.handler.stats.queries);
src/server/resolver_integration_test.zig
+1 -1
View File
@@ -176,7 +176,7 @@ const EntryStorage = struct {
.priority = priority,
.enabled = true,
.health = .init,
.sem = .{ .permits = self.slots.len },
.admission = .{ .permits = self.slots.len },
.reuse_recoveries = &self.recoveries,
};
}
src/upstream/dot_client_integration_test.zig
+1 -1
View File
@@ -458,7 +458,7 @@ test "a session the upstream closed is recovered by one redial and counted, not
.priority = 10,
.enabled = true,
.health = .init,
.sem = .{ .permits = slots.len },
.admission = .{ .permits = slots.len },
.reuse_recoveries = &fixture.recoveries,
}};
var pool: pool_mod.Pool = .init(&entries, .{
src/upstream/owner.zig
+1 -1
View File
@@ -500,7 +500,7 @@ const Upstreams = struct {
.priority = server.priority,
.enabled = true,
.health = .init,
.sem = .{ .permits = entry_slots.len },
.admission = .{ .permits = entry_slots.len },
.reuse_recoveries = &self.recovery_counters[self.used],
};
self.used += 1;
src/upstream/pool.zig
+814 -139
View File
File diff suppressed because it is too large Load Diff
src/upstream/transport.zig
+152 -30
View File
@@ -1,4 +1,4 @@
//! Shared vocabulary for every upstream client: endpoint URLs, the three
//! Shared vocabulary for every upstream client: endpoint URLs, the four
//! disjoint failure groups, the `Client` interface, and response validation.
//!
//! Everything here except the `Client` vtable is pure. `validateResponse` takes
@@ -8,8 +8,10 @@
//!
//! The failure classification is the reason this file exists. Health and
//! backoff must count only what the peer did wrong: a local `OutOfMemory` says
//! nothing about the upstream, and `error.Canceled` says nothing at all. The
//! three error sets below are disjoint by construction and `group` switches
//! nothing about the upstream, `error.Canceled` says nothing at all, and
//! `error.BudgetExhausted` says the caller ran out of time before the peer was
//! given its interval. The four error sets below are disjoint by construction
//! and `group` switches
//! over them exhaustively, so a new failure mode cannot silently land in the
//! wrong bucket.
@@ -188,9 +190,15 @@ pub const LocalResource = error{
pub const Cancellation = error{Canceled};
pub const ExchangeError = PeerFault || LocalResource || Cancellation;
/// The caller's own time ran out before any peer could be given the observation
/// interval it was configured to get. Evidence about this process's budget, not
/// about any endpoint, so it is never recorded against health — that is the
/// whole reason it is not a `PeerFault`.
pub const BudgetFault = error{BudgetExhausted};
pub const Group = enum { peer_fault, local_resource, cancellation };
pub const ExchangeError = PeerFault || LocalResource || Cancellation || BudgetFault;
pub const Group = enum { peer_fault, local_resource, cancellation, budget_exhausted };
/// Exhaustive switch over `ExchangeError` — no `else` arm. A new error member
/// must break the build here, so no failure can silently land in the wrong
@@ -218,6 +226,8 @@ pub fn group(err: ExchangeError) Group {
=> .local_resource,
error.Canceled => .cancellation,
error.BudgetExhausted => .budget_exhausted,
};
}
@@ -269,29 +279,29 @@ pub fn closeBlocked(io: std.Io, target: anytype) void {
}
}
/// The payload of `f`'s return type, which `raceWithin` requires to be
/// The payload of `f`'s return type, which the race harness 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)));
if (info != .@"fn") @compileError("the race harness needs a function, found " ++ @typeName(@TypeOf(f)));
const Return = info.@"fn".return_type orelse
@compileError("raceWithin needs a function with a concrete return type");
@compileError("the race harness 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)),
else => @compileError("the race harness needs `ExchangeError!T`, found " ++ @typeName(Return)),
};
if (union_info.error_set != ExchangeError)
@compileError("raceWithin needs `ExchangeError!T`, found " ++ @typeName(Return));
@compileError("the race harness 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.
/// task rather than a socket option. `raceUntilTagged` is the one copy of that
/// harness; this is the untagged wrapper for callers that own no deadline and
/// only need "a bound on this one operation".
///
/// `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
@@ -303,33 +313,69 @@ pub fn raceWithin(
comptime f: anytype,
args: anytype,
) ExchangeError!RacedPayload(f) {
const Outcome = union(enum) {
var outcome: RaceOutcome = .completed;
return raceUntilTagged(io, .fromNow(io, budget), &outcome, f, args);
}
/// Which side of the race ended the call.
///
/// `completed` means the raced operation itself returned — including when what
/// it returned is `error.Timeout`, which is then the peer's own timeout and
/// real evidence about that peer. `expired` means the caller's clock ran out
/// with the operation still in flight, which is evidence about the budget only.
pub const RaceOutcome = enum { completed, expired };
/// `raceWithin` with the two timer origins told apart, and with an ABSOLUTE
/// expiry rather than a duration.
///
/// The timestamp is the point of the whole function. A duration recomputed from
/// a deadline and then slept re-anchors at "now", so every re-race drifts a
/// little past the caller's real deadline and a truncated attempt is then
/// indistinguishable from a full one. The caller that owns the deadline
/// computes the instant once and passes it here.
///
/// `outcome` is written before this returns on both racing paths. It is left
/// untouched when the race cannot start at all (`error.SystemResources`) or
/// when the whole task is being canceled, since neither is an observation about
/// this budget; callers initialize it to the value they want in those cases.
pub fn raceUntilTagged(
io: std.Io,
expiry_at: std.Io.Clock.Timestamp,
outcome: *RaceOutcome,
comptime f: anytype,
args: anytype,
) ExchangeError!RacedPayload(f) {
const Slot = 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);
var slots: [2]Slot = undefined;
var race: std.Io.Select(Slot) = .init(io, &slots);
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) {
race.concurrent(.expiry, expire, .{ io, expiry_at }) catch |err| switch (err) {
error.ConcurrencyUnavailable => return error.SystemResources,
};
switch (try race.await()) {
.raced => |result| return result,
.raced => |result| {
outcome.* = .completed;
return result;
},
.expiry => |result| {
try result;
outcome.* = .expired;
return error.Timeout;
},
}
}
fn expire(io: std.Io, budget: std.Io.Clock.Duration) std.Io.Cancelable!void {
return budget.sleep(io);
fn expire(io: std.Io, expiry_at: std.Io.Clock.Timestamp) std.Io.Cancelable!void {
return expiry_at.wait(io);
}
/// A thing that sends one DNS message and returns one validated DNS message.
@@ -347,13 +393,21 @@ pub const Client = struct {
/// Returns a prefix of `response_buf`. The returned message has already
/// passed `validateResponse` against `query`.
///
/// `selected` names the resolver the exchange used. An implementation
/// writes it *before* each attempt, never after, so a failed exchange still
/// names the last resolver it tried — a SERVFAIL row without its resolver
/// explains nothing. The slice must outlive the call; every implementation
/// borrows storage it owns for at least the query's duration. Callers
/// initialize it to null: a `null` after the call means no resolver was
/// reached at all.
/// `selected` names the resolver the exchange used. A single-endpoint
/// implementation (DoH, DoT, the forward client, test fakes) may write it
/// *before* each attempt: it has one resolver and records no health, so
/// "the one I tried" is an honest answer even for a failure, and a SERVFAIL
/// row without its resolver explains nothing.
///
/// `Pool` is stricter, and documents the rule on `Pool.exchange`: it names
/// only endpoints whose outcome it recorded, so a query that ran out of
/// budget blames nobody and may leave this `null`. Both are within this
/// contract — the guarantee here is that a non-null value names a resolver
/// this exchange really used, never that a failure leaves one behind.
///
/// The slice must outlive the call; every implementation borrows storage it
/// owns for at least the query's duration. Callers initialize it to null: a
/// `null` after the call means no resolver is being reported.
pub fn exchange(
self: Client,
io: std.Io,
@@ -530,8 +584,8 @@ test "parse rejects a fragment" {
try testing.expectError(error.BadUrl, Endpoint.parse("tls://dns.google#f"));
}
test "the three error groups are disjoint" {
const sets = .{ PeerFault, LocalResource, Cancellation };
test "the four error groups are disjoint" {
const sets = .{ PeerFault, LocalResource, Cancellation, BudgetFault };
inline for (sets, 0..) |a, i| {
inline for (sets, 0..) |b, j| {
if (i >= j) continue;
@@ -548,7 +602,8 @@ test "the three error groups are disjoint" {
// member added to two sets at once cannot pass unnoticed.
const total = @typeInfo(PeerFault).error_set.?.len +
@typeInfo(LocalResource).error_set.?.len +
@typeInfo(Cancellation).error_set.?.len;
@typeInfo(Cancellation).error_set.?.len +
@typeInfo(BudgetFault).error_set.?.len;
try testing.expectEqual(total, @typeInfo(ExchangeError).error_set.?.len);
}
@@ -558,6 +613,7 @@ test "group classifies each member" {
try testing.expectEqual(Group.local_resource, group(error.OutOfMemory));
try testing.expectEqual(Group.local_resource, group(error.BufferTooSmall));
try testing.expectEqual(Group.cancellation, group(error.Canceled));
try testing.expectEqual(Group.budget_exhausted, group(error.BudgetExhausted));
}
test "mapLocal folds only local and cancellation errors" {
@@ -641,6 +697,72 @@ test "raceWithin passes the raced task's own failure through" {
);
}
test "raceUntilTagged tells a leaf Timeout apart from an expiry" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
// The leaf's own timeout: it returned, so the peer really did time out and
// the outcome is `completed` even though the error is the same one an
// expiry produces.
var outcome: RaceOutcome = .expired;
const far: std.Io.Clock.Timestamp = .fromNow(io, .{ .raw = .fromSeconds(30), .clock = .awake });
try testing.expectError(
error.Timeout,
raceUntilTagged(io, far, &outcome, racedReply, .{
io, 0, @as(ExchangeError!usize, error.Timeout),
}),
);
try testing.expectEqual(RaceOutcome.completed, outcome);
// The expiry side, distinguishable only through the tag.
outcome = .completed;
const soon: std.Io.Clock.Timestamp = .fromNow(io, .{ .raw = .fromMilliseconds(20), .clock = .awake });
try testing.expectError(
error.Timeout,
raceUntilTagged(io, soon, &outcome, racedReply, .{
io, 30_000, @as(ExchangeError!usize, 7),
}),
);
try testing.expectEqual(RaceOutcome.expired, outcome);
}
test "raceUntilTagged tags a successful completion" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var outcome: RaceOutcome = .expired;
const far: std.Io.Clock.Timestamp = .fromNow(io, .{ .raw = .fromSeconds(30), .clock = .awake });
const len = try raceUntilTagged(io, far, &outcome, racedReply, .{
io, 0, @as(ExchangeError!usize, 7),
});
try testing.expectEqual(@as(usize, 7), len);
try testing.expectEqual(RaceOutcome.completed, outcome);
}
test "raceUntilTagged honours an expiry already in the past" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
// A deadline the caller has already spent: the expiry wins immediately
// rather than re-anchoring a duration at "now" and granting a fresh budget.
const past = std.Io.Clock.Timestamp.now(io, .awake)
.subDuration(.{ .raw = .fromSeconds(1), .clock = .awake });
var outcome: RaceOutcome = .completed;
const started = std.Io.Clock.awake.now(io);
try testing.expectError(
error.Timeout,
raceUntilTagged(io, past, &outcome, racedReply, .{
io, 30_000, @as(ExchangeError!usize, 7),
}),
);
try testing.expectEqual(RaceOutcome.expired, outcome);
const elapsed_ns = std.Io.Clock.awake.now(io).nanoseconds - started.nanoseconds;
try testing.expect(elapsed_ns < @as(i96, 5) * std.time.ns_per_s);
}
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
src/web/metrics.zig
+39 -2
View File
@@ -166,6 +166,12 @@ pub const Sample = struct {
udp_listener: ?udp_server.Snapshot = null,
tcp_listener: ?tcp_server.Snapshot = null,
upstreams: []const UpstreamSample = &.{},
/// Exchanges the pool gave up on because the request's own budget ran out.
/// Pool-wide rather than per-upstream on purpose: budget exhaustion is
/// evidence about the pool, never about an endpoint, so it carries no url
/// label and cannot live in `UpstreamSample`. Absent while no pool is
/// wired, like every other collaborator.
upstream_budget_exhausted_total: ?u64 = null,
};
pub fn handle(
@@ -271,7 +277,10 @@ pub fn collect(state: *server.WebState, io: std.Io, arena: Allocator) Allocator.
if (state.upstreams) |owner| {
const generation = owner.acquire(io);
defer owner.release(io, generation);
if (generation.pool) |pool| sample.upstreams = try upstreams(pool, io, arena);
if (generation.pool) |pool| {
sample.upstreams = try upstreams(pool, io, arena);
sample.upstream_budget_exhausted_total = pool.budgetExhaustedTotal();
}
}
return sample;
@@ -465,6 +474,15 @@ pub fn render(w: *std.Io.Writer, sample: Sample) std.Io.Writer.Error!void {
if (sample.doh_certs != null or sample.dot_certs != null) try renderCerts(w, sample);
if (sample.upstreams.len != 0) try renderUpstreams(w, sample.upstreams);
if (sample.upstream_budget_exhausted_total) |total| {
try labeledHead(
w,
"nxdns_upstream_budget_exhausted_total",
"Exchanges that ran out of their own total budget before any upstream answered.",
"counter",
);
try w.print("nxdns_upstream_budget_exhausted_total {d}\n", .{total});
}
}
fn renderCerts(w: *std.Io.Writer, sample: Sample) std.Io.Writer.Error!void {
@@ -1540,7 +1558,7 @@ test "the queue families carry what a real pool recorded, through the real snaps
.priority = 10,
.enabled = true,
.health = .init,
.sem = .{ .permits = slots.len },
.admission = .{ .permits = slots.len },
.reuse_recoveries = &recoveries,
}};
var pool: pool_mod.Pool = .init(&entries, .{}, .{
@@ -1662,3 +1680,22 @@ fn fieldIndex(comptime name: []const u8) usize {
}
@compileError("no such counter: " ++ name);
}
test "the budget-exhausted counter renders pool-wide, without a url label" {
const text = try renderToString(testing.allocator, .{ .upstream_budget_exhausted_total = 7 });
defer testing.allocator.free(text);
try testing.expect(std.mem.containsAtLeast(
u8,
text,
1,
"# TYPE nxdns_upstream_budget_exhausted_total counter\n",
));
try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_upstream_budget_exhausted_total 7\n"));
// No pool, no series: an operator tells "no upstreams wired" from "zero
// exhausted budgets" the same way every other collaborator is told.
const absent = try renderToString(testing.allocator, .{});
defer testing.allocator.free(absent);
try testing.expect(!std.mem.containsAtLeast(u8, absent, 1, "nxdns_upstream_budget_exhausted_total"));
}
src/web/web_integration_test.zig
+1 -1
View File
@@ -495,7 +495,7 @@ const Env = struct {
.priority = 1,
.enabled = true,
.health = .init,
.sem = .{ .permits = self.pool_slots.len },
.admission = .{ .permits = self.pool_slots.len },
.reuse_recoveries = &self.pool_recoveries,
}};
self.pool_owner = .{};