milestone 16: behavioral fixes for silent failures, locks, counters and the query log

src/web/sse.zig

@@ -29,14 +29,18 @@ 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 };
+/// 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 {
     /// 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,
     slots: [max_subscribers]Slot,
 
     const Slot = struct {
@@ -56,6 +60,7 @@ pub const Hub = struct {
     /// 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;
@@ -99,6 +104,22 @@ pub const Hub = struct {
         slot.head = 0;
     }
 
+    /// Milestone-16 ruling 11. Ends every live stream.
+    ///
+    /// Without this, a graceful drain waits out one heartbeat interval per idle
+    /// subscriber: shutting the sockets down does not reach a task parked inside
+    /// `wait`, which is waiting on an event, not on the peer. Called before the
+    /// web listener begins its shutdown.
+    pub fn close(self: *Hub, io: std.Io) void {
+        self.mutex.lockUncancelable(io);
+        defer self.mutex.unlock(io);
+
+        self.closing = true;
+        for (&self.slots) |*slot| {
+            if (slot.active) slot.event.set(io);
+        }
+    }
+
     /// 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 {
@@ -140,8 +161,9 @@ pub const Hub = struct {
         return self.slotOf(id).overflowed;
     }
 
-    /// Blocks until something is ready for this subscriber or `timeout`
-    /// passes; `.timeout` is the heartbeat's cue.
+    /// Blocks until something is ready for this subscriber, `timeout` passes,
+    /// or the hub closes; `.timeout` is the heartbeat's cue and `.closed` ends
+    /// the stream.
     ///
     /// 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
@@ -157,6 +179,10 @@ pub const Hub = struct {
         timeout: std.Io.Clock.Duration,
     ) std.Io.Cancelable!Wake {
         self.mutex.lockUncancelable(io);
+        if (self.closing) {
+            self.mutex.unlock(io);
+            return .closed;
+        }
         const slot = self.slotOf(id);
         if (slot.len > 0 or slot.overflowed) {
             self.mutex.unlock(io);
@@ -169,7 +195,13 @@ pub const Hub = struct {
             error.Timeout => return .timeout,
             error.Canceled => |e| return e,
         };
-        return .ready;
+
+        // `close` wakes the same event a publish does, so the flag is what tells
+        // the two apart. Reading it here rather than on the next call through
+        // keeps the drain from writing one more frame into a dying connection.
+        self.mutex.lockUncancelable(io);
+        defer self.mutex.unlock(io);
+        return if (self.closing) .closed else .ready;
     }
 
     fn slotOf(self: *Hub, id: SubscriberId) *Slot {
@@ -401,6 +433,44 @@ test "an overflow wakes a waiting subscriber" {
     try testing.expect(hub.overflowed(io, id));
 }
 
+test "close wakes a parked subscriber instead of leaving it on the heartbeat" {
+    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);
+
+    // 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 });
+
+    hub.close(io);
+
+    try testing.expectEqual(Wake.closed, try future.await(io));
+}
+
+test "wait on a closed hub returns at once, with no entry pending" {
+    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.close(io);
+
+    const long: std.Io.Clock.Duration = .{ .raw = .fromSeconds(600), .clock = .awake };
+    try testing.expectEqual(Wake.closed, try hub.wait(io, id, long));
+}
+
 test "publishing while subscribers come and go reaches only the live ones" {
     var threaded: std.Io.Threaded = .init(testing.allocator, .{});
     defer threaded.deinit();