nxdns/src/upstream/pool.zig

//! Priority-ordered sequential failover across upstream endpoints, with two
//! deadlines, health tracking and backoff (PLAN §9).
//!
//! The pool is itself a `transport.Client`, so the handler above it sees one
//! interface and knows nothing about how many upstreams exist.
//!
//! Two deadlines, nested. `timeouts.attempt` bounds one exchange against one
//! entry; `timeouts.total` bounds the whole failover loop, every attempt
//! together. The outer one is what the client asking the question actually
//! 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 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
//! backoff check: every endpoint being in backoff must not turn into SERVFAIL
//! for every client. A probe is better than a guaranteed failure, and the probe
//! is how backoff recovers.
//!
//! Concurrency (PLAN §9, spec S4.2): several handler tasks share one `Pool`,
//! and they hold two different locks at two different scopes.
//!
//! `Pool.mutex` guards health mutation, the jitter RNG and `snapshot`. It is
//! never held across an exchange, so a slow upstream cannot block a health read
//! or an attempt on some other entry.
//!
//! 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 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 `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");

const events = @import("../storage/events.zig");
const health = @import("health.zig");
const safe_url = @import("../safe_url.zig");
const transport = @import("transport.zig");

const log = std.log.scoped(.upstream);

/// The one line `exchange` writes about a failed attempt, as a value.
///
/// It is a value rather than a format string at the call site for the same
/// reason `dot_client.Diagnostic` is: a `std.log` line is not readable from a
/// unit test under the default test runner, so the test below reads this
/// instead of stderr. An upstream url is operator-supplied and a DoH one carries
/// its credential in the path — `https://dns.nextdns.io/abcd12` is a whole
/// NextDNS account identifier — so it reaches the line through
/// `safe_url.redactQuoted`. Quoted rather than bare because the error name
/// follows it: a redacted authority may still hold a space and a `:`, so
/// unquoted, a url ending `ok failed: Timeout` would report a failure that did
/// not happen.
const AttemptFailure = struct {
    endpoint: transport.Endpoint,
    err: transport.ExchangeError,

    pub fn format(self: AttemptFailure, w: *std.Io.Writer) std.Io.Writer.Error!void {
        try w.print("upstream {f} failed: {t}", .{ safe_url.redactQuoted(self.endpoint.url), self.err });
    }
};

/// 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,
    /// 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,
    /// 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
/// the `/api/health` upstream condition.
pub const Snapshot = struct {
    /// Whole, not redacted. `GET /api/upstreams` already serves the same url in
    /// full to a session, so redacting here would hide nothing from that reader
    /// and would make two responses of one API disagree. A consumer reachable
    /// without a session has to redact it itself.
    url: []const u8,
    enabled: bool,
    available: bool,
    consecutive_failures: u32,
    total_successes: u64,
    total_failures: u64,
    success_rate: f32,
    last_success_at: ?std.Io.Timestamp,
    last_error_at: ?std.Io.Timestamp,
    /// 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
/// two parameters in a row could be swapped at a call site and still compile,
/// and the swap would be invisible until an operator watched a query take five
/// attempts of two and a half seconds each.
pub const Timeouts = struct {
    /// Bounds one exchange against one entry.
    attempt: std.Io.Clock.Duration,
    /// Bounds the whole failover loop.
    total: std.Io.Clock.Duration,
};

pub const Pool = struct {
    /// Caller-owned, sorted ascending by priority in `init`.
    entries: []Entry,
    cfg: health.Config,
    /// Both on the `.awake` clock, so a suspended Pi does not burn a budget.
    timeouts: Timeouts,
    mutex: std.Io.Mutex,
    rng: std.Random.DefaultPrng,
    /// The diagnostics store. Defaulted rather than an `init` parameter: the
    /// composition root wires it after the pool exists, and the pool is fully
    /// usable without it — `nxdns check` and every unit test here run with no
    /// store at all. Every emit here sits outside `mutex`; see
    /// `recordDiagnostics`.
    diagnostics: ?*events.Store = null,

    pub fn init(
        entries: []Entry,
        cfg: health.Config,
        timeouts: Timeouts,
        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 .{
            .entries = entries,
            .cfg = cfg,
            .timeouts = timeouts,
            .mutex = .init,
            .rng = .init(seed),
        };
    }

    fn byPriority(_: void, a: Entry, b: Entry) bool {
        return a.priority < b.priority;
    }

    pub fn client(self: *Pool) transport.Client {
        return .{ .ptr = self, .exchangeFn = exchangeErased };
    }

    fn exchangeErased(
        ptr: *anyopaque,
        io: std.Io,
        query: []const u8,
        response_buf: []u8,
        selected: *?[]const u8,
    ) transport.ExchangeError![]u8 {
        const self: *Pool = @ptrCast(@alignCast(ptr));
        return self.exchange(io, query, response_buf, selected);
    }

    /// `response_buf` is handed to each attempt in turn, so a failed attempt
    /// may have written into it. The returned slice is only meaningful on
    /// 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 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, no permit lost and no counter
    /// half-written.
    pub fn exchange(
        self: *Pool,
        io: std.Io,
        query: []const u8,
        response_buf: []u8,
        selected: *?[]const u8,
    ) transport.ExchangeError![]u8 {
        const len = try transport.raceWithin(io, self.timeouts.total, exchangeLoopLen, .{
            self, io, query, response_buf, selected,
        });
        return response_buf[0..len];
    }

    /// The two-pass failover loop, as a raceable task. It returns the reply's
    /// length rather than its slice for the reason `Attributed` in the tests
    /// below does: `Io.concurrent` stores the future's return value, so the
    /// bytes are read back out of the caller's `response_buf` by `exchange`.
    fn exchangeLoopLen(
        self: *Pool,
        io: std.Io,
        query: []const u8,
        response_buf: []u8,
        selected: *?[]const u8,
    ) transport.ExchangeError!usize {
        const now = std.Io.Clock.awake.now(io);
        var last_fault: ?transport.ExchangeError = null;
        var attempted = false;

        var pass: u8 = 0;
        while (pass < 2) : (pass += 1) {
            for (self.entries) |*entry| {
                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 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 — 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);
                    if (!self.entryAvailable(io, entry, recheck)) continue;
                }
                attempted = true;

                // Before the attempt, not after it: a failover reports whoever
                // answered, an all-failed exchange reports the last endpoint
                // tried, and a cancellation mid-flight reports the endpoint the
                // query was in. The url is owned by the Endpoint, which outlives
                // the pool, so the borrow stays valid past this loop.
                selected.* = entry.endpoint.url;

                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)) {
                    .peer_fault => {
                        log.debug("{f}", .{AttemptFailure{ .endpoint = entry.endpoint, .err = err }});
                        self.recordFailure(io, entry, completed_at, err);
                        last_fault = err;
                        continue;
                    },
                    // The next upstream would hit the same wall, and this says
                    // nothing about any peer, so it is never recorded.
                    .local_resource => return err,
                    .cancellation => return error.Canceled,
                };

                self.recordSuccess(io, entry, completed_at);
                return response.len;
            }
            if (attempted) break;
        }

        // Guaranteed non-null whenever any entry is enabled: pass two attempts
        // every enabled entry regardless of backoff.
        if (last_fault) |err| return err;
        return error.ConnectFailed;
    }

    /// Copies health into `out` in pool order; returns the number written.
    pub fn snapshot(self: *Pool, io: std.Io, out: []Snapshot) std.Io.Cancelable!usize {
        const now = std.Io.Clock.awake.now(io);
        try self.mutex.lock(io);
        defer self.mutex.unlock(io);

        const count = @min(self.entries.len, out.len);
        for (self.entries[0..count], out[0..count]) |*entry, *slot| {
            slot.* = .{
                .url = entry.endpoint.url,
                .enabled = entry.enabled,
                .available = entry.enabled and entry.health.available(now),
                .consecutive_failures = entry.health.consecutive_failures,
                .total_successes = entry.health.total_successes,
                .total_failures = entry.health.total_failures,
                .success_rate = entry.health.successRate(),
                .last_success_at = entry.health.last_success_at,
                .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;
    }

    /// One exchange raced against the per-attempt budget.
    ///
    /// The leaf client reports an identity of its own, which the pool discards:
    /// the entry's endpoint is the pool's own naming of the same resolver, and
    /// it is what the caller was handed. A leaf writing its identity into the
    /// caller's slot would let a test fake overwrite the endpoint that actually
    /// answered.
    fn attempt(
        self: *Pool,
        io: std.Io,
        entry_client: transport.Client,
        query: []const u8,
        response_buf: []u8,
    ) transport.ExchangeError![]u8 {
        var leaf_selected: ?[]const u8 = null;
        return transport.raceWithin(io, self.timeouts.attempt, transport.Client.exchange, .{
            entry_client, io, query, response_buf, &leaf_selected,
        });
    }

    fn entryAvailable(
        self: *Pool,
        io: std.Io,
        entry: *const Entry,
        now: std.Io.Timestamp,
    ) bool {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);
        return entry.health.available(now);
    }

    fn recordSuccess(self: *Pool, io: std.Io, entry: *Entry, at: std.Io.Timestamp) void {
        {
            // Uncancelable: this section takes no Io and never blocks on a
            // peer. Losing the bookkeeping for a completed exchange to a
            // cancellation that arrives one instruction later would corrupt
            // health for good.
            self.mutex.lockUncancelable(io);
            defer self.mutex.unlock(io);
            entry.health.recordSuccess(at);
        }
        // The block above closes before this line, and that ordering is the
        // constraint: the store takes a mutex of its own, and no task may hold
        // one of the two while it takes the other.
        self.recordDiagnostics(io, entry, .success);
    }

    fn recordFailure(
        self: *Pool,
        io: std.Io,
        entry: *Entry,
        at: std.Io.Timestamp,
        err: transport.ExchangeError,
    ) void {
        {
            self.mutex.lockUncancelable(io);
            defer self.mutex.unlock(io);
            entry.health.recordFailure(at, @errorName(err), self.cfg, self.rng.random().int(u32));
        }
        // After the pool mutex is released, for the reason `recordSuccess`
        // states.
        self.recordDiagnostics(io, entry, .{ .failure = @errorName(err) });
    }

    const Outcome = union(enum) { success, failure: []const u8 };

    /// The store takes a mutex of its own, so this runs after the pool's is
    /// released.
    ///
    /// A success is the steady state of the whole program, so `resolve` is
    /// built to issue no SQL when nothing is open (`storage/events.zig`).
    fn recordDiagnostics(self: *Pool, io: std.Io, entry: *Entry, outcome: Outcome) void {
        const store = self.diagnostics orelse return;
        const url = entry.endpoint.url;
        const now_s = std.Io.Clock.real.now(io).toSeconds();
        switch (outcome) {
            .success => store.resolve(io, now_s, .upstream_exchange, url),
            .failure => |name| {
                var label_buf: [events.Store.max_subject_label_len]u8 = undefined;
                const label = std.fmt.bufPrint(&label_buf, "{f}", .{safe_url.redact(url)}) catch &label_buf;
                var detail_buf: [events.Store.max_detail_len]u8 = undefined;
                const detail = std.fmt.bufPrint(&detail_buf, "upstream {f} failed: {s}", .{
                    safe_url.redactQuoted(url),
                    name,
                }) catch &detail_buf;
                store.report(io, now_s, .upstream_exchange, url, label, .warning, detail);
            },
        }
    }
};

