diff --git a/README.md b/README.md index 96105a5..040a6ab 100644 --- a/README.md +++ b/README.md @@ -48,6 +48,8 @@ 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 # lifecycle crash + RocksDB fault recovery matrix +just powerloss-image # one-time Linux NBD/ext4 tool-image bootstrap +just powerloss-oracle # real power-cut recovery, offline after bootstrap just differential-oracle # exact pinned upstream semantics, fully offline ``` @@ -56,6 +58,17 @@ The Zig build pins and statically compiles its native dependencies, including RocksDB, zstd, xxHash, and websocket.zig's zlib 1.3.2; it does not depend on a system zlib installation. +The strict power-loss tier is deliberately separate from ordinary unit and +process tests. It cross-builds the production ReleaseSafe Linux binary, runs +RocksDB and JSS on ext4 over a kernel NBD device, kills the block backend before +the Stream process, and reconstructs the device only from writes acknowledged +by FLUSH/FUA. It therefore requires privileged Linux containers and an +available `/dev/nbd3`. The pinned Ubuntu tool image needs network access once +for `just powerloss-image`; `just powerloss-oracle` itself refuses image pulls +and runs from cached Zig, Python, simulator, and container inputs. The current +recipe targets `aarch64-linux-gnu`, matching the development host used for its +recorded receipt. + Production bootstrap controls: ``` diff --git a/docs/bootstrap-semantic-parity.md b/docs/bootstrap-semantic-parity.md index 32cfa82..c8de71e 100644 --- a/docs/bootstrap-semantic-parity.md +++ b/docs/bootstrap-semantic-parity.md @@ -24,7 +24,7 @@ every experiment. | Steady retry host semantics | persist actual post-redirect PDS host; 16 global / 4 per-host workers; 429 parks only that host | redirected `Request.uri` is normalized and persisted; worker gates and durable host-specific parks match upstream limits; a real redirected 429 fixture proves the final host replaces the predicted host and only the final host is parked | closed | | Retry diagnostics | persist attempts, last error/class, host, retry count, next attempt, and durable per-host aggregates | RocksDB row v5 persists every repo field and backward-decodes v2/v3/v4; retry, terminal resync, active-flip, and crash-recovery RMW paths preserve identity/latest fields, while success clears stale repo failure state. The same synced transition batch maintains normalized `host/` current counts plus cumulative error-class counts and five bounded recent samples; host moves remove the repo from the old bucket, successful retry retains diagnostic history, and a one-time migration rebuilds current aggregates without inventing history. The public host view reopens those durable rows through the production binary and exercises both upstream sort modes plus restart, text/HTML, filtering, and escaping offline. | closed | | Deadline coverage | getRepo streaming read + CAR load/completeness check, excluding handler/durability | same `prepared_fetch.zig` future covers both and returns a prepared repo | closed | -| Oracle/adversity suite | truncation, hanging server, terminal errors, crash/restart, active flips | focused offline fixtures cover truncation, hanging fetches, terminal errors, retry/429 interaction, final-host parking, and hostile embedded DIDs; the fresh offline crash matrix survived all 8 lifecycle seams and rejects shutdown OOM/fatal-handler evidence | closed; rerun against the exact candidate before deployment | +| Oracle/adversity suite | truncation, hanging server, terminal errors, crash/restart, active flips, and power loss at canonical lifecycle boundaries | focused offline fixtures cover truncation, hanging fetches, terminal errors, retry/429 interaction, final-host parking, and hostile embedded DIDs; the ordinary crash matrix survived all 8 local lifecycle seams and rejects shutdown OOM/fatal-handler evidence. The strict Linux oracle runs production RocksDB/JSS on ext4 over calibrated NBD storage, kills storage before the process, reconstructs only FLUSH/FUA-acknowledged bytes, and recovered all 8 canonical lifecycle schedules plus the two-crash restart-after-cleanup guard through public archive and V2 replay. | closed; rerun both oracles against the exact candidate before deployment | The deleted July experiment is not evidence for closing these rows. A fresh experiment may be provisioned only after every blocker is closed and the gate diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index 1ed30e5..89dd0f6 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. The upstream merge-cursor store-fault contract 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. Segment-file fault parity is also closed: every production JSS write/fsync/commit-rename crosses the same operation taxonomy, the real process proves ENOSPC/EIO/short-write failure and recovery, Patch and Rewrite exhaustively sweep every local ordinal against real files, and a compaction rename failure preserves both the source bytes and unadvanced watermark before a successful retry. The audit also added the missing active-segment parent-directory fsync and repaired temp-file cleanup. Still open: strict power-loss coverage, broader 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 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. Segment-file fault parity is also closed: every production JSS write/fsync/commit-rename crosses the same operation taxonomy, the real process proves ENOSPC/EIO/short-write failure and recovery, Patch and Rewrite exhaustively sweep every local ordinal against real files, and a compaction rename failure preserves both the source bytes and unadvanced watermark before a successful retry. The strict lifecycle power-loss tier now runs the production Linux binary, RocksDB, and JSS on ext4 over a calibrated NBD volatile-write-cache model. It kills storage before the process and reconstructs solely from FLUSH/FUA-acknowledged bytes; all eight canonical lifecycle schedules plus upstream's two-crash restart-after-cleanup guard recover through real XRPC and public V2 replay. The campaign also closed directory-fsync, cleanup-cursor ordering, Linux subscriber-stack, pre-bootstrap serving, and cold-cursor defects. Still open: the broader compaction power-loss mutation campaign and multi-seed determinism. | ## Experiment admission diff --git a/docs/upstream-harness.md b/docs/upstream-harness.md index 16211a1..54466b3 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), deterministic store faults (a selected real RocksDB mutation fails, then the same disk state is reopened), segment faults (real JSS write/fsync/rename calls fail at a selected process-wide ordinal, with exhaustive Patch/Rewrite sweeps and fault-free reopen), 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), segment faults (real JSS write/fsync/rename calls fail at a selected process-wide ordinal, with exhaustive Patch/Rewrite sweeps and fault-free reopen), and power loss. Upstream models power loss with its strict in-memory VFS; Stream exercises the production Linux filesystem path with ext4 on a strict NBD write-cache backend. detection power measured by mutation campaign. ## ingest / live consumer (upstream reference) @@ -34,7 +34,7 @@ driver boots real server against simulator, walks lifecycle gated on durable-app ## zig oracle v1 (2026-07-12) -`just oracle` (tests/oracle.py): crash-matrix restart harness against the +`just oracle` (`tests/oracle.py`): crash-matrix restart harness against the simulator. arms `--crashpoint=` (src/internal/crashpoint.zig — abort() at durable commit boundaries incl. mid-download), lets the real binary die, restarts clean, asserts convergence: serving ungates, listSegments non-empty, @@ -47,9 +47,10 @@ 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. +does not publish or strand a segment durability boundary. Event-log and final +state equivalence are covered by the pinned differential oracle below; strict +lifecycle power loss is covered by the Linux tier below. Broader compaction +mutation schedules and multi-seed determinism remain open. The harness is intentionally runnable without internet after dependencies are cached. Start the pinned simulator with `GOPROXY=off go run ./cmd/simulator @@ -64,6 +65,50 @@ seams. It also caught a later live-repair regression that recreated healing now have disjoint scratch namespaces, and every case again proves the lifecycle-owned `backfill/` tree is absent at steady state. +## strict Linux power-loss oracle (2026-07-20) + +`just powerloss-oracle` runs the real `aarch64-linux-gnu` ReleaseSafe binary, +RocksDB, and JSS on ext4 mounted over a kernel NBD device. The Python nbdkit +backend has separate live and durable images: ordinary writes affect only the +live image, while NBD FLUSH/FUA copies dirty extents to the durable image and +fsyncs it. At the selected boundary the harness kills nbdkit first, SIGKILLs +Stream second, recreates the live device exclusively from the durable image, +runs `e2fsck`, remounts it, and starts an uninstrumented recovery process. + +The calibration is fail-closed: a raw fsynced `OLD` block must survive while a +later direct, readable, unflushed `NEW` block must disappear, and the flush +receipt must remain unchanged. Every lifecycle case requires a unique cutpoint +receipt, at least one pre-cut storage flush, real RocksDB and segment trees, +nonempty archive XRPC results, healthy serving, and 100 ordered public +`/subscribe-v2` rows (or the complete archive when smaller). The passing +candidate recovered all eight schedules: `after-repo-complete` ordinals 1 and +2, destination-flush-before-source-commit, destination-seal-before-discovery, +discovery-before-cleanup, cleanup-complete, bootstrap-live-close-before-seal, +and steady-phase-before-steady-run. + +An additional two-crash schedule covers upstream's restart-after-cleanup +guard: the first process is killed after removing `backfill/` but before the +directory fsync; the second observes the cached absence, must emit another +durability flush before deleting merge cursors, and is then power-cut at +cleanup-complete. Recovery must still find the source tree absent and every +survivor exactly once. + +This campaign found production defects that process-local crash tests could +not: Linux directory handles opened with `O_PATH` made the required cleanup +fsync fail; cleanup cursor deletion—including its restart guard—was not coupled +to durable directory removal; a 256 KiB ping-thread stack failed on Linux; the listener briefly +served before bootstrap gating; and leading sequence-zero hot-tail rows could +misclassify an archive cursor as resident. The Linux process regression now +observes repeated `503` responses throughout bootstrap and then replays +sequences 1 through 10 from the fresh archive without restart. + +The one-time `just powerloss-image` step builds the digest-pinned Ubuntu tool +image containing nbdkit, `nbd-client`, and ext4 utilities. Actual oracle runs +use `--pull=never`, `uv --offline`, and the pinned local simulator. They require +privileged Linux container access and `/dev/nbd3`; failure evidence is retained +under the printed `stream-powerloss-*` temporary directory instead of being +deleted. + ## archive client contract `just archive-contract` creates a temporary three-segment, 5,000-row JSS diff --git a/justfile b/justfile index e60b1cf..66a5eb7 100644 --- a/justfile +++ b/justfile @@ -169,6 +169,21 @@ oracle: zig build -Doptimize=ReleaseSafe uv run tests/oracle.py +# One-time networked bootstrap for the pinned strict-storage tool image. +# Actual power-loss runs never pull or install anything. +powerloss-image: + docker build --platform=linux/arm64 --pull=false \ + -t stream-powerloss-oracle:ubuntu24.04 tests -f tests/powerloss.Dockerfile + +# Real Linux binary + RocksDB + ext4 + kernel NBD. Expects `just simulator`; +# Go/Zig/uv and Docker are all forced to their existing offline state. +powerloss-oracle: + rm -rf /tmp/stream-powerloss-linux + zig build -Doptimize=ReleaseSafe -Dtarget=aarch64-linux-gnu \ + --prefix /tmp/stream-powerloss-linux + STREAM_POWERLOSS_BIN=/tmp/stream-powerloss-linux/bin/stream \ + uv run --offline tests/powerloss_oracle.py + # Deterministic, fully offline semantic comparison using the exact pinned # upstream simulator, JSS reader, oracle model, and public Go client. differential-oracle: diff --git a/src/internal/bootstrap/engine.zig b/src/internal/bootstrap/engine.zig index d439c8b..880943d 100644 --- a/src/internal/bootstrap/engine.zig +++ b/src/internal/bootstrap/engine.zig @@ -573,6 +573,13 @@ pub fn run( return err; }; + // Canonical upstream boundary: archive rows were flushed before the + // synced completion batch above, so a restart may skip this repo + // without losing its segment rows. A batched commit fires once per + // completed repo, matching upstream's completion batcher ordinals. + for (results.items) |result| + if (result.status == .complete) crashpoint.hit(.after_repo_complete); + var durable_completions: usize = 0; const durable_awake_us = Io.Timestamp.now(io, .awake).toMicroseconds(); for (results.items) |result| switch (result.status) { diff --git a/src/internal/bootstrap/lifecycle.zig b/src/internal/bootstrap/lifecycle.zig index adae5b8..11693a3 100644 --- a/src/internal/bootstrap/lifecycle.zig +++ b/src/internal/bootstrap/lifecycle.zig @@ -93,7 +93,11 @@ pub fn run( var live_dir_buf: [256]u8 = undefined; const live_data_dir = try std.fmt.bufPrint(&live_dir_buf, "{s}/backfill/live", .{opts.data_dir}); - var meta_dir = try Io.Dir.cwd().createDirPathOpen(io, opts.data_dir, .{}); + // `iterate=true` is required even though this handle is not iterated: + // Zig 0.16 otherwise opens Linux directories with O_PATH, and fsync on an + // O_PATH descriptor fails with EBADF. Cleanup below needs a real fd so + // removal of the backfill dirent is actually durable. + var meta_dir = try Io.Dir.cwd().createDirPathOpen(io, opts.data_dir, .{ .open_options = .{ .iterate = true } }); defer meta_dir.close(io); var repo_st = try repo_store_mod.Store.init(allocator, io, meta); defer repo_st.deinit(); @@ -169,6 +173,7 @@ pub fn run( // serving can ungate, every resident header/index must match disk. const reconciled = try dst.reconcileManifest(); _ = stats.compaction_manifest_reconciled_total.fetchAdd(reconciled, .monotonic); + crashpoint.hit(.after_merge_dst_seal_before_discovery); crashpoint.hit(.after_seal_before_discovery); if (have_live) { @@ -181,6 +186,7 @@ pub fn run( // outage here fails the merge (upstream couples them the same // way) — re-entry is idempotent. try runDiscovery(allocator, io, opts.relay_http, &repo_st, stats); + crashpoint.hit(.after_merge_discovery_before_cleanup); crashpoint.hit(.after_discovery_before_cleanup); // cleanup: remove the backfill tree (including the discovery @@ -188,10 +194,34 @@ pub fn run( // (upstream fsyncs the data dir here; std.Io.Dir has no sync, so // fsync the directory fd directly — the Threaded backend's Dir // handle is a posix fd) - try meta_dir.deleteTree(io, "backfill"); - const dir_file: Io.File = .{ .handle = meta_dir.handle, .flags = .{ .nonblocking = false } }; - dir_file.sync(io) catch {}; + meta_dir.deleteTree(io, "backfill") catch |err| { + log.err("merge cleanup: remove backfill tree: {s}", .{@errorName(err)}); + return err; + }; + crashpoint.hit(.after_cleanup_remove_before_sync); + } + // Always sync, including restart-after-cleanup. A prior process may have + // removed the tree and died before this fsync; observing the cached + // absence on this boot does not prove the dirent deletion is durable. + const dir_file: Io.File = .{ .handle = meta_dir.handle, .flags = .{ .nonblocking = false } }; + dir_file.sync(io) catch |err| { + log.err("merge cleanup: sync data directory: {s}", .{@errorName(err)}); + return err; + }; + // Cleanup's dirent must be durable before these synced cursor deletes. + // Otherwise a power cut can resurrect the source tree with cursor zero + // and duplicate already-merged survivors on restart. + { + var cleanup = meta.batch(); + defer cleanup.deinit(); + cleanup.delete("merge/next_source_idx"); + cleanup.delete("bootstrap/last_listrepos_cursor"); + cleanup.commitDurable() catch |err| { + log.err("merge cleanup: delete durable cursors: {s}", .{@errorName(err)}); + return err; + }; } + crashpoint.hit(.after_merge_cleanup_complete); crashpoint.hit(.after_cleanup_before_phase_write); // Bootstrap-live used a temporary local sequence space. Publish the // merged destination's durable counters before steady_state can become @@ -215,6 +245,7 @@ pub fn run( stats.observeOrchestratorState(.write_phase_steady, elapsedUs(io, write_started)); stats.observeOrchestratorTransition(.merging, .steady_state); stats.setOrchestratorPhase(.steady_state); + crashpoint.hit(.after_steady_phase_before_steady_run); } /// resync every pending repo into `dst`. per-repo failure defers to @@ -444,6 +475,7 @@ fn runBootstrap( // closing the archive those callbacks write to. const seal_started = Io.Timestamp.now(io, .awake); try pipe.drainAndStop(); + crashpoint.hit(.after_bootstrap_live_close_before_seal); crashpoint.hit(.after_live_close_before_seal); live_archive.sealAndClose() catch |err| { log.warn("bootstrap: live archive seal failed: {s}", .{@errorName(err)}); diff --git a/src/internal/bootstrap/merge.zig b/src/internal/bootstrap/merge.zig index f690dc6..0c37be2 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 crashpoint = @import("../crashpoint.zig"); const metrics = @import("../metrics.zig"); const repo_store = @import("repo_store.zig"); const segment = @import("../segment.zig"); @@ -118,6 +119,7 @@ pub fn run( 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(); + crashpoint.hit(.after_merge_dst_flush_before_source_commit); const updates = try source_arena.allocator().alloc(repo_store.RevUpdate, revs.count()); var update_it = revs.iterator(); var update_index: usize = 0; diff --git a/src/internal/crashpoint.zig b/src/internal/crashpoint.zig index f7974f8..c320bd1 100644 --- a/src/internal/crashpoint.zig +++ b/src/internal/crashpoint.zig @@ -11,10 +11,21 @@ //! the closest in-process approximation of SIGKILL. const std = @import("std"); +const builtin = @import("builtin"); const log = std.log.scoped(.stream); pub const Point = enum { + // Canonical upstream Jetstream V2 lifecycle boundaries. + after_repo_complete, + after_merge_dst_flush_before_source_commit, + after_merge_dst_seal_before_discovery, + after_merge_discovery_before_cleanup, + after_merge_cleanup_complete, + after_bootstrap_live_close_before_seal, + after_steady_phase_before_steady_run, + + // Additional local boundaries retained by the broader restart oracle. mid_backfill_download, after_backfill_before_phase_merging, after_live_close_before_seal, @@ -22,14 +33,57 @@ pub const Point = enum { after_pending_pass, after_seal_before_discovery, after_discovery_before_cleanup, + after_cleanup_remove_before_sync, after_cleanup_before_phase_write, }; +const Action = enum { abort_process, await_power_cut }; + var armed: ?Point = null; +var action: Action = .abort_process; +var target_ordinal: u64 = 1; +var occurrence: std.atomic.Value(u64) = .init(0); + +pub fn pointName(point: Point) []const u8 { + return switch (point) { + .after_repo_complete => "after-repo-complete", + .after_merge_dst_flush_before_source_commit => "after-merge-dst-flush-before-source-commit", + .after_merge_dst_seal_before_discovery => "after-merge-dst-seal-before-discovery", + .after_merge_discovery_before_cleanup => "after-merge-discovery-before-cleanup", + .after_merge_cleanup_complete => "after-merge-cleanup-complete", + .after_bootstrap_live_close_before_seal => "after-bootstrap-live-close-before-seal", + .after_steady_phase_before_steady_run => "after-steady-phase-before-steady-run", + else => @tagName(point), + }; +} + +fn parse(name: []const u8) ?Point { + inline for (@typeInfo(Point).@"enum".fields) |field| { + const point: Point = @enumFromInt(field.value); + if (std.mem.eql(u8, name, pointName(point)) or std.mem.eql(u8, name, field.name)) return point; + } + return null; +} /// arm from the CLI (debug flag); returns false for an unknown name pub fn arm(name: []const u8) bool { - armed = std.meta.stringToEnum(Point, name) orelse return false; + armed = parse(name) orelse return false; + action = .abort_process; + target_ordinal = 1; + occurrence.store(0, .release); + return true; +} + +/// Arm a lifecycle boundary for the strict block-device oracle. The real +/// process stops at the requested occurrence without running defers; the +/// external harness then removes power from its backing device and SIGKILLs +/// the process. This action is intentionally Linux-only. +pub fn armPowerCut(name: []const u8, ordinal: u64) bool { + if (ordinal == 0) return false; + armed = parse(name) orelse return false; + action = .await_power_cut; + target_ordinal = ordinal; + occurrence.store(0, .release); return true; } @@ -44,6 +98,20 @@ pub fn wouldHit(p: Point) bool { pub fn hit(p: Point) void { const a = armed orelse return; if (a != p) return; - log.warn("crashpoint {s}: aborting", .{@tagName(p)}); - std.process.abort(); + const ordinal = occurrence.fetchAdd(1, .acq_rel) + 1; + if (ordinal != target_ordinal) return; + switch (action) { + .abort_process => { + log.warn("crashpoint {s}: aborting", .{pointName(p)}); + std.process.abort(); + }, + .await_power_cut => { + log.warn("power cutpoint {s} ordinal {d}: ready", .{ pointName(p), ordinal }); + if (comptime builtin.os.tag == .linux) { + while (true) _ = std.os.linux.pause(); + } else { + std.process.abort(); + } + }, + } } diff --git a/src/internal/server.zig b/src/internal/server.zig index aaac880..0ddc413 100644 --- a/src/internal/server.zig +++ b/src/internal/server.zig @@ -33,7 +33,7 @@ const log = std.log.scoped(.stream); /// 8 MB, matching the Io.Threaded stacks (docs/lessons-from-zlay.md #1) const subscriber_stack_size: usize = 8 * 1024 * 1024; -const ping_stack_size: usize = 256 * 1024; +const ping_stack_size: usize = subscriber_stack_size; const slow_window_us: i64 = 60 * std.time.us_per_s; const slow_lag_threshold: u64 = 100_000; const slow_min_rate: u64 = 5; @@ -135,6 +135,9 @@ const Subscriber = struct { v2: bool = false, thread: ?std.Thread = null, ping_thread: ?std.Thread = null, + /// Neither worker may touch the connection or trigger Handler.close until + /// both join handles are installed and the subscriber is published. + start_gate: Io.Event = .unset, ping_stop: Io.Event = .unset, slow: SlowDetector = .{}, @@ -142,6 +145,8 @@ const Subscriber = struct { /// thread — the subscriber never frees itself, so the handler can always /// safely signal it. fn run(self: *Subscriber) void { + self.start_gate.waitUncancelable(self.hub.io); + if (!self.alive.load(.acquire)) return; defer self.workerDone(); if (self.cold_from_seq) |from_seq| if (!self.coldPhaseSeq(from_seq)) return; while (self.alive.load(.acquire)) { @@ -167,6 +172,8 @@ const Subscriber = struct { } fn pingLoop(self: *Subscriber) void { + self.start_gate.waitUncancelable(self.hub.io); + if (!self.alive.load(.acquire)) return; while (self.alive.load(.acquire)) { self.ping_stop.waitTimeout(self.hub.io, .{ .duration = ping_interval }) catch |err| switch (err) { error.Timeout => {}, @@ -521,28 +528,29 @@ pub const Handler = struct { .compressor = compressor, }; self.pending_filter = null; // ownership moved to the subscriber - self.subscriber = sub; const scheme = sub.compression(); - _ = hub.active.fetchAdd(1, .monotonic); - _ = hub.stats.subscribe_active.getPtr(scheme).fetchAdd(1, .monotonic); sub.thread = std.Thread.spawn(.{ .stack_size = subscriber_stack_size }, Subscriber.run, .{sub}) catch |err| { - _ = hub.active.fetchSub(1, .monotonic); - _ = hub.stats.subscribe_active.getPtr(scheme).fetchSub(1, .monotonic); sub.filter.deinit(); hub.allocator.destroy(sub); - self.subscriber = null; return err; }; sub.ping_thread = std.Thread.spawn(.{ .stack_size = ping_stack_size }, Subscriber.pingLoop, .{sub}) catch |err| { - self.signalStop(sub); + log.err("subscriber ping thread spawn failed: {s}", .{@errorName(err)}); + sub.alive.store(false, .release); + sub.stopped.store(true, .release); + sub.start_gate.set(hub.io); sub.thread.?.join(); sub.filter.deinit(); - _ = hub.active.fetchSub(1, .monotonic); - _ = hub.stats.subscribe_active.getPtr(scheme).fetchSub(1, .monotonic); hub.allocator.destroy(sub); - self.subscriber = null; return err; }; + // Publish only after both join handles exist. A replay failure may + // interrupt the server reader immediately after this release; close() + // can now always join both workers exactly once. + self.subscriber = sub; + _ = hub.active.fetchAdd(1, .monotonic); + _ = hub.stats.subscribe_active.getPtr(scheme).fetchAdd(1, .monotonic); + sub.start_gate.set(hub.io); } /// client → server messages: SubscriberSourcedMessage envelope. diff --git a/src/internal/tail.zig b/src/internal/tail.zig index 0955f4f..c31ad3f 100644 --- a/src/internal/tail.zig +++ b/src/internal/tail.zig @@ -264,15 +264,26 @@ pub const Tail = struct { pub fn indexForSeq(self: *Tail, cursor: u64) ?u64 { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); - if (self.entries.items.len == 0) return self.base + self.entries.items.len; + if (self.entries.items.len == 0) return null; + var saw_seq = false; for (self.entries.items, 0..) |e, i| { if (self.durability_gated and e.seq != 0 and e.seq > self.published_seq) break; + // Non-archived diagnostic/control rows do not define the retained + // sequence floor. A leading seq=0 row must not make an actually + // too-old cursor look resident in the hot tail. + if (e.seq == 0) continue; + if (!saw_seq) { + saw_seq = true; + if (e.seq > cursor and cursor > 0) return null; + } if (e.seq >= cursor) { - // below-floor detection: first retained entry already past it - if (i == 0 and e.seq > cursor and cursor > 0) return null; return self.base + i; } } + // A tail containing only seq=0 control rows has no sequence floor. + // The caller must consult the archive rather than treating its tip as + // a successful hot-cursor resolution. + if (!saw_seq) return null; if (!self.durability_gated) return self.base + self.entries.items.len; var visible: usize = 0; for (self.entries.items) |entry| { @@ -341,7 +352,7 @@ test "durability gate hides appended frames until publication" { try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 100, 1, "{\"n\":1}", "{\"v2\":1}", .none); try testing.expectEqual(@as(u64, 1), t.tip()); try testing.expectEqual(@as(u64, 0), t.publishedTip()); - try testing.expectEqual(@as(?u64, 0), t.indexForSeq(1)); + try testing.expectEqual(@as(?u64, null), t.indexForSeq(1)); t.publishDurable(1); try testing.expectEqual(@as(u64, 1), t.publishedTip()); @@ -352,6 +363,19 @@ test "durability gate hides appended frames until publication" { try testing.expectEqualStrings("{\"n\":1}", json); } +test "seq cursor below the hot floor ignores leading unarchived rows" { + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + var t = Tail.init(testing.allocator, threaded.io(), 1 << 20); + defer t.deinit(); + + try t.append(.identity, "did:plc:a", "", 100, 0, "{}", "{}", .none); + try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 200, 42, "{}", "{}", .none); + + try testing.expectEqual(@as(?u64, null), t.indexForSeq(1)); + try testing.expectEqual(@as(?u64, 1), t.indexForSeq(42)); +} + test "undurable rows pin readable log beyond budget until publication" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); diff --git a/src/main.zig b/src/main.zig index 65e23e7..541cd4e 100644 --- a/src/main.zig +++ b/src/main.zig @@ -166,6 +166,16 @@ 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, "--power-cutpoint=")) { + // Linux strict-storage oracle only: NAME or NAME:ORDINAL. + const spec = arg["--power-cutpoint=".len..]; + const split = std.mem.lastIndexOfScalar(u8, spec, ':'); + const name = if (split) |i| spec[0..i] else spec; + const ordinal = if (split) |i| try std.fmt.parseInt(u64, spec[i + 1 ..], 10) else 1; + if (!crashpoint.armPowerCut(name, ordinal)) { + log.err("invalid power cutpoint: {s}", .{spec}); + 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..]; @@ -249,6 +259,23 @@ pub fn main(init: std.process.Init.Minimal) !void { defer repo_diagnostics.deinit(); hub.repo_status = &repo_diagnostics; + // Resolve the durable lifecycle before the listener can accept a request. + // Starting the server first creates a real bootstrap race: a client can + // observe the default serving=true and upgrade before the gate closes. + const stored_phase = blk: { + var ps = phase_mod.Store.init(&meta); + 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; + const bootstrap_requested = do_backfill or backfill_max_repos > 0 or backfill_repos.len > 0; + if (mid_lifecycle or bootstrap_requested) hub.serving.store(false, .release); + var import_dir_buffer: [Io.Dir.max_path_bytes]u8 = undefined; const import_dir = timestamp_import_dir_arg orelse try std.fmt.bufPrint(&import_dir_buffer, "{s}/imports", .{data_dir}); @@ -331,19 +358,7 @@ pub fn main(init: std.process.Init.Minimal) !void { // mid-merge — always resume it. otherwise only --backfill (full crawl) // or --backfill-max-repos (debug) start one; serving is 503-gated // until the lifecycle commits steady_state. - const stored_phase = blk: { - var ps = phase_mod.Store.init(&meta); - 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); + if (mid_lifecycle or bootstrap_requested) { lifecycle.run(allocator, io, .{ .data_dir = data_dir, .upstream = upstream, diff --git a/tests/oracle.py b/tests/oracle.py index ec2e65a..a69314a 100644 --- a/tests/oracle.py +++ b/tests/oracle.py @@ -286,11 +286,17 @@ def run_store_fault_case(): def run_segment_fault_cases(): """Real JSS write/fsync failures must fail loud and recover cleanly.""" cases = [ - ("write:1:enospc", "NoSpaceLeft", True), - ("sync:2:eio", "InputOutput", False), - ("write:3:shortwrite", "ShortWrite", False), + ("write:1:enospc", ("NoSpaceLeft",), True), + ("sync:2:eio", ("InputOutput",), False), + # This ordinal lands inside a per-repository archive append. The + # repository boundary intentionally maps every local persistence + # failure to the fatal ArchiveAppendFailed class, while the concurrent + # live archive propagates ShortWrite directly. Either archive may win + # the process-wide ordinal; the marker below proves the exact leaf + # fault and both accepted boundaries are fatal. + ("write:3:shortwrite", ("ShortWrite", "ArchiveAppendFailed"), False), ] - for spec, error_name, wants_disk_message in cases: + for spec, error_names, wants_disk_message in cases: shutil.rmtree(DATA, ignore_errors=True) label = spec.replace(":", "-") log_path = pathlib.Path(f"/tmp/oracle-segment-fault-{label}.log") @@ -303,7 +309,8 @@ def run_segment_fault_cases(): op, ordinal, kind = spec.split(":") marker = f"segment fault: injected {kind} on {op} at ordinal {ordinal}" assert marker in logged, f"{spec}: segment fault never fired (anti-vacuity)" - assert error_name in logged, f"{spec}: injected segment failure did not reach process boundary" + assert any(name in logged for name in error_names), \ + f"{spec}: injected segment failure did not reach process boundary" if wants_disk_message: assert "fatal persistence error: disk full" in logged assert str(DATA) in logged diff --git a/tests/powerloss.Dockerfile b/tests/powerloss.Dockerfile new file mode 100644 index 0000000..f43f244 --- /dev/null +++ b/tests/powerloss.Dockerfile @@ -0,0 +1,14 @@ +FROM ubuntu@sha256:4fbb8e6a8395de5a7550b33509421a2bafbc0aab6c06ba2cef9ebffbc7092d90 + +RUN apt-get update \ + && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \ + curl \ + e2fsprogs \ + nbd-client \ + nbdkit \ + nbdkit-plugin-python \ + procps \ + python3 \ + util-linux \ + && rm -rf /var/lib/apt/lists/* + diff --git a/tests/powerloss_oracle.py b/tests/powerloss_oracle.py new file mode 100644 index 0000000..c4bc2ad --- /dev/null +++ b/tests/powerloss_oracle.py @@ -0,0 +1,439 @@ +# /// script +# dependencies = ["websockets"] +# /// +"""Strict power-loss oracle for the production Linux Stream binary. + +The process uses RocksDB and ext4 on a kernel NBD device. The nbdkit backend +keeps acknowledged writes in a volatile image and copies dirty extents to the +durable image only for FLUSH/FUA. At a canonical upstream lifecycle boundary +this driver kills nbdkit first, then Stream, reconstructs the device solely +from the durable image, repairs ext4's journal, and requires a clean restart. + +The runtime target is offline: it refuses image pulls and expects the pinned +simulator on the host. Build the one-time tool image with +`just powerloss-image`, then run `just powerloss-oracle` while `just simulator` +is active. +""" + +import asyncio +import json +import os +import pathlib +import shutil +import socket +import subprocess +import tempfile +import time +import urllib.error +import urllib.request + +import websockets + + +ROOT = pathlib.Path(__file__).resolve().parents[1] +IMAGE = os.environ.get("STREAM_POWERLOSS_IMAGE", "stream-powerloss-oracle:ubuntu24.04") +BIN = pathlib.Path(os.environ.get("STREAM_POWERLOSS_BIN", "/tmp/stream-powerloss-linux/bin/stream")).resolve() +DEVICE = os.environ.get("STREAM_POWERLOSS_DEVICE", "/dev/nbd3") +PORT = int(os.environ.get("STREAM_POWERLOSS_PORT", "16099")) +UPSTREAM_HOST = os.environ.get("STREAM_POWERLOSS_UPSTREAM_HOST", "host.docker.internal") +CONTAINER = f"stream-powerloss-oracle-{os.getpid()}" +SIZE = 768 * 1024 * 1024 +BASE = f"http://127.0.0.1:{PORT}" + +CASES = [ + ("after-repo-complete", 1), + ("after-repo-complete", 2), + ("after-merge-dst-flush-before-source-commit", 1), + ("after-merge-dst-seal-before-discovery", 1), + ("after-merge-discovery-before-cleanup", 1), + ("after-merge-cleanup-complete", 1), + ("after-bootstrap-live-close-before-seal", 1), + ("after-steady-phase-before-steady-run", 1), +] + + +def run(args, *, check=True, capture=True, timeout=60): + result = subprocess.run( + [str(a) for a in args], + text=True, + stdout=subprocess.PIPE if capture else None, + stderr=subprocess.STDOUT if capture else None, + timeout=timeout, + ) + if check and result.returncode != 0: + raise AssertionError(f"command failed ({result.returncode}): {' '.join(map(str, args))}\n{result.stdout}") + return result + + +def dexec(*args, check=True, capture=True, timeout=60): + return run(["docker", "exec", CONTAINER, *args], check=check, capture=capture, timeout=timeout) + + +def dexec_sh(script, *, check=True, timeout=60): + return dexec("sh", "-lc", script, check=check, timeout=timeout) + + +def docker_image_ready(): + result = run(["docker", "image", "inspect", IMAGE], check=False) + if result.returncode != 0: + raise SystemExit(f"strict power-loss image missing: {IMAGE}; run `just powerloss-image` once while online") + + +def require_simulator(): + result = dexec( + "curl", "-fsS", "--max-time", "3", + f"http://{UPSTREAM_HOST}:7777/xrpc/com.atproto.sync.listRepos?limit=1", + check=False, + ) + if result.returncode != 0: + raise SystemExit("pinned simulator is not reachable; start it with `just simulator`") + body = json.loads(result.stdout) + assert body.get("repos"), "simulator returned no repositories" + + +def wait_log(path, marker, timeout=300): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if path.exists(): + text = path.read_text(errors="replace") + if marker in text: + return text + alive = dexec_sh("test -f /state/stream.pid && kill -0 $(cat /state/stream.pid)", check=False) + if alive.returncode != 0: + raise AssertionError(f"Stream exited before {marker!r}\n{path.read_text(errors='replace') if path.exists() else ''}") + time.sleep(0.1) + raise AssertionError(f"timed out waiting for {marker!r}\n{path.read_text(errors='replace') if path.exists() else ''}") + + +def http_json(path): + try: + with urllib.request.urlopen(BASE + path, timeout=3) as response: + return response.status, json.loads(response.read()) + except urllib.error.HTTPError as error: + return error.code, None + except OSError: + return None, None + + +def http_status(path): + try: + with urllib.request.urlopen(BASE + path, timeout=3) as response: + response.read() + return response.status + except urllib.error.HTTPError as error: + return error.code + except OSError: + return None + + +def wait_serving(log_path, timeout=300): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + status, body = http_json("/xrpc/network.bsky.jetstream.listSegments") + if status == 200: + return body + alive = dexec_sh("test -f /state/stream.pid && kill -0 $(cat /state/stream.pid)", check=False) + if alive.returncode != 0: + raise AssertionError(f"recovery exited before serving\n{log_path.read_text(errors='replace')}") + time.sleep(0.2) + raise AssertionError(f"recovery did not serve\n{log_path.read_text(errors='replace')}") + + +async def replay(count, *, require_unique=False): + previous = 0 + fingerprints = set() + url = f"ws://127.0.0.1:{PORT}/subscribe-v2?cursor=0" + async with websockets.connect(url) as websocket: + for _ in range(count): + event = json.loads(await asyncio.wait_for(websocket.recv(), timeout=15)) + seq = event.get("seq") + assert isinstance(seq, int) and seq > previous, (previous, event) + assert event.get("did", "").startswith("did:"), event + if require_unique: + semantic = dict(event) + semantic.pop("seq", None) + semantic.pop("cursor", None) + fingerprint = json.dumps(semantic, sort_keys=True, separators=(",", ":")) + assert fingerprint not in fingerprints, f"duplicate recovered event at seq {seq}" + fingerprints.add(fingerprint) + previous = seq + + +def start_backend_detached(): + run([ + "docker", "exec", "-d", CONTAINER, + "sh", "-lc", + "exec env STRICT_NBD_DURABLE=/state/durable.img " + "STRICT_NBD_LIVE=/state/live.img " + f"STRICT_NBD_SIZE={SIZE} " + "STRICT_NBD_RECEIPT=/state/receipt " + "nbdkit -f -i 127.0.0.1 -p 10809 --tls=off python /oracle/strict_nbd.py " + ">/state/nbdkit.log 2>&1", + ]) + for _ in range(100): + if dexec_sh("pgrep -r RSD -x nbdkit >/dev/null", check=False).returncode == 0: + return + time.sleep(0.05) + raise AssertionError("nbdkit did not start") + + +def connect_device(): + dexec_sh(f"nbd-client -d {DEVICE} >/dev/null 2>&1 || true") + for _ in range(100): + attached = dexec( + "nbd-client", "127.0.0.1", "10809", DEVICE, "-b", "4096", + check=False, + ) + if attached.returncode == 0 and dexec("nbd-client", "-c", DEVICE, check=False).returncode == 0: + return + time.sleep(0.05) + detail = dexec_sh("dmesg | tail -20", check=False).stdout + raise AssertionError(f"failed to attach {DEVICE}\n{detail}") + + +def stop_device(*, graceful=False): + if graceful: + dexec_sh("sync", check=False) + dexec_sh("pkill -9 -x nbdkit 2>/dev/null || true", check=False) + dexec_sh("timeout 5 umount -l /mnt/strict 2>/dev/null || true", check=False) + dexec_sh(f"timeout 5 nbd-client -d {DEVICE} >/dev/null 2>&1 || true", check=False) + dexec_sh("pkill -9 -x nbd-client 2>/dev/null || true", check=False) + + +def calibrate(state): + dexec_sh("rm -rf /state/*") + start_backend_detached() + connect_device() + dexec_sh( + f"python3 - <<'PY'\n" + "import os\n" + f"fd=os.open('{DEVICE}', os.O_RDWR)\n" + "os.pwrite(fd, b'OLD'+bytes(4093), 16*1024*1024)\n" + "os.fsync(fd)\n" + "os.close(fd)\n" + "open('/tmp/new.block','wb').write(b'NEW'+bytes(4093))\n" + "PY\n" + f"before=$(wc -l /dev/null 2>&1 || true", check=False) + dexec_sh("pkill -9 -x nbd-client 2>/dev/null || true", check=False) + start_backend_detached() + connect_device() + recovered = dexec_sh( + f"dd if={DEVICE} bs=4096 skip=4096 count=1 iflag=direct status=none | head -c3" + ).stdout + assert recovered == "OLD", f"strict cache calibration recovered {recovered!r}, not OLD" + assert len((state / "receipt").read_text().splitlines()) == 1 + stop_device() + print("powerloss: calibrated durable FLUSH vs discarded volatile write") + + +def stream_args(point=None, ordinal=1): + args = ( + "/oracle/stream --port=6099 --data-dir=/mnt/strict/data " + f"--upstream=ws://{UPSTREAM_HOST}:7777 " + f"--relay-http=http://{UPSTREAM_HOST}:7777 " + f"--plc=http://{UPSTREAM_HOST}:7777 " + "--backfill-max-repos=25 --compaction-interval=0 --retry-interval=0 --no-verify" + ) + if point: + args += f" --power-cutpoint={point}:{ordinal}" + return args + + +def start_stream(log_name, point=None, ordinal=1): + command = f"echo $$ >/state/stream.pid; exec {stream_args(point, ordinal)} >/state/{log_name} 2>&1" + run(["docker", "exec", "-d", CONTAINER, "sh", "-lc", command]) + + +def run_case(state, point, ordinal): + label = f"{point}-ordinal-{ordinal}" + dexec_sh("rm -rf /state/*; mkdir -p /mnt/strict") + start_backend_detached() + connect_device() + dexec_sh( + f"mkfs.ext4 -F {DEVICE} >/state/mkfs.log 2>&1; " + f"mount -o barrier=1,commit=600 {DEVICE} /mnt/strict; mkdir /mnt/strict/data" + ) + start_stream("first.log", point, ordinal) + marker = f"power cutpoint {point} ordinal {ordinal}: ready" + first_log = wait_log(state / "first.log", marker) + assert first_log.count(marker) == 1, "cutpoint receipt was not unique" + if point == "after-repo-complete" and ordinal == 1: + # The first canonical seam holds the bootstrap process in place, so + # this is a deterministic process-level proof that the listener never + # exposes archive/subscribe readiness before steady state. Repeating + # the request catches a transient open-then-close gate. + assert http_status("/healthz") == 200 + for _ in range(3): + assert http_status("/xrpc/network.bsky.jetstream.listSegments") == 503 + flushes_before = len((state / "receipt").read_text().splitlines()) + assert flushes_before > 0, "filesystem never emitted a durability flush" + + # Ordering is the power cut: storage disappears before the process can run + # a defer, close RocksDB, or flush a segment. + dexec_sh("pkill -9 -x nbdkit; kill -9 $(cat /state/stream.pid)", check=False) + dexec_sh("timeout 5 umount -l /mnt/strict 2>/dev/null || true", check=False) + dexec_sh(f"timeout 5 nbd-client -d {DEVICE} >/dev/null 2>&1 || true", check=False) + dexec_sh("pkill -9 -x nbd-client 2>/dev/null || true", check=False) + + start_backend_detached() + connect_device() + fsck = dexec_sh(f"e2fsck -fy {DEVICE}", check=False, timeout=120) + assert fsck.returncode in (0, 1), f"e2fsck failed after {label}:\n{fsck.stdout}" + dexec_sh(f"mount -o barrier=1,commit=600 {DEVICE} /mnt/strict") + dexec_sh("test -d /mnt/strict/data/meta.rocksdb && test -d /mnt/strict/data/segments") + start_stream("recovery.log") + listing = wait_serving(state / "recovery.log") + segments = listing.get("segments", []) + events = sum(segment.get("eventCount", 0) for segment in segments) + assert segments and events > 0, listing + assert http_status("/healthz") == 200 + asyncio.run(replay(min(events, 100))) + dexec_sh( + "test ! -e /mnt/strict/data/backfill; " + "test ! -e /mnt/strict/data/phase; test ! -e /mnt/strict/data/repos.log" + ) + recovery_log = (state / "recovery.log").read_text(errors="replace") + assert "OutOfMemory" not in recovery_log + assert "fatal event handling failure" not in recovery_log + print(f"powerloss: {label} recovered ({events} events, {flushes_before} pre-cut flushes)") + + dexec_sh("kill -TERM $(cat /state/stream.pid) 2>/dev/null || true", check=False) + for _ in range(100): + if dexec_sh("kill -0 $(cat /state/stream.pid) 2>/dev/null", check=False).returncode != 0: + break + time.sleep(0.05) + dexec_sh("kill -9 $(cat /state/stream.pid) 2>/dev/null || true", check=False) + stop_device(graceful=True) + + +def run_restart_after_cleanup_guard_case(state): + """Crash before dir fsync, restart on cached state, then lose power. + + This is upstream's two-crash guard-path schedule. The second process sees + the live tree as absent, but must fsync the data directory again before it + durably deletes the merge cursors. Otherwise the subsequent power loss can + resurrect the source tree with cursor zero and duplicate survivors. + """ + label = "restart-after-cleanup-guard" + dexec_sh("rm -rf /state/*; mkdir -p /mnt/strict") + start_backend_detached() + connect_device() + dexec_sh( + f"mkfs.ext4 -F {DEVICE} >/state/mkfs.log 2>&1; " + f"mount -o barrier=1,commit=600 {DEVICE} /mnt/strict; mkdir /mnt/strict/data" + ) + + local_point = "after_cleanup_remove_before_sync" + start_stream("guard-first.log", local_point) + wait_log(state / "guard-first.log", f"power cutpoint {local_point} ordinal 1: ready") + dexec_sh("test ! -e /mnt/strict/data/backfill") + before_count = int(dexec_sh( + "python3 - <<'PY'\n" + "import glob\n" + "total = 0\n" + "for path in glob.glob('/mnt/strict/data/segments/*.jss'):\n" + " with open(path, 'rb') as handle:\n" + " header = handle.read(22)\n" + " if len(header) == 22 and header[:4] == b'jss0' and int.from_bytes(header[4:12], 'little'):\n" + " total += int.from_bytes(header[18:22], 'little')\n" + "print(total)\n" + "PY" + ).stdout) + assert before_count > 0, "guard fixture had no sealed destination rows" + flushes_before_guard = len((state / "receipt").read_text().splitlines()) + dexec_sh("kill -9 $(cat /state/stream.pid)", check=False) + + canonical_point = "after-merge-cleanup-complete" + start_stream("guard-second.log", canonical_point) + marker = f"power cutpoint {canonical_point} ordinal 1: ready" + second_log = wait_log(state / "guard-second.log", marker) + assert second_log.count(marker) == 1 + flushes_after_guard = len((state / "receipt").read_text().splitlines()) + assert flushes_after_guard > flushes_before_guard, \ + "restart-after-cleanup guard emitted no durability flush" + + dexec_sh("pkill -9 -x nbdkit; kill -9 $(cat /state/stream.pid)", check=False) + dexec_sh("timeout 5 umount -l /mnt/strict 2>/dev/null || true", check=False) + dexec_sh(f"timeout 5 nbd-client -d {DEVICE} >/dev/null 2>&1 || true", check=False) + dexec_sh("pkill -9 -x nbd-client 2>/dev/null || true", check=False) + + start_backend_detached() + connect_device() + fsck = dexec_sh(f"e2fsck -fy {DEVICE}", check=False, timeout=120) + assert fsck.returncode in (0, 1), f"e2fsck failed after {label}:\n{fsck.stdout}" + dexec_sh(f"mount -o barrier=1,commit=600 {DEVICE} /mnt/strict") + start_stream("guard-recovery.log") + listing = wait_serving(state / "guard-recovery.log") + segments = listing.get("segments", []) + events = sum(segment.get("eventCount", 0) for segment in segments) + assert segments and events > 0, listing + assert events == before_count, \ + f"guard recovery changed sealed event count: before={before_count} after={events}" + asyncio.run(replay(events, require_unique=True)) + dexec_sh( + "test ! -e /mnt/strict/data/backfill; " + "test ! -e /mnt/strict/data/phase; test ! -e /mnt/strict/data/repos.log" + ) + print( + f"powerloss: {label} recovered " + f"({events} events, {flushes_after_guard - flushes_before_guard} guard flushes)" + ) + + dexec_sh("kill -TERM $(cat /state/stream.pid) 2>/dev/null || true", check=False) + for _ in range(100): + if dexec_sh("kill -0 $(cat /state/stream.pid) 2>/dev/null", check=False).returncode != 0: + break + time.sleep(0.05) + dexec_sh("kill -9 $(cat /state/stream.pid) 2>/dev/null || true", check=False) + stop_device(graceful=True) + + +def main(): + docker_image_ready() + if not BIN.is_file(): + raise SystemExit(f"Linux Stream binary missing: {BIN}") + with socket.socket() as sock: + try: + sock.bind(("127.0.0.1", PORT)) + except OSError as error: + raise SystemExit(f"power-loss oracle port {PORT} is busy: {error}") + + state_dir = pathlib.Path(tempfile.mkdtemp(prefix="stream-powerloss-")) + succeeded = False + try: + run([ + "docker", "run", "-d", "--pull=never", "--privileged", + "--name", CONTAINER, + "-p", f"127.0.0.1:{PORT}:6099", + "-v", f"{BIN}:/oracle/stream:ro", + "-v", f"{ROOT / 'tests' / 'strict_nbd.py'}:/oracle/strict_nbd.py:ro", + "-v", f"{state_dir}:/state", + IMAGE, "sleep", "infinity", + ]) + require_simulator() + calibrate(state_dir) + for point, ordinal in CASES: + run_case(state_dir, point, ordinal) + run_restart_after_cleanup_guard_case(state_dir) + print( + f"powerloss: all {len(CASES)} canonical lifecycle schedules " + "plus restart-after-cleanup guard recovered" + ) + succeeded = True + finally: + dexec_sh("pkill -9 stream 2>/dev/null || true; pkill -9 nbdkit 2>/dev/null || true", check=False) + dexec_sh(f"timeout 5 umount -l /mnt/strict 2>/dev/null || true; timeout 5 nbd-client -d {DEVICE} >/dev/null 2>&1 || true", check=False) + run(["docker", "rm", "-f", CONTAINER], check=False) + if succeeded: + shutil.rmtree(state_dir, ignore_errors=True) + else: + print(f"powerloss: preserved failure evidence at {state_dir}") + + +if __name__ == "__main__": + main() diff --git a/tests/strict_nbd.py b/tests/strict_nbd.py new file mode 100644 index 0000000..2226dd6 --- /dev/null +++ b/tests/strict_nbd.py @@ -0,0 +1,187 @@ +"""nbdkit block backend that can discard every write after the last flush. + +This is not a Stream storage substitute. The production Linux binary still +uses RocksDB, std.Io, and ext4 through /dev/nbdN. This backend models a drive +with volatile write cache: writes update ``live``; NBD FLUSH/FUA copies dirty +extents to ``durable``. Killing nbdkit and starting it again reconstructs +``live`` exclusively from ``durable``, which is the power-loss operation. + +Environment: + STRICT_NBD_DURABLE durable image path + STRICT_NBD_LIVE volatile image path + STRICT_NBD_SIZE bytes (used when creating a new image) + STRICT_NBD_RECEIPT append-only flush receipt path +""" + +import os +import shutil +import threading + +import nbdkit + +API_VERSION = 2 + +durable_path = os.environ["STRICT_NBD_DURABLE"] +live_path = os.environ["STRICT_NBD_LIVE"] +receipt_path = os.environ["STRICT_NBD_RECEIPT"] +size = int(os.environ["STRICT_NBD_SIZE"]) +live_fd = -1 +durable_fd = -1 +dirty = [] +flush_count = 0 +lock = threading.RLock() + + +def _write_all(fd, data, offset): + view = memoryview(data) + written = 0 + while written < len(view): + n = os.pwrite(fd, view[written:], offset + written) + if n <= 0: + raise OSError("short pwrite") + written += n + + +def _mark_dirty(offset, count): + dirty.append((offset, offset + count)) + + +def _coalesced_dirty(): + if not dirty: + return [] + ordered = sorted(dirty) + out = [list(ordered[0])] + for start, end in ordered[1:]: + prior = out[-1] + if start <= prior[1]: + prior[1] = max(prior[1], end) + else: + out.append([start, end]) + return out + + +def _persist_dirty(): + global flush_count + for start, end in _coalesced_dirty(): + offset = start + while offset < end: + chunk = os.pread(live_fd, min(1024 * 1024, end - offset), offset) + if not chunk: + raise OSError("short pread while flushing") + _write_all(durable_fd, chunk, offset) + offset += len(chunk) + os.fsync(durable_fd) + dirty.clear() + flush_count += 1 + receipt_fd = os.open(receipt_path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600) + try: + _write_all(receipt_fd, f"flush {flush_count}\n".encode("ascii"), os.lseek(receipt_fd, 0, os.SEEK_END)) + os.fsync(receipt_fd) + finally: + os.close(receipt_fd) + + +def _initialize(): + global live_fd, durable_fd + if not os.path.exists(durable_path): + fd = os.open(durable_path, os.O_RDWR | os.O_CREAT | os.O_EXCL, 0o600) + try: + os.posix_fallocate(fd, 0, size) + os.fsync(fd) + finally: + os.close(fd) + elif os.path.getsize(durable_path) != size: + raise RuntimeError("durable image size does not match STRICT_NBD_SIZE") + shutil.copyfile(durable_path, live_path) + live_fd = os.open(live_path, os.O_RDWR) + durable_fd = os.open(durable_path, os.O_RDWR) + + +def cleanup(): + # Deliberately never flush here. SIGKILL is the normal power-cut path, but + # even orderly plugin teardown must not accidentally make dirty data safe. + if live_fd >= 0: + os.close(live_fd) + if durable_fd >= 0: + os.close(durable_fd) + + +def open(readonly): + if readonly: + raise RuntimeError("strict NBD oracle requires a writable export") + return 1 + + +def get_size(_handle): + return size + + +def can_flush(_handle): + return True + + +def can_fua(_handle): + return nbdkit.FUA_NATIVE + + +def can_trim(_handle): + return True + + +def can_zero(_handle): + return True + + +def pread(_handle, buf, offset, flags): + assert flags == 0 + with lock: + data = os.pread(live_fd, len(buf), offset) + if len(data) != len(buf): + raise OSError("short pread") + buf[:] = data + + +def pwrite(_handle, buf, offset, flags): + with lock: + _write_all(live_fd, buf, offset) + _mark_dirty(offset, len(buf)) + if flags & nbdkit.FLAG_FUA: + _persist_dirty() + + +def flush(_handle, flags): + assert flags == 0 + with lock: + _persist_dirty() + + +def _zero_range(count, offset): + zeros = bytes(min(1024 * 1024, count)) + remaining = count + while remaining: + chunk = zeros[: min(len(zeros), remaining)] + _write_all(live_fd, chunk, offset) + offset += len(chunk) + remaining -= len(chunk) + + +def trim(_handle, count, offset, flags): + # Logical zeroing is a conservative implementation of discard semantics. + with lock: + _zero_range(count, offset) + _mark_dirty(offset, count) + if flags & nbdkit.FLAG_FUA: + _persist_dirty() + + +def zero(_handle, count, offset, flags): + with lock: + _zero_range(count, offset) + _mark_dirty(offset, count) + if flags & nbdkit.FLAG_FUA: + _persist_dirty() + + +# The nbdkit Python API intentionally has no load callback. Top-level code is +# the documented initialization mechanism; cleanup above is invoked at exit. +_initialize()