diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4ce0853..0ea1eb1 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -19,6 +19,8 @@ Sections are written by hand. Nothing here is generated from commit messages: th
 - **Blocklists take less memory, and a refresh peaks lower.** A list that names a domain both on its own and with its subdomains, as Adblock-style lists do for every entry, now stores that name once instead of twice. A reload reads one list's compiled files at a time instead of all of them at once, and the startup pass no longer rebuilds the blocklists when no list was due for a refresh.
 - **The query-history buffer defaults to 1024 queries** instead of 10000, under 2 MiB instead of about 17. It only holds queries that arrive while a write is in progress.
 - **A live query stream takes memory only while someone watches it.** Each of the 32 possible streams reserved its buffer at startup, about 3.6 MiB in all. A stream that cannot get its buffer is refused with its own message and counted in `nxdns_live_stream_allocation_failures_total`.
+- **All DNS listeners share one budget of 32 queries at a time**, about 2.4 MiB, where each UDP and TCP listener used to reserve room for 64 of its own, about 27 MiB across the four. The Pi's busiest moment over 37 hours was 27 queries at once. A query beyond the budget is dropped and counted rather than queued, so a device retries it: watch `nxdns_udp_server_dropped_no_slot_total` and the stream listeners' new `dropped_no_permit_total`. TCP, DoT and DoH each accept 16 connections at a time instead of 64, and the DoT and DoH listeners now report `rejected_at_capacity_total` too.
+- **The server runs a bounded number of threads**, sized from those caps, instead of starting a new one whenever all were busy. Work that could not get a thread used to fail as a generic error or not be counted at all; `/metrics` now counts it as `nxdns_concurrency_unavailable_total`, by where it happened.
 - **Each database connection keeps a page cache sized for its job** instead of SQLite's 2 MiB default on all seven: 256 KiB for the configuration database and for retention, 1 MiB for the query-history writer and reader.
 - **The documentation is written for people running nxdns.** It starts from Get started, which downloads a release instead of building from source, and drops the API reference, the troubleshooting page and the performance pages. It is published at <https://nxdns.sh/docs/>, and `INSTALL.md` in the tarball links there.
 