const events_fixture = @import("../storage/events_fixture.zig");

const testing = std.testing;

/// 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 question answered differently, so a reply identifies the entry that
/// produced it: one A record, a different address.
const alt_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\x0a\x00\x00\x01";

/// Stands in for a DoH or DoT client. Every behaviour the pool has to react to
/// is one variant, and every call is counted so a test can assert that an entry
/// in backoff was not touched.
const Fake = struct {
    behavior: Behavior,
    /// 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.
    then: ?Behavior = null,
    /// 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 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.
        reply: []const u8,
        /// Fail with this error.
        fail: transport.ExchangeError,
        /// Sleep, then reply. Used to outrun the pool's attempt budget.
        slow: struct { duration: std.Io.Clock.Duration, reply: []const u8 },
        /// Sleep, then fail. Holds the entry's lock long enough for a second
        /// task to queue on it before the failure opens a backoff window.
        slow_fail: struct { duration: std.Io.Clock.Duration, err: transport.ExchangeError },
    };

    fn exchangeFn(
        ptr: *anyopaque,
        io: std.Io,
        query: []const u8,
        response_buf: []u8,
        selected: *?[]const u8,
    ) transport.ExchangeError![]u8 {
        _ = query;
        // Deliberately not the endpoint url: the pool must report its own
        // entry, so a test can tell the two apart.
        selected.* = "fake://leaf";
        const self: *Fake = @ptrCast(@alignCast(ptr));
        const entrants = self.in_flight.fetchAdd(1, .acq_rel) + 1;
        defer _ = self.in_flight.fetchSub(1, .acq_rel);
        _ = self.peak_in_flight.fetchMax(entrants, .acq_rel);

        _ = 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,
            .slow => |slow| {
                try slow.duration.sleep(io);
                return copy(slow.reply, response_buf);
            },
            .slow_fail => |slow| {
                try slow.duration.sleep(io);
                return slow.err;
            },
        }
    }

    fn copy(bytes: []const u8, response_buf: []u8) transport.ExchangeError![]u8 {
        if (bytes.len > response_buf.len) return error.ResponseTooLarge;
        @memcpy(response_buf[0..bytes.len], bytes);
        return response_buf[0..bytes.len];
    }

    fn client(self: *Fake) transport.Client {
        return .{ .ptr = self, .exchangeFn = exchangeFn };
    }
};

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,
        .slots = slots,
        .priority = priority,
        .enabled = true,
        .health = .init,
        .sem = .{ .permits = slot_count },
        .reuse_recoveries = &fake.recoveries,
    };
}

