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.
//!
//! One deadline, computed once. `Pool.exchange` turns `timeouts.total` into an
//! absolute instant and the loop spends it: admission waits against it, and
//! every attempt is raced against `min(now + timeouts.attempt, deadline)`. No
//! outer race wraps the loop. Two timers aimed at the same instant race each
//! other, and the outer one wins by cancelling the loop from outside, which
//! throws away the loop's own classification of what just happened.
//!
//! What the deadline expiring means depends on which side of it ran. An attempt
//! that got its full `timeouts.attempt` and did not answer is evidence about
//! that peer, recorded as a failure. An attempt the deadline cut short, and a
//! wait for a slot that never ended, are evidence about this process's budget:
//! they are `error.BudgetExhausted`, counted by `budgetExhaustedTotal` and
//! recorded against no endpoint. That distinction is why the pool needs
//! `transport.raceUntilTagged` rather than a bare timeout.
//!
//! 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.admission` 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.
//!
//! Priority orders the entries that can be admitted *now*. A saturated entry is
//! stepped over while a lower-priority one has a free slot, because blocking on
//! the higher-priority entry spends the caller's whole budget in a queue while
//! an idle standby watches. Only when no eligible entry can admit immediately
//! does a task block, and then on the highest-priority eligible entry alone.
//!
//! Because a saturated entry can still make a task 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;

/// Admission control for one entry: one permit per slot, a try-acquire, and an
/// acquire bounded by the caller's deadline. `std.Io.Semaphore` offers neither
/// in 0.16, and `std.Io.Condition` has no timed wait, so this mirrors the
/// standard semaphore's mutex/count/condition protocol with the two operations
/// the failover loop needs added. The stdlib is not patched and nothing spins.
///
/// The mutex owns the permit count and the decision to take one, and the caller
/// takes its permit only *after* the timed race has resolved. That ordering is
/// the whole design: racing a bare `Semaphore.wait` against an expiry can
/// decrement the count in the same instant the expiry wins, and the discarded
/// success never reaches the caller's release, so the permit is gone for good. A
/// wait discarded here loses a notification instead, which the exiting waiter
/// repairs by re-signalling.
///
/// Bound worth knowing: `acquireUntil` blocks on one entry. A permit freeing on
/// some *other* entry does not wake this waiter, so the pool's no-head-of-line
/// guarantee covers only the capacity its `tryAcquire` sweep observed before it
/// blocked. Waking on any of several entries needs multi-wait machinery the pool
/// does not have.
pub const Admission = struct {
    mutex: std.Io.Mutex = .init,
    cond: std.Io.Condition = .init,
    /// One per slot at build time; `Pool.init` asserts it.
    permits: usize = 0,

    /// Which side of `acquireUntil` ended the wait. `.expired` holds no permit.
    pub const Outcome = enum { acquired, expired };

    /// Takes a permit if one is free this instant; never blocks.
    ///
    /// A contended mutex reads as "no permit now" rather than as a reason to
    /// wait. The failover loop calls this on every candidate in turn, so
    /// blocking here for a descheduled permit holder would stall the whole
    /// sweep and spend the caller's budget outside `acquireUntil`, which is the
    /// only step allowed to consume it.
    pub fn tryAcquire(self: *Admission, io: std.Io) bool {
        if (!self.mutex.tryLock()) return false;
        defer self.mutex.unlock(io);
        if (self.permits == 0) return false;
        self.permits -= 1;
        return true;
    }

    /// Returns a permit. Uncancelable, so a torn-down attempt still gives its
    /// slot back.
    pub fn release(self: *Admission, io: std.Io) void {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);
        self.permits += 1;
        self.cond.signal(io);
    }

    /// Waits for a permit until `deadline`. Returns `.acquired` holding one,
    /// `.expired` holding none, or an error holding none — `error.Canceled`
    /// when the whole task is torn down, `error.SystemResources` when the race
    /// cannot start.
    pub fn acquireUntil(
        self: *Admission,
        io: std.Io,
        deadline: std.Io.Clock.Timestamp,
    ) transport.ExchangeError!Outcome {
        while (true) {
            // Every take is non-blocking, including the retries. A take that
            // waits on the mutex spends time no deadline bounds: before the
            // first race it carries the call past the deadline outright, and
            // after a wake it can return `.acquired` late. A miss instead
            // re-enters the race below, which is aimed at the same absolute
            // instant every time, so the cycle ends at the deadline with
            // `.expired` — the correct answer, not a spin.
            if (self.tryAcquire(io)) return .acquired;
            var race: transport.RaceOutcome = .completed;
            _ = transport.raceUntilTagged(io, deadline, &race, waitForPermit, .{ self, io }) catch |err| {
                // This task may be leaving with a notification meant for a peer:
                // `Condition`'s cancel path consumes a pending signal before it
                // reports the cancellation, and a wait whose success the race
                // discarded consumed one outright. Hand it back before exiting.
                self.resignal(io);
                if (err == error.Timeout and race == .expired) return .expired;
                return err;
            };
            // The wait saw a permit but did not take one, so another task may
            // have taken it first. The loop re-checks rather than assuming, and
            // does not spin: `waitForPermit` blocks whenever the count is zero.
        }
    }

    /// Returns once the count is non-zero, without touching it.
    fn waitForPermit(self: *Admission, io: std.Io) transport.ExchangeError!void {
        self.mutex.lock(io) catch return error.Canceled;
        defer self.mutex.unlock(io);
        while (self.permits == 0) self.cond.wait(io, &self.mutex) catch return error.Canceled;
    }

    /// Only when a permit is actually there to hand over: an unconditional
    /// signal would wake a peer that would find nothing.
    fn resignal(self: *Admission, io: std.Io) void {
        self.mutex.lockUncancelable(io);
        defer self.mutex.unlock(io);
        if (self.permits > 0) self.cond.signal(io);
    }
};

