From d951d03b7d99dad3785ece489e52d58ab7f2c7a6 Mon Sep 17 00:00:00 2001 From: m5r Date: Wed, 21 Oct 2026 22:38:53 +0200 Subject: [PATCH] querylog: the writer queue defaults to 1024 entries, and a live stream allocates its ring on subscribe instead of at boot --- CHANGELOG.md | 2 + PLAN.md | 2 +- docs/reference/configuration.md | 2 +- src/app.zig | 14 +- src/config/model.zig | 6 +- src/server/handler.zig | 8 +- src/server/query_sink.zig | 24 +-- src/storage/logger.zig | 4 +- src/web/handlers/live.zig | 6 +- src/web/metrics.zig | 24 +++ src/web/sse.zig | 267 ++++++++++++++++++++----------- src/web/web_integration_test.zig | 4 +- 12 files changed, 234 insertions(+), 129 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3ba8c58..9516833 100644 --- a/CHANGELOG.md +++ b/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 , and `INSTALL.md` in the tarball links there. diff --git a/PLAN.md b/PLAN.md index 7f03d12..2e84cb8 100644 --- a/PLAN.md +++ b/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. diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index b77e9aa..1d7e5a2 100644 --- a/docs/reference/configuration.md +++ b/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. | diff --git a/src/app.zig b/src/app.zig index a74fabd..27df25b 100644 --- a/src/app.zig +++ b/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) diff --git a/src/config/model.zig b/src/config/model.zig index 0cbd819..3199dbe 100644 --- a/src/config/model.zig +++ b/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. diff --git a/src/server/handler.zig b/src/server/handler.zig index a9ca641..99d9c99 100644 --- a/src/server/handler.zig +++ b/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; diff --git a/src/server/query_sink.zig b/src/server/query_sink.zig index 6b0a03b..8759971 100644 --- a/src/server/query_sink.zig +++ b/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")); diff --git a/src/storage/logger.zig b/src/storage/logger.zig index 2af7e47..b4a89ec 100644 --- a/src/storage/logger.zig +++ b/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); diff --git a/src/web/handlers/live.zig b/src/web/handlers/live.zig index db71ce2..8cac3d5 100644 --- a/src/web/handlers/live.zig +++ b/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(&.{}, .{ diff --git a/src/web/metrics.zig b/src/web/metrics.zig index f082e8f..90ae93f 100644 --- a/src/web/metrics.zig +++ b/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")); +} diff --git a/src/web/sse.zig b/src/web/sse.zig index 729570b..5903718 100644 --- a/src/web/sse.zig +++ b/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)); +} diff --git a/src/web/web_integration_test.zig b/src/web/web_integration_test.zig index d742ac1..9f0d3fa 100644 --- a/src/web/web_integration_test.zig +++ b/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);