querylog: the writer queue defaults to 1024 entries, and a live stream allocates its ring on subscribe instead of at boot

CHANGELOG.md

@@ -14,6 +14,8 @@ Sections are written by hand. Nothing here is generated from commit messages: th
 ### Changed
 
 - **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`.
 - **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.
 

PLAN.md

@@ -489,7 +489,7 @@ The sketch above is the original shape; `src/storage/querylog_schema.zig` is the
 
 ### 11.4 Query Logger
 
-- In-memory buffer, mutex guarded, hard cap `query_log_buffer_max` (default 10000).
+- In-memory buffer, mutex guarded, hard cap `query_log_buffer_max` (default 1024, from the measured queue peak; see specs/memory-budget.md item 4).
 - Flush: batch size (100, comptime) or max interval `query_log_flush_interval_s` (default 60s, 0–3600, 0 = do not wait). One transaction per interval: at household query rates a per-query commit costs orders of magnitude more disk writes than the rows are worth. The interval is also roughly what a crash costs, while the writer is healthy and the disk gate is open — a gated or lock-delayed batch is older, so it is a normal case, not a bound.
 - Privacy transforms (hide_domains / hide_client_ips) applied before persist + SSE fanout.
 - Backpressure: buffer full → drop oldest unflushed entry, increment monotonic `queries_dropped` (exposed in `/api/health` + `/metrics`). SSE fanout precedes buffer insert, so live viewers still see dropped-from-persistence entries.

docs/reference/configuration.md

@@ -124,7 +124,7 @@ The process log and the query history.
 |---|---|---|---|---|
 | `logging.level` | `.err`, `.warn`, `.info` or `.debug` | `.info` | | The least important log line nxdns writes. The file says `.err`; the web interface says `error`. |
 | `logging.retention_days` | integer | 30 | at least 1 | How long the query history is kept, and how long a device that stopped asking stays in **Clients**. |
-| `logging.query_log_buffer_max` | integer | 10000 | 1–37449 | How many queries nxdns holds in memory while it waits to write them. |
+| `logging.query_log_buffer_max` | integer | 1024 | 1–37449 | How many queries nxdns holds in memory while it waits to write them, each about 1.8 KiB. The default covers over a minute and a half of queries at 10 a second, far more than a household asks while a write is in progress. When the buffer is full, the oldest waiting query is dropped from the history and counted. |
 | `logging.query_log_flush_interval_s` | integer | 60 | 0–3600 s | How often the query history is written to disk; see below. |
 | `logging.hide_domains` | bool | false | | Stores a placeholder instead of the name in the query history. |
 | `logging.hide_client_ips` | bool | false | | Stores a placeholder instead of the client's address in the query history. |

src/app.zig

@@ -584,16 +584,10 @@ fn serve(r: cli.Runner, args: cli.RunArgs) !u8 {
 
     // Milestone 8 fans every logged query out to the SSE hub as well. The hub
     // exists only when the web interface does (ruling 6) — without it the sink
-    // costs the query path one null check. Its rings are ~900 KiB, so it lives
-    // on the heap and initializes in place; a by-value init would copy the
-    // whole of it through this frame.
-    var hub: ?*sse.Hub = null;
-    defer if (hub) |hub_ptr| gpa.destroy(hub_ptr);
-    if (cfg.web.enabled) {
-        const created = try gpa.create(sse.Hub);
-        created.init();
-        hub = created;
-    }
+    // costs the query path one null check.
+    var hub_storage: sse.Hub = .init(gpa);
+    defer hub_storage.deinit();
+    const hub: ?*sse.Hub = if (cfg.web.enabled) &hub_storage else null;
 
     // -----------------------------------------------------------------------
     // disk, retention and the remaining connections (ruling 21)

src/config/model.zig

@@ -232,7 +232,11 @@ pub const LogOutput = enum {
 pub const Logging = struct {
     level: LogLevel = .info,
     retention_days: u16 = 30,
-    query_log_buffer_max: u32 = 10000,
+    /// The queue is allocated in full, about 1.8 KiB an entry. It only holds
+    /// queries that arrive while a batch commits or the disk gate waits: the
+    /// Pi's peak was 7 over 37 hours at 0.35 queries a second, so 1024 is
+    /// over a minute and a half of queries at 10 a second.
+    query_log_buffer_max: u32 = 1024,
     /// How long the query-log writer gathers entries before it commits them.
     /// `0` does not wait at all: it flushes the entry that woke the writer plus
     /// whatever is already queued.

src/server/handler.zig

@@ -4281,9 +4281,9 @@ test "the sink both streams and logs the query the handler answered" {
     defer t.deinit();
     const io = t.io();
 
-    const hub = try testing.allocator.create(sse.Hub);
-    defer testing.allocator.destroy(hub);
-    hub.init();
+    var hub_storage: sse.Hub = .init(testing.allocator);
+    defer hub_storage.deinit();
+    const hub = &hub_storage;
 
     var queue_buf: [4]logger_mod.Entry = undefined;
     var lg: logger_mod.Logger = .init(.{}, &queue_buf);
@@ -4295,7 +4295,7 @@ test "the sink both streams and logs the query the handler answered" {
     var h = bare(h_owner.client(fake.client()));
     h.sink = &sink;
 
-    const id = hub.subscribe(io).?;
+    const id = try hub.subscribe(io);
     defer hub.unsubscribe(io, id);
 
     var buf: [udp_limit_min]u8 = undefined;

src/server/query_sink.zig

@@ -66,16 +66,16 @@ test "the sink publishes and logs the same entry" {
     defer threaded.deinit();
     const io = threaded.io();
 
-    const hub = try testing.allocator.create(sse.Hub);
-    defer testing.allocator.destroy(hub);
-    hub.init();
+    var hub_storage: sse.Hub = .init(testing.allocator);
+    defer hub_storage.deinit();
+    const hub = &hub_storage;
 
     var queue_buf: [4]logger.Entry = undefined;
     var query_logger: logger.Logger = .init(.{}, &queue_buf);
     var owner: logger_controller.Borrowed = .{};
     var sink: QuerySink = .init(owner.over(&query_logger), hub);
 
-    const id = hub.subscribe(io).?;
+    const id = try hub.subscribe(io);
     defer hub.unsubscribe(io, id);
 
     sink.log(io, sampleEntry(11, "example.com"));
@@ -94,16 +94,16 @@ test "fanout does not depend on the entry reaching the queue" {
     defer threaded.deinit();
     const io = threaded.io();
 
-    const hub = try testing.allocator.create(sse.Hub);
-    defer testing.allocator.destroy(hub);
-    hub.init();
+    var hub_storage: sse.Hub = .init(testing.allocator);
+    defer hub_storage.deinit();
+    const hub = &hub_storage;
 
     var queue_buf: [4]logger.Entry = undefined;
     var query_logger: logger.Logger = .init(.{}, &queue_buf);
     var owner: logger_controller.Borrowed = .{};
     var sink: QuerySink = .init(owner.over(&query_logger), hub);
 
-    const id = hub.subscribe(io).?;
+    const id = try hub.subscribe(io);
     defer hub.unsubscribe(io, id);
 
     // A closed queue drops what it is handed. The stream still carries the
@@ -120,9 +120,9 @@ test "the privacy transforms run once, before both consumers" {
     defer threaded.deinit();
     const io = threaded.io();
 
-    const hub = try testing.allocator.create(sse.Hub);
-    defer testing.allocator.destroy(hub);
-    hub.init();
+    var hub_storage: sse.Hub = .init(testing.allocator);
+    defer hub_storage.deinit();
+    const hub = &hub_storage;
 
     var queue_buf: [4]logger.Entry = undefined;
     var query_logger: logger.Logger = .init(
@@ -132,7 +132,7 @@ test "the privacy transforms run once, before both consumers" {
     var owner: logger_controller.Borrowed = .{};
     var sink: QuerySink = .init(owner.over(&query_logger), hub);
 
-    const id = hub.subscribe(io).?;
+    const id = try hub.subscribe(io);
     defer hub.unsubscribe(io, id);
 
     sink.log(io, sampleEntry(4, "tracker.example"));

src/storage/logger.zig

@@ -268,7 +268,7 @@ pub const Entry = struct {
 
 /// The memory budget the queue is allowed to occupy. `Entry` travels by value,
 /// so the composition root allocates `query_log_buffer_max` of them in full at
-/// boot (`app.zig`) and the SSE hub embeds a ring of them per subscriber.
+/// boot (`app.zig`), and each live SSE subscriber holds a ring of them.
 const queue_budget_bytes = 64 * 1024 * 1024;
 
 /// The ceiling `config/validate.zig` enforces on `logging.query_log_buffer_max`,
@@ -1335,7 +1335,7 @@ test "the entry queue's worst case stays inside its byte budget" {
 
     // The shipped default has to be comfortably inside the budget, or the
     // out-of-the-box configuration is the one that spends it. At the widths
-    // above it costs about 17 MiB, roughly a quarter of the ceiling.
+    // above it costs under 2 MiB.
     const default_max: usize = (model.Logging{}).query_log_buffer_max;
     try testing.expect(default_max <= query_log_buffer_max);
     try testing.expect(default_max * @sizeOf(Entry) <= queue_budget_bytes / 2);

src/web/handlers/live.zig

@@ -72,8 +72,10 @@ pub fn stream(
     }
     defer if (state.limiter) |limiter| limiter.releaseSse(io, client);
 
-    const id = hub.subscribe(io) orelse
-        return http_util.respondError(request, .service_unavailable, "live stream is full");
+    const id = hub.subscribe(io) catch |err| return switch (err) {
+        error.Full => http_util.respondError(request, .service_unavailable, "live stream is full"),
+        error.OutOfMemory => http_util.respondError(request, .service_unavailable, "no memory for a live stream"),
+    };
     defer hub.unsubscribe(io, id);
 
     var response = try request.http.respondStreaming(&.{}, .{

src/web/metrics.zig

@@ -189,6 +189,9 @@ pub const Sample = struct {
     /// Exchanges the pool ended on this process's own resources. Pool-wide for
     /// the same reason as `upstream_budget_exhausted_total`.
     upstream_local_resource_total: ?u64 = null,
+    /// Live streams refused because their ring could not be allocated. Absent
+    /// while no hub is wired.
+    live_stream_allocation_failures_total: ?u64 = null,
     peaks: PeakSample = .{},
 };
 
@@ -307,6 +310,8 @@ pub fn collect(state: *server.WebState, io: std.Io, arena: Allocator) Allocator.
         }
     }
 
+    if (state.hub) |hub| sample.live_stream_allocation_failures_total = hub.ringAllocationFailures(io);
+
     return sample;
 }
 
@@ -525,6 +530,14 @@ pub fn render(w: *std.Io.Writer, sample: Sample) std.Io.Writer.Error!void {
         );
         try w.print("nxdns_upstream_local_resource_total {d}\n", .{total});
     }
+    if (sample.live_stream_allocation_failures_total) |total| {
+        try counter(
+            w,
+            "nxdns_live_stream_allocation_failures_total",
+            "Live query streams refused because their buffer could not be allocated.",
+            total,
+        );
+    }
 }
 
 fn renderPeaks(w: *std.Io.Writer, peaks: PeakSample) std.Io.Writer.Error!void {
@@ -1818,3 +1831,14 @@ test "the local-resource counter renders pool-wide, without a url label" {
     defer testing.allocator.free(absent);
     try testing.expect(!std.mem.containsAtLeast(u8, absent, 1, "nxdns_upstream_local_resource_total"));
 }
+
+test "live stream allocation failures render as a counter, absent without a hub" {
+    const text = try renderToString(testing.allocator, .{ .live_stream_allocation_failures_total = 2 });
+    defer testing.allocator.free(text);
+    try testing.expect(std.mem.containsAtLeast(u8, text, 1, "# TYPE nxdns_live_stream_allocation_failures_total counter\n"));
+    try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_live_stream_allocation_failures_total 2\n"));
+
+    const absent = try renderToString(testing.allocator, .{});
+    defer testing.allocator.free(absent);
+    try testing.expect(!std.mem.containsAtLeast(u8, absent, 1, "nxdns_live_stream_allocation_failures"));
+}

src/web/sse.zig

@@ -4,6 +4,8 @@
 //! before it hands the same entry to the logger: the event stream must never
 //! wait on a database. `publish` therefore copies and returns — it allocates
 //! nothing, touches no I/O, and holds one mutex across a scan of 32 slots.
+//! A subscriber's ring is allocated by `subscribe` and freed by `unsubscribe`,
+//! both off the query path.
 //!
 //! A subscriber that cannot keep up loses its stream rather than the queries:
 //! a full ring sets `overflowed`, the subscriber task sees the flag and ends
@@ -19,97 +21,129 @@ const logger = @import("../storage/logger.zig");
 pub const Entry = logger.Entry;
 
 /// Concurrent live streams. The per-IP cap (`web.sse_max_connections_per_ip`)
-/// keeps one client from taking all of them; `subscribe` returning null is the
-/// backstop and answers 503.
+/// keeps one client from taking all of them; `error.Full` from `subscribe` is
+/// the backstop and answers 503.
 pub const max_subscribers = 32;
 
 /// Entries one subscriber may fall behind by. At household query rates this is
 /// several seconds of slack on a stalled TCP connection.
 ///
-/// The ring is embedded in the slot, so this multiplies `@sizeOf(logger.Entry)`
-/// — about 1.8 KiB once milestone 28 widened it for provenance. One subscriber
-/// therefore costs roughly 115 KiB of ring and the whole hub roughly 3.6 MiB,
-/// allocated once for the life of the process. That is the reason to keep both
-/// this and `max_subscribers` small: they are paid whether or not anyone is
-/// watching. `logger.query_log_buffer_max` bounds the other consumer of the
-/// same width, the writer queue.
+/// One ring is `ring_capacity * @sizeOf(logger.Entry)`, about 115 KiB, held
+/// only while its subscriber is connected.
 pub const ring_capacity = 64;
 
 pub const SubscriberId = enum(u8) { _ };
 
+pub const SubscribeError = error{
+    /// Every slot holds a live stream.
+    Full,
+    /// No memory for the ring. Counted in `Hub.ring_allocation_failures`.
+    OutOfMemory,
+};
+
 /// What `wait` returns: an entry (or the overflow flag) is ready, the caller's
 /// timeout passed and it owes the client a heartbeat, or the hub is closing and
 /// the subscriber must end its response now.
 pub const Wake = enum { ready, timeout, closed };
 
 pub const Hub = struct {
+    gpa: std.mem.Allocator,
     /// Guards every field of every slot. `publish` runs on the DNS hot path,
     /// so the critical section is copies and flag writes only.
     mutex: std.Io.Mutex,
     /// Milestone-16 ruling 11. Set once, never cleared: a hub that is closing
     /// belongs to a server that is going away.
     closing: bool,
+    /// Subscriptions refused because the ring could not be allocated.
+    ring_allocation_failures: u64,
     slots: [max_subscribers]Slot,
 
     const Slot = struct {
-        active: bool,
+        /// The subscriber's ring; empty while the slot is free. Its storage
+        /// stays undefined: `len` says which entries are live.
+        ring: []Entry,
         /// Set by `publish` when the ring is full. Never cleared while the
         /// subscriber lives: the stream it belongs to is over.
         overflowed: bool,
         head: u32,
         len: u32,
         event: std.Io.Event,
-        ring: [ring_capacity]Entry,
+
+        fn active(slot: *const Slot) bool {
+            return slot.ring.len != 0;
+        }
     };
 
-    /// Initializes in place. The rings are close to a megabyte, which a
-    /// by-value `init` would copy through the caller's frame.
-    ///
-    /// The ring storage stays undefined: `len` says which slots hold entries.
-    pub fn init(self: *Hub) void {
-        self.mutex = .init;
-        self.closing = false;
-        for (&self.slots) |*slot| {
-            slot.active = false;
-            slot.overflowed = false;
-            slot.head = 0;
-            slot.len = 0;
-            slot.event = .unset;
-        }
+    pub fn init(gpa: std.mem.Allocator) Hub {
+        return .{
+            .gpa = gpa,
+            .mutex = .init,
+            .closing = false,
+            .ring_allocation_failures = 0,
+            .slots = @splat(.{ .ring = &.{}, .overflowed = false, .head = 0, .len = 0, .event = .unset }),
+        };
     }
 
-    /// Claims a slot, or null when all 32 are taken.
+    /// Every subscriber must have unsubscribed.
+    pub fn deinit(self: *Hub) void {
+        for (&self.slots) |*slot| std.debug.assert(!slot.active());
+        self.* = undefined;
+    }
+
+    /// Claims a slot with a fresh ring.
+    ///
+    /// The ring is allocated and, when every slot is taken, freed outside the
+    /// mutex, so `publish` never waits on the allocator.
     ///
     /// `lockUncancelable` throughout this file: `publish`'s caller is
     /// `Handler.handle`, which has no error union to carry `error.Canceled`
     /// out of (the same reasoning as `clients.Tracker.track`), and the rest of
     /// the surface shares the mutex with it.
-    pub fn subscribe(self: *Hub, io: std.Io) ?SubscriberId {
+    pub fn subscribe(self: *Hub, io: std.Io) SubscribeError!SubscriberId {
+        const ring = self.gpa.alloc(Entry, ring_capacity) catch {
+            self.mutex.lockUncancelable(io);
+            defer self.mutex.unlock(io);
+            self.ring_allocation_failures += 1;
+            return error.OutOfMemory;
+        };
+
+        return self.claim(io, ring) orelse {
+            self.gpa.free(ring);
+            return error.Full;
+        };
+    }
+
+    fn claim(self: *Hub, io: std.Io, ring: []Entry) ?SubscriberId {
         self.mutex.lockUncancelable(io);
         defer self.mutex.unlock(io);
 
         for (&self.slots, 0..) |*slot, index| {
-            if (slot.active) continue;
-            slot.active = true;
-            slot.overflowed = false;
-            slot.head = 0;
-            slot.len = 0;
-            slot.event = .unset;
+            if (slot.active()) continue;
+            slot.* = .{ .ring = ring, .overflowed = false, .head = 0, .len = 0, .event = .unset };
             return @enumFromInt(index);
         }
         return null;
     }
 
-    /// Releases the slot. The caller must not be waiting on it.
+    /// Releases the slot and frees its ring. The caller must not be waiting
+    /// on it.
     pub fn unsubscribe(self: *Hub, io: std.Io, id: SubscriberId) void {
         self.mutex.lockUncancelable(io);
-        defer self.mutex.unlock(io);
-
         const slot = self.slotOf(id);
-        slot.active = false;
+        const ring = slot.ring;
+        slot.ring = &.{};
         slot.overflowed = false;
         slot.len = 0;
         slot.head = 0;
+        self.mutex.unlock(io);
+
+        self.gpa.free(ring);
+    }
+
+    pub fn ringAllocationFailures(self: *Hub, io: std.Io) u64 {
+        self.mutex.lockUncancelable(io);
+        defer self.mutex.unlock(io);
+        return self.ring_allocation_failures;
     }
 
     /// Milestone-16 ruling 11. Ends every live stream.
@@ -124,7 +158,7 @@ pub const Hub = struct {
 
         self.closing = true;
         for (&self.slots) |*slot| {
-            if (slot.active) slot.event.set(io);
+            if (slot.active()) slot.event.set(io);
         }
     }
 
@@ -135,7 +169,7 @@ pub const Hub = struct {
         defer self.mutex.unlock(io);
 
         for (&self.slots) |*slot| {
-            if (!slot.active or slot.overflowed) continue;
+            if (!slot.active() or slot.overflowed) continue;
             if (slot.len == ring_capacity) {
                 slot.overflowed = true;
             } else {
@@ -214,7 +248,7 @@ pub const Hub = struct {
 
     fn slotOf(self: *Hub, id: SubscriberId) *Slot {
         const slot = &self.slots[@intFromEnum(id)];
-        std.debug.assert(slot.active);
+        std.debug.assert(slot.active());
         return slot;
     }
 };
@@ -234,21 +268,15 @@ fn sampleEntry(timestamp: i64, domain: []const u8) Entry {
     });
 }
 
-fn newHub(gpa: std.mem.Allocator) !*Hub {
-    const hub = try gpa.create(Hub);
-    hub.init();
-    return hub;
-}
-
 test "a subscriber reads what was published, oldest first" {
     var threaded: std.Io.Threaded = .init(testing.allocator, .{});
     defer threaded.deinit();
     const io = threaded.io();
 
-    const hub = try newHub(testing.allocator);
-    defer testing.allocator.destroy(hub);
+    var hub: Hub = .init(testing.allocator);
+    defer hub.deinit();
 
-    const id = hub.subscribe(io).?;
+    const id = try hub.subscribe(io);
     defer hub.unsubscribe(io, id);
 
     hub.publish(io, sampleEntry(1, "first.example"));
@@ -265,12 +293,12 @@ test "an entry published before a subscription is not delivered" {
     defer threaded.deinit();
     const io = threaded.io();
 
-    const hub = try newHub(testing.allocator);
-    defer testing.allocator.destroy(hub);
+    var hub: Hub = .init(testing.allocator);
+    defer hub.deinit();
 
     hub.publish(io, sampleEntry(1, "early.example"));
 
-    const id = hub.subscribe(io).?;
+    const id = try hub.subscribe(io);
     defer hub.unsubscribe(io, id);
     try testing.expect(hub.next(io, id) == null);
 }
@@ -280,11 +308,11 @@ test "every live subscriber receives its own copy" {
     defer threaded.deinit();
     const io = threaded.io();
 
-    const hub = try newHub(testing.allocator);
-    defer testing.allocator.destroy(hub);
+    var hub: Hub = .init(testing.allocator);
+    defer hub.deinit();
 
-    const first = hub.subscribe(io).?;
-    const second = hub.subscribe(io).?;
+    const first = try hub.subscribe(io);
+    const second = try hub.subscribe(io);
     defer hub.unsubscribe(io, first);
     defer hub.unsubscribe(io, second);
 
@@ -299,16 +327,17 @@ test "the hub hands out every slot and then refuses" {
     defer threaded.deinit();
     const io = threaded.io();
 
-    const hub = try newHub(testing.allocator);
-    defer testing.allocator.destroy(hub);
+    var hub: Hub = .init(testing.allocator);
+    defer hub.deinit();
 
     var ids: [max_subscribers]SubscriberId = undefined;
-    for (&ids) |*id| id.* = hub.subscribe(io).?;
-    try testing.expect(hub.subscribe(io) == null);
+    for (&ids) |*id| id.* = try hub.subscribe(io);
+    try testing.expectError(error.Full, hub.subscribe(io));
 
     hub.unsubscribe(io, ids[3]);
-    const reused = hub.subscribe(io).?;
+    const reused = try hub.subscribe(io);
     try testing.expectEqual(ids[3], reused);
+    for (ids) |id| hub.unsubscribe(io, id);
 }
 
 test "a full ring marks the subscriber overflowed and stops copying" {
@@ -316,10 +345,10 @@ test "a full ring marks the subscriber overflowed and stops copying" {
     defer threaded.deinit();
     const io = threaded.io();
 
-    const hub = try newHub(testing.allocator);
-    defer testing.allocator.destroy(hub);
+    var hub: Hub = .init(testing.allocator);
+    defer hub.deinit();
 
-    const id = hub.subscribe(io).?;
+    const id = try hub.subscribe(io);
     defer hub.unsubscribe(io, id);
 
     for (0..ring_capacity) |i| hub.publish(io, sampleEntry(@intCast(i), "fill.example"));
@@ -343,10 +372,10 @@ test "the ring wraps around its head" {
     defer threaded.deinit();
     const io = threaded.io();
 
-    const hub = try newHub(testing.allocator);
-    defer testing.allocator.destroy(hub);
+    var hub: Hub = .init(testing.allocator);
+    defer hub.deinit();
 
-    const id = hub.subscribe(io).?;
+    const id = try hub.subscribe(io);
     defer hub.unsubscribe(io, id);
 
     // Two and a half laps, consuming as we go: the head passes the end of the
@@ -368,10 +397,10 @@ test "wait returns as soon as an entry is waiting" {
     defer threaded.deinit();
     const io = threaded.io();
 
-    const hub = try newHub(testing.allocator);
-    defer testing.allocator.destroy(hub);
+    var hub: Hub = .init(testing.allocator);
+    defer hub.deinit();
 
-    const id = hub.subscribe(io).?;
+    const id = try hub.subscribe(io);
     defer hub.unsubscribe(io, id);
 
     const long: std.Io.Clock.Duration = .{ .raw = .fromSeconds(60), .clock = .awake };
@@ -384,10 +413,10 @@ test "wait times out on an idle subscriber so the heartbeat can go out" {
     defer threaded.deinit();
     const io = threaded.io();
 
-    const hub = try newHub(testing.allocator);
-    defer testing.allocator.destroy(hub);
+    var hub: Hub = .init(testing.allocator);
+    defer hub.deinit();
 
-    const id = hub.subscribe(io).?;
+    const id = try hub.subscribe(io);
     defer hub.unsubscribe(io, id);
 
     const brief: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(20), .clock = .awake };
@@ -400,14 +429,14 @@ test "a publish wakes a waiting subscriber" {
     defer threaded.deinit();
     const io = threaded.io();
 
-    const hub = try newHub(testing.allocator);
-    defer testing.allocator.destroy(hub);
+    var hub: Hub = .init(testing.allocator);
+    defer hub.deinit();
 
-    const id = hub.subscribe(io).?;
+    const id = try hub.subscribe(io);
     defer hub.unsubscribe(io, id);
 
     const long: std.Io.Clock.Duration = .{ .raw = .fromSeconds(60), .clock = .awake };
-    var future = try io.concurrent(Hub.wait, .{ hub, io, id, long });
+    var future = try io.concurrent(Hub.wait, .{ &hub, io, id, long });
 
     hub.publish(io, sampleEntry(5, "late.example"));
 
@@ -420,10 +449,10 @@ test "an overflow wakes a waiting subscriber" {
     defer threaded.deinit();
     const io = threaded.io();
 
-    const hub = try newHub(testing.allocator);
-    defer testing.allocator.destroy(hub);
+    var hub: Hub = .init(testing.allocator);
+    defer hub.deinit();
 
-    const id = hub.subscribe(io).?;
+    const id = try hub.subscribe(io);
     defer hub.unsubscribe(io, id);
 
     for (0..ring_capacity) |i| hub.publish(io, sampleEntry(@intCast(i), "fill.example"));
@@ -434,7 +463,7 @@ test "an overflow wakes a waiting subscriber" {
     for (0..ring_capacity) |i| hub.publish(io, sampleEntry(@intCast(i), "fill.example"));
 
     const long: std.Io.Clock.Duration = .{ .raw = .fromSeconds(60), .clock = .awake };
-    var future = try io.concurrent(Hub.wait, .{ hub, io, id, long });
+    var future = try io.concurrent(Hub.wait, .{ &hub, io, id, long });
     hub.publish(io, sampleEntry(999, "lost.example"));
 
     try testing.expectEqual(Wake.ready, try future.await(io));
@@ -446,16 +475,16 @@ test "close wakes a parked subscriber instead of leaving it on the heartbeat" {
     defer threaded.deinit();
     const io = threaded.io();
 
-    const hub = try newHub(testing.allocator);
-    defer testing.allocator.destroy(hub);
+    var hub: Hub = .init(testing.allocator);
+    defer hub.deinit();
 
-    const id = hub.subscribe(io).?;
+    const id = try hub.subscribe(io);
     defer hub.unsubscribe(io, id);
 
     // Far longer than the 15 s heartbeat the handler passes, so a pass here
     // cannot come from the timeout arm.
     const long: std.Io.Clock.Duration = .{ .raw = .fromSeconds(600), .clock = .awake };
-    var future = try io.concurrent(Hub.wait, .{ hub, io, id, long });
+    var future = try io.concurrent(Hub.wait, .{ &hub, io, id, long });
 
     hub.close(io);
 
@@ -467,10 +496,10 @@ test "wait on a closed hub returns at once, with no entry pending" {
     defer threaded.deinit();
     const io = threaded.io();
 
-    const hub = try newHub(testing.allocator);
-    defer testing.allocator.destroy(hub);
+    var hub: Hub = .init(testing.allocator);
+    defer hub.deinit();
 
-    const id = hub.subscribe(io).?;
+    const id = try hub.subscribe(io);
     defer hub.unsubscribe(io, id);
 
     hub.close(io);
@@ -484,13 +513,13 @@ test "publishing while subscribers come and go reaches only the live ones" {
     defer threaded.deinit();
     const io = threaded.io();
 
-    const hub = try newHub(testing.allocator);
-    defer testing.allocator.destroy(hub);
+    var hub: Hub = .init(testing.allocator);
+    defer hub.deinit();
 
-    const steady = hub.subscribe(io).?;
+    const steady = try hub.subscribe(io);
     defer hub.unsubscribe(io, steady);
 
-    var churner = try io.concurrent(churn, .{ hub, io });
+    var churner = try io.concurrent(churn, .{ &hub, io });
 
     var published: usize = 0;
     while (published < 500) : (published += 1) {
@@ -504,14 +533,64 @@ test "publishing while subscribers come and go reaches only the live ones" {
 
     // Every slot the churner used is free again.
     var ids: [max_subscribers - 1]SubscriberId = undefined;
-    for (&ids) |*id| id.* = hub.subscribe(io).?;
+    for (&ids) |*id| id.* = try hub.subscribe(io);
     for (ids) |id| hub.unsubscribe(io, id);
 }
 
 fn churn(hub: *Hub, io: std.Io) void {
     for (0..200) |i| {
-        const id = hub.subscribe(io) orelse continue;
+        const id = hub.subscribe(io) catch continue;
         if (i % 3 == 0) _ = hub.next(io, id);
         hub.unsubscribe(io, id);
     }
 }
+
+test "a subscription holds a ring only while it lives" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var counting: std.testing.FailingAllocator = .init(testing.allocator, .{});
+    var hub: Hub = .init(counting.allocator());
+    defer hub.deinit();
+    try testing.expectEqual(@as(usize, 0), counting.allocated_bytes);
+
+    const id = try hub.subscribe(io);
+    try testing.expectEqual(ring_capacity * @sizeOf(Entry), counting.allocated_bytes - counting.freed_bytes);
+    hub.unsubscribe(io, id);
+    try testing.expectEqual(counting.allocated_bytes, counting.freed_bytes);
+}
+
+test "a ring that cannot be allocated is counted and leaves the slot reusable" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var failing: std.testing.FailingAllocator = .init(testing.allocator, .{ .fail_index = 0 });
+    var hub: Hub = .init(failing.allocator());
+    defer hub.deinit();
+
+    try testing.expectError(error.OutOfMemory, hub.subscribe(io));
+    try testing.expectEqual(@as(u64, 1), hub.ringAllocationFailures(io));
+
+    failing.fail_index = std.math.maxInt(usize);
+    var ids: [max_subscribers]SubscriberId = undefined;
+    for (&ids) |*id| id.* = try hub.subscribe(io);
+    for (ids) |id| hub.unsubscribe(io, id);
+}
+
+test "a subscription refused for a full hub frees the ring it allocated" {
+    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    var hub: Hub = .init(testing.allocator);
+    defer hub.deinit();
+
+    var ids: [max_subscribers]SubscriberId = undefined;
+    for (&ids) |*id| id.* = try hub.subscribe(io);
+    defer for (ids) |id| hub.unsubscribe(io, id);
+    // `testing.allocator` reports the leak if the refused ring is kept.
+    try testing.expectError(error.Full, hub.subscribe(io));
+    try testing.expectEqual(@as(u64, 0), hub.ringAllocationFailures(io));
+}

src/web/web_integration_test.zig

@@ -477,10 +477,9 @@ const Env = struct {
         });
         errdefer self.limiter.deinit();
 
-        // ~900 KiB: never a stack local (W5 rule).
         self.hub = try gpa.create(sse.Hub);
         errdefer gpa.destroy(self.hub);
-        self.hub.init();
+        self.hub.* = .init(gpa);
 
         self.pauser = .{};
         self.tables = .empty;
@@ -682,6 +681,7 @@ const Env = struct {
         self.state.proxies.deinit(gpa);
         self.state.overview_cache.deinit(gpa);
         self.tables.deinit(gpa);
+        self.hub.deinit();
         gpa.destroy(self.hub);
         self.limiter.deinit();
         self.mgr.deinit(ioh);