/// 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.
    admission: Admission,
    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),
    /// Blocked admissions, not a queue length: a task counts itself queued
    /// exactly when it has run out of immediately admissible entries and blocks
    /// on this one, whichever way that block ends. A `tryAcquire` — hit or miss
    /// — is not a queue, so it never counts.
    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 every exit of a blocking wait for a permit. A waiter
    /// that burned its budget in the queue is the field failure this records, so
    /// an expiry and a cancellation count 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;
    }
};

/// One exchange, raced until `expiry_at` and tagged with which side won.
///
/// `race` distinguishes an expiry from the leaf returning `error.Timeout` of
/// its own, which are the same error and opposite evidence: the first is about
/// the clock the caller set, the second is about the peer.
///
/// 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 attemptUntil(
    io: std.Io,
    entry_client: transport.Client,
    query: []const u8,
    response_buf: []u8,
    expiry_at: std.Io.Clock.Timestamp,
    race: *transport.RaceOutcome,
) transport.ExchangeError![]u8 {
    var leaf_selected: ?[]const u8 = null;
    return transport.raceUntilTagged(io, expiry_at, race, transport.Client.exchange, .{
        entry_client, io, query, response_buf, &leaf_selected,
    });
}

/// 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,
    /// Exchanges that ran out of budget, counted once each by `exchange`. A
    /// pool-wide number rather than a per-entry one on purpose: budget
    /// exhaustion is this process failing to give any peer its interval, so
    /// attributing it to an endpoint would be the very lie this milestone
    /// removes. Read through `budgetExhaustedTotal`, since `Snapshot` is
    /// per-entry and has nowhere to put it.
    budget_exhausted_total: std.atomic.Value(u64) = .init(0),

    /// The failover loop tracks which entries it has already spent in a bitset
    /// on its stack, so the number of upstreams one pool may hold is bounded.
    /// A household config has two; `web/metrics.zig` already renders at most
    /// this many.
    pub const max_entries = 64;

    pub fn init(
        entries: []Entry,
        cfg: health.Config,
        timeouts: Timeouts,
        seed: u64,
    ) Pool {
        std.debug.assert(entries.len > 0);
        std.debug.assert(entries.len <= max_entries);
        for (entries) |*entry| {
            std.debug.assert(entry.slots.len >= 1);
            std.debug.assert(entry.admission.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.
    ///
    /// `selected` follows a stricter rule here than the `transport.Client`
    /// contract requires: the pool names the last endpoint whose outcome it
    /// actually recorded, written after the attempt rather than before it. An
    /// exchange that ran out of budget while an attempt was still in flight
    /// therefore leaves `selected` at the previous endpoint, or at `null` if it
    /// never got that far, because "the endpoint that was in the way when the
    /// clock ran out" is not evidence about that endpoint and the query log must
    /// not print it as though it were.
    ///
    /// The whole call is bounded by one deadline derived from `timeouts.total`
    /// here and spent by the loop below: no outer race, so an expiry is always
    /// classified by the step that owned it.
    pub fn exchange(
        self: *Pool,
        io: std.Io,
        query: []const u8,
        response_buf: []u8,
        selected: *?[]const u8,
    ) transport.ExchangeError![]u8 {
        const deadline: std.Io.Clock.Timestamp = .fromNow(io, self.timeouts.total);
        return self.failover(io, query, response_buf, selected, deadline) catch |err| {
            // Once per exhausted exchange, wherever the exhaustion was
            // detected — the loop's own waits and attempts, or a leaf that
            // reported one of its own.
            if (err == error.BudgetExhausted) {
                _ = self.budget_exhausted_total.fetchAdd(1, .monotonic);
            }
            return err;
        };
    }

    /// Exchanges that ended in `error.BudgetExhausted`, pool-wide.
    pub fn budgetExhaustedTotal(self: *const Pool) u64 {
        return self.budget_exhausted_total.load(.acquire);
    }

    /// The two-pass failover loop.
    ///
    /// Pass one sweeps the entries health says are available, in priority
    /// order, admitting through `tryAcquire` so a saturated entry is stepped
    /// over rather than waited on. It blocks only when the sweep found no
    /// capacity anywhere, and then on the highest-priority candidate alone.
    /// Pass two runs when pass one attempted nothing and probes every enabled
    /// entry regardless of backoff, which is how backoff recovers; it keeps
    /// today's one-at-a-time blocking, since probing a backed-off entry is
    /// already a last resort.
    fn failover(
        self: *Pool,
        io: std.Io,
        query: []const u8,
        response_buf: []u8,
        selected: *?[]const u8,
        deadline: std.Io.Clock.Timestamp,
    ) transport.ExchangeError![]u8 {
        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) {
            // Entries this call is finished with: attempted already, or found
            // unavailable by the post-admission recheck. Cleared per pass,
            // because pass two only runs when pass one attempted nothing and
            // must reconsider everything it skipped.
            var spent: std.StaticBitSet(max_entries) = .initEmpty();

            while (true) {
                // An admission that never came ends the exchange: nothing ran,
                // so nothing is attributable and `selected` stays as it is.
                const index = try self.admit(io, spent, pass, now, deadline) orelse break;
                spent.set(index);
                const entry = &self.entries[index];

                // 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.admission.release(io);
                }

                // Health was read before the permit was taken, and the attempts
                // this one waited behind may have failed the entry into backoff
                // meanwhile. Pass one must not touch an entry that is
                // unavailable now, so re-read 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.
                if (pass == 0) {
                    const recheck = std.Io.Clock.awake.now(io);
                    if (!self.entryAvailable(io, entry, recheck)) continue;
                }
                // Nothing is left to observe with. Admission and the recheck
                // above can each land after the deadline has already passed,
                // and starting an attempt against an expired timestamp would
                // race the leaf against a timer that has already fired: an
                // instantly-completing leaf would win nondeterministically and
                // manufacture a success, or evidence about a peer, out of a
                // budget that was already gone. The defer above returns the
                // permit; no endpoint is named, because none was observed.
                const started_at: std.Io.Clock.Timestamp = .now(io, .awake);
                if (deadline.compare(.lte, started_at)) return error.BudgetExhausted;
                attempted = true;

                // The full attempt budget unless the deadline lands first, in
                // which case the peer is being given less than the configured
                // observation interval and an expiry says nothing about it.
                const full_expiry = started_at.addDuration(self.timeouts.attempt);
                const truncated = deadline.compare(.lt, full_expiry);
                const expiry_at = if (truncated) deadline else full_expiry;

                var race: transport.RaceOutcome = .completed;
                const result = attemptUntil(io, slot.client, query, response_buf, expiry_at, &race);
                const completed_at = std.Io.Clock.awake.now(io);

                const response = result catch |err| {
                    // A censored observation: the pool cut the attempt short,
                    // so the expiry is evidence about this budget and about no
                    // peer. A leaf's own `error.Timeout` is not this case — the
                    // race tag is what tells the two apart.
                    if (truncated and race == .expired) return error.BudgetExhausted;
                    switch (transport.group(err)) {
                        .peer_fault => {
                            log.debug("{f}", .{AttemptFailure{ .endpoint = entry.endpoint, .err = err }});
                            self.recordFailure(io, entry, completed_at, err);
                            selected.* = entry.endpoint.url;
                            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,
                        // The caller's budget, not this endpoint's conduct.
                        .budget_exhausted => return err,
                    }
                };

                self.recordSuccess(io, entry, completed_at);
                selected.* = entry.endpoint.url;
                return response;
            }
            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;
    }

    /// Takes a permit on the next entry to attempt, and returns its index.
    /// `null` means this pass has no candidate left.
    ///
    /// Pass one sweeps: every eligible candidate is offered a non-blocking
    /// `tryAcquire` in priority order, including ones an earlier sweep step
    /// skipped, since a slot may have freed since. That sweep is what keeps a
    /// saturated entry from blocking the query. Blocking is the fallback, not
    /// the first move, and it waits on the highest-priority candidate only.
    ///
    /// Pass two probes instead: it takes the next enabled entry in priority
    /// order and blocks for its permit, one entry at a time. Probing a
    /// backed-off entry is already a last resort, so the pass keeps the order
    /// the operator configured rather than letting a saturated higher-priority
    /// candidate be skipped over — the deadline, not a lock miss, is what ends
    /// it.
    fn admit(
        self: *Pool,
        io: std.Io,
        spent: std.StaticBitSet(max_entries),
        pass: u8,
        now: std.Io.Timestamp,
        deadline: std.Io.Clock.Timestamp,
    ) transport.ExchangeError!?usize {
        if (pass != 0) {
            for (0..self.entries.len) |index| {
                if (spent.isSet(index)) continue;
                if (!self.entries[index].enabled) continue;
                return try self.admitBlocking(io, index, deadline);
            }
            return null;
        }

        var first_eligible: ?usize = null;
        for (self.entries, 0..) |*entry, index| {
            if (spent.isSet(index)) continue;
            if (!entry.enabled) continue;
            if (!self.entryAvailable(io, entry, now)) continue;
            if (entry.admission.tryAcquire(io)) return index;
            if (first_eligible == null) first_eligible = index;
        }

        const index = first_eligible orelse return null;
        return try self.admitBlocking(io, index, deadline);
    }

    /// Blocks for one entry's permit until `deadline`, counting the wait.
    fn admitBlocking(
        self: *Pool,
        io: std.Io,
        index: usize,
        deadline: std.Io.Clock.Timestamp,
    ) transport.ExchangeError!usize {
        const entry = &self.entries[index];
        const started = std.Io.Clock.awake.now(io);
        defer entry.recordQueued(io, started);
        switch (try entry.admission.acquireUntil(io, deadline)) {
            .acquired => return index,
            .expired => return error.BudgetExhausted,
        }
    }

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

    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 },
        /// Wait for the gate, then reply. A test that needs two calls provably
        /// in flight together cannot get that from a duration: a loaded
        /// scheduler can start the second call after the first one's sleep
        /// expired, and no poll can recover an overlap that already passed.
        /// The gate makes the test, not the clock, decide when a call returns.
        hold: struct { gate: *std.Io.Semaphore, reply: []const u8 },
        /// Wait for the gate, then fail. The failing half of `hold`, with the
        /// same constraint: the test releases the call once it has observed
        /// whatever had to be true while the call was still in flight.
        hold_fail: struct { gate: *std.Io.Semaphore, 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));
        _ = self.calls.fetchAdd(1, .acq_rel);
        // The behaviour is taken before `in_flight` rises: a test that waits
        // on the count to know a call is inside this fake also needs to know
        // which behaviour that call drew, or a second call could still take
        // the first behaviour from under it.
        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;
        };
        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);
        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;
            },
            .hold => |held| {
                try held.gate.wait(io);
                return copy(held.reply, response_buf);
            },
            .hold_fail => |held| {
                try held.gate.wait(io);
                return held.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,
        .admission = .{ .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 "a canceled admission waiter holds nothing and its peer still wakes" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var admission: Admission = .{ .permits = 0 };
    const far: std.Io.Clock.Timestamp = .fromNow(io, .{ .raw = .fromSeconds(30), .clock = .awake });

    var canceled = io.concurrent(Admission.acquireUntil, .{ &admission, io, far }) catch |err| switch (err) {
        error.ConcurrencyUnavailable => return error.SkipZigTest,
    };
    var peer = io.concurrent(Admission.acquireUntil, .{ &admission, io, far }) catch |err| switch (err) {
        error.ConcurrencyUnavailable => return error.SkipZigTest,
    };
    defer _ = peer.await(io) catch Admission.Outcome.expired;

    // Long enough that both tasks are provably inside the wait: neither can
    // reach a permit, because none exists yet.
    const settle: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(20), .clock = .awake };
    try settle.sleep(io);
    try testing.expectError(error.Canceled, canceled.cancel(io));

    // One permit, one signal. The canceled task may have absorbed that signal
    // on its way out; if it did not hand it back, the peer would wait here
    // forever and this test would hang.
    admission.release(io);
    try testing.expectEqual(Admission.Outcome.acquired, try peer.await(io));
    try testing.expectEqual(@as(usize, 0), admission.permits);
}

