diff --git a/build.zig.zon b/build.zig.zon index 53464c2..89052dd 100644 --- a/build.zig.zon +++ b/build.zig.zon @@ -5,8 +5,8 @@ .minimum_zig_version = "0.16.0", .dependencies = .{ .zat = .{ - .url = "git+https://tangled.org/zat.dev/zat?ref=v0.3.16#42093a6f8ad71acdcecbb8dd80a87b21d963e6be", - .hash = "zat-0.3.16-5PuC7nkkCwD898jK6JR2d1BMth0ss8_ueAB8gd6iqqj0", + .url = "git+https://tangled.org/zat.dev/zat#6fb7ac137610db71ce2b2636955fab710e34a587", + .hash = "zat-0.3.16-5PuC7o4pCwBjGxkaZdIt-qcGW-u3sM4mgsV1xRTrSgxI", }, .websocket = .{ .url = "https://tangled.org/zzstoatzz.io/websocket.zig/archive/v0.1.9.tar.gz", diff --git a/docs/grafana-dashboard.md b/docs/grafana-dashboard.md index e9b6b68..cde5311 100644 --- a/docs/grafana-dashboard.md +++ b/docs/grafana-dashboard.md @@ -37,6 +37,12 @@ The following dashboard families already have non-placeholder producers: - `jetstream_build_info` - `jetstream_livestream_events_received_total` - `jetstream_livestream_decode_errors_total` +- `jetstream_livestream_reconnects_total` +- `jetstream_livestream_sequence_gaps_total` +- `jetstream_livestream_sequence_gap_missed_seqs_total` +- `jetstream_livestream_unknown_events_total` +- `jetstream_livestream_stream_error_frames_total` +- `jetstream_livestream_last_seen_upstream_event_timestamp_seconds` - `jetstream_livestream_replayed_account_events_dropped_total` - `jetstream_livestream_replayed_identity_events_dropped_total` - `jetstream_ingest_events_appended_total` (counted in the shared archive diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index 4c2fea9..1661a56 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-receive/archive-append/drop/verifier/backfill/subscriber/timestamp-import families have real producers; **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/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.** | | 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/ingest.zig b/src/internal/ingest.zig index 013c090..e6044f5 100644 --- a/src/internal/ingest.zig +++ b/src/internal/ingest.zig @@ -24,6 +24,45 @@ const Io = std.Io; const Allocator = std.mem.Allocator; const log = std.log.scoped(.stream); +const StreamErrorCode = enum { future_cursor, consumer_too_slow, other, missing }; +const InspectedFrame = union(enum) { + message: struct { seq: ?i64 }, + unknown: struct { seq: ?i64 }, + stream_error: StreamErrorCode, + malformed, +}; + +fn inspectFrame(allocator: Allocator, data: []const u8) InspectedFrame { + var arena = std.heap.ArenaAllocator.init(allocator); + defer arena.deinit(); + const header = zat.cbor.decode(arena.allocator(), data) catch return .malformed; + const op = header.value.getInt("op") orelse return .malformed; + const payload = zat.cbor.decodeAll(arena.allocator(), data[header.consumed..]) catch return .malformed; + if (op == -1) { + const code = payload.getString("error") orelse return .{ .stream_error = .missing }; + if (std.mem.eql(u8, code, "FutureCursor")) return .{ .stream_error = .future_cursor }; + if (std.mem.eql(u8, code, "ConsumerTooSlow")) return .{ .stream_error = .consumer_too_slow }; + return .{ .stream_error = .other }; + } + const seq = payload.getInt("seq"); + if (op != 1) return .{ .unknown = .{ .seq = seq } }; + const tag = header.value.getString("t") orelse return .malformed; + const known = std.mem.eql(u8, tag, "#commit") or + std.mem.eql(u8, tag, "#sync") or + std.mem.eql(u8, tag, "#identity") or + std.mem.eql(u8, tag, "#account") or + std.mem.eql(u8, tag, "#info"); + return if (known) .{ .message = .{ .seq = seq } } else .{ .unknown = .{ .seq = seq } }; +} + +fn testFrame(allocator: Allocator, header: zat.cbor.Value, payload: zat.cbor.Value) ![]u8 { + const header_bytes = try zat.cbor.encodeAlloc(allocator, header); + defer allocator.free(header_bytes); + const payload_bytes = try zat.cbor.encodeAlloc(allocator, payload); + defer allocator.free(payload_bytes); + return std.mem.concat(allocator, u8, &.{ header_bytes, payload_bytes }); +} + pub const Consumer = struct { allocator: Allocator, io: Io, @@ -49,6 +88,7 @@ pub const Consumer = struct { write_archive_sequence_metadata: bool = true, fatal_mu: Io.Mutex = .init, fatal_error: ?anyerror = null, + last_wire_seq: ?i64 = null, /// blocks forever; zat handles reconnect with backoff and host rotation. pub fn run(self: *Consumer) !void { @@ -67,6 +107,19 @@ pub const Consumer = struct { /// emitVerified in completion order across DIDs and FIFO order per DID. pub fn onRawFrame(self: *Consumer, data: []const u8) void { const pipe = self.pipeline orelse return; + switch (inspectFrame(self.allocator, data)) { + .message => |message| self.observeWireSeq(message.seq), + .unknown => |unknown| { + self.observeWireSeq(unknown.seq); + if (self.stats) |stats| _ = stats.livestream_unknown_events_total.fetchAdd(1, .monotonic); + return; + }, + .stream_error => |code| { + self.observeStreamError(code); + return; + }, + .malformed => {}, + } pipe.submit(data) catch |err| { // Cancellation can interrupt a producer waiting for one of the // 32 active-DID worker slots during intentional shutdown. Every @@ -77,6 +130,39 @@ pub const Consumer = struct { }; } + pub fn onError(_: *Consumer, err: anyerror) void { + log.err("firehose error: {s}, reconnecting", .{@errorName(err)}); + } + + pub fn onReconnect(self: *Consumer) void { + if (self.stats) |stats| _ = stats.livestream_reconnects_total.fetchAdd(1, .monotonic); + } + + fn observeWireSeq(self: *Consumer, maybe_seq: ?i64) void { + const seq = maybe_seq orelse return; + const previous = self.last_wire_seq orelse self.cursor; + if (previous) |last| { + const delta = std.math.sub(i64, seq, last) catch std.math.maxInt(i64); + if (delta > 1) if (self.stats) |stats| { + _ = stats.livestream_sequence_gaps_total.fetchAdd(1, .monotonic); + _ = stats.livestream_sequence_gap_missed_seqs_total.fetchAdd(@intCast(delta - 1), .monotonic); + }; + if (seq <= last) return; + } + self.last_wire_seq = seq; + } + + fn observeStreamError(self: *Consumer, code: StreamErrorCode) void { + const stats = self.stats orelse return; + const counter = switch (code) { + .future_cursor => &stats.livestream_stream_error_future_cursor_total, + .consumer_too_slow => &stats.livestream_stream_error_consumer_too_slow_total, + .other => &stats.livestream_stream_error_other_total, + .missing => &stats.livestream_stream_error_missing_total, + }; + _ = counter.fetchAdd(1, .monotonic); + } + pub fn shouldStop(self: *Consumer) bool { if (self.getFatal() != null) return true; return if (self.pipeline) |pipeline| pipeline.fatalError() != null else false; @@ -216,7 +302,13 @@ pub const Consumer = struct { const time_us: i64 = Io.Timestamp.now(self.io, .real).toMicroseconds(); if (!replayed_after_repair) { - if (self.stats) |st| _ = st.events_total.fetchAdd(1, .monotonic); + if (self.stats) |st| { + _ = st.events_total.fetchAdd(1, .monotonic); + st.livestream_last_seen_upstream_event_timestamp_seconds.store( + @divFloor(time_us, std.time.us_per_s), + .monotonic, + ); + } } switch (event) { @@ -732,3 +824,51 @@ test "verifier silent drop waits for a later emitted cursor watermark" { try archive.flushBlock(); try testing.expectEqual(@as(?i64, 12), cursor.load(io)); } + +test "raw firehose inspection classifies unknown and bounded error frames" { + const testing = std.testing; + const unknown = try testFrame(testing.allocator, .{ .map = &.{ + .{ .key = "op", .value = .{ .unsigned = 1 } }, + .{ .key = "t", .value = .{ .text = "#futureThing" } }, + } }, .{ .map = &.{.{ .key = "seq", .value = .{ .unsigned = 2 } }} }); + defer testing.allocator.free(unknown); + try testing.expectEqual(@as(?i64, 2), inspectFrame(testing.allocator, unknown).unknown.seq); + + const future = try testFrame(testing.allocator, .{ .map = &.{ + .{ .key = "op", .value = .{ .negative = -1 } }, + } }, .{ .map = &.{.{ .key = "error", .value = .{ .text = "FutureCursor" } }} }); + defer testing.allocator.free(future); + try testing.expectEqual(StreamErrorCode.future_cursor, inspectFrame(testing.allocator, future).stream_error); + + const hostile = try testFrame(testing.allocator, .{ .map = &.{ + .{ .key = "op", .value = .{ .negative = -1 } }, + } }, .{ .map = &.{.{ .key = "error", .value = .{ .text = "unbounded-attacker-value" } }} }); + defer testing.allocator.free(hostile); + try testing.expectEqual(StreamErrorCode.other, inspectFrame(testing.allocator, hostile).stream_error); +} + +test "wire sequence accounting ignores replay and sizes forward gaps" { + const testing = std.testing; + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + var tail = tail_mod.Tail.init(testing.allocator, threaded.io(), 1024); + defer tail.deinit(); + var stats: metrics.Stats = .{}; + var consumer: Consumer = .{ + .allocator = testing.allocator, + .io = threaded.io(), + .tail = &tail, + .upstream = "ws://offline.invalid", + .cursor = 10, + .stats = &stats, + }; + consumer.observeWireSeq(11); + consumer.observeWireSeq(9); + consumer.observeWireSeq(15); + consumer.onReconnect(); + + try testing.expectEqual(@as(u64, 1), stats.livestream_sequence_gaps_total.load(.monotonic)); + try testing.expectEqual(@as(u64, 3), stats.livestream_sequence_gap_missed_seqs_total.load(.monotonic)); + try testing.expectEqual(@as(u64, 1), stats.livestream_reconnects_total.load(.monotonic)); + try testing.expectEqual(@as(?i64, 15), consumer.last_wire_seq); +} diff --git a/src/internal/metrics.zig b/src/internal/metrics.zig index 56ce3cc..36d945d 100644 --- a/src/internal/metrics.zig +++ b/src/internal/metrics.zig @@ -45,6 +45,15 @@ pub const Stats = struct { verify_queue_dropped_total: Counter = .init(0), replayed_account_events_dropped_total: Counter = .init(0), replayed_identity_events_dropped_total: Counter = .init(0), + livestream_reconnects_total: Counter = .init(0), + livestream_sequence_gaps_total: Counter = .init(0), + livestream_sequence_gap_missed_seqs_total: Counter = .init(0), + livestream_unknown_events_total: Counter = .init(0), + livestream_stream_error_future_cursor_total: Counter = .init(0), + livestream_stream_error_consumer_too_slow_total: Counter = .init(0), + livestream_stream_error_other_total: Counter = .init(0), + livestream_stream_error_missing_total: Counter = .init(0), + livestream_last_seen_upstream_event_timestamp_seconds: std.atomic.Value(i64) = .init(0), /// highest physically-compacted seq (lag = stream_upstream_seq - this) compaction_watermark: Counter = .init(0), retry_resynced_total: Counter = .init(0), @@ -174,6 +183,21 @@ pub fn format( \\jetstream_livestream_events_received_total {d} \\# TYPE jetstream_livestream_decode_errors_total counter \\jetstream_livestream_decode_errors_total {d} + \\# TYPE jetstream_livestream_reconnects_total counter + \\jetstream_livestream_reconnects_total {d} + \\# TYPE jetstream_livestream_sequence_gaps_total counter + \\jetstream_livestream_sequence_gaps_total {d} + \\# TYPE jetstream_livestream_sequence_gap_missed_seqs_total counter + \\jetstream_livestream_sequence_gap_missed_seqs_total {d} + \\# TYPE jetstream_livestream_unknown_events_total counter + \\jetstream_livestream_unknown_events_total {d} + \\# TYPE jetstream_livestream_stream_error_frames_total counter + \\jetstream_livestream_stream_error_frames_total{{code="FutureCursor"}} {d} + \\jetstream_livestream_stream_error_frames_total{{code="ConsumerTooSlow"}} {d} + \\jetstream_livestream_stream_error_frames_total{{code="other"}} {d} + \\jetstream_livestream_stream_error_frames_total{{code="missing"}} {d} + \\# TYPE jetstream_livestream_last_seen_upstream_event_timestamp_seconds gauge + \\jetstream_livestream_last_seen_upstream_event_timestamp_seconds {d} \\# TYPE jetstream_livestream_replayed_account_events_dropped_total counter \\jetstream_livestream_replayed_account_events_dropped_total {d} \\# TYPE jetstream_livestream_replayed_identity_events_dropped_total counter @@ -200,6 +224,15 @@ pub fn format( build_options.optimize, stats.events_total.load(.monotonic), stats.livestream_decode_errors_total.load(.monotonic), + stats.livestream_reconnects_total.load(.monotonic), + stats.livestream_sequence_gaps_total.load(.monotonic), + stats.livestream_sequence_gap_missed_seqs_total.load(.monotonic), + stats.livestream_unknown_events_total.load(.monotonic), + stats.livestream_stream_error_future_cursor_total.load(.monotonic), + stats.livestream_stream_error_consumer_too_slow_total.load(.monotonic), + stats.livestream_stream_error_other_total.load(.monotonic), + stats.livestream_stream_error_missing_total.load(.monotonic), + stats.livestream_last_seen_upstream_event_timestamp_seconds.load(.monotonic), stats.replayed_account_events_dropped_total.load(.monotonic), stats.replayed_identity_events_dropped_total.load(.monotonic), stats.ingest_events_appended_total.load(.monotonic), @@ -368,6 +401,12 @@ test "format renders counters" { _ = stats.ingest_append_errors_total.fetchAdd(8, .monotonic); stats.ingest_active_segment_bytes.store(9000, .monotonic); _ = stats.livestream_decode_errors_total.fetchAdd(2, .monotonic); + _ = stats.livestream_reconnects_total.fetchAdd(3, .monotonic); + _ = stats.livestream_sequence_gaps_total.fetchAdd(4, .monotonic); + _ = stats.livestream_sequence_gap_missed_seqs_total.fetchAdd(5, .monotonic); + _ = stats.livestream_unknown_events_total.fetchAdd(6, .monotonic); + _ = stats.livestream_stream_error_future_cursor_total.fetchAdd(7, .monotonic); + stats.livestream_last_seen_upstream_event_timestamp_seconds.store(1234, .monotonic); _ = stats.verify_queue_dropped_total.fetchAdd(9, .monotonic); _ = stats.replayed_account_events_dropped_total.fetchAdd(3, .monotonic); _ = stats.replayed_identity_events_dropped_total.fetchAdd(4, .monotonic); @@ -399,6 +438,12 @@ test "format renders counters" { try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_ingest_append_errors_total 8") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_ingest_active_segment_bytes 9000") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_livestream_decode_errors_total 2") != null); + try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_livestream_reconnects_total 3") != null); + try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_livestream_sequence_gaps_total 4") != null); + try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_livestream_sequence_gap_missed_seqs_total 5") != null); + try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_livestream_unknown_events_total 6") != null); + try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_livestream_stream_error_frames_total{code=\"FutureCursor\"} 7") != null); + try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_livestream_last_seen_upstream_event_timestamp_seconds 1234") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_livestream_verify_queue_dropped_events_total 9") != null); 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);