const Endpoint = transport.Endpoint;

/// Long enough that no test can outlive a backoff it just set, and short enough
/// that nothing waits on it.
const test_cfg: health.Config = .{
    .failure_threshold = 2,
    .base_backoff_ms = 60_000,
    .max_backoff_ms = 60_000,
};

/// Long enough that nothing in a test reaches either budget unless the test is
/// about a budget, and short enough that a stuck test still ends.
const test_timeouts: Timeouts = .{
    .attempt = .{ .raw = .fromSeconds(10), .clock = .awake },
    .total = .{ .raw = .fromSeconds(30), .clock = .awake },
};

fn expectFailureLine(expected: []const u8, url: []const u8, err: transport.ExchangeError) !void {
    var buf: [8 * safe_url.max_len]u8 = undefined;
    const line = try std.fmt.bufPrint(&buf, "{f}", .{AttemptFailure{
        .endpoint = try .parse(url),
        .err = err,
    }});
    try testing.expectEqualStrings(expected, line);
}

test "the failed-attempt line names an upstream by a url carrying no credential" {
    // A NextDNS DoH upstream puts the whole account identifier in the path, and
    // this line ran at `debug` on every peer fault, so a debug-level operator
    // persisted it to the journal once per failure.
    try expectFailureLine(
        "upstream 'https://dns.nextdns.io' failed: Timeout",
        "https://dns.nextdns.io/abcd12",
        error.Timeout,
    );
    try expectFailureLine(
        "upstream 'https://cdn.example:8443' failed: TlsFailed",
        "https://cdn.example:8443/d/hunter2/dns-query",
        error.TlsFailed,
    );
    // What it still says, because an operator reading a failover has to know
    // which upstream failed: the scheme, the host and the port.
    try expectFailureLine(
        "upstream 'tls://9.9.9.9:853' failed: ConnectFailed",
        "tls://9.9.9.9:853",
        error.ConnectFailed,
    );
}

test "Pool satisfies the Client interface" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var fake: Fake = .{ .behavior = .{ .reply = response_bytes } };
    var entries = [_]Entry{testEntry("https://a.example/dns-query", &fake, 10)};
    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);

    var buf: [512]u8 = undefined;
    var selected: ?[]const u8 = null;
    const reply = try pool.client().exchange(io, query_bytes, &buf, &selected);
    try testing.expectEqualSlices(u8, response_bytes, reply);
}

test "the pool reports the endpoint that answered, not the leaf client's own name" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var fake: Fake = .{ .behavior = .{ .reply = response_bytes } };
    var entries = [_]Entry{testEntry("https://a.example/dns-query", &fake, 10)};
    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);

    var buf: [512]u8 = undefined;
    var selected: ?[]const u8 = null;
    _ = try pool.exchange(io, query_bytes, &buf, &selected);
    try testing.expectEqualStrings("https://a.example/dns-query", selected.?);
}

test "a failover reports the endpoint that answered, not the first one tried" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var bad: Fake = .{ .behavior = .{ .fail = error.Timeout } };
    var good: Fake = .{ .behavior = .{ .reply = response_bytes } };
    var entries = [_]Entry{
        testEntry("https://bad.example/dns-query", &bad, 10),
        testEntry("https://good.example/dns-query", &good, 20),
    };
    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);

    var buf: [512]u8 = undefined;
    var selected: ?[]const u8 = null;
    _ = try pool.exchange(io, query_bytes, &buf, &selected);
    try testing.expectEqualStrings("https://good.example/dns-query", selected.?);
}

test "an all-failed exchange reports the last endpoint attempted" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var first: Fake = .{ .behavior = .{ .fail = error.ConnectFailed } };
    var last: Fake = .{ .behavior = .{ .fail = error.Timeout } };
    var entries = [_]Entry{
        testEntry("https://first.example/dns-query", &first, 10),
        testEntry("https://last.example/dns-query", &last, 20),
    };
    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);

    var buf: [512]u8 = undefined;
    var selected: ?[]const u8 = null;
    try testing.expectError(error.Timeout, pool.exchange(io, query_bytes, &buf, &selected));
    try testing.expectEqualStrings("https://last.example/dns-query", selected.?);
}

