From 76f7eb03cb1d4ccaf4a45534a32c6056296ceb90 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Tue, 21 Jul 2026 15:13:47 -0500 Subject: [PATCH] bound live repair wire memory --- docs/semantic-parity.md | 2 +- docs/upstream-harness.md | 1 + src/internal/ingest.zig | 11 ++- src/internal/repair_integration_test.zig | 92 ++++++++++++++++++++++++ 4 files changed, 103 insertions(+), 3 deletions(-) diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index 3e43a36..eb80920 100644 --- a/docs/semantic-parity.md +++ b/docs/semantic-parity.md @@ -17,7 +17,7 @@ this document or `bootstrap-semantic-parity.md` is open. | Archive XRPC | Official Go client plus Stream conformance replay over listSegments/getSegment/getBlock/planBackfill | **closed.** With the pinned Zig dependencies cached, `just archive-contract` seeds three real sealed JSS segments and runs without network access. The pinned Go client consumed 5,000 ordered whole-segment rows, the exact 2,500-row DID block plan, 50 required identity sentinels for an empty collection match, the 234-row `(1000,1234]` bound, and a duplicate-free archive-to-live cutover at cursor 5000. Direct conformance also proves checksums/ETags, byte-identical segments, raw block decode, error names, DID/collection blooms, plan modes, and HTTP ranges. The planner uses refcounted manifest-resident headers, block envelopes, DID blooms, and collection summaries; the offline contract hides every JSS file after startup and still requires an exact plan, proving the request path performs no segment-file I/O. The harness rejects any upstream checkout not at the recorded pin. | | Subscribe v1/v2 | Wire/filter/cursor/compression unit and local e2e coverage | Cursor parsing/resolution precedes upgrade; both endpoints use the seq/time-us magnitude split, v1 clamps below-floor seqs, and v2 rejects them. Timestamp translation uses bounded positional reads of real sealed JSS headers, block indexes, and one selected block before upgrade; disk/decode/index faults are retryable generic 503s, logged internally and counted as `resolve_failed`. Hot/cold scan, skip, encode, options-update, oversize-parser, clean peer/server close, cursor-resolution, and sustained adversarial-rate boundaries feed canonical metrics, with deterministic offline tests derived from upstream. Subscriber transport sends a ping every 30 seconds, applies a kernel-enforced five-second deadline to every frame write, and sends close code 1001 before intentional server shutdown while keeping transport failure out of clean-disconnect accounting. A failed delivery, cold read, ping, or adversarial-rate check interrupts the server reader and removes the connection instead of leaving a ping-only zombie. V1 now negotiates genuine RFC 7692 deflate with the upstream 128-byte threshold and 32 KiB context takeover; v1 custom zstd and all v2 connections explicitly decline it. V2 dictionary negotiation, exact pinned dictionary download/validators, and v2 binary delivery remain independent of the frozen v1 dictionary contract; pinned official Go clients have decoded both compression families offline. | | Sync 1.1 live verification | Upstream requires durable per-DID chain/hosting state, MST inversion, op-CID consistency, rev replay/future guards, default acceptance of legacy-shaped commits, transparent whole-repo repair, and Atmos delivery scheduling | **closed at the implementation boundary.** Diff-CAR/MST inversion (including upstream's narrow default lenient carve-out), post-state op-CID proof, signed inner/outer consistency, exact decimal size/future-rev/replay gates, exact legacy-shape detection with default `LegacyAccept`, two-phase durable chain/hosting state, and account/identity replay ratchets are implemented. Recoverable decode/inversion/duplicate-path/op-CID/chain failures route to repair; signature failures and outer/inner producer-integrity mismatches bypass it, and no failed event is archived. `prepareRepo` retains and authenticates the canonical signed complete head without a second multi-gigabyte parse/walk. The live path now uses Atmos's 32 worker slots, one worker-held FIFO chain per DID, a 64-pending drop-oldest boundary, completion-order batches of 50/500 ms across DIDs, and `min(inflight)-1` cursor watermarks. Dropped and silent events leave the inflight set without inventing a delivery batch; cursor advancement remains coupled to durable archive boundaries. Offline receipts use real encoded frames, real RocksDB verifier state, and actual worker blocking rather than fixture verdicts or scheduler mocks. See `live-scheduler.md`. | -| Sync 1.1 resync ordering | Upstream serializes per DID, drops stale async repairs, and emits a sync tombstone plus authoritative replacements | **closed at the implementation boundary.** The live coordinator has 32 real fetch workers, a 64-job queue, a five-minute fetch budget plus the 1,000 B/s-for-30-seconds slow-transfer guard, a bounded 16,384-entry 5/minute-per-DID limiter, 2,048 pending commits with drop-oldest overflow, fetch outside the DID lane, authenticated apply with older/equal-contradictory head rejection, ordered pending replay, and trigger-ticket-gated outbox delivery. Commit divergence schedules async repair; `#sync` divergence attempts inline repair and queues one retry only for transient failure, preserving the original relay cursor only on inline success. Matching Atmos's gate ownership, a divergent `#sync` repairs before envelope/inner validation; an invalid envelope rev then durably advances only the authenticated chain and drops the original tombstone plus every replacement row as one `live/invalid_rev` event. Valid repairs emit `sync` then bounded 1,024-row `create_resync` batches and stage fetched/pending chain checkpoints under the archive lock before any matching fsync. The shared hot tail is an amortized physical deque: mass replacement eviction cannot shift the full 256 MiB log once per row, and a 20,000-entry burst regression locks that boundary. 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. | +| Sync 1.1 resync ordering | Upstream serializes per DID, drops stale async repairs, and emits a sync tombstone plus authoritative replacements | **closed at the implementation boundary.** The live coordinator has 32 real fetch workers, a 64-job queue, a five-minute fetch budget plus the 1,000 B/s-for-30-seconds slow-transfer guard, a bounded 16,384-entry 5/minute-per-DID limiter, 2,048 pending commits with drop-oldest overflow, fetch outside the DID lane, authenticated apply with older/equal-contradictory head rejection, ordered pending replay, and trigger-ticket-gated outbox delivery. Commit divergence schedules async repair; `#sync` divergence attempts inline repair and queues one retry only for transient failure, preserving the original relay cursor only on inline success. Matching Atmos's gate ownership, a divergent `#sync` repairs before envelope/inner validation; an invalid envelope rev then durably advances only the authenticated chain and drops the original tombstone plus every replacement row as one `live/invalid_rev` event. Valid repairs emit `sync` then bounded 1,024-row `create_resync` batches and stage fetched/pending chain checkpoints under the archive lock before any matching fsync. The shared hot tail is an amortized physical deque: mass replacement eviction cannot shift the full 256 MiB log once per row, and a 20,000-entry burst regression locks that boundary. Wire encoding uses a per-row scratch arena before the tail takes its durable copy, matching upstream's bounded append/publish lifetime instead of retaining v1 and v2 JSON for the entire fetched repository. A real 4,096-record MST repair now completes under a 1 MiB repair-scratch cap; the prior ownership model deterministically fails the same test with `ArchiveAppendFailed`. Loopback-only tests exercise a real HTTP getRepo mmap, real DID resolution/signature, buffering while the response is held open, slow-transfer cancellation, durable RocksDB promotion/cursor coupling, v2 tail publication, and sealed JSS row order. | | Timestamp import | Canonical dashboard and API include timestamp import lifecycle/rewrite metrics and behavior | **Closed at the implementation boundary.** The format preserves distinct `witnessed_at` and sentinel-zero `indexed_at` columns. Subscriber v1/v2 encoding applies upstream's `indexed_at != 0 ? indexed_at : witnessed_at` display rule while ranges and timestamp cursors remain anchored to immutable witness envelopes. The strict seekable RFC 4180 parser matches upstream's header, row rejection, byte-offset, CID, RFC3339, 64 KiB, and bounded-sampling contracts in a forward-only allocation-free pass. A separate Bloom-filtered RocksDB rule store bulk-loads sorted external SST chunks with CSV-order last-write-wins, reconstructs its resident collection gate on open, and implements specific-CID precedence with all-version fallback. Its central archive-lock hook stamps live, repair, bootstrap, and merge materializations before sequence assignment, and feeds the mutated rows to immediate delivery so hot and cold encodings agree. The sealed-JSS patch primitive preserves block topology/envelopes, clean frames, and the opaque bloom/collection footer while atomically changing only `indexed_at`; it is exercised against an upstream-produced segment. A concrete sealed-segment bloom catalog, bounded DID/FD bucketer, fsynced packed offsets, positioned revalidation, and collision-safe per-segment patch plan implement Phase B/C with specific-CID precedence and idempotent disk-to-disk coverage. A dedicated archive rewrite mutex serializes the real delete compactor with timestamp patching while leaving live append available. The unified metadata DB synchronously persists the current-job pointer, complete job record, and per-segment done set. The real runner activates rules before force-seal, durably crosses the bucketed handoff, checkpoints only after patch fsync/rename/dir-sync, retains partial cancellation progress, and resumes after full archive/rule/metadata reopen. The background manager canonicalizes symlinks, confines regular files, enforces one durable nonterminal job, and adopts it on startup. Bearer-gated import/status XRPC matches the pinned lexicons, fixed-width digest comparison, error names/statuses, steady-state admission, and current/by-id status behavior; a real loopback HTTP run exercised it against the local simulator. Every canonical import metric is produced at its real parse, route, patch, or durable-terminal boundary; cancellation remains a resumable pause and is not counted as terminal failure. The strict Linux campaign cuts all four patch write/fsync/rename/dir-sync boundaries and proves exact target and untargeted timestamps after daemon recovery and public V2 replay; it also exposed and closed a Linux `O_PATH` directory-fsync defect. | | Status/diagnostics | Repo attempts/error class/final PDS and per-host aggregates survive restart and are operator-queryable | **closed at the implementation boundary.** Durable v5 repo rows distinguish initial Backfill.Rev from latest Rev/UpdatedAt, retain resolved Handle/PDS and reserved record/byte fields, and atomically maintain `handle/`. Merge source cursor commits refresh latest revisions without mutating backfill watermarks. Same-batch `host/` totals, active/status counts, cumulative error classes, and five bounded recent samples survive restart; host moves and active flips are atomic with the source repo row. `/status?tab=hosts` reads those rows without a whole-network scan and matches upstream ordering and error presentation. Account queries preserve resolver-first handle semantics, the legacy `did`/`handle` aliases, and missing-identity hydration, then reconstruct records last-writer-wins from checksum-verified, DID-bloom-pruned sealed JSS plus rotation-safe active/pending and bootstrap-live snapshots. The resulting canonical MST is compared with the PDS commit root through real Sync 1.1 `getLatestCommit`/`getBlocks`; the page renders the same match fields and error state as upstream. `HEAD` performs lookup without verification; successful status responses are `no-store` and carry a generation timestamp. Expensive verification uses upstream's exact 4/source-IP/minute fixed-window limiter, 4,096-entry bound, stale pruning, oldest eviction, and explicit disable option. The offline receipt uses a real HTTP Stream route, real RocksDB row, real JSS archive, signed commit, and loopback PLC/PDS; four separate connections verify successfully, a fifth changing only its ephemeral port is blocked without another upstream request, HEAD remains side-effect-free, and disabling the limiter performs a new authoritative verification. | | Prometheus/Grafana | Every canonical dashboard query maps to a real, same-semantics producer; runtime-specific panels use honest Zig/process metrics | **closed.** The exact checksum-pinned upstream dashboard is vendored with an offline fail-closed scrape contract. Canonical build, live-firehose, archive, verifier, subscribe, backfill, timestamp-import, and compaction families all have same-semantics producers. Compaction measures pass success/failure/duration, early triggers, live tombstone size, folded reasons, worker examination/rewrite/clean outcomes, dropped-row reasons, bytes, manifest heals, durable watermark, and witnessed-time lag at the corresponding real boundaries. Real JSS tests cover bloom avoidance, bounded concurrent rewrites, malformed account failure, reason accounting, manifest-before-watermark ordering, and durable eviction. The ReleaseSafe production receipt reports `required=104 present=104 missing=0`; the runtime-only dashboard row remains an intentional checksum-pinned Zig/process adaptation with no synthetic `go_*` aliases. See `grafana-dashboard.md`. | diff --git a/docs/upstream-harness.md b/docs/upstream-harness.md index ba5fd6d..8188fa6 100644 --- a/docs/upstream-harness.md +++ b/docs/upstream-harness.md @@ -30,6 +30,7 @@ driver boots real server against simulator, walks lifecycle gated on durable-app - field limits (upstream columnar format): did ≤65535, collection/rkey/rev ≤255, payload ≤u32 - invariants: fsync data before committing cursor; seq assigned at append under writer mutex, starts at 1 (0 = sentinel); per-DID order preserved; never crash on upstream data / crash loud on own corruption - readable log (hot tail): writer-owned deque, deep-copied entries at seq allocation, 256MiB budget, evicts only below durable watermark (pinned above), notify channel per append, encode-once wire memo per entry shared across fan-out; cursor below floor → cold reader over sealed segments. Stream's tail enforces the same pin: an undurable suffix is retained even when it overruns the byte budget, `publishDurable` releases only the newly durable byte suffix before eviction, and scrape-time readable/pinned/overrun gauges are O(1) accumulators rather than a scan under the hot lock. Its physical deque retires prefix entries in O(1) and compacts only after a large amortized prefix; the 20,000-row burst regression catches the former `orderedRemove(0)` implementation that made a real 8,192-row repair quadratic and stalled global ingest. +- resync memory lifetime: upstream appends each converted replacement and immediately forwards it to the bounded readable log; it does not retain a second whole-repository wire representation. Stream batches physical archive writes in groups of 1,024 but releases the complete v1/v2 encoder arena after each row is copied into the tail. The offline 4,096-record MST receipt is capped at 1 MiB of repair scratch and mutation-checks the former repair-lifetime JSON ownership by reproducing its exact `ArchiveAppendFailed`. - archive health: free space is collected from the filesystem containing the open archive directory on every scrape; segment-seal latency starts before the pending-block flush and is observed only after footer write/fsync plus finalized-header pwrite/fsync succeeds, using upstream's `0.01 × 2^n` bucket boundaries. ## zig oracle v1 (2026-07-12) diff --git a/src/internal/ingest.zig b/src/internal/ingest.zig index c31c9ca..b95f620 100644 --- a/src/internal/ingest.zig +++ b/src/internal/ingest.zig @@ -233,8 +233,15 @@ pub const Consumer = struct { 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 ""; + // Tail.append deep-copies both wire encodings. Allocate + // their complete CBOR/CID scratch graph in a per-row + // arena; the repair-lifetime arena must not retain two + // whole-repo JSON copies on top of the CAR and MST. + var wire_arena = std.heap.ArenaAllocator.init(sink.consumer.allocator); + defer wire_arena.deinit(); + const wire_alloc = wire_arena.allocator(); + const json = (try cold.encodeEvent(wire_alloc, stamped)) orelse ""; + const json_v2 = (try cold.encodeEventV2(wire_alloc, stamped)) orelse ""; const kind: @import("wire.zig").Kind = switch (row.kind) { .create, .update, .delete, .create_resync => .commit, .identity => .identity, diff --git a/src/internal/repair_integration_test.zig b/src/internal/repair_integration_test.zig index 6fab659..b6cee54 100644 --- a/src/internal/repair_integration_test.zig +++ b/src/internal/repair_integration_test.zig @@ -117,3 +117,95 @@ test "repair completion durably emits tombstone then replacement offline" { try testing.expectEqual(@as(u64, 1), block.events[0].seq); try testing.expectEqual(@as(u64, 2), block.events[1].seq); } + +test "large repair keeps wire encoding scratch bounded" { + 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(); + // Make every 1,024-row batch durable so the tiny tail can retire it. The + // cap below therefore measures repair-owned scratch, not the configured + // subscriber retention window. + archive.flush_interval_us = 0; + var tail = tail_mod.Tail.init(testing.allocator, io, 64 * 1024); + defer tail.deinit(); + tail.enableDurabilityGate(0); + var cursor = cursor_mod.Store.init(&meta); + + var repair_alloc = std.heap.DebugAllocator(.{ .enable_memory_limit = true }){}; + defer testing.expect(repair_alloc.deinit() == .ok) catch @panic("repair allocator leak"); + repair_alloc.requested_memory_limit = 1 * 1024 * 1024; + var consumer: ingest.Consumer = .{ + .allocator = repair_alloc.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:large-durable-repair"; + const body = [_]u8{'x'} ** 512; + const record = try zat.cbor.encodeAlloc(alloc, .{ .map = &.{ + .{ .key = "$type", .value = .{ .text = "app.bsky.feed.post" } }, + .{ .key = "text", .value = .{ .text = &body } }, + } }); + const record_cid = try zat.cbor.Cid.forDagCbor(alloc, record); + var tree = zat.mst.Mst.init(alloc); + for (0..4096) |i| { + const key = try std.fmt.allocPrint(alloc, "app.bsky.feed.post/r{d}", .{i}); + try tree.put(key, record_cid); + } + const data_cid = try tree.rootCid(); + const keypair = try zat.Keypair.fromSecretKey(.p256, .{29} ** 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 = 4096, + }; + + try ingest.Consumer.emitRepair(&consumer, &completion); + try testing.expectEqual(@as(u64, 4097), archive.committed_seq.load(.acquire)); + try testing.expectEqual(@as(?i64, 4096), cursor.load(io)); + try testing.expect(repair_alloc.total_requested_bytes < 256 * 1024); +} -- 2.51.2