From 8bb96d1839dd6c124fc44bbc730e345599804194 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Mon, 20 Jul 2026 14:23:09 -0500 Subject: [PATCH] prove merge cursor store fault recovery --- README.md | 2 +- docs/semantic-parity.md | 2 +- docs/upstream-harness.md | 2 +- src/internal/meta_store.zig | 91 +++++++++++++++++++++++++++++++++++++ src/main.zig | 13 ++++++ tests/oracle.py | 47 +++++++++++++++++++ 6 files changed, 154 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 4a41866..96105a5 100644 --- a/README.md +++ b/README.md @@ -47,7 +47,7 @@ just run-sim # stream against it just e2e # python wire checks just archive-contract # archive XRPCs + resident manifest + pinned Go client just status-contract # durable host rows + public HTTP view, offline -just oracle # crash-matrix: abort at lifecycle seams, assert recovery +just oracle # lifecycle crash + RocksDB fault recovery matrix just differential-oracle # exact pinned upstream semantics, fully offline ``` diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index 43945e9..a6076dd 100644 --- a/docs/semantic-parity.md +++ b/docs/semantic-parity.md @@ -21,7 +21,7 @@ this document or `bootstrap-semantic-parity.md` is open. | 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 | **closed.** The exact checksum-pinned upstream dashboard is vendored with an offline fail-closed scrape contract. Canonical build, live-firehose, archive, verifier, subscribe, backfill, timestamp-import, and compaction families all have same-semantics producers. Compaction measures pass success/failure/duration, early triggers, live tombstone size, folded reasons, worker examination/rewrite/clean outcomes, dropped-row reasons, bytes, manifest heals, durable watermark, and witnessed-time lag at the corresponding real boundaries. Real JSS tests cover bloom avoidance, bounded concurrent rewrites, malformed account failure, reason accounting, manifest-before-watermark ordering, and durable eviction. The ReleaseSafe production receipt reports `required=104 present=104 missing=0`; the runtime-only dashboard row remains an intentional checksum-pinned Zig/process adaptation with no synthetic `go_*` aliases. See `grafana-dashboard.md`. | -| Oracle | Event-log equivalence, final-state convergence, crash/power-loss, verifier repair, hostile input, and anti-vacuity receipts | **partially closed.** `just differential-oracle` reuses the exact pinned upstream simulator, physical JSS reader, event-log normalizer/comparator, independent MST ground truth, invariant checker, fault scheduler, and public Go client through a fail-closed Go overlay. A real ReleaseSafe Stream child now proves bootstrap convergence; exact normalized durable equivalence for controlled create/update/delete/identity/account/sync/create_resync traffic; duplicate and regressed relay windows with account lifecycle events; the complete upstream frame-fault campaign (malformed/error/oversized/unknown/swallowed/partial-CAR); exact reconnect, sequence-gap, decode, unknown, missing-block, and repair metrics; all-kind anti-vacuity; mutation detection; clean SIGTERM drain; same-disk restart; and public plan/getSegment/getBlock/subscribe-v2 reconstruction. The complete upstream adversarial ingest matrix is also closed: live sibling-preserving invalid collection/rkey/field-width drops, divergent invalid-rev sync repair-then-whole-event-drop, signed non-TID verifier rejection plus follow-up repair, DID-scoped PLC failure without sibling wedge, and backfill CAR invalid UTF-8/path/width drops all run through real sockets, RocksDB, signed repos, and physical JSS with source/reason counter floors and filtered MST convergence. Still open: power-loss/storage fault schedules, compaction mutation coverage, and multi-seed determinism. | +| Oracle | Event-log equivalence, final-state convergence, crash/power-loss, verifier repair, hostile input, and anti-vacuity receipts | **partially closed.** `just differential-oracle` reuses the exact pinned upstream simulator, physical JSS reader, event-log normalizer/comparator, independent MST ground truth, invariant checker, fault scheduler, and public Go client through a fail-closed Go overlay. A real ReleaseSafe Stream child now proves bootstrap convergence; exact normalized durable equivalence for controlled create/update/delete/identity/account/sync/create_resync traffic; duplicate and regressed relay windows with account lifecycle events; the complete upstream frame-fault campaign (malformed/error/oversized/unknown/swallowed/partial-CAR); exact reconnect, sequence-gap, decode, unknown, missing-block, and repair metrics; all-kind anti-vacuity; mutation detection; clean SIGTERM drain; same-disk restart; and public plan/getSegment/getBlock/subscribe-v2 reconstruction. The complete upstream adversarial ingest matrix is also closed: live sibling-preserving invalid collection/rkey/field-width drops, divergent invalid-rev sync repair-then-whole-event-drop, signed non-TID verifier rejection plus follow-up repair, DID-scoped PLC failure without sibling wedge, and backfill CAR invalid UTF-8/path/width drops all run through real sockets, RocksDB, signed repos, and physical JSS with source/reason counter floors and filtered MST convergence. The upstream merge-cursor store-fault contract now also runs through a real process and RocksDB: the selected durable batch fails loudly, proves it fired, and a fault-free reopen converges from the unadvanced cursor. Still open: power-loss and segment-file fault schedules, compaction mutation coverage, and multi-seed determinism. | ## Experiment admission diff --git a/docs/upstream-harness.md b/docs/upstream-harness.md index 3583b2e..1123d2b 100644 --- a/docs/upstream-harness.md +++ b/docs/upstream-harness.md @@ -19,7 +19,7 @@ zig usage: point stream at `ws://localhost:7777/xrpc/com.atproto.sync.subscribeR ## oracle (what an equivalent zig harness would check) -driver boots real server against simulator, walks lifecycle gated on durable-append acks (never sleeps). the simulator world doubles as independent ground truth. contracts: seq ordering, final-state vs world, event-log equivalence (catches lost intermediate events), compaction watermark rules, replay==archive, fold-convergence, anti-vacuity on injected faults (every scheduled fault must prove it fired). tiers incl. crash/restart (SIGKILL real children at crashpoints) and power-loss (strict-mem vfs dropping unsynced writes). detection power measured by mutation campaign. +driver boots real server against simulator, walks lifecycle gated on durable-append acks (never sleeps). the simulator world doubles as independent ground truth. contracts: seq ordering, final-state vs world, event-log equivalence (catches lost intermediate events), compaction watermark rules, replay==archive, fold-convergence, anti-vacuity on injected faults (every scheduled fault must prove it fired). tiers incl. crash/restart (SIGKILL real children at crashpoints), deterministic store faults (a selected real RocksDB mutation fails, then the same disk state is reopened), and power-loss (strict-mem vfs dropping unsynced writes). detection power measured by mutation campaign. ## ingest / live consumer (upstream reference) diff --git a/src/internal/meta_store.zig b/src/internal/meta_store.zig index 8039396..17e7768 100644 --- a/src/internal/meta_store.zig +++ b/src/internal/meta_store.zig @@ -13,6 +13,41 @@ const Allocator = std.mem.Allocator; const Io = std.Io; const log = std.log.scoped(.stream_meta); +/// Deterministic, process-wide metadata fault seam used only by the offline +/// restart oracle. Production leaves this unarmed. The operation still runs +/// through the real Store and RocksDB schema; the selected durable mutation +/// returns an I/O failure immediately before it reaches RocksDB so callers' +/// fail-loud and restart behavior can be exercised without corrupting bytes. +const Fault = struct { + prefix: ?[]const u8 = null, + ordinal: u64 = 0, + seen: std.atomic.Value(u64) = .init(0), + + fn check(self: *Fault, key: []const u8) !void { + const prefix = self.prefix orelse return; + if (!std.mem.startsWith(u8, key, prefix)) return; + const seen = self.seen.fetchAdd(1, .monotonic) + 1; + if (seen != self.ordinal) return; + log.warn("store fault: injected I/O failure for prefix {s} at ordinal {d}", .{ prefix, seen }); + return error.InjectedStoreFault; + } +}; + +var fault: Fault = .{}; + +/// Arm one matching durable metadata operation. Must be called during process +/// startup, before worker threads exist. Returns false for an empty prefix or +/// zero ordinal so an invalid oracle configuration fails closed. +pub fn armFault(prefix: []const u8, ordinal: u64) bool { + if (prefix.len == 0 or ordinal == 0) return false; + fault = .{ .prefix = prefix, .ordinal = ordinal }; + return true; +} + +pub fn disarmFault() void { + fault = .{}; +} + pub const subdir = "meta.rocksdb"; pub const Store = struct { @@ -130,6 +165,10 @@ pub const Store = struct { pub fn putDurable(self: *Store, key: []const u8, value: []const u8) !void { const started_at = self.started(); + fault.check(key) catch |err| { + self.observe(.set, .error_result, started_at); + return err; + }; var err_data: ?rocksdb.Data = null; defer if (err_data) |e| e.deinit(); self.db.putWithOptions(self.default_cf, key, value, .{ .sync = true }, &err_data) catch |err| { @@ -142,6 +181,10 @@ pub const Store = struct { pub fn deleteDurable(self: *Store, key: []const u8) !void { const started_at = self.started(); + fault.check(key) catch |err| { + self.observe(.delete, .error_result, started_at); + return err; + }; var err_data: ?rocksdb.Data = null; defer if (err_data) |e| e.deinit(); self.db.deleteWithOptions(self.default_cf, key, .{ .sync = true }, &err_data) catch |err| { @@ -177,21 +220,32 @@ pub const Store = struct { pub const Batch = struct { store: *Store, inner: rocksdb.WriteBatch, + fault_key: ?[]const u8 = null, pub fn deinit(self: *Batch) void { self.inner.deinit(); } pub fn put(self: *Batch, key: []const u8, value: []const u8) void { + if (fault.prefix) |prefix| { + if (self.fault_key == null and std.mem.startsWith(u8, key, prefix)) self.fault_key = key; + } self.inner.put(self.store.default_cf, key, value); } pub fn delete(self: *Batch, key: []const u8) void { + if (fault.prefix) |prefix| { + if (self.fault_key == null and std.mem.startsWith(u8, key, prefix)) self.fault_key = key; + } self.inner.delete(self.store.default_cf, key); } pub fn commitDurable(self: *Batch) !void { const started_at = self.store.started(); + if (self.fault_key) |key| fault.check(key) catch |err| { + self.store.observe(.batch_commit, .error_result, started_at); + return err; + }; var err_data: ?rocksdb.Data = null; defer if (err_data) |e| e.deinit(); self.store.db.writeWithOptions(self.inner, .{ .sync = true }, &err_data) catch |err| { @@ -260,6 +314,43 @@ test "real RocksDB operations populate canonical store metric outcomes" { try testing.expectEqual(1, stats.store_op_duration_count[storeMetricIndex(.batch_commit, .ok)].load(.monotonic)); } +test "armed durable store fault fires once at matching batch ordinal" { + const testing = std.testing; + disarmFault(); + defer disarmFault(); + 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 store = try Store.open(testing.allocator, path_buf[0..path_len]); + defer store.deinit(); + + try testing.expect(armFault("merge/next_source_idx", 2)); + for (0..2) |i| { + var writes = store.batch(); + defer writes.deinit(); + writes.put("repo/did:plc:alice", "complete"); + writes.put("merge/next_source_idx", "1"); + if (i == 0) { + try writes.commitDurable(); + } else { + try testing.expectError(error.InjectedStoreFault, writes.commitDurable()); + } + } + + // The rejected batch never reached RocksDB. Disarming and retrying the + // exact real batch succeeds, which is the recovery contract used by the + // process oracle. + disarmFault(); + var retry = store.batch(); + defer retry.deinit(); + retry.put("merge/next_source_idx", "2"); + try retry.commitDurable(); + const cursor = (try store.getAlloc(testing.allocator, "merge/next_source_idx")).?; + defer testing.allocator.free(cursor); + try testing.expectEqualStrings("2", cursor); +} + fn storeMetricIndex(op: metrics.StoreOp, status: metrics.StoreStatus) usize { return @intFromEnum(op) * @typeInfo(metrics.StoreStatus).@"enum".fields.len + @intFromEnum(status); } diff --git a/src/main.zig b/src/main.zig index 6e2fc61..7f5bef1 100644 --- a/src/main.zig +++ b/src/main.zig @@ -122,6 +122,8 @@ pub fn main(init: std.process.Init.Minimal) !void { var timestamp_import_token: []const u8 = ""; var timestamp_import_dir_arg: ?[]const u8 = null; var repo_action_rate_limits = true; + var store_fault_prefix: ?[]const u8 = null; + var store_fault_ordinal: u64 = 1; var arg_it = init.args.iterate(); _ = arg_it.next(); // program name @@ -162,6 +164,11 @@ pub fn main(init: std.process.Init.Minimal) !void { log.err("unknown crashpoint: {s}", .{arg["--crashpoint=".len..]}); return error.BadArgs; } + } else if (std.mem.startsWith(u8, arg, "--store-fault-prefix=")) { + // debug/oracle only: fail one matching durable RocksDB mutation + store_fault_prefix = arg["--store-fault-prefix=".len..]; + } else if (std.mem.startsWith(u8, arg, "--store-fault-ordinal=")) { + store_fault_ordinal = try std.fmt.parseInt(u64, arg["--store-fault-ordinal=".len..], 10); } else if (std.mem.eql(u8, arg, "--backfill")) { do_backfill = true; } else if (std.mem.eql(u8, arg, "--no-verify")) { @@ -175,6 +182,12 @@ pub fn main(init: std.process.Init.Minimal) !void { return error.BadArgs; } } + if (store_fault_prefix) |prefix| { + if (!meta_store.armFault(prefix, store_fault_ordinal)) { + log.err("invalid store fault: prefix must be non-empty and ordinal must be positive", .{}); + return error.BadArgs; + } + } const backfill_repos: []const []const u8 = if (backfill_repos_raw) |raw| try parseBackfillRepos(allocator, raw) else &.{}; defer if (backfill_repos.len > 0) allocator.free(backfill_repos); if (backfill_repos.len > 0 and backfill_max_repos > 0) return error.ConflictingBackfillSelection; diff --git a/tests/oracle.py b/tests/oracle.py index a5c2b6e..09325db 100644 --- a/tests/oracle.py +++ b/tests/oracle.py @@ -240,6 +240,49 @@ def run_case(point): log.close() +def run_store_fault_case(): + """A merge-cursor RocksDB failure must abort loudly, then recover.""" + shutil.rmtree(DATA, ignore_errors=True) + log_path = pathlib.Path("/tmp/oracle-store-fault.log") + log = open(log_path, "w") + + p = subprocess.Popen(ARGS + [ + "--store-fault-prefix=merge/next_source_idx", + "--store-fault-ordinal=1", + ], stdout=log, stderr=log) + rc = p.wait(timeout=300) + assert rc != 0, "armed store-fault run completed instead of failing loud" + log.flush() + logged = log_path.read_text() + marker = "store fault: injected I/O failure for prefix merge/next_source_idx at ordinal 1" + assert marker in logged, "store fault never fired (anti-vacuity)" + assert "InjectedStoreFault" in logged, "injected store failure did not reach the process boundary" + + # The failed batch did not advance the source cursor. Reopen the same real + # RocksDB/JSS state without a fault and require full lifecycle convergence. + p = subprocess.Popen(ARGS, stdout=log, stderr=log) + try: + wait_serving(p) + st, body = http_status("/xrpc/network.bsky.jetstream.listSegments") + assert st == 200, st + segs = json.loads(body)["segments"] + assert segs and sum(s["eventCount"] for s in segs) > 0, segs + # A bounded prefix is enough here: the ordinary crash matrix already + # sweeps long replay, while this case is specifically the store-fault + # fail-loud/reopen contract. + replay_count = min(50, sum(s["eventCount"] for s in segs)) + assert replay_count > 0 + assert asyncio.run(replay_invariants(replay_count)) == replay_count + assert not (DATA / "backfill").exists(), "recovery did not finish cleanup" + finally: + p.send_signal(signal.SIGTERM) + try: + p.wait(timeout=15) + except subprocess.TimeoutExpired: + p.kill() + log.close() + + def run_backfill_metrics_receipt(): """A clean production run must exercise, not merely register, backfill metrics.""" shutil.rmtree(DATA, ignore_errors=True) @@ -297,6 +340,10 @@ def main(): run_backfill_metrics_receipt() print(f"ok backfill metrics ({time.time() - t0:.1f}s)") + t0 = time.time() + run_store_fault_case() + print(f"ok store fault recovery ({time.time() - t0:.1f}s)") + for point in CRASHPOINTS: t0 = time.time() run_case(point) -- 2.51.2