test "a timeout mid-flight reports the endpoint the query was in" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var stalling: Fake = .{ .behavior = .{ .slow = .{
        .duration = .{ .raw = .fromSeconds(30), .clock = .awake },
        .reply = response_bytes,
    } } };
    var untouched: Fake = .{ .behavior = .{ .reply = response_bytes } };
    var entries = [_]Entry{
        testEntry("https://stalling.example/dns-query", &stalling, 10),
        testEntry("https://untouched.example/dns-query", &untouched, 20),
    };
    var pool: Pool = .init(&entries, test_cfg, .{
        .attempt = .{ .raw = .fromMilliseconds(200), .clock = .awake },
        .total = .{ .raw = .fromMilliseconds(60), .clock = .awake },
    }, 1);

    var buf: [512]u8 = undefined;
    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.load(.acquire));
}

test "an exchange that attempted nothing reports no endpoint" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var fake: Fake = .{ .behavior = .{ .reply = response_bytes } };
    var entries = [_]Entry{testEntry("https://a.example/dns-query", &fake, 10)};
    entries[0].enabled = false;
    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);

    var buf: [512]u8 = undefined;
    var selected: ?[]const u8 = null;
    try testing.expectError(error.ConnectFailed, pool.exchange(io, query_bytes, &buf, &selected));
    try testing.expect(selected == null);
}

test "entries are tried in ascending priority order" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var low: Fake = .{ .behavior = .{ .reply = response_bytes } };
    var high: Fake = .{ .behavior = .{ .reply = response_bytes } };
    // Registered out of order: priority, not position, decides.
    var entries = [_]Entry{
        testEntry("https://high.example/dns-query", &high, 100),
        testEntry("https://low.example/dns-query", &low, 10),
    };
    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);

    try testing.expectEqual(@as(i32, 10), entries[0].priority);

    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.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" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var bad: Fake = .{ .behavior = .{ .fail = error.Timeout } };
    var good: Fake = .{ .behavior = .{ .reply = response_bytes } };
    var entries = [_]Entry{
        testEntry("https://bad.example/dns-query", &bad, 10),
        testEntry("https://good.example/dns-query", &good, 20),
    };
    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);

    var buf: [512]u8 = undefined;
    var selected: ?[]const u8 = null;
    const reply = try pool.exchange(io, query_bytes, &buf, &selected);
    try testing.expectEqualSlices(u8, response_bytes, reply);

    try testing.expectEqual(@as(u32, 1), entries[0].health.consecutive_failures);
    try testing.expectEqual(@as(u64, 1), entries[0].health.total_failures);
    try testing.expectEqualStrings("Timeout", entries[0].health.lastError());
    try testing.expectEqual(@as(u64, 1), entries[1].health.total_successes);
}

test "an entry in backoff is skipped while another is available" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var bad: Fake = .{ .behavior = .{ .fail = error.ConnectFailed } };
    var good: Fake = .{ .behavior = .{ .reply = response_bytes } };
    var entries = [_]Entry{
        testEntry("https://bad.example/dns-query", &bad, 10),
        testEntry("https://good.example/dns-query", &good, 20),
    };
    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);

    var buf: [512]u8 = undefined;
    // Two failures reach `failure_threshold` and open a backoff window.
    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.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.load(.acquire));
    try testing.expectEqual(@as(usize, 3), good.calls.load(.acquire));
}

test "every entry in backoff is still probed" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var first: Fake = .{ .behavior = .{ .fail = error.ConnectFailed } };
    var second: Fake = .{ .behavior = .{ .fail = error.BadResponse } };
    var entries = [_]Entry{
        testEntry("https://first.example/dns-query", &first, 10),
        testEntry("https://second.example/dns-query", &second, 20),
    };
    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);

    var buf: [512]u8 = undefined;
    var selected: ?[]const u8 = null;
    try testing.expectError(error.BadResponse, pool.exchange(io, query_bytes, &buf, &selected));
    try testing.expectError(error.BadResponse, pool.exchange(io, query_bytes, &buf, &selected));
    try testing.expect(entries[0].health.backoff_until != null);
    try testing.expect(entries[1].health.backoff_until != null);

    // 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.load(.acquire));
    try testing.expectEqual(@as(usize, 3), second.calls.load(.acquire));
}

test "a local resource error short-circuits and records nothing" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var broke: Fake = .{ .behavior = .{ .fail = error.OutOfMemory } };
    var good: Fake = .{ .behavior = .{ .reply = response_bytes } };
    var entries = [_]Entry{
        testEntry("https://broke.example/dns-query", &broke, 10),
        testEntry("https://good.example/dns-query", &good, 20),
    };
    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);

    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.load(.acquire));
    try testing.expectEqual(@as(u64, 0), entries[0].health.total_failures);
    try testing.expectEqual(@as(u32, 0), entries[0].health.consecutive_failures);
}

