From 1c76e0554c1bc23c1cc5fc3d4bb703e7cc085d7f Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Sun, 19 Jul 2026 11:23:16 -0500 Subject: [PATCH] close account diagnostics parity --- README.md | 15 +- build.zig.zon | 8 +- docs/semantic-parity.md | 2 +- docs/upstream-harness.md | 12 +- justfile | 2 +- src/internal/repo_action_limiter.zig | 118 +++++ src/internal/repo_export.zig | 644 +++++++++++++++++++++++++++ src/internal/server.zig | 276 +++++++++++- src/internal/status_page.zig | 442 +++++++++++++++++- src/main.zig | 7 + src/tests.zig | 2 + tests/status_contract.py | 40 +- 12 files changed, 1546 insertions(+), 22 deletions(-) create mode 100644 src/internal/repo_action_limiter.zig create mode 100644 src/internal/repo_export.zig diff --git a/README.md b/README.md index bc39be4..7b45cc5 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,9 @@ the live, phone-sized field report is at - crash-resumable timestamp imports with a durable RocksDB rule map, topology-preserving sealed-segment repair, and bearer-gated operator XRPC - browser and terminal operations views at `/status`, including durable PDS - diagnostics at `/status?tab=hosts`, plus Prometheus metrics at `/metrics` + diagnostics at `/status?tab=hosts` and per-repository reconstruction and + authoritative root verification at `/status?tab=accounts`, plus Prometheus + metrics at `/metrics` Stream is still converging on complete Jetstream V2 parity. The precise remaining deployment gates live in [`docs/semantic-parity.md`](docs/semantic-parity.md). @@ -92,8 +94,15 @@ 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. +refreshes. Account lookup resolves handles through the configured identity +directory, reconstructs the repository from sealed JSS segments, the active +writer, pending rows, and bootstrap live segments at a rotation-safe boundary, +then compares the canonical local MST root with the PDS commit root obtained +through Sync 1.1 `getLatestCommit` and `getBlocks`. Verification runs +automatically for a found account. Matching upstream, expensive repo actions +are limited to four per source IP per minute with a bounded 4,096-source map; +trusted operator deployments can explicitly pass +`--disable-repo-action-rate-limits`. Timestamp import is disabled to remote callers unless a non-empty bearer token is configured. Stage plain CSVs under the confined import directory (default diff --git a/build.zig.zon b/build.zig.zon index 858569e..1c1b31b 100644 --- a/build.zig.zon +++ b/build.zig.zon @@ -5,12 +5,12 @@ .minimum_zig_version = "0.16.0", .dependencies = .{ .zat = .{ - .url = "git+https://tangled.org/zat.dev/zat#7c447c0bdcb7e31d68513080c013ed59a2164060", - .hash = "zat-0.3.16-5PuC7o8pCwD9p78jkB1jXhpNoouP8u7-Ph9APtn7Y0lQ", + .url = "git+https://tangled.org/zat.dev/zat#5ef2d7712f8776f08d426503ce26a99a445b9327", + .hash = "zat-0.3.17-5PuC7owpCwDY__0ulmTTtBV0uApYFGjp5jjXvpPsyzHv", }, .websocket = .{ - .url = "https://tangled.org/zzstoatzz.io/websocket.zig/archive/bf9de23.tar.gz", - .hash = "websocket-0.1.9-ZPISdfjABACZlbEjUhDGnNlsutPGgtlHsvxD-Vs3ojnc", + .url = "https://github.com/zzstoatzz/websocket.zig/archive/d8e6149.tar.gz", + .hash = "websocket-0.1.10-ZPISdSDLBABPDTfutnraAXkxog3PDrDe-6HaVteRt7CF", }, .zstd = .{ .url = "https://github.com/facebook/zstd/releases/download/v1.5.7/zstd-1.5.7.tar.gz", diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index 818eb61..ea23c2d 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 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.** | +| Status/diagnostics | Repo attempts/error class/final PDS and per-host aggregates survive restart and are operator-queryable | **closed at the implementation boundary.** Durable v5 repo rows distinguish initial Backfill.Rev from latest Rev/UpdatedAt, retain resolved Handle/PDS and reserved record/byte fields, and atomically maintain `handle/`. Merge source cursor commits refresh latest revisions without mutating backfill watermarks. Same-batch `host/` totals, active/status counts, cumulative error classes, and five bounded recent samples survive restart; host moves and active flips are atomic with the source repo row. `/status?tab=hosts` reads those rows without a whole-network scan and matches upstream ordering and error presentation. Account queries preserve resolver-first handle semantics, the legacy `did`/`handle` aliases, and missing-identity hydration, then reconstruct records last-writer-wins from checksum-verified, DID-bloom-pruned sealed JSS plus rotation-safe active/pending and bootstrap-live snapshots. The resulting canonical MST is compared with the PDS commit root through real Sync 1.1 `getLatestCommit`/`getBlocks`; the page renders the same match fields and error state as upstream. `HEAD` performs lookup without verification; successful status responses are `no-store` and carry a generation timestamp. Expensive verification uses upstream's exact 4/source-IP/minute fixed-window limiter, 4,096-entry bound, stale pruning, oldest eviction, and explicit disable option. The offline receipt uses a real HTTP Stream route, real RocksDB row, real JSS archive, signed commit, and loopback PLC/PDS; four separate connections verify successfully, a fifth changing only its ephemeral port is blocked without another upstream request, HEAD remains side-effect-free, and disabling the limiter performs a new authoritative verification. | | Prometheus/Grafana | Every canonical dashboard query maps to a real, same-semantics producer; runtime-specific panels use honest Zig/process metrics | 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-harness.md b/docs/upstream-harness.md index f548557..bbd5bac 100644 --- a/docs/upstream-harness.md +++ b/docs/upstream-harness.md @@ -73,7 +73,7 @@ Zig's checksum-pinned dependencies must already be in its global cache. A missing pin, Go toolchain/module, Python environment, row, sentinel, or cursor transition fails the command rather than weakening the assertion. -## host diagnostics contract +## status diagnostics contract `just status-contract` seeds repository transitions through the real RocksDB repo store, closes the seed process, and reopens the same data directory with @@ -83,6 +83,10 @@ current counts and bounded error history, safe escaping, and the browser filter contract. The seed includes a healthy larger host and a smaller host with repeated failures, so either sort or history silently disappearing makes the receipt fail. Browser checks additionally exercise the filter and the -horizontal table layout at desktop and phone-sized viewports. Account links -are intentionally absent until the upstream-equivalent account lookup and -automatic repo-verification path exists. +horizontal table layout at desktop and phone-sized viewports. The same +restarted binary is queried through the Account tab and legacy `did` alias, +including durable fields, unknown identity presentation, automatic real +verification failure presentation, no-store headers, and side-effect-free +HEAD. A separate production-route test uses a real JSS archive and signed +commit data from loopback PLC/PDS endpoints to prove successful verification +and the port-free per-source-IP limiter boundary. diff --git a/justfile b/justfile index 3b5058a..609d089 100644 --- a/justfile +++ b/justfile @@ -53,7 +53,7 @@ archive-contract: STREAM_UPSTREAM_REPO='{{upstream}}' python3 tests/archive_client_e2e.py # Seed the real RocksDB repo/host schema, reopen it through the production -# binary, and exercise the public host view without network access. +# binary, and exercise the public host/account views without network access. status-contract: #!/usr/bin/env bash set -euo pipefail diff --git a/src/internal/repo_action_limiter.zig b/src/internal/repo_action_limiter.zig new file mode 100644 index 0000000..8abaa24 --- /dev/null +++ b/src/internal/repo_action_limiter.zig @@ -0,0 +1,118 @@ +//! Fixed-window per-source-IP limiter for expensive status-page repo actions. +//! Semantics match upstream Jetstream V2's internal/web implementation. + +const std = @import("std"); + +const Io = std.Io; +const Allocator = std.mem.Allocator; + +pub const max_entries: usize = 4096; +pub const default_limit: usize = 4; +pub const default_window_us: i64 = std.time.us_per_min; + +const Bucket = struct { + window_start_us: ?i64 = null, + last_seen_us: i64, + used: usize = 0, +}; + +pub const Limiter = struct { + mutex: Io.Mutex = .init, + entries: std.StringHashMapUnmanaged(Bucket) = .empty, + limit: usize = default_limit, + window_us: i64 = default_window_us, + + pub fn deinit(self: *Limiter, allocator: Allocator) void { + var it = self.entries.keyIterator(); + while (it.next()) |key| allocator.free(key.*); + self.entries.deinit(allocator); + self.* = .{}; + } + + pub fn allow(self: *Limiter, allocator: Allocator, io: Io, ip: []const u8, now_us: i64) !bool { + self.mutex.lockUncancelable(io); + defer self.mutex.unlock(io); + + if (self.entries.count() >= max_entries) self.pruneLocked(allocator, now_us); + if (self.entries.count() >= max_entries) self.evictOldestLocked(allocator); + + const bucket = self.entries.getPtr(ip) orelse bucket: { + const owned_ip = try allocator.dupe(u8, ip); + errdefer allocator.free(owned_ip); + try self.entries.put(allocator, owned_ip, .{ .last_seen_us = now_us }); + break :bucket self.entries.getPtr(owned_ip).?; + }; + if (bucket.window_start_us == null or now_us - bucket.window_start_us.? >= self.window_us) { + bucket.window_start_us = now_us; + bucket.used = 0; + } + bucket.last_seen_us = now_us; + if (bucket.used >= self.limit) return false; + bucket.used += 1; + return true; + } + + fn pruneLocked(self: *Limiter, allocator: Allocator, now_us: i64) void { + var it = self.entries.iterator(); + while (it.next()) |entry| { + if (now_us - entry.value_ptr.last_seen_us < 2 * self.window_us) continue; + const owned_ip = entry.key_ptr.*; + _ = self.entries.remove(owned_ip); + allocator.free(owned_ip); + it = self.entries.iterator(); + } + } + + fn evictOldestLocked(self: *Limiter, allocator: Allocator) void { + var oldest_ip: ?[]const u8 = null; + var oldest_seen: i64 = 0; + var it = self.entries.iterator(); + while (it.next()) |entry| { + if (oldest_ip == null or entry.value_ptr.last_seen_us < oldest_seen) { + oldest_ip = entry.key_ptr.*; + oldest_seen = entry.value_ptr.last_seen_us; + } + } + if (oldest_ip) |ip| { + _ = self.entries.remove(ip); + allocator.free(ip); + } + } +}; + +test "repo action limiter matches upstream fixed-window behavior" { + const testing = std.testing; + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var limiter: Limiter = .{}; + defer limiter.deinit(testing.allocator); + + for (0..default_limit) |_| try testing.expect(try limiter.allow(testing.allocator, io, "203.0.113.10", 1)); + try testing.expect(!try limiter.allow(testing.allocator, io, "203.0.113.10", default_window_us - 1)); + try testing.expect(try limiter.allow(testing.allocator, io, "203.0.113.11", default_window_us - 1)); + try testing.expect(try limiter.allow(testing.allocator, io, "203.0.113.10", default_window_us + 1)); +} + +test "repo action limiter prunes stale entries and bounds cardinality" { + const testing = std.testing; + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var limiter: Limiter = .{ .window_us = 1_000_000_000 }; + defer limiter.deinit(testing.allocator); + + var ip_buf: [64]u8 = undefined; + for (0..max_entries) |index| { + const ip = try std.fmt.bufPrint(&ip_buf, "source-{d}", .{index}); + try testing.expect(try limiter.allow(testing.allocator, io, ip, @intCast(index))); + } + try testing.expectEqual(max_entries, limiter.entries.count()); + try testing.expect(try limiter.allow(testing.allocator, io, "newest", @intCast(max_entries))); + try testing.expectEqual(max_entries, limiter.entries.count()); + try testing.expect(!limiter.entries.contains("source-0")); + + limiter.window_us = 1; + try testing.expect(try limiter.allow(testing.allocator, io, "after-prune", 2 * @as(i64, max_entries) + 10)); + try testing.expectEqual(@as(usize, 1), limiter.entries.count()); +} diff --git a/src/internal/repo_export.zig b/src/internal/repo_export.zig new file mode 100644 index 0000000..5ed723a --- /dev/null +++ b/src/internal/repo_export.zig @@ -0,0 +1,644 @@ +//! Reconstruct and verify one repository from Stream's real JSS archive. +//! +//! This is the Zig equivalent of upstream `internal/repoexport`: candidate +//! sealed segments are pruned by their resident-format DID blooms, the active +//! writer and its pending block are snapshotted under the archive lock before +//! sealed selection (closing the rotation race), and repository records are +//! folded last-writer-wins into a canonical AT Protocol MST. + +const std = @import("std"); +const zat = @import("zat"); +const archive_mod = @import("archive.zig"); +const gloom = @import("gloom.zig"); +const segment = @import("segment.zig"); +const writer_mod = @import("segment_writer.zig"); + +const Io = std.Io; +const Allocator = std.mem.Allocator; + +pub const no_local_repo_message = "repoexport: no local commit events for DID"; +pub const root_mismatch_message = "local reconstructed MST root does not match authoritative repo root"; + +pub const Snapshot = struct { + did: []const u8, + latest_rev: []const u8, + root: zat.cbor.Cid, + record_count: usize, +}; + +pub const VerifyReport = struct { + did: []const u8, + match: bool = false, + authoritative_rev: []const u8 = "", + authoritative_root: []const u8 = "", + local_latest_rev: []const u8 = "", + local_root: []const u8 = "", + local_record_count: usize = 0, + message: []const u8 = "", +}; + +/// Compare the PDS's signed commit data root with a reconstruction from the +/// local event archive. The authoritative side deliberately uses the same +/// two PDS-only Sync 1.1 calls as upstream: getLatestCommit, then getBlocks +/// for that commit CID. Only the matching, hash-valid commit block among the +/// first eight CAR blocks is accepted. +pub fn verify( + allocator: Allocator, + io: Io, + archive: *archive_mod.Archive, + data_dir: []const u8, + plc_url: []const u8, + did: []const u8, +) !VerifyReport { + const parsed_did = zat.Did.parse(did) orelse return error.InvalidDid; + var resolver = zat.DidResolver.init(io, allocator); + defer resolver.deinit(); + resolver.plc_url = plc_url; + var doc = try resolver.resolve(parsed_did); + defer doc.deinit(); + const endpoint = std.mem.trimEnd(u8, doc.pdsEndpoint() orelse return error.NoAuthoritativePDS, "/"); + if (endpoint.len == 0) return error.NoAuthoritativePDS; + + const latest_url = try xrpcUrlAlloc(allocator, endpoint, "com.atproto.sync.getLatestCommit", &.{.{ .name = "did", .value = did }}); + defer allocator.free(latest_url); + var transport = zat.HttpTransport.init(io, allocator); + defer transport.deinit(); + const latest_response = try transport.fetch(.{ .url = latest_url, .max_response_size = 64 * 1024, .redirect_behavior = .not_allowed }); + defer allocator.free(latest_response.body); + if (latest_response.status.class() != .success) return error.GetLatestCommitFailed; + const latest_json = try std.json.parseFromSlice(std.json.Value, allocator, latest_response.body, .{}); + defer latest_json.deinit(); + const rev_raw = zat.json.getString(latest_json.value, "rev") orelse return error.InvalidLatestCommit; + const cid_raw = zat.json.getString(latest_json.value, "cid") orelse return error.InvalidLatestCommit; + const rev = try allocator.dupe(u8, rev_raw); + const commit_cid_string = try allocator.dupe(u8, cid_raw); + const commit_cid = try zat.mst.parseCidString(allocator, commit_cid_string); + + const blocks_url = try xrpcUrlAlloc(allocator, endpoint, "com.atproto.sync.getBlocks", &.{ + .{ .name = "did", .value = did }, + .{ .name = "cids", .value = commit_cid_string }, + }); + defer allocator.free(blocks_url); + const blocks_response = try transport.fetch(.{ + .url = blocks_url, + .max_response_size = 8 * (1024 * 1024 + 128) + 64 * 1024, + .redirect_behavior = .not_allowed, + }); + defer allocator.free(blocks_response.body); + if (blocks_response.status.class() != .success) return error.GetCommitBlockFailed; + const commit_data = try findAuthoritativeCommitBlock(allocator, blocks_response.body, commit_cid); + + const roots = [_]zat.cbor.Cid{commit_cid}; + const blocks = [_]zat.car.Block{.{ .cid_raw = commit_cid.raw, .data = commit_data }}; + const single_car = try zat.car.writeAlloc(allocator, .{ .roots = &roots, .blocks = &blocks }); + defer allocator.free(single_car); + const loaded = try zat.loadCommitFromCAR(allocator, single_car); + if (!std.mem.eql(u8, loaded.commit.did, did)) return error.AuthoritativeCommitDidMismatch; + if (!std.mem.eql(u8, loaded.commit.rev, rev)) return error.AuthoritativeCommitRevMismatch; + const authoritative_root = try zat.multibase.base32lower.encode(allocator, loaded.commit.data_cid); + + var report: VerifyReport = .{ + .did = did, + .authoritative_rev = rev, + .authoritative_root = authoritative_root, + }; + const local = reconstruct(allocator, io, archive, data_dir, did) catch |err| switch (err) { + error.NoLocalRepo => { + report.message = try std.fmt.allocPrint(allocator, "{s}: {s}", .{ no_local_repo_message, did }); + return report; + }, + else => return err, + }; + defer freeSnapshot(allocator, local); + report.local_latest_rev = try allocator.dupe(u8, local.latest_rev); + report.local_root = try zat.multibase.base32lower.encode(allocator, local.root.raw); + report.local_record_count = local.record_count; + report.match = std.mem.eql(u8, report.authoritative_root, report.local_root); + if (!report.match) report.message = root_mismatch_message; + return report; +} + +const QueryParam = struct { name: []const u8, value: []const u8 }; + +fn xrpcUrlAlloc(allocator: Allocator, endpoint: []const u8, nsid: []const u8, params: []const QueryParam) ![]u8 { + var out: std.ArrayList(u8) = .empty; + errdefer out.deinit(allocator); + try out.print(allocator, "{s}/xrpc/{s}", .{ endpoint, nsid }); + for (params, 0..) |param, index| { + try out.append(allocator, if (index == 0) '?' else '&'); + try out.appendSlice(allocator, param.name); + try out.append(allocator, '='); + for (param.value) |byte| { + if (std.ascii.isAlphanumeric(byte) or byte == '-' or byte == '_' or byte == '.' or byte == '~') + try out.append(allocator, byte) + else + try out.print(allocator, "%{X:0>2}", .{byte}); + } + } + return out.toOwnedSlice(allocator); +} + +fn findAuthoritativeCommitBlock(allocator: Allocator, car_bytes: []const u8, wanted: zat.cbor.Cid) ![]const u8 { + var pos: usize = 0; + const header_len_u64 = zat.cbor.readUvarint(car_bytes, &pos) orelse return error.InvalidAuthoritativeCar; + const header_len = std.math.cast(usize, header_len_u64) orelse return error.InvalidAuthoritativeCar; + if (header_len > 1024 * 1024 or pos + header_len > car_bytes.len) return error.InvalidAuthoritativeCar; + var header_arena = std.heap.ArenaAllocator.init(allocator); + defer header_arena.deinit(); + const header = try zat.cbor.decodeAll(header_arena.allocator(), car_bytes[pos..][0..header_len]); + if (header.getUint("version") != 1) return error.InvalidAuthoritativeCar; + pos += header_len; + var blocks_read: usize = 0; + while (pos < car_bytes.len and blocks_read < 8) : (blocks_read += 1) { + const block_len_u64 = zat.cbor.readUvarint(car_bytes, &pos) orelse return error.InvalidAuthoritativeCar; + const block_len = std.math.cast(usize, block_len_u64) orelse return error.InvalidAuthoritativeCar; + if (block_len == 0 or block_len > 1024 * 1024 or pos + block_len > car_bytes.len) return error.InvalidAuthoritativeCar; + const raw = car_bytes[pos..][0..block_len]; + const cid_len = cidLength(raw) orelse return error.InvalidAuthoritativeCar; + if (cid_len > raw.len) return error.InvalidAuthoritativeCar; + const claimed: zat.cbor.Cid = .{ .raw = raw[0..cid_len] }; + const data = raw[cid_len..]; + if (claimed.version() != 1 or claimed.hashFn() != zat.cbor.HashFn.sha2_256) return error.InvalidAuthoritativeCar; + const computed = try zat.cbor.Cid.create(header_arena.allocator(), 1, claimed.codec() orelse return error.InvalidAuthoritativeCar, zat.cbor.HashFn.sha2_256, data); + if (!std.mem.eql(u8, computed.raw, claimed.raw)) return error.AuthoritativeBlockCidMismatch; + pos += block_len; + if (std.mem.eql(u8, claimed.raw, wanted.raw)) return data; + } + // Upstream classifies any eight non-matching blocks as the bounded-search + // limit, even when the eighth block is also the end of the response. + if (blocks_read >= 8) return error.AuthoritativeCommitBlockLimit; + return error.AuthoritativeCommitBlockNotFound; +} + +fn cidLength(data: []const u8) ?usize { + if (data.len < 2) return null; + if (data[0] == 0x12 and data[1] == 0x20) return if (data.len >= 34) 34 else null; + var pos: usize = 0; + _ = zat.cbor.readUvarint(data, &pos) orelse return null; + _ = zat.cbor.readUvarint(data, &pos) orelse return null; + _ = zat.cbor.readUvarint(data, &pos) orelse return null; + const digest_len = zat.cbor.readUvarint(data, &pos) orelse return null; + const digest_len_usize = std.math.cast(usize, digest_len) orelse return null; + if (pos + digest_len_usize > data.len) return null; + return pos + digest_len_usize; +} + +const ReplayState = struct { + allocator: Allocator, + did: []const u8, + records: std.StringHashMap([]const u8), + latest_rev: []const u8 = "", + seen_commit: bool = false, + + fn init(allocator: Allocator, did: []const u8) ReplayState { + return .{ .allocator = allocator, .did = did, .records = std.StringHashMap([]const u8).init(allocator) }; + } + + fn deinit(self: *ReplayState) void { + var it = self.records.iterator(); + while (it.next()) |entry| { + self.allocator.free(entry.key_ptr.*); + self.allocator.free(entry.value_ptr.*); + } + self.records.deinit(); + if (self.latest_rev.len > 0) self.allocator.free(@constCast(self.latest_rev)); + } + + fn apply(self: *ReplayState, ev: segment.Event, watermark: []const u8) !void { + if (!std.mem.eql(u8, ev.did, self.did)) return; + switch (ev.kind) { + .create, .update, .create_resync => { + if (watermark.len > 0 and std.mem.order(u8, ev.rev, watermark) != .gt) return; + const key = try std.fmt.allocPrint(self.allocator, "{s}/{s}", .{ ev.collection, ev.rkey }); + errdefer self.allocator.free(key); + const payload = try self.allocator.dupe(u8, ev.payload); + errdefer self.allocator.free(payload); + if (self.records.getPtr(key)) |old_payload| { + self.allocator.free(key); + self.allocator.free(@constCast(old_payload.*)); + old_payload.* = payload; + } else { + try self.records.put(key, payload); + } + }, + .delete => { + if (watermark.len > 0 and std.mem.order(u8, ev.rev, watermark) != .gt) return; + const key = try std.fmt.allocPrint(self.allocator, "{s}/{s}", .{ ev.collection, ev.rkey }); + defer self.allocator.free(key); + if (self.records.fetchRemove(key)) |entry| { + self.allocator.free(entry.key); + self.allocator.free(entry.value); + } + }, + else => return, + } + self.seen_commit = true; + if (self.latest_rev.len > 0) self.allocator.free(@constCast(self.latest_rev)); + self.latest_rev = try self.allocator.dupe(u8, ev.rev); + } +}; + +const ActiveSnapshot = struct { + bytes: []u8, + pending: []segment.Event, + + fn deinit(self: *ActiveSnapshot, allocator: Allocator) void { + allocator.free(self.bytes); + for (self.pending) |ev| { + allocator.free(@constCast(ev.did)); + allocator.free(@constCast(ev.collection)); + allocator.free(@constCast(ev.rkey)); + allocator.free(@constCast(ev.rev)); + allocator.free(@constCast(ev.payload)); + } + allocator.free(self.pending); + } +}; + +/// Reconstruct at a rotation-safe point in time. `data_dir` is used only for +/// the bootstrap live archive, which is absent after steady-state cleanup. +pub fn reconstruct(allocator: Allocator, io: Io, archive: *archive_mod.Archive, data_dir: []const u8, did: []const u8) !Snapshot { + if (zat.Did.parse(did) == null) return error.InvalidDid; + var state = ReplayState.init(allocator, did); + defer state.deinit(); + + // Upstream's critical ordering: active paths before sealed selection. We + // make the boundary stronger by copying the active frames and pending rows + // together under the writer lock, then scan the sealed directory. + var active = try snapshotActive(allocator, io, archive); + defer active.deinit(allocator); + + var names: std.ArrayList([]u8) = .empty; + defer { + for (names.items) |name| allocator.free(name); + names.deinit(allocator); + } + var it = archive.dir.iterate(); + while (try it.next(io)) |entry| { + if (archive_mod.parseSegmentIndex(entry.name) == null) continue; + try names.append(allocator, try allocator.dupe(u8, entry.name)); + } + std.mem.sort([]u8, names.items, {}, struct { + fn less(_: void, a: []u8, b: []u8) bool { + return std.mem.order(u8, a, b) == .lt; + } + }.less); + for (names.items) |name| try replaySealedCandidate(allocator, io, archive.dir, name, &state); + try replayActiveBytes(allocator, active.bytes, "", &state); + + const primary_watermark = if (state.latest_rev.len > 0) try allocator.dupe(u8, state.latest_rev) else ""; + defer if (primary_watermark.len > 0) allocator.free(@constCast(primary_watermark)); + try replayBootstrapLive(allocator, io, data_dir, primary_watermark, &state); + for (active.pending) |ev| try state.apply(ev, ""); + + if (!state.seen_commit) return error.NoLocalRepo; + var tree_arena = std.heap.ArenaAllocator.init(allocator); + defer tree_arena.deinit(); + const tree_allocator = tree_arena.allocator(); + var tree = zat.mst.Mst.init(tree_allocator); + var keys: std.ArrayList([]const u8) = .empty; + defer keys.deinit(allocator); + var key_it = state.records.keyIterator(); + while (key_it.next()) |key| try keys.append(allocator, key.*); + std.mem.sort([]const u8, keys.items, {}, struct { + fn less(_: void, a: []const u8, b: []const u8) bool { + return std.mem.order(u8, a, b) == .lt; + } + }.less); + for (keys.items) |key| { + const cid = try zat.cbor.Cid.forDagCbor(tree_allocator, state.records.get(key).?); + try tree.putBorrowed(key, cid); + } + const root = try tree.rootCid(); + return .{ + .did = try allocator.dupe(u8, did), + .latest_rev = try allocator.dupe(u8, state.latest_rev), + .root = .{ .raw = try allocator.dupe(u8, root.raw) }, + .record_count = state.records.count(), + }; +} + +pub fn freeSnapshot(allocator: Allocator, snapshot: Snapshot) void { + allocator.free(snapshot.did); + allocator.free(snapshot.latest_rev); + allocator.free(snapshot.root.raw); +} + +fn snapshotActive(allocator: Allocator, io: Io, archive: *archive_mod.Archive) !ActiveSnapshot { + archive.mu.lockUncancelable(io); + defer archive.mu.unlock(io); + const bytes = try allocator.dupe(u8, archive.writer.bytes()); + errdefer allocator.free(bytes); + const pending = try allocator.alloc(segment.Event, archive.writer.pending.items.len); + errdefer allocator.free(pending); + var copied: usize = 0; + errdefer for (pending[0..copied]) |ev| { + allocator.free(@constCast(ev.did)); + allocator.free(@constCast(ev.collection)); + allocator.free(@constCast(ev.rkey)); + allocator.free(@constCast(ev.rev)); + allocator.free(@constCast(ev.payload)); + }; + for (archive.writer.pending.items, 0..) |ev, index| { + pending[index] = try dupeEvent(allocator, ev); + copied += 1; + } + return .{ .bytes = bytes, .pending = pending }; +} + +fn dupeEvent(allocator: Allocator, ev: segment.Event) !segment.Event { + const did = try allocator.dupe(u8, ev.did); + errdefer allocator.free(did); + const collection = try allocator.dupe(u8, ev.collection); + errdefer allocator.free(collection); + const rkey = try allocator.dupe(u8, ev.rkey); + errdefer allocator.free(rkey); + const rev = try allocator.dupe(u8, ev.rev); + errdefer allocator.free(rev); + const payload = try allocator.dupe(u8, ev.payload); + return .{ + .seq = ev.seq, + .witnessed_at = ev.witnessed_at, + .indexed_at = ev.indexed_at, + .kind = ev.kind, + .did = did, + .collection = collection, + .rkey = rkey, + .rev = rev, + .payload = payload, + }; +} + +fn replaySealedCandidate(allocator: Allocator, io: Io, dir: Io.Dir, name: []const u8, state: *ReplayState) !void { + var file = try dir.openFile(io, name, .{}); + defer file.close(io); + var header_bytes: [segment.header_size]u8 = undefined; + if (try file.readPositionalAll(io, &header_bytes, 0) != header_bytes.len) return error.Truncated; + const header = segment.Header.decode(&header_bytes) catch |err| switch (err) { + error.ActiveSegment => return, + else => return err, + }; + if (header.block_did_bloom_offset < header.did_bloom_offset) return error.BadFooter; + const bloom_len = header.block_did_bloom_offset - header.did_bloom_offset; + if (bloom_len > 1 << 30) return error.LimitExceeded; + const bloom_bytes = try allocator.alloc(u8, @intCast(bloom_len)); + defer allocator.free(bloom_bytes); + if (try file.readPositionalAll(io, bloom_bytes, header.did_bloom_offset) != bloom_bytes.len) return error.Truncated; + var filter = try gloom.unmarshal(allocator, bloom_bytes); + defer filter.deinit(allocator); + if (!filter.contains(state.did)) return; + + const bytes = try dir.readFileAlloc(io, name, allocator, .limited(1 << 31)); + defer allocator.free(bytes); + var sealed = try segment.Sealed.parse(allocator, bytes); + defer sealed.deinit(allocator); + const blobs = try sealed.blockBloomBlobs(allocator); + defer allocator.free(blobs); + for (blobs, 0..) |blob, index| { + var block_filter = gloom.unmarshal(allocator, blob) catch null; + if (block_filter) |*candidate| { + defer candidate.deinit(allocator); + if (!candidate.contains(state.did)) continue; + } + var block = try sealed.readBlock(allocator, index); + defer block.deinit(allocator); + for (block.events) |ev| try state.apply(ev, ""); + } +} + +fn replayActiveBytes(allocator: Allocator, bytes: []const u8, watermark: []const u8, state: *ReplayState) !void { + var it: writer_mod.ActiveIterator = .{ .bytes = bytes }; + while (try it.next(allocator)) |block_value| { + var block = block_value; + defer block.deinit(allocator); + for (block.events) |ev| try state.apply(ev, watermark); + } +} + +fn replayBootstrapLive(allocator: Allocator, io: Io, data_dir: []const u8, watermark: []const u8, state: *ReplayState) !void { + var path_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const path = std.fmt.bufPrint(&path_buf, "{s}/backfill/live/segments", .{data_dir}) catch return error.NameTooLong; + var dir = Io.Dir.cwd().openDir(io, path, .{ .iterate = true }) catch |err| switch (err) { + error.FileNotFound => return, + else => return err, + }; + defer dir.close(io); + var names: std.ArrayList([]u8) = .empty; + defer { + for (names.items) |name| allocator.free(name); + names.deinit(allocator); + } + var it = dir.iterate(); + while (try it.next(io)) |entry| if (archive_mod.parseSegmentIndex(entry.name) != null) + try names.append(allocator, try allocator.dupe(u8, entry.name)); + std.mem.sort([]u8, names.items, {}, struct { + fn less(_: void, a: []u8, b: []u8) bool { + return std.mem.order(u8, a, b) == .lt; + } + }.less); + for (names.items) |name| { + const bytes = try dir.readFileAlloc(io, name, allocator, .limited(1 << 31)); + defer allocator.free(bytes); + if (segment.Header.decode(bytes)) |_| { + var sealed = try segment.Sealed.parse(allocator, bytes); + defer sealed.deinit(allocator); + for (sealed.block_index, 0..) |_, index| { + var block = try sealed.readBlock(allocator, index); + defer block.deinit(allocator); + for (block.events) |ev| try state.apply(ev, watermark); + } + } else |err| switch (err) { + error.ActiveSegment => try replayActiveBytes(allocator, bytes, watermark, state), + else => return err, + } + } +} + +const testing = std.testing; + +fn fixtureEvent(did: []const u8, kind: segment.Kind, rkey: []const u8, rev: []const u8, payload: []const u8) segment.Event { + return .{ + .seq = 0, + .witnessed_at = 1_750_000_000_000_000, + .indexed_at = 0, + .kind = kind, + .did = did, + .collection = "app.bsky.feed.post", + .rkey = rkey, + .rev = rev, + .payload = payload, + }; +} + +test "reconstruct prunes real sealed JSS and includes flushed active plus pending events" { + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var root_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const root_len = try tmp.dir.realPath(io, &root_buf); + const data_dir = root_buf[0..root_len]; + var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir); + defer archive.deinit(); + archive.max_events_per_block = 2; + archive.writer.max_events_per_block = 2; + + const did = "did:plc:reconstructfixture"; + const other = "did:plc:otherfixture"; + const first = "\xa1\x64text\x65first"; + const second = "\xa1\x64text\x66second"; + const latest = "\xa1\x64text\x66latest"; + _ = try archive.append(fixtureEvent(did, .create, "one", "3la", first), 1); + _ = try archive.append(fixtureEvent(other, .create, "other", "3la", second), 1); + try archive.rotate(); + // This pair flushes into the active file. The final row stays only in the + // writer's pending block, exercising all three production sources. + _ = try archive.append(fixtureEvent(did, .update, "one", "3lb", second), 2); + _ = try archive.append(fixtureEvent(other, .update, "other", "3lb", first), 2); + _ = try archive.append(fixtureEvent(did, .create, "two", "3lc", latest), 2); + _ = try archive.append(fixtureEvent(did, .delete, "one", "3ld", ""), 2); + + const snapshot = try reconstruct(testing.allocator, io, &archive, data_dir, did); + defer freeSnapshot(testing.allocator, snapshot); + try testing.expectEqualStrings("3ld", snapshot.latest_rev); + try testing.expectEqual(@as(usize, 1), snapshot.record_count); + + var arena = std.heap.ArenaAllocator.init(testing.allocator); + defer arena.deinit(); + const alloc = arena.allocator(); + var expected = zat.mst.Mst.init(alloc); + const record_cid = try zat.cbor.Cid.forDagCbor(alloc, latest); + try expected.put("app.bsky.feed.post/two", record_cid); + const expected_root = try expected.rootCid(); + try testing.expectEqualSlices(u8, expected_root.raw, snapshot.root.raw); +} + +test "reconstruct distinguishes an absent DID from an empty repository" { + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var root_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const root_len = try tmp.dir.realPath(io, &root_buf); + const data_dir = root_buf[0..root_len]; + var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir); + defer archive.deinit(); + try testing.expectError(error.NoLocalRepo, reconstruct(testing.allocator, io, &archive, data_dir, "did:plc:absentfixture")); + + const did = "did:plc:emptyfixture"; + _ = try archive.append(fixtureEvent(did, .delete, "gone", "3la", ""), 1); + const snapshot = try reconstruct(testing.allocator, io, &archive, data_dir, did); + defer freeSnapshot(testing.allocator, snapshot); + try testing.expectEqual(@as(usize, 0), snapshot.record_count); + try testing.expectEqualStrings("3la", snapshot.latest_rev); +} + +test "verify compares a real JSS reconstruction with real PLC and PDS Sync responses offline" { + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var fixture_arena = std.heap.ArenaAllocator.init(testing.allocator); + defer fixture_arena.deinit(); + const alloc = fixture_arena.allocator(); + + const did = "did:plc:verifyfixture"; + const rev = "3k2abcdefghij"; + const record = try zat.cbor.encodeAlloc(alloc, .{ .map = &.{ + .{ .key = "$type", .value = .{ .text = "app.bsky.feed.post" } }, + .{ .key = "text", .value = .{ .text = "offline verification" } }, + } }); + const record_cid = try zat.cbor.Cid.forDagCbor(alloc, record); + var authoritative_tree = zat.mst.Mst.init(alloc); + try authoritative_tree.put("app.bsky.feed.post/fixture", record_cid); + const data_root = try authoritative_tree.rootCid(); + const keypair = try zat.Keypair.fromSecretKey(.p256, .{41} ** 32); + const signed = try zat.signCommit(alloc, .{ .did = did, .rev = rev, .data = data_root }, &keypair); + const commit_cid_string = try zat.multibase.base32lower.encode(alloc, signed.cid.raw); + const decoy_data = "decoy"; + const decoy_cid = try zat.cbor.Cid.forDagCbor(alloc, decoy_data); + // PDS getBlocks legitimately uses an empty-roots CAR; put a valid decoy + // first to prove the verifier searches rather than trusting block zero. + const get_blocks_car = try zat.car.writeAlloc(alloc, .{ + .roots = &.{}, + .blocks = &.{ + .{ .cid_raw = decoy_cid.raw, .data = decoy_data }, + .{ .cid_raw = signed.cid.raw, .data = signed.bytes }, + }, + }); + + var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; + defer listener.deinit(io); + var endpoint_buf: [96]u8 = undefined; + const endpoint = try std.fmt.bufPrint(&endpoint_buf, "http://127.0.0.1:{d}", .{listener.socket.address.getPort()}); + const did_doc = try std.fmt.allocPrint(alloc, + \\{{"id":"{s}","alsoKnownAs":["at://verify.example"],"verificationMethod":[],"service":[{{"id":"#atproto_pds","type":"AtprotoPersonalDataServer","serviceEndpoint":"{s}"}}]}} + , .{ did, endpoint }); + const latest_json = try std.fmt.allocPrint(alloc, "{{\"cid\":\"{s}\",\"rev\":\"{s}\"}}", .{ commit_cid_string, rev }); + const Fixture = struct { + fn run(server: *Io.net.Server, task_io: Io, doc: []const u8, latest: []const u8, blocks_car: []const u8) !void { + for (0..3) |_| { + 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 = std.http.Server.init(&reader.interface, &writer.interface); + var request = try http.receiveHead(); + const target = request.head.target; + const body, const content_type = if (std.mem.indexOf(u8, target, "getLatestCommit") != null) + .{ latest, "application/json" } + else if (std.mem.indexOf(u8, target, "getBlocks") != null) + .{ blocks_car, "application/vnd.ipld.car" } + else + .{ doc, "application/json" }; + try request.respond(body, .{ + .status = .ok, + .keep_alive = false, + .extra_headers = &.{.{ .name = "content-type", .value = content_type }}, + }); + } + } + }; + var server_future = try io.concurrent(Fixture.run, .{ &listener, io, did_doc, latest_json, get_blocks_car }); + defer _ = server_future.cancel(io) catch {}; + + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var root_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const root_len = try tmp.dir.realPath(io, &root_buf); + const data_dir = root_buf[0..root_len]; + var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir); + defer archive.deinit(); + _ = try archive.append(fixtureEvent(did, .create, "fixture", rev, record), 1); + + var verify_arena = std.heap.ArenaAllocator.init(testing.allocator); + defer verify_arena.deinit(); + const report = try verify(verify_arena.allocator(), io, &archive, data_dir, endpoint, did); + try testing.expect(report.match); + try testing.expectEqualStrings(rev, report.authoritative_rev); + try testing.expectEqualStrings(rev, report.local_latest_rev); + try testing.expectEqual(@as(usize, 1), report.local_record_count); + try testing.expectEqualStrings(report.authoritative_root, report.local_root); + try testing.expectEqualStrings("", report.message); + _ = try server_future.await(io); +} + +test "authoritative commit search matches upstream eight-block boundary" { + var arena = std.heap.ArenaAllocator.init(testing.allocator); + defer arena.deinit(); + const alloc = arena.allocator(); + var blocks: [8]zat.car.Block = undefined; + for (&blocks, 0..) |*block, index| { + const data = try std.fmt.allocPrint(alloc, "decoy-{d}", .{index}); + const cid = try zat.cbor.Cid.forDagCbor(alloc, data); + block.* = .{ .cid_raw = cid.raw, .data = data }; + } + const wanted = try zat.cbor.Cid.forDagCbor(alloc, "wanted"); + const seven = try zat.car.writeAlloc(alloc, .{ .roots = &.{}, .blocks = blocks[0..7] }); + try testing.expectError(error.AuthoritativeCommitBlockNotFound, findAuthoritativeCommitBlock(alloc, seven, wanted)); + const eight = try zat.car.writeAlloc(alloc, .{ .roots = &.{}, .blocks = &blocks }); + try testing.expectError(error.AuthoritativeCommitBlockLimit, findAuthoritativeCommitBlock(alloc, eight, wanted)); +} diff --git a/src/internal/server.zig b/src/internal/server.zig index 084686d..04f7a18 100644 --- a/src/internal/server.zig +++ b/src/internal/server.zig @@ -10,11 +10,14 @@ const std = @import("std"); const websocket = @import("websocket"); +const zat = @import("zat"); const archive_mod = @import("archive.zig"); const cold = @import("cold.zig"); const filter_mod = @import("filter.zig"); const homepage = @import("homepage.zig"); const metrics = @import("metrics.zig"); +const repo_action_limiter = @import("repo_action_limiter.zig"); +const repo_export = @import("repo_export.zig"); const repo_store = @import("bootstrap/repo_store.zig"); const status_page = @import("status_page.zig"); const tail_mod = @import("tail.zig"); @@ -79,6 +82,14 @@ pub const Hub = struct { archive: ?*archive_mod.Archive = null, xrpc: ?*xrpcapi.Api = null, repo_status: ?*repo_store.Store = null, + data_dir: []const u8 = ".", + /// Identity directory used by the public account diagnostics collector. + /// Production always configures it, matching upstream's resolver-first + /// handle semantics; tests may disable resolution explicitly. + plc_url: []const u8 = "https://plc.directory", + status_identity_resolution: bool = true, + repo_action_rate_limits: bool = true, + repo_action_limiter: repo_action_limiter.Limiter = .{}, pipeline_gauges: ?*const fn (*anyopaque) [3]u64 = null, pipeline_ctx: ?*anyopaque = null, active: std.atomic.Value(u32) = .init(0), @@ -89,6 +100,10 @@ pub const Hub = struct { /// steady_state — subscribe + xrpc return 503 (upstream contract); /// /healthz and /metrics stay reachable serving: std.atomic.Value(bool) = .init(true), + + pub fn deinit(self: *Hub) void { + self.repo_action_limiter.deinit(self.allocator); + } }; const Subscriber = struct { @@ -645,7 +660,8 @@ pub const Handler = struct { headers.get("if-range"), )) return; } - if (!std.mem.eql(u8, method, "GET")) return respond(conn, "405 Method Not Allowed", "text/plain", "method not allowed\n"); + const status_head = std.mem.eql(u8, method, "HEAD") and std.mem.eql(u8, path, "/status"); + if (!std.mem.eql(u8, method, "GET") and !status_head) return respond(conn, "405 Method Not Allowed", "text/plain", "method not allowed\n"); if (std.mem.eql(u8, path, "/healthz")) return respond(conn, "200 OK", "text/plain", "ok\n"); if (std.mem.eql(u8, path, "/")) { const now_s: i64 = @divTrunc(Io.Timestamp.now(hub.io, .real).toMicroseconds(), std.time.us_per_s); @@ -664,6 +680,57 @@ pub const Handler = struct { } if (std.mem.eql(u8, path, "/status")) { const wants_html = if (headers.get("accept")) |a| std.mem.indexOf(u8, a, "text/html") != null else false; + const account_query = status_page.accountFromQueryAlloc(hub.allocator, query) catch + return respond(conn, "400 Bad Request", "text/plain", "invalid account query\n"); + if (account_query) |raw_account| { + defer hub.allocator.free(raw_account); + const store = hub.repo_status orelse return respond(conn, "500 Internal Server Error", "text/plain", "account diagnostics unavailable\n"); + var arena = std.heap.ArenaAllocator.init(hub.allocator); + defer arena.deinit(); + const account = status_page.collectAccount( + arena.allocator(), + hub.io, + store, + hub.plc_url, + raw_account, + hub.status_identity_resolution, + ) catch return respond(conn, "500 Internal Server Error", "text/plain", "failed to read account diagnostics\n"); + const now_us = Io.Timestamp.now(hub.io, .real).toMicroseconds(); + var verification: ?repo_export.VerifyReport = null; + var verification_error: []const u8 = ""; + if (!status_head and account.found) if (hub.archive) |archive| verify: { + if (hub.repo_action_rate_limits) { + var peer_ip_buf: [64]u8 = undefined; + const peer_ip = conn.peerIp(&peer_ip_buf) catch { + verification_error = "unable to identify source IP for repo verification"; + break :verify; + }; + const allowed = hub.repo_action_limiter.allow(hub.allocator, hub.io, peer_ip, now_us) catch { + verification_error = "unable to apply repo action rate limit"; + break :verify; + }; + if (!allowed) { + verification_error = "repo action rate limit exceeded"; + break :verify; + } + } + verification = repo_export.verify( + arena.allocator(), + hub.io, + archive, + hub.data_dir, + hub.plc_url, + account.did, + ) catch |err| { + verification_error = std.fmt.allocPrint(arena.allocator(), "repoexport: verify {s}: {s}", .{ account.did, @errorName(err) }) catch "repo verification failed"; + break :verify; + }; + }; + const page = status_page.renderAccountAlloc(hub.allocator, account, now_us, wants_html, verification, verification_error) catch + return respond(conn, "500 Internal Server Error", "text/plain", "failed to render account diagnostics\n"); + defer hub.allocator.free(page); + return respondStatus(conn, if (wants_html) "text/html; charset=utf-8" else "text/plain; charset=utf-8", page, status_head, now_us); + } if (status_page.hostSortFromQuery(query)) |sort| { const store = hub.repo_status orelse return respond(conn, "500 Internal Server Error", "text/plain", "host diagnostics unavailable\n"); const statuses = store.listHostStatuses(hub.allocator) catch @@ -673,7 +740,7 @@ pub const Handler = struct { const page = status_page.renderHostsAlloc(hub.allocator, statuses, sort, now_us, wants_html) catch return respond(conn, "500 Internal Server Error", "text/plain", "failed to render host diagnostics\n"); defer hub.allocator.free(page); - return respond(conn, "200 OK", if (wants_html) "text/html; charset=utf-8" else "text/plain; charset=utf-8", page); + return respondStatus(conn, if (wants_html) "text/html; charset=utf-8" else "text/plain; charset=utf-8", page, status_head, now_us); } const now_s: i64 = @divTrunc(Io.Timestamp.now(hub.io, .real).toMicroseconds(), std.time.us_per_s); const args: homepage.StatusArgs = .{ @@ -691,8 +758,8 @@ pub const Handler = struct { }; var page_buf: [32 * 1024]u8 = undefined; if (wants_html) - return respond(conn, "200 OK", "text/html; charset=utf-8", homepage.renderStatusHtml(&page_buf, args)); - return respond(conn, "200 OK", "text/plain; charset=utf-8", homepage.renderStatus(&page_buf, args)); + return respondStatus(conn, "text/html; charset=utf-8", homepage.renderStatusHtml(&page_buf, args), status_head, now_s * std.time.us_per_s); + return respondStatus(conn, "text/plain; charset=utf-8", homepage.renderStatus(&page_buf, args), status_head, now_s * std.time.us_per_s); } if (std.mem.eql(u8, path, "/metrics")) { const g = hub.tail.gauges(); @@ -868,6 +935,37 @@ fn respond(conn: *websocket.Conn, status: []const u8, content_type: []const u8, if (resp_body.len > 0) conn.writeFramed(resp_body) catch return; } +fn respondStatus(conn: *websocket.Conn, content_type: []const u8, body: []const u8, head: bool, generated_us: i64) void { + var time_buf: [40]u8 = undefined; + const generated_at = formatRfc3339(&time_buf, generated_us); + var header_buf: [512]u8 = undefined; + const header = std.fmt.bufPrint( + &header_buf, + "HTTP/1.1 200 OK\r\nContent-Type: {s}\r\nContent-Length: {d}\r\nCache-Control: no-store\r\nX-Status-Generated-At: {s}\r\nConnection: close\r\nServer: stream\r\n\r\n", + .{ content_type, body.len, generated_at }, + ) catch return; + conn.writeFramed(header) catch return; + if (!head) conn.writeFramed(body) catch {}; +} + +fn formatRfc3339(buf: []u8, micros: i64) []const u8 { + const secs: u64 = @intCast(@divFloor(micros, std.time.us_per_s)); + const frac: u64 = @intCast(@mod(micros, std.time.us_per_s)); + const epoch_secs: std.time.epoch.EpochSeconds = .{ .secs = secs }; + const year_day = epoch_secs.getEpochDay().calculateYearDay(); + const month_day = year_day.calculateMonthDay(); + const day_secs = epoch_secs.getDaySeconds(); + return std.fmt.bufPrint(buf, "{d:0>4}-{d:0>2}-{d:0>2}T{d:0>2}:{d:0>2}:{d:0>2}.{d:0>6}Z", .{ + year_day.year, + month_day.month.numeric(), + month_day.day_index + 1, + day_secs.getHoursIntoDay(), + day_secs.getMinutesIntoHour(), + day_secs.getSecondsIntoMinute(), + frac, + }) catch unreachable; +} + fn addAll( filter: *filter_mod.Filter, v: std.json.Value, @@ -1061,3 +1159,173 @@ test "cursor resolution is pre-upgrade and preserves v1-v2 floor semantics" { try std.testing.expectError(error.InvalidCursor, resolveCursor(&hub, true, "nope")); try std.testing.expectError(error.InvalidCursor, resolveCursor(&hub, true, "-1")); } + +test "accounts HTTP route verifies real archive state and rate limits by port-free peer IP" { + const testing = std.testing; + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var fixture_arena = std.heap.ArenaAllocator.init(testing.allocator); + defer fixture_arena.deinit(); + const alloc = fixture_arena.allocator(); + + const did = "did:plc:statusverifyfixture"; + const rev = "3k2abcdefghij"; + const record = try zat.cbor.encodeAlloc(alloc, .{ .map = &.{ + .{ .key = "$type", .value = .{ .text = "app.bsky.feed.post" } }, + .{ .key = "text", .value = .{ .text = "status route verification" } }, + } }); + const record_cid = try zat.cbor.Cid.forDagCbor(alloc, record); + var authoritative_tree = zat.mst.Mst.init(alloc); + try authoritative_tree.put("app.bsky.feed.post/fixture", record_cid); + const data_root = try authoritative_tree.rootCid(); + const keypair = try zat.Keypair.fromSecretKey(.p256, .{43} ** 32); + const signed = try zat.signCommit(alloc, .{ .did = did, .rev = rev, .data = data_root }, &keypair); + const commit_cid_string = try zat.multibase.base32lower.encode(alloc, signed.cid.raw); + const get_blocks_car = try zat.car.writeAlloc(alloc, .{ + .roots = &.{}, + .blocks = &.{.{ .cid_raw = signed.cid.raw, .data = signed.bytes }}, + }); + + var identity_listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; + defer identity_listener.deinit(io); + var identity_endpoint_buf: [96]u8 = undefined; + const identity_endpoint = try std.fmt.bufPrint(&identity_endpoint_buf, "http://127.0.0.1:{d}", .{identity_listener.socket.address.getPort()}); + const did_doc = try std.fmt.allocPrint(alloc, + \\{{"id":"{s}","alsoKnownAs":["at://status.example"],"verificationMethod":[],"service":[{{"id":"#atproto_pds","type":"AtprotoPersonalDataServer","serviceEndpoint":"{s}"}}]}} + , .{ did, identity_endpoint }); + const latest_json = try std.fmt.allocPrint(alloc, "{{\"cid\":\"{s}\",\"rev\":\"{s}\"}}", .{ commit_cid_string, rev }); + var identity_calls: std.atomic.Value(u32) = .init(0); + const IdentityFixture = struct { + fn run(server: *Io.net.Server, task_io: Io, calls: *std.atomic.Value(u32), doc: []const u8, latest: []const u8, blocks_car: []const u8) !void { + while (true) { + 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 = std.http.Server.init(&reader.interface, &writer.interface); + var request = try http.receiveHead(); + _ = calls.fetchAdd(1, .monotonic); + const target = request.head.target; + const response_body, const content_type = if (std.mem.indexOf(u8, target, "getLatestCommit") != null) + .{ latest, "application/json" } + else if (std.mem.indexOf(u8, target, "getBlocks") != null) + .{ blocks_car, "application/vnd.ipld.car" } + else + .{ doc, "application/json" }; + try request.respond(response_body, .{ + .status = .ok, + .keep_alive = false, + .extra_headers = &.{.{ .name = "content-type", .value = content_type }}, + }); + } + } + }; + var identity_future = try io.concurrent(IdentityFixture.run, .{ &identity_listener, io, &identity_calls, did_doc, latest_json, get_blocks_car }); + defer _ = identity_future.cancel(io) catch {}; + + 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(); + _ = try archive.append(.{ + .seq = 1, + .witnessed_at = 1_750_000_000_000_000, + .indexed_at = 0, + .kind = .create, + .did = did, + .collection = "app.bsky.feed.post", + .rkey = "fixture", + .rev = rev, + .payload = record, + }, 1); + + const meta_store = @import("meta_store.zig"); + var meta = try meta_store.Store.open(testing.allocator, data_dir); + defer meta.deinit(); + var repo_status = try repo_store.Store.init(testing.allocator, io, &meta); + defer repo_status.deinit(); + try repo_status.put(did, .{ + .status = .complete, + .active = true, + .rev = rev, + .latest_rev = rev, + .handle = "status.example", + .pds = identity_endpoint, + .host = "127.0.0.1", + .record_count = 1, + }); + + var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20); + defer tail.deinit(); + var stats: metrics.Stats = .{}; + var hub: Hub = .{ + .allocator = testing.allocator, + .io = io, + .tail = &tail, + .stats = &stats, + .archive = &archive, + .repo_status = &repo_status, + .data_dir = data_dir, + .plc_url = identity_endpoint, + }; + defer hub.deinit(); + + var stream_listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; + defer stream_listener.deinit(io); + const stream_port = stream_listener.socket.address.getPort(); + var ws_server = try websocket.Server(Handler).init(testing.allocator, io, .{ + .port = stream_port, + .address = "127.0.0.1", + .max_conn = 16, + .handshake = .{ .max_size = 16 * 1024, .max_headers = 64 }, + }); + defer ws_server.deinit(); + const StreamFixture = struct { + fn run(server: *websocket.Server(Handler), listener: *Io.net.Server, fixture_hub: *Hub) !void { + server.runIo(listener, fixture_hub); + } + }; + var stream_future = try io.concurrent(StreamFixture.run, .{ &ws_server, &stream_listener, &hub }); + defer _ = stream_future.cancel(io) catch {}; + + var status_url_buf: [256]u8 = undefined; + const status_url = try std.fmt.bufPrint(&status_url_buf, "http://127.0.0.1:{d}/status?tab=accounts&account={s}", .{ stream_port, did }); + var transport = zat.HttpTransport.init(io, testing.allocator); + transport.keep_alive = false; + defer transport.deinit(); + for (0..4) |_| { + const response = try transport.fetch(.{ .url = status_url, .accept = "text/html", .max_response_size = 256 * 1024 }); + defer testing.allocator.free(response.body); + try testing.expectEqual(std.http.Status.ok, response.status); + try testing.expect(std.mem.indexOf(u8, response.body, "Repo verification") != null); + try testing.expect(std.mem.indexOf(u8, response.body, "match-ok\">Match") != null); + try testing.expect(std.mem.indexOf(u8, response.body, "repo action rate limit exceeded") == null); + } + try testing.expectEqual(@as(u32, 12), identity_calls.load(.monotonic)); + + const limited = try transport.fetch(.{ .url = status_url, .accept = "text/html", .max_response_size = 256 * 1024 }); + defer testing.allocator.free(limited.body); + try testing.expectEqual(std.http.Status.ok, limited.status); + try testing.expect(std.mem.indexOf(u8, limited.body, "repo action rate limit exceeded") != null); + try testing.expectEqual(@as(u32, 12), identity_calls.load(.monotonic)); + + const head = try transport.fetch(.{ .url = status_url, .method = .HEAD, .accept = "text/html", .max_response_size = 256 * 1024 }); + defer testing.allocator.free(head.body); + try testing.expectEqual(std.http.Status.ok, head.status); + try testing.expectEqual(@as(usize, 0), head.body.len); + try testing.expectEqual(@as(u32, 12), identity_calls.load(.monotonic)); + + hub.repo_action_rate_limits = false; + const trusted = try transport.fetch(.{ .url = status_url, .accept = "text/html", .max_response_size = 256 * 1024 }); + defer testing.allocator.free(trusted.body); + try testing.expectEqual(std.http.Status.ok, trusted.status); + try testing.expect(std.mem.indexOf(u8, trusted.body, "match-ok\">Match") != null); + try testing.expect(std.mem.indexOf(u8, trusted.body, "repo action rate limit exceeded") == null); + try testing.expectEqual(@as(u32, 15), identity_calls.load(.monotonic)); +} diff --git a/src/internal/status_page.zig b/src/internal/status_page.zig index e39ddae..6af67c0 100644 --- a/src/internal/status_page.zig +++ b/src/internal/status_page.zig @@ -6,13 +6,187 @@ //! whole-network accounts. const std = @import("std"); +const zat = @import("zat"); +const repo_export = @import("repo_export.zig"); const repo_store = @import("bootstrap/repo_store.zig"); +const Io = std.Io; const Allocator = std.mem.Allocator; const Writer = std.Io.Writer; pub const HostSort = enum { largest, failing }; +pub const AccountLookup = struct { + query: []const u8 = "", + query_kind: []const u8 = "", + found: bool = false, + error_message: []const u8 = "", + did: []const u8 = "", + handle: []const u8 = "", + pds: []const u8 = "", + host: []const u8 = "", + active: bool = false, + backfill: []const u8 = "", + attempts: u32 = 0, + last_error: []const u8 = "", + backfill_rev: []const u8 = "", + latest_rev: []const u8 = "", + updated_us: i64 = 0, + last_attempt_us: i64 = 0, + record_count: i64 = 0, + total_bytes: i64 = 0, +}; + +pub fn accountFromQueryAlloc(allocator: Allocator, query: []const u8) !?[]u8 { + var is_accounts = false; + var account: ?[]u8 = null; + var did: ?[]u8 = null; + var handle: ?[]u8 = null; + errdefer { + if (account) |value| allocator.free(value); + if (did) |value| allocator.free(value); + if (handle) |value| allocator.free(value); + } + var fields = std.mem.splitScalar(u8, query, '&'); + while (fields.next()) |field| { + const split = std.mem.indexOfScalar(u8, field, '=') orelse field.len; + const key = field[0..split]; + const value = if (split < field.len) field[split + 1 ..] else ""; + if (std.mem.eql(u8, key, "tab")) is_accounts = std.mem.eql(u8, value, "accounts"); + // net/url.Values.Get, used upstream, selects the first occurrence. + if (std.mem.eql(u8, key, "account") and account == null) account = try percentDecodeQueryAlloc(allocator, value); + if (std.mem.eql(u8, key, "did") and did == null) did = try percentDecodeQueryAlloc(allocator, value); + if (std.mem.eql(u8, key, "handle") and handle == null) handle = try percentDecodeQueryAlloc(allocator, value); + } + if (!is_accounts) { + if (account) |value| allocator.free(value); + if (did) |value| allocator.free(value); + if (handle) |value| allocator.free(value); + return null; + } + if (account) |value| { + if (std.mem.trim(u8, value, " \t\r\n").len > 0) { + if (did) |old| allocator.free(old); + if (handle) |old| allocator.free(old); + return value; + } + allocator.free(value); + } + if (did) |value| { + if (std.mem.trim(u8, value, " \t\r\n").len > 0) { + if (handle) |old| allocator.free(old); + return value; + } + allocator.free(value); + } + return handle orelse try allocator.dupe(u8, ""); +} + +/// Collect one account exactly like upstream's status collector. When an +/// identity resolver is configured (the production path), a handle lookup +/// MUST resolve authoritatively and an error MUST NOT fall through to the +/// local declared-handle index. The index exists only for resolver-less use. +pub fn collectAccount( + allocator: Allocator, + io: Io, + store: *repo_store.Store, + plc_url: []const u8, + raw_account: []const u8, + use_identity_resolver: bool, +) !AccountLookup { + const account = std.mem.trim(u8, raw_account, " \t\r\n"); + if (account.len == 0) return .{}; + + var result: AccountLookup = .{}; + var did: []const u8 = undefined; + if (std.ascii.startsWithIgnoreCase(account, "did:")) { + result.query_kind = "did"; + result.query = account; + if (zat.Did.parse(account) == null) { + result.error_message = try std.fmt.allocPrint(allocator, "invalid DID: {s}", .{account}); + return result; + } + did = account; + } else { + const without_at = if (std.mem.startsWith(u8, account, "@")) account[1..] else account; + const parsed = zat.Handle.parse(without_at) orelse { + result.query_kind = "handle"; + result.query = without_at; + result.error_message = try std.fmt.allocPrint(allocator, "invalid handle: {s}", .{without_at}); + return result; + }; + const normalized = try asciiLowerAlloc(allocator, parsed.str()); + result.query_kind = "handle"; + result.query = normalized; + if (use_identity_resolver) { + var resolver = zat.HandleResolver.init(io, allocator); + defer resolver.deinit(); + did = resolver.resolve(.{ .raw = normalized }) catch |err| { + result.error_message = try std.fmt.allocPrint(allocator, "resolve handle \"{s}\": {s}", .{ normalized, @errorName(err) }); + return result; + }; + } else { + did = (try store.lookupDIDByHandle(allocator, normalized)) orelse return result; + } + } + + const state = (try store.get(allocator, did)) orelse { + result.did = did; + return result; + }; + result = .{ + .query = result.query, + .query_kind = result.query_kind, + .found = true, + .did = did, + .handle = state.handle, + .pds = state.pds, + .host = state.host, + .active = state.active, + .backfill = @tagName(state.status), + .attempts = state.attempts, + .last_error = state.last_error, + .backfill_rev = state.rev, + .latest_rev = state.latest_rev, + .updated_us = state.updated_us, + .last_attempt_us = state.last_attempt_us, + .record_count = state.record_count, + .total_bytes = state.total_bytes, + }; + + if (use_identity_resolver and (result.handle.len == 0 or result.pds.len == 0)) { + var resolver = zat.DidResolver.init(io, allocator); + defer resolver.deinit(); + resolver.plc_url = plc_url; + const parsed_did = zat.Did.parse(did).?; + var doc = resolver.resolve(parsed_did) catch null; + if (doc) |*resolved| { + defer resolved.deinit(); + if (result.handle.len == 0) result.handle = if (resolved.handle()) |handle| try allocator.dupe(u8, handle) else ""; + if (result.pds.len == 0) result.pds = if (resolved.pdsEndpoint()) |pds| try allocator.dupe(u8, pds) else ""; + } + } + if (result.handle.len == 0 and std.mem.eql(u8, result.query_kind, "handle")) result.handle = result.query; + return result; +} + +pub fn renderAccountAlloc( + allocator: Allocator, + account: AccountLookup, + now_us: i64, + html: bool, + verification: ?repo_export.VerifyReport, + verification_error: []const u8, +) ![]u8 { + var output: Writer.Allocating = .init(allocator); + errdefer output.deinit(); + if (html) + try renderAccountHtml(&output.writer, account, now_us, verification, verification_error) + else + try renderAccountText(&output.writer, account, now_us, verification, verification_error); + return output.toOwnedSlice(); +} + pub fn hostSortFromQuery(query: []const u8) ?HostSort { var tab_hosts = false; var sort: HostSort = .largest; @@ -48,6 +222,139 @@ pub fn renderHostsAlloc(allocator: Allocator, statuses: []repo_store.HostStatus, return output.toOwnedSlice(); } +fn renderAccountText( + w: *Writer, + account: AccountLookup, + now_us: i64, + verification: ?repo_export.VerifyReport, + verification_error: []const u8, +) !void { + try w.writeAll("stream account diagnostics\n\n"); + if (account.query.len == 0) return w.writeAll("Supply a DID or handle with ?tab=accounts&account=...\n"); + try w.print("query\t{s}\t{s}\n", .{ account.query_kind, account.query }); + if (account.error_message.len > 0) return w.print("error\t{s}\n", .{account.error_message}); + if (account.did.len > 0) try w.print("did\t{s}\n", .{account.did}); + if (!account.found) return w.writeAll("found\tno\n"); + try w.writeAll("found\tyes\n"); + try w.print("handle\t{s}\npds\t{s}\nhost\t{s}\nactive\t{}\nbackfill\t{s}\nattempts\t{d}\n", .{ + account.handle, + account.pds, + account.host, + account.active, + account.backfill, + account.attempts, + }); + try w.writeAll("last attempted\t"); + try writeAge(w, now_us, account.last_attempt_us); + try w.print("\nbackfill rev\t{s}\nlatest rev\t{s}\n", .{ account.backfill_rev, account.latest_rev }); + if (account.record_count != 0) try w.print("records\t{d}\n", .{account.record_count}); + if (account.total_bytes != 0) try w.print("bytes\t{d}\n", .{account.total_bytes}); + if (account.last_error.len > 0) try w.print("last error\t{s}\n", .{account.last_error}); + if (verification) |report| { + try w.print("\nrepo verification\nresult\t{s}\nauthoritative rev\t{s}\nauthoritative root\t{s}\nlocal rev\t{s}\nlocal root\t{s}\nlocal records\t{d}\n", .{ + if (report.match) "Match" else "Match failure", + report.authoritative_rev, + report.authoritative_root, + report.local_latest_rev, + report.local_root, + report.local_record_count, + }); + if (report.message.len > 0) try w.print("message\t{s}\n", .{report.message}); + } else if (verification_error.len > 0) { + try w.print("\nrepo verification\nerror\t{s}\n", .{verification_error}); + } +} + +fn renderAccountHtml( + w: *Writer, + account: AccountLookup, + now_us: i64, + verification: ?repo_export.VerifyReport, + verification_error: []const u8, +) !void { + try w.writeAll( + \\ + \\ + \\stream · accounts + \\
+ \\

Account diagnostics

+ \\

Inspect durable backfill state for one repository. Handle queries resolve through the network identity directory before the local status row is loaded.

+ \\
"); + if (account.query.len == 0) return w.writeAll("

Enter a DID or handle.

"); + if (account.error_message.len > 0) { + try w.writeAll("

"); + try writeHtmlEscaped(w, account.error_message); + return w.writeAll("

"); + } + if (!account.found) { + try w.writeAll("
No local repository status was found"); + if (account.did.len > 0) { + try w.writeAll(" for "); + try writeHtmlEscaped(w, account.did); + try w.writeAll(""); + } + return w.writeAll(".
"); + } + try w.writeAll("
DID
"); + try writeHtmlEscaped(w, account.did); + try w.writeAll("
Declared handle
"); + try writeHtmlCodeOrUnknown(w, account.handle); + try w.writeAll("
PDS endpoint
"); + try writeHtmlCodeOrUnknown(w, account.pds); + try w.writeAll("
Active
"); + try w.writeAll(if (account.active) "true" else "false"); + try w.writeAll("
Backfill
"); + try writeHtmlEscaped(w, account.backfill); + try w.print("
Retries
{d}
Last attempted
", .{account.attempts}); + try writeAge(w, now_us, account.last_attempt_us); + try w.writeAll("
Backfill rev
"); + try writeHtmlCodeOrUnknown(w, account.backfill_rev); + try w.writeAll("
Latest rev
"); + try writeHtmlCodeOrUnknown(w, account.latest_rev); + try w.writeAll("
"); + if (account.record_count != 0) try w.print("
Records
{d}
", .{account.record_count}); + if (account.total_bytes != 0) try w.print("
Bytes
{d}
", .{account.total_bytes}); + if (account.last_error.len > 0) { + try w.writeAll("
Last error
"); + try writeHtmlEscaped(w, account.last_error); + try w.writeAll("
"); + } + try w.writeAll("
"); + if (verification) |report| { + try w.writeAll("

Repo verification

Result
Match" else "match-fail\">Match failure"); + try w.writeAll("
authoritative rev
"); + try writeHtmlCodeOrUnknown(w, report.authoritative_rev); + try w.writeAll("
authoritative root
"); + try writeHtmlCodeOrUnknown(w, report.authoritative_root); + try w.writeAll("
local rev
"); + try writeHtmlCodeOrUnknown(w, report.local_latest_rev); + try w.writeAll("
local root
"); + try writeHtmlCodeOrUnknown(w, report.local_root); + try w.print("
local records
{d}
", .{report.local_record_count}); + if (report.message.len > 0) { + try w.writeAll("
Message
"); + try writeHtmlEscaped(w, report.message); + try w.writeAll("
"); + } + try w.writeAll("
"); + } else if (verification_error.len > 0) { + try w.writeAll("

Repo verification

"); + try writeHtmlEscaped(w, verification_error); + try w.writeAll("
"); + } + try w.writeAll(""); +} + fn renderHostsText(w: *Writer, statuses: []const repo_store.HostStatus, sort: HostSort, now_us: i64) !void { try w.print("stream host diagnostics\nsort: {s}\n\n", .{@tagName(sort)}); if (statuses.len == 0) { @@ -91,7 +398,7 @@ fn renderHostsHtml(w: *Writer, statuses: []const repo_store.HostStatus, sort: Ho \\h1{margin:0;font-size:clamp(30px,7vw,52px);font-weight:500;letter-spacing:-.04em}nav{display:flex;gap:12px}nav a,.toggle a{padding:7px 9px;border:1px solid var(--line);text-decoration:none;font:12px ui-monospace,SFMono-Regular,monospace}.active{background:var(--water);color:var(--deep)!important;border-color:var(--water)!important} \\.lede{color:var(--dim);max-width:70ch;margin:14px 0 24px}.toggle{display:flex;gap:8px;margin:22px 0 14px}.filter{width:min(440px,100%);margin:0 0 14px;padding:9px 11px;color:var(--ink);background:var(--panel);border:1px solid var(--line);border-radius:0;font:13px ui-monospace,SFMono-Regular,monospace}.filter:focus{outline:1px solid var(--water);border-color:var(--water)}.table-wrap{overflow:auto;border:1px solid var(--line)}table{border-collapse:collapse;width:100%;min-width:880px}th,td{text-align:left;padding:10px 11px;border-bottom:1px solid var(--line);vertical-align:top}th{color:var(--dim);font-size:11px;letter-spacing:.08em;text-transform:uppercase;background:#0a1d25;position:sticky;top:0}.num{text-align:right;font-variant-numeric:tabular-nums}code{color:var(--foam);font:12px ui-monospace,SFMono-Regular,monospace}.error{max-width:390px;overflow-wrap:anywhere}.samples td{padding:0 11px 13px;background:var(--panel)}.samples ul{list-style:none;margin:0;padding:8px 0 0}.samples li{display:grid;grid-template-columns:minmax(180px,1fr) 110px 90px minmax(220px,2fr);gap:10px;padding:7px 0;border-top:1px solid var(--line)}.bad{color:var(--bad)}.empty{border:1px solid var(--line);background:var(--panel);padding:18px;color:var(--dim)} \\@media(max-width:620px){header{display:block}nav{margin-top:14px}.samples li{grid-template-columns:1fr}.toggle{flex-wrap:wrap}}
- \\

Host diagnostics

+ \\

Host diagnostics

\\

Durable PDS aggregates from repository backfill and healing. Counts update atomically with each account row; historical errors remain visible after recovery.

); try w.writeAll(if (sort == .largest) ""); + try writeHtmlEscaped(w, raw); + try w.writeAll(""); +} + +fn asciiLowerAlloc(allocator: Allocator, raw: []const u8) ![]u8 { + const out = try allocator.alloc(u8, raw.len); + for (raw, out) |byte, *dst| dst.* = std.ascii.toLower(byte); + return out; +} + +fn percentDecodeQueryAlloc(allocator: Allocator, raw: []const u8) ![]u8 { + var out: std.ArrayList(u8) = .empty; + errdefer out.deinit(allocator); + var i: usize = 0; + while (i < raw.len) { + if (raw[i] == '+') { + try out.append(allocator, ' '); + i += 1; + } else if (raw[i] == '%') { + if (i + 2 >= raw.len) return error.InvalidQueryEncoding; + const byte = std.fmt.parseInt(u8, raw[i + 1 .. i + 3], 16) catch return error.InvalidQueryEncoding; + try out.append(allocator, byte); + i += 3; + } else { + try out.append(allocator, raw[i]); + i += 1; + } + } + return out.toOwnedSlice(allocator); +} + test "host query and upstream sort modes are exact" { const testing = std.testing; try testing.expectEqual(HostSort.largest, hostSortFromQuery("tab=hosts").?); @@ -221,6 +562,103 @@ test "host rendering is unbounded, escaped, and preserves diagnostic history" { try testing.expect(std.mem.indexOf(u8, html, "host-filter") != null); try testing.expect(std.mem.indexOf(u8, html, "id=\"hosts-table\"") != null); try testing.expect(std.mem.indexOf(u8, html, "host-errors-row") != null); - try testing.expect(std.mem.indexOf(u8, html, "tab=accounts") == null); + try testing.expect(std.mem.indexOf(u8, html, "tab=accounts") != null); try testing.expect(std.mem.indexOf(u8, html, "1s ago") != null); } + +test "account query decoding and normalization select only the accounts tab" { + const testing = std.testing; + try testing.expect(try accountFromQueryAlloc(testing.allocator, "tab=hosts&account=ignored") == null); + const account = (try accountFromQueryAlloc(testing.allocator, "account=%40Alice.Example&tab=accounts")).?; + defer testing.allocator.free(account); + try testing.expectEqualStrings("@Alice.Example", account); + const legacy_did = (try accountFromQueryAlloc(testing.allocator, "tab=accounts&did=did%3Aplc%3Alegacy&handle=ignored.example")).?; + defer testing.allocator.free(legacy_did); + try testing.expectEqualStrings("did:plc:legacy", legacy_did); + const explicit_wins = (try accountFromQueryAlloc(testing.allocator, "tab=accounts&account=chosen.example&did=did%3Aplc%3Aignored")).?; + defer testing.allocator.free(explicit_wins); + try testing.expectEqualStrings("chosen.example", explicit_wins); +} + +test "resolver-less account lookup uses the durable handle index and renders exact state" { + const testing = std.testing; + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var path_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const path_len = try tmp.dir.realPath(io, &path_buf); + const meta_store = @import("meta_store.zig"); + var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]); + defer meta.deinit(); + var store = try repo_store.Store.init(testing.allocator, io, &meta); + defer store.deinit(); + const did = "did:plc:accountfixture"; + try store.put(did, .{ + .status = .complete, + .active = true, + .rev = "3lbackfill", + .latest_rev = "3llatest", + .handle = "alice.example", + .pds = "https://pds.example", + .host = "pds.example", + .attempts = 2, + .last_attempt_us = 9 * std.time.us_per_s, + .record_count = 7, + .total_bytes = 99, + }); + + var arena = std.heap.ArenaAllocator.init(testing.allocator); + defer arena.deinit(); + const account = try collectAccount(arena.allocator(), io, &store, "http://unused", " @ALICE.EXAMPLE ", false); + try testing.expect(account.found); + try testing.expectEqualStrings("handle", account.query_kind); + try testing.expectEqualStrings("alice.example", account.query); + try testing.expectEqualStrings(did, account.did); + try testing.expectEqualStrings("3lbackfill", account.backfill_rev); + try testing.expectEqualStrings("3llatest", account.latest_rev); + + const verification: repo_export.VerifyReport = .{ + .did = did, + .match = true, + .authoritative_rev = "3llatest", + .authoritative_root = "bafyroot", + .local_latest_rev = "3llatest", + .local_root = "bafyroot", + .local_record_count = 7, + }; + const html = try renderAccountAlloc(testing.allocator, account, 10 * std.time.us_per_s, true, verification, ""); + defer testing.allocator.free(html); + try testing.expect(std.mem.indexOf(u8, html, "Account diagnostics") != null); + try testing.expect(std.mem.indexOf(u8, html, "alice.example") != null); + try testing.expect(std.mem.indexOf(u8, html, "3llatest") != null); + try testing.expect(std.mem.indexOf(u8, html, "7") != null); + try testing.expect(std.mem.indexOf(u8, html, "1s ago") != null); + try testing.expect(std.mem.indexOf(u8, html, "Repo verification") != null); + try testing.expect(std.mem.indexOf(u8, html, "match-ok\">Match") != null); + try testing.expect(std.mem.indexOf(u8, html, "bafyroot") != null); +} + +test "account lookup validates DID and reports a resolved DID missing locally" { + const testing = std.testing; + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var path_buf: [Io.Dir.max_path_bytes]u8 = undefined; + const path_len = try tmp.dir.realPath(io, &path_buf); + const meta_store = @import("meta_store.zig"); + var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]); + defer meta.deinit(); + var store = try repo_store.Store.init(testing.allocator, io, &meta); + defer store.deinit(); + var arena = std.heap.ArenaAllocator.init(testing.allocator); + defer arena.deinit(); + const invalid = try collectAccount(arena.allocator(), io, &store, "http://unused", "DID:plc:wrongcase", false); + try testing.expect(std.mem.startsWith(u8, invalid.error_message, "invalid DID:")); + const missing = try collectAccount(arena.allocator(), io, &store, "http://unused", "did:plc:missing", false); + try testing.expect(!missing.found); + try testing.expectEqualStrings("did:plc:missing", missing.did); +} diff --git a/src/main.zig b/src/main.zig index 1053d20..98b78ac 100644 --- a/src/main.zig +++ b/src/main.zig @@ -92,6 +92,7 @@ pub fn main(init: std.process.Init.Minimal) !void { var retry_interval_s: u64 = 4 * 3600; var timestamp_import_token: []const u8 = ""; var timestamp_import_dir_arg: ?[]const u8 = null; + var repo_action_rate_limits = true; var arg_it = init.args.iterate(); _ = arg_it.next(); // program name @@ -136,6 +137,8 @@ pub fn main(init: std.process.Init.Minimal) !void { do_backfill = true; } else if (std.mem.eql(u8, arg, "--no-verify")) { do_verify = false; + } else if (std.mem.eql(u8, arg, "--disable-repo-action-rate-limits")) { + repo_action_rate_limits = false; } else if (std.mem.eql(u8, arg, "--stdout")) { to_stdout = true; } else { @@ -160,7 +163,11 @@ pub fn main(init: std.process.Init.Minimal) !void { .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, + .plc_url = plc_url, + .data_dir = data_dir, + .repo_action_rate_limits = repo_action_rate_limits, }; + defer hub.deinit(); var archive = try archive_mod.Archive.init(allocator, io, data_dir); defer { diff --git a/src/tests.zig b/src/tests.zig index 1caaed4..2339cae 100644 --- a/src/tests.zig +++ b/src/tests.zig @@ -41,6 +41,8 @@ comptime { _ = @import("internal/tail.zig"); _ = @import("internal/server.zig"); _ = @import("internal/status_page.zig"); + _ = @import("internal/repo_export.zig"); + _ = @import("internal/repo_action_limiter.zig"); _ = @import("internal/xrpcapi.zig"); _ = @import("internal/homepage.zig"); _ = @import("internal/timestamp/parse.zig"); diff --git a/tests/status_contract.py b/tests/status_contract.py index 5c9e280..700a07c 100644 --- a/tests/status_contract.py +++ b/tests/status_contract.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Offline HTTP contract for durable host diagnostics.""" +"""Offline HTTP contract for durable host and account diagnostics.""" import os import urllib.request @@ -37,7 +37,41 @@ assert 'id="host-filter"' in html assert 'id="hosts-table"' in html assert 'class="samples host-errors-row"' in html assert "document.querySelectorAll" in html -assert "tab=accounts" not in html +assert 'href="/status?tab=accounts"' in html assert html.endswith("") -print("HOST STATUS HTTP CONTRACT PASS (durable rows, restart, sorts, filtering, text/html, escaping)") +content_type, account_text = get( + "/status?tab=accounts&did=did:plc:large0", "text/plain" +) +assert content_type == "text/plain" +assert "query\tdid\tdid:plc:large0" in account_text +assert "found\tyes" in account_text +assert "backfill\tcomplete" in account_text +assert "repo verification" in account_text +assert "error\trepoexport: verify did:plc:large0:" in account_text + +content_type, account_html = get( + "/status?tab=accounts&account=did:plc:large0", "text/html" +) +assert content_type == "text/html" +assert "Account diagnostics" in account_html +assert "did:plc:large0" in account_html +assert "
Active
true
" in account_html +assert "
Declared handle
unknown
" in account_html +assert "Repo verification" in account_html + +head = urllib.request.Request( + base + "/status?tab=accounts&account=did:plc:large0", + headers={"Accept": "text/html"}, + method="HEAD", +) +with urllib.request.urlopen(head, timeout=5) as response: + assert response.status == 200 + assert response.headers["Cache-Control"] == "no-store" + assert response.headers["X-Status-Generated-At"].endswith("Z") + assert response.read() == b"" + +print( + "STATUS HTTP CONTRACT PASS " + "(durable host/account rows, restart, sorts, filtering, aliases, HEAD, escaping)" +) -- 2.51.2