From ab591aeeacb156fb2f9f3b7e1b18daaedf8cb35f Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Thu, 16 Jul 2026 17:15:48 -0500 Subject: [PATCH] implement authenticated live repair --- README.md | 3 +- docs/live-repair.md | 62 ++ docs/semantic-parity.md | 4 +- src/internal/archive.zig | 15 + src/internal/bootstrap/capture.zig | 2 +- src/internal/bootstrap/fetched_car.zig | 205 ++++- src/internal/bootstrap/lifecycle.zig | 19 + src/internal/bootstrap/prepared_fetch.zig | 17 +- src/internal/bootstrap/resync.zig | 2 +- src/internal/cold.zig | 4 +- src/internal/ingest.zig | 294 ++++++-- src/internal/metrics.zig | 35 + src/internal/pipeline.zig | 115 ++- src/internal/repair.zig | 867 ++++++++++++++++++++++ src/internal/repair_integration_test.zig | 118 +++ src/internal/verify.zig | 66 +- src/main.zig | 17 + src/tests.zig | 2 + 18 files changed, 1769 insertions(+), 78 deletions(-) create mode 100644 docs/live-repair.md create mode 100644 src/internal/repair.zig create mode 100644 src/internal/repair_integration_test.zig diff --git a/README.md b/README.md index 8e21ae2..aa69f1b 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,8 @@ the live, phone-sized field report is at - full-network bootstrap with concurrent live capture, deterministic merge, and crash-safe cutover into steady-state ingestion - Sync 1.1 commit verification, including PLC key rotation, MST inversion, - op-CID checks, replay protection, and durable repository-chain state + op-CID checks, replay protection, durable repository-chain state, and + transparent authenticated whole-repo repair with ordered pending replay - durable metadata and archive publication through RocksDB transactions, coupled to segment fsync boundaries - tombstone-aware compaction and continuing repair of repositories that were diff --git a/docs/live-repair.md b/docs/live-repair.md new file mode 100644 index 0000000..71e005d --- /dev/null +++ b/docs/live-repair.md @@ -0,0 +1,62 @@ +# Live Sync 1.1 repair + +Stream follows the pinned Atmos/Jetstream V2 repair contract rather than +inventing a second recovery model. + +## Contract + +- Recoverable chain, inversion, CAR-decode, duplicate-path, and op-CID + failures withhold the offending event and schedule authoritative `getRepo` + through the configured relay. Signature failures, future revisions, + oversized input, and outer/inner envelope disagreement bypass repair. +- A `#sync` divergence first performs that fetch synchronously, with the DID + lane released during network I/O, and queues one async retry only for a + retryable fetch failure. A successful inline repair advances the original + firehose cursor at the replacement set's durable boundary; the deferred + retry is synthetic and carries no cursor. +- The pool is 32 workers behind a 64-job bounded queue. A fetch owns a + five-minute budget and does not hold the DID lane. Commits arriving during + the fetch enter a 2,048-frame FIFO; overflow drops and reports the oldest. +- A bounded 16,384-DID token-bucket cache permits five repairs per minute with + burst five. Rate-limit eviction only resets a DID's repair allowance; it + never drops repository state. +- Apply re-authenticates the prepared complete repository, rejects a head + older than the current chain, and rejects an equal revision with a different + data CID. Pending commits replay against the fetched head under the same DID + lane. Replay failures are reported and never recursively enqueue repair. +- Completion is registered under the DID lane, but cannot pass the archive + writer until its trigger ticket has completed. The durable row order is a + `sync` DID tombstone, the authoritative `create_resync` materialization, + then any valid pending commits. +- A withheld trigger or pending commit does not advance the relay cursor by + itself. As in Atmos, a later normally emitted upstream event may move the + cursor watermark beyond those silent drops; the in-memory repair queue is + not a crash journal. A repair completion is synthetic and carries no relay + cursor. This preserves the pinned upstream restart behavior rather than + presenting a stronger durability guarantee that Jetstream V2 does not make. +- Archive sequence assignment and verifier/cursor/hosting metadata staging + share one writer-lock transaction. Any fsync callback therefore sees the + matching checkpoint already registered; tail publication remains gated on + that same durable boundary. + +Jetstream runs Atmos with its default hosting policy (`HostingTrack`), so +account state is persisted but does not gate commit or repair fetches. + +## Offline receipts + +`zig build test` requires no public network. The repair tests use loopback HTTP +and generated cryptographic material to exercise the production paths: + +1. a signed, complete repository CAR is streamed to scratch, mmapped, + completeness-checked, resolved through a real DID document, and verified; +2. the HTTP body is held open while a same-DID encoded firehose frame enters + the pending FIFO, proving network fetch does not own the apply lane; +3. limiter, overflow, stale-head, contradictory-head, and trigger-ticket + ordering boundaries are checked deterministically; +4. a real synchronous-`#sync` completion is written through + `Consumer.emitRepair`, then read back + as sealed JSS rows while its RocksDB chain checkpoint and v2 tail frames are + asserted with its original relay cursor at the same durable sequence + boundary; and +5. verifier silent-drops leave the durable relay cursor unchanged until a + later ordinary event is archived, matching Atmos's watermark behavior. diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index ef4cb94..3ae6129 100644 --- a/docs/semantic-parity.md +++ b/docs/semantic-parity.md @@ -16,8 +16,8 @@ this document or `bootstrap-semantic-parity.md` is open. | JSS v1 storage | Upstream-produced fixture, reciprocal segment parsing, checksum/index/gloom vectors | closed for sealed format; rerun reciprocal corpus before experiment | | Archive XRPC | Official Go client plus Stream conformance replay over listSegments/getSegment/getBlock/planBackfill | implementation present; fresh full suite required | | Subscribe v1/v2 | Wire/filter/cursor/zstd unit and local e2e coverage | implementation present; official-client replay and below-floor/adversarial suite required | -| 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, and transparent whole-repo repair | Diff-CAR/MST inversion (including upstream's narrow default lenient carve-out), post-state op-CID proof (including a tampered-envelope test), signed inner/outer consistency, size/future-rev/replay gates, default `LegacyAccept`, ordered per-DID lanes, two-phase durable chain/hosting state, and account/identity replay ratchets are implemented. Recoverable decode/inversion/duplicate-path/op-CID/chain failures now route explicitly to repair; signature failures and outer/inner producer-integrity mismatches bypass repair, and no failed event is archived. `prepareRepo` retains the canonical unsigned commit and signature while indexing and completeness-checking the full CAR, so repair authentication reuses that exact prepared head instead of reparsing and re-walking a potentially multi-gigabyte repository. **Open: connection of authenticated fetches to transparent repair** | -| Sync 1.1 resync ordering | Upstream serializes per DID, drops stale async repairs, and emits a sync tombstone plus authoritative replacements | Failed-repo replacements pre-validate then append bounded 1,024-row batches, matching upstream and avoiding both a repository-sized row array and a whole-repo live-writer stall. **Open — 32-worker async coordinator, five-minute fetch budget, 5/minute per-DID limiter, 2,048-entry pending replay, stale apply guard, and durable state/outbox coupling** | +| 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, and transparent whole-repo repair | The verifier core is 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. Offline receipts use real signed CARs and real verifier state, not fixture verdicts. **Open:** replace the global ticket/backpressure scheduler with upstream's 32-worker per-DID FIFO, 64-event per-DID drop-oldest overflow, cross-DID completion delivery, and smallest-inflight cursor watermark. | +| 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. The writer emits `sync` then bounded 1,024-row `create_resync` batches and stages the 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 | **open — inventory and implementation receipt required** | | Status/diagnostics | Repo attempts/error class/final PDS and per-host aggregates survive restart and are operator-queryable | durable repo fields implemented; **operator repo/host views open** | | Prometheus/Grafana | Every canonical dashboard query maps to a real, same-semantics producer; runtime-specific panels use honest Zig/process metrics | pinned dashboard installer exists and fails closed; canonical build/live-receive/archive-append/drop/verifier/backfill/subscriber families now have real producers; **remaining metric families and Go-row adaptation open** | diff --git a/src/internal/archive.zig b/src/internal/archive.zig index e7769d0..8781339 100644 --- a/src/internal/archive.zig +++ b/src/internal/archive.zig @@ -184,6 +184,20 @@ pub const Archive = struct { /// lock. This is the transaction boundary relied on by frame seq /// assignment (`first_seq + op_index`) and verifier checkpoints. pub fn appendBatch(self: *Archive, events: []const segment.Event, now_us: i64) !?u64 { + return self.appendBatchBeforeDurable(events, now_us, null, null); + } + + /// Append rows and run `before_durable(first,last)` under the writer lock + /// before this append is allowed to trigger an fsync. The ingest writer + /// uses this to register cursor/verifier checkpoints against assigned + /// sequence numbers, closing the rows-before-metadata crash window. + pub fn appendBatchBeforeDurable( + self: *Archive, + events: []const segment.Event, + now_us: i64, + hook_ctx: ?*anyopaque, + before_durable: ?*const fn (*anyopaque, u64, u64) anyerror!void, + ) !?u64 { if (events.len == 0) return null; self.mu.lockUncancelable(self.io); defer self.mu.unlock(self.io); @@ -193,6 +207,7 @@ pub const Archive = struct { for (events) |ev| { _ = try self.appendStampedLocked(ev); } + if (before_durable) |hook| try hook(hook_ctx.?, first_seq, self.next_seq - 1); if (self.writer.pending.items.len > 0 and now_us - self.last_sync_us >= self.flush_interval_us) { try self.writer.flush(); // partial block: durability over block size } diff --git a/src/internal/bootstrap/capture.zig b/src/internal/bootstrap/capture.zig index f575780..b7eedcb 100644 --- a/src/internal/bootstrap/capture.zig +++ b/src/internal/bootstrap/capture.zig @@ -34,7 +34,7 @@ pub const Capture = struct { var client = zat.FirehoseClient.init(self.io, self.allocator, .{ .hosts = &.{self.upstream}, .cursor = self.last_seq, - .max_message_size = 10_000_000, + .max_message_size = 2 * 1024 * 1024, }); defer client.deinit(); client.subscribe(self) catch |err| { diff --git a/src/internal/bootstrap/fetched_car.zig b/src/internal/bootstrap/fetched_car.zig index 9c6554f..54651a0 100644 --- a/src/internal/bootstrap/fetched_car.zig +++ b/src/internal/bootstrap/fetched_car.zig @@ -17,6 +17,14 @@ pub const attempt_timeout: Io.Clock.Duration = .{ .raw = Io.Duration.fromSeconds(5 * 60), .clock = .awake, }; +pub const low_speed_bytes_per_second: u64 = 1000; +pub const low_speed_seconds: u16 = 30; + +pub const TransferProgress = struct { + body_started: std.atomic.Value(bool) = .init(false), + body_finished: std.atomic.Value(bool) = .init(false), + bytes: std.atomic.Value(u64) = .init(0), +}; pub const Outcome = union(enum) { ok: FetchedCar, @@ -74,6 +82,18 @@ pub fn fetchOnce( relay_http: []const u8, did: []const u8, final_host: *FinalHost, +) !Outcome { + return fetchOnceTracked(allocator, io, scratch_dir, relay_http, did, final_host, null); +} + +pub fn fetchOnceTracked( + allocator: Allocator, + io: Io, + scratch_dir: []const u8, + relay_http: []const u8, + did: []const u8, + final_host: *FinalHost, + progress: ?*TransferProgress, ) !Outcome { const url = try std.fmt.allocPrint(allocator, "{s}/xrpc/com.atproto.sync.getRepo?did={s}", .{ relay_http, did }); defer allocator.free(url); @@ -107,7 +127,19 @@ pub fn fetchOnce( var write_buf: [64 * 1024]u8 = undefined; var writer = Io.File.Writer.initStreaming(file, io, &write_buf); const reader = response.reader(&.{}); - _ = try reader.streamRemaining(&writer.interface); + if (progress) |p| p.body_started.store(true, .release); + var read_buf: [64 * 1024]u8 = undefined; + while (true) { + var slices = [_][]u8{&read_buf}; + const n = reader.readVec(&slices) catch |err| switch (err) { + error.EndOfStream => break, + else => return err, + }; + if (n == 0) continue; + try writer.interface.writeAll(read_buf[0..n]); + if (progress) |p| _ = p.bytes.fetchAdd(n, .monotonic); + } + if (progress) |p| p.body_finished.store(true, .release); try writer.end(); const file_len = try file.length(io); @@ -140,7 +172,93 @@ pub fn fetch( did: []const u8, final_host: *FinalHost, ) !Outcome { - return fetchWithTimeout(allocator, io, scratch_dir, relay_http, did, final_host, attempt_timeout); + const Task = struct { + allocator: Allocator, + io: Io, + scratch_dir: []const u8, + relay_http: []const u8, + did: []const u8, + final_host: *FinalHost, + progress: *TransferProgress, + done: Io.Event = .unset, + result: ?(anyerror!Outcome) = null, + + fn run(self: *@This()) void { + self.result = fetchOnceTracked(self.allocator, self.io, self.scratch_dir, self.relay_http, self.did, self.final_host, self.progress); + self.done.set(self.io); + } + }; + var progress: TransferProgress = .{}; + var task: Task = .{ + .allocator = allocator, + .io = io, + .scratch_dir = scratch_dir, + .relay_http = relay_http, + .did = did, + .final_host = final_host, + .progress = &progress, + }; + var future = try io.concurrent(Task.run, .{&task}); + try waitForTransfer(io, &task.done, &future, &progress); + _ = future.await(io); + return try task.result.?; +} + +pub fn waitForTransfer(io: Io, done: *Io.Event, future: *Io.Future(void), progress: *TransferProgress) !void { + return waitForTransferPolicy( + io, + done, + future, + progress, + .{ .raw = Io.Duration.fromSeconds(1), .clock = .awake }, + low_speed_bytes_per_second, + low_speed_seconds, + 5 * 60, + ); +} + +fn waitForTransferPolicy( + io: Io, + done: *Io.Event, + future: *Io.Future(void), + progress: *TransferProgress, + sample_interval: Io.Clock.Duration, + minimum_bytes: u64, + maximum_slow_samples: u16, + maximum_samples: u16, +) !void { + var prior_bytes: u64 = 0; + var slow_samples: u16 = 0; + var samples: u16 = 0; + while (true) { + done.waitTimeout(io, .{ .duration = sample_interval }) catch |err| switch (err) { + error.Timeout => { + samples += 1; + if (progress.body_started.load(.acquire) and !progress.body_finished.load(.acquire)) { + const current = progress.bytes.load(.monotonic); + if (current - prior_bytes < minimum_bytes) + slow_samples += 1 + else + slow_samples = 0; + prior_bytes = current; + if (slow_samples >= maximum_slow_samples) { + _ = future.cancel(io); + return error.TransferTooSlow; + } + } + if (samples >= maximum_samples) { + _ = future.cancel(io); + return error.AttemptTimeout; + } + continue; + }, + else => { + _ = future.cancel(io); + return err; + }, + }; + return; + } } fn fetchWithTimeout( @@ -326,3 +444,86 @@ test "stalled getRepo is canceled at the attempt deadline" { fetchWithTimeout(std.testing.allocator, io, scratch[0..scratch_len], url, "did:plc:stalled", &final_host, timeout), ); } + +test "trickling getRepo is canceled by the low speed guard" { + const testing = std.testing; + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var server = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; + defer server.deinit(io); + const Fixture = struct { + fn run(listener: *Io.net.Server, task_io: Io) !void { + var stream = try listener.accept(task_io); + defer stream.close(task_io); + var read_buf: [4096]u8 = undefined; + var write_buf: [4096]u8 = undefined; + var reader = stream.reader(task_io, &read_buf); + var writer = stream.writer(task_io, &write_buf); + var http_server = std.http.Server.init(&reader.interface, &writer.interface); + _ = try http_server.receiveHead(); + try writer.interface.writeAll( + "HTTP/1.1 200 OK\r\nContent-Type: application/vnd.ipld.car\r\nContent-Length: 10000\r\nConnection: close\r\n\r\n", + ); + try writer.interface.flush(); + while (true) { + writer.interface.writeByte(0) catch return; + writer.interface.flush() catch return; + task_io.sleep(Io.Duration.fromMilliseconds(5), .awake) catch return; + } + } + }; + var server_future = try io.concurrent(Fixture.run, .{ &server, io }); + defer _ = server_future.cancel(io) catch {}; + + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var scratch: [Io.Dir.max_path_bytes]u8 = undefined; + const scratch_len = try tmp.dir.realPath(io, &scratch); + var url_buf: [96]u8 = undefined; + const url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}", .{server.socket.address.getPort()}); + var final_host: FinalHost = .{}; + var progress: TransferProgress = .{}; + const Fetch = struct { + allocator: Allocator, + io: Io, + scratch: []const u8, + url: []const u8, + host: *FinalHost, + progress: *TransferProgress, + done: Io.Event = .unset, + result: ?(anyerror!Outcome) = null, + + fn run(self: *@This()) void { + self.result = fetchOnceTracked( + self.allocator, + self.io, + self.scratch, + self.url, + "did:plc:trickle", + self.host, + self.progress, + ); + self.done.set(self.io); + } + }; + var fetch_task: Fetch = .{ + .allocator = testing.allocator, + .io = io, + .scratch = scratch[0..scratch_len], + .url = url, + .host = &final_host, + .progress = &progress, + }; + var fetch_future = try io.concurrent(Fetch.run, .{&fetch_task}); + try testing.expectError(error.TransferTooSlow, waitForTransferPolicy( + io, + &fetch_task.done, + &fetch_future, + &progress, + .{ .raw = Io.Duration.fromMilliseconds(10), .clock = .awake }, + 100, + 3, + 100, + )); +} diff --git a/src/internal/bootstrap/lifecycle.zig b/src/internal/bootstrap/lifecycle.zig index 3ae9a21..3cc6d84 100644 --- a/src/internal/bootstrap/lifecycle.zig +++ b/src/internal/bootstrap/lifecycle.zig @@ -29,6 +29,7 @@ const metrics = @import("../metrics.zig"); const meta_store = @import("../meta_store.zig"); const phase_mod = @import("phase.zig"); const pipeline_mod = @import("../pipeline.zig"); +const repair_mod = @import("../repair.zig"); const repo_store_mod = @import("repo_store.zig"); const repos = @import("repos.zig"); const resync = @import("resync.zig"); @@ -322,7 +323,25 @@ fn runBootstrap( .stats = stats, .emit_ctx = &consumer, .emit = ingest.Consumer.emitVerified, + .emit_repair = ingest.Consumer.emitRepair, }; + var repair_scratch_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const repair_scratch = try std.fmt.bufPrint(&repair_scratch_buf, "{s}/repair-scratch", .{live_data_dir}); + var repair_coordinator: ?repair_mod.Coordinator = if (verifier) |v| .{ + .allocator = allocator, + .io = io, + .verifier = v, + .relay_http = opts.relay_http, + .scratch_dir = repair_scratch, + .stats = stats, + .publish_ctx = &pipe, + .publish = pipeline_mod.Pipeline.publishRepair, + } else null; + if (repair_coordinator) |*coordinator| { + try coordinator.start(); + pipe.repair = coordinator; + } + defer if (repair_coordinator) |*coordinator| coordinator.deinit(); try pipe.start(); defer pipe.stop(); consumer.pipeline = &pipe; diff --git a/src/internal/bootstrap/prepared_fetch.zig b/src/internal/bootstrap/prepared_fetch.zig index 6d95e67..869ec0d 100644 --- a/src/internal/bootstrap/prepared_fetch.zig +++ b/src/internal/bootstrap/prepared_fetch.zig @@ -48,6 +48,7 @@ pub fn fetch( relay_http: []const u8, did: []const u8, final_host: *fetched_car.FinalHost, + progress: *fetched_car.TransferProgress, done: Io.Event = .unset, result: ?(anyerror!Outcome) = null, @@ -57,13 +58,14 @@ pub fn fetch( } fn runAttempt(self: *@This()) !Outcome { - var outcome = try fetched_car.fetchOnce( + var outcome = try fetched_car.fetchOnceTracked( self.allocator, self.io, self.scratch_dir, self.relay_http, self.did, self.final_host, + self.progress, ); return switch (outcome) { .ok => |*fetched| blk: { @@ -80,6 +82,7 @@ pub fn fetch( } }; + var progress: fetched_car.TransferProgress = .{}; var task: Task = .{ .allocator = allocator, .parse_allocator = parse_allocator, @@ -88,18 +91,10 @@ pub fn fetch( .relay_http = relay_http, .did = did, .final_host = final_host, + .progress = &progress, }; var future = try io.concurrent(Task.run, .{&task}); - task.done.waitTimeout(io, .{ .duration = fetched_car.attempt_timeout }) catch |err| switch (err) { - error.Timeout => { - _ = future.cancel(io); - return error.AttemptTimeout; - }, - else => { - _ = future.cancel(io); - return err; - }, - }; + try fetched_car.waitForTransfer(io, &task.done, &future, &progress); _ = future.await(io); return try task.result.?; } diff --git a/src/internal/bootstrap/resync.zig b/src/internal/bootstrap/resync.zig index 481dfd8..2e3b0ee 100644 --- a/src/internal/bootstrap/resync.zig +++ b/src/internal/bootstrap/resync.zig @@ -148,7 +148,7 @@ pub fn resyncRepo( /// synthetic #sync payload matching what the live path archives (a CBOR /// map the cold reader pulls did/rev/time from; seq and blocks default) -fn syncTombstoneEvent(alloc: Allocator, did: []const u8, rev: []const u8, now_us: i64) !segment.Event { +pub fn syncTombstoneEvent(alloc: Allocator, did: []const u8, rev: []const u8, now_us: i64) !segment.Event { var time_buf: [40]u8 = undefined; const time_str = try alloc.dupe(u8, formatRfc3339(&time_buf, now_us)); const entries = try alloc.dupe(zat.cbor.Value.MapEntry, &.{ diff --git a/src/internal/cold.zig b/src/internal/cold.zig index 474e318..946c403 100644 --- a/src/internal/cold.zig +++ b/src/internal/cold.zig @@ -149,7 +149,7 @@ pub fn replaySeq( } /// one archived row → one v2 wire frame -fn encodeEventV2(alloc: Allocator, ev: segment.Event) !?[]u8 { +pub fn encodeEventV2(alloc: Allocator, ev: segment.Event) !?[]u8 { var out: Io.Writer.Allocating = .init(alloc); switch (ev.kind) { .create, .update, .delete, .create_resync => { @@ -214,7 +214,7 @@ fn lessThan(_: void, a: []u8, b: []u8) bool { } /// one archived row → one v1 wire frame (null for rows v1 never emits) -fn encodeEvent(alloc: Allocator, ev: segment.Event) !?[]u8 { +pub fn encodeEvent(alloc: Allocator, ev: segment.Event) !?[]u8 { var out: Io.Writer.Allocating = .init(alloc); switch (ev.kind) { .create, .update, .delete => { diff --git a/src/internal/ingest.zig b/src/internal/ingest.zig index c20cfc8..2b7cf6e 100644 --- a/src/internal/ingest.zig +++ b/src/internal/ingest.zig @@ -7,12 +7,16 @@ const std = @import("std"); const zat = @import("zat"); const archive_mod = @import("archive.zig"); +const cold = @import("cold.zig"); const convert = @import("convert.zig"); const segment = @import("segment.zig"); const cursor_mod = @import("cursor.zig"); const metrics = @import("metrics.zig"); const meta_store = @import("meta_store.zig"); const pipeline_mod = @import("pipeline.zig"); +const repair = @import("repair.zig"); +const resync = @import("bootstrap/resync.zig"); +const repos = @import("bootstrap/repos.zig"); const tail_mod = @import("tail.zig"); const verify = @import("verify.zig"); @@ -51,6 +55,7 @@ pub const Consumer = struct { var client = zat.FirehoseClient.init(self.io, self.allocator, .{ .hosts = &.{self.upstream}, .cursor = self.cursor, + .max_message_size = 2 * 1024 * 1024, }); defer client.deinit(); try client.subscribe(self); @@ -82,13 +87,107 @@ pub const Consumer = struct { /// worker; its backing arena stays alive for the duration of this call. pub fn emitVerified(ctx: *anyopaque, event: zat.FirehoseEvent, verdict: verify.Result) !void { const self: *Consumer = @ptrCast(@alignCast(ctx)); - self.handleEvent(event, verdict) catch |err| { + self.handleEvent(event, verdict, false) catch |err| { log.err("fatal event handling failure: {s}", .{@errorName(err)}); self.setFatal(err); return err; }; } + pub fn emitRepair(ctx: *anyopaque, completion: *repair.Completion) !void { + const self: *Consumer = @ptrCast(@alignCast(ctx)); + const archive = self.archive orelse return error.RepairRequiresArchive; + var arena = std.heap.ArenaAllocator.init(self.allocator); + defer arena.deinit(); + const alloc = arena.allocator(); + const now_us = completion.witnessed_at; + + const Sink = struct { + consumer: *Consumer, + archive: *archive_mod.Archive, + allocator: Allocator, + now_us: i64, + batch: std.ArrayList(segment.Event) = .empty, + last_seq: u64 = 0, + fetched_chain: verify.ChainUpdate, + did: []const u8, + upstream_seq: ?i64, + stage_chain_on_flush: bool = false, + + fn stageChain(hook_ctx: *anyopaque, _: u64, last: u64) !void { + const sink: *@This() = @ptrCast(@alignCast(hook_ctx)); + const owned_did = try sink.consumer.allocator.dupe(u8, sink.did); + errdefer sink.consumer.allocator.free(owned_did); + try sink.consumer.pending_chains.append(sink.consumer.allocator, .{ + .local_seq = last, + .did = owned_did, + .state = sink.fetched_chain, + }); + if (sink.upstream_seq) |seq| { + if (sink.consumer.stats) |stats| stats.upstream_seq.store(seq, .monotonic); + try sink.consumer.pending_cursors.append(sink.consumer.allocator, .{ + .local_seq = last, + .upstream_seq = seq, + }); + } + } + + fn flush(sink: *@This()) !void { + if (sink.batch.items.len == 0) return; + const first = (if (sink.stage_chain_on_flush) + try sink.archive.appendBatchBeforeDurable(sink.batch.items, sink.now_us, sink, stageChain) + else + try sink.archive.appendBatch(sink.batch.items, sink.now_us)) orelse return; + for (sink.batch.items, 0..) |row, i| { + var stamped = row; + stamped.seq = first + i; + const json = (try cold.encodeEvent(sink.allocator, stamped)) orelse ""; + const json_v2 = (try cold.encodeEventV2(sink.allocator, stamped)) orelse ""; + const kind: @import("wire.zig").Kind = switch (row.kind) { + .create, .update, .delete, .create_resync => .commit, + .identity => .identity, + .account => .account, + .sync => .sync, + }; + try sink.consumer.tail.append(kind, row.did, row.collection, row.witnessed_at, stamped.seq, json, json_v2); + } + sink.last_seq = first + sink.batch.items.len - 1; + sink.batch.clearRetainingCapacity(); + } + + fn emit(sink: *@This(), row: segment.Event) anyerror!void { + if (sink.batch.items.len == 1024) try sink.flush(); + try sink.batch.append(sink.allocator, row); + } + }; + var sink: Sink = .{ + .consumer = self, + .archive = archive, + .allocator = alloc, + .now_us = now_us, + .fetched_chain = completion.fetched_chain, + .did = completion.did, + .upstream_seq = completion.upstream_seq, + }; + defer sink.batch.deinit(alloc); + try sink.emit(try resync.syncTombstoneEvent(alloc, completion.did, completion.prepared.repo.rev, now_us)); + var drops: repos.DropCounts = .{}; + _ = try repos.emitPrepared( + completion.prepared_arena.allocator(), + &completion.prepared.repo, + completion.did, + now_us, + .create_resync, + &drops, + &sink, + Sink.emit, + ); + sink.stage_chain_on_flush = true; + try sink.flush(); + for (completion.pending.items) |item| try self.handleEvent(item.event, item.verdict, true); + try archive.flushBlock(); + } + fn getFatal(self: *Consumer) ?anyerror { self.fatal_mu.lockUncancelable(self.io); defer self.fatal_mu.unlock(self.io); @@ -101,48 +200,62 @@ pub const Consumer = struct { if (self.fatal_error == null) self.fatal_error = err; } - fn handleEvent(self: *Consumer, event: zat.FirehoseEvent, verdict: verify.Result) !void { + fn handleEvent(self: *Consumer, event: zat.FirehoseEvent, verdict: verify.Result, replayed_after_repair: bool) !void { var arena = std.heap.ArenaAllocator.init(self.allocator); defer arena.deinit(); const alloc = arena.allocator(); const time_us: i64 = Io.Timestamp.now(self.io, .real).toMicroseconds(); - if (self.stats) |st| _ = st.events_total.fetchAdd(1, .monotonic); + if (!replayed_after_repair) { + if (self.stats) |st| _ = st.events_total.fetchAdd(1, .monotonic); + } switch (event) { .commit => |c| switch (verdict.outcome) { - .invalid_signature, .invalid_commit, .unverified, .needs_resync => { + .invalid_signature, .invalid_commit, .unverified => { // Never contaminate the archive with a commit we could - // not prove. Full Sync 1.1 parity will transparently - // resync recoverable failures instead of dropping them. + // not prove. These failures bypass authoritative repair. self.drops.getPtr(.invalid_signature).* += 1; if (self.stats) |st| _ = st.drops.getPtr(.invalid_signature).fetchAdd(1, .monotonic); log.warn("rejected unverified commit: {s} seq={d} reason={s}", .{ c.repo, c.seq, @tagName(verdict.outcome) }); - return self.noteSeq(event, time_us, if (self.archive) |a| a.lastSeq() else 0); + return; + }, + .needs_resync => { + // The coordinator owns this DID now. The trigger is + // replaced by its authoritative full repo; later commits + // are already in the bounded pending FIFO. + log.debug("commit withheld for repair: {s} seq={d}", .{ c.repo, c.seq }); + return; }, .valid => {}, - .replay => return self.noteSeq(event, time_us, if (self.archive) |a| a.lastSeq() else 0), + .replay => { + return; + }, }, .sync => switch (verdict.outcome) { .valid => {}, - .replay, .invalid_signature, .invalid_commit, .unverified, .needs_resync => return self.noteSeq(event, time_us, if (self.archive) |a| a.lastSeq() else 0), + .replay, .invalid_signature, .invalid_commit, .unverified, .needs_resync => { + return; + }, }, .identity => |identity| { if (identity.seq <= try self.appliedSeq(.identity, identity.did)) { if (self.stats) |stats| _ = stats.replayed_identity_events_dropped_total.fetchAdd(1, .monotonic); - return self.noteSeq(event, time_us, if (self.archive) |a| a.lastSeq() else 0); + if (!replayed_after_repair) self.noteSeq(event, time_us, if (self.archive) |a| a.lastSeq() else 0); + return; } // rotation signal: next commit re-resolves the key if (self.verifier) |v| v.evict(identity.did); }, .account => |account| if (account.seq <= try self.appliedSeq(.account, account.did)) { if (self.stats) |stats| _ = stats.replayed_account_events_dropped_total.fetchAdd(1, .monotonic); - return self.noteSeq(event, time_us, if (self.archive) |a| a.lastSeq() else 0); + if (!replayed_after_repair) self.noteSeq(event, time_us, if (self.archive) |a| a.lastSeq() else 0); + return; }, else => {}, } var persisted: Persisted = .{}; - if (self.archive) |a| persisted = self.persist(a, event) catch |err| { + if (self.archive) |a| persisted = self.persist(a, event, verdict, !replayed_after_repair) catch |err| { // our own storage failing is fatal-worthy, not drop-worthy log.err("archive append failed: {s}", .{@errorName(err)}); return err; @@ -161,40 +274,18 @@ pub const Consumer = struct { try self.tail.append(frame.kind, frame.did, frame.collection, frame.time_us, frame.seq, frame.json, frame.json_v2); } - if (verdict.chain) |chain| if (event == .commit or event == .sync) { - const barrier = if (persisted.last != 0) persisted.last else if (self.archive) |a| a.lastSeq() else 0; - const did = switch (event) { - .commit => |value| value.repo, - .sync => |value| value.did, - else => unreachable, - }; - const owned_did = try self.allocator.dupe(u8, did); - errdefer self.allocator.free(owned_did); - try self.pending_chains.append(self.allocator, .{ - .local_seq = barrier, - .did = owned_did, - .state = chain, - }); - }; - if (verdict.hosting) |hosting| if (event == .account) { - const barrier = if (persisted.last != 0) persisted.last else if (self.archive) |a| a.lastSeq() else 0; - const owned_did = try self.allocator.dupe(u8, event.account.did); - errdefer self.allocator.free(owned_did); - try self.pending_hosting.append(self.allocator, .{ - .local_seq = barrier, - .did = owned_did, - .state = hosting, - }); - }; - - switch (event) { - .identity => |identity| try self.stageRatchet(.identity, identity.did, identity.seq, persisted.last), - .account => |account| try self.stageRatchet(.account, account.did, account.seq, persisted.last), - else => {}, + if (persisted.last == 0) { + if (self.archive) |a| { + const barrier = a.lastSeq(); + try self.stageEventBoundary(event, verdict, barrier, !replayed_after_repair); + try self.commitDurableMetadata( + a.committed_seq.load(.acquire), + a.committed_seq.load(.acquire) + 1, + if (!replayed_after_repair) event.seq() else null, + ); + } else if (!replayed_after_repair) self.noteSeq(event, time_us, 0); } - self.noteSeq(event, time_us, if (persisted.last != 0) persisted.last else if (self.archive) |a| a.lastSeq() else 0); - if (self.to_stdout and frames.items.len > 0) { var stdout_buf: [64 * 1024]u8 = undefined; var stdout = std.Io.File.stdout().writer(self.io, &stdout_buf); @@ -313,7 +404,13 @@ pub const Consumer = struct { /// identity/account (payload = the envelope re-encoded as CBOR). /// returns the local seq of the FIRST row appended (0 if none); rows /// for one upstream event are contiguous, so frame i has seq first+i - fn persist(self: *Consumer, a: *archive_mod.Archive, event: zat.FirehoseEvent) !Persisted { + fn persist( + self: *Consumer, + a: *archive_mod.Archive, + event: zat.FirehoseEvent, + verdict: verify.Result, + stage_cursor: bool, + ) !Persisted { var arena = std.heap.ArenaAllocator.init(self.allocator); defer arena.deinit(); const alloc = arena.allocator(); @@ -388,10 +485,64 @@ pub const Consumer = struct { }, .info => {}, } - const first = (try a.appendBatch(rows.items, now_us)) orelse return .{}; + const Boundary = struct { + consumer: *Consumer, + event: zat.FirehoseEvent, + verdict: verify.Result, + stage_cursor: bool, + + fn stage(ctx: *anyopaque, _: u64, last: u64) !void { + const boundary: *@This() = @ptrCast(@alignCast(ctx)); + try boundary.consumer.stageEventBoundary(boundary.event, boundary.verdict, last, boundary.stage_cursor); + } + }; + var boundary: Boundary = .{ + .consumer = self, + .event = event, + .verdict = verdict, + // Repair replay is synthetic archive work. Its original relay + // frame was a verifier silent-drop, so (matching Atmos) it does + // not advance the cursor by itself; a later normally emitted + // upstream event advances the watermark beyond it. + .stage_cursor = stage_cursor and event.seq() != null, + }; + const first = (try a.appendBatchBeforeDurable(rows.items, now_us, &boundary, Boundary.stage)) orelse return .{}; return .{ .first = first, .last = first + rows.items.len - 1 }; } + fn stageEventBoundary( + self: *Consumer, + event: zat.FirehoseEvent, + verdict: verify.Result, + local_seq: u64, + stage_cursor: bool, + ) !void { + if (verdict.chain) |chain| if (event == .commit or event == .sync) { + const did = switch (event) { + .commit => |value| value.repo, + .sync => |value| value.did, + else => unreachable, + }; + const owned_did = try self.allocator.dupe(u8, did); + errdefer self.allocator.free(owned_did); + try self.pending_chains.append(self.allocator, .{ .local_seq = local_seq, .did = owned_did, .state = chain }); + }; + if (verdict.hosting) |hosting| if (event == .account) { + const owned_did = try self.allocator.dupe(u8, event.account.did); + errdefer self.allocator.free(owned_did); + try self.pending_hosting.append(self.allocator, .{ .local_seq = local_seq, .did = owned_did, .state = hosting }); + }; + switch (event) { + .identity => |identity| try self.stageRatchet(.identity, identity.did, identity.seq, local_seq), + .account => |account| try self.stageRatchet(.account, account.did, account.seq, local_seq), + else => {}, + } + if (stage_cursor) if (event.seq()) |seq| { + if (self.stats) |stats| stats.upstream_seq.store(seq, .monotonic); + try self.pending_cursors.append(self.allocator, .{ .local_seq = local_seq, .upstream_seq = seq }); + }; + } + pub fn deinit(self: *Consumer) void { self.pending_cursors.deinit(self.allocator); for (self.pending_chains.items) |point| self.allocator.free(point.did); @@ -503,3 +654,54 @@ test "applied identity and account ratchets commit with durable boundary" { try testing.expectEqual(@as(i64, 10), try consumer.appliedSeq(.identity, "did:plc:test")); try testing.expectEqual(@as(i64, 12), try consumer.appliedSeq(.account, "did:plc:test")); } + +test "verifier silent drop waits for a later emitted cursor watermark" { + const testing = std.testing; + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var path_buf: [std.fs.max_path_bytes]u8 = undefined; + const path_len = try tmp.dir.realPath(io, &path_buf); + const root = path_buf[0..path_len]; + var meta = try meta_store.Store.open(testing.allocator, root); + defer meta.deinit(); + var cursor = cursor_mod.Store.init(&meta); + var archive = try archive_mod.Archive.init(testing.allocator, io, root); + defer archive.deinit(); + var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20); + defer tail.deinit(); + tail.enableDurabilityGate(0); + var consumer: Consumer = .{ + .allocator = testing.allocator, + .io = io, + .tail = &tail, + .upstream = "ws://offline.invalid", + .cursor_store = &cursor, + .archive = &archive, + }; + defer consumer.deinit(); + archive.on_durable_ctx = &consumer; + archive.on_durable = Consumer.onArchiveDurable; + + const trigger: zat.FirehoseEvent = .{ .commit = .{ + .seq = 11, + .repo = "did:plc:repair-cursor", + .rev = "3l3qo2vutsw2b", + .time = "2024-09-09T19:46:02.102Z", + .ops = &.{}, + } }; + try consumer.handleEvent(trigger, .{ .outcome = .needs_resync }, false); + try testing.expectEqual(@as(?i64, null), cursor.load(io)); + + const later: zat.FirehoseEvent = .{ .identity = .{ + .seq = 12, + .did = "did:plc:later-event", + .time = "2024-09-09T19:46:03.102Z", + .handle = "later.test", + } }; + try consumer.handleEvent(later, .{}, false); + try archive.flushBlock(); + try testing.expectEqual(@as(?i64, 12), cursor.load(io)); +} diff --git a/src/internal/metrics.zig b/src/internal/metrics.zig index bcf49d8..1f27ccf 100644 --- a/src/internal/metrics.zig +++ b/src/internal/metrics.zig @@ -32,6 +32,14 @@ pub const Stats = struct { verify_unverified: Counter = .init(0), verify_cache_hits: Counter = .init(0), verify_cache_misses: Counter = .init(0), + resync_completed_total: Counter = .init(0), + resync_failed_total: Counter = .init(0), + resync_pending_dropped_total: Counter = .init(0), + resync_rate_limited_total: Counter = .init(0), + resync_stale_total: Counter = .init(0), + resync_active: Counter = .init(0), + resync_queue_depth: std.atomic.Value(usize) = .init(0), + resync_pending: std.atomic.Value(usize) = .init(0), livestream_decode_errors_total: Counter = .init(0), replayed_account_events_dropped_total: Counter = .init(0), replayed_identity_events_dropped_total: Counter = .init(0), @@ -108,6 +116,33 @@ pub fn format( tail.tip, }) catch {}; + w.print( + \\# TYPE stream_resync_total counter + \\stream_resync_total{{result="completed"}} {d} + \\stream_resync_total{{result="failed"}} {d} + \\stream_resync_total{{result="rate_limited"}} {d} + \\# TYPE stream_resync_pending_dropped_total counter + \\stream_resync_pending_dropped_total {d} + \\# TYPE stream_resync_workers_active gauge + \\stream_resync_workers_active {d} + \\# TYPE stream_resync_queue_depth gauge + \\stream_resync_queue_depth {d} + \\# TYPE stream_resync_pending gauge + \\stream_resync_pending {d} + \\# TYPE jetstream_livestream_stale_resyncs_dropped_total counter + \\jetstream_livestream_stale_resyncs_dropped_total {d} + \\ + , .{ + stats.resync_completed_total.load(.monotonic), + stats.resync_failed_total.load(.monotonic), + stats.resync_rate_limited_total.load(.monotonic), + stats.resync_pending_dropped_total.load(.monotonic), + stats.resync_active.load(.monotonic), + stats.resync_queue_depth.load(.monotonic), + stats.resync_pending.load(.monotonic), + stats.resync_stale_total.load(.monotonic), + }) catch {}; + // Canonical upstream names. These are aliases only where Stream owns the // same lifecycle boundary; absent families stay absent so the dashboard // contract cannot be satisfied by placeholder zeroes. diff --git a/src/internal/pipeline.zig b/src/internal/pipeline.zig index 49d70a1..899b130 100644 --- a/src/internal/pipeline.zig +++ b/src/internal/pipeline.zig @@ -12,6 +12,7 @@ const std = @import("std"); const zat = @import("zat"); const verify = @import("verify.zig"); +const repair = @import("repair.zig"); const meta_store = @import("meta_store.zig"); const metrics = @import("metrics.zig"); @@ -19,7 +20,7 @@ const Io = std.Io; const Allocator = std.mem.Allocator; const log = std.log.scoped(.stream); -const worker_count = 4; +const worker_count = 32; /// Per-DID verification is serialized through ordered hash lanes. Tickets /// register in relay order, then different lanes may verify concurrently. /// Hash collisions only reduce parallelism; they cannot reorder a repo. @@ -44,11 +45,13 @@ pub const Pipeline = struct { allocator: Allocator, io: Io, verifier: ?*verify.Verifier, + repair: ?*repair.Coordinator = null, stats: ?*metrics.Stats = null, /// emit(ctx, frame_bytes, verdict) — runs on the writer thread, in /// ticket order; owns nothing (pipeline frees the frame after). emit_ctx: *anyopaque, emit: *const fn (*anyopaque, zat.FirehoseEvent, verify.Result) anyerror!void, + emit_repair: ?*const fn (*anyopaque, *repair.Completion) anyerror!void = null, mutex: Io.Mutex = .init, /// workers wait for work; reader waits for space; writer waits for done @@ -62,10 +65,13 @@ pub const Pipeline = struct { next_emit: u64 = 0, // next ticket the writer will emit accepting: bool = true, alive: bool = true, + stopped: bool = false, fatal_error: ?anyerror = null, next_verify_register: u64 = 0, verify_lane_tail: [verification_lane_count]?u64 = @splat(null), verify_lane_completed: [verification_lane_count]?u64 = @splat(null), + repair_queue: std.ArrayList(*repair.Completion) = .empty, + repair_inflight: bool = false, // workers + writer run as io.concurrent tasks, never raw std.Threads: // they call io-backed I/O (DID resolution, archive writes), and Io @@ -100,6 +106,9 @@ pub const Pipeline = struct { } pub fn stop(self: *Pipeline) void { + if (self.stopped) return; + self.stopped = true; + if (self.repair) |coordinator| coordinator.stop(); self.signalStop(); for (&self.futures) |*f| { if (f.*) |*future| _ = future.cancel(self.io); @@ -109,6 +118,8 @@ pub const Pipeline = struct { if (slot.*) |s| self.freeSlot(s); slot.* = null; } + for (self.repair_queue.items) |completion| completion.deinit(); + self.repair_queue.deinit(self.allocator); } /// Stop accepting frames, wait until every already-accepted ticket has @@ -132,6 +143,11 @@ pub const Pipeline = struct { while (self.alive and self.next_emit != self.next_submit) { self.space_cond.wait(self.io, &self.mutex) catch {}; } + self.mutex.unlock(self.io); + if (self.repair) |coordinator| try coordinator.drainAndStop(); + self.mutex.lockUncancelable(self.io); + while (self.alive and (self.repair_queue.items.len != 0 or self.repair_inflight)) + self.space_cond.wait(self.io, &self.mutex) catch {}; const failure = self.fatal_error; self.mutex.unlock(self.io); self.stop(); @@ -149,7 +165,7 @@ pub const Pipeline = struct { pub fn fatalError(self: *Pipeline) ?anyerror { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); - return self.fatal_error; + return self.fatal_error orelse if (self.repair) |coordinator| coordinator.fatalError() else null; } /// called by the ws reader: takes ownership of nothing — dupes `data`. @@ -219,7 +235,12 @@ pub const Pipeline = struct { const turn = self.beginVerification(ticket, if (event) |ev| eventDid(ev) else null); if (turn.proceed) { defer self.endVerification(ticket, turn.lane); - if (event) |ev| verdict = switch (ev) { + if (event) |ev| verdict = if (self.repair) |coordinator| + coordinator.process(if (resolver_storage) |*r| r else undefined, frame, ev, ticket) catch |err| blk: { + failure = err; + break :blk .{}; + } + else switch (ev) { .commit => |c| v.verifyCommit(if (resolver_storage) |*r| r else undefined, c) catch |err| blk: { failure = err; break :blk .{}; @@ -307,6 +328,7 @@ pub const Pipeline = struct { while (true) { self.mutex.lockUncancelable(self.io); while (self.alive) { + if (self.eligibleRepairIndex() != null) break; const slot = self.slots[self.next_emit % capacity]; if (slot != null and slot.?.done and slot.?.ticket == self.next_emit) break; self.done_cond.wait(self.io, &self.mutex) catch {}; @@ -315,6 +337,27 @@ pub const Pipeline = struct { self.mutex.unlock(self.io); return; } + if (self.eligibleRepairIndex()) |repair_index| { + const completion = self.repair_queue.orderedRemove(repair_index); + self.repair_inflight = true; + self.mutex.unlock(self.io); + const callback = self.emit_repair orelse { + completion.deinit(); + self.failRepair(error.MissingRepairEmitter); + return; + }; + callback(self.emit_ctx, completion) catch |err| { + completion.deinit(); + self.failRepair(err); + return; + }; + completion.deinit(); + self.mutex.lockUncancelable(self.io); + self.repair_inflight = false; + self.space_cond.broadcast(self.io); + self.mutex.unlock(self.io); + continue; + } const slot = self.slots[self.next_emit % capacity].?; self.mutex.unlock(self.io); @@ -341,6 +384,43 @@ pub const Pipeline = struct { } } + /// Caller holds `mutex`. Choose the oldest trigger whose ordinary ticket + /// has already crossed the writer callback; a faster later repair cannot + /// jump ahead of earlier relay input. + fn eligibleRepairIndex(self: *Pipeline) ?usize { + var best: ?usize = null; + for (self.repair_queue.items, 0..) |completion, i| { + if (completion.after_ticket >= self.next_emit) continue; + if (best == null or completion.after_ticket < self.repair_queue.items[best.?].after_ticket) + best = i; + } + return best; + } + + /// Repair workers call this while still holding their per-DID lane. The + /// writer mutex makes registration atomic with selection of the next + /// ordinary ticket, preserving upstream's ordered outbox guarantee. + pub fn publishRepair(ctx: *anyopaque, completion: *repair.Completion) !void { + const self: *Pipeline = @ptrCast(@alignCast(ctx)); + self.mutex.lockUncancelable(self.io); + defer self.mutex.unlock(self.io); + if (!self.alive) return error.Closed; + try self.repair_queue.append(self.allocator, completion); + self.done_cond.signal(self.io); + } + + fn failRepair(self: *Pipeline, err: anyerror) void { + self.mutex.lockUncancelable(self.io); + if (self.fatal_error == null) self.fatal_error = err; + self.repair_inflight = false; + self.alive = false; + self.work_cond.broadcast(self.io); + self.space_cond.broadcast(self.io); + self.done_cond.broadcast(self.io); + self.verify_cond.broadcast(self.io); + self.mutex.unlock(self.io); + } + fn failSlot(self: *Pipeline, slot: Slot, err: anyerror) void { self.mutex.lockUncancelable(self.io); self.slots[self.next_emit % capacity] = null; @@ -416,6 +496,35 @@ test "pipeline preserves submission order under parallel workers" { } } +test "repair outbox waits for trigger ticket and selects relay order" { + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + var collector: Collector = .{ .allocator = testing.allocator }; + var pipeline: Pipeline = .{ + .allocator = testing.allocator, + .io = threaded.io(), + .verifier = null, + .emit_ctx = &collector, + .emit = Collector.emit, + }; + // Only `after_ticket` is observed by eligibleRepairIndex; these never + // enter writer ownership and are deliberately stack-backed. + var later: repair.Completion = undefined; + later.after_ticket = 9; + var earlier: repair.Completion = undefined; + earlier.after_ticket = 4; + try pipeline.repair_queue.append(testing.allocator, &later); + try pipeline.repair_queue.append(testing.allocator, &earlier); + defer pipeline.repair_queue.deinit(testing.allocator); + + pipeline.next_emit = 4; + try testing.expect(pipeline.eligibleRepairIndex() == null); + pipeline.next_emit = 5; + try testing.expectEqual(@as(?usize, 1), pipeline.eligibleRepairIndex()); + pipeline.next_emit = 10; + try testing.expectEqual(@as(?usize, 1), pipeline.eligibleRepairIndex()); +} + test "pipeline exposes emit failure without advancing emitted ticket" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); diff --git a/src/internal/repair.zig b/src/internal/repair.zig new file mode 100644 index 0000000..4ae5ad2 --- /dev/null +++ b/src/internal/repair.zig @@ -0,0 +1,867 @@ +//! Authenticated asynchronous whole-repo repair for live Sync 1.1 ingest. +//! +//! This is the Atmos/Jetstream V2 state machine, expressed in Stream's +//! ordered pipeline: 32 fetch workers, a 64-job bounded queue, five repairs +//! per DID per minute (burst five), 2,048 pending commits per DID, unlocked +//! five-minute getRepo fetches, stale-head rejection, authenticated apply, +//! and ordered replay before the per-DID lane is released. + +const std = @import("std"); +const zat = @import("zat"); +const fetched_car = @import("bootstrap/fetched_car.zig"); +const prepared_fetch = @import("bootstrap/prepared_fetch.zig"); +const verify = @import("verify.zig"); +const metrics = @import("metrics.zig"); +const lru = @import("lru.zig"); +const meta_store = @import("meta_store.zig"); + +const Io = std.Io; +const Allocator = std.mem.Allocator; +const log = std.log.scoped(.stream); + +pub const worker_count = 32; +pub const queue_capacity = worker_count * 2; +pub const pending_capacity = 2048; +const lane_count = 4096; +const limiter_capacity = 16_384; +const refill_us = 12 * std.time.us_per_s; // five tokens per minute + +const Bucket = struct { tokens: u8 = 5, last_us: i64 }; + +pub const PendingVerified = struct { + arena: *std.heap.ArenaAllocator, + event: zat.FirehoseEvent, + verdict: verify.Result, +}; + +/// Ownership handed to the single archive writer. The fetched mmap and parse +/// arena remain live until all replacement rows have been emitted. +pub const Completion = struct { + allocator: Allocator, + io: Io, + did: []u8, + prepared_arena: *std.heap.ArenaAllocator, + prepared: prepared_fetch.Prepared, + fetched_chain: verify.ChainUpdate, + pending: std.ArrayList(PendingVerified) = .empty, + witnessed_at: i64, + /// Successful synchronous #sync repair carries the original firehose + /// cursor. Commit-triggered and retry repairs are synthetic and carry nil. + upstream_seq: ?i64 = null, + /// The trigger ticket must complete ordered emission before this repair + /// may enter the archive. This preserves the verifier collector's global + /// result order while per-DID apply remains asynchronous. + after_ticket: u64, + + pub fn deinit(self: *Completion) void { + for (self.pending.items) |item| { + item.arena.deinit(); + self.allocator.destroy(item.arena); + } + self.pending.deinit(self.allocator); + self.prepared.deinit(self.io); + self.prepared_arena.deinit(); + self.allocator.destroy(self.prepared_arena); + self.allocator.free(self.did); + self.allocator.destroy(self); + } +}; + +const Status = enum { idle, resyncing }; +const State = struct { + did: []u8, + status: Status = .idle, + pending: std.ArrayList([]u8) = .empty, + trigger_ticket: u64 = 0, +}; +const Job = struct { did: []const u8, upstream_seq: ?i64 = null }; + +pub const Coordinator = struct { + allocator: Allocator, + io: Io, + verifier: *verify.Verifier, + relay_http: []const u8, + scratch_dir: []const u8, + stats: ?*metrics.Stats = null, + emit_failure_logs: bool = true, + publish_ctx: *anyopaque, + publish: *const fn (*anyopaque, *Completion) anyerror!void, + + lanes: [lane_count]Io.Mutex = @splat(.init), + states_mu: Io.Mutex = .init, + states: std.StringHashMapUnmanaged(*State) = .empty, + limiter_mu: Io.Mutex = .init, + limiters: ?lru.LruCache(Bucket) = null, + + queue_mu: Io.Mutex = .init, + queue_cond: Io.Condition = .init, + queue_space: Io.Condition = .init, + queue: [queue_capacity]?Job = @splat(null), + queue_read: u64 = 0, + queue_write: u64 = 0, + accepting: bool = true, + alive: bool = true, + fatal_error: ?anyerror = null, + futures: [worker_count]?Io.Future(void) = @splat(null), + + pub fn start(self: *Coordinator) !void { + self.limiters = lru.LruCache(Bucket).init(self.allocator, limiter_capacity, self.io); + var spawned: usize = 0; + errdefer { + self.stop(); + for (self.futures[0..spawned]) |*slot| { + if (slot.*) |*future| _ = future.cancel(self.io); + } + } + for (&self.futures) |*slot| { + slot.* = try self.io.concurrent(workerLoop, .{self}); + spawned += 1; + } + } + + pub fn stop(self: *Coordinator) void { + self.queue_mu.lockUncancelable(self.io); + self.accepting = false; + self.alive = false; + self.queue_cond.broadcast(self.io); + self.queue_space.broadcast(self.io); + self.queue_mu.unlock(self.io); + for (&self.futures) |*slot| { + if (slot.*) |*future| _ = future.cancel(self.io); + slot.* = null; + } + } + + /// Close the job queue and wait for accepted repairs to publish their + /// completion. This is part of the archive cutover durability boundary. + pub fn drainAndStop(self: *Coordinator) !void { + self.queue_mu.lockUncancelable(self.io); + self.accepting = false; + self.queue_cond.broadcast(self.io); + self.queue_mu.unlock(self.io); + for (&self.futures) |*slot| { + if (slot.*) |*future| _ = future.await(self.io); + slot.* = null; + } + self.queue_mu.lockUncancelable(self.io); + const failure = self.fatal_error; + self.alive = false; + self.queue_mu.unlock(self.io); + if (failure) |err| return err; + } + + pub fn deinit(self: *Coordinator) void { + self.stop(); + var it = self.states.iterator(); + while (it.next()) |entry| self.freeState(entry.value_ptr.*); + self.states.deinit(self.allocator); + if (self.limiters) |*limiters| limiters.deinit(); + self.limiters = null; + } + + pub fn fatalError(self: *Coordinator) ?anyerror { + self.queue_mu.lockUncancelable(self.io); + defer self.queue_mu.unlock(self.io); + return self.fatal_error; + } + + /// Verify one decoded frame while holding the same bounded per-DID lane + /// used by repair apply. A commit arriving during fetch is copied into the + /// bounded pending FIFO; the reader and other DIDs continue immediately. + pub fn process( + self: *Coordinator, + resolver: *zat.DidResolver, + frame: []const u8, + event: zat.FirehoseEvent, + ticket: u64, + ) !verify.Result { + const did = eventDid(event) orelse return .{}; + const lane = self.laneFor(did); + lane.lockUncancelable(self.io); + var lane_locked = true; + defer if (lane_locked) lane.unlock(self.io); + + if (event == .commit) { + if (self.getState(did)) |state| if (state.status == .resyncing) { + try self.bufferPending(state, frame); + return .{ .outcome = .needs_resync }; + }; + } + + const result = switch (event) { + .commit => |value| try self.verifier.verifyCommit(resolver, value), + .sync => |value| try self.verifier.verifySync(resolver, value), + .identity => |value| blk: { + self.verifier.evict(value.did); + break :blk verify.Result{}; + }, + .account => |value| try self.verifier.onAccount(value), + .info => verify.Result{}, + }; + if (result.outcome != .needs_resync) return result; + + // Unlike a broken #commit, Atmos first repairs #sync inline. The + // network fetch is unlocked, and only a retry enters the async queue. + if (event == .sync) { + if (!try self.beginSynchronousRepairLocked(did, ticket)) return result; + const state = self.getState(did).?; + lane.unlock(self.io); + lane_locked = false; + self.runJob(resolver, .{ .did = state.did, .upstream_seq = event.sync.seq }) catch |err| { + if (self.emit_failure_logs) + log.err("synchronous sync repair failed for {s}: {s}", .{ did, @errorName(err) }); + self.failJob(did, err); + if (retryableSyncFailure(err)) { + lane.lockUncancelable(self.io); + lane_locked = true; + try self.scheduleLocked(did, ticket); + } + }; + return result; + } + + try self.scheduleLocked(did, ticket); + return result; + } + + /// Begin the inline #sync attempt. Caller holds the DID lane. A repair + /// already in flight is sufficient, so the sync directive is withheld. + fn beginSynchronousRepairLocked(self: *Coordinator, did: []const u8, ticket: u64) !bool { + const existing = self.getState(did); + if (existing) |state| if (state.status == .resyncing) return false; + const state = existing orelse try self.createState(did); + state.status = .resyncing; + state.trigger_ticket = ticket; + return true; + } + + fn scheduleLocked(self: *Coordinator, did: []const u8, ticket: u64) !void { + const existing = self.getState(did); + const state = existing orelse try self.createState(did); + if (state.status == .resyncing) return; + state.status = .resyncing; + state.trigger_ticket = ticket; + + self.queue_mu.lockUncancelable(self.io); + defer self.queue_mu.unlock(self.io); + while (self.alive and self.accepting and self.queue_write - self.queue_read >= queue_capacity) + try self.queue_space.wait(self.io, &self.queue_mu); + if (!self.alive or !self.accepting) return error.Closed; + self.queue[self.queue_write % queue_capacity] = .{ .did = state.did }; + self.queue_write += 1; + self.queue_cond.signal(self.io); + if (self.stats) |stats| stats.resync_queue_depth.store(@intCast(self.queue_write - self.queue_read), .monotonic); + } + + fn bufferPending(self: *Coordinator, state: *State, frame: []const u8) !void { + const owned = try self.allocator.dupe(u8, frame); + const grew = state.pending.items.len != pending_capacity; + if (!grew) { + const dropped = state.pending.orderedRemove(0); + self.allocator.free(dropped); + if (self.stats) |stats| _ = stats.resync_pending_dropped_total.fetchAdd(1, .monotonic); + if (self.emit_failure_logs) + log.err("repair pending overflow: {s}; dropped oldest commit", .{state.did}); + } + try state.pending.append(self.allocator, owned); + if (grew) { + if (self.stats) |stats| _ = stats.resync_pending.fetchAdd(1, .monotonic); + } + } + + fn workerLoop(self: *Coordinator) void { + var resolver = self.verifier.makeResolver(); + defer resolver.deinit(); + while (self.takeJob()) |job| self.runJob(&resolver, job) catch |err| { + if (self.emit_failure_logs) + log.err("repair worker failed for {s}: {s}", .{ job.did, @errorName(err) }); + self.failJob(job.did, err); + }; + } + + fn takeJob(self: *Coordinator) ?Job { + self.queue_mu.lockUncancelable(self.io); + defer self.queue_mu.unlock(self.io); + while (self.alive and self.queue_read == self.queue_write and self.accepting) + self.queue_cond.wait(self.io, &self.queue_mu) catch {}; + if (!self.alive or (self.queue_read == self.queue_write and !self.accepting)) return null; + const index = self.queue_read % queue_capacity; + const job = self.queue[index].?; + self.queue[index] = null; + self.queue_read += 1; + self.queue_space.signal(self.io); + if (self.stats) |stats| stats.resync_queue_depth.store(@intCast(self.queue_write - self.queue_read), .monotonic); + return job; + } + + fn runJob(self: *Coordinator, resolver: *zat.DidResolver, job: Job) !void { + if (self.stats) |stats| _ = stats.resync_active.fetchAdd(1, .monotonic); + defer { + if (self.stats) |stats| _ = stats.resync_active.fetchSub(1, .monotonic); + } + + const now_us = Io.Timestamp.now(self.io, .real).toMicroseconds(); + if (!try self.takeRepairToken(job.did, now_us)) { + if (self.stats) |stats| _ = stats.resync_rate_limited_total.fetchAdd(1, .monotonic); + log.warn("repair rate limited: {s}", .{job.did}); + return error.RepairRateLimited; + } + + var parse_arena = try self.allocator.create(std.heap.ArenaAllocator); + parse_arena.* = std.heap.ArenaAllocator.init(self.allocator); + errdefer { + parse_arena.deinit(); + self.allocator.destroy(parse_arena); + } + var final_host: fetched_car.FinalHost = .{}; + const outcome = try prepared_fetch.fetch( + self.allocator, + parse_arena.allocator(), + self.io, + self.scratch_dir, + self.relay_http, + job.did, + &final_host, + ); + var prepared = switch (outcome) { + .ok => |value| value, + else => return error.GetRepoFailed, + }; + errdefer prepared.deinit(self.io); + + const lane = self.laneFor(job.did); + lane.lockUncancelable(self.io); + defer lane.unlock(self.io); + const state = self.getState(job.did) orelse return error.RepairStateMissing; + if (state.status != .resyncing) return error.RepairStateMissing; + const fetched = try self.verifier.verifyFetchedRepo(resolver, job.did, &prepared.repo); + const fetched_chain = fetched.chain orelse return error.RepairAuthenticationFailed; + try validateFetchedHead(try self.verifier.chainFor(job.did), fetched_chain); + try self.verifier.rememberChain(job.did, fetched_chain); + + const completion = try self.allocator.create(Completion); + errdefer self.allocator.destroy(completion); + completion.* = .{ + .allocator = self.allocator, + .io = self.io, + .did = try self.allocator.dupe(u8, job.did), + .prepared_arena = parse_arena, + .prepared = prepared, + .fetched_chain = fetched_chain, + .witnessed_at = Io.Timestamp.now(self.io, .real).toMicroseconds(), + .after_ticket = state.trigger_ticket, + .upstream_seq = job.upstream_seq, + }; + errdefer { + self.allocator.free(completion.did); + for (completion.pending.items) |item| { + item.arena.deinit(); + self.allocator.destroy(item.arena); + } + completion.pending.deinit(self.allocator); + } + + for (state.pending.items) |raw| { + const arena = try self.allocator.create(std.heap.ArenaAllocator); + arena.* = std.heap.ArenaAllocator.init(self.allocator); + const event = zat.firehose.decodeFrame(arena.allocator(), raw) catch { + arena.deinit(); + self.allocator.destroy(arena); + continue; + }; + const verdict = switch (event) { + .commit => |value| try self.verifier.verifyCommit(resolver, value), + else => { + arena.deinit(); + self.allocator.destroy(arena); + continue; + }, + }; + switch (verdict.outcome) { + .valid => try completion.pending.append(self.allocator, .{ + .arena = arena, + .event = event, + .verdict = verdict, + }), + .replay => { + arena.deinit(); + self.allocator.destroy(arena); + }, + else => { + log.err("pending repair replay rejected: {s} reason={s}", .{ job.did, @tagName(verdict.outcome) }); + arena.deinit(); + self.allocator.destroy(arena); + }, + } + } + // Registration under the DID lane is the ordered outbox boundary: + // no later same-DID verification can finish before this callback. + try self.publish(self.publish_ctx, completion); + parse_arena = undefined; + prepared = undefined; + self.removeState(state); + if (self.stats) |stats| _ = stats.resync_completed_total.fetchAdd(1, .monotonic); + } + + fn failJob(self: *Coordinator, did: []const u8, err: anyerror) void { + const lane = self.laneFor(did); + lane.lockUncancelable(self.io); + defer lane.unlock(self.io); + if (self.getState(did)) |state| { + if (state.pending.items.len == 0) + self.removeState(state) + else + state.status = .idle; + } + if (self.stats) |stats| _ = stats.resync_failed_total.fetchAdd(1, .monotonic); + if (err == error.StaleRepair) { + if (self.stats) |stats| _ = stats.resync_stale_total.fetchAdd(1, .monotonic); + } + // Local allocation/write/publication failures are fatal. Remote and + // validation failures remain observable and allow a later retry. + switch (err) { + error.OutOfMemory, error.WriteFailed, error.ArchiveAppendFailed => self.setFatal(err), + else => {}, + } + } + + fn setFatal(self: *Coordinator, err: anyerror) void { + self.queue_mu.lockUncancelable(self.io); + defer self.queue_mu.unlock(self.io); + if (self.fatal_error == null) self.fatal_error = err; + } + + fn laneFor(self: *Coordinator, did: []const u8) *Io.Mutex { + return &self.lanes[std.hash.Wyhash.hash(0, did) % lane_count]; + } + + fn getState(self: *Coordinator, did: []const u8) ?*State { + self.states_mu.lockUncancelable(self.io); + defer self.states_mu.unlock(self.io); + return self.states.get(did); + } + + fn createState(self: *Coordinator, did: []const u8) !*State { + const state = try self.allocator.create(State); + errdefer self.allocator.destroy(state); + state.* = .{ .did = try self.allocator.dupe(u8, did) }; + errdefer self.allocator.free(state.did); + self.states_mu.lockUncancelable(self.io); + defer self.states_mu.unlock(self.io); + try self.states.put(self.allocator, state.did, state); + return state; + } + + fn removeState(self: *Coordinator, state: *State) void { + self.states_mu.lockUncancelable(self.io); + _ = self.states.remove(state.did); + self.states_mu.unlock(self.io); + if (self.stats) |stats| _ = stats.resync_pending.fetchSub(state.pending.items.len, .monotonic); + self.freeState(state); + } + + fn freeState(self: *Coordinator, state: *State) void { + for (state.pending.items) |frame| self.allocator.free(frame); + state.pending.deinit(self.allocator); + self.allocator.free(state.did); + self.allocator.destroy(state); + } + + fn takeRepairToken(self: *Coordinator, did: []const u8, now_us: i64) !bool { + self.limiter_mu.lockUncancelable(self.io); + defer self.limiter_mu.unlock(self.io); + const limiters = &(self.limiters orelse return error.CoordinatorNotStarted); + var bucket = limiters.get(did) orelse Bucket{ .last_us = now_us }; + const allowed = takeToken(&bucket, now_us); + try limiters.put(did, bucket); + return allowed; + } +}; + +fn takeToken(bucket: *Bucket, now_us: i64) bool { + if (now_us > bucket.last_us) { + const steps: u64 = @intCast(@divFloor(now_us - bucket.last_us, refill_us)); + if (steps != 0) { + bucket.tokens = @intCast(@min(@as(u64, bucket.tokens) + steps, 5)); + bucket.last_us += @as(i64, @intCast(steps)) * refill_us; + } + } + if (bucket.tokens == 0) return false; + bucket.tokens -= 1; + return true; +} + +fn validateFetchedHead(current: ?verify.ChainUpdate, fetched: verify.ChainUpdate) !void { + const seen = current orelse return; + switch (std.mem.order(u8, &fetched.rev, &seen.rev)) { + .lt => return error.StaleRepair, + .eq => if (!std.mem.eql(u8, fetched.dataSlice(), seen.dataSlice())) + return error.ContradictoryRepair, + .gt => {}, + } +} + +/// Match Atmos's one deferred retry for transport/rate-limit failures of a +/// synchronous #sync repair. Authentication and rev-regression failures are +/// deterministic and deliberately do not retry the same authoritative head. +fn retryableSyncFailure(err: anyerror) bool { + return switch (err) { + error.GetRepoFailed, + error.AttemptTimeout, + error.TransferTooSlow, + error.RepairRateLimited, + error.ConnectionRefused, + error.ConnectionResetByPeer, + error.NetworkUnreachable, + error.TemporaryNameServerFailure, + => true, + else => false, + }; +} + +fn eventDid(event: zat.FirehoseEvent) ?[]const u8 { + return switch (event) { + .commit => |value| value.repo, + .sync => |value| value.did, + .identity => |value| value.did, + .account => |value| value.did, + .info => null, + }; +} + +test "repair limiter matches upstream five per minute burst five" { + var bucket: Bucket = .{ .last_us = 1_000_000 }; + for (0..5) |_| try std.testing.expect(takeToken(&bucket, 1_000_000)); + try std.testing.expect(!takeToken(&bucket, 1_000_000)); + try std.testing.expect(!takeToken(&bucket, 1_000_000 + 11 * std.time.us_per_s)); + try std.testing.expect(takeToken(&bucket, 1_000_000 + 12 * std.time.us_per_s)); + try std.testing.expect(!takeToken(&bucket, 1_000_000 + 12 * std.time.us_per_s)); + try std.testing.expect(takeToken(&bucket, 1_000_000 + 72 * std.time.us_per_s)); + try std.testing.expect(takeToken(&bucket, 1_000_000 + 72 * std.time.us_per_s)); + try std.testing.expect(takeToken(&bucket, 1_000_000 + 72 * std.time.us_per_s)); + try std.testing.expect(takeToken(&bucket, 1_000_000 + 72 * std.time.us_per_s)); + try std.testing.expect(takeToken(&bucket, 1_000_000 + 72 * std.time.us_per_s)); + try std.testing.expect(!takeToken(&bucket, 1_000_000 + 72 * std.time.us_per_s)); +} + +test "repair apply rejects older and contradictory authoritative heads" { + var current: verify.ChainUpdate = .{ .rev = undefined, .data_len = 3 }; + @memcpy(¤t.rev, "3k2abcdefghij"); + @memcpy(current.data[0..3], "one"); + var fetched = current; + try validateFetchedHead(current, fetched); + @memcpy(fetched.data[0..3], "two"); + try std.testing.expectError(error.ContradictoryRepair, validateFetchedHead(current, fetched)); + @memcpy(&fetched.rev, "3k2abc0000000"); + try std.testing.expectError(error.StaleRepair, validateFetchedHead(current, fetched)); + @memcpy(&fetched.rev, "3k2zzzzzzzzzz"); + try validateFetchedHead(current, fetched); +} + +test "pending FIFO is bounded at 2048 and drops the oldest real frame" { + const testing = std.testing; + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + var stats: metrics.Stats = .{}; + var coordinator: Coordinator = .{ + .allocator = testing.allocator, + .io = threaded.io(), + .verifier = undefined, + .relay_http = "http://offline.invalid", + .scratch_dir = "unused", + .stats = &stats, + .emit_failure_logs = false, + .publish_ctx = undefined, + .publish = undefined, + }; + var state: State = .{ .did = @constCast("did:plc:pending-fixture") }; + defer { + for (state.pending.items) |frame| testing.allocator.free(frame); + state.pending.deinit(testing.allocator); + } + var frame_buf: [16]u8 = undefined; + for (0..pending_capacity + 1) |i| { + const frame = try std.fmt.bufPrint(&frame_buf, "frame-{d}", .{i}); + try coordinator.bufferPending(&state, frame); + } + try testing.expectEqual(@as(usize, pending_capacity), state.pending.items.len); + try testing.expectEqualStrings("frame-1", state.pending.items[0]); + try testing.expectEqualStrings("frame-2048", state.pending.items[pending_capacity - 1]); + try testing.expectEqual(@as(u64, 1), stats.resync_pending_dropped_total.load(.monotonic)); + try testing.expectEqual(@as(usize, pending_capacity), stats.resync_pending.load(.monotonic)); +} + +test "offline worker pool performs exactly 32 concurrent real HTTP fetches" { + const testing = std.testing; + const request_count = 40; + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var server = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; + defer server.deinit(io); + var active: std.atomic.Value(usize) = .init(0); + var maximum: std.atomic.Value(usize) = .init(0); + var reached_pool: Io.Event = .unset; + var release: Io.Event = .unset; + + const Fixture = struct { + fn serveOne( + stream_value: Io.net.Stream, + task_io: Io, + active_count: *std.atomic.Value(usize), + max_count: *std.atomic.Value(usize), + reached: *Io.Event, + release_gate: *Io.Event, + ) void { + var stream = stream_value; + defer stream.close(task_io); + var read_buf: [2048]u8 = undefined; + var write_buf: [2048]u8 = undefined; + var reader = stream.reader(task_io, &read_buf); + var writer = stream.writer(task_io, &write_buf); + var http_server = std.http.Server.init(&reader.interface, &writer.interface); + var request = http_server.receiveHead() catch return; + const now = active_count.fetchAdd(1, .acq_rel) + 1; + var seen = max_count.load(.acquire); + while (now > seen) { + seen = max_count.cmpxchgWeak(seen, now, .acq_rel, .acquire) orelse break; + } + if (now == worker_count) reached.set(task_io); + release_gate.wait(task_io) catch return; + request.respond("", .{ .status = .service_unavailable, .keep_alive = false }) catch return; + _ = active_count.fetchSub(1, .acq_rel); + } + + fn run( + listener: *Io.net.Server, + task_io: Io, + active_count: *std.atomic.Value(usize), + max_count: *std.atomic.Value(usize), + reached: *Io.Event, + release_gate: *Io.Event, + ) !void { + var futures: [request_count]?Io.Future(void) = @splat(null); + for (&futures) |*slot| { + const stream = try listener.accept(task_io); + slot.* = try task_io.concurrent(serveOne, .{ stream, task_io, active_count, max_count, reached, release_gate }); + } + for (&futures) |*slot| { + if (slot.*) |*future| _ = future.await(task_io); + } + } + }; + var server_future = try io.concurrent(Fixture.run, .{ &server, io, &active, &maximum, &reached_pool, &release }); + defer _ = server_future.cancel(io) catch {}; + var url_buf: [96]u8 = undefined; + const url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}", .{server.socket.address.getPort()}); + + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var root_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const root_len = try tmp.dir.realPath(io, &root_buf); + const root = root_buf[0..root_len]; + var meta = try meta_store.Store.open(testing.allocator, root); + defer meta.deinit(); + var verifier = verify.Verifier.init(testing.allocator, io, "http://offline.invalid", &meta); + defer verifier.deinit(); + const NoPublish = struct { + fn publish(_: *anyopaque, _: *Completion) !void { + return error.UnexpectedCompletion; + } + }; + var scratch_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const scratch = try std.fmt.bufPrint(&scratch_buf, "{s}/scratch", .{root}); + var coordinator: Coordinator = .{ + .allocator = testing.allocator, + .io = io, + .verifier = &verifier, + .relay_http = url, + .scratch_dir = scratch, + .emit_failure_logs = false, + .publish_ctx = undefined, + .publish = NoPublish.publish, + }; + defer coordinator.deinit(); + try coordinator.start(); + for (0..request_count) |i| { + var did_buf: [64]u8 = undefined; + const did = try std.fmt.bufPrint(&did_buf, "did:plc:worker-{d}", .{i}); + const lane = coordinator.laneFor(did); + lane.lockUncancelable(io); + try coordinator.scheduleLocked(did, i); + lane.unlock(io); + } + try reached_pool.waitTimeout(io, .{ .duration = .{ .raw = Io.Duration.fromSeconds(2), .clock = .awake } }); + try testing.expectEqual(@as(usize, worker_count), active.load(.acquire)); + release.set(io); + try coordinator.drainAndStop(); + _ = try server_future.await(io); + try testing.expectEqual(@as(usize, worker_count), maximum.load(.acquire)); +} + +test "offline coordinator fetches and authenticates a real complete repo" { + const testing = std.testing; + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var arena = std.heap.ArenaAllocator.init(testing.allocator); + defer arena.deinit(); + const alloc = arena.allocator(); + + const keypair = try zat.Keypair.fromSecretKey(.p256, .{21} ** 32); + const did = "did:plc:repairfixture"; + const did_key = try keypair.did(alloc); + const multibase = did_key["did:key:".len..]; + const record = try zat.cbor.encodeAlloc(alloc, .{ .map = &.{ + .{ .key = "$type", .value = .{ .text = "app.bsky.feed.post" } }, + .{ .key = "text", .value = .{ .text = "offline repair" } }, + } }); + const record_cid = try zat.cbor.Cid.forDagCbor(alloc, record); + var tree = zat.mst.Mst.init(alloc); + try tree.put("app.bsky.feed.post/3k2abcdefghij", record_cid); + const data_cid = try tree.rootCid(); + const signed = try zat.signCommit(alloc, .{ + .did = did, + .rev = "3k2abcdefghij", + .data = data_cid, + }, &keypair); + var blocks: std.ArrayList(zat.car.Block) = .empty; + try blocks.append(alloc, .{ .cid_raw = signed.cid.raw, .data = signed.bytes }); + try tree.collectBlocks(&blocks); + try blocks.append(alloc, .{ .cid_raw = record_cid.raw, .data = record }); + const car_bytes = try zat.car.writeAlloc(alloc, .{ .roots = &.{signed.cid}, .blocks = blocks.items }); + const did_doc = try std.fmt.allocPrint(alloc, + \\{{"id":"{s}","verificationMethod":[{{"id":"{s}#atproto","type":"Multikey","controller":"{s}","publicKeyMultibase":"{s}"}}]}} + , .{ did, did, did, multibase }); + + var server = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; + defer server.deinit(io); + var fetch_started: Io.Event = .unset; + var release_fetch: Io.Event = .unset; + const Fixture = struct { + fn run(listener: *Io.net.Server, task_io: Io, doc: []const u8, car: []const u8, started: *Io.Event, release: *Io.Event) !void { + for (0..2) |_| { + var stream = try listener.accept(task_io); + defer stream.close(task_io); + var read_buf: [4096]u8 = undefined; + var write_buf: [4096]u8 = undefined; + var reader = stream.reader(task_io, &read_buf); + var writer = stream.writer(task_io, &write_buf); + var http_server = std.http.Server.init(&reader.interface, &writer.interface); + var request = try http_server.receiveHead(); + const is_repo = std.mem.startsWith(u8, request.head.target, "/xrpc/"); + if (is_repo) { + started.set(task_io); + try release.wait(task_io); + } + const body = if (is_repo) car else doc; + try request.respond(body, .{ + .status = .ok, + .keep_alive = false, + .extra_headers = &.{.{ + .name = "content-type", + .value = if (is_repo) "application/vnd.ipld.car" else "application/json", + }}, + }); + } + } + }; + var server_future = try io.concurrent(Fixture.run, .{ &server, io, did_doc, car_bytes, &fetch_started, &release_fetch }); + defer _ = server_future.cancel(io) catch {}; + var url_buf: [96]u8 = undefined; + const url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}", .{server.socket.address.getPort()}); + + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var root_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const root_len = try tmp.dir.realPath(io, &root_buf); + const root = root_buf[0..root_len]; + var meta = try meta_store.Store.open(testing.allocator, root); + defer meta.deinit(); + var verifier = verify.Verifier.init(testing.allocator, io, url, &meta); + defer verifier.deinit(); + + const Published = struct { + completion: ?*Completion = null, + fn publish(ctx: *anyopaque, completion: *Completion) !void { + const self: *@This() = @ptrCast(@alignCast(ctx)); + if (self.completion != null) return error.DuplicateCompletion; + self.completion = completion; + } + }; + var published: Published = .{}; + var scratch_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const scratch = try std.fmt.bufPrint(&scratch_buf, "{s}/scratch", .{root}); + var coordinator: Coordinator = .{ + .allocator = testing.allocator, + .io = io, + .verifier = &verifier, + .relay_http = url, + .scratch_dir = scratch, + .publish_ctx = &published, + .publish = Published.publish, + }; + defer coordinator.deinit(); + try coordinator.start(); + var resolver = verifier.makeResolver(); + defer resolver.deinit(); + const sync_event: zat.FirehoseEvent = .{ .sync = .{ + .seq = 1, + .did = did, + .rev = "3k2abcdefghij", + .time = "2026-07-16T00:00:00Z", + .blocks = &.{}, + } }; + const SyncTask = struct { + coordinator: *Coordinator, + resolver: *zat.DidResolver, + event: zat.FirehoseEvent, + result: ?(anyerror!verify.Result) = null, + + fn run(self: *@This()) void { + self.result = self.coordinator.process(self.resolver, "", self.event, 0); + } + }; + var sync_task: SyncTask = .{ .coordinator = &coordinator, .resolver = &resolver, .event = sync_event }; + var sync_future = try io.concurrent(SyncTask.run, .{&sync_task}); + defer _ = sync_future.cancel(io); + try fetch_started.wait(io); + + // The HTTP response is deliberately held open. A same-DID commit must + // still return immediately into the real pending FIFO: fetch I/O never + // owns the per-DID apply lane. + const pending_frame = try zat.firehose.encodeCommitEvent(alloc, .{ + .seq = 2, + .repo_did = did, + .commit_cid = signed.cid, + .rev = "3k2abcdefghij", + .blocks = car_bytes, + .ops = &.{.{ + .action = .create, + .collection = "app.bsky.feed.post", + .rkey = "3k2abcdefghij", + .cid = record_cid, + }}, + .time = "2026-07-16T00:00:01Z", + }); + var pending_arena = std.heap.ArenaAllocator.init(testing.allocator); + defer pending_arena.deinit(); + const pending_event = try zat.firehose.decodeFrame(pending_arena.allocator(), pending_frame); + const buffered = try coordinator.process(&resolver, pending_frame, pending_event, 1); + try testing.expectEqual(verify.Outcome.needs_resync, buffered.outcome); + release_fetch.set(io); + _ = sync_future.await(io); + const result = try sync_task.result.?; + try testing.expectEqual(verify.Outcome.needs_resync, result.outcome); + try coordinator.drainAndStop(); + _ = try server_future.await(io); + + const completion = published.completion orelse return error.MissingCompletion; + defer completion.deinit(); + try testing.expectEqualStrings(did, completion.did); + try testing.expectEqualStrings("3k2abcdefghij", &completion.fetched_chain.rev); + try testing.expectEqualSlices(u8, data_cid.raw, completion.fetched_chain.dataSlice()); + try testing.expectEqual(@as(usize, 0), completion.pending.items.len); + try testing.expectEqual(@as(?i64, 1), completion.upstream_seq); + try testing.expectEqual(@as(?anyerror, null), coordinator.fatalError()); +} diff --git a/src/internal/repair_integration_test.zig b/src/internal/repair_integration_test.zig new file mode 100644 index 0000000..4df2969 --- /dev/null +++ b/src/internal/repair_integration_test.zig @@ -0,0 +1,118 @@ +const std = @import("std"); +const zat = @import("zat"); +const archive_mod = @import("archive.zig"); +const cursor_mod = @import("cursor.zig"); +const ingest = @import("ingest.zig"); +const meta_store = @import("meta_store.zig"); +const repair = @import("repair.zig"); +const repos = @import("bootstrap/repos.zig"); +const segment = @import("segment.zig"); +const tail_mod = @import("tail.zig"); +const verify = @import("verify.zig"); + +const Io = std.Io; + +test "repair completion durably emits tombstone then replacement offline" { + const testing = std.testing; + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var root_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const root_len = try tmp.dir.realPath(io, &root_buf); + const root = root_buf[0..root_len]; + + var meta = try meta_store.Store.open(testing.allocator, root); + defer meta.deinit(); + var archive = try archive_mod.Archive.init(testing.allocator, io, root); + defer archive.deinit(); + var tail = tail_mod.Tail.init(testing.allocator, io, 8 * 1024 * 1024); + defer tail.deinit(); + tail.enableDurabilityGate(0); + var cursor = cursor_mod.Store.init(&meta); + var consumer: ingest.Consumer = .{ + .allocator = testing.allocator, + .io = io, + .tail = &tail, + .upstream = "ws://offline.invalid", + .cursor_store = &cursor, + .archive = &archive, + }; + defer consumer.deinit(); + archive.on_durable_ctx = &consumer; + archive.on_durable = ingest.Consumer.onArchiveDurable; + + const prepared_arena = try testing.allocator.create(std.heap.ArenaAllocator); + prepared_arena.* = std.heap.ArenaAllocator.init(testing.allocator); + defer { + prepared_arena.deinit(); + testing.allocator.destroy(prepared_arena); + } + const alloc = prepared_arena.allocator(); + const did = "did:plc:durable-repair"; + const record = try zat.cbor.encodeAlloc(alloc, .{ .map = &.{ + .{ .key = "$type", .value = .{ .text = "app.bsky.feed.post" } }, + .{ .key = "text", .value = .{ .text = "durable offline repair" } }, + } }); + const record_cid = try zat.cbor.Cid.forDagCbor(alloc, record); + var tree = zat.mst.Mst.init(alloc); + try tree.put("app.bsky.feed.post/3k2abcdefghij", record_cid); + const data_cid = try tree.rootCid(); + const keypair = try zat.Keypair.fromSecretKey(.p256, .{31} ** 32); + const signed = try zat.signCommit(alloc, .{ + .did = did, + .rev = "3k2abcdefghij", + .data = data_cid, + }, &keypair); + var blocks: std.ArrayList(zat.car.Block) = .empty; + try blocks.append(alloc, .{ .cid_raw = signed.cid.raw, .data = signed.bytes }); + try tree.collectBlocks(&blocks); + try blocks.append(alloc, .{ .cid_raw = record_cid.raw, .data = record }); + const car_bytes = try zat.car.writeAlloc(alloc, .{ .roots = &.{signed.cid}, .blocks = blocks.items }); + const prepared = try repos.prepareRepo(alloc, car_bytes); + var chain: verify.ChainUpdate = .{ .rev = undefined, .data_len = @intCast(data_cid.raw.len) }; + @memcpy(&chain.rev, "3k2abcdefghij"); + @memcpy(chain.data[0..data_cid.raw.len], data_cid.raw); + var completion: repair.Completion = .{ + .allocator = testing.allocator, + .io = io, + .did = @constCast(did), + .prepared_arena = prepared_arena, + .prepared = .{ .fetched = undefined, .repo = prepared }, + .fetched_chain = chain, + .witnessed_at = 1_784_160_000_000_000, + .after_ticket = 0, + .upstream_seq = 77, + }; + try ingest.Consumer.emitRepair(&consumer, &completion); + + try testing.expectEqual(@as(u64, 2), archive.committed_seq.load(.acquire)); + try testing.expectEqual(@as(?i64, 77), cursor.load(io)); + try testing.expectEqual(@as(usize, 2), tail.gauges().entries); + try testing.expectEqual(@as(usize, 0), tail.entries.items[0].json.len); + try testing.expect(tail.entries.items[0].json_v2.len > 0); + try testing.expectEqual(@as(usize, 0), tail.entries.items[1].json.len); + try testing.expect(tail.entries.items[1].json_v2.len > 0); + + var reloaded = verify.Verifier.init(testing.allocator, io, "http://offline.invalid", &meta); + defer reloaded.deinit(); + const durable_chain = (try reloaded.chainFor(did)).?; + try testing.expectEqualStrings("3k2abcdefghij", &durable_chain.rev); + try testing.expectEqualSlices(u8, data_cid.raw, durable_chain.dataSlice()); + + try archive.rotate(); + var name_buf: [64]u8 = undefined; + const name = archive_mod.formatSegmentName(&name_buf, 0); + const bytes = try archive.dir.readFileAlloc(io, name, testing.allocator, .limited(16 * 1024 * 1024)); + defer testing.allocator.free(bytes); + var sealed = try segment.Sealed.parse(testing.allocator, bytes); + defer sealed.deinit(testing.allocator); + var block = try sealed.readBlock(testing.allocator, 0); + defer block.deinit(testing.allocator); + try testing.expectEqual(@as(usize, 2), block.events.len); + try testing.expectEqual(segment.Kind.sync, block.events[0].kind); + try testing.expectEqual(segment.Kind.create_resync, block.events[1].kind); + try testing.expectEqual(@as(u64, 1), block.events[0].seq); + try testing.expectEqual(@as(u64, 2), block.events[1].seq); +} diff --git a/src/internal/verify.zig b/src/internal/verify.zig index 6ff1a0a..b2789b9 100644 --- a/src/internal/verify.zig +++ b/src/internal/verify.zig @@ -24,6 +24,8 @@ const repos = @import("bootstrap/repos.zig"); const Io = std.Io; const Allocator = std.mem.Allocator; const log = std.log.scoped(.stream); +const max_commit_blocks_bytes = 2_000_000; +const max_firehose_message_bytes = 2 * 1024 * 1024; const CachedKey = struct { key_type: zat.multicodec.KeyType, @@ -37,14 +39,15 @@ pub const Outcome = enum { replay, /// signature failed against a freshly resolved key — drop the event invalid_signature, - /// could not resolve the DID — reject (full parity will resync/retry) + /// could not resolve the DID — reject; authoritative repair cannot be + /// authenticated without the same identity material unverified, /// input-integrity failure which an authoritative fetch cannot repair: /// future rev, size violation, or outer/inner field mismatch invalid_commit, /// Chain state cannot be advanced from this event alone. Upstream - /// performs authoritative getRepo here; Stream rejects until that path - /// is wired rather than accepting a guessed state. + /// performs authoritative getRepo here; Stream's coordinator consumes + /// this outcome and withholds the triggering event from the archive. needs_resync, }; @@ -217,7 +220,7 @@ pub const Verifier = struct { const now_us = Io.Timestamp.now(self.io, .real).toMicroseconds(); if (rev.timestamp() > @as(u64, @intCast(@max(now_us, 0))) + 5 * 60 * std.time.us_per_s) return self.rejectInvalid(); - if (commit.blocks.len > 2 * 1024 * 1024 or commit.ops.len > 200) + if (commit.blocks.len > max_commit_blocks_bytes or commit.ops.len > 200) return self.rejectInvalid(); // key lookup/resolution under the lock; signature crypto outside it @@ -230,12 +233,12 @@ pub const Verifier = struct { if (commit.prev_data) |claimed| { if (!std.mem.eql(u8, claimed.raw, state.dataSlice())) return self.needsResync(); - } else { + } else if (isLegacyCommit(commit)) { // Atmos defaults to LegacyAccept: Sync-1.0-shaped commits // cannot prove continuity, but still pass signature and // post-state op-CID verification and advance the chain. legacy = true; - } + } else return self.needsResync(); } const key = self.lockedGetKey(resolver, commit.repo, false) orelse { @@ -374,7 +377,7 @@ pub const Verifier = struct { const result = zat.verifyCommitCar(allocator, sync_event.blocks, public_key, .{ .verify_mst = false, .expected_did = sync_event.did, - .max_car_size = 2 * 1024 * 1024, + .max_car_size = max_firehose_message_bytes, }) catch return false; return std.mem.eql(u8, result.commit_rev, sync_event.rev); } @@ -462,7 +465,7 @@ pub const Verifier = struct { public_key, .{ .expected_did = null, - .max_car_size = 2 * 1024 * 1024, + .max_car_size = max_commit_blocks_bytes, }, ) catch |err| switch (err) { // Match upstream's default LenientInversion carve-out narrowly: @@ -479,7 +482,7 @@ pub const Verifier = struct { .{ .expected_did = null, .skip_inversion = true, - .max_car_size = 2 * 1024 * 1024, + .max_car_size = max_commit_blocks_bytes, }, ) else @@ -609,6 +612,15 @@ pub const Verifier = struct { } }; +fn isLegacyCommit(commit: zat.firehose.CommitEvent) bool { + if (commit.prev_data != null) return false; + for (commit.ops) |op| switch (op.action) { + .update, .delete => if (op.prev != null) return false, + .create => {}, + }; + return true; +} + test "durable verifier chain reloads from metadata store" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); @@ -697,6 +709,42 @@ test "op CID proof rejects an envelope that disagrees with post-state MST" { try testing.expectEqual(CommitFailureClass.resync, classifyCommitFailure(error.OpCidMismatch)); } +test "commit block limit is upstream decimal two million bytes" { + const testing = std.testing; + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var path_buf: [std.fs.max_path_bytes]u8 = undefined; + const path_len = try tmp.dir.realPath(threaded.io(), &path_buf); + var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]); + defer meta.deinit(); + var verifier = Verifier.init(testing.allocator, threaded.io(), "http://offline.invalid", &meta); + defer verifier.deinit(); + const keypair = try zat.Keypair.fromSecretKey(.p256, .{29} ** 32); + const public_key = try keypair.publicKey(); + var cached: CachedKey = .{ .key_type = .p256, .raw = undefined, .len = @intCast(public_key.len) }; + @memcpy(cached.raw[0..public_key.len], &public_key); + try verifier.cache.put("did:plc:size-boundary", cached); + const blocks = try testing.allocator.alloc(u8, max_commit_blocks_bytes + 1); + defer testing.allocator.free(blocks); + @memset(blocks, 0); + const base: zat.firehose.CommitEvent = .{ + .seq = 1, + .repo = "did:plc:size-boundary", + .rev = "3k2abcdefghij", + .time = "2026-07-16T00:00:00Z", + .blocks = blocks[0..max_commit_blocks_bytes], + .ops = &.{}, + }; + const at_limit = try verifier.verifyCommit(undefined, base); + try testing.expectEqual(Outcome.needs_resync, at_limit.outcome); + var over = base; + over.blocks = blocks; + const rejected = try verifier.verifyCommit(undefined, over); + try testing.expectEqual(Outcome.invalid_commit, rejected.outcome); +} + test "signed chain failures follow upstream repair policy and legacy default" { const testing = std.testing; var arena = std.heap.ArenaAllocator.init(testing.allocator); diff --git a/src/main.zig b/src/main.zig index f1272f1..f3704d5 100644 --- a/src/main.zig +++ b/src/main.zig @@ -15,6 +15,7 @@ const xrpcapi = @import("internal/xrpcapi.zig"); const metrics = @import("internal/metrics.zig"); const meta_store = @import("internal/meta_store.zig"); const pipeline_mod = @import("internal/pipeline.zig"); +const repair_mod = @import("internal/repair.zig"); const server_mod = @import("internal/server.zig"); const steady_mod = @import("internal/compact/steady.zig"); const tail_mod = @import("internal/tail.zig"); @@ -251,7 +252,23 @@ pub fn main(init: std.process.Init.Minimal) !void { .stats = &stats, .emit_ctx = &consumer, .emit = ingest.Consumer.emitVerified, + .emit_repair = ingest.Consumer.emitRepair, }; + var repair_coordinator: ?repair_mod.Coordinator = if (verifier) |*v| .{ + .allocator = allocator, + .io = io, + .verifier = v, + .relay_http = relay_http, + .scratch_dir = repo_scratch, + .stats = &stats, + .publish_ctx = &pipe, + .publish = pipeline_mod.Pipeline.publishRepair, + } else null; + if (repair_coordinator) |*coordinator| { + try coordinator.start(); + pipe.repair = coordinator; + } + defer if (repair_coordinator) |*coordinator| coordinator.deinit(); try pipe.start(); defer pipe.stop(); consumer.pipeline = &pipe; diff --git a/src/tests.zig b/src/tests.zig index 154be5b..35a8e91 100644 --- a/src/tests.zig +++ b/src/tests.zig @@ -5,6 +5,8 @@ comptime { _ = @import("internal/meta_store.zig"); _ = @import("internal/metrics.zig"); _ = @import("internal/pipeline.zig"); + _ = @import("internal/repair.zig"); + _ = @import("internal/repair_integration_test.zig"); _ = @import("internal/segment.zig"); _ = @import("internal/zstd.zig"); _ = @import("internal/xxhash128.zig"); -- 2.51.2