milestone 8: web server, rest api, sse, auth, metrics and static assets

2026-08-02 00:54:13 +02:00
parent a8092bb1b9
commit 5253c47303
59 changed files with 19640 additions and 150 deletions
@@ -0,0 +1,439 @@
+//! Live query fanout for `GET /api/queries/live` (PLAN §11.4:455).
+//!
+//! The DNS query path publishes through `QuerySink`, which calls `publish`
+//! 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 that cannot keep up loses its stream rather than the queries:
+//! a full ring sets `overflowed`, the subscriber task sees the flag and ends
+//! the response, and the browser's `EventSource` reconnects on its own.
+//!
+//! `logger.Entry` carries its own bytes, so a ring slot is a plain copy with
+//! nothing borrowed from the query that produced it.
+
+const std = @import("std");
+
+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.
+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.
+pub const ring_capacity = 64;
+
+pub const SubscriberId = enum(u8) { _ };
+
+/// What `wait` returns: an entry (or the overflow flag) is ready, or the
+/// caller's timeout passed and it owes the client a heartbeat.
+pub const Wake = enum { ready, timeout };
+
+pub const Hub = struct {
+    /// 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,
+    slots: [max_subscribers]Slot,
+
+    const Slot = struct {
+        active: bool,
+        /// 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,
+    };
+
+    /// 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;
+        for (&self.slots) |*slot| {
+            slot.active = false;
+            slot.overflowed = false;
+            slot.head = 0;
+            slot.len = 0;
+            slot.event = .unset;
+        }
+    }
+
+    /// Claims a slot, or null when all 32 are taken.
+    ///
+    /// `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 {
+        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;
+            return @enumFromInt(index);
+        }
+        return null;
+    }
+
+    /// Releases the slot. 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;
+        slot.overflowed = false;
+        slot.len = 0;
+        slot.head = 0;
+    }
+
+    /// Copies `entry` into every live ring and wakes its subscriber. Called
+    /// once per logged query.
+    pub fn publish(self: *Hub, io: std.Io, entry: Entry) void {
+        self.mutex.lockUncancelable(io);
+        defer self.mutex.unlock(io);
+
+        for (&self.slots) |*slot| {
+            if (!slot.active or slot.overflowed) continue;
+            if (slot.len == ring_capacity) {
+                slot.overflowed = true;
+            } else {
+                slot.ring[(slot.head + slot.len) % ring_capacity] = entry;
+                slot.len += 1;
+            }
+            slot.event.set(io);
+        }
+    }
+
+    /// The oldest entry this subscriber has not seen, or null when its ring is
+    /// empty. Check `overflowed` first: entries that predate the overflow are
+    /// still readable, but the stream must end once they run out.
+    pub fn next(self: *Hub, io: std.Io, id: SubscriberId) ?Entry {
+        self.mutex.lockUncancelable(io);
+        defer self.mutex.unlock(io);
+
+        const slot = self.slotOf(id);
+        if (slot.len == 0) return null;
+        const entry = slot.ring[slot.head];
+        slot.head = (slot.head + 1) % ring_capacity;
+        slot.len -= 1;
+        return entry;
+    }
+
+    /// True once this subscriber missed an entry. The subscriber task ends the
+    /// response when it sees this.
+    pub fn overflowed(self: *Hub, io: std.Io, id: SubscriberId) bool {
+        self.mutex.lockUncancelable(io);
+        defer self.mutex.unlock(io);
+        return self.slotOf(id).overflowed;
+    }
+
+    /// Blocks until something is ready for this subscriber or `timeout`
+    /// passes; `.timeout` is the heartbeat's cue.
+    ///
+    /// The event is reset under the mutex and only while the ring is empty, so
+    /// a `publish` that lands between the check and the wait sets the event
+    /// again and the wait returns at once. Only the owning subscriber task
+    /// calls this, which is what `Event.reset` requires (`Io.zig:1866`).
+    ///
+    /// A spurious futex wakeup reports `.timeout` (`Io.zig:1824`): the caller
+    /// sends one heartbeat it did not strictly owe.
+    pub fn wait(
+        self: *Hub,
+        io: std.Io,
+        id: SubscriberId,
+        timeout: std.Io.Clock.Duration,
+    ) std.Io.Cancelable!Wake {
+        self.mutex.lockUncancelable(io);
+        const slot = self.slotOf(id);
+        if (slot.len > 0 or slot.overflowed) {
+            self.mutex.unlock(io);
+            return .ready;
+        }
+        slot.event.reset();
+        self.mutex.unlock(io);
+
+        slot.event.waitTimeout(io, .{ .duration = timeout }) catch |err| switch (err) {
+            error.Timeout => return .timeout,
+            error.Canceled => |e| return e,
+        };
+        return .ready;
+    }
+
+    fn slotOf(self: *Hub, id: SubscriberId) *Slot {
+        const slot = &self.slots[@intFromEnum(id)];
+        std.debug.assert(slot.active);
+        return slot;
+    }
+};
+
+// ---------------------------------------------------------------------------
+// tests
+// ---------------------------------------------------------------------------
+
+const testing = std.testing;
+
+fn sampleEntry(timestamp: i64, domain: []const u8) Entry {
+    return .init(.{
+        .timestamp = timestamp,
+        .domain = domain,
+        .client_ip = "192.0.2.10",
+        .qtype = 1,
+    });
+}
+
+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);
+
+    const id = hub.subscribe(io).?;
+    defer hub.unsubscribe(io, id);
+
+    hub.publish(io, sampleEntry(1, "first.example"));
+    hub.publish(io, sampleEntry(2, "second.example"));
+
+    try testing.expectEqualStrings("first.example", hub.next(io, id).?.domain());
+    try testing.expectEqualStrings("second.example", hub.next(io, id).?.domain());
+    try testing.expect(hub.next(io, id) == null);
+    try testing.expect(!hub.overflowed(io, id));
+}
+
+test "an entry published before a subscription is not delivered" {
+    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);
+
+    hub.publish(io, sampleEntry(1, "early.example"));
+
+    const id = hub.subscribe(io).?;
+    defer hub.unsubscribe(io, id);
+    try testing.expect(hub.next(io, id) == null);
+}
+
+test "every live subscriber receives its own copy" {
+    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);
+
+    const first = hub.subscribe(io).?;
+    const second = hub.subscribe(io).?;
+    defer hub.unsubscribe(io, first);
+    defer hub.unsubscribe(io, second);
+
+    hub.publish(io, sampleEntry(7, "shared.example"));
+
+    try testing.expectEqualStrings("shared.example", hub.next(io, first).?.domain());
+    try testing.expectEqualStrings("shared.example", hub.next(io, second).?.domain());
+}
+
+test "the hub hands out every slot and then refuses" {
+    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 ids: [max_subscribers]SubscriberId = undefined;
+    for (&ids) |*id| id.* = hub.subscribe(io).?;
+    try testing.expect(hub.subscribe(io) == null);
+
+    hub.unsubscribe(io, ids[3]);
+    const reused = hub.subscribe(io).?;
+    try testing.expectEqual(ids[3], reused);
+}
+
+test "a full ring marks the subscriber overflowed and stops copying" {
+    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);
+
+    const id = hub.subscribe(io).?;
+    defer hub.unsubscribe(io, id);
+
+    for (0..ring_capacity) |i| hub.publish(io, sampleEntry(@intCast(i), "fill.example"));
+    try testing.expect(!hub.overflowed(io, id));
+
+    hub.publish(io, sampleEntry(999, "lost.example"));
+    try testing.expect(hub.overflowed(io, id));
+
+    // What the ring already held is still readable; the entry that overflowed
+    // it is not, and the flag stays set.
+    var drained: usize = 0;
+    while (hub.next(io, id)) |entry| : (drained += 1) {
+        try testing.expectEqualStrings("fill.example", entry.domain());
+    }
+    try testing.expectEqual(@as(usize, ring_capacity), drained);
+    try testing.expect(hub.overflowed(io, id));
+}
+
+test "the ring wraps around its head" {
+    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);
+
+    const id = hub.subscribe(io).?;
+    defer hub.unsubscribe(io, id);
+
+    // Two and a half laps, consuming as we go: the head passes the end of the
+    // storage twice and no entry is lost.
+    for (0..ring_capacity * 2 + ring_capacity / 2) |i| {
+        var buf: [32]u8 = undefined;
+        const domain = try std.fmt.bufPrint(&buf, "d{d}.example", .{i});
+        hub.publish(io, sampleEntry(@intCast(i), domain));
+
+        const got = hub.next(io, id).?;
+        try testing.expectEqualStrings(domain, got.domain());
+        try testing.expectEqual(@as(i64, @intCast(i)), got.timestamp);
+    }
+    try testing.expect(!hub.overflowed(io, id));
+}
+
+test "wait returns as soon as an entry is waiting" {
+    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);
+
+    const id = hub.subscribe(io).?;
+    defer hub.unsubscribe(io, id);
+
+    const long: std.Io.Clock.Duration = .{ .raw = .fromSeconds(60), .clock = .awake };
+    hub.publish(io, sampleEntry(1, "ready.example"));
+    try testing.expectEqual(Wake.ready, try hub.wait(io, id, long));
+}
+
+test "wait times out on an idle subscriber so the heartbeat can go out" {
+    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);
+
+    const id = hub.subscribe(io).?;
+    defer hub.unsubscribe(io, id);
+
+    const brief: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(20), .clock = .awake };
+    try testing.expectEqual(Wake.timeout, try hub.wait(io, id, brief));
+    try testing.expect(hub.next(io, id) == null);
+}
+
+test "a publish wakes a waiting subscriber" {
+    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);
+
+    const id = 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 });
+
+    hub.publish(io, sampleEntry(5, "late.example"));
+
+    try testing.expectEqual(Wake.ready, try future.await(io));
+    try testing.expectEqualStrings("late.example", hub.next(io, id).?.domain());
+}
+
+test "an overflow wakes a waiting subscriber" {
+    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);
+
+    const id = hub.subscribe(io).?;
+    defer hub.unsubscribe(io, id);
+
+    for (0..ring_capacity) |i| hub.publish(io, sampleEntry(@intCast(i), "fill.example"));
+    while (hub.next(io, id)) |_| {}
+
+    // The ring is empty again but its head sits mid-storage; refill it and
+    // overflow, so the wake comes from the flag rather than from an entry.
+    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 });
+    hub.publish(io, sampleEntry(999, "lost.example"));
+
+    try testing.expectEqual(Wake.ready, try future.await(io));
+    try testing.expect(hub.overflowed(io, id));
+}
+
+test "publishing while subscribers come and go reaches only the live ones" {
+    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);
+
+    const steady = hub.subscribe(io).?;
+    defer hub.unsubscribe(io, steady);
+
+    var churner = try io.concurrent(churn, .{ hub, io });
+
+    var published: usize = 0;
+    while (published < 500) : (published += 1) {
+        hub.publish(io, sampleEntry(@intCast(published), "churn.example"));
+        // Keep the steady subscriber under its ring cap: this test is about
+        // the churn, not about overflow.
+        while (hub.next(io, steady)) |_| {}
+    }
+    churner.await(io);
+    try testing.expect(!hub.overflowed(io, steady));
+
+    // 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| hub.unsubscribe(io, id);
+}
+
+fn churn(hub: *Hub, io: std.Io) void {
+    for (0..200) |i| {
+        const id = hub.subscribe(io) orelse continue;
+        if (i % 3 == 0) _ = hub.next(io, id);
+        hub.unsubscribe(io, id);
+    }
+}