test "a cancellation short-circuits and records nothing" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var canceled: Fake = .{ .behavior = .{ .fail = error.Canceled } };
    var good: Fake = .{ .behavior = .{ .reply = response_bytes } };
    var entries = [_]Entry{
        testEntry("https://canceled.example/dns-query", &canceled, 10),
        testEntry("https://good.example/dns-query", &good, 20),
    };
    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);

    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.load(.acquire));
    try testing.expectEqual(@as(u64, 0), entries[0].health.total_failures);
}

test "an attempt that outruns the budget is a recorded Timeout" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var slow: Fake = .{ .behavior = .{ .slow = .{
        .duration = .{ .raw = .fromSeconds(30), .clock = .awake },
        .reply = response_bytes,
    } } };
    var good: Fake = .{ .behavior = .{ .reply = response_bytes } };
    var entries = [_]Entry{
        testEntry("https://slow.example/dns-query", &slow, 10),
        testEntry("https://good.example/dns-query", &good, 20),
    };
    // A tight attempt budget under a loose total one, so the attempt deadline
    // is the only one that can fire and the failover after it has room to run.
    var pool: Pool = .init(&entries, test_cfg, .{
        .attempt = .{ .raw = .fromMilliseconds(20), .clock = .awake },
        .total = .{ .raw = .fromSeconds(30), .clock = .awake },
    }, 1);

    var buf: [512]u8 = undefined;
    var selected: ?[]const u8 = null;
    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.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);
}

test "two stalling upstreams cost the total budget, not one budget each" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    // Both entries stall past their own attempt budget, so without the outer
    // race the exchange would take two attempt budgets and then some. The
    // total is set below two attempts, which is what makes the assertion mean
    // something: only the outer deadline can end this call in time.
    var first: Fake = .{ .behavior = .{ .slow = .{
        .duration = .{ .raw = .fromSeconds(30), .clock = .awake },
        .reply = response_bytes,
    } } };
    var second: Fake = .{ .behavior = .{ .slow = .{
        .duration = .{ .raw = .fromSeconds(30), .clock = .awake },
        .reply = response_bytes,
    } } };
    var entries = [_]Entry{
        testEntry("https://first.example/dns-query", &first, 10),
        testEntry("https://second.example/dns-query", &second, 20),
    };
    var pool: Pool = .init(&entries, test_cfg, .{
        .attempt = .{ .raw = .fromMilliseconds(200), .clock = .awake },
        .total = .{ .raw = .fromMilliseconds(60), .clock = .awake },
    }, 1);

    var buf: [512]u8 = undefined;
    const started = std.Io.Clock.awake.now(io);
    var selected: ?[]const u8 = null;
    try testing.expectError(error.Timeout, pool.exchange(io, query_bytes, &buf, &selected));
    const elapsed_ns = std.Io.Clock.awake.now(io).nanoseconds - started.nanoseconds;

    // Under one attempt budget, so the outer deadline is provably what fired.
    // The bound is generous against a loaded CI box; the failure it catches is
    // a whole extra attempt, not a scheduling hiccup.
    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.load(.acquire));
}

test "every entry disabled yields ConnectFailed without waiting out the total budget" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var one: Fake = .{ .behavior = .{ .reply = response_bytes } };
    var two: Fake = .{ .behavior = .{ .reply = response_bytes } };
    var entries = [_]Entry{
        testEntry("https://one.example/dns-query", &one, 10),
        testEntry("https://two.example/dns-query", &two, 20),
    };
    for (&entries) |*entry| entry.enabled = false;
    // A total budget long enough that waiting it out would be unmistakable.
    // The outer expiry returns `error.Timeout` unconditionally, so this pins
    // that the fast-failure path still wins its own race.
    var pool: Pool = .init(&entries, test_cfg, .{
        .attempt = .{ .raw = .fromSeconds(10), .clock = .awake },
        .total = .{ .raw = .fromSeconds(30), .clock = .awake },
    }, 1);

    var buf: [512]u8 = undefined;
    const started = std.Io.Clock.awake.now(io);
    var selected: ?[]const u8 = null;
    try testing.expectError(error.ConnectFailed, pool.exchange(io, query_bytes, &buf, &selected));
    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.load(.acquire));
    try testing.expectEqual(@as(usize, 0), two.calls.load(.acquire));
}

