touch the systemPublic surfaces
diff --git a/src/internal/server.zig b/src/internal/server.zig
index f10a09d..b2b3084 100644
--- a/src/internal/server.zig
+++ b/src/internal/server.zig
@@ -26,9 +26,12 @@ const log = std.log.scoped(.stream);
/// 8 MB, matching the Io.Threaded stacks (docs/lessons-from-zlay.md #1)
const subscriber_stack_size: usize = 8 * 1024 * 1024;
+const ping_stack_size: usize = 256 * 1024;
const slow_window_us: i64 = 60 * std.time.us_per_s;
const slow_lag_threshold: u64 = 100_000;
const slow_min_rate: u64 = 5;
+const ping_interval: Io.Clock.Duration = .{ .raw = Io.Duration.fromSeconds(30), .clock = .awake };
+const frame_write_timeout_ms: u32 = 5_000;
const SlowDetector = struct {
streak_start_us: i64 = 0,
@@ -113,14 +116,17 @@ const Subscriber = struct {
/// /subscribe-v2: superset wire (seq, record_cbor, #sync), strict cursors
v2: bool = false,
thread: ?std.Thread = null,
+ ping_thread: ?std.Thread = null,
+ ping_stop: Io.Event = .unset,
slow: SlowDetector = .{},
/// the pull loop. cleanup happens in Handler.close, which joins this
/// thread — the subscriber never frees itself, so the handler can always
/// safely signal it.
fn run(self: *Subscriber) void {
- if (self.cold_from_seq) |from_seq| self.coldPhaseSeq(from_seq);
- if (self.cold_from_us) |from_us| self.coldPhase(from_us);
+ defer self.workerDone();
+ if (self.cold_from_seq) |from_seq| if (!self.coldPhaseSeq(from_seq)) return;
+ if (self.cold_from_us) |from_us| if (!self.coldPhase(from_us)) return;
while (self.alive.load(.acquire)) {
const json = self.hub.tail.next(self.hub.allocator, &self.idx, &self.stopped, self.v2, self, hotWantsPred) catch return orelse return;
defer self.hub.allocator.free(json);
@@ -143,6 +149,45 @@ const Subscriber = struct {
}
}
+ fn pingLoop(self: *Subscriber) void {
+ while (self.alive.load(.acquire)) {
+ self.ping_stop.waitTimeout(self.hub.io, .{ .duration = ping_interval }) catch |err| switch (err) {
+ error.Timeout => {},
+ error.Canceled => return,
+ };
+ if (!self.alive.load(.acquire)) return;
+ self.write_mutex.lockUncancelable(self.hub.io);
+ if (!self.alive.load(.acquire)) {
+ self.write_mutex.unlock(self.hub.io);
+ return;
+ }
+ var payload: [0]u8 = .{};
+ self.conn.writePing(&payload) catch {
+ const interrupt = self.alive.swap(false, .acq_rel);
+ self.stopped.store(true, .release);
+ self.write_mutex.unlock(self.hub.io);
+ self.ping_stop.set(self.hub.io);
+ self.hub.tail.wake();
+ if (interrupt) self.conn.interruptRead();
+ return;
+ };
+ self.write_mutex.unlock(self.hub.io);
+ }
+ }
+
+ /// A subscriber-worker failure must end the connection, not leave a
+ /// ping-only zombie. Half-shutdown wakes the server reader; that reader
+ /// remains the sole owner of final descriptor cleanup.
+ fn workerDone(self: *Subscriber) void {
+ self.write_mutex.lockUncancelable(self.hub.io);
+ const interrupt = self.alive.swap(false, .acq_rel);
+ self.stopped.store(true, .release);
+ self.write_mutex.unlock(self.hub.io);
+ self.ping_stop.set(self.hub.io);
+ self.hub.tail.wake();
+ if (interrupt) self.conn.interruptRead();
+ }
+
fn write(self: *Subscriber, json: []const u8) !cold.WriteResult {
self.write_mutex.lockUncancelable(self.hub.io);
defer self.write_mutex.unlock(self.hub.io);
@@ -170,9 +215,9 @@ const Subscriber = struct {
return if (self.conn.compression != null) .deflate else .none;
}
- fn coldPhase(self: *Subscriber, from_us: i64) void {
+ fn coldPhase(self: *Subscriber, from_us: i64) bool {
const hub = self.hub;
- const archive = hub.archive orelse return;
+ const archive = hub.archive orelse return true;
const until_us = hub.tail.oldestTime() orelse std.math.maxInt(i64);
_ = hub.stats.subscribe_cold_reads_total.fetchAdd(1, .monotonic);
const n = cold.replay(
@@ -188,14 +233,15 @@ const Subscriber = struct {
until_us,
) catch |err| {
log.warn("cold replay failed: {s}", .{@errorName(err)});
- return;
+ return false;
};
log.debug("cold replay served {d} frames", .{n});
+ return true;
}
- fn coldPhaseSeq(self: *Subscriber, from_seq: u64) void {
+ fn coldPhaseSeq(self: *Subscriber, from_seq: u64) bool {
const hub = self.hub;
- const archive = hub.archive orelse return;
+ const archive = hub.archive orelse return true;
_ = hub.stats.subscribe_cold_reads_total.fetchAdd(1, .monotonic);
const n = cold.replaySeq(
hub.allocator,
@@ -210,9 +256,10 @@ const Subscriber = struct {
from_seq,
) catch |err| {
log.warn("seq cold replay failed: {s}", .{@errorName(err)});
- return;
+ return false;
};
log.debug("seq cold replay served {d} frames", .{n});
+ return true;
}
fn coldWrite(self: *Subscriber, json: []const u8) anyerror!cold.WriteResult {
@@ -287,6 +334,7 @@ pub const Handler = struct {
awaiting_hello: bool,
compress: bool,
v2: bool,
+ clean_disconnect_counted: bool = false,
pub fn init(handshake: *const websocket.Handshake, conn: *websocket.Conn, hub: *Hub) !Handler {
const url = handshake.url;
@@ -340,6 +388,7 @@ pub const Handler = struct {
if (compress) if (handshake.headers.get("sec-websocket-extensions")) |extensions| {
if (offersPerMessageDeflate(extensions)) return error.InvalidRequest;
};
+ try conn.writeTimeout(frame_write_timeout_ms);
return .{
.hub = hub,
@@ -399,6 +448,16 @@ pub const Handler = struct {
self.subscriber = null;
return err;
};
+ sub.ping_thread = std.Thread.spawn(.{ .stack_size = ping_stack_size }, Subscriber.pingLoop, .{sub}) catch |err| {
+ self.signalStop(sub);
+ sub.thread.?.join();
+ sub.filter.deinit();
+ _ = hub.active.fetchSub(1, .monotonic);
+ _ = hub.stats.subscribe_active.getPtr(scheme).fetchSub(1, .monotonic);
+ hub.allocator.destroy(sub);
+ self.subscriber = null;
+ return err;
+ };
}
/// client → server messages: SubscriberSourcedMessage envelope.
@@ -496,16 +555,24 @@ pub const Handler = struct {
pub fn clientClose(self: *Handler, _: []const u8) !void {
if (self.subscriber) |sub| self.signalStop(sub);
- _ = self.hub.stats.subscribe_clean_disconnects_total.fetchAdd(1, .monotonic);
+ self.noteCleanDisconnect();
self.conn.close(.{}) catch {};
}
+ /// Called only by the websocket server's intentional shutdown path.
+ pub fn serverClose(self: *Handler) void {
+ if (self.subscriber) |sub| self.signalStop(sub);
+ self.conn.writeClose(.{ .code = 1001, .reason = "server shutting down" }) catch {};
+ self.noteCleanDisconnect();
+ }
+
/// always called by the server when the connection ends — the single
/// place the subscriber thread is joined and freed.
pub fn close(self: *Handler) void {
if (self.subscriber) |sub| {
self.signalStop(sub);
if (sub.thread) |t| t.join();
+ if (sub.ping_thread) |t| t.join();
if (sub.compressor) |*comp| comp.deinit();
sub.filter.deinit();
_ = self.hub.active.fetchSub(1, .monotonic);
@@ -521,9 +588,16 @@ pub const Handler = struct {
sub.alive.store(false, .release);
sub.stopped.store(true, .release);
sub.write_mutex.unlock(self.hub.io);
+ sub.ping_stop.set(self.hub.io);
self.hub.tail.wake();
}
+ fn noteCleanDisconnect(self: *Handler) void {
+ if (self.clean_disconnect_counted) return;
+ self.clean_disconnect_counted = true;
+ _ = self.hub.stats.subscribe_clean_disconnects_total.fetchAdd(1, .monotonic);
+ }
+
/// plain-HTTP requests on the websocket port: /metrics and /healthz
pub fn httpFallback(
conn: *websocket.Conn,