From bdb9a793c29de21abe8911298d89f30655408250 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Fri, 24 Jul 2026 02:27:54 -0500 Subject: [PATCH] fail closed on merge restart state --- docs/bootstrap-semantic-parity.md | 6 ++-- docs/semantic-parity.md | 2 +- docs/upstream-bootstrap-spec.md | 10 +++--- src/internal/bootstrap/lifecycle.zig | 41 +++++++++++++++++---- src/internal/bootstrap/merge.zig | 51 ++++++++++++++++++++++++--- src/internal/bootstrap/repo_store.zig | 9 +++-- src/internal/meta_store.zig | 11 ++++++ 7 files changed, 108 insertions(+), 22 deletions(-) diff --git a/docs/bootstrap-semantic-parity.md b/docs/bootstrap-semantic-parity.md index cef0273..e3785bd 100644 --- a/docs/bootstrap-semantic-parity.md +++ b/docs/bootstrap-semantic-parity.md @@ -60,7 +60,7 @@ and retry behavior; those still prevent another deployment. | listRepos cursor read | **verified** | `loadRelayListReposCursor` propagates RocksDB errors and distinguishes a missing key. | Valid cursor length remains artificially bounded by fixed buffers during enumeration; initial enumeration fails loud, while merge discovery silently stops. | | Lifecycle phase read | **verified** | Phase writes are synced. Reads distinguish a missing key from RocksDB failure and reject every value outside `bootstrap`, `merging`, and `steady_state`; focused tests inject both corruption and a real Store read error. The production orchestrator uses the same fallible read before any fresh-directory write. | Persisted phase-entry timing remains a separate gap below. | | Bootstrap → merging commit point | **blocked** | Stream eventually persists merging before entering the merge driver. | Upstream writes merging immediately after backfill drains and before stopping bootstrap-live capture. Stream stops/joins/flushed/seals capture inside `runBootstrap`, returns, and only then writes merging. Port the ordering and test both sides of the write. | -| Merging → steady_state commit point | **partial** | Stream drains, compacts, reconciles the manifest, discovers, removes the source tree, syncs the data directory, deletes cursors, publishes seq metadata, then writes steady_state. Named write-side crash tests cover several seams. | Phase read corruption and the fail-open source-existence guard can enter this sequence from the wrong state. | +| Merging → steady_state commit point | **partial** | Stream drains, compacts, reconciles the manifest, discovers, removes the source tree, syncs the data directory, deletes cursors, publishes seq metadata, then writes steady_state. Phase reads fail closed, and the restart guard distinguishes only `FileNotFound` from every other filesystem error before taking the cleanup-complete path. Named write-side crash tests cover several seams. | Exact persisted phase-entry timing remains absent, and other merge/discovery rows remain partial or blocked. | | Exact persisted phase timing | **blocked** | Metrics record process-local state durations. | Upstream persists/serves phase entry and backfill timing. Stream has no equivalent entered-at state, so status and restart timing cannot be equivalent. | ## Bootstrap-live capture and merge @@ -70,11 +70,11 @@ and retry behavior; those still prevent another deployment. | Concurrent bootstrap-live attachment | **partial** | The production lifecycle starts a separate capture archive and crash/oracle fixtures send live events during bootstrap. | The phase commit ordering differs, and the firehose dependency can terminate during reconnect sleep. | | Bootstrap-live failure propagation | **partial** | The lifecycle now observes an unexpectedly completed capture future and treats it as fatal. | `HttpConnectionClosing` can still escape Zat's reconnect loop and terminate the process; the exact dependency behavior needs a regression test on Linux. | | Merge source existence guard | **blocked** | A genuinely absent source directory is intended to mean cleanup completed before the previous phase write. | Stream converts every `openDir` error to absence. Upstream distinguishes not-found from permission/I/O failures. Inject EACCES/EIO and require startup failure without cursor deletion or phase advancement. | -| Merge source cursor read | **blocked** | Successful source completion commits cursor and latest-revision updates together. | Read errors and malformed values become source zero, permitting duplicate re-drain. | +| Merge source cursor read | **verified** | Successful source completion atomically commits upstream's `[version=1][uint64 LE]` cursor and latest-revision updates. Reads distinguish absence from Store failure and reject wrong width/version; focused tests inject each failure, and startup migrates Stream's prior valid 8-byte cursor. | This row proves cursor decoding and commit atomicity, not every source-file read. | | Merge row filtering | **verified** | Rows are dropped only when the repository is complete and the source revision is at/below its backfill watermark; missing lookups are cached. Kept/dropped fixtures exercise this. | The surrounding cursor recovery remains blocked. | | Latest revision refresh | **partial** | Kept rev-bearing rows update latest revision in the same batch as the source cursor while preserving the backfill revision. | Missing repo rows are skipped; this matches the intended defensive behavior, but corrupt cursor state can replay updates and rows. | | Pending-repository pass | **blocked** | Pending repositories are repaired after captured live rows are merged, preserving intended row order. | Stream materializes pending DIDs and processes them through a separate path rather than upstream's bounded retry runner with the same global/per-host gates and retry semantics. | -| Merge-tail compaction and manifest reconcile | **partial** | Destination sealing precedes delete/update compaction; manifest reconciliation happens before serving is enabled. Physical rewrite tests cover named write failures, and the versioned compaction watermark now propagates Store errors and rejects malformed/unknown encodings without reinitializing the key. | The fail-open merge source-existence guard and other cutover rows remain independent blockers. | +| Merge-tail compaction and manifest reconcile | **partial** | Destination sealing precedes delete/update compaction; manifest reconciliation happens before serving is enabled. Physical rewrite tests cover named write failures; compaction watermark and merge source admission now fail closed on metadata/filesystem errors. | Other cutover and post-bootstrap discovery rows remain independent blockers. | | Post-bootstrap discovery | **blocked** | It resumes from the last non-empty bootstrap cursor and creates failed rows for unknown active repositories on the normal path. | Unknown inactive repositories are skipped instead of recorded. A next cursor longer than 512 bytes silently exits the loop and cleanup proceeds, omitting the remaining network. | | Cleanup durability | **verified** | On the successful path, the backfill tree is removed, the data directory is synced, and merge/discovery cursors are deleted in a synced RocksDB batch afterward. Restart-after-cleanup repeats the directory sync. | This row assumes the source was correctly classified as absent; that guard is separately blocked. | diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index ba0672b..1492f47 100644 --- a/docs/semantic-parity.md +++ b/docs/semantic-parity.md @@ -72,7 +72,7 @@ is in [bootstrap-semantic-parity.md](bootstrap-semantic-parity.md). | OpenTelemetry | **partial** | Provider configuration, propagation, sampling, batching, TLS/mTLS, and representative production spans have real collector tests. | “Every pinned production span” has not been independently re-audited in this pass. Treat the enumerated tested spans as evidence, not a blanket closure. | | Inspect/version command surfaces | **partial** | `version`, sealed `inspect-segment`, active inspection, and representative `inspect-all` reports have fixtures and golden comparisons. | Inspection does not compensate for missing online status tabs, and the current audit did not rerun every golden against the current commit. | | Differential oracle | **blocked as an admission proof** | It provides useful event-log, final-state, restart, repair, and public-client coverage against a pinned upstream simulator. Focused production-boundary tests now prove per-repository writer-gated completion and same-disk reopen with a sibling held incomplete. | The oracle still needs an invariant-based process-kill world that scales this beyond a two-repository fixture; its old four-account `after_repo_complete` schedule could not detect a dispatch-scale replay defect. | -| Strict power-loss oracle | **partial** | It exercises acknowledged-write reconstruction at named write/fsync/rename boundaries. Focused tests separately inject Store read failures for lifecycle phase, relay cursor, and compaction watermark; the completion reopen test proves completed-versus-interrupted repository classification around an unchanged listRepos cursor. | It does not yet inject merge-cursor read failures, missing manifest files during replay, long listRepos cursors, or storage loss specifically between archive fsync and the completion metadata commit. | +| Strict power-loss oracle | **partial** | It exercises acknowledged-write reconstruction at named write/fsync/rename boundaries. Focused tests separately inject Store read failures for lifecycle phase, relay cursor, compaction watermark, and merge cursor; the merge restart guard also distinguishes source absence from filesystem failure. The completion reopen test proves completed-versus-interrupted repository classification around an unchanged listRepos cursor. | It does not yet inject missing manifest files during replay, long listRepos cursors, or storage loss specifically between archive fsync and the completion metadata commit. | | Exact artifact admission | **blocked** | Commit `4d163e8` had a recorded suite and immutable OCI digest. | The experiment deployed `e1926f3`, which included source and dependency changes after that receipt. No mechanism prevented deployment of a non-admitted digest. | ## Evidence that must not be overread diff --git a/docs/upstream-bootstrap-spec.md b/docs/upstream-bootstrap-spec.md index cc968cf..c353589 100644 --- a/docs/upstream-bootstrap-spec.md +++ b/docs/upstream-bootstrap-spec.md @@ -89,9 +89,11 @@ boundary, and fully reopen the store without replaying the completed sibling. The same audit found fail-open phase, relay-cursor, and compaction-watermark reads. All three now fail closed on Store errors and malformed or unknown -encodings, with injected read-fault tests. The audit also found a fail-open merge -source guard, incomplete post-bootstrap discovery, an unbounded retry candidate -scan, and retry-supervisor failures hidden from health. Current Stream status -and the evidence needed to close each remaining gap are maintained in +encodings, with injected read-fault tests. The merge source cursor now uses the +same strict versioned encoding, and the restart-after-cleanup guard treats only +`FileNotFound` as proof of absence rather than swallowing arbitrary filesystem +errors. The audit also found incomplete post-bootstrap discovery, an unbounded +retry candidate scan, and retry-supervisor failures hidden from health. Current +Stream status and the evidence needed to close each remaining gap are maintained in [bootstrap-semantic-parity.md](bootstrap-semantic-parity.md) and [semantic-parity.md](semantic-parity.md). diff --git a/src/internal/bootstrap/lifecycle.zig b/src/internal/bootstrap/lifecycle.zig index 7732ca0..9b25b96 100644 --- a/src/internal/bootstrap/lifecycle.zig +++ b/src/internal/bootstrap/lifecycle.zig @@ -174,13 +174,9 @@ pub fn runTraced( // merging: restart-after-cleanup guard — live dir gone means a prior // run finished the drain and cleanup but died before the phase write - var backfill_dir = Io.Dir.cwd().createDirPathOpen(io, opts.data_dir, .{}) catch unreachable; + var backfill_dir = try Io.Dir.cwd().createDirPathOpen(io, opts.data_dir, .{}); defer backfill_dir.close(io); - const have_live = blk: { - var d = Io.Dir.cwd().openDir(io, live_data_dir, .{}) catch break :blk false; - d.close(io); - break :blk true; - }; + const have_live = try mergeSourceExists(io, live_data_dir); var bf_path_buf: [256]u8 = undefined; const bf_path = try std.fmt.bufPrint(&bf_path_buf, "{s}/backfill", .{opts.data_dir}); if (have_live) { @@ -625,3 +621,36 @@ fn orchestratorPhase(phase: phase_mod.Phase) metrics.OrchestratorPhase { fn elapsedUs(io: Io, started_at: Io.Timestamp) u64 { return @intCast(@max(started_at.durationTo(Io.Timestamp.now(io, .awake)).toMicroseconds(), 0)); } + +fn mergeSourceExists(io: Io, path: []const u8) !bool { + var dir = Io.Dir.cwd().openDir(io, path, .{}) catch |err| switch (err) { + error.FileNotFound => return false, + else => return err, + }; + dir.close(io); + return true; +} + +test "merge restart guard distinguishes missing source from filesystem failure" { + const testing = std.testing; + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var root_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const root_len = try tmp.dir.realPath(testing.io, &root_buf); + const root = root_buf[0..root_len]; + var missing_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const missing = try std.fmt.bufPrint(&missing_buf, "{s}/missing", .{root}); + try testing.expect(!try mergeSourceExists(testing.io, missing)); + + var present_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const present = try std.fmt.bufPrint(&present_buf, "{s}/present", .{root}); + var present_dir = try Io.Dir.cwd().createDirPathOpen(testing.io, present, .{}); + present_dir.close(testing.io); + try testing.expect(try mergeSourceExists(testing.io, present)); + + var not_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const not_dir = try std.fmt.bufPrint(¬_dir_buf, "{s}/not-a-directory", .{root}); + var file = try Io.Dir.cwd().createFile(testing.io, not_dir, .{}); + file.close(testing.io); + try testing.expectError(error.NotDir, mergeSourceExists(testing.io, not_dir)); +} diff --git a/src/internal/bootstrap/merge.zig b/src/internal/bootstrap/merge.zig index 143cb6e..263c920 100644 --- a/src/internal/bootstrap/merge.zig +++ b/src/internal/bootstrap/merge.zig @@ -121,7 +121,7 @@ pub fn runTraced( } std.mem.sort(u64, indices.items, {}, std.sort.asc(u64)); - const from_idx = loadCursor(store); + const from_idx = try loadCursor(store); var stats: Stats = .{}; var cache = RepoStatusCache.init(allocator, store, metric_stats); defer cache.deinit(); @@ -220,13 +220,15 @@ fn drainOne( } const cursor_key = "merge/next_source_idx"; +const cursor_version: u8 = 1; -fn loadCursor(store: *repo_store.Store) u64 { - const maybe_bytes = store.meta.getAlloc(store.allocator, cursor_key) catch return 0; +fn loadCursor(store: *repo_store.Store) !u64 { + const maybe_bytes = try store.meta.getAlloc(store.allocator, cursor_key); const bytes = maybe_bytes orelse return 0; defer store.allocator.free(bytes); - if (bytes.len != 8) return 0; - return std.mem.readInt(u64, bytes[0..8], .little); + if (bytes.len != 9) return error.InvalidMergeCursorLength; + if (bytes[0] != cursor_version) return error.UnsupportedMergeCursorVersion; + return std.mem.readInt(u64, bytes[1..9], .little); } // === tests === @@ -267,6 +269,45 @@ test "shouldKeep mirrors upstream merge_filter" { try testing.expect(shouldKeep(row(.create, "did:plc:x", "3l3qo2vutsw2a"), .{ .status = .complete, .rev = "" })); } +test "merge cursor corruption and storage errors fail closed" { + const meta_store = @import("../meta_store.zig"); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var path_buf: [std.fs.max_path_bytes]u8 = undefined; + const path_len = try tmp.dir.realPath(testing.io, &path_buf); + var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]); + defer meta.deinit(); + var store = try repo_store.Store.init(testing.allocator, testing.io, &meta); + defer store.deinit(); + + try testing.expectEqual(@as(u64, 0), try loadCursor(&store)); + try meta.putDurable(cursor_key, &.{ 1, 2, 3 }); + try testing.expectError(error.InvalidMergeCursorLength, loadCursor(&store)); + try meta.putDurable(cursor_key, &.{ 2, 0, 0, 0, 0, 0, 0, 0, 0 }); + try testing.expectError(error.UnsupportedMergeCursorVersion, loadCursor(&store)); + try testing.expect(meta_store.armReadFault(cursor_key, 1)); + defer meta_store.disarmReadFault(); + try testing.expectError(error.InjectedStoreFault, loadCursor(&store)); +} + +test "legacy bare merge cursor is migrated before strict decoding" { + const meta_store = @import("../meta_store.zig"); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var path_buf: [std.fs.max_path_bytes]u8 = undefined; + const path_len = try tmp.dir.realPath(testing.io, &path_buf); + const path = path_buf[0..path_len]; + var meta = try meta_store.Store.open(testing.allocator, path); + defer meta.deinit(); + var legacy: [8]u8 = undefined; + std.mem.writeInt(u64, &legacy, 73, .little); + try meta.putDurable(cursor_key, &legacy); + try meta.migrateLegacy(testing.io, path); + var store = try repo_store.Store.init(testing.allocator, testing.io, &meta); + defer store.deinit(); + try testing.expectEqual(@as(u64, 73), try loadCursor(&store)); +} + test "merge repo status cache performs one real RocksDB lookup per DID including misses" { const meta_store = @import("../meta_store.zig"); var tmp = testing.tmpDir(.{}); diff --git a/src/internal/bootstrap/repo_store.zig b/src/internal/bootstrap/repo_store.zig index 19e74a3..4f61973 100644 --- a/src/internal/bootstrap/repo_store.zig +++ b/src/internal/bootstrap/repo_store.zig @@ -334,8 +334,9 @@ pub const Store = struct { var writes = self.meta.batch(); defer writes.deinit(); - var cursor_bytes: [8]u8 = undefined; - std.mem.writeInt(u64, &cursor_bytes, next_idx, .little); + var cursor_bytes: [9]u8 = undefined; + cursor_bytes[0] = 1; + std.mem.writeInt(u64, cursor_bytes[1..9], next_idx, .little); writes.put(cursor_key, &cursor_bytes); for (updates) |update| { const key = try repoKey(self.allocator, update.did); @@ -895,7 +896,9 @@ test "merge source commit refreshes only latest repo state with its cursor" { const cursor = (try meta.getAlloc(testing.allocator, "merge/next_source_idx")).?; defer testing.allocator.free(cursor); - try testing.expectEqual(@as(u64, 7), std.mem.readInt(u64, cursor[0..8], .little)); + try testing.expectEqual(@as(usize, 9), cursor.len); + try testing.expectEqual(@as(u8, 1), cursor[0]); + try testing.expectEqual(@as(u64, 7), std.mem.readInt(u64, cursor[1..9], .little)); const known = (try store.get(testing.allocator, "did:plc:known")).?; defer store.freeState(testing.allocator, known); try testing.expectEqualStrings("backfill-rev", known.rev); diff --git a/src/internal/meta_store.zig b/src/internal/meta_store.zig index 337d5ed..c2906d9 100644 --- a/src/internal/meta_store.zig +++ b/src/internal/meta_store.zig @@ -173,6 +173,17 @@ pub const Store = struct { if (value.len == bytes.len) try self.putDurable("compaction/seq", value); } else |_| {} } + + if (try self.getAlloc(self.allocator, "merge/next_source_idx")) |existing| { + defer self.allocator.free(existing); + if (existing.len == 8) { + var versioned: [9]u8 = undefined; + versioned[0] = 1; + @memcpy(versioned[1..9], existing); + try self.putDurable("merge/next_source_idx", &versioned); + log.info("migrated merge cursor to versioned encoding", .{}); + } + } } pub fn getAlloc(self: *Store, allocator: Allocator, key: []const u8) !?[]u8 { -- 2.51.2