diff --git a/CHANGELOG.md b/CHANGELOG.md index a82b6eb..b18b9e1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,21 @@ All notable changes to nxdns are recorded here. The format follows [Keep a Chang Sections are written by hand. Nothing here is generated from commit messages: the point of the file is to say what changed for an operator, which a commit subject rarely does. +## [0.0.13] - 2026-08-27 + +The upstream query budget becomes one honest deadline. A busy network no longer blames a healthy standby for running out of time, and a query burst no longer queues invisibly until everything answers SERVFAIL at once. + +### Fixed + +- **The per-query upstream budget is now one absolute deadline, spent by everything that blocks.** Waiting for a free slot on a saturated upstream now spends the query's `upstream.total_timeout_ms` budget just like the exchange itself, instead of being invisible to it — under a burst, queries used to wait out their whole budget in the queue and then start attempts they could never finish. An attempt near the end of the budget runs truncated, and when a truncated attempt runs out of time that is evidence about the budget, not the upstream: it no longer counts against that upstream's health or success rate, and the query log no longer names an upstream that was given no fair chance. A field incident produced 279 rows blaming a standby whose health counters read zero for zero; those rows now attribute nothing. +- **A query no longer blocks behind a saturated upstream while another has capacity.** Admission sweeps the upstreams in priority order and takes the first free slot; priority now means the order among upstreams that can be admitted right now, and a query blocks only when nothing has capacity — on the highest-priority eligible upstream, bounded by the remaining budget. +- **Conditional forward zones spend `upstream.read_timeout_ms` once per query.** A UDP attempt, a truncated answer and the TCP retry now share the one budget instead of taking a fresh one each, so a slow zone resolver can no longer stretch a single query to several times the configured timeout. + +### Added + +- **`nxdns_upstream_budget_exhausted_total`.** A pool-wide counter of queries whose budget ran out — in the queue or mid-attempt — before any upstream answered. It carries no per-upstream label on purpose: running out of budget is a fact about the pool. +- **A configuration with more than 64 enabled upstreams is rejected at validation** with a clear message, instead of tripping an internal limit at startup. + ## [0.0.12] - 2026-08-27 Overview stops re-reading the whole query log. One endpoint, one snapshot, pre-aggregated buckets — a 30-day view now costs the same on a month of history as on a day of it. Read the upgrade note first: it resets your query history. diff --git a/docs/how-to/troubleshoot.md b/docs/how-to/troubleshoot.md index 1629bc6..1abefeb 100644 --- a/docs/how-to/troubleshoot.md +++ b/docs/how-to/troubleshoot.md @@ -267,11 +267,11 @@ cat /etc/resolv.conf 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_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 given up on in the queue and answered SERVFAIL. `nxdns_upstream_budget_exhausted_total` counts those give-ups pool-wide — queries whose own budget ran out, in the queue or mid-attempt, before any upstream answered. It carries no `url` label on purpose: running out of budget is a fact about the pool, so the exhausted attempt is never charged to an upstream's health and is never attributed in the query log, although the row can still name the last endpoint whose success or recorded failure preceded it. `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. +**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 helps only with saturation, not with latency: failover is strict priority, not load spreading. A query does take the next upstream when the first one's slots are all busy — priority orders the candidates that can be admitted right now — but once every eligible upstream is full it blocks on the highest-priority one, and it still sends one query to one upstream at a time. The per-upstream slot count is compiled, not configured, so there is no knob to widen one upstream either. ## The disk is filling up diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index b5d5793..48d9b37 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -37,12 +37,14 @@ Timeouts for talking to upstream resolvers. | Key | Type | Default | Unit | Validation | Consumed by | |---|---|---|---|---|---| | `upstream.attempt_timeout_ms` | u32 | 2500 | ms | 100–120000, and not above `total_timeout_ms` | deadline on one attempt against one upstream inside the pool's failover loop (`src/upstream/pool.zig`), the whole attempt including the connect | -| `upstream.read_timeout_ms` | u32 | 3000 | ms | 100–120000 | read deadline on conditional-forward-zone exchanges (`src/local/forward_client.zig`) | +| `upstream.read_timeout_ms` | u32 | 3000 | ms | 100–120000 | budget for a whole conditional-forward-zone exchange (`src/local/forward_client.zig`): the UDP attempt, a TC=1 fallback and the TCP retry together | | `upstream.total_timeout_ms` | u32 | 5000 | ms | 100–120000 | per-query budget of the upstream pool (`src/upstream/pool.zig`): every failover attempt together, not one of them; also the `nxdns check` probe deadline | The two pool budgets nest. `attempt_timeout_ms` bounds one try against one upstream; when it expires the pool records the failure and moves to the next candidate. `total_timeout_ms` bounds the whole loop, so a query against five unreachable upstreams costs the total budget once, not five attempt budgets in a row. When the total expires the in-flight attempt is canceled and the query fails with a timeout. -`read_timeout_ms` is unrelated to both. It bounds a different subsystem — the conditional-forward-zone client — so no cross-check relates it to the pool's budgets, and it is free to sit above either of them. +Budget semantics. The deadline is an instant, computed once when the query enters the pool, and every blocking step spends against it — waiting for a free slot on an upstream as much as the exchange itself. An attempt therefore runs against `min(now + attempt_timeout_ms, deadline)`, which near the end of the budget is *truncated*: shorter than the configured attempt. Attribution follows from that. A truncated attempt that expires is evidence about the budget, not about the upstream, so it leaves that upstream's health and success rate untouched and the query log's `upstream` field unchanged — the row may still name the endpoint of the preceding attributable attempt, and is null only when there was none — and it increments the pool-wide `nxdns_upstream_budget_exhausted_total` metric. An attempt that expires on its full budget, or that fails outright (a refused connection, a bad answer, the peer's own timeout), is evidence about the upstream: it is recorded against that upstream's health and names it in the query log. Either way the client is answered SERVFAIL. Forward-zone queries are not affected — they have one configured resolver, so a timeout there always names it. + +`read_timeout_ms` is unrelated to both pool budgets. It bounds a different subsystem — the conditional-forward-zone client — so no cross-check relates it to the pool's budgets, and it is free to sit above either of them. Within that subsystem it is one budget for the whole exchange: a query that goes out over UDP, comes back truncated and is retried over TCP has the two legs and the fallback share `read_timeout_ms`, never one each. ### dns @@ -318,7 +320,7 @@ Zones resolved by a specific resolver instead of the configured upstreams, for L | `zone` | string | required | a valid domain name; unique | | `resolver` | string | required | `udp://IP:port` or `tcp://IP:port`; the host must be an IP literal and the port is mandatory | -The resolver host must be an IP literal because resolving the resolver's own name would be a bootstrap problem. Matching is longest suffix (`src/local/forward_zones.zig`); the exchange is UDP then TCP (`src/local/forward_client.zig`) with `upstream.read_timeout_ms` as the read deadline. +The resolver host must be an IP literal because resolving the resolver's own name would be a bootstrap problem. Matching is longest suffix (`src/local/forward_zones.zig`); the exchange is UDP then TCP (`src/local/forward_client.zig`), with `upstream.read_timeout_ms` bounding the whole exchange rather than each leg. Reverse zones are declared the same way, and one is the prerequisite for [learned client names](#learned-names): diff --git a/specs/milestone-37.md b/specs/milestone-37.md new file mode 100644 index 0000000..7541eeb --- /dev/null +++ b/specs/milestone-37.md @@ -0,0 +1,337 @@ +# Milestone 37: Upstream failover budget — deadline ownership and honest attribution + +Make the pool's failover loop own one deadline (admission included), revive the +standby under primary saturation, and stop blaming endpoints for budget +exhaustion. + +## The defect (field-verified on the Pi, 0.0.11) + +With defaults attempt=2500 ms / total=5000 ms and two upstreams: + +1. `Pool.exchangeLoopLen` (pool.zig:277) never computes remaining time. The + semaphore wait consumes the total budget invisibly: under a traffic burst + the primary's slots saturate, a new request queues behind them while the + standby sits idle with free slots, and the outer `raceWithin(total)` cancels + whatever finally runs. 155 requests died at the 5 s cap; 706 of 717 + SERVFAILs came from one bursty client. On an idle pool a fast standby works + today — the queue is the killer, not the timer arithmetic alone. +2. The reporting contradicts itself. `selected` is written before the attempt + runs (pool.zig:348), so the query log blamed the standby on 279 rows; + cancelled attempts are excluded from health (pool.zig:363), so its counters + read 0/0. Nothing records "the request exhausted its own budget." + +Design reviewed and converged with Codex (thread of 2026-08-27). Defaults do +not change. + +## Sessions + +Three, strictly sequential: A (transport vocabulary) → B (pool) → C +(handler, forward client, surfaces). + +--- + +## Session A: transport vocabulary + +### A.1 `error.BudgetExhausted` and a fourth group (src/upstream/transport.zig) + +- Add `BudgetExhausted` to `ExchangeError`. +- Add `.budget_exhausted` to `Group`; `group()` maps the new error to it. The + switch stays exhaustive with no `else` — every switch over `Group` breaks at + compile time until it handles the new group, which is the point. The + production switches are `pool.zig` (:353), `handler.zig` (:677, :735) and + `forward_client.zig` (:124). Session A owns a mechanical placeholder arm at + each (pool and forward_client: propagate the error without recording + health; handler: `=> ctx.servFail()`), plus any test switches the compiler + flags, so A's `zig build test` passes. B and C then own the real behavior + at their sites. (health.zig has no `group` call; the doh/dot occurrences + are tests, not switches.) + +### A.2 Timer-origin race (src/upstream/transport.zig) + +`raceWithin` collapses "the expiry task won" and "the raced operation itself +returned error.Timeout" into one `error.Timeout`. Add: + +```zig +pub const RaceOutcome = enum { completed, expired }; + +pub fn raceUntilTagged( + io: std.Io, + expiry_at: std.Io.Clock.Timestamp, + outcome: *RaceOutcome, + comptime f: anytype, + args: anytype, +) ExchangeError!RacedPayload(f) +``` + +The expiry parameter is an ABSOLUTE timestamp, not a duration: a duration +computed from `deadline.toDurationFromNow()` and then slept re-anchors at +"now", drifting past the total deadline and misclassifying a nominally full +attempt. The pool computes `expiry_at = min(now + attempt, total_deadline)` +and passes the timestamp. On the expiry side the function sets `outcome.* = +.expired` and returns `error.Timeout`; on completion it sets `.completed` and +returns the raced result (which may itself be `error.Timeout` from the leaf — +that is a completed peer timeout, not an expiry). `raceWithin` keeps its +public API and behavior but delegates to the same internal harness (compute +the absolute deadline, discard the outcome) — one `Select` harness in the +file, not two copies. + +### A.3 Acceptance + +- [ ] `zig build test` green. +- [ ] Unit tests: tagged race distinguishes leaf `error.Timeout` (completed) + from expiry (`expired`); the untagged wrapper is unchanged behavior. + +--- + +## Session B: the pool (src/upstream/pool.zig) + +### B.1 One deadline owns the loop + +- `Pool.exchange` establishes `deadline = std.Io.Timeout{ .duration = + self.timeouts.total }.toDeadline(io)` (`.awake` clock, as the loop uses + today) and passes it down. The equal-deadline outer `raceWithin` at + pool.zig:267 is REMOVED — the loop owns the deadline; two timers aimed at + the same instant race each other and let the outer cancellation bypass the + loop's classification. No replacement watchdog (the outer race never bounded + uncancelable health writes anyway; a later-firing watchdog is scope creep). +- Every blocking step consumes the deadline: + - Admission: through the ownership-safe protocol of B.2 — never a bare + `Semaphore.wait` raced against an expiry. Racing the wait with + `Select.cancelDiscard` can leak a permit: the wait may have decremented + the count in the same instant the expiry wins, and the discarded success + never reaches the caller's release-defer. If no time remains before + admission, return `error.BudgetExhausted` without waiting. + - Attempt: raced via `raceUntilTagged` against + `expiry_at = min(now + timeouts.attempt, total_deadline)`. +- `total < attempt` is already rejected by validate.zig; `total == attempt` + (and any admission overhead) simply yields truncated attempts, handled by + B.3. + +### B.2 Admission without head-of-line blocking + +Zig 0.16's `Semaphore` has neither try-acquire nor a timed wait, so the pool +gains a local admission helper mirroring the standard semaphore's own +mutex/decrement/condition protocol (never by patching `../zig`, never by +spinning): + +- `tryAcquire()` — take a permit if one is immediately available, else fail + without blocking. +- `acquireUntil(deadline)` — timed acquisition; returns Acquired (holding a + permit), Expired (holding none), or `error.Canceled` (holding none). + `std.Io.Condition` has no timed wait in 0.16, so this is a NEW pool-local + primitive, specified exactly: state lives under one mutex (permit count + + waiter bookkeeping); the wait itself may race `Condition.wait` against a + sleep via `Select`, because a permit is only ever taken under the mutex + AFTER the race resolves — a discarded wake is a lost notification, not a + lost permit. To keep that lost notification from stranding another waiter, + an exiting waiter that may have absorbed a signal (expiry or cancellation + path) re-signals the condition before returning. Permit conservation is by + construction: the decrement and the "did I win" decision happen under the + same mutex. Every take is non-blocking (the mutex's `tryLock`): a lock + miss reads as "no permit now", and `acquireUntil` re-enters the + absolute-deadline race on each miss, so contention on the admission mutex + never carries a call past its deadline (review round 2026-08-27). Tests: + an expiry/acquisition tie leaves the permit count exact; a cancelled + waiter holds nothing and a peer waiter still wakes; a contended admission + mutex does not carry `acquireUntil` past its deadline. + +Loop semantics — priority means ordering among immediately admissible +candidates: + +1. Pass one, first sweep in priority order: `tryAcquire` on each available + entry; the first immediate success is attempted. A saturated entry is + skipped while another eligible entry has capacity. +2. If the attempted entry fails (peer fault), the sweep continues from the + next entry, still by `tryAcquire`; previously skipped saturated entries + are re-tried by `tryAcquire` on each subsequent sweep step (a slot may + have freed). +3. Only when no eligible entry has an immediate permit does the loop block: + `acquireUntil(remaining deadline)` on the highest-priority eligible + entry. Expired → `error.BudgetExhausted`. The no-head-of-line guarantee + is deliberately scoped to capacity observed during the sweep: once + blocked, a lower-priority slot freeing does not wake this waiter + (any-entry wakeups need multi-wait machinery this milestone does not + buy). Record this bound in the admission helper's doc comment. +4. Pass two (backoff probing) keeps today's in-order probing but admits + through the same helper bounded by the remaining deadline — it + deliberately retains blocking, one entry at a time, because probing a + backed-off entry is already a last resort. +5. The post-admission health recheck survives the refactor: after acquiring + a permit by either path in pass one, re-read health against a fresh + `now`; an entry that entered backoff while this task waited is released + (permit returned) and the sweep resumes. The existing regression test + for this recheck is retained. +6. Before any attempt starts — immediate admission included — the loop + re-reads the clock; a deadline already passed returns + `error.BudgetExhausted` with the permit returned and no endpoint named, + so an instantly-completing leaf can never manufacture evidence after + exhaustion (review round 2026-08-27). Test: an exchange whose deadline + is already gone starts no attempt. +7. The compiled pool bound (`Pool.max_entries`, 64) is enforced at config + validation: more than 64 ENABLED upstreams is `TooManyUpstreams` at path + `upstreams`, so a valid config can never trip the pool assert (review + round 2026-08-27). Tests: 64 enabled passes, 65 fails, 65 listed with + 64 disabled passes. +8. Queue accounting keeps its meaning: `queued_total` and + `queued_seconds_total` count only the blocking `acquireUntil` path, + recorded whether it ends in acquisition, expiry or cancellation; a + `tryAcquire` — hit or miss — never counts as queued. Acceptance test + retained. + +### B.3 Classification (uses A.2's tagged race) + +| Attempt outcome | Budget it ran with | Health | `selected` | Loop action | +| --- | --- | --- | --- | --- | +| success | any | success | this endpoint | return answer | +| expiry (`expired`) | full `attempt` | failure | this endpoint | continue failover | +| expiry (`expired`) | truncated | untouched | unchanged | return `error.BudgetExhausted` | +| completed peer fault (incl. leaf Timeout) | any, even truncated | failure | this endpoint | continue failover | +| local_resource | — | untouched | unchanged | return err (as today) | +| cancellation | — | untouched | unchanged | return `error.Canceled` | +| wait exhausts deadline | — | untouched | unchanged (null if nothing ran) | return `error.BudgetExhausted` | +| completed attempt returns `error.BudgetExhausted` (a leaf may emit it once it exists) | any | untouched | unchanged | return it; counter increments once at the outer pool | + +A truncated expiry is a censored observation: the pool did not grant the +configured observation interval, so it is evidence about the pool's budget, +never about the peer. No minimum-attempt floor exists. + +### B.4 `selected` = last attributable endpoint + +Assign `selected.*` only in the success row and the two health-recording +failure rows above — after the attempt completes, not before it starts. This +attribution rule is POOL-SPECIFIC: leaf clients (DoH, DoT, ForwardClient, +test fakes) keep their write-before-attempt behavior — they have one +endpoint and record no health, so "the endpoint I tried" is honest there. +The `Client.exchange` doc comment in transport.zig is amended to state both +contracts: implementations may write before each attempt; `Pool` documents +its stricter last-attributable rule on `Pool.exchange` itself. The pool-level +tests at pool.zig:748-771 (selected-before-attempt) and :910 invert into the +new contract's tests; leaf-client tests are untouched. + +### B.5 The counter + +`Pool` gains one pool-level counter, `budget_exhausted_total` (atomic u64, +incremented once per exchange that returns `error.BudgetExhausted`, never per +endpoint). `Pool.snapshot` returns per-entry rows and cannot carry a +pool-wide number without duplicating it — so the counter is exposed through a +separate getter, `Pool.budgetExhaustedTotal()`, and B owns the mechanical +plumbing at every existing snapshot call site its change touches so B's own +gate passes before C. No per-stage split (queue vs attempt) — +fixed-cardinality stage labels are deferred until an operator needs them. + +### B.6 Acceptance (the decisive regressions first) + +- [ ] Fast standby at shipped defaults: entry 0 stalls its full attempt + budget, entry 1 answers instantly, attempt = total/2 → entry 1 answers. +- [ ] Saturated primary, free standby: entry 0's slots all held by stalled + exchanges, entry 1 free and fast → entry 1 answers well inside the + deadline. This is the Pi reproduction; it must FAIL against the current + code and pass after B.2. +- [ ] Truncated expiry mutates no health, returns BudgetExhausted, leaves + `selected` at the last attributable endpoint. +- [ ] Truncated attempt failing with ConnectionRefused mutates health and + updates `selected`. +- [ ] Queue-only exhaustion → BudgetExhausted with `selected == null`. +- [ ] All-backoff pass two runs under the same deadline. +- [ ] External cancellation returns Canceled, counts no budget, mutates + nothing. +- [ ] `budget_exhausted_total` increments once per exhausted exchange. +- [ ] Existing invariants hold: cancelled waiter returns its permit; two + stalling upstreams cost ≤ total, not one budget each. + +--- + +## Session C: handler, forward client, surfaces + +### C.1 Handler (src/server/handler.zig) + +Both `transport.group` switches (:677, :735) gain `.budget_exhausted => +ctx.servFail()` — SERVFAIL on the wire, same as peer faults; no rcode fits +better. No per-query warn log (spam); the counter is the record. + +### C.2 Forward client (src/local/forward_client.zig) + +One outer budget for the whole exchange: wrap UDP attempt → truncation +fallback → TCP in a single tagged race against `read_timeout` (the UDP +receive's internal deadline and the TCP `raceWithin` at :217 collapse into +the one outer bound — remove the fresh TCP budget). Every timeout here +concerns the single configured resolver, so expiry stays `error.Timeout` +(peer evidence), never BudgetExhausted — the budget/peer distinction is pool +policy. Test: truncated-UDP-then-stalled-TCP completes or expires within one +`read_timeout`, not two. Update the doc comment on `read_timeout_ms` in the +config (validate.zig:360 note and the settings description) to say it bounds +the whole forward-zone exchange. The key is NOT renamed (anti-requirement). + +### C.3 Surfaces + +- Metrics: `nxdns_upstream_budget_exhausted_total` (pool-wide) wherever + `nxdns_upstream_*` counters render, read via `Pool.budgetExhaustedTotal()`. +- `/api/health` is NOT changed (decision, not omission): the counter is an + operator metric, it never changes health status, and adding it to the API + would drag openapi.yaml, admin types, fixtures, contract samples and the + admin gates into a milestone that owes them nothing. Metrics only. +- docs: `docs/reference/configuration.md` — every statement describing + `read_timeout_ms` as a per-read or per-attempt bound is rewritten to the + whole-exchange contract; the upstream timeouts section gains one paragraph + on budget semantics (deadline, truncation, attribution). The stale contract + text in `src/config/model.zig` (the setting's doc comment) and the note at + `src/config/validate.zig:360` are updated to match. + +### C.4 Acceptance + +- [ ] `zig build test` and `zig build test -Dintegration` green. +- [ ] Metrics test covers the new counter's rendering. +- [ ] Forward-client single-budget test per C.2. + +--- + +## File ownership + +- A: `src/upstream/transport.zig`, plus mechanical placeholder arms at the + broken `Group` switches (`src/upstream/pool.zig`, `src/server/handler.zig`, + `src/local/forward_client.zig`, test switches the compiler flags) — + sequential ownership, A runs alone; C takes forward_client.zig and + handler.zig over later in sequence. +- B: `src/upstream/pool.zig` (+ the `Client.exchange` doc contract in + transport.zig and snapshot-caller plumbing for the new getter — B runs + alone after A). +- C: `src/server/handler.zig`, `src/local/forward_client.zig`, the metrics + rendering files, `src/config/model.zig` + `src/config/validate.zig` doc + text, `docs/reference/configuration.md`. +- Orchestrator: CHANGELOG.md. + +## Acceptance criteria (milestone complete) + +- [ ] All session criteria; both suites green; `zig fmt --check` clean. +- [x] The saturated-primary regression demonstrably fails on pre-milestone + code and passes after. Evidence (orchestrator-run mutation check, + 2026-08-27): with the pass-one `tryAcquire` sweep disabled in + `Pool.admit` — restoring pre-fix head-of-line blocking — the test + "a saturated primary defers to a standby that has capacity" fails with + `error.BudgetExhausted` out of the blocking admission path, and four + queue-accounting/attribution tests fail with it (5 failed, seed + 0x91ce5df5). Sweep restored: 30/30 steps, 1892/2067 passed, 0 failed. +- [ ] Changelog: failover now works under primary saturation; SERVFAILs from + budget exhaustion are counted, not blamed on an upstream; for + upstream-pool queries the query-log `upstream` field now names only + endpoints whose outcome was recorded (may be null) — forward-zone + queries keep naming their single configured resolver as before. + +## Anti-requirements + +- No default timeout changes. +- No parallel/racing fan-out to multiple upstreams. Priority failover stays, + with priority defined as ordering among immediately admissible candidates + (B.2) — a saturated higher-priority entry defers to an admissible + lower-priority one; it does not outrank an idle standby by blocking on it. +- No `/api/health` or admin changes; the counter surfaces in metrics only. +- No patching of the vendored/system Zig stdlib; the admission helper is + pool-local. +- No rename of `upstream.read_timeout_ms` (doc fix only). +- No minimum-attempt floor constant. +- No per-stage split of the budget counter; no per-query budget log lines. +- No watchdog replacing the removed outer race. +- No fake upstream identity (e.g. "budget") in the query log; no new + query-log column. +- No changes to backoff policy, slot counts, or health scoring beyond the + attribution rules above. diff --git a/src/cli.zig b/src/cli.zig index bbcbc9b..40d0657 100644 --- a/src/cli.zig +++ b/src/cli.zig @@ -1012,7 +1012,7 @@ fn probeUpstreams(r: Runner, cfg: model.Config) !usize { .priority = server.priority, .enabled = true, .health = .init, - .sem = .{ .permits = slots.len }, + .admission = .{ .permits = slots.len }, .reuse_recoveries = &recoveries, }}; var single: pool.Pool = .init(&entries, .{}, timeouts, seed); diff --git a/src/config/model.zig b/src/config/model.zig index 81cadb9..0cbd819 100644 --- a/src/config/model.zig +++ b/src/config/model.zig @@ -57,9 +57,11 @@ pub const Config = struct { pub const Upstream = struct { /// Bounds one attempt against one upstream inside the pool's failover loop. attempt_timeout_ms: u32 = 2500, - /// The forward-zone client's read deadline, and nothing else. It bounds a - /// different subsystem from the two above (`src/local/forward_client.zig`), - /// so no cross-check relates it to them. + /// The forward-zone client's whole-exchange budget, and nothing else: one + /// bound covers the UDP attempt, a TC=1 fallback and the TCP retry + /// together, not each of them. It bounds a different subsystem from the two + /// above (`src/local/forward_client.zig`), so no cross-check relates it to + /// them. read_timeout_ms: u32 = 3000, /// The whole-exchange budget: every failover attempt together, not one of /// them. The pool races the entire loop against it. diff --git a/src/config/validate.zig b/src/config/validate.zig index 97bf4f7..38065a3 100644 --- a/src/config/validate.zig +++ b/src/config/validate.zig @@ -52,6 +52,7 @@ const limits = @import("limits.zig"); const logger = @import("../storage/logger.zig"); const regex = @import("../filter/regex.zig"); const safe_url = @import("../safe_url.zig"); +const pool = @import("../upstream/pool.zig"); const transport = @import("../upstream/transport.zig"); const Config = model.Config; @@ -69,6 +70,7 @@ const Prefix = address.Prefix; /// like every other resource failure. pub const ValidateError = error{ NoUpstreams, + TooManyUpstreams, BadUpstreamUrl, UpstreamHostNotIpLiteral, DuplicateUpstreamUrl, @@ -357,8 +359,9 @@ fn checkScalars(cfg: Config, diags: *Diagnostics) error{OutOfMemory}!void { // The only cross-check that relates two knobs of one subsystem: the pool // races one attempt against `attempt` and the whole failover loop against // `total`, so an attempt budget above the total one can never be reached. - // `read_timeout_ms` belongs to the forward-zone client and is deliberately - // unrelated to both. + // `read_timeout_ms` bounds the forward-zone client's whole exchange — + // UDP attempt, TC=1 fallback and TCP retry under one budget — and is + // deliberately unrelated to both. if (up.attempt_timeout_ms > up.total_timeout_ms) { try diags.add( error.BadTimeout, @@ -823,6 +826,21 @@ fn checkCollections(cfg: Config, diags: *Diagnostics, scratch: Allocator) error{ .{}, ); } + // Each enabled upstream becomes one pool entry, and the failover loop + // tracks the entries it has spent in a fixed bitset of `Pool.max_entries` + // bits. Without this check a config past that bound reaches an assert and + // panics at startup, which is the wrong way to tell an operator that a + // number is too large. Disabled upstreams are not counted: they never + // become entries. + if (enabled_upstreams > pool.Pool.max_entries) { + try diags.add( + error.TooManyUpstreams, + "upstreams", + .{}, + "{d} upstreams are enabled; nxdns is built for at most {d}", + .{ enabled_upstreams, pool.Pool.max_entries }, + ); + } var client_ips: IndexSet = .empty; for (cfg.clients, 0..) |client, i| { @@ -1323,6 +1341,52 @@ test "error.NoUpstreams when nothing is enabled" { try expectProblem(cfg, error.NoUpstreams, "upstreams"); } +/// `count` distinct enabled upstreams. Generated rather than written out +/// because the bound this exercises is 64, and a hand-written list that long +/// would say less than the loop does. +fn ManyUpstreams(comptime count: usize) type { + return struct { + const list: [count]model.UpstreamServer = blk: { + var built: [count]model.UpstreamServer = undefined; + for (&built, 0..) |*server, i| { + server.* = .{ .url = std.fmt.comptimePrint("https://u{d}.example/dns-query", .{i}) }; + } + break :blk built; + }; + }; +} + +fn manyUpstreams(comptime count: usize) []const model.UpstreamServer { + return &ManyUpstreams(count).list; +} + +test "as many enabled upstreams as the pool holds validates cleanly" { + var cfg = baseConfig(); + cfg.upstreams = manyUpstreams(pool.Pool.max_entries); + try expectClean(cfg); +} + +test "error.TooManyUpstreams one enabled upstream past the pool's bound" { + // The pool asserts this bound, so without the check here a valid-looking + // config panics at startup instead of being reported. + var cfg = baseConfig(); + cfg.upstreams = manyUpstreams(pool.Pool.max_entries + 1); + try expectProblem(cfg, error.TooManyUpstreams, "upstreams"); +} + +test "upstreams past the pool's bound are fine while they are disabled" { + // Only enabled upstreams become pool entries, so a long list with a small + // enabled subset is not near the bound at all. + var cfg = baseConfig(); + cfg.upstreams = comptime blk: { + var list = manyUpstreams(pool.Pool.max_entries + 1)[0 .. pool.Pool.max_entries + 1].*; + for (list[1..]) |*server| server.enabled = false; + const frozen = list; + break :blk &frozen; + }; + try expectClean(cfg); +} + test "error.BadUpstreamUrl on an unsupported scheme" { var cfg = baseConfig(); cfg.upstreams = &.{.{ .url = "ftp://dns.example/" }}; diff --git a/src/local/forward_client.zig b/src/local/forward_client.zig index 9d84160..94cb0a2 100644 --- a/src/local/forward_client.zig +++ b/src/local/forward_client.zig @@ -109,6 +109,11 @@ pub const ForwardClient = struct { /// `.udp` resolvers send one datagram and fall back to TCP when the answer /// comes back with TC=1. `.tcp` resolvers skip straight to the TCP path. + /// + /// `read_timeout` bounds the WHOLE exchange, truncation fallback included: + /// the instant is computed once here and every blocking step inside runs + /// against it, so a truncated UDP answer followed by a stalled TCP retry + /// costs one budget rather than two. pub fn exchange( self: *ForwardClient, io: std.Io, @@ -120,10 +125,22 @@ pub const ForwardClient = struct { if (response_buf.len == 0) return error.BufferTooSmall; self.stats.queries += 1; - return self.route(io, query, response_buf) catch |err| { + const expiry_at: std.Io.Clock.Timestamp = .fromNow(io, self.read_timeout); + // The outcome is deliberately discarded. A forward zone has exactly one + // configured resolver, so an expiry here is still that resolver failing + // to answer in time: `error.Timeout` is peer evidence, and the + // budget/peer distinction is upstream-pool policy. + var outcome: transport.RaceOutcome = .completed; + return transport.raceUntilTagged(io, expiry_at, &outcome, route, .{ + self, + io, + query, + response_buf, + expiry_at, + }) catch |err| { switch (transport.group(err)) { .peer_fault, .local_resource => self.stats.failures += 1, - .cancellation => {}, + .cancellation, .budget_exhausted => {}, } return err; }; @@ -134,11 +151,12 @@ pub const ForwardClient = struct { io: std.Io, query: []const u8, response_buf: []u8, + expiry_at: std.Io.Clock.Timestamp, ) transport.ExchangeError![]u8 { if (self.resolver.scheme == .udp) { - if (try self.exchangeUdp(io, query, response_buf)) |reply| return reply; + if (try self.exchangeUdp(io, query, response_buf, expiry_at)) |reply| return reply; } - return self.exchangeTcp(io, query, response_buf); + return self.tcpOnce(io, query, response_buf); } /// `null` means the resolver set TC=1 and the caller must retry over TCP. @@ -151,6 +169,7 @@ pub const ForwardClient = struct { io: std.Io, query: []const u8, response_buf: []u8, + expiry_at: std.Io.Clock.Timestamp, ) transport.ExchangeError!?[]u8 { const dest = self.destination(); const local = wildcardFor(dest); @@ -166,9 +185,10 @@ pub const ForwardClient = struct { return transport.mapPhase(err, error.SendFailed); }; - // A deadline, not a duration: a discarded foreign datagram restarts the - // receive, and a duration would hand each retry the full budget again. - const deadline = (std.Io.Timeout{ .duration = self.read_timeout }).toDeadline(io); + // The exchange-wide instant, not a fresh duration: a discarded foreign + // datagram restarts the receive, and a duration would hand each retry + // the full budget again. + const deadline: std.Io.Timeout = .{ .deadline = expiry_at }; while (true) { const msg = socket.receiveTimeout(io, response_buf, deadline) catch |err| switch (err) { @@ -205,18 +225,11 @@ pub const ForwardClient = struct { } } - /// The read budget bounds the whole TCP exchange through - /// `transport.raceWithin`. `ConnectOptions.timeout` is never set: the - /// Threaded backend panics on it (Threaded.zig:12076). - fn exchangeTcp( - self: *ForwardClient, - io: std.Io, - query: []const u8, - response_buf: []u8, - ) transport.ExchangeError![]u8 { - return transport.raceWithin(io, self.read_timeout, tcpOnce, .{ self, io, query, response_buf }); - } - + /// Unbounded on its own: `exchange` runs it inside the exchange-wide race, + /// which is what cancels a stalled connect or read. It takes no budget of + /// its own, so a truncation fallback does not start a second one. + /// `ConnectOptions.timeout` is never set: the Threaded backend panics on it + /// (Threaded.zig:12076). fn tcpOnce( self: *ForwardClient, io: std.Io, @@ -307,6 +320,11 @@ fn receiveFailure(stream_reader: *const net.Stream.Reader, err: anyerror) transp } const testing = std.testing; +const build_options = @import("build_options"); +const name_mod = @import("../dns/name.zig"); +const packet = @import("../dns/packet.zig"); +const question = @import("../dns/question.zig"); +const types = @import("../dns/types.zig"); fn testBuf() [min_frame_buf]u8 { return undefined; @@ -467,3 +485,113 @@ test "a stashed stream error is preferred over the collapsed one" { receiveFailure(&stream_reader, error.EndOfStream), ); } + +const one_budget_ms = 400; +/// Late enough in the budget that a second, fresh budget for the TCP leg would +/// be unmistakable in the elapsed time. +const truncate_after_ms = 300; + +/// Answers the first datagram late in the budget with TC=1, which sends the +/// client to TCP — where a listener that never accepts leaves it stalled. +fn truncatingThenStallingResolver(io: std.Io, socket: *const net.Socket) void { + var buf: [2048]u8 = undefined; + const msg = socket.receive(io, &buf) catch return; + const request = packet.parse(msg.data) catch return; + + (std.Io.Clock.Duration{ + .raw = .fromMilliseconds(truncate_after_ms), + .clock = .awake, + }).sleep(io) catch return; + + // The question has to be echoed: `transport.validateResponse` runs before + // the client reads the TC bit, so a bare header would come back as + // `BadResponse` and never reach the TCP fallback this case is about. + const q = packet.firstQuestion(request) orelse return; + var reply_buf: [512]u8 = undefined; + var b = packet.ResponseBuilder.init(&reply_buf, request.header, q) catch return; + const reply = b.finish(); + + var parsed = dns_header.parse(reply) catch return; + parsed.flags.tc = true; + dns_header.encode(parsed, reply[0..types.header_len]); + socket.send(io, &msg.from, reply) catch return; +} + +fn testQuery(io: std.Io, buf: []u8) ![]const u8 { + var id_bytes: [2]u8 = undefined; + io.random(&id_bytes); + dns_header.encode(.{ + .id = std.mem.readInt(u16, &id_bytes, .big), + .flags = .{ + .rcode = .no_error, + .z = 0, + .ra = false, + .rd = true, + .tc = false, + .aa = false, + .opcode = .query, + .qr = false, + }, + .qdcount = 1, + .ancount = 0, + .nscount = 0, + .arcount = 0, + }, buf[0..types.header_len]); + + var w: std.Io.Writer = .fixed(buf[types.header_len..]); + try question.encode(.{ + .name = try name_mod.fromText("nas.lan"), + .qtype = .a, + .qclass = .in, + }, &w); + return buf[0 .. types.header_len + w.buffered().len]; +} + +test "one budget covers the udp leg, the TC=1 fallback and the tcp leg" { + if (!build_options.integration) return error.SkipZigTest; + + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + const bind_address: net.IpAddress = try .parse("127.0.0.1", 0); + const socket = try bind_address.bind(io, .{ .mode = .dgram }); + defer socket.close(io); + const port = socket.address.ip4.port; + + // Bound but never accepted: the connect completes out of the kernel's + // backlog and the read then waits forever, which is the stall a second + // budget would be spent on. + const tcp_address: net.IpAddress = try .parse("127.0.0.1", port); + var tcp_listener = try tcp_address.listen(io, .{ .reuse_address = true }); + defer tcp_listener.deinit(io); + + var group: std.Io.Group = .init; + defer group.cancel(io); + try group.concurrent(io, truncatingThenStallingResolver, .{ io, &socket }); + + var url_buf: [64]u8 = undefined; + const url = try std.fmt.bufPrint(&url_buf, "udp://127.0.0.1:{d}", .{port}); + + var frame_buf: [min_frame_buf]u8 = undefined; + var fc: ForwardClient = .init( + try validate.parseResolver(url), + &frame_buf, + .{ .raw = .fromMilliseconds(one_budget_ms), .clock = .awake }, + ); + + var query_buf: [types.header_len + types.max_name_len + 4]u8 = undefined; + const query = try testQuery(io, &query_buf); + var response_buf: [2048]u8 = undefined; + + const started = std.Io.Clock.awake.now(io); + try testing.expectError(error.Timeout, fc.exchange(io, query, &response_buf)); + const elapsed = started.durationTo(std.Io.Clock.awake.now(io)).nanoseconds; + + // Two budgets would spend 300 ms on the UDP leg and then a fresh 400 ms on + // the stalled TCP leg. The bound sits between one budget and that sum, so + // the double-budget shape cannot pass. + try testing.expect(elapsed < @as(i96, one_budget_ms + truncate_after_ms / 2) * std.time.ns_per_ms); + try testing.expectEqual(@as(u64, 1), fc.stats.udp_truncated); + try testing.expectEqual(@as(u64, 1), fc.stats.failures); +} diff --git a/src/server/handler.zig b/src/server/handler.zig index c80ca58..3d6a4c8 100644 --- a/src/server/handler.zig +++ b/src/server/handler.zig @@ -676,7 +676,10 @@ const Context = struct { const answer = client.exchange(ctx.io, ctx.query, ctx.response_buf) catch |err| { return switch (transport.group(err)) { .cancellation => .drop, - .peer_fault, .local_resource => ctx.servFail(), + // A budget that ran out is SERVFAIL like any other failure: no + // rcode says "I gave up in time". It is not logged per query — + // `nxdns_upstream_budget_exhausted_total` is the record. + .peer_fault, .local_resource, .budget_exhausted => ctx.servFail(), }; }; @@ -737,7 +740,13 @@ const Context = struct { // the client is about to lose the socket anyway; the listener's // own counters record the abandoned datagram. .cancellation => return .drop, - .peer_fault, .local_resource => return ctx.servFail(), + // A budget that ran out is SERVFAIL like any other failure: no + // rcode says "I gave up in time". It is not logged per query — + // `nxdns_upstream_budget_exhausted_total` is the record — and + // the pool leaves `selected` unchanged, so the row still names + // the last attributable endpoint if there was one, and names + // none only when no attributable attempt happened. + .peer_fault, .local_resource, .budget_exhausted => return ctx.servFail(), }; }; bump(&ctx.handler.stats.queries); diff --git a/src/server/resolver_integration_test.zig b/src/server/resolver_integration_test.zig index b0c9ab0..60caf90 100644 --- a/src/server/resolver_integration_test.zig +++ b/src/server/resolver_integration_test.zig @@ -176,7 +176,7 @@ const EntryStorage = struct { .priority = priority, .enabled = true, .health = .init, - .sem = .{ .permits = self.slots.len }, + .admission = .{ .permits = self.slots.len }, .reuse_recoveries = &self.recoveries, }; } diff --git a/src/upstream/dot_client_integration_test.zig b/src/upstream/dot_client_integration_test.zig index 614176e..b318d7f 100644 --- a/src/upstream/dot_client_integration_test.zig +++ b/src/upstream/dot_client_integration_test.zig @@ -458,7 +458,7 @@ test "a session the upstream closed is recovered by one redial and counted, not .priority = 10, .enabled = true, .health = .init, - .sem = .{ .permits = slots.len }, + .admission = .{ .permits = slots.len }, .reuse_recoveries = &fixture.recoveries, }}; var pool: pool_mod.Pool = .init(&entries, .{ diff --git a/src/upstream/owner.zig b/src/upstream/owner.zig index f85a967..b9f303e 100644 --- a/src/upstream/owner.zig +++ b/src/upstream/owner.zig @@ -500,7 +500,7 @@ const Upstreams = struct { .priority = server.priority, .enabled = true, .health = .init, - .sem = .{ .permits = entry_slots.len }, + .admission = .{ .permits = entry_slots.len }, .reuse_recoveries = &self.recovery_counters[self.used], }; self.used += 1; diff --git a/src/upstream/pool.zig b/src/upstream/pool.zig index af3b991..f328507 100644 --- a/src/upstream/pool.zig +++ b/src/upstream/pool.zig @@ -4,15 +4,20 @@ //! The pool is itself a `transport.Client`, so the handler above it sees one //! interface and knows nothing about how many upstreams exist. //! -//! Two deadlines, nested. `timeouts.attempt` bounds one exchange against one -//! entry; `timeouts.total` bounds the whole failover loop, every attempt -//! together. The outer one is what the client asking the question actually -//! waits for: without it, N unreachable upstreams cost N × attempt, and the -//! resolver above has already given up. The outer expiry is `error.Timeout` -//! unconditionally — the fast-failure paths (no entry enabled) return long -//! before the budget, so an expiry means either that an attempt was in flight -//! or that the task died waiting for a slot on a saturated entry. -//! `Entry.queued_total` is what tells those two apart. +//! One deadline, computed once. `Pool.exchange` turns `timeouts.total` into an +//! absolute instant and the loop spends it: admission waits against it, and +//! every attempt is raced against `min(now + timeouts.attempt, deadline)`. No +//! outer race wraps the loop. Two timers aimed at the same instant race each +//! other, and the outer one wins by cancelling the loop from outside, which +//! throws away the loop's own classification of what just happened. +//! +//! What the deadline expiring means depends on which side of it ran. An attempt +//! that got its full `timeouts.attempt` and did not answer is evidence about +//! that peer, recorded as a failure. An attempt the deadline cut short, and a +//! wait for a slot that never ended, are evidence about this process's budget: +//! they are `error.BudgetExhausted`, counted by `budgetExhaustedTotal` and +//! recorded against no endpoint. That distinction is why the pool needs +//! `transport.raceUntilTagged` rather than a bare timeout. //! //! Two passes, not one. Pass one walks the enabled entries that health says are //! available. Pass two runs only when pass one attempted nothing, and skips the @@ -28,7 +33,7 @@ //! or an attempt on some other entry. //! //! An entry owns one leaf client per slot, each behind its own `Slot.busy` -//! mutex, plus an `Entry.sem` carrying one permit per slot. A task takes a +//! mutex, plus an `Entry.admission` carrying one permit per slot. A task takes a //! permit, then the first free slot's mutex, and holds that mutex for a whole //! attempt against the entry — the exchange and the timeout race around it — //! before the failover loop moves to the next candidate. The mutex is what @@ -36,9 +41,15 @@ //! transfer buffer, `DotClient` owns four TLS buffers and the stream state //! built on them, so one client may be inside `exchange` only once at a time. //! -//! Because a saturated entry still makes tasks wait, pass one re-reads health -//! after it acquires a slot: the attempts a task queued behind can fail the -//! entry into backoff while it waits, and pass one must not attempt an entry +//! Priority orders the entries that can be admitted *now*. A saturated entry is +//! stepped over while a lower-priority one has a free slot, because blocking on +//! the higher-priority entry spends the caller's whole budget in a queue while +//! an idle standby watches. Only when no eligible entry can admit immediately +//! does a task block, and then on the highest-priority eligible entry alone. +//! +//! Because a saturated entry can still make a task wait, pass one re-reads +//! health after it acquires a slot: the attempts a task queued behind can fail +//! the entry into backoff while it waits, and pass one must not attempt an entry //! that is unavailable by the time it gets its turn. Lock order is always //! `Slot.busy` then `Pool.mutex`, never the reverse. //! @@ -89,6 +100,108 @@ const AttemptFailure = struct { /// shared `std.http.Client`. pub const slots_per_entry = 8; +/// Admission control for one entry: one permit per slot, a try-acquire, and an +/// acquire bounded by the caller's deadline. `std.Io.Semaphore` offers neither +/// in 0.16, and `std.Io.Condition` has no timed wait, so this mirrors the +/// standard semaphore's mutex/count/condition protocol with the two operations +/// the failover loop needs added. The stdlib is not patched and nothing spins. +/// +/// The mutex owns the permit count and the decision to take one, and the caller +/// takes its permit only *after* the timed race has resolved. That ordering is +/// the whole design: racing a bare `Semaphore.wait` against an expiry can +/// decrement the count in the same instant the expiry wins, and the discarded +/// success never reaches the caller's release, so the permit is gone for good. A +/// wait discarded here loses a notification instead, which the exiting waiter +/// repairs by re-signalling. +/// +/// Bound worth knowing: `acquireUntil` blocks on one entry. A permit freeing on +/// some *other* entry does not wake this waiter, so the pool's no-head-of-line +/// guarantee covers only the capacity its `tryAcquire` sweep observed before it +/// blocked. Waking on any of several entries needs multi-wait machinery the pool +/// does not have. +pub const Admission = struct { + mutex: std.Io.Mutex = .init, + cond: std.Io.Condition = .init, + /// One per slot at build time; `Pool.init` asserts it. + permits: usize = 0, + + /// Which side of `acquireUntil` ended the wait. `.expired` holds no permit. + pub const Outcome = enum { acquired, expired }; + + /// Takes a permit if one is free this instant; never blocks. + /// + /// A contended mutex reads as "no permit now" rather than as a reason to + /// wait. The failover loop calls this on every candidate in turn, so + /// blocking here for a descheduled permit holder would stall the whole + /// sweep and spend the caller's budget outside `acquireUntil`, which is the + /// only step allowed to consume it. + pub fn tryAcquire(self: *Admission, io: std.Io) bool { + if (!self.mutex.tryLock()) return false; + defer self.mutex.unlock(io); + if (self.permits == 0) return false; + self.permits -= 1; + return true; + } + + /// Returns a permit. Uncancelable, so a torn-down attempt still gives its + /// slot back. + pub fn release(self: *Admission, io: std.Io) void { + self.mutex.lockUncancelable(io); + defer self.mutex.unlock(io); + self.permits += 1; + self.cond.signal(io); + } + + /// Waits for a permit until `deadline`. Returns `.acquired` holding one, + /// `.expired` holding none, or an error holding none — `error.Canceled` + /// when the whole task is torn down, `error.SystemResources` when the race + /// cannot start. + pub fn acquireUntil( + self: *Admission, + io: std.Io, + deadline: std.Io.Clock.Timestamp, + ) transport.ExchangeError!Outcome { + while (true) { + // Every take is non-blocking, including the retries. A take that + // waits on the mutex spends time no deadline bounds: before the + // first race it carries the call past the deadline outright, and + // after a wake it can return `.acquired` late. A miss instead + // re-enters the race below, which is aimed at the same absolute + // instant every time, so the cycle ends at the deadline with + // `.expired` — the correct answer, not a spin. + if (self.tryAcquire(io)) return .acquired; + var race: transport.RaceOutcome = .completed; + _ = transport.raceUntilTagged(io, deadline, &race, waitForPermit, .{ self, io }) catch |err| { + // This task may be leaving with a notification meant for a peer: + // `Condition`'s cancel path consumes a pending signal before it + // reports the cancellation, and a wait whose success the race + // discarded consumed one outright. Hand it back before exiting. + self.resignal(io); + if (err == error.Timeout and race == .expired) return .expired; + return err; + }; + // The wait saw a permit but did not take one, so another task may + // have taken it first. The loop re-checks rather than assuming, and + // does not spin: `waitForPermit` blocks whenever the count is zero. + } + } + + /// Returns once the count is non-zero, without touching it. + fn waitForPermit(self: *Admission, io: std.Io) transport.ExchangeError!void { + self.mutex.lock(io) catch return error.Canceled; + defer self.mutex.unlock(io); + while (self.permits == 0) self.cond.wait(io, &self.mutex) catch return error.Canceled; + } + + /// Only when a permit is actually there to hand over: an unconditional + /// signal would wake a peer that would find nothing. + fn resignal(self: *Admission, io: std.Io) void { + self.mutex.lockUncancelable(io); + defer self.mutex.unlock(io); + if (self.permits > 0) self.cond.signal(io); + } +}; + /// One leaf client of an entry, and the lock that keeps one task inside it. pub const Slot = struct { client: transport.Client, @@ -112,15 +225,14 @@ pub const Entry = struct { /// `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, + admission: Admission, in_flight: std.atomic.Value(u32) = .init(0), /// Test-only: the most exchanges this entry ever had in flight at once. peak_in_flight: std.atomic.Value(u32) = .init(0), - /// 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. + /// Blocked admissions, not a queue length: a task counts itself queued + /// exactly when it has run out of immediately admissible entries and blocks + /// on this one, whichever way that block ends. A `tryAcquire` — hit or miss + /// — is not a queue, so it never counts. queued_total: std.atomic.Value(u64) = .init(0), queued_ns_total: std.atomic.Value(u64) = .init(0), /// Points into the composition root's stable counter storage, never into an @@ -130,9 +242,9 @@ pub const Entry = struct { /// 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. + /// Both counters, on every exit of a blocking wait for a permit. A waiter + /// that burned its budget in the queue is the field failure this records, so + /// an expiry and a cancellation count exactly like an acquisition. fn recordQueued(self: *Entry, io: std.Io, started: std.Io.Timestamp) void { const elapsed = std.Io.Clock.awake.now(io).nanoseconds - started.nanoseconds; _ = self.queued_total.fetchAdd(1, .monotonic); @@ -149,6 +261,30 @@ pub const Entry = struct { } }; +/// One exchange, raced until `expiry_at` and tagged with which side won. +/// +/// `race` distinguishes an expiry from the leaf returning `error.Timeout` of +/// its own, which are the same error and opposite evidence: the first is about +/// the clock the caller set, the second is about the peer. +/// +/// The leaf client reports an identity of its own, which the pool discards: the +/// entry's endpoint is the pool's own naming of the same resolver, and it is +/// what the caller was handed. A leaf writing its identity into the caller's +/// slot would let a test fake overwrite the endpoint that actually answered. +fn attemptUntil( + io: std.Io, + entry_client: transport.Client, + query: []const u8, + response_buf: []u8, + expiry_at: std.Io.Clock.Timestamp, + race: *transport.RaceOutcome, +) transport.ExchangeError![]u8 { + var leaf_selected: ?[]const u8 = null; + return transport.raceUntilTagged(io, expiry_at, race, transport.Client.exchange, .{ + entry_client, io, query, response_buf, &leaf_selected, + }); +} + /// A copy of one entry's health, taken under the mutex. Feeds `/metrics` and /// the `/api/health` upstream condition. pub const Snapshot = struct { @@ -205,6 +341,19 @@ pub const Pool = struct { /// store at all. Every emit here sits outside `mutex`; see /// `recordDiagnostics`. diagnostics: ?*events.Store = null, + /// Exchanges that ran out of budget, counted once each by `exchange`. A + /// pool-wide number rather than a per-entry one on purpose: budget + /// exhaustion is this process failing to give any peer its interval, so + /// attributing it to an endpoint would be the very lie this milestone + /// removes. Read through `budgetExhaustedTotal`, since `Snapshot` is + /// per-entry and has nowhere to put it. + budget_exhausted_total: std.atomic.Value(u64) = .init(0), + + /// The failover loop tracks which entries it has already spent in a bitset + /// on its stack, so the number of upstreams one pool may hold is bounded. + /// A household config has two; `web/metrics.zig` already renders at most + /// this many. + pub const max_entries = 64; pub fn init( entries: []Entry, @@ -213,9 +362,10 @@ pub const Pool = struct { seed: u64, ) Pool { std.debug.assert(entries.len > 0); + std.debug.assert(entries.len <= max_entries); for (entries) |*entry| { std.debug.assert(entry.slots.len >= 1); - std.debug.assert(entry.sem.permits == entry.slots.len); + std.debug.assert(entry.admission.permits == entry.slots.len); } // Stable, so entries sharing a priority keep their configured order. std.mem.sort(Entry, entries, {}, byPriority); @@ -251,12 +401,18 @@ pub const Pool = struct { /// may have written into it. The returned slice is only meaningful on /// success; on error the buffer's contents are undefined. /// - /// The failover loop runs raced against `timeouts.total`. Losing that race - /// cancels the loop, which unwinds whatever attempt was in flight: the wait - /// for a slot is cancelable on purpose and every slot is released by defer, - /// and the health bookkeeping around a completed exchange is uncancelable, - /// so a canceled attempt leaves no lock held, no permit lost and no counter - /// half-written. + /// `selected` follows a stricter rule here than the `transport.Client` + /// contract requires: the pool names the last endpoint whose outcome it + /// actually recorded, written after the attempt rather than before it. An + /// exchange that ran out of budget while an attempt was still in flight + /// therefore leaves `selected` at the previous endpoint, or at `null` if it + /// never got that far, because "the endpoint that was in the way when the + /// clock ran out" is not evidence about that endpoint and the query log must + /// not print it as though it were. + /// + /// The whole call is bounded by one deadline derived from `timeouts.total` + /// here and spent by the loop below: no outer race, so an expiry is always + /// classified by the step that owned it. pub fn exchange( self: *Pool, io: std.Io, @@ -264,55 +420,59 @@ pub const Pool = struct { response_buf: []u8, selected: *?[]const u8, ) transport.ExchangeError![]u8 { - const len = try transport.raceWithin(io, self.timeouts.total, exchangeLoopLen, .{ - self, io, query, response_buf, selected, - }); - return response_buf[0..len]; + const deadline: std.Io.Clock.Timestamp = .fromNow(io, self.timeouts.total); + return self.failover(io, query, response_buf, selected, deadline) catch |err| { + // Once per exhausted exchange, wherever the exhaustion was + // detected — the loop's own waits and attempts, or a leaf that + // reported one of its own. + if (err == error.BudgetExhausted) { + _ = self.budget_exhausted_total.fetchAdd(1, .monotonic); + } + return err; + }; } - /// The two-pass failover loop, as a raceable task. It returns the reply's - /// length rather than its slice for the reason `Attributed` in the tests - /// below does: `Io.concurrent` stores the future's return value, so the - /// bytes are read back out of the caller's `response_buf` by `exchange`. - fn exchangeLoopLen( + /// Exchanges that ended in `error.BudgetExhausted`, pool-wide. + pub fn budgetExhaustedTotal(self: *const Pool) u64 { + return self.budget_exhausted_total.load(.acquire); + } + + /// The two-pass failover loop. + /// + /// Pass one sweeps the entries health says are available, in priority + /// order, admitting through `tryAcquire` so a saturated entry is stepped + /// over rather than waited on. It blocks only when the sweep found no + /// capacity anywhere, and then on the highest-priority candidate alone. + /// Pass two runs when pass one attempted nothing and probes every enabled + /// entry regardless of backoff, which is how backoff recovers; it keeps + /// today's one-at-a-time blocking, since probing a backed-off entry is + /// already a last resort. + fn failover( self: *Pool, io: std.Io, query: []const u8, response_buf: []u8, selected: *?[]const u8, - ) transport.ExchangeError!usize { + deadline: std.Io.Clock.Timestamp, + ) transport.ExchangeError![]u8 { const now = std.Io.Clock.awake.now(io); var last_fault: ?transport.ExchangeError = null; var attempted = false; var pass: u8 = 0; while (pass < 2) : (pass += 1) { - for (self.entries) |*entry| { - if (!entry.enabled) continue; - if (pass == 0 and !self.entryAvailable(io, entry, now)) continue; + // Entries this call is finished with: attempted already, or found + // unavailable by the post-admission recheck. Cleared per pass, + // because pass two only runs when pass one attempted nothing and + // must reconsider everything it skipped. + var spent: std.StaticBitSet(max_entries) = .initEmpty(); - // Sampled before the wait, so a task that is about to queue is - // counted as queued whichever way the wait ends. The read races - // the releasers by one slot on purpose; see `Entry`. - const queued_at: ?std.Io.Timestamp = if (entry.in_flight.load(.acquire) >= entry.slots.len) - std.Io.Clock.awake.now(io) - else - null; - - // Cancelable, unlike the health-bookkeeping locks below: a task - // waiting its turn on a saturated upstream has done nothing that - // a cancellation could corrupt, so it gives up here rather than - // queueing behind exchanges it will not use. `wait` either - // returns holding a permit or returns `error.Canceled` holding - // none. The wait itself is bounded by the holders' - // `timeouts.attempt`; the waiter's own budget only starts once - // it has a slot, and the whole loop is bounded by - // `timeouts.total` regardless. - entry.sem.wait(io) catch |err| { - if (queued_at) |started| entry.recordQueued(io, started); - return err; - }; - if (queued_at) |started| entry.recordQueued(io, started); + while (true) { + // An admission that never came ends the exchange: nothing ran, + // so nothing is attributable and `selected` stays as it is. + const index = try self.admit(io, spent, pass, now, deadline) orelse break; + spent.set(index); + const entry = &self.entries[index]; // A permit means a slot's mutex is free, so failing to find one // is a protocol bug in this loop, not a runtime condition. @@ -324,47 +484,68 @@ pub const Pool = struct { _ = entry.in_flight.fetchSub(1, .acq_rel); // Uncancelable, so a canceled attempt still returns its // permit. - entry.sem.post(io); + entry.admission.release(io); } - // The check above ran before the wait, and the attempts this one - // queued behind may have failed the entry into backoff while it - // waited. Pass one must not touch an entry that is unavailable - // now, so re-read health against a fresh `now` and move on if it - // is — the defer above releases the slot before the `continue`. - // Pass two skips this on purpose: it probes regardless of - // backoff, which is how backoff recovers. + // Health was read before the permit was taken, and the attempts + // this one waited behind may have failed the entry into backoff + // meanwhile. Pass one must not touch an entry that is + // unavailable now, so re-read against a fresh `now` and move on + // if it is — the defer above releases the slot before the + // `continue`. Pass two skips this on purpose. if (pass == 0) { const recheck = std.Io.Clock.awake.now(io); if (!self.entryAvailable(io, entry, recheck)) continue; } + // Nothing is left to observe with. Admission and the recheck + // above can each land after the deadline has already passed, + // and starting an attempt against an expired timestamp would + // race the leaf against a timer that has already fired: an + // instantly-completing leaf would win nondeterministically and + // manufacture a success, or evidence about a peer, out of a + // budget that was already gone. The defer above returns the + // permit; no endpoint is named, because none was observed. + const started_at: std.Io.Clock.Timestamp = .now(io, .awake); + if (deadline.compare(.lte, started_at)) return error.BudgetExhausted; attempted = true; - // Before the attempt, not after it: a failover reports whoever - // answered, an all-failed exchange reports the last endpoint - // tried, and a cancellation mid-flight reports the endpoint the - // query was in. The url is owned by the Endpoint, which outlives - // the pool, so the borrow stays valid past this loop. - selected.* = entry.endpoint.url; + // The full attempt budget unless the deadline lands first, in + // which case the peer is being given less than the configured + // observation interval and an expiry says nothing about it. + const full_expiry = started_at.addDuration(self.timeouts.attempt); + const truncated = deadline.compare(.lt, full_expiry); + const expiry_at = if (truncated) deadline else full_expiry; - const result = self.attempt(io, slot.client, query, response_buf); + var race: transport.RaceOutcome = .completed; + const result = attemptUntil(io, slot.client, query, response_buf, expiry_at, &race); const completed_at = std.Io.Clock.awake.now(io); - const response = result catch |err| switch (transport.group(err)) { - .peer_fault => { - log.debug("{f}", .{AttemptFailure{ .endpoint = entry.endpoint, .err = err }}); - self.recordFailure(io, entry, completed_at, err); - last_fault = err; - continue; - }, - // The next upstream would hit the same wall, and this says - // nothing about any peer, so it is never recorded. - .local_resource => return err, - .cancellation => return error.Canceled, + const response = result catch |err| { + // A censored observation: the pool cut the attempt short, + // so the expiry is evidence about this budget and about no + // peer. A leaf's own `error.Timeout` is not this case — the + // race tag is what tells the two apart. + if (truncated and race == .expired) return error.BudgetExhausted; + switch (transport.group(err)) { + .peer_fault => { + log.debug("{f}", .{AttemptFailure{ .endpoint = entry.endpoint, .err = err }}); + self.recordFailure(io, entry, completed_at, err); + selected.* = entry.endpoint.url; + last_fault = err; + continue; + }, + // The next upstream would hit the same wall, and this + // says nothing about any peer, so it is never recorded. + .local_resource => return err, + .cancellation => return error.Canceled, + // The caller's budget, not this endpoint's conduct. + .budget_exhausted => return err, + } }; self.recordSuccess(io, entry, completed_at); - return response.len; + selected.* = entry.endpoint.url; + return response; } if (attempted) break; } @@ -375,6 +556,67 @@ pub const Pool = struct { return error.ConnectFailed; } + /// Takes a permit on the next entry to attempt, and returns its index. + /// `null` means this pass has no candidate left. + /// + /// Pass one sweeps: every eligible candidate is offered a non-blocking + /// `tryAcquire` in priority order, including ones an earlier sweep step + /// skipped, since a slot may have freed since. That sweep is what keeps a + /// saturated entry from blocking the query. Blocking is the fallback, not + /// the first move, and it waits on the highest-priority candidate only. + /// + /// Pass two probes instead: it takes the next enabled entry in priority + /// order and blocks for its permit, one entry at a time. Probing a + /// backed-off entry is already a last resort, so the pass keeps the order + /// the operator configured rather than letting a saturated higher-priority + /// candidate be skipped over — the deadline, not a lock miss, is what ends + /// it. + fn admit( + self: *Pool, + io: std.Io, + spent: std.StaticBitSet(max_entries), + pass: u8, + now: std.Io.Timestamp, + deadline: std.Io.Clock.Timestamp, + ) transport.ExchangeError!?usize { + if (pass != 0) { + for (0..self.entries.len) |index| { + if (spent.isSet(index)) continue; + if (!self.entries[index].enabled) continue; + return try self.admitBlocking(io, index, deadline); + } + return null; + } + + var first_eligible: ?usize = null; + for (self.entries, 0..) |*entry, index| { + if (spent.isSet(index)) continue; + if (!entry.enabled) continue; + if (!self.entryAvailable(io, entry, now)) continue; + if (entry.admission.tryAcquire(io)) return index; + if (first_eligible == null) first_eligible = index; + } + + const index = first_eligible orelse return null; + return try self.admitBlocking(io, index, deadline); + } + + /// Blocks for one entry's permit until `deadline`, counting the wait. + fn admitBlocking( + self: *Pool, + io: std.Io, + index: usize, + deadline: std.Io.Clock.Timestamp, + ) transport.ExchangeError!usize { + const entry = &self.entries[index]; + const started = std.Io.Clock.awake.now(io); + defer entry.recordQueued(io, started); + switch (try entry.admission.acquireUntil(io, deadline)) { + .acquired => return index, + .expired => return error.BudgetExhausted, + } + } + /// Copies health into `out` in pool order; returns the number written. pub fn snapshot(self: *Pool, io: std.Io, out: []Snapshot) std.Io.Cancelable!usize { const now = std.Io.Clock.awake.now(io); @@ -405,26 +647,6 @@ pub const Pool = struct { return count; } - /// One exchange raced against the per-attempt budget. - /// - /// The leaf client reports an identity of its own, which the pool discards: - /// the entry's endpoint is the pool's own naming of the same resolver, and - /// it is what the caller was handed. A leaf writing its identity into the - /// caller's slot would let a test fake overwrite the endpoint that actually - /// answered. - fn attempt( - self: *Pool, - io: std.Io, - entry_client: transport.Client, - query: []const u8, - response_buf: []u8, - ) transport.ExchangeError![]u8 { - var leaf_selected: ?[]const u8 = null; - return transport.raceWithin(io, self.timeouts.attempt, transport.Client.exchange, .{ - entry_client, io, query, response_buf, &leaf_selected, - }); - } - fn entryAvailable( self: *Pool, io: std.Io, @@ -623,7 +845,7 @@ fn testEntrySlots(url: []const u8, fake: *Fake, priority: i32, slot_count: usize .priority = priority, .enabled = true, .health = .init, - .sem = .{ .permits = slot_count }, + .admission = .{ .permits = slot_count }, .reuse_recoveries = &fake.recoveries, }; } @@ -677,6 +899,97 @@ test "the failed-attempt line names an upstream by a url carrying no credential" ); } +test "a canceled admission waiter holds nothing and its peer still wakes" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var admission: Admission = .{ .permits = 0 }; + const far: std.Io.Clock.Timestamp = .fromNow(io, .{ .raw = .fromSeconds(30), .clock = .awake }); + + var canceled = io.concurrent(Admission.acquireUntil, .{ &admission, io, far }) catch |err| switch (err) { + error.ConcurrencyUnavailable => return error.SkipZigTest, + }; + var peer = io.concurrent(Admission.acquireUntil, .{ &admission, io, far }) catch |err| switch (err) { + error.ConcurrencyUnavailable => return error.SkipZigTest, + }; + defer _ = peer.await(io) catch Admission.Outcome.expired; + + // Long enough that both tasks are provably inside the wait: neither can + // reach a permit, because none exists yet. + const settle: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(20), .clock = .awake }; + try settle.sleep(io); + try testing.expectError(error.Canceled, canceled.cancel(io)); + + // One permit, one signal. The canceled task may have absorbed that signal + // on its way out; if it did not hand it back, the peer would wait here + // forever and this test would hang. + admission.release(io); + try testing.expectEqual(Admission.Outcome.acquired, try peer.await(io)); + try testing.expectEqual(@as(usize, 0), admission.permits); +} + +fn holdAdmissionMutex(admission: *Admission, io: std.Io, hold: std.Io.Clock.Duration) void { + admission.mutex.lockUncancelable(io); + defer admission.mutex.unlock(io); + hold.sleep(io) catch {}; +} + +test "a contended admission mutex does not carry acquireUntil past its deadline" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + // One permit, but the mutex guarding it is held far past the caller's + // deadline. Every take must therefore run inside the deadline race: a + // blocking lock before the race starts would return `.acquired` only once + // the holder let go, long after the budget was gone. + var admission: Admission = .{ .permits = 1 }; + const hold: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(200), .clock = .awake }; + var holder = io.concurrent(holdAdmissionMutex, .{ &admission, io, hold }) catch |err| switch (err) { + error.ConcurrencyUnavailable => return error.SkipZigTest, + }; + defer holder.await(io); + + const settle: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(20), .clock = .awake }; + try settle.sleep(io); + + const deadline: std.Io.Clock.Timestamp = .fromNow(io, settle); + // `.expired` is the whole assertion: the permit is free, so a take that + // waited on the mutex outside the race would report `.acquired` instead, + // hundreds of milliseconds after the deadline. The exit path itself still + // waits for the mutex, uncancelably, to re-signal the condition. + try testing.expectEqual(Admission.Outcome.expired, try admission.acquireUntil(io, deadline)); + try testing.expectEqual(@as(usize, 1), admission.permits); +} + +test "an admission expiry racing a release leaves the permit count exact" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + // The tie the design exists for: the permit is taken by the caller under + // the mutex only after the race resolves, so a discarded wait can lose a + // notification but never a permit. Repeated because the interleaving is a + // race — a leak would show as a count that drifts below one. + const tick: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(2), .clock = .awake }; + for (0..50) |_| { + var admission: Admission = .{ .permits = 0 }; + const deadline: std.Io.Clock.Timestamp = .fromNow(io, tick); + var waiter = io.concurrent(Admission.acquireUntil, .{ &admission, io, deadline }) catch |err| switch (err) { + error.ConcurrencyUnavailable => return error.SkipZigTest, + }; + try tick.sleep(io); + admission.release(io); + + switch (try waiter.await(io)) { + .acquired => admission.release(io), + .expired => {}, + } + try testing.expectEqual(@as(usize, 1), admission.permits); + } +} + test "Pool satisfies the Client interface" { var threaded: std.Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); @@ -745,11 +1058,14 @@ test "an all-failed exchange reports the last endpoint attempted" { try testing.expectEqualStrings("https://last.example/dns-query", selected.?); } -test "a timeout mid-flight reports the endpoint the query was in" { +test "a truncated attempt blames the budget, not the endpoint it was in" { var threaded: std.Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); + // The total budget is shorter than one attempt, so the first attempt is + // truncated: the endpoint never got the observation interval it was + // configured to get, and an expiry says nothing about it. var stalling: Fake = .{ .behavior = .{ .slow = .{ .duration = .{ .raw = .fromSeconds(30), .clock = .awake }, .reply = response_bytes, @@ -766,9 +1082,45 @@ test "a timeout mid-flight reports the endpoint the query was in" { var buf: [512]u8 = undefined; var selected: ?[]const u8 = null; - try testing.expectError(error.Timeout, pool.exchange(io, query_bytes, &buf, &selected)); - try testing.expectEqualStrings("https://stalling.example/dns-query", selected.?); + try testing.expectError( + error.BudgetExhausted, + pool.exchange(io, query_bytes, &buf, &selected), + ); + // Nothing is attributable: no outcome was recorded, so no endpoint is + // named and no health moved. + try testing.expect(selected == null); + try testing.expectEqual(@as(u64, 0), entries[0].health.total_failures); + try testing.expectEqual(@as(u32, 0), entries[0].health.consecutive_failures); + try testing.expectEqual(@as(?std.Io.Timestamp, null), entries[0].health.last_error_at); try testing.expectEqual(@as(usize, 0), untouched.calls.load(.acquire)); + try testing.expectEqual(@as(u64, 1), pool.budgetExhaustedTotal()); +} + +test "a truncated attempt that fails on its own still blames the endpoint" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + // Same truncated budget, but the attempt completes with a peer fault of + // its own before the deadline. A completed failure is real evidence + // whatever budget it ran with, so health moves and `selected` names it. + var refusing: Fake = .{ .behavior = .{ .fail = error.ConnectFailed } }; + var entries = [_]Entry{testEntry("https://refusing.example/dns-query", &refusing, 10)}; + var pool: Pool = .init(&entries, test_cfg, .{ + .attempt = .{ .raw = .fromMilliseconds(200), .clock = .awake }, + .total = .{ .raw = .fromMilliseconds(60), .clock = .awake }, + }, 1); + + var buf: [512]u8 = undefined; + var selected: ?[]const u8 = null; + try testing.expectError( + error.ConnectFailed, + pool.exchange(io, query_bytes, &buf, &selected), + ); + try testing.expectEqualStrings("https://refusing.example/dns-query", selected.?); + try testing.expectEqual(@as(u64, 1), entries[0].health.total_failures); + try testing.expectEqualStrings("ConnectFailed", entries[0].health.lastError()); + try testing.expectEqual(@as(u64, 0), pool.budgetExhaustedTotal()); } test "an exchange that attempted nothing reports no endpoint" { @@ -925,6 +1277,9 @@ test "a cancellation short-circuits and records nothing" { try testing.expectError(error.Canceled, pool.exchange(io, query_bytes, &buf, &selected)); try testing.expectEqual(@as(usize, 0), good.calls.load(.acquire)); try testing.expectEqual(@as(u64, 0), entries[0].health.total_failures); + // A torn-down task is not a budget that ran out, and it names no endpoint. + try testing.expectEqual(@as(u64, 0), pool.budgetExhaustedTotal()); + try testing.expect(selected == null); } test "an attempt that outruns the budget is a recorded Timeout" { @@ -959,15 +1314,251 @@ test "an attempt that outruns the budget is a recorded Timeout" { try testing.expectEqual(@as(u64, 1), entries[1].health.total_successes); } +test "a stalled primary fails over to a fast standby inside the total budget" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + // The shipped shape of the defaults: attempt is half of total, so the + // primary gets one whole attempt and the standby still has room to answer + // inside the same deadline. + var stalled: Fake = .{ .behavior = .{ .slow = .{ + .duration = .{ .raw = .fromSeconds(30), .clock = .awake }, + .reply = response_bytes, + } } }; + var standby: Fake = .{ .behavior = .{ .reply = alt_response_bytes } }; + var entries = [_]Entry{ + testEntry("https://stalled.example/dns-query", &stalled, 10), + testEntry("https://standby.example/dns-query", &standby, 20), + }; + var pool: Pool = .init(&entries, test_cfg, .{ + .attempt = .{ .raw = .fromMilliseconds(150), .clock = .awake }, + .total = .{ .raw = .fromMilliseconds(300), .clock = .awake }, + }, 1); + + var buf: [512]u8 = undefined; + var selected: ?[]const u8 = null; + const reply = try pool.exchange(io, query_bytes, &buf, &selected); + + try testing.expectEqualSlices(u8, alt_response_bytes, reply); + try testing.expectEqualStrings("https://standby.example/dns-query", selected.?); + // The primary got its full configured interval and did not answer, which is + // evidence about the primary — so it is recorded, unlike a truncated one. + try testing.expectEqual(@as(u32, 1), entries[0].health.consecutive_failures); + try testing.expectEqualStrings("Timeout", entries[0].health.lastError()); + try testing.expectEqual(@as(u64, 0), pool.budgetExhaustedTotal()); +} + +test "pass two probes every backed-off entry under the one deadline" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + const cfg: health.Config = .{ + .failure_threshold = 1, + .base_backoff_ms = 60_000, + .max_backoff_ms = 60_000, + }; + var first: Fake = .{ .behavior = .{ .fail = error.ConnectFailed } }; + var second: Fake = .{ .behavior = .{ .fail = error.ConnectFailed } }; + var entries = [_]Entry{ + testEntry("https://first.example/dns-query", &first, 10), + testEntry("https://second.example/dns-query", &second, 20), + }; + + // One exchange under generous budgets puts both entries into backoff. + var seeding_pool: Pool = .init(&entries, cfg, test_timeouts, 1); + var buf: [512]u8 = undefined; + var selected: ?[]const u8 = null; + try testing.expectError( + error.ConnectFailed, + seeding_pool.exchange(io, query_bytes, &buf, &selected), + ); + try testing.expect(entries[0].health.backoff_until != null); + try testing.expect(entries[1].health.backoff_until != null); + + // Now both stall. Pass one has no candidate at all, so pass two probes + // them in order — sharing the one deadline rather than granting each probe + // a fresh attempt budget. + first.behavior = .{ .slow = .{ + .duration = .{ .raw = .fromSeconds(30), .clock = .awake }, + .reply = response_bytes, + } }; + second.behavior = first.behavior; + var probing_pool: Pool = .init(&entries, cfg, .{ + .attempt = .{ .raw = .fromMilliseconds(100), .clock = .awake }, + .total = .{ .raw = .fromMilliseconds(150), .clock = .awake }, + }, 1); + + const started = std.Io.Clock.awake.now(io); + selected = null; + try testing.expectError( + error.BudgetExhausted, + probing_pool.exchange(io, query_bytes, &buf, &selected), + ); + const elapsed_ns = std.Io.Clock.awake.now(io).nanoseconds - started.nanoseconds; + + // Both were probed, and the pair cost the total budget rather than one + // attempt budget each. + try testing.expectEqual(@as(usize, 2), first.calls.load(.acquire)); + try testing.expectEqual(@as(usize, 2), second.calls.load(.acquire)); + try testing.expect(elapsed_ns < @as(i96, 200) * std.time.ns_per_ms); + // The first probe ran a full attempt and is recorded; the second was cut + // short by the deadline and is not, so `selected` still names the first. + try testing.expectEqualStrings("https://first.example/dns-query", selected.?); + try testing.expectEqual(@as(u64, 2), entries[0].health.total_failures); + try testing.expectEqual(@as(u64, 1), entries[1].health.total_failures); + try testing.expectEqual(@as(u64, 1), probing_pool.budgetExhaustedTotal()); +} + +test "an exchange whose deadline is already gone starts no attempt" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + // A budget of nothing: the entry is healthy and its permit is free, so + // admission and the health recheck both succeed, and only the remaining-time + // check stands between the loop and an attempt. Racing a leaf against a + // timestamp that has already passed is a coin toss — an instantly answering + // fake can win it — so a pool that omits the check manufactures a success + // out of a budget that was already spent. + var fake: Fake = .{ .behavior = .{ .reply = response_bytes } }; + var entries = [_]Entry{testEntry("https://only.example/dns-query", &fake, 10)}; + var pool: Pool = .init(&entries, test_cfg, .{ + .attempt = .{ .raw = .fromMilliseconds(0), .clock = .awake }, + .total = .{ .raw = .fromMilliseconds(0), .clock = .awake }, + }, 1); + + var buf: [512]u8 = undefined; + var selected: ?[]const u8 = null; + try testing.expectError( + error.BudgetExhausted, + pool.exchange(io, query_bytes, &buf, &selected), + ); + try testing.expectEqual(@as(usize, 0), fake.calls.load(.acquire)); + try testing.expect(selected == null); + try testing.expectEqual(@as(u64, 0), entries[0].health.total_failures); + try testing.expectEqual(@as(u64, 0), entries[0].health.total_successes); + try testing.expectEqual(@as(u64, 1), pool.budgetExhaustedTotal()); + // The permit was released on the way out, so the entry is usable again. + try testing.expectEqual(@as(usize, 1), entries[0].admission.permits); +} + +/// The pass-two probing test's numbers: the holder's stall outlives the +/// probe's whole budget, so the probe provably died waiting rather than after +/// being served. +const probe_holder_stall_ms = 800; +const probe_order_total_ms = 200; + +test "pass two probes in priority order rather than stepping over a saturated entry" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + const cfg: health.Config = .{ + .failure_threshold = 1, + .base_backoff_ms = 60_000, + .max_backoff_ms = 60_000, + }; + var first: Fake = .{ .behavior = .{ .fail = error.ConnectFailed } }; + var second: Fake = .{ .behavior = .{ .fail = error.ConnectFailed } }; + var entries = [_]Entry{ + testEntry("https://first.example/dns-query", &first, 10), + testEntry("https://second.example/dns-query", &second, 20), + }; + + // One exchange under generous budgets puts both entries into backoff, so + // every exchange after it reaches pass two. + var seeding_pool: Pool = .init(&entries, cfg, test_timeouts, 1); + var buf: [512]u8 = undefined; + var selected: ?[]const u8 = null; + try testing.expectError( + error.ConnectFailed, + seeding_pool.exchange(io, query_bytes, &buf, &selected), + ); + try testing.expect(entries[0].health.backoff_until != null); + try testing.expect(entries[1].health.backoff_until != null); + + first.behavior = .{ .slow = .{ + .duration = .{ .raw = .fromMilliseconds(probe_holder_stall_ms), .clock = .awake }, + .reply = response_bytes, + } }; + second.behavior = .{ .reply = alt_response_bytes }; + + var probe_pool: Pool = .init(&entries, cfg, .{ + .attempt = .{ .raw = .fromMilliseconds(probe_order_total_ms), .clock = .awake }, + .total = .{ .raw = .fromMilliseconds(probe_order_total_ms), .clock = .awake }, + }, 1); + + var holder_buf: [512]u8 = undefined; + var holder = io.concurrent(exchangeAttributed, .{ &seeding_pool, io, &holder_buf }) catch |err| switch (err) { + error.ConcurrencyUnavailable => return error.SkipZigTest, + }; + defer _ = holder.await(io) catch Attributed.discarded; + // The first entry's one permit is provably taken before the probe asks. + try awaitInFlight(io, &first, 1); + + const calls_before = second.calls.load(.acquire); + var probe_buf: [512]u8 = undefined; + selected = null; + try testing.expectError( + error.BudgetExhausted, + probe_pool.exchange(io, query_bytes, &probe_buf, &selected), + ); + + // The claim: pass two waits for the higher-priority entry it is probing. + // A sweep that stepped over the saturated entry would have been answered + // instantly by the second one instead. + try testing.expectEqual(calls_before, second.calls.load(.acquire)); + try testing.expect(selected == null); + try testing.expectEqual(@as(u64, 1), probe_pool.budgetExhaustedTotal()); + + const held = try holder.await(io); + try testing.expectEqualSlices(u8, response_bytes, holder_buf[0..held.reply_len]); +} + +test "budget_exhausted_total counts exhausted exchanges once each" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var stalling: Fake = .{ .behavior = .{ .slow = .{ + .duration = .{ .raw = .fromSeconds(30), .clock = .awake }, + .reply = response_bytes, + } } }; + var entries = [_]Entry{testEntry("https://stalling.example/dns-query", &stalling, 10)}; + var pool: Pool = .init(&entries, test_cfg, .{ + .attempt = .{ .raw = .fromMilliseconds(200), .clock = .awake }, + .total = .{ .raw = .fromMilliseconds(40), .clock = .awake }, + }, 1); + + var buf: [512]u8 = undefined; + var selected: ?[]const u8 = null; + try testing.expectEqual(@as(u64, 0), pool.budgetExhaustedTotal()); + for (1..4) |expected| { + try testing.expectError( + error.BudgetExhausted, + pool.exchange(io, query_bytes, &buf, &selected), + ); + // Once per exchange, never once per endpoint attempted inside it. + try testing.expectEqual(@as(u64, expected), pool.budgetExhaustedTotal()); + } + + // An exchange that answers adds nothing. + stalling.behavior = .{ .reply = response_bytes }; + _ = try pool.exchange(io, query_bytes, &buf, &selected); + try testing.expectEqual(@as(u64, 3), pool.budgetExhaustedTotal()); +} + test "two stalling upstreams cost the total budget, not one budget each" { var threaded: std.Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); - // Both entries stall past their own attempt budget, so without the outer - // race the exchange would take two attempt budgets and then some. The + // Both entries stall past their own attempt budget, so an exchange that + // gave each one a fresh budget would cost two of them and then some. The // total is set below two attempts, which is what makes the assertion mean - // something: only the outer deadline can end this call in time. + // something: only the one deadline the loop owns can end this call in time. var first: Fake = .{ .behavior = .{ .slow = .{ .duration = .{ .raw = .fromSeconds(30), .clock = .awake }, .reply = response_bytes, @@ -988,15 +1579,18 @@ test "two stalling upstreams cost the total budget, not one budget each" { var buf: [512]u8 = undefined; const started = std.Io.Clock.awake.now(io); var selected: ?[]const u8 = null; - try testing.expectError(error.Timeout, pool.exchange(io, query_bytes, &buf, &selected)); + try testing.expectError( + error.BudgetExhausted, + pool.exchange(io, query_bytes, &buf, &selected), + ); const elapsed_ns = std.Io.Clock.awake.now(io).nanoseconds - started.nanoseconds; - // Under one attempt budget, so the outer deadline is provably what fired. - // The bound is generous against a loaded CI box; the failure it catches is - // a whole extra attempt, not a scheduling hiccup. + // Under one attempt budget, so the deadline is provably what fired. The + // bound is generous against a loaded CI box; the failure it catches is a + // whole extra attempt, not a scheduling hiccup. try testing.expect(elapsed_ns < @as(i96, 200) * std.time.ns_per_ms); - // The second entry was never reached: the loop was canceled inside the - // first attempt. + // The second entry was never reached: the deadline landed inside the first + // attempt, which is a truncated one and so ends the exchange. try testing.expectEqual(@as(usize, 0), second.calls.load(.acquire)); } @@ -1263,7 +1857,7 @@ test "a task that waits for a saturated entry is counted as queued" { 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" { +test "a waiter that spends its whole budget queueing blames nobody but the budget" { var threaded: std.Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); @@ -1301,15 +1895,24 @@ test "a waiter canceled by the total budget is counted and still returns its per // 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. + // This call never reaches the fake: it is the only entry, so there is no + // admissible alternative and the call waits for the entry's one permit + // until its own budget expires. That is the field failure — a burst dying + // in the queue — and it has to be visible. var selected: ?[]const u8 = null; - try testing.expectError(error.Timeout, waiter_pool.exchange(io, query_bytes, &buf_b, &selected)); + try testing.expectError( + error.BudgetExhausted, + waiter_pool.exchange(io, query_bytes, &buf_b, &selected), + ); + // Nothing ran, so nothing is named. A query log row blaming this upstream + // for a queue it never entered is the reporting defect this replaces. + try testing.expect(selected == null); + try testing.expectEqual(@as(u64, 1), waiter_pool.budgetExhaustedTotal()); + try testing.expectEqual(@as(u64, 0), entries[0].health.total_failures); // Where it died, not just that it died: the holder's call is still the only - // one any leaf client has seen, so the cancellation landed inside - // `Semaphore.wait` rather than inside an attempt. + // one any leaf client has seen, so the budget ran out inside the admission + // wait rather than inside an attempt. try testing.expectEqual(@as(usize, 1), fake.calls.load(.acquire)); try testing.expectEqual(@as(u64, 1), entries[0].queued_total.load(.acquire)); try testing.expect(entries[0].queued_ns_total.load(.acquire) > 0); @@ -1319,7 +1922,7 @@ test "a waiter canceled by the total budget is counted and still returns its per 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 + // The expired waiter returned the permit it never held, so the entry is // usable again — and the fresh call is the second one to reach the fake. fake.behavior = .{ .reply = response_bytes }; const reply = try waiter_pool.exchange(io, query_bytes, &buf_b, &selected); @@ -1327,6 +1930,72 @@ test "a waiter canceled by the total budget is counted and still returns its per try testing.expectEqual(@as(usize, 2), fake.calls.load(.acquire)); } +/// The saturated-primary test's numbers. The holders stall longer than the +/// probe's whole budget, so the probe provably cannot be waiting for one of +/// their slots; the stall still ends inside the test so the holders' futures +/// can be awaited rather than abandoned. +const saturating_stall_ms = 800; +const probe_attempt_ms = 150; +const probe_total_ms = 300; + +test "a saturated primary defers to a standby that has capacity" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + // The Pi reproduction: the primary is healthy but every slot is held by a + // slow exchange, and the standby is idle. Priority orders the entries that + // can be admitted now; it does not entitle a saturated entry to hold a + // query until the whole budget is gone. + var primary: Fake = .{ .behavior = .{ .slow = .{ + .duration = .{ .raw = .fromMilliseconds(saturating_stall_ms), .clock = .awake }, + .reply = response_bytes, + } } }; + var standby: Fake = .{ .behavior = .{ .reply = alt_response_bytes } }; + var entries = [_]Entry{ + testEntrySlots("https://primary.example/dns-query", &primary, 10, 2), + testEntry("https://standby.example/dns-query", &standby, 20), + }; + + // Two pools over one `entries` slice, as the queued-waiter test above does: + // the holders need a budget long enough to finish, the probe a budget short + // enough that waiting for a slot would be fatal. + var holder_pool: Pool = .init(&entries, test_cfg, test_timeouts, 1); + var probe_pool: Pool = .init(&entries, test_cfg, .{ + .attempt = .{ .raw = .fromMilliseconds(probe_attempt_ms), .clock = .awake }, + .total = .{ .raw = .fromMilliseconds(probe_total_ms), .clock = .awake }, + }, 1); + + var bufs: [2][512]u8 = undefined; + var holders: [2]std.Io.Future(transport.ExchangeError!Attributed) = undefined; + for (&holders, &bufs) |*holder, *buf| { + holder.* = io.concurrent(exchangeAttributed, .{ &holder_pool, io, buf }) catch |err| switch (err) { + error.ConcurrencyUnavailable => return error.SkipZigTest, + }; + } + defer for (&holders) |*holder| { + _ = holder.await(io) catch Attributed.discarded; + }; + // Both permits provably taken before the probe asks for one. + try awaitInFlight(io, &primary, 2); + + var probe_buf: [512]u8 = undefined; + var selected: ?[]const u8 = null; + const started = std.Io.Clock.awake.now(io); + const reply = try probe_pool.exchange(io, query_bytes, &probe_buf, &selected); + const elapsed_ns = std.Io.Clock.awake.now(io).nanoseconds - started.nanoseconds; + + try testing.expectEqualSlices(u8, alt_response_bytes, reply); + try testing.expectEqualStrings("https://standby.example/dns-query", selected.?); + // Well inside the probe's own budget: it never queued on the primary. + try testing.expect(elapsed_ns < @as(i96, probe_total_ms) * std.time.ns_per_ms); + try testing.expectEqual(@as(usize, 1), standby.calls.load(.acquire)); + // The primary was never entered a third time, and never counted a failure: + // it was skipped for want of capacity, not blamed for anything. + try testing.expectEqual(@as(usize, 2), primary.calls.load(.acquire)); + try testing.expectEqual(@as(u64, 0), entries[0].health.total_failures); +} + /// The two entries of the test below, each answering with bytes only it /// produces so a call's reply proves which entry served it independently of /// what the pool reported. @@ -1349,16 +2018,20 @@ test "overlapping exchanges each report the entry that answered that call" { defer threaded.deinit(); const io = threaded.io(); - // The first entry fails the task that reaches it first, slowly enough that - // the second task is queued behind it, then answers that second task. One - // failure is under `test_cfg`'s threshold of two, so no backoff steers the - // waiting task away and the two calls end on different entries. + // The first entry has room for both tasks and treats them differently: it + // fails whichever reaches it first and answers the other, slowly enough + // that the two calls are in flight together. One failure is under + // `test_cfg`'s threshold of two, so no backoff steers the failed-over task + // away and the two calls end on different entries. var first_entry: Fake = .{ .behavior = .{ .slow_fail = .{ .duration = .{ .raw = .fromMilliseconds(50), .clock = .awake }, .err = error.ConnectFailed, } }, - .then = .{ .reply = alt_response_bytes }, + .then = .{ .slow = .{ + .duration = .{ .raw = .fromMilliseconds(100), .clock = .awake }, + .reply = alt_response_bytes, + } }, }; // Slow too, so the failed-over call is still in flight while the other call // is being answered — a shared identity would be overwritten under it. @@ -1367,7 +2040,7 @@ test "overlapping exchanges each report the entry that answered that call" { .reply = response_bytes, } } }; var entries = [_]Entry{ - testEntry(divergent_first_url, &first_entry, 10), + testEntrySlots(divergent_first_url, &first_entry, 10, 2), testEntry(divergent_second_url, &second_entry, 20), }; var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1); @@ -1396,7 +2069,9 @@ test "overlapping exchanges each report the entry that answered that call" { 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)); + // Both tasks were admitted to the first entry at once, which is what makes + // them concurrent rather than serialized behind one permit. + try testing.expectEqual(@as(u32, 2), first_entry.peak_in_flight.load(.acquire)); try testing.expectEqual(@as(u64, 1), entries[0].health.total_failures); try testing.expectEqual(@as(u64, 1), entries[0].health.total_successes); try testing.expectEqual(@as(u64, 1), entries[1].health.total_successes); diff --git a/src/upstream/transport.zig b/src/upstream/transport.zig index 2ddabf4..3993cd5 100644 --- a/src/upstream/transport.zig +++ b/src/upstream/transport.zig @@ -1,4 +1,4 @@ -//! Shared vocabulary for every upstream client: endpoint URLs, the three +//! Shared vocabulary for every upstream client: endpoint URLs, the four //! disjoint failure groups, the `Client` interface, and response validation. //! //! Everything here except the `Client` vtable is pure. `validateResponse` takes @@ -8,8 +8,10 @@ //! //! The failure classification is the reason this file exists. Health and //! backoff must count only what the peer did wrong: a local `OutOfMemory` says -//! nothing about the upstream, and `error.Canceled` says nothing at all. The -//! three error sets below are disjoint by construction and `group` switches +//! nothing about the upstream, `error.Canceled` says nothing at all, and +//! `error.BudgetExhausted` says the caller ran out of time before the peer was +//! given its interval. The four error sets below are disjoint by construction +//! and `group` switches //! over them exhaustively, so a new failure mode cannot silently land in the //! wrong bucket. @@ -188,9 +190,15 @@ pub const LocalResource = error{ pub const Cancellation = error{Canceled}; -pub const ExchangeError = PeerFault || LocalResource || Cancellation; +/// The caller's own time ran out before any peer could be given the observation +/// interval it was configured to get. Evidence about this process's budget, not +/// about any endpoint, so it is never recorded against health — that is the +/// whole reason it is not a `PeerFault`. +pub const BudgetFault = error{BudgetExhausted}; -pub const Group = enum { peer_fault, local_resource, cancellation }; +pub const ExchangeError = PeerFault || LocalResource || Cancellation || BudgetFault; + +pub const Group = enum { peer_fault, local_resource, cancellation, budget_exhausted }; /// Exhaustive switch over `ExchangeError` — no `else` arm. A new error member /// must break the build here, so no failure can silently land in the wrong @@ -218,6 +226,8 @@ pub fn group(err: ExchangeError) Group { => .local_resource, error.Canceled => .cancellation, + + error.BudgetExhausted => .budget_exhausted, }; } @@ -269,29 +279,29 @@ pub fn closeBlocked(io: std.Io, target: anytype) void { } } -/// The payload of `f`'s return type, which `raceWithin` requires to be +/// The payload of `f`'s return type, which the race harness requires to be /// `ExchangeError!T`. A raced function with any other error set would let a /// failure reach the pool without passing through `group`. fn RacedPayload(comptime f: anytype) type { const info = @typeInfo(@TypeOf(f)); - if (info != .@"fn") @compileError("raceWithin needs a function, found " ++ @typeName(@TypeOf(f))); + if (info != .@"fn") @compileError("the race harness needs a function, found " ++ @typeName(@TypeOf(f))); const Return = info.@"fn".return_type orelse - @compileError("raceWithin needs a function with a concrete return type"); + @compileError("the race harness needs a function with a concrete return type"); const union_info = switch (@typeInfo(Return)) { .error_union => |u| u, - else => @compileError("raceWithin needs `ExchangeError!T`, found " ++ @typeName(Return)), + else => @compileError("the race harness needs `ExchangeError!T`, found " ++ @typeName(Return)), }; if (union_info.error_set != ExchangeError) - @compileError("raceWithin needs `ExchangeError!T`, found " ++ @typeName(Return)); + @compileError("the race harness needs `ExchangeError!T`, found " ++ @typeName(Return)); return union_info.payload; } /// Runs `f(args...)` raced against `budget`, and cancels the loser. /// /// No stream read or write in 0.16.0 takes a timeout, so a deadline is a second -/// task rather than a socket option. This is the one copy of that harness: the -/// pool races an attempt and its whole failover loop through it, and the -/// forward client races its TCP exchange. +/// task rather than a socket option. `raceUntilTagged` is the one copy of that +/// harness; this is the untagged wrapper for callers that own no deadline and +/// only need "a bound on this one operation". /// /// `error.Timeout` means the budget won. A canceled sleep means the whole task /// is being torn down rather than the budget running out, so it stays @@ -303,33 +313,69 @@ pub fn raceWithin( comptime f: anytype, args: anytype, ) ExchangeError!RacedPayload(f) { - const Outcome = union(enum) { + var outcome: RaceOutcome = .completed; + return raceUntilTagged(io, .fromNow(io, budget), &outcome, f, args); +} + +/// Which side of the race ended the call. +/// +/// `completed` means the raced operation itself returned — including when what +/// it returned is `error.Timeout`, which is then the peer's own timeout and +/// real evidence about that peer. `expired` means the caller's clock ran out +/// with the operation still in flight, which is evidence about the budget only. +pub const RaceOutcome = enum { completed, expired }; + +/// `raceWithin` with the two timer origins told apart, and with an ABSOLUTE +/// expiry rather than a duration. +/// +/// The timestamp is the point of the whole function. A duration recomputed from +/// a deadline and then slept re-anchors at "now", so every re-race drifts a +/// little past the caller's real deadline and a truncated attempt is then +/// indistinguishable from a full one. The caller that owns the deadline +/// computes the instant once and passes it here. +/// +/// `outcome` is written before this returns on both racing paths. It is left +/// untouched when the race cannot start at all (`error.SystemResources`) or +/// when the whole task is being canceled, since neither is an observation about +/// this budget; callers initialize it to the value they want in those cases. +pub fn raceUntilTagged( + io: std.Io, + expiry_at: std.Io.Clock.Timestamp, + outcome: *RaceOutcome, + comptime f: anytype, + args: anytype, +) ExchangeError!RacedPayload(f) { + const Slot = union(enum) { raced: ExchangeError!RacedPayload(f), expiry: std.Io.Cancelable!void, }; - var outcomes: [2]Outcome = undefined; - var race: std.Io.Select(Outcome) = .init(io, &outcomes); + var slots: [2]Slot = undefined; + var race: std.Io.Select(Slot) = .init(io, &slots); defer race.cancelDiscard(); race.concurrent(.raced, f, args) catch |err| switch (err) { error.ConcurrencyUnavailable => return error.SystemResources, }; - race.concurrent(.expiry, expire, .{ io, budget }) catch |err| switch (err) { + race.concurrent(.expiry, expire, .{ io, expiry_at }) catch |err| switch (err) { error.ConcurrencyUnavailable => return error.SystemResources, }; switch (try race.await()) { - .raced => |result| return result, + .raced => |result| { + outcome.* = .completed; + return result; + }, .expiry => |result| { try result; + outcome.* = .expired; return error.Timeout; }, } } -fn expire(io: std.Io, budget: std.Io.Clock.Duration) std.Io.Cancelable!void { - return budget.sleep(io); +fn expire(io: std.Io, expiry_at: std.Io.Clock.Timestamp) std.Io.Cancelable!void { + return expiry_at.wait(io); } /// A thing that sends one DNS message and returns one validated DNS message. @@ -347,13 +393,21 @@ pub const Client = struct { /// Returns a prefix of `response_buf`. The returned message has already /// passed `validateResponse` against `query`. /// - /// `selected` names the resolver the exchange used. An implementation - /// writes it *before* each attempt, never after, so a failed exchange still - /// names the last resolver it tried — a SERVFAIL row without its resolver - /// explains nothing. The slice must outlive the call; every implementation - /// borrows storage it owns for at least the query's duration. Callers - /// initialize it to null: a `null` after the call means no resolver was - /// reached at all. + /// `selected` names the resolver the exchange used. A single-endpoint + /// implementation (DoH, DoT, the forward client, test fakes) may write it + /// *before* each attempt: it has one resolver and records no health, so + /// "the one I tried" is an honest answer even for a failure, and a SERVFAIL + /// row without its resolver explains nothing. + /// + /// `Pool` is stricter, and documents the rule on `Pool.exchange`: it names + /// only endpoints whose outcome it recorded, so a query that ran out of + /// budget blames nobody and may leave this `null`. Both are within this + /// contract — the guarantee here is that a non-null value names a resolver + /// this exchange really used, never that a failure leaves one behind. + /// + /// The slice must outlive the call; every implementation borrows storage it + /// owns for at least the query's duration. Callers initialize it to null: a + /// `null` after the call means no resolver is being reported. pub fn exchange( self: Client, io: std.Io, @@ -530,8 +584,8 @@ test "parse rejects a fragment" { try testing.expectError(error.BadUrl, Endpoint.parse("tls://dns.google#f")); } -test "the three error groups are disjoint" { - const sets = .{ PeerFault, LocalResource, Cancellation }; +test "the four error groups are disjoint" { + const sets = .{ PeerFault, LocalResource, Cancellation, BudgetFault }; inline for (sets, 0..) |a, i| { inline for (sets, 0..) |b, j| { if (i >= j) continue; @@ -548,7 +602,8 @@ test "the three error groups are disjoint" { // member added to two sets at once cannot pass unnoticed. const total = @typeInfo(PeerFault).error_set.?.len + @typeInfo(LocalResource).error_set.?.len + - @typeInfo(Cancellation).error_set.?.len; + @typeInfo(Cancellation).error_set.?.len + + @typeInfo(BudgetFault).error_set.?.len; try testing.expectEqual(total, @typeInfo(ExchangeError).error_set.?.len); } @@ -558,6 +613,7 @@ test "group classifies each member" { try testing.expectEqual(Group.local_resource, group(error.OutOfMemory)); try testing.expectEqual(Group.local_resource, group(error.BufferTooSmall)); try testing.expectEqual(Group.cancellation, group(error.Canceled)); + try testing.expectEqual(Group.budget_exhausted, group(error.BudgetExhausted)); } test "mapLocal folds only local and cancellation errors" { @@ -641,6 +697,72 @@ test "raceWithin passes the raced task's own failure through" { ); } +test "raceUntilTagged tells a leaf Timeout apart from an expiry" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + // The leaf's own timeout: it returned, so the peer really did time out and + // the outcome is `completed` even though the error is the same one an + // expiry produces. + var outcome: RaceOutcome = .expired; + const far: std.Io.Clock.Timestamp = .fromNow(io, .{ .raw = .fromSeconds(30), .clock = .awake }); + try testing.expectError( + error.Timeout, + raceUntilTagged(io, far, &outcome, racedReply, .{ + io, 0, @as(ExchangeError!usize, error.Timeout), + }), + ); + try testing.expectEqual(RaceOutcome.completed, outcome); + + // The expiry side, distinguishable only through the tag. + outcome = .completed; + const soon: std.Io.Clock.Timestamp = .fromNow(io, .{ .raw = .fromMilliseconds(20), .clock = .awake }); + try testing.expectError( + error.Timeout, + raceUntilTagged(io, soon, &outcome, racedReply, .{ + io, 30_000, @as(ExchangeError!usize, 7), + }), + ); + try testing.expectEqual(RaceOutcome.expired, outcome); +} + +test "raceUntilTagged tags a successful completion" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var outcome: RaceOutcome = .expired; + const far: std.Io.Clock.Timestamp = .fromNow(io, .{ .raw = .fromSeconds(30), .clock = .awake }); + const len = try raceUntilTagged(io, far, &outcome, racedReply, .{ + io, 0, @as(ExchangeError!usize, 7), + }); + try testing.expectEqual(@as(usize, 7), len); + try testing.expectEqual(RaceOutcome.completed, outcome); +} + +test "raceUntilTagged honours an expiry already in the past" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + // A deadline the caller has already spent: the expiry wins immediately + // rather than re-anchoring a duration at "now" and granting a fresh budget. + const past = std.Io.Clock.Timestamp.now(io, .awake) + .subDuration(.{ .raw = .fromSeconds(1), .clock = .awake }); + var outcome: RaceOutcome = .completed; + const started = std.Io.Clock.awake.now(io); + try testing.expectError( + error.Timeout, + raceUntilTagged(io, past, &outcome, racedReply, .{ + io, 30_000, @as(ExchangeError!usize, 7), + }), + ); + try testing.expectEqual(RaceOutcome.expired, outcome); + const elapsed_ns = std.Io.Clock.awake.now(io).nanoseconds - started.nanoseconds; + try testing.expect(elapsed_ns < @as(i96, 5) * std.time.ns_per_s); +} + test "closeBlocked closes a target of either close shape" { // The two shapes the transports use: a socket or a plain stream, which // closes through the `Io`, and a `TlsStream`, which owns the one it was diff --git a/src/web/metrics.zig b/src/web/metrics.zig index 1fac2a0..41d9c8f 100644 --- a/src/web/metrics.zig +++ b/src/web/metrics.zig @@ -166,6 +166,12 @@ pub const Sample = struct { udp_listener: ?udp_server.Snapshot = null, tcp_listener: ?tcp_server.Snapshot = null, upstreams: []const UpstreamSample = &.{}, + /// Exchanges the pool gave up on because the request's own budget ran out. + /// Pool-wide rather than per-upstream on purpose: budget exhaustion is + /// evidence about the pool, never about an endpoint, so it carries no url + /// label and cannot live in `UpstreamSample`. Absent while no pool is + /// wired, like every other collaborator. + upstream_budget_exhausted_total: ?u64 = null, }; pub fn handle( @@ -271,7 +277,10 @@ pub fn collect(state: *server.WebState, io: std.Io, arena: Allocator) Allocator. if (state.upstreams) |owner| { const generation = owner.acquire(io); defer owner.release(io, generation); - if (generation.pool) |pool| sample.upstreams = try upstreams(pool, io, arena); + if (generation.pool) |pool| { + sample.upstreams = try upstreams(pool, io, arena); + sample.upstream_budget_exhausted_total = pool.budgetExhaustedTotal(); + } } return sample; @@ -465,6 +474,15 @@ pub fn render(w: *std.Io.Writer, sample: Sample) std.Io.Writer.Error!void { if (sample.doh_certs != null or sample.dot_certs != null) try renderCerts(w, sample); if (sample.upstreams.len != 0) try renderUpstreams(w, sample.upstreams); + if (sample.upstream_budget_exhausted_total) |total| { + try labeledHead( + w, + "nxdns_upstream_budget_exhausted_total", + "Exchanges that ran out of their own total budget before any upstream answered.", + "counter", + ); + try w.print("nxdns_upstream_budget_exhausted_total {d}\n", .{total}); + } } fn renderCerts(w: *std.Io.Writer, sample: Sample) std.Io.Writer.Error!void { @@ -1540,7 +1558,7 @@ test "the queue families carry what a real pool recorded, through the real snaps .priority = 10, .enabled = true, .health = .init, - .sem = .{ .permits = slots.len }, + .admission = .{ .permits = slots.len }, .reuse_recoveries = &recoveries, }}; var pool: pool_mod.Pool = .init(&entries, .{}, .{ @@ -1662,3 +1680,22 @@ fn fieldIndex(comptime name: []const u8) usize { } @compileError("no such counter: " ++ name); } + +test "the budget-exhausted counter renders pool-wide, without a url label" { + const text = try renderToString(testing.allocator, .{ .upstream_budget_exhausted_total = 7 }); + defer testing.allocator.free(text); + + try testing.expect(std.mem.containsAtLeast( + u8, + text, + 1, + "# TYPE nxdns_upstream_budget_exhausted_total counter\n", + )); + try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_upstream_budget_exhausted_total 7\n")); + + // No pool, no series: an operator tells "no upstreams wired" from "zero + // exhausted budgets" the same way every other collaborator is told. + const absent = try renderToString(testing.allocator, .{}); + defer testing.allocator.free(absent); + try testing.expect(!std.mem.containsAtLeast(u8, absent, 1, "nxdns_upstream_budget_exhausted_total")); +} diff --git a/src/web/web_integration_test.zig b/src/web/web_integration_test.zig index 1818a31..4e1568c 100644 --- a/src/web/web_integration_test.zig +++ b/src/web/web_integration_test.zig @@ -495,7 +495,7 @@ const Env = struct { .priority = 1, .enabled = true, .health = .init, - .sem = .{ .permits = self.pool_slots.len }, + .admission = .{ .permits = self.pool_slots.len }, .reuse_recoveries = &self.pool_recoveries, }}; self.pool_owner = .{};