diff --git a/docs/bootstrap-semantic-parity.md b/docs/bootstrap-semantic-parity.md index 10254ff..c1ae0fc 100644 --- a/docs/bootstrap-semantic-parity.md +++ b/docs/bootstrap-semantic-parity.md @@ -35,7 +35,7 @@ and retry behavior; those still prevent another deployment. | Terminal repository errors | **verified** | Structured RepoNotFound becomes complete with no rows; deactivated/suspended/takendown become unavailable. Focused tests inspect the durable status. | None known for the classified responses. | | Inactive repository during initial discovery | **verified** | A new inactive entry is stored as not_started/active=false and is not dispatched; later active flips preserve the row. | Post-bootstrap discovery does not preserve the same rule; see below. | | Explicit selected backfill | **partial** | The selected DID list bypasses listRepos, preserves caller order, conflicts with max-repos, clears stale discovery state, resolves identity metadata, and maintains the handle index in focused tests. | The exact current artifact has not rerun the full selected-repo receipt, and selected completion still uses the blocked batch completion machinery. | -| Max-repos debug mode | **partial** | The flag truncates selected work and avoids committing a resumable whole-network cursor. | Its recovery behavior remains coupled to blocked phase/cursor reads. It is not evidence for whole-network behavior. | +| Max-repos debug mode | **partial** | The flag truncates selected work and avoids committing a resumable whole-network cursor. Phase and relay-cursor reads now fail closed. | It remains a debug path and is not evidence for whole-network discovery or merge behavior. | ## Repository emission and durable completion @@ -56,9 +56,9 @@ and retry behavior; those still prevent another deployment. | Invariant | Status | Audited evidence | Gap or required proof | |---|---|---|---| | Relay listRepos cursor checkpoint | **verified** | Stream atomically writes the relay cursor and last non-empty discovery cursor only after the dispatch batch drains. Final empty cursor handling has unit coverage, while the straggler/reopen test proves individual repository completion can become durable with the page cursor still absent. | The 100,000-entry unit is checkpoint/re-enumeration cadence, not repository completion durability. | -| Relay firehose cursor read | **blocked** | Durable cursor writes are coupled to archive metadata batches on the normal path. | Read error, malformed width, or a negative value is accepted as “no cursor”/a value. Upstream rejects corruption and negative encoding to avoid skipping the interval to current relay tip. | +| Relay firehose cursor read | **verified** | Durable cursor writes remain coupled to archive metadata batches. Reads now propagate RocksDB errors and require upstream's `[version=1][uint64 LE]` encoding, rejecting wrong width, unknown version, and values above `MaxInt64`; writes reject negative cursors. Focused tests exercise every rejection, and a startup migration converts Stream's prior valid 8-byte encoding without accepting other malformed values. | This row does not close the independent Zat reconnect failure. | | 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 | **blocked** | Phase writes are synced. | RocksDB read failure, malformed bytes, and an unknown phase all return null and are treated as a fresh data directory. Upstream propagates read/decode errors. | +| 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. | | 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. | diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index 7d50f67..dbc8555 100644 --- a/docs/semantic-parity.md +++ b/docs/semantic-parity.md @@ -57,8 +57,8 @@ is in [bootstrap-semantic-parity.md](bootstrap-semantic-parity.md). | `planBackfill` | **blocked** | Normal DID/collection/sequence filtering, whole-segment versus block mode, pagination, and configured limits match the pinned client fixtures. | `matchBlocks` errors are handled with `catch continue`, creating a successful plan with a false negative. Upstream's planner returns the error. Add an injected allocation/internal-error contract. | | Subscribe v1/v2 handshake and wire formats | **partial** | Cursor parsing, v1/v2 filter shapes, WebSocket framing, v1 deflate, v2 dictionary zstd, dictionary validators, size caps, pings, write deadlines, and shutdown close frames have production-path tests. | These tests do not establish gap-free cold replay or remote-data encoding isolation. | | Cold archive replay | **blocked** | Normal sealed and active-prefix replay reaches the hot tail in order in existing fixtures. Timestamp cursor corruption is surfaced for the selected block. | Both cold walkers skip a manifest-listed segment when `openFile` fails. They may then advance beyond durable events never delivered. Upstream returns an error and enforces progress/hole invariants below the readable-log floor. | -| Live Sync 1.1 verification | **partial** | DID/signature checks, MST inversion, op-CID checks, replay/future-rev gates, durable chain/hosting updates, and several repair-trigger cases use real RocksDB, CARs, and sockets. | The surface cannot be called closed while an accepted live row can persist and then kill the process during eager wire encoding. Metadata cursor corruption also fails open. | -| Live scheduling and cursor durability | **partial** | Per-DID ordering, bounded pending work, worker admission, completion-order emission, and archive-before-cursor write ordering are represented in production code and focused tests. | The Zat reconnect sleep can propagate `HttpConnectionClosing`; production observed two process exits. Relay cursor load errors/malformed values become absence, which can omit the interval between the durable cursor and current relay tip. | +| Live Sync 1.1 verification | **partial** | DID/signature checks, MST inversion, op-CID checks, replay/future-rev gates, durable chain/hosting updates, and several repair-trigger cases use real RocksDB, CARs, and sockets. Relay cursor corruption and read failures now fail closed. | The surface cannot be called closed while an accepted live row can persist and then kill the process during eager wire encoding. | +| Live scheduling and cursor durability | **partial** | Per-DID ordering, bounded pending work, worker admission, completion-order emission, archive-before-cursor write ordering, strict versioned cursor decoding, and cursor-read failure propagation are represented in production code and focused tests. | The Zat reconnect sleep can propagate `HttpConnectionClosing`; production observed two process exits. | | Whole-repository repair/resync | **partial** | Fetched repositories are authenticated before replacement emission; valid repair emits sync then bounded replacement batches; repair-tail encoding errors are isolated per row. Capacity tests exercise 32 active plus 64 queued repairs. | The isolation applies only to repair-tail publication. Ordinary live publication still eagerly encodes and can terminate the process. Retry-subsystem terminal failures are not tied to health. | | Delete/update compaction | **partial** | Rewrite selection, survivor correctness, manifest-before-watermark ordering, fsync/rename cutpoints, bounded workers, and several reason counters have physical-file tests. | Compaction watermark read/corruption is treated as missing and reinitialized. A read-error fault contract against the production metadata store is required. | | Timestamp import | **partial** | CSV parsing, rule precedence, patch topology preservation, job persistence, restart, HTTP authentication, and write/fsync/rename power-loss cuts have concrete tests. | Public visibility after import still depends on the blocked cold-replay path. No audit has yet proved behavior when timestamp metadata reads are corrupt rather than absent. | @@ -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, and the focused reopen test proves completed-versus-interrupted repository classification around an unchanged listRepos cursor. | It does not inject correctness-metadata 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 and relay cursor, and the completion reopen test proves completed-versus-interrupted repository classification around an unchanged listRepos cursor. | It does not yet inject compaction/merge metadata read failures, 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 679dcd7..61b8f8b 100644 --- a/docs/upstream-bootstrap-spec.md +++ b/docs/upstream-bootstrap-spec.md @@ -87,9 +87,11 @@ upstream's empty-CAR, RepoNotFound, duplicate-DID, captured-timestamp, and forced-drain rules. Focused tests hold one sibling incomplete, cross the writer boundary, and fully reopen the store without replaying the completed sibling. -The same audit found fail-open phase/cursor/watermark reads, a fail-open merge +The same audit found fail-open phase, relay-cursor, and compaction-watermark +reads. Phase and relay-cursor decoding now fail closed and have injected read +fault tests; the watermark gap remains. 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 gap are maintained in +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 1f9a9a6..7732ca0 100644 --- a/src/internal/bootstrap/lifecycle.zig +++ b/src/internal/bootstrap/lifecycle.zig @@ -99,7 +99,7 @@ pub fn runTraced( var phase_store = phase_mod.Store.init(meta); defer phase_store.deinit(io); - var phase = phase_store.read(io) orelse blk: { + var phase = (try phase_store.read(io)) orelse blk: { try phase_store.write(io, .bootstrap); break :blk phase_mod.Phase.bootstrap; }; @@ -492,7 +492,7 @@ fn runBootstrap( .emit = ingest.Consumer.emitVerified, .emit_repair = ingest.Consumer.emitRepair, .emit_cursor = ingest.Consumer.emitCursor, - .cursor_watermark = cursor_store.load(io) orelse 0, + .cursor_watermark = (try cursor_store.load(io)) orelse 0, .trace_context = trace_context, }; var repair_scratch_buf: [Io.Dir.max_path_bytes]u8 = undefined; @@ -521,7 +521,7 @@ fn runBootstrap( .io = io, .upstream = opts.upstream, .consumer = &consumer, - .last_seq = cursor_store.load(io), + .last_seq = try cursor_store.load(io), }; const BootstrapRace = Io.Select(union(enum) { capture: void, diff --git a/src/internal/bootstrap/phase.zig b/src/internal/bootstrap/phase.zig index a93b1b9..b2bca7c 100644 --- a/src/internal/bootstrap/phase.zig +++ b/src/internal/bootstrap/phase.zig @@ -16,11 +16,11 @@ pub const Store = struct { } pub fn deinit(_: *Store, _: std.Io) void {} - pub fn read(self: *Store, _: std.Io) ?Phase { - const maybe_bytes = self.meta.getAlloc(self.meta.allocator, key) catch return null; + pub fn read(self: *Store, _: std.Io) !?Phase { + const maybe_bytes = try self.meta.getAlloc(self.meta.allocator, key); const bytes = maybe_bytes orelse return null; defer self.meta.allocator.free(bytes); - return std.meta.stringToEnum(Phase, bytes); + return std.meta.stringToEnum(Phase, bytes) orelse error.InvalidLifecyclePhase; } pub fn write(self: *Store, _: std.Io, phase: Phase) !void { @@ -38,9 +38,26 @@ test "phase round-trip and first-boot null" { var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]); defer meta.deinit(); var store = Store.init(&meta); - try testing.expectEqual(@as(?Phase, null), store.read(testing.io)); + try testing.expectEqual(@as(?Phase, null), try store.read(testing.io)); try store.write(testing.io, .bootstrap); - try testing.expectEqual(@as(?Phase, .bootstrap), store.read(testing.io)); + try testing.expectEqual(@as(?Phase, .bootstrap), try store.read(testing.io)); try store.write(testing.io, .steady_state); - try testing.expectEqual(@as(?Phase, .steady_state), store.read(testing.io)); + try testing.expectEqual(@as(?Phase, .steady_state), try store.read(testing.io)); +} + +test "phase corruption and storage errors fail closed" { + const testing = std.testing; + 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 = Store.init(&meta); + + try meta.putDurable(key, "banana"); + try testing.expectError(error.InvalidLifecyclePhase, store.read(testing.io)); + try testing.expect(meta_store.armReadFault(key, 1)); + defer meta_store.disarmReadFault(); + try testing.expectError(error.InjectedStoreFault, store.read(testing.io)); } diff --git a/src/internal/cursor.zig b/src/internal/cursor.zig index ea9fb65..463eaba 100644 --- a/src/internal/cursor.zig +++ b/src/internal/cursor.zig @@ -5,6 +5,7 @@ const meta_store = @import("meta_store.zig"); const log = std.log.scoped(.stream); const key = "relay/cursor"; +const encoding_version: u8 = 1; pub const flush_interval_us: i64 = 5 * std.time.us_per_s; pub const Store = struct { @@ -18,12 +19,15 @@ pub const Store = struct { pub fn deinit(_: *Store, _: std.Io) void {} - pub fn load(self: *Store, _: std.Io) ?i64 { - const maybe_bytes = self.meta.getAlloc(self.meta.allocator, key) catch return null; + pub fn load(self: *Store, _: std.Io) !?i64 { + const maybe_bytes = try self.meta.getAlloc(self.meta.allocator, key); const bytes = maybe_bytes orelse return null; defer self.meta.allocator.free(bytes); - if (bytes.len != 8) return null; - return std.mem.readInt(i64, bytes[0..8], .little); + if (bytes.len != 9) return error.InvalidRelayCursorLength; + if (bytes[0] != encoding_version) return error.UnsupportedRelayCursorVersion; + const raw = std.mem.readInt(u64, bytes[1..9], .little); + if (raw > std.math.maxInt(i64)) return error.NegativeRelayCursor; + return @intCast(raw); } pub fn update(self: *Store, io: std.Io, seq: i64, now_us: i64) void { @@ -35,8 +39,13 @@ pub const Store = struct { pub fn flush(self: *Store, _: std.Io) void { const seq = self.pending orelse return; - var bytes: [8]u8 = undefined; - std.mem.writeInt(i64, &bytes, seq, .little); + if (seq < 0) { + log.err("refusing to flush negative cursor {d}", .{seq}); + return; + } + var bytes: [9]u8 = undefined; + bytes[0] = encoding_version; + std.mem.writeInt(u64, bytes[1..9], @intCast(seq), .little); self.meta.putDurable(key, &bytes) catch |err| { log.warn("cursor flush failed: {s}", .{@errorName(err)}); return; @@ -44,9 +53,11 @@ pub const Store = struct { self.pending = null; } - pub fn stage(writes: *meta_store.Batch, seq: i64) void { - var bytes: [8]u8 = undefined; - std.mem.writeInt(i64, &bytes, seq, .little); + pub fn stage(writes: *meta_store.Batch, seq: i64) !void { + if (seq < 0) return error.NegativeRelayCursor; + var bytes: [9]u8 = undefined; + bytes[0] = encoding_version; + std.mem.writeInt(u64, bytes[1..9], @intCast(seq), .little); writes.put(key, &bytes); } }; @@ -61,11 +72,52 @@ test "cursor round-trip and interval flushing" { defer meta.deinit(); var store = Store.init(&meta); - try testing.expect(store.load(testing.io) == null); + try testing.expect(try store.load(testing.io) == null); store.update(testing.io, 41, 10 * std.time.us_per_s); - try testing.expectEqual(@as(?i64, 41), store.load(testing.io)); + try testing.expectEqual(@as(?i64, 41), try store.load(testing.io)); store.update(testing.io, 42, 11 * std.time.us_per_s); - try testing.expectEqual(@as(?i64, 41), store.load(testing.io)); + try testing.expectEqual(@as(?i64, 41), try store.load(testing.io)); store.flush(testing.io); - try testing.expectEqual(@as(?i64, 42), store.load(testing.io)); + try testing.expectEqual(@as(?i64, 42), try store.load(testing.io)); +} + +test "cursor rejects malformed version width negative values and storage errors" { + const testing = std.testing; + 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 = Store.init(&meta); + + try meta.putDurable(key, &.{ 1, 2, 3 }); + try testing.expectError(error.InvalidRelayCursorLength, store.load(testing.io)); + try meta.putDurable(key, &.{ 2, 0, 0, 0, 0, 0, 0, 0, 0 }); + try testing.expectError(error.UnsupportedRelayCursorVersion, store.load(testing.io)); + try meta.putDurable(key, &.{ 1, 0, 0, 0, 0, 0, 0, 0, 0x80 }); + try testing.expectError(error.NegativeRelayCursor, store.load(testing.io)); + var writes = meta.batch(); + defer writes.deinit(); + try testing.expectError(error.NegativeRelayCursor, Store.stage(&writes, -1)); + try testing.expect(meta_store.armReadFault(key, 1)); + defer meta_store.disarmReadFault(); + try testing.expectError(error.InjectedStoreFault, store.load(testing.io)); +} + +test "legacy bare cursor is migrated before strict decoding" { + const testing = std.testing; + 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(i64, &legacy, 73, .little); + try meta.putDurable(key, &legacy); + try meta.migrateLegacy(testing.io, path); + var store = Store.init(&meta); + try testing.expectEqual(@as(?i64, 73), try store.load(testing.io)); } diff --git a/src/internal/ingest.zig b/src/internal/ingest.zig index 5043849..23fed0d 100644 --- a/src/internal/ingest.zig +++ b/src/internal/ingest.zig @@ -520,7 +520,7 @@ pub const Consumer = struct { std.mem.writeInt(u64, &durable_bytes, durable_seq, .little); writes.put("seq/durable", &durable_bytes); } - if (upstream_seq) |seq| cursor_mod.Store.stage(&writes, seq); + if (upstream_seq) |seq| try cursor_mod.Store.stage(&writes, seq); var durable_chains: usize = 0; for (self.pending_chains.items) |point| { if (point.local_seq > durable_seq) break; @@ -1007,7 +1007,7 @@ test "verifier silent drop waits for a later emitted cursor watermark" { .ops = &.{}, } }; try consumer.handleEvent(trigger, .{ .outcome = .needs_resync }, false, observability.root_context); - try testing.expectEqual(@as(?i64, null), cursor.load(io)); + try testing.expectEqual(@as(?i64, null), try cursor.load(io)); const later: zat.FirehoseEvent = .{ .identity = .{ .seq = 12, @@ -1017,7 +1017,7 @@ test "verifier silent drop waits for a later emitted cursor watermark" { } }; try consumer.handleEvent(later, .{}, false, observability.root_context); try archive.flushBlock(); - try testing.expectEqual(@as(?i64, 12), cursor.load(io)); + try testing.expectEqual(@as(?i64, 12), try cursor.load(io)); } test "raw firehose inspection classifies unknown and bounded error frames" { diff --git a/src/internal/meta_store.zig b/src/internal/meta_store.zig index 17e7768..337d5ed 100644 --- a/src/internal/meta_store.zig +++ b/src/internal/meta_store.zig @@ -34,6 +34,7 @@ const Fault = struct { }; var fault: Fault = .{}; +var read_fault: Fault = .{}; /// Arm one matching durable metadata operation. Must be called during process /// startup, before worker threads exist. Returns false for an empty prefix or @@ -48,6 +49,19 @@ pub fn disarmFault() void { fault = .{}; } +/// Independent read-side fault seam. Keeping it separate from durable mutation +/// faults preserves the existing crash schedules while allowing startup code +/// to prove that a storage error is not misclassified as a missing key. +pub fn armReadFault(prefix: []const u8, ordinal: u64) bool { + if (prefix.len == 0 or ordinal == 0) return false; + read_fault = .{ .prefix = prefix, .ordinal = ordinal }; + return true; +} + +pub fn disarmReadFault() void { + read_fault = .{}; +} + pub const subdir = "meta.rocksdb"; pub const Store = struct { @@ -107,15 +121,30 @@ pub const Store = struct { /// keys always win, making this safe on every restart. pub fn migrateLegacy(self: *Store, io: std.Io, data_dir: []const u8) !void { if (try self.getAlloc(self.allocator, "relay/cursor")) |existing| { - self.allocator.free(existing); + defer self.allocator.free(existing); + // Stream previously stored a bare signed little-endian value. + // Convert that known schema exactly once; all other malformed or + // unknown-version values remain untouched so the strict loader + // fails closed instead of laundering corruption. + if (existing.len == 8) { + const seq = std.mem.readInt(i64, existing[0..8], .little); + if (seq < 0) return error.NegativeRelayCursor; + var versioned: [9]u8 = undefined; + versioned[0] = 1; + std.mem.writeInt(u64, versioned[1..9], @intCast(seq), .little); + try self.putDurable("relay/cursor", &versioned); + log.info("migrated relay cursor to versioned encoding", .{}); + } } else { var path_buf: [std.fs.max_path_bytes]u8 = undefined; const path = try std.fmt.bufPrint(&path_buf, "{s}/cursor", .{data_dir}); var text_buf: [64]u8 = undefined; if (std.Io.Dir.cwd().readFile(io, path, &text_buf)) |text| { if (std.fmt.parseInt(i64, std.mem.trim(u8, text, " \n\t"), 10)) |seq| { - var bytes: [8]u8 = undefined; - std.mem.writeInt(i64, &bytes, seq, .little); + if (seq < 0) return error.NegativeRelayCursor; + var bytes: [9]u8 = undefined; + bytes[0] = 1; + std.mem.writeInt(u64, bytes[1..9], @intCast(seq), .little); try self.putDurable("relay/cursor", &bytes); log.info("imported legacy relay cursor {d}", .{seq}); } else |_| {} @@ -148,6 +177,10 @@ pub const Store = struct { pub fn getAlloc(self: *Store, allocator: Allocator, key: []const u8) !?[]u8 { const started_at = self.started(); + read_fault.check(key) catch |err| { + self.observe(.get, .error_result, started_at); + return err; + }; var err_data: ?rocksdb.Data = null; defer if (err_data) |e| e.deinit(); const value = self.db.get(self.default_cf, key, &err_data) catch |err| { diff --git a/src/internal/repair_integration_test.zig b/src/internal/repair_integration_test.zig index 2dcb88b..691a2f5 100644 --- a/src/internal/repair_integration_test.zig +++ b/src/internal/repair_integration_test.zig @@ -88,7 +88,7 @@ test "repair completion durably emits tombstone then replacement offline" { try ingest.Consumer.emitRepair(&consumer, &completion, @import("observability.zig").root_context); try testing.expectEqual(@as(u64, 2), archive.committed_seq.load(.acquire)); - try testing.expectEqual(@as(?i64, 77), cursor.load(io)); + try testing.expectEqual(@as(?i64, 77), try cursor.load(io)); try testing.expectEqual(@as(usize, 2), tail.gauges().entries); const retained = tail.entries.items[tail.head..]; try testing.expectEqual(@as(usize, 0), retained[0].json.len); @@ -206,6 +206,6 @@ test "large repair keeps wire encoding scratch bounded" { try ingest.Consumer.emitRepair(&consumer, &completion, @import("observability.zig").root_context); try testing.expectEqual(@as(u64, 4097), archive.committed_seq.load(.acquire)); - try testing.expectEqual(@as(?i64, 4096), cursor.load(io)); + try testing.expectEqual(@as(?i64, 4096), try cursor.load(io)); try testing.expect(repair_alloc.total_requested_bytes < 256 * 1024); } diff --git a/src/main.zig b/src/main.zig index c7ca0a1..0ff2f8f 100644 --- a/src/main.zig +++ b/src/main.zig @@ -982,7 +982,7 @@ pub fn main(init: std.process.Init.Minimal) !void { const stored_phase = blk: { var ps = phase_mod.Store.init(&meta); defer ps.deinit(io); - break :blk ps.read(io); + break :blk try ps.read(io); }; if (stored_phase) |phase| stats.setOrchestratorPhase(switch (phase) { .bootstrap => .bootstrap, @@ -1191,7 +1191,7 @@ pub fn main(init: std.process.Init.Minimal) !void { hub.serving.store(true, .release); } if (cursor == null) { - cursor = cursor_store.load(io); + cursor = try cursor_store.load(io); if (cursor) |c| log.info("resuming from persisted cursor {d}", .{c}); }