From 025edbb0935bc7163be4fd43ce8ef858c9b6baff Mon Sep 17 00:00:00 2001 From: m5r Date: Sat, 22 Aug 2026 19:54:02 +0200 Subject: [PATCH] milestone 31: concurrent upstream exchanges, dot session reuse, queue metrics --- CHANGELOG.md | 2 + PLAN.md | 4 +- docs/how-to/troubleshoot.md | 16 + specs/milestone-31.md | 201 ++++++ src/app.zig | 175 +++-- src/cli.zig | 66 +- src/server/resolver_integration_test.zig | 41 +- src/tests.zig | 1 + src/upstream/dot_client.zig | 375 ++++++++++- src/upstream/dot_client_integration_test.zig | 651 +++++++++++++++++++ src/upstream/dot_client_live_test.zig | 5 + src/upstream/pool.zig | 472 +++++++++++--- src/web/metrics.zig | 266 +++++++- src/web/web_integration_test.zig | 12 +- 14 files changed, 2099 insertions(+), 188 deletions(-) create mode 100644 specs/milestone-31.md create mode 100644 src/upstream/dot_client_integration_test.zig diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a39441..d5ab93a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,7 @@ Query provenance: every logged query becomes exactly explainable — what the po - **Diagnostics can be scoped to an absolute window.** `/diagnostics?since=…&until=…` now validates and applies both bounds to the active and resolved lists, and the page states the window it is showing with a way to clear it. A query's detail page links here with the five minutes either side of that query, which is where the underlying failure text for a SERVFAIL lives. - **The query log and the stats endpoints say how far back the history goes.** `GET /api/queries` and all five `/api/stats*` endpoints each carry a `coverage` object: `available_since`, the first second the file can answer for, and `complete`, whether the window you asked for begins inside it. A period that starts before the query log does now says so instead of charting the missing part as zero — which is what a recreate, a retention pass or a fresh install would otherwise look like. - **Three new period breakdowns: `GET /api/stats/types`, `/api/stats/routes` and `/api/stats/clients`.** They take the same `period` parameter as `/api/stats` and report over the same UTC-aligned window, so every panel of one page describes the same span. `types` counts queries per DNS type, with the queries that recorded no type kept as their own row instead of dropped — the numeric type only, since naming types is the admin's job and a second table in the server would drift out of agreement with it. `routes` counts queries by how they were answered, grouping upstream rows by the answering resolver and forward-zone rows by the zone, with blocked, cache, local and rejected answers carrying no source. `clients` returns one bucketed series per client, aligned exactly like `/api/stats/timeseries` so the two charts share an x-axis: the eight busiest clients in the window are named and everything else sums into an `other` series, which is always present and always the same length as the named ones. +- **`/metrics` says whether an upstream is queueing.** Five new per-upstream series, labelled by index and redacted url like the existing ones: `nxdns_upstream_in_flight` and `nxdns_upstream_slots` are the exchanges in flight against an upstream and the ceiling they cannot cross, `nxdns_upstream_queued_total` and `nxdns_upstream_queued_seconds_total` count the exchanges that had to wait for a slot and the time they spent waiting — including the ones that were cancelled while waiting, which is exactly the query that ends in SERVFAIL — and `nxdns_upstream_reuse_recoveries_total` counts the stale DoT connections that were redialled, so connection churn is a number instead of log noise. The two queue counters are approximate: they are sampled when a query is admitted, not measured as a queue length. `/api/health` is unchanged. - **Each of those responses is read atomically.** Every window-bounded read — the five stats endpoints and `GET /api/queries` — now takes its rows and its coverage watermark inside one SQLite read transaction. A retention pass that runs mid-response can no longer hand back rows from before the prune tagged with an `available_since` from after it, and the clients breakdown ranks and buckets from one database state rather than two. The transaction is a deferred read, so it never blocks the query logger or retention. ### Removed @@ -38,6 +39,7 @@ Query provenance: every logged query becomes exactly explainable — what the po ### Fixed +- **A burst of concurrent queries no longer resolves one at a time, and no longer ends in SERVFAIL.** The pool held an upstream for the whole of an exchange, so every query against one upstream waited for the one before it, and the DoT client dialled a fresh TCP connection and ran a full TLS handshake for each query on top of that. Thirty concurrent names against one DoT upstream resolved as a staircase at about 92 ms per query on a Pi, and the queries at the back of the queue burned the five-second total budget waiting and were answered SERVFAIL — with nothing on any surface saying a queue existed. Two changes fix it: each upstream now runs up to eight exchanges at once, each on its own leaf client behind a semaphore, so a query waits for a free slot rather than for the whole upstream; and a DoT client keeps its TLS session open across exchanges instead of handshaking per query. A reused connection that the resolver closed while it was idle is detected at use — never by a keepalive timer — and redialled once, and that redial is invisible to health and to Diagnostics because an idle close is normal, not a fault. The slot count is compiled, not configured: there is no new knob. - **A UDP reply that has to be truncated keeps the answer's RCODE.** When an answer does not fit the client's UDP buffer, nxdns replaces it with an empty reply carrying the TC bit, which tells the client to retry over TCP. That replacement was always built as NOERROR, whatever the answer said — so an oversized NXDOMAIN reached the client as a success, and an EDNS extended RCODE above 15 lost the eight upper bits it needs an OPT record to carry. The truncated reply now carries the full twelve-bit code the answer had, split across the header and the reply's OPT record where the code needs it, and the query-log row records the code the client actually saw. The retry over TCP always returned the right RCODE; this was the UDP answer that preceded it. ## [0.0.8] - 2026-08-21 diff --git a/PLAN.md b/PLAN.md index cf9168d..91a346a 100644 --- a/PLAN.md +++ b/PLAN.md @@ -324,7 +324,9 @@ Per-group boolean. Rewrites known engine domains to their safe-search CNAME targ ## 9. Upstream Resolution - Schemes: `https://…` → DoH, `tls://host:853` → DoT. -- Ordered by priority; sequential attempt; per-upstream failure counters; exponential backoff with jitter; success resets. +- Ordered by priority; one query tries upstreams sequentially; per-upstream failure counters; exponential backoff with jitter; success resets. +- Concurrency is per upstream, not per pool: each entry owns `slots_per_entry = 8` leaf clients behind a semaphore, so at most 8 exchanges are in flight against one upstream at a time and the rest wait for a slot rather than for the whole entry. The slot count is compiled, not configured. A task that finds every slot taken counts itself in `queued_total`/`queued_seconds_total` on both exits — acquisition and cancellation — so a query that burned its total budget waiting is visible on `/metrics` (`nxdns_upstream_queued_total`, `nxdns_upstream_in_flight`, `nxdns_upstream_slots`) instead of being an unexplained SERVFAIL. The counters are admission samples, not an exact queue length. +- DoT keeps its connection: a `DotClient` holds one TLS session open across exchanges rather than dialing and handshaking per query. Staleness is detected at use, never by a keepalive timer — a reused session that fails before the first response byte with a connection-lifecycle error is redialed once and retried, and that recovery counts in `nxdns_upstream_reuse_recoveries_total` without touching health or emitting a diagnostics episode. Every other failure closes the session and classifies as before. - `UpstreamHealth` per upstream: last_success_at, last_error_at, last_error_message, rolling success rate, consecutive failures, backoff-until. This is routing state: it drives failover and backoff, and is exposed through `/metrics` and `nxdns check`. There is no per-upstream API surface: `/api/health` reports the pool as available-of-enabled, and a failing upstream is a Diagnostics episode (`upstream.exchange`). - DoH client: `std.http.Client` with `content-type/accept: application/dns-message`; strict status + payload checks. - `platform/tls_client.zig` enforces per-connection read/write deadlines, classifies TLS errors explicitly, retries with backoff. Integration tests cover timeout/hang scenarios so compiler upgrades can't silently regress them. diff --git a/docs/how-to/troubleshoot.md b/docs/how-to/troubleshoot.md index 0b11cda..1629bc6 100644 --- a/docs/how-to/troubleshoot.md +++ b/docs/how-to/troubleshoot.md @@ -257,6 +257,22 @@ cat /etc/resolv.conf **Fix.** If it points at the nxdns container, repoint it at a real resolver. The container resolves its upstream DoH and DoT hostnames through the host's DNS configuration, so pointing that at nxdns makes nxdns depend on itself to start. LAN clients point at nxdns; the container's own host does not. +## SERVFAIL in bursts, only under concurrent load + +**Symptom.** Single queries answer normally, but a burst of unique names — a page load, a device waking up, a `dig` loop — returns SERVFAIL for some of them, and the upstream itself is healthy on `/api/health`. + +**Diagnosis.** Scrape the queue counters: + +```sh +curl -s http://127.0.0.1:8080/metrics | grep nxdns_upstream_ +``` + +`nxdns_upstream_in_flight` against `nxdns_upstream_slots` is how much of an upstream's concurrency is in use right now, and `nxdns_upstream_queued_total` counts exchanges that had to wait for a slot (with `nxdns_upstream_queued_seconds_total` for how long they waited in total). Both queue counters are approximate — they are sampled when a query is admitted, not measured as a queue length. A `queued_total` climbing with each burst means queries are waiting on the upstream rather than failing at it, and a query that waits past `upstream.total_timeout_ms` is canceled in the queue and answered SERVFAIL. + +`nxdns_upstream_reuse_recoveries_total` is the other half of the picture: it counts DoT connections that went stale between exchanges and were redialed. A few are normal — a resolver is free to close an idle connection. One per query means the connection is never being reused, and every query is paying a full TLS handshake. + +**Fix.** Raise `upstream.total_timeout_ms` if the queue drains but drains too slowly for the budget. Otherwise the queue is telling you the upstream is slow: an upstream answering in a few milliseconds does not fill eight concurrent slots at household query rates, so sustained queueing points at the resolver you configured, and a faster one is the fix. Adding a second upstream is not: failover is strict priority, not load spreading — a query waits for a slot on the first available upstream and only reaches the next one when that upstream fails or is in backoff, so a second entry adds no concurrent capacity to the first. The per-upstream slot count is compiled, not configured, so there is no knob to widen one upstream either. + ## The disk is filling up **Symptom.** Writes stop but DNS keeps answering. The journal shows the transition: diff --git a/specs/milestone-31.md b/specs/milestone-31.md new file mode 100644 index 0000000..8f0883a --- /dev/null +++ b/specs/milestone-31.md @@ -0,0 +1,201 @@ +# Milestone 31: upstream concurrency + +Fix the field defect the Pi deployment reproduced: with one enabled upstream, 30 concurrent unique-name queries resolve as a serial staircase (~92 ms per query), because the pool serializes all exchanges per entry (`Entry.busy` held across the whole exchange, pool.zig:226-247) and the DoT client dials a fresh TCP connection and full TLS handshake per query (dot_client.zig:158-165). Under load, waiters burn the 5 s total budget in the queue and cancel into SERVFAIL, and nothing on any surface shows the queue. Three changes: N concurrent in-flight exchanges per entry, DoT session reuse across exchanges, and queue visibility on `/metrics`. + +Acceptance is the Pi burst repro: 30 concurrent unique names against one enabled DoT upstream must complete without SERVFAIL and without a serial staircase. That run happens on the Pi after deploy (coordinated with the `rpi-nixos-iac` session); the in-repo gates below are the hermetic equivalents. + +Codex design review round 1 folded in; the corrections are marked (C1..C12) where they changed the shape. + +## Sessions + +S1: per-entry slots in the pool (Zig). S2: DoT session reuse (Zig). S3: metrics surface + docs + changelog. Sequential: S1 → S2 → S3. S1 and S2 both touch `src/app.zig` and `src/cli.zig`; S2 and S3 both touch `src/upstream/pool.zig` — sequencing removes every overlap. + +--- + +## Session S1: per-entry concurrency in the pool + +### S1.1 Slot structure + +`pool.zig`. Replace `Entry.client` + `Entry.busy` with: + +```zig +pub const Slot = struct { + client: transport.Client, + busy: std.Io.Mutex = .init, +}; + +pub const Entry = struct { + endpoint: transport.Endpoint, + slots: []Slot, // caller-owned, len >= 1 + priority: i32, + enabled: bool, + health: health.State, + sem: std.Io.Semaphore, // .{ .permits = slots.len } at build time + in_flight: std.atomic.Value(u32) = .init(0), + peak_in_flight: std.atomic.Value(u32) = .init(0), + queued_total: std.atomic.Value(u64) = .init(0), + queued_ns_total: std.atomic.Value(u64) = .init(0), + /// Points into `Upstreams`' stable counter storage (S1.3). Incremented by a + /// DoT slot client when a stale reused session was recovered by the single + /// redial (S2). Zero forever for DoH entries. + reuse_recoveries: *std.atomic.Value(u64), +}; +``` + +**(R2-1)** The recovery counter lives *outside* the entry: `Pool.init` sorts entries by value, so a pointer into an `Entry` taken before the sort would dangle onto a different upstream's field. `Upstreams.build` allocates a stable `recovery_counters: []std.atomic.Value(u64)` (one per enabled upstream, zero-initialized, freed in `deinit`); both the `Entry` and its DoT slot clients hold pointers into it, and the sort copies those pointers intact. `cli.probeUpstreams` points its one-entry pool at a local counter. + +`Pool.init` keeps its sort; sorting copies entries by value, and that stays legal because it runs before any task can reach the pool (same argument the current `busy` default makes). The semaphore's mutex/cond are plain values; assert in `init` that every entry has `slots.len >= 1` and `sem.permits == slots.len`. + +The compiled slot count lives here: `pub const slots_per_entry = 8;`. No config knob (anti-requirement A2). Rationale recorded in the doc comment: listeners admit up to 64 UDP in-flight + 64 TCP connections per family; 8 concurrent exchanges per upstream clears the Pi burst (30 concurrent) in ≤4 waves even before session reuse, and costs ~533 KiB of TLS buffers per DoT entry (8 × 4 × `tls.Client.min_buffer_len`, min_buffer_len = `tls.max_ciphertext_record_len`). + +### S1.2 Acquisition protocol + +In `exchangeLoopLen`, per candidate entry, replacing the `busy.lock` block: + +1. Read `in_flight` with `.acquire`; if `>= slots.len`, this task counts as queued: take a start timestamp (`.awake` clock) now. +2. `try entry.sem.wait(io)` — cancelable, exactly like the old `busy.lock(io)` (`sem.wait` either decrements a permit or returns `error.Canceled`; it cannot return holding one). **(C5)** If step 1 marked this task queued, the counters land on *both* exits: on `error.Canceled`, add `queued_total += 1` and `queued_ns_total += elapsed` (plain atomic adds, nothing cancelable) *before* propagating — a waiter that burned its budget in the queue is exactly the field failure and must be counted. +3. On successful acquisition with step 1 marked: same two adds. +4. **(C6)** The queued detection is approximate by construction: between the releaser's `in_flight` decrement and its `sem.post` (and the mirror window on acquire) a reader can misjudge saturation. The counters are admission samples, not an exact queue length. This is stated in the `Entry` doc comment and in the metric HELP text (S3.1); no attempt at an exact count — that would put a lock around the hot path to size a household queue. +5. Scan `slots` with `busy.tryLock()`; after a semaphore permit, at least one slot mutex is free, so the scan must succeed — `unreachable` if it does not (that is a protocol bug, not a runtime condition). +6. `in_flight.fetchAdd(1, .acq_rel)`, `peak_in_flight.fetchMax(...)`. +7. Run the existing attempt against `slot.client` (same `attempt()` race, same health bookkeeping, same `selected` write, same pass-0 health re-check after acquisition). +8. Release in a defer, in this order: `busy.unlock()`, `in_flight.fetchSub(1, .acq_rel)`, `sem.post(io)`. `post` is uncancelable, so a canceled attempt still returns its permit. + +The pass-0 re-check keeps its current meaning: an entry that entered backoff while this task queued is skipped after acquisition (the defer releases the slot before `continue`). + +**(C12)** Module doc comment rewrite (pool.zig:7-46): the guarantee becomes "at most `slots.len` in-flight exchanges per entry"; the "two concurrent queries do serialize / right trade at household scale" paragraph is deleted (it is the defect); and lines 11-13 ("an outer expiry always means an attempt was in flight") are corrected — a total-budget expiry can now also mean the task died waiting in an entry's queue, which is precisely what `queued_total` records. + +### S1.3 Composition root fan-out + +`src/app.zig` (`Upstreams.build`, app.zig:1154-1270). Each enabled upstream gets `slots_per_entry` leaf clients instead of one: + +- DoH: `slots_per_entry` `DohClient` values per entry, each with its own `request_buf` (1024) + `transfer_buf` (4096); all share the one `std.http.Client` (its connection pool already handles concurrent requests; `DohClient`'s only mutable state is its two buffers, doh_client.zig:105-139, so one client per slot makes each slot single-flight by construction). +- DoT: `slots_per_entry` `DotClient` values per entry, each with its own four `tls.Client.min_buffer_len` buffers, sharing `bundle`/`bundle_lock`/gpa as today. Each DoT client also receives its upstream's `*std.atomic.Value(u64)` from `recovery_counters` (R2-1, S2.1) — wired here even in S1, as a pointer parameter that S1's `DotClient` accepts and stores unused, so S2 does not touch `app.zig`'s allocation shape again. +- **(C11)** New allocations, named: `slot_storage: []pool_mod.Slot` of `enabled * slots_per_entry`, sliced per entry into `Entry.slots` (never pointing into the client arrays), and `recovery_counters` (R2-1). **(R2-2)** `Upstreams` stores the initialized extents — `dot_used` and `doh_used` counts (malformed/skipped upstreams leave the tails of the over-allocated client arrays uninitialized) — and every teardown path iterates only `dot[0..dot_used]`. `Upstreams.deinit` order: close every *initialized* DoT client first (S2's `DotClient.close`), then free `slot_storage`, `recovery_counters`, the client arrays and the buffer blocks. `build`'s failure-path `errdefer`s use the same extents, so a half-built `Upstreams` never closes a session that was never opened nor reads an undefined `DotClient`. In S1 (no sessions yet) the close pass is a loop over the initialized extent whose body S2 fills in. +- Client and slot arrays are never resized after wiring: S2 pins live TLS state inside `DotClient`. + +**(C9)** `cli.probeUpstreams` (cli.zig:935-1010) stays sequential and builds one-slot entries (`slots.len == 1`), but does not stay otherwise unchanged: once S2 lands, a successful DoT probe leaves a session open on a client the loop is about to overwrite, whose buffers the next probe reuses. The probe loop closes its `DotClient` at the end of every iteration, on success and on every error path (`defer dot.close(io)` scoped per iteration). S1 introduces the `close` call site with S1's no-op close; S2 gives it meaning. + +### S1.4 Tests (pool.zig) + +**(C10)** The `Fake` first: multiple slots wrapping one fake removes the serialization that made `calls`, `behavior` and `then` safe to mutate plainly. `calls` becomes `std.atomic.Value(usize)`; `behavior`/`then` swap moves under a `Fake`-owned `std.Io.Mutex` (or the test builds one fake per slot where the scenario allows). Existing tests keep their assertions; the counters they read change type. + +- Concurrent exchanges through one entry overlap: with `slots.len = 2` and a slow fake, two concurrent calls reach `peak_in_flight == 2` on the fake and finish in ~one sleep, not two. (Replaces the current "do not overlap" test, whose claim inverts: the new claim is "at most `slots.len` in flight", pinned by a third concurrent call — fake `peak_in_flight` never exceeds `slots.len`.) +- Queue accounting: with `slots.len = 1` and a slow fake, a second concurrent call lands `queued_total == 1` and non-zero `queued_ns_total`; an uncontended call adds neither. +- **(C5)** A queued waiter canceled by the total budget still counts: exhaust the single slot with a stalling fake, race a second call with a total budget shorter than the stall — it fails `error.Timeout` *and* `queued_total`/`queued_ns_total` advanced. +- A canceled waiter leaves the permit count intact: after the test above, a fresh call still acquires and completes. +- The existing suite (failover, backoff-while-queued, attribution, budgets, diagnostics episodes) passes with entries built through a test helper that wraps a fake into a 1..N-slot entry. + +### S1.5 Acceptance (S1) + +- [ ] `zig build test` green; `zig build test -Dintegration` green. +- [ ] The S1.4 tests exist and pass. +- [ ] `cli.zig` probe path builds, probes each upstream exactly once, and closes its DoT client every iteration. + +--- + +## Session S2: DoT session reuse + +### S2.1 Persistent session state + +`dot_client.zig`. `DotClient` gains an optional open session it owns: + +```zig +const Session = struct { + stream: net.Stream, + tls: tls_client.TlsStream, +}; +session: ?Session = null, +/// Points at the owning pool entry's counter; null in `nxdns check` probes. +reuse_recoveries: ?*std.atomic.Value(u64), +``` + +**(C1)** `TlsStream` self-pins (it hands `tls.Client` pointers into its own reader/writer fields, tls_client.zig:141-147), so a `Session` must never be built locally and copied into `self.session`. Emplacement order is part of the contract: connect the socket; set `self.session = .{ .stream = , .tls = undefined }`; then run `TlsStream.init` *in place* through a pointer to the stored payload (`&self.session.?.tls` via an `if (self.session) |*s|` capture). **(R2-3)** On handshake failure the order is: extract the concrete cause from the in-place `TlsStream` fields first (`concreteHandshake` reads `stream_reader.err`/`stream_writer.err` — resetting the optional first would destroy the only storage holding a cancellation or local-resource cause), then close the raw socket, then set `self.session = null`, then log/map/return the captured cause. No partially initialized session may survive the call. The `DotClient` itself never moves after wiring (S1.3 guarantees the array is final), so the pinned pointers stay valid across exchanges. + +`DotClient.close(io)`: if a session is open, `transport.closeBlocked` the TLS stream then the socket, set `session = null`; tolerate an absent session. `Upstreams.deinit` and the probe loop call it (S1.3). + +### S2.2 Exchange protocol with reuse + +`exchange` becomes: + +1. If `session == null`: dial + handshake with the emplacement order above; failures classify exactly as today through `transport.mapPhase` **(R2-6)** — peer fault unless the concrete cause is local-resource or cancellation, which keep their groups. A session dialed by this call is *fresh* for the rest of these rules. +2. Send the framed query, read the framed response through the stored session. **(C4)** The two-byte length prefix is read byte-at-a-time (or through an explicit received-byte count), so "no response byte received yet" is an observable predicate, not an inference from `readSliceAll`'s all-or-nothing surface — a failure after one prefix byte is *after* first byte. +3. On success: keep the session open and return the reply. `validateResponse` failures (`BadResponse`, `ResponseMismatch`, `ResponseTooLarge`) close the session before returning — the stream position after a bad frame is untrustworthy. +4. **(C3)** Retry-once eligibility is three conditions, all required: the session was **reused** (existed before this call); **no response byte** of this exchange was received yet; and the **concrete unwrapped cause** (via the existing `concreteRead`/`concreteWrite` helpers) is a connection-lifecycle error: `error.EndOfStream`, `error.TlsConnectionTruncated`, `error.ConnectionResetByPeer`, `error.BrokenPipe`. Then: close the session, dial fresh once, retry the exchange once; the retry's outcome is the exchange's outcome. The recovered failure is never recorded against health, never emits a diagnostics episode, and logs at debug only — an upstream closing an idle connection is normal (RFC 7858 keepalive is advisory). On a *successful* recovery, increment `reuse_recoveries` (the pointer from S2.1) — the churn is visible as a counter, not as log spam. +5. **(C2, C3)** Every other failure is final and classifies exactly as today (local resource and cancellation propagate through `transport.mapPhase` untouched; TLS protocol faults, cert faults and post-first-byte faults are peer faults) — and every final I/O or validation failure **closes the session before returning**: after any failed send, receive or validate, framing and TLS state are untrustworthy, and the next exchange must start from a fresh dial. +6. Cancellation (`error.Canceled` unwrapped anywhere) closes the session — mid-frame state is unknown — and propagates. + +The attempt budget covers the whole of `exchange` including the one redial (the pool's `raceWithin` already wraps it; on budget expiry the pool cancels the attempt, and rule 6 closes the session). + +The eligibility decision is a pure function — `retryDecision(reused: bool, bytes_received: bool, cause: anyerror) enum { retry, final }` or equivalent — so the matrix is table-testable without a wire. + +### S2.3 Tests + +**(C8)** Flag discipline: `-Dintegration` is hermetic loopback only; external hosts stay under the live/manual flag (`dot_client_live_test.zig`'s existing gating). The reuse and recovery *wire* claims are therefore in-repo loopback tests, not live-network tests, and they gate the milestone: + +- A loopback DoT fixture: the `-Dintegration` mbedTLS `tls_server` loopback (already built for `platform/tls_server.zig`) serves framed DNS responses. **(R2-4)** The fixture certificate is self-signed and `DotClient` hardcodes `.ca = .system`, so the test preloads the fixture certificate into the test's `Certificate.Bundle` and constructs the client with `tls_name` matching the fixture cert's dNSName SAN — no insecure-verification option is added to the production client (that would be out of scope and a hole). Against it: two sequential exchanges through one client complete over one connection (`session != null` after the first; the server observed one accept); a server-side close between exchanges recovers via the single redial with no health-visible fault, no diagnostics episode (assert via `events_fixture` as in pool.zig:1143), and `reuse_recoveries == 1`; **(C4, R2-5)** the one-prefix-byte case runs on a *reused* session — one successful exchange first, then the server sends exactly one prefix byte and closes — and asserts a final failure with no second accept (no redial): on the fresh path "no retry" would hold vacuously. +- Table tests on the pure decision function: fresh-dial failure → final; reused + pre-byte + lifecycle cause → retry; retry failure → final; reused + post-byte → final; reused + pre-byte + non-lifecycle cause (e.g. `error.TlsAlert`, `error.SystemResources`, `error.Canceled`) → final; validate failure → close+final. +- Cancellation closes the session (stub technique as `stubStream`). +- Live tests (existing flag) may add a real-host reuse smoke but nothing in acceptance depends on them. + +### S2.4 Acceptance (S2) + +- [ ] `zig build test` and `-Dintegration` green, including the loopback reuse/recovery/one-prefix-byte tests. +- [ ] The decision-function table tests pass. +- [ ] No health counter and no diagnostics episode from a recovered stale-reuse failure; `reuse_recoveries` counts it instead. +- [ ] `Upstreams.deinit` and the probe loop close open sessions (no fd leak across probe iterations — assert the probe loop's close runs on the error path too). + +--- + +## Session S3: visibility, docs, changelog + +### S3.1 Snapshot + /metrics + +- `Pool.Snapshot` gains `in_flight: u32`, `slots: u32`, `queued_total: u64`, `queued_seconds_total: f64` (from `queued_ns_total`), `reuse_recoveries_total: u64` **(C7)**. `snapshot()` reads the atomics with `.acquire`; `peak_in_flight` stays test-only. +- `src/web/metrics.zig`: extend `UpstreamSample`/`renderUpstreams` (metrics.zig:105-116, 477-512) with `nxdns_upstream_in_flight` (gauge), `nxdns_upstream_slots` (gauge), `nxdns_upstream_queued_total` (counter), `nxdns_upstream_queued_seconds_total` (counter), `nxdns_upstream_reuse_recoveries_total` (counter), same index+redacted-url labels. **(C6)** The two queued HELP strings say "approximate; sampled at admission". +- `/api/health` upstreams condition: unchanged (anti-requirement A3). + +### S3.2 Docs + changelog + +- PLAN.md §9: concurrency paragraph updated to the slot model and DoT reuse. +- `docs/how-to/troubleshoot.md`: a short section — SERVFAIL bursts under concurrent load, read `nxdns_upstream_queued_total`/`in_flight` for queueing and `nxdns_upstream_reuse_recoveries_total` for connection churn. +- CHANGELOG.md Unreleased: the defect (serialized upstream exchanges, per-query DoT handshakes), the fix, the new metrics. +- `docs/reference/api.md`: only if it documents /metrics series today — match the existing level of detail; otherwise untouched. + +### S3.3 Acceptance (S3) + +- [ ] `zig build test -Dintegration` green; `zig build -Dadmin-dist` green (no admin changes expected — the build gate only proves nothing broke). +- [ ] A metrics test (existing metrics.zig test style) pins the five new series render with the fixture pool. + +--- + +## File ownership + +Sequential sessions. S1: `src/upstream/pool.zig`, `src/app.zig`, `src/cli.zig`, `src/upstream/dot_client.zig` (the unused `reuse_recoveries` parameter + no-op `close` only), `src/upstream/doh_client.zig` (only if a buffer-len constant moves; behavior unchanged). S2: `src/upstream/dot_client.zig`, `src/upstream/dot_client_live_test.zig`, the loopback fixture file(s), `src/app.zig`/`src/cli.zig` (teardown call sites only). S3: `src/upstream/pool.zig` (Snapshot), `src/web/metrics.zig`, PLAN.md, docs/**, CHANGELOG.md. + +## Anti-requirements + +- A1: no upstream name resolution, no new transports, no changes to failover order or backoff semantics. +- A2: no config knobs for concurrency — `slots_per_entry` is compiled; the listener caps stay compiled too. +- A3: no /api/health schema change, no openapi change, no admin change. +- A4: no TLS session tickets/resumption — `tls.Client` in 0.16.0 has no such API here; reuse means keeping the connection open, nothing more. +- A5: no connection keepalive timers or background probes; staleness is detected at use. +- A6: no DoH changes beyond per-slot buffer fan-out — `std.http.Client` already pools connections. +- A7: no exact queue-length accounting — the queued counters are admission samples (S1.2 step 4). + +## Acceptance (milestone complete) + +- [ ] All three sessions' gates green. +- [ ] Deployed to the Pi and the burst repro re-run by the `rpi-nixos-iac` session: 30 concurrent unique names, zero SERVFAIL, no serial staircase (wall clock for the burst well under 30 × single-query latency). +- [ ] `/metrics` on the Pi shows the five new series. + +## Implementation notes (post-build sync) + +Deviations the build kept, judged defensible in review: + +- `Upstreams.deinit` frees in reverse-allocation order rather than the C11 listing order; the constraint that matters — initialized DoT sessions close first — holds, and nothing reads the passive allocations after that. +- The S2.3 cancellation "stub technique" became a real loopback test: a stubbed `TlsStream` cannot survive `close` (it calls `client.end()` through an undefined `tls.Client`), so the wire test proves the claim directly. The pure table test also covers cancellation-is-final. +- `error.BrokenPipe` in the C3 lifecycle set is unreachable from `net.Stream.Writer.Error` in 0.16.0; it stays in `retryDecision` per the contract, reached only by the table test today. +- `Certificate.Bundle` has no in-memory PEM entry point in 0.16.0; the loopback test mirrors `addCertsFromFile`'s decode+parse calls to preload the fixture cert. +- The S2.4 probe-close criterion is pinned as far as the repo can observe it: the failed-probe path runs the per-iteration deferred `close` (cli test, honestly named), and close-after-failure/double-close safety is pinned in the DoT integration tests. Close after a *successful* probe is unreachable in-repo (the probe verifies against the system trust store, and the only in-repo DoT peer is self-signed); no production surface was added to force it. +- The metrics S3.3 "fixture pool" test drives a real `Pool` through the real `snapshot()` → `upstreams()` → render path with distinct per-field values. diff --git a/src/app.zig b/src/app.zig index 7f8a6c9..dff909a 100644 --- a/src/app.zig +++ b/src/app.zig @@ -529,8 +529,8 @@ fn serve(r: cli.Runner, args: cli.RunArgs) !u8 { defer bundle.deinit(gpa); var bundle_lock: std.Io.RwLock = .init; - var upstreams = try Upstreams.build(gpa, cfg.upstreams, &dns_http, &bundle, &bundle_lock, &config_load); - defer upstreams.deinit(gpa); + var upstreams = try Upstreams.build(io, gpa, cfg.upstreams, &dns_http, &bundle, &bundle_lock, &config_load); + defer upstreams.deinit(io, gpa); var pool: pool_mod.Pool = .init( upstreams.active(), @@ -1146,16 +1146,34 @@ fn maintenanceOnce( /// The pool's entries and everything they point into. /// -/// `Pool.Entry.client` is a type-erased pointer into `doh` or `dot`, and each -/// client borrows a slice of `doh_buf`/`dot_buf`, so all five allocations live -/// exactly as long as the pool does. One entry is used by one task at a time -/// (`Entry.busy`), which is why the buffers are per client and not shared the -/// way `cli.probeUpstreams` shares them. +/// Every enabled upstream gets `pool_mod.slots_per_entry` leaf clients, one per +/// slot of its entry, so that many exchanges can be in flight against it at +/// once. `Slot.client` is a type-erased pointer into `doh` or `dot`, each of +/// those clients borrows a slice of `doh_buf`/`dot_buf`, and each entry borrows +/// a run of `slot_storage` and one counter of `recovery_counters` — so every +/// allocation here lives exactly as long as the pool does, and none of them is +/// ever resized. One slot is used by one task at a time, which is why the +/// buffers are per client and not shared the way `cli.probeUpstreams` shares +/// them. const Upstreams = struct { entries: []pool_mod.Entry, used: usize, + /// Sliced per entry into `Entry.slots`, never pointing into the client + /// arrays: `Pool.init` sorts entries and the slices have to survive it. + slot_storage: []pool_mod.Slot, + /// One per enabled upstream, and the reason it is a separate allocation: + /// `Pool.init` sorts entries by value, so a counter living inside an entry + /// would be pointed at by the wrong upstream's clients after the sort. + recovery_counters: []std.atomic.Value(u64), doh: []doh_client.DohClient, dot: []dot_client.DotClient, + /// How much of `doh`/`dot` was actually initialized. A malformed or skipped + /// upstream leaves the tail of an over-allocated array undefined, and both + /// `deinit` and `build`'s failure paths iterate only the initialized + /// prefix — reading a `DotClient` that was never built, or closing a + /// session that was never opened, is what these two counts prevent. + doh_used: usize, + dot_used: usize, doh_buf: []u8, dot_buf: []u8, @@ -1163,6 +1181,7 @@ const Upstreams = struct { /// skipped, because one bad row in a table of four must not take DNS down. /// No usable row at all is a configuration fault. fn build( + io: std.Io, gpa: Allocator, servers: []const model.UpstreamServer, http: *std.http.Client, @@ -1177,24 +1196,30 @@ const Upstreams = struct { if (enabled == 0) return error.NoUsableUpstreams; const chunk = tls.Client.min_buffer_len; + const slots = pool_mod.slots_per_entry; + const leaf_clients = enabled * slots; var self: Upstreams = .{ .entries = try gpa.alloc(pool_mod.Entry, enabled), .used = 0, + .slot_storage = &.{}, + .recovery_counters = &.{}, .doh = &.{}, .dot = &.{}, + .doh_used = 0, + .dot_used = 0, .doh_buf = &.{}, .dot_buf = &.{}, }; - errdefer self.deinit(gpa); + errdefer self.deinit(io, gpa); - self.doh = try gpa.alloc(doh_client.DohClient, enabled); - self.dot = try gpa.alloc(dot_client.DotClient, enabled); - self.doh_buf = try gpa.alloc(u8, enabled * (doh_request_buf_len + doh_transfer_buf_len)); - self.dot_buf = try gpa.alloc(u8, enabled * 4 * chunk); - - var doh_count: usize = 0; - var dot_count: usize = 0; + self.slot_storage = try gpa.alloc(pool_mod.Slot, leaf_clients); + self.recovery_counters = try gpa.alloc(std.atomic.Value(u64), enabled); + for (self.recovery_counters) |*counter| counter.* = .init(0); + self.doh = try gpa.alloc(doh_client.DohClient, leaf_clients); + self.dot = try gpa.alloc(dot_client.DotClient, leaf_clients); + self.doh_buf = try gpa.alloc(u8, leaf_clients * (doh_request_buf_len + doh_transfer_buf_len)); + self.dot_buf = try gpa.alloc(u8, leaf_clients * 4 * chunk); for (servers) |server| { if (!server.enabled) continue; @@ -1208,46 +1233,27 @@ const Upstreams = struct { continue; }; - const client: transport.Client = switch (endpoint.scheme) { - .doh => doh: { - const base = doh_count * (doh_request_buf_len + doh_transfer_buf_len); - const slot = &self.doh[doh_count]; - slot.* = doh_client.DohClient.init( - http, - endpoint, - self.doh_buf[base..][0..doh_request_buf_len], - self.doh_buf[base + doh_request_buf_len ..][0..doh_transfer_buf_len], - ) catch { - log.warn( - "upstream {f} is not a usable DoH url; skipped", - .{safe_url.redactQuoted(server.url)}, - ); - noteUpstream(config_load, server.url, "not a usable DoH url; skipped"); - continue; - }; - doh_count += 1; - break :doh slot.client(); + const entry_slots = self.slot_storage[self.used * slots ..][0..slots]; + switch (endpoint.scheme) { + .doh => if (!self.wireDoh(http, endpoint, entry_slots)) { + log.warn( + "upstream {f} is not a usable DoH url; skipped", + .{safe_url.redactQuoted(server.url)}, + ); + noteUpstream(config_load, server.url, "not a usable DoH url; skipped"); + continue; }, - .dot => dot: { - const base = dot_count * 4 * chunk; - const slot = &self.dot[dot_count]; - slot.* = dot_client.DotClient.init(endpoint, server.tls_name, gpa, bundle, bundle_lock, .{ - .tls_read = self.dot_buf[base..][0..chunk], - .tls_write = self.dot_buf[base + chunk ..][0..chunk], - .stream_read = self.dot_buf[base + 2 * chunk ..][0..chunk], - .stream_write = self.dot_buf[base + 3 * chunk ..][0..chunk], - }); - dot_count += 1; - break :dot slot.client(); - }, - }; + .dot => self.wireDot(gpa, endpoint, server.tls_name, bundle, bundle_lock, entry_slots), + } self.entries[self.used] = .{ .endpoint = endpoint, - .client = client, + .slots = entry_slots, .priority = server.priority, .enabled = true, .health = .init, + .sem = .{ .permits = entry_slots.len }, + .reuse_recoveries = &self.recovery_counters[self.used], }; self.used += 1; } @@ -1256,17 +1262,88 @@ const Upstreams = struct { return self; } + /// One `DohClient` per slot, all sharing the one `std.http.Client`: its + /// connection pool already serves concurrent requests, and a `DohClient`'s + /// only mutable state is the two buffers this gives each slot its own of. + /// + /// False means the url is not a usable DoH url, which `DohClient.init` + /// decides from the url alone — so it fails on the first slot or on none. + /// `doh_used` still advances per client rather than per entry: it means + /// "initialized", and a skipped entry's clients are simply never reached. + fn wireDoh( + self: *Upstreams, + http: *std.http.Client, + endpoint: transport.Endpoint, + slots: []pool_mod.Slot, + ) bool { + for (slots) |*slot| { + const index = self.doh_used; + const base = index * (doh_request_buf_len + doh_transfer_buf_len); + self.doh[index] = doh_client.DohClient.init( + http, + endpoint, + self.doh_buf[base..][0..doh_request_buf_len], + self.doh_buf[base + doh_request_buf_len ..][0..doh_transfer_buf_len], + ) catch return false; + self.doh_used = index + 1; + slot.* = .{ .client = self.doh[index].client() }; + } + return true; + } + + /// One `DotClient` per slot, each with its own four TLS buffers and all + /// sharing the trust store. Every client of one entry reports its stale-reuse + /// recoveries through that entry's counter. + fn wireDot( + self: *Upstreams, + gpa: Allocator, + endpoint: transport.Endpoint, + tls_name: []const u8, + bundle: *Certificate.Bundle, + bundle_lock: *std.Io.RwLock, + slots: []pool_mod.Slot, + ) void { + const chunk = tls.Client.min_buffer_len; + const recoveries = &self.recovery_counters[self.used]; + for (slots) |*slot| { + const index = self.dot_used; + const base = index * 4 * chunk; + self.dot[index] = dot_client.DotClient.init( + endpoint, + tls_name, + gpa, + bundle, + bundle_lock, + recoveries, + .{ + .tls_read = self.dot_buf[base..][0..chunk], + .tls_write = self.dot_buf[base + chunk ..][0..chunk], + .stream_read = self.dot_buf[base + 2 * chunk ..][0..chunk], + .stream_write = self.dot_buf[base + 3 * chunk ..][0..chunk], + }, + ); + self.dot_used = index + 1; + slot.* = .{ .client = self.dot[index].client() }; + } + } + /// The prefix `Pool.init` is given. The rest of `entries` is allocated but /// never filled, which is what keeps `deinit` able to free the whole block. fn active(self: *Upstreams) []pool_mod.Entry { return self.entries[0..self.used]; } - fn deinit(self: *Upstreams, gpa: Allocator) void { + /// Connections first, memory second: a `DotClient` holds a socket its + /// buffers belong to, so nothing it points at may be freed before it is + /// closed. + fn deinit(self: *Upstreams, io: std.Io, gpa: Allocator) void { + for (self.dot[0..self.dot_used]) |*client| client.close(io); gpa.free(self.dot_buf); gpa.free(self.doh_buf); gpa.free(self.dot); gpa.free(self.doh); + gpa.free(self.recovery_counters); + gpa.free(self.slot_storage); gpa.free(self.entries); self.* = undefined; } diff --git a/src/cli.zig b/src/cli.zig index dec88f9..061eaa6 100644 --- a/src/cli.zig +++ b/src/cli.zig @@ -974,10 +974,16 @@ fn probeUpstreams(r: Runner, cfg: model.Config) !usize { continue; }; - // Both clients are pinned for the pool's lifetime: `Entry.client` is an - // erased pointer into one of them. + // Both clients are pinned for the pool's lifetime: the entry's one slot + // holds an erased pointer into one of them. var doh: doh_client.DohClient = undefined; var dot: dot_client.DotClient = undefined; + // A DoT probe leaves a connection open on a client whose buffers the + // next iteration reuses, so it is closed at the end of every iteration — + // on the FAIL paths and on any error out of the loop too. The flag is + // what keeps the close off `dot` while it is still undefined. + var dot_wired = false; + defer if (dot_wired) dot.close(r.io); const client: transport.Client = switch (endpoint.scheme) { .doh => doh: { doh = doh_client.DohClient.init(&http, endpoint, &request_buf, &transfer_buf) catch { @@ -988,22 +994,31 @@ fn probeUpstreams(r: Runner, cfg: model.Config) !usize { break :doh doh.client(); }, .dot => dot: { - dot = dot_client.DotClient.init(endpoint, server.tls_name, r.gpa, &bundle, &bundle_lock, .{ + dot = dot_client.DotClient.init(endpoint, server.tls_name, r.gpa, &bundle, &bundle_lock, null, .{ .tls_read = tls_buffers[0..chunk], .tls_write = tls_buffers[chunk .. 2 * chunk], .stream_read = tls_buffers[2 * chunk .. 3 * chunk], .stream_write = tls_buffers[3 * chunk ..], }); + dot_wired = true; break :dot dot.client(); }, }; + // One slot: the probe is sequential by design, and one exchange per + // upstream is the whole of it. + var slots = [_]pool.Slot{.{ .client = client }}; + // The pool's counter has to point somewhere, and a probe has no + // process-wide counter storage to point it at. + var recoveries: std.atomic.Value(u64) = .init(0); var entries = [_]pool.Entry{.{ .endpoint = endpoint, - .client = client, + .slots = &slots, .priority = server.priority, .enabled = true, .health = .init, + .sem = .{ .permits = slots.len }, + .reuse_recoveries = &recoveries, }}; var single: pool.Pool = .init(&entries, .{}, timeouts, seed); @@ -1746,6 +1761,49 @@ test "the upstream probe redacts a url it cannot parse, without leaving the mach try testing.expect(!std.mem.containsAtLeast(u8, text, 1, "efgh34")); } +test "a failed DoT probe reaches close through the per-iteration defer without a session" { + // Exactly that and no more. This is the only test that reaches the DoT + // branch at all: it arms `dot_wired`, fails the dial, and runs the deferred + // `DotClient.close` on a client that never opened a session — twice, so the + // second iteration's dial happens after the first close. What that proves is + // that the path compiles and survives: `dot_wired` set only after `init`, + // and `close` tolerant of a sessionless client. It does *not* prove the + // defer is load-bearing — with no session open, deleting the `defer` would + // leave this test green. + // + // The property that matters — a probe's connection not surviving into the + // next iteration's shared TLS buffers — needs a session to exist, so it is + // pinned one level down on the client itself, by the close-after-failure + // assertions in `dot_client_integration_test.zig`. A successful DoT probe + // cannot be pinned in-repo at all: it needs a real handshake, and the only + // DoT peer this repo has is the self-signed `-Dintegration` loopback + // fixture, which `probeUpstreams` cannot verify because it builds its bundle + // from the system trust store. + // + // `tls://` with a host that is not an IP literal fails inside `dial` before + // any socket is opened or any CA store is read: upstream name resolution is + // out of scope, so `resolveAddress` refuses it. That keeps the test off the + // network and off the host's configuration. + var captured: Captured = .init(testing.allocator); + defer captured.deinit(); + const r = captured.runner(); + + const cfg: model.Config = .{ + .groups = &.{.{ .name = "default" }}, + .upstreams = &.{ + .{ .url = "tls://dns.example:853", .tls_name = "dns.example" }, + .{ .url = "tls://other.example:853", .tls_name = "other.example" }, + }, + }; + + try testing.expectEqual(@as(usize, 2), try probeUpstreams(r, cfg)); + try testing.expectEqualStrings( + "FAIL upstreams[0] 'tls://dns.example:853': ConnectFailed\n" ++ + "FAIL upstreams[1] 'tls://other.example:853': ConnectFailed\n", + captured.out.written(), + ); +} + test "a successful import prints the warnings the file earned" { // D5, second half. `check` printed this WARN and `import` recorded it and // threw it away, because diagnostics were written on the failure path only. diff --git a/src/server/resolver_integration_test.zig b/src/server/resolver_integration_test.zig index 6ede0a9..277482f 100644 --- a/src/server/resolver_integration_test.zig +++ b/src/server/resolver_integration_test.zig @@ -155,15 +155,32 @@ const FaultyUpstream = struct { } }; -fn testEntry(url: []const u8, upstream_client: transport.Client, priority: i32) pool.Entry { - return .{ - .endpoint = transport.Endpoint.parse(url) catch unreachable, - .client = upstream_client, - .priority = priority, - .enabled = true, - .health = .init, - }; -} +/// One-slot entry storage: the slot the entry points at and the recovery +/// counter it borrows, declared by the test so both outlive the pool. These +/// tests drive failover and deadlines, not concurrency within one upstream, so +/// one slot per entry is the whole story here. +const EntryStorage = struct { + slots: [1]pool.Slot = undefined, + recoveries: std.atomic.Value(u64) = .init(0), + + fn entry( + self: *EntryStorage, + url: []const u8, + upstream_client: transport.Client, + priority: i32, + ) pool.Entry { + self.slots[0] = .{ .client = upstream_client }; + return .{ + .endpoint = transport.Endpoint.parse(url) catch unreachable, + .slots = &self.slots, + .priority = priority, + .enabled = true, + .health = .init, + .sem = .{ .permits = self.slots.len }, + .reuse_recoveries = &self.recoveries, + }; + } +}; const Outcome = union(enum) { work: anyerror!void, @@ -250,9 +267,11 @@ test "the whole resolver answers over udp and tcp and fails over to a healthy up .fail_first = std.math.maxInt(u64), }; var good: GoodUpstream = .{}; + var bad_storage: EntryStorage = .{}; + var good_storage: EntryStorage = .{}; var entries = [_]pool.Entry{ - testEntry("https://bad.example/dns-query", bad.client(), 10), - testEntry("tls://good.example", good.client(), 20), + bad_storage.entry("https://bad.example/dns-query", bad.client(), 10), + good_storage.entry("tls://good.example", good.client(), 20), }; var upstreams: pool.Pool = .init(&entries, test_cfg, pool_timeouts, 1); diff --git a/src/tests.zig b/src/tests.zig index fb81de5..5dcfe72 100644 --- a/src/tests.zig +++ b/src/tests.zig @@ -24,6 +24,7 @@ comptime { _ = @import("upstream/pool.zig"); _ = @import("upstream/dot_client.zig"); _ = @import("upstream/dot_client_live_test.zig"); + _ = @import("upstream/dot_client_integration_test.zig"); _ = @import("server/listener.zig"); _ = @import("server/handler.zig"); _ = @import("server/udp_server.zig"); diff --git a/src/upstream/dot_client.zig b/src/upstream/dot_client.zig index 6303c54..81fe4ec 100644 --- a/src/upstream/dot_client.zig +++ b/src/upstream/dot_client.zig @@ -10,6 +10,16 @@ //! local resource error or a cancellation must never reach the pool as a peer //! fault, so the concrete error is unwrapped from `error.ReadFailed` / //! `error.WriteFailed` before it is classified. +//! +//! A client keeps its connection open between exchanges (RFC 7858 §3.4). An +//! upstream is free to drop an idle one at any time and says nothing first, so +//! staleness is detected at use: an exchange that fails on a *reused* session, +//! before any byte of its response arrived, with a connection-lifecycle cause, +//! is retried exactly once on a fresh dial and counted through +//! `reuse_recoveries` rather than against the upstream's health. Every other +//! failure is final, and every final failure closes the session — after a +//! failed send, receive or validation, the framing and TLS state are +//! untrustworthy and the next exchange must start from a fresh dial. const std = @import("std"); const net = std.Io.net; @@ -41,6 +51,9 @@ const Diagnostic = struct { connect_failed: anyerror, handshake_failed: Handshake, bundle_load_failed: anyerror, + /// A reused session turned out to be dead. Normal operation, so this + /// one is only ever logged at debug; the counter is the real surface. + stale_session: anyerror, const Handshake = struct { verify_name: []const u8, @@ -65,6 +78,10 @@ const Diagnostic = struct { tls_client.classify(hs.cause), }), .bundle_load_failed => |err| try w.print("CA bundle load failed: {s}", .{@errorName(err)}), + .stale_session => |err| try w.print( + "reused session was stale ({s}), redialing once", + .{@errorName(err)}, + ), } } }; @@ -93,6 +110,25 @@ pub const DotClient = struct { bundle_lock: *std.Io.RwLock, /// Caller-owned. One `DotClient` is used by one task at a time. buffers: Buffers, + /// Points at the owning pool entry's counter, which lives in the + /// composition root's stable storage; null in `nxdns check` probes, which + /// have no pool to report through. Incremented when a stale reused session + /// is recovered by a redial. + reuse_recoveries: ?*std.atomic.Value(u64), + /// The connection this client keeps open between exchanges, or null when it + /// holds none. Owned here: nothing else may close either half. + /// + /// Emplaced, never assigned from a local: `TlsStream` hands `tls.Client` + /// pointers into its own reader and writer fields, so a `Session` built on + /// the stack and copied in would leave the TLS client pointing at the dead + /// copy. `dial` sets this to `.{ .stream = …, .tls = undefined }` first and + /// runs the handshake through the stored payload. + session: ?Session = null, + + const Session = struct { + stream: net.Stream, + tls: tls_client.TlsStream, + }; pub const Buffers = struct { /// Plaintext read buffer. @@ -116,6 +152,7 @@ pub const DotClient = struct { gpa: std.mem.Allocator, bundle: *Certificate.Bundle, bundle_lock: *std.Io.RwLock, + reuse_recoveries: ?*std.atomic.Value(u64), buffers: Buffers, ) DotClient { std.debug.assert(endpoint.scheme == .dot); @@ -130,9 +167,25 @@ pub const DotClient = struct { .bundle = bundle, .bundle_lock = bundle_lock, .buffers = buffers, + .reuse_recoveries = reuse_recoveries, }; } + /// Releases whatever this client holds open between exchanges: TLS first, + /// so the peer gets a close_notify, then the socket underneath it. + /// + /// Idempotent and safe with no session open, because that is how every + /// caller uses it — `Upstreams.deinit` closes clients that may never have + /// dialed, the `nxdns check` probe loop closes on its error paths, and + /// `exchange` closes after a failure it has already classified. + pub fn close(self: *DotClient, io: std.Io) void { + if (self.session) |*session| { + transport.closeBlocked(io, &session.tls); + transport.closeBlocked(io, &session.stream); + self.session = null; + } + } + pub fn client(self: *DotClient) transport.Client { return .{ .ptr = self, .exchangeFn = exchangeFn }; } @@ -155,14 +208,15 @@ pub const DotClient = struct { return self.exchange(io, query, response_buf); } - /// One TCP connection and one TLS handshake per exchange, both closed - /// before returning. + /// One framed query and one framed reply over a session this client keeps + /// open, dialing one only when it holds none. /// - /// Connection reuse is deliberately not built. At household query rates the - /// saved round trips are worth less than what a per-exchange connection - /// buys: the pool's per-attempt budget stays a plain race against one task, - /// the failover path never has to reason about a half-dead pooled socket, - /// and every failure is attributable to exactly one exchange. + /// A reused session may already be dead: the upstream is entitled to close + /// an idle connection and RFC 7858 keepalive is advisory, so the first sign + /// is this exchange failing. That case — and only that case — is retried + /// once on a fresh dial; see `retryDecision` for the three conditions. The + /// caller's attempt budget covers the whole call including that redial, + /// because the pool races this function as a whole. pub fn exchange( self: *DotClient, io: std.Io, @@ -174,6 +228,49 @@ pub const DotClient = struct { // local error rather than a silently truncated frame. if (query.len > transport.max_message_len) return error.BufferTooSmall; + const reused = self.session != null; + if (!reused) try self.dial(io); + + const failure = switch (self.transact(query, response_buf)) { + .ok => |reply| return reply, + .failed => |failure| failure, + }; + + switch (retryDecision(reused, failure.received_any, failure.cause)) { + .final => { + self.close(io); + return transport.mapPhase(failure.cause, failure.phase); + }, + .retry => {}, + } + + // Debug, not warn: an upstream dropping an idle connection is routine, + // and one line per idle timeout on a household resolver is log spam. + // `reuse_recoveries` is what makes the churn visible. + log.debug("{f}", .{self.diagnose(.{ .stale_session = failure.cause })}); + self.close(io); + try self.dial(io); + + switch (self.transact(query, response_buf)) { + .ok => |reply| { + if (self.reuse_recoveries) |counter| _ = counter.fetchAdd(1, .monotonic); + return reply; + }, + // The retry's outcome is the exchange's outcome: one redial, never + // two. + .failed => |retried| { + self.close(io); + return transport.mapPhase(retried.cause, retried.phase); + }, + } + } + + /// Opens a session and leaves it in `self.session`, or leaves `self.session` + /// null and returns the classified failure. No partially initialized + /// session ever survives this call. + fn dial(self: *DotClient, io: std.Io) transport.ExchangeError!void { + std.debug.assert(self.session == null); + const address = resolveAddress(self.endpoint) catch |err| { log.warn("{f}", .{self.diagnose(.not_an_ip_literal)}); return err; @@ -181,22 +278,24 @@ pub const DotClient = struct { try self.ensureBundle(io); - var stream = address.connect(io, .{ .mode = .stream }) catch |err| { + const stream = address.connect(io, .{ .mode = .stream }) catch |err| { log.debug("{f}", .{self.diagnose(.{ .connect_failed = err })}); return transport.mapPhase(err, error.ConnectFailed); }; - defer transport.closeBlocked(io, &stream); - // `TlsStream` is pinned: it holds its reader and writer by value and the - // TLS client points at them, so it must not move after `init`. - var tls_stream: tls_client.TlsStream = undefined; + // Emplaced before the handshake, never built beside it and copied in: + // `TlsStream` is pinned by the pointers `tls.Client` holds into its own + // reader and writer fields, so `init` has to run at the address the + // session will live at. + self.session = .{ .stream = stream, .tls = undefined }; + const session = &self.session.?; // `concreteHandshake` reads these two fields when the handshake reports // `error.ReadFailed` / `error.WriteFailed`. `TlsStream.init` sets them // before it can produce either error, but clearing them here keeps that // out of this file's correctness argument. - tls_stream.stream_reader.err = null; - tls_stream.stream_writer.err = null; - tls_stream.init(io, &stream, self.bundle, self.bundle_lock, self.gpa, .{ + session.tls.stream_reader.err = null; + session.tls.stream_writer.err = null; + session.tls.init(io, &session.stream, self.bundle, self.bundle_lock, self.gpa, .{ .host = self.verify_name, .ca = .system, .read_buffer = self.buffers.tls_read, @@ -204,36 +303,72 @@ pub const DotClient = struct { .stream_read_buffer = self.buffers.stream_read, .stream_write_buffer = self.buffers.stream_write, }) catch |err| { - const cause = concreteHandshake(&tls_stream, err); + // Order matters: the concrete cause lives in the in-place + // `TlsStream`'s two error fields, so clearing the optional first + // would destroy the only record that a cancellation or a local + // resource failure — not the peer — ended the handshake. + const cause = concreteHandshake(&session.tls, err); + transport.closeBlocked(io, &session.stream); + self.session = null; log.warn("{f}", .{self.diagnose(.{ .handshake_failed = .{ .verify_name = self.verify_name, .cause = cause, } })}); return transport.mapPhase(cause, error.TlsFailed); }; - defer transport.closeBlocked(io, &tls_stream); + } + + /// One query and one reply on the open session, with enough detail on + /// failure for `retryDecision` to rule on it. Leaves the session open + /// either way; closing a failed one is `exchange`'s job, because only it + /// knows whether the failure is final. + fn transact(self: *DotClient, query: []const u8, response_buf: []u8) Transact { + const session = &self.session.?; + const tls_stream = &session.tls; const writer = tls_stream.writer(); const prefix = transport.framePrefix(@intCast(query.len)); - writer.writeAll(&prefix) catch |err| return sendFailure(&tls_stream, err); - writer.writeAll(query) catch |err| return sendFailure(&tls_stream, err); + writer.writeAll(&prefix) catch |err| return sendFailed(tls_stream, err); + writer.writeAll(query) catch |err| return sendFailed(tls_stream, err); // `TlsStream.flush`, not `writer.flush`: the latter leaves the encrypted // record in the socket writer's buffer and the query never leaves this // process. - tls_stream.flush() catch |err| return sendFailure(&tls_stream, err); + tls_stream.flush() catch |err| return sendFailed(tls_stream, err); const reader = tls_stream.reader(); + // Byte at a time, not `readSliceAll`: that call is all-or-nothing, so a + // failure after the first prefix byte would be indistinguishable from + // one before it, and "no response byte received yet" is one of the three + // conditions a retry needs. var prefix_bytes: [transport.prefix_len]u8 = undefined; - reader.readSliceAll(&prefix_bytes) catch |err| return receiveFailure(&tls_stream, err); + for (&prefix_bytes, 0..) |*byte, received| { + byte.* = reader.takeByte() catch |err| + return receiveFailed(tls_stream, err, received != 0); + } const len = transport.parsePrefix(prefix_bytes); - if (len == 0) return error.BadResponse; - if (len > response_buf.len) return error.ResponseTooLarge; + if (len == 0) return .{ .failed = .{ + .cause = error.BadResponse, + .phase = error.BadResponse, + .received_any = true, + } }; + if (len > response_buf.len) return .{ .failed = .{ + .cause = error.ResponseTooLarge, + .phase = error.ResponseTooLarge, + .received_any = true, + } }; reader.readSliceAll(response_buf[0..len]) catch |err| - return receiveFailure(&tls_stream, err); + return receiveFailed(tls_stream, err, true); - try transport.validateResponse(query, response_buf[0..len]); - return response_buf[0..len]; + transport.validateResponse(query, response_buf[0..len]) catch |err| return .{ .failed = .{ + .cause = err, + .phase = switch (err) { + error.BadResponse => error.BadResponse, + error.ResponseMismatch => error.ResponseMismatch, + }, + .received_any = true, + } }; + return .{ .ok = response_buf[0..len] }; } /// Loads the system CA bundle before the handshake, so that a failure to @@ -301,14 +436,72 @@ fn concreteWrite(stream: *tls_client.TlsStream, err: anyerror) anyerror { return err; } -fn sendFailure(stream: *tls_client.TlsStream, err: anyerror) transport.ExchangeError { - return transport.mapPhase(concreteWrite(stream, err), error.SendFailed); +/// The outcome of one `transact`, as a value: a reply, or everything +/// `retryDecision` needs to rule on the failure. +const Transact = union(enum) { + ok: []u8, + failed: Failure, + + const Failure = struct { + /// Already unwrapped out of `error.ReadFailed` / `error.WriteFailed`. + cause: anyerror, + /// The peer fault this phase means, unless `cause` belongs to this + /// process. + phase: transport.PeerFault, + /// Whether any byte of this exchange's response had arrived. + received_any: bool, + }; +}; + +fn sendFailed(stream: *tls_client.TlsStream, err: anyerror) Transact { + return .{ + .failed = .{ + .cause = concreteWrite(stream, err), + .phase = error.SendFailed, + // Nothing is read before the query is out, so a send failure is always + // pre-first-byte. + .received_any = false, + }, + }; } -fn receiveFailure(stream: *tls_client.TlsStream, err: anyerror) transport.ExchangeError { - return transport.mapPhase(concreteRead(stream, err), error.ReceiveFailed); +fn receiveFailed(stream: *tls_client.TlsStream, err: anyerror, received_any: bool) Transact { + return .{ .failed = .{ + .cause = concreteRead(stream, err), + .phase = error.ReceiveFailed, + .received_any = received_any, + } }; } +/// Whether a failed exchange may be retried once on a fresh dial. +/// +/// Pure, and separate from the wire for that reason: the three conditions are +/// the whole of the reuse contract, and a table test is the only way to see all +/// of them at once. +/// +/// `reused` — a session this call dialed itself was never idle, so its failure +/// is the upstream's answer, not a stale connection. `bytes_received` — once a +/// reply has started, re-sending the query would be a second question, and the +/// failure is the upstream's. `cause` — only the ways a connection ends +/// (`error.EndOfStream` is a clean close_notify, `error.TlsConnectionTruncated` +/// a close without one, `error.ConnectionResetByPeer` and `error.BrokenPipe` +/// the socket-level pair). A TLS alert, a certificate fault, a local resource +/// failure and a cancellation all keep their meaning and are final. +fn retryDecision(reused: bool, bytes_received: bool, cause: anyerror) RetryDecision { + if (!reused) return .final; + if (bytes_received) return .final; + return switch (cause) { + error.EndOfStream, + error.TlsConnectionTruncated, + error.ConnectionResetByPeer, + error.BrokenPipe, + => .retry, + else => .final, + }; +} + +const RetryDecision = enum { retry, final }; + const testing = std.testing; fn expectDiagnostic(expected: []const u8, url: []const u8, detail: Diagnostic.Detail) !void { @@ -343,6 +536,11 @@ test "every diagnostic line redacts the url it names the upstream by" { "tls://dns.example/abcd12?apikey=s3cr3t", .{ .bundle_load_failed = error.FileNotFound }, ); + try expectDiagnostic( + "dot upstream 'tls://dns.example:853': reused session was stale (EndOfStream), redialing once", + "tls://dns.example:853/", + .{ .stale_session = error.EndOfStream }, + ); try expectDiagnostic( "dot upstream 'tls://dns.example': TLS handshake as 'one.one.one.one' failed: " ++ "CertificateHostMismatch (certificate)", @@ -497,27 +695,126 @@ test "the handshake unwrap passes other errors through untouched" { ); } +/// What `exchange` would return for a failure `transact` reported. +fn mappedFailure(outcome: Transact) transport.ExchangeError { + return transport.mapPhase(outcome.failed.cause, outcome.failed.phase); +} + test "the send and receive unwraps prefer the stored cause" { var send = stubStream(null, error.Canceled, null); try testing.expectEqual( transport.ExchangeError.Canceled, - sendFailure(&send, error.WriteFailed), + mappedFailure(sendFailed(&send, error.WriteFailed)), ); // The TLS client's own error wins over the socket reader's. var receive = stubStream(error.ConnectionResetByPeer, null, error.TlsAlert); try testing.expectEqual( transport.ExchangeError.ReceiveFailed, - receiveFailure(&receive, error.ReadFailed), + mappedFailure(receiveFailed(&receive, error.ReadFailed, false)), ); var socket = stubStream(error.SystemResources, null, null); try testing.expectEqual( transport.ExchangeError.SystemResources, - receiveFailure(&socket, error.ReadFailed), + mappedFailure(receiveFailed(&socket, error.ReadFailed, true)), ); } +test "a send failure is always pre-first-byte, and a receive failure reports what it read" { + // The retry rule reads `received_any`, so where it comes from is part of the + // contract rather than an incidental field: nothing is read before the query + // is on the wire, and the receive side is told by its caller. + var stream = stubStream(error.ConnectionResetByPeer, error.ConnectionResetByPeer, null); + try testing.expect(!sendFailed(&stream, error.WriteFailed).failed.received_any); + try testing.expect(!receiveFailed(&stream, error.ReadFailed, false).failed.received_any); + try testing.expect(receiveFailed(&stream, error.ReadFailed, true).failed.received_any); +} + +test "a fresh session is never retried, whatever failed" { + // The redial has nothing to fix: this call dialed the connection itself, so + // the failure is the upstream's answer rather than a stale socket. + for ([_]anyerror{ + error.EndOfStream, + error.TlsConnectionTruncated, + error.ConnectionResetByPeer, + error.BrokenPipe, + error.TlsAlert, + error.Canceled, + }) |cause| { + try testing.expectEqual(RetryDecision.final, retryDecision(false, false, cause)); + try testing.expectEqual(RetryDecision.final, retryDecision(false, true, cause)); + } +} + +test "a reused session is retried once for the ways a connection ends" { + for ([_]anyerror{ + error.EndOfStream, + error.TlsConnectionTruncated, + error.ConnectionResetByPeer, + error.BrokenPipe, + }) |cause| { + try testing.expectEqual(RetryDecision.retry, retryDecision(true, false, cause)); + } +} + +test "a reused session that already answered is never retried" { + // Re-sending the query after a byte of the reply arrived would ask the + // upstream a second question, so a mid-reply failure is final however the + // connection died. + for ([_]anyerror{ + error.EndOfStream, + error.TlsConnectionTruncated, + error.ConnectionResetByPeer, + error.BrokenPipe, + }) |cause| { + try testing.expectEqual(RetryDecision.final, retryDecision(true, true, cause)); + } +} + +test "a reused session is not retried for a fault that says something" { + // A TLS alert, a certificate fault, a local resource failure, a + // cancellation and a bad frame all keep their meaning: none of them is an + // idle connection going away, so none is worth a second dial. + for ([_]anyerror{ + error.TlsAlert, + error.TlsBadRecordMac, + error.CertificateExpired, + error.SystemResources, + error.OutOfMemory, + error.Canceled, + error.BadResponse, + error.ResponseMismatch, + error.ResponseTooLarge, + error.ConnectionRefused, + error.Timeout, + }) |cause| { + try testing.expectEqual(RetryDecision.final, retryDecision(true, false, cause)); + } +} + +test "a validation failure is final and its phase survives the mapping" { + // `transact` reports these with `received_any` set, so the decision is final + // by two of the three conditions at once, and the peer fault the pool + // records is the validation error itself rather than a receive failure. + const outcomes = [_]struct { cause: anyerror, phase: transport.PeerFault }{ + .{ .cause = error.BadResponse, .phase = error.BadResponse }, + .{ .cause = error.ResponseMismatch, .phase = error.ResponseMismatch }, + .{ .cause = error.ResponseTooLarge, .phase = error.ResponseTooLarge }, + }; + for (outcomes) |outcome| { + try testing.expectEqual(RetryDecision.final, retryDecision(true, true, outcome.cause)); + try testing.expectEqual( + @as(transport.ExchangeError, outcome.phase), + mappedFailure(.{ .failed = .{ + .cause = outcome.cause, + .phase = outcome.phase, + .received_any = true, + } }), + ); + } +} + test "a CA bundle scan failure keeps local resource errors out of the peer fault group" { // `Certificate.Bundle.rescan` reaches these through `Allocator.Error`, // `Io.File.OpenError` and `Io.UnexpectedError`. @@ -565,7 +862,7 @@ test "DotClient satisfies the Client interface" { // `init` asserts `endpoint.scheme == .dot`; a `.doh` endpoint trips // `std.debug.assert`, which a test cannot catch in-process. - var dot: DotClient = .init(try .parse("tls://9.9.9.9:853"), "", gpa, &bundle, &bundle_lock, .{ + var dot: DotClient = .init(try .parse("tls://9.9.9.9:853"), "", gpa, &bundle, &bundle_lock, null, .{ .tls_read = buffer[0..chunk], .tls_write = buffer[chunk .. 2 * chunk], .stream_read = buffer[2 * chunk .. 3 * chunk], @@ -574,6 +871,12 @@ test "DotClient satisfies the Client interface" { try testing.expectEqual(transport.Scheme.dot, dot.endpoint.scheme); try testing.expectEqualStrings("9.9.9.9", dot.endpoint.host); + // A client is wired with nothing open, so `init` can return by value and + // the composition root can copy the result into its array: the pinned TLS + // state only exists once `exchange` has dialed. + try testing.expect(dot.session == null); + dot.close(undefined); + try testing.expect(dot.session == null); const iface: transport.Client = dot.client(); try testing.expectEqual(@as(*anyopaque, @ptrCast(&dot)), iface.ptr); @@ -598,7 +901,7 @@ test "a tls_name replaces the verification name and leaves the dial target alone }; const endpoint: transport.Endpoint = try .parse("tls://1.1.1.1:853"); - const named: DotClient = .init(endpoint, "one.one.one.one", gpa, &bundle, &bundle_lock, buffers); + const named: DotClient = .init(endpoint, "one.one.one.one", gpa, &bundle, &bundle_lock, null, buffers); try testing.expectEqualStrings("one.one.one.one", named.verify_name); try testing.expectEqualStrings("1.1.1.1", named.endpoint.host); @@ -606,6 +909,6 @@ test "a tls_name replaces the verification name and leaves the dial target alone try testing.expectEqualSlices(u8, &.{ 1, 1, 1, 1 }, &address.ip4.bytes); try testing.expectEqual(@as(u16, 853), address.ip4.port); - const plain: DotClient = .init(endpoint, "", gpa, &bundle, &bundle_lock, buffers); + const plain: DotClient = .init(endpoint, "", gpa, &bundle, &bundle_lock, null, buffers); try testing.expectEqualStrings("1.1.1.1", plain.verify_name); } diff --git a/src/upstream/dot_client_integration_test.zig b/src/upstream/dot_client_integration_test.zig new file mode 100644 index 0000000..614176e --- /dev/null +++ b/src/upstream/dot_client_integration_test.zig @@ -0,0 +1,651 @@ +//! Hermetic loopback tests for DoT session reuse (`dot_client.zig`). +//! +//! This lives in its own file because it needs `@import("build_options")`, +//! which only exists when the compilation is driven by build.zig, and because a +//! DoT server is a whole fixture rather than a stub. `-Dintegration` gates it; +//! nothing here leaves the machine and nothing here resolves a name. +//! +//! The peer is the same mbedTLS `platform/tls_server.zig` the nxdns listener +//! uses, serving RFC 1035 §4.2.2 framed DNS on the plaintext side. Its +//! certificate is the committed self-signed fixture, which is why every test +//! preloads that certificate into its own `Certificate.Bundle`: `DotClient` +//! hardcodes `.ca = .system` and there is deliberately no way to ask it to skip +//! verification — a fixture-only trust anchor is a test's business, an +//! insecure-verify switch in the production client would be a hole in it. +//! +//! Every test races its client work against a budget. The failures under test +//! are "the client waits for a reply that never comes" shaped, and without a +//! deadline those hang the suite instead of failing it. + +const std = @import("std"); +const build_options = @import("build_options"); +const tls = std.crypto.tls; +const net = std.Io.net; +const Certificate = std.crypto.Certificate; + +const dot_client = @import("dot_client.zig"); +const pool_mod = @import("pool.zig"); +const transport = @import("transport.zig"); +const tls_server = @import("../platform/tls_server.zig"); +const events_fixture = @import("../storage/events_fixture.zig"); + +const testing = std.testing; + +/// Long enough that no handshake or exchange on loopback needs it, short enough +/// that a hang ends the test rather than the suite. +const budget: std.Io.Clock.Duration = .{ .raw = .fromSeconds(10), .clock = .awake }; + +/// The fixture certificate carries `DNS:localhost`, and +/// `Certificate.Parsed.verifyHostName` matches dNSName SANs only, so this is +/// the only name a `DotClient` can verify it as. The dial target stays the +/// loopback IP literal. +const fixture_host = "localhost"; + +/// A query for example.com A: id 0x1234, RD set, one question. +const query_bytes = + "\x12\x34\x01\x00\x00\x01\x00\x00\x00\x00\x00\x00" ++ + "\x07example\x03com\x00\x00\x01\x00\x01"; + +/// The matching response: the question echoed plus one A record. +const response_bytes = + "\x12\x34\x81\x80\x00\x01\x00\x01\x00\x00\x00\x00" ++ + "\x07example\x03com\x00\x00\x01\x00\x01" ++ + "\xc0\x0c\x00\x01\x00\x01\x00\x00\x01\x2c\x00\x04\x5d\xb8\xd8\x22"; + +/// The same answer under a different transaction id. Well-framed and the length +/// the prefix claims, so the client reads the whole of it and only +/// `transport.validateResponse` can reject it — which is the point: a validation +/// failure is not an I/O failure, and it has to close the session all the same. +const mismatched_response_bytes = + "\x99\x99\x81\x80\x00\x01\x00\x01\x00\x00\x00\x00" ++ + "\x07example\x03com\x00\x00\x01\x00\x01" ++ + "\xc0\x0c\x00\x01\x00\x01\x00\x00\x01\x2c\x00\x04\x5d\xb8\xd8\x22"; + +// --------------------------------------------------------------------------- +// Server side +// --------------------------------------------------------------------------- + +/// The listening half: one mbedTLS context, one loopback listener, and the +/// count of connections the scripts below have accepted. That count is what +/// "one connection" and "no redial" are asserted on. +const Server = struct { + ctx: tls_server.ServerContext, + listener: net.Server, + /// Every dial a script chose to `accept`. The listener stays up for the + /// whole of every test on purpose: a refused dial is invisible to this + /// counter, so a script that stopped listening could not tell "the client + /// did not dial again" from "it dialed and was refused". A dial beyond what + /// the script accepts sits un-handshaken in the backlog instead, stalling + /// the client into its budget — the test fails either way, but only an + /// accepted dial shows up in this count. + accepts: std.atomic.Value(u32) = .init(0), + /// Framed queries taken off the wire. A test that has to know the client is + /// blocked waiting for a reply waits on this rather than on a sleep: a + /// Debug-build ECDSA handshake on loopback is slow enough that any fixed + /// gap is either a flake or a stall. + queries: std.atomic.Value(u32) = .init(0), + + fn init(self: *Server, gpa: std.mem.Allocator, io: std.Io) !void { + const fixtures = @import("test_fixtures"); + self.* = .{ + .ctx = try .init(gpa, fixtures.cert_pem, fixtures.key_pem, null), + .listener = try (net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ + .reuse_address = true, + }), + }; + } + + fn deinit(self: *Server, gpa: std.mem.Allocator, io: std.Io) void { + self.listener.deinit(io); + self.ctx.deinit(gpa); + } + + fn address(self: *const Server) net.IpAddress { + return self.listener.socket.address; + } +}; + +/// One accepted connection. Pinned: `ServerStream` recovers its `std.Io` +/// interfaces from their addresses inside itself and hands mbedTLS a pointer to +/// itself as the BIO context, and its plaintext buffers live here too. +const Accepted = struct { + server: *Server, + stream: net.Stream, + tls: tls_server.ServerStream, + read_buffer: [4096]u8, + write_buffer: [4096]u8, + + fn accept(self: *Accepted, gpa: std.mem.Allocator, server: *Server, io: std.Io) !void { + self.server = server; + self.stream = try server.listener.accept(io); + _ = server.accepts.fetchAdd(1, .acq_rel); + errdefer self.stream.close(io); + try self.tls.accept(gpa, &server.ctx, io, &self.stream, &self.read_buffer, &self.write_buffer); + } + + fn close(self: *Accepted, gpa: std.mem.Allocator, io: std.Io) void { + self.tls.close(gpa); + self.stream.close(io); + } + + /// Reads one framed query and writes `response_bytes` back. + fn answerOnce(self: *Accepted) !void { + var query_buf: [2048]u8 = undefined; + _ = try self.readFramed(&query_buf); + try self.writeFramed(response_bytes); + } + + fn readFramed(self: *Accepted, buf: []u8) ![]u8 { + var prefix: [transport.prefix_len]u8 = undefined; + try self.tls.reader().readSliceAll(&prefix); + const len = transport.parsePrefix(prefix); + if (len > buf.len) return error.QueryTooLarge; + try self.tls.reader().readSliceAll(buf[0..len]); + _ = self.server.queries.fetchAdd(1, .acq_rel); + return buf[0..len]; + } + + fn writeFramed(self: *Accepted, message: []const u8) !void { + const prefix = transport.framePrefix(@intCast(message.len)); + try self.tls.writer().writeAll(&prefix); + try self.tls.writer().writeAll(message); + try self.tls.writer().flush(); + } +}; + +/// Answers `count` queries on one connection and holds it open until the test +/// tears it down. +fn serveOnOneConnection( + gpa: std.mem.Allocator, + server: *Server, + io: std.Io, + count: usize, +) anyerror!void { + var session: Accepted = undefined; + try session.accept(gpa, server, io); + defer session.close(gpa, io); + for (0..count) |_| try session.answerOnce(); +} + +/// Answers one query, drops the connection the way an upstream reaps an idle +/// one, then accepts a second and answers one more. +fn serveThenCloseThenServe(gpa: std.mem.Allocator, server: *Server, io: std.Io) anyerror!void { + for (0..2) |_| { + var session: Accepted = undefined; + try session.accept(gpa, server, io); + defer session.close(gpa, io); + try session.answerOnce(); + } +} + +/// Answers one query, then replies to the next with a single length-prefix byte +/// and closes. The second reply has started, so the client may not redial. +fn serveThenTruncate(gpa: std.mem.Allocator, server: *Server, io: std.Io) anyerror!void { + var session: Accepted = undefined; + try session.accept(gpa, server, io); + defer session.close(gpa, io); + + try session.answerOnce(); + + var query_buf: [2048]u8 = undefined; + _ = try session.readFramed(&query_buf); + try session.tls.writer().writeAll(&[_]u8{0x00}); + try session.tls.writer().flush(); +} + +/// Answers one query, then answers the next with a frame the client can read +/// whole and `transport.validateResponse` must still reject. +fn serveThenAnswerWithWrongId(gpa: std.mem.Allocator, server: *Server, io: std.Io) anyerror!void { + var session: Accepted = undefined; + try session.accept(gpa, server, io); + defer session.close(gpa, io); + + try session.answerOnce(); + + var query_buf: [2048]u8 = undefined; + _ = try session.readFramed(&query_buf); + try session.writeFramed(mismatched_response_bytes); +} + +/// Answers one query, drops the connection, then takes the redial and drops that +/// one too — after reading its query and before answering it. +/// +/// The redial *succeeds*, which is the point: the exchange that fails is the +/// retry, running on a session this call dialed itself. `retryDecision` gives a +/// fresh session no second chance, so the client owes exactly two dials. +fn serveThenCloseThenFailTheRetry(gpa: std.mem.Allocator, server: *Server, io: std.Io) anyerror!void { + { + var session: Accepted = undefined; + try session.accept(gpa, server, io); + defer session.close(gpa, io); + try session.answerOnce(); + } + { + var session: Accepted = undefined; + try session.accept(gpa, server, io); + defer session.close(gpa, io); + // Read the query before dropping the connection: the failure under test + // is the client's *receive*, and a peer that closed before taking the + // query could fail its send instead. + var query_buf: [2048]u8 = undefined; + _ = try session.readFramed(&query_buf); + } +} + +/// Completes the handshake, takes the query and never answers it. +fn serveSilently(gpa: std.mem.Allocator, server: *Server, io: std.Io) anyerror!void { + var session: Accepted = undefined; + try session.accept(gpa, server, io); + defer session.close(gpa, io); + + var query_buf: [2048]u8 = undefined; + _ = try session.readFramed(&query_buf); + const forever: std.Io.Clock.Duration = .{ .raw = .fromSeconds(3600), .clock = .awake }; + try forever.sleep(io); +} + +// --------------------------------------------------------------------------- +// Client side +// --------------------------------------------------------------------------- + +/// Everything one `DotClient` borrows, plus the client itself. +/// +/// Built in place: `DotClient` pins the TLS state of an open session inside +/// itself, and its endpoint and buffers are borrowed from this struct. +const ClientFixture = struct { + gpa: std.mem.Allocator, + bundle: Certificate.Bundle = .empty, + bundle_lock: std.Io.RwLock = .init, + buffers: [4 * tls.Client.min_buffer_len]u8 = undefined, + url_buf: [32]u8 = undefined, + recoveries: std.atomic.Value(u64) = .init(0), + dot: dot_client.DotClient = undefined, + + fn init(self: *ClientFixture, gpa: std.mem.Allocator, io: std.Io, address: net.IpAddress) !void { + self.* = .{ .gpa = gpa }; + errdefer self.bundle.deinit(gpa); + try self.preloadFixtureCert(io); + + const url = try std.fmt.bufPrint(&self.url_buf, "tls://127.0.0.1:{d}", .{address.ip4.port}); + const chunk = tls.Client.min_buffer_len; + self.dot = .init( + try .parse(url), + fixture_host, + gpa, + &self.bundle, + &self.bundle_lock, + &self.recoveries, + .{ + .tls_read = self.buffers[0..chunk], + .tls_write = self.buffers[chunk .. 2 * chunk], + .stream_read = self.buffers[2 * chunk .. 3 * chunk], + .stream_write = self.buffers[3 * chunk ..], + }, + ); + } + + fn deinit(self: *ClientFixture, io: std.Io) void { + self.dot.close(io); + self.bundle.deinit(self.gpa); + } + + /// Makes the self-signed fixture certificate the one trust anchor this + /// client has. + /// + /// A non-empty bundle is what both `DotClient.ensureBundle` and + /// `TlsStream.init` check before they would scan the system store, so + /// preloading it also keeps the test off the host's CA directory entirely. + /// The decode mirrors `Certificate.Bundle.addCertsFromFile`, which is the + /// only PEM entry point the stdlib exposes and takes a file; writing the + /// committed fixture back out to disk to read it in again would be the + /// longer way round to the same three calls. + fn preloadFixtureCert(self: *ClientFixture, io: std.Io) !void { + const fixtures = @import("test_fixtures"); + const begin_marker = "-----BEGIN CERTIFICATE-----"; + const end_marker = "-----END CERTIFICATE-----"; + + const body_start = (std.mem.find(u8, fixtures.cert_pem, begin_marker) orelse + return error.MissingBeginCertificateMarker) + begin_marker.len; + const body_end = std.mem.findPos(u8, fixtures.cert_pem, body_start, end_marker) orelse + return error.MissingEndCertificateMarker; + const encoded = std.mem.trim(u8, fixtures.cert_pem[body_start..body_end], " \t\r\n"); + + const decoder = std.base64.standard.decoderWithIgnore(" \t\r\n"); + try self.bundle.bytes.ensureUnusedCapacity(self.gpa, encoded.len / 4 * 3 + 3); + const decoded_start: u32 = @intCast(self.bundle.bytes.items.len); + const written = try decoder.decode( + self.bundle.bytes.allocatedSlice()[decoded_start..], + encoded, + ); + self.bundle.bytes.items.len += written; + try self.bundle.parseCert(self.gpa, decoded_start, std.Io.Clock.real.now(io).toSeconds()); + try testing.expect(self.bundle.map.count() == 1); + } +}; + +/// Blocks until the server has taken `count` framed queries off the wire. +/// +/// Bounded by the same budget as everything else here: a server that died +/// before it read one would otherwise hang the suite. +fn awaitQueries(io: std.Io, server: *Server, count: u32) !void { + const step: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(10), .clock = .awake }; + const steps = 1000; + for (0..steps) |_| { + if (server.queries.load(.acquire) >= count) return; + try step.sleep(io); + } + return error.DotFixtureTimedOut; +} + +fn expire(io: std.Io, duration: std.Io.Clock.Duration) std.Io.Cancelable!void { + return duration.sleep(io); +} + +const Outcome = union(enum) { + client: anyerror!void, + expiry: std.Io.Cancelable!void, +}; + +/// Runs `client_work` against the budget and tears the server task down either +/// way: a client that fails before it connects would otherwise leave the server +/// blocked in `accept` forever. +fn runAgainstServer( + io: std.Io, + server_task: *std.Io.Future(anyerror!void), + comptime client_work: anytype, + args: anytype, +) !void { + var outcomes: [2]Outcome = undefined; + var race: std.Io.Select(Outcome) = .init(io, &outcomes); + defer race.cancelDiscard(); + try race.concurrent(.client, client_work, args); + try race.concurrent(.expiry, expire, .{ io, budget }); + + const client_result: anyerror!void = switch (try race.await()) { + .client => |result| result, + .expiry => |result| blk: { + try result; + break :blk error.DotFixtureTimedOut; + }, + }; + + const server_result = if (client_result) |_| + server_task.await(io) + else |_| + server_task.cancel(io); + + try client_result; + server_result catch |err| switch (err) { + // The scripts that hold a connection open past their last answer are + // torn down by the cancel above, which is the intended end for them. + error.Canceled => {}, + else => return err, + }; +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +fn twoExchangesOverOneSession(io: std.Io, fixture: *ClientFixture) anyerror!void { + var buf: [512]u8 = undefined; + + const first = try fixture.dot.exchange(io, query_bytes, &buf); + try testing.expectEqualSlices(u8, response_bytes, first); + // The point of the milestone: the connection outlives the exchange. + try testing.expect(fixture.dot.session != null); + + const second = try fixture.dot.exchange(io, query_bytes, &buf); + try testing.expectEqualSlices(u8, response_bytes, second); + try testing.expect(fixture.dot.session != null); +} + +test "two exchanges through one DoT client share one connection" { + if (!build_options.integration) return error.SkipZigTest; + + const gpa = testing.allocator; + var threaded: std.Io.Threaded = .init(gpa, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var server: Server = undefined; + try server.init(gpa, io); + defer server.deinit(gpa, io); + + var fixture: ClientFixture = undefined; + try fixture.init(gpa, io, server.address()); + defer fixture.deinit(io); + + var server_task = try io.concurrent(serveOnOneConnection, .{ gpa, &server, io, @as(usize, 2) }); + try runAgainstServer(io, &server_task, twoExchangesOverOneSession, .{ io, &fixture }); + + try testing.expectEqual(@as(u32, 1), server.accepts.load(.acquire)); + try testing.expectEqual(@as(u64, 0), fixture.recoveries.load(.acquire)); +} + +fn twoPoolExchanges(io: std.Io, pool: *pool_mod.Pool) anyerror!void { + var buf: [512]u8 = undefined; + var selected: ?[]const u8 = null; + const first = try pool.exchange(io, query_bytes, &buf, &selected); + try testing.expectEqualSlices(u8, response_bytes, first); + const second = try pool.exchange(io, query_bytes, &buf, &selected); + try testing.expectEqualSlices(u8, response_bytes, second); +} + +test "a session the upstream closed is recovered by one redial and counted, not blamed" { + if (!build_options.integration) return error.SkipZigTest; + + const gpa = testing.allocator; + var threaded: std.Io.Threaded = .init(gpa, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var server: Server = undefined; + try server.init(gpa, io); + defer server.deinit(gpa, io); + + var fixture: ClientFixture = undefined; + try fixture.init(gpa, io, server.address()); + defer fixture.deinit(io); + + // Through a real pool entry, because the claim is about what the pool does + // *not* see: an upstream reaping an idle connection must not cost it health + // or raise an operational event. + var slots = [_]pool_mod.Slot{.{ .client = fixture.dot.client() }}; + var entries = [_]pool_mod.Entry{.{ + .endpoint = fixture.dot.endpoint, + .slots = &slots, + .priority = 10, + .enabled = true, + .health = .init, + .sem = .{ .permits = slots.len }, + .reuse_recoveries = &fixture.recoveries, + }}; + var pool: pool_mod.Pool = .init(&entries, .{ + .failure_threshold = 2, + .base_backoff_ms = 60_000, + .max_backoff_ms = 60_000, + }, .{ + .attempt = .{ .raw = .fromSeconds(10), .clock = .awake }, + .total = .{ .raw = .fromSeconds(30), .clock = .awake }, + }, 1); + + var fx: events_fixture.Fixture = .{}; + try fx.init(io, 1000); + defer fx.deinit(); + pool.diagnostics = &fx.store; + + var server_task = try io.concurrent(serveThenCloseThenServe, .{ gpa, &server, io }); + try runAgainstServer(io, &server_task, twoPoolExchanges, .{ io, &pool }); + + try testing.expectEqual(@as(u32, 2), server.accepts.load(.acquire)); + try testing.expectEqual(@as(u64, 1), fixture.recoveries.load(.acquire)); + try testing.expectEqual(@as(u64, 2), entries[0].health.total_successes); + try testing.expectEqual(@as(u64, 0), entries[0].health.total_failures); + try testing.expectEqual(@as(u32, 0), entries[0].health.consecutive_failures); + try testing.expectEqual(@as(?std.Io.Timestamp, null), entries[0].health.backoff_until); + try testing.expectEqual( + @as(i64, 0), + try fx.count("SELECT count(*) FROM operational_events"), + ); +} + +fn exchangeThenReadTruncatedReply(io: std.Io, fixture: *ClientFixture) anyerror!void { + var buf: [512]u8 = undefined; + _ = try fixture.dot.exchange(io, query_bytes, &buf); + + // One prefix byte arrived, so the reply had started: re-sending the query on + // a fresh connection would be a second question, not a recovery. + try testing.expectError(error.ReceiveFailed, fixture.dot.exchange(io, query_bytes, &buf)); + try testing.expect(fixture.dot.session == null); +} + +test "a reused session that failed after one response byte is not redialed" { + if (!build_options.integration) return error.SkipZigTest; + + const gpa = testing.allocator; + var threaded: std.Io.Threaded = .init(gpa, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var server: Server = undefined; + try server.init(gpa, io); + defer server.deinit(gpa, io); + + var fixture: ClientFixture = undefined; + try fixture.init(gpa, io, server.address()); + defer fixture.deinit(io); + + var server_task = try io.concurrent(serveThenTruncate, .{ gpa, &server, io }); + try runAgainstServer(io, &server_task, exchangeThenReadTruncatedReply, .{ io, &fixture }); + + // The whole claim: the client never came back for a second connection. + try testing.expectEqual(@as(u32, 1), server.accepts.load(.acquire)); + try testing.expectEqual(@as(u64, 0), fixture.recoveries.load(.acquire)); +} + +fn exchangeThenReadWrongId(io: std.Io, fixture: *ClientFixture) anyerror!void { + var buf: [512]u8 = undefined; + _ = try fixture.dot.exchange(io, query_bytes, &buf); + try testing.expect(fixture.dot.session != null); + + // The read succeeded; only the bytes are wrong. Nothing about that says the + // connection is stale, so it is final — but the stream position after a + // frame this client will not trust is unknowable, so the session goes. + try testing.expectError(error.ResponseMismatch, fixture.dot.exchange(io, query_bytes, &buf)); + try testing.expect(fixture.dot.session == null); +} + +test "a reply that fails validation is final and clears the session without redialing" { + if (!build_options.integration) return error.SkipZigTest; + + const gpa = testing.allocator; + var threaded: std.Io.Threaded = .init(gpa, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var server: Server = undefined; + try server.init(gpa, io); + defer server.deinit(gpa, io); + + var fixture: ClientFixture = undefined; + try fixture.init(gpa, io, server.address()); + defer fixture.deinit(io); + + var server_task = try io.concurrent(serveThenAnswerWithWrongId, .{ gpa, &server, io }); + try runAgainstServer(io, &server_task, exchangeThenReadWrongId, .{ io, &fixture }); + + // Both halves of the claim. One accept: the reused session read a whole + // frame, so nothing here is a lifecycle failure and no redial is owed. And + // nothing was recovered, so the counter stays where it was. + try testing.expectEqual(@as(u32, 1), server.accepts.load(.acquire)); + try testing.expectEqual(@as(u64, 0), fixture.recoveries.load(.acquire)); +} + +fn exchangeThenFailTheRetry(io: std.Io, fixture: *ClientFixture) anyerror!void { + var buf: [512]u8 = undefined; + _ = try fixture.dot.exchange(io, query_bytes, &buf); + try testing.expect(fixture.dot.session != null); + + // Stale session, no response byte, lifecycle cause: the one redial is owed, + // taken, and connected. The exchange on that fresh session then fails its + // read, and the retry's outcome is the exchange's outcome — no third dial, + // because a session this call dialed itself is never retried. + try testing.expectError(error.ReceiveFailed, fixture.dot.exchange(io, query_bytes, &buf)); + try testing.expect(fixture.dot.session == null); + + // What the `nxdns check` probe loop relies on: closing a client whose + // exchange already failed is a no-op, not a double close. + fixture.dot.close(io); + try testing.expect(fixture.dot.session == null); +} + +test "a stale session whose retry fails is final and leaves no session behind" { + if (!build_options.integration) return error.SkipZigTest; + + const gpa = testing.allocator; + var threaded: std.Io.Threaded = .init(gpa, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var server: Server = undefined; + try server.init(gpa, io); + defer server.deinit(gpa, io); + + var fixture: ClientFixture = undefined; + try fixture.init(gpa, io, server.address()); + defer fixture.deinit(io); + + var server_task = try io.concurrent(serveThenCloseThenFailTheRetry, .{ gpa, &server, io }); + try runAgainstServer(io, &server_task, exchangeThenFailTheRetry, .{ io, &fixture }); + + // Exactly two: the first exchange and the one redial. A client that retried + // its retry would need a third dial; the script accepts no third connection, + // so that dial would stall un-handshaken until the client's budget failed + // the test — it cannot succeed silently. + try testing.expectEqual(@as(u32, 2), server.accepts.load(.acquire)); + // The redial connected but the exchange on it failed, so nothing was + // recovered and nothing is counted. + try testing.expectEqual(@as(u64, 0), fixture.recoveries.load(.acquire)); +} + +fn oneExchange(io: std.Io, fixture: *ClientFixture) transport.ExchangeError!void { + var buf: [512]u8 = undefined; + _ = try fixture.dot.exchange(io, query_bytes, &buf); +} + +test "a canceled exchange leaves no session open" { + if (!build_options.integration) return error.SkipZigTest; + + const gpa = testing.allocator; + var threaded: std.Io.Threaded = .init(gpa, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var server: Server = undefined; + try server.init(gpa, io); + defer server.deinit(gpa, io); + + var fixture: ClientFixture = undefined; + try fixture.init(gpa, io, server.address()); + defer fixture.deinit(io); + + var server_task = try io.concurrent(serveSilently, .{ gpa, &server, io }); + var client_task = try io.concurrent(oneExchange, .{ io, &fixture }); + + // The query is on the wire and no answer is coming, so the client is + // provably blocked in its read — the state a total-budget expiry cancels an + // exchange in. + try awaitQueries(io, &server, 1); + try testing.expectError(error.Canceled, client_task.cancel(io)); + + // Mid-frame state is unknowable after a cancellation, so the next exchange + // has to start from a fresh dial. + try testing.expect(fixture.dot.session == null); + try testing.expectEqual(@as(u64, 0), fixture.recoveries.load(.acquire)); + + server_task.cancel(io) catch |err| switch (err) { + error.Canceled => {}, + else => return err, + }; +} diff --git a/src/upstream/dot_client_live_test.zig b/src/upstream/dot_client_live_test.zig index c60c261..d955c59 100644 --- a/src/upstream/dot_client_live_test.zig +++ b/src/upstream/dot_client_live_test.zig @@ -55,8 +55,13 @@ fn runExchange(io: std.Io, params: Params) anyerror!usize { params.gpa, params.bundle, params.bundle_lock, + null, params.buffers, ); + // The client owns the session it dials and keeps it open past the exchange, + // so the task that built the client is what has to close it. Without this + // the socket and its TLS state outlive the test. + defer client.close(io); var selected: ?[]const u8 = null; const reply = try client.client().exchange(io, query_bytes, params.response_buf, &selected); std.debug.assert(std.mem.eql(u8, selected.?, endpoint.url)); diff --git a/src/upstream/pool.zig b/src/upstream/pool.zig index 1e179f2..af3b991 100644 --- a/src/upstream/pool.zig +++ b/src/upstream/pool.zig @@ -10,7 +10,9 @@ //! waits for: without it, N unreachable upstreams cost N × attempt, and the //! resolver above has already given up. The outer expiry is `error.Timeout` //! unconditionally — the fast-failure paths (no entry enabled) return long -//! before the budget, so an expiry always means an attempt was in flight. +//! before the budget, so an expiry means either that an attempt was in flight +//! or that the task died waiting for a slot on a saturated entry. +//! `Entry.queued_total` is what tells those two apart. //! //! Two passes, not one. Pass one walks the enabled entries that health says are //! available. Pass two runs only when pass one attempted nothing, and skips the @@ -25,25 +27,24 @@ //! never held across an exchange, so a slow upstream cannot block a health read //! or an attempt on some other entry. //! -//! `Entry.busy` guards one entry's `client`. A task holds it for a whole -//! attempt against that entry — the exchange and the timeout race around it — -//! and drops it before the failover loop moves to the next candidate. This is -//! what makes the pool safe to share: `DohClient` owns a request buffer and a +//! An entry owns one leaf client per slot, each behind its own `Slot.busy` +//! mutex, plus an `Entry.sem` carrying one permit per slot. A task takes a +//! permit, then the first free slot's mutex, and holds that mutex for a whole +//! attempt against the entry — the exchange and the timeout race around it — +//! before the failover loop moves to the next candidate. The mutex is what +//! makes the pool safe to share: `DohClient` owns a request buffer and a //! transfer buffer, `DotClient` owns four TLS buffers and the stream state //! built on them, so one client may be inside `exchange` only once at a time. //! -//! Because that wait is serializing, pass one re-reads health after it takes -//! `busy`: the attempts a task queued behind can fail the entry into backoff -//! while it waits, and pass one must not attempt an entry that is unavailable by -//! the time it gets its turn. Lock order is always `busy` then `Pool.mutex`, -//! never the reverse. +//! Because a saturated entry still makes tasks wait, pass one re-reads health +//! after it acquires a slot: the attempts a task queued behind can fail the +//! entry into backoff while it waits, and pass one must not attempt an entry +//! that is unavailable by the time it gets its turn. Lock order is always +//! `Slot.busy` then `Pool.mutex`, never the reverse. //! -//! The guarantee is therefore: at most one in-flight exchange per entry, any -//! number of entries in flight at once, and bookkeeping that never waits on a -//! peer. Two concurrent queries that resolve to the same sole upstream do -//! serialize. At household scale that is the right trade — the alternative is a -//! client instance per listener task, which multiplies TLS buffers and -//! connections for a query rate that never needed them. +//! The guarantee is therefore: at most `slots.len` in-flight exchanges per +//! entry, any number of entries in flight at once, and bookkeeping that never +//! waits on a peer. const std = @import("std"); @@ -75,18 +76,77 @@ const AttemptFailure = struct { } }; +/// How many leaf clients the composition root builds per enabled upstream, and +/// therefore how many exchanges one upstream may have in flight at once. +/// +/// Compiled, not configured (anti-requirement A2). The listeners admit up to 64 +/// UDP queries in flight and 64 TCP connections per family, so a burst reaches +/// the pool tens wide; 8 concurrent exchanges per upstream clear the 30-query +/// burst the Pi reproduced in at most four waves even before DoT session reuse +/// removes the per-exchange handshake. The cost is bounded and paid once at +/// startup: a DoT entry holds 8 × 4 × `tls.Client.min_buffer_len` of TLS +/// buffers, about 533 KiB, and a DoH entry two small buffers per slot over one +/// shared `std.http.Client`. +pub const slots_per_entry = 8; + +/// One leaf client of an entry, and the lock that keeps one task inside it. +pub const Slot = struct { + client: transport.Client, + /// Held for the whole of one attempt, so `client` is never re-entered while + /// it is using its own buffers. Defaulted because `Pool.init` sorts + /// `entries` by value, which may only copy unlocked mutexes, and it runs + /// before the pool is reachable by any task. + busy: std.Io.Mutex = .init, +}; + pub const Entry = struct { endpoint: transport.Endpoint, - client: transport.Client, + /// Caller-owned, `len >= 1`, never resized: `DotClient` pins live TLS state + /// and `Slot.client` is an erased pointer at it. + slots: []Slot, /// Lower is tried first (PLAN §11.2 `upstreams.priority`). priority: i32, enabled: bool, health: health.State, - /// Held for the whole of one attempt against this entry, so `client` is - /// never re-entered while it is using its own buffers. Defaulted because - /// `init` sorts `entries` by value, which may only copy unlocked mutexes, - /// and it runs before the pool is reachable by any task. - busy: std.Io.Mutex = .init, + /// One permit per slot, so a task waits here rather than spinning over + /// `Slot.busy`. `.{ .permits = slots.len }` at build time; `Pool.init` + /// asserts it. Its mutex and condition are plain values, so the sort in + /// `init` may copy it. + sem: std.Io.Semaphore, + in_flight: std.atomic.Value(u32) = .init(0), + /// Test-only: the most exchanges this entry ever had in flight at once. + peak_in_flight: std.atomic.Value(u32) = .init(0), + /// Admission samples, not a queue length: a task counts itself queued when + /// it finds `in_flight` saturated before it waits for a permit, and between + /// a releaser's `in_flight` decrement and its `sem.post` (and the mirror + /// window on acquire) that read can misjudge saturation by one. Exactness + /// would cost a lock on the hot path to size a household queue. + queued_total: std.atomic.Value(u64) = .init(0), + queued_ns_total: std.atomic.Value(u64) = .init(0), + /// Points into the composition root's stable counter storage, never into an + /// `Entry`: `Pool.init` sorts entries by value, so a pointer taken into one + /// before the sort would dangle onto a different upstream's field. + /// Incremented by a DoT slot client when a stale reused session was + /// recovered by its single redial. Zero forever for DoH entries. + reuse_recoveries: *std.atomic.Value(u64), + + /// Both counters, on either exit of the wait for a permit. A waiter that + /// burned its budget in the queue is the field failure this records, so a + /// cancellation counts exactly like an acquisition. + fn recordQueued(self: *Entry, io: std.Io, started: std.Io.Timestamp) void { + const elapsed = std.Io.Clock.awake.now(io).nanoseconds - started.nanoseconds; + _ = self.queued_total.fetchAdd(1, .monotonic); + _ = self.queued_ns_total.fetchAdd(@intCast(@max(elapsed, 0)), .monotonic); + } + + /// The first slot whose client is free. Called only while holding a permit, + /// where a free slot is guaranteed. + fn takeSlot(self: *Entry) ?*Slot { + for (self.slots) |*slot| { + if (slot.busy.tryLock()) return slot; + } + return null; + } }; /// A copy of one entry's health, taken under the mutex. Feeds `/metrics` and @@ -108,6 +168,16 @@ pub const Snapshot = struct { /// Borrowed from the entry; valid until that entry's next failure. last_error: []const u8, backoff_until: ?std.Io.Timestamp, + /// Exchanges in flight against this entry at the instant of the read, and + /// the ceiling they cannot cross (`slots.len`). `peak_in_flight` stays out: + /// it is a test assertion, not an operator's number. + in_flight: u32, + slots: u32, + /// Admission samples; see `Entry.queued_total`. Seconds rather than + /// nanoseconds because Prometheus counts time in seconds. + queued_total: u64, + queued_seconds_total: f64, + reuse_recoveries_total: u64, }; /// The two budgets, named rather than positional: they are the same type, so @@ -143,6 +213,10 @@ pub const Pool = struct { seed: u64, ) Pool { std.debug.assert(entries.len > 0); + for (entries) |*entry| { + std.debug.assert(entry.slots.len >= 1); + std.debug.assert(entry.sem.permits == entry.slots.len); + } // Stable, so entries sharing a priority keep their configured order. std.mem.sort(Entry, entries, {}, byPriority); return .{ @@ -178,10 +252,11 @@ pub const Pool = struct { /// success; on error the buffer's contents are undefined. /// /// The failover loop runs raced against `timeouts.total`. Losing that race - /// cancels the loop, which unwinds whatever attempt was in flight: the - /// `Entry.busy` lock is taken cancelably on purpose and released by defer, + /// cancels the loop, which unwinds whatever attempt was in flight: the wait + /// for a slot is cancelable on purpose and every slot is released by defer, /// and the health bookkeeping around a completed exchange is uncancelable, - /// so a canceled attempt leaves no lock held and no counter half-written. + /// so a canceled attempt leaves no lock held, no permit lost and no counter + /// half-written. pub fn exchange( self: *Pool, io: std.Io, @@ -216,21 +291,48 @@ pub const Pool = struct { if (!entry.enabled) continue; if (pass == 0 and !self.entryAvailable(io, entry, now)) continue; + // Sampled before the wait, so a task that is about to queue is + // counted as queued whichever way the wait ends. The read races + // the releasers by one slot on purpose; see `Entry`. + const queued_at: ?std.Io.Timestamp = if (entry.in_flight.load(.acquire) >= entry.slots.len) + std.Io.Clock.awake.now(io) + else + null; + // Cancelable, unlike the health-bookkeeping locks below: a task - // waiting its turn on a busy upstream has done nothing that a - // cancellation could corrupt, so it gives up here rather than - // queueing behind an exchange it will not use. The wait itself - // is bounded by the holder's `timeouts.attempt`; the waiter's - // own budget only starts once it has the lock, and the whole - // loop is bounded by `timeouts.total` regardless. - try entry.busy.lock(io); - defer entry.busy.unlock(io); + // waiting its turn on a saturated upstream has done nothing that + // a cancellation could corrupt, so it gives up here rather than + // queueing behind exchanges it will not use. `wait` either + // returns holding a permit or returns `error.Canceled` holding + // none. The wait itself is bounded by the holders' + // `timeouts.attempt`; the waiter's own budget only starts once + // it has a slot, and the whole loop is bounded by + // `timeouts.total` regardless. + entry.sem.wait(io) catch |err| { + if (queued_at) |started| entry.recordQueued(io, started); + return err; + }; + if (queued_at) |started| entry.recordQueued(io, started); + + // A permit means a slot's mutex is free, so failing to find one + // is a protocol bug in this loop, not a runtime condition. + const slot = entry.takeSlot() orelse unreachable; + const entrants = entry.in_flight.fetchAdd(1, .acq_rel) + 1; + _ = entry.peak_in_flight.fetchMax(entrants, .acq_rel); + defer { + slot.busy.unlock(io); + _ = entry.in_flight.fetchSub(1, .acq_rel); + // Uncancelable, so a canceled attempt still returns its + // permit. + entry.sem.post(io); + } // The check above ran before the wait, and the attempts this one // queued behind may have failed the entry into backoff while it // waited. Pass one must not touch an entry that is unavailable // now, so re-read health against a fresh `now` and move on if it - // is. Pass two skips this on purpose: it probes regardless of + // is — the defer above releases the slot before the `continue`. + // Pass two skips this on purpose: it probes regardless of // backoff, which is how backoff recovers. if (pass == 0) { const recheck = std.Io.Clock.awake.now(io); @@ -245,7 +347,7 @@ pub const Pool = struct { // the pool, so the borrow stays valid past this loop. selected.* = entry.endpoint.url; - const result = self.attempt(io, entry.client, query, response_buf); + const result = self.attempt(io, slot.client, query, response_buf); const completed_at = std.Io.Clock.awake.now(io); const response = result catch |err| switch (transport.group(err)) { @@ -293,6 +395,11 @@ pub const Pool = struct { .last_error_at = entry.health.last_error_at, .last_error = entry.health.lastError(), .backoff_until = entry.health.backoff_until, + .in_flight = entry.in_flight.load(.acquire), + .slots = @intCast(entry.slots.len), + .queued_total = entry.queued_total.load(.acquire), + .queued_seconds_total = @as(f64, @floatFromInt(entry.queued_ns_total.load(.acquire))) / std.time.ns_per_s, + .reuse_recoveries_total = entry.reuse_recoveries.load(.acquire), }; } return count; @@ -419,13 +526,20 @@ const Fake = struct { /// Replaces `behavior` after the first call. One entry that fails the task /// which reaches it first and answers the next is what makes two concurrent /// exchanges end on different entries; a single behaviour cannot say that. - /// Mutated under the entry's `busy` lock, like `calls`. then: ?Behavior = null, - calls: usize = 0, + /// Guards `behavior` and `then`. One fake now backs every slot of an entry, + /// so the swap runs under as many tasks as the entry has slots. + mutex: std.Io.Mutex = .init, + calls: std.atomic.Value(usize) = .init(0), in_flight: std.atomic.Value(u32) = .init(0), - /// The most tasks ever inside `exchangeFn` at once. The per-entry lock is - /// only doing its job while this stays at 1. + /// The most tasks ever inside `exchangeFn` at once. The entry's slot count + /// is the ceiling this must never cross. peak_in_flight: std.atomic.Value(u32) = .init(0), + /// Slot storage for the entry this fake backs, and the recovery counter + /// that entry points at. Both live here so one declared `Fake` is one whole + /// test upstream; `testEntry` wires as many of the slots as a test wants. + slots: [slots_per_entry]Slot = undefined, + recoveries: std.atomic.Value(u64) = .init(0), const Behavior = union(enum) { /// Copy these bytes into the caller's buffer and return them. @@ -455,12 +569,17 @@ const Fake = struct { defer _ = self.in_flight.fetchSub(1, .acq_rel); _ = self.peak_in_flight.fetchMax(entrants, .acq_rel); - self.calls += 1; - const behavior = self.behavior; - if (self.then) |next| { - self.behavior = next; - self.then = null; - } + _ = self.calls.fetchAdd(1, .acq_rel); + const behavior = behavior: { + self.mutex.lockUncancelable(io); + defer self.mutex.unlock(io); + const current = self.behavior; + if (self.then) |next| { + self.behavior = next; + self.then = null; + } + break :behavior current; + }; switch (behavior) { .reply => |bytes| return copy(bytes, response_buf), .fail => |err| return err, @@ -487,12 +606,25 @@ const Fake = struct { }; fn testEntry(url: []const u8, fake: *Fake, priority: i32) Entry { + return testEntrySlots(url, fake, priority, 1); +} + +/// An entry whose `slot_count` slots all run through one fake, borrowing that +/// fake's slot storage and recovery counter. One slot unless a test is about +/// concurrency: the assertions below read the fake's own counters, and one fake +/// per entry is what makes those readable. +fn testEntrySlots(url: []const u8, fake: *Fake, priority: i32, slot_count: usize) Entry { + std.debug.assert(slot_count >= 1 and slot_count <= fake.slots.len); + const slots = fake.slots[0..slot_count]; + for (slots) |*slot| slot.* = .{ .client = fake.client() }; return .{ .endpoint = Endpoint.parse(url) catch unreachable, - .client = fake.client(), + .slots = slots, .priority = priority, .enabled = true, .health = .init, + .sem = .{ .permits = slot_count }, + .reuse_recoveries = &fake.recoveries, }; } @@ -636,7 +768,7 @@ test "a timeout mid-flight reports the endpoint the query was in" { var selected: ?[]const u8 = null; try testing.expectError(error.Timeout, pool.exchange(io, query_bytes, &buf, &selected)); try testing.expectEqualStrings("https://stalling.example/dns-query", selected.?); - try testing.expectEqual(@as(usize, 0), untouched.calls); + try testing.expectEqual(@as(usize, 0), untouched.calls.load(.acquire)); } test "an exchange that attempted nothing reports no endpoint" { @@ -674,8 +806,8 @@ test "entries are tried in ascending priority order" { var buf: [512]u8 = undefined; var selected: ?[]const u8 = null; _ = try pool.exchange(io, query_bytes, &buf, &selected); - try testing.expectEqual(@as(usize, 1), low.calls); - try testing.expectEqual(@as(usize, 0), high.calls); + try testing.expectEqual(@as(usize, 1), low.calls.load(.acquire)); + try testing.expectEqual(@as(usize, 0), high.calls.load(.acquire)); } test "a peer fault fails over to the next entry and is recorded" { @@ -720,12 +852,12 @@ test "an entry in backoff is skipped while another is available" { var selected: ?[]const u8 = null; _ = try pool.exchange(io, query_bytes, &buf, &selected); _ = try pool.exchange(io, query_bytes, &buf, &selected); - try testing.expectEqual(@as(usize, 2), bad.calls); + try testing.expectEqual(@as(usize, 2), bad.calls.load(.acquire)); try testing.expect(entries[0].health.backoff_until != null); _ = try pool.exchange(io, query_bytes, &buf, &selected); - try testing.expectEqual(@as(usize, 2), bad.calls); - try testing.expectEqual(@as(usize, 3), good.calls); + try testing.expectEqual(@as(usize, 2), bad.calls.load(.acquire)); + try testing.expectEqual(@as(usize, 3), good.calls.load(.acquire)); } test "every entry in backoff is still probed" { @@ -750,8 +882,8 @@ test "every entry in backoff is still probed" { // Pass one now has no candidate at all. Pass two probes both anyway. try testing.expectError(error.BadResponse, pool.exchange(io, query_bytes, &buf, &selected)); - try testing.expectEqual(@as(usize, 3), first.calls); - try testing.expectEqual(@as(usize, 3), second.calls); + try testing.expectEqual(@as(usize, 3), first.calls.load(.acquire)); + try testing.expectEqual(@as(usize, 3), second.calls.load(.acquire)); } test "a local resource error short-circuits and records nothing" { @@ -770,7 +902,7 @@ test "a local resource error short-circuits and records nothing" { var buf: [512]u8 = undefined; var selected: ?[]const u8 = null; try testing.expectError(error.OutOfMemory, pool.exchange(io, query_bytes, &buf, &selected)); - try testing.expectEqual(@as(usize, 0), good.calls); + try testing.expectEqual(@as(usize, 0), good.calls.load(.acquire)); try testing.expectEqual(@as(u64, 0), entries[0].health.total_failures); try testing.expectEqual(@as(u32, 0), entries[0].health.consecutive_failures); } @@ -791,7 +923,7 @@ test "a cancellation short-circuits and records nothing" { var buf: [512]u8 = undefined; var selected: ?[]const u8 = null; try testing.expectError(error.Canceled, pool.exchange(io, query_bytes, &buf, &selected)); - try testing.expectEqual(@as(usize, 0), good.calls); + try testing.expectEqual(@as(usize, 0), good.calls.load(.acquire)); try testing.expectEqual(@as(u64, 0), entries[0].health.total_failures); } @@ -821,7 +953,7 @@ test "an attempt that outruns the budget is a recorded Timeout" { const reply = try pool.exchange(io, query_bytes, &buf, &selected); try testing.expectEqualSlices(u8, response_bytes, reply); - try testing.expectEqual(@as(usize, 1), slow.calls); + try testing.expectEqual(@as(usize, 1), slow.calls.load(.acquire)); try testing.expectEqual(@as(u32, 1), entries[0].health.consecutive_failures); try testing.expectEqualStrings("Timeout", entries[0].health.lastError()); try testing.expectEqual(@as(u64, 1), entries[1].health.total_successes); @@ -865,7 +997,7 @@ test "two stalling upstreams cost the total budget, not one budget each" { try testing.expect(elapsed_ns < @as(i96, 200) * std.time.ns_per_ms); // The second entry was never reached: the loop was canceled inside the // first attempt. - try testing.expectEqual(@as(usize, 0), second.calls); + try testing.expectEqual(@as(usize, 0), second.calls.load(.acquire)); } test "every entry disabled yields ConnectFailed without waiting out the total budget" { @@ -895,8 +1027,8 @@ test "every entry disabled yields ConnectFailed without waiting out the total bu const elapsed_ns = std.Io.Clock.awake.now(io).nanoseconds - started.nanoseconds; try testing.expect(elapsed_ns < @as(i96, 5) * std.time.ns_per_s); - try testing.expectEqual(@as(usize, 0), one.calls); - try testing.expectEqual(@as(usize, 0), two.calls); + try testing.expectEqual(@as(usize, 0), one.calls.load(.acquire)); + try testing.expectEqual(@as(usize, 0), two.calls.load(.acquire)); } test "snapshot reports the counters in pool order" { @@ -939,6 +1071,15 @@ test "snapshot reports the counters in pool order" { try testing.expect(out[1].last_success_at != null); try testing.expectEqual(@as(?std.Io.Timestamp, null), out[1].backoff_until); + // The concurrency fields: nothing is in flight once both exchanges have + // returned, the ceiling is the entry's slot count, and an uncontended pool + // queued nobody and recovered no session. + try testing.expectEqual(@as(u32, 0), out[1].in_flight); + try testing.expectEqual(@as(u32, 1), out[1].slots); + try testing.expectEqual(@as(u64, 0), out[1].queued_total); + try testing.expectEqual(@as(f64, 0), out[1].queued_seconds_total); + try testing.expectEqual(@as(u64, 0), out[1].reuse_recoveries_total); + // A short `out` truncates rather than overflowing. var one: [1]Snapshot = undefined; try testing.expectEqual(@as(usize, 1), try pool.snapshot(io, &one)); @@ -970,18 +1111,117 @@ fn exchangeAttributed(pool: *Pool, io: std.Io, buf: []u8) transport.ExchangeErro return .{ .reply_len = reply.len, .selected = selected }; } -test "concurrent exchanges through one entry do not overlap" { +/// Blocks until `count` tasks are inside `fake.exchangeFn` at once. +/// +/// `io.concurrent` does not promise the task it spawns has started, let alone +/// reached the leaf client, so a sleep proves nothing about who holds an entry's +/// permits. The queued counters are admission samples, and a test about them has +/// to *know* the entry is occupied before it starts the task that queues — +/// otherwise the waiter may be the one that wins the permit and the test passes +/// on the wrong interleaving. Bounded so a fake that never runs fails the test +/// instead of hanging the suite. +fn awaitInFlight(io: std.Io, fake: *Fake, count: u32) !void { + const step: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(1), .clock = .awake }; + const steps = 10_000; + for (0..steps) |_| { + if (fake.in_flight.load(.acquire) >= count) return; + try step.sleep(io); + } + return error.FakeNeverEnteredExchange; +} + +/// The stall one test's slow fake sleeps for. Long enough that two tasks +/// serialized through one slot would take twice it, short enough that a test +/// suite still ends promptly. +const overlap_stall_ms = 100; + +test "concurrent exchanges through one entry overlap" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var fake: Fake = .{ .behavior = .{ .slow = .{ + .duration = .{ .raw = .fromMilliseconds(overlap_stall_ms), .clock = .awake }, + .reply = response_bytes, + } } }; + var entries = [_]Entry{testEntrySlots("https://only.example/dns-query", &fake, 10, 2)}; + var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1); + + var buf_a: [512]u8 = undefined; + var buf_b: [512]u8 = undefined; + + const started = std.Io.Clock.awake.now(io); + var first = io.concurrent(exchangeAttributed, .{ &pool, io, &buf_a }) catch |err| switch (err) { + error.ConcurrencyUnavailable => return error.SkipZigTest, + }; + defer _ = first.await(io) catch Attributed.discarded; + var second = io.concurrent(exchangeAttributed, .{ &pool, io, &buf_b }) catch |err| switch (err) { + error.ConcurrencyUnavailable => return error.SkipZigTest, + }; + defer _ = second.await(io) catch Attributed.discarded; + + const result_a = try first.await(io); + const result_b = try second.await(io); + const elapsed_ns = std.Io.Clock.awake.now(io).nanoseconds - started.nanoseconds; + + try testing.expectEqualSlices(u8, response_bytes, buf_a[0..result_a.reply_len]); + try testing.expectEqualSlices(u8, response_bytes, buf_b[0..result_b.reply_len]); + try testing.expectEqualStrings("https://only.example/dns-query", result_a.selected.?); + try testing.expectEqualStrings("https://only.example/dns-query", result_b.selected.?); + try testing.expectEqual(@as(usize, 2), fake.calls.load(.acquire)); + // The defect this milestone fixes: both tasks were inside the fake at once, + // and the pair cost one stall rather than two. + try testing.expectEqual(@as(u32, 2), fake.peak_in_flight.load(.acquire)); + try testing.expect(elapsed_ns < @as(i96, 2 * overlap_stall_ms) * std.time.ns_per_ms); + try testing.expectEqual(@as(u64, 2), entries[0].health.total_successes); + // Neither task found the entry saturated, so neither queued. + try testing.expectEqual(@as(u64, 0), entries[0].queued_total.load(.acquire)); +} + +test "an entry never runs more exchanges at once than it has slots" { var threaded: std.Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); - // Slow enough that an unserialized second task would still be inside the - // fake when the first one is, and short enough to stay far under the - // 10-second attempt budget even when the two run back to back. var fake: Fake = .{ .behavior = .{ .slow = .{ .duration = .{ .raw = .fromMilliseconds(50), .clock = .awake }, .reply = response_bytes, } } }; + var entries = [_]Entry{testEntrySlots("https://only.example/dns-query", &fake, 10, 2)}; + var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1); + + var bufs: [3][512]u8 = undefined; + var calls: [3]std.Io.Future(transport.ExchangeError!Attributed) = undefined; + for (&calls, &bufs) |*call, *buf| { + call.* = io.concurrent(exchangeAttributed, .{ &pool, io, buf }) catch |err| switch (err) { + error.ConcurrencyUnavailable => return error.SkipZigTest, + }; + } + // Every future is awaited before the first assertion: a failed assertion + // between two awaits would leave a task running past the end of the test. + var results: [3]transport.ExchangeError!Attributed = undefined; + for (&calls, &results) |*call, *result| result.* = call.await(io); + for (results, &bufs) |result, *buf| { + const value = try result; + try testing.expectEqualSlices(u8, response_bytes, buf[0..value.reply_len]); + } + + try testing.expectEqual(@as(usize, 3), fake.calls.load(.acquire)); + // The third task waited for a permit rather than entering a third client: + // the slot count, not the task count, is the ceiling. + try testing.expectEqual(@as(u32, 2), fake.peak_in_flight.load(.acquire)); + try testing.expectEqual(@as(u32, 2), entries[0].peak_in_flight.load(.acquire)); +} + +test "a task that waits for a saturated entry is counted as queued" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var fake: Fake = .{ .behavior = .{ .slow = .{ + .duration = .{ .raw = .fromMilliseconds(overlap_stall_ms), .clock = .awake }, + .reply = response_bytes, + } } }; var entries = [_]Entry{testEntry("https://only.example/dns-query", &fake, 10)}; var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1); @@ -992,21 +1232,99 @@ test "concurrent exchanges through one entry do not overlap" { error.ConcurrencyUnavailable => return error.SkipZigTest, }; defer _ = first.await(io) catch Attributed.discarded; + // The counter is an admission sample, so the second task has to find the + // entry already occupied. Waiting for the first task to be inside the fake + // is what makes that true of every run. + try awaitInFlight(io, &fake, 1); var second = io.concurrent(exchangeAttributed, .{ &pool, io, &buf_b }) catch |err| switch (err) { error.ConcurrencyUnavailable => return error.SkipZigTest, }; defer _ = second.await(io) catch Attributed.discarded; - const result_a = try first.await(io); - const result_b = try second.await(io); + _ = try first.await(io); + _ = try second.await(io); - try testing.expectEqualSlices(u8, response_bytes, buf_a[0..result_a.reply_len]); - try testing.expectEqualSlices(u8, response_bytes, buf_b[0..result_b.reply_len]); - try testing.expectEqualStrings("https://only.example/dns-query", result_a.selected.?); - try testing.expectEqualStrings("https://only.example/dns-query", result_b.selected.?); - try testing.expectEqual(@as(usize, 2), fake.calls); - try testing.expectEqual(@as(u32, 1), fake.peak_in_flight.load(.acquire)); - try testing.expectEqual(@as(u64, 2), entries[0].health.total_successes); + try testing.expectEqual(@as(u64, 1), entries[0].queued_total.load(.acquire)); + try testing.expect(entries[0].queued_ns_total.load(.acquire) > 0); + + // An uncontended call adds neither. + const queued_ns = entries[0].queued_ns_total.load(.acquire); + fake.behavior = .{ .reply = response_bytes }; + var selected: ?[]const u8 = null; + _ = try pool.exchange(io, query_bytes, &buf_a, &selected); + try testing.expectEqual(@as(u64, 1), entries[0].queued_total.load(.acquire)); + try testing.expectEqual(queued_ns, entries[0].queued_ns_total.load(.acquire)); +} + +/// The stall the holder of the entry's one permit sits in below, and the whole +/// budget the task queued behind it gets. The gap between them is the test's +/// whole argument: the holder is still inside the fake long after the waiter's +/// budget is gone, so the waiter can only have died waiting for a permit. +const holder_stall_ms = 600; +const waiter_total_ms = 100; + +test "a waiter canceled by the total budget is counted and still returns its permit" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var fake: Fake = .{ .behavior = .{ .slow = .{ + .duration = .{ .raw = .fromMilliseconds(holder_stall_ms), .clock = .awake }, + .reply = response_bytes, + } } }; + var entries = [_]Entry{testEntry("https://only.example/dns-query", &fake, 10)}; + + // Two pools over one `entries` slice. A budget is a property of the pool, + // not of a call, and this test needs two: a generous one for the task that + // holds the permit and a short one for the task that queues behind it. A + // `Pool` borrows `entries`, so both see the same semaphore, the same queue + // counters and the same health; only `mutex` and the jitter RNG are + // per-pool. Sharing health across two pool mutexes is sound here because + // nothing writes it concurrently: the waiter dies before it attempts + // anything, and the holder's success is written before the last call runs. + var holder_pool: Pool = .init(&entries, test_cfg, test_timeouts, 1); + var waiter_pool: Pool = .init(&entries, test_cfg, .{ + .attempt = .{ .raw = .fromSeconds(30), .clock = .awake }, + .total = .{ .raw = .fromMilliseconds(waiter_total_ms), .clock = .awake }, + }, 1); + + var buf_a: [512]u8 = undefined; + var buf_b: [512]u8 = undefined; + + var holder = io.concurrent(exchangeAttributed, .{ &holder_pool, io, &buf_a }) catch |err| switch (err) { + error.ConcurrencyUnavailable => return error.SkipZigTest, + }; + defer _ = holder.await(io) catch Attributed.discarded; + // The whole test rests on the holder owning the entry's one permit before + // the waiter asks for one. `io.concurrent` does not order those two, so the + // wait is on the fake itself: once a task is inside `exchangeFn` the permit + // is provably taken, and the call below can only queue. + try awaitInFlight(io, &fake, 1); + + // This call never reaches the fake: it waits for the entry's one permit + // until its own total budget expires. That is the field failure — a burst + // dying in the queue — and it has to be visible. + var selected: ?[]const u8 = null; + try testing.expectError(error.Timeout, waiter_pool.exchange(io, query_bytes, &buf_b, &selected)); + + // Where it died, not just that it died: the holder's call is still the only + // one any leaf client has seen, so the cancellation landed inside + // `Semaphore.wait` rather than inside an attempt. + try testing.expectEqual(@as(usize, 1), fake.calls.load(.acquire)); + try testing.expectEqual(@as(u64, 1), entries[0].queued_total.load(.acquire)); + try testing.expect(entries[0].queued_ns_total.load(.acquire) > 0); + + // The holder's own budget is generous, so it answers rather than expiring: + // the permit comes back from a completed attempt. + const held = try holder.await(io); + try testing.expectEqualSlices(u8, response_bytes, buf_a[0..held.reply_len]); + + // The canceled waiter returned the permit it never held, so the entry is + // usable again — and the fresh call is the second one to reach the fake. + fake.behavior = .{ .reply = response_bytes }; + const reply = try waiter_pool.exchange(io, query_bytes, &buf_b, &selected); + try testing.expectEqualSlices(u8, response_bytes, reply); + try testing.expectEqual(@as(usize, 2), fake.calls.load(.acquire)); } /// The two entries of the test below, each answering with bytes only it @@ -1076,8 +1394,8 @@ test "overlapping exchanges each report the entry that answered that call" { try expectAnsweredByReporter(result_a, &buf_a); try expectAnsweredByReporter(result_b, &buf_b); - try testing.expectEqual(@as(usize, 2), first_entry.calls); - try testing.expectEqual(@as(usize, 1), second_entry.calls); + try testing.expectEqual(@as(usize, 2), first_entry.calls.load(.acquire)); + try testing.expectEqual(@as(usize, 1), second_entry.calls.load(.acquire)); try testing.expectEqual(@as(u32, 1), first_entry.peak_in_flight.load(.acquire)); try testing.expectEqual(@as(u64, 1), entries[0].health.total_failures); try testing.expectEqual(@as(u64, 1), entries[0].health.total_successes); @@ -1131,11 +1449,11 @@ test "an entry that enters backoff while a task waits on it is not attempted" { try testing.expectEqualSlices(u8, response_bytes, buf_b[0..result_b.reply_len]); try testing.expectEqualStrings("https://good.example/dns-query", result_a.selected.?); try testing.expectEqualStrings("https://good.example/dns-query", result_b.selected.?); - try testing.expectEqual(@as(usize, 2), good.calls); + try testing.expectEqual(@as(usize, 2), good.calls.load(.acquire)); // The point of the test: the entry was attempted once, not twice. Without // the re-check the second task would take the lock and attempt it anyway. - try testing.expectEqual(@as(usize, 1), slow_bad.calls); + try testing.expectEqual(@as(usize, 1), slow_bad.calls.load(.acquire)); try testing.expectEqual(@as(u64, 1), entries[0].health.total_failures); try testing.expect(entries[0].health.backoff_until != null); } diff --git a/src/web/metrics.zig b/src/web/metrics.zig index a25c40d..84b7d88 100644 --- a/src/web/metrics.zig +++ b/src/web/metrics.zig @@ -113,6 +113,11 @@ pub const UpstreamSample = struct { total_successes: u64, total_failures: u64, success_rate: f32, + in_flight: u32, + slots: u32, + queued_total: u64, + queued_seconds_total: f64, + reuse_recoveries_total: u64, }; /// The diagnostics store, as one scrape sees it. Two gauges and a counter, @@ -318,6 +323,11 @@ fn upstreams(pool: *pool_mod.Pool, io: std.Io, arena: Allocator) Allocator.Error .total_successes = entry.total_successes, .total_failures = entry.total_failures, .success_rate = entry.success_rate, + .in_flight = entry.in_flight, + .slots = entry.slots, + .queued_total = entry.queued_total, + .queued_seconds_total = entry.queued_seconds_total, + .reuse_recoveries_total = entry.reuse_recoveries_total, }; } return out; @@ -510,6 +520,47 @@ fn renderUpstreams(w: *std.Io.Writer, list: []const UpstreamSample) std.Io.Write for (list, 0..) |entry, i| { try labeledValue(w, "nxdns_upstream_failures_total", i, entry.url, entry.total_failures); } + + try labeledHead(w, "nxdns_upstream_in_flight", "Exchanges in flight against an upstream.", "gauge"); + for (list, 0..) |entry, i| { + try labeledValue(w, "nxdns_upstream_in_flight", i, entry.url, entry.in_flight); + } + + try labeledHead(w, "nxdns_upstream_slots", "Concurrent exchanges an upstream allows.", "gauge"); + for (list, 0..) |entry, i| { + try labeledValue(w, "nxdns_upstream_slots", i, entry.url, entry.slots); + } + + try labeledHead( + w, + "nxdns_upstream_queued_total", + "Exchanges that waited for a slot. Approximate; sampled at admission.", + "counter", + ); + for (list, 0..) |entry, i| { + try labeledValue(w, "nxdns_upstream_queued_total", i, entry.url, entry.queued_total); + } + + try labeledHead( + w, + "nxdns_upstream_queued_seconds_total", + "Seconds exchanges spent waiting for a slot. Approximate; sampled at admission.", + "counter", + ); + for (list, 0..) |entry, i| { + try writeUpstreamLabels(w, "nxdns_upstream_queued_seconds_total", i, entry.url); + try w.print(" {d:.4}\n", .{entry.queued_seconds_total}); + } + + try labeledHead( + w, + "nxdns_upstream_reuse_recoveries_total", + "Stale reused DoT connections recovered by a redial.", + "counter", + ); + for (list, 0..) |entry, i| { + try labeledValue(w, "nxdns_upstream_reuse_recoveries_total", i, entry.url, entry.reuse_recoveries_total); + } } /// Every field of a plain counter struct, under one prefix. @@ -553,7 +604,7 @@ fn labeledValue( } /// The label set every upstream family shares, up to and including the closing -/// brace. One definition, because six families have to agree on it exactly: +/// brace. One definition, because eleven families have to agree on it exactly: /// Prometheus identifies a series by its name and its whole label set, so a /// family that labelled its samples differently would be a different series. /// @@ -651,6 +702,7 @@ const db = @import("../storage/db.zig"); const local_tables = @import("../server/local_tables.zig"); const logger_mod = @import("../storage/logger.zig"); const migrations = @import("../storage/migrations.zig"); +const transport = @import("../upstream/transport.zig"); const testing = std.testing; /// A handler with no upstream reachable: every test here reads counters and @@ -684,6 +736,11 @@ test "a full sample renders the whole exposition, byte for byte" { .total_successes = 9, .total_failures = 1, .success_rate = 0.9, + .in_flight = 2, + .slots = 8, + .queued_total = 3, + .queued_seconds_total = 0.25, + .reuse_recoveries_total = 1, }, }; @@ -760,10 +817,16 @@ test "a full sample renders the whole exposition, byte for byte" { 1, "nxdns_upstream_success_rate{index=\"0\",url=\"https://dns.example\"} 0.9000\n", )); + try testing.expect(std.mem.containsAtLeast( + u8, + text, + 1, + "nxdns_upstream_failures_total{index=\"0\",url=\"https://dns.example\"} 1\n", + )); try testing.expect(std.mem.endsWith( u8, text, - "nxdns_upstream_failures_total{index=\"0\",url=\"https://dns.example\"} 1\n", + "nxdns_upstream_reuse_recoveries_total{index=\"0\",url=\"https://dns.example\"} 1\n", )); } @@ -1115,12 +1178,18 @@ test "a label value escapes the characters the format reserves" { .total_successes = 0, .total_failures = 2, .success_rate = 0, + .in_flight = 1, + .slots = 8, + .queued_total = 4, + .queued_seconds_total = 1.5, + .reuse_recoveries_total = 2, }}; const text = try renderToString(testing.allocator, .{ .upstreams = &upstream_list }); defer testing.allocator.free(text); // The url wrote no line of its own, and lost none: every line is a - // comment or a sample, and the six families contribute six samples. + // comment or a sample, and the eleven families contribute eleven + // samples. var samples: usize = 0; var lines = std.mem.splitScalar(u8, text, '\n'); while (lines.next()) |line| { @@ -1131,7 +1200,7 @@ test "a label value escapes the characters the format reserves" { // Both labels are there, and both values close where they opened. try testing.expectEqual(@as(?usize, 2), labelPairs(line)); } - try testing.expectEqual(@as(usize, 6), samples); + try testing.expectEqual(@as(usize, 11), samples); } } @@ -1149,6 +1218,11 @@ test "two upstreams on one host stay two series" { .total_successes = 5, .total_failures = 0, .success_rate = 1, + .in_flight = 0, + .slots = 8, + .queued_total = 0, + .queued_seconds_total = 0, + .reuse_recoveries_total = 0, }, .{ .url = "https://dns.nextdns.io/efgh34", @@ -1158,6 +1232,11 @@ test "two upstreams on one host stay two series" { .total_successes = 9, .total_failures = 3, .success_rate = 0.75, + .in_flight = 3, + .slots = 8, + .queued_total = 7, + .queued_seconds_total = 2, + .reuse_recoveries_total = 5, }, }; const text = try renderToString(testing.allocator, .{ .upstreams = &upstream_list }); @@ -1206,7 +1285,7 @@ test "two upstreams on one host stay two series" { try testing.expectEqualStrings("0", up_values[1]); // No two samples in the scrape share a series key, whatever the urls were. - var keys: [32][]const u8 = undefined; + var keys: [64][]const u8 = undefined; var count: usize = 0; var lines = std.mem.splitScalar(u8, text, '\n'); while (lines.next()) |line| { @@ -1216,7 +1295,7 @@ test "two upstreams on one host stay two series" { keys[count] = key; count += 1; } - try testing.expectEqual(@as(usize, 12), count); + try testing.expectEqual(@as(usize, 22), count); } test "an upstream url is redacted before it reaches an open endpoint's label" { @@ -1233,6 +1312,11 @@ test "an upstream url is redacted before it reaches an open endpoint's label" { .total_successes = 3, .total_failures = 0, .success_rate = 1, + .in_flight = 0, + .slots = 8, + .queued_total = 0, + .queued_seconds_total = 0, + .reuse_recoveries_total = 0, }, .{ .url = "https://user:hunter2@dns.example:8443/dns-query?apikey=s3cr3t#frag", @@ -1242,6 +1326,11 @@ test "an upstream url is redacted before it reaches an open endpoint's label" { .total_successes = 0, .total_failures = 4, .success_rate = 0, + .in_flight = 0, + .slots = 8, + .queued_total = 0, + .queued_seconds_total = 0, + .reuse_recoveries_total = 0, }, }; const text = try renderToString(testing.allocator, .{ .upstreams = &upstream_list }); @@ -1254,7 +1343,8 @@ test "an upstream url is redacted before it reaches an open endpoint's label" { try testing.expect(!std.mem.containsAtLeast(u8, text, 1, "frag")); try testing.expect(!std.mem.containsAtLeast(u8, text, 1, "dns-query")); - // Every family carries the label, so none of the six may keep the whole url. + // Every family carries the label, so none of the eleven may keep the whole + // url. try testing.expect(std.mem.containsAtLeast( u8, text, @@ -1302,6 +1392,54 @@ test "an upstream url is redacted before it reaches an open endpoint's label" { )); } +test "the queue families render with the label set every upstream family shares" { + // The surface the Pi burst defect is read from: whether an upstream is + // queueing (`queued_total`/`in_flight` against `slots`) and whether it is + // churning connections (`reuse_recoveries_total`). + const upstream_list = [_]UpstreamSample{.{ + .url = "tls://dns.example:853", + .enabled = true, + .available = true, + .consecutive_failures = 0, + .total_successes = 30, + .total_failures = 0, + .success_rate = 1, + .in_flight = 6, + .slots = 8, + .queued_total = 22, + .queued_seconds_total = 1.5, + .reuse_recoveries_total = 3, + }}; + const text = try renderToString(testing.allocator, .{ .upstreams = &upstream_list }); + defer testing.allocator.free(text); + + const label = "{index=\"0\",url=\"tls://dns.example:853\"}"; + try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_upstream_in_flight" ++ label ++ " 6\n")); + try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_upstream_slots" ++ label ++ " 8\n")); + try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_upstream_queued_total" ++ label ++ " 22\n")); + try testing.expect(std.mem.containsAtLeast( + u8, + text, + 1, + "nxdns_upstream_queued_seconds_total" ++ label ++ " 1.5000\n", + )); + try testing.expect(std.mem.containsAtLeast( + u8, + text, + 1, + "nxdns_upstream_reuse_recoveries_total" ++ label ++ " 3\n", + )); + + // The types a scrape reads them as, and the approximation the two queue + // counters carry stated where an operator meets them. + try testing.expect(std.mem.containsAtLeast(u8, text, 1, "# TYPE nxdns_upstream_in_flight gauge\n")); + try testing.expect(std.mem.containsAtLeast(u8, text, 1, "# TYPE nxdns_upstream_slots gauge\n")); + try testing.expect(std.mem.containsAtLeast(u8, text, 1, "# TYPE nxdns_upstream_queued_total counter\n")); + try testing.expect(std.mem.containsAtLeast(u8, text, 1, "# TYPE nxdns_upstream_queued_seconds_total counter\n")); + try testing.expect(std.mem.containsAtLeast(u8, text, 1, "# TYPE nxdns_upstream_reuse_recoveries_total counter\n")); + try testing.expectEqual(@as(usize, 2), std.mem.count(u8, text, "Approximate; sampled at admission.")); +} + test "a url longer than the redaction bound cannot run past it" { const long_host = "h" ** (4 * safe_url.max_len); const upstream_list = [_]UpstreamSample{.{ @@ -1312,6 +1450,11 @@ test "a url longer than the redaction bound cannot run past it" { .total_successes = 0, .total_failures = 0, .success_rate = 1, + .in_flight = 0, + .slots = 8, + .queued_total = 0, + .queued_seconds_total = 0, + .reuse_recoveries_total = 0, }}; const text = try renderToString(testing.allocator, .{ .upstreams = &upstream_list }); defer testing.allocator.free(text); @@ -1325,6 +1468,115 @@ test "a url longer than the redaction bound cannot run past it" { )); } +/// A leaf client that answers, and counts that it was asked. +/// +/// `pool.zig`'s own fake is private to that file and models failure modes this +/// test has no use for; all this one has to do is let a real `Pool.exchange` +/// complete, so the health half of the sample below is the pool's own +/// bookkeeping rather than a literal. +const AnsweringClient = struct { + calls: std.atomic.Value(u32) = .init(0), + + const reply = "\x12\x34\x81\x80"; + + fn exchangeFn( + ptr: *anyopaque, + io: std.Io, + query: []const u8, + response_buf: []u8, + selected: *?[]const u8, + ) transport.ExchangeError![]u8 { + _ = io; + _ = query; + const self: *AnsweringClient = @ptrCast(@alignCast(ptr)); + _ = self.calls.fetchAdd(1, .acq_rel); + selected.* = "fake://leaf"; + @memcpy(response_buf[0..reply.len], reply); + return response_buf[0..reply.len]; + } + + fn client(self: *AnsweringClient) transport.Client { + return .{ .ptr = self, .exchangeFn = exchangeFn }; + } +}; + +test "the queue families carry what a real pool recorded, through the real snapshot" { + // The test above renders a hand-built `UpstreamSample`, so it proves the + // exposition and nothing else. This one drives a real `Pool` and goes + // through `Pool.snapshot` and `upstreams`, which is where a crossed field + // would live. Every driven value is distinct for that reason: a swap + // anywhere along the path renders the wrong number rather than a match. + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var leaf: AnsweringClient = .{}; + var slots = [_]pool_mod.Slot{ + .{ .client = leaf.client() }, + .{ .client = leaf.client() }, + }; + var recoveries: std.atomic.Value(u64) = .init(0); + var entries = [_]pool_mod.Entry{.{ + .endpoint = try .parse("tls://dns.example:853"), + .slots = &slots, + .priority = 10, + .enabled = true, + .health = .init, + .sem = .{ .permits = slots.len }, + .reuse_recoveries = &recoveries, + }}; + var pool: pool_mod.Pool = .init(&entries, .{}, .{ + .attempt = .{ .raw = .fromSeconds(10), .clock = .awake }, + .total = .{ .raw = .fromSeconds(30), .clock = .awake }, + }, 1); + + var buf: [512]u8 = undefined; + var selected: ?[]const u8 = null; + _ = try pool.exchange(io, "\x12\x34\x01\x00", &buf, &selected); + try testing.expectEqual(@as(u32, 1), leaf.calls.load(.acquire)); + + // The counters a burst would move, written straight into the entry: this + // test is about what the snapshot path carries, not about reproducing a + // queue. The exchange above has already returned, so nothing else is + // touching them. + entries[0].in_flight.store(5, .release); + entries[0].queued_total.store(7, .release); + entries[0].queued_ns_total.store(1_500_000_000, .release); + recoveries.store(9, .release); + + var arena: std.heap.ArenaAllocator = .init(testing.allocator); + defer arena.deinit(); + const list = try upstreams(&pool, io, arena.allocator()); + const text = try renderToString(testing.allocator, .{ .upstreams = list }); + defer testing.allocator.free(text); + + const label = "{index=\"0\",url=\"tls://dns.example:853\"}"; + try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_upstream_in_flight" ++ label ++ " 5\n")); + try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_upstream_slots" ++ label ++ " 2\n")); + try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_upstream_queued_total" ++ label ++ " 7\n")); + try testing.expect(std.mem.containsAtLeast( + u8, + text, + 1, + "nxdns_upstream_queued_seconds_total" ++ label ++ " 1.5000\n", + )); + try testing.expect(std.mem.containsAtLeast( + u8, + text, + 1, + "nxdns_upstream_reuse_recoveries_total" ++ label ++ " 9\n", + )); + + // The one series the pool wrote by itself, so the five above are read + // against an entry the pool really did use. + try testing.expect(std.mem.containsAtLeast( + u8, + text, + 1, + "nxdns_upstream_successes_total" ++ label ++ " 1\n", + )); +} + test "collect reads the live counters of the components it is given" { var threaded: std.Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); diff --git a/src/web/web_integration_test.zig b/src/web/web_integration_test.zig index d32eeb0..232f3ac 100644 --- a/src/web/web_integration_test.zig +++ b/src/web/web_integration_test.zig @@ -313,6 +313,8 @@ const Env = struct { pauser: pause_mod.Pause, tables: local_tables_mod.LocalTables, pool_entries: [1]pool_mod.Entry, + pool_slots: [1]pool_mod.Slot, + pool_recoveries: std.atomic.Value(u64), pool: pool_mod.Pool, state: server.WebState, web: server.Server, @@ -387,14 +389,18 @@ const Env = struct { self.pauser = .{}; self.tables = .empty; + // Never exchanged with: the pool feeds `/metrics` and the + // `/api/health` upstream condition only. + self.pool_slots = .{.{ .client = .{ .ptr = undefined, .exchangeFn = undefined } }}; + self.pool_recoveries = .init(0); self.pool_entries = .{.{ .endpoint = transport.Endpoint.parse("https://dns.example/dns-query") catch unreachable, - // Never exchanged with: the pool feeds `/metrics` and the - // `/api/health` upstream condition only. - .client = .{ .ptr = undefined, .exchangeFn = undefined }, + .slots = &self.pool_slots, .priority = 1, .enabled = true, .health = .init, + .sem = .{ .permits = self.pool_slots.len }, + .reuse_recoveries = &self.pool_recoveries, }}; self.pool = .init(&self.pool_entries, .{}, .{ .attempt = .{ .raw = .fromMilliseconds(50), .clock = .awake },