fn holdAdmissionMutex(admission: *Admission, io: std.Io, hold: std.Io.Clock.Duration) void {
    admission.mutex.lockUncancelable(io);
    defer admission.mutex.unlock(io);
    hold.sleep(io) catch {};
}

test "a contended admission mutex does not carry acquireUntil past its deadline" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    // One permit, but the mutex guarding it is held far past the caller's
    // deadline. Every take must therefore run inside the deadline race: a
    // blocking lock before the race starts would return `.acquired` only once
    // the holder let go, long after the budget was gone.
    var admission: Admission = .{ .permits = 1 };
    const hold: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(200), .clock = .awake };
    var holder = io.concurrent(holdAdmissionMutex, .{ &admission, io, hold }) catch |err| switch (err) {
        error.ConcurrencyUnavailable => return error.SkipZigTest,
    };
    defer holder.await(io);

    const settle: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(20), .clock = .awake };
    try settle.sleep(io);

    const deadline: std.Io.Clock.Timestamp = .fromNow(io, settle);
    // `.expired` is the whole assertion: the permit is free, so a take that
    // waited on the mutex outside the race would report `.acquired` instead,
    // hundreds of milliseconds after the deadline. The exit path itself still
    // waits for the mutex, uncancelably, to re-signal the condition.
    try testing.expectEqual(Admission.Outcome.expired, try admission.acquireUntil(io, deadline));
    try testing.expectEqual(@as(usize, 1), admission.permits);
}

