//! shared counters + prometheus text formatting //! //! observability from day one (docs/lessons-from-zlay.md #8, #11): the //! build_info canary proves which binary runs, and verification counters //! prove verification is actually verifying — a check that silently passes //! everything is indistinguishable from a working one without them. //! //! all counters are atomics: single-writer ingest thread, read by whichever //! server thread serves /metrics. const std = @import("std"); const build_options = @import("build_options"); const convert = @import("../ingest/convert.zig"); const process_metrics = @import("process_metrics.zig"); 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 OptionsErrorReason = enum { oversize, bad_envelope_json, bad_payload_json, invalid_options }; pub const CursorMode = enum { live, seq, time_us, clamped, disabled, unavailable, too_old, resolve_failed }; pub const HttpHandler = enum { root, status, subscribe, subscribe_v2, xrpc, }; 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) — benign tail_closed, /// the connection's own read side closed it (client hangup discovered /// by the server reader, or a proxy dropping the middle) — the reason /// the 2026-08-13 eval drops were invisible inside tail_closed client_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 }; pub const OrchestratorState = enum { drain_bootstrap, seal_bootstrap, close_backfill, write_phase_merging, merge, write_phase_steady, }; pub const CompactionPassResult = enum { ok, error_result }; pub const CompactionReason = enum { record, sync, account }; pub const PipelineFailureStage = enum { scheduler, emit, repair, cursor }; const http_series_count = @typeInfo(HttpHandler).@"enum".field_names.len * @typeInfo(HttpMethod).@"enum".field_names.len * @typeInfo(HttpCode).@"enum".field_names.len; const store_series_count = @typeInfo(StoreOp).@"enum".field_names.len * @typeInfo(StoreStatus).@"enum".field_names.len; const orchestrator_phase_count = @typeInfo(OrchestratorPhase).@"enum".field_names.len; pub const Stats = struct { start_time_s: i64 = 0, events_total: Counter = .init(0), frames_total: Counter = .init(0), ingest_events_appended_total: Counter = .init(0), ingest_blocks_flushed_total: Counter = .init(0), ingest_segments_rotated_total: Counter = .init(0), ingest_append_errors_total: Counter = .init(0), ingest_active_segment_bytes: std.atomic.Value(usize) = .init(0), segment_seal_duration_count: Counter = .init(0), segment_seal_duration_sum_us: Counter = .init(0), segment_seal_duration_buckets: [15]Counter = @splat(.init(0)), http_request_duration_count: [http_series_count]Counter = @splat(.init(0)), 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), getblock_duration_buckets: [12]Counter = @splat(.init(0)), manifest_segments_loaded: std.atomic.Value(usize) = .init(0), manifest_resident_bytes: std.atomic.Value(usize) = .init(0), manifest_block_index_cache_hits_total: Counter = .init(0), manifest_block_index_cache_misses_total: Counter = .init(0), manifest_block_index_load_count: Counter = .init(0), manifest_block_index_load_sum_us: Counter = .init(0), manifest_block_index_load_buckets: [8]Counter = @splat(.init(0)), store_op_duration_count: [store_series_count]Counter = @splat(.init(0)), store_op_duration_sum_us: [store_series_count]Counter = @splat(.init(0)), store_op_duration_buckets: [store_series_count][15]Counter = @splat(@splat(.init(0))), orchestrator_phase: std.atomic.Value(usize) = .init(0), orchestrator_phase_transitions: [orchestrator_phase_count * orchestrator_phase_count]Counter = @splat(.init(0)), orchestrator_state_duration_count: [@typeInfo(OrchestratorState).@"enum".field_names.len]Counter = @splat(.init(0)), orchestrator_state_duration_sum_us: [@typeInfo(OrchestratorState).@"enum".field_names.len]Counter = @splat(.init(0)), orchestrator_state_duration_buckets: [@typeInfo(OrchestratorState).@"enum".field_names.len][14]Counter = @splat(@splat(.init(0))), orchestrator_merge_events_kept_total: Counter = .init(0), orchestrator_merge_events_dropped_total: Counter = .init(0), orchestrator_merge_segments_consumed_total: Counter = .init(0), orchestrator_merge_did_lookups_total: Counter = .init(0), orchestrator_merge_repo_revs_updated_total: Counter = .init(0), orchestrator_merge_dids_discovered_post_bootstrap_total: Counter = .init(0), upstream_seq: std.atomic.Value(i64) = .init(0), drops: std.enums.EnumArray(convert.DropReason, Counter) = .initFill(.init(0)), backfill_drops: std.enums.EnumArray(convert.DropReason, Counter) = .initFill(.init(0)), verify_valid: Counter = .init(0), verify_invalid: Counter = .init(0), verify_unverified: Counter = .init(0), verify_cache_hits: Counter = .init(0), verify_cache_misses: Counter = .init(0), resync_completed_total: Counter = .init(0), resync_failed_total: Counter = .init(0), resync_pending_dropped_total: Counter = .init(0), resync_rate_limited_total: Counter = .init(0), resync_stale_total: Counter = .init(0), resync_active: Counter = .init(0), resync_queue_depth: std.atomic.Value(usize) = .init(0), resync_pending: std.atomic.Value(usize) = .init(0), livestream_decode_errors_total: Counter = .init(0), 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), upstream_slow_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), pipeline_submitted: Counter = .init(0), pipeline_claimed: Counter = .init(0), /// Tickets settled by drop-oldest before a verifier worker claimed them. pipeline_unclaimed_dropped: Counter = .init(0), /// Historical `emitted` name: every settled ticket, including drops and /// verifier rejections. This is the scheduler's durable drain boundary. pipeline_emitted: Counter = .init(0), pipeline_fatal: Counter = .init(0), pipeline_failures: std.enums.EnumArray(PipelineFailureStage, Counter) = .initFill(.init(0)), compaction_watermark: Counter = .init(0), compaction_watermark_lag_us: Counter = .init(0), compaction_passes: std.enums.EnumArray(CompactionPassResult, Counter) = .initFill(.init(0)), compaction_pass_duration_count: Counter = .init(0), compaction_pass_duration_sum_us: Counter = .init(0), compaction_pass_duration_buckets: [16]Counter = @splat(.init(0)), compaction_passes_early_total: Counter = .init(0), compaction_tombstones_collected: std.enums.EnumArray(CompactionReason, Counter) = .initFill(.init(0)), compaction_tombstone_set_entries: std.atomic.Value(usize) = .init(0), compaction_tombstone_set_bytes: std.atomic.Value(usize) = .init(0), compaction_segments_examined_total: Counter = .init(0), compaction_segments_rewritten_total: Counter = .init(0), compaction_segments_skipped_clean_total: Counter = .init(0), compaction_manifest_reconciled_total: Counter = .init(0), compaction_rows_dropped: std.enums.EnumArray(CompactionReason, Counter) = .initFill(.init(0)), compaction_bytes_rewritten_total: Counter = .init(0), /// Unix seconds when the last compaction pass finished (ok or error); set /// to boot time at startup so "never ran since boot" reads as growing /// age, not as a healthy zero. Alert when age exceeds 2x the interval. compaction_last_pass_unix_s: std.atomic.Value(i64) = .init(0), /// Segments folded (tombstone-collected) by the running pass, live. /// The fold phase of a backlog pass reads the whole archive and /// previously reported nothing until chunk commit. compaction_fold_segments_total: Counter = .init(0), compaction_rewritten_age_count: Counter = .init(0), compaction_rewritten_age_sum_s: Counter = .init(0), compaction_rewritten_age_buckets: [8]Counter = @splat(.init(0)), rebloom_segments_total: Counter = .init(0), rebloom_bytes_reclaimed_total: Counter = .init(0), retry_resynced_total: Counter = .init(0), retry_failed_total: Counter = .init(0), backfill_discovered_total: Counter = .init(0), backfill_completed_total: Counter = .init(0), backfill_failed_total: Counter = .init(0), backfill_active_flips_total: Counter = .init(0), backfill_on_fail_store_errors_total: Counter = .init(0), backfill_handle_repo_duration_count: Counter = .init(0), backfill_handle_repo_duration_sum_us: Counter = .init(0), backfill_handle_repo_duration_buckets: [15]Counter = @splat(.init(0)), backfill_progress_completed: std.atomic.Value(usize) = .init(0), backfill_completion_queued_total: Counter = .init(0), backfill_completion_queue_depth: std.atomic.Value(usize) = .init(0), backfill_completion_durable_batches_total: Counter = .init(0), backfill_completion_durable_repos_total: Counter = .init(0), backfill_completion_stage_errors_total: Counter = .init(0), backfill_completion_queue_wait_count: Counter = .init(0), backfill_completion_queue_wait_sum_us: Counter = .init(0), backfill_completion_queue_wait_buckets: [15]Counter = @splat(.init(0)), backfill_forced_checkpoint_flushes_total: Counter = .init(0), retry_passes_total: Counter = .init(0), retry_terminal_failures_total: Counter = .init(0), retry_candidates_total: Counter = .init(0), retry_attempts_total: Counter = .init(0), retry_skipped_host_parked_total: Counter = .init(0), retry_pds_direct_total: Counter = .init(0), retry_relay_fallback_total: Counter = .init(0), backfill_workers_active: std.atomic.Value(usize) = .init(0), backfill_queue_depth: std.atomic.Value(usize) = .init(0), import_jobs_ok_total: Counter = .init(0), import_jobs_error_total: Counter = .init(0), import_job_duration_count: Counter = .init(0), import_job_duration_sum_us: Counter = .init(0), import_job_duration_buckets: [16]Counter = @splat(.init(0)), import_phase: Counter = .init(0), import_rows_parsed_total: Counter = .init(0), import_rows_rejected: std.enums.EnumArray(timestamp_parse.RejectReason, Counter) = .initFill(.init(0)), import_rows_mutated_total: Counter = .init(0), import_rows_matched_all_total: Counter = .init(0), import_rows_matched_specific_total: Counter = .init(0), import_rows_corrupt_offset_total: Counter = .init(0), import_dids_matched_total: Counter = .init(0), 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), subscribe_events_skipped_sync_total: Counter = .init(0), subscribe_events_skipped_resync_total: Counter = .init(0), subscribe_encode_errors_total: Counter = .init(0), subscribe_options_update_errors: std.enums.EnumArray(OptionsErrorReason, Counter) = .initFill(.init(0)), subscribe_cursor_requests: std.enums.EnumArray(CursorMode, Counter) = .initFill(.init(0)), subscribe_cursor_resolve_count: Counter = .init(0), subscribe_cursor_resolve_sum_us: Counter = .init(0), subscribe_cursor_resolve_buckets: [8]Counter = @splat(.init(0)), subscribe_hot_reads_total: Counter = .init(0), subscribe_cold_reads_total: Counter = .init(0), subscribe_cold_stalls_total: Counter = .init(0), subscribe_cold_contended_total: Counter = .init(0), subscribe_capacity_rejects_total: Counter = .init(0), subscribe_adversarial_drops_total: Counter = .init(0), subscribe_clean_disconnects_total: Counter = .init(0), subscribe_options_updates_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| { const delta = after.get(reason) - before.get(reason); if (delta > 0) _ = self.drops.getPtr(reason).fetchAdd(delta, .monotonic); } } pub fn observeCursorResolve(self: *Stats, duration_us: u64) void { _ = self.subscribe_cursor_resolve_count.fetchAdd(1, .monotonic); _ = self.subscribe_cursor_resolve_sum_us.fetchAdd(duration_us, .monotonic); const bounds = [_]u64{ 100, 400, 1_600, 6_400, 25_600, 102_400, 409_600, 1_638_400 }; inline for (bounds, 0..) |upper, i| { if (duration_us <= upper) _ = self.subscribe_cursor_resolve_buckets[i].fetchAdd(1, .monotonic); } } pub fn observeSegmentSeal(self: *Stats, duration_us: u64) void { _ = self.segment_seal_duration_count.fetchAdd(1, .monotonic); _ = self.segment_seal_duration_sum_us.fetchAdd(duration_us, .monotonic); inline for (0..15) |i| { const upper_us = @as(u64, 10_000) << @intCast(i); if (duration_us <= upper_us) _ = self.segment_seal_duration_buckets[i].fetchAdd(1, .monotonic); } } pub fn observeHttp(self: *Stats, handler: HttpHandler, method: HttpMethod, code: u16, duration_us: u64) void { const index = httpIndex(handler, method, httpCode(code)); _ = self.http_request_duration_count[index].fetchAdd(1, .monotonic); _ = self.http_request_duration_sum_us[index].fetchAdd(duration_us, .monotonic); const bounds_us = [_]u64{ 5_000, 10_000, 25_000, 50_000, 100_000, 250_000, 500_000, 1_000_000, 2_500_000, 5_000_000, 10_000_000 }; inline for (bounds_us, 0..) |upper, i| { if (duration_us <= upper) _ = self.http_request_duration_buckets[index][i].fetchAdd(1, .monotonic); } } pub fn observeGetBlock(self: *Stats, code: u16, served_bytes: u64, duration_us: u64) void { const result: GetBlockResult = switch (code) { 200, 206, 304 => .ok, 400 => .bad_request, 404 => .not_found, else => .error_result, }; _ = self.getblock_requests.getPtr(result).fetchAdd(1, .monotonic); if (code == 200 and served_bytes > 0) _ = self.getblock_served_bytes_total.fetchAdd(served_bytes, .monotonic); _ = self.getblock_duration_count.fetchAdd(1, .monotonic); _ = self.getblock_duration_sum_us.fetchAdd(duration_us, .monotonic); inline for (0..12) |i| { const upper_us = @as(u64, 100) << @intCast(i); if (duration_us <= upper_us) _ = self.getblock_duration_buckets[i].fetchAdd(1, .monotonic); } } pub fn observeManifestLoad(self: *Stats, duration_us: u64) void { _ = self.manifest_block_index_load_count.fetchAdd(1, .monotonic); _ = self.manifest_block_index_load_sum_us.fetchAdd(duration_us, .monotonic); inline for (0..8) |index| { const upper_us = @as(u64, 100) << @intCast(index * 2); if (duration_us <= upper_us) _ = self.manifest_block_index_load_buckets[index].fetchAdd(1, .monotonic); } } /// Bucket bounds in seconds: 1h, 6h, 1d, 3d, 7d, 30d, 90d, 365d. The /// distribution answers whether steady-state rewrites cluster in recent /// segments or scatter across deep history (storage-tiering go/no-go). pub const rewritten_age_bounds_s = [8]u64{ 3600, 21600, 86400, 259200, 604800, 2592000, 7776000, 31536000 }; pub fn observeCompactionRewriteAge(self: *Stats, age_s: u64) void { _ = self.compaction_rewritten_age_count.fetchAdd(1, .monotonic); _ = self.compaction_rewritten_age_sum_s.fetchAdd(age_s, .monotonic); inline for (rewritten_age_bounds_s, 0..) |upper_s, index| { if (age_s <= upper_s) _ = self.compaction_rewritten_age_buckets[index].fetchAdd(1, .monotonic); } } pub fn observeStore(self: *Stats, op: StoreOp, status: StoreStatus, duration_us: u64) void { const index = storeIndex(op, status); _ = self.store_op_duration_count[index].fetchAdd(1, .monotonic); _ = self.store_op_duration_sum_us[index].fetchAdd(duration_us, .monotonic); inline for (0..15) |bucket_index| { const upper_us = @as(u64, 100) << @intCast(bucket_index); if (duration_us <= upper_us) _ = self.store_op_duration_buckets[index][bucket_index].fetchAdd(1, .monotonic); } } pub fn setOrchestratorPhase(self: *Stats, phase: OrchestratorPhase) void { self.orchestrator_phase.store(@backingInt(phase) + 1, .monotonic); } pub fn observeOrchestratorTransition(self: *Stats, from: OrchestratorPhase, to: OrchestratorPhase) void { _ = self.orchestrator_phase_transitions[orchestratorTransitionIndex(from, to)].fetchAdd(1, .monotonic); } pub fn observeOrchestratorState(self: *Stats, state: OrchestratorState, duration_us: u64) void { const index = @backingInt(state); _ = self.orchestrator_state_duration_count[index].fetchAdd(1, .monotonic); _ = self.orchestrator_state_duration_sum_us[index].fetchAdd(duration_us, .monotonic); inline for (0..14) |bucket_index| { const upper_us = @as(u64, 10_000) << @intCast(bucket_index); if (duration_us <= upper_us) _ = self.orchestrator_state_duration_buckets[index][bucket_index].fetchAdd(1, .monotonic); } } pub fn observeBackfillHandleRepo(self: *Stats, duration_us: u64) void { observeSlowHistogram( &self.backfill_handle_repo_duration_count, &self.backfill_handle_repo_duration_sum_us, &self.backfill_handle_repo_duration_buckets, duration_us, ); } pub fn observeBackfillCompletionWait(self: *Stats, duration_us: u64) void { observeSlowHistogram( &self.backfill_completion_queue_wait_count, &self.backfill_completion_queue_wait_sum_us, &self.backfill_completion_queue_wait_buckets, duration_us, ); } pub fn observeCompactionPass(self: *Stats, result: CompactionPassResult, duration_us: u64) void { _ = self.compaction_passes.getPtr(result).fetchAdd(1, .monotonic); _ = self.compaction_pass_duration_count.fetchAdd(1, .monotonic); _ = self.compaction_pass_duration_sum_us.fetchAdd(duration_us, .monotonic); inline for (0..16) |index| { const upper_us = @as(u64, 100_000) << @intCast(index); if (duration_us <= upper_us) _ = self.compaction_pass_duration_buckets[index].fetchAdd(1, .monotonic); } } }; fn observeSlowHistogram(count: *Counter, sum_us: *Counter, buckets: *[15]Counter, duration_us: u64) void { _ = count.fetchAdd(1, .monotonic); _ = sum_us.fetchAdd(duration_us, .monotonic); inline for (0..15) |index| { const upper_us = @as(u64, 10_000) << @intCast(index); if (duration_us <= upper_us) _ = buckets[index].fetchAdd(1, .monotonic); } } fn storeIndex(op: StoreOp, status: StoreStatus) usize { return @backingInt(op) * @typeInfo(StoreStatus).@"enum".field_names.len + @backingInt(status); } fn orchestratorTransitionIndex(from: OrchestratorPhase, to: OrchestratorPhase) usize { return @backingInt(from) * orchestrator_phase_count + @backingInt(to); } fn httpIndex(handler: HttpHandler, method: HttpMethod, code: HttpCode) usize { const method_count = @typeInfo(HttpMethod).@"enum".field_names.len; const code_count = @typeInfo(HttpCode).@"enum".field_names.len; return (@backingInt(handler) * method_count + @backingInt(method)) * code_count + @backingInt(code); } fn httpCode(code: u16) HttpCode { return switch (code) { 200 => .c200, 201 => .c201, 206 => .c206, 304 => .c304, 400 => .c400, 401 => .c401, 404 => .c404, 405 => .c405, 409 => .c409, 412 => .c412, 416 => .c416, 500 => .c500, 501 => .c501, 503 => .c503, else => .other, }; } /// Mirror upstream's public mux, including its middleware boundaries. The /// debug /metrics and /healthz routes and unmatched public requests are not /// wrapped. All XRPC methods share the stable `xrpc/` handler label because /// upstream registers one `/xrpc/` subtree handler. pub fn httpHandler(path: []const u8, method: []const u8) ?HttpHandler { // the v2 subscription lives on an /xrpc/ path but keeps its own handler // label (continuity with the /subscribe-v2 era dashboards) if (std.mem.eql(u8, path, "/xrpc/network.bsky.jetstream.subscribeEvents")) return .subscribe_v2; if (std.mem.startsWith(u8, path, "/xrpc/")) return .xrpc; if (std.mem.eql(u8, method, "GET")) { if (std.mem.eql(u8, path, "/")) return .root; if (std.mem.eql(u8, path, "/status")) return .status; if (std.mem.eql(u8, path, "/subscribe")) return .subscribe; } if (std.mem.eql(u8, method, "HEAD") and std.mem.eql(u8, path, "/status")) return .status; return null; } pub fn httpMethod(method: []const u8) HttpMethod { if (std.mem.eql(u8, method, "GET")) return .get; if (std.mem.eql(u8, method, "HEAD")) return .head; if (std.mem.eql(u8, method, "POST")) return .post; if (std.mem.eql(u8, method, "PUT")) return .put; if (std.mem.eql(u8, method, "PATCH")) return .patch; if (std.mem.eql(u8, method, "DELETE")) return .delete; if (std.mem.eql(u8, method, "OPTIONS")) return .options; if (std.mem.eql(u8, method, "CONNECT")) return .connect; if (std.mem.eql(u8, method, "TRACE")) return .trace; return .other; } pub const TailGauges = struct { entries: usize, bytes: usize, pinned_bytes: usize, pinned_overrun_bytes: usize, base: u64, tip: u64, }; /// Durable repository counts, read from the metadata store at scrape time. /// These are the only backfill progress figures that survive a restart: the /// process-local counters reset to zero and climb again, which is what made a /// restart look like progress on the July dashboard. Null when no repository /// state exists, which must not render as zero. pub const DurableRepos = struct { complete: u64 = 0, pending: u64 = 0, failed: u64 = 0, unavailable: u64 = 0, not_started: u64 = 0, pub fn total(self: DurableRepos) u64 { return self.complete + self.pending + self.failed + self.unavailable + self.not_started; } }; pub fn format( buf: []u8, stats: *const Stats, subscribers_active: u32, tail: TailGauges, now_s: i64, process: process_metrics.Snapshot, data_dir_free_bytes: ?u64, durable_repos: ?DurableRepos, ) []const u8 { @setEvalBranchQuota(2_000); var w: Io.Writer = .fixed(buf); // canary: proves what binary is running (lessons #8) w.print( \\# TYPE stream_build_info gauge \\stream_build_info{{git_sha="{s}",optimize="{s}"}} 1 \\# TYPE stream_uptime_seconds gauge \\stream_uptime_seconds {d} \\# TYPE stream_events_total counter \\stream_events_total {d} \\# TYPE stream_frames_total counter \\stream_frames_total {d} \\# TYPE stream_upstream_seq gauge \\stream_upstream_seq {d} \\# TYPE stream_subscribers_active gauge \\stream_subscribers_active {d} \\# TYPE stream_tail_entries gauge \\stream_tail_entries {d} \\# TYPE stream_tail_bytes gauge \\stream_tail_bytes {d} \\# TYPE jetstream_ingest_readable_log_bytes gauge \\jetstream_ingest_readable_log_bytes {d} \\# TYPE jetstream_ingest_readable_log_pinned_bytes gauge \\jetstream_ingest_readable_log_pinned_bytes {d} \\# TYPE jetstream_ingest_readable_log_pinned_overrun_bytes gauge \\jetstream_ingest_readable_log_pinned_overrun_bytes {d} \\# TYPE stream_tail_base gauge \\stream_tail_base {d} \\# TYPE stream_tail_tip gauge \\stream_tail_tip {d} \\ , .{ build_options.git_sha, build_options.optimize, now_s - stats.start_time_s, stats.events_total.load(.monotonic), stats.frames_total.load(.monotonic), stats.upstream_seq.load(.monotonic), subscribers_active, tail.entries, tail.bytes, tail.bytes, tail.pinned_bytes, tail.pinned_overrun_bytes, tail.base, tail.tip, }) catch {}; const pipeline_submitted = stats.pipeline_submitted.load(.monotonic); const pipeline_claimed = stats.pipeline_claimed.load(.monotonic); const pipeline_unclaimed_dropped = stats.pipeline_unclaimed_dropped.load(.monotonic); const pipeline_emitted = stats.pipeline_emitted.load(.monotonic); const pipeline_claimed_settled = pipeline_emitted -| pipeline_unclaimed_dropped; w.print( \\# TYPE stream_pipeline_ticket gauge \\stream_pipeline_ticket{{stage="submitted"}} {d} \\stream_pipeline_ticket{{stage="claimed"}} {d} \\stream_pipeline_ticket{{stage="emitted"}} {d} \\# TYPE stream_pipeline_stage_gap gauge \\stream_pipeline_stage_gap{{boundary="submitted_to_claimed"}} {d} \\stream_pipeline_stage_gap{{boundary="claimed_to_emitted"}} {d} \\stream_pipeline_stage_gap{{boundary="submitted_to_emitted"}} {d} \\# TYPE stream_pipeline_fatal gauge \\stream_pipeline_fatal {d} \\# TYPE stream_pipeline_failures_total counter \\ , .{ pipeline_submitted, pipeline_claimed, pipeline_emitted, pipeline_submitted -| pipeline_claimed -| pipeline_unclaimed_dropped, pipeline_claimed -| pipeline_claimed_settled, pipeline_submitted -| pipeline_emitted, stats.pipeline_fatal.load(.monotonic), }) catch {}; inline for (comptime std.enums.values(PipelineFailureStage)) |stage| { w.print("stream_pipeline_failures_total{{stage=\"{s}\"}} {d}\n", .{ @tagName(stage), stats.pipeline_failures.getPtrConst(stage).load(.monotonic), }) catch {}; } w.print( \\# TYPE process_start_time_seconds gauge \\process_start_time_seconds {d} \\ , .{stats.start_time_s}) catch {}; if (process.cpu_us) |value| w.print( \\# TYPE process_cpu_seconds_total counter \\process_cpu_seconds_total {d}.{d:0>6} \\ , .{ value / std.time.us_per_s, value % std.time.us_per_s }) catch {}; if (process.resident_memory_bytes) |value| w.print( \\# TYPE process_resident_memory_bytes gauge \\process_resident_memory_bytes {d} \\ , .{value}) catch {}; if (process.virtual_memory_bytes) |value| w.print( \\# TYPE process_virtual_memory_bytes gauge \\process_virtual_memory_bytes {d} \\ , .{value}) catch {}; // all-arena glibc heap picture (zlay lesson 8: balanced allocs + growing // RSS = arena fragmentation; free_total is the recoverable-in-principle // slice, arena_heaps the chained 64 MiB units that pin it) if (process.malloc_system_bytes) |value| w.print( \\# TYPE stream_malloc_system_bytes gauge \\stream_malloc_system_bytes {d} \\ , .{value}) catch {}; if (process.malloc_free_total_bytes) |value| w.print( \\# TYPE stream_malloc_free_total_bytes gauge \\stream_malloc_free_total_bytes {d} \\ , .{value}) catch {}; if (process.malloc_arena_heaps) |value| w.print( \\# TYPE stream_malloc_arena_heaps gauge \\stream_malloc_arena_heaps {d} \\ , .{value}) catch {}; if (process.threads) |value| w.print( \\# TYPE stream_process_threads gauge \\stream_process_threads {d} \\ , .{value}) catch {}; if (process.open_fds) |value| w.print( \\# TYPE process_open_fds gauge \\process_open_fds {d} \\ , .{value}) catch {}; if (process.max_fds) |value| w.print( \\# TYPE process_max_fds gauge \\process_max_fds {d} \\ , .{value}) catch {}; if (process.minor_page_faults) |value| w.print( \\# TYPE process_minor_page_faults_total counter \\process_minor_page_faults_total {d} \\ , .{value}) catch {}; if (process.major_page_faults) |value| w.print( \\# TYPE process_major_page_faults_total counter \\process_major_page_faults_total {d} \\ , .{value}) catch {}; if (process.voluntary_context_switches) |value| w.print( \\# TYPE process_voluntary_context_switches_total counter \\process_voluntary_context_switches_total {d} \\ , .{value}) catch {}; if (process.involuntary_context_switches) |value| w.print( \\# TYPE process_involuntary_context_switches_total counter \\process_involuntary_context_switches_total {d} \\ , .{value}) catch {}; if (data_dir_free_bytes) |value| w.print( \\# TYPE jetstream_data_dir_free_bytes gauge \\jetstream_data_dir_free_bytes {d} \\ , .{value}) catch {}; w.print( \\# TYPE stream_resync_total counter \\stream_resync_total{{result="completed"}} {d} \\stream_resync_total{{result="failed"}} {d} \\stream_resync_total{{result="rate_limited"}} {d} \\# TYPE stream_resync_pending_dropped_total counter \\stream_resync_pending_dropped_total {d} \\# TYPE stream_resync_workers_active gauge \\stream_resync_workers_active {d} \\# TYPE stream_resync_queue_depth gauge \\stream_resync_queue_depth {d} \\# TYPE stream_resync_pending gauge \\stream_resync_pending {d} \\# TYPE jetstream_livestream_stale_resyncs_dropped_total counter \\jetstream_livestream_stale_resyncs_dropped_total {d} \\# TYPE jetstream_livestream_verify_queue_dropped_events_total counter \\jetstream_livestream_verify_queue_dropped_events_total {d} \\ , .{ stats.resync_completed_total.load(.monotonic), stats.resync_failed_total.load(.monotonic), stats.resync_rate_limited_total.load(.monotonic), stats.resync_pending_dropped_total.load(.monotonic), stats.resync_active.load(.monotonic), stats.resync_queue_depth.load(.monotonic), stats.resync_pending.load(.monotonic), stats.resync_stale_total.load(.monotonic), stats.verify_queue_dropped_total.load(.monotonic), }) catch {}; const seal_sum_us = stats.segment_seal_duration_sum_us.load(.monotonic); const seal_count = stats.segment_seal_duration_count.load(.monotonic); w.print( \\# TYPE jetstream_segment_seal_duration_seconds histogram \\jetstream_segment_seal_duration_seconds_sum {d}.{d:0>6} \\jetstream_segment_seal_duration_seconds_count {d} \\ , .{ seal_sum_us / std.time.us_per_s, seal_sum_us % std.time.us_per_s, seal_count }) catch {}; const seal_bounds = [_][]const u8{ "0.01", "0.02", "0.04", "0.08", "0.16", "0.32", "0.64", "1.28", "2.56", "5.12", "10.24", "20.48", "40.96", "81.92", "163.84", }; inline for (seal_bounds, 0..) |upper, i| { w.print("jetstream_segment_seal_duration_seconds_bucket{{le=\"{s}\"}} {d}\n", .{ upper, stats.segment_seal_duration_buckets[i].load(.monotonic), }) catch {}; } w.print("jetstream_segment_seal_duration_seconds_bucket{{le=\"+Inf\"}} {d}\n", .{seal_count}) catch {}; w.print("# TYPE jetstream_http_request_duration_seconds histogram\n", .{}) catch {}; for (comptime std.enums.values(HttpHandler)) |handler| { for (comptime std.enums.values(HttpMethod)) |method| { for (comptime std.enums.values(HttpCode)) |code| { const index = httpIndex(handler, method, code); const count = stats.http_request_duration_count[index].load(.monotonic); if (count != 0) { const labels = .{ httpHandlerLabel(handler), httpMethodLabel(method), httpCodeLabel(code) }; const sum_us = stats.http_request_duration_sum_us[index].load(.monotonic); w.print("jetstream_http_request_duration_seconds_sum{{handler=\"{s}\",method=\"{s}\",code=\"{s}\"}} {d}.{d:0>6}\n", .{ labels[0], labels[1], labels[2], sum_us / std.time.us_per_s, sum_us % std.time.us_per_s, }) catch {}; w.print("jetstream_http_request_duration_seconds_count{{handler=\"{s}\",method=\"{s}\",code=\"{s}\"}} {d}\n", .{ labels[0], labels[1], labels[2], count, }) catch {}; const bounds = [_][]const u8{ "0.005", "0.01", "0.025", "0.05", "0.1", "0.25", "0.5", "1", "2.5", "5", "10" }; for (bounds, 0..) |upper, i| w.print("jetstream_http_request_duration_seconds_bucket{{handler=\"{s}\",method=\"{s}\",code=\"{s}\",le=\"{s}\"}} {d}\n", .{ labels[0], labels[1], labels[2], upper, stats.http_request_duration_buckets[index][i].load(.monotonic), }) catch {}; w.print("jetstream_http_request_duration_seconds_bucket{{handler=\"{s}\",method=\"{s}\",code=\"{s}\",le=\"+Inf\"}} {d}\n", .{ labels[0], labels[1], labels[2], count, }) catch {}; } } } } w.print("# TYPE jetstream_getblock_requests_total counter\n", .{}) catch {}; inline for (comptime std.enums.values(GetBlockResult)) |result| { const count = stats.getblock_requests.getPtrConst(result).load(.monotonic); if (count != 0) w.print( "jetstream_getblock_requests_total{{result=\"{s}\"}} {d}\n", .{ 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} \\# TYPE jetstream_getblock_duration_seconds histogram \\jetstream_getblock_duration_seconds_sum {d}.{d:0>6} \\jetstream_getblock_duration_seconds_count {d} \\ , .{ stats.getblock_served_bytes_total.load(.monotonic), stats.getblock_duration_sum_us.load(.monotonic) / std.time.us_per_s, stats.getblock_duration_sum_us.load(.monotonic) % std.time.us_per_s, stats.getblock_duration_count.load(.monotonic), }) catch {}; const getblock_bounds = [_][]const u8{ "0.0001", "0.0002", "0.0004", "0.0008", "0.0016", "0.0032", "0.0064", "0.0128", "0.0256", "0.0512", "0.1024", "0.2048" }; inline for (getblock_bounds, 0..) |upper, i| w.print( "jetstream_getblock_duration_seconds_bucket{{le=\"{s}\"}} {d}\n", .{ upper, stats.getblock_duration_buckets[i].load(.monotonic) }, ) catch {}; w.print("jetstream_getblock_duration_seconds_bucket{{le=\"+Inf\"}} {d}\n", .{ stats.getblock_duration_count.load(.monotonic), }) catch {}; const manifest_load_sum_us = stats.manifest_block_index_load_sum_us.load(.monotonic); const manifest_load_count = stats.manifest_block_index_load_count.load(.monotonic); w.print( \\# TYPE jetstream_manifest_segments_loaded gauge \\jetstream_manifest_segments_loaded {d} \\# TYPE stream_manifest_resident_bytes gauge \\stream_manifest_resident_bytes {d} \\# TYPE jetstream_manifest_block_index_cache_hits_total counter \\jetstream_manifest_block_index_cache_hits_total {d} \\# TYPE jetstream_manifest_block_index_cache_misses_total counter \\jetstream_manifest_block_index_cache_misses_total {d} \\# TYPE jetstream_manifest_block_index_load_seconds histogram \\jetstream_manifest_block_index_load_seconds_sum {d}.{d:0>6} \\jetstream_manifest_block_index_load_seconds_count {d} \\ , .{ stats.manifest_segments_loaded.load(.monotonic), stats.manifest_resident_bytes.load(.monotonic), stats.manifest_block_index_cache_hits_total.load(.monotonic), stats.manifest_block_index_cache_misses_total.load(.monotonic), manifest_load_sum_us / std.time.us_per_s, manifest_load_sum_us % std.time.us_per_s, manifest_load_count, }) catch {}; const manifest_load_bounds = [_][]const u8{ "0.0001", "0.0004", "0.0016", "0.0064", "0.0256", "0.1024", "0.4096", "1.6384" }; inline for (manifest_load_bounds, 0..) |upper, index| w.print( "jetstream_manifest_block_index_load_seconds_bucket{{le=\"{s}\"}} {d}\n", .{ upper, stats.manifest_block_index_load_buckets[index].load(.monotonic) }, ) catch {}; w.print("jetstream_manifest_block_index_load_seconds_bucket{{le=\"+Inf\"}} {d}\n", .{manifest_load_count}) catch {}; w.print("# TYPE jetstream_store_op_duration_seconds histogram\n", .{}) catch {}; const store_bounds = [_][]const u8{ "0.0001", "0.0002", "0.0004", "0.0008", "0.0016", "0.0032", "0.0064", "0.0128", "0.0256", "0.0512", "0.1024", "0.2048", "0.4096", "0.8192", "1.6384", }; inline for (comptime std.enums.values(StoreOp)) |op| { inline for (comptime std.enums.values(StoreStatus)) |status| { if (op != .get and status == .notfound) continue; const index = storeIndex(op, status); const count = stats.store_op_duration_count[index].load(.monotonic); const sum_us = stats.store_op_duration_sum_us[index].load(.monotonic); inline for (store_bounds, 0..) |upper, bucket_index| w.print( "jetstream_store_op_duration_seconds_bucket{{op=\"{s}\",status=\"{s}\",le=\"{s}\"}} {d}\n", .{ storeOpLabel(op), storeStatusLabel(status), upper, stats.store_op_duration_buckets[index][bucket_index].load(.monotonic) }, ) catch {}; w.print( "jetstream_store_op_duration_seconds_bucket{{op=\"{s}\",status=\"{s}\",le=\"+Inf\"}} {d}\n", .{ storeOpLabel(op), storeStatusLabel(status), count }, ) catch {}; w.print( "jetstream_store_op_duration_seconds_sum{{op=\"{s}\",status=\"{s}\"}} {d}.{d:0>6}\n", .{ storeOpLabel(op), storeStatusLabel(status), sum_us / std.time.us_per_s, sum_us % std.time.us_per_s }, ) catch {}; w.print( "jetstream_store_op_duration_seconds_count{{op=\"{s}\",status=\"{s}\"}} {d}\n", .{ storeOpLabel(op), storeStatusLabel(status), count }, ) catch {}; } } w.print( \\# TYPE jetstream_orchestrator_phase gauge \\jetstream_orchestrator_phase {d} \\# TYPE jetstream_orchestrator_phase_transitions_total counter \\ , .{stats.orchestrator_phase.load(.monotonic)}) catch {}; inline for (comptime std.enums.values(OrchestratorPhase)) |from| { inline for (comptime std.enums.values(OrchestratorPhase)) |to| { const count = stats.orchestrator_phase_transitions[orchestratorTransitionIndex(from, to)].load(.monotonic); if (count != 0) w.print( "jetstream_orchestrator_phase_transitions_total{{from=\"{s}\",to=\"{s}\"}} {d}\n", .{ @tagName(from), @tagName(to), count }, ) catch {}; } } w.print("# TYPE jetstream_orchestrator_state_duration_seconds histogram\n", .{}) catch {}; const orchestrator_state_bounds = [_][]const u8{ "0.01", "0.02", "0.04", "0.08", "0.16", "0.32", "0.64", "1.28", "2.56", "5.12", "10.24", "20.48", "40.96", "81.92", }; inline for (comptime std.enums.values(OrchestratorState)) |state| { const index = @backingInt(state); const count = stats.orchestrator_state_duration_count[index].load(.monotonic); if (count != 0) { const sum_us = stats.orchestrator_state_duration_sum_us[index].load(.monotonic); inline for (orchestrator_state_bounds, 0..) |upper, bucket_index| w.print( "jetstream_orchestrator_state_duration_seconds_bucket{{state=\"{s}\",le=\"{s}\"}} {d}\n", .{ @tagName(state), upper, stats.orchestrator_state_duration_buckets[index][bucket_index].load(.monotonic) }, ) catch {}; w.print( "jetstream_orchestrator_state_duration_seconds_bucket{{state=\"{s}\",le=\"+Inf\"}} {d}\n", .{ @tagName(state), count }, ) catch {}; w.print( "jetstream_orchestrator_state_duration_seconds_sum{{state=\"{s}\"}} {d}.{d:0>6}\n", .{ @tagName(state), sum_us / std.time.us_per_s, sum_us % std.time.us_per_s }, ) catch {}; w.print( "jetstream_orchestrator_state_duration_seconds_count{{state=\"{s}\"}} {d}\n", .{ @tagName(state), count }, ) catch {}; } } w.print( \\# TYPE jetstream_orchestrator_merge_events_kept_total counter \\jetstream_orchestrator_merge_events_kept_total {d} \\# TYPE jetstream_orchestrator_merge_events_dropped_total counter \\jetstream_orchestrator_merge_events_dropped_total {d} \\# TYPE jetstream_orchestrator_merge_segments_consumed_total counter \\jetstream_orchestrator_merge_segments_consumed_total {d} \\# TYPE jetstream_orchestrator_merge_did_lookups_total counter \\jetstream_orchestrator_merge_did_lookups_total {d} \\# TYPE jetstream_orchestrator_merge_repo_revs_updated_total counter \\jetstream_orchestrator_merge_repo_revs_updated_total {d} \\# TYPE jetstream_orchestrator_merge_dids_discovered_post_bootstrap_total counter \\jetstream_orchestrator_merge_dids_discovered_post_bootstrap_total {d} \\ , .{ stats.orchestrator_merge_events_kept_total.load(.monotonic), stats.orchestrator_merge_events_dropped_total.load(.monotonic), stats.orchestrator_merge_segments_consumed_total.load(.monotonic), stats.orchestrator_merge_did_lookups_total.load(.monotonic), stats.orchestrator_merge_repo_revs_updated_total.load(.monotonic), stats.orchestrator_merge_dids_discovered_post_bootstrap_total.load(.monotonic), }) catch {}; // Canonical upstream names. These are aliases only where Stream owns the // same lifecycle boundary; absent families stay absent so the dashboard // contract cannot be satisfied by placeholder zeroes. w.print( \\# TYPE jetstream_build_info gauge \\jetstream_build_info{{version="{s}",optimize="{s}"}} 1 \\# TYPE jetstream_livestream_events_received_total counter \\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 \\jetstream_livestream_replayed_identity_events_dropped_total {d} \\# TYPE jetstream_ingest_events_appended_total counter \\jetstream_ingest_events_appended_total {d} \\# TYPE jetstream_ingest_blocks_flushed_total counter \\jetstream_ingest_blocks_flushed_total {d} \\# TYPE jetstream_ingest_segments_rotated_total counter \\jetstream_ingest_segments_rotated_total {d} \\# TYPE jetstream_ingest_append_errors_total counter \\jetstream_ingest_append_errors_total {d} \\# TYPE jetstream_ingest_active_segment_bytes gauge \\jetstream_ingest_active_segment_bytes {d} \\ , .{ build_options.git_sha, 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), stats.ingest_blocks_flushed_total.load(.monotonic), stats.ingest_segments_rotated_total.load(.monotonic), stats.ingest_append_errors_total.load(.monotonic), stats.ingest_active_segment_bytes.load(.monotonic), }) 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( \\# TYPE jetstream_subscribe_events_skipped_total counter \\jetstream_subscribe_events_skipped_total{{reason="sync"}} {d} \\jetstream_subscribe_events_skipped_total{{reason="resync_replacement"}} {d} \\# TYPE jetstream_subscribe_encode_errors_total counter \\jetstream_subscribe_encode_errors_total {d} \\# TYPE jetstream_subscribe_hot_reads_total counter \\jetstream_subscribe_hot_reads_total {d} \\# TYPE jetstream_subscribe_cold_reads_total counter \\jetstream_subscribe_cold_reads_total {d} \\# TYPE stream_subscribe_cold_stalls_total counter \\stream_subscribe_cold_stalls_total {d} \\# TYPE stream_subscribe_cold_contended_total counter \\stream_subscribe_cold_contended_total {d} \\# TYPE stream_subscribe_capacity_rejects_total counter \\stream_subscribe_capacity_rejects_total {d} \\# TYPE jetstream_subscribe_adversarial_drops_total counter \\jetstream_subscribe_adversarial_drops_total {d} \\# TYPE jetstream_subscribe_clean_disconnects_total counter \\jetstream_subscribe_clean_disconnects_total {d} \\# TYPE jetstream_subscribe_options_updates_total counter \\jetstream_subscribe_options_updates_total {d} \\ , .{ stats.subscribe_events_skipped_sync_total.load(.monotonic), stats.subscribe_events_skipped_resync_total.load(.monotonic), stats.subscribe_encode_errors_total.load(.monotonic), stats.subscribe_hot_reads_total.load(.monotonic), stats.subscribe_cold_reads_total.load(.monotonic), stats.subscribe_cold_stalls_total.load(.monotonic), stats.subscribe_cold_contended_total.load(.monotonic), stats.subscribe_capacity_rejects_total.load(.monotonic), stats.subscribe_adversarial_drops_total.load(.monotonic), stats.subscribe_clean_disconnects_total.load(.monotonic), stats.subscribe_options_updates_total.load(.monotonic), }) catch {}; w.print( \\# TYPE stream_upstream_slow_reconnects_total counter \\stream_upstream_slow_reconnects_total {d} \\ , .{stats.upstream_slow_reconnects_total.load(.monotonic)}) catch {}; w.print("# TYPE jetstream_subscribe_options_update_errors_total counter\n", .{}) catch {}; inline for (comptime std.enums.values(OptionsErrorReason)) |reason| { w.print("jetstream_subscribe_options_update_errors_total{{reason=\"{s}\"}} {d}\n", .{ @tagName(reason), stats.subscribe_options_update_errors.getPtrConst(reason).load(.monotonic), }) catch {}; } w.print("# TYPE jetstream_subscribe_cursor_requests_total counter\n", .{}) catch {}; inline for (comptime std.enums.values(CursorMode)) |mode| { w.print("jetstream_subscribe_cursor_requests_total{{mode=\"{s}\"}} {d}\n", .{ @tagName(mode), stats.subscribe_cursor_requests.getPtrConst(mode).load(.monotonic), }) catch {}; } const cursor_sum_us = stats.subscribe_cursor_resolve_sum_us.load(.monotonic); w.print( \\# TYPE jetstream_subscribe_cursor_resolve_seconds histogram \\jetstream_subscribe_cursor_resolve_seconds_sum {d}.{d:0>6} \\jetstream_subscribe_cursor_resolve_seconds_count {d} \\ , .{ cursor_sum_us / std.time.us_per_s, cursor_sum_us % std.time.us_per_s, stats.subscribe_cursor_resolve_count.load(.monotonic), }) catch {}; const cursor_bucket_us = [_]u64{ 100, 400, 1_600, 6_400, 25_600, 102_400, 409_600, 1_638_400 }; inline for (cursor_bucket_us, 0..) |upper_us, i| { w.print("jetstream_subscribe_cursor_resolve_seconds_bucket{{le=\"{d}.{d:0>6}\"}} {d}\n", .{ upper_us / std.time.us_per_s, upper_us % std.time.us_per_s, stats.subscribe_cursor_resolve_buckets[i].load(.monotonic), }) catch {}; } w.print("jetstream_subscribe_cursor_resolve_seconds_bucket{{le=\"+Inf\"}} {d}\n", .{ stats.subscribe_cursor_resolve_count.load(.monotonic), }) catch {}; w.print( \\# TYPE stream_verify_total counter \\stream_verify_total{{result="valid"}} {d} \\stream_verify_total{{result="invalid"}} {d} \\stream_verify_total{{result="unverified"}} {d} \\# TYPE stream_verify_key_cache_total counter \\stream_verify_key_cache_total{{result="hit"}} {d} \\stream_verify_key_cache_total{{result="miss"}} {d} \\ , .{ stats.verify_valid.load(.monotonic), stats.verify_invalid.load(.monotonic), stats.verify_unverified.load(.monotonic), stats.verify_cache_hits.load(.monotonic), stats.verify_cache_misses.load(.monotonic), }) catch {}; w.print( \\# TYPE stream_compaction_watermark_seq gauge \\stream_compaction_watermark_seq {d} \\# TYPE jetstream_compaction_watermark_seq gauge \\jetstream_compaction_watermark_seq {d} \\# TYPE jetstream_compaction_watermark_lag_seconds gauge \\jetstream_compaction_watermark_lag_seconds {d}.{d:0>6} \\# TYPE jetstream_compaction_passes_total counter \\jetstream_compaction_passes_total{{result="ok"}} {d} \\jetstream_compaction_passes_total{{result="error"}} {d} \\# TYPE jetstream_compaction_passes_early_total counter \\jetstream_compaction_passes_early_total {d} \\# TYPE jetstream_compaction_tombstone_set_entries gauge \\jetstream_compaction_tombstone_set_entries {d} \\# TYPE jetstream_compaction_tombstone_set_bytes gauge \\jetstream_compaction_tombstone_set_bytes {d} \\# TYPE jetstream_compaction_segments_examined_total counter \\jetstream_compaction_segments_examined_total {d} \\# TYPE jetstream_compaction_segments_rewritten_total counter \\jetstream_compaction_segments_rewritten_total {d} \\# TYPE jetstream_compaction_segments_skipped_clean_total counter \\jetstream_compaction_segments_skipped_clean_total {d} \\# TYPE jetstream_compaction_manifest_reconciled_total counter \\jetstream_compaction_manifest_reconciled_total {d} \\# TYPE jetstream_compaction_bytes_rewritten_total counter \\jetstream_compaction_bytes_rewritten_total {d} \\# TYPE stream_rebloom_segments_total counter \\stream_rebloom_segments_total {d} \\# TYPE stream_rebloom_bytes_reclaimed_total counter \\stream_rebloom_bytes_reclaimed_total {d} \\ , .{ stats.compaction_watermark.load(.monotonic), stats.compaction_watermark.load(.monotonic), stats.compaction_watermark_lag_us.load(.monotonic) / std.time.us_per_s, stats.compaction_watermark_lag_us.load(.monotonic) % std.time.us_per_s, stats.compaction_passes.getPtrConst(.ok).load(.monotonic), stats.compaction_passes.getPtrConst(.error_result).load(.monotonic), stats.compaction_passes_early_total.load(.monotonic), stats.compaction_tombstone_set_entries.load(.monotonic), stats.compaction_tombstone_set_bytes.load(.monotonic), stats.compaction_segments_examined_total.load(.monotonic), stats.compaction_segments_rewritten_total.load(.monotonic), stats.compaction_segments_skipped_clean_total.load(.monotonic), stats.compaction_manifest_reconciled_total.load(.monotonic), stats.compaction_bytes_rewritten_total.load(.monotonic), stats.rebloom_segments_total.load(.monotonic), stats.rebloom_bytes_reclaimed_total.load(.monotonic), }) catch {}; w.print( \\# HELP stream_compaction_last_pass_timestamp_seconds Unix time of the last completed compaction pass; boot time until one completes. Alert when now minus this exceeds 2x the compaction interval. \\# TYPE stream_compaction_last_pass_timestamp_seconds gauge \\stream_compaction_last_pass_timestamp_seconds {d} \\# TYPE stream_compaction_fold_segments_total counter \\stream_compaction_fold_segments_total {d} \\ , .{ stats.compaction_last_pass_unix_s.load(.monotonic), stats.compaction_fold_segments_total.load(.monotonic), }) catch {}; w.print( \\# TYPE stream_compaction_rewritten_age_seconds histogram \\stream_compaction_rewritten_age_seconds_sum {d} \\stream_compaction_rewritten_age_seconds_count {d} \\ , .{ stats.compaction_rewritten_age_sum_s.load(.monotonic), stats.compaction_rewritten_age_count.load(.monotonic), }) catch {}; inline for (Stats.rewritten_age_bounds_s, 0..) |upper_s, i| w.print( "stream_compaction_rewritten_age_seconds_bucket{{le=\"{d}\"}} {d}\n", .{ upper_s, stats.compaction_rewritten_age_buckets[i].load(.monotonic) }, ) catch {}; w.print("stream_compaction_rewritten_age_seconds_bucket{{le=\"+Inf\"}} {d}\n", .{ stats.compaction_rewritten_age_count.load(.monotonic), }) catch {}; w.print( \\# TYPE stream_retry_repos_total counter \\stream_retry_repos_total{{result="resynced"}} {d} \\stream_retry_repos_total{{result="failed"}} {d} \\# TYPE jetstream_backfill_discovered_total counter \\jetstream_backfill_discovered_total {d} \\# TYPE jetstream_backfill_completed_total counter \\jetstream_backfill_completed_total {d} \\# TYPE jetstream_backfill_failed_total counter \\jetstream_backfill_failed_total {d} \\# TYPE jetstream_backfill_active_flips_total counter \\jetstream_backfill_active_flips_total {d} \\# TYPE jetstream_backfill_on_fail_store_errors_total counter \\jetstream_backfill_on_fail_store_errors_total {d} \\# TYPE jetstream_backfill_progress_completed gauge \\jetstream_backfill_progress_completed {d} \\# TYPE jetstream_backfill_completion_queued_total counter \\jetstream_backfill_completion_queued_total {d} \\# TYPE jetstream_backfill_completion_queue_depth gauge \\jetstream_backfill_completion_queue_depth {d} \\# TYPE jetstream_backfill_completion_durable_batches_total counter \\jetstream_backfill_completion_durable_batches_total {d} \\# TYPE jetstream_backfill_completion_durable_repos_total counter \\jetstream_backfill_completion_durable_repos_total {d} \\# TYPE jetstream_backfill_completion_stage_errors_total counter \\jetstream_backfill_completion_stage_errors_total {d} \\# TYPE jetstream_backfill_forced_checkpoint_flushes_total counter \\jetstream_backfill_forced_checkpoint_flushes_total {d} \\# TYPE jetstream_backfill_failed_repo_retry_passes_total counter \\jetstream_backfill_failed_repo_retry_passes_total {d} \\# TYPE stream_backfill_failed_repo_retry_terminal_failures_total counter \\stream_backfill_failed_repo_retry_terminal_failures_total {d} \\# TYPE jetstream_backfill_failed_repo_retry_candidates_total counter \\jetstream_backfill_failed_repo_retry_candidates_total {d} \\# TYPE jetstream_backfill_failed_repo_retry_attempts_total counter \\jetstream_backfill_failed_repo_retry_attempts_total {d} \\# TYPE jetstream_backfill_failed_repo_retry_succeeded_total counter \\jetstream_backfill_failed_repo_retry_succeeded_total {d} \\# TYPE jetstream_backfill_failed_repo_retry_failed_total counter \\jetstream_backfill_failed_repo_retry_failed_total {d} \\# TYPE jetstream_backfill_failed_repo_retry_skipped_host_parked_total counter \\jetstream_backfill_failed_repo_retry_skipped_host_parked_total {d} \\# TYPE stream_backfill_failed_repo_retry_pds_direct_total counter \\stream_backfill_failed_repo_retry_pds_direct_total {d} \\# TYPE stream_backfill_failed_repo_retry_relay_fallback_total counter \\stream_backfill_failed_repo_retry_relay_fallback_total {d} \\# TYPE stream_backfill_repos_total counter \\stream_backfill_repos_total{{result="complete"}} {d} \\stream_backfill_repos_total{{result="failed"}} {d} \\# TYPE stream_backfill_workers_active gauge \\stream_backfill_workers_active {d} \\# TYPE stream_backfill_queue_depth gauge \\stream_backfill_queue_depth {d} \\ , .{ stats.retry_resynced_total.load(.monotonic), stats.retry_failed_total.load(.monotonic), stats.backfill_discovered_total.load(.monotonic), stats.backfill_completed_total.load(.monotonic), stats.backfill_failed_total.load(.monotonic), stats.backfill_active_flips_total.load(.monotonic), stats.backfill_on_fail_store_errors_total.load(.monotonic), stats.backfill_progress_completed.load(.monotonic), stats.backfill_completion_queued_total.load(.monotonic), stats.backfill_completion_queue_depth.load(.monotonic), stats.backfill_completion_durable_batches_total.load(.monotonic), stats.backfill_completion_durable_repos_total.load(.monotonic), stats.backfill_completion_stage_errors_total.load(.monotonic), stats.backfill_forced_checkpoint_flushes_total.load(.monotonic), stats.retry_passes_total.load(.monotonic), stats.retry_terminal_failures_total.load(.monotonic), stats.retry_candidates_total.load(.monotonic), stats.retry_attempts_total.load(.monotonic), stats.retry_resynced_total.load(.monotonic), stats.retry_failed_total.load(.monotonic), stats.retry_skipped_host_parked_total.load(.monotonic), stats.retry_pds_direct_total.load(.monotonic), stats.retry_relay_fallback_total.load(.monotonic), stats.backfill_completed_total.load(.monotonic), stats.backfill_failed_total.load(.monotonic), stats.backfill_workers_active.load(.monotonic), stats.backfill_queue_depth.load(.monotonic), }) catch {}; w.print("# TYPE jetstream_compaction_tombstones_collected_total counter\n", .{}) catch {}; w.print("# TYPE jetstream_compaction_rows_dropped_total counter\n", .{}) catch {}; inline for (comptime std.enums.values(CompactionReason)) |reason| { w.print("jetstream_compaction_tombstones_collected_total{{kind=\"{s}\"}} {d}\n", .{ @tagName(reason), stats.compaction_tombstones_collected.getPtrConst(reason).load(.monotonic), }) catch {}; w.print("jetstream_compaction_rows_dropped_total{{reason=\"{s}\"}} {d}\n", .{ @tagName(reason), stats.compaction_rows_dropped.getPtrConst(reason).load(.monotonic), }) catch {}; } const compaction_count = stats.compaction_pass_duration_count.load(.monotonic); const compaction_sum_us = stats.compaction_pass_duration_sum_us.load(.monotonic); w.print("# TYPE jetstream_compaction_pass_duration_seconds histogram\n", .{}) catch {}; inline for (0..16) |index| { const upper_us = @as(u64, 100_000) << @intCast(index); w.print("jetstream_compaction_pass_duration_seconds_bucket{{le=\"{d}.{d:0>6}\"}} {d}\n", .{ upper_us / std.time.us_per_s, upper_us % std.time.us_per_s, stats.compaction_pass_duration_buckets[index].load(.monotonic), }) catch {}; } w.print( \\jetstream_compaction_pass_duration_seconds_bucket{{le="+Inf"}} {d} \\jetstream_compaction_pass_duration_seconds_sum {d}.{d:0>6} \\jetstream_compaction_pass_duration_seconds_count {d} \\ , .{ compaction_count, compaction_sum_us / std.time.us_per_s, compaction_sum_us % std.time.us_per_s, compaction_count }) catch {}; const backfill_slow_bounds = [_][]const u8{ "0.01", "0.02", "0.04", "0.08", "0.16", "0.32", "0.64", "1.28", "2.56", "5.12", "10.24", "20.48", "40.96", "81.92", "163.84", }; const handle_count = stats.backfill_handle_repo_duration_count.load(.monotonic); const handle_sum_us = stats.backfill_handle_repo_duration_sum_us.load(.monotonic); w.print( "# TYPE jetstream_backfill_handle_repo_duration_seconds histogram\n", .{}, ) catch {}; inline for (backfill_slow_bounds, 0..) |upper, index| w.print( "jetstream_backfill_handle_repo_duration_seconds_bucket{{le=\"{s}\"}} {d}\n", .{ upper, stats.backfill_handle_repo_duration_buckets[index].load(.monotonic) }, ) catch {}; w.print( \\jetstream_backfill_handle_repo_duration_seconds_bucket{{le="+Inf"}} {d} \\jetstream_backfill_handle_repo_duration_seconds_sum {d}.{d:0>6} \\jetstream_backfill_handle_repo_duration_seconds_count {d} \\ , .{ handle_count, handle_sum_us / std.time.us_per_s, handle_sum_us % std.time.us_per_s, handle_count }) catch {}; const completion_wait_count = stats.backfill_completion_queue_wait_count.load(.monotonic); const completion_wait_sum_us = stats.backfill_completion_queue_wait_sum_us.load(.monotonic); w.print( "# TYPE jetstream_backfill_completion_queue_wait_seconds histogram\n", .{}, ) catch {}; inline for (backfill_slow_bounds, 0..) |upper, index| w.print( "jetstream_backfill_completion_queue_wait_seconds_bucket{{le=\"{s}\"}} {d}\n", .{ upper, stats.backfill_completion_queue_wait_buckets[index].load(.monotonic) }, ) catch {}; w.print( \\jetstream_backfill_completion_queue_wait_seconds_bucket{{le="+Inf"}} {d} \\jetstream_backfill_completion_queue_wait_seconds_sum {d}.{d:0>6} \\jetstream_backfill_completion_queue_wait_seconds_count {d} \\ , .{ completion_wait_count, completion_wait_sum_us / std.time.us_per_s, completion_wait_sum_us % std.time.us_per_s, completion_wait_count }) catch {}; w.print("# TYPE stream_dropped_events_total counter\n", .{}) catch {}; w.print("# TYPE jetstream_ingest_dropped_events_total counter\n", .{}) catch {}; inline for (comptime std.enums.values(convert.DropReason)) |reason| { w.print("stream_dropped_events_total{{reason=\"{s}\"}} {d}\n", .{ @tagName(reason), stats.drops.getPtrConst(reason).load(.monotonic), }) catch {}; w.print("jetstream_ingest_dropped_events_total{{source=\"live\",reason=\"{s}\"}} {d}\n", .{ @tagName(reason), stats.drops.getPtrConst(reason).load(.monotonic), }) catch {}; w.print("jetstream_ingest_dropped_events_total{{source=\"backfill\",reason=\"{s}\"}} {d}\n", .{ @tagName(reason), stats.backfill_drops.getPtrConst(reason).load(.monotonic), }) catch {}; } w.print( \\# TYPE jetstream_verifier_failures_total counter \\jetstream_verifier_failures_total{{kind="invalid"}} {d} \\jetstream_verifier_failures_total{{kind="unverified"}} {d} \\ , .{ stats.verify_invalid.load(.monotonic), stats.verify_unverified.load(.monotonic), }) catch {}; w.print( \\# TYPE jetstream_import_jobs_total counter \\jetstream_import_jobs_total{{result="ok"}} {d} \\jetstream_import_jobs_total{{result="error"}} {d} \\# TYPE jetstream_import_phase gauge \\jetstream_import_phase {d} \\# TYPE jetstream_import_rows_parsed_total counter \\jetstream_import_rows_parsed_total {d} \\# TYPE jetstream_import_rows_mutated_total counter \\jetstream_import_rows_mutated_total {d} \\# TYPE jetstream_import_rows_matched_total counter \\jetstream_import_rows_matched_total{{scope="all_versions"}} {d} \\jetstream_import_rows_matched_total{{scope="specific"}} {d} \\# TYPE jetstream_import_rows_corrupt_offset_total counter \\jetstream_import_rows_corrupt_offset_total {d} \\# TYPE jetstream_import_dids_matched_total counter \\jetstream_import_dids_matched_total {d} \\# TYPE jetstream_import_segments_examined_total counter \\jetstream_import_segments_examined_total {d} \\# TYPE jetstream_import_segments_patched_total counter \\jetstream_import_segments_patched_total {d} \\# TYPE jetstream_import_bytes_rewritten_total counter \\jetstream_import_bytes_rewritten_total {d} \\# TYPE jetstream_import_job_duration_seconds histogram \\jetstream_import_job_duration_seconds_sum {d}.{d:0>6} \\jetstream_import_job_duration_seconds_count {d} \\ , .{ stats.import_jobs_ok_total.load(.monotonic), stats.import_jobs_error_total.load(.monotonic), stats.import_phase.load(.monotonic), stats.import_rows_parsed_total.load(.monotonic), stats.import_rows_mutated_total.load(.monotonic), stats.import_rows_matched_all_total.load(.monotonic), stats.import_rows_matched_specific_total.load(.monotonic), stats.import_rows_corrupt_offset_total.load(.monotonic), stats.import_dids_matched_total.load(.monotonic), stats.import_segments_examined_total.load(.monotonic), stats.import_segments_patched_total.load(.monotonic), stats.import_bytes_rewritten_total.load(.monotonic), stats.import_job_duration_sum_us.load(.monotonic) / std.time.us_per_s, stats.import_job_duration_sum_us.load(.monotonic) % std.time.us_per_s, stats.import_job_duration_count.load(.monotonic), }) catch {}; w.print("# TYPE jetstream_import_rows_rejected_total counter\n", .{}) catch {}; inline for (comptime std.enums.values(timestamp_parse.RejectReason)) |reason| { w.print("jetstream_import_rows_rejected_total{{reason=\"{s}\"}} {d}\n", .{ @tagName(reason), stats.import_rows_rejected.getPtrConst(reason).load(.monotonic), }) catch {}; } inline for (0..16) |bucket_index| { const upper = @as(u64, 1) << @intCast(bucket_index); w.print("jetstream_import_job_duration_seconds_bucket{{le=\"{d}\"}} {d}\n", .{ upper, stats.import_job_duration_buckets[bucket_index].load(.monotonic), }) catch {}; } w.print("jetstream_import_job_duration_seconds_bucket{{le=\"+Inf\"}} {d}\n", .{ stats.import_job_duration_count.load(.monotonic), }) catch {}; // Durable repository progress. Sourced from the metadata store rather than // from process-local counters, so a restart continues the curve instead of // restarting it. Emitted only when repository state exists: absent is not // zero, and a progress panel reading zero-of-zero is worse than a gap. if (durable_repos) |repos| { w.print( \\# TYPE stream_backfill_repos_durable gauge \\stream_backfill_repos_durable{{status="complete"}} {d} \\stream_backfill_repos_durable{{status="pending"}} {d} \\stream_backfill_repos_durable{{status="failed"}} {d} \\stream_backfill_repos_durable{{status="unavailable"}} {d} \\stream_backfill_repos_durable{{status="not_started"}} {d} \\# TYPE stream_backfill_repos_durable_total gauge \\stream_backfill_repos_durable_total {d} \\ , .{ repos.complete, repos.pending, repos.failed, repos.unavailable, repos.not_started, repos.total(), }) catch {}; } return w.buffered(); } fn httpHandlerLabel(value: HttpHandler) []const u8 { return switch (value) { .root => "root", .status => "status", .subscribe => "subscribe", .subscribe_v2 => "subscribe-v2", .xrpc => "xrpc/", }; } fn httpMethodLabel(value: HttpMethod) []const u8 { return switch (value) { .get => "GET", .head => "HEAD", .post => "POST", .put => "PUT", .patch => "PATCH", .delete => "DELETE", .options => "OPTIONS", .connect => "CONNECT", .trace => "TRACE", .other => "OTHER", }; } fn httpCodeLabel(value: HttpCode) []const u8 { return switch (value) { .c200 => "200", .c201 => "201", .c206 => "206", .c304 => "304", .c400 => "400", .c401 => "401", .c404 => "404", .c405 => "405", .c409 => "409", .c412 => "412", .c416 => "416", .c500 => "500", .c501 => "501", .c503 => "503", .other => "other", }; } fn getBlockResultLabel(value: GetBlockResult) []const u8 { return switch (value) { .ok => "ok", .not_found => "not_found", .bad_request => "bad_request", .error_result => "error", }; } fn storeOpLabel(value: StoreOp) []const u8 { return switch (value) { .get => "get", .set => "set", .delete => "delete", .batch_commit => "batch_commit", }; } fn storeStatusLabel(value: StoreStatus) []const u8 { return switch (value) { .ok => "ok", .notfound => "notfound", .error_result => "error", }; } // === tests === test "http handler labels match upstream public middleware boundaries" { try std.testing.expectEqual(HttpHandler.root, httpHandler("/", "GET").?); try std.testing.expectEqual(HttpHandler.status, httpHandler("/status", "GET").?); try std.testing.expectEqual(HttpHandler.status, httpHandler("/status", "HEAD").?); try std.testing.expectEqual(HttpHandler.subscribe, httpHandler("/subscribe", "GET").?); try std.testing.expectEqual(HttpHandler.subscribe_v2, httpHandler("/xrpc/network.bsky.jetstream.subscribeEvents", "GET").?); try std.testing.expectEqual(HttpHandler.xrpc, httpHandler("/xrpc/network.bsky.jetstream.getBlock", "GET").?); try std.testing.expectEqual(HttpHandler.xrpc, httpHandler("/xrpc/not-implemented", "POST").?); try std.testing.expectEqual(null, httpHandler("/metrics", "GET")); try std.testing.expectEqual(null, httpHandler("/healthz", "GET")); try std.testing.expectEqual(null, httpHandler("/", "POST")); try std.testing.expectEqual(null, httpHandler("/not-found", "GET")); } test "getBlock metrics preserve upstream result and served-byte semantics" { var stats: Stats = .{}; stats.observeGetBlock(200, 100, 99); stats.observeGetBlock(206, 50, 100); stats.observeGetBlock(304, 0, 101); stats.observeGetBlock(400, 0, 102); stats.observeGetBlock(404, 0, 103); stats.observeGetBlock(412, 0, 104); stats.observeGetBlock(416, 0, 105); try std.testing.expectEqual(3, stats.getblock_requests.getPtrConst(.ok).load(.monotonic)); try std.testing.expectEqual(1, stats.getblock_requests.getPtrConst(.bad_request).load(.monotonic)); try std.testing.expectEqual(1, stats.getblock_requests.getPtrConst(.not_found).load(.monotonic)); try std.testing.expectEqual(2, stats.getblock_requests.getPtrConst(.error_result).load(.monotonic)); try std.testing.expectEqual(100, stats.getblock_served_bytes_total.load(.monotonic)); 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); _ = stats.verify_valid.fetchAdd(3, .monotonic); _ = stats.drops.getPtr(.invalid_rev).fetchAdd(1, .monotonic); _ = stats.backfill_drops.getPtr(.invalid_rkey).fetchAdd(2, .monotonic); _ = stats.backfill_completed_total.fetchAdd(11, .monotonic); _ = stats.ingest_blocks_flushed_total.fetchAdd(6, .monotonic); _ = stats.ingest_segments_rotated_total.fetchAdd(7, .monotonic); _ = 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); _ = stats.retry_attempts_total.fetchAdd(5, .monotonic); _ = stats.backfill_discovered_total.fetchAdd(2, .monotonic); _ = stats.backfill_active_flips_total.fetchAdd(3, .monotonic); _ = stats.backfill_on_fail_store_errors_total.fetchAdd(4, .monotonic); stats.observeBackfillHandleRepo(20_000); stats.backfill_progress_completed.store(11, .monotonic); _ = stats.backfill_completion_queued_total.fetchAdd(12, .monotonic); stats.backfill_completion_queue_depth.store(1, .monotonic); _ = stats.backfill_completion_durable_batches_total.fetchAdd(6, .monotonic); _ = stats.backfill_completion_durable_repos_total.fetchAdd(11, .monotonic); _ = stats.backfill_completion_stage_errors_total.fetchAdd(1, .monotonic); stats.observeBackfillCompletionWait(40_000); _ = stats.backfill_forced_checkpoint_flushes_total.fetchAdd(7, .monotonic); _ = stats.retry_passes_total.fetchAdd(8, .monotonic); _ = stats.retry_candidates_total.fetchAdd(9, .monotonic); _ = stats.retry_skipped_host_parked_total.fetchAdd(10, .monotonic); _ = stats.retry_pds_direct_total.fetchAdd(11, .monotonic); _ = stats.retry_relay_fallback_total.fetchAdd(12, .monotonic); stats.backfill_workers_active.store(4, .monotonic); stats.backfill_queue_depth.store(23, .monotonic); _ = 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); _ = stats.subscribe_events_skipped_sync_total.fetchAdd(7, .monotonic); _ = stats.subscribe_events_skipped_resync_total.fetchAdd(8, .monotonic); _ = stats.subscribe_encode_errors_total.fetchAdd(9, .monotonic); _ = stats.subscribe_options_update_errors.getPtr(.invalid_options).fetchAdd(10, .monotonic); _ = stats.subscribe_cursor_requests.getPtr(.seq).fetchAdd(11, .monotonic); stats.observeCursorResolve(400); _ = stats.subscribe_hot_reads_total.fetchAdd(12, .monotonic); _ = stats.subscribe_cold_reads_total.fetchAdd(13, .monotonic); _ = stats.subscribe_adversarial_drops_total.fetchAdd(14, .monotonic); _ = stats.subscribe_clean_disconnects_total.fetchAdd(15, .monotonic); _ = stats.subscribe_options_updates_total.fetchAdd(16, .monotonic); stats.observeSegmentSeal(20_000); stats.observeHttp(.xrpc, .get, 200, 12_000); stats.observeGetBlock(200, 321, 800); stats.observeGetBlock(404, 0, 200); stats.manifest_segments_loaded.store(3, .monotonic); stats.manifest_resident_bytes.store(4096, .monotonic); stats.observeCompactionRewriteAge(100_000); // 1d < 100k s <= 3d bucket stats.compaction_last_pass_unix_s.store(1_786_300_000, .monotonic); _ = stats.compaction_fold_segments_total.fetchAdd(7, .monotonic); _ = stats.manifest_block_index_cache_hits_total.fetchAdd(4, .monotonic); _ = stats.manifest_block_index_cache_misses_total.fetchAdd(5, .monotonic); stats.observeManifestLoad(400); stats.observeStore(.get, .ok, 200); stats.observeStore(.get, .notfound, 400); stats.observeStore(.set, .error_result, 800); stats.setOrchestratorPhase(.steady_state); stats.observeOrchestratorTransition(.bootstrap, .merging); stats.observeOrchestratorState(.merge, 20_000); _ = stats.orchestrator_merge_events_kept_total.fetchAdd(17, .monotonic); _ = stats.orchestrator_merge_events_dropped_total.fetchAdd(18, .monotonic); _ = stats.orchestrator_merge_segments_consumed_total.fetchAdd(19, .monotonic); _ = stats.orchestrator_merge_did_lookups_total.fetchAdd(20, .monotonic); _ = stats.orchestrator_merge_repo_revs_updated_total.fetchAdd(21, .monotonic); _ = stats.orchestrator_merge_dids_discovered_post_bootstrap_total.fetchAdd(22, .monotonic); stats.pipeline_submitted.store(20, .monotonic); stats.pipeline_claimed.store(17, .monotonic); stats.pipeline_emitted.store(11, .monotonic); stats.pipeline_fatal.store(1, .monotonic); _ = stats.pipeline_failures.getPtr(.cursor).fetchAdd(2, .monotonic); var buf: [64 * 1024]u8 = undefined; const out = format(&buf, &stats, 2, .{ .entries = 5, .bytes = 100, .pinned_bytes = 70, .pinned_overrun_bytes = 6, .base = 1, .tip = 6, }, 60, .{ .cpu_us = 1_250_000, .resident_memory_bytes = 4096, .virtual_memory_bytes = 8192, .threads = 7, .open_fds = 11, .max_fds = 1024, .minor_page_faults = 31, .major_page_faults = 2, .malloc_system_bytes = 5555, .malloc_free_total_bytes = 444, .malloc_arena_heaps = 3, .voluntary_context_switches = 41, .involuntary_context_switches = 3, }, 123_456, .{ .complete = 90, .pending = 6, .failed = 3, .unavailable = 1, .not_started = 0 }); try std.testing.expect(std.mem.indexOf(u8, out, "stream_events_total 7") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_verify_total{result=\"valid\"} 3") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_dropped_events_total{reason=\"invalid_rev\"} 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_ingest_dropped_events_total{source=\"live\",reason=\"invalid_rev\"} 1") != null); // Durable progress is the restart-stable series: it comes from the store, // not from this process's counters. try std.testing.expect(std.mem.indexOf(u8, out, "stream_backfill_repos_durable{status=\"complete\"} 90") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_backfill_repos_durable_total 100") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_ingest_dropped_events_total{source=\"backfill\",reason=\"invalid_rkey\"} 2") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_build_info{git_sha=") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_subscribers_active 2") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_ingest_readable_log_bytes 100") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_ingest_readable_log_pinned_bytes 70") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_ingest_readable_log_pinned_overrun_bytes 6") != null); try std.testing.expect(std.mem.indexOf(u8, out, "process_start_time_seconds 0") != null); try std.testing.expect(std.mem.indexOf(u8, out, "process_cpu_seconds_total 1.250000") != null); try std.testing.expect(std.mem.indexOf(u8, out, "process_resident_memory_bytes 4096") != null); try std.testing.expect(std.mem.indexOf(u8, out, "process_virtual_memory_bytes 8192") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_malloc_system_bytes 5555") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_malloc_free_total_bytes 444") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_malloc_arena_heaps 3") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_process_threads 7") != null); try std.testing.expect(std.mem.indexOf(u8, out, "process_open_fds 11") != null); try std.testing.expect(std.mem.indexOf(u8, out, "process_max_fds 1024") != null); try std.testing.expect(std.mem.indexOf(u8, out, "process_minor_page_faults_total 31") != null); try std.testing.expect(std.mem.indexOf(u8, out, "process_major_page_faults_total 2") != null); try std.testing.expect(std.mem.indexOf(u8, out, "process_voluntary_context_switches_total 41") != null); try std.testing.expect(std.mem.indexOf(u8, out, "process_involuntary_context_switches_total 3") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_data_dir_free_bytes 123456") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_segment_seal_duration_seconds_sum 0.020000") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_segment_seal_duration_seconds_bucket{le=\"0.02\"} 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_segment_seal_duration_seconds_count 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_http_request_duration_seconds_count{handler=\"xrpc/\",method=\"GET\",code=\"200\"} 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_http_request_duration_seconds_bucket{handler=\"xrpc/\",method=\"GET\",code=\"200\",le=\"0.025\"} 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_getblock_requests_total{result=\"ok\"} 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_getblock_requests_total{result=\"not_found\"} 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_getblock_served_bytes_total 321") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_getblock_duration_seconds_count 2") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_manifest_segments_loaded 3") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_manifest_resident_bytes 4096") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_compaction_rewritten_age_seconds_count 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_compaction_last_pass_timestamp_seconds 1786300000") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_compaction_fold_segments_total 7") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_compaction_rewritten_age_seconds_bucket{le=\"86400\"} 0") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_compaction_rewritten_age_seconds_bucket{le=\"259200\"} 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_compaction_rewritten_age_seconds_bucket{le=\"+Inf\"} 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_manifest_block_index_cache_hits_total 4") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_manifest_block_index_cache_misses_total 5") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_manifest_block_index_load_seconds_bucket{le=\"0.0004\"} 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_manifest_block_index_load_seconds_count 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_store_op_duration_seconds_bucket{op=\"get\",status=\"ok\",le=\"0.0002\"} 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_store_op_duration_seconds_count{op=\"get\",status=\"notfound\"} 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_store_op_duration_seconds_count{op=\"set\",status=\"error\"} 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_store_op_duration_seconds_count{op=\"delete\",status=\"error\"} 0") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_orchestrator_phase 3") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_orchestrator_phase_transitions_total{from=\"bootstrap\",to=\"merging\"} 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_orchestrator_state_duration_seconds_bucket{state=\"merge\",le=\"0.02\"} 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_orchestrator_state_duration_seconds_sum{state=\"merge\"} 0.020000") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_orchestrator_merge_events_kept_total 17") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_orchestrator_merge_events_dropped_total 18") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_orchestrator_merge_segments_consumed_total 19") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_orchestrator_merge_did_lookups_total 20") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_orchestrator_merge_repo_revs_updated_total 21") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_orchestrator_merge_dids_discovered_post_bootstrap_total 22") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_pipeline_ticket{stage=\"submitted\"} 20") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_pipeline_stage_gap{boundary=\"submitted_to_claimed\"} 3") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_pipeline_stage_gap{boundary=\"claimed_to_emitted\"} 6") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_pipeline_stage_gap{boundary=\"submitted_to_emitted\"} 9") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_pipeline_fatal 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_pipeline_failures_total{stage=\"cursor\"} 2") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_backfill_repos_total{result=\"complete\"} 11") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_backfill_workers_active 4") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_backfill_queue_depth 23") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_build_info{version=") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_completed_total 11") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_discovered_total 2") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_active_flips_total 3") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_on_fail_store_errors_total 4") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_handle_repo_duration_seconds_bucket{le=\"0.02\"} 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_handle_repo_duration_seconds_count 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_progress_completed 11") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_completion_queued_total 12") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_completion_queue_depth 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_completion_durable_batches_total 6") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_completion_durable_repos_total 11") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_completion_stage_errors_total 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_completion_queue_wait_seconds_bucket{le=\"0.04\"} 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_completion_queue_wait_seconds_count 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_forced_checkpoint_flushes_total 7") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_failed_repo_retry_passes_total 8") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_failed_repo_retry_candidates_total 9") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_backfill_failed_repo_retry_skipped_host_parked_total 10") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_backfill_failed_repo_retry_pds_direct_total 11") != null); try std.testing.expect(std.mem.indexOf(u8, out, "stream_backfill_failed_repo_retry_relay_fallback_total 12") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_ingest_blocks_flushed_total 6") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_ingest_segments_rotated_total 7") != null); 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); 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_subscribe_events_skipped_total{reason=\"sync\"} 7") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_subscribe_events_skipped_total{reason=\"resync_replacement\"} 8") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_subscribe_encode_errors_total 9") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_subscribe_options_update_errors_total{reason=\"invalid_options\"} 10") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_subscribe_cursor_requests_total{mode=\"seq\"} 11") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_subscribe_cursor_resolve_seconds_bucket{le=\"0.000400\"} 1") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_subscribe_hot_reads_total 12") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_subscribe_cold_reads_total 13") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_subscribe_adversarial_drops_total 14") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_subscribe_clean_disconnects_total 15") != null); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_subscribe_options_updates_total 16") != 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); try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_verifier_failures_total{kind=\"invalid\"}") != null); }