diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index 9e86cb1..c12d7b5 100644 --- a/docs/semantic-parity.md +++ b/docs/semantic-parity.md @@ -53,7 +53,7 @@ blockers remain. The detailed bootstrap audit is in | Lifecycle phase machine | **verified** | Phase reads distinguish absence from RocksDB failure, reject unknown values, and use synced writes. Bootstrap writes `merging` after backfill drains while bootstrap-live is still running, exactly at the pinned upstream boundary; process-kill crashpoints exercise both sides of that write. Cleanup writes `steady_state` only after directory sync and cursor deletion. | Exact persisted phase-entry timestamps are a separate status-surface gap; this row establishes transition state and ordering only. | | Whole-network bootstrap | **partial** | listRepos pages form a page-aligned dispatch/checkpoint unit; eligible repositories are shuffled; download concurrency and CAR preparation are real. Successful repositories become durable independently at archive-writer durability boundaries, including while a sibling remains incomplete and the listRepos cursor remains unchanged. | Resource-exhaustion cleanup at every preparation/emission ownership transfer is not yet proved, and selected/debug paths still need exact-artifact admission receipts. See the detailed audit. | | Merge and post-bootstrap discovery | **blocked** | Source rows are filtered by the backfill revision; source cursor/latest-revision updates share a synced batch; merge cursor corruption/read errors fail closed; the restart guard treats only `FileNotFound` as cleanup-complete; discovery records active and inactive unknown repositories and follows arbitrary-length cursors with loop detection; successful cleanup is directory-synced. | The pending-repository pass still materializes all DIDs and does not use upstream's bounded retry runner. Latest-revision refresh and source-file failure coverage remain partial. | -| Failed-repository healing | **blocked** | A real global/per-host worker gate exists, final redirect hosts are recorded, and retry state is stored. | Pass-level RocksDB/archive failures are logged and hidden from service health. Every failed DID and host gate is materialized in RAM before work. Persisted host parking is not upstream behavior, and park-read errors mean “not parked.” | +| Failed-repository healing | **blocked** | A real global/per-host worker gate exists, final redirect hosts are recorded, and retry state is stored. A pass that fails for our own reasons (store, archive, memory) now latches a terminal error, counts `retry_terminal_failures_total`, and requests process shutdown instead of sleeping until the next interval — matching upstream, whose errgroup cancels steady state. Per-repo remote failures still become backoff and never reach that path; a real read fault drives the test. | Every failed DID and host gate is still materialized in RAM before work. Persisted host parking is not upstream behavior, and park-read errors still mean “not parked.” | | JSS sealed format interoperability | **verified** | Upstream-produced sealed fixtures are parsed; Stream-produced sealed files are consumed by the pinned Go reader; header/footer, block index, blooms, collections, compression, and checksums have reciprocal fixtures. | This verifies sealed-format compatibility only. It does not verify startup recovery or replay completeness. | | Archive startup recovery | **verified** | Startup validates the active header, walks complete frames, propagates read/decode/allocation failures, truncates and fsyncs only a framing-torn suffix, reconstructs block/event/sequence state, and resumes the same active file and segment index. Sealed high-water floors are accepted only through the checksum-verifying parser, including the empty-active/lower-sealed case. Physical tests prove repeated same-file restart, exact torn-tail truncation, byte-preserving failure on a complete corrupt frame, and rejection of a forged sealed `max_seq`. | This row establishes startup recovery. Cold replay and post-startup rewrite paths remain separate surfaces. | | `listSegments`, `getSegment`, `getBlock` | **verified** | The pinned Go client and direct HTTP tests cover normal responses, byte identity, checksums/ETags, conditional requests, ranges, cache headers, and storage-error responses through the production server. `listSegments` rendering was extracted so encode failure propagates; a fail-index sweep over the production handler proves every response is the exact baseline list or `5xx`, never a truncated 200 and never a dropped request. | The claim is limited to these three methods and the tested file states. | diff --git a/src/internal/bootstrap/retry.zig b/src/internal/bootstrap/retry.zig index facd5b4..ec44ccc 100644 --- a/src/internal/bootstrap/retry.zig +++ b/src/internal/bootstrap/retry.zig @@ -76,6 +76,17 @@ pub const Retry = struct { emit_failure_logs: bool = true, jitter_ctx: ?*anyopaque = null, jitter: ?*const fn (?*anyopaque, u64) u64 = null, + /// Set when a pass fails for a reason that is ours rather than a remote's. + /// Upstream runs this loop inside an errgroup, so returning an error there + /// cancels the steady-state consumer; this is how that reaches the process. + shutdown: ?*std.atomic.Value(bool) = null, + /// Latched terminal error, exposed so health can report why we are going + /// down rather than leaving a dead subsystem behind a green check. + terminal_error: std.atomic.Value(bool) = .init(false), + + pub fn failed(self: *const Retry) bool { + return self.terminal_error.load(.acquire); + } pub fn init(allocator: Allocator, io: Io, archive: *archive_mod.Archive, meta: *meta_store.Store, data_dir: []const u8, relay_http: []const u8) Retry { return .{ @@ -95,8 +106,9 @@ pub const Retry = struct { self.stop_event.set(self.io); } - /// timer loop; spawn via io.concurrent, stop via requestStop. a - /// failed pass logs and retries next interval — never tears down. + /// timer loop; spawn via io.concurrent, stop via requestStop. a pass that + /// fails for our own reasons latches terminal_error and requests shutdown, + /// mirroring upstream's errgroup; per-repo remote failures never get here. pub fn run(self: *Retry) void { if (self.interval_ns == 0) return; const interval: Io.Clock.Duration = .{ @@ -116,8 +128,19 @@ pub const Retry = struct { } if (self.stop_flag.load(.acquire)) return; const stats = self.runPass() catch |err| { - log.warn("failed-repo retry pass failed: {s} (retrying next interval)", .{@errorName(err)}); - continue; + // Remote failures never reach here: processDid converts them + // into per-repo backoff. An error escaping a pass means our own + // store, archive, or memory failed. Upstream returns it out of + // RunFailedRepoRetry and its errgroup takes the service down; + // continuing would leave repositories silently unhealed behind + // a green health check. + if (self.stop_flag.load(.acquire)) return; + if (self.emit_failure_logs) + log.err("failed-repo retry pass failed: {s}; stopping stream", .{@errorName(err)}); + if (self.stats) |s| _ = s.retry_terminal_failures_total.fetchAdd(1, .monotonic); + self.terminal_error.store(true, .release); + if (self.shutdown) |flag| flag.store(true, .release); + return; }; if (stats.candidates > 0) log.info("failed-repo retry pass: {d} candidates, {d} resynced, {d} re-deferred, {d} skipped", .{ @@ -420,6 +443,56 @@ test "runPass with no failed repos is a no-op" { try testing.expectEqual(@as(u64, 0), instrumentation.retry_candidates_total.load(.monotonic)); } +// Upstream runs this loop in an errgroup: an error out of RunFailedRepoRetry +// cancels steady state and the process goes down. Logging and continuing would +// leave failed repositories permanently unhealed while health stayed green -- +// the loud-and-small failure turned silent and large. +test "a terminal retry pass failure stops the service instead of looping quietly" { + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var path_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const n = try tmp.dir.realPath(io, &path_buf); + var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const data_dir = try std.fmt.bufPrint(&data_dir_buf, "{s}/data", .{path_buf[0..n]}); + + var a = try archive_mod.Archive.init(testing.allocator, io, data_dir); + defer a.deinit(); + var meta = try meta_store.Store.open(testing.allocator, data_dir); + defer meta.deinit(); + + var seed = try repo_store.Store.init(testing.allocator, io, &meta); + defer seed.deinit(); + try seed.put("did:plc:retryfatal", .{ .status = .failed, .active = true }); + + var instrumentation: metrics.Stats = .{}; + var shutdown: std.atomic.Value(bool) = .init(false); + var r = Retry.init(testing.allocator, io, &a, &meta, data_dir, "http://127.0.0.1:9"); + defer r.deinit(); + r.stats = &instrumentation; + r.shutdown = &shutdown; + r.interval_ns = std.time.ns_per_ms; + r.emit_failure_logs = false; // the terminal log is asserted by the metric + + // Our own store read fails: not a remote's fault, so the pass must not + // simply be retried next interval. + try testing.expect(meta_store.armReadFault("repo/", 1)); + defer meta_store.disarmReadFault(); + + // With the old log-and-continue behavior this call never returns: the pass + // fails, the loop sleeps, and the subsystem stays dead while health is green. + r.run(); + + try testing.expect(r.failed()); + try testing.expect(shutdown.load(.acquire)); + try testing.expectEqual( + @as(u64, 1), + instrumentation.retry_terminal_failures_total.load(.monotonic), + ); +} + test "host parking is isolated and durable by final PDS host" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); diff --git a/src/internal/metrics.zig b/src/internal/metrics.zig index 12e566d..595f425 100644 --- a/src/internal/metrics.zig +++ b/src/internal/metrics.zig @@ -166,6 +166,7 @@ pub const Stats = struct { 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), @@ -1043,6 +1044,8 @@ pub fn format( \\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 @@ -1081,6 +1084,7 @@ pub fn format( 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), diff --git a/src/main.zig b/src/main.zig index 0ff2f8f..e446392 100644 --- a/src/main.zig +++ b/src/main.zig @@ -1244,6 +1244,7 @@ pub fn main(init: std.process.Init.Minimal) !void { retry.max_delay_ns = retry_max_delay_ns; retry.stats = &stats; retry.trace_context = steady_trace.context(); + retry.shutdown = &shutdown_requested; if (retry_interval_ns != 0) log.info("failed-repo retry loop on (interval {d}ns, workers {d}, host workers {d}, max delay {d}ns)", .{ retry_interval_ns, retry_workers,