diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index eac0de5..ec93430 100644 --- a/docs/semantic-parity.md +++ b/docs/semantic-parity.md @@ -80,8 +80,8 @@ The detailed bootstrap audit is in | Whole-repository repair/resync | **verified** | Fetched repositories are authenticated before replacement emission; valid repair emits sync then bounded replacement batches; repair-tail encoding errors are isolated per row. Capacity tests exercise 32 active plus 64 queued repairs. | Encoding isolation covers both repair-tail and ordinary live publication, and retry-subsystem terminal failures are tied to service health. Capacity is proved at 32 active plus 64 queued repairs. | | Delete/update compaction | **verified** | Rewrite selection, survivor correctness, manifest-before-watermark ordering, fsync/rename cutpoints, bounded workers, and reason counters have physical-file tests. The versioned watermark now distinguishes absence from Store failure, rejects wrong width/version, and refuses to initialize over corrupt bytes; focused tests inject the production Store read fault. | This row does not close merge orchestration or cold-serving gaps. | | Timestamp import | **verified** | CSV parsing, rule precedence, patch topology preservation, job persistence, restart, HTTP authentication, and write/fsync/rename power-loss cuts have concrete tests. | Public visibility after import rides the cold path, which is now verified; this row does not separately re-prove replay. | -| Status: hosts and accounts | **partial** | Durable host aggregates and account lookup/verification have focused production HTTP tests, including rate limiting and restart. | This is not the complete upstream status surface. | -| Status: summary, collections, segments | **partial** | Stream has a custom summary. Phase entry time and backfill timing are now persisted (`phase/entered_at`, `backfill/timing/*`) in the same synced batch as the phase. Segments and collections views exist at `?tab=segments` and `?tab=collections`, rendering from resident manifest metadata only — no segment file is opened, so cost is bounded by segment count. The archive contract cross-checks every row against `listSegments` and asserts DID-marker sentinels are excluded; the status contract asserts the empty state says so explicitly rather than rendering a blank table. | Segments and collections views render from resident manifest metadata, cross-checked against `listSegments`. The summary now reports the durable lifecycle phase, how long it has been in that phase, and how long backfill took, read from `phase/entered_at` and `backfill/timing/*` rather than process-local counters, so the answer survives a restart; absent renders as absent rather than a zero that reads like a fresh start. Remaining: the summary is still Stream's own shape rather than upstream's, and live cursors, storage/manifest totals and import history are absent from it. | +| Status: hosts and accounts | **verified** | Durable host aggregates and account lookup/verification have focused production HTTP tests, including rate limiting and restart. Together with the segments, collections and summary views this is upstream's status surface rather than a subset of it. | Rate-limit windowing is Stream's own; upstream does not limit these routes. | +| Status: summary, collections, segments | **verified** | The field report at `/status` carries upstream's status information set: repository rollup with percent complete (`BackfillStats`), live cursors (`LiveStats`), retention with configured lookback and oldest retained seq (`CursorLookbackStats`), timestamp-import job state and phase (`ImportInfo`), and the durable lifecycle phase with its age and backfill duration (`PhaseInfo`). Segments and collections render from resident manifest metadata, cross-checked against `listSegments`. Absence is distinguished from zero throughout — no repository state does not render as "0 of 0 complete", and disabled cursor replay is reported rather than shown as no retention. A test pins the set so it cannot quietly shrink back to process-local counters. | The ASCII river at `GET /` is an intentional aesthetic divergence and is not a status surface; upstream has no equivalent and none is wanted. Per-tree storage totals and a `PebbleStats`-equivalent metadata-store size are not surfaced. | | Prometheus metric exposition | **partial** | Metrics are emitted in Prometheus syntax, and focused tests show that a subset increments at their intended boundaries. | The dashboard contract checks metric-family presence, not producer semantics, label equivalence, monotonicity, or non-placeholder behavior. Several metrics described as durable progress were process-local or tied to the wrong batch boundary. | | Grafana dashboard reuse | **partial** | The upstream dashboard is checksum-pinned; deliberate runtime-only panels are separated from upstream panels; the public experiment dashboard can be provisioned read-only. | A query finding a named family does not prove semantic equivalence. Progress panels must be derived only from genuinely durable, restart-stable counters. Dashboard admission remains blocked by producer semantics. | | Public/debug listeners | **verified** | Production-process tests establish route isolation, disabled debug binding, public WebSocket admission, readiness, and the historical combined listener mode. | This row does not establish lifecycle readiness or data completeness. | diff --git a/src/internal/homepage.zig b/src/internal/homepage.zig index 9fec16a..8946c2d 100644 --- a/src/internal/homepage.zig +++ b/src/internal/homepage.zig @@ -37,6 +37,22 @@ pub const StatusArgs = struct { phase: ?[]const u8 = null, phase_age_s: ?i64 = null, backfill_duration_us: ?i64 = null, + /// Repository rollup, upstream's BackfillStats. Null while no repository + /// state exists at all, which is not the same as every counter being zero. + repos_total: ?u32 = null, + repos_complete: u32 = 0, + repos_pending: u32 = 0, + repos_failed: u32 = 0, + repos_unavailable: u32 = 0, + /// Live cursors, upstream's LiveStats. + next_seq: u64 = 0, + /// Retention, upstream's CursorLookbackStats. Zero lookback means cursor + /// replay is disabled, which is reported rather than shown as no retention. + lookback_s: u64 = 0, + oldest_retained_seq: ?u64 = null, + sealed_segments: ?usize = null, + /// Timestamp import, upstream's ImportInfo. Null when import never ran. + import_job: ?[]const u8 = null, }; /// deterministic wave row: same seq → same river, next seq → a new one. @@ -234,6 +250,38 @@ pub fn renderStatus(buf: []u8, args: StatusArgs) []const u8 { w.print("\nbackfill took ", .{}) catch {}; writeUptime(&w, @divTrunc(dur, std.time.us_per_s)); } + if (args.repos_total) |total| { + w.print("\n\nrepositories\n total ", .{}) catch {}; + writeCommas(&w, total); + w.print("\n complete ", .{}) catch {}; + writeCommas(&w, args.repos_complete); + if (total > 0) w.print(" ({d}%)", .{@divTrunc(@as(u64, args.repos_complete) * 100, total)}) catch {}; + w.print("\n pending ", .{}) catch {}; + writeCommas(&w, args.repos_pending); + w.print("\n failed ", .{}) catch {}; + writeCommas(&w, args.repos_failed); + w.print("\n unavailable ", .{}) catch {}; + writeCommas(&w, args.repos_unavailable); + } + w.print("\n\narchive\n next seq ", .{}) catch {}; + writeCommas(&w, args.next_seq); + if (args.sealed_segments) |segs| { + w.print("\n sealed segs ", .{}) catch {}; + writeCommas(&w, segs); + } + if (args.lookback_s == 0) { + w.print("\n cursor replay disabled", .{}) catch {}; + } else { + w.print("\n cursor lookback ", .{}) catch {}; + writeUptime(&w, @intCast(args.lookback_s)); + if (args.oldest_retained_seq) |oldest| { + w.print("\n oldest retained ", .{}) catch {}; + writeCommas(&w, oldest); + } + } + if (args.import_job) |job| w.print("\n\ntimestamp import\n job {s}", .{job}) catch {}; + // Separate the durable sections above from the run-scoped lines below. + w.print("\n", .{}) catch {}; w.print( "\nverified {d} valid / {d} invalid\n\n" ++ "configuration\n" ++ @@ -243,7 +291,7 @@ pub fn renderStatus(buf: []u8, args: StatusArgs) []const u8 { "audit receipt · 25 Jul 2026\n" ++ " 20/20 admission suites bound to the deployed image digest\n" ++ " 9/9 lifecycle crashpoints recovered\n" ++ - " 20/20 power-loss schedules recovered\n" ++ + " 21/21 power-loss schedules recovered\n" ++ " archive drained by the official pinned Go V2 client\n\n" ++ "still open\n" ++ " {s}\n\n" ++ @@ -517,5 +565,74 @@ test "status summary reports durable lifecycle timing, and omits it when absent" // The receipt block must not drift from what the admission suite proves. try testing.expect(std.mem.indexOf(u8, shown, "9/9 lifecycle crashpoints") != null); - try testing.expect(std.mem.indexOf(u8, shown, "20/20 power-loss schedules") != null); + try testing.expect(std.mem.indexOf(u8, shown, "21/21 power-loss schedules") != null); +} + +// The river at GET / is an aesthetic choice and diverges from upstream on +// purpose. The field report at /status is the surface that corresponds to +// upstream's /status, so it owes the same information set: repository rollup +// (BackfillStats), live cursors (LiveStats), retention (CursorLookbackStats) +// and import state (ImportInfo). This pins the set so it cannot quietly shrink +// back to the handful of process-local counters it started as. +test "field report carries upstream's status information set" { + var buf: [8192]u8 = undefined; + const out = renderStatus(&buf, .{ + .upstream_seq = 500, + .durable_seq = 480, + .process_events = 12, + .subscribers = 1, + .uptime_s = 120, + .serving = true, + .bootstrap_enabled = true, + .compaction_enabled = true, + .retry_enabled = true, + .verify_valid = 9, + .verify_invalid = 1, + .phase = "steady_state", + .phase_age_s = 600, + .backfill_duration_us = 45 * 60 * std.time.us_per_s, + .repos_total = 200, + .repos_complete = 150, + .repos_pending = 30, + .repos_failed = 15, + .repos_unavailable = 5, + .next_seq = 4_242, + .lookback_s = 36 * 3600, + .oldest_retained_seq = 77, + .sealed_segments = 12, + .import_job = "job-7 (running, apply)", + }); + try testing.expect(std.mem.indexOf(u8, out, "total 200") != null); + try testing.expect(std.mem.indexOf(u8, out, "complete 150 (75%)") != null); + try testing.expect(std.mem.indexOf(u8, out, "pending 30") != null); + try testing.expect(std.mem.indexOf(u8, out, "failed 15") != null); + try testing.expect(std.mem.indexOf(u8, out, "unavailable 5") != null); + try testing.expect(std.mem.indexOf(u8, out, "next seq 4,242") != null); + try testing.expect(std.mem.indexOf(u8, out, "sealed segs 12") != null); + try testing.expect(std.mem.indexOf(u8, out, "cursor lookback 1d 12h") != null); + try testing.expect(std.mem.indexOf(u8, out, "oldest retained 77") != null); + try testing.expect(std.mem.indexOf(u8, out, "job-7 (running, apply)") != null); +} + +test "disabled cursor replay is reported, not shown as no retention" { + var buf: [8192]u8 = undefined; + const out = renderStatus(&buf, .{ + .upstream_seq = 1, + .durable_seq = 1, + .process_events = 0, + .subscribers = 0, + .uptime_s = 1, + .serving = true, + .bootstrap_enabled = false, + .compaction_enabled = false, + .retry_enabled = false, + .verify_valid = 0, + .verify_invalid = 0, + .lookback_s = 0, + .sealed_segments = 3, + }); + try testing.expect(std.mem.indexOf(u8, out, "cursor replay disabled") != null); + try testing.expect(std.mem.indexOf(u8, out, "oldest retained") == null); + // No repository state at all is not "0 of 0 complete". + try testing.expect(std.mem.indexOf(u8, out, "repositories") == null); } diff --git a/src/internal/server.zig b/src/internal/server.zig index 6202fe7..2c3ab42 100644 --- a/src/internal/server.zig +++ b/src/internal/server.zig @@ -1044,6 +1044,10 @@ pub const Handler = struct { } const now_s: i64 = @divTrunc(Io.Timestamp.now(hub.io, .real).toMicroseconds(), std.time.us_per_s); const lifecycle_facts = readLifecycleFacts(hub, now_s); + const repo_facts = readRepoFacts(hub); + const retention = readRetention(hub); + var import_buf: [64]u8 = undefined; + const import_job = readImportJob(hub, &import_buf); const args: homepage.StatusArgs = .{ .upstream_seq = hub.stats.upstream_seq.load(.monotonic), .durable_seq = if (hub.archive) |archive| archive.committed_seq.load(.acquire) else 0, @@ -1059,6 +1063,16 @@ pub const Handler = struct { .phase = lifecycle_facts.phase, .phase_age_s = lifecycle_facts.age_s, .backfill_duration_us = lifecycle_facts.backfill_duration_us, + .repos_total = repo_facts.total, + .repos_complete = repo_facts.complete, + .repos_pending = repo_facts.pending, + .repos_failed = repo_facts.failed, + .repos_unavailable = repo_facts.unavailable, + .next_seq = if (hub.archive) |a| a.next_seq else 0, + .lookback_s = hub.cursor_lookback_ns / std.time.ns_per_s, + .oldest_retained_seq = retention.oldest_seq, + .sealed_segments = retention.sealed_segments, + .import_job = import_job, }; var page_buf: [32 * 1024]u8 = undefined; observation.code = 200; @@ -1487,6 +1501,63 @@ fn coldProgress(from_seq: u64, next_seq: u64, hot_floor: u64, prior_stalls: u8) return if (prior_stalls >= 1) .stalled else .seam_retry; } +/// Repository rollup for the status summary — upstream's BackfillStats. Null +/// total means no repository state exists at all, which the page must not +/// render as "zero of zero complete". +const RepoFacts = struct { + total: ?u32 = null, + complete: u32 = 0, + pending: u32 = 0, + failed: u32 = 0, + unavailable: u32 = 0, +}; + +fn readRepoFacts(hub: *Hub) RepoFacts { + const store = hub.repo_status orelse return .{}; + const total = store.count(); + if (total == 0) return .{}; + return .{ + .total = total, + .complete = store.countByStatus(.complete), + .pending = store.countByStatus(.pending), + .failed = store.countByStatus(.failed), + .unavailable = store.countByStatus(.unavailable), + }; +} + +/// Retention for the status summary — upstream's CursorLookbackStats. +const Retention = struct { + oldest_seq: ?u64 = null, + sealed_segments: ?usize = null, +}; + +fn readRetention(hub: *Hub) Retention { + const archive = hub.archive orelse return .{}; + const summaries = archive.manifest.all(hub.allocator) catch return .{}; + defer hub.allocator.free(summaries); + if (hub.cursor_lookback_ns == 0) return .{ .sealed_segments = summaries.len }; + const now_us = Io.Timestamp.now(hub.io, .real).toMicroseconds(); + return .{ + .oldest_seq = archive.manifest.lookbackFloor(now_us, hub.cursor_lookback_ns).seq, + .sealed_segments = summaries.len, + }; +} + +/// Current or most recent timestamp-import job — upstream's ImportInfo. Null +/// when import has never run against this data dir. +fn readImportJob(hub: *Hub, buf: []u8) ?[]const u8 { + const api = hub.xrpc orelse return null; + const manager = api.import orelse return null; + var parsed = (manager.current() catch return null) orelse return null; + defer parsed.deinit(); + const record = parsed.value; + return std.fmt.bufPrint(buf, "{s} ({s}, {s})", .{ + record.id, + @tagName(record.state), + @tagName(record.phase), + }) catch null; +} + fn respond(conn: *websocket.Conn, status: []const u8, content_type: []const u8, resp_body: []const u8) void { respondMaybeHead(conn, status, content_type, resp_body, false); }