From f7af4110807ad23b7d21ee6cee2284dc5515f181 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Tue, 21 Jul 2026 12:49:40 -0500 Subject: [PATCH] prove compaction power-loss parity --- README.md | 6 +- docs/semantic-parity.md | 4 +- docs/upstream-harness.md | 22 +- justfile | 2 + src/internal/compact/pass.zig | 4 + src/internal/crashpoint.zig | 20 ++ src/internal/homepage.zig | 1 + src/internal/segment_patch.zig | 5 + src/internal/segment_rewrite.zig | 5 + src/internal/timestamp/bucket.zig | 5 +- src/internal/timestamp/manager.zig | 5 +- src/internal/timestamp/runner.zig | 31 ++- tests/powerloss_oracle.py | 311 ++++++++++++++++++++++++++--- tests/timestamp-import.csv | 2 + 14 files changed, 385 insertions(+), 38 deletions(-) create mode 100644 tests/timestamp-import.csv diff --git a/README.md b/README.md index 040a6ab..480ac9a 100644 --- a/README.md +++ b/README.md @@ -62,7 +62,11 @@ 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 +by FLUSH/FUA. Its campaign covers eight lifecycle cuts, the two-crash cleanup +guard, six delete-compaction cuts, and four timestamp-import rewrite cuts. The +mutation cases use archives written by the production JSS writer and verify +recovery through the production daemon and public V2 replay. 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 diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index 89dd0f6..33910f6 100644 --- a/docs/semantic-parity.md +++ b/docs/semantic-parity.md @@ -18,10 +18,10 @@ this document or `bootstrap-semantic-parity.md` is open. | Subscribe v1/v2 | Wire/filter/cursor/compression unit and local e2e coverage | Cursor parsing/resolution precedes upgrade; both endpoints use the seq/time-us magnitude split, v1 clamps below-floor seqs, and v2 rejects them. Timestamp translation uses bounded positional reads of real sealed JSS headers, block indexes, and one selected block before upgrade; disk/decode/index faults are retryable generic 503s, logged internally and counted as `resolve_failed`. Hot/cold scan, skip, encode, options-update, oversize-parser, clean peer/server close, cursor-resolution, and sustained adversarial-rate boundaries feed canonical metrics, with deterministic offline tests derived from upstream. Subscriber transport sends a ping every 30 seconds, applies a kernel-enforced five-second deadline to every frame write, and sends close code 1001 before intentional server shutdown while keeping transport failure out of clean-disconnect accounting. A failed delivery, cold read, ping, or adversarial-rate check interrupts the server reader and removes the connection instead of leaving a ping-only zombie. V1 now negotiates genuine RFC 7692 deflate with the upstream 128-byte threshold and 32 KiB context takeover; v1 custom zstd and all v2 connections explicitly decline it. V2 dictionary negotiation, exact pinned dictionary download/validators, and v2 binary delivery remain independent of the frozen v1 dictionary contract; pinned official Go clients have decoded both compression families offline. | | Sync 1.1 live verification | Upstream requires durable per-DID chain/hosting state, MST inversion, op-CID consistency, rev replay/future guards, default acceptance of legacy-shaped commits, transparent whole-repo repair, and Atmos delivery scheduling | **closed at the implementation boundary.** Diff-CAR/MST inversion (including upstream's narrow default lenient carve-out), post-state op-CID proof, signed inner/outer consistency, exact decimal size/future-rev/replay gates, exact legacy-shape detection with default `LegacyAccept`, two-phase durable chain/hosting state, and account/identity replay ratchets are implemented. Recoverable decode/inversion/duplicate-path/op-CID/chain failures route to repair; signature failures and outer/inner producer-integrity mismatches bypass it, and no failed event is archived. `prepareRepo` retains and authenticates the canonical signed complete head without a second multi-gigabyte parse/walk. The live path now uses Atmos's 32 worker slots, one worker-held FIFO chain per DID, a 64-pending drop-oldest boundary, completion-order batches of 50/500 ms across DIDs, and `min(inflight)-1` cursor watermarks. Dropped and silent events leave the inflight set without inventing a delivery batch; cursor advancement remains coupled to durable archive boundaries. Offline receipts use real encoded frames, real RocksDB verifier state, and actual worker blocking rather than fixture verdicts or scheduler mocks. See `live-scheduler.md`. | | Sync 1.1 resync ordering | Upstream serializes per DID, drops stale async repairs, and emits a sync tombstone plus authoritative replacements | **closed at the implementation boundary.** The live coordinator has 32 real fetch workers, a 64-job queue, a five-minute fetch budget plus the 1,000 B/s-for-30-seconds slow-transfer guard, a bounded 16,384-entry 5/minute-per-DID limiter, 2,048 pending commits with drop-oldest overflow, fetch outside the DID lane, authenticated apply with older/equal-contradictory head rejection, ordered pending replay, and trigger-ticket-gated outbox delivery. Commit divergence schedules async repair; `#sync` divergence attempts inline repair and queues one retry only for transient failure, preserving the original relay cursor only on inline success. Matching Atmos's gate ownership, a divergent `#sync` repairs before envelope/inner validation; an invalid envelope rev then durably advances only the authenticated chain and drops the original tombstone plus every replacement row as one `live/invalid_rev` event. Valid repairs emit `sync` then bounded 1,024-row `create_resync` batches and stage fetched/pending chain checkpoints under the archive lock before any matching fsync. Loopback-only tests exercise a real HTTP getRepo mmap, real DID resolution/signature, buffering while the response is held open, slow-transfer cancellation, durable RocksDB promotion/cursor coupling, v2 tail publication, and sealed JSS row order. | -| Timestamp import | Canonical dashboard and API include timestamp import lifecycle/rewrite metrics and behavior | **Closed at the implementation boundary.** The format preserves distinct `witnessed_at` and sentinel-zero `indexed_at` columns. Subscriber v1/v2 encoding applies upstream's `indexed_at != 0 ? indexed_at : witnessed_at` display rule while ranges and timestamp cursors remain anchored to immutable witness envelopes. The strict seekable RFC 4180 parser matches upstream's header, row rejection, byte-offset, CID, RFC3339, 64 KiB, and bounded-sampling contracts in a forward-only allocation-free pass. A separate Bloom-filtered RocksDB rule store bulk-loads sorted external SST chunks with CSV-order last-write-wins, reconstructs its resident collection gate on open, and implements specific-CID precedence with all-version fallback. Its central archive-lock hook stamps live, repair, bootstrap, and merge materializations before sequence assignment, and feeds the mutated rows to immediate delivery so hot and cold encodings agree. The sealed-JSS patch primitive preserves block topology/envelopes, clean frames, and the opaque bloom/collection footer while atomically changing only `indexed_at`; it is exercised against an upstream-produced segment. A concrete sealed-segment bloom catalog, bounded DID/FD bucketer, fsynced packed offsets, positioned revalidation, and collision-safe per-segment patch plan implement Phase B/C with specific-CID precedence and idempotent disk-to-disk coverage. A dedicated archive rewrite mutex serializes the real delete compactor with timestamp patching while leaving live append available. The unified metadata DB synchronously persists the current-job pointer, complete job record, and per-segment done set. The real runner activates rules before force-seal, durably crosses the bucketed handoff, checkpoints only after patch fsync/rename/dir-sync, retains partial cancellation progress, and resumes after full archive/rule/metadata reopen. The background manager canonicalizes symlinks, confines regular files, enforces one durable nonterminal job, and adopts it on startup. Bearer-gated import/status XRPC matches the pinned lexicons, fixed-width digest comparison, error names/statuses, steady-state admission, and current/by-id status behavior; a real loopback HTTP run exercised it against the local simulator. Every canonical import metric is produced at its real parse, route, patch, or durable-terminal boundary; cancellation remains a resumable pause and is not counted as terminal failure. | +| 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. The strict Linux campaign cuts all four patch write/fsync/rename/dir-sync boundaries and proves exact target and untargeted timestamps after daemon recovery and public V2 replay; it also exposed and closed a Linux `O_PATH` directory-fsync defect. | | 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 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. | +| 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 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. Six compaction schedules cover rewrite write/fsync/rename/dir-sync and both sides of the durable watermark; four timestamp-import schedules cover patch write/fsync/rename/dir-sync. All ten converge to exact row/timestamp results without stale temporary files. The campaign closed directory-fsync, cleanup-cursor ordering, Linux subscriber-stack, pre-bootstrap serving, cold-cursor, and Linux directory-handle defects. Still open: multi-seed determinism. | ## Experiment admission diff --git a/docs/upstream-harness.md b/docs/upstream-harness.md index 54466b3..98df9ba 100644 --- a/docs/upstream-harness.md +++ b/docs/upstream-harness.md @@ -49,8 +49,8 @@ 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. 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. +lifecycle, compaction, and timestamp-import power loss are covered by the Linux +tier below. Multi-seed determinism remains open. The harness is intentionally runnable without internet after dependencies are cached. Start the pinned simulator with `GOPROXY=off go run ./cmd/simulator @@ -86,6 +86,19 @@ candidate recovered all eight schedules: `after-repo-complete` ordinals 1 and discovery-before-cleanup, cleanup-complete, bootstrap-live-close-before-seal, and steady-phase-before-steady-run. +The same production-binary campaign covers upstream's mutation boundaries. +Six delete-compaction schedules cut at each rewrite write/fsync/rename/dir-fsync +boundary and on both sides of the durable watermark; every recovery converged +to watermark 2500 and exactly 1,507 retained events after dropping 993 rows. +Four timestamp-import schedules cut at each sealed-segment patch boundary; the +target rows recovered with the imported timestamp while untargeted rows kept +their exact witness-derived time. All ten cases reject stale temporary files, +require complete duplicate-free public V2 replay, and use fixtures created by +the real Stream JSS writer rather than synthetic segment bytes. This campaign +found a Linux-only `O_PATH` parent-directory handle whose `fsync` failed with +`EBADF`; rewrite directories are now opened as readable directory handles and +the full campaign passes on Linux. + 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 @@ -173,8 +186,9 @@ bootstrap CARs carry invalid UTF-8, no-slash, dot-dot, and 300-byte keys; only valid records reach physical JSS, filtered MST truth converges, and every canonical `(source, reason)` counter meets the upstream ledger floor. -This tier does not yet claim upstream's storage/power-loss fault schedules, -compaction mutation campaign, or multi-seed determinism; those remain explicit +The differential tier alone does not claim filesystem durability. The strict +Linux tier closes the pinned storage, lifecycle-power-loss, compaction, and +timestamp-import mutation schedules. Multi-seed determinism remains explicit work in `semantic-parity.md`. ## HTTP and getBlock metrics contract diff --git a/justfile b/justfile index 66a5eb7..b075464 100644 --- a/justfile +++ b/justfile @@ -178,10 +178,12 @@ powerloss-image: # 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: + zig build write-sample -- /tmp/stream-powerloss-fixture.jss 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 \ + STREAM_POWERLOSS_FIXTURE=/tmp/stream-powerloss-fixture.jss \ uv run --offline tests/powerloss_oracle.py # Deterministic, fully offline semantic comparison using the exact pinned diff --git a/src/internal/compact/pass.zig b/src/internal/compact/pass.zig index 535c74d..e8162bd 100644 --- a/src/internal/compact/pass.zig +++ b/src/internal/compact/pass.zig @@ -18,6 +18,7 @@ const std = @import("std"); const archive_mod = @import("../archive.zig"); +const crashpoint = @import("../crashpoint.zig"); const meta_store = @import("../meta_store.zig"); const metrics = @import("../metrics.zig"); const rewrite_mod = @import("../segment_rewrite.zig"); @@ -210,6 +211,8 @@ fn runInner( _ = s.compaction_tombstones_collected.getPtr(.account).fetchAdd(collected.account, .monotonic); } + crashpoint.hit(.after_compaction_rewrite_before_watermark); + // durable commit, then eviction (rebuild-from-disk makes a crash // between the two immaterial) try wm.save(io, chunk_end); @@ -221,6 +224,7 @@ fn runInner( committed(opts.on_committed_ctx, chunk_end) else if (live) |l| l.evict(chunk_end); + crashpoint.hit(.after_compaction_chunk_watermark); stats.chunks += 1; current = chunk_end; } diff --git a/src/internal/crashpoint.zig b/src/internal/crashpoint.zig index c320bd1..ae0849c 100644 --- a/src/internal/crashpoint.zig +++ b/src/internal/crashpoint.zig @@ -24,6 +24,16 @@ pub const Point = enum { after_merge_cleanup_complete, after_bootstrap_live_close_before_seal, after_steady_phase_before_steady_run, + after_compaction_rewrite_before_watermark, + after_compaction_chunk_watermark, + after_segment_rewrite_temp_written, + after_segment_rewrite_temp_synced, + after_segment_rewrite_renamed, + after_segment_rewrite_dir_synced, + after_segment_patch_temp_written, + after_segment_patch_temp_synced, + after_segment_patch_renamed, + after_segment_patch_dir_synced, // Additional local boundaries retained by the broader restart oracle. mid_backfill_download, @@ -53,6 +63,16 @@ pub fn pointName(point: Point) []const u8 { .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", + .after_compaction_rewrite_before_watermark => "after-compaction-rewrite-before-watermark", + .after_compaction_chunk_watermark => "after-compaction-chunk-watermark", + .after_segment_rewrite_temp_written => "after-segment-rewrite-temp-written", + .after_segment_rewrite_temp_synced => "after-segment-rewrite-temp-synced", + .after_segment_rewrite_renamed => "after-segment-rewrite-renamed", + .after_segment_rewrite_dir_synced => "after-segment-rewrite-dir-synced", + .after_segment_patch_temp_written => "after-segment-patch-temp-written", + .after_segment_patch_temp_synced => "after-segment-patch-temp-synced", + .after_segment_patch_renamed => "after-segment-patch-renamed", + .after_segment_patch_dir_synced => "after-segment-patch-dir-synced", else => @tagName(point), }; } diff --git a/src/internal/homepage.zig b/src/internal/homepage.zig index 233ec61..33c8c2f 100644 --- a/src/internal/homepage.zig +++ b/src/internal/homepage.zig @@ -227,6 +227,7 @@ pub fn renderStatus(buf: []u8, args: StatusArgs) []const u8 { " 20/20 live protocol checks\n" ++ " 803,987/803,987 rows drained by the official Go V2 client\n" ++ " 8/8 lifecycle crashpoints recovered\n" ++ + " 10/10 compaction/import power cuts recovered\n" ++ " 16 throttled archive downloads stayed below 40 MiB RSS\n\n" ++ "still open\n" ++ " {s}\n\n" ++ diff --git a/src/internal/segment_patch.zig b/src/internal/segment_patch.zig index 2b4a287..7aadfe0 100644 --- a/src/internal/segment_patch.zig +++ b/src/internal/segment_patch.zig @@ -8,6 +8,7 @@ //! + parent-directory fsync. const std = @import("std"); +const crashpoint = @import("crashpoint.zig"); const gloom = @import("gloom.zig"); const segment = @import("segment.zig"); const writer = @import("segment_writer.zig"); @@ -229,13 +230,17 @@ pub fn patchFile( dir.deleteFile(io, tmp_name) catch {}; } try segment_io.writeStreamingAll(&file, io, new_bytes); + crashpoint.hit(.after_segment_patch_temp_written); try segment_io.sync(&file, io); + crashpoint.hit(.after_segment_patch_temp_synced); file.close(io); file_open = false; try segment_io.rename(dir, tmp_name, dir, name, io); + crashpoint.hit(.after_segment_patch_renamed); const dir_file: Io.File = .{ .handle = dir.handle, .flags = .{ .nonblocking = false } }; var sync_file = dir_file; try segment_io.sync(&sync_file, io); + crashpoint.hit(.after_segment_patch_dir_synced); return result; } diff --git a/src/internal/segment_rewrite.zig b/src/internal/segment_rewrite.zig index 66a7fee..0d802b2 100644 --- a/src/internal/segment_rewrite.zig +++ b/src/internal/segment_rewrite.zig @@ -15,6 +15,7 @@ //! ("clean" — rewritten=false). const std = @import("std"); +const crashpoint = @import("crashpoint.zig"); const footer_mod = @import("segment_footer.zig"); const gloom = @import("gloom.zig"); const segment = @import("segment.zig"); @@ -165,15 +166,19 @@ pub fn rewriteFile( dir.deleteFile(io, tmp_name) catch {}; } try segment_io.writeStreamingAll(&f, io, new_bytes); + crashpoint.hit(.after_segment_rewrite_temp_written); try segment_io.sync(&f, io); + crashpoint.hit(.after_segment_rewrite_temp_synced); f.close(io); file_open = false; try segment_io.rename(dir, tmp_name, dir, name, io); + crashpoint.hit(.after_segment_rewrite_renamed); // make the rename durable: fsync the directory fd (std.Io.Dir has no // sync; the Threaded backend's Dir handle is a posix fd) const dir_file: Io.File = .{ .handle = dir.handle, .flags = .{ .nonblocking = false } }; var sync_file = dir_file; try segment_io.sync(&sync_file, io); + crashpoint.hit(.after_segment_rewrite_dir_synced); return result; } diff --git a/src/internal/timestamp/bucket.zig b/src/internal/timestamp/bucket.zig index 2af32f6..902782e 100644 --- a/src/internal/timestamp/bucket.zig +++ b/src/internal/timestamp/bucket.zig @@ -288,7 +288,10 @@ pub const Bucketer = struct { if (first_error) |err| return err; try syncDir(self.io, self.job_dir); const parent_path = std.fs.path.dirname(self.job_path) orelse "."; - var parent = try Io.Dir.cwd().openDir(self.io, parent_path, .{}); + // On Linux Zig's default directory handle may be O_PATH, which cannot + // be fsynced (EBADF/Unexpected). Request an iterable directory so the + // parent-dir durability barrier uses a real readable fd. + var parent = try Io.Dir.cwd().openDir(self.io, parent_path, .{ .iterate = true }); defer parent.close(self.io); try syncDir(self.io, parent); self.closed = true; diff --git a/src/internal/timestamp/manager.zig b/src/internal/timestamp/manager.zig index e86d6ae..c61b84f 100644 --- a/src/internal/timestamp/manager.zig +++ b/src/internal/timestamp/manager.zig @@ -12,6 +12,7 @@ const runner = @import("runner.zig"); const Allocator = std.mem.Allocator; const Io = std.Io; +const log = std.log.scoped(.stream_timestamp_import); pub const Config = struct { allocator: Allocator, @@ -162,7 +163,9 @@ pub const Manager = struct { var config = self.runner_config; config.scratch_root = self.scratch_root; config.cancelled = &self.cancelled; - runner.run(config, self.active_id.?) catch {}; + runner.run(config, self.active_id.?) catch |err| { + log.err("timestamp import {s} stopped: {s}", .{ self.active_id.?, @errorName(err) }); + }; self.mutex.lockUncancelable(self.io); self.active = false; self.mutex.unlock(self.io); diff --git a/src/internal/timestamp/runner.zig b/src/internal/timestamp/runner.zig index a8705c5..7cb89b2 100644 --- a/src/internal/timestamp/runner.zig +++ b/src/internal/timestamp/runner.zig @@ -17,6 +17,7 @@ const metrics = @import("../metrics.zig"); const Allocator = std.mem.Allocator; const Io = std.Io; +const log = std.log.scoped(.stream_timestamp_import); pub const Config = struct { allocator: Allocator, @@ -101,14 +102,26 @@ fn runInner(config: Config, record: *jobs_mod.Record) !void { var bucket_parse_observed = false; defer if (!bucket_parse_observed) observeParse(config, &rule_result.parse); if (rule_error) |err| return err; + log.info("timestamp import {s}: rules committed", .{record.id}); // Rules are live before this boundary. Pre-existing active rows become // sealed and enter the catalog; later rows are stamped at append time. - try config.archive.rotate(); - var catalog = try bucket.Catalog.open(config.allocator, config.io, config.archive.dir); + config.archive.rotate() catch |err| { + log.err("timestamp import {s}: archive rotate failed: {s}", .{ record.id, @errorName(err) }); + return err; + }; + log.info("timestamp import {s}: archive rotated", .{record.id}); + var catalog = bucket.Catalog.open(config.allocator, config.io, config.archive.dir) catch |err| { + log.err("timestamp import {s}: catalog open failed: {s}", .{ record.id, @errorName(err) }); + return err; + }; defer catalog.deinit(); - var bucketer = try bucket.Bucketer.init(config.allocator, config.io, &catalog, job_path, .{}); + var bucketer = bucket.Bucketer.init(config.allocator, config.io, &catalog, job_path, .{}) catch |err| { + log.err("timestamp import {s}: bucketer init failed: {s}", .{ record.id, @errorName(err) }); + return err; + }; defer bucketer.deinit(); + log.info("timestamp import {s}: bucketer initialized", .{record.id}); var csv = try std.Io.Dir.cwd().openFile(config.io, record.csv_path, .{}); defer csv.close(config.io); var read_buffer: [64 * 1024]u8 = undefined; @@ -148,12 +161,19 @@ fn runInner(config: Config, record: *jobs_mod.Record) !void { bucketer.statsSnapshot().rows_routed, .monotonic, ); - if (parse_error) |err| return err; - if (close_error) |err| return err; + if (parse_error) |err| { + log.err("timestamp import {s}: CSV bucketing failed: {s}", .{ record.id, @errorName(err) }); + return err; + } + if (close_error) |err| { + log.err("timestamp import {s}: bucket durability close failed: {s}", .{ record.id, @errorName(err) }); + return err; + } record.bucketed = true; record.phase = .apply; record.segments_to_apply = bucketer.statsSnapshot().segments_touched; try config.jobs.put(record.*); + log.info("timestamp import {s}: bucketing committed ({d} segments)", .{ record.id, record.segments_to_apply }); } if (isCancelled(config)) return error.ImportCancelled; @@ -189,6 +209,7 @@ fn runInner(config: Config, record: *jobs_mod.Record) !void { item.segment_name, &plan, ); + log.info("timestamp import {s}: segment {d} patch completed (patched={})", .{ record.id, item.index, patch.patched }); const plan_stats = plan.stats(); const applied = plan.applied(); record.segments_examined += 1; diff --git a/tests/powerloss_oracle.py b/tests/powerloss_oracle.py index c4bc2ad..5dc7218 100644 --- a/tests/powerloss_oracle.py +++ b/tests/powerloss_oracle.py @@ -25,6 +25,7 @@ import subprocess import tempfile import time import urllib.error +import urllib.parse import urllib.request import websockets @@ -33,6 +34,8 @@ 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() +FIXTURE = pathlib.Path(os.environ.get("STREAM_POWERLOSS_FIXTURE", "/tmp/stream-powerloss-fixture.jss")).resolve() +IMPORT_CSV = ROOT / "tests" / "timestamp-import.csv" 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") @@ -51,6 +54,26 @@ CASES = [ ("after-steady-phase-before-steady-run", 1), ] +COMPACTION_CASES = [ + "after-segment-rewrite-temp-written", + "after-segment-rewrite-temp-synced", + "after-segment-rewrite-renamed", + "after-segment-rewrite-dir-synced", + "after-compaction-rewrite-before-watermark", + "after-compaction-chunk-watermark", +] + +PATCH_CASES = [ + "after-segment-patch-temp-written", + "after-segment-patch-temp-synced", + "after-segment-patch-renamed", + "after-segment-patch-dir-synced", +] + +FIXTURE_EVENTS = 2500 +COMPACTED_EVENTS = 1507 +IMPORTED_TIMESTAMP_US = 1_704_164_645_000_000 + def run(args, *, check=True, capture=True, timeout=60): result = subprocess.run( @@ -115,6 +138,20 @@ def http_json(path): return None, None +def import_json(path): + request = urllib.request.Request( + BASE + path, + headers={"Authorization": "Bearer powerloss-test-token"}, + ) + try: + with urllib.request.urlopen(request, 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: @@ -126,6 +163,43 @@ def http_status(path): return None +def http_text(path): + try: + with urllib.request.urlopen(BASE + path, timeout=3) as response: + return response.status, response.read().decode() + except urllib.error.HTTPError as error: + return error.code, error.read().decode(errors="replace") + except OSError: + return None, "" + + +def post_import(): + request = urllib.request.Request( + BASE + "/xrpc/network.bsky.jetstream.importTimestamps", + data=json.dumps({"path": "job.csv"}).encode(), + headers={ + "Authorization": "Bearer powerloss-test-token", + "Content-Type": "application/json", + }, + method="POST", + ) + with urllib.request.urlopen(request, timeout=5) as response: + body = json.loads(response.read()) + assert response.status == 200 and body.get("job"), body + return body["job"] + + +def metric_value(name): + status, text = http_text("/metrics") + if status != 200: + return None + prefix = name + " " + for line in text.splitlines(): + if line.startswith(prefix): + return int(float(line[len(prefix):])) + return None + + def wait_serving(log_path, timeout=300): deadline = time.monotonic() + timeout while time.monotonic() < deadline: @@ -141,7 +215,7 @@ def wait_serving(log_path, timeout=300): async def replay(count, *, require_unique=False): previous = 0 - fingerprints = set() + fingerprints = {} url = f"ws://127.0.0.1:{PORT}/subscribe-v2?cursor=0" async with websockets.connect(url) as websocket: for _ in range(count): @@ -154,8 +228,11 @@ async def replay(count, *, require_unique=False): 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) + assert fingerprint not in fingerprints, ( + f"duplicate recovered event at seq {seq}; " + f"first local seq {fingerprints.get(fingerprint)}; event={semantic}" + ) + fingerprints[fingerprint] = seq previous = seq @@ -249,6 +326,183 @@ def start_stream(log_name, point=None, ordinal=1): run(["docker", "exec", "-d", CONTAINER, "sh", "-lc", command]) +def mutation_stream_args(kind, point=None): + interval = 1 if kind == "compaction" else 0 + args = ( + "/oracle/stream --port=6099 --data-dir=/mnt/strict/data " + "--upstream=ws://127.0.0.1:1 --relay-http=http://127.0.0.1:1 " + "--plc=http://127.0.0.1:1 --retry-interval=0 --no-verify " + f"--compaction-interval={interval} " + "--timestamp-import-dir=/mnt/strict/imports " + "--timestamp-import-token=powerloss-test-token" + ) + if point: + args += f" --power-cutpoint={point}:1" + return args + + +def start_mutation_stream(log_name, kind, point=None): + command = ( + f"echo $$ >/state/stream.pid; exec {mutation_stream_args(kind, point)} " + f">/state/{log_name} 2>&1" + ) + run(["docker", "exec", "-d", CONTAINER, "sh", "-lc", command]) + + +def prepare_mutation_filesystem(): + 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 -p /mnt/strict/data/segments /mnt/strict/imports; " + "cp /oracle/fixture.jss /mnt/strict/data/segments/seg_0000000000.jss; " + "cp /oracle/timestamp-import.csv /mnt/strict/imports/job.csv; sync" + ) + + +def cut_power_and_recover(): + 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 mutation cut:\n{fsck.stdout}" + dexec_sh(f"mount -o barrier=1,commit=600 {DEVICE} /mnt/strict") + + +def stop_mutation_case(): + 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 sealed_archive_identity(): + encoded = dexec_sh( + "python3 - <<'PY'\n" + "import glob, hashlib, json\n" + "total = 0\n" + "digest = hashlib.sha256()\n" + "sealed = 0\n" + "for path in sorted(glob.glob('/mnt/strict/data/segments/*.jss')):\n" + " data = open(path, 'rb').read()\n" + " if len(data) < 22 or data[:4] != b'jss0' or int.from_bytes(data[4:12], 'little') == 0:\n" + " continue\n" + " sealed += 1\n" + " total += int.from_bytes(data[18:22], 'little')\n" + " digest.update(path.rsplit('/', 1)[-1].encode() + b'\\0')\n" + " digest.update(data)\n" + "print(json.dumps({'segments': sealed, 'events': total, 'sha256': digest.hexdigest()}))\n" + "PY" + ).stdout + return json.loads(encoded) + + +def wait_compacted(log_path, timeout=120): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + listing = wait_serving(log_path, timeout=5) + events = sum(item.get("eventCount", 0) for item in listing.get("segments", [])) + if metric_value("jetstream_compaction_watermark_seq") == FIXTURE_EVENTS and events == COMPACTED_EVENTS: + return listing + time.sleep(0.2) + raise AssertionError(f"compaction did not converge\n{log_path.read_text(errors='replace')}") + + +def run_compaction_powerloss_case(state, point): + prepare_mutation_filesystem() + start_mutation_stream("mutation-first.log", "compaction", point) + marker = f"power cutpoint {point} ordinal 1: ready" + first_log = wait_log(state / "mutation-first.log", marker, timeout=120) + assert first_log.count(marker) == 1 + assert len((state / "receipt").read_text().splitlines()) > 0 + cut_power_and_recover() + start_mutation_stream("mutation-recovery.log", "compaction") + wait_compacted(state / "mutation-recovery.log") + dexec_sh("test ! -e /mnt/strict/data/segments/seg_0000000000.jss.tmp") + asyncio.run(replay(COMPACTED_EVENTS, require_unique=True)) + print(f"powerloss: {point} converged to {COMPACTED_EVENTS} compacted events") + stop_mutation_case() + + +async def verify_timestamp_import(): + targeted = 0 + untouched = 0 + url = f"ws://127.0.0.1:{PORT}/subscribe-v2?cursor=0" + async with websockets.connect(url) as websocket: + for _ in range(FIXTURE_EVENTS): + event = json.loads(await asyncio.wait_for(websocket.recv(), timeout=15)) + commit = event.get("commit", {}) + materialization = event.get("kind") == "commit" and commit.get("operation") != "delete" + if materialization and event.get("did") == "did:plc:streamsampleaccount" and commit.get("collection") == "app.bsky.feed.post": + assert event.get("time_us") == IMPORTED_TIMESTAMP_US, event + targeted += 1 + elif materialization: + expected = 1_700_000_000_000_000 + event["seq"] * 1000 + assert event.get("time_us") == expected, event + untouched += 1 + assert targeted > 0 and untouched > 0, (targeted, untouched) + + +def wait_import_complete(log_path, timeout=120): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + status, body = import_json("/xrpc/network.bsky.jetstream.getImportStatus") + if status == 200 and body.get("state") == "complete": + return body + alive = dexec_sh("kill -0 $(cat /state/stream.pid)", check=False) + if alive.returncode != 0: + raise AssertionError(f"import recovery exited\n{log_path.read_text(errors='replace')}") + time.sleep(0.2) + raise AssertionError(f"timestamp import did not complete\n{log_path.read_text(errors='replace')}") + + +def wait_patch_marker(log_path, marker, job, timeout=120): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if log_path.exists(): + text = log_path.read_text(errors="replace") + if marker in text: + return text + status, body = import_json( + "/xrpc/network.bsky.jetstream.getImportStatus?job=" + + urllib.parse.quote(job) + ) + if status == 200 and body.get("state") == "failed": + raise AssertionError(f"timestamp import failed before {marker}: {body}") + alive = dexec_sh("kill -0 $(cat /state/stream.pid)", check=False) + if alive.returncode != 0: + raise AssertionError(f"Stream exited before {marker}\n{log_path.read_text(errors='replace')}") + time.sleep(0.1) + raise AssertionError(f"timed out waiting for {marker}\n{log_path.read_text(errors='replace')}") + + +def run_patch_powerloss_case(state, point): + prepare_mutation_filesystem() + start_mutation_stream("patch-first.log", "patch", point) + wait_serving(state / "patch-first.log") + job = post_import() + marker = f"power cutpoint {point} ordinal 1: ready" + first_log = wait_patch_marker(state / "patch-first.log", marker, job, timeout=120) + assert first_log.count(marker) == 1 + cut_power_and_recover() + start_mutation_stream("patch-recovery.log", "patch") + wait_serving(state / "patch-recovery.log") + result = wait_import_complete(state / "patch-recovery.log") + assert result.get("segmentsApplied", 0) > 0, result + dexec_sh("test ! -e /mnt/strict/data/segments/seg_0000000000.jss.tmp") + asyncio.run(verify_timestamp_import()) + print(f"powerloss: {point} timestamp import recovered") + stop_mutation_case() + + def run_case(state, point, ordinal): label = f"{point}-ordinal-{ordinal}" dexec_sh("rm -rf /state/*; mkdir -p /mnt/strict") @@ -332,18 +586,8 @@ def run_restart_after_cleanup_guard_case(state): 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) + before = sealed_archive_identity() + before_count = before["events"] 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) @@ -374,7 +618,9 @@ def run_restart_after_cleanup_guard_case(state): 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)) + after = sealed_archive_identity() + assert after == before, f"guard recovery changed sealed archive bytes: before={before} after={after}" + asyncio.run(replay(events)) dexec_sh( "test ! -e /mnt/strict/data/backfill; " "test ! -e /mnt/strict/data/phase; test ! -e /mnt/strict/data/repos.log" @@ -397,6 +643,8 @@ def main(): docker_image_ready() if not BIN.is_file(): raise SystemExit(f"Linux Stream binary missing: {BIN}") + if not FIXTURE.is_file(): + raise SystemExit(f"power-loss JSS fixture missing: {FIXTURE}") with socket.socket() as sock: try: sock.bind(("127.0.0.1", PORT)) @@ -404,6 +652,12 @@ def main(): raise SystemExit(f"power-loss oracle port {PORT} is busy: {error}") state_dir = pathlib.Path(tempfile.mkdtemp(prefix="stream-powerloss-")) + scope = os.environ.get("STREAM_POWERLOSS_SCOPE", "all") + if scope not in ("all", "lifecycle", "guard", "mutation"): + raise SystemExit("STREAM_POWERLOSS_SCOPE must be all, lifecycle, guard, or mutation") + mutation_kind = os.environ.get("STREAM_POWERLOSS_MUTATION_KIND", "all") + if mutation_kind not in ("all", "compaction", "patch"): + raise SystemExit("STREAM_POWERLOSS_MUTATION_KIND must be all, compaction, or patch") succeeded = False try: run([ @@ -411,19 +665,28 @@ def main(): "--name", CONTAINER, "-p", f"127.0.0.1:{PORT}:6099", "-v", f"{BIN}:/oracle/stream:ro", + "-v", f"{FIXTURE}:/oracle/fixture.jss:ro", + "-v", f"{IMPORT_CSV}:/oracle/timestamp-import.csv:ro", "-v", f"{ROOT / 'tests' / 'strict_nbd.py'}:/oracle/strict_nbd.py:ro", "-v", f"{state_dir}:/state", IMAGE, "sleep", "infinity", ]) - require_simulator() + if scope != "mutation": + 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" - ) + if scope in ("all", "lifecycle"): + for point, ordinal in CASES: + run_case(state_dir, point, ordinal) + if scope in ("all", "lifecycle", "guard"): + run_restart_after_cleanup_guard_case(state_dir) + if scope in ("all", "mutation"): + if mutation_kind != "patch": + for point in COMPACTION_CASES: + run_compaction_powerloss_case(state_dir, point) + if mutation_kind != "compaction": + for point in PATCH_CASES: + run_patch_powerloss_case(state_dir, point) + print(f"powerloss: selected {scope} schedules recovered") succeeded = True finally: dexec_sh("pkill -9 stream 2>/dev/null || true; pkill -9 nbdkit 2>/dev/null || true", check=False) diff --git a/tests/timestamp-import.csv b/tests/timestamp-import.csv new file mode 100644 index 0000000..0432d9c --- /dev/null +++ b/tests/timestamp-import.csv @@ -0,0 +1,2 @@ +uri,timestamp +at://did:plc:streamsampleaccount/app.bsky.feed.post/3l3qo2vuowo2b,2024-01-02T03:04:05Z -- 2.51.2