Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826//! 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);}