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

milestone 31: concurrent upstream exchanges, dot session reuse, queue metrics

Browse Source
This commit is contained in:
mokhtar
2026-08-22 19:54:02 +02:00
parent 623667e475
commit c99a37d170
14 changed files with 2099 additions and 188 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/app.zig
+126 -49
View File
@@ -529,8 +529,8 @@ fn serve(r: cli.Runner, args: cli.RunArgs) !u8 {
defer bundle.deinit(gpa);
var bundle_lock: std.Io.RwLock = .init;
var upstreams = try Upstreams.build(gpa, cfg.upstreams, &dns_http, &bundle, &bundle_lock, &config_load);
defer upstreams.deinit(gpa);
var upstreams = try Upstreams.build(io, gpa, cfg.upstreams, &dns_http, &bundle, &bundle_lock, &config_load);
defer upstreams.deinit(io, gpa);
var pool: pool_mod.Pool = .init(
upstreams.active(),
@@ -1146,16 +1146,34 @@ fn maintenanceOnce(
/// The pool's entries and everything they point into.
///
/// `Pool.Entry.client` is a type-erased pointer into `doh` or `dot`, and each
/// client borrows a slice of `doh_buf`/`dot_buf`, so all five allocations live
/// exactly as long as the pool does. One entry is used by one task at a time
/// (`Entry.busy`), which is why the buffers are per client and not shared the
/// way `cli.probeUpstreams` shares them.
/// Every enabled upstream gets `pool_mod.slots_per_entry` leaf clients, one per
/// slot of its entry, so that many exchanges can be in flight against it at
/// once. `Slot.client` is a type-erased pointer into `doh` or `dot`, each of
/// those clients borrows a slice of `doh_buf`/`dot_buf`, and each entry borrows
/// a run of `slot_storage` and one counter of `recovery_counters` — so every
/// allocation here lives exactly as long as the pool does, and none of them is
/// ever resized. One slot is used by one task at a time, which is why the
/// buffers are per client and not shared the way `cli.probeUpstreams` shares
/// them.
const Upstreams = struct {
entries: []pool_mod.Entry,
used: usize,
/// Sliced per entry into `Entry.slots`, never pointing into the client
/// arrays: `Pool.init` sorts entries and the slices have to survive it.
slot_storage: []pool_mod.Slot,
/// One per enabled upstream, and the reason it is a separate allocation:
/// `Pool.init` sorts entries by value, so a counter living inside an entry
/// would be pointed at by the wrong upstream's clients after the sort.
recovery_counters: []std.atomic.Value(u64),
doh: []doh_client.DohClient,
dot: []dot_client.DotClient,
/// How much of `doh`/`dot` was actually initialized. A malformed or skipped
/// upstream leaves the tail of an over-allocated array undefined, and both
/// `deinit` and `build`'s failure paths iterate only the initialized
/// prefix — reading a `DotClient` that was never built, or closing a
/// session that was never opened, is what these two counts prevent.
doh_used: usize,
dot_used: usize,
doh_buf: []u8,
dot_buf: []u8,
@@ -1163,6 +1181,7 @@ const Upstreams = struct {
/// skipped, because one bad row in a table of four must not take DNS down.
/// No usable row at all is a configuration fault.
fn build(
io: std.Io,
gpa: Allocator,
servers: []const model.UpstreamServer,
http: *std.http.Client,
@@ -1177,24 +1196,30 @@ const Upstreams = struct {
if (enabled == 0) return error.NoUsableUpstreams;
const chunk = tls.Client.min_buffer_len;
const slots = pool_mod.slots_per_entry;
const leaf_clients = enabled * slots;
var self: Upstreams = .{
.entries = try gpa.alloc(pool_mod.Entry, enabled),
.used = 0,
.slot_storage = &.{},
.recovery_counters = &.{},
.doh = &.{},
.dot = &.{},
.doh_used = 0,
.dot_used = 0,
.doh_buf = &.{},
.dot_buf = &.{},
};
errdefer self.deinit(gpa);
errdefer self.deinit(io, gpa);
self.doh = try gpa.alloc(doh_client.DohClient, enabled);
self.dot = try gpa.alloc(dot_client.DotClient, enabled);
self.doh_buf = try gpa.alloc(u8, enabled * (doh_request_buf_len + doh_transfer_buf_len));
self.dot_buf = try gpa.alloc(u8, enabled * 4 * chunk);
var doh_count: usize = 0;
var dot_count: usize = 0;
self.slot_storage = try gpa.alloc(pool_mod.Slot, leaf_clients);
self.recovery_counters = try gpa.alloc(std.atomic.Value(u64), enabled);
for (self.recovery_counters) |*counter| counter.* = .init(0);
self.doh = try gpa.alloc(doh_client.DohClient, leaf_clients);
self.dot = try gpa.alloc(dot_client.DotClient, leaf_clients);
self.doh_buf = try gpa.alloc(u8, leaf_clients * (doh_request_buf_len + doh_transfer_buf_len));
self.dot_buf = try gpa.alloc(u8, leaf_clients * 4 * chunk);
for (servers) |server| {
if (!server.enabled) continue;
@@ -1208,46 +1233,27 @@ const Upstreams = struct {
continue;
};
const client: transport.Client = switch (endpoint.scheme) {
.doh => doh: {
const base = doh_count * (doh_request_buf_len + doh_transfer_buf_len);
const slot = &self.doh[doh_count];
slot.* = doh_client.DohClient.init(
http,
endpoint,
self.doh_buf[base..][0..doh_request_buf_len],
self.doh_buf[base + doh_request_buf_len ..][0..doh_transfer_buf_len],
) catch {
log.warn(
"upstream {f} is not a usable DoH url; skipped",
.{safe_url.redactQuoted(server.url)},
);
noteUpstream(config_load, server.url, "not a usable DoH url; skipped");
continue;
};
doh_count += 1;
break :doh slot.client();
const entry_slots = self.slot_storage[self.used * slots ..][0..slots];
switch (endpoint.scheme) {
.doh => if (!self.wireDoh(http, endpoint, entry_slots)) {
log.warn(
"upstream {f} is not a usable DoH url; skipped",
.{safe_url.redactQuoted(server.url)},
);
noteUpstream(config_load, server.url, "not a usable DoH url; skipped");
continue;
},
.dot => dot: {
const base = dot_count * 4 * chunk;
const slot = &self.dot[dot_count];
slot.* = dot_client.DotClient.init(endpoint, server.tls_name, gpa, bundle, bundle_lock, .{
.tls_read = self.dot_buf[base..][0..chunk],
.tls_write = self.dot_buf[base + chunk ..][0..chunk],
.stream_read = self.dot_buf[base + 2 * chunk ..][0..chunk],
.stream_write = self.dot_buf[base + 3 * chunk ..][0..chunk],
});
dot_count += 1;
break :dot slot.client();
},
};
.dot => self.wireDot(gpa, endpoint, server.tls_name, bundle, bundle_lock, entry_slots),
}
self.entries[self.used] = .{
.endpoint = endpoint,
.client = client,
.slots = entry_slots,
.priority = server.priority,
.enabled = true,
.health = .init,
.sem = .{ .permits = entry_slots.len },
.reuse_recoveries = &self.recovery_counters[self.used],
};
self.used += 1;
}
@@ -1256,17 +1262,88 @@ const Upstreams = struct {
return self;
}
/// One `DohClient` per slot, all sharing the one `std.http.Client`: its
/// connection pool already serves concurrent requests, and a `DohClient`'s
/// only mutable state is the two buffers this gives each slot its own of.
///
/// False means the url is not a usable DoH url, which `DohClient.init`
/// decides from the url alone — so it fails on the first slot or on none.
/// `doh_used` still advances per client rather than per entry: it means
/// "initialized", and a skipped entry's clients are simply never reached.
fn wireDoh(
self: *Upstreams,
http: *std.http.Client,
endpoint: transport.Endpoint,
slots: []pool_mod.Slot,
) bool {
for (slots) |*slot| {
const index = self.doh_used;
const base = index * (doh_request_buf_len + doh_transfer_buf_len);
self.doh[index] = doh_client.DohClient.init(
http,
endpoint,
self.doh_buf[base..][0..doh_request_buf_len],
self.doh_buf[base + doh_request_buf_len ..][0..doh_transfer_buf_len],
) catch return false;
self.doh_used = index + 1;
slot.* = .{ .client = self.doh[index].client() };
}
return true;
}
/// One `DotClient` per slot, each with its own four TLS buffers and all
/// sharing the trust store. Every client of one entry reports its stale-reuse
/// recoveries through that entry's counter.
fn wireDot(
self: *Upstreams,
gpa: Allocator,
endpoint: transport.Endpoint,
tls_name: []const u8,
bundle: *Certificate.Bundle,
bundle_lock: *std.Io.RwLock,
slots: []pool_mod.Slot,
) void {
const chunk = tls.Client.min_buffer_len;
const recoveries = &self.recovery_counters[self.used];
for (slots) |*slot| {
const index = self.dot_used;
const base = index * 4 * chunk;
self.dot[index] = dot_client.DotClient.init(
endpoint,
tls_name,
gpa,
bundle,
bundle_lock,
recoveries,
.{
.tls_read = self.dot_buf[base..][0..chunk],
.tls_write = self.dot_buf[base + chunk ..][0..chunk],
.stream_read = self.dot_buf[base + 2 * chunk ..][0..chunk],
.stream_write = self.dot_buf[base + 3 * chunk ..][0..chunk],
},
);
self.dot_used = index + 1;
slot.* = .{ .client = self.dot[index].client() };
}
}
/// The prefix `Pool.init` is given. The rest of `entries` is allocated but
/// never filled, which is what keeps `deinit` able to free the whole block.
fn active(self: *Upstreams) []pool_mod.Entry {
return self.entries[0..self.used];
}
fn deinit(self: *Upstreams, gpa: Allocator) void {
/// Connections first, memory second: a `DotClient` holds a socket its
/// buffers belong to, so nothing it points at may be freed before it is
/// closed.
fn deinit(self: *Upstreams, io: std.Io, gpa: Allocator) void {
for (self.dot[0..self.dot_used]) |*client| client.close(io);
gpa.free(self.dot_buf);
gpa.free(self.doh_buf);
gpa.free(self.dot);
gpa.free(self.doh);
gpa.free(self.recovery_counters);
gpa.free(self.slot_storage);
gpa.free(self.entries);
self.* = undefined;
}
src/cli.zig
+62 -4
View File
@@ -974,10 +974,16 @@ fn probeUpstreams(r: Runner, cfg: model.Config) !usize {
continue;
};
// Both clients are pinned for the pool's lifetime: `Entry.client` is an
// erased pointer into one of them.
// Both clients are pinned for the pool's lifetime: the entry's one slot
// holds an erased pointer into one of them.
var doh: doh_client.DohClient = undefined;
var dot: dot_client.DotClient = undefined;
// A DoT probe leaves a connection open on a client whose buffers the
// next iteration reuses, so it is closed at the end of every iteration —
// on the FAIL paths and on any error out of the loop too. The flag is
// what keeps the close off `dot` while it is still undefined.
var dot_wired = false;
defer if (dot_wired) dot.close(r.io);
const client: transport.Client = switch (endpoint.scheme) {
.doh => doh: {
doh = doh_client.DohClient.init(&http, endpoint, &request_buf, &transfer_buf) catch {
@@ -988,22 +994,31 @@ fn probeUpstreams(r: Runner, cfg: model.Config) !usize {
break :doh doh.client();
},
.dot => dot: {
dot = dot_client.DotClient.init(endpoint, server.tls_name, r.gpa, &bundle, &bundle_lock, .{
dot = dot_client.DotClient.init(endpoint, server.tls_name, r.gpa, &bundle, &bundle_lock, null, .{
.tls_read = tls_buffers[0..chunk],
.tls_write = tls_buffers[chunk .. 2 * chunk],
.stream_read = tls_buffers[2 * chunk .. 3 * chunk],
.stream_write = tls_buffers[3 * chunk ..],
});
dot_wired = true;
break :dot dot.client();
},
};
// One slot: the probe is sequential by design, and one exchange per
// upstream is the whole of it.
var slots = [_]pool.Slot{.{ .client = client }};
// The pool's counter has to point somewhere, and a probe has no
// process-wide counter storage to point it at.
var recoveries: std.atomic.Value(u64) = .init(0);
var entries = [_]pool.Entry{.{
.endpoint = endpoint,
.client = client,
.slots = &slots,
.priority = server.priority,
.enabled = true,
.health = .init,
.sem = .{ .permits = slots.len },
.reuse_recoveries = &recoveries,
}};
var single: pool.Pool = .init(&entries, .{}, timeouts, seed);
@@ -1746,6 +1761,49 @@ test "the upstream probe redacts a url it cannot parse, without leaving the mach
try testing.expect(!std.mem.containsAtLeast(u8, text, 1, "efgh34"));
}
test "a failed DoT probe reaches close through the per-iteration defer without a session" {
// Exactly that and no more. This is the only test that reaches the DoT
// branch at all: it arms `dot_wired`, fails the dial, and runs the deferred
// `DotClient.close` on a client that never opened a session — twice, so the
// second iteration's dial happens after the first close. What that proves is
// that the path compiles and survives: `dot_wired` set only after `init`,
// and `close` tolerant of a sessionless client. It does *not* prove the
// defer is load-bearing — with no session open, deleting the `defer` would
// leave this test green.
//
// The property that matters — a probe's connection not surviving into the
// next iteration's shared TLS buffers — needs a session to exist, so it is
// pinned one level down on the client itself, by the close-after-failure
// assertions in `dot_client_integration_test.zig`. A successful DoT probe
// cannot be pinned in-repo at all: it needs a real handshake, and the only
// DoT peer this repo has is the self-signed `-Dintegration` loopback
// fixture, which `probeUpstreams` cannot verify because it builds its bundle
// from the system trust store.
//
// `tls://` with a host that is not an IP literal fails inside `dial` before
// any socket is opened or any CA store is read: upstream name resolution is
// out of scope, so `resolveAddress` refuses it. That keeps the test off the
// network and off the host's configuration.
var captured: Captured = .init(testing.allocator);
defer captured.deinit();
const r = captured.runner();
const cfg: model.Config = .{
.groups = &.{.{ .name = "default" }},
.upstreams = &.{
.{ .url = "tls://dns.example:853", .tls_name = "dns.example" },
.{ .url = "tls://other.example:853", .tls_name = "other.example" },
},
};
try testing.expectEqual(@as(usize, 2), try probeUpstreams(r, cfg));
try testing.expectEqualStrings(
"FAIL upstreams[0] 'tls://dns.example:853': ConnectFailed\n" ++
"FAIL upstreams[1] 'tls://other.example:853': ConnectFailed\n",
captured.out.written(),
);
}
test "a successful import prints the warnings the file earned" {
// D5, second half. `check` printed this WARN and `import` recorded it and
// threw it away, because diagnostics were written on the failure path only.
src/server/resolver_integration_test.zig
+30 -11
View File
@@ -155,15 +155,32 @@ const FaultyUpstream = struct {
}
};
fn testEntry(url: []const u8, upstream_client: transport.Client, priority: i32) pool.Entry {
return .{
.endpoint = transport.Endpoint.parse(url) catch unreachable,
.client = upstream_client,
.priority = priority,
.enabled = true,
.health = .init,
};
}
/// One-slot entry storage: the slot the entry points at and the recovery
/// counter it borrows, declared by the test so both outlive the pool. These
/// tests drive failover and deadlines, not concurrency within one upstream, so
/// one slot per entry is the whole story here.
const EntryStorage = struct {
slots: [1]pool.Slot = undefined,
recoveries: std.atomic.Value(u64) = .init(0),
fn entry(
self: *EntryStorage,
url: []const u8,
upstream_client: transport.Client,
priority: i32,
) pool.Entry {
self.slots[0] = .{ .client = upstream_client };
return .{
.endpoint = transport.Endpoint.parse(url) catch unreachable,
.slots = &self.slots,
.priority = priority,
.enabled = true,
.health = .init,
.sem = .{ .permits = self.slots.len },
.reuse_recoveries = &self.recoveries,
};
}
};
const Outcome = union(enum) {
work: anyerror!void,
@@ -250,9 +267,11 @@ test "the whole resolver answers over udp and tcp and fails over to a healthy up
.fail_first = std.math.maxInt(u64),
};
var good: GoodUpstream = .{};
var bad_storage: EntryStorage = .{};
var good_storage: EntryStorage = .{};
var entries = [_]pool.Entry{
testEntry("https://bad.example/dns-query", bad.client(), 10),
testEntry("tls://good.example", good.client(), 20),
bad_storage.entry("https://bad.example/dns-query", bad.client(), 10),
good_storage.entry("tls://good.example", good.client(), 20),
};
var upstreams: pool.Pool = .init(&entries, test_cfg, pool_timeouts, 1);
src/tests.zig
+1
View File
@@ -24,6 +24,7 @@ comptime {
_ = @import("upstream/pool.zig");
_ = @import("upstream/dot_client.zig");
_ = @import("upstream/dot_client_live_test.zig");
_ = @import("upstream/dot_client_integration_test.zig");
_ = @import("server/listener.zig");
_ = @import("server/handler.zig");
_ = @import("server/udp_server.zig");
src/upstream/dot_client.zig
+339 -36
View File
@@ -10,6 +10,16 @@
//! local resource error or a cancellation must never reach the pool as a peer
//! fault, so the concrete error is unwrapped from `error.ReadFailed` /
//! `error.WriteFailed` before it is classified.
//!
//! A client keeps its connection open between exchanges (RFC 7858 §3.4). An
//! upstream is free to drop an idle one at any time and says nothing first, so
//! staleness is detected at use: an exchange that fails on a *reused* session,
//! before any byte of its response arrived, with a connection-lifecycle cause,
//! is retried exactly once on a fresh dial and counted through
//! `reuse_recoveries` rather than against the upstream's health. Every other
//! failure is final, and every final failure closes the session — after a
//! failed send, receive or validation, the framing and TLS state are
//! untrustworthy and the next exchange must start from a fresh dial.
const std = @import("std");
const net = std.Io.net;
@@ -41,6 +51,9 @@ const Diagnostic = struct {
connect_failed: anyerror,
handshake_failed: Handshake,
bundle_load_failed: anyerror,
/// A reused session turned out to be dead. Normal operation, so this
/// one is only ever logged at debug; the counter is the real surface.
stale_session: anyerror,
const Handshake = struct {
verify_name: []const u8,
@@ -65,6 +78,10 @@ const Diagnostic = struct {
tls_client.classify(hs.cause),
}),
.bundle_load_failed => |err| try w.print("CA bundle load failed: {s}", .{@errorName(err)}),
.stale_session => |err| try w.print(
"reused session was stale ({s}), redialing once",
.{@errorName(err)},
),
}
}
};
@@ -93,6 +110,25 @@ pub const DotClient = struct {
bundle_lock: *std.Io.RwLock,
/// Caller-owned. One `DotClient` is used by one task at a time.
buffers: Buffers,
/// Points at the owning pool entry's counter, which lives in the
/// composition root's stable storage; null in `nxdns check` probes, which
/// have no pool to report through. Incremented when a stale reused session
/// is recovered by a redial.
reuse_recoveries: ?*std.atomic.Value(u64),
/// The connection this client keeps open between exchanges, or null when it
/// holds none. Owned here: nothing else may close either half.
///
/// Emplaced, never assigned from a local: `TlsStream` hands `tls.Client`
/// pointers into its own reader and writer fields, so a `Session` built on
/// the stack and copied in would leave the TLS client pointing at the dead
/// copy. `dial` sets this to `.{ .stream = …, .tls = undefined }` first and
/// runs the handshake through the stored payload.
session: ?Session = null,
const Session = struct {
stream: net.Stream,
tls: tls_client.TlsStream,
};
pub const Buffers = struct {
/// Plaintext read buffer.
@@ -116,6 +152,7 @@ pub const DotClient = struct {
gpa: std.mem.Allocator,
bundle: *Certificate.Bundle,
bundle_lock: *std.Io.RwLock,
reuse_recoveries: ?*std.atomic.Value(u64),
buffers: Buffers,
) DotClient {
std.debug.assert(endpoint.scheme == .dot);
@@ -130,9 +167,25 @@ pub const DotClient = struct {
.bundle = bundle,
.bundle_lock = bundle_lock,
.buffers = buffers,
.reuse_recoveries = reuse_recoveries,
};
}
/// Releases whatever this client holds open between exchanges: TLS first,
/// so the peer gets a close_notify, then the socket underneath it.
///
/// Idempotent and safe with no session open, because that is how every
/// caller uses it — `Upstreams.deinit` closes clients that may never have
/// dialed, the `nxdns check` probe loop closes on its error paths, and
/// `exchange` closes after a failure it has already classified.
pub fn close(self: *DotClient, io: std.Io) void {
if (self.session) |*session| {
transport.closeBlocked(io, &session.tls);
transport.closeBlocked(io, &session.stream);
self.session = null;
}
}
pub fn client(self: *DotClient) transport.Client {
return .{ .ptr = self, .exchangeFn = exchangeFn };
}
@@ -155,14 +208,15 @@ pub const DotClient = struct {
return self.exchange(io, query, response_buf);
}
/// One TCP connection and one TLS handshake per exchange, both closed
/// before returning.
/// One framed query and one framed reply over a session this client keeps
/// open, dialing one only when it holds none.
///
/// Connection reuse is deliberately not built. At household query rates the
/// saved round trips are worth less than what a per-exchange connection
/// buys: the pool's per-attempt budget stays a plain race against one task,
/// the failover path never has to reason about a half-dead pooled socket,
/// and every failure is attributable to exactly one exchange.
/// A reused session may already be dead: the upstream is entitled to close
/// an idle connection and RFC 7858 keepalive is advisory, so the first sign
/// is this exchange failing. That case — and only that case — is retried
/// once on a fresh dial; see `retryDecision` for the three conditions. The
/// caller's attempt budget covers the whole call including that redial,
/// because the pool races this function as a whole.
pub fn exchange(
self: *DotClient,
io: std.Io,
@@ -174,6 +228,49 @@ pub const DotClient = struct {
// local error rather than a silently truncated frame.
if (query.len > transport.max_message_len) return error.BufferTooSmall;
const reused = self.session != null;
if (!reused) try self.dial(io);
const failure = switch (self.transact(query, response_buf)) {
.ok => |reply| return reply,
.failed => |failure| failure,
};
switch (retryDecision(reused, failure.received_any, failure.cause)) {
.final => {
self.close(io);
return transport.mapPhase(failure.cause, failure.phase);
},
.retry => {},
}
// Debug, not warn: an upstream dropping an idle connection is routine,
// and one line per idle timeout on a household resolver is log spam.
// `reuse_recoveries` is what makes the churn visible.
log.debug("{f}", .{self.diagnose(.{ .stale_session = failure.cause })});
self.close(io);
try self.dial(io);
switch (self.transact(query, response_buf)) {
.ok => |reply| {
if (self.reuse_recoveries) |counter| _ = counter.fetchAdd(1, .monotonic);
return reply;
},
// The retry's outcome is the exchange's outcome: one redial, never
// two.
.failed => |retried| {
self.close(io);
return transport.mapPhase(retried.cause, retried.phase);
},
}
}
/// Opens a session and leaves it in `self.session`, or leaves `self.session`
/// null and returns the classified failure. No partially initialized
/// session ever survives this call.
fn dial(self: *DotClient, io: std.Io) transport.ExchangeError!void {
std.debug.assert(self.session == null);
const address = resolveAddress(self.endpoint) catch |err| {
log.warn("{f}", .{self.diagnose(.not_an_ip_literal)});
return err;
@@ -181,22 +278,24 @@ pub const DotClient = struct {
try self.ensureBundle(io);
var stream = address.connect(io, .{ .mode = .stream }) catch |err| {
const stream = address.connect(io, .{ .mode = .stream }) catch |err| {
log.debug("{f}", .{self.diagnose(.{ .connect_failed = err })});
return transport.mapPhase(err, error.ConnectFailed);
};
defer transport.closeBlocked(io, &stream);
// `TlsStream` is pinned: it holds its reader and writer by value and the
// TLS client points at them, so it must not move after `init`.
var tls_stream: tls_client.TlsStream = undefined;
// Emplaced before the handshake, never built beside it and copied in:
// `TlsStream` is pinned by the pointers `tls.Client` holds into its own
// reader and writer fields, so `init` has to run at the address the
// session will live at.
self.session = .{ .stream = stream, .tls = undefined };
const session = &self.session.?;
// `concreteHandshake` reads these two fields when the handshake reports
// `error.ReadFailed` / `error.WriteFailed`. `TlsStream.init` sets them
// before it can produce either error, but clearing them here keeps that
// out of this file's correctness argument.
tls_stream.stream_reader.err = null;
tls_stream.stream_writer.err = null;
tls_stream.init(io, &stream, self.bundle, self.bundle_lock, self.gpa, .{
session.tls.stream_reader.err = null;
session.tls.stream_writer.err = null;
session.tls.init(io, &session.stream, self.bundle, self.bundle_lock, self.gpa, .{
.host = self.verify_name,
.ca = .system,
.read_buffer = self.buffers.tls_read,
@@ -204,36 +303,72 @@ pub const DotClient = struct {
.stream_read_buffer = self.buffers.stream_read,
.stream_write_buffer = self.buffers.stream_write,
}) catch |err| {
const cause = concreteHandshake(&tls_stream, err);
// Order matters: the concrete cause lives in the in-place
// `TlsStream`'s two error fields, so clearing the optional first
// would destroy the only record that a cancellation or a local
// resource failure — not the peer — ended the handshake.
const cause = concreteHandshake(&session.tls, err);
transport.closeBlocked(io, &session.stream);
self.session = null;
log.warn("{f}", .{self.diagnose(.{ .handshake_failed = .{
.verify_name = self.verify_name,
.cause = cause,
} })});
return transport.mapPhase(cause, error.TlsFailed);
};
defer transport.closeBlocked(io, &tls_stream);
}
/// One query and one reply on the open session, with enough detail on
/// failure for `retryDecision` to rule on it. Leaves the session open
/// either way; closing a failed one is `exchange`'s job, because only it
/// knows whether the failure is final.
fn transact(self: *DotClient, query: []const u8, response_buf: []u8) Transact {
const session = &self.session.?;
const tls_stream = &session.tls;
const writer = tls_stream.writer();
const prefix = transport.framePrefix(@intCast(query.len));
writer.writeAll(&prefix) catch |err| return sendFailure(&tls_stream, err);
writer.writeAll(query) catch |err| return sendFailure(&tls_stream, err);
writer.writeAll(&prefix) catch |err| return sendFailed(tls_stream, err);
writer.writeAll(query) catch |err| return sendFailed(tls_stream, err);
// `TlsStream.flush`, not `writer.flush`: the latter leaves the encrypted
// record in the socket writer's buffer and the query never leaves this
// process.
tls_stream.flush() catch |err| return sendFailure(&tls_stream, err);
tls_stream.flush() catch |err| return sendFailed(tls_stream, err);
const reader = tls_stream.reader();
// Byte at a time, not `readSliceAll`: that call is all-or-nothing, so a
// failure after the first prefix byte would be indistinguishable from
// one before it, and "no response byte received yet" is one of the three
// conditions a retry needs.
var prefix_bytes: [transport.prefix_len]u8 = undefined;
reader.readSliceAll(&prefix_bytes) catch |err| return receiveFailure(&tls_stream, err);
for (&prefix_bytes, 0..) |*byte, received| {
byte.* = reader.takeByte() catch |err|
return receiveFailed(tls_stream, err, received != 0);
}
const len = transport.parsePrefix(prefix_bytes);
if (len == 0) return error.BadResponse;
if (len > response_buf.len) return error.ResponseTooLarge;
if (len == 0) return .{ .failed = .{
.cause = error.BadResponse,
.phase = error.BadResponse,
.received_any = true,
} };
if (len > response_buf.len) return .{ .failed = .{
.cause = error.ResponseTooLarge,
.phase = error.ResponseTooLarge,
.received_any = true,
} };
reader.readSliceAll(response_buf[0..len]) catch |err|
return receiveFailure(&tls_stream, err);
return receiveFailed(tls_stream, err, true);
try transport.validateResponse(query, response_buf[0..len]);
return response_buf[0..len];
transport.validateResponse(query, response_buf[0..len]) catch |err| return .{ .failed = .{
.cause = err,
.phase = switch (err) {
error.BadResponse => error.BadResponse,
error.ResponseMismatch => error.ResponseMismatch,
},
.received_any = true,
} };
return .{ .ok = response_buf[0..len] };
}
/// Loads the system CA bundle before the handshake, so that a failure to
@@ -301,14 +436,72 @@ fn concreteWrite(stream: *tls_client.TlsStream, err: anyerror) anyerror {
return err;
}
fn sendFailure(stream: *tls_client.TlsStream, err: anyerror) transport.ExchangeError {
return transport.mapPhase(concreteWrite(stream, err), error.SendFailed);
/// The outcome of one `transact`, as a value: a reply, or everything
/// `retryDecision` needs to rule on the failure.
const Transact = union(enum) {
ok: []u8,
failed: Failure,
const Failure = struct {
/// Already unwrapped out of `error.ReadFailed` / `error.WriteFailed`.
cause: anyerror,
/// The peer fault this phase means, unless `cause` belongs to this
/// process.
phase: transport.PeerFault,
/// Whether any byte of this exchange's response had arrived.
received_any: bool,
};
};
fn sendFailed(stream: *tls_client.TlsStream, err: anyerror) Transact {
return .{
.failed = .{
.cause = concreteWrite(stream, err),
.phase = error.SendFailed,
// Nothing is read before the query is out, so a send failure is always
// pre-first-byte.
.received_any = false,
},
};
}
fn receiveFailure(stream: *tls_client.TlsStream, err: anyerror) transport.ExchangeError {
return transport.mapPhase(concreteRead(stream, err), error.ReceiveFailed);
fn receiveFailed(stream: *tls_client.TlsStream, err: anyerror, received_any: bool) Transact {
return .{ .failed = .{
.cause = concreteRead(stream, err),
.phase = error.ReceiveFailed,
.received_any = received_any,
} };
}
/// Whether a failed exchange may be retried once on a fresh dial.
///
/// Pure, and separate from the wire for that reason: the three conditions are
/// the whole of the reuse contract, and a table test is the only way to see all
/// of them at once.
///
/// `reused` — a session this call dialed itself was never idle, so its failure
/// is the upstream's answer, not a stale connection. `bytes_received` — once a
/// reply has started, re-sending the query would be a second question, and the
/// failure is the upstream's. `cause` — only the ways a connection ends
/// (`error.EndOfStream` is a clean close_notify, `error.TlsConnectionTruncated`
/// a close without one, `error.ConnectionResetByPeer` and `error.BrokenPipe`
/// the socket-level pair). A TLS alert, a certificate fault, a local resource
/// failure and a cancellation all keep their meaning and are final.
fn retryDecision(reused: bool, bytes_received: bool, cause: anyerror) RetryDecision {
if (!reused) return .final;
if (bytes_received) return .final;
return switch (cause) {
error.EndOfStream,
error.TlsConnectionTruncated,
error.ConnectionResetByPeer,
error.BrokenPipe,
=> .retry,
else => .final,
};
}
const RetryDecision = enum { retry, final };
const testing = std.testing;
fn expectDiagnostic(expected: []const u8, url: []const u8, detail: Diagnostic.Detail) !void {
@@ -343,6 +536,11 @@ test "every diagnostic line redacts the url it names the upstream by" {
"tls://dns.example/abcd12?apikey=s3cr3t",
.{ .bundle_load_failed = error.FileNotFound },
);
try expectDiagnostic(
"dot upstream 'tls://dns.example:853': reused session was stale (EndOfStream), redialing once",
"tls://dns.example:853/",
.{ .stale_session = error.EndOfStream },
);
try expectDiagnostic(
"dot upstream 'tls://dns.example': TLS handshake as 'one.one.one.one' failed: " ++
"CertificateHostMismatch (certificate)",
@@ -497,27 +695,126 @@ test "the handshake unwrap passes other errors through untouched" {
);
}
/// What `exchange` would return for a failure `transact` reported.
fn mappedFailure(outcome: Transact) transport.ExchangeError {
return transport.mapPhase(outcome.failed.cause, outcome.failed.phase);
}
test "the send and receive unwraps prefer the stored cause" {
var send = stubStream(null, error.Canceled, null);
try testing.expectEqual(
transport.ExchangeError.Canceled,
sendFailure(&send, error.WriteFailed),
mappedFailure(sendFailed(&send, error.WriteFailed)),
);
// The TLS client's own error wins over the socket reader's.
var receive = stubStream(error.ConnectionResetByPeer, null, error.TlsAlert);
try testing.expectEqual(
transport.ExchangeError.ReceiveFailed,
receiveFailure(&receive, error.ReadFailed),
mappedFailure(receiveFailed(&receive, error.ReadFailed, false)),
);
var socket = stubStream(error.SystemResources, null, null);
try testing.expectEqual(
transport.ExchangeError.SystemResources,
receiveFailure(&socket, error.ReadFailed),
mappedFailure(receiveFailed(&socket, error.ReadFailed, true)),
);
}
test "a send failure is always pre-first-byte, and a receive failure reports what it read" {
// The retry rule reads `received_any`, so where it comes from is part of the
// contract rather than an incidental field: nothing is read before the query
// is on the wire, and the receive side is told by its caller.
var stream = stubStream(error.ConnectionResetByPeer, error.ConnectionResetByPeer, null);
try testing.expect(!sendFailed(&stream, error.WriteFailed).failed.received_any);
try testing.expect(!receiveFailed(&stream, error.ReadFailed, false).failed.received_any);
try testing.expect(receiveFailed(&stream, error.ReadFailed, true).failed.received_any);
}
test "a fresh session is never retried, whatever failed" {
// The redial has nothing to fix: this call dialed the connection itself, so
// the failure is the upstream's answer rather than a stale socket.
for ([_]anyerror{
error.EndOfStream,
error.TlsConnectionTruncated,
error.ConnectionResetByPeer,
error.BrokenPipe,
error.TlsAlert,
error.Canceled,
}) |cause| {
try testing.expectEqual(RetryDecision.final, retryDecision(false, false, cause));
try testing.expectEqual(RetryDecision.final, retryDecision(false, true, cause));
}
}
test "a reused session is retried once for the ways a connection ends" {
for ([_]anyerror{
error.EndOfStream,
error.TlsConnectionTruncated,
error.ConnectionResetByPeer,
error.BrokenPipe,
}) |cause| {
try testing.expectEqual(RetryDecision.retry, retryDecision(true, false, cause));
}
}
test "a reused session that already answered is never retried" {
// Re-sending the query after a byte of the reply arrived would ask the
// upstream a second question, so a mid-reply failure is final however the
// connection died.
for ([_]anyerror{
error.EndOfStream,
error.TlsConnectionTruncated,
error.ConnectionResetByPeer,
error.BrokenPipe,
}) |cause| {
try testing.expectEqual(RetryDecision.final, retryDecision(true, true, cause));
}
}
test "a reused session is not retried for a fault that says something" {
// A TLS alert, a certificate fault, a local resource failure, a
// cancellation and a bad frame all keep their meaning: none of them is an
// idle connection going away, so none is worth a second dial.
for ([_]anyerror{
error.TlsAlert,
error.TlsBadRecordMac,
error.CertificateExpired,
error.SystemResources,
error.OutOfMemory,
error.Canceled,
error.BadResponse,
error.ResponseMismatch,
error.ResponseTooLarge,
error.ConnectionRefused,
error.Timeout,
}) |cause| {
try testing.expectEqual(RetryDecision.final, retryDecision(true, false, cause));
}
}
test "a validation failure is final and its phase survives the mapping" {
// `transact` reports these with `received_any` set, so the decision is final
// by two of the three conditions at once, and the peer fault the pool
// records is the validation error itself rather than a receive failure.
const outcomes = [_]struct { cause: anyerror, phase: transport.PeerFault }{
.{ .cause = error.BadResponse, .phase = error.BadResponse },
.{ .cause = error.ResponseMismatch, .phase = error.ResponseMismatch },
.{ .cause = error.ResponseTooLarge, .phase = error.ResponseTooLarge },
};
for (outcomes) |outcome| {
try testing.expectEqual(RetryDecision.final, retryDecision(true, true, outcome.cause));
try testing.expectEqual(
@as(transport.ExchangeError, outcome.phase),
mappedFailure(.{ .failed = .{
.cause = outcome.cause,
.phase = outcome.phase,
.received_any = true,
} }),
);
}
}
test "a CA bundle scan failure keeps local resource errors out of the peer fault group" {
// `Certificate.Bundle.rescan` reaches these through `Allocator.Error`,
// `Io.File.OpenError` and `Io.UnexpectedError`.
@@ -565,7 +862,7 @@ test "DotClient satisfies the Client interface" {
// `init` asserts `endpoint.scheme == .dot`; a `.doh` endpoint trips
// `std.debug.assert`, which a test cannot catch in-process.
var dot: DotClient = .init(try .parse("tls://9.9.9.9:853"), "", gpa, &bundle, &bundle_lock, .{
var dot: DotClient = .init(try .parse("tls://9.9.9.9:853"), "", gpa, &bundle, &bundle_lock, null, .{
.tls_read = buffer[0..chunk],
.tls_write = buffer[chunk .. 2 * chunk],
.stream_read = buffer[2 * chunk .. 3 * chunk],
@@ -574,6 +871,12 @@ test "DotClient satisfies the Client interface" {
try testing.expectEqual(transport.Scheme.dot, dot.endpoint.scheme);
try testing.expectEqualStrings("9.9.9.9", dot.endpoint.host);
// A client is wired with nothing open, so `init` can return by value and
// the composition root can copy the result into its array: the pinned TLS
// state only exists once `exchange` has dialed.
try testing.expect(dot.session == null);
dot.close(undefined);
try testing.expect(dot.session == null);
const iface: transport.Client = dot.client();
try testing.expectEqual(@as(*anyopaque, @ptrCast(&dot)), iface.ptr);
@@ -598,7 +901,7 @@ test "a tls_name replaces the verification name and leaves the dial target alone
};
const endpoint: transport.Endpoint = try .parse("tls://1.1.1.1:853");
const named: DotClient = .init(endpoint, "one.one.one.one", gpa, &bundle, &bundle_lock, buffers);
const named: DotClient = .init(endpoint, "one.one.one.one", gpa, &bundle, &bundle_lock, null, buffers);
try testing.expectEqualStrings("one.one.one.one", named.verify_name);
try testing.expectEqualStrings("1.1.1.1", named.endpoint.host);
@@ -606,6 +909,6 @@ test "a tls_name replaces the verification name and leaves the dial target alone
try testing.expectEqualSlices(u8, &.{ 1, 1, 1, 1 }, &address.ip4.bytes);
try testing.expectEqual(@as(u16, 853), address.ip4.port);
const plain: DotClient = .init(endpoint, "", gpa, &bundle, &bundle_lock, buffers);
const plain: DotClient = .init(endpoint, "", gpa, &bundle, &bundle_lock, null, buffers);
try testing.expectEqualStrings("1.1.1.1", plain.verify_name);
}
src/upstream/dot_client_integration_test.zig
+651
View File
@@ -0,0 +1,651 @@
//! Hermetic loopback tests for DoT session reuse (`dot_client.zig`).
//!
//! This lives in its own file because it needs `@import("build_options")`,
//! which only exists when the compilation is driven by build.zig, and because a
//! DoT server is a whole fixture rather than a stub. `-Dintegration` gates it;
//! nothing here leaves the machine and nothing here resolves a name.
//!
//! The peer is the same mbedTLS `platform/tls_server.zig` the nxdns listener
//! uses, serving RFC 1035 §4.2.2 framed DNS on the plaintext side. Its
//! certificate is the committed self-signed fixture, which is why every test
//! preloads that certificate into its own `Certificate.Bundle`: `DotClient`
//! hardcodes `.ca = .system` and there is deliberately no way to ask it to skip
//! verification — a fixture-only trust anchor is a test's business, an
//! insecure-verify switch in the production client would be a hole in it.
//!
//! Every test races its client work against a budget. The failures under test
//! are "the client waits for a reply that never comes" shaped, and without a
//! deadline those hang the suite instead of failing it.
const std = @import("std");
const build_options = @import("build_options");
const tls = std.crypto.tls;
const net = std.Io.net;
const Certificate = std.crypto.Certificate;
const dot_client = @import("dot_client.zig");
const pool_mod = @import("pool.zig");
const transport = @import("transport.zig");
const tls_server = @import("../platform/tls_server.zig");
const events_fixture = @import("../storage/events_fixture.zig");
const testing = std.testing;
/// Long enough that no handshake or exchange on loopback needs it, short enough
/// that a hang ends the test rather than the suite.
const budget: std.Io.Clock.Duration = .{ .raw = .fromSeconds(10), .clock = .awake };
/// The fixture certificate carries `DNS:localhost`, and
/// `Certificate.Parsed.verifyHostName` matches dNSName SANs only, so this is
/// the only name a `DotClient` can verify it as. The dial target stays the
/// loopback IP literal.
const fixture_host = "localhost";
/// A query for example.com A: id 0x1234, RD set, one question.
const query_bytes =
"\x12\x34\x01\x00\x00\x01\x00\x00\x00\x00\x00\x00" ++
"\x07example\x03com\x00\x00\x01\x00\x01";
/// The matching response: the question echoed plus one A record.
const response_bytes =
"\x12\x34\x81\x80\x00\x01\x00\x01\x00\x00\x00\x00" ++
"\x07example\x03com\x00\x00\x01\x00\x01" ++
"\xc0\x0c\x00\x01\x00\x01\x00\x00\x01\x2c\x00\x04\x5d\xb8\xd8\x22";
/// The same answer under a different transaction id. Well-framed and the length
/// the prefix claims, so the client reads the whole of it and only
/// `transport.validateResponse` can reject it — which is the point: a validation
/// failure is not an I/O failure, and it has to close the session all the same.
const mismatched_response_bytes =
"\x99\x99\x81\x80\x00\x01\x00\x01\x00\x00\x00\x00" ++
"\x07example\x03com\x00\x00\x01\x00\x01" ++
"\xc0\x0c\x00\x01\x00\x01\x00\x00\x01\x2c\x00\x04\x5d\xb8\xd8\x22";
// ---------------------------------------------------------------------------
// Server side
// ---------------------------------------------------------------------------
/// The listening half: one mbedTLS context, one loopback listener, and the
/// count of connections the scripts below have accepted. That count is what
/// "one connection" and "no redial" are asserted on.
const Server = struct {
ctx: tls_server.ServerContext,
listener: net.Server,
/// Every dial a script chose to `accept`. The listener stays up for the
/// whole of every test on purpose: a refused dial is invisible to this
/// counter, so a script that stopped listening could not tell "the client
/// did not dial again" from "it dialed and was refused". A dial beyond what
/// the script accepts sits un-handshaken in the backlog instead, stalling
/// the client into its budget — the test fails either way, but only an
/// accepted dial shows up in this count.
accepts: std.atomic.Value(u32) = .init(0),
/// Framed queries taken off the wire. A test that has to know the client is
/// blocked waiting for a reply waits on this rather than on a sleep: a
/// Debug-build ECDSA handshake on loopback is slow enough that any fixed
/// gap is either a flake or a stall.
queries: std.atomic.Value(u32) = .init(0),
fn init(self: *Server, gpa: std.mem.Allocator, io: std.Io) !void {
const fixtures = @import("test_fixtures");
self.* = .{
.ctx = try .init(gpa, fixtures.cert_pem, fixtures.key_pem, null),
.listener = try (net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{
.reuse_address = true,
}),
};
}
fn deinit(self: *Server, gpa: std.mem.Allocator, io: std.Io) void {
self.listener.deinit(io);
self.ctx.deinit(gpa);
}
fn address(self: *const Server) net.IpAddress {
return self.listener.socket.address;
}
};
/// One accepted connection. Pinned: `ServerStream` recovers its `std.Io`
/// interfaces from their addresses inside itself and hands mbedTLS a pointer to
/// itself as the BIO context, and its plaintext buffers live here too.
const Accepted = struct {
server: *Server,
stream: net.Stream,
tls: tls_server.ServerStream,
read_buffer: [4096]u8,
write_buffer: [4096]u8,
fn accept(self: *Accepted, gpa: std.mem.Allocator, server: *Server, io: std.Io) !void {
self.server = server;
self.stream = try server.listener.accept(io);
_ = server.accepts.fetchAdd(1, .acq_rel);
errdefer self.stream.close(io);
try self.tls.accept(gpa, &server.ctx, io, &self.stream, &self.read_buffer, &self.write_buffer);
}
fn close(self: *Accepted, gpa: std.mem.Allocator, io: std.Io) void {
self.tls.close(gpa);
self.stream.close(io);
}
/// Reads one framed query and writes `response_bytes` back.
fn answerOnce(self: *Accepted) !void {
var query_buf: [2048]u8 = undefined;
_ = try self.readFramed(&query_buf);
try self.writeFramed(response_bytes);
}
fn readFramed(self: *Accepted, buf: []u8) ![]u8 {
var prefix: [transport.prefix_len]u8 = undefined;
try self.tls.reader().readSliceAll(&prefix);
const len = transport.parsePrefix(prefix);
if (len > buf.len) return error.QueryTooLarge;
try self.tls.reader().readSliceAll(buf[0..len]);
_ = self.server.queries.fetchAdd(1, .acq_rel);
return buf[0..len];
}
fn writeFramed(self: *Accepted, message: []const u8) !void {
const prefix = transport.framePrefix(@intCast(message.len));
try self.tls.writer().writeAll(&prefix);
try self.tls.writer().writeAll(message);
try self.tls.writer().flush();
}
};
/// Answers `count` queries on one connection and holds it open until the test
/// tears it down.
fn serveOnOneConnection(
gpa: std.mem.Allocator,
server: *Server,
io: std.Io,
count: usize,
) anyerror!void {
var session: Accepted = undefined;
try session.accept(gpa, server, io);
defer session.close(gpa, io);
for (0..count) |_| try session.answerOnce();
}
/// Answers one query, drops the connection the way an upstream reaps an idle
/// one, then accepts a second and answers one more.
fn serveThenCloseThenServe(gpa: std.mem.Allocator, server: *Server, io: std.Io) anyerror!void {
for (0..2) |_| {
var session: Accepted = undefined;
try session.accept(gpa, server, io);
defer session.close(gpa, io);
try session.answerOnce();
}
}
/// Answers one query, then replies to the next with a single length-prefix byte
/// and closes. The second reply has started, so the client may not redial.
fn serveThenTruncate(gpa: std.mem.Allocator, server: *Server, io: std.Io) anyerror!void {
var session: Accepted = undefined;
try session.accept(gpa, server, io);
defer session.close(gpa, io);
try session.answerOnce();
var query_buf: [2048]u8 = undefined;
_ = try session.readFramed(&query_buf);
try session.tls.writer().writeAll(&[_]u8{0x00});
try session.tls.writer().flush();
}
/// Answers one query, then answers the next with a frame the client can read
/// whole and `transport.validateResponse` must still reject.
fn serveThenAnswerWithWrongId(gpa: std.mem.Allocator, server: *Server, io: std.Io) anyerror!void {
var session: Accepted = undefined;
try session.accept(gpa, server, io);
defer session.close(gpa, io);
try session.answerOnce();
var query_buf: [2048]u8 = undefined;
_ = try session.readFramed(&query_buf);
try session.writeFramed(mismatched_response_bytes);
}
/// Answers one query, drops the connection, then takes the redial and drops that
/// one too — after reading its query and before answering it.
///
/// The redial *succeeds*, which is the point: the exchange that fails is the
/// retry, running on a session this call dialed itself. `retryDecision` gives a
/// fresh session no second chance, so the client owes exactly two dials.
fn serveThenCloseThenFailTheRetry(gpa: std.mem.Allocator, server: *Server, io: std.Io) anyerror!void {
{
var session: Accepted = undefined;
try session.accept(gpa, server, io);
defer session.close(gpa, io);
try session.answerOnce();
}
{
var session: Accepted = undefined;
try session.accept(gpa, server, io);
defer session.close(gpa, io);
// Read the query before dropping the connection: the failure under test
// is the client's *receive*, and a peer that closed before taking the
// query could fail its send instead.
var query_buf: [2048]u8 = undefined;
_ = try session.readFramed(&query_buf);
}
}
/// Completes the handshake, takes the query and never answers it.
fn serveSilently(gpa: std.mem.Allocator, server: *Server, io: std.Io) anyerror!void {
var session: Accepted = undefined;
try session.accept(gpa, server, io);
defer session.close(gpa, io);
var query_buf: [2048]u8 = undefined;
_ = try session.readFramed(&query_buf);
const forever: std.Io.Clock.Duration = .{ .raw = .fromSeconds(3600), .clock = .awake };
try forever.sleep(io);
}
// ---------------------------------------------------------------------------
// Client side
// ---------------------------------------------------------------------------
/// Everything one `DotClient` borrows, plus the client itself.
///
/// Built in place: `DotClient` pins the TLS state of an open session inside
/// itself, and its endpoint and buffers are borrowed from this struct.
const ClientFixture = struct {
gpa: std.mem.Allocator,
bundle: Certificate.Bundle = .empty,
bundle_lock: std.Io.RwLock = .init,
buffers: [4 * tls.Client.min_buffer_len]u8 = undefined,
url_buf: [32]u8 = undefined,
recoveries: std.atomic.Value(u64) = .init(0),
dot: dot_client.DotClient = undefined,
fn init(self: *ClientFixture, gpa: std.mem.Allocator, io: std.Io, address: net.IpAddress) !void {
self.* = .{ .gpa = gpa };
errdefer self.bundle.deinit(gpa);
try self.preloadFixtureCert(io);
const url = try std.fmt.bufPrint(&self.url_buf, "tls://127.0.0.1:{d}", .{address.ip4.port});
const chunk = tls.Client.min_buffer_len;
self.dot = .init(
try .parse(url),
fixture_host,
gpa,
&self.bundle,
&self.bundle_lock,
&self.recoveries,
.{
.tls_read = self.buffers[0..chunk],
.tls_write = self.buffers[chunk .. 2 * chunk],
.stream_read = self.buffers[2 * chunk .. 3 * chunk],
.stream_write = self.buffers[3 * chunk ..],
},
);
}
fn deinit(self: *ClientFixture, io: std.Io) void {
self.dot.close(io);
self.bundle.deinit(self.gpa);
}
/// Makes the self-signed fixture certificate the one trust anchor this
/// client has.
///
/// A non-empty bundle is what both `DotClient.ensureBundle` and
/// `TlsStream.init` check before they would scan the system store, so
/// preloading it also keeps the test off the host's CA directory entirely.
/// The decode mirrors `Certificate.Bundle.addCertsFromFile`, which is the
/// only PEM entry point the stdlib exposes and takes a file; writing the
/// committed fixture back out to disk to read it in again would be the
/// longer way round to the same three calls.
fn preloadFixtureCert(self: *ClientFixture, io: std.Io) !void {
const fixtures = @import("test_fixtures");
const begin_marker = "-----BEGIN CERTIFICATE-----";
const end_marker = "-----END CERTIFICATE-----";
const body_start = (std.mem.find(u8, fixtures.cert_pem, begin_marker) orelse
return error.MissingBeginCertificateMarker) + begin_marker.len;
const body_end = std.mem.findPos(u8, fixtures.cert_pem, body_start, end_marker) orelse
return error.MissingEndCertificateMarker;
const encoded = std.mem.trim(u8, fixtures.cert_pem[body_start..body_end], " \t\r\n");
const decoder = std.base64.standard.decoderWithIgnore(" \t\r\n");
try self.bundle.bytes.ensureUnusedCapacity(self.gpa, encoded.len / 4 * 3 + 3);
const decoded_start: u32 = @intCast(self.bundle.bytes.items.len);
const written = try decoder.decode(
self.bundle.bytes.allocatedSlice()[decoded_start..],
encoded,
);
self.bundle.bytes.items.len += written;
try self.bundle.parseCert(self.gpa, decoded_start, std.Io.Clock.real.now(io).toSeconds());
try testing.expect(self.bundle.map.count() == 1);
}
};
/// Blocks until the server has taken `count` framed queries off the wire.
///
/// Bounded by the same budget as everything else here: a server that died
/// before it read one would otherwise hang the suite.
fn awaitQueries(io: std.Io, server: *Server, count: u32) !void {
const step: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(10), .clock = .awake };
const steps = 1000;
for (0..steps) |_| {
if (server.queries.load(.acquire) >= count) return;
try step.sleep(io);
}
return error.DotFixtureTimedOut;
}
fn expire(io: std.Io, duration: std.Io.Clock.Duration) std.Io.Cancelable!void {
return duration.sleep(io);
}
const Outcome = union(enum) {
client: anyerror!void,
expiry: std.Io.Cancelable!void,
};
/// Runs `client_work` against the budget and tears the server task down either
/// way: a client that fails before it connects would otherwise leave the server
/// blocked in `accept` forever.
fn runAgainstServer(
io: std.Io,
server_task: *std.Io.Future(anyerror!void),
comptime client_work: anytype,
args: anytype,
) !void {
var outcomes: [2]Outcome = undefined;
var race: std.Io.Select(Outcome) = .init(io, &outcomes);
defer race.cancelDiscard();
try race.concurrent(.client, client_work, args);
try race.concurrent(.expiry, expire, .{ io, budget });
const client_result: anyerror!void = switch (try race.await()) {
.client => |result| result,
.expiry => |result| blk: {
try result;
break :blk error.DotFixtureTimedOut;
},
};
const server_result = if (client_result) |_|
server_task.await(io)
else |_|
server_task.cancel(io);
try client_result;
server_result catch |err| switch (err) {
// The scripts that hold a connection open past their last answer are
// torn down by the cancel above, which is the intended end for them.
error.Canceled => {},
else => return err,
};
}
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
fn twoExchangesOverOneSession(io: std.Io, fixture: *ClientFixture) anyerror!void {
var buf: [512]u8 = undefined;
const first = try fixture.dot.exchange(io, query_bytes, &buf);
try testing.expectEqualSlices(u8, response_bytes, first);
// The point of the milestone: the connection outlives the exchange.
try testing.expect(fixture.dot.session != null);
const second = try fixture.dot.exchange(io, query_bytes, &buf);
try testing.expectEqualSlices(u8, response_bytes, second);
try testing.expect(fixture.dot.session != null);
}
test "two exchanges through one DoT client share one connection" {
if (!build_options.integration) return error.SkipZigTest;
const gpa = testing.allocator;
var threaded: std.Io.Threaded = .init(gpa, .{});
defer threaded.deinit();
const io = threaded.io();
var server: Server = undefined;
try server.init(gpa, io);
defer server.deinit(gpa, io);
var fixture: ClientFixture = undefined;
try fixture.init(gpa, io, server.address());
defer fixture.deinit(io);
var server_task = try io.concurrent(serveOnOneConnection, .{ gpa, &server, io, @as(usize, 2) });
try runAgainstServer(io, &server_task, twoExchangesOverOneSession, .{ io, &fixture });
try testing.expectEqual(@as(u32, 1), server.accepts.load(.acquire));
try testing.expectEqual(@as(u64, 0), fixture.recoveries.load(.acquire));
}
fn twoPoolExchanges(io: std.Io, pool: *pool_mod.Pool) anyerror!void {
var buf: [512]u8 = undefined;
var selected: ?[]const u8 = null;
const first = try pool.exchange(io, query_bytes, &buf, &selected);
try testing.expectEqualSlices(u8, response_bytes, first);
const second = try pool.exchange(io, query_bytes, &buf, &selected);
try testing.expectEqualSlices(u8, response_bytes, second);
}
test "a session the upstream closed is recovered by one redial and counted, not blamed" {
if (!build_options.integration) return error.SkipZigTest;
const gpa = testing.allocator;
var threaded: std.Io.Threaded = .init(gpa, .{});
defer threaded.deinit();
const io = threaded.io();
var server: Server = undefined;
try server.init(gpa, io);
defer server.deinit(gpa, io);
var fixture: ClientFixture = undefined;
try fixture.init(gpa, io, server.address());
defer fixture.deinit(io);
// Through a real pool entry, because the claim is about what the pool does
// *not* see: an upstream reaping an idle connection must not cost it health
// or raise an operational event.
var slots = [_]pool_mod.Slot{.{ .client = fixture.dot.client() }};
var entries = [_]pool_mod.Entry{.{
.endpoint = fixture.dot.endpoint,
.slots = &slots,
.priority = 10,
.enabled = true,
.health = .init,
.sem = .{ .permits = slots.len },
.reuse_recoveries = &fixture.recoveries,
}};
var pool: pool_mod.Pool = .init(&entries, .{
.failure_threshold = 2,
.base_backoff_ms = 60_000,
.max_backoff_ms = 60_000,
}, .{
.attempt = .{ .raw = .fromSeconds(10), .clock = .awake },
.total = .{ .raw = .fromSeconds(30), .clock = .awake },
}, 1);
var fx: events_fixture.Fixture = .{};
try fx.init(io, 1000);
defer fx.deinit();
pool.diagnostics = &fx.store;
var server_task = try io.concurrent(serveThenCloseThenServe, .{ gpa, &server, io });
try runAgainstServer(io, &server_task, twoPoolExchanges, .{ io, &pool });
try testing.expectEqual(@as(u32, 2), server.accepts.load(.acquire));
try testing.expectEqual(@as(u64, 1), fixture.recoveries.load(.acquire));
try testing.expectEqual(@as(u64, 2), entries[0].health.total_successes);
try testing.expectEqual(@as(u64, 0), entries[0].health.total_failures);
try testing.expectEqual(@as(u32, 0), entries[0].health.consecutive_failures);
try testing.expectEqual(@as(?std.Io.Timestamp, null), entries[0].health.backoff_until);
try testing.expectEqual(
@as(i64, 0),
try fx.count("SELECT count(*) FROM operational_events"),
);
}
fn exchangeThenReadTruncatedReply(io: std.Io, fixture: *ClientFixture) anyerror!void {
var buf: [512]u8 = undefined;
_ = try fixture.dot.exchange(io, query_bytes, &buf);
// One prefix byte arrived, so the reply had started: re-sending the query on
// a fresh connection would be a second question, not a recovery.
try testing.expectError(error.ReceiveFailed, fixture.dot.exchange(io, query_bytes, &buf));
try testing.expect(fixture.dot.session == null);
}
test "a reused session that failed after one response byte is not redialed" {
if (!build_options.integration) return error.SkipZigTest;
const gpa = testing.allocator;
var threaded: std.Io.Threaded = .init(gpa, .{});
defer threaded.deinit();
const io = threaded.io();
var server: Server = undefined;
try server.init(gpa, io);
defer server.deinit(gpa, io);
var fixture: ClientFixture = undefined;
try fixture.init(gpa, io, server.address());
defer fixture.deinit(io);
var server_task = try io.concurrent(serveThenTruncate, .{ gpa, &server, io });
try runAgainstServer(io, &server_task, exchangeThenReadTruncatedReply, .{ io, &fixture });
// The whole claim: the client never came back for a second connection.
try testing.expectEqual(@as(u32, 1), server.accepts.load(.acquire));
try testing.expectEqual(@as(u64, 0), fixture.recoveries.load(.acquire));
}
fn exchangeThenReadWrongId(io: std.Io, fixture: *ClientFixture) anyerror!void {
var buf: [512]u8 = undefined;
_ = try fixture.dot.exchange(io, query_bytes, &buf);
try testing.expect(fixture.dot.session != null);
// The read succeeded; only the bytes are wrong. Nothing about that says the
// connection is stale, so it is final — but the stream position after a
// frame this client will not trust is unknowable, so the session goes.
try testing.expectError(error.ResponseMismatch, fixture.dot.exchange(io, query_bytes, &buf));
try testing.expect(fixture.dot.session == null);
}
test "a reply that fails validation is final and clears the session without redialing" {
if (!build_options.integration) return error.SkipZigTest;
const gpa = testing.allocator;
var threaded: std.Io.Threaded = .init(gpa, .{});
defer threaded.deinit();
const io = threaded.io();
var server: Server = undefined;
try server.init(gpa, io);
defer server.deinit(gpa, io);
var fixture: ClientFixture = undefined;
try fixture.init(gpa, io, server.address());
defer fixture.deinit(io);
var server_task = try io.concurrent(serveThenAnswerWithWrongId, .{ gpa, &server, io });
try runAgainstServer(io, &server_task, exchangeThenReadWrongId, .{ io, &fixture });
// Both halves of the claim. One accept: the reused session read a whole
// frame, so nothing here is a lifecycle failure and no redial is owed. And
// nothing was recovered, so the counter stays where it was.
try testing.expectEqual(@as(u32, 1), server.accepts.load(.acquire));
try testing.expectEqual(@as(u64, 0), fixture.recoveries.load(.acquire));
}
fn exchangeThenFailTheRetry(io: std.Io, fixture: *ClientFixture) anyerror!void {
var buf: [512]u8 = undefined;
_ = try fixture.dot.exchange(io, query_bytes, &buf);
try testing.expect(fixture.dot.session != null);
// Stale session, no response byte, lifecycle cause: the one redial is owed,
// taken, and connected. The exchange on that fresh session then fails its
// read, and the retry's outcome is the exchange's outcome — no third dial,
// because a session this call dialed itself is never retried.
try testing.expectError(error.ReceiveFailed, fixture.dot.exchange(io, query_bytes, &buf));
try testing.expect(fixture.dot.session == null);
// What the `nxdns check` probe loop relies on: closing a client whose
// exchange already failed is a no-op, not a double close.
fixture.dot.close(io);
try testing.expect(fixture.dot.session == null);
}
test "a stale session whose retry fails is final and leaves no session behind" {
if (!build_options.integration) return error.SkipZigTest;
const gpa = testing.allocator;
var threaded: std.Io.Threaded = .init(gpa, .{});
defer threaded.deinit();
const io = threaded.io();
var server: Server = undefined;
try server.init(gpa, io);
defer server.deinit(gpa, io);
var fixture: ClientFixture = undefined;
try fixture.init(gpa, io, server.address());
defer fixture.deinit(io);
var server_task = try io.concurrent(serveThenCloseThenFailTheRetry, .{ gpa, &server, io });
try runAgainstServer(io, &server_task, exchangeThenFailTheRetry, .{ io, &fixture });
// Exactly two: the first exchange and the one redial. A client that retried
// its retry would need a third dial; the script accepts no third connection,
// so that dial would stall un-handshaken until the client's budget failed
// the test — it cannot succeed silently.
try testing.expectEqual(@as(u32, 2), server.accepts.load(.acquire));
// The redial connected but the exchange on it failed, so nothing was
// recovered and nothing is counted.
try testing.expectEqual(@as(u64, 0), fixture.recoveries.load(.acquire));
}
fn oneExchange(io: std.Io, fixture: *ClientFixture) transport.ExchangeError!void {
var buf: [512]u8 = undefined;
_ = try fixture.dot.exchange(io, query_bytes, &buf);
}
test "a canceled exchange leaves no session open" {
if (!build_options.integration) return error.SkipZigTest;
const gpa = testing.allocator;
var threaded: std.Io.Threaded = .init(gpa, .{});
defer threaded.deinit();
const io = threaded.io();
var server: Server = undefined;
try server.init(gpa, io);
defer server.deinit(gpa, io);
var fixture: ClientFixture = undefined;
try fixture.init(gpa, io, server.address());
defer fixture.deinit(io);
var server_task = try io.concurrent(serveSilently, .{ gpa, &server, io });
var client_task = try io.concurrent(oneExchange, .{ io, &fixture });
// The query is on the wire and no answer is coming, so the client is
// provably blocked in its read — the state a total-budget expiry cancels an
// exchange in.
try awaitQueries(io, &server, 1);
try testing.expectError(error.Canceled, client_task.cancel(io));
// Mid-frame state is unknowable after a cancellation, so the next exchange
// has to start from a fresh dial.
try testing.expect(fixture.dot.session == null);
try testing.expectEqual(@as(u64, 0), fixture.recoveries.load(.acquire));
server_task.cancel(io) catch |err| switch (err) {
error.Canceled => {},
else => return err,
};
}
src/upstream/dot_client_live_test.zig
+5
View File
@@ -55,8 +55,13 @@ fn runExchange(io: std.Io, params: Params) anyerror!usize {
params.gpa,
params.bundle,
params.bundle_lock,
null,
params.buffers,
);
// The client owns the session it dials and keeps it open past the exchange,
// so the task that built the client is what has to close it. Without this
// the socket and its TLS state outlive the test.
defer client.close(io);
var selected: ?[]const u8 = null;
const reply = try client.client().exchange(io, query_bytes, params.response_buf, &selected);
std.debug.assert(std.mem.eql(u8, selected.?, endpoint.url));
src/upstream/pool.zig
+395 -77
View File
@@ -10,7 +10,9 @@
//! waits for: without it, N unreachable upstreams cost N × attempt, and the
//! resolver above has already given up. The outer expiry is `error.Timeout`
//! unconditionally — the fast-failure paths (no entry enabled) return long
//! before the budget, so an expiry always means an attempt was in flight.
//! before the budget, so an expiry means either that an attempt was in flight
//! or that the task died waiting for a slot on a saturated entry.
//! `Entry.queued_total` is what tells those two apart.
//!
//! Two passes, not one. Pass one walks the enabled entries that health says are
//! available. Pass two runs only when pass one attempted nothing, and skips the
@@ -25,25 +27,24 @@
//! never held across an exchange, so a slow upstream cannot block a health read
//! or an attempt on some other entry.
//!
//! `Entry.busy` guards one entry's `client`. A task holds it for a whole
//! attempt against that entry — the exchange and the timeout race around it —
//! and drops it before the failover loop moves to the next candidate. This is
//! what makes the pool safe to share: `DohClient` owns a request buffer and a
//! An entry owns one leaf client per slot, each behind its own `Slot.busy`
//! mutex, plus an `Entry.sem` carrying one permit per slot. A task takes a
//! permit, then the first free slot's mutex, and holds that mutex for a whole
//! attempt against the entry — the exchange and the timeout race around it —
//! before the failover loop moves to the next candidate. The mutex is what
//! makes the pool safe to share: `DohClient` owns a request buffer and a
//! transfer buffer, `DotClient` owns four TLS buffers and the stream state
//! built on them, so one client may be inside `exchange` only once at a time.
//!
//! Because that wait is serializing, pass one re-reads health after it takes
//! `busy`: the attempts a task queued behind can fail the entry into backoff
//! while it waits, and pass one must not attempt an entry that is unavailable by
//! the time it gets its turn. Lock order is always `busy` then `Pool.mutex`,
//! never the reverse.
//! Because a saturated entry still makes tasks wait, pass one re-reads health
//! after it acquires a slot: the attempts a task queued behind can fail the
//! entry into backoff while it waits, and pass one must not attempt an entry
//! that is unavailable by the time it gets its turn. Lock order is always
//! `Slot.busy` then `Pool.mutex`, never the reverse.
//!
//! The guarantee is therefore: at most one in-flight exchange per entry, any
//! number of entries in flight at once, and bookkeeping that never waits on a
//! peer. Two concurrent queries that resolve to the same sole upstream do
//! serialize. At household scale that is the right trade — the alternative is a
//! client instance per listener task, which multiplies TLS buffers and
//! connections for a query rate that never needed them.
//! The guarantee is therefore: at most `slots.len` in-flight exchanges per
//! entry, any number of entries in flight at once, and bookkeeping that never
//! waits on a peer.
const std = @import("std");
@@ -75,18 +76,77 @@ const AttemptFailure = struct {
}
};
/// How many leaf clients the composition root builds per enabled upstream, and
/// therefore how many exchanges one upstream may have in flight at once.
///
/// Compiled, not configured (anti-requirement A2). The listeners admit up to 64
/// UDP queries in flight and 64 TCP connections per family, so a burst reaches
/// the pool tens wide; 8 concurrent exchanges per upstream clear the 30-query
/// burst the Pi reproduced in at most four waves even before DoT session reuse
/// removes the per-exchange handshake. The cost is bounded and paid once at
/// startup: a DoT entry holds 8 × 4 × `tls.Client.min_buffer_len` of TLS
/// buffers, about 533 KiB, and a DoH entry two small buffers per slot over one
/// shared `std.http.Client`.
pub const slots_per_entry = 8;
/// One leaf client of an entry, and the lock that keeps one task inside it.
pub const Slot = struct {
client: transport.Client,
/// Held for the whole of one attempt, so `client` is never re-entered while
/// it is using its own buffers. Defaulted because `Pool.init` sorts
/// `entries` by value, which may only copy unlocked mutexes, and it runs
/// before the pool is reachable by any task.
busy: std.Io.Mutex = .init,
};
pub const Entry = struct {
endpoint: transport.Endpoint,
client: transport.Client,
/// Caller-owned, `len >= 1`, never resized: `DotClient` pins live TLS state
/// and `Slot.client` is an erased pointer at it.
slots: []Slot,
/// Lower is tried first (PLAN §11.2 `upstreams.priority`).
priority: i32,
enabled: bool,
health: health.State,
/// Held for the whole of one attempt against this entry, so `client` is
/// never re-entered while it is using its own buffers. Defaulted because
/// `init` sorts `entries` by value, which may only copy unlocked mutexes,
/// and it runs before the pool is reachable by any task.
busy: std.Io.Mutex = .init,
/// One permit per slot, so a task waits here rather than spinning over
/// `Slot.busy`. `.{ .permits = slots.len }` at build time; `Pool.init`
/// asserts it. Its mutex and condition are plain values, so the sort in
/// `init` may copy it.
sem: std.Io.Semaphore,
in_flight: std.atomic.Value(u32) = .init(0),
/// Test-only: the most exchanges this entry ever had in flight at once.
peak_in_flight: std.atomic.Value(u32) = .init(0),
/// Admission samples, not a queue length: a task counts itself queued when
/// it finds `in_flight` saturated before it waits for a permit, and between
/// a releaser's `in_flight` decrement and its `sem.post` (and the mirror
/// window on acquire) that read can misjudge saturation by one. Exactness
/// would cost a lock on the hot path to size a household queue.
queued_total: std.atomic.Value(u64) = .init(0),
queued_ns_total: std.atomic.Value(u64) = .init(0),
/// Points into the composition root's stable counter storage, never into an
/// `Entry`: `Pool.init` sorts entries by value, so a pointer taken into one
/// before the sort would dangle onto a different upstream's field.
/// Incremented by a DoT slot client when a stale reused session was
/// recovered by its single redial. Zero forever for DoH entries.
reuse_recoveries: *std.atomic.Value(u64),
/// Both counters, on either exit of the wait for a permit. A waiter that
/// burned its budget in the queue is the field failure this records, so a
/// cancellation counts exactly like an acquisition.
fn recordQueued(self: *Entry, io: std.Io, started: std.Io.Timestamp) void {
const elapsed = std.Io.Clock.awake.now(io).nanoseconds - started.nanoseconds;
_ = self.queued_total.fetchAdd(1, .monotonic);
_ = self.queued_ns_total.fetchAdd(@intCast(@max(elapsed, 0)), .monotonic);
}
/// The first slot whose client is free. Called only while holding a permit,
/// where a free slot is guaranteed.
fn takeSlot(self: *Entry) ?*Slot {
for (self.slots) |*slot| {
if (slot.busy.tryLock()) return slot;
}
return null;
}
};
/// A copy of one entry's health, taken under the mutex. Feeds `/metrics` and
@@ -108,6 +168,16 @@ pub const Snapshot = struct {
/// Borrowed from the entry; valid until that entry's next failure.
last_error: []const u8,
backoff_until: ?std.Io.Timestamp,
/// Exchanges in flight against this entry at the instant of the read, and
/// the ceiling they cannot cross (`slots.len`). `peak_in_flight` stays out:
/// it is a test assertion, not an operator's number.
in_flight: u32,
slots: u32,
/// Admission samples; see `Entry.queued_total`. Seconds rather than
/// nanoseconds because Prometheus counts time in seconds.
queued_total: u64,
queued_seconds_total: f64,
reuse_recoveries_total: u64,
};
/// The two budgets, named rather than positional: they are the same type, so
@@ -143,6 +213,10 @@ pub const Pool = struct {
seed: u64,
) Pool {
std.debug.assert(entries.len > 0);
for (entries) |*entry| {
std.debug.assert(entry.slots.len >= 1);
std.debug.assert(entry.sem.permits == entry.slots.len);
}
// Stable, so entries sharing a priority keep their configured order.
std.mem.sort(Entry, entries, {}, byPriority);
return .{
@@ -178,10 +252,11 @@ pub const Pool = struct {
/// success; on error the buffer's contents are undefined.
///
/// The failover loop runs raced against `timeouts.total`. Losing that race
/// cancels the loop, which unwinds whatever attempt was in flight: the
/// `Entry.busy` lock is taken cancelably on purpose and released by defer,
/// cancels the loop, which unwinds whatever attempt was in flight: the wait
/// for a slot is cancelable on purpose and every slot is released by defer,
/// and the health bookkeeping around a completed exchange is uncancelable,
/// so a canceled attempt leaves no lock held and no counter half-written.
/// so a canceled attempt leaves no lock held, no permit lost and no counter
/// half-written.
pub fn exchange(
self: *Pool,
io: std.Io,
@@ -216,21 +291,48 @@ pub const Pool = struct {
if (!entry.enabled) continue;
if (pass == 0 and !self.entryAvailable(io, entry, now)) continue;
// Sampled before the wait, so a task that is about to queue is
// counted as queued whichever way the wait ends. The read races
// the releasers by one slot on purpose; see `Entry`.
const queued_at: ?std.Io.Timestamp = if (entry.in_flight.load(.acquire) >= entry.slots.len)
std.Io.Clock.awake.now(io)
else
null;
// Cancelable, unlike the health-bookkeeping locks below: a task
// waiting its turn on a busy upstream has done nothing that a
// cancellation could corrupt, so it gives up here rather than
// queueing behind an exchange it will not use. The wait itself
// is bounded by the holder's `timeouts.attempt`; the waiter's
// own budget only starts once it has the lock, and the whole
// loop is bounded by `timeouts.total` regardless.
try entry.busy.lock(io);
defer entry.busy.unlock(io);
// waiting its turn on a saturated upstream has done nothing that
// a cancellation could corrupt, so it gives up here rather than
// queueing behind exchanges it will not use. `wait` either
// returns holding a permit or returns `error.Canceled` holding
// none. The wait itself is bounded by the holders'
// `timeouts.attempt`; the waiter's own budget only starts once
// it has a slot, and the whole loop is bounded by
// `timeouts.total` regardless.
entry.sem.wait(io) catch |err| {
if (queued_at) |started| entry.recordQueued(io, started);
return err;
};
if (queued_at) |started| entry.recordQueued(io, started);
// A permit means a slot's mutex is free, so failing to find one
// is a protocol bug in this loop, not a runtime condition.
const slot = entry.takeSlot() orelse unreachable;
const entrants = entry.in_flight.fetchAdd(1, .acq_rel) + 1;
_ = entry.peak_in_flight.fetchMax(entrants, .acq_rel);
defer {
slot.busy.unlock(io);
_ = entry.in_flight.fetchSub(1, .acq_rel);
// Uncancelable, so a canceled attempt still returns its
// permit.
entry.sem.post(io);
}
// The check above ran before the wait, and the attempts this one
// queued behind may have failed the entry into backoff while it
// waited. Pass one must not touch an entry that is unavailable
// now, so re-read health against a fresh `now` and move on if it
// is. Pass two skips this on purpose: it probes regardless of
// is — the defer above releases the slot before the `continue`.
// Pass two skips this on purpose: it probes regardless of
// backoff, which is how backoff recovers.
if (pass == 0) {
const recheck = std.Io.Clock.awake.now(io);
@@ -245,7 +347,7 @@ pub const Pool = struct {
// the pool, so the borrow stays valid past this loop.
selected.* = entry.endpoint.url;
const result = self.attempt(io, entry.client, query, response_buf);
const result = self.attempt(io, slot.client, query, response_buf);
const completed_at = std.Io.Clock.awake.now(io);
const response = result catch |err| switch (transport.group(err)) {
@@ -293,6 +395,11 @@ pub const Pool = struct {
.last_error_at = entry.health.last_error_at,
.last_error = entry.health.lastError(),
.backoff_until = entry.health.backoff_until,
.in_flight = entry.in_flight.load(.acquire),
.slots = @intCast(entry.slots.len),
.queued_total = entry.queued_total.load(.acquire),
.queued_seconds_total = @as(f64, @floatFromInt(entry.queued_ns_total.load(.acquire))) / std.time.ns_per_s,
.reuse_recoveries_total = entry.reuse_recoveries.load(.acquire),
};
}
return count;
@@ -419,13 +526,20 @@ const Fake = struct {
/// Replaces `behavior` after the first call. One entry that fails the task
/// which reaches it first and answers the next is what makes two concurrent
/// exchanges end on different entries; a single behaviour cannot say that.
/// Mutated under the entry's `busy` lock, like `calls`.
then: ?Behavior = null,
calls: usize = 0,
/// Guards `behavior` and `then`. One fake now backs every slot of an entry,
/// so the swap runs under as many tasks as the entry has slots.
mutex: std.Io.Mutex = .init,
calls: std.atomic.Value(usize) = .init(0),
in_flight: std.atomic.Value(u32) = .init(0),
/// The most tasks ever inside `exchangeFn` at once. The per-entry lock is
/// only doing its job while this stays at 1.
/// The most tasks ever inside `exchangeFn` at once. The entry's slot count
/// is the ceiling this must never cross.
peak_in_flight: std.atomic.Value(u32) = .init(0),
/// Slot storage for the entry this fake backs, and the recovery counter
/// that entry points at. Both live here so one declared `Fake` is one whole
/// test upstream; `testEntry` wires as many of the slots as a test wants.
slots: [slots_per_entry]Slot = undefined,
recoveries: std.atomic.Value(u64) = .init(0),
const Behavior = union(enum) {
/// Copy these bytes into the caller's buffer and return them.
@@ -455,12 +569,17 @@ const Fake = struct {
defer _ = self.in_flight.fetchSub(1, .acq_rel);
_ = self.peak_in_flight.fetchMax(entrants, .acq_rel);
self.calls += 1;
const behavior = self.behavior;
if (self.then) |next| {
self.behavior = next;
self.then = null;
}
_ = self.calls.fetchAdd(1, .acq_rel);
const behavior = behavior: {
self.mutex.lockUncancelable(io);
defer self.mutex.unlock(io);
const current = self.behavior;
if (self.then) |next| {
self.behavior = next;
self.then = null;
}
break :behavior current;
};
switch (behavior) {
.reply => |bytes| return copy(bytes, response_buf),
.fail => |err| return err,
@@ -487,12 +606,25 @@ const Fake = struct {
};
fn testEntry(url: []const u8, fake: *Fake, priority: i32) Entry {
return testEntrySlots(url, fake, priority, 1);
}
/// An entry whose `slot_count` slots all run through one fake, borrowing that
/// fake's slot storage and recovery counter. One slot unless a test is about
/// concurrency: the assertions below read the fake's own counters, and one fake
/// per entry is what makes those readable.
fn testEntrySlots(url: []const u8, fake: *Fake, priority: i32, slot_count: usize) Entry {
std.debug.assert(slot_count >= 1 and slot_count <= fake.slots.len);
const slots = fake.slots[0..slot_count];
for (slots) |*slot| slot.* = .{ .client = fake.client() };
return .{
.endpoint = Endpoint.parse(url) catch unreachable,
.client = fake.client(),
.slots = slots,
.priority = priority,
.enabled = true,
.health = .init,
.sem = .{ .permits = slot_count },
.reuse_recoveries = &fake.recoveries,
};
}
@@ -636,7 +768,7 @@ test "a timeout mid-flight reports the endpoint the query was in" {
var selected: ?[]const u8 = null;
try testing.expectError(error.Timeout, pool.exchange(io, query_bytes, &buf, &selected));
try testing.expectEqualStrings("https://stalling.example/dns-query", selected.?);
try testing.expectEqual(@as(usize, 0), untouched.calls);
try testing.expectEqual(@as(usize, 0), untouched.calls.load(.acquire));
}
test "an exchange that attempted nothing reports no endpoint" {
@@ -674,8 +806,8 @@ test "entries are tried in ascending priority order" {
var buf: [512]u8 = undefined;
var selected: ?[]const u8 = null;
_ = try pool.exchange(io, query_bytes, &buf, &selected);
try testing.expectEqual(@as(usize, 1), low.calls);
try testing.expectEqual(@as(usize, 0), high.calls);
try testing.expectEqual(@as(usize, 1), low.calls.load(.acquire));
try testing.expectEqual(@as(usize, 0), high.calls.load(.acquire));
}
test "a peer fault fails over to the next entry and is recorded" {
@@ -720,12 +852,12 @@ test "an entry in backoff is skipped while another is available" {
var selected: ?[]const u8 = null;
_ = try pool.exchange(io, query_bytes, &buf, &selected);
_ = try pool.exchange(io, query_bytes, &buf, &selected);
try testing.expectEqual(@as(usize, 2), bad.calls);
try testing.expectEqual(@as(usize, 2), bad.calls.load(.acquire));
try testing.expect(entries[0].health.backoff_until != null);
_ = try pool.exchange(io, query_bytes, &buf, &selected);
try testing.expectEqual(@as(usize, 2), bad.calls);
try testing.expectEqual(@as(usize, 3), good.calls);
try testing.expectEqual(@as(usize, 2), bad.calls.load(.acquire));
try testing.expectEqual(@as(usize, 3), good.calls.load(.acquire));
}
test "every entry in backoff is still probed" {
@@ -750,8 +882,8 @@ test "every entry in backoff is still probed" {
// Pass one now has no candidate at all. Pass two probes both anyway.
try testing.expectError(error.BadResponse, pool.exchange(io, query_bytes, &buf, &selected));
try testing.expectEqual(@as(usize, 3), first.calls);
try testing.expectEqual(@as(usize, 3), second.calls);
try testing.expectEqual(@as(usize, 3), first.calls.load(.acquire));
try testing.expectEqual(@as(usize, 3), second.calls.load(.acquire));
}
test "a local resource error short-circuits and records nothing" {
@@ -770,7 +902,7 @@ test "a local resource error short-circuits and records nothing" {
var buf: [512]u8 = undefined;
var selected: ?[]const u8 = null;
try testing.expectError(error.OutOfMemory, pool.exchange(io, query_bytes, &buf, &selected));
try testing.expectEqual(@as(usize, 0), good.calls);
try testing.expectEqual(@as(usize, 0), good.calls.load(.acquire));
try testing.expectEqual(@as(u64, 0), entries[0].health.total_failures);
try testing.expectEqual(@as(u32, 0), entries[0].health.consecutive_failures);
}
@@ -791,7 +923,7 @@ test "a cancellation short-circuits and records nothing" {
var buf: [512]u8 = undefined;
var selected: ?[]const u8 = null;
try testing.expectError(error.Canceled, pool.exchange(io, query_bytes, &buf, &selected));
try testing.expectEqual(@as(usize, 0), good.calls);
try testing.expectEqual(@as(usize, 0), good.calls.load(.acquire));
try testing.expectEqual(@as(u64, 0), entries[0].health.total_failures);
}
@@ -821,7 +953,7 @@ test "an attempt that outruns the budget is a recorded Timeout" {
const reply = try pool.exchange(io, query_bytes, &buf, &selected);
try testing.expectEqualSlices(u8, response_bytes, reply);
try testing.expectEqual(@as(usize, 1), slow.calls);
try testing.expectEqual(@as(usize, 1), slow.calls.load(.acquire));
try testing.expectEqual(@as(u32, 1), entries[0].health.consecutive_failures);
try testing.expectEqualStrings("Timeout", entries[0].health.lastError());
try testing.expectEqual(@as(u64, 1), entries[1].health.total_successes);
@@ -865,7 +997,7 @@ test "two stalling upstreams cost the total budget, not one budget each" {
try testing.expect(elapsed_ns < @as(i96, 200) * std.time.ns_per_ms);
// The second entry was never reached: the loop was canceled inside the
// first attempt.
try testing.expectEqual(@as(usize, 0), second.calls);
try testing.expectEqual(@as(usize, 0), second.calls.load(.acquire));
}
test "every entry disabled yields ConnectFailed without waiting out the total budget" {
@@ -895,8 +1027,8 @@ test "every entry disabled yields ConnectFailed without waiting out the total bu
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);
try testing.expectEqual(@as(usize, 0), one.calls);
try testing.expectEqual(@as(usize, 0), two.calls);
try testing.expectEqual(@as(usize, 0), one.calls.load(.acquire));
try testing.expectEqual(@as(usize, 0), two.calls.load(.acquire));
}
test "snapshot reports the counters in pool order" {
@@ -939,6 +1071,15 @@ test "snapshot reports the counters in pool order" {
try testing.expect(out[1].last_success_at != null);
try testing.expectEqual(@as(?std.Io.Timestamp, null), out[1].backoff_until);
// The concurrency fields: nothing is in flight once both exchanges have
// returned, the ceiling is the entry's slot count, and an uncontended pool
// queued nobody and recovered no session.
try testing.expectEqual(@as(u32, 0), out[1].in_flight);
try testing.expectEqual(@as(u32, 1), out[1].slots);
try testing.expectEqual(@as(u64, 0), out[1].queued_total);
try testing.expectEqual(@as(f64, 0), out[1].queued_seconds_total);
try testing.expectEqual(@as(u64, 0), out[1].reuse_recoveries_total);
// A short `out` truncates rather than overflowing.
var one: [1]Snapshot = undefined;
try testing.expectEqual(@as(usize, 1), try pool.snapshot(io, &one));
@@ -970,18 +1111,117 @@ fn exchangeAttributed(pool: *Pool, io: std.Io, buf: []u8) transport.ExchangeErro
return .{ .reply_len = reply.len, .selected = selected };
}
test "concurrent exchanges through one entry do not overlap" {
/// Blocks until `count` tasks are inside `fake.exchangeFn` at once.
///
/// `io.concurrent` does not promise the task it spawns has started, let alone
/// reached the leaf client, so a sleep proves nothing about who holds an entry's
/// permits. The queued counters are admission samples, and a test about them has
/// to *know* the entry is occupied before it starts the task that queues —
/// otherwise the waiter may be the one that wins the permit and the test passes
/// on the wrong interleaving. Bounded so a fake that never runs fails the test
/// instead of hanging the suite.
fn awaitInFlight(io: std.Io, fake: *Fake, count: u32) !void {
const step: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(1), .clock = .awake };
const steps = 10_000;
for (0..steps) |_| {
if (fake.in_flight.load(.acquire) >= count) return;
try step.sleep(io);
}
return error.FakeNeverEnteredExchange;
}
/// The stall one test's slow fake sleeps for. Long enough that two tasks
/// serialized through one slot would take twice it, short enough that a test
/// suite still ends promptly.
const overlap_stall_ms = 100;
test "concurrent exchanges through one entry overlap" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var fake: Fake = .{ .behavior = .{ .slow = .{
.duration = .{ .raw = .fromMilliseconds(overlap_stall_ms), .clock = .awake },
.reply = response_bytes,
} } };
var entries = [_]Entry{testEntrySlots("https://only.example/dns-query", &fake, 10, 2)};
var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);
var buf_a: [512]u8 = undefined;
var buf_b: [512]u8 = undefined;
const started = std.Io.Clock.awake.now(io);
var first = io.concurrent(exchangeAttributed, .{ &pool, io, &buf_a }) catch |err| switch (err) {
error.ConcurrencyUnavailable => return error.SkipZigTest,
};
defer _ = first.await(io) catch Attributed.discarded;
var second = io.concurrent(exchangeAttributed, .{ &pool, io, &buf_b }) catch |err| switch (err) {
error.ConcurrencyUnavailable => return error.SkipZigTest,
};
defer _ = second.await(io) catch Attributed.discarded;
const result_a = try first.await(io);
const result_b = try second.await(io);
const elapsed_ns = std.Io.Clock.awake.now(io).nanoseconds - started.nanoseconds;
try testing.expectEqualSlices(u8, response_bytes, buf_a[0..result_a.reply_len]);
try testing.expectEqualSlices(u8, response_bytes, buf_b[0..result_b.reply_len]);
try testing.expectEqualStrings("https://only.example/dns-query", result_a.selected.?);
try testing.expectEqualStrings("https://only.example/dns-query", result_b.selected.?);
try testing.expectEqual(@as(usize, 2), fake.calls.load(.acquire));
// The defect this milestone fixes: both tasks were inside the fake at once,
// and the pair cost one stall rather than two.
try testing.expectEqual(@as(u32, 2), fake.peak_in_flight.load(.acquire));
try testing.expect(elapsed_ns < @as(i96, 2 * overlap_stall_ms) * std.time.ns_per_ms);
try testing.expectEqual(@as(u64, 2), entries[0].health.total_successes);
// Neither task found the entry saturated, so neither queued.
try testing.expectEqual(@as(u64, 0), entries[0].queued_total.load(.acquire));
}
test "an entry never runs more exchanges at once than it has slots" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
// Slow enough that an unserialized second task would still be inside the
// fake when the first one is, and short enough to stay far under the
// 10-second attempt budget even when the two run back to back.
var fake: Fake = .{ .behavior = .{ .slow = .{
.duration = .{ .raw = .fromMilliseconds(50), .clock = .awake },
.reply = response_bytes,
} } };
var entries = [_]Entry{testEntrySlots("https://only.example/dns-query", &fake, 10, 2)};
var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);
var bufs: [3][512]u8 = undefined;
var calls: [3]std.Io.Future(transport.ExchangeError!Attributed) = undefined;
for (&calls, &bufs) |*call, *buf| {
call.* = io.concurrent(exchangeAttributed, .{ &pool, io, buf }) catch |err| switch (err) {
error.ConcurrencyUnavailable => return error.SkipZigTest,
};
}
// Every future is awaited before the first assertion: a failed assertion
// between two awaits would leave a task running past the end of the test.
var results: [3]transport.ExchangeError!Attributed = undefined;
for (&calls, &results) |*call, *result| result.* = call.await(io);
for (results, &bufs) |result, *buf| {
const value = try result;
try testing.expectEqualSlices(u8, response_bytes, buf[0..value.reply_len]);
}
try testing.expectEqual(@as(usize, 3), fake.calls.load(.acquire));
// The third task waited for a permit rather than entering a third client:
// the slot count, not the task count, is the ceiling.
try testing.expectEqual(@as(u32, 2), fake.peak_in_flight.load(.acquire));
try testing.expectEqual(@as(u32, 2), entries[0].peak_in_flight.load(.acquire));
}
test "a task that waits for a saturated entry is counted as queued" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var fake: Fake = .{ .behavior = .{ .slow = .{
.duration = .{ .raw = .fromMilliseconds(overlap_stall_ms), .clock = .awake },
.reply = response_bytes,
} } };
var entries = [_]Entry{testEntry("https://only.example/dns-query", &fake, 10)};
var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);
@@ -992,21 +1232,99 @@ test "concurrent exchanges through one entry do not overlap" {
error.ConcurrencyUnavailable => return error.SkipZigTest,
};
defer _ = first.await(io) catch Attributed.discarded;
// The counter is an admission sample, so the second task has to find the
// entry already occupied. Waiting for the first task to be inside the fake
// is what makes that true of every run.
try awaitInFlight(io, &fake, 1);
var second = io.concurrent(exchangeAttributed, .{ &pool, io, &buf_b }) catch |err| switch (err) {
error.ConcurrencyUnavailable => return error.SkipZigTest,
};
defer _ = second.await(io) catch Attributed.discarded;
const result_a = try first.await(io);
const result_b = try second.await(io);
_ = try first.await(io);
_ = try second.await(io);
try testing.expectEqualSlices(u8, response_bytes, buf_a[0..result_a.reply_len]);
try testing.expectEqualSlices(u8, response_bytes, buf_b[0..result_b.reply_len]);
try testing.expectEqualStrings("https://only.example/dns-query", result_a.selected.?);
try testing.expectEqualStrings("https://only.example/dns-query", result_b.selected.?);
try testing.expectEqual(@as(usize, 2), fake.calls);
try testing.expectEqual(@as(u32, 1), fake.peak_in_flight.load(.acquire));
try testing.expectEqual(@as(u64, 2), entries[0].health.total_successes);
try testing.expectEqual(@as(u64, 1), entries[0].queued_total.load(.acquire));
try testing.expect(entries[0].queued_ns_total.load(.acquire) > 0);
// An uncontended call adds neither.
const queued_ns = entries[0].queued_ns_total.load(.acquire);
fake.behavior = .{ .reply = response_bytes };
var selected: ?[]const u8 = null;
_ = try pool.exchange(io, query_bytes, &buf_a, &selected);
try testing.expectEqual(@as(u64, 1), entries[0].queued_total.load(.acquire));
try testing.expectEqual(queued_ns, entries[0].queued_ns_total.load(.acquire));
}
/// The stall the holder of the entry's one permit sits in below, and the whole
/// budget the task queued behind it gets. The gap between them is the test's
/// whole argument: the holder is still inside the fake long after the waiter's
/// budget is gone, so the waiter can only have died waiting for a permit.
const holder_stall_ms = 600;
const waiter_total_ms = 100;
test "a waiter canceled by the total budget is counted and still returns its permit" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var fake: Fake = .{ .behavior = .{ .slow = .{
.duration = .{ .raw = .fromMilliseconds(holder_stall_ms), .clock = .awake },
.reply = response_bytes,
} } };
var entries = [_]Entry{testEntry("https://only.example/dns-query", &fake, 10)};
// Two pools over one `entries` slice. A budget is a property of the pool,
// not of a call, and this test needs two: a generous one for the task that
// holds the permit and a short one for the task that queues behind it. A
// `Pool` borrows `entries`, so both see the same semaphore, the same queue
// counters and the same health; only `mutex` and the jitter RNG are
// per-pool. Sharing health across two pool mutexes is sound here because
// nothing writes it concurrently: the waiter dies before it attempts
// anything, and the holder's success is written before the last call runs.
var holder_pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);
var waiter_pool: Pool = .init(&entries, test_cfg, .{
.attempt = .{ .raw = .fromSeconds(30), .clock = .awake },
.total = .{ .raw = .fromMilliseconds(waiter_total_ms), .clock = .awake },
}, 1);
var buf_a: [512]u8 = undefined;
var buf_b: [512]u8 = undefined;
var holder = io.concurrent(exchangeAttributed, .{ &holder_pool, io, &buf_a }) catch |err| switch (err) {
error.ConcurrencyUnavailable => return error.SkipZigTest,
};
defer _ = holder.await(io) catch Attributed.discarded;
// The whole test rests on the holder owning the entry's one permit before
// the waiter asks for one. `io.concurrent` does not order those two, so the
// wait is on the fake itself: once a task is inside `exchangeFn` the permit
// is provably taken, and the call below can only queue.
try awaitInFlight(io, &fake, 1);
// This call never reaches the fake: it waits for the entry's one permit
// until its own total budget expires. That is the field failure — a burst
// dying in the queue — and it has to be visible.
var selected: ?[]const u8 = null;
try testing.expectError(error.Timeout, waiter_pool.exchange(io, query_bytes, &buf_b, &selected));
// Where it died, not just that it died: the holder's call is still the only
// one any leaf client has seen, so the cancellation landed inside
// `Semaphore.wait` rather than inside an attempt.
try testing.expectEqual(@as(usize, 1), fake.calls.load(.acquire));
try testing.expectEqual(@as(u64, 1), entries[0].queued_total.load(.acquire));
try testing.expect(entries[0].queued_ns_total.load(.acquire) > 0);
// The holder's own budget is generous, so it answers rather than expiring:
// the permit comes back from a completed attempt.
const held = try holder.await(io);
try testing.expectEqualSlices(u8, response_bytes, buf_a[0..held.reply_len]);
// The canceled waiter returned the permit it never held, so the entry is
// usable again — and the fresh call is the second one to reach the fake.
fake.behavior = .{ .reply = response_bytes };
const reply = try waiter_pool.exchange(io, query_bytes, &buf_b, &selected);
try testing.expectEqualSlices(u8, response_bytes, reply);
try testing.expectEqual(@as(usize, 2), fake.calls.load(.acquire));
}
/// The two entries of the test below, each answering with bytes only it
@@ -1076,8 +1394,8 @@ test "overlapping exchanges each report the entry that answered that call" {
try expectAnsweredByReporter(result_a, &buf_a);
try expectAnsweredByReporter(result_b, &buf_b);
try testing.expectEqual(@as(usize, 2), first_entry.calls);
try testing.expectEqual(@as(usize, 1), second_entry.calls);
try testing.expectEqual(@as(usize, 2), first_entry.calls.load(.acquire));
try testing.expectEqual(@as(usize, 1), second_entry.calls.load(.acquire));
try testing.expectEqual(@as(u32, 1), first_entry.peak_in_flight.load(.acquire));
try testing.expectEqual(@as(u64, 1), entries[0].health.total_failures);
try testing.expectEqual(@as(u64, 1), entries[0].health.total_successes);
@@ -1131,11 +1449,11 @@ test "an entry that enters backoff while a task waits on it is not attempted" {
try testing.expectEqualSlices(u8, response_bytes, buf_b[0..result_b.reply_len]);
try testing.expectEqualStrings("https://good.example/dns-query", result_a.selected.?);
try testing.expectEqualStrings("https://good.example/dns-query", result_b.selected.?);
try testing.expectEqual(@as(usize, 2), good.calls);
try testing.expectEqual(@as(usize, 2), good.calls.load(.acquire));
// The point of the test: the entry was attempted once, not twice. Without
// the re-check the second task would take the lock and attempt it anyway.
try testing.expectEqual(@as(usize, 1), slow_bad.calls);
try testing.expectEqual(@as(usize, 1), slow_bad.calls.load(.acquire));
try testing.expectEqual(@as(u64, 1), entries[0].health.total_failures);
try testing.expect(entries[0].health.backoff_until != null);
}
src/web/metrics.zig
+259 -7
View File
@@ -113,6 +113,11 @@ pub const UpstreamSample = struct {
total_successes: u64,
total_failures: u64,
success_rate: f32,
in_flight: u32,
slots: u32,
queued_total: u64,
queued_seconds_total: f64,
reuse_recoveries_total: u64,
};
/// The diagnostics store, as one scrape sees it. Two gauges and a counter,
@@ -318,6 +323,11 @@ fn upstreams(pool: *pool_mod.Pool, io: std.Io, arena: Allocator) Allocator.Error
.total_successes = entry.total_successes,
.total_failures = entry.total_failures,
.success_rate = entry.success_rate,
.in_flight = entry.in_flight,
.slots = entry.slots,
.queued_total = entry.queued_total,
.queued_seconds_total = entry.queued_seconds_total,
.reuse_recoveries_total = entry.reuse_recoveries_total,
};
}
return out;
@@ -510,6 +520,47 @@ fn renderUpstreams(w: *std.Io.Writer, list: []const UpstreamSample) std.Io.Write
for (list, 0..) |entry, i| {
try labeledValue(w, "nxdns_upstream_failures_total", i, entry.url, entry.total_failures);
}
try labeledHead(w, "nxdns_upstream_in_flight", "Exchanges in flight against an upstream.", "gauge");
for (list, 0..) |entry, i| {
try labeledValue(w, "nxdns_upstream_in_flight", i, entry.url, entry.in_flight);
}
try labeledHead(w, "nxdns_upstream_slots", "Concurrent exchanges an upstream allows.", "gauge");
for (list, 0..) |entry, i| {
try labeledValue(w, "nxdns_upstream_slots", i, entry.url, entry.slots);
}
try labeledHead(
w,
"nxdns_upstream_queued_total",
"Exchanges that waited for a slot. Approximate; sampled at admission.",
"counter",
);
for (list, 0..) |entry, i| {
try labeledValue(w, "nxdns_upstream_queued_total", i, entry.url, entry.queued_total);
}
try labeledHead(
w,
"nxdns_upstream_queued_seconds_total",
"Seconds exchanges spent waiting for a slot. Approximate; sampled at admission.",
"counter",
);
for (list, 0..) |entry, i| {
try writeUpstreamLabels(w, "nxdns_upstream_queued_seconds_total", i, entry.url);
try w.print(" {d:.4}\n", .{entry.queued_seconds_total});
}
try labeledHead(
w,
"nxdns_upstream_reuse_recoveries_total",
"Stale reused DoT connections recovered by a redial.",
"counter",
);
for (list, 0..) |entry, i| {
try labeledValue(w, "nxdns_upstream_reuse_recoveries_total", i, entry.url, entry.reuse_recoveries_total);
}
}
/// Every field of a plain counter struct, under one prefix.
@@ -553,7 +604,7 @@ fn labeledValue(
}
/// The label set every upstream family shares, up to and including the closing
/// brace. One definition, because six families have to agree on it exactly:
/// brace. One definition, because eleven families have to agree on it exactly:
/// Prometheus identifies a series by its name and its whole label set, so a
/// family that labelled its samples differently would be a different series.
///
@@ -651,6 +702,7 @@ const db = @import("../storage/db.zig");
const local_tables = @import("../server/local_tables.zig");
const logger_mod = @import("../storage/logger.zig");
const migrations = @import("../storage/migrations.zig");
const transport = @import("../upstream/transport.zig");
const testing = std.testing;
/// A handler with no upstream reachable: every test here reads counters and
@@ -684,6 +736,11 @@ test "a full sample renders the whole exposition, byte for byte" {
.total_successes = 9,
.total_failures = 1,
.success_rate = 0.9,
.in_flight = 2,
.slots = 8,
.queued_total = 3,
.queued_seconds_total = 0.25,
.reuse_recoveries_total = 1,
},
};
@@ -760,10 +817,16 @@ test "a full sample renders the whole exposition, byte for byte" {
1,
"nxdns_upstream_success_rate{index=\"0\",url=\"https://dns.example\"} 0.9000\n",
));
try testing.expect(std.mem.containsAtLeast(
u8,
text,
1,
"nxdns_upstream_failures_total{index=\"0\",url=\"https://dns.example\"} 1\n",
));
try testing.expect(std.mem.endsWith(
u8,
text,
"nxdns_upstream_failures_total{index=\"0\",url=\"https://dns.example\"} 1\n",
"nxdns_upstream_reuse_recoveries_total{index=\"0\",url=\"https://dns.example\"} 1\n",
));
}
@@ -1115,12 +1178,18 @@ test "a label value escapes the characters the format reserves" {
.total_successes = 0,
.total_failures = 2,
.success_rate = 0,
.in_flight = 1,
.slots = 8,
.queued_total = 4,
.queued_seconds_total = 1.5,
.reuse_recoveries_total = 2,
}};
const text = try renderToString(testing.allocator, .{ .upstreams = &upstream_list });
defer testing.allocator.free(text);
// The url wrote no line of its own, and lost none: every line is a
// comment or a sample, and the six families contribute six samples.
// comment or a sample, and the eleven families contribute eleven
// samples.
var samples: usize = 0;
var lines = std.mem.splitScalar(u8, text, '\n');
while (lines.next()) |line| {
@@ -1131,7 +1200,7 @@ test "a label value escapes the characters the format reserves" {
// Both labels are there, and both values close where they opened.
try testing.expectEqual(@as(?usize, 2), labelPairs(line));
}
try testing.expectEqual(@as(usize, 6), samples);
try testing.expectEqual(@as(usize, 11), samples);
}
}
@@ -1149,6 +1218,11 @@ test "two upstreams on one host stay two series" {
.total_successes = 5,
.total_failures = 0,
.success_rate = 1,
.in_flight = 0,
.slots = 8,
.queued_total = 0,
.queued_seconds_total = 0,
.reuse_recoveries_total = 0,
},
.{
.url = "https://dns.nextdns.io/efgh34",
@@ -1158,6 +1232,11 @@ test "two upstreams on one host stay two series" {
.total_successes = 9,
.total_failures = 3,
.success_rate = 0.75,
.in_flight = 3,
.slots = 8,
.queued_total = 7,
.queued_seconds_total = 2,
.reuse_recoveries_total = 5,
},
};
const text = try renderToString(testing.allocator, .{ .upstreams = &upstream_list });
@@ -1206,7 +1285,7 @@ test "two upstreams on one host stay two series" {
try testing.expectEqualStrings("0", up_values[1]);
// No two samples in the scrape share a series key, whatever the urls were.
var keys: [32][]const u8 = undefined;
var keys: [64][]const u8 = undefined;
var count: usize = 0;
var lines = std.mem.splitScalar(u8, text, '\n');
while (lines.next()) |line| {
@@ -1216,7 +1295,7 @@ test "two upstreams on one host stay two series" {
keys[count] = key;
count += 1;
}
try testing.expectEqual(@as(usize, 12), count);
try testing.expectEqual(@as(usize, 22), count);
}
test "an upstream url is redacted before it reaches an open endpoint's label" {
@@ -1233,6 +1312,11 @@ test "an upstream url is redacted before it reaches an open endpoint's label" {
.total_successes = 3,
.total_failures = 0,
.success_rate = 1,
.in_flight = 0,
.slots = 8,
.queued_total = 0,
.queued_seconds_total = 0,
.reuse_recoveries_total = 0,
},
.{
.url = "https://user:hunter2@dns.example:8443/dns-query?apikey=s3cr3t#frag",
@@ -1242,6 +1326,11 @@ test "an upstream url is redacted before it reaches an open endpoint's label" {
.total_successes = 0,
.total_failures = 4,
.success_rate = 0,
.in_flight = 0,
.slots = 8,
.queued_total = 0,
.queued_seconds_total = 0,
.reuse_recoveries_total = 0,
},
};
const text = try renderToString(testing.allocator, .{ .upstreams = &upstream_list });
@@ -1254,7 +1343,8 @@ test "an upstream url is redacted before it reaches an open endpoint's label" {
try testing.expect(!std.mem.containsAtLeast(u8, text, 1, "frag"));
try testing.expect(!std.mem.containsAtLeast(u8, text, 1, "dns-query"));
// Every family carries the label, so none of the six may keep the whole url.
// Every family carries the label, so none of the eleven may keep the whole
// url.
try testing.expect(std.mem.containsAtLeast(
u8,
text,
@@ -1302,6 +1392,54 @@ test "an upstream url is redacted before it reaches an open endpoint's label" {
));
}
test "the queue families render with the label set every upstream family shares" {
// The surface the Pi burst defect is read from: whether an upstream is
// queueing (`queued_total`/`in_flight` against `slots`) and whether it is
// churning connections (`reuse_recoveries_total`).
const upstream_list = [_]UpstreamSample{.{
.url = "tls://dns.example:853",
.enabled = true,
.available = true,
.consecutive_failures = 0,
.total_successes = 30,
.total_failures = 0,
.success_rate = 1,
.in_flight = 6,
.slots = 8,
.queued_total = 22,
.queued_seconds_total = 1.5,
.reuse_recoveries_total = 3,
}};
const text = try renderToString(testing.allocator, .{ .upstreams = &upstream_list });
defer testing.allocator.free(text);
const label = "{index=\"0\",url=\"tls://dns.example:853\"}";
try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_upstream_in_flight" ++ label ++ " 6\n"));
try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_upstream_slots" ++ label ++ " 8\n"));
try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_upstream_queued_total" ++ label ++ " 22\n"));
try testing.expect(std.mem.containsAtLeast(
u8,
text,
1,
"nxdns_upstream_queued_seconds_total" ++ label ++ " 1.5000\n",
));
try testing.expect(std.mem.containsAtLeast(
u8,
text,
1,
"nxdns_upstream_reuse_recoveries_total" ++ label ++ " 3\n",
));
// The types a scrape reads them as, and the approximation the two queue
// counters carry stated where an operator meets them.
try testing.expect(std.mem.containsAtLeast(u8, text, 1, "# TYPE nxdns_upstream_in_flight gauge\n"));
try testing.expect(std.mem.containsAtLeast(u8, text, 1, "# TYPE nxdns_upstream_slots gauge\n"));
try testing.expect(std.mem.containsAtLeast(u8, text, 1, "# TYPE nxdns_upstream_queued_total counter\n"));
try testing.expect(std.mem.containsAtLeast(u8, text, 1, "# TYPE nxdns_upstream_queued_seconds_total counter\n"));
try testing.expect(std.mem.containsAtLeast(u8, text, 1, "# TYPE nxdns_upstream_reuse_recoveries_total counter\n"));
try testing.expectEqual(@as(usize, 2), std.mem.count(u8, text, "Approximate; sampled at admission."));
}
test "a url longer than the redaction bound cannot run past it" {
const long_host = "h" ** (4 * safe_url.max_len);
const upstream_list = [_]UpstreamSample{.{
@@ -1312,6 +1450,11 @@ test "a url longer than the redaction bound cannot run past it" {
.total_successes = 0,
.total_failures = 0,
.success_rate = 1,
.in_flight = 0,
.slots = 8,
.queued_total = 0,
.queued_seconds_total = 0,
.reuse_recoveries_total = 0,
}};
const text = try renderToString(testing.allocator, .{ .upstreams = &upstream_list });
defer testing.allocator.free(text);
@@ -1325,6 +1468,115 @@ test "a url longer than the redaction bound cannot run past it" {
));
}
/// A leaf client that answers, and counts that it was asked.
///
/// `pool.zig`'s own fake is private to that file and models failure modes this
/// test has no use for; all this one has to do is let a real `Pool.exchange`
/// complete, so the health half of the sample below is the pool's own
/// bookkeeping rather than a literal.
const AnsweringClient = struct {
calls: std.atomic.Value(u32) = .init(0),
const reply = "\x12\x34\x81\x80";
fn exchangeFn(
ptr: *anyopaque,
io: std.Io,
query: []const u8,
response_buf: []u8,
selected: *?[]const u8,
) transport.ExchangeError![]u8 {
_ = io;
_ = query;
const self: *AnsweringClient = @ptrCast(@alignCast(ptr));
_ = self.calls.fetchAdd(1, .acq_rel);
selected.* = "fake://leaf";
@memcpy(response_buf[0..reply.len], reply);
return response_buf[0..reply.len];
}
fn client(self: *AnsweringClient) transport.Client {
return .{ .ptr = self, .exchangeFn = exchangeFn };
}
};
test "the queue families carry what a real pool recorded, through the real snapshot" {
// The test above renders a hand-built `UpstreamSample`, so it proves the
// exposition and nothing else. This one drives a real `Pool` and goes
// through `Pool.snapshot` and `upstreams`, which is where a crossed field
// would live. Every driven value is distinct for that reason: a swap
// anywhere along the path renders the wrong number rather than a match.
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var leaf: AnsweringClient = .{};
var slots = [_]pool_mod.Slot{
.{ .client = leaf.client() },
.{ .client = leaf.client() },
};
var recoveries: std.atomic.Value(u64) = .init(0);
var entries = [_]pool_mod.Entry{.{
.endpoint = try .parse("tls://dns.example:853"),
.slots = &slots,
.priority = 10,
.enabled = true,
.health = .init,
.sem = .{ .permits = slots.len },
.reuse_recoveries = &recoveries,
}};
var pool: pool_mod.Pool = .init(&entries, .{}, .{
.attempt = .{ .raw = .fromSeconds(10), .clock = .awake },
.total = .{ .raw = .fromSeconds(30), .clock = .awake },
}, 1);
var buf: [512]u8 = undefined;
var selected: ?[]const u8 = null;
_ = try pool.exchange(io, "\x12\x34\x01\x00", &buf, &selected);
try testing.expectEqual(@as(u32, 1), leaf.calls.load(.acquire));
// The counters a burst would move, written straight into the entry: this
// test is about what the snapshot path carries, not about reproducing a
// queue. The exchange above has already returned, so nothing else is
// touching them.
entries[0].in_flight.store(5, .release);
entries[0].queued_total.store(7, .release);
entries[0].queued_ns_total.store(1_500_000_000, .release);
recoveries.store(9, .release);
var arena: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena.deinit();
const list = try upstreams(&pool, io, arena.allocator());
const text = try renderToString(testing.allocator, .{ .upstreams = list });
defer testing.allocator.free(text);
const label = "{index=\"0\",url=\"tls://dns.example:853\"}";
try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_upstream_in_flight" ++ label ++ " 5\n"));
try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_upstream_slots" ++ label ++ " 2\n"));
try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_upstream_queued_total" ++ label ++ " 7\n"));
try testing.expect(std.mem.containsAtLeast(
u8,
text,
1,
"nxdns_upstream_queued_seconds_total" ++ label ++ " 1.5000\n",
));
try testing.expect(std.mem.containsAtLeast(
u8,
text,
1,
"nxdns_upstream_reuse_recoveries_total" ++ label ++ " 9\n",
));
// The one series the pool wrote by itself, so the five above are read
// against an entry the pool really did use.
try testing.expect(std.mem.containsAtLeast(
u8,
text,
1,
"nxdns_upstream_successes_total" ++ label ++ " 1\n",
));
}
test "collect reads the live counters of the components it is given" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
src/web/web_integration_test.zig
+9 -3
View File
@@ -313,6 +313,8 @@ const Env = struct {
pauser: pause_mod.Pause,
tables: local_tables_mod.LocalTables,
pool_entries: [1]pool_mod.Entry,
pool_slots: [1]pool_mod.Slot,
pool_recoveries: std.atomic.Value(u64),
pool: pool_mod.Pool,
state: server.WebState,
web: server.Server,
@@ -387,14 +389,18 @@ const Env = struct {
self.pauser = .{};
self.tables = .empty;
// Never exchanged with: the pool feeds `/metrics` and the
// `/api/health` upstream condition only.
self.pool_slots = .{.{ .client = .{ .ptr = undefined, .exchangeFn = undefined } }};
self.pool_recoveries = .init(0);
self.pool_entries = .{.{
.endpoint = transport.Endpoint.parse("https://dns.example/dns-query") catch unreachable,
// Never exchanged with: the pool feeds `/metrics` and the
// `/api/health` upstream condition only.
.client = .{ .ptr = undefined, .exchangeFn = undefined },
.slots = &self.pool_slots,
.priority = 1,
.enabled = true,
.health = .init,
.sem = .{ .permits = self.pool_slots.len },
.reuse_recoveries = &self.pool_recoveries,
}};
self.pool = .init(&self.pool_entries, .{}, .{
.attempt = .{ .raw = .fromMilliseconds(50), .clock = .awake },