Fix full repo export reachability · zat.dev/zds@8fae81d

+15 -10

GETREPO_NOTES.md

··· 7 7 Current shape: 8 8 9 9 - ZDS serves raw CAR bytes and does not appear to add application-layer gzip, so it avoids the Node compression issue described in the operator thread. 10 - - Full repo export is built from SQLite `repo_blocks`, ordered by CID. 10 + - As of the 2026-06-29 getRepo pass, full repo export starts at the latest commit block and uses `zat.mst.collectReachableBlocks` to include only current MST and record blocks reachable from the commit's `data` CID. 11 11 - Incremental export with `since` filters by `repo_rev`. 12 12 - Repo writes already use lazy MST loading backed by `repo_blocks`, so ZDS is closer to a Hubble-style block/CAR-serving shape than an eager full-repo rebuild path. 13 - - `subscribeRepos` has a connection cap, but `getRepo` does not appear to have route-specific backpressure. 13 + - `subscribeRepos` has a connection cap. Full `getRepo` now has its own lightweight route-local concurrency cap so backup/backfill exports cannot consume unbounded handler slots. Operators can tune it with `ZDS_MAX_CONCURRENT_REPO_EXPORTS`; the default is `4`, matching Tranquil's current full-export default. 14 14 15 15 Open questions: 16 16 17 - - Should full `getRepo` walk the current commit/MST and include only reachable blocks, instead of exporting every retained block for the DID? 18 - - Do retained old MST or record blocks make the current full export disclose deleted or stale record data? 19 - - Should `getRepo` have route-specific concurrency or rate limits separate from normal public reads? 20 - - Should the benchmark suite compare the current range-scan export against a reachability-based export for small, medium, and large repos? 17 + - Should `HEAD com.atproto.sync.getRepo` stay semantically cheap, or should it compute/export-sized metadata? Today it is treated as a probe and still shares the same handler path. 18 + - Should `getRepo` backpressure eventually become per-account/IP rate limiting in addition to process-local concurrency? 19 + - Should the benchmark suite keep a historical range-scan comparison around, now that production full export is reachability-based? 20 + 21 + Done in this pass: 22 + 23 + 1. Adopted `zat` 0.3.9 for `collectReachableBlocks`. 24 + 2. Added a regression test for update/delete followed by full `getRepo`; stale record CIDs are excluded from full export. 25 + 3. Changed full export from whole-DID `repo_blocks` scan to current-root reachability. 26 + 4. Added lightweight route backpressure for full `getRepo`, configurable with `ZDS_MAX_CONCURRENT_REPO_EXPORTS`. 27 + 5. Ran `just bench repo-size`; reachable full export measured about 1.4 ms for 100 records, 16.8 ms for 1,000 records, and 95.7 ms for 5,000 records on the local synthetic benchmark. 21 28 22 29 Suggested next pass: 23 30 24 - 1. Add a regression test for delete/update followed by full `getRepo`, checking whether stale record blocks are exported. 25 - 2. If stale blocks are visible, change full export to current-root reachability rather than whole-DID block scan. 26 - 3. Add lightweight route backpressure for full `getRepo`. 27 - 4. Re-run `bench repo-size` and keep raw range-scan numbers separate from reachable-export numbers. 31 + 1. Decide whether `GETREPO_NOTES.md` should stay as a root handoff note or move into `docs/`. 32 + 2. Add a larger deliberate `just bench repo-large 100000` run when we want stress numbers comparable to large public repos.

+1

README.md

··· 79 79 ZDS_BLOBSTORE_PATH=/var/lib/zds/blobs \ 80 80 ZDS_HANDLE_DOMAINS='.example.com,example.com' \ 81 81 ZDS_CRAWLERS='https://bsky.network,https://vsky.network' \ 82 + ZDS_MAX_CONCURRENT_REPO_EXPORTS=4 \ 82 83 ZDS_PLC_ROTATION_KEY='64-hex-secp256k1-secret-or-private-multikey' \ 83 84 ZDS_RECOVERY_DID_KEY='did:key:optionalRecoveryKey' \ 84 85 ZDS_JWT_SECRET='at-least-32-random-bytes-here' \

+14

bench/README.md

··· 107 107 fixtures before they become stable benchmarks. Do not make routine CI or 108 108 pre-commit checks depend on live PDS availability. 109 109 110 + June 29, 2026 local reachable-export check after switching full 111 + `com.atproto.sync.getRepo` from whole-DID `repo_blocks` scans to current-root 112 + reachability via `zat.mst.collectReachableBlocks`: 113 + 114 + | repo tier | records | list | full CAR export | 115 + |---|---:|---:|---:| 116 + | small | 100 | 13.4k ops/s, 15.0 ms | 3.6k ops/s, 72.1 MB/s, 1.4 ms | 117 + | medium | 1,000 | 12.4k ops/s, 16.1 ms | 297 ops/s, 59.6 MB/s, 16.8 ms | 118 + | large | 5,000 | 11.9k ops/s, 16.8 ms | 52 ops/s, 53.1 MB/s, 95.7 ms | 119 + 120 + These are synthetic local trend numbers, not cross-PDS comparisons. The old 121 + whole-DID range-scan export path is intentionally no longer the production full 122 + export behavior because it could include retained stale blocks. 123 + 110 124 ## methodology 111 125 112 126 - Build ZDS with `-Doptimize=ReleaseFast`.

+2 -2

build.zig.zon

··· 5 5 .minimum_zig_version = "0.16.0", 6 6 .dependencies = .{ 7 7 .zat = .{ 8 - .url = "https://tangled.org/zat.dev/zat/archive/v0.3.8.tar.gz", 9 - .hash = "zat-0.3.8-5PuC7glgCgCVn18R0k6AuXiNok-jjqDRGgaBFj5JsI_N", 8 + .url = "https://tangled.org/zat.dev/zat/archive/v0.3.9.tar.gz", 9 + .hash = "zat-0.3.9-5PuC7v6aCgDbfEdNEUxbSbbpq27V5upmAsa6J-kqdCXx", 10 10 }, 11 11 .zqlite = .{ 12 12 .url = "git+https://github.com/karlseguin/zqlite.zig?ref=master#05a88d6758753e1c63fdd45b211dde2057094b0c",

+4

docs/operations.md

··· 105 105 - `ZDS_BLOB_UPLOAD_LIMIT`: upload body limit. Default: `100000000`. 106 106 - `ZDS_BLOBSTORE_PATH`: disk blobstore root. 107 107 - `ZDS_CRAWLERS`: comma-separated relay crawl targets. 108 + - `ZDS_MAX_CONCURRENT_REPO_EXPORTS`: maximum concurrent full 109 + `com.atproto.sync.getRepo` exports. Default: `4`. Incremental 110 + `getRepo?since=...` exports are not intended to consume this full-export 111 + backpressure budget. 108 112 - `ZDS_ADMIN_TOKEN`: bearer token for privileged local PDS administration, 109 113 including invite-code minting. 110 114 - `ZDS_INVITE_REQUIRED`: set to `true` to require invite codes for account

+16

src/atproto/sync.zig

··· 13 13 const notify_threshold_ms = 20 * 60 * 1000; 14 14 15 15 var subscribe_repos_connections: usize = 0; 16 + var get_repo_connections: usize = 0; 16 17 var crawler_last_notified_ms: std.atomic.Value(i64) = .init(0); 17 18 var crawler_notify_in_flight: std.atomic.Value(bool) = .init(false); 18 19 ··· 207 208 var since_buf: [256]u8 = undefined; 208 209 const since = http_api.queryParam(request.url.raw, "since", &since_buf); 209 210 try requirePublicRepoAvailable(request, did); 211 + 212 + const full_export = since == null; 213 + if (full_export) { 214 + const active = @atomicRmw(usize, &get_repo_connections, .Add, 1, .monotonic); 215 + const max_connections = config.maxConcurrentRepoExports(); 216 + if (active >= max_connections) { 217 + _ = @atomicRmw(usize, &get_repo_connections, .Sub, 1, .monotonic); 218 + log.debug("sync getRepo rejected too_many_connections active={d} max={d} did={s}\n", .{ active + 1, max_connections, did }); 219 + return http_api.xrpcError(request, .too_many_requests, "RateLimitExceeded", "too many full getRepo exports"); 220 + } 221 + } 222 + defer { 223 + if (full_export) _ = @atomicRmw(usize, &get_repo_connections, .Sub, 1, .monotonic); 224 + } 225 + 210 226 const body = store.writeRepoCarSince(allocator, did, since) catch { 211 227 return http_api.xrpcError(request, .not_found, "RepoNotFound", "Repo not found"); 212 228 };

+9

src/core/config.zig

··· 14 14 var blobstore_path_value: []const u8 = "dev/blobs"; 15 15 var handle_domains_value: []const u8 = ".test"; 16 16 var crawlers_value: []const u8 = "https://bsky.network,https://vsky.network"; 17 + var max_concurrent_repo_exports_value: usize = 4; 17 18 var proxy_service_did_value: []const u8 = "did:web:api.bsky.app"; 18 19 var proxy_service_id_value: []const u8 = "bsky_appview"; 19 20 var proxy_service_url_value: []const u8 = "https://api.bsky.app"; ··· 89 90 90 91 pub fn crawlers() []const u8 { 91 92 return crawlers_value; 93 + } 94 + 95 + pub fn maxConcurrentRepoExports() usize { 96 + return max_concurrent_repo_exports_value; 92 97 } 93 98 94 99 pub fn proxyServiceDid() []const u8 { ··· 185 190 186 191 pub fn setCrawlers(value: []const u8) void { 187 192 crawlers_value = value; 193 + } 194 + 195 + pub fn setMaxConcurrentRepoExports(value: usize) void { 196 + max_concurrent_repo_exports_value = value; 188 197 } 189 198 190 199 pub fn setProxyServiceDid(value: []const u8) void {

+13 -1

src/internal/cli.zig

··· 18 18 blobstore_path: ?[]const u8 = null, 19 19 handle_domains: ?[]const u8 = null, 20 20 crawlers: ?[]const u8 = null, 21 + max_concurrent_repo_exports: ?usize = null, 21 22 proxy_service_did: ?[]const u8 = null, 22 23 proxy_service_id: ?[]const u8 = null, 23 24 proxy_service_url: ?[]const u8 = null, ··· 35 36 MissingBlobUploadLimit, 36 37 MissingBlobstorePath, 37 38 MissingCrawlers, 39 + MissingMaxConcurrentRepoExports, 38 40 MissingProxyServiceDid, 39 41 MissingProxyServiceId, 40 42 MissingProxyServiceUrl, ··· 78 80 .blobstore_path = env("ZDS_BLOBSTORE_PATH"), 79 81 .handle_domains = env("ZDS_HANDLE_DOMAINS"), 80 82 .crawlers = env("ZDS_CRAWLERS"), 83 + .max_concurrent_repo_exports = try envUsize("ZDS_MAX_CONCURRENT_REPO_EXPORTS"), 81 84 .proxy_service_did = env("ZDS_PROXY_SERVICE_DID"), 82 85 .proxy_service_id = env("ZDS_PROXY_SERVICE_ID"), 83 86 .proxy_service_url = env("ZDS_PROXY_SERVICE_URL"), ··· 108 111 \\ [--plc-rotation-key KEY] [--recovery-did-key DIDKEY] 109 112 \\ [--mail-provider comail|resend] [--email-from ADDRESS] 110 113 \\ [--blob-upload-limit BYTES] [--blobstore-path PATH] 111 - \\ [--handle-domains DOMAINS] [--crawlers URLS] [--jwt-secret SECRET] 114 + \\ [--handle-domains DOMAINS] [--crawlers URLS] 115 + \\ [--max-concurrent-repo-exports N] [--jwt-secret SECRET] 112 116 \\ [--dpop-secret SECRET] 113 117 \\ [--proxy-service-did DID] [--proxy-service-id ID] [--proxy-service-url URL] 114 118 \\ [--admin-token TOKEN] [--invite-required] ··· 199 203 options.crawlers = args.next() orelse return error.MissingCrawlers; 200 204 return true; 201 205 } 206 + if (std.mem.eql(u8, arg, "--max-concurrent-repo-exports")) { 207 + options.max_concurrent_repo_exports = try std.fmt.parseInt(usize, args.next() orelse return error.MissingMaxConcurrentRepoExports, 10); 208 + return true; 209 + } 202 210 if (std.mem.eql(u8, arg, "--proxy-service-did")) { 203 211 options.proxy_service_did = args.next() orelse return error.MissingProxyServiceDid; 204 212 return true; ··· 243 251 } 244 252 if (std.mem.startsWith(u8, arg, "--blob-upload-limit=")) { 245 253 options.blob_upload_limit = try std.fmt.parseInt(usize, arg["--blob-upload-limit=".len..], 10); 254 + return true; 255 + } 256 + if (std.mem.startsWith(u8, arg, "--max-concurrent-repo-exports=")) { 257 + options.max_concurrent_repo_exports = try std.fmt.parseInt(usize, arg["--max-concurrent-repo-exports=".len..], 10); 246 258 return true; 247 259 } 248 260 return false;

+1

src/main.zig

+116 -25

src/storage/store.zig

··· 427 427 const RepoBlockReader = struct { 428 428 allocator: std.mem.Allocator, 429 429 did: []const u8, 430 + locked: bool = false, 430 431 431 432 fn reader(self: *RepoBlockReader) zat.mst.BlockReader { 432 433 return .{ ··· 439 440 const self: *RepoBlockReader = @ptrCast(@alignCast(ctx)); 440 441 const cid = try cidText(self.allocator, cid_raw); 441 442 443 + if (self.locked) { 444 + return repoBlockDataLocked(self.allocator, self.did, cid); 445 + } 446 + 442 447 db_mutex.lockUncancelable(store_io); 443 448 defer db_mutex.unlock(store_io); 444 449 try requireInitialized(); 445 450 446 - const row = try conn.row( 447 - \\SELECT data 448 - \\FROM repo_blocks 449 - \\WHERE did = ? AND cid = ? 450 - \\LIMIT 1 451 - , .{ self.did, cid }); 452 - if (row == null) return null; 453 - defer row.?.deinit(); 454 - 455 - const data = row.?.nullableBlob(0) orelse return null; 456 - return try self.allocator.dupe(u8, data); 451 + return repoBlockDataLocked(self.allocator, self.did, cid); 457 452 } 458 453 }; 454 + 455 + fn repoBlockDataLocked(allocator: std.mem.Allocator, did: []const u8, cid: []const u8) !?[]const u8 { 456 + const row = try conn.row( 457 + \\SELECT data 458 + \\FROM repo_blocks 459 + \\WHERE did = ? AND cid = ? 460 + \\LIMIT 1 461 + , .{ did, cid }); 462 + if (row == null) return null; 463 + defer row.?.deinit(); 464 + 465 + const data = row.?.nullableBlob(0) orelse return null; 466 + return try allocator.dupe(u8, data); 467 + } 459 468 460 469 var conn: zqlite.Conn = undefined; 461 470 var initialized = false; ··· 2684 2693 } 2685 2694 2686 2695 pub fn writeRepoCarSince(allocator: std.mem.Allocator, did: []const u8, since: ?[]const u8) ![]const u8 { 2696 + if (since == null) return writeRepoCarFull(allocator, did); 2697 + 2687 2698 db_mutex.lockUncancelable(store_io); 2688 2699 defer db_mutex.unlock(store_io); 2689 2700 try requireInitialized(); ··· 2692 2703 const root_raw = try zat.multibase.base32lower.decode(allocator, root.cid[1..]); 2693 2704 const car_root = zat.cbor.Cid{ .raw = root_raw }; 2694 2705 2695 - var rows = if (since) |since_rev| 2696 - try conn.rows( 2697 - \\SELECT cid, data 2698 - \\FROM repo_blocks 2699 - \\WHERE did = ? AND (repo_rev IS NULL OR repo_rev > ?) 2700 - \\ORDER BY repo_rev DESC, cid DESC 2701 - , .{ did, since_rev }) 2702 - else 2703 - try conn.rows( 2704 - \\SELECT cid, data 2705 - \\FROM repo_blocks 2706 - \\WHERE did = ? 2707 - \\ORDER BY cid ASC 2708 - , .{did}); 2706 + const since_rev = since.?; 2707 + var rows = try conn.rows( 2708 + \\SELECT cid, data 2709 + \\FROM repo_blocks 2710 + \\WHERE did = ? AND (repo_rev IS NULL OR repo_rev > ?) 2711 + \\ORDER BY repo_rev DESC, cid DESC 2712 + , .{ did, since_rev }); 2709 2713 defer rows.deinit(); 2710 2714 2711 2715 var blocks: std.ArrayList(zat.car.Block) = .empty; ··· 2724 2728 .blocks = blocks.items, 2725 2729 }; 2726 2730 return zat.car.writeAlloc(allocator, c); 2731 + } 2732 + 2733 + fn writeRepoCarFull(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { 2734 + db_mutex.lockUncancelable(store_io); 2735 + defer db_mutex.unlock(store_io); 2736 + try requireInitialized(); 2737 + 2738 + const root = try latestCommitRawLocked(allocator, did) orelse return Error.RepoNotFound; 2739 + var blocks: std.ArrayList(zat.car.Block) = .empty; 2740 + try blocks.append(allocator, .{ 2741 + .cid_raw = root.commit_cid_raw, 2742 + .data = root.commit_data, 2743 + }); 2744 + 2745 + var repo_block_reader = RepoBlockReader{ 2746 + .allocator = allocator, 2747 + .did = did, 2748 + .locked = true, 2749 + }; 2750 + try zat.mst.collectReachableBlocks( 2751 + allocator, 2752 + root.data_cid_raw, 2753 + repo_block_reader.reader(), 2754 + &blocks, 2755 + .{ .include_records = true }, 2756 + ); 2757 + 2758 + return zat.car.writeAlloc(allocator, .{ 2759 + .roots = &.{.{ .raw = root.commit_cid_raw }}, 2760 + .blocks = blocks.items, 2761 + }); 2727 2762 } 2728 2763 2729 2764 pub fn writeRepoListJson(allocator: std.mem.Allocator, cursor: ?[]const u8, limit: usize) ![]const u8 { ··· 5876 5911 try std.testing.expect(diff_car.blocks.len > 0); 5877 5912 try std.testing.expectEqual(@as(usize, 0), empty_diff_car.blocks.len); 5878 5913 try std.testing.expectEqualStrings(second_result.commit.cid, try cidText(allocator, empty_diff_car.roots[0].raw)); 5914 + } 5915 + 5916 + fn carContainsCid(allocator: std.mem.Allocator, c: zat.car.Car, cid: []const u8) !bool { 5917 + for (c.blocks) |block| { 5918 + if (std.mem.eql(u8, cid, try cidText(allocator, block.cid_raw))) return true; 5919 + } 5920 + return false; 5921 + } 5922 + 5923 + test "full getRepo exports only current reachable record blocks" { 5924 + var arena = std.heap.ArenaAllocator.init(std.testing.allocator); 5925 + defer arena.deinit(); 5926 + const allocator = arena.allocator(); 5927 + 5928 + try init(std.Options.debug_io, ":memory:"); 5929 + defer close(); 5930 + 5931 + const account = try createAccount( 5932 + allocator, 5933 + "reachable.test", 5934 + "reachable@test.com", 5935 + "password", 5936 + "did:plc:reachablecheck", 5937 + true, 5938 + ); 5939 + 5940 + const first = try std.json.parseFromSlice(std.json.Value, allocator, "{\"$type\":\"app.bsky.feed.post\",\"text\":\"first\"}", .{}); 5941 + defer first.deinit(); 5942 + const first_result = try applyWritesWithOptions(allocator, account, &.{.{ .create = .{ 5943 + .collection = "app.bsky.feed.post", 5944 + .rkey = "3zreach", 5945 + .value = first.value, 5946 + } }}, .{}); 5947 + const first_cid = first_result.records[0].cid; 5948 + 5949 + const second = try std.json.parseFromSlice(std.json.Value, allocator, "{\"$type\":\"app.bsky.feed.post\",\"text\":\"second\"}", .{}); 5950 + defer second.deinit(); 5951 + const second_result = try applyWritesWithOptions(allocator, account, &.{.{ .update = .{ 5952 + .collection = "app.bsky.feed.post", 5953 + .rkey = "3zreach", 5954 + .value = second.value, 5955 + } }}, .{}); 5956 + const second_cid = second_result.records[0].cid; 5957 + 5958 + const updated_car = try zat.car.read(allocator, try writeRepoCar(allocator, account.did)); 5959 + try std.testing.expect(!try carContainsCid(allocator, updated_car, first_cid)); 5960 + try std.testing.expect(try carContainsCid(allocator, updated_car, second_cid)); 5961 + 5962 + _ = try applyWritesWithOptions(allocator, account, &.{.{ .delete = .{ 5963 + .collection = "app.bsky.feed.post", 5964 + .rkey = "3zreach", 5965 + } }}, .{}); 5966 + 5967 + const deleted_car = try zat.car.read(allocator, try writeRepoCar(allocator, account.did)); 5968 + try std.testing.expect(!try carContainsCid(allocator, deleted_car, first_cid)); 5969 + try std.testing.expect(!try carContainsCid(allocator, deleted_car, second_cid)); 5879 5970 } 5880 5971 5881 5972 test "permissioned spaces store self-owned records outside public repo" {

Configure Feed

Configure Feed