test "snapshot reports the counters in pool order" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var bad: Fake = .{ .behavior = .{ .fail = error.Timeout } };
    var good: Fake = .{ .behavior = .{ .reply = response_bytes } };
    var entries = [_]Entry{
        testEntry("https://bad.example/dns-query", &bad, 10),
        testEntry("https://good.example/dns-query", &good, 20),
    };
    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);

    var buf: [512]u8 = undefined;
    var selected: ?[]const u8 = null;
    _ = try pool.exchange(io, query_bytes, &buf, &selected);
    _ = try pool.exchange(io, query_bytes, &buf, &selected);

    var out: [4]Snapshot = undefined;
    const written = try pool.snapshot(io, &out);
    try testing.expectEqual(@as(usize, 2), written);

    try testing.expectEqualStrings("https://bad.example/dns-query", out[0].url);
    try testing.expect(out[0].enabled);
    try testing.expect(!out[0].available);
    try testing.expectEqual(@as(u32, 2), out[0].consecutive_failures);
    try testing.expectEqual(@as(u64, 2), out[0].total_failures);
    try testing.expectEqual(@as(u64, 0), out[0].total_successes);
    try testing.expectEqual(@as(f32, 0.0), out[0].success_rate);
    try testing.expectEqualStrings("Timeout", out[0].last_error);
    try testing.expect(out[0].last_error_at != null);
    try testing.expect(out[0].backoff_until != null);

    try testing.expectEqualStrings("https://good.example/dns-query", out[1].url);
    try testing.expect(out[1].available);
    try testing.expectEqual(@as(u64, 2), out[1].total_successes);
    try testing.expectEqual(@as(f32, 1.0), out[1].success_rate);
    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));
}

/// One concurrent call's whole result: how much of its buffer the reply filled,
/// and which resolver the pool said answered it.
///
/// `Io.concurrent` stores the return value in the future, so the reply slice is
/// reduced to its length here and the bytes are read back out of the caller's
/// buffer. `selected` survives the trip because it borrows an `Endpoint.url`,
/// which outlives the pool.
const Attributed = struct {
    reply_len: usize,
    selected: ?[]const u8,

    /// What a teardown `await` discards into once the assertions above it have
    /// already taken the value.
    const discarded: Attributed = .{ .reply_len = 0, .selected = null };
};

/// Each call keeps its own `selected` out-value rather than dropping it: the
/// pointer is written per call, on the stack of the task that made it, and a
/// pool that hung it off `*Pool` instead would hand one call's identity to
/// another. Nothing but a per-call capture can see that.
fn exchangeAttributed(pool: *Pool, io: std.Io, buf: []u8) transport.ExchangeError!Attributed {
    var selected: ?[]const u8 = null;
    const reply = try pool.exchange(io, query_bytes, buf, &selected);
    return .{ .reply_len = reply.len, .selected = selected };
}

/// 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();

    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);

    var buf_a: [512]u8 = undefined;
    var buf_b: [512]u8 = undefined;

    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;
    // 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;

    _ = try first.await(io);
    _ = try second.await(io);

    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
/// produces so a call's reply proves which entry served it independently of
/// what the pool reported.
const divergent_first_url = "https://first.example/dns-query";
const divergent_second_url = "https://second.example/dns-query";

fn expectAnsweredByReporter(result: Attributed, buf: []const u8) !void {
    const url = result.selected orelse return error.TestExpectedSelectedResolver;
    const reply = if (std.mem.eql(u8, url, divergent_first_url))
        alt_response_bytes
    else if (std.mem.eql(u8, url, divergent_second_url))
        response_bytes
    else
        return error.TestUnexpectedResolver;
    try testing.expectEqualSlices(u8, reply, buf[0..result.reply_len]);
}

test "overlapping exchanges each report the entry that answered that call" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    // The first entry fails the task that reaches it first, slowly enough that
    // the second task is queued behind it, then answers that second task. One
    // failure is under `test_cfg`'s threshold of two, so no backoff steers the
    // waiting task away and the two calls end on different entries.
    var first_entry: Fake = .{
        .behavior = .{ .slow_fail = .{
            .duration = .{ .raw = .fromMilliseconds(50), .clock = .awake },
            .err = error.ConnectFailed,
        } },
        .then = .{ .reply = alt_response_bytes },
    };
    // Slow too, so the failed-over call is still in flight while the other call
    // is being answered — a shared identity would be overwritten under it.
    var second_entry: Fake = .{ .behavior = .{ .slow = .{
        .duration = .{ .raw = .fromMilliseconds(50), .clock = .awake },
        .reply = response_bytes,
    } } };
    var entries = [_]Entry{
        testEntry(divergent_first_url, &first_entry, 10),
        testEntry(divergent_second_url, &second_entry, 20),
    };
    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);

    var buf_a: [512]u8 = undefined;
    var buf_b: [512]u8 = undefined;

    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);

    // Which task lands where depends on the order they take the first entry's
    // lock, so the claim is over the pair: one identity each, and each one
    // matching the bytes that call received.
    try testing.expect(!std.mem.eql(u8, result_a.selected.?, result_b.selected.?));
    try expectAnsweredByReporter(result_a, &buf_a);
    try expectAnsweredByReporter(result_b, &buf_b);

    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);
    try testing.expectEqual(@as(u64, 1), entries[1].health.total_successes);
}

test "an entry that enters backoff while a task waits on it is not attempted" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    // One failure is enough to open the backoff window, so the first task's
    // failure makes the entry unavailable to the second task, which by then is
    // already past its own availability check and waiting on `busy`.
    const cfg: health.Config = .{
        .failure_threshold = 1,
        .base_backoff_ms = 60_000,
        .max_backoff_ms = 60_000,
    };

    var slow_bad: Fake = .{ .behavior = .{ .slow_fail = .{
        .duration = .{ .raw = .fromMilliseconds(50), .clock = .awake },
        .err = error.ConnectFailed,
    } } };
    var good: Fake = .{ .behavior = .{ .reply = response_bytes } };
    var entries = [_]Entry{
        testEntry("https://slow-bad.example/dns-query", &slow_bad, 10),
        testEntry("https://good.example/dns-query", &good, 20),
    };
    var pool: Pool = .init(&entries, cfg, test_timeouts, 1);

    var buf_a: [512]u8 = undefined;
    var buf_b: [512]u8 = undefined;

    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);

    // Both tasks fail over to the healthy entry and get an answer, and both
    // report it: the identity is the entry that answered, not the one that
    // failed on the way there.
    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://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.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.load(.acquire));
    try testing.expectEqual(@as(u64, 1), entries[0].health.total_failures);
    try testing.expect(entries[0].health.backoff_until != null);
}

test "a successful exchange with nothing open costs the store no statement" {
    if (@FieldType(events.Store, "statements") != u64) return error.SkipZigTest;

    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var good: Fake = .{ .behavior = .{ .reply = response_bytes } };
    var entries = [_]Entry{testEntry("https://good.example/dns-query", &good, 10)};
    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);

    var fx: events_fixture.Fixture = .{};
    try fx.init(io, 1000);
    defer fx.deinit();
    pool.diagnostics = &fx.store;

    var buf: [512]u8 = undefined;
    const before = fx.store.statements;
    var selected: ?[]const u8 = null;
    for (0..20) |_| _ = try pool.exchange(io, query_bytes, &buf, &selected);
    try testing.expectEqual(before, fx.store.statements);
    try testing.expectEqual(@as(i64, 0), try fx.count("SELECT count(*) FROM operational_events"));
}

test "a failing then recovering upstream leaves exactly one resolved episode" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var flaky: Fake = .{ .behavior = .{ .fail = error.Timeout } };
    var standby: Fake = .{ .behavior = .{ .reply = response_bytes } };
    var entries = [_]Entry{
        testEntry("https://flaky.example/dns-query", &flaky, 10),
        testEntry("https://standby.example/dns-query", &standby, 20),
    };
    var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);

    var fx: events_fixture.Fixture = .{};
    try fx.init(io, 1000);
    defer fx.deinit();
    pool.diagnostics = &fx.store;

    var buf: [512]u8 = undefined;
    var selected: ?[]const u8 = null;
    _ = try pool.exchange(io, query_bytes, &buf, &selected);
    // Backoff would park the failing entry, so the second failure is driven
    // through `recordFailure` itself rather than through another exchange.
    pool.recordFailure(io, &entries[0], std.Io.Clock.awake.now(io), error.ConnectFailed);

    try testing.expectEqual(@as(i64, 1), try fx.count("SELECT count(*) FROM operational_events"));
    try testing.expectEqual(@as(i64, 2), try fx.count("SELECT occurrences FROM operational_events"));
    try testing.expectEqualStrings("upstream.exchange", try fx.text("SELECT code FROM operational_events"));
    try testing.expectEqualStrings(
        "https://flaky.example/dns-query",
        try fx.text("SELECT subject_key FROM operational_events"),
    );

    flaky.behavior = .{ .reply = response_bytes };
    pool.recordSuccess(io, &entries[0], std.Io.Clock.awake.now(io));

    try testing.expectEqual(@as(i64, 1), try fx.count("SELECT count(*) FROM operational_events"));
    try testing.expectEqual(
        @as(i64, 0),
        try fx.count("SELECT count(*) FROM operational_events WHERE resolved_at IS NULL"),
    );
    try testing.expectEqual(@as(i64, 2), try fx.count("SELECT occurrences FROM operational_events"));
}