diff --git a/docs/grafana-dashboard.md b/docs/grafana-dashboard.md index 55e8a2a..edf35b2 100644 --- a/docs/grafana-dashboard.md +++ b/docs/grafana-dashboard.md @@ -107,6 +107,15 @@ The following dashboard families already have non-placeholder producers: - `jetstream_manifest_block_index_cache_misses_total` - `jetstream_manifest_block_index_load_seconds` - `jetstream_store_op_duration_seconds` +- `jetstream_orchestrator_phase` +- `jetstream_orchestrator_phase_transitions_total` +- `jetstream_orchestrator_state_duration_seconds` +- `jetstream_orchestrator_merge_events_kept_total` +- `jetstream_orchestrator_merge_events_dropped_total` +- `jetstream_orchestrator_merge_segments_consumed_total` +- `jetstream_orchestrator_merge_did_lookups_total` +- `jetstream_orchestrator_merge_repo_revs_updated_total` +- `jetstream_orchestrator_merge_dids_discovered_post_bootstrap_total` - `jetstream_verifier_failures_total` - `jetstream_backfill_completed_total` - `jetstream_backfill_failed_total` diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index e1508ef..90c15e6 100644 --- a/docs/semantic-parity.md +++ b/docs/semantic-parity.md @@ -20,7 +20,7 @@ this document or `bootstrap-semantic-parity.md` is open. | Sync 1.1 resync ordering | Upstream serializes per DID, drops stale async repairs, and emits a sync tombstone plus authoritative replacements | **closed at the implementation boundary.** The live coordinator has 32 real fetch workers, a 64-job queue, a five-minute fetch budget plus the 1,000 B/s-for-30-seconds slow-transfer guard, a bounded 16,384-entry 5/minute-per-DID limiter, 2,048 pending commits with drop-oldest overflow, fetch outside the DID lane, authenticated apply with older/equal-contradictory head rejection, ordered pending replay, and trigger-ticket-gated outbox delivery. Commit divergence schedules async repair; `#sync` divergence attempts inline repair and queues one retry only for transient failure, preserving the original relay cursor only on inline success. The writer emits `sync` then bounded 1,024-row `create_resync` batches and stages the fetched/pending chain checkpoints under the archive lock before any matching fsync. Loopback-only tests exercise a real HTTP getRepo mmap, real DID resolution/signature, buffering while the response is held open, slow-transfer cancellation, durable RocksDB promotion/cursor coupling, v2 tail publication, and sealed JSS row order. | | Timestamp import | Canonical dashboard and API include timestamp import lifecycle/rewrite metrics and behavior | **Closed at the implementation boundary.** The format preserves distinct `witnessed_at` and sentinel-zero `indexed_at` columns. Subscriber v1/v2 encoding applies upstream's `indexed_at != 0 ? indexed_at : witnessed_at` display rule while ranges and timestamp cursors remain anchored to immutable witness envelopes. The strict seekable RFC 4180 parser matches upstream's header, row rejection, byte-offset, CID, RFC3339, 64 KiB, and bounded-sampling contracts in a forward-only allocation-free pass. A separate Bloom-filtered RocksDB rule store bulk-loads sorted external SST chunks with CSV-order last-write-wins, reconstructs its resident collection gate on open, and implements specific-CID precedence with all-version fallback. Its central archive-lock hook stamps live, repair, bootstrap, and merge materializations before sequence assignment, and feeds the mutated rows to immediate delivery so hot and cold encodings agree. The sealed-JSS patch primitive preserves block topology/envelopes, clean frames, and the opaque bloom/collection footer while atomically changing only `indexed_at`; it is exercised against an upstream-produced segment. A concrete sealed-segment bloom catalog, bounded DID/FD bucketer, fsynced packed offsets, positioned revalidation, and collision-safe per-segment patch plan implement Phase B/C with specific-CID precedence and idempotent disk-to-disk coverage. A dedicated archive rewrite mutex serializes the real delete compactor with timestamp patching while leaving live append available. The unified metadata DB synchronously persists the current-job pointer, complete job record, and per-segment done set. The real runner activates rules before force-seal, durably crosses the bucketed handoff, checkpoints only after patch fsync/rename/dir-sync, retains partial cancellation progress, and resumes after full archive/rule/metadata reopen. The background manager canonicalizes symlinks, confines regular files, enforces one durable nonterminal job, and adopts it on startup. Bearer-gated import/status XRPC matches the pinned lexicons, fixed-width digest comparison, error names/statuses, steady-state admission, and current/by-id status behavior; a real loopback HTTP run exercised it against the local simulator. Every canonical import metric is produced at its real parse, route, patch, or durable-terminal boundary; cancellation remains a resumable pause and is not counted as terminal failure. | | Status/diagnostics | Repo attempts/error class/final PDS and per-host aggregates survive restart and are operator-queryable | **closed at the implementation boundary.** Durable v5 repo rows distinguish initial Backfill.Rev from latest Rev/UpdatedAt, retain resolved Handle/PDS and reserved record/byte fields, and atomically maintain `handle/`. Merge source cursor commits refresh latest revisions without mutating backfill watermarks. Same-batch `host/` totals, active/status counts, cumulative error classes, and five bounded recent samples survive restart; host moves and active flips are atomic with the source repo row. `/status?tab=hosts` reads those rows without a whole-network scan and matches upstream ordering and error presentation. Account queries preserve resolver-first handle semantics, the legacy `did`/`handle` aliases, and missing-identity hydration, then reconstruct records last-writer-wins from checksum-verified, DID-bloom-pruned sealed JSS plus rotation-safe active/pending and bootstrap-live snapshots. The resulting canonical MST is compared with the PDS commit root through real Sync 1.1 `getLatestCommit`/`getBlocks`; the page renders the same match fields and error state as upstream. `HEAD` performs lookup without verification; successful status responses are `no-store` and carry a generation timestamp. Expensive verification uses upstream's exact 4/source-IP/minute fixed-window limiter, 4,096-entry bound, stale pruning, oldest eviction, and explicit disable option. The offline receipt uses a real HTTP Stream route, real RocksDB row, real JSS archive, signed commit, and loopback PLC/PDS; four separate connections verify successfully, a fifth changing only its ephemeral port is blocked without another upstream request, HEAD remains side-effect-free, and disabling the limiter performs a new authoritative verification. | -| Prometheus/Grafana | Every canonical dashboard query maps to a real, same-semantics producer; runtime-specific panels use honest Zig/process metrics | exact checksum-pinned upstream dashboard is vendored with an offline fail-closed scrape contract; canonical build/live-firehose/archive-append/drop/verifier/backfill/timestamp-import producers are present. The complete canonical subscriber family set is exported from real scan, delivery, rejection, cursor, and disconnect boundaries; `none`, `deflate`, and `zstd` labels now arise only from the actual negotiated delivery path. Live frame inspection distinguishes gaps, bounded-label server errors, forward-compatible unknowns, and malformed CBOR at the raw transport boundary; a Zat reconnect callback measures attempts rather than connection failures. The only dashboard divergence is now a deterministic, checksum-pinned process-runtime row: same-semantics start/CPU/RSS/FD measurements, real VM/thread/page-fault/context-switch signals, no `go_*` aliases, and no host-network counters mislabeled as process I/O. A production-binary loopback receipt proves values and cumulative monotonicity on independent scrapes. The hot readable log now matches upstream's durability pin: rows above the archive watermark may exceed the byte budget but cannot be evicted, publication releases their pinned-byte accounting, and only then may retention advance the floor; all three canonical readable-log gauges come from that exact state. Data-directory free bytes are collected fresh from the archive directory's filesystem at scrape time, and the canonical seal histogram measures only successful flush-through-footer/header-fsync work with upstream's exact 15 slow-latency buckets. HTTP duration uses upstream's public-mux boundary and stable labels, including inline WebSocket lifetime and the single `xrpc/` subtree; debug and unmatched routes remain unobserved. getBlock outcomes, duration, and served bytes come from actual response completion, with bytes counted only for complete 200 responses. A ReleaseSafe offline receipt proves 200/206/304/400/404/416 behavior and WebSocket close accounting. Sealed headers and block indexes are now genuinely resident: startup and verified refresh latency, loaded count, and lookup hits/misses use upstream's exact boundaries. Refcounted snapshots remain memory-safe across refresh, while fresh-file offsets prevent cross-generation splicing; the pinned official client proves cold replay produces hits. RocksDB point reads, durable point writes/deletes, and atomic batch commits populate upstream's exact `{op,status}` store histogram and fast-latency buckets; real-database and production-startup receipts prove the non-error paths. **Remaining resident bloom/collection planning, orchestrator, backfill, and compaction families open.** | +| Prometheus/Grafana | Every canonical dashboard query maps to a real, same-semantics producer; runtime-specific panels use honest Zig/process metrics | exact checksum-pinned upstream dashboard is vendored with an offline fail-closed scrape contract; canonical build/live-firehose/archive-append/drop/verifier/backfill/timestamp-import producers are present. The complete canonical subscriber family set is exported from real scan, delivery, rejection, cursor, and disconnect boundaries; `none`, `deflate`, and `zstd` labels now arise only from the actual negotiated delivery path. Live frame inspection distinguishes gaps, bounded-label server errors, forward-compatible unknowns, and malformed CBOR at the raw transport boundary; a Zat reconnect callback measures attempts rather than connection failures. The only dashboard divergence is now a deterministic, checksum-pinned process-runtime row: same-semantics start/CPU/RSS/FD measurements, real VM/thread/page-fault/context-switch signals, no `go_*` aliases, and no host-network counters mislabeled as process I/O. A production-binary loopback receipt proves values and cumulative monotonicity on independent scrapes. The hot readable log now matches upstream's durability pin: rows above the archive watermark may exceed the byte budget but cannot be evicted, publication releases their pinned-byte accounting, and only then may retention advance the floor; all three canonical readable-log gauges come from that exact state. Data-directory free bytes are collected fresh from the archive directory's filesystem at scrape time, and the canonical seal histogram measures only successful flush-through-footer/header-fsync work with upstream's exact 15 slow-latency buckets. HTTP duration uses upstream's public-mux boundary and stable labels, including inline WebSocket lifetime and the single `xrpc/` subtree; debug and unmatched routes remain unobserved. getBlock outcomes, duration, and served bytes come from actual response completion, with bytes counted only for complete 200 responses. A ReleaseSafe offline receipt proves 200/206/304/400/404/416 behavior and WebSocket close accounting. Sealed headers and block indexes are now genuinely resident: startup and verified refresh latency, loaded count, and lookup hits/misses use upstream's exact boundaries. Refcounted snapshots remain memory-safe across refresh, while fresh-file offsets prevent cross-generation splicing; the pinned official client proves cold replay produces hits. RocksDB point reads, durable point writes/deletes, and atomic batch commits populate upstream's exact `{op,status}` store histogram and fast-latency buckets; real-database and production-startup receipts prove the non-error paths. Durable lifecycle phase/transition metrics, all six cutover-state timings, and merge counters are wired to the actual bootstrap/merge state machine. Merge now performs one cached Backfill.Rev lookup per DID, including misses; real JSS/RocksDB and eight-crash production receipts prove counter and restart behavior. **Remaining resident bloom/collection planning, backfill, and compaction families open.** | | Oracle | Event-log equivalence, final-state convergence, crash/power-loss, verifier repair, hostile input, and anti-vacuity receipts | **open — current crash matrix is necessary but substantially smaller than upstream's oracle** | ## Experiment admission diff --git a/docs/upstream-bootstrap-spec.md b/docs/upstream-bootstrap-spec.md index 0f5e36c..97c2c15 100644 --- a/docs/upstream-bootstrap-spec.md +++ b/docs/upstream-bootstrap-spec.md @@ -58,7 +58,9 @@ starts concurrently with backfill; no tombstone machinery; not user-visible. open dst on segments/ (real seq space); drain sources in index order with merge/next_source_idx cursor; keep/drop per row: drop iff DID complete AND row.Rev <= Backfill.Rev (lexicographic TID); re-stamp witnessed_at = now; -per-source atomic commit (cursor + repo Rev refresh). then: pending-repo +the Backfill.Rev lookup is cached once per DID for the complete merge run, +including missing rows; per-source atomic commit (cursor + repo Rev refresh). +then: pending-repo retry pass (HandleRepoResync: KindSync tombstone + KindCreateResync rows, pre-validated), dst seal, merge-tail delete/update compaction, discovery from bootstrap/last_listrepos_cursor (unknown DIDs → failed rows for steady retry), @@ -96,4 +98,8 @@ until steady_state. failed-repo retry loop: 4h interval, 16 workers, 4/host, - the segment durable hook fsyncs event bytes before committing `seq/next`, `seq/durable`, and the covered relay cursor; the durable watermark is published only after that metadata batch succeeds. +- canonical orchestrator phase, transition, state-duration, and merge counters + are updated only at these durable lifecycle boundaries. The offline crash + matrix checks their restart semantics, while a deterministic real-JSS and + real-RocksDB merge receipt proves kept/dropped/source/lookup/revision counts. - reproduce crashpoint seams at every commit boundary for future oracle tests. diff --git a/docs/upstream-harness.md b/docs/upstream-harness.md index 010da96..be98ab9 100644 --- a/docs/upstream-harness.md +++ b/docs/upstream-harness.md @@ -41,7 +41,12 @@ restarts clean, asserts convergence: serving ungates, listSegments non-empty, /subscribe-v2 replay seq-monotonic, serving proves the RocksDB phase reached steady_state, backfill tree cleaned, no legacy metadata stores reappear, and interrupted repos are repaired by the pending pass (anti-vacuity via log -markers). A separate injected metadata-commit failure test proves the archive +markers). Every restart also scrapes the production metrics endpoint: phase 3, +the real merging→steady transition, merge/write-state histograms, and all six +registered merge counters are mandatory; bootstrap-resume cases additionally +prove bootstrap→merging plus drain/seal/close/write timings, and the interrupted +download case must perform nonzero source, DID-lookup, and event work. A +separate injected metadata-commit failure test proves the archive does not publish or strand a segment durability boundary. NOT yet ported from upstream's oracle: power-loss vfs, event-log equivalence vs simulator ground truth, mutation campaign. diff --git a/src/internal/bootstrap/engine.zig b/src/internal/bootstrap/engine.zig index bf4b1b4..4dd9009 100644 --- a/src/internal/bootstrap/engine.zig +++ b/src/internal/bootstrap/engine.zig @@ -553,8 +553,15 @@ pub fn run( if (config.max_repos == 0) try saveListReposCursor(store.meta, next); } - try archive.flushBlock(); - try store.flush(); + { + const close_started = Io.Timestamp.now(io, .awake); + defer if (config.metrics) |m| m.observeOrchestratorState(.close_backfill, @intCast(@max( + close_started.durationTo(Io.Timestamp.now(io, .awake)).toMicroseconds(), + 0, + ))); + try archive.flushBlock(); + try store.flush(); + } if (config.max_repos == 0 and config.selected_repos.len == 0) try markListReposComplete(store.meta); work.publishGauges(); return stats; diff --git a/src/internal/bootstrap/lifecycle.zig b/src/internal/bootstrap/lifecycle.zig index 7909dc5..ce5db61 100644 --- a/src/internal/bootstrap/lifecycle.zig +++ b/src/internal/bootstrap/lifecycle.zig @@ -77,6 +77,7 @@ pub fn run( try phase_store.write(io, .bootstrap); break :blk phase_mod.Phase.bootstrap; }; + stats.setOrchestratorPhase(orchestratorPhase(phase)); if (phase == .steady_state) return; // watermark floor (upstream initCompactionWatermarkFloor, when the @@ -97,10 +98,18 @@ pub fn run( if (phase == .bootstrap) { try runBootstrap(allocator, io, opts, live_data_dir, dst, cursor_store, verifier, stats, &repo_st); crashpoint.hit(.after_backfill_before_phase_merging); + const write_started = Io.Timestamp.now(io, .awake); try phase_store.write(io, .merging); + stats.observeOrchestratorState(.write_phase_merging, elapsedUs(io, write_started)); + stats.observeOrchestratorTransition(.bootstrap, .merging); + stats.setOrchestratorPhase(.merging); phase = .merging; } + const merge_started = Io.Timestamp.now(io, .awake); + var merge_observed = false; + defer if (!merge_observed) stats.observeOrchestratorState(.merge, elapsedUs(io, merge_started)); + // merging: restart-after-cleanup guard — live dir gone means a prior // run finished the drain and cleanup but died before the phase write var backfill_dir = Io.Dir.cwd().createDirPathOpen(io, opts.data_dir, .{}) catch unreachable; @@ -116,7 +125,7 @@ pub fn run( var bf_dir = try Io.Dir.cwd().openDir(io, bf_path, .{}); defer bf_dir.close(io); - const m = try merge.run(allocator, io, live_data_dir, dst, &repo_st); + const m = try merge.run(allocator, io, live_data_dir, dst, &repo_st, stats); log.info("merge: {d} kept, {d} dropped across {d} sources", .{ m.kept, m.dropped, m.sources }); try dst.flushBlock(); crashpoint.hit(.after_merge_drain_before_pending); @@ -156,7 +165,7 @@ pub fn run( // become failed rows the steady retry loop downloads. a relay // outage here fails the merge (upstream couples them the same // way) — re-entry is idempotent. - try runDiscovery(allocator, io, opts.relay_http, &repo_st); + try runDiscovery(allocator, io, opts.relay_http, &repo_st, stats); crashpoint.hit(.after_discovery_before_cleanup); // cleanup: remove the backfill tree (including the discovery @@ -184,7 +193,13 @@ pub fn run( writes.put("seq/durable", &durable_bytes); try writes.commitDurable(); } + stats.observeOrchestratorState(.merge, elapsedUs(io, merge_started)); + merge_observed = true; + const write_started = Io.Timestamp.now(io, .awake); try phase_store.write(io, .steady_state); + stats.observeOrchestratorState(.write_phase_steady, elapsedUs(io, write_started)); + stats.observeOrchestratorTransition(.merging, .steady_state); + stats.setOrchestratorPhase(.steady_state); } /// resync every pending repo into `dst`. per-repo failure defers to @@ -254,6 +269,7 @@ fn runDiscovery( io: Io, relay_http: []const u8, repo_st: *repo_store_mod.Store, + stats: *metrics.Stats, ) !void { const start = (try engine.loadListReposCursor(repo_st.meta, allocator)) orelse { log.info("merge discovery: no bootstrap cursor, skipping", .{}); @@ -284,6 +300,7 @@ fn runDiscovery( } try repo_st.put(ref.did, .{ .status = .failed }); discovered += 1; + _ = stats.orchestrator_merge_dids_discovered_post_bootstrap_total.fetchAdd(1, .monotonic); } const next = page.cursor orelse break; if (seen.contains(next)) return error.ListReposCursorLoop; @@ -402,16 +419,32 @@ fn runBootstrap( // cancel unblocks an idle read), then seal its archive so the merge // reads a fully-sealed source tree (the merge's seal guard covers a // crash landing between these two steps) + const drain_started = Io.Timestamp.now(io, .awake); capture.requestStop(); _ = capture_future.cancel(io); + stats.observeOrchestratorState(.drain_bootstrap, elapsedUs(io, drain_started)); // cancel joins the websocket reader, so no new submissions can race this // boundary. Drain every frame it already handed to the pipeline before // closing the archive those callbacks write to. + const seal_started = Io.Timestamp.now(io, .awake); try pipe.drainAndStop(); crashpoint.hit(.after_live_close_before_seal); live_archive.sealAndClose() catch |err| { log.warn("bootstrap: live archive seal failed: {s}", .{@errorName(err)}); return err; }; + stats.observeOrchestratorState(.seal_bootstrap, elapsedUs(io, seal_started)); live_sealed = true; } + +fn orchestratorPhase(phase: phase_mod.Phase) metrics.OrchestratorPhase { + return switch (phase) { + .bootstrap => .bootstrap, + .merging => .merging, + .steady_state => .steady_state, + }; +} + +fn elapsedUs(io: Io, started_at: Io.Timestamp) u64 { + return @intCast(@max(started_at.durationTo(Io.Timestamp.now(io, .awake)).toMicroseconds(), 0)); +} diff --git a/src/internal/bootstrap/merge.zig b/src/internal/bootstrap/merge.zig index 8859330..f690dc6 100644 --- a/src/internal/bootstrap/merge.zig +++ b/src/internal/bootstrap/merge.zig @@ -16,6 +16,7 @@ const std = @import("std"); const archive_mod = @import("../archive.zig"); +const metrics = @import("../metrics.zig"); const repo_store = @import("repo_store.zig"); const segment = @import("../segment.zig"); @@ -29,6 +30,35 @@ pub const Stats = struct { sources: u64 = 0, }; +/// One RocksDB repo-row lookup per DID for the entire merge run, including +/// cached misses. Backfill.Rev is immutable during merge, so cached states +/// remain the authoritative filter watermark while source commits refresh +/// only the separate latest-revision fields. +const RepoStatusCache = struct { + arena: std.heap.ArenaAllocator, + states: std.StringHashMapUnmanaged(?repo_store.RepoState) = .empty, + store: *repo_store.Store, + stats: *metrics.Stats, + + fn init(allocator: Allocator, store: *repo_store.Store, stats: *metrics.Stats) RepoStatusCache { + return .{ .arena = .init(allocator), .store = store, .stats = stats }; + } + + fn deinit(self: *RepoStatusCache) void { + self.arena.deinit(); + } + + fn get(self: *RepoStatusCache, did: []const u8) !?repo_store.RepoState { + if (self.states.get(did)) |state| return state; + const allocator = self.arena.allocator(); + const owned_did = try allocator.dupe(u8, did); + const state = try self.store.get(allocator, did); + try self.states.put(allocator, owned_did, state); + _ = self.stats.orchestrator_merge_did_lookups_total.fetchAdd(1, .monotonic); + return state; + } +}; + /// true unless the row is superseded by the backfill watermark pub fn shouldKeep(ev: segment.Event, state: ?repo_store.RepoState) bool { switch (ev.kind) { @@ -49,6 +79,7 @@ pub fn run( live_data_dir: []const u8, dst: *archive_mod.Archive, store: *repo_store.Store, + metric_stats: *metrics.Stats, ) !Stats { // seal guard: a crash between live-consumer teardown and seal leaves // an active trailing source. Archive recovery seals torn actives; @@ -74,6 +105,8 @@ pub fn run( const from_idx = loadCursor(store); var stats: Stats = .{}; + var cache = RepoStatusCache.init(allocator, store, metric_stats); + defer cache.deinit(); var expect = from_idx; for (indices.items) |idx| { if (idx < from_idx) continue; @@ -82,7 +115,7 @@ pub fn run( var source_arena = std.heap.ArenaAllocator.init(allocator); defer source_arena.deinit(); - const revs = try drainOne(source_arena.allocator(), io, src_dir, idx, dst, store, &stats); + const revs = try drainOne(source_arena.allocator(), io, src_dir, idx, dst, &cache, &stats, metric_stats); // durability ordering: dst rows fsynced before the cursor advances try dst.flushBlock(); const updates = try source_arena.allocator().alloc(repo_store.RevUpdate, revs.count()); @@ -92,6 +125,8 @@ pub fn run( updates[update_index] = .{ .did = entry.key_ptr.*, .rev = entry.value_ptr.* }; try store.commitMergeSource(cursor_key, idx + 1, updates, Io.Timestamp.now(io, .real).toMicroseconds()); stats.sources += 1; + _ = metric_stats.orchestrator_merge_segments_consumed_total.fetchAdd(1, .monotonic); + _ = metric_stats.orchestrator_merge_repo_revs_updated_total.fetchAdd(updates.len, .monotonic); } return stats; } @@ -102,8 +137,9 @@ fn drainOne( src_dir: Io.Dir, idx: u64, dst: *archive_mod.Archive, - store: *repo_store.Store, + cache: *RepoStatusCache, stats: *Stats, + metric_stats: *metrics.Stats, ) !std.StringHashMapUnmanaged([]const u8) { var name_buf: [64]u8 = undefined; const name = archive_mod.formatSegmentName(&name_buf, idx); @@ -118,16 +154,17 @@ fn drainOne( var block = try sealed.readBlock(allocator, i); defer block.deinit(allocator); for (block.events) |ev| { - const state = try store.get(allocator, ev.did); - defer if (state) |s| store.freeState(allocator, s); + const state = try cache.get(ev.did); if (!shouldKeep(ev, state)) { stats.dropped += 1; + _ = metric_stats.orchestrator_merge_events_dropped_total.fetchAdd(1, .monotonic); continue; } var stamped = ev; stamped.witnessed_at = now_us; // upstream §3.4 re-stamp _ = try dst.append(stamped, now_us); stats.kept += 1; + _ = metric_stats.orchestrator_merge_events_kept_total.fetchAdd(1, .monotonic); switch (ev.kind) { .create, .update, .delete, .create_resync, .sync => if (ev.rev.len > 0) { const result = try per_did.getOrPut(allocator, ev.did); @@ -188,3 +225,76 @@ test "shouldKeep mirrors upstream merge_filter" { try testing.expect(shouldKeep(row(.create, "did:plc:x", ""), complete)); try testing.expect(shouldKeep(row(.create, "did:plc:x", "3l3qo2vutsw2a"), .{ .status = .complete, .rev = "" })); } + +test "merge repo status cache performs one real RocksDB lookup per DID including misses" { + const meta_store = @import("../meta_store.zig"); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var path_buf: [std.fs.max_path_bytes]u8 = undefined; + const path_len = try tmp.dir.realPath(testing.io, &path_buf); + var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]); + defer meta.deinit(); + var store = try repo_store.Store.init(testing.allocator, testing.io, &meta); + defer store.deinit(); + try store.put("did:plc:known", .{ .status = .complete, .rev = "3lknown" }); + + var metric_stats: metrics.Stats = .{}; + var cache = RepoStatusCache.init(testing.allocator, &store, &metric_stats); + defer cache.deinit(); + try testing.expectEqualStrings("3lknown", (try cache.get("did:plc:known")).?.rev); + try testing.expectEqualStrings("3lknown", (try cache.get("did:plc:known")).?.rev); + try testing.expectEqual(null, try cache.get("did:plc:missing")); + try testing.expectEqual(null, try cache.get("did:plc:missing")); + try testing.expectEqual(2, metric_stats.orchestrator_merge_did_lookups_total.load(.monotonic)); +} + +test "real merge records canonical counters at durable source boundaries" { + const meta_store = @import("../meta_store.zig"); + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var root_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const root_len = try tmp.dir.realPath(io, &root_buf); + const root = root_buf[0..root_len]; + var live_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const live_root = try std.fmt.bufPrint(&live_buf, "{s}/live", .{root}); + var dst_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const dst_root = try std.fmt.bufPrint(&dst_buf, "{s}/dst", .{root}); + + var meta = try meta_store.Store.open(testing.allocator, root); + defer meta.deinit(); + var store = try repo_store.Store.init(testing.allocator, io, &meta); + defer store.deinit(); + try store.put("did:plc:known", .{ .status = .complete, .rev = "3lmiddle" }); + + var source = try archive_mod.Archive.init(testing.allocator, io, live_root); + _ = try source.append(row(.create, "did:plc:known", "3la"), 1); + _ = try source.append(row(.update, "did:plc:known", "3lz"), 2); + _ = try source.append(row(.identity, "did:plc:known", ""), 3); + _ = try source.append(row(.create, "did:plc:unknown", "3lx"), 4); + try source.sealAndClose(); + + var dst = try archive_mod.Archive.init(testing.allocator, io, dst_root); + defer { + dst.close(); + dst.deinit(); + } + var metric_stats: metrics.Stats = .{}; + const result = try run(testing.allocator, io, live_root, &dst, &store, &metric_stats); + try testing.expectEqual(@as(u64, 3), result.kept); + try testing.expectEqual(@as(u64, 1), result.dropped); + try testing.expectEqual(@as(u64, 1), result.sources); + try testing.expectEqual(3, metric_stats.orchestrator_merge_events_kept_total.load(.monotonic)); + try testing.expectEqual(1, metric_stats.orchestrator_merge_events_dropped_total.load(.monotonic)); + try testing.expectEqual(1, metric_stats.orchestrator_merge_segments_consumed_total.load(.monotonic)); + try testing.expectEqual(2, metric_stats.orchestrator_merge_did_lookups_total.load(.monotonic)); + try testing.expectEqual(2, metric_stats.orchestrator_merge_repo_revs_updated_total.load(.monotonic)); + + const updated = (try store.get(testing.allocator, "did:plc:known")).?; + defer store.freeState(testing.allocator, updated); + try testing.expectEqualStrings("3lz", updated.latest_rev); + try testing.expectEqualStrings("3lmiddle", updated.rev); + try testing.expectEqual(null, try store.get(testing.allocator, "did:plc:unknown")); +} diff --git a/src/internal/metrics.zig b/src/internal/metrics.zig index c5cbf98..3bfd6e8 100644 --- a/src/internal/metrics.zig +++ b/src/internal/metrics.zig @@ -32,11 +32,21 @@ pub const HttpCode = enum { c200, c201, c206, c304, c400, c401, c404, c405, c409 pub const GetBlockResult = enum { ok, not_found, bad_request, error_result }; 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, +}; const http_series_count = @typeInfo(HttpHandler).@"enum".fields.len * @typeInfo(HttpMethod).@"enum".fields.len * @typeInfo(HttpCode).@"enum".fields.len; const store_series_count = @typeInfo(StoreOp).@"enum".fields.len * @typeInfo(StoreStatus).@"enum".fields.len; +const orchestrator_phase_count = @typeInfo(OrchestratorPhase).@"enum".fields.len; pub const Stats = struct { start_time_s: i64 = 0, @@ -67,6 +77,17 @@ pub const Stats = struct { 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".fields.len]Counter = @splat(.init(0)), + orchestrator_state_duration_sum_us: [@typeInfo(OrchestratorState).@"enum".fields.len]Counter = @splat(.init(0)), + orchestrator_state_duration_buckets: [@typeInfo(OrchestratorState).@"enum".fields.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)), verify_valid: Counter = .init(0), @@ -218,12 +239,35 @@ pub const Stats = struct { _ = self.store_op_duration_buckets[index][bucket_index].fetchAdd(1, .monotonic); } } + + pub fn setOrchestratorPhase(self: *Stats, phase: OrchestratorPhase) void { + self.orchestrator_phase.store(@intFromEnum(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 = @intFromEnum(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); + } + } }; fn storeIndex(op: StoreOp, status: StoreStatus) usize { return @intFromEnum(op) * @typeInfo(StoreStatus).@"enum".fields.len + @intFromEnum(status); } +fn orchestratorTransitionIndex(from: OrchestratorPhase, to: OrchestratorPhase) usize { + return @intFromEnum(from) * orchestrator_phase_count + @intFromEnum(to); +} + fn httpIndex(handler: HttpHandler, method: HttpMethod, code: HttpCode) usize { const method_count = @typeInfo(HttpMethod).@"enum".fields.len; const code_count = @typeInfo(HttpCode).@"enum".fields.len; @@ -570,6 +614,72 @@ pub fn format( } } + 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 = @intFromEnum(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. @@ -1029,6 +1139,15 @@ test "format renders counters" { 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); var buf: [64 * 1024]u8 = undefined; const out = format(&buf, &stats, 2, .{ @@ -1088,6 +1207,16 @@ test "format renders counters" { 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_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); diff --git a/src/main.zig b/src/main.zig index e446fe9..bd514e0 100644 --- a/src/main.zig +++ b/src/main.zig @@ -279,6 +279,11 @@ pub fn main(init: std.process.Init.Minimal) !void { defer ps.deinit(io); break :blk ps.read(io); }; + if (stored_phase) |phase| stats.setOrchestratorPhase(switch (phase) { + .bootstrap => .bootstrap, + .merging => .merging, + .steady_state => .steady_state, + }); const mid_lifecycle = if (stored_phase) |ph| ph != .steady_state else false; if (mid_lifecycle or do_backfill or backfill_max_repos > 0 or backfill_repos.len > 0) { hub.serving.store(false, .release); diff --git a/tests/oracle.py b/tests/oracle.py index 53f3b36..7989226 100644 --- a/tests/oracle.py +++ b/tests/oracle.py @@ -49,6 +49,9 @@ ARGS = [ "--retry-interval=0", ] +SAMPLE_RE = re.compile(r"^([a-zA-Z_:][a-zA-Z0-9_:]*)(?:\{([^}]*)\})?\s+([^\s]+)$") +LABEL_RE = re.compile(r'(\w+)="([^"]*)"') + def http_status(path): try: @@ -73,6 +76,25 @@ def wait_serving(proc, timeout=120): raise AssertionError("timed out waiting for steady_state serving") +def metrics_snapshot(): + st, body = http_status("/metrics") + assert st == 200, st + samples = {} + for line in body.decode().splitlines(): + if line.startswith("#"): + continue + match = SAMPLE_RE.match(line) + if not match: + continue + name, labels, value = match.groups() + samples[(name, tuple(sorted(LABEL_RE.findall(labels or ""))))] = float(value) + return samples + + +def metric(samples, name, **labels): + return samples.get((name, tuple(sorted(labels.items()))), 0.0) + + async def replay_invariants(n): """seqs strictly increasing from a cold cursor; frames well-formed""" url = f"ws://127.0.0.1:{PORT}/subscribe-v2?cursor=0" @@ -114,6 +136,60 @@ def run_case(point): st, _ = http_status("/healthz") assert st == 200, st + metrics = metrics_snapshot() + assert metric(metrics, "jetstream_orchestrator_phase") == 3 + assert metric( + metrics, + "jetstream_orchestrator_phase_transitions_total", + **{"from": "merging", "to": "steady_state"}, + ) == 1 + for state in ("merge", "write_phase_steady"): + assert metric( + metrics, + "jetstream_orchestrator_state_duration_seconds_count", + state=state, + ) == 1 + merge_counters = ( + "jetstream_orchestrator_merge_events_kept_total", + "jetstream_orchestrator_merge_events_dropped_total", + "jetstream_orchestrator_merge_segments_consumed_total", + "jetstream_orchestrator_merge_did_lookups_total", + "jetstream_orchestrator_merge_repo_revs_updated_total", + "jetstream_orchestrator_merge_dids_discovered_post_bootstrap_total", + ) + for name in merge_counters: + assert (name, ()) in metrics, f"missing registered orchestrator counter {name}" + + if point in { + "mid_backfill_download", + "after_backfill_before_phase_merging", + "after_live_close_before_seal", + }: + assert metric( + metrics, + "jetstream_orchestrator_phase_transitions_total", + **{"from": "bootstrap", "to": "merging"}, + ) == 1 + for state in ( + "drain_bootstrap", + "seal_bootstrap", + "close_backfill", + "write_phase_merging", + ): + assert metric( + metrics, + "jetstream_orchestrator_state_duration_seconds_count", + state=state, + ) == 1 + + if point == "mid_backfill_download": + assert metric(metrics, "jetstream_orchestrator_merge_segments_consumed_total") > 0 + assert metric(metrics, "jetstream_orchestrator_merge_did_lookups_total") > 0 + assert ( + metric(metrics, "jetstream_orchestrator_merge_events_kept_total") + + metric(metrics, "jetstream_orchestrator_merge_events_dropped_total") + ) > 0 + # A fresh simulator has small repos; older simulator state can have # thousands of rows. Verify the complete replay when it is small and a # bounded 200-event prefix when it is large. Waiting for an arbitrary @@ -140,6 +216,21 @@ def run_case(point): logged = pathlib.Path(f"/tmp/oracle-{point}.log").read_text() match = re.search(r"pending-repo pass: (\d+) resynced", logged) assert match and int(match.group(1)) > 0, "pending pass never repaired interrupted repos" + + # A process born directly into durable steady_state must restore + # phase=3 without inventing transitions that occurred in an older + # process lifetime. + p.send_signal(signal.SIGTERM) + p.wait(timeout=15) + p = subprocess.Popen(ARGS, stdout=log, stderr=log) + wait_serving(p) + restarted_metrics = metrics_snapshot() + assert metric(restarted_metrics, "jetstream_orchestrator_phase") == 3 + assert metric( + restarted_metrics, + "jetstream_orchestrator_phase_transitions_total", + **{"from": "merging", "to": "steady_state"}, + ) == 0 finally: p.send_signal(signal.SIGTERM) try: