From 654df48d5dc485536aee66edac678b88b69465d6 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Fri, 17 Jul 2026 16:49:57 -0500 Subject: [PATCH] persist upstream account diagnostics state --- README.md | 12 + build.zig | 5 + docs/bootstrap-semantic-parity.md | 6 +- docs/semantic-parity.md | 2 +- docs/upstream-bootstrap-spec.md | 16 +- src/internal/bootstrap/engine.zig | 469 +++++++++++++++++++++++--- src/internal/bootstrap/lifecycle.zig | 14 + src/internal/bootstrap/merge.zig | 35 +- src/internal/bootstrap/repo_store.zig | 376 ++++++++++++++++++++- src/internal/bootstrap/resync.zig | 83 ++++- src/internal/bootstrap/retry.zig | 7 + src/main.zig | 41 ++- src/tests.zig | 3 + 13 files changed, 980 insertions(+), 89 deletions(-) diff --git a/README.md b/README.md index 128b971..bc39be4 100644 --- a/README.md +++ b/README.md @@ -83,6 +83,18 @@ recent-error content in the browser. This implementation is not admission to another full-network run by itself: `docs/semantic-parity.md` remains authoritative. +The same durable account row retains the distinct initial-backfill and latest +known revisions, update time, declared handle, PDS endpoint, and reserved +record/byte counters. Matching upstream, the debug-only explicit +`--backfill-repos=did:plc:example[,did:web:example.com]` path processes the +requested DIDs serially, resolves their real DID documents, and removes stale +full-network merge-discovery state. Ordinary whole-network `listRepos` +bootstrap does not add millions of extra identity requests. +`handle/` is maintained atomically with every repo transition, and +each merge-source commit advances its cursor together with latest-revision +refreshes. These are the storage prerequisites for the Account diagnostics +view, not a claim that its reconstruction/verification surface is complete. + Timestamp import is disabled to remote callers unless a non-empty bearer token is configured. Stage plain CSVs under the confined import directory (default `/imports`), submit a relative path, then poll the returned job id: diff --git a/build.zig b/build.zig index f6648e8..77b4669 100644 --- a/build.zig +++ b/build.zig @@ -147,8 +147,13 @@ pub fn build(b: *std.Build) void { test_mod.link_libc = true; test_mod.link_libcpp = true; linkVendoredC(test_mod, b, target, optimize); + const test_filters: []const []const u8 = if (b.option([]const u8, "test-filter", "Only compile tests whose name contains this text")) |filter| + b.allocator.dupe([]const u8, &.{filter}) catch @panic("out of memory") + else + &.{}; const t = b.addTest(.{ .root_module = test_mod, + .filters = test_filters, }); test_step.dependOn(&b.addRunArtifact(t).step); } diff --git a/docs/bootstrap-semantic-parity.md b/docs/bootstrap-semantic-parity.md index b66bfc5..32cfa82 100644 --- a/docs/bootstrap-semantic-parity.md +++ b/docs/bootstrap-semantic-parity.md @@ -15,12 +15,14 @@ every experiment. | Rate-limit retry | Twenty additional attempts; honor reset/Retry-After independently | independent counter; `RateLimit-Reset` precedence; 330s bootstrap clamp; the mixed redirected fixture proves a 429 does not consume the ordinary transient budget; steady retry persists/clamps the server deadline | closed | | Terminal errors | RepoNotFound → complete/no rows; deactivated/suspended/takendown → unavailable | exact structured-error classification tests and durable statuses | closed | | DID attribution | listRepos/handler DID is authoritative, never embedded commit DID | a complete, valid, signed CAR whose embedded DID differs from the listRepos DID emits every row under the listRepos DID | closed | -| Inactive discovery | persist not_started + active=false; do not dispatch; preserve status on active flips | RocksDB row encoding v4, backward v2/v3 decode, restart test | closed | +| Inactive discovery | persist not_started + active=false; do not dispatch; preserve status on active flips | RocksDB row encoding v5, backward v2/v3/v4 decode, restart test | closed | +| Explicit selected backfill and account identity metadata | `--backfill-repos` processes the caller's order serially, bypasses listRepos, cannot combine with max-repos, deletes stale merge-discovery state, resolves each selected DID even when its repo row is already complete, retains declared handle/PDS, and atomically maintains the normalized local handle index. Ordinary whole-network backfill does not perform this extra resolution. | Stream parses and deduplicates the explicit DID list, synthesizes one ordered selected page without touching listRepos, requires the configured real `DidResolver`, deletes the stale bootstrap discovery watermark, and never advances listRepos state. `repo/` v5 persists Handle/PDS with same-batch `handle/` transitions. A real loopback PLC + getRepo run fails on any listRepos request and proves terminal status plus index persistence; handle moves/collisions, rebuild, and restart reads are also exercised offline. | closed | +| Merge latest revision | For every kept rev-bearing row, refresh top-level Rev/UpdatedAt while preserving Backfill.Rev; commit those rows atomically with the per-source merge cursor | merge collects the last kept rev per DID and `commitMergeSource` stages cursor + repo refreshes in one synced RocksDB batch; missing repo rows are skipped rather than synthesized malformed | closed | | Cursor durability | advance after page data is durable; resume cursor drains empty; discovery cursor retains tail | separate final resume-cursor commit and regression test | closed | | Resource exhaustion | capacity failure is fatal, never a durable per-repo “too large” skip | OOM is lifecycle-fatal and tested | closed | | Resync ordering | validate replacement first, then append sync + replacement, then durable completion | `prepareRepo` precedes emission; sync plus replacement rows append in bounded 1,024-row batches, matching upstream. Memory does not require a second repository-sized event array; archive durability precedes RocksDB completion | closed | | Steady retry host semantics | persist actual post-redirect PDS host; 16 global / 4 per-host workers; 429 parks only that host | redirected `Request.uri` is normalized and persisted; worker gates and durable host-specific parks match upstream limits; a real redirected 429 fixture proves the final host replaces the predicted host and only the final host is parked | closed | -| Retry diagnostics | persist attempts, last error/class, host, retry count, next attempt, and durable per-host aggregates | RocksDB row v4 persists every repo field and backward-decodes v2/v3; success clears stale repo failure state. The same synced transition batch maintains normalized `host/` current counts plus cumulative error-class counts and five bounded recent samples; host moves remove the repo from the old bucket, successful retry retains diagnostic history, and a one-time migration rebuilds current aggregates without inventing history. The public host view reopens those durable rows through the production binary and exercises both upstream sort modes plus restart, text/HTML, filtering, and escaping offline. | closed | +| Retry diagnostics | persist attempts, last error/class, host, retry count, next attempt, and durable per-host aggregates | RocksDB row v5 persists every repo field and backward-decodes v2/v3/v4; retry, terminal resync, active-flip, and crash-recovery RMW paths preserve identity/latest fields, while success clears stale repo failure state. The same synced transition batch maintains normalized `host/` current counts plus cumulative error-class counts and five bounded recent samples; host moves remove the repo from the old bucket, successful retry retains diagnostic history, and a one-time migration rebuilds current aggregates without inventing history. The public host view reopens those durable rows through the production binary and exercises both upstream sort modes plus restart, text/HTML, filtering, and escaping offline. | closed | | Deadline coverage | getRepo streaming read + CAR load/completeness check, excluding handler/durability | same `prepared_fetch.zig` future covers both and returns a prepared repo | closed | | Oracle/adversity suite | truncation, hanging server, terminal errors, crash/restart, active flips | focused offline fixtures cover truncation, hanging fetches, terminal errors, retry/429 interaction, final-host parking, and hostile embedded DIDs; the fresh offline crash matrix survived all 8 lifecycle seams and rejects shutdown OOM/fatal-handler evidence | closed; rerun against the exact candidate before deployment | diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index b509658..818eb61 100644 --- a/docs/semantic-parity.md +++ b/docs/semantic-parity.md @@ -19,7 +19,7 @@ this document or `bootstrap-semantic-parity.md` is open. | 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. 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 | **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. | -| Status/diagnostics | Repo attempts/error class/final PDS and per-host aggregates survive restart and are operator-queryable | Durable repo fields and same-batch `host/` aggregates are implemented. 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 the durable host rows without a whole-network repo scan, matches upstream's largest/failing ordering and bounded recent-error presentation, supports browser filtering, and has an offline real-RocksDB/restart HTTP contract. **The account lookup, identity fallback, and automatic reconstructed-repo verification surface remain open.** | +| Status/diagnostics | Repo attempts/error class/final PDS and per-host aggregates survive restart and are operator-queryable | Durable v5 repo rows now 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 the durable host rows without a whole-network repo scan, matches upstream's largest/failing ordering and bounded recent-error presentation, supports browser filtering, and has an offline real-RocksDB/restart HTTP contract. **The account lookup, identity fallback, and automatic reconstructed-repo verification surface remain open.** | | Prometheus/Grafana | Every canonical dashboard query maps to a real, same-semantics producer; runtime-specific panels use honest Zig/process metrics | exact checksum-pinned upstream dashboard is vendored with an offline fail-closed scrape contract; canonical build/live-firehose/archive-append/drop/verifier/backfill/timestamp-import producers are present. The complete canonical subscriber family set is exported from real scan, delivery, rejection, cursor, and disconnect boundaries; `none`, `deflate`, and `zstd` labels now arise only from the actual negotiated delivery path. Live frame inspection distinguishes gaps, bounded-label server errors, forward-compatible unknowns, and malformed CBOR at the raw transport boundary; a Zat reconnect callback measures attempts rather than connection failures. **Remaining storage/API/orchestrator families and Go-row adaptation open.** | | Oracle | Event-log equivalence, final-state convergence, crash/power-loss, verifier repair, hostile input, and anti-vacuity receipts | **open — current crash matrix is necessary but substantially smaller than upstream's oracle** | diff --git a/docs/upstream-bootstrap-spec.md b/docs/upstream-bootstrap-spec.md index 4a0e250..0f5e36c 100644 --- a/docs/upstream-bootstrap-spec.md +++ b/docs/upstream-bootstrap-spec.md @@ -34,12 +34,20 @@ fsync data before pebble commit, always; fsync dir before deleting durable state fails (retryable); bad NSID/rkey path → drop record only; field-too-long → drop w/ own counter; missing block → repo fails. - per-DID status rows: not_started|pending|complete|failed|unavailable + - Backfill.Rev watermark; completion committed in the same durable batch as - segment data. interrupted-bootstrap rows (pre-existing not_started, not + Backfill.Rev watermark. Selected backfill resolves the DID document and + stores declared Handle, PDS endpoint, and normalized Host; handle index + changes commit atomically with the repo row. Completion sets both immutable + Backfill.Rev and top-level latest Rev. Completion is committed after segment + data. interrupted-bootstrap rows (pre-existing not_started, not discovered this run) defer to pending — merge repairs after live drain (re-downloading now would put low seqs under stale tombstones). -- --max-backfill-repos: debug, selected set, no durable cursor advance, - auto-skip merge discovery. +- --max-backfill-repos: debug truncation of listRepos, no durable cursor + advance, auto-skip merge discovery. +- --backfill-repos: debug explicit DID set, mutually exclusive with max-repos; + process the caller's order serially, bypass listRepos, delete stale merge + discovery state, resolve Handle/PDS metadata even for already-complete rows, + maintain the local handle index, and auto-skip merge discovery. Ordinary + whole-network listRepos backfill does not perform per-DID identity resolution. ## live capture during bootstrap second live consumer → backfill/live_segments/ with THROWAWAY seq space diff --git a/src/internal/bootstrap/engine.zig b/src/internal/bootstrap/engine.zig index b342979..bf4b1b4 100644 --- a/src/internal/bootstrap/engine.zig +++ b/src/internal/bootstrap/engine.zig @@ -43,7 +43,13 @@ pub const default_max_inflight_bytes: usize = 8 * 1024 * 1024 * 1024; pub const Config = struct { relay_http: []const u8, + /// Identity resolver used to persist upstream's account diagnostics + /// metadata. Null is reserved for focused offline unit tests. + plc_url: ?[]const u8 = null, data_dir: []const u8, + /// Upstream's debug-only explicit-DID mode. It bypasses listRepos, + /// resolves identity metadata, and skips merge discovery. + selected_repos: []const []const u8 = &.{}, /// 0 = unlimited (production); >0 = debug partial crawl max_repos: usize = 0, workers: usize = default_workers, @@ -64,16 +70,96 @@ const Result = struct { did: []const u8, status: repo_store.Status, rev: []u8 = &.{}, + latest_rev: []u8 = &.{}, host: []u8 = &.{}, last_error: []u8 = &.{}, error_class: repo_store.ErrorClass = .unknown, attempts: u32 = 0, - last_attempt_us: i64 = 0, started_us: i64 = 0, rows: u64 = 0, drops: repos.DropCounts = .{}, }; +const IdentityMetadata = struct { + handle: []u8 = &.{}, + pds: []u8 = &.{}, + + fn deinit(self: *IdentityMetadata, allocator: Allocator) void { + if (self.handle.len > 0) allocator.free(self.handle); + if (self.pds.len > 0) allocator.free(self.pds); + self.* = .{}; + } +}; + +fn resolveIdentityMetadata(allocator: Allocator, resolver: *zat.DidResolver, raw_did: []const u8) !?IdentityMetadata { + const did = zat.Did.parse(raw_did) orelse return null; + var doc = resolver.resolve(did) catch |err| { + log.warn("backfill identity resolution {s} failed: {s}", .{ raw_did, @errorName(err) }); + return null; + }; + defer doc.deinit(); + + var out: IdentityMetadata = .{}; + errdefer out.deinit(allocator); + if (doc.handle()) |raw_handle| if (zat.Handle.parse(raw_handle) != null) { + out.handle = try allocator.dupe(u8, raw_handle); + for (out.handle) |*byte| byte.* = std.ascii.toLower(byte.*); + }; + if (doc.pdsEndpoint()) |pds| out.pds = try allocator.dupe(u8, pds); + return out; +} + +fn selectedReposPage(allocator: Allocator, dids: []const []const u8) !repos.ListReposPage { + const refs = try allocator.alloc(repos.RepoRef, dids.len); + for (dids, refs) |did, *ref| ref.* = .{ .did = did, .rev = "", .active = true }; + return .{ .repos = refs, .cursor = null }; +} + +fn stateWithStatus(state: repo_store.RepoState, status: repo_store.Status) repo_store.RepoState { + var next = state; + next.status = status; + return next; +} + +fn terminalState(result: Result, previous: ?repo_store.RepoState, terminal_us: i64) repo_store.RepoState { + const downloaded = result.status == .complete and result.rev.len > 0; + return .{ + .status = result.status, + .active = if (previous) |state| state.active else true, + .rev = if (result.rev.len > 0) result.rev else if (previous) |state| state.rev else "", + .latest_rev = if (result.latest_rev.len > 0) result.latest_rev else if (previous) |state| state.latest_rev else "", + .host = if (result.host.len > 0) result.host else if (previous) |state| state.host else "", + .handle = if (previous) |state| state.handle else "", + .pds = if (previous) |state| state.pds else "", + .last_error = if (result.status == .failed) result.last_error else "", + .error_class = if (result.status == .failed) result.error_class else .unknown, + .attempts = if (result.status == .failed) result.attempts else 0, + .last_attempt_us = terminal_us, + .started_us = if (previous) |state| if (state.started_us != 0) state.started_us else result.started_us else result.started_us, + .completed_us = if (result.status == .complete) terminal_us else if (previous) |state| state.completed_us else 0, + .retry_count = if (result.status == .failed) if (previous) |state| state.retry_count else 0 else 0, + .next_attempt_us = if (result.status == .failed) if (previous) |state| state.next_attempt_us else 0 else 0, + .updated_us = if (downloaded) terminal_us else if (previous) |state| state.updated_us else 0, + .record_count = if (previous) |state| state.record_count else 0, + .total_bytes = if (previous) |state| state.total_bytes else 0, + }; +} + +fn resolveAndRecordSelectedIdentity(allocator: Allocator, io: Io, plc_url: []const u8, store: *repo_store.Store, did: []const u8, state: repo_store.RepoState) !void { + var resolver = zat.DidResolver.init(io, allocator); + defer resolver.deinit(); + resolver.plc_url = plc_url; + var identity = (try resolveIdentityMetadata(allocator, &resolver, did)) orelse return; + defer identity.deinit(allocator); + const host = try repo_store.hostFromPDS(allocator, identity.pds); + defer if (host.len > 0) allocator.free(host); + var next = state; + next.handle = identity.handle; + next.pds = identity.pds; + next.host = host; + try store.put(did, next); +} + const JobQueue = struct { buffer: []Job, head: usize = 0, @@ -233,7 +319,6 @@ const Work = struct { .limit = self.config.max_inflight_bytes, }; const allocator = bounded.allocator(); - while (self.queue.pop(self.io)) |job| { self.publishGauges(); if (self.getFatal() != null) { @@ -313,6 +398,8 @@ pub fn run( archive: *archive_mod.Archive, store: *repo_store.Store, ) !Stats { + if (config.selected_repos.len > 0 and config.max_repos > 0) return error.ConflictingBackfillSelection; + if (config.selected_repos.len > 0 and config.plc_url == null) return error.SelectedBackfillRequiresIdentityResolver; var scratch_buf: [Io.Dir.max_path_bytes]u8 = undefined; const scratch_dir = try std.fmt.bufPrint(&scratch_buf, "{s}/backfill/repo-scratch", .{config.data_dir}); try fetched_car.resetScratch(io, scratch_dir); @@ -320,14 +407,25 @@ pub fn run( var list_client = zat.XrpcClient.init(io, allocator, config.relay_http); defer list_client.deinit(); - var work = try Work.init(allocator, io, config, archive); + // Upstream's explicit-DID runner is intentionally serial and preserves + // the caller's order. It is a diagnostic/reconciliation path, not the + // parallel whole-network engine. + var work_config = config; + if (config.selected_repos.len > 0) work_config.workers = 1; + var work = try Work.init(allocator, io, work_config, archive); defer work.deinit(); try work.start(); defer work.shutdown(); var stats: Stats = .{}; var selected: usize = 0; - const cursor_owned: ?[]u8 = if (config.max_repos == 0) try loadListReposCursor(store.meta, allocator) else null; + // An explicit selection is a self-contained debug run. It must not leave + // an old full-network discovery watermark for merge to consume later. + if (config.selected_repos.len > 0) try store.meta.deleteDurable(bootstrap_cursor_key); + const cursor_owned: ?[]u8 = if (config.max_repos == 0 and config.selected_repos.len == 0) + try loadListReposCursor(store.meta, allocator) + else + null; defer if (cursor_owned) |owned| allocator.free(owned); var cursor: ?[]const u8 = if (cursor_owned) |owned| owned else null; var cursor_buf: [512]u8 = undefined; @@ -337,7 +435,13 @@ pub fn run( defer arena.deinit(); const alloc = arena.allocator(); - const page = try listReposPage(alloc, &list_client, cursor); + const selected_page = config.selected_repos.len > 0 and cursor == null; + const page = if (selected_page) + try selectedReposPage(alloc, config.selected_repos) + else if (config.selected_repos.len > 0) + break + else + try listReposPage(alloc, &list_client, cursor); if (page.repos.len == 0) break; var jobs: std.ArrayList(Job) = .empty; var dispatched: std.ArrayList(repo_store.Transition) = .empty; @@ -345,14 +449,33 @@ pub fn run( for (page.repos) |ref| { if (config.max_repos > 0 and selected >= config.max_repos) break; - const existing = try store.get(allocator, ref.did); - defer if (existing) |state| store.freeState(allocator, state); + var existing = try store.get(alloc, ref.did); + if (selected_page) { + if (existing == null) { + try store.put(ref.did, .{ .status = .not_started, .active = true, .started_us = Io.Timestamp.now(io, .real).toMicroseconds() }); + existing = try store.get(alloc, ref.did); + } + // Upstream reconciles an inactive selected row before + // recording identity metadata, so a crash cannot briefly + // attribute the newly resolved host as inactive. + if (existing) |state| if (state.active != ref.active) { + var next = state; + next.active = ref.active; + try store.put(ref.did, next); + existing = try store.get(alloc, ref.did); + }; + try resolveAndRecordSelectedIdentity(alloc, io, config.plc_url.?, store, ref.did, existing.?); + existing = try store.get(alloc, ref.did); + } if (existing) |state| { if (state.active != ref.active) try store.put(ref.did, .{ .status = state.status, .active = ref.active, .rev = state.rev, + .latest_rev = state.latest_rev, .host = state.host, + .handle = state.handle, + .pds = state.pds, .last_error = state.last_error, .error_class = state.error_class, .attempts = state.attempts, @@ -361,6 +484,9 @@ pub fn run( .completed_us = state.completed_us, .retry_count = state.retry_count, .next_attempt_us = state.next_attempt_us, + .updated_us = state.updated_us, + .record_count = state.record_count, + .total_bytes = state.total_bytes, }); } if (!ref.active) { @@ -372,7 +498,7 @@ pub fn run( if (state.status != .failed and state.status != .not_started) continue; } try jobs.append(alloc, .{ .did = ref.did }); - try dispatched.append(alloc, .{ .did = ref.did, .state = .{ .status = .not_started } }); + try dispatched.append(alloc, .{ .did = ref.did, .state = if (existing) |state| stateWithStatus(state, .not_started) else .{ .status = .not_started } }); selected += 1; } @@ -383,8 +509,10 @@ pub fn run( if (jobs.items.len > 0 and crashpoint.wouldHit(.mid_backfill_download)) crashpoint.hit(.mid_backfill_download); - var prng = std.Random.DefaultPrng.init(@bitCast(Io.Timestamp.now(io, .awake).toMicroseconds())); - prng.random().shuffle(Job, jobs.items); + if (!selected_page) { + var prng = std.Random.DefaultPrng.init(@bitCast(Io.Timestamp.now(io, .awake).toMicroseconds())); + prng.random().shuffle(Job, jobs.items); + } for (jobs.items) |job| try work.submit(job); const target = work.submittedCount(); work.waitThrough(target); @@ -400,24 +528,17 @@ pub fn run( try archive.flushBlock(); const transitions = try alloc.alloc(repo_store.Transition, results.items.len); for (results.items, transitions) |result, *transition| { - transition.* = .{ .did = result.did, .state = .{ - .status = result.status, - .rev = result.rev, - .host = result.host, - .last_error = result.last_error, - .error_class = result.error_class, - .attempts = result.attempts, - .last_attempt_us = result.last_attempt_us, - .started_us = result.started_us, - .completed_us = if (result.status == .complete) Io.Timestamp.now(io, .real).toMicroseconds() else 0, - } }; - if (result.status == .complete) { + const existing = try store.get(alloc, result.did); + const terminal_us = Io.Timestamp.now(io, .real).toMicroseconds(); + const downloaded = result.status == .complete and result.rev.len > 0; + transition.* = .{ .did = result.did, .state = terminalState(result, existing, terminal_us) }; + if (downloaded) { stats.completed += 1; stats.rows += result.rows; stats.drops.invalid_path += result.drops.invalid_path; stats.drops.field_too_long += result.drops.field_too_long; if (config.metrics) |m| _ = m.backfill_completed_total.fetchAdd(1, .monotonic); - } else { + } else if (result.status == .failed) { stats.failed += 1; if (config.metrics) |m| _ = m.backfill_failed_total.fetchAdd(1, .monotonic); } @@ -434,11 +555,188 @@ pub fn run( try archive.flushBlock(); try store.flush(); - if (config.max_repos == 0) try markListReposComplete(store.meta); + if (config.max_repos == 0 and config.selected_repos.len == 0) try markListReposComplete(store.meta); work.publishGauges(); return stats; } +test "bootstrap identity metadata comes from a real DID resolution response" { + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; + defer listener.deinit(io); + const Fixture = struct { + fn serve(server: *Io.net.Server, task_io: Io) !void { + var stream = try server.accept(task_io); + defer stream.close(task_io); + var read_buf: [2048]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(); + try request.respond( + "{\"id\":\"did:plc:abcdefghijklmnopqrstuvwx\",\"alsoKnownAs\":[\"at://Alice.Test\"],\"verificationMethod\":[],\"service\":[{\"id\":\"#atproto_pds\",\"type\":\"AtprotoPersonalDataServer\",\"serviceEndpoint\":\"https://pds.example\"}]}", + .{ .status = .ok, .keep_alive = false, .extra_headers = &.{.{ .name = "content-type", .value = "application/json" }} }, + ); + } + }; + var server_future = try io.concurrent(Fixture.serve, .{ &listener, io }); + defer _ = server_future.cancel(io) catch {}; + var plc_buf: [96]u8 = undefined; + const plc_url = try std.fmt.bufPrint(&plc_buf, "http://127.0.0.1:{d}", .{listener.socket.address.getPort()}); + var resolver = zat.DidResolver.init(io, testing.allocator); + defer resolver.deinit(); + resolver.plc_url = plc_url; + + var identity = (try resolveIdentityMetadata(testing.allocator, &resolver, "did:plc:abcdefghijklmnopqrstuvwx")).?; + defer identity.deinit(testing.allocator); + try testing.expectEqualStrings("alice.test", identity.handle); + try testing.expectEqualStrings("https://pds.example", identity.pds); + try server_future.await(io); +} + +test "explicit selected backfill resolves identity, bypasses listRepos, and persists its index" { + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; + defer listener.deinit(io); + const Fixture = struct { + fn serve(server: *Io.net.Server, task_io: Io) !void { + for (0..2) |_| { + var stream = try server.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(); + if (std.mem.startsWith(u8, request.head.target, "/did:plc:")) { + try request.respond( + "{\"id\":\"did:plc:abcdefghijklmnopqrstuvwx\",\"alsoKnownAs\":[\"at://selected.test\"],\"verificationMethod\":[],\"service\":[{\"id\":\"#atproto_pds\",\"type\":\"AtprotoPersonalDataServer\",\"serviceEndpoint\":\"https://PDS.EXAMPLE:443\"}]}", + .{ .status = .ok, .keep_alive = false }, + ); + } else if (std.mem.startsWith(u8, request.head.target, "/xrpc/com.atproto.sync.getRepo?")) { + try request.respond("{\"error\":\"RepoNotFound\"}", .{ + .status = .not_found, + .keep_alive = false, + .extra_headers = &.{.{ .name = "content-type", .value = "application/json" }}, + }); + } else { + // A listRepos call would prove this is only API-shaped, + // rather than upstream's distinct selected-DID path. + return error.UnexpectedSelectedBackfillRequest; + } + } + } + }; + var server_future = try io.concurrent(Fixture.serve, .{ &listener, io }); + 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}", .{listener.socket.address.getPort()}); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const data_dir_len = try tmp.dir.realPath(io, &data_dir_buf); + const data_dir = data_dir_buf[0..data_dir_len]; + var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir); + defer archive.deinit(); + var meta = try meta_store.Store.open(testing.allocator, data_dir); + defer meta.deinit(); + var store = try repo_store.Store.init(testing.allocator, io, &meta); + defer store.deinit(); + try meta.putDurable(bootstrap_cursor_key, "stale-full-network-cursor"); + const selected = [_][]const u8{"did:plc:abcdefghijklmnopqrstuvwx"}; + const stats = try run(testing.allocator, io, .{ + .relay_http = url, + .plc_url = url, + .data_dir = data_dir, + .selected_repos = &selected, + .workers = 1, + .max_inflight_bytes = 1024 * 1024, + }, &archive, &store); + // RepoNotFound is terminal-complete state, but upstream does not count it + // as a downloaded repository completion. + try testing.expectEqual(@as(u64, 0), stats.completed); + const state = (try store.get(testing.allocator, selected[0])).?; + defer store.freeState(testing.allocator, state); + try testing.expectEqual(repo_store.Status.complete, state.status); + try testing.expectEqual(@as(u32, 0), state.attempts); + try testing.expect(state.completed_us > 0); + try testing.expectEqualStrings("selected.test", state.handle); + try testing.expectEqualStrings("https://PDS.EXAMPLE:443", state.pds); + const indexed = (try store.lookupDIDByHandle(testing.allocator, "SELECTED.TEST")).?; + defer testing.allocator.free(indexed); + try testing.expectEqualStrings(selected[0], indexed); + try testing.expect(try meta.getAlloc(testing.allocator, list_repos_cursor_key) == null); + try testing.expect(try meta.getAlloc(testing.allocator, bootstrap_cursor_key) == null); + try server_future.await(io); +} + +test "explicit selected backfill repairs identity metadata without redownloading a complete repo" { + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; + defer listener.deinit(io); + const Fixture = struct { + fn serve(server: *Io.net.Server, task_io: Io) !void { + var stream = try server.accept(task_io); + defer stream.close(task_io); + var read_buf: [2048]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(); + if (!std.mem.startsWith(u8, request.head.target, "/did:plc:")) return error.UnexpectedSelectedBackfillRequest; + try request.respond( + "{\"id\":\"did:plc:abcdefghijklmnopqrstuvwx\",\"alsoKnownAs\":[\"at://repaired.test\"],\"verificationMethod\":[],\"service\":[{\"id\":\"#atproto_pds\",\"type\":\"AtprotoPersonalDataServer\",\"serviceEndpoint\":\"https://repaired-pds.example\"}]}", + .{ .status = .ok, .keep_alive = false }, + ); + } + }; + var server_future = try io.concurrent(Fixture.serve, .{ &listener, io }); + 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}", .{listener.socket.address.getPort()}); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const data_dir_len = try tmp.dir.realPath(io, &data_dir_buf); + const data_dir = data_dir_buf[0..data_dir_len]; + var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir); + defer archive.deinit(); + var meta = try meta_store.Store.open(testing.allocator, data_dir); + defer meta.deinit(); + var store = try repo_store.Store.init(testing.allocator, io, &meta); + defer store.deinit(); + const did = "did:plc:abcdefghijklmnopqrstuvwx"; + try store.put(did, .{ .status = .complete, .active = false, .rev = "backfill", .latest_rev = "latest" }); + const selected = [_][]const u8{did}; + const stats = try run(testing.allocator, io, .{ + .relay_http = url, + .plc_url = url, + .data_dir = data_dir, + .selected_repos = &selected, + .workers = 1, + .max_inflight_bytes = 1024 * 1024, + }, &archive, &store); + try testing.expectEqual(@as(u64, 0), stats.completed); + const state = (try store.get(testing.allocator, did)).?; + defer store.freeState(testing.allocator, state); + try testing.expectEqual(repo_store.Status.complete, state.status); + try testing.expect(state.active); + try testing.expectEqualStrings("backfill", state.rev); + try testing.expectEqualStrings("latest", state.latest_rev); + try testing.expectEqualStrings("repaired.test", state.handle); + try testing.expectEqualStrings("https://repaired-pds.example", state.pds); + try server_future.await(io); +} + test "worker limits cannot deadlock the shared byte budget" { const limits = effectiveWorkerLimits(16, default_max_inflight_bytes); try std.testing.expectEqual(@as(usize, 4), limits.count); @@ -449,6 +747,85 @@ test "worker limits cannot deadlock the shared byte budget" { try std.testing.expectEqual(@as(usize, 512 * 1024 * 1024), small.bytes); } +test "terminal transitions preserve upstream RepoStatus diagnostics exactly" { + const previous: repo_store.RepoState = .{ + .status = .failed, + .active = false, + .rev = "initial-rev", + .latest_rev = "prior-live-rev", + .host = "old-pds.example", + .handle = "alice.test", + .pds = "https://identity-pds.example", + .last_error = "old failure", + .error_class = .timeout, + .attempts = 4, + .last_attempt_us = 10, + .started_us = 5, + .completed_us = 7, + .retry_count = 3, + .next_attempt_us = 99, + .updated_us = 8, + .record_count = 42, + .total_bytes = 1024, + }; + + const completed = terminalState(.{ + .did = "did:plc:alice", + .status = .complete, + .rev = @constCast("new-rev"), + .latest_rev = @constCast("new-rev"), + .host = @constCast("new-pds.example"), + .attempts = 2, + .started_us = 20, + }, previous, 100); + try testing.expect(!completed.active); + try testing.expectEqualStrings("new-rev", completed.rev); + try testing.expectEqualStrings("new-rev", completed.latest_rev); + try testing.expectEqualStrings("new-pds.example", completed.host); + try testing.expectEqualStrings("alice.test", completed.handle); + try testing.expectEqualStrings("https://identity-pds.example", completed.pds); + try testing.expectEqualStrings("", completed.last_error); + try testing.expectEqual(repo_store.ErrorClass.unknown, completed.error_class); + try testing.expectEqual(@as(u32, 0), completed.attempts); + try testing.expectEqual(@as(u32, 0), completed.retry_count); + try testing.expectEqual(@as(i64, 0), completed.next_attempt_us); + try testing.expectEqual(@as(i64, 5), completed.started_us); + try testing.expectEqual(@as(i64, 100), completed.last_attempt_us); + try testing.expectEqual(@as(i64, 100), completed.completed_us); + try testing.expectEqual(@as(i64, 100), completed.updated_us); + try testing.expectEqual(@as(i64, 42), completed.record_count); + try testing.expectEqual(@as(i64, 1024), completed.total_bytes); + + const failed = terminalState(.{ + .did = "did:plc:alice", + .status = .failed, + .last_error = @constCast("new failure"), + .error_class = .http_5xx, + .attempts = 2, + .started_us = 20, + }, previous, 101); + try testing.expectEqualStrings("initial-rev", failed.rev); + try testing.expectEqualStrings("prior-live-rev", failed.latest_rev); + try testing.expectEqualStrings("old-pds.example", failed.host); + try testing.expectEqualStrings("new failure", failed.last_error); + try testing.expectEqual(@as(u32, 2), failed.attempts); + try testing.expectEqual(@as(u32, 3), failed.retry_count); + try testing.expectEqual(@as(i64, 99), failed.next_attempt_us); + try testing.expectEqual(@as(i64, 7), failed.completed_us); + try testing.expectEqual(@as(i64, 8), failed.updated_us); + + const not_found = terminalState(.{ + .did = "did:plc:alice", + .status = .complete, + .started_us = 20, + }, previous, 102); + try testing.expectEqualStrings("initial-rev", not_found.rev); + try testing.expectEqualStrings("prior-live-rev", not_found.latest_rev); + try testing.expectEqual(@as(i64, 8), not_found.updated_us); + try testing.expectEqual(@as(i64, 102), not_found.completed_us); + try testing.expectEqual(@as(u32, 0), not_found.attempts); +} + test "resource exhaustion is fatal rather than a durable large-repo skip" { try std.testing.expect(isFatalDownloadError(error.OutOfMemory)); try std.testing.expect(isFatalDownloadError(error.ArchiveAppendFailed)); @@ -510,28 +887,27 @@ fn downloadOne( defer arena.deinit(); const alloc = arena.allocator(); attempts += 1; - const attempted_at = Io.Timestamp.now(io, .real).toMicroseconds(); var outcome = prepared_fetch.fetch(result_allocator, alloc, io, scratch_dir, relay_http, did, &final_host) catch |err| { // Upstream deliberately fails a five-minute attempt immediately // for this run; retrying it in-place can occupy a worker for // another 15 minutes. The durable steady pass will revisit it. - if (err == error.AttemptTimeout) return failureResult(result_allocator, did, &final_host, attempts, attempted_at, started_us, .timeout, @errorName(err)); - if (transient_attempt >= transient_retries) return failureResult(result_allocator, did, &final_host, attempts, attempted_at, started_us, repo_store.classifyAttemptError(err), @errorName(err)); + if (err == error.AttemptTimeout) return failureResult(result_allocator, did, &final_host, attempts, started_us, .timeout, @errorName(err)); + if (transient_attempt >= transient_retries) return failureResult(result_allocator, did, &final_host, attempts, started_us, repo_store.classifyAttemptError(err), @errorName(err)); try sleepBackoff(io, transient_attempt, 30); transient_attempt += 1; continue; }; switch (outcome) { - .not_found => return successResult(result_allocator, did, .complete, &final_host, attempts, attempted_at, started_us), - .unavailable => return successResult(result_allocator, did, .unavailable, &final_host, attempts, attempted_at, started_us), - .permanent => return failureResult(result_allocator, did, &final_host, attempts, attempted_at, started_us, .unknown, "permanent getRepo response"), + .not_found => return successResult(result_allocator, did, .complete, &final_host, attempts, started_us), + .unavailable => return successResult(result_allocator, did, .unavailable, &final_host, attempts, started_us), + .permanent => return failureResult(result_allocator, did, &final_host, attempts, started_us, .unknown, "permanent getRepo response"), .transient => { - if (transient_attempt >= transient_retries) return failureResult(result_allocator, did, &final_host, attempts, attempted_at, started_us, .http_5xx, "transient getRepo response"); + if (transient_attempt >= transient_retries) return failureResult(result_allocator, did, &final_host, attempts, started_us, .http_5xx, "transient getRepo response"); try sleepBackoff(io, transient_attempt, 30); transient_attempt += 1; }, .rate_limited => |retry_after| { - if (rate_attempt >= rate_limit_retries) return failureResult(result_allocator, did, &final_host, attempts, attempted_at, started_us, .http_429, "rate limited"); + if (rate_attempt >= rate_limit_retries) return failureResult(result_allocator, did, &final_host, attempts, started_us, .http_429, "rate limited"); const delay = @min(retry_after orelse (@as(u64, 1) << @min(rate_attempt, 8)), rate_limit_ceiling_s); try io.sleep(Io.Duration.fromSeconds(@max(delay, 1)), .awake); rate_attempt += 1; @@ -556,20 +932,26 @@ fn downloadOne( const head_rev = repos.emitPrepared(alloc, &prepared.repo, did, now_us, .create, &local.drops, &sink, Sink.emit) catch |err| switch (err) { error.OutOfMemory, error.ArchiveAppendFailed => return err, error.MissingBlock, error.IncompleteCar => { - if (transient_attempt >= transient_retries) return failureResult(result_allocator, did, &final_host, attempts, attempted_at, started_us, .car, @errorName(err)); + if (transient_attempt >= transient_retries) return failureResult(result_allocator, did, &final_host, attempts, started_us, .car, @errorName(err)); try sleepBackoff(io, transient_attempt, 30); transient_attempt += 1; continue; }, - else => return failureResult(result_allocator, did, &final_host, attempts, attempted_at, started_us, .car, @errorName(err)), + else => return failureResult(result_allocator, did, &final_host, attempts, started_us, .car, @errorName(err)), }; + const owned_rev = try result_allocator.dupe(u8, head_rev); + errdefer result_allocator.free(owned_rev); + const owned_latest_rev = try result_allocator.dupe(u8, head_rev); + errdefer result_allocator.free(owned_latest_rev); + const owned_host = try result_allocator.dupe(u8, final_host.slice()); + errdefer if (owned_host.len > 0) result_allocator.free(owned_host); return .{ .did = did, .status = .complete, - .rev = try result_allocator.dupe(u8, head_rev), - .host = try result_allocator.dupe(u8, final_host.slice()), + .rev = owned_rev, + .latest_rev = owned_latest_rev, + .host = owned_host, .attempts = attempts, - .last_attempt_us = attempted_at, .started_us = started_us, .rows = sink.rows, .drops = local.drops, @@ -579,34 +961,35 @@ fn downloadOne( } } -fn successResult(allocator: Allocator, did: []const u8, status: repo_store.Status, host: *const fetched_car.FinalHost, attempts: u32, attempted_at: i64, started_us: i64) !Result { +fn successResult(allocator: Allocator, did: []const u8, status: repo_store.Status, host: *const fetched_car.FinalHost, attempts: u32, started_us: i64) !Result { return .{ .did = did, .status = status, .host = try allocator.dupe(u8, host.slice()), .attempts = attempts, - .last_attempt_us = attempted_at, .started_us = started_us, }; } -fn failureResult(allocator: Allocator, did: []const u8, host: *const fetched_car.FinalHost, attempts: u32, attempted_at: i64, started_us: i64, class: repo_store.ErrorClass, message: []const u8) !Result { +fn failureResult(allocator: Allocator, did: []const u8, host: *const fetched_car.FinalHost, attempts: u32, started_us: i64, class: repo_store.ErrorClass, message: []const u8) !Result { const owned_host = try allocator.dupe(u8, host.slice()); - errdefer allocator.free(owned_host); + errdefer if (owned_host.len > 0) allocator.free(owned_host); + const owned_error = try allocator.dupe(u8, message); + errdefer if (owned_error.len > 0) allocator.free(owned_error); return .{ .did = did, .status = .failed, .host = owned_host, - .last_error = try allocator.dupe(u8, message), + .last_error = owned_error, .error_class = class, .attempts = attempts, - .last_attempt_us = attempted_at, .started_us = started_us, }; } fn freeResult(allocator: Allocator, result: Result) void { if (result.rev.len > 0) allocator.free(result.rev); + if (result.latest_rev.len > 0) allocator.free(result.latest_rev); if (result.host.len > 0) allocator.free(result.host); if (result.last_error.len > 0) allocator.free(result.last_error); } diff --git a/src/internal/bootstrap/lifecycle.zig b/src/internal/bootstrap/lifecycle.zig index 921a5a9..7909dc5 100644 --- a/src/internal/bootstrap/lifecycle.zig +++ b/src/internal/bootstrap/lifecycle.zig @@ -48,6 +48,8 @@ pub const Options = struct { data_dir: []const u8, upstream: []const u8, relay_http: []const u8, + plc_url: []const u8, + selected_repos: []const []const u8 = &.{}, /// 0 = full crawl; >0 = upstream --max-backfill-repos debug semantics max_repos: usize = 0, workers: usize = engine.default_workers, @@ -219,11 +221,21 @@ fn runPendingPass( .status = .failed, .active = if (old) |state| state.active else true, .rev = if (old) |state| state.rev else "", + .latest_rev = if (old) |state| state.latest_rev else "", .host = if (details.final_host.len > 0) details.final_host.slice() else if (old) |state| state.host else "", + .handle = if (old) |state| state.handle else "", + .pds = if (old) |state| state.pds else "", .last_error = @errorName(err), .error_class = details.error_class, .attempts = (if (old) |state| state.attempts else 0) +| 1, .last_attempt_us = Io.Timestamp.now(io, .real).toMicroseconds(), + .started_us = if (old) |state| state.started_us else 0, + .completed_us = if (old) |state| state.completed_us else 0, + .retry_count = if (old) |state| state.retry_count else 0, + .next_attempt_us = if (old) |state| state.next_attempt_us else 0, + .updated_us = if (old) |state| state.updated_us else 0, + .record_count = if (old) |state| state.record_count else 0, + .total_bytes = if (old) |state| state.total_bytes else 0, }); deferred += 1; continue; @@ -367,7 +379,9 @@ fn runBootstrap( const bf = engine.run(allocator, io, .{ .relay_http = opts.relay_http, + .plc_url = opts.plc_url, .data_dir = opts.data_dir, + .selected_repos = opts.selected_repos, .max_repos = opts.max_repos, .workers = opts.workers, .max_inflight_bytes = opts.max_inflight_bytes, diff --git a/src/internal/bootstrap/merge.zig b/src/internal/bootstrap/merge.zig index 2e7ae44..8859330 100644 --- a/src/internal/bootstrap/merge.zig +++ b/src/internal/bootstrap/merge.zig @@ -11,10 +11,8 @@ //! commit-shaped or sync (the kinds that carry a repo rev), the DID's //! backfill is complete, both revs are non-empty, and row.rev <= //! Backfill.Rev (TIDs sort lexicographically). identity/account rows -//! always survive. upstream's per-source top-level Rev refresh is -//! skipped: our repo store keeps only the immutable Backfill.Rev the -//! filter compares against; the refreshed rev feeds steady-state -//! resync, which we don't have yet. +//! always survive. Each source commit atomically advances the merge cursor and +//! refreshes top-level latest Rev/UpdatedAt while preserving Backfill.Rev. const std = @import("std"); const archive_mod = @import("../archive.zig"); @@ -82,10 +80,17 @@ pub fn run( if (idx != expect) return error.SourceIndexGap; expect += 1; - try drainOne(allocator, io, src_dir, idx, dst, store, &stats); + var source_arena = std.heap.ArenaAllocator.init(allocator); + defer source_arena.deinit(); + const revs = try drainOne(source_arena.allocator(), io, src_dir, idx, dst, store, &stats); // durability ordering: dst rows fsynced before the cursor advances try dst.flushBlock(); - try storeCursor(store, idx + 1); + const updates = try source_arena.allocator().alloc(repo_store.RevUpdate, revs.count()); + var update_it = revs.iterator(); + var update_index: usize = 0; + while (update_it.next()) |entry| : (update_index += 1) + updates[update_index] = .{ .did = entry.key_ptr.*, .rev = entry.value_ptr.* }; + try store.commitMergeSource(cursor_key, idx + 1, updates, Io.Timestamp.now(io, .real).toMicroseconds()); stats.sources += 1; } return stats; @@ -99,13 +104,14 @@ fn drainOne( dst: *archive_mod.Archive, store: *repo_store.Store, stats: *Stats, -) !void { +) !std.StringHashMapUnmanaged([]const u8) { var name_buf: [64]u8 = undefined; const name = archive_mod.formatSegmentName(&name_buf, idx); const bytes = try src_dir.readFileAlloc(io, name, allocator, .limited(1 << 31)); defer allocator.free(bytes); var sealed = try segment.Sealed.parse(allocator, bytes); defer sealed.deinit(allocator); + var per_did: std.StringHashMapUnmanaged([]const u8) = .empty; const now_us: i64 = Io.Timestamp.now(io, .real).toMicroseconds(); for (0..sealed.block_index.len) |i| { @@ -122,8 +128,17 @@ fn drainOne( stamped.witnessed_at = now_us; // upstream §3.4 re-stamp _ = try dst.append(stamped, now_us); stats.kept += 1; + switch (ev.kind) { + .create, .update, .delete, .create_resync, .sync => if (ev.rev.len > 0) { + const result = try per_did.getOrPut(allocator, ev.did); + if (!result.found_existing) result.key_ptr.* = try allocator.dupe(u8, ev.did); + result.value_ptr.* = try allocator.dupe(u8, ev.rev); + }, + .identity, .account => {}, + } } } + return per_did; } const cursor_key = "merge/next_source_idx"; @@ -136,12 +151,6 @@ fn loadCursor(store: *repo_store.Store) u64 { return std.mem.readInt(u64, bytes[0..8], .little); } -fn storeCursor(store: *repo_store.Store, next_idx: u64) !void { - var bytes: [8]u8 = undefined; - std.mem.writeInt(u64, &bytes, next_idx, .little); - try store.meta.putDurable(cursor_key, &bytes); -} - // === tests === const testing = std.testing; diff --git a/src/internal/bootstrap/repo_store.zig b/src/internal/bootstrap/repo_store.zig index 7f9c1af..19e74a3 100644 --- a/src/internal/bootstrap/repo_store.zig +++ b/src/internal/bootstrap/repo_store.zig @@ -15,10 +15,14 @@ const Allocator = std.mem.Allocator; const repo_prefix = "repo/"; const counts_key = "backfill/counts"; const host_prefix = "host/"; +const handle_prefix = "handle/"; const hosts_version_key = "backfill/host-status-version"; const hosts_version: u8 = 1; -const encoding_version: u8 = 4; -const header_len = 52; +const handles_version_key = "backfill/handle-index-version"; +const handles_version: u8 = 1; +const encoding_version: u8 = 5; +const header_len = 84; +const v4_header_len = 52; const legacy_header_len = 14; pub const Status = host_store.RepoStatus; @@ -36,13 +40,45 @@ pub fn classifyAttemptError(err: anyerror) ErrorClass { }; } +/// Convert a resolved PDS service endpoint to upstream's durable host bucket. +/// Malformed or non-HTTP endpoints are retained in the explicit invalid-pds +/// bucket so selected-backfill diagnostics never silently lose the account. +pub fn hostFromPDS(allocator: Allocator, endpoint: []const u8) ![]u8 { + const uri = std.Uri.parse(std.mem.trim(u8, endpoint, " \t\r\n")) catch + return allocator.dupe(u8, "invalid-pds"); + if (!std.ascii.eqlIgnoreCase(uri.scheme, "http") and !std.ascii.eqlIgnoreCase(uri.scheme, "https")) + return allocator.dupe(u8, "invalid-pds"); + var host_buf: [std.Io.net.HostName.max_len]u8 = undefined; + const host = uri.getHost(&host_buf) catch return allocator.dupe(u8, "invalid-pds"); + if (host.bytes.len == 0) return allocator.dupe(u8, "invalid-pds"); + const bare_host = if (host.bytes.len >= 2 and host.bytes[0] == '[' and host.bytes[host.bytes.len - 1] == ']') + host.bytes[1 .. host.bytes.len - 1] + else + host.bytes; + const default_port: ?u16 = if (std.ascii.eqlIgnoreCase(uri.scheme, "https")) 443 else 80; + const raw = if (uri.port != null and uri.port != default_port) + if (std.mem.indexOfScalar(u8, bare_host, ':') != null) + try std.fmt.allocPrint(allocator, "[{s}]:{d}", .{ bare_host, uri.port.? }) + else + try std.fmt.allocPrint(allocator, "{s}:{d}", .{ bare_host, uri.port.? }) + else + try allocator.dupe(u8, bare_host); + defer allocator.free(raw); + return host_store.normalizeAuthority(allocator, raw); +} + pub const RepoState = struct { status: Status, active: bool = true, /// Owned by the caller when returned from `get`; literals passed to put /// remain caller-owned. rev: []const u8 = "", + /// Latest known repo revision. This equals `rev` at initial backfill and + /// advances independently as captured live events are merged. + latest_rev: []const u8 = "", host: []const u8 = "", + handle: []const u8 = "", + pds: []const u8 = "", last_error: []const u8 = "", error_class: ErrorClass = .unknown, attempts: u32 = 0, @@ -51,9 +87,13 @@ pub const RepoState = struct { completed_us: i64 = 0, retry_count: u32 = 0, next_attempt_us: i64 = 0, + updated_us: i64 = 0, + record_count: i64 = 0, + total_bytes: i64 = 0, }; pub const Transition = struct { did: []const u8, state: RepoState }; +pub const RevUpdate = struct { did: []const u8, rev: []const u8 }; pub const Store = struct { allocator: Allocator, @@ -69,6 +109,7 @@ pub const Store = struct { try self.rebuildCounts(); } try self.ensureHostStatuses(); + try self.ensureHandleIndex(); // A durable not_started row means the previous process died after // dispatch and before completion. Defer it to the merge pending pass; // never re-download it into the low bootstrap sequence range. @@ -89,7 +130,10 @@ pub const Store = struct { .status = .pending, .active = true, .rev = state.rev, + .latest_rev = state.latest_rev, .host = state.host, + .handle = state.handle, + .pds = state.pds, .last_error = state.last_error, .error_class = state.error_class, .attempts = state.attempts, @@ -98,6 +142,9 @@ pub const Store = struct { .completed_us = state.completed_us, .retry_count = state.retry_count, .next_attempt_us = state.next_attempt_us, + .updated_us = state.updated_us, + .record_count = state.record_count, + .total_bytes = state.total_bytes, } }; n += 1; } @@ -116,15 +163,24 @@ pub const Store = struct { const state = try decode(bytes); const rev = if (state.rev.len > 0) try allocator.dupe(u8, state.rev) else ""; errdefer if (rev.len > 0) allocator.free(@constCast(rev)); + const latest_rev = if (state.latest_rev.len > 0) try allocator.dupe(u8, state.latest_rev) else ""; + errdefer if (latest_rev.len > 0) allocator.free(@constCast(latest_rev)); const host = if (state.host.len > 0) try allocator.dupe(u8, state.host) else ""; errdefer if (host.len > 0) allocator.free(@constCast(host)); + const handle = if (state.handle.len > 0) try allocator.dupe(u8, state.handle) else ""; + errdefer if (handle.len > 0) allocator.free(@constCast(handle)); + const pds = if (state.pds.len > 0) try allocator.dupe(u8, state.pds) else ""; + errdefer if (pds.len > 0) allocator.free(@constCast(pds)); const last_error = if (state.last_error.len > 0) try allocator.dupe(u8, state.last_error) else ""; allocator.free(bytes); return .{ .status = state.status, .active = state.active, .rev = rev, + .latest_rev = latest_rev, .host = host, + .handle = handle, + .pds = pds, .last_error = last_error, .error_class = state.error_class, .attempts = state.attempts, @@ -133,12 +189,18 @@ pub const Store = struct { .completed_us = state.completed_us, .retry_count = state.retry_count, .next_attempt_us = state.next_attempt_us, + .updated_us = state.updated_us, + .record_count = state.record_count, + .total_bytes = state.total_bytes, }; } pub fn freeState(_: *Store, allocator: Allocator, state: RepoState) void { if (state.rev.len > 0) allocator.free(@constCast(state.rev)); + if (state.latest_rev.len > 0) allocator.free(@constCast(state.latest_rev)); if (state.host.len > 0) allocator.free(@constCast(state.host)); + if (state.handle.len > 0) allocator.free(@constCast(state.handle)); + if (state.pds.len > 0) allocator.free(@constCast(state.pds)); if (state.last_error.len > 0) allocator.free(@constCast(state.last_error)); } @@ -205,6 +267,11 @@ pub const Store = struct { try host_store.recordError(diagnostic, self.allocator, transition.did, transition.state.last_attempt_us, transition.state.error_class, transition.state.last_error); } + if (!had_old or !handlesEqual(old.handle, transition.state.handle)) { + if (had_old) try self.stageHandleDeleteIfMatches(&writes, old.handle, transition.did); + try stageHandleSet(self.allocator, &writes, transition.state.handle, transition.did); + } + var canonical = transition.state; canonical.host = new_host; const value = try encode(self.allocator, canonical); @@ -250,6 +317,41 @@ pub const Store = struct { allocator.free(statuses); } + /// Lookup through the durable declared-handle index. Like upstream, this + /// operator convenience is normalized but is not bidirectionally verified. + pub fn lookupDIDByHandle(self: *Store, allocator: Allocator, handle: []const u8) !?[]u8 { + const key = try handleKey(self.allocator, handle) orelse return null; + defer self.allocator.free(key); + return self.meta.getAlloc(allocator, key); + } + + /// Upstream's per-source merge commit: advance the source cursor and + /// refresh only top-level latest revision/time in one synced metadata + /// batch. Backfill `rev` and every lifecycle/identity field are preserved. + pub fn commitMergeSource(self: *Store, cursor_key: []const u8, next_idx: u64, updates: []const RevUpdate, updated_us: i64) !void { + self.mu.lockUncancelable(self.io); + defer self.mu.unlock(self.io); + + var writes = self.meta.batch(); + defer writes.deinit(); + var cursor_bytes: [8]u8 = undefined; + std.mem.writeInt(u64, &cursor_bytes, next_idx, .little); + writes.put(cursor_key, &cursor_bytes); + for (updates) |update| { + const key = try repoKey(self.allocator, update.did); + defer self.allocator.free(key); + const bytes = try self.meta.getAlloc(self.allocator, key) orelse continue; + defer self.allocator.free(bytes); + var state = try decode(bytes); + state.latest_rev = update.rev; + state.updated_us = updated_us; + const encoded = try encode(self.allocator, state); + defer self.allocator.free(encoded); + writes.put(key, encoded); + } + try writes.commitDurable(); + } + /// Compatibility with the former buffered store. Writes are already /// synchronously durable, so there is nothing left to flush. pub fn flush(_: *Store) !void {} @@ -350,6 +452,36 @@ pub const Store = struct { try writes.commitDurable(); } + fn ensureHandleIndex(self: *Store) !void { + if (try self.meta.getAlloc(self.allocator, handles_version_key)) |version| { + defer self.allocator.free(version); + if (version.len == 1 and version[0] == handles_version) return; + return error.UnsupportedHandleIndexVersion; + } + + var writes = self.meta.batch(); + defer writes.deinit(); + var err_data: ?rocksdb.Data = null; + defer if (err_data) |e| e.deinit(); + var it = self.meta.iterator(repo_prefix); + defer it.deinit(); + while (try it.next(&err_data)) |entry| { + if (!std.mem.startsWith(u8, entry[0].data, repo_prefix)) break; + const state = try decode(entry[1].data); + try stageHandleSet(self.allocator, &writes, state.handle, entry[0].data[repo_prefix.len..]); + } + writes.put(handles_version_key, &.{handles_version}); + try writes.commitDurable(); + } + + fn stageHandleDeleteIfMatches(self: *Store, writes: *meta_store.Batch, handle: []const u8, did: []const u8) !void { + const key = try handleKey(self.allocator, handle) orelse return; + defer self.allocator.free(key); + const indexed = try self.meta.getAlloc(self.allocator, key) orelse return; + defer self.allocator.free(indexed); + if (std.mem.eql(u8, indexed, did)) writes.delete(key); + } + fn cachedHost(self: *Store, cache: *std.StringHashMap(HostStatus), normalized_host: []const u8) !*HostStatus { if (cache.getPtr(normalized_host)) |status| return status; const key = try hostKey(self.allocator, normalized_host); @@ -391,10 +523,31 @@ fn hostKey(allocator: Allocator, host: []const u8) ![]u8 { return std.mem.concat(allocator, u8, &.{ host_prefix, host }); } +fn handleKey(allocator: Allocator, handle: []const u8) !?[]u8 { + const normalized = std.mem.trim(u8, handle, " \t\r\n"); + if (normalized.len == 0) return null; + const out = try allocator.alloc(u8, handle_prefix.len + normalized.len); + @memcpy(out[0..handle_prefix.len], handle_prefix); + for (normalized, out[handle_prefix.len..]) |byte, *dst| dst.* = std.ascii.toLower(byte); + return out; +} + +fn handlesEqual(a: []const u8, b: []const u8) bool { + return std.ascii.eqlIgnoreCase(std.mem.trim(u8, a, " \t\r\n"), std.mem.trim(u8, b, " \t\r\n")); +} + +fn stageHandleSet(allocator: Allocator, writes: *meta_store.Batch, handle: []const u8, did: []const u8) !void { + const key = try handleKey(allocator, handle) orelse return; + defer allocator.free(key); + writes.put(key, did); +} + fn encode(allocator: Allocator, state: RepoState) ![]u8 { - if (state.host.len > std.math.maxInt(u16) or state.last_error.len > std.math.maxInt(u16) or state.rev.len > std.math.maxInt(u32)) + if (state.host.len > std.math.maxInt(u16) or state.handle.len > std.math.maxInt(u16) or + state.pds.len > std.math.maxInt(u16) or state.last_error.len > std.math.maxInt(u16) or + state.rev.len > std.math.maxInt(u32) or state.latest_rev.len > std.math.maxInt(u32)) return error.RepoStateFieldTooLong; - const out = try allocator.alloc(u8, header_len + state.rev.len + state.host.len + state.last_error.len); + const out = try allocator.alloc(u8, header_len + state.rev.len + state.latest_rev.len + state.host.len + state.handle.len + state.pds.len + state.last_error.len); out[0] = encoding_version; out[1] = @intFromEnum(state.status); out[2] = @intFromBool(state.active); @@ -404,15 +557,27 @@ fn encode(allocator: Allocator, state: RepoState) ![]u8 { std.mem.writeInt(i64, out[19..27], state.last_attempt_us, .little); std.mem.writeInt(i64, out[27..35], state.started_us, .little); std.mem.writeInt(i64, out[35..43], state.completed_us, .little); - std.mem.writeInt(u32, out[43..47], @intCast(state.rev.len), .little); - std.mem.writeInt(u16, out[47..49], @intCast(state.host.len), .little); - std.mem.writeInt(u16, out[49..51], @intCast(state.last_error.len), .little); - out[51] = @intFromEnum(state.error_class); + std.mem.writeInt(i64, out[43..51], state.updated_us, .little); + std.mem.writeInt(i64, out[51..59], state.record_count, .little); + std.mem.writeInt(i64, out[59..67], state.total_bytes, .little); + std.mem.writeInt(u32, out[67..71], @intCast(state.rev.len), .little); + std.mem.writeInt(u32, out[71..75], @intCast(state.latest_rev.len), .little); + std.mem.writeInt(u16, out[75..77], @intCast(state.host.len), .little); + std.mem.writeInt(u16, out[77..79], @intCast(state.handle.len), .little); + std.mem.writeInt(u16, out[79..81], @intCast(state.pds.len), .little); + std.mem.writeInt(u16, out[81..83], @intCast(state.last_error.len), .little); + out[83] = @intFromEnum(state.error_class); var offset: usize = header_len; @memcpy(out[offset..][0..state.rev.len], state.rev); offset += state.rev.len; + @memcpy(out[offset..][0..state.latest_rev.len], state.latest_rev); + offset += state.latest_rev.len; @memcpy(out[offset..][0..state.host.len], state.host); offset += state.host.len; + @memcpy(out[offset..][0..state.handle.len], state.handle); + offset += state.handle.len; + @memcpy(out[offset..][0..state.pds.len], state.pds); + offset += state.pds.len; @memcpy(out[offset..][0..state.last_error.len], state.last_error); return out; } @@ -426,6 +591,7 @@ fn decode(bytes: []const u8) !RepoState { .retry_count = std.mem.readInt(u32, bytes[2..6], .little), .next_attempt_us = std.mem.readInt(i64, bytes[6..14], .little), .rev = bytes[legacy_header_len..], + .latest_rev = bytes[legacy_header_len..], }; } if (bytes.len >= 15 and bytes[0] == 3) { @@ -436,31 +602,70 @@ fn decode(bytes: []const u8) !RepoState { .retry_count = std.mem.readInt(u32, bytes[3..7], .little), .next_attempt_us = std.mem.readInt(i64, bytes[7..15], .little), .rev = bytes[15..], + .latest_rev = bytes[15..], + }; + } + if (bytes.len >= v4_header_len and bytes[0] == 4) { + if (bytes[1] >= std.meta.fields(Status).len) return error.InvalidRepoState; + if (bytes[51] >= std.meta.fields(ErrorClass).len) return error.InvalidRepoState; + const rev_len = std.mem.readInt(u32, bytes[43..47], .little); + const host_len = std.mem.readInt(u16, bytes[47..49], .little); + const error_len = std.mem.readInt(u16, bytes[49..51], .little); + const total = v4_header_len + @as(usize, rev_len) + @as(usize, host_len) + @as(usize, error_len); + if (total != bytes.len) return error.InvalidRepoState; + const rev_end = v4_header_len + rev_len; + const host_end = rev_end + host_len; + return .{ + .status = @enumFromInt(bytes[1]), + .active = bytes[2] != 0, + .rev = bytes[v4_header_len..rev_end], + .latest_rev = bytes[v4_header_len..rev_end], + .host = bytes[rev_end..host_end], + .last_error = bytes[host_end..total], + .error_class = @enumFromInt(bytes[51]), + .retry_count = std.mem.readInt(u32, bytes[3..7], .little), + .next_attempt_us = std.mem.readInt(i64, bytes[7..15], .little), + .attempts = std.mem.readInt(u32, bytes[15..19], .little), + .last_attempt_us = std.mem.readInt(i64, bytes[19..27], .little), + .started_us = std.mem.readInt(i64, bytes[27..35], .little), + .completed_us = std.mem.readInt(i64, bytes[35..43], .little), }; } if (bytes.len < header_len or bytes[0] != encoding_version) return error.InvalidRepoState; if (bytes[1] >= std.meta.fields(Status).len) return error.InvalidRepoState; - if (bytes[51] >= std.meta.fields(ErrorClass).len) return error.InvalidRepoState; - const rev_len = std.mem.readInt(u32, bytes[43..47], .little); - const host_len = std.mem.readInt(u16, bytes[47..49], .little); - const error_len = std.mem.readInt(u16, bytes[49..51], .little); - const total = header_len + @as(usize, rev_len) + @as(usize, host_len) + @as(usize, error_len); + if (bytes[83] >= std.meta.fields(ErrorClass).len) return error.InvalidRepoState; + const rev_len = std.mem.readInt(u32, bytes[67..71], .little); + const latest_rev_len = std.mem.readInt(u32, bytes[71..75], .little); + const host_len = std.mem.readInt(u16, bytes[75..77], .little); + const handle_len = std.mem.readInt(u16, bytes[77..79], .little); + const pds_len = std.mem.readInt(u16, bytes[79..81], .little); + const error_len = std.mem.readInt(u16, bytes[81..83], .little); + const total = header_len + @as(usize, rev_len) + @as(usize, latest_rev_len) + @as(usize, host_len) + @as(usize, handle_len) + @as(usize, pds_len) + @as(usize, error_len); if (total != bytes.len) return error.InvalidRepoState; const rev_end = header_len + rev_len; - const host_end = rev_end + host_len; + const latest_rev_end = rev_end + latest_rev_len; + const host_end = latest_rev_end + host_len; + const handle_end = host_end + handle_len; + const pds_end = handle_end + pds_len; return .{ .status = @enumFromInt(bytes[1]), .active = bytes[2] != 0, .rev = bytes[header_len..rev_end], - .host = bytes[rev_end..host_end], - .last_error = bytes[host_end..total], - .error_class = @enumFromInt(bytes[51]), + .latest_rev = bytes[rev_end..latest_rev_end], + .host = bytes[latest_rev_end..host_end], + .handle = bytes[host_end..handle_end], + .pds = bytes[handle_end..pds_end], + .last_error = bytes[pds_end..total], + .error_class = @enumFromInt(bytes[83]), .retry_count = std.mem.readInt(u32, bytes[3..7], .little), .next_attempt_us = std.mem.readInt(i64, bytes[7..15], .little), .attempts = std.mem.readInt(u32, bytes[15..19], .little), .last_attempt_us = std.mem.readInt(i64, bytes[19..27], .little), .started_us = std.mem.readInt(i64, bytes[27..35], .little), .completed_us = std.mem.readInt(i64, bytes[35..43], .little), + .updated_us = std.mem.readInt(i64, bytes[43..51], .little), + .record_count = std.mem.readInt(i64, bytes[51..59], .little), + .total_bytes = std.mem.readInt(i64, bytes[59..67], .little), }; } @@ -481,11 +686,17 @@ test "rows and counts are durable and atomically updated" { .status = .complete, .active = false, .rev = "3lrev", + .latest_rev = "3llatest", .host = "pds.example", + .handle = "Alice.Test", + .pds = "https://pds.example", .last_error = "previous timeout", .error_class = .timeout, .attempts = 2, .last_attempt_us = 17, + .updated_us = 18, + .record_count = 42, + .total_bytes = 1024, }); try store.put("did:plc:aaa", .{ .status = .failed }); try testing.expectEqual(@as(u32, 2), store.count()); @@ -497,11 +708,22 @@ test "rows and counts are durable and atomically updated" { try testing.expectEqual(Status.complete, bbb.status); try testing.expect(!bbb.active); try testing.expectEqualStrings("3lrev", bbb.rev); + try testing.expectEqualStrings("3llatest", bbb.latest_rev); try testing.expectEqualStrings("pds.example", bbb.host); + try testing.expectEqualStrings("Alice.Test", bbb.handle); + try testing.expectEqualStrings("https://pds.example", bbb.pds); try testing.expectEqualStrings("previous timeout", bbb.last_error); try testing.expectEqual(ErrorClass.timeout, bbb.error_class); try testing.expectEqual(@as(u32, 2), bbb.attempts); try testing.expectEqual(@as(i64, 17), bbb.last_attempt_us); + try testing.expectEqual(@as(i64, 18), bbb.updated_us); + try testing.expectEqual(@as(i64, 42), bbb.record_count); + try testing.expectEqual(@as(i64, 1024), bbb.total_bytes); + const indexed = (try store.lookupDIDByHandle(testing.allocator, " @ignored ")); + try testing.expect(indexed == null); + const alice = (try store.lookupDIDByHandle(testing.allocator, " alice.TEST ")).?; + defer testing.allocator.free(alice); + try testing.expectEqualStrings("did:plc:bbb", alice); const failed = try store.collectByStatus(testing.allocator, .failed); defer store.freeDids(testing.allocator, failed); @@ -605,6 +827,107 @@ test "repo transitions atomically maintain durable upstream-shaped host diagnost try testing.expectEqual(@as(u64, 1), rebuilt[0].count(.complete)); } +test "repo transitions atomically maintain the normalized declared-handle index" { + const testing = std.testing; + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var path_buf: [std.fs.max_path_bytes]u8 = undefined; + const path_len = try tmp.dir.realPath(testing.io, &path_buf); + + var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]); + defer meta.deinit(); + var store = try Store.init(testing.allocator, testing.io, &meta); + defer store.deinit(); + + try store.put("did:plc:alice", .{ .status = .complete, .handle = " Alice.Test " }); + var did = (try store.lookupDIDByHandle(testing.allocator, "ALICE.test")).?; + try testing.expectEqualStrings("did:plc:alice", did); + testing.allocator.free(did); + + // Moving a DID removes its old index only when that index still names it. + try store.put("did:plc:bob", .{ .status = .complete, .handle = "alice.test" }); + try store.put("did:plc:alice", .{ .status = .complete, .handle = "alice.new" }); + did = (try store.lookupDIDByHandle(testing.allocator, "alice.test")).?; + try testing.expectEqualStrings("did:plc:bob", did); + testing.allocator.free(did); + did = (try store.lookupDIDByHandle(testing.allocator, "alice.new")).?; + try testing.expectEqualStrings("did:plc:alice", did); + testing.allocator.free(did); + + try store.put("did:plc:alice", .{ .status = .complete, .handle = "" }); + try testing.expect(try store.lookupDIDByHandle(testing.allocator, "alice.new") == null); + + // Rebuilding from repo/ rows is deterministic and survives restart. + try meta.deleteDurable(handles_version_key); + try meta.deleteDurable("handle/alice.test"); + var rebuilt = try Store.init(testing.allocator, testing.io, &meta); + defer rebuilt.deinit(); + did = (try rebuilt.lookupDIDByHandle(testing.allocator, "alice.test")).?; + defer testing.allocator.free(did); + try testing.expectEqualStrings("did:plc:bob", did); +} + +test "merge source commit refreshes only latest repo state with its cursor" { + const testing = std.testing; + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var path_buf: [std.fs.max_path_bytes]u8 = undefined; + const path_len = try tmp.dir.realPath(testing.io, &path_buf); + var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]); + defer meta.deinit(); + var store = try Store.init(testing.allocator, testing.io, &meta); + defer store.deinit(); + + try store.put("did:plc:known", .{ + .status = .complete, + .rev = "backfill-rev", + .latest_rev = "old-live-rev", + .handle = "known.test", + .pds = "https://pds.example", + .record_count = 9, + }); + try store.commitMergeSource("merge/next_source_idx", 7, &.{ + .{ .did = "did:plc:known", .rev = "new-live-rev" }, + // Upstream deliberately does not synthesize malformed lifecycle rows + // for accounts absent from the bootstrap status map. + .{ .did = "did:plc:unknown", .rev = "ignored" }, + }, 1234); + + const cursor = (try meta.getAlloc(testing.allocator, "merge/next_source_idx")).?; + defer testing.allocator.free(cursor); + try testing.expectEqual(@as(u64, 7), std.mem.readInt(u64, cursor[0..8], .little)); + const known = (try store.get(testing.allocator, "did:plc:known")).?; + defer store.freeState(testing.allocator, known); + try testing.expectEqualStrings("backfill-rev", known.rev); + try testing.expectEqualStrings("new-live-rev", known.latest_rev); + try testing.expectEqual(@as(i64, 1234), known.updated_us); + try testing.expectEqualStrings("known.test", known.handle); + try testing.expectEqualStrings("https://pds.example", known.pds); + try testing.expectEqual(@as(i64, 9), known.record_count); + try testing.expect(try store.get(testing.allocator, "did:plc:unknown") == null); +} + +test "v4 rows migrate latest revision without losing lifecycle state" { + const testing = std.testing; + var bytes: [v4_header_len + 3 + 4 + 5]u8 = @splat(0); + bytes[0] = 4; + bytes[1] = @intFromEnum(Status.complete); + bytes[2] = 1; + std.mem.writeInt(u32, bytes[43..47], 3, .little); + std.mem.writeInt(u16, bytes[47..49], 4, .little); + std.mem.writeInt(u16, bytes[49..51], 5, .little); + bytes[51] = @intFromEnum(ErrorClass.timeout); + @memcpy(bytes[52..55], "rev"); + @memcpy(bytes[55..59], "host"); + @memcpy(bytes[59..64], "error"); + const state = try decode(&bytes); + try testing.expectEqualStrings("rev", state.rev); + try testing.expectEqualStrings("rev", state.latest_rev); + try testing.expectEqualStrings("host", state.host); + try testing.expectEqualStrings("error", state.last_error); + try testing.expectEqual(ErrorClass.timeout, state.error_class); +} + test "failure classes match upstream diagnostic buckets" { const testing = std.testing; try testing.expectEqual(ErrorClass.timeout, classifyAttemptError(error.TransferTooSlow)); @@ -613,3 +936,22 @@ test "failure classes match upstream diagnostic buckets" { try testing.expectEqual(ErrorClass.local_write, classifyAttemptError(error.ArchiveAppendFailed)); try testing.expectEqual(ErrorClass.unknown, classifyAttemptError(error.ConnectionResetByPeer)); } + +test "resolved PDS endpoints produce upstream host buckets" { + const testing = std.testing; + const default_https = try hostFromPDS(testing.allocator, "HTTPS://PDS.EXAMPLE:443"); + defer testing.allocator.free(default_https); + try testing.expectEqualStrings("pds.example", default_https); + const custom = try hostFromPDS(testing.allocator, "http://PDS.EXAMPLE:8080/path"); + defer testing.allocator.free(custom); + try testing.expectEqualStrings("pds.example:8080", custom); + const ipv6 = try hostFromPDS(testing.allocator, "https://[2001:DB8::1]:8443"); + defer testing.allocator.free(ipv6); + try testing.expectEqualStrings("[2001:db8::1]:8443", ipv6); + const default_ipv6 = try hostFromPDS(testing.allocator, "https://[2001:DB8::1]:443"); + defer testing.allocator.free(default_ipv6); + try testing.expectEqualStrings("2001:db8::1", default_ipv6); + const invalid = try hostFromPDS(testing.allocator, "javascript:wat"); + defer testing.allocator.free(invalid); + try testing.expectEqualStrings("invalid-pds", invalid); +} diff --git a/src/internal/bootstrap/resync.zig b/src/internal/bootstrap/resync.zig index ec49cb0..e199578 100644 --- a/src/internal/bootstrap/resync.zig +++ b/src/internal/bootstrap/resync.zig @@ -56,10 +56,8 @@ pub fn resyncRepo( defer arena.deinit(); const alloc = arena.allocator(); - const attempted_at = Io.Timestamp.now(io, .real).toMicroseconds(); const previous = try store.get(allocator, did); defer if (previous) |state| store.freeState(allocator, state); - const attempts = if (previous) |state| state.attempts +| 1 else 1; const active = if (previous) |state| state.active else true; const outcome = prepared_fetch.fetch(allocator, alloc, io, scratch_dir, relay_http, did, &details.final_host) catch |err| { details.error_class = repo_store.classifyAttemptError(err); @@ -73,11 +71,11 @@ pub fn resyncRepo( return error.RateLimited; }, .not_found => { - try store.put(did, .{ .status = .complete, .active = active, .host = details.final_host.slice(), .attempts = attempts, .last_attempt_us = attempted_at, .completed_us = attempted_at }); + try store.put(did, terminalState(previous, .complete, active, details.final_host.slice(), Io.Timestamp.now(io, .real).toMicroseconds())); return .{}; }, .unavailable => { - try store.put(did, .{ .status = .unavailable, .active = active, .host = details.final_host.slice(), .attempts = attempts, .last_attempt_us = attempted_at, .completed_us = attempted_at }); + try store.put(did, terminalState(previous, .unavailable, active, details.final_host.slice(), Io.Timestamp.now(io, .real).toMicroseconds())); return .{}; }, .transient => { @@ -126,19 +124,50 @@ pub fn resyncRepo( // durability ordering: every replacement batch is durable before the // completion row that describes it. + const completed_us = Io.Timestamp.now(io, .real).toMicroseconds(); try store.put(did, .{ .status = .complete, .active = active, .rev = head_rev, + .latest_rev = head_rev, .host = details.final_host.slice(), - .attempts = attempts, - .last_attempt_us = attempted_at, - .completed_us = Io.Timestamp.now(io, .real).toMicroseconds(), + .handle = if (previous) |state| state.handle else "", + .pds = if (previous) |state| state.pds else "", + .attempts = 0, + .last_attempt_us = completed_us, + .started_us = if (previous) |state| state.started_us else 0, + .completed_us = completed_us, + .retry_count = 0, + .next_attempt_us = 0, + .updated_us = completed_us, + .record_count = if (previous) |state| state.record_count else 0, + .total_bytes = if (previous) |state| state.total_bytes else 0, }); try store.flush(); return .{ .rows = sink.row_count + 1 }; } +fn terminalState(previous: ?repo_store.RepoState, status: repo_store.Status, active: bool, host: []const u8, terminal_us: i64) repo_store.RepoState { + return .{ + .status = status, + .active = active, + .rev = if (previous) |state| state.rev else "", + .latest_rev = if (previous) |state| state.latest_rev else "", + .host = if (host.len > 0) host else if (previous) |state| state.host else "", + .handle = if (previous) |state| state.handle else "", + .pds = if (previous) |state| state.pds else "", + .attempts = 0, + .last_attempt_us = terminal_us, + .started_us = if (previous) |state| state.started_us else 0, + .completed_us = if (status == .complete) terminal_us else if (previous) |state| state.completed_us else 0, + .retry_count = 0, + .next_attempt_us = 0, + .updated_us = if (previous) |state| state.updated_us else 0, + .record_count = if (previous) |state| state.record_count else 0, + .total_bytes = if (previous) |state| state.total_bytes else 0, + }; +} + /// synthetic #sync payload matching what the live path archives (a CBOR /// map the cold reader pulls did/rev/time from; seq and blocks default) pub fn syncTombstoneEvent(alloc: Allocator, did: []const u8, rev: []const u8, now_us: i64) !segment.Event { @@ -190,6 +219,46 @@ pub fn formatRfc3339(buf: []u8, micros: i64) []const u8 { const testing = std.testing; +test "terminal no-repo resync states match upstream OnFail bookkeeping" { + const previous: repo_store.RepoState = .{ + .status = .failed, + .active = true, + .rev = "initial", + .latest_rev = "latest", + .host = "old.example", + .handle = "alice.test", + .pds = "https://identity.example", + .last_error = "old failure", + .error_class = .timeout, + .attempts = 4, + .started_us = 3, + .completed_us = 5, + .retry_count = 6, + .next_attempt_us = 7, + .updated_us = 8, + .record_count = 9, + .total_bytes = 10, + }; + const missing = terminalState(previous, .complete, true, "new.example", 100); + try testing.expectEqual(@as(u32, 0), missing.attempts); + try testing.expectEqual(@as(u32, 0), missing.retry_count); + try testing.expectEqual(@as(i64, 0), missing.next_attempt_us); + try testing.expectEqual(@as(i64, 100), missing.completed_us); + try testing.expectEqual(@as(i64, 8), missing.updated_us); + try testing.expectEqualStrings("initial", missing.rev); + try testing.expectEqualStrings("latest", missing.latest_rev); + try testing.expectEqualStrings("", missing.last_error); + try testing.expectEqualStrings("alice.test", missing.handle); + + const unavailable = terminalState(previous, .unavailable, true, "", 101); + try testing.expectEqual(@as(i64, 5), unavailable.completed_us); + try testing.expectEqual(@as(i64, 101), unavailable.last_attempt_us); + try testing.expectEqualStrings("old.example", unavailable.host); + try testing.expectEqual(@as(u32, 0), unavailable.attempts); + try testing.expectEqual(@as(i64, 9), unavailable.record_count); + try testing.expectEqual(@as(i64, 10), unavailable.total_bytes); +} + test "formatRfc3339" { var buf: [40]u8 = undefined; // 1783900800 = 2026-07-13T00:00:00Z diff --git a/src/internal/bootstrap/retry.zig b/src/internal/bootstrap/retry.zig index 768a22b..138a8fd 100644 --- a/src/internal/bootstrap/retry.zig +++ b/src/internal/bootstrap/retry.zig @@ -242,7 +242,10 @@ const RetryWork = struct { .status = .failed, .active = state.active, .rev = state.rev, + .latest_rev = state.latest_rev, .host = final_host, + .handle = state.handle, + .pds = state.pds, .last_error = @errorName(err), .error_class = details.error_class, .attempts = state.attempts +| 1, @@ -250,6 +253,10 @@ const RetryWork = struct { .started_us = state.started_us, .retry_count = count, .next_attempt_us = next_attempt, + .completed_us = state.completed_us, + .updated_us = state.updated_us, + .record_count = state.record_count, + .total_bytes = state.total_bytes, }); if (err == error.RateLimited) try self.retry.saveHostPark(final_host, next_attempt); if (self.retry.emit_failure_logs) { diff --git a/src/main.zig b/src/main.zig index 04a7f3c..1053d20 100644 --- a/src/main.zig +++ b/src/main.zig @@ -1,6 +1,7 @@ //! stream — a zig jetstream. entrypoint: parse args, wire components. const std = @import("std"); +const zat = @import("zat"); const websocket = @import("websocket"); const archive_mod = @import("internal/archive.zig"); const lifecycle = @import("internal/bootstrap/lifecycle.zig"); @@ -34,6 +35,34 @@ const default_tail_bytes: usize = 256 * 1024 * 1024; /// (docs/lessons-from-zlay.md #1) const default_stack_size: usize = 8 * 1024 * 1024; +fn parseBackfillRepos(allocator: std.mem.Allocator, raw: []const u8) ![]const []const u8 { + if (std.mem.trim(u8, raw, " \t\r\n").len == 0) return &.{}; + var out: std.ArrayList([]const u8) = .empty; + errdefer out.deinit(allocator); + var it = std.mem.splitScalar(u8, raw, ','); + while (it.next()) |part| { + const did = std.mem.trim(u8, part, " \t\r\n"); + if (did.len == 0 or zat.Did.parse(did) == null) return error.InvalidBackfillRepo; + for (out.items) |prior| if (std.mem.eql(u8, prior, did)) return error.DuplicateBackfillRepo; + try out.append(allocator, did); + } + if (out.items.len == 0) return error.InvalidBackfillRepo; + return out.toOwnedSlice(allocator); +} + +test "explicit backfill repo parsing matches upstream selection rules" { + const testing = std.testing; + const parsed = try parseBackfillRepos(testing.allocator, "did:plc:aaa, did:web:example.com"); + defer testing.allocator.free(parsed); + try testing.expectEqual(@as(usize, 2), parsed.len); + try testing.expectEqualStrings("did:plc:aaa", parsed[0]); + try testing.expectEqualStrings("did:web:example.com", parsed[1]); + try testing.expectEqual(@as(usize, 0), (try parseBackfillRepos(testing.allocator, " \t ")).len); + try testing.expectError(error.DuplicateBackfillRepo, parseBackfillRepos(testing.allocator, "did:plc:aaa,did:plc:aaa")); + try testing.expectError(error.InvalidBackfillRepo, parseBackfillRepos(testing.allocator, "did:plc:aaa,")); + try testing.expectError(error.InvalidBackfillRepo, parseBackfillRepos(testing.allocator, "not-a-did")); +} + pub fn main(init: std.process.Init.Minimal) !void { const allocator = std.heap.c_allocator; @@ -52,6 +81,7 @@ pub fn main(init: std.process.Init.Minimal) !void { var do_verify = true; var max_segment_bytes: usize = archive_mod.default_max_segment_bytes; var backfill_max_repos: usize = 0; + var backfill_repos_raw: ?[]const u8 = null; var do_backfill = false; var backfill_workers: usize = backfill_engine.default_workers; var backfill_max_inflight_bytes: usize = backfill_engine.default_max_inflight_bytes; @@ -80,6 +110,8 @@ pub fn main(init: std.process.Init.Minimal) !void { max_segment_bytes = try std.fmt.parseInt(usize, arg["--max-segment-bytes=".len..], 10); } else if (std.mem.startsWith(u8, arg, "--backfill-max-repos=")) { backfill_max_repos = try std.fmt.parseInt(usize, arg["--backfill-max-repos=".len..], 10); + } else if (std.mem.startsWith(u8, arg, "--backfill-repos=")) { + backfill_repos_raw = arg["--backfill-repos=".len..]; } else if (std.mem.startsWith(u8, arg, "--backfill-workers=")) { backfill_workers = try std.fmt.parseInt(usize, arg["--backfill-workers=".len..], 10); } else if (std.mem.startsWith(u8, arg, "--backfill-max-inflight-bytes=")) { @@ -111,6 +143,9 @@ pub fn main(init: std.process.Init.Minimal) !void { return error.BadArgs; } } + const backfill_repos: []const []const u8 = if (backfill_repos_raw) |raw| try parseBackfillRepos(allocator, raw) else &.{}; + defer if (backfill_repos.len > 0) allocator.free(backfill_repos); + if (backfill_repos.len > 0 and backfill_max_repos > 0) return error.ConflictingBackfillSelection; var hot_tail = tail_mod.Tail.init(allocator, io, default_tail_bytes); defer hot_tail.deinit(); @@ -122,7 +157,7 @@ pub fn main(init: std.process.Init.Minimal) !void { .io = io, .tail = &hot_tail, .stats = &stats, - .bootstrap_enabled = do_backfill, + .bootstrap_enabled = do_backfill or backfill_max_repos > 0 or backfill_repos.len > 0, .compaction_enabled = compaction_interval_s > 0, .retry_enabled = retry_interval_s > 0, }; @@ -239,13 +274,15 @@ pub fn main(init: std.process.Init.Minimal) !void { break :blk ps.read(io); }; const mid_lifecycle = if (stored_phase) |ph| ph != .steady_state else false; - if (mid_lifecycle or do_backfill or backfill_max_repos > 0) { + if (mid_lifecycle or do_backfill or backfill_max_repos > 0 or backfill_repos.len > 0) { hub.serving.store(false, .release); try lifecycle.run(allocator, io, .{ .data_dir = data_dir, .upstream = upstream, .relay_http = relay_http, + .plc_url = plc_url, .max_repos = backfill_max_repos, + .selected_repos = backfill_repos, .workers = backfill_workers, .max_inflight_bytes = backfill_max_inflight_bytes, }, &archive, &meta, &cursor_store, if (verifier) |*v| v else null, &stats); diff --git a/src/tests.zig b/src/tests.zig index 04ef7b5..1caaed4 100644 --- a/src/tests.zig +++ b/src/tests.zig @@ -17,6 +17,9 @@ comptime { _ = @import("internal/bootstrap/repos.zig"); _ = @import("internal/bootstrap/fetched_car.zig"); _ = @import("internal/bootstrap/prepared_fetch.zig"); + // Import directly: Zig does not collect a module's test declarations when + // it is reached only transitively through main.zig. + _ = @import("internal/bootstrap/engine.zig"); _ = @import("internal/bootstrap/merge.zig"); _ = @import("internal/bootstrap/resync.zig"); _ = @import("internal/bootstrap/retry.zig"); -- 2.54.0