test "an admission expiry racing a release leaves the permit count exact" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    // The tie the design exists for: the permit is taken by the caller under
    // the mutex only after the race resolves, so a discarded wait can lose a
    // notification but never a permit. Repeated because the interleaving is a
    // race — a leak would show as a count that drifts below one.
    const tick: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(2), .clock = .awake };
    for (0..50) |_| {
        var admission: Admission = .{ .permits = 0 };
        const deadline: std.Io.Clock.Timestamp = .fromNow(io, tick);
        var waiter = io.concurrent(Admission.acquireUntil, .{ &admission, io, deadline }) catch |err| switch (err) {
            error.ConcurrencyUnavailable => return error.SkipZigTest,
        };
        try tick.sleep(io);
        admission.release(io);

        switch (try waiter.await(io)) {
            .acquired => admission.release(io),
            .expired => {},
        }
        try testing.expectEqual(@as(usize, 1), admission.permits);
    }
}

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 truncated attempt blames the budget, not the endpoint it was in" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    // The total budget is shorter than one attempt, so the first attempt is
    // truncated: the endpoint never got the observation interval it was
    // configured to get, and an expiry says nothing about it.
    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.BudgetExhausted,
        pool.exchange(io, query_bytes, &buf, &selected),
    );
    // Nothing is attributable: no outcome was recorded, so no endpoint is
    // named and no health moved.
    try testing.expect(selected == null);
    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.last_error_at);
    try testing.expectEqual(@as(usize, 0), untouched.calls.load(.acquire));
    try testing.expectEqual(@as(u64, 1), pool.budgetExhaustedTotal());
}

test "a truncated attempt that fails on its own still blames the endpoint" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    // Same truncated budget, but the attempt completes with a peer fault of
    // its own before the deadline. A completed failure is real evidence
    // whatever budget it ran with, so health moves and `selected` names it.
    var refusing: Fake = .{ .behavior = .{ .fail = error.ConnectFailed } };
    var entries = [_]Entry{testEntry("https://refusing.example/dns-query", &refusing, 10)};
    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.ConnectFailed,
        pool.exchange(io, query_bytes, &buf, &selected),
    );
    try testing.expectEqualStrings("https://refusing.example/dns-query", selected.?);
    try testing.expectEqual(@as(u64, 1), entries[0].health.total_failures);
    try testing.expectEqualStrings("ConnectFailed", entries[0].health.lastError());
    try testing.expectEqual(@as(u64, 0), pool.budgetExhaustedTotal());
}

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);
    // A torn-down task is not a budget that ran out, and it names no endpoint.
    try testing.expectEqual(@as(u64, 0), pool.budgetExhaustedTotal());
    try testing.expect(selected == null);
}

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 "a stalled primary fails over to a fast standby inside the total budget" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    // The shipped shape of the defaults: attempt is half of total, so the
    // primary gets one whole attempt and the standby still has room to answer
    // inside the same deadline.
    var stalled: Fake = .{ .behavior = .{ .slow = .{
        .duration = .{ .raw = .fromSeconds(30), .clock = .awake },
        .reply = response_bytes,
    } } };
    var standby: Fake = .{ .behavior = .{ .reply = alt_response_bytes } };
    var entries = [_]Entry{
        testEntry("https://stalled.example/dns-query", &stalled, 10),
        testEntry("https://standby.example/dns-query", &standby, 20),
    };
    var pool: Pool = .init(&entries, test_cfg, .{
        .attempt = .{ .raw = .fromMilliseconds(150), .clock = .awake },
        .total = .{ .raw = .fromMilliseconds(300), .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, alt_response_bytes, reply);
    try testing.expectEqualStrings("https://standby.example/dns-query", selected.?);
    // The primary got its full configured interval and did not answer, which is
    // evidence about the primary — so it is recorded, unlike a truncated one.
    try testing.expectEqual(@as(u32, 1), entries[0].health.consecutive_failures);
    try testing.expectEqualStrings("Timeout", entries[0].health.lastError());
    try testing.expectEqual(@as(u64, 0), pool.budgetExhaustedTotal());
}

test "pass two probes every backed-off entry under the one deadline" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const cfg: health.Config = .{
        .failure_threshold = 1,
        .base_backoff_ms = 60_000,
        .max_backoff_ms = 60_000,
    };
    var first: Fake = .{ .behavior = .{ .fail = error.ConnectFailed } };
    var second: Fake = .{ .behavior = .{ .fail = error.ConnectFailed } };
    var entries = [_]Entry{
        testEntry("https://first.example/dns-query", &first, 10),
        testEntry("https://second.example/dns-query", &second, 20),
    };

    // One exchange under generous budgets puts both entries into backoff.
    var seeding_pool: Pool = .init(&entries, cfg, test_timeouts, 1);
    var buf: [512]u8 = undefined;
    var selected: ?[]const u8 = null;
    try testing.expectError(
        error.ConnectFailed,
        seeding_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);

    // Now both stall. Pass one has no candidate at all, so pass two probes
    // them in order — sharing the one deadline rather than granting each probe
    // a fresh attempt budget.
    first.behavior = .{ .slow = .{
        .duration = .{ .raw = .fromSeconds(30), .clock = .awake },
        .reply = response_bytes,
    } };
    second.behavior = first.behavior;
    var probing_pool: Pool = .init(&entries, cfg, .{
        .attempt = .{ .raw = .fromMilliseconds(100), .clock = .awake },
        .total = .{ .raw = .fromMilliseconds(150), .clock = .awake },
    }, 1);

    const started = std.Io.Clock.awake.now(io);
    selected = null;
    try testing.expectError(
        error.BudgetExhausted,
        probing_pool.exchange(io, query_bytes, &buf, &selected),
    );
    const elapsed_ns = std.Io.Clock.awake.now(io).nanoseconds - started.nanoseconds;

    // Both were probed, and the pair cost the total budget rather than one
    // attempt budget each.
    try testing.expectEqual(@as(usize, 2), first.calls.load(.acquire));
    try testing.expectEqual(@as(usize, 2), second.calls.load(.acquire));
    try testing.expect(elapsed_ns < @as(i96, 200) * std.time.ns_per_ms);
    // The first probe ran a full attempt and is recorded; the second was cut
    // short by the deadline and is not, so `selected` still names the first.
    try testing.expectEqualStrings("https://first.example/dns-query", selected.?);
    try testing.expectEqual(@as(u64, 2), entries[0].health.total_failures);
    try testing.expectEqual(@as(u64, 1), entries[1].health.total_failures);
    try testing.expectEqual(@as(u64, 1), probing_pool.budgetExhaustedTotal());
}

test "an exchange whose deadline is already gone starts no attempt" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    // A budget of nothing: the entry is healthy and its permit is free, so
    // admission and the health recheck both succeed, and only the remaining-time
    // check stands between the loop and an attempt. Racing a leaf against a
    // timestamp that has already passed is a coin toss — an instantly answering
    // fake can win it — so a pool that omits the check manufactures a success
    // out of a budget that was already spent.
    var fake: Fake = .{ .behavior = .{ .reply = response_bytes } };
    var entries = [_]Entry{testEntry("https://only.example/dns-query", &fake, 10)};
    var pool: Pool = .init(&entries, test_cfg, .{
        .attempt = .{ .raw = .fromMilliseconds(0), .clock = .awake },
        .total = .{ .raw = .fromMilliseconds(0), .clock = .awake },
    }, 1);

    var buf: [512]u8 = undefined;
    var selected: ?[]const u8 = null;
    try testing.expectError(
        error.BudgetExhausted,
        pool.exchange(io, query_bytes, &buf, &selected),
    );
    try testing.expectEqual(@as(usize, 0), fake.calls.load(.acquire));
    try testing.expect(selected == null);
    try testing.expectEqual(@as(u64, 0), entries[0].health.total_failures);
    try testing.expectEqual(@as(u64, 0), entries[0].health.total_successes);
    try testing.expectEqual(@as(u64, 1), pool.budgetExhaustedTotal());
    // The permit was released on the way out, so the entry is usable again.
    try testing.expectEqual(@as(usize, 1), entries[0].admission.permits);
}

/// The pass-two probing test's numbers: the holder's stall outlives the
/// probe's whole budget, so the probe provably died waiting rather than after
/// being served.
const probe_holder_stall_ms = 800;
const probe_order_total_ms = 200;

test "pass two probes in priority order rather than stepping over a saturated entry" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const cfg: health.Config = .{
        .failure_threshold = 1,
        .base_backoff_ms = 60_000,
        .max_backoff_ms = 60_000,
    };
    var first: Fake = .{ .behavior = .{ .fail = error.ConnectFailed } };
    var second: Fake = .{ .behavior = .{ .fail = error.ConnectFailed } };
    var entries = [_]Entry{
        testEntry("https://first.example/dns-query", &first, 10),
        testEntry("https://second.example/dns-query", &second, 20),
    };

    // One exchange under generous budgets puts both entries into backoff, so
    // every exchange after it reaches pass two.
    var seeding_pool: Pool = .init(&entries, cfg, test_timeouts, 1);
    var buf: [512]u8 = undefined;
    var selected: ?[]const u8 = null;
    try testing.expectError(
        error.ConnectFailed,
        seeding_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);

    first.behavior = .{ .slow = .{
        .duration = .{ .raw = .fromMilliseconds(probe_holder_stall_ms), .clock = .awake },
        .reply = response_bytes,
    } };
    second.behavior = .{ .reply = alt_response_bytes };

    var probe_pool: Pool = .init(&entries, cfg, .{
        .attempt = .{ .raw = .fromMilliseconds(probe_order_total_ms), .clock = .awake },
        .total = .{ .raw = .fromMilliseconds(probe_order_total_ms), .clock = .awake },
    }, 1);

    var holder_buf: [512]u8 = undefined;
    var holder = io.concurrent(exchangeAttributed, .{ &seeding_pool, io, &holder_buf }) catch |err| switch (err) {
        error.ConcurrencyUnavailable => return error.SkipZigTest,
    };
    defer _ = holder.await(io) catch Attributed.discarded;
    // The first entry's one permit is provably taken before the probe asks.
    try awaitInFlight(io, &first, 1);

    const calls_before = second.calls.load(.acquire);
    var probe_buf: [512]u8 = undefined;
    selected = null;
    try testing.expectError(
        error.BudgetExhausted,
        probe_pool.exchange(io, query_bytes, &probe_buf, &selected),
    );

    // The claim: pass two waits for the higher-priority entry it is probing.
    // A sweep that stepped over the saturated entry would have been answered
    // instantly by the second one instead.
    try testing.expectEqual(calls_before, second.calls.load(.acquire));
    try testing.expect(selected == null);
    try testing.expectEqual(@as(u64, 1), probe_pool.budgetExhaustedTotal());

    const held = try holder.await(io);
    try testing.expectEqualSlices(u8, response_bytes, holder_buf[0..held.reply_len]);
}

test "budget_exhausted_total counts exhausted exchanges once each" {
    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 entries = [_]Entry{testEntry("https://stalling.example/dns-query", &stalling, 10)};
    var pool: Pool = .init(&entries, test_cfg, .{
        .attempt = .{ .raw = .fromMilliseconds(200), .clock = .awake },
        .total = .{ .raw = .fromMilliseconds(40), .clock = .awake },
    }, 1);

    var buf: [512]u8 = undefined;
    var selected: ?[]const u8 = null;
    try testing.expectEqual(@as(u64, 0), pool.budgetExhaustedTotal());
    for (1..4) |expected| {
        try testing.expectError(
            error.BudgetExhausted,
            pool.exchange(io, query_bytes, &buf, &selected),
        );
        // Once per exchange, never once per endpoint attempted inside it.
        try testing.expectEqual(@as(u64, expected), pool.budgetExhaustedTotal());
    }

    // An exchange that answers adds nothing.
    stalling.behavior = .{ .reply = response_bytes };
    _ = try pool.exchange(io, query_bytes, &buf, &selected);
    try testing.expectEqual(@as(u64, 3), pool.budgetExhaustedTotal());
}

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 an exchange that
    // gave each one a fresh budget would cost two of them and then some. The
    // total is set below two attempts, which is what makes the assertion mean
    // something: only the one deadline the loop owns 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.BudgetExhausted,
        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 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 deadline landed inside the first
    // attempt, which is a truncated one and so ends the exchange.
    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 that spends its whole budget queueing blames nobody but the budget" {
    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 is the only entry, so there is no
    // admissible alternative and the call waits for the entry's one permit
    // until its own 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.BudgetExhausted,
        waiter_pool.exchange(io, query_bytes, &buf_b, &selected),
    );
    // Nothing ran, so nothing is named. A query log row blaming this upstream
    // for a queue it never entered is the reporting defect this replaces.
    try testing.expect(selected == null);
    try testing.expectEqual(@as(u64, 1), waiter_pool.budgetExhaustedTotal());
    try testing.expectEqual(@as(u64, 0), entries[0].health.total_failures);

    // Where it died, not just that it died: the holder's call is still the only
    // one any leaf client has seen, so the budget ran out inside the admission
    // 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 expired 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 saturated-primary test's budgets. The holders do not stall on a clock at
/// all: they are held on a gate the test releases, so the probe's own budget is
/// the only duration the test depends on.
const probe_attempt_ms = 150;
const probe_total_ms = 300;

test "a saturated primary defers to a standby that has capacity" {
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    // The Pi reproduction: the primary is healthy but every slot is held by a
    // slow exchange, and the standby is idle. Priority orders the entries that
    // can be admitted now; it does not entitle a saturated entry to hold a
    // query until the whole budget is gone.
    // The holders are released by the test, not by a clock: a fixed stall can
    // expire before a loaded scheduler lets the poller observe both calls, and
    // an overlap that already passed cannot be recovered by polling.
    var gate: std.Io.Semaphore = .{};
    var primary: Fake = .{ .behavior = .{ .hold = .{ .gate = &gate, .reply = response_bytes } } };
    var standby: Fake = .{ .behavior = .{ .reply = alt_response_bytes } };
    var entries = [_]Entry{
        testEntrySlots("https://primary.example/dns-query", &primary, 10, 2),
        testEntry("https://standby.example/dns-query", &standby, 20),
    };

    // Two pools over one `entries` slice, as the queued-waiter test above does:
    // the holders need a budget long enough to finish, the probe a budget short
    // enough that waiting for a slot would be fatal.
    var holder_pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);
    var probe_pool: Pool = .init(&entries, test_cfg, .{
        .attempt = .{ .raw = .fromMilliseconds(probe_attempt_ms), .clock = .awake },
        .total = .{ .raw = .fromMilliseconds(probe_total_ms), .clock = .awake },
    }, 1);

    var bufs: [2][512]u8 = undefined;
    var holders: [2]std.Io.Future(transport.ExchangeError!Attributed) = undefined;
    var spawned: usize = 0;
    // Every spawned holder is released and awaited on any exit, including the
    // skip when the second spawn is refused: a holder left blocked on the gate
    // would outlive the stack it reads.
    defer for (holders[0..spawned]) |*holder| {
        _ = holder.await(io) catch Attributed.discarded;
    };
    defer for (0..spawned) |_| gate.post(io);
    for (&holders, &bufs) |*holder, *buf| {
        holder.* = io.concurrent(exchangeAttributed, .{ &holder_pool, io, buf }) catch |err| switch (err) {
            error.ConcurrencyUnavailable => return error.SkipZigTest,
        };
        spawned += 1;
    }
    // Both permits provably taken before the probe asks for one.
    try awaitInFlight(io, &primary, 2);

    var probe_buf: [512]u8 = undefined;
    var selected: ?[]const u8 = null;
    const started = std.Io.Clock.awake.now(io);
    const reply = try probe_pool.exchange(io, query_bytes, &probe_buf, &selected);
    const elapsed_ns = std.Io.Clock.awake.now(io).nanoseconds - started.nanoseconds;

    try testing.expectEqualSlices(u8, alt_response_bytes, reply);
    try testing.expectEqualStrings("https://standby.example/dns-query", selected.?);
    // Well inside the probe's own budget: it never queued on the primary.
    try testing.expect(elapsed_ns < @as(i96, probe_total_ms) * std.time.ns_per_ms);
    try testing.expectEqual(@as(usize, 1), standby.calls.load(.acquire));
    // The primary was never entered a third time, and never counted a failure:
    // it was skipped for want of capacity, not blamed for anything.
    try testing.expectEqual(@as(usize, 2), primary.calls.load(.acquire));
    try testing.expectEqual(@as(u64, 0), entries[0].health.total_failures);
}

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

    // Two gates, one per call the first entry takes. Nothing in this test
    // sleeps: the test releases each call itself, so the schedule it asserts
    // on is the schedule it built.
    var gate_a: std.Io.Semaphore = .{};
    var gate_b: std.Io.Semaphore = .{};

    // The first entry has room for both tasks and treats them differently: it
    // fails whichever reaches it first and answers the other. One failure is
    // under `test_cfg`'s threshold of two, so no backoff steers the failed-over
    // task away and the two calls end on different entries.
    var first_entry: Fake = .{
        .behavior = .{ .hold_fail = .{ .gate = &gate_a, .err = error.ConnectFailed } },
        .then = .{ .hold = .{ .gate = &gate_b, .reply = alt_response_bytes } },
    };
    var second_entry: Fake = .{ .behavior = .{ .reply = response_bytes } };
    var entries = [_]Entry{
        testEntrySlots(divergent_first_url, &first_entry, 10, 2),
        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;
    // Registered right after the task and after its await, so it runs before
    // the await: any early return below (a failed wait, a skipped second
    // task, a failed assertion) releases the held call first and the await
    // then returns, instead of the test deadlocking on a gate nobody posted.
    // Extra permits on an already-released gate are harmless.
    defer gate_a.post(io);
    // A count of one means the first call is inside the entry with its
    // behaviour already drawn, so it holds gate A and the second call will draw
    // gate B.
    try awaitInFlight(io, &first_entry, 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;
    defer gate_b.post(io);
    // Both tasks are now inside the first entry, the first on `hold_fail` and
    // the second on `hold`. That is the overlap every assertion below rests
    // on, and it is established here rather than assumed.
    try awaitInFlight(io, &first_entry, 2);

    // Release the first call only. It fails, fails over to the second entry
    // and is answered there while the second call is still held inside the
    // first entry, so a shared identity would be overwritten under it.
    gate_a.post(io);
    const result_a = try first.await(io);
    gate_b.post(io);
    const result_b = try second.await(io);

    // The first task reached the first entry first, but the claim stays 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));
    // Both tasks were admitted to the first entry at once, which is what makes
    // them concurrent rather than serialized behind one permit. The wait above
    // already required this; the counter records that it held to the end.
    try testing.expectEqual(@as(u32, 2), 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"));
}