diff --git a/specs/memory-budget.md b/specs/memory-budget.md
index 264c5bf..7d40a05 100644
--- a/specs/memory-budget.md
+++ b/specs/memory-budget.md
@@ -62,6 +62,7 @@ Within one source, as compiled entries, an exact `x` in `.list` together with `x
 - UDP keeps today's counted drop when no permit is free (`dropped_no_slot`). The kernel buffer does not absorb overload invisibly.
 - The upstream exchange keeps its whole-operation `Select` deadline and its distinction between peer result, budget expiry and cancellation. The thread high-water falls because the query budget is smaller, not because the race changes.
 - UDP slots and TCP connection buffers are sized from these bounds, not 64 per listener.
+- As built: 32 query permits for the whole process (`query_budget.permits`; the Pi's `nxdns_dns_in_flight_peak` was 27 over 37 h), each holding the datagram, the reply buffer and the handler scratch, about 2.4 MiB in all. 16 connections per DNS stream listener (`listener.dns_max_connections`), a policy rather than a measurement: the Pi's TCP peak was 0 and it runs neither DoT nor DoH. A stream query that finds no permit closes its connection and is counted as `dropped_no_permit`, on DoH too, where a 503 would need a reply buffer the query does not have. `async_limit` is `.nothing`, so every pool task is a counted `concurrent` one, and `app.concurrent_limit` enumerates them. `ConcurrencyUnavailable` is `nxdns_concurrency_unavailable_total{site}`, one label per boundary (`platform/concurrency.zig`).
 
 ### 4. Size the query-log buffers by use
 
diff --git a/src/app.zig b/src/app.zig
index b9e62e5..898220f 100644
--- a/src/app.zig
+++ b/src/app.zig
@@ -35,6 +35,7 @@ const cert_store = @import("server/cert_store.zig");
 const cli = @import("cli.zig");
 const client_names = @import("server/client_names.zig");
 const clients = @import("server/clients.zig");
+const concurrency = @import("platform/concurrency.zig");
 const config_export = @import("config/export.zig");
 const db = @import("storage/db.zig");
 const disk_monitor = @import("storage/disk_monitor.zig");
@@ -47,6 +48,7 @@ const fetcher = @import("filter/fetcher.zig");
 const forward_zones = @import("local/forward_zones.zig");
 const handler = @import("server/handler.zig");
 const http_util = @import("web/http_util.zig");
+const listener = @import("server/listener.zig");
 const loader = @import("config/loader.zig");
 const local_records = @import("local/records.zig");
 const local_tables = @import("server/local_tables.zig");
@@ -57,6 +59,7 @@ const migrations = @import("storage/migrations.zig");
 const model = @import("config/model.zig");
 const pause = @import("server/pause.zig");
 const queries_repo = @import("storage/repositories/queries_repo.zig");
+const query_budget = @import("server/query_budget.zig");
 const query_sink = @import("server/query_sink.zig");
 const querylog_migrations = @import("storage/querylog_migrations.zig");
 const querylog_schema = @import("storage/querylog_schema.zig");
@@ -86,6 +89,40 @@ const maintenance_interval_s = 60;
 /// takes longer than this is not going to finish at all.
 const download_budget_s = 300;
 
+/// The most tasks the process's `std.Io.Threaded` runs at once, which `main`
+/// sets as its `concurrent_limit` (specs/memory-budget.md item 3). Each
+/// `Threaded` worker holds a task and is never retired, so this also bounds
+/// the worker threads. `main` runs `async` work inline, so every task is one of
+/// the `concurrent` ones counted here; a term that is wrong shows up as
+/// `nxdns_concurrency_unavailable_total`, not as a hang.
+///
+/// An upstream or forward-zone exchange, an upstream admission wait, a stream
+/// read or write and a TLS handshake each race one task against a timer task,
+/// and a task runs at most one such race at a time.
+pub const concurrent_limit =
+    // A UDP query: its `respondOne` task and the race of its exchange. A stream
+    // query runs on its connection's task and is counted below.
+    query_budget.permits * 3 +
+    // tcp6, tcp4, dot and doh: each connection's task and its race.
+    4 * @as(usize, listener.dns_max_connections) * 3 +
+    // A web connection's task. It runs no race of its own; the two below are
+    // shared with the services that start them.
+    web_server.default_max_connections +
+    // A blocklist download and its timer, on the scheduler's task or a web
+    // connection's, one at a time under the manager's refresh lock.
+    2 +
+    // A client-naming PTR lookup and its timer, on the tracker's task, and
+    // inside it the forward exchange's own race.
+    2 + 2 +
+    // The services `serve` starts: six DNS listeners, two certificate
+    // watchers, retention, the disk monitor, the blocklist scheduler, the
+    // client tracker, maintenance and the web listener.
+    14 +
+    // The query-log writers: the live one and, across a resize, the one before
+    // or after it, each waiting for its next entry against a timer; and the
+    // reaper.
+    2 * 3 + 1;
+
 pub fn run(runner: cli.Runner, args: cli.RunArgs) u8 {
     const code = serve(runner, args) catch |err| code: {
         runner.err.print("nxdns run failed: {s}\n", .{@errorName(err)}) catch {};
@@ -792,10 +829,15 @@ fn serve(r: cli.Runner, args: cli.RunArgs) !u8 {
     var failed_listeners: [2][]const u8 = undefined;
     var failed_listener_count: usize = 0;
 
+    // Declared before every listener, so its deinit runs after theirs: each
+    // listener returns its permits before it is gone.
+    var budget: query_budget.QueryBudget = try .init(gpa, query_budget.permits);
+    defer budget.deinit();
+
     var doh: ?doh_server.DohServer = null;
     defer if (doh) |*server| server.deinit(io);
     if (doh_certs) |*store| {
-        doh = bindDoh(gpa, io, cfg.doh_server, &h, store, event_store, boot_now_s);
+        doh = bindDoh(gpa, io, cfg.doh_server, &h, &budget, store, event_store, boot_now_s);
         if (doh == null) {
             failed_listeners[failed_listener_count] = "doh";
             failed_listener_count += 1;
@@ -805,7 +847,7 @@ fn serve(r: cli.Runner, args: cli.RunArgs) !u8 {
     var dot: ?dot_server.DotServer = null;
     defer if (dot) |*server| server.deinit(io);
     if (dot_certs) |*store| {
-        dot = bindDot(gpa, io, cfg.dot_server, &h, store, event_store, boot_now_s);
+        dot = bindDot(gpa, io, cfg.dot_server, &h, &budget, store, event_store, boot_now_s);
         if (dot == null) {
             failed_listeners[failed_listener_count] = "dot";
             failed_listener_count += 1;
@@ -883,30 +925,30 @@ fn serve(r: cli.Runner, args: cli.RunArgs) !u8 {
     const v4_bind = parseBind(r, cfg.dns.bind_ipv4, cfg.dns.port, "dns.bind_ipv4", .ip4) catch |err| return err;
 
     // IPv6 first, and the order is load-bearing — see `Listeners`.
-    var udp6: ?udp_server.UdpServer = udp_server.UdpServer.bind(gpa, io, v6_bind, &h, .{}) catch |err| bound: {
+    var udp6: ?udp_server.UdpServer = udp_server.UdpServer.bind(io, v6_bind, &h, &budget) catch |err| bound: {
         if (!ipv6Unavailable(err)) return reportBind(r, "udp", v6_bind, err);
         log.warn("this system has no IPv6; serving IPv4 only", .{});
         config_load.note("dns.bind_ipv6", "dns.bind_ipv6", "this system has no IPv6; serving IPv4 only");
         break :bound null;
     };
-    defer if (udp6) |*s| s.deinit(gpa, io);
+    defer if (udp6) |*s| s.deinit(io);
 
     const v6_is_wildcard = udp6 != null and isWildcard(v6_bind);
 
-    var udp4: ?udp_server.UdpServer = udp_server.UdpServer.bind(gpa, io, v4_bind, &h, .{}) catch |err| bound: {
+    var udp4: ?udp_server.UdpServer = udp_server.UdpServer.bind(io, v4_bind, &h, &budget) catch |err| bound: {
         if (err != error.AddressInUse or !v6_is_wildcard) return reportBind(r, "udp", v4_bind, err);
         log.info("the IPv6 UDP listener is dual-stack and already serves IPv4", .{});
         break :bound null;
     };
-    defer if (udp4) |*s| s.deinit(gpa, io);
+    defer if (udp4) |*s| s.deinit(io);
 
-    var tcp6: ?tcp_server.TcpServer = tcp_server.TcpServer.listen(gpa, io, v6_bind, &h, .{}) catch |err| bound: {
+    var tcp6: ?tcp_server.TcpServer = tcp_server.TcpServer.listen(gpa, io, v6_bind, &h, &budget, .{}) catch |err| bound: {
         if (!ipv6Unavailable(err)) return reportBind(r, "tcp", v6_bind, err);
         break :bound null;
     };
     defer if (tcp6) |*s| s.deinit(io);
 
-    var tcp4: ?tcp_server.TcpServer = tcp_server.TcpServer.listen(gpa, io, v4_bind, &h, .{}) catch |err| bound: {
+    var tcp4: ?tcp_server.TcpServer = tcp_server.TcpServer.listen(gpa, io, v4_bind, &h, &budget, .{}) catch |err| bound: {
         if (err != error.AddressInUse or !(tcp6 != null and isWildcard(v6_bind))) {
             return reportBind(r, "tcp", v4_bind, err);
         }
@@ -991,27 +1033,27 @@ fn serve(r: cli.Runner, args: cli.RunArgs) !u8 {
         log_controller.shutdown(io);
     }
 
-    if (udp6) |*s| try group.concurrent(io, udp_server.UdpServer.serve, .{ s, io });
-    if (udp4) |*s| try group.concurrent(io, udp_server.UdpServer.serve, .{ s, io });
-    if (tcp6) |*s| try group.concurrent(io, tcp_server.TcpServer.serve, .{ s, io });
-    if (tcp4) |*s| try group.concurrent(io, tcp_server.TcpServer.serve, .{ s, io });
-    if (doh) |*s| try group.concurrent(io, doh_server.DohServer.serve, .{ s, io });
-    if (dot) |*s| try group.concurrent(io, dot_server.DotServer.serve, .{ s, io });
-    if (doh_certs) |*store| try group.concurrent(io, cert_store.CertStore.watch, .{ store, io });
-    if (dot_certs) |*store| try group.concurrent(io, cert_store.CertStore.watch, .{ store, io });
+    if (udp6) |*s| try concurrency.counted(.service, group.concurrent(io, udp_server.UdpServer.serve, .{ s, io }));
+    if (udp4) |*s| try concurrency.counted(.service, group.concurrent(io, udp_server.UdpServer.serve, .{ s, io }));
+    if (tcp6) |*s| try concurrency.counted(.service, group.concurrent(io, tcp_server.TcpServer.serve, .{ s, io }));
+    if (tcp4) |*s| try concurrency.counted(.service, group.concurrent(io, tcp_server.TcpServer.serve, .{ s, io }));
+    if (doh) |*s| try concurrency.counted(.service, group.concurrent(io, doh_server.DohServer.serve, .{ s, io }));
+    if (dot) |*s| try concurrency.counted(.service, group.concurrent(io, dot_server.DotServer.serve, .{ s, io }));
+    if (doh_certs) |*store| try concurrency.counted(.service, group.concurrent(io, cert_store.CertStore.watch, .{ store, io }));
+    if (dot_certs) |*store| try concurrency.counted(.service, group.concurrent(io, cert_store.CertStore.watch, .{ store, io }));
 
-    try group.concurrent(io, retention_mod.Retention.run, .{ &retention, io, &querylog_retention_db, gate, event_store });
-    try group.concurrent(io, disk_monitor.Monitor.run, .{ &monitor, io, event_store });
-    try group.concurrent(io, manager_mod.Manager.runScheduler, .{ &manager, io });
-    try group.concurrent(io, clients.Tracker.run, .{ &tracker, io, &tracker_db, gate, &client_names_resolver });
-    try group.concurrent(io, runMaintenance, .{ &h, if (web_limiter) |*l| l else null, io });
+    try concurrency.counted(.service, group.concurrent(io, retention_mod.Retention.run, .{ &retention, io, &querylog_retention_db, gate, event_store }));
+    try concurrency.counted(.service, group.concurrent(io, disk_monitor.Monitor.run, .{ &monitor, io, event_store }));
+    try concurrency.counted(.service, group.concurrent(io, manager_mod.Manager.runScheduler, .{ &manager, io }));
+    try concurrency.counted(.service, group.concurrent(io, clients.Tracker.run, .{ &tracker, io, &tracker_db, gate, &client_names_resolver }));
+    try concurrency.counted(.service, group.concurrent(io, runMaintenance, .{ &h, if (web_limiter) |*l| l else null, io }));
 
     // Started last (ruling 26), canceled by the same `group.cancel`; its inner
     // connection group is canceled, not awaited (ruling 4), so an idle
     // keep-alive client cannot hold shutdown open. A web bind failure is not
     // fatal: `web_server.serve` warns and returns, and the DNS side — the thing
     // this box exists for — keeps serving.
-    if (cfg.web.enabled) try group.concurrent(io, web_server.serve, .{ &web_state, io });
+    if (cfg.web.enabled) try concurrency.counted(.service, group.concurrent(io, web_server.serve, .{ &web_state, io }));
 
     logStartup(io, authority, &manager, upstream_count, .{
         .udp6 = if (udp6) |*s| s.boundAddress() else null,
@@ -1111,6 +1153,7 @@ fn bindDoh(
     io: std.Io,
     endpoint: model.TlsEndpoint,
     h: *handler.Handler,
+    budget: *query_budget.QueryBudget,
     store: *cert_store.CertStore,
     diagnostics: ?*events.Store,
     now_s: i64,
@@ -1120,7 +1163,7 @@ fn bindDoh(
         reportListener(diagnostics, io, now_s, "doh", "bind '{s}' is not an IP address", .{endpoint.bind});
         return null;
     };
-    const server = doh_server.DohServer.listen(gpa, io, bind_address, h, store, .{}) catch |err| {
+    const server = doh_server.DohServer.listen(gpa, io, bind_address, h, budget, store, .{}) catch |err| {
         log.warn("doh listener cannot listen on {s}:{d}: {t}", .{ endpoint.bind, endpoint.port, err });
         reportListener(diagnostics, io, now_s, "doh", "cannot listen on {s}:{d}: {t}", .{ endpoint.bind, endpoint.port, err });
         return null;
@@ -1134,6 +1177,7 @@ fn bindDot(
     io: std.Io,
     endpoint: model.TlsEndpoint,
     h: *handler.Handler,
+    budget: *query_budget.QueryBudget,
     store: *cert_store.CertStore,
     diagnostics: ?*events.Store,
     now_s: i64,
@@ -1143,7 +1187,7 @@ fn bindDot(
         reportListener(diagnostics, io, now_s, "dot", "bind '{s}' is not an IP address", .{endpoint.bind});
         return null;
     };
-    const server = dot_server.DotServer.listen(gpa, io, bind_address, h, store, .{}) catch |err| {
+    const server = dot_server.DotServer.listen(gpa, io, bind_address, h, budget, store, .{}) catch |err| {
         log.warn("dot listener cannot listen on {s}:{d}: {t}", .{ endpoint.bind, endpoint.port, err });
         reportListener(diagnostics, io, now_s, "dot", "cannot listen on {s}:{d}: {t}", .{ endpoint.bind, endpoint.port, err });
         return null;
diff --git a/src/filter/manager.zig b/src/filter/manager.zig
index 30be39f..6f325c7 100644
--- a/src/filter/manager.zig
+++ b/src/filter/manager.zig
@@ -52,6 +52,7 @@ const std = @import("std");
 const Allocator = std.mem.Allocator;
 
 const model = @import("../config/model.zig");
+const concurrency = @import("../platform/concurrency.zig");
 const safe_url = @import("../safe_url.zig");
 const db = @import("../storage/db.zig");
 const clients_repo = @import("../storage/repositories/clients_repo.zig");
@@ -1412,10 +1413,16 @@ pub const Manager = struct {
         defer race.cancelDiscard();
 
         race.concurrent(.fetch, fetcher.Fetcher.fetch, .{ self.fetcher, io, url, w }) catch |err| switch (err) {
-            error.ConcurrencyUnavailable => return error.SystemResources,
+            error.ConcurrencyUnavailable => {
+                concurrency.unavailable(.blocklist_fetch);
+                return error.SystemResources;
+            },
         };
         race.concurrent(.expiry, expire, .{ io, self.total_budget }) catch |err| switch (err) {
-            error.ConcurrencyUnavailable => return error.SystemResources,
+            error.ConcurrencyUnavailable => {
+                concurrency.unavailable(.blocklist_fetch);
+                return error.SystemResources;
+            },
         };
 
         // Read before the deferred `cancelDiscard`, which tears the loser down:
diff --git a/src/local/forward_client.zig b/src/local/forward_client.zig
index 94cb0a2..687a407 100644
--- a/src/local/forward_client.zig
+++ b/src/local/forward_client.zig
@@ -20,6 +20,7 @@
 const std = @import("std");
 const net = std.Io.net;
 
+const concurrency = @import("../platform/concurrency.zig");
 const transport = @import("../upstream/transport.zig");
 const validate = @import("../config/validate.zig");
 const dns_header = @import("../dns/header.zig");
@@ -193,7 +194,10 @@ pub const ForwardClient = struct {
         while (true) {
             const msg = socket.receiveTimeout(io, response_buf, deadline) catch |err| switch (err) {
                 error.Timeout => return error.Timeout,
-                error.ConcurrencyUnavailable => return error.SystemResources,
+                error.ConcurrencyUnavailable => {
+                    concurrency.unavailable(.forward_receive);
+                    return error.SystemResources;
+                },
                 else => return transport.mapPhase(err, error.ReceiveFailed),
             };
 
diff --git a/src/main.zig b/src/main.zig
index a95f2f5..248cb08 100644
--- a/src/main.zig
+++ b/src/main.zig
@@ -3,6 +3,8 @@
 //! parameters and is therefore testable without a process.
 
 const std = @import("std");
+const builtin = @import("builtin");
+const app = @import("app.zig");
 const cli = @import("cli.zig");
 const logging = @import("platform/logging.zig");
 
@@ -40,28 +42,51 @@ fn runnerWriter(file: std.Io.File, io: std.Io, buffer: []u8) std.Io.File.Writer
     return file.writerStreaming(io, buffer);
 }
 
-pub fn main(init: std.process.Init) u8 {
+/// The allocator `start.zig` would choose: leak-checking in Debug, and libc's
+/// otherwise, because SQLite allocates through `malloc` either way.
+const use_debug_allocator = builtin.mode == .Debug;
+
+/// `main` builds the `Threaded` itself, which the full `std.process.Init` would
+/// build with neither limit set (specs/memory-budget.md item 3).
+pub fn main(init: std.process.Init.Minimal) u8 {
+    var debug_allocator: std.heap.DebugAllocator(.{}) = .init;
+    defer if (use_debug_allocator) {
+        _ = debug_allocator.deinit();
+    };
+    const gpa = if (use_debug_allocator) debug_allocator.allocator() else std.heap.c_allocator;
+
+    var threaded: std.Io.Threaded = .init(gpa, .{
+        .argv0 = .init(init.args),
+        .environ = init.environ,
+        // Inline, so no `async` task holds a worker beside the counted
+        // `concurrent` ones.
+        .async_limit = .nothing,
+        .concurrent_limit = .limited(app.concurrent_limit),
+    });
+    defer threaded.deinit();
+    const io = threaded.io();
+
     // Both `File.Writer` values are self-referential and must not move, so they
     // stay in these `var` slots for the whole of `main`.
     var out_buffer: [4096]u8 = undefined;
     var err_buffer: [4096]u8 = undefined;
-    var out = runnerWriter(std.Io.File.stdout(), init.io, &out_buffer);
-    var err = runnerWriter(std.Io.File.stderr(), init.io, &err_buffer);
+    var out = runnerWriter(std.Io.File.stdout(), io, &out_buffer);
+    var err = runnerWriter(std.Io.File.stderr(), io, &err_buffer);
 
     const runner: cli.Runner = .{
-        .io = init.io,
-        .gpa = init.gpa,
+        .io = io,
+        .gpa = gpa,
         .out = &out.interface,
         .err = &err.interface,
     };
 
     var argv: std.ArrayList([]const u8) = .empty;
-    defer argv.deinit(init.gpa);
+    defer argv.deinit(gpa);
 
-    var args = init.minimal.args.iterate();
+    var args = init.args.iterate();
     _ = args.skip();
     while (args.next()) |arg| {
-        argv.append(init.gpa, arg) catch return cli.exit_runtime;
+        argv.append(gpa, arg) catch return cli.exit_runtime;
     }
 
     const command = cli.parseArgs(argv.items) catch |e| return cli.runUsageError(runner, e);
diff --git a/src/platform/concurrency.zig b/src/platform/concurrency.zig
new file mode 100644
index 0000000..52d576b
--- /dev/null
+++ b/src/platform/concurrency.zig
@@ -0,0 +1,91 @@
+//! Where `error.ConcurrencyUnavailable` surfaced (specs/memory-budget.md item 3).
+//!
+//! `main` caps the process's one `std.Io.Threaded` at a task count the
+//! accounting in `app.zig` says is never reached, so this error means the
+//! accounting is wrong or the system refused a thread. Either way the work it
+//! cost is lost, and each boundary that can lose work to it counts here under
+//! its own site rather than folding it into a generic failure.
+//!
+//! Process-wide, like the log sink's counters: the cap is a property of the
+//! one `Threaded` every subsystem shares, not of any one of them, and a
+//! counter beside each call site keeps transport helpers free of a stats
+//! pointer they would otherwise thread through for this alone. Never reset,
+//! so a test reads a delta, not a value.
+
+const std = @import("std");
+
+pub const Site = enum {
+    /// A UDP datagram that held a query permit but got no task to answer it.
+    /// The datagram is dropped.
+    udp_query,
+    /// A UDP receive whose timed wait could not start.
+    udp_receive,
+    /// An accepted stream connection that got no task. The stream is closed.
+    connection,
+    /// A stream listener's read, write or TLS handshake that could not start
+    /// its race against the idle budget. The connection ends.
+    listener_race,
+    /// An upstream or forward-zone exchange, or an upstream admission wait,
+    /// that could not start its race against the query's deadline. The
+    /// exchange fails as a local resource fault.
+    exchange_race,
+    /// A forward-zone receive whose timed wait could not start.
+    forward_receive,
+    /// A blocklist download that could not start its race against its budget.
+    blocklist_fetch,
+    /// The query-log writer's wait for the next entry. The writer flushes what
+    /// it holds instead of waiting out the interval.
+    querylog_wait,
+    /// A long-running task the server starts at boot or on a query-log resize.
+    /// The start or the resize fails.
+    service,
+};
+
+var counts: std.enums.EnumArray(Site, std.atomic.Value(u64)) = .initFill(.init(0));
+
+pub fn unavailable(site: Site) void {
+    _ = counts.getPtr(site).fetchAdd(1, .monotonic);
+}
+
+/// `result` unchanged, with a `ConcurrencyUnavailable` in it counted under
+/// `site`: for a spawn whose failure its caller propagates.
+pub fn counted(site: Site, result: anytype) @TypeOf(result) {
+    if (result) |_| {} else |err| switch (err) {
+        error.ConcurrencyUnavailable => unavailable(site),
+    }
+    return result;
+}
+
+pub const Snapshot = std.enums.EnumArray(Site, u64);
+
+pub fn snapshot() Snapshot {
+    var out: Snapshot = .initFill(0);
+    for (std.enums.values(Site)) |site| out.set(site, counts.getPtrConst(site).load(.monotonic));
+    return out;
+}
+
+test "a propagated spawn failure is counted and passed through" {
+    const before = snapshot().get(.service);
+    const failed: error{ConcurrencyUnavailable}!u8 = error.ConcurrencyUnavailable;
+    try std.testing.expectError(error.ConcurrencyUnavailable, counted(.service, failed));
+    const started: error{ConcurrencyUnavailable}!u8 = 7;
+    try std.testing.expectEqual(@as(u8, 7), try counted(.service, started));
+    try std.testing.expectEqual(@as(u64, 1), snapshot().get(.service) - before);
+}
+
+test "each site counts on its own" {
+    const before = snapshot();
+    unavailable(.querylog_wait);
+    unavailable(.querylog_wait);
+    unavailable(.udp_query);
+    const after = snapshot();
+
+    for (std.enums.values(Site)) |site| {
+        const expected: u64 = switch (site) {
+            .querylog_wait => 2,
+            .udp_query => 1,
+            else => 0,
+        };
+        try std.testing.expectEqual(expected, after.get(site) - before.get(site));
+    }
+}
diff --git a/src/server/doh_server.zig b/src/server/doh_server.zig
index 07f5182..e8d5569 100644
--- a/src/server/doh_server.zig
+++ b/src/server/doh_server.zig
@@ -11,7 +11,7 @@
 //!
 //! The handshake runs under the same race budget tcp_server applies to its
 //! reads (ruling 3's rationale): a client that connects and never handshakes
-//! must not pin one of the 64 slots forever.
+//! must not pin a slot forever.
 //!
 //! Over capacity the raw TCP stream is closed and counted, with no response.
 //! The web listener's 503 is not possible here: HTTP exists only on the far
@@ -19,6 +19,11 @@
 //! certificate acquisition — exactly what an over-capacity listener does not
 //! have to spend. Closing is the cheapest honest refusal.
 //!
+//! A query holds a permit from the shared query budget from its decoded body
+//! to its sent response. With none free the connection is closed unanswered
+//! and counted, the TCP/53 rule: a 503 would be the one DNS refusal that
+//! differs by transport.
+//!
 //! Only `/dns-query` exists. This is not the admin API, so errors are plain
 //! text, not JSON, and there are no cache-control headers (LAN, no
 //! intermediaries).
@@ -34,6 +39,7 @@ const doh_client = @import("../upstream/doh_client.zig");
 const handler = @import("handler.zig");
 const listener = @import("listener.zig");
 const model = @import("../config/model.zig");
+const query_budget = @import("query_budget.zig");
 const tls_server = @import("../platform/tls_server.zig");
 const transport = @import("../upstream/transport.zig");
 const upstream_owner = @import("../upstream/owner.zig");
@@ -52,12 +58,10 @@ pub const alpn_protocols: [*:null]const ?[*:0]const u8 = &.{"http/1.1"};
 const recv_buffer_len = 8 * 1024;
 const send_buffer_len = 4 * 1024;
 
-pub const default_max_connections: u16 = 64;
-
 const allow_header: http.Header = .{ .name = "allow", .value = "GET, POST" };
 
 pub const Options = struct {
-    max_connections: u16 = default_max_connections,
+    max_connections: u16 = listener.dns_max_connections,
     /// The budget for the TLS handshake and for the wait on the next request
     /// head of a keep-alive connection (milestone-16 ruling 10). A request
     /// already being served has no timeout, like the web listener: the port is
@@ -87,27 +91,23 @@ pub const Snapshot = struct {
     idle_timeouts: u64,
     connection_errors: u64,
     bad_requests: u64,
+    dropped_no_permit: u64,
 };
 
 pub const DohServer = struct {
     core: listener.Core(Config),
     handler: *handler.Handler,
+    budget: *query_budget.QueryBudget,
     certs: *cert_store.CertStore,
     options: Options,
     stats: Stats,
 
-    /// One slot is ~150 KiB, so the default 64 connections cost ~9.4 MiB. The
-    /// two message buffers cannot shrink: a POST body and the reply both go up
-    /// to the 65535 bytes a DNS message can be. The `ServerStream` plaintext
-    /// buffers belong to the core; its `read_buf` doubles as the HTTP head cap
-    /// (see `recv_buffer_len`).
+    /// The `ServerStream` plaintext buffers belong to the core; its `read_buf`
+    /// doubles as the HTTP head cap (see `recv_buffer_len`).
     pub const Payload = struct {
-        /// The decoded query: a POST body or a GET `dns` parameter.
+        /// The decoded query: a POST body or a GET `dns` parameter, up to the
+        /// 65535 bytes a DNS message can be. Read before a permit is taken.
         query: [transport.max_message_len]u8,
-        reply: [transport.max_message_len]u8,
-        /// The handler's per-query working memory. A connection is answered
-        /// serially, so one query uses it at a time.
-        scratch: handler.Scratch,
         /// Valid between a successful `ServerStream.accept` and the
         /// `close(gpa)` in `serveOne`'s defer.
         tls: tls_server.ServerStream,
@@ -131,12 +131,14 @@ pub const DohServer = struct {
         io: std.Io,
         listen_address: net.IpAddress,
         h: *handler.Handler,
+        budget: *query_budget.QueryBudget,
         certs: *cert_store.CertStore,
         options: Options,
     ) ListenError!DohServer {
         return .{
             .core = try listener.Core(Config).listen(gpa, io, listen_address, options.max_connections),
             .handler = h,
+            .budget = budget,
             .certs = certs,
             .options = options,
             .stats = .{},
@@ -163,6 +165,7 @@ pub const DohServer = struct {
             .idle_timeouts = core.idle_timeouts.load(.monotonic),
             .connection_errors = core.connection_errors.load(.monotonic),
             .bad_requests = self.stats.bad_requests.load(.monotonic),
+            .dropped_no_permit = core.dropped_no_permit.load(.monotonic),
         };
     }
 
@@ -192,7 +195,7 @@ pub const DohServer = struct {
         const stage: Handshake = .{ .conn = conn, .gpa = gpa, .ctx = &entry.ctx, .io = io };
         switch (listener.handshakeStage(io, self.options.idle_timeout, stage)) {
             .ok => {},
-            .canceled => return,
+            .canceled, .unavailable => return,
             // Milestone-16 ruling 9: a stalled handshake is refused like a broken
             // one, the DoT arrangement. `idle_timeouts` belongs to the keep-alive
             // wait below, so the two listeners export the same names for the
@@ -220,9 +223,10 @@ pub const DohServer = struct {
                     listener.bump(&stats.idle_timeouts);
                     return;
                 },
-                // Cancellation is shutdown; `.failed` here is only the wrapper
-                // failing to start, which costs nothing and counts as nothing.
-                .canceled, .failed => return,
+                // Cancellation is shutdown, and `.unavailable` is counted by
+                // the race. `.failed` cannot happen: `receiveHeadInto` stores
+                // the head's error instead of returning it.
+                .canceled, .unavailable, .failed => return,
             }
 
             var request = head catch |err| switch (err) {
@@ -385,13 +389,19 @@ pub const DohServer = struct {
         request: *http.Server.Request,
         query: []const u8,
     ) error{ WriteFailed, HttpExpectationFailed }!Next {
+        const work = self.budget.tryAcquire(io) orelse {
+            listener.bump(&self.core.stats.dropped_no_permit);
+            return .close;
+        };
+        defer self.budget.release(io, work);
+
         const outcome = self.handler.handle(
             io,
             .tcp,
             address.NetAddress.fromIp(conn.peer),
             query,
-            &conn.payload.reply,
-            &conn.payload.scratch,
+            &work.reply,
+            &work.scratch,
         );
         switch (outcome) {
             .drop => return self.refuse(request, .bad_request, "bad request\n", &.{}, false),
@@ -626,6 +636,7 @@ const Harness = struct {
     upstream: FailingUpstream,
     upstream_owner: upstream_owner.Borrowed,
     h: handler.Handler,
+    permits: query_budget.QueryBudget,
     server: DohServer,
     group: std.Io.Group,
 
@@ -659,8 +670,11 @@ const Harness = struct {
             .local_tables = &hx.tables,
         };
 
+        hx.permits = try .init(testing.allocator, 4);
+        errdefer hx.permits.deinit();
+
         const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
-        hx.server = try DohServer.listen(testing.allocator, hio, listen_address, &hx.h, &hx.store, options);
+        hx.server = try DohServer.listen(testing.allocator, hio, listen_address, &hx.h, &hx.permits, &hx.store, options);
         errdefer hx.server.deinit(hio);
 
         hx.group = .init;
@@ -673,6 +687,7 @@ const Harness = struct {
         hx.group.await(hio) catch |err| switch (err) {
             error.Canceled => unreachable,
         };
+        hx.permits.deinit();
         hx.store.deinit(hio);
         hx.tables.deinit(testing.allocator);
         hx.tmp.cleanup();
diff --git a/src/server/dot_server.zig b/src/server/dot_server.zig
index 07224e8..f801407 100644
--- a/src/server/dot_server.zig
+++ b/src/server/dot_server.zig
@@ -18,6 +18,8 @@
 //!   TCP/53 each connection costs one allocation. The pool itself is still
 //!   fixed and pre-allocated.
 //!
+//! A query holds a permit from the shared query budget exactly as on TCP/53.
+//!
 //! ALPN belongs to the `CertStore`'s `ServerContext` (set at store init);
 //! this listener only acquires entries.
 
@@ -26,17 +28,18 @@ const address = @import("../platform/address.zig");
 const cert_store = @import("cert_store.zig");
 const handler = @import("handler.zig");
 const listener = @import("listener.zig");
+const query_budget = @import("query_budget.zig");
 const tls_server = @import("../platform/tls_server.zig");
 const transport = @import("../upstream/transport.zig");
 const upstream_owner = @import("../upstream/owner.zig");
 
 /// Plaintext staging for `ServerStream`: the framing bytes and the decrypted
 /// record tail pass through here, while whole messages go straight to
-/// `Payload.query`/`Payload.reply`.
+/// `Payload.query` and from the permit's reply.
 const stream_buffer_len = 1024;
 
 pub const Options = struct {
-    max_connections: u16 = 64,
+    max_connections: u16 = listener.dns_max_connections,
     /// RFC 7766 §6.2.3 recommends a few seconds of idle tolerance, and the
     /// handshake runs under the same budget.
     idle_timeout: std.Io.Clock.Duration = .{ .raw = .fromSeconds(10), .clock = .awake },
@@ -53,28 +56,24 @@ pub const Stats = struct {
 /// walks for the `nxdns_dot_server_*` families.
 pub const StatsSnapshot = struct {
     connections: u64,
+    rejected_at_capacity: u64,
     tls_handshake_failures: u64,
     idle_timeouts: u64,
     connection_errors: u64,
+    dropped_no_permit: u64,
 };
 
 pub const DotServer = struct {
     core: listener.Core(Config),
     handler: *handler.Handler,
+    budget: *query_budget.QueryBudget,
     certs: *cert_store.CertStore,
     options: Options,
     stats: Stats,
 
-    /// One slot is ~137 KiB — the same two message ceilings as TCP/53 plus the
-    /// `ServerStream` bookkeeping — so the default 64 connections stay inside
-    /// the PLAN §18 budget.
     pub const Payload = struct {
+        /// Read before a permit is taken, as on TCP/53.
         query: [transport.max_message_len]u8,
-        reply: [transport.max_message_len]u8,
-        /// The handler's per-query working memory. It belongs to the slot so
-        /// that answering a message allocates nothing, and a connection is
-        /// answered serially, so one query uses it at a time.
-        scratch: handler.Scratch,
         /// Pinned once its `accept` succeeds: mbedTLS holds a pointer to it,
         /// and the slot never moves.
         tls: tls_server.ServerStream,
@@ -98,12 +97,14 @@ pub const DotServer = struct {
         io: std.Io,
         listen_address: std.Io.net.IpAddress,
         h: *handler.Handler,
+        budget: *query_budget.QueryBudget,
         certs: *cert_store.CertStore,
         options: Options,
     ) ListenError!DotServer {
         return .{
             .core = try listener.Core(Config).listen(gpa, io, listen_address, options.max_connections),
             .handler = h,
+            .budget = budget,
             .certs = certs,
             .options = options,
             .stats = .{},
@@ -133,9 +134,11 @@ pub const DotServer = struct {
         const core = &self.core.stats;
         return .{
             .connections = core.connections.load(.monotonic),
+            .rejected_at_capacity = core.rejected_at_capacity.load(.monotonic),
             .tls_handshake_failures = self.stats.tls_handshake_failures.load(.monotonic),
             .idle_timeouts = core.idle_timeouts.load(.monotonic),
             .connection_errors = core.connection_errors.load(.monotonic),
+            .dropped_no_permit = core.dropped_no_permit.load(.monotonic),
         };
     }
 
@@ -157,7 +160,7 @@ pub const DotServer = struct {
         const stage: Handshake = .{ .conn = conn, .gpa = gpa, .ctx = &entry.ctx, .io = io };
         switch (listener.handshakeStage(io, budget, stage)) {
             .ok => {},
-            .canceled => return,
+            .canceled, .unavailable => return,
             // A stalled handshake is refused like a broken one: it must not
             // pin a connection slot for longer than the idle budget.
             .timed_out, .failed => {
@@ -181,7 +184,7 @@ pub const DotServer = struct {
                     listener.bump(&stats.idle_timeouts);
                     return;
                 },
-                .canceled => return,
+                .canceled, .unavailable => return,
                 // A transport EOF without close_notify lands here too: the
                 // stream reads it as a truncation, never as a clean end.
                 .failed => {
@@ -208,7 +211,7 @@ pub const DotServer = struct {
 
             switch (listener.race(io, budget, listener.readBody, .{ reader, conn.payload.query[0..len] })) {
                 .ok => {},
-                .canceled => return,
+                .canceled, .unavailable => return,
                 // A half-sent message is a broken peer, not an idle one.
                 .timed_out, .failed => {
                     listener.bump(&stats.connection_errors);
@@ -216,13 +219,20 @@ pub const DotServer = struct {
                 },
             }
 
+            // There is no framing for "no answer", so the connection ends.
+            const work = self.budget.tryAcquire(io) orelse {
+                listener.bump(&stats.dropped_no_permit);
+                return;
+            };
+            defer self.budget.release(io, work);
+
             const outcome = self.handler.handle(
                 io,
                 .tcp,
                 address.NetAddress.fromIp(conn.peer),
                 conn.payload.query[0..len],
-                &conn.payload.reply,
-                &conn.payload.scratch,
+                &work.reply,
+                &work.scratch,
             );
             const bytes = switch (outcome) {
                 // There is no framing for "no answer", so the connection ends.
@@ -233,7 +243,7 @@ pub const DotServer = struct {
             const out = transport.framePrefix(@intCast(bytes.len));
             switch (listener.race(io, budget, listener.writeReply, .{ writer, &out, bytes })) {
                 .ok => {},
-                .canceled => return,
+                .canceled, .unavailable => return,
                 .timed_out, .failed => {
                     listener.bump(&stats.connection_errors);
                     return;
@@ -577,7 +587,9 @@ test "dot: two framed queries share one TLS connection" {
     var h = bareHandler(h_owner.client(fake.client()));
 
     const listen_address: std.Io.net.IpAddress = try .parse("127.0.0.1", 0);
-    var server = try DotServer.listen(gpa, io, listen_address, &h, &env.store, .{ .max_connections = 2 });
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try DotServer.listen(gpa, io, listen_address, &h, &permits, &env.store, .{ .max_connections = 2 });
     const server_address = server.boundAddress();
 
     var group: std.Io.Group = .init;
@@ -621,7 +633,9 @@ test "dot: a transport EOF without close_notify is a connection error, not a cra
     var h = bareHandler(h_owner.client(fake.client()));
 
     const listen_address: std.Io.net.IpAddress = try .parse("127.0.0.1", 0);
-    var server = try DotServer.listen(gpa, io, listen_address, &h, &env.store, .{ .max_connections = 2 });
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try DotServer.listen(gpa, io, listen_address, &h, &permits, &env.store, .{ .max_connections = 2 });
     const server_address = server.boundAddress();
 
     var group: std.Io.Group = .init;
@@ -661,7 +675,9 @@ test "dot: plain TCP bytes fail the handshake and are counted" {
     var h = bareHandler(h_owner.client(fake.client()));
 
     const listen_address: std.Io.net.IpAddress = try .parse("127.0.0.1", 0);
-    var server = try DotServer.listen(gpa, io, listen_address, &h, &env.store, .{ .max_connections = 2 });
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try DotServer.listen(gpa, io, listen_address, &h, &permits, &env.store, .{ .max_connections = 2 });
     const server_address = server.boundAddress();
 
     var group: std.Io.Group = .init;
@@ -749,7 +765,9 @@ test "dot: a reload serves new handshakes without breaking the old connection" {
     var h = bareHandler(h_owner.client(fake.client()));
 
     const listen_address: std.Io.net.IpAddress = try .parse("127.0.0.1", 0);
-    var server = try DotServer.listen(gpa, io, listen_address, &h, &env.store, .{ .max_connections = 2 });
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try DotServer.listen(gpa, io, listen_address, &h, &permits, &env.store, .{ .max_connections = 2 });
     const server_address = server.boundAddress();
 
     var group: std.Io.Group = .init;
@@ -841,7 +859,9 @@ test "dot: a cert path change serves the new certificate on the next handshake"
     var h = bareHandler(h_owner.client(fake.client()));
 
     const listen_address: std.Io.net.IpAddress = try .parse("127.0.0.1", 0);
-    var server = try DotServer.listen(gpa, io, listen_address, &h, &env.store, .{ .max_connections = 2 });
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try DotServer.listen(gpa, io, listen_address, &h, &permits, &env.store, .{ .max_connections = 2 });
     const server_address = server.boundAddress();
 
     var group: std.Io.Group = .init;
@@ -883,7 +903,9 @@ test "dot: an idle connection is closed with close_notify and counted" {
     var h = bareHandler(h_owner.client(fake.client()));
 
     const listen_address: std.Io.net.IpAddress = try .parse("127.0.0.1", 0);
-    var server = try DotServer.listen(gpa, io, listen_address, &h, &env.store, .{
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try DotServer.listen(gpa, io, listen_address, &h, &permits, &env.store, .{
         .max_connections = 2,
         .idle_timeout = short_idle,
     });
diff --git a/src/server/handler.zig b/src/server/handler.zig
index 22c3064..96d3c57 100644
--- a/src/server/handler.zig
+++ b/src/server/handler.zig
@@ -468,7 +468,7 @@ pub const Handler = struct {
     }
 };
 
-/// Per-query working memory, owned by the listener's slot so that no query path
+/// Per-query working memory, owned by the query's permit (`query_budget.Work`) so that no query path
 /// allocates. Every buffer is used by one query at a time.
 pub const Scratch = struct {
     normalize: [types.max_name_len]u8,
diff --git a/src/server/listener.zig b/src/server/listener.zig
index 39ca972..a5314c7 100644
--- a/src/server/listener.zig
+++ b/src/server/listener.zig
@@ -28,8 +28,16 @@ const std = @import("std");
 const net = std.Io.net;
 const Allocator = std.mem.Allocator;
 
+const concurrency = @import("../platform/concurrency.zig");
 const transport = @import("../upstream/transport.zig");
 
+/// The connection cap of each DNS stream listener: TCP per address family, DoT
+/// and DoH. A connection holds memory and a task, never a query permit, while
+/// it waits for a query. The Pi saw no concurrent TCP connection in 37 hours
+/// and runs neither DoT nor DoH, so this is household policy, not a
+/// measurement: a few stub resolvers retrying truncated answers, with room.
+pub const dns_max_connections: u16 = 16;
+
 /// Lifecycle of the accept loop. `serve` claims `.serving`, `deinit` publishes
 /// `.closing`, and the two meet at `stopped` so no task touches a connection
 /// slot after it is freed.
@@ -83,6 +91,10 @@ pub const CoreStats = struct {
     accept_errors: std.atomic.Value(u64) = .init(0),
     connection_errors: std.atomic.Value(u64) = .init(0),
     idle_timeouts: std.atomic.Value(u64) = .init(0),
+    /// A DNS listener read a whole query and found every permit of the shared
+    /// query budget taken, so the connection ended unanswered. The web
+    /// listener answers no DNS and keeps this at zero.
+    dropped_no_permit: std.atomic.Value(u64) = .init(0),
     /// The most slots that have been taken at once (specs/memory-budget.md
     /// item 0). A gauge, so no listener's `Snapshot` carries it: those export
     /// every field as a counter. Written only under `Core.mutex`.
@@ -98,7 +110,10 @@ pub const Outcome = union(enum) {
     expiry: std.Io.Cancelable!void,
 };
 
-pub const Result = enum { ok, timed_out, failed, canceled };
+/// `unavailable` is the race that could not start for want of a task. It is
+/// counted here, under `concurrency.Site.listener_race`, so a caller ends the
+/// connection without counting it as the peer's failure.
+pub const Result = enum { ok, timed_out, failed, canceled, unavailable };
 
 /// Runs one connection operation against a budget and cancels the loser. No
 /// stream read or write in 0.16.0 accepts a timeout, so every wait on a peer
@@ -114,10 +129,16 @@ pub fn race(
     defer select.cancelDiscard();
 
     select.concurrent(.op, f, args) catch |err| switch (err) {
-        error.ConcurrencyUnavailable => return .failed,
+        error.ConcurrencyUnavailable => {
+            concurrency.unavailable(.listener_race);
+            return .unavailable;
+        },
     };
     select.concurrent(.expiry, expire, .{ io, budget }) catch |err| switch (err) {
-        error.ConcurrencyUnavailable => return .failed,
+        error.ConcurrencyUnavailable => {
+            concurrency.unavailable(.listener_race);
+            return .unavailable;
+        },
     };
 
     return switch (select.await() catch return .canceled) {
@@ -421,7 +442,7 @@ pub fn Core(comptime Cfg: type) type {
 
                 group.concurrent(io, runConn, .{ self, io, index }) catch |err| switch (err) {
                     error.ConcurrencyUnavailable => {
-                        bump(&self.stats.rejected_at_capacity);
+                        concurrency.unavailable(.connection);
                         self.finish(io, index);
                         continue;
                     },
@@ -524,6 +545,20 @@ fn testConns(count: usize) ![]TestSlot {
     return conns;
 }
 
+fn neverCalled() anyerror!void {
+    unreachable;
+}
+
+test "a race that cannot start is unavailable and counted, not a failure" {
+    var threaded: std.Io.Threaded = .init(std.testing.allocator, .{ .concurrent_limit = .nothing });
+    defer threaded.deinit();
+
+    const before = concurrency.snapshot().get(.listener_race);
+    const budget: std.Io.Clock.Duration = .{ .raw = .fromSeconds(60), .clock = .awake };
+    try std.testing.expectEqual(Result.unavailable, race(threaded.io(), budget, neverCalled, .{}));
+    try std.testing.expectEqual(@as(u64, 1), concurrency.snapshot().get(.listener_race) - before);
+}
+
 test "the connection pool hands out every slot once" {
     const conns = try testConns(3);
     defer testing.allocator.free(conns);
diff --git a/src/server/phase7_integration_test.zig b/src/server/phase7_integration_test.zig
index 99a686a..62e4e08 100644
--- a/src/server/phase7_integration_test.zig
+++ b/src/server/phase7_integration_test.zig
@@ -53,6 +53,7 @@ const shutdown = @import("shutdown.zig");
 const transport = @import("../upstream/transport.zig");
 const upstream_owner = @import("../upstream/owner.zig");
 const types = @import("../dns/types.zig");
+const query_budget = @import("query_budget.zig");
 const udp_server = @import("udp_server.zig");
 
 const testing = std.testing;
@@ -96,20 +97,29 @@ fn baseHandler(up: *upstream_owner.Owner) handler.Handler {
 /// holds a pointer into it.
 const Loop = struct {
     server: udp_server.UdpServer,
+    /// On the heap because the server keeps a pointer to it and `bind` returns
+    /// the loop by value.
+    permits: *query_budget.QueryBudget,
     group: std.Io.Group,
     client: net.Socket,
     server_address: net.IpAddress,
 
     fn bind(gpa: Allocator, io: std.Io, h: *handler.Handler) !Loop {
+        const permits = try gpa.create(query_budget.QueryBudget);
+        errdefer gpa.destroy(permits);
+        permits.* = try .init(gpa, 4);
+        errdefer permits.deinit();
+
         const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
-        var server = try udp_server.UdpServer.bind(gpa, io, listen_address, h, .{ .max_in_flight = 4 });
-        errdefer server.deinit(gpa, io);
+        var server = try udp_server.UdpServer.bind(io, listen_address, h, permits);
+        errdefer server.deinit(io);
 
         const client_address: net.IpAddress = try .parse("127.0.0.1", 0);
         const client = try client_address.bind(io, .{ .mode = .dgram });
 
         return .{
             .server = server,
+            .permits = permits,
             .group = .init,
             .client = client,
             .server_address = server.boundAddress(),
@@ -128,9 +138,11 @@ const Loop = struct {
     }
 
     fn stop(self: *Loop, gpa: Allocator, io: std.Io) void {
-        self.server.deinit(gpa, io);
+        self.server.deinit(io);
         self.group.cancel(io);
         self.client.close(io);
+        self.permits.deinit();
+        gpa.destroy(self.permits);
     }
 };
 
diff --git a/src/server/query_budget.zig b/src/server/query_budget.zig
new file mode 100644
index 0000000..d5e7dfb
--- /dev/null
+++ b/src/server/query_budget.zig
@@ -0,0 +1,151 @@
+//! The DNS work budget every listener shares (specs/memory-budget.md item 3).
+//!
+//! A permit is the memory one query needs while it is answered: the reply
+//! buffer, the handler's scratch and, for UDP, the datagram. Both address
+//! families of UDP and TCP, DoT and DoH all draw from one pool, allocated once
+//! at boot. A query takes a permit only once it has arrived whole and returns
+//! it when its reply is written, so a keep-alive connection between requests
+//! holds none.
+//!
+//! Nothing waits for a permit. A listener that finds the pool empty drops the
+//! query and counts it, because a wait would only turn a visible drop into
+//! invisible latency, and every permit is also the tasks its query may start,
+//! which `app.concurrent_limit` adds up from `permits`.
+
+const std = @import("std");
+const Allocator = std.mem.Allocator;
+
+const handler = @import("handler.zig");
+const transport = @import("../upstream/transport.zig");
+
+/// The most DNS queries answered at once, across every listener. The Pi's
+/// highest reading of `nxdns_dns_in_flight_peak` was 27 over 37 hours of
+/// household traffic, and a query beyond this is dropped and counted.
+pub const permits = 32;
+
+/// The EDNS ceiling the UDP listener advertises. A larger datagram arrives
+/// truncated and is dropped rather than guessed at.
+pub const max_datagram = 4096;
+
+comptime {
+    // `handler.handle` asserts this range on the buffers it is given.
+    std.debug.assert(max_datagram >= handler.udp_limit_min);
+    std.debug.assert(max_datagram >= handler.udp_limit_max);
+    std.debug.assert(transport.max_message_len >= handler.udp_limit_min);
+}
+
+pub const Work = struct {
+    /// UDP's copy of the datagram, so its listener can receive the next one
+    /// while this query is answered. A stream query stays in its connection,
+    /// which reads it before taking a permit: a slow sender holds none.
+    datagram: [max_datagram]u8,
+    /// A whole DNS message, not a whole datagram. A UDP reply between
+    /// `max_datagram` and `transport.max_message_len` must reach the handler
+    /// intact, because only the handler can turn it into the TC=1 reply that
+    /// sends the client to TCP; the datagram it sends stays bounded by
+    /// `udpLimit`.
+    reply: [transport.max_message_len]u8,
+    scratch: handler.Scratch,
+};
+
+pub const QueryBudget = struct {
+    gpa: Allocator,
+    work: []Work,
+    mutex: std.Io.Mutex,
+    /// Guarded by `mutex`: the indices of the free permits are
+    /// `free[0..free_len]`.
+    free: []u16,
+    free_len: usize,
+
+    pub fn init(gpa: Allocator, count: u16) Allocator.Error!QueryBudget {
+        std.debug.assert(count > 0);
+        const work = try gpa.alloc(Work, count);
+        errdefer gpa.free(work);
+        const free = try gpa.alloc(u16, count);
+        for (free, 0..) |*slot, index| slot.* = @intCast(index);
+        return .{ .gpa = gpa, .work = work, .mutex = .init, .free = free, .free_len = count };
+    }
+
+    /// Every listener has stopped, so every permit is back.
+    pub fn deinit(self: *QueryBudget) void {
+        std.debug.assert(self.free_len == self.work.len);
+        self.gpa.free(self.free);
+        self.gpa.free(self.work);
+        self.* = undefined;
+    }
+
+    /// A permit, or null when every one is taken. Waits for the bookkeeping
+    /// mutex, which no holder keeps across any Io, but never for a permit:
+    /// treating a contended lock as an empty pool would drop queries the
+    /// budget has room for.
+    pub fn tryAcquire(self: *QueryBudget, io: std.Io) ?*Work {
+        self.mutex.lockUncancelable(io);
+        defer self.mutex.unlock(io);
+        if (self.free_len == 0) return null;
+        self.free_len -= 1;
+        return &self.work[self.free[self.free_len]];
+    }
+
+    pub fn release(self: *QueryBudget, io: std.Io, work: *Work) void {
+        const index = (@intFromPtr(work) - @intFromPtr(self.work.ptr)) / @sizeOf(Work);
+        std.debug.assert(&self.work[index] == work);
+
+        self.mutex.lockUncancelable(io);
+        defer self.mutex.unlock(io);
+        std.debug.assert(self.free_len < self.work.len);
+        self.free[self.free_len] = @intCast(index);
+        self.free_len += 1;
+    }
+
+    /// Permits held right now.
+    pub fn taken(self: *QueryBudget, io: std.Io) usize {
+        self.mutex.lockUncancelable(io);
+        defer self.mutex.unlock(io);
+        return self.work.len - self.free_len;
+    }
+};
+
+const testing = std.testing;
+
+test "a budget hands out every permit once, then refuses" {
+    var budget: QueryBudget = try .init(testing.allocator, 3);
+    defer budget.deinit();
+    const io = testing.io;
+
+    var held: [3]*Work = undefined;
+    for (&held) |*slot| slot.* = budget.tryAcquire(io) orelse return error.TestUnexpectedResult;
+    try testing.expect(held[0] != held[1] and held[1] != held[2] and held[0] != held[2]);
+    try testing.expectEqual(@as(?*Work, null), budget.tryAcquire(io));
+    try testing.expectEqual(@as(usize, 3), budget.taken(io));
+
+    for (held) |work| budget.release(io, work);
+    try testing.expectEqual(@as(usize, 0), budget.taken(io));
+}
+
+test "a released permit is handed out again" {
+    var budget: QueryBudget = try .init(testing.allocator, 2);
+    defer budget.deinit();
+    const io = testing.io;
+
+    const first = budget.tryAcquire(io).?;
+    const second = budget.tryAcquire(io).?;
+    budget.release(io, first);
+    try testing.expectEqual(first, budget.tryAcquire(io).?);
+    budget.release(io, first);
+    budget.release(io, second);
+}
+
+test "init survives every allocation failure" {
+    try testing.checkAllAllocationFailures(testing.allocator, struct {
+        fn run(gpa: Allocator) !void {
+            var budget: QueryBudget = try .init(gpa, 2);
+            budget.deinit();
+        }
+    }.run, .{});
+}
+
+test "the default budget stays small" {
+    // 4096 + 65535 + scratch per permit: about 2.4 MiB for the whole process,
+    // where each UDP and TCP listener used to hold 4.8 and 8.7 MiB of its own.
+    try testing.expect(@sizeOf(Work) * permits < 3 * 1024 * 1024);
+}
diff --git a/src/server/resolver_integration_test.zig b/src/server/resolver_integration_test.zig
index 8125140..6f5067d 100644
--- a/src/server/resolver_integration_test.zig
+++ b/src/server/resolver_integration_test.zig
@@ -16,6 +16,7 @@ const build_options = @import("build_options");
 const net = std.Io.net;
 
 const handler = @import("handler.zig");
+const query_budget = @import("query_budget.zig");
 const tcp_server = @import("tcp_server.zig");
 const udp_server = @import("udp_server.zig");
 const model = @import("../config/model.zig");
@@ -282,8 +283,10 @@ test "the whole resolver answers over udp and tcp and fails over to a healthy up
     var h = bareHandler(h_owner.client(upstreams.client()));
 
     const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
-    var udp = try udp_server.UdpServer.bind(gpa, io, listen_address, &h, .{ .max_in_flight = 4 });
-    var tcp = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, .{ .max_connections = 2 });
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var udp = try udp_server.UdpServer.bind(io, listen_address, &h, &permits);
+    var tcp = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, &permits, .{ .max_connections = 2 });
     const udp_address = udp.boundAddress();
     const tcp_address = tcp.boundAddress();
 
@@ -293,7 +296,7 @@ test "the whole resolver answers over udp and tcp and fails over to a healthy up
     // Deferred so a failed check still stops both servers: a `serve` left in
     // accept makes `threaded.deinit` wait for it forever.
     defer {
-        udp.deinit(gpa, io);
+        udp.deinit(io);
         tcp.deinit(io);
         group.cancel(io);
     }
diff --git a/src/server/tcp_server.zig b/src/server/tcp_server.zig
index f7f6fea..8d4ecbf 100644
--- a/src/server/tcp_server.zig
+++ b/src/server/tcp_server.zig
@@ -9,7 +9,9 @@
 //! `listener.Core`'s (milestone-18 ruling 1); this file is the per-connection
 //! loop and nothing else. Connection slots are fixed and pre-allocated. Over
 //! capacity the listener closes the new stream immediately and counts it; it
-//! never queues, and it never allocates per connection.
+//! never queues, and it never allocates per connection. Each query holds a
+//! permit from the shared query budget from its last byte read to its reply's
+//! last byte written; with none free the connection ends and is counted.
 //!
 //! No stream read or write in 0.16.0 accepts a timeout, so every per-connection
 //! operation is raced against `Options.idle_timeout` through `listener.race`
@@ -19,20 +21,21 @@ const std = @import("std");
 const address = @import("../platform/address.zig");
 const handler = @import("handler.zig");
 const listener = @import("listener.zig");
+const query_budget = @import("query_budget.zig");
 const transport = @import("../upstream/transport.zig");
 
 /// The stream buffers only stage the framing bytes. A message longer than this
-/// is read straight into `Conn.query` and written straight from `Conn.reply`,
-/// so making them larger would buy nothing.
+/// is read straight into `Payload.query` and written straight from the
+/// permit's reply, so making them larger would buy nothing.
 const stream_buffer_len = 1024;
 
 pub const Options = struct {
-    max_connections: u16 = 64,
+    max_connections: u16 = listener.dns_max_connections,
     /// RFC 7766 §6.2.3 recommends a few seconds of idle tolerance.
     idle_timeout: std.Io.Clock.Duration = .{ .raw = .fromSeconds(10), .clock = .awake },
 };
 
-/// TCP/53 keeps no counter of its own: the shared six are exactly what it
+/// TCP/53 keeps no counter of its own: the shared ones are exactly what it
 /// counts.
 pub const Stats = listener.CoreStats;
 
@@ -47,24 +50,20 @@ pub const Snapshot = struct {
     accept_errors: u64,
     connection_errors: u64,
     idle_timeouts: u64,
+    dropped_no_permit: u64,
 };
 
 pub const TcpServer = struct {
     core: listener.Core(Config),
     handler: *handler.Handler,
+    budget: *query_budget.QueryBudget,
     options: Options,
 
-    /// One slot is ~137 KiB, so the default 64 connections cost ~8.8 MiB, which
-    /// is inside the PLAN §18 budget. The two message buffers cannot be shared
-    /// or shrunk: the handler holds the query while the reply is built, and
-    /// both ceilings are the 65535 bytes the length prefix can express.
+    /// The query is read here, before a permit is taken, so a client that
+    /// sends slowly holds connection memory and no query permit. 65535 bytes,
+    /// the most the length prefix can express.
     pub const Payload = struct {
         query: [transport.max_message_len]u8,
-        reply: [transport.max_message_len]u8,
-        /// The handler's per-query working memory. It belongs to the slot so
-        /// that answering a message allocates nothing, and a connection is
-        /// answered serially, so one query uses it at a time.
-        scratch: handler.Scratch,
     };
 
     const Config = struct {
@@ -85,11 +84,13 @@ pub const TcpServer = struct {
         io: std.Io,
         listen_address: std.Io.net.IpAddress,
         h: *handler.Handler,
+        budget: *query_budget.QueryBudget,
         options: Options,
     ) ListenError!TcpServer {
         return .{
             .core = try listener.Core(Config).listen(gpa, io, listen_address, options.max_connections),
             .handler = h,
+            .budget = budget,
             .options = options,
         };
     }
@@ -115,6 +116,7 @@ pub const TcpServer = struct {
             .accept_errors = stats.accept_errors.load(.monotonic),
             .connection_errors = stats.connection_errors.load(.monotonic),
             .idle_timeouts = stats.idle_timeouts.load(.monotonic),
+            .dropped_no_permit = stats.dropped_no_permit.load(.monotonic),
         };
     }
 
@@ -147,7 +149,7 @@ pub const TcpServer = struct {
                     listener.bump(&stats.idle_timeouts);
                     return;
                 },
-                .canceled => return,
+                .canceled, .unavailable => return,
                 .failed => {
                     listener.bump(&stats.connection_errors);
                     return;
@@ -172,7 +174,7 @@ pub const TcpServer = struct {
 
             switch (listener.race(io, budget, listener.readBody, .{ &reader.interface, conn.payload.query[0..len] })) {
                 .ok => {},
-                .canceled => return,
+                .canceled, .unavailable => return,
                 // A half-sent message is a broken peer, not an idle one.
                 .timed_out, .failed => {
                     listener.bump(&stats.connection_errors);
@@ -180,13 +182,20 @@ pub const TcpServer = struct {
                 },
             }
 
+            // There is no framing for "no answer", so the connection ends.
+            const work = self.budget.tryAcquire(io) orelse {
+                listener.bump(&stats.dropped_no_permit);
+                return;
+            };
+            defer self.budget.release(io, work);
+
             const outcome = self.handler.handle(
                 io,
                 .tcp,
                 address.NetAddress.fromIp(conn.peer),
                 conn.payload.query[0..len],
-                &conn.payload.reply,
-                &conn.payload.scratch,
+                &work.reply,
+                &work.scratch,
             );
             const bytes = switch (outcome) {
                 // There is no framing for "no answer", so the connection ends.
@@ -197,7 +206,7 @@ pub const TcpServer = struct {
             const out = transport.framePrefix(@intCast(bytes.len));
             switch (listener.race(io, budget, listener.writeReply, .{ &writer.interface, &out, bytes })) {
                 .ok => {},
-                .canceled => return,
+                .canceled, .unavailable => return,
                 .timed_out, .failed => {
                     listener.bump(&stats.connection_errors);
                     return;
diff --git a/src/server/tcp_server_integration_test.zig b/src/server/tcp_server_integration_test.zig
index 677d369..7293719 100644
--- a/src/server/tcp_server_integration_test.zig
+++ b/src/server/tcp_server_integration_test.zig
@@ -13,7 +13,9 @@ const std = @import("std");
 const build_options = @import("build_options");
 const net = std.Io.net;
 
+const concurrency = @import("../platform/concurrency.zig");
 const handler = @import("handler.zig");
+const query_budget = @import("query_budget.zig");
 const tcp_server = @import("tcp_server.zig");
 const model = @import("../config/model.zig");
 const response = @import("../filter/response.zig");
@@ -179,7 +181,9 @@ test "two length-prefixed queries share one connection" {
     var h = bareHandler(h_owner.client(fake.client()));
 
     const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
-    var server = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, .{ .max_connections = 2 });
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, &permits, .{ .max_connections = 2 });
     const server_address = server.boundAddress();
 
     var group: std.Io.Group = .init;
@@ -213,7 +217,9 @@ test "the claimed slot records the connecting client" {
     var h = bareHandler(h_owner.client(fake.client()));
 
     const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
-    var server = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, .{ .max_connections = 2 });
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, &permits, .{ .max_connections = 2 });
     const server_address = server.boundAddress();
 
     var group: std.Io.Group = .init;
@@ -295,7 +301,9 @@ test "a canceled serve does not wait for a live connection" {
     // The idle budget is the whole time a drain would have to wait out, so it
     // is set far beyond any patience this test run has: if `serve` waits for
     // the connection instead of canceling it, the wait never ends.
-    var server = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, .{
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, &permits, .{
         .max_connections = 2,
         .idle_timeout = .{ .raw = .fromSeconds(600), .clock = .awake },
     });
@@ -346,7 +354,9 @@ test "the connections peak is the most slots held at once" {
     var h = bareHandler(h_owner.client(fake.client()));
 
     const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
-    var server = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, .{
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, &permits, .{
         .max_connections = 3,
         .idle_timeout = .{ .raw = .fromSeconds(600), .clock = .awake },
     });
@@ -401,7 +411,9 @@ test "an idle connection is closed and counted" {
     var h = bareHandler(h_owner.client(fake.client()));
 
     const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
-    var server = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, .{
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, &permits, .{
         .max_connections = 2,
         .idle_timeout = short_idle,
     });
@@ -436,7 +448,9 @@ test "a zero-length message is a connection error" {
     var h = bareHandler(h_owner.client(fake.client()));
 
     const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
-    var server = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, .{
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, &permits, .{
         .max_connections = 2,
         .idle_timeout = short_idle,
     });
@@ -490,7 +504,9 @@ test "deinit ends a serve loop that is blocked on accept" {
     var h = bareHandler(h_owner.client(fake.client()));
 
     const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
-    var server = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, .{ .max_connections = 2 });
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, &permits, .{ .max_connections = 2 });
 
     var group: std.Io.Group = .init;
     try group.concurrent(io, tcp_server.TcpServer.serve, .{ &server, io });
@@ -501,3 +517,167 @@ test "deinit ends a serve loop that is blocked on accept" {
         error.Canceled => unreachable,
     };
 }
+
+/// Writes one framed query and returns the frame the server answered with.
+fn askOn(reader: *std.Io.Reader, writer: *std.Io.Writer) ![]const u8 {
+    try writer.writeAll(&transport.framePrefix(@intCast(query_bytes.len)));
+    try writer.writeAll(query_bytes);
+    try writer.flush();
+    const len = transport.parsePrefix((try reader.takeArray(transport.prefix_len)).*);
+    return reader.take(len);
+}
+
+/// One answered query, then the connection idles with every permit back in the
+/// budget; a second query that finds none taken ends the connection unanswered.
+fn keepAliveThenStarved(io: std.Io, address: net.IpAddress, permits: *query_budget.QueryBudget) anyerror!void {
+    var stream = try address.connect(io, .{ .mode = .stream });
+    defer stream.close(io);
+
+    var read_buf: [1024]u8 = undefined;
+    var write_buf: [1024]u8 = undefined;
+    var reader = stream.reader(io, &read_buf);
+    var writer = stream.writer(io, &write_buf);
+
+    try expectAnswersQuery(try askOn(&reader.interface, &writer.interface));
+
+    // The server returns the permit just after its reply's last byte, which can
+    // reach this side first. `bounded` caps the wait.
+    const tick: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(1), .clock = .awake };
+    while (permits.taken(io) != 0) try tick.sleep(io);
+
+    const held = permits.tryAcquire(io).?;
+    defer permits.release(io, held);
+
+    try writer.interface.writeAll(&transport.framePrefix(@intCast(query_bytes.len)));
+    try writer.interface.writeAll(query_bytes);
+    try writer.interface.flush();
+
+    var sink: [64]u8 = undefined;
+    if (try reader.interface.readSliceShort(&sink) != 0) return error.TestUnexpectedBytes;
+}
+
+test "a keep-alive connection holds no permit between queries, and a query without one ends it" {
+    if (!build_options.integration) return error.SkipZigTest;
+
+    const gpa = testing.allocator;
+    var threaded: std.Io.Threaded = .init(gpa, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var fake: FakeUpstream = .{ .reply = response_bytes };
+    var h_owner: upstream_owner.Borrowed = .{};
+    var h = bareHandler(h_owner.client(fake.client()));
+
+    const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
+    var permits: query_budget.QueryBudget = try .init(gpa, 1);
+    defer permits.deinit();
+    var server = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, &permits, .{ .max_connections = 2 });
+    const server_address = server.boundAddress();
+
+    var group: std.Io.Group = .init;
+    try group.concurrent(io, tcp_server.TcpServer.serve, .{ &server, io });
+    defer {
+        server.deinit(io);
+        group.await(io) catch |err| switch (err) {
+            error.Canceled => unreachable,
+        };
+    }
+
+    try bounded(io, keepAliveThenStarved, .{ io, server_address, &permits });
+
+    try testing.expectEqual(@as(u64, 1), server.core.stats.dropped_no_permit.load(.monotonic));
+    try testing.expectEqual(@as(u64, 0), server.core.stats.connection_errors.load(.monotonic));
+    try testing.expectEqual(@as(u64, 1), h.stats.queries.load(.monotonic));
+}
+
+/// Holds the only slot with an answered connection, then opens a second one,
+/// which the server closes without reading.
+fn oneHeldOneRefused(io: std.Io, address: net.IpAddress) anyerror!void {
+    var held = try address.connect(io, .{ .mode = .stream });
+    defer held.close(io);
+    var read_buf: [1024]u8 = undefined;
+    var write_buf: [1024]u8 = undefined;
+    var reader = held.reader(io, &read_buf);
+    var writer = held.writer(io, &write_buf);
+    try expectAnswersQuery(try askOn(&reader.interface, &writer.interface));
+
+    var refused = try address.connect(io, .{ .mode = .stream });
+    defer refused.close(io);
+    var refused_buf: [64]u8 = undefined;
+    var refused_reader = refused.reader(io, &refused_buf);
+    var sink: [64]u8 = undefined;
+    if (try refused_reader.interface.readSliceShort(&sink) != 0) return error.TestUnexpectedBytes;
+}
+
+test "a listener at its connection cap refuses and counts" {
+    if (!build_options.integration) return error.SkipZigTest;
+
+    const gpa = testing.allocator;
+    var threaded: std.Io.Threaded = .init(gpa, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var fake: FakeUpstream = .{ .reply = response_bytes };
+    var h_owner: upstream_owner.Borrowed = .{};
+    var h = bareHandler(h_owner.client(fake.client()));
+
+    const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, &permits, .{ .max_connections = 1 });
+    const server_address = server.boundAddress();
+
+    var group: std.Io.Group = .init;
+    try group.concurrent(io, tcp_server.TcpServer.serve, .{ &server, io });
+    defer {
+        server.deinit(io);
+        group.await(io) catch |err| switch (err) {
+            error.Canceled => unreachable,
+        };
+    }
+
+    try bounded(io, oneHeldOneRefused, .{ io, server_address });
+
+    try testing.expectEqual(@as(u64, 1), server.core.stats.rejected_at_capacity.load(.monotonic));
+    try testing.expectEqual(@as(u64, 1), server.core.stats.connections.load(.monotonic));
+}
+
+test "a connection that gets no task is closed and counted, not refused at capacity" {
+    if (!build_options.integration) return error.SkipZigTest;
+
+    const gpa = testing.allocator;
+    // The accept loop is the one task the server's pool allows. The client
+    // runs on a pool of its own, so `bounded` has the tasks it needs.
+    var server_threaded: std.Io.Threaded = .init(gpa, .{ .concurrent_limit = .limited(1) });
+    defer server_threaded.deinit();
+    const server_io = server_threaded.io();
+    var client_threaded: std.Io.Threaded = .init(gpa, .{});
+    defer client_threaded.deinit();
+    const io = client_threaded.io();
+
+    var fake: FakeUpstream = .{ .reply = response_bytes };
+    var h_owner: upstream_owner.Borrowed = .{};
+    var h = bareHandler(h_owner.client(fake.client()));
+
+    const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try tcp_server.TcpServer.listen(gpa, server_io, listen_address, &h, &permits, .{ .max_connections = 2 });
+    const server_address = server.boundAddress();
+
+    var group: std.Io.Group = .init;
+    try group.concurrent(server_io, tcp_server.TcpServer.serve, .{ &server, server_io });
+    defer {
+        server.deinit(server_io);
+        group.await(server_io) catch |err| switch (err) {
+            error.Canceled => unreachable,
+        };
+    }
+
+    const before = concurrency.snapshot().get(.connection);
+    try bounded(io, waitForServerClose, .{ io, server_address });
+
+    try testing.expectEqual(@as(u64, 1), concurrency.snapshot().get(.connection) - before);
+    try testing.expectEqual(@as(u64, 0), server.core.stats.rejected_at_capacity.load(.monotonic));
+    try testing.expectEqual(@as(u64, 0), server.core.stats.connections.load(.monotonic));
+}
diff --git a/src/server/udp_server.zig b/src/server/udp_server.zig
index b1833a4..e1e0a6f 100644
--- a/src/server/udp_server.zig
+++ b/src/server/udp_server.zig
@@ -1,49 +1,39 @@
 //! The UDP/53 listener.
 //!
-//! One receive loop feeds a bounded pool of in-flight tasks. A datagram that
-//! finds no free slot is dropped and counted, never queued: an unbounded queue
-//! only moves the failure from "visible drop" to "invisible latency", which is
-//! the failure mode PLAN §1 names.
+//! One receive loop hands each datagram to a task that holds a permit from the
+//! shared query budget. A datagram that finds no permit is dropped and counted,
+//! never queued: an unbounded queue only moves the failure from "visible drop"
+//! to "invisible latency", which is the failure mode PLAN §1 names.
 //!
-//! Every slot carries its own query and reply buffer, so answering a datagram
-//! allocates nothing. The pool is sized once in `bind` and never grows.
+//! The permit carries the query, the reply and the handler's scratch, so
+//! answering a datagram allocates nothing.
 
 const std = @import("std");
 const address = @import("../platform/address.zig");
+const concurrency = @import("../platform/concurrency.zig");
 const handler = @import("handler.zig");
-const transport = @import("../upstream/transport.zig");
+const query_budget = @import("query_budget.zig");
 
 const log = std.log.scoped(.udp_server);
 
-/// The EDNS ceiling this server advertises. A larger datagram arrives
-/// truncated and is dropped rather than guessed at.
-pub const max_datagram = 4096;
-
-comptime {
-    // `handler.handle` asserts this range on the buffers it is given.
-    std.debug.assert(max_datagram >= handler.udp_limit_min);
-    std.debug.assert(max_datagram >= handler.udp_limit_max);
-    std.debug.assert(transport.max_message_len >= handler.udp_limit_min);
-}
+pub const max_datagram = query_budget.max_datagram;
 
 /// How long a receive waits before the loop re-reads the stop flag. `deinit`
 /// therefore returns within this much time, without depending on a blocked
 /// `receive` being unblocked by a close — a guarantee POSIX does not make.
 const poll_interval: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(100), .clock = .awake };
 
-pub const Options = struct {
-    max_in_flight: u16 = 64,
-};
-
 pub const Stats = struct {
     received: std.atomic.Value(u64) = .init(0),
     /// The datagram arrived with `flags.trunc`: the kernel discarded its tail.
     dropped_oversize: std.atomic.Value(u64) = .init(0),
+    /// Every permit of the shared query budget was taken.
     dropped_no_slot: std.atomic.Value(u64) = .init(0),
     /// The handler answered `.drop` — an upstream exchange canceled by
     /// shutdown. The handler counts its own malformed drops; this counter is
     /// what keeps the listener's "every dropped datagram is counted" rule true
-    /// for the drops it cannot see.
+    /// for the drops it cannot see. A datagram that got a permit but no task
+    /// is counted under `nxdns_concurrency_unavailable_total{site="udp_query"}`.
     dropped_handler: std.atomic.Value(u64) = .init(0),
     receive_errors: std.atomic.Value(u64) = .init(0),
     send_errors: std.atomic.Value(u64) = .init(0),
@@ -63,69 +53,31 @@ pub const Snapshot = struct {
 };
 
 /// Lifecycle of the receive loop. `serve` claims `.serving`, `deinit` publishes
-/// `.closing`, and the two meet at `stopped` so no task touches the slots after
-/// they are freed.
+/// `.closing`, and the two meet at `stopped` so no task touches the socket
+/// after it is closed.
 const State = enum(u32) { idle, serving, closing };
 
 pub const UdpServer = struct {
     socket: std.Io.net.Socket,
     handler: *handler.Handler,
-    slots: []Slot,
-    mutex: std.Io.Mutex,
+    budget: *query_budget.QueryBudget,
     stats: Stats,
     state: std.atomic.Value(State),
     stopped: std.Io.Event,
 
-    pub const Slot = struct {
-        query: [max_datagram]u8,
-        /// The buffer the handler receives the upstream answer into, so it
-        /// holds a whole DNS message, not a whole datagram. A reply between
-        /// `max_datagram` and `transport.max_message_len` must reach the
-        /// handler intact: only the handler can turn it into the TC=1 reply
-        /// that makes the client retry over TCP (S5.3). A 4096-byte buffer
-        /// would instead fail the exchange with `error.ResponseTooLarge` and
-        /// answer SERVFAIL, which is wrong.
-        ///
-        /// The datagram actually sent stays bounded by `udpLimit` inside the
-        /// handler, so nothing larger than 4096 bytes leaves this socket.
-        ///
-        /// Cost: 4096 + 65535 + the scratch below ≈ 74 KiB per slot, so the
-        /// default 64 slots hold ≈ 4.6 MiB. The PLAN §18 budget is 100 MiB with
-        /// ~1M blocked domains, so this pool takes about 5% of it.
-        reply: [transport.max_message_len]u8,
-        /// The handler's per-query working memory. It belongs to the slot so
-        /// that answering a datagram still allocates nothing, and one slot
-        /// serves one query at a time.
-        scratch: handler.Scratch,
-        from: std.Io.net.IpAddress,
-        len: usize,
-        /// Guarded by `UdpServer.mutex`.
-        in_use: bool,
-    };
-
-    pub const BindError = std.Io.net.IpAddress.BindError || error{OutOfMemory};
+    pub const BindError = std.Io.net.IpAddress.BindError;
 
     pub fn bind(
-        gpa: std.mem.Allocator,
         io: std.Io,
         bind_address: std.Io.net.IpAddress,
         h: *handler.Handler,
-        options: Options,
+        budget: *query_budget.QueryBudget,
     ) BindError!UdpServer {
-        std.debug.assert(options.max_in_flight > 0);
-
-        const slots = try gpa.alloc(Slot, options.max_in_flight);
-        errdefer gpa.free(slots);
-        for (slots) |*slot| slot.in_use = false;
-
-        const local = bind_address;
-        const socket = try local.bind(io, .{ .mode = .dgram });
-
+        const socket = try bind_address.bind(io, .{ .mode = .dgram });
         return .{
             .socket = socket,
             .handler = h,
-            .slots = slots,
-            .mutex = .init,
+            .budget = budget,
             .stats = .{},
             .state = .init(.idle),
             .stopped = .unset,
@@ -139,7 +91,7 @@ pub const UdpServer = struct {
 
     /// The counters, read one at a time. A scrape that lands mid-datagram can
     /// see a receive counted before its drop is; a lock would buy a consistency
-    /// no consumer needs, and the receive loop takes that lock per datagram.
+    /// no consumer needs.
     pub fn snapshotStats(self: *const UdpServer) Snapshot {
         return .{
             .received = self.stats.received.load(.monotonic),
@@ -169,172 +121,88 @@ pub const UdpServer = struct {
         self.stopped.set(io);
     }
 
-    pub fn deinit(self: *UdpServer, gpa: std.mem.Allocator, io: std.Io) void {
+    pub fn deinit(self: *UdpServer, io: std.Io) void {
         // `serve` reads this between receives, so it stops on its own and the
         // socket is closed only once nothing can be reading from it.
         if (self.state.swap(.closing, .acq_rel) == .serving) self.stopped.waitUncancelable(io);
         self.socket.close(io);
-        gpa.free(self.slots);
         self.* = undefined;
     }
 
     fn receiveLoop(self: *UdpServer, io: std.Io, group: *std.Io.Group) void {
-        // A datagram that finds no free slot still has to leave the socket
-        // buffer, or the loop would spin on the same undeliverable packet.
-        var overflow: [max_datagram]u8 = undefined;
+        // The datagram is received here, without a permit, so an idle listener
+        // holds none; a query that gets one is copied into it.
+        var datagram: [max_datagram]u8 = undefined;
 
         while (self.state.load(.acquire) == .serving) {
-            const claimed = self.claim(io);
-            const buffer = if (claimed) |index| &self.slots[index].query else &overflow;
-
-            const msg = self.socket.receiveTimeout(io, buffer, .{ .duration = poll_interval }) catch |err| {
-                if (claimed) |index| self.release(io, index);
+            const msg = self.socket.receiveTimeout(io, &datagram, .{ .duration = poll_interval }) catch |err| {
                 switch (err) {
                     error.Canceled => return,
                     // The stop flag is re-read at the top of the loop.
                     error.Timeout => continue,
+                    error.ConcurrencyUnavailable => concurrency.unavailable(.udp_receive),
                     else => {
                         bump(&self.stats.receive_errors);
                         log.debug("udp receive failed: {t}", .{err});
-                        // One bad datagram must not kill the listener, and a
-                        // persistent error must not turn it into a spin.
-                        poll_interval.sleep(io) catch return;
-                        continue;
                     },
                 }
+                // One bad datagram must not kill the listener, and a persistent
+                // error must not turn it into a spin.
+                poll_interval.sleep(io) catch return;
+                continue;
             };
 
             bump(&self.stats.received);
 
-            const index = claimed orelse {
-                bump(&self.stats.dropped_no_slot);
-                continue;
-            };
-
             // The kernel already threw the tail away, so the message cannot be
             // parsed and any answer would be a guess.
             if (msg.flags.trunc) {
                 bump(&self.stats.dropped_oversize);
-                self.release(io, index);
                 continue;
             }
 
-            const slot = &self.slots[index];
-            slot.from = msg.from;
-            slot.len = msg.data.len;
+            const work = self.budget.tryAcquire(io) orelse {
+                bump(&self.stats.dropped_no_slot);
+                continue;
+            };
+            @memcpy(work.datagram[0..msg.data.len], msg.data);
 
-            group.concurrent(io, respondOne, .{ self, io, index }) catch |err| switch (err) {
+            group.concurrent(io, respondOne, .{ self, io, work, msg.from, msg.data.len }) catch |err| switch (err) {
                 error.ConcurrencyUnavailable => {
-                    bump(&self.stats.dropped_no_slot);
-                    self.release(io, index);
+                    concurrency.unavailable(.udp_query);
+                    self.budget.release(io, work);
                 },
             };
         }
     }
 
-    fn respondOne(self: *UdpServer, io: std.Io, index: usize) void {
-        const slot = &self.slots[index];
-        defer self.release(io, index);
+    fn respondOne(
+        self: *UdpServer,
+        io: std.Io,
+        work: *query_budget.Work,
+        from: std.Io.net.IpAddress,
+        len: usize,
+    ) void {
+        defer self.budget.release(io, work);
 
         const outcome = self.handler.handle(
             io,
             .udp,
-            address.NetAddress.fromIp(slot.from),
-            slot.query[0..slot.len],
-            &slot.reply,
-            &slot.scratch,
+            address.NetAddress.fromIp(from),
+            work.datagram[0..len],
+            &work.reply,
+            &work.scratch,
         );
         switch (outcome) {
             .drop => bump(&self.stats.dropped_handler),
-            .reply => |bytes| self.socket.send(io, &slot.from, bytes) catch |err| {
+            .reply => |bytes| self.socket.send(io, &from, bytes) catch |err| {
                 bump(&self.stats.send_errors);
                 log.debug("udp send to a client failed: {t}", .{err});
             },
         }
     }
-
-    fn claim(self: *UdpServer, io: std.Io) ?usize {
-        // Uncancelable: this section takes no Io and never blocks on a peer, so
-        // it cannot deadlock, and losing the lock mid-update would leak a slot.
-        self.mutex.lockUncancelable(io);
-        defer self.mutex.unlock(io);
-
-        const index = firstFree(self.slots) orelse return null;
-        self.slots[index].in_use = true;
-        return index;
-    }
-
-    fn release(self: *UdpServer, io: std.Io, index: usize) void {
-        self.mutex.lockUncancelable(io);
-        defer self.mutex.unlock(io);
-        self.slots[index].in_use = false;
-    }
 };
 
-/// The capacity rule, without the mutex, so it is testable without a backend.
-fn firstFree(slots: []const UdpServer.Slot) ?usize {
-    for (slots, 0..) |*slot, index| {
-        if (!slot.in_use) return index;
-    }
-    return null;
-}
-
 fn bump(counter: *std.atomic.Value(u64)) void {
     _ = counter.fetchAdd(1, .monotonic);
 }
-
-const testing = std.testing;
-
-fn testSlots(count: usize) ![]UdpServer.Slot {
-    const slots = try testing.allocator.alloc(UdpServer.Slot, count);
-    for (slots) |*slot| slot.in_use = false;
-    return slots;
-}
-
-test "the slot pool hands out every index once" {
-    const slots = try testSlots(3);
-    defer testing.allocator.free(slots);
-
-    for (0..slots.len) |expected| {
-        const index = firstFree(slots) orelse return error.TestUnexpectedResult;
-        try testing.expectEqual(expected, index);
-        slots[index].in_use = true;
-    }
-}
-
-test "a full slot pool refuses instead of growing" {
-    const slots = try testSlots(2);
-    defer testing.allocator.free(slots);
-
-    for (slots) |*slot| slot.in_use = true;
-    try testing.expectEqual(@as(?usize, null), firstFree(slots));
-}
-
-test "a slot's reply buffer holds a whole DNS message" {
-    const slots = try testSlots(1);
-    defer testing.allocator.free(slots);
-
-    // The handler asserts both ends of this range on the buffer it is given,
-    // and an upstream reply of up to 65535 bytes has to fit so the handler can
-    // truncate it instead of the exchange failing.
-    try testing.expectEqual(@as(usize, transport.max_message_len), slots[0].reply.len);
-    try testing.expect(slots[0].reply.len >= handler.udp_limit_min);
-    try testing.expect(slots[0].reply.len > max_datagram);
-}
-
-test "the default slot pool stays inside the memory budget" {
-    // 4096 + 65535 + scratch ≈ 74 KiB per slot; 64 slots ≈ 4.6 MiB, against the
-    // 100 MiB of PLAN §18.
-    const options: Options = .{};
-    const pool_bytes = @sizeOf(UdpServer.Slot) * @as(usize, options.max_in_flight);
-    try testing.expect(pool_bytes < 8 * 1024 * 1024);
-}
-
-test "a released slot is handed out again" {
-    const slots = try testSlots(2);
-    defer testing.allocator.free(slots);
-
-    for (slots) |*slot| slot.in_use = true;
-    slots[1].in_use = false;
-    try testing.expectEqual(@as(?usize, 1), firstFree(slots));
-}
diff --git a/src/server/udp_server_integration_test.zig b/src/server/udp_server_integration_test.zig
index 2f9d0c0..6856518 100644
--- a/src/server/udp_server_integration_test.zig
+++ b/src/server/udp_server_integration_test.zig
@@ -12,7 +12,9 @@ const std = @import("std");
 const build_options = @import("build_options");
 const net = std.Io.net;
 
+const concurrency = @import("../platform/concurrency.zig");
 const handler = @import("handler.zig");
+const query_budget = @import("query_budget.zig");
 const udp_server = @import("udp_server.zig");
 const model = @import("../config/model.zig");
 const response = @import("../filter/response.zig");
@@ -116,7 +118,9 @@ test "a udp query is answered on the loopback" {
     var h = bareHandler(h_owner.client(fake.client()));
 
     const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
-    var server = try udp_server.UdpServer.bind(gpa, io, listen_address, &h, .{ .max_in_flight = 4 });
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try udp_server.UdpServer.bind(io, listen_address, &h, &permits);
     const server_address = server.boundAddress();
 
     var group: std.Io.Group = .init;
@@ -124,7 +128,7 @@ test "a udp query is answered on the loopback" {
     // Deferred so a failed check still stops the server: a `serve` left in
     // accept makes `threaded.deinit` wait for it forever.
     defer {
-        server.deinit(gpa, io);
+        server.deinit(io);
         group.await(io) catch |err| switch (err) {
             error.Canceled => unreachable,
         };
@@ -158,13 +162,15 @@ test "a runt datagram is dropped and no reply is sent" {
     var h = bareHandler(h_owner.client(fake.client()));
 
     const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
-    var server = try udp_server.UdpServer.bind(gpa, io, listen_address, &h, .{ .max_in_flight = 4 });
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try udp_server.UdpServer.bind(io, listen_address, &h, &permits);
     const server_address = server.boundAddress();
 
     var group: std.Io.Group = .init;
     try group.concurrent(io, udp_server.UdpServer.serve, .{ &server, io });
     defer {
-        server.deinit(gpa, io);
+        server.deinit(io);
         group.await(io) catch |err| switch (err) {
             error.Canceled => unreachable,
         };
@@ -195,13 +201,15 @@ test "an oversize datagram arrives truncated and is dropped" {
     var h = bareHandler(h_owner.client(fake.client()));
 
     const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
-    var server = try udp_server.UdpServer.bind(gpa, io, listen_address, &h, .{ .max_in_flight = 4 });
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try udp_server.UdpServer.bind(io, listen_address, &h, &permits);
     const server_address = server.boundAddress();
 
     var group: std.Io.Group = .init;
     try group.concurrent(io, udp_server.UdpServer.serve, .{ &server, io });
     defer {
-        server.deinit(gpa, io);
+        server.deinit(io);
         group.await(io) catch |err| switch (err) {
             error.Canceled => unreachable,
         };
@@ -238,14 +246,101 @@ test "deinit ends a serve loop that is blocked on receive" {
     var h = bareHandler(h_owner.client(fake.client()));
 
     const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
-    var server = try udp_server.UdpServer.bind(gpa, io, listen_address, &h, .{ .max_in_flight = 4 });
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try udp_server.UdpServer.bind(io, listen_address, &h, &permits);
 
     var group: std.Io.Group = .init;
     try group.concurrent(io, udp_server.UdpServer.serve, .{ &server, io });
 
     // No datagram ever arrives, so `serve` is inside a receive when this runs.
-    server.deinit(gpa, io);
+    server.deinit(io);
     group.await(io) catch |err| switch (err) {
         error.Canceled => unreachable,
     };
 }
+
+test "with every permit taken a udp query is dropped and counted" {
+    if (!build_options.integration) return error.SkipZigTest;
+
+    const gpa = testing.allocator;
+    var threaded: std.Io.Threaded = .init(gpa, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var fake: FakeUpstream = .{ .reply = response_bytes };
+    var h_owner: upstream_owner.Borrowed = .{};
+    var h = bareHandler(h_owner.client(fake.client()));
+
+    const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
+    var permits: query_budget.QueryBudget = try .init(gpa, 1);
+    defer permits.deinit();
+    var server = try udp_server.UdpServer.bind(io, listen_address, &h, &permits);
+    const server_address = server.boundAddress();
+
+    var group: std.Io.Group = .init;
+    try group.concurrent(io, udp_server.UdpServer.serve, .{ &server, io });
+    defer {
+        server.deinit(io);
+        group.await(io) catch |err| switch (err) {
+            error.Canceled => unreachable,
+        };
+    }
+
+    // Another listener's query, as far as this one can tell.
+    const held = permits.tryAcquire(io).?;
+    defer permits.release(io, held);
+
+    const client_address: net.IpAddress = try .parse("127.0.0.1", 0);
+    var client = try client_address.bind(io, .{ .mode = .dgram });
+    defer client.close(io);
+
+    try client.send(io, &server_address, query_bytes);
+
+    var buf: [udp_server.max_datagram]u8 = undefined;
+    try testing.expectError(error.Timeout, client.receiveTimeout(io, &buf, silence));
+    try testing.expectEqual(@as(u64, 1), server.stats.dropped_no_slot.load(.monotonic));
+    try testing.expectEqual(@as(u64, 0), h.stats.queries.load(.monotonic));
+}
+
+test "a udp query with a permit but no task is dropped, counted, and returns its permit" {
+    if (!build_options.integration) return error.SkipZigTest;
+
+    const gpa = testing.allocator;
+    // The serve task is the one task this pool allows.
+    var threaded: std.Io.Threaded = .init(gpa, .{ .concurrent_limit = .limited(1) });
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var fake: FakeUpstream = .{ .reply = response_bytes };
+    var h_owner: upstream_owner.Borrowed = .{};
+    var h = bareHandler(h_owner.client(fake.client()));
+
+    const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
+    var permits: query_budget.QueryBudget = try .init(gpa, 4);
+    defer permits.deinit();
+    var server = try udp_server.UdpServer.bind(io, listen_address, &h, &permits);
+    const server_address = server.boundAddress();
+
+    var group: std.Io.Group = .init;
+    try group.concurrent(io, udp_server.UdpServer.serve, .{ &server, io });
+    defer {
+        server.deinit(io);
+        group.await(io) catch |err| switch (err) {
+            error.Canceled => unreachable,
+        };
+    }
+
+    const client_address: net.IpAddress = try .parse("127.0.0.1", 0);
+    var client = try client_address.bind(io, .{ .mode = .dgram });
+    defer client.close(io);
+
+    const before = concurrency.snapshot().get(.udp_query);
+    try client.send(io, &server_address, query_bytes);
+
+    var buf: [udp_server.max_datagram]u8 = undefined;
+    try testing.expectError(error.Timeout, client.receiveTimeout(io, &buf, silence));
+    try testing.expectEqual(@as(u64, 1), concurrency.snapshot().get(.udp_query) - before);
+    try testing.expectEqual(@as(usize, 0), permits.taken(io));
+    try testing.expectEqual(@as(u64, 0), server.stats.dropped_no_slot.load(.monotonic));
+}
diff --git a/src/storage/logger.zig b/src/storage/logger.zig
index b4a89ec..3485e0c 100644
--- a/src/storage/logger.zig
+++ b/src/storage/logger.zig
@@ -33,6 +33,7 @@ const std = @import("std");
 const Allocator = std.mem.Allocator;
 const builtin = @import("builtin");
 
+const concurrency = @import("../platform/concurrency.zig");
 const db = @import("db.zig");
 const disk_monitor = @import("disk_monitor.zig");
 const events = @import("events.zig");
@@ -817,10 +818,16 @@ pub const Logger = struct {
         race.concurrent(.entry, takeOne, .{ &self.queue, io }) catch |err| switch (err) {
             // No second unit of concurrency: the caller flushes what it holds
             // rather than block past the interval.
-            error.ConcurrencyUnavailable => return null,
+            error.ConcurrencyUnavailable => {
+                concurrency.unavailable(.querylog_wait);
+                return null;
+            },
         };
         race.concurrent(.expiry, expire, .{ io, budget }) catch |err| switch (err) {
-            error.ConcurrencyUnavailable => return drainRace(&race),
+            error.ConcurrencyUnavailable => {
+                concurrency.unavailable(.querylog_wait);
+                return drainRace(&race);
+            },
         };
 
         const first = race.await() catch |err| {
@@ -1259,6 +1266,20 @@ test "log applies both privacy transforms before the entry reaches the queue" {
     try testing.expectEqual(@as(u64, 0), logger.queries_dropped.load(.monotonic));
 }
 
+test "a wait that cannot start its race is counted and ends the batch early" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{ .concurrent_limit = .nothing });
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var buf: [4]Entry = undefined;
+    var logger: Logger = .init(.{}, &buf);
+
+    const before = concurrency.snapshot().get(.querylog_wait);
+    const long: std.Io.Clock.Duration = .{ .raw = .fromSeconds(60), .clock = .awake };
+    try testing.expectEqual(@as(?Entry, null), try logger.getWithin(io, long, Gate.initial));
+    try testing.expectEqual(@as(u64, 1), concurrency.snapshot().get(.querylog_wait) - before);
+}
+
 test "hide_domains hides every query-derived name and leaves the labels alone" {
     var threaded: std.Io.Threaded = .init(testing.allocator, .{});
     defer threaded.deinit();
diff --git a/src/storage/logger_controller.zig b/src/storage/logger_controller.zig
index 01d4291..4f3ffbe 100644
--- a/src/storage/logger_controller.zig
+++ b/src/storage/logger_controller.zig
@@ -38,6 +38,7 @@
 const std = @import("std");
 const Allocator = std.mem.Allocator;
 
+const concurrency = @import("../platform/concurrency.zig");
 const db = @import("db.zig");
 const disk_monitor = @import("disk_monitor.zig");
 const events = @import("events.zig");
@@ -249,19 +250,19 @@ pub const Controller = struct {
             .live = generation,
         };
 
-        owned.writer = try io.concurrent(logger.Logger.runWriter, .{
+        owned.writer = try concurrency.counted(.service, io.concurrent(logger.Logger.runWriter, .{
             generation.logger,
             io,
             opts.gpa,
             &owned.database,
             opts.monitor,
-        });
+        }));
         errdefer {
             generation.logger.shutdown(io);
             owned.writer.await(io) catch {};
         }
 
-        self.reaper = try io.concurrent(runReaper, .{ self, io });
+        self.reaper = try concurrency.counted(.service, io.concurrent(runReaper, .{ self, io }));
     }
 
     /// Stops every generation in the order the writer deadlock demands, and
@@ -444,7 +445,7 @@ pub const Controller = struct {
         errdefer generation.deinit(self.gpa);
         const owned = &generation.owned.?;
 
-        owned.writer = try io.concurrent(runParkedWriter, .{ generation, io, self.gpa, self.monitor });
+        owned.writer = try concurrency.counted(.service, io.concurrent(runParkedWriter, .{ generation, io, self.gpa, self.monitor }));
 
         // The statements are prepared before anything is published, so a
         // failure here is a refused settings change rather than a writer that
diff --git a/src/tests.zig b/src/tests.zig
index ce5a256..22ff799 100644
--- a/src/tests.zig
+++ b/src/tests.zig
@@ -14,6 +14,7 @@ comptime {
     _ = @import("dns/packet.zig");
     _ = @import("dns/dns.zig");
     _ = @import("platform/address.zig");
+    _ = @import("platform/concurrency.zig");
     _ = @import("platform/tls_client.zig");
     _ = @import("platform/tls_client_integration_test.zig");
     _ = @import("platform/tls_server.zig");
@@ -83,6 +84,7 @@ comptime {
     _ = @import("local/reverse_name.zig");
     _ = @import("cache/dns_cache.zig");
     _ = @import("server/rate_limiter.zig");
+    _ = @import("server/query_budget.zig");
     _ = @import("storage/repositories/queries_repo.zig");
     _ = @import("storage/repositories/events_repo.zig");
     _ = @import("storage/events.zig");
diff --git a/src/upstream/transport.zig b/src/upstream/transport.zig
index d551b7d..d77d68b 100644
--- a/src/upstream/transport.zig
+++ b/src/upstream/transport.zig
@@ -16,6 +16,7 @@
 //! wrong bucket.
 
 const std = @import("std");
+const concurrency = @import("../platform/concurrency.zig");
 const packet = @import("../dns/packet.zig");
 const name = @import("../dns/name.zig");
 
@@ -390,10 +391,16 @@ pub fn raceUntilTagged(
     defer race.cancelDiscard();
 
     race.concurrent(.raced, f, args) catch |err| switch (err) {
-        error.ConcurrencyUnavailable => return error.SystemResources,
+        error.ConcurrencyUnavailable => {
+            concurrency.unavailable(.exchange_race);
+            return error.SystemResources;
+        },
     };
     race.concurrent(.expiry, expire, .{ io, expiry_at }) catch |err| switch (err) {
-        error.ConcurrencyUnavailable => return error.SystemResources,
+        error.ConcurrencyUnavailable => {
+            concurrency.unavailable(.exchange_race);
+            return error.SystemResources;
+        },
     };
 
     switch (try race.await()) {
@@ -919,6 +926,22 @@ test "raceUntilTagged tells a raced Timeout apart from an expiry" {
     try testing.expectEqual(RaceOutcome.expired, outcome);
 }
 
+test "raceUntilTagged that cannot start is a local resource fault, counted" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{ .concurrent_limit = .nothing });
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    const before = concurrency.snapshot().get(.exchange_race);
+    var outcome: RaceOutcome = .completed;
+    const far: std.Io.Clock.Timestamp = .fromNow(io, .{ .raw = .fromSeconds(30), .clock = .awake });
+    try testing.expectError(
+        error.SystemResources,
+        raceUntilTagged(io, far, &outcome, racedReply, .{ io, 0, @as(ExchangeError!usize, 7) }),
+    );
+    try testing.expectEqual(RaceOutcome.completed, outcome);
+    try testing.expectEqual(@as(u64, 1), concurrency.snapshot().get(.exchange_race) - before);
+}
+
 test "raceUntilTagged tags a successful completion" {
     var threaded: std.Io.Threaded = .init(testing.allocator, .{});
     defer threaded.deinit();
diff --git a/src/web/metrics.zig b/src/web/metrics.zig
index d9d8599..c7d281c 100644
--- a/src/web/metrics.zig
+++ b/src/web/metrics.zig
@@ -26,6 +26,7 @@ const Allocator = std.mem.Allocator;
 const cert_store = @import("../server/cert_store.zig");
 const client_names = @import("../server/client_names.zig");
 const clients = @import("../server/clients.zig");
+const concurrency = @import("../platform/concurrency.zig");
 const dns_cache = @import("../cache/dns_cache.zig");
 const dns_handler = @import("../server/handler.zig");
 const disk_monitor = @import("../storage/disk_monitor.zig");
@@ -93,15 +94,18 @@ pub const DiskSample = struct {
 };
 
 /// The DoH listener counters this exposition exports (milestone-10 ruling 10):
-/// the four every TLS listener keeps, plus DoH's `bad_requests`. A subset of
-/// `doh_server.Snapshot` on purpose — the accept-side refusal counters stay
-/// internal, exactly as they do for the DoT listener and TCP/53.
+/// the ones every TLS listener keeps, plus DoH's `bad_requests`. A subset of
+/// `doh_server.Snapshot` on purpose: of the accept-side refusals only
+/// `rejected_at_capacity` is exported, the one a full connection cap explains,
+/// exactly as for the DoT listener.
 pub const DohListenerSample = struct {
     connections: u64,
+    rejected_at_capacity: u64,
     tls_handshake_failures: u64,
     idle_timeouts: u64,
     connection_errors: u64,
     bad_requests: u64,
+    dropped_no_permit: u64,
 };
 
 /// One upstream, with every string owned by the caller's arena.
@@ -152,6 +156,9 @@ pub const Sample = struct {
     dns: DnsCounters = @splat(0),
     logger: LoggerCounters = .{},
     log_sink: logging.Stats = .{},
+    /// Process-wide, like the log sink: every `concurrency.Site`, always
+    /// rendered, so a zero reads as "never happened" rather than "not wired".
+    concurrency: concurrency.Snapshot = .initFill(0),
     cache: ?CacheSample = null,
     limiter: ?LimiterSample = null,
     tracker: ?TrackerSample = null,
@@ -212,7 +219,7 @@ pub fn handle(
 }
 
 pub fn collect(state: *server.WebState, io: std.Io, arena: Allocator) Allocator.Error!Sample {
-    var sample: Sample = .{ .log_sink = logging.stats() };
+    var sample: Sample = .{ .log_sink = logging.stats(), .concurrency = concurrency.snapshot() };
 
     if (state.handler) |handler| {
         sample.dns = dnsCounters(&handler.stats);
@@ -290,10 +297,12 @@ pub fn collect(state: *server.WebState, io: std.Io, arena: Allocator) Allocator.
         const snapshot = listener.snapshotStats();
         sample.doh_listener = .{
             .connections = snapshot.connections,
+            .rejected_at_capacity = snapshot.rejected_at_capacity,
             .tls_handshake_failures = snapshot.tls_handshake_failures,
             .idle_timeouts = snapshot.idle_timeouts,
             .connection_errors = snapshot.connection_errors,
             .bad_requests = snapshot.bad_requests,
+            .dropped_no_permit = snapshot.dropped_no_permit,
         };
     }
     if (state.dot_listener) |listener| sample.dot_listener = listener.snapshotStats();
@@ -414,6 +423,16 @@ pub fn render(w: *std.Io.Writer, sample: Sample) std.Io.Writer.Error!void {
     try counterGroup(w, "nxdns_querylog_", "Query log writer counter", sample.logger);
     try counterGroup(w, "nxdns_log_", "Diagnostic log sink counter", sample.log_sink);
 
+    try labeledHead(
+        w,
+        "nxdns_concurrency_unavailable_total",
+        "Work lost because the task pool was at its limit or could not start a thread, by where it happened.",
+        "counter",
+    );
+    for (std.enums.values(concurrency.Site)) |site| {
+        try w.print("nxdns_concurrency_unavailable_total{{site=\"{t}\"}} {d}\n", .{ site, sample.concurrency.get(site) });
+    }
+
     if (sample.cache) |cache| {
         try counterGroup(w, "nxdns_cache_", "DNS cache counter", cache.stats);
         try gauge(w, "nxdns_cache_entries", "Responses currently held in the DNS cache.", cache.entries);
@@ -1048,16 +1067,20 @@ test "every HELP line has a TYPE line and a sample, and every sample a name" {
             try testing.expect(std.mem.startsWith(u8, line, "nxdns_"));
         }
     }
+    // Every family a bare state has is one sample, except the concurrency
+    // family, which has one per site.
+    const sites = std.enums.values(concurrency.Site).len;
     try testing.expectEqual(helps, types);
-    try testing.expectEqual(helps, samples);
+    try testing.expectEqual(helps - 1 + sites, samples);
     // The families a bare state still has: `nxdns_up`, every DNS counter, the
-    // query log writer's, and the diagnostic log sink's. Derived rather than
-    // counted, so a new counter in any of those structs extends the exposition
-    // and this assertion together.
+    // query log writer's, the diagnostic log sink's and the concurrency sites.
+    // Derived rather than counted, so a new counter in any of those structs
+    // extends the exposition and this assertion together.
     try testing.expectEqual(
         1 + dns_stat_fields.len +
             @typeInfo(LoggerCounters).@"struct".fields.len +
-            @typeInfo(logging.Stats).@"struct".fields.len,
+            @typeInfo(logging.Stats).@"struct".fields.len +
+            sites,
         samples,
     );
     // The reflective walk names the counters, so a renamed `Handler.Stats`
@@ -1141,6 +1164,7 @@ test "the plain-DNS listener families carry every counter of both listeners" {
             .accept_errors = 8,
             .connection_errors = 9,
             .idle_timeouts = 10,
+            .dropped_no_permit = 11,
         },
     });
     defer testing.allocator.free(text);
@@ -1155,6 +1179,7 @@ test "the plain-DNS listener families carry every counter of both listeners" {
     try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_tcp_server_connections_total 12\n"));
     try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_tcp_server_rejected_at_capacity_total 6\n"));
     try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_tcp_server_rejected_at_shutdown_total 7\n"));
+    try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_tcp_server_dropped_no_permit_total 11\n"));
     try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_tcp_server_accept_errors_total 8\n"));
     try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_tcp_server_connection_errors_total 9\n"));
     try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_tcp_server_idle_timeouts_total 10\n"));
@@ -1223,10 +1248,12 @@ test "listener counters render only for the wired servers" {
     const doh_only = try renderToString(testing.allocator, .{
         .doh_listener = .{
             .connections = 9,
+            .rejected_at_capacity = 6,
             .tls_handshake_failures = 2,
             .idle_timeouts = 1,
             .connection_errors = 0,
             .bad_requests = 4,
+            .dropped_no_permit = 7,
         },
     });
     defer testing.allocator.free(doh_only);
@@ -1236,14 +1263,18 @@ test "listener counters render only for the wired servers" {
     try testing.expect(std.mem.containsAtLeast(u8, doh_only, 1, "nxdns_doh_server_idle_timeouts_total 1\n"));
     try testing.expect(std.mem.containsAtLeast(u8, doh_only, 1, "nxdns_doh_server_connection_errors_total 0\n"));
     try testing.expect(std.mem.containsAtLeast(u8, doh_only, 1, "nxdns_doh_server_bad_requests_total 4\n"));
+    try testing.expect(std.mem.containsAtLeast(u8, doh_only, 1, "nxdns_doh_server_rejected_at_capacity_total 6\n"));
+    try testing.expect(std.mem.containsAtLeast(u8, doh_only, 1, "nxdns_doh_server_dropped_no_permit_total 7\n"));
     try testing.expect(!std.mem.containsAtLeast(u8, doh_only, 1, "nxdns_dot_server_"));
 
     const dot_only = try renderToString(testing.allocator, .{
         .dot_listener = .{
             .connections = 5,
+            .rejected_at_capacity = 2,
             .tls_handshake_failures = 0,
             .idle_timeouts = 3,
             .connection_errors = 1,
+            .dropped_no_permit = 4,
         },
     });
     defer testing.allocator.free(dot_only);
@@ -1251,11 +1282,28 @@ test "listener counters render only for the wired servers" {
     try testing.expect(std.mem.containsAtLeast(u8, dot_only, 1, "nxdns_dot_server_connections_total 5\n"));
     try testing.expect(std.mem.containsAtLeast(u8, dot_only, 1, "nxdns_dot_server_idle_timeouts_total 3\n"));
     try testing.expect(std.mem.containsAtLeast(u8, dot_only, 1, "nxdns_dot_server_connection_errors_total 1\n"));
+    try testing.expect(std.mem.containsAtLeast(u8, dot_only, 1, "nxdns_dot_server_rejected_at_capacity_total 2\n"));
+    try testing.expect(std.mem.containsAtLeast(u8, dot_only, 1, "nxdns_dot_server_dropped_no_permit_total 4\n"));
     // The DoT listener has no HTTP layer, so no bad_requests family.
     try testing.expect(!std.mem.containsAtLeast(u8, dot_only, 1, "nxdns_dot_server_bad_requests_total"));
     try testing.expect(!std.mem.containsAtLeast(u8, dot_only, 1, "nxdns_doh_server_"));
 }
 
+test "every concurrency site renders under its own label, zero included" {
+    var sample: Sample = .{};
+    sample.concurrency.set(.querylog_wait, 3);
+    const text = try renderToString(testing.allocator, sample);
+    defer testing.allocator.free(text);
+
+    try testing.expect(std.mem.containsAtLeast(u8, text, 1, "# TYPE nxdns_concurrency_unavailable_total counter\n"));
+    try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_concurrency_unavailable_total{site=\"querylog_wait\"} 3\n"));
+    for (std.enums.values(concurrency.Site)) |site| {
+        var line: [96]u8 = undefined;
+        const prefix = try std.fmt.bufPrint(&line, "nxdns_concurrency_unavailable_total{{site=\"{t}\"}} ", .{site});
+        try testing.expect(std.mem.containsAtLeast(u8, text, 1, prefix));
+    }
+}
+
 test "cert reload counters render per endpoint, only for the wired stores" {
     const one = try renderToString(testing.allocator, .{
         .doh_certs = .{ .reloads = 2, .reload_failures = 1, .last_reload_unix = 1_700_000_000 },
@@ -1474,7 +1522,7 @@ test "two upstreams on one host stay two series" {
         keys[count] = key;
         count += 1;
     }
-    try testing.expectEqual(@as(usize, 22), count);
+    try testing.expectEqual(@as(usize, 22 + std.enums.values(concurrency.Site).len), count);
 }
 
 test "an upstream url is redacted before it reaches an open endpoint's label" {