diff --git a/docs/grafana-dashboard.md b/docs/grafana-dashboard.md index cde5311..55c48f0 100644 --- a/docs/grafana-dashboard.md +++ b/docs/grafana-dashboard.md @@ -59,6 +59,11 @@ The following dashboard families already have non-placeholder producers: - `jetstream_backfill_failed_repo_retry_succeeded_total` - `jetstream_backfill_failed_repo_retry_failed_total` - `jetstream_subscribe_subscribers` +- `jetstream_subscribe_events_sent_total` +- `jetstream_subscribe_bytes_sent_total` +- `jetstream_subscribe_bytes_encoded_total` +- `jetstream_subscribe_events_filtered_total` +- `jetstream_subscribe_events_oversize_total` - `jetstream_import_jobs_total` - `jetstream_import_job_duration_seconds` - `jetstream_import_phase` @@ -72,10 +77,6 @@ The following dashboard families already have non-placeholder producers: - `jetstream_import_segments_patched_total` - `jetstream_import_bytes_rewritten_total` -The subscriber total is currently labeled `compression="unknown"`; the -canonical per-compression split is still open and the contract must not treat -that aggregate as completion of the dashboard's compression panels. - The checksum-pinned deployment contract remains red until every other queried family is backed by its real lifecycle boundary. The current binary is not a dashboard-complete release. diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index 1661a56..45d096e 100644 --- a/docs/semantic-parity.md +++ b/docs/semantic-parity.md @@ -20,7 +20,7 @@ this document or `bootstrap-semantic-parity.md` is open. | Sync 1.1 resync ordering | Upstream serializes per DID, drops stale async repairs, and emits a sync tombstone plus authoritative replacements | **closed at the implementation boundary.** The live coordinator has 32 real fetch workers, a 64-job queue, a five-minute fetch budget plus the 1,000 B/s-for-30-seconds slow-transfer guard, a bounded 16,384-entry 5/minute-per-DID limiter, 2,048 pending commits with drop-oldest overflow, fetch outside the DID lane, authenticated apply with older/equal-contradictory head rejection, ordered pending replay, and trigger-ticket-gated outbox delivery. Commit divergence schedules async repair; `#sync` divergence attempts inline repair and queues one retry only for transient failure, preserving the original relay cursor only on inline success. The writer emits `sync` then bounded 1,024-row `create_resync` batches and stages the fetched/pending chain checkpoints under the archive lock before any matching fsync. Loopback-only tests exercise a real HTTP getRepo mmap, real DID resolution/signature, buffering while the response is held open, slow-transfer cancellation, durable RocksDB promotion/cursor coupling, v2 tail publication, and sealed JSS row order. | | Timestamp import | Canonical dashboard and API include timestamp import lifecycle/rewrite metrics and behavior | **Closed at the implementation boundary.** The format preserves distinct `witnessed_at` and sentinel-zero `indexed_at` columns. Subscriber v1/v2 encoding applies upstream's `indexed_at != 0 ? indexed_at : witnessed_at` display rule while ranges and timestamp cursors remain anchored to immutable witness envelopes. The strict seekable RFC 4180 parser matches upstream's header, row rejection, byte-offset, CID, RFC3339, 64 KiB, and bounded-sampling contracts in a forward-only allocation-free pass. A separate Bloom-filtered RocksDB rule store bulk-loads sorted external SST chunks with CSV-order last-write-wins, reconstructs its resident collection gate on open, and implements specific-CID precedence with all-version fallback. Its central archive-lock hook stamps live, repair, bootstrap, and merge materializations before sequence assignment, and feeds the mutated rows to immediate delivery so hot and cold encodings agree. The sealed-JSS patch primitive preserves block topology/envelopes, clean frames, and the opaque bloom/collection footer while atomically changing only `indexed_at`; it is exercised against an upstream-produced segment. A concrete sealed-segment bloom catalog, bounded DID/FD bucketer, fsynced packed offsets, positioned revalidation, and collision-safe per-segment patch plan implement Phase B/C with specific-CID precedence and idempotent disk-to-disk coverage. A dedicated archive rewrite mutex serializes the real delete compactor with timestamp patching while leaving live append available. The unified metadata DB synchronously persists the current-job pointer, complete job record, and per-segment done set. The real runner activates rules before force-seal, durably crosses the bucketed handoff, checkpoints only after patch fsync/rename/dir-sync, retains partial cancellation progress, and resumes after full archive/rule/metadata reopen. The background manager canonicalizes symlinks, confines regular files, enforces one durable nonterminal job, and adopts it on startup. Bearer-gated import/status XRPC matches the pinned lexicons, fixed-width digest comparison, error names/statuses, steady-state admission, and current/by-id status behavior; a real loopback HTTP run exercised it against the local simulator. Every canonical import metric is produced at its real parse, route, patch, or durable-terminal boundary; cancellation remains a resumable pause and is not counted as terminal failure. | | Status/diagnostics | Repo attempts/error class/final PDS and per-host aggregates survive restart and are operator-queryable | durable repo fields implemented; **operator repo/host views open** | -| Prometheus/Grafana | Every canonical dashboard query maps to a real, same-semantics producer; runtime-specific panels use honest Zig/process metrics | exact checksum-pinned upstream dashboard is vendored with an offline fail-closed scrape contract; canonical build/live-firehose/archive-append/drop/verifier/backfill/subscriber/timestamp-import families have real producers. Live frame inspection distinguishes gaps, bounded-label server errors, forward-compatible unknowns, and malformed CBOR at the raw transport boundary; a Zat reconnect callback measures attempts rather than connection failures. **Remaining metric families and Go-row adaptation open.** | +| Prometheus/Grafana | Every canonical dashboard query maps to a real, same-semantics producer; runtime-specific panels use honest Zig/process metrics | exact checksum-pinned upstream dashboard is vendored with an offline fail-closed scrape contract; canonical build/live-firehose/archive-append/drop/verifier/backfill/timestamp-import producers are present. Subscriber gauges and successful-delivery bytes/events are split by actual `none`/`zstd` compression, and filter/oversize drops are counted at their decision boundaries. Live frame inspection distinguishes gaps, bounded-label server errors, forward-compatible unknowns, and malformed CBOR at the raw transport boundary; a Zat reconnect callback measures attempts rather than connection failures. **Remaining subscriber/storage/API/orchestrator metric families and Go-row adaptation open.** | | Oracle | Event-log equivalence, final-state convergence, crash/power-loss, verifier repair, hostile input, and anti-vacuity receipts | **open — current crash matrix is necessary but substantially smaller than upstream's oracle** | ## Experiment admission diff --git a/src/internal/metrics.zig b/src/internal/metrics.zig index 36d945d..7a3afc6 100644 --- a/src/internal/metrics.zig +++ b/src/internal/metrics.zig @@ -16,6 +16,7 @@ const timestamp_parse = @import("timestamp/parse.zig"); const Io = std.Io; pub const Counter = std.atomic.Value(u64); +pub const Compression = enum { none, deflate, zstd }; pub const Stats = struct { start_time_s: i64 = 0, @@ -81,6 +82,12 @@ pub const Stats = struct { import_segments_examined_total: Counter = .init(0), import_segments_patched_total: Counter = .init(0), import_bytes_rewritten_total: Counter = .init(0), + subscribe_active: std.enums.EnumArray(Compression, Counter) = .initFill(.init(0)), + subscribe_events_sent: std.enums.EnumArray(Compression, Counter) = .initFill(.init(0)), + subscribe_bytes_sent: std.enums.EnumArray(Compression, Counter) = .initFill(.init(0)), + subscribe_bytes_encoded: std.enums.EnumArray(Compression, Counter) = .initFill(.init(0)), + subscribe_events_filtered_total: Counter = .init(0), + subscribe_events_oversize_total: Counter = .init(0), pub fn addDrops(self: *Stats, before: *const convert.Drops, after: *const convert.Drops) void { inline for (comptime std.enums.values(convert.DropReason)) |reason| { @@ -216,8 +223,6 @@ pub fn format( \\jetstream_backfill_completed_total {d} \\# TYPE jetstream_backfill_failed_total counter \\jetstream_backfill_failed_total {d} - \\# TYPE jetstream_subscribe_subscribers gauge - \\jetstream_subscribe_subscribers{{compression="unknown"}} {d} \\ , .{ build_options.git_sha, @@ -242,7 +247,35 @@ pub fn format( stats.ingest_active_segment_bytes.load(.monotonic), stats.backfill_completed_total.load(.monotonic), stats.backfill_failed_total.load(.monotonic), - subscribers_active, + }) catch {}; + + w.print("# TYPE jetstream_subscribe_subscribers gauge\n", .{}) catch {}; + w.print("# TYPE jetstream_subscribe_events_sent_total counter\n", .{}) catch {}; + w.print("# TYPE jetstream_subscribe_bytes_sent_total counter\n", .{}) catch {}; + w.print("# TYPE jetstream_subscribe_bytes_encoded_total counter\n", .{}) catch {}; + inline for (comptime std.enums.values(Compression)) |scheme| { + w.print("jetstream_subscribe_subscribers{{compression=\"{s}\"}} {d}\n", .{ + @tagName(scheme), stats.subscribe_active.getPtrConst(scheme).load(.monotonic), + }) catch {}; + w.print("jetstream_subscribe_events_sent_total{{compression=\"{s}\"}} {d}\n", .{ + @tagName(scheme), stats.subscribe_events_sent.getPtrConst(scheme).load(.monotonic), + }) catch {}; + w.print("jetstream_subscribe_bytes_sent_total{{compression=\"{s}\"}} {d}\n", .{ + @tagName(scheme), stats.subscribe_bytes_sent.getPtrConst(scheme).load(.monotonic), + }) catch {}; + w.print("jetstream_subscribe_bytes_encoded_total{{compression=\"{s}\"}} {d}\n", .{ + @tagName(scheme), stats.subscribe_bytes_encoded.getPtrConst(scheme).load(.monotonic), + }) catch {}; + } + w.print( + \\# TYPE jetstream_subscribe_events_filtered_total counter + \\jetstream_subscribe_events_filtered_total {d} + \\# TYPE jetstream_subscribe_events_oversize_total counter + \\jetstream_subscribe_events_oversize_total {d} + \\ + , .{ + stats.subscribe_events_filtered_total.load(.monotonic), + stats.subscribe_events_oversize_total.load(.monotonic), }) catch {}; w.print( @@ -418,6 +451,12 @@ test "format renders counters" { _ = stats.import_jobs_ok_total.fetchAdd(1, .monotonic); stats.import_phase.store(2, .monotonic); _ = stats.import_rows_mutated_total.fetchAdd(12, .monotonic); + stats.subscribe_active.getPtr(.zstd).store(2, .monotonic); + _ = stats.subscribe_events_sent.getPtr(.zstd).fetchAdd(3, .monotonic); + _ = stats.subscribe_bytes_sent.getPtr(.zstd).fetchAdd(400, .monotonic); + _ = stats.subscribe_bytes_encoded.getPtr(.zstd).fetchAdd(800, .monotonic); + _ = stats.subscribe_events_filtered_total.fetchAdd(5, .monotonic); + _ = stats.subscribe_events_oversize_total.fetchAdd(6, .monotonic); var buf: [16 * 1024]u8 = undefined; const out = format(&buf, &stats, 2, .{ .entries = 5, .bytes = 100, .base = 1, .tip = 6 }, 60); @@ -448,6 +487,12 @@ test "format renders counters" { try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_import_jobs_total{result=\"ok\"} 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_import_phase 2") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_import_rows_mutated_total 12") != null); + try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_subscribe_subscribers{compression=\"zstd\"} 2") != null); + try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_subscribe_events_sent_total{compression=\"zstd\"} 3") != null); + try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_subscribe_bytes_sent_total{compression=\"zstd\"} 400") != null); + try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_subscribe_bytes_encoded_total{compression=\"zstd\"} 800") != null); + try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_subscribe_events_filtered_total 5") != null); + try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_subscribe_events_oversize_total 6") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_livestream_replayed_account_events_dropped_total 3") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_livestream_replayed_identity_events_dropped_total 4") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_failed_repo_retry_attempts_total 5") != null); diff --git a/src/internal/server.zig b/src/internal/server.zig index d4021f7..2061807 100644 --- a/src/internal/server.zig +++ b/src/internal/server.zig @@ -56,6 +56,10 @@ const Subscriber = struct { max_msg_size: std.atomic.Value(u64), idx: u64, alive: std.atomic.Value(bool) = .init(true), + /// Serializes socket writes with close signaling. A normal peer close must + /// wait for an in-flight write and prevent the next one before closing the + /// descriptor; otherwise Threaded treats the resulting BADF as a bug. + write_mutex: Io.Mutex = .init, /// inverse of alive, in the shape tail.next wants for its stop check stopped: std.atomic.Value(bool) = .init(false), /// set when the cursor fell below the hot tail's floor: replay from @@ -83,19 +87,35 @@ const Subscriber = struct { // oversize events are silently skipped, cursor advances (v1 quirk: // cap compares the uncompressed JSON length) const cap = self.max_msg_size.load(.monotonic); - if (cap != 0 and json.len > cap) continue; + if (cap != 0 and json.len > cap) { + _ = self.hub.stats.subscribe_events_oversize_total.fetchAdd(1, .monotonic); + continue; + } self.write(json) catch return; } } fn write(self: *Subscriber, json: []const u8) !void { + self.write_mutex.lockUncancelable(self.hub.io); + defer self.write_mutex.unlock(self.hub.io); + if (!self.alive.load(.acquire)) return error.Closed; + const scheme = self.compression(); if (self.compressor) |*comp| { const frame = try comp.compress(self.hub.allocator, json); defer self.hub.allocator.free(frame); try self.conn.writeBin(frame); + _ = self.hub.stats.subscribe_events_sent.getPtr(scheme).fetchAdd(1, .monotonic); + _ = self.hub.stats.subscribe_bytes_sent.getPtr(scheme).fetchAdd(@intCast(frame.len), .monotonic); } else { try self.conn.writeText(json); + _ = self.hub.stats.subscribe_events_sent.getPtr(scheme).fetchAdd(1, .monotonic); + _ = self.hub.stats.subscribe_bytes_sent.getPtr(scheme).fetchAdd(@intCast(json.len), .monotonic); } + _ = self.hub.stats.subscribe_bytes_encoded.getPtr(scheme).fetchAdd(@intCast(json.len), .monotonic); + } + + fn compression(self: *const Subscriber) metrics.Compression { + return if (self.compressor != null) .zstd else .none; } fn coldPhase(self: *Subscriber, from_us: i64) void { @@ -141,14 +161,19 @@ const Subscriber = struct { fn coldWrite(self: *Subscriber, json: []const u8) anyerror!void { if (!self.alive.load(.acquire)) return error.Closed; const cap = self.max_msg_size.load(.monotonic); - if (cap != 0 and json.len > cap) return; + if (cap != 0 and json.len > cap) { + _ = self.hub.stats.subscribe_events_oversize_total.fetchAdd(1, .monotonic); + return; + } try self.write(json); } fn wantsPred(self: *Subscriber, kind: wire.Kind, did: []const u8, collection: []const u8) bool { self.opts_mutex.lockUncancelable(self.hub.io); defer self.opts_mutex.unlock(self.hub.io); - return self.filter.wants(kind, did, collection); + const wanted = self.filter.wants(kind, did, collection); + if (!wanted) _ = self.hub.stats.subscribe_events_filtered_total.fetchAdd(1, .monotonic); + return wanted; } /// atomically replace the whole filter (options_update contract) @@ -288,9 +313,12 @@ pub const Handler = struct { }; self.pending_filter = null; // ownership moved to the subscriber self.subscriber = sub; + const scheme = sub.compression(); _ = hub.active.fetchAdd(1, .monotonic); + _ = hub.stats.subscribe_active.getPtr(scheme).fetchAdd(1, .monotonic); sub.thread = std.Thread.spawn(.{ .stack_size = subscriber_stack_size }, Subscriber.run, .{sub}) catch |err| { _ = hub.active.fetchSub(1, .monotonic); + _ = hub.stats.subscribe_active.getPtr(scheme).fetchSub(1, .monotonic); sub.filter.deinit(); hub.allocator.destroy(sub); self.subscriber = null; @@ -379,6 +407,7 @@ pub const Handler = struct { if (sub.compressor) |*comp| comp.deinit(); sub.filter.deinit(); _ = self.hub.active.fetchSub(1, .monotonic); + _ = self.hub.stats.subscribe_active.getPtr(sub.compression()).fetchSub(1, .monotonic); self.hub.allocator.destroy(sub); self.subscriber = null; } @@ -386,8 +415,10 @@ pub const Handler = struct { } fn signalStop(self: *Handler, sub: *Subscriber) void { + sub.write_mutex.lockUncancelable(self.hub.io); sub.alive.store(false, .release); sub.stopped.store(true, .release); + sub.write_mutex.unlock(self.hub.io); self.hub.tail.wake(); } diff --git a/tests/e2e.py b/tests/e2e.py index dd73129..181bb51 100644 --- a/tests/e2e.py +++ b/tests/e2e.py @@ -108,6 +108,17 @@ def check_metrics(): assert valid and float(valid[0].split(" ")[1]) > 0, "verification never verified anything" invalid = [l for l in body.splitlines() if l.startswith('stream_verify_total{result="invalid"}')] assert float(invalid[0].split(" ")[1]) == 0, "honest traffic produced invalid signatures" + def labeled(name, label): + prefix = f'{name}{{{label}}} ' + line = next((line for line in body.splitlines() if line.startswith(prefix)), None) + assert line is not None, f"metric {name}{{{label}}} missing" + return float(line.split(" ")[1]) + assert labeled("jetstream_subscribe_events_sent_total", 'compression="none"') > 0 + assert labeled("jetstream_subscribe_events_sent_total", 'compression="zstd"') > 0 + assert labeled("jetstream_subscribe_bytes_encoded_total", 'compression="zstd"') > 0 + assert labeled("jetstream_subscribe_bytes_sent_total", 'compression="zstd"') > 0 + assert val("jetstream_subscribe_events_filtered_total") > 0 + assert val("jetstream_subscribe_events_oversize_total") > 0 health = urllib.request.urlopen("http://localhost:6008/healthz").read().decode() assert health.strip() == "ok" print("metrics: PASS")