diff --git a/docs/upstream-subscribe-protocol.md b/docs/upstream-subscribe-protocol.md index eef0d07..1cb9027 100644 --- a/docs/upstream-subscribe-protocol.md +++ b/docs/upstream-subscribe-protocol.md @@ -168,3 +168,15 @@ text frame: `{"type":"options_update","payload":{"wantedCollections":[],"wantedD 8. requireHello exact-literal "true" only 9. below-floor cursor clamped, never rejected 10. read limit 10_000_000 decimal + +## stream-only observability additions + +- `stream_subscribe_worker_exits_total{reason}` — why a subscriber's delivery + worker ended (tail_closed, read_batch_error, write_error, write_closed, + ping_write_error, too_slow, cold_failed, cold_stall). Upstream has no + equivalent: its handler exits are as silent as ours were. Added 2026-08-13 + after eval connections died during compaction passes with no log line and + no counter anywhere. Every reason renders even at zero, so an absent series + means "not scraped", never "never happened". The wedged-writer kill itself + (`frame_write_timeout_ms = 5s`) IS upstream behavior (`frameWriteTimeout`, + internal/subscribe/handler.go) — only the accounting is ours. diff --git a/src/internal/runtime/metrics.zig b/src/internal/runtime/metrics.zig index b6106b5..9b6468d 100644 --- a/src/internal/runtime/metrics.zig +++ b/src/internal/runtime/metrics.zig @@ -30,6 +30,28 @@ pub const HttpHandler = enum { pub const HttpMethod = enum { get, head, post, put, patch, delete, options, connect, trace, other }; pub const HttpCode = enum { c200, c201, c206, c304, c400, c401, c404, c405, c409, c412, c416, c500, c501, c503, other }; pub const GetBlockResult = enum { ok, not_found, bad_request, error_result }; +/// why a subscriber's delivery worker ended. Found 2026-08-13: connections +/// were dying during compaction passes with no log line and no counter — +/// every worker exit must name itself so the next incident self-identifies. +pub const SubscriberExit = enum { + /// tail shut down (process stopping) — the only benign reason here + tail_closed, + /// hot-tail readBatch returned an error (allocation or internal) + read_batch_error, + /// socket write failed while the connection was still considered live + write_error, + /// write found the connection already closed (handler/ping tore it down) + write_closed, + /// the keepalive ping could not be written — dead socket found idle + ping_write_error, + /// slow-consumer policy disconnected the subscriber (also counted in + /// subscribe_adversarial_drops_total) + too_slow, + /// cold replay failed (archive read error) + cold_failed, + /// rotation-seam invariant violated (also subscribe_cold_stalls_total) + cold_stall, +}; pub const StoreOp = enum { get, set, delete, batch_commit }; pub const StoreStatus = enum { ok, notfound, error_result }; pub const OrchestratorPhase = enum { bootstrap, merging, steady_state }; @@ -67,6 +89,7 @@ pub const Stats = struct { http_request_duration_sum_us: [http_series_count]Counter = @splat(.init(0)), http_request_duration_buckets: [http_series_count][11]Counter = @splat(@splat(.init(0))), getblock_requests: std.enums.EnumArray(GetBlockResult, Counter) = .initFill(.init(0)), + subscribe_worker_exits: std.enums.EnumArray(SubscriberExit, Counter) = .initFill(.init(0)), getblock_served_bytes_total: Counter = .init(0), getblock_duration_count: Counter = .init(0), getblock_duration_sum_us: Counter = .init(0), @@ -700,6 +723,17 @@ pub fn format( .{ getBlockResultLabel(result), count }, ) catch {}; } + + // Always emit every reason, zeros included: this family exists to make + // subscriber deaths legible, and an absent series reads as "not scraped" + // rather than "never happened". + w.print("# TYPE stream_subscribe_worker_exits_total counter\n", .{}) catch {}; + inline for (comptime std.enums.values(SubscriberExit)) |reason| { + w.print( + "stream_subscribe_worker_exits_total{{reason=\"{s}\"}} {d}\n", + .{ @tagName(reason), stats.subscribe_worker_exits.getPtrConst(reason).load(.monotonic) }, + ) catch {}; + } w.print( \\# TYPE jetstream_getblock_served_bytes_total counter \\jetstream_getblock_served_bytes_total {d} @@ -1465,6 +1499,41 @@ test "getBlock metrics preserve upstream result and served-byte semantics" { try std.testing.expectEqual(7, stats.getblock_duration_count.load(.monotonic)); } +// Every worker-exit reason renders, zeros included: this family exists so a +// subscriber death during an incident names itself; an absent series must +// mean "not scraped", never "never happened". (2026-08-13: eval connections +// died during compaction passes with no log and no counter.) +test "subscriber worker exits render every reason, zeros included" { + var stats: Stats = .{}; + _ = stats.subscribe_worker_exits.getPtr(.write_error).fetchAdd(2, .monotonic); + var buf: [64 * 1024]u8 = undefined; + const out = format(&buf, &stats, 0, .{ + .entries = 0, + .bytes = 0, + .pinned_bytes = 0, + .pinned_overrun_bytes = 0, + .base = 0, + .tip = 0, + }, 0, .{ + .cpu_us = 0, + .resident_memory_bytes = 0, + .virtual_memory_bytes = 0, + .threads = 0, + .open_fds = 0, + .max_fds = 0, + .minor_page_faults = 0, + .major_page_faults = 0, + .voluntary_context_switches = 0, + .involuntary_context_switches = 0, + }, null, null); + try std.testing.expect(std.mem.indexOf(u8, out, "stream_subscribe_worker_exits_total{reason=\"write_error\"} 2") != null); + inline for (comptime std.enums.values(SubscriberExit)) |reason| { + var line_buf: [128]u8 = undefined; + const line = try std.fmt.bufPrint(&line_buf, "stream_subscribe_worker_exits_total{{reason=\"{s}\"}}", .{@tagName(reason)}); + try std.testing.expect(std.mem.indexOf(u8, out, line) != null); + } +} + test "format renders counters" { var stats: Stats = .{}; _ = stats.events_total.fetchAdd(7, .monotonic); diff --git a/src/internal/serve/server.zig b/src/internal/serve/server.zig index 5f7d2fd..f4bcdf9 100644 --- a/src/internal/serve/server.zig +++ b/src/internal/serve/server.zig @@ -238,11 +238,11 @@ const Subscriber = struct { if (self.cold_stall_passes >= 2) { log.err("subscriber cursor {d} unserveable: cold exhausted below committed {d} (rotation seam invariant violated)", .{ self.cursor_seq, committed }); _ = self.hub.stats.subscribe_cold_stalls_total.fetchAdd(1, .monotonic); - return; + return self.noteExit(.cold_stall); } continue; }, - .failed => return, + .failed => return self.noteExit(.cold_failed), } } @@ -254,9 +254,12 @@ const Subscriber = struct { &self.stopped, self.v2, self.hub.subscribe_read_batch, - ) catch return; + ) catch |err| { + log.warn("subscriber read batch failed: {s}", .{@errorName(err)}); + return self.noteExit(.read_batch_error); + }; switch (result) { - .closed => return, + .closed => return self.noteExit(.tail_closed), .fell_behind => { // evicted from under us: the next pass re-decides from // cursor_seq (seq-unknown subscribers clamp to the floor, @@ -274,15 +277,24 @@ const Subscriber = struct { _ = self.hub.stats.subscribe_events_oversize_total.fetchAdd(1, .monotonic); continue; } - if ((self.write(entry.json) catch return) == .encode_error) + const wrote = self.write(entry.json) catch |err| { + if (err == error.Closed) return self.noteExit(.write_closed); + log.warn("subscriber write failed: {s}", .{@errorName(err)}); + return self.noteExit(.write_error); + }; + if (wrote == .encode_error) _ = self.hub.stats.subscribe_encode_errors_total.fetchAdd(1, .monotonic); } }, } - if (self.tooSlow()) return; + if (self.tooSlow()) return self.noteExit(.too_slow); } } + fn noteExit(self: *Subscriber, reason: metrics.SubscriberExit) void { + _ = self.hub.stats.subscribe_worker_exits.getPtr(reason).fetchAdd(1, .monotonic); + } + fn pingLoop(self: *Subscriber) void { self.start_gate.waitUncancelable(self.hub.io); if (!self.alive.load(.acquire)) return; @@ -299,6 +311,7 @@ const Subscriber = struct { } var payload: [0]u8 = .{}; self.conn.writePing(&payload) catch { + self.noteExit(.ping_write_error); const interrupt = self.alive.swap(false, .acq_rel); self.stopped.store(true, .release); self.write_mutex.unlock(self.hub.io);