atproto pds in zig
pds.zat.dev
pds
atproto
bench: add focused getRepo export benchmark
+116
-3
+11
-1
bench/README.md
+11
-1
bench/README.md
···
24
24
just bench get-record 10 1000
25
25
just bench list-records 10 1000
26
26
just bench repo-size 5000
27
+
just bench get-repo 5000 4 5
27
28
just bench repo-large 100000
28
29
just bench space 1000
29
30
just bench http
···
54
55
`listRecords(limit=100)` and full repo CAR materialization. The default Just
55
56
target keeps this moderate; use `repo-large` for intentional 100k+ record
56
57
runs.
58
+
- `get-repo`: focused `com.atproto.sync.getRepo` storage-pressure probe. It
59
+
measures full repo CAR construction, `since` diff CAR construction after one
60
+
additional commit, and concurrent full exports against one seeded repo.
57
61
- `space`: experimental permissioned-data storage probes for space discovery,
58
62
private record writes, private record reads/lists, repo oplog catch-up, and
59
63
blob readback.
···
121
125
whole-DID range-scan export path is intentionally no longer the production full
122
126
export behavior because it could include retained stale blocks.
123
127
128
+
`just bench get-repo` is the narrower operator-facing lane for the same sync
129
+
surface. It belongs here first because ZDS owns the storage and route behavior
130
+
for serving repo exports. Cross-PDS HTTP comparisons, compression behavior, and
131
+
live backfill traffic should live in `atproto-bench` once this local seam is
132
+
stable enough to compare against other implementations.
133
+
124
134
## methodology
125
135
126
136
- Build ZDS with `-Doptimize=ReleaseFast`.
···
201
211
|---|---|
202
212
| repo writes | `write`, `metastore`, `write-profile` |
203
213
| repo reads | `get-cid`, `get-block`, `get-record`, `list-records` |
204
-
| repo export/sync snapshots | `repo`, `repo-size` |
214
+
| repo export/sync snapshots | `repo`, `repo-size`, `get-repo` |
205
215
| sync event storage | `metastore` write path plus smoke tests |
206
216
| blob storage | `blob`, `space` blob readback |
207
217
| HTTP routing/serialization/auth overhead | `http` |
+4
bench/justfile
+4
bench/justfile
···
61
61
repo-size max_records="5000":
62
62
{{zig}} build bench -Doptimize=ReleaseFast -- --scenario repo-size --records {{max_records}}
63
63
64
+
# benchmark sync.getRepo storage pressure: full export, since export, and concurrent full exports
65
+
get-repo records="5000" callers="4" ops_per_caller="5":
66
+
{{zig}} build bench -Doptimize=ReleaseFast -- --scenario get-repo --records {{records}} --callers {{callers}} --ops-per-caller {{ops_per_caller}}
67
+
64
68
# deliberately slow large-repo lane for Paul/Jerry-shaped synthetic repositories
65
69
repo-large records="100000":
66
70
{{zig}} build bench -Doptimize=ReleaseFast -- --scenario repo-size --records {{records}}
+98
-1
bench/main.zig
+98
-1
bench/main.zig
···
15
15
get_record,
16
16
list_records,
17
17
repo_size,
18
+
get_repo,
18
19
space,
19
20
write_profile,
20
21
};
···
123
124
.get_record => try benchGetRecord(allocator, options),
124
125
.list_records => try benchListRecords(allocator, options),
125
126
.repo_size => try benchRepoSizes(allocator, options.records),
127
+
.get_repo => try benchGetRepo(allocator, options),
126
128
.space => try benchSpace(allocator, options),
127
129
.write_profile => try benchWriteProfile(allocator, options),
128
130
}
···
544
546
};
545
547
}
546
548
549
+
fn benchGetRepo(allocator: std.mem.Allocator, options: Options) !void {
550
+
const level: ConcurrencyLevel = .{
551
+
.callers = options.callers orelse 4,
552
+
.ops_per_caller = options.ops_per_caller orelse if (options.records >= 100_000) 1 else 5,
553
+
};
554
+
555
+
std.debug.print("\n=== zds getRepo benchmarks ===\n", .{});
556
+
std.debug.print("seeding repo with {d} records\n", .{options.records});
557
+
var state = try initBench(allocator);
558
+
defer state.deinit();
559
+
try seedIndexedRecords(allocator, state.account, options.records);
560
+
561
+
const since_rev = try appendIndexedRecordRev(allocator, state.account, options.records);
562
+
defer allocator.free(since_rev);
563
+
const latest_rev = try appendIndexedRecordRev(allocator, state.account, options.records + 1);
564
+
defer allocator.free(latest_rev);
565
+
566
+
(try benchRepoCarForRepo(allocator, state.account, options.records + 2)).print();
567
+
(try benchRepoCarSince(allocator, state.account, since_rev)).print();
568
+
(try benchConcurrentRepoCar(allocator, state.account, level)).print();
569
+
}
570
+
571
+
fn benchRepoCarSince(
572
+
allocator: std.mem.Allocator,
573
+
account: zds.auth.tokens.Account,
574
+
since_rev: []const u8,
575
+
) !BenchResult {
576
+
var arena = std.heap.ArenaAllocator.init(allocator);
577
+
defer arena.deinit();
578
+
const iterations: usize = 20;
579
+
var total_bytes: usize = 0;
580
+
const start = nowNs();
581
+
for (0..iterations) |_| {
582
+
_ = arena.reset(.retain_capacity);
583
+
const car = try zds.storage.store.writeRepoCarSince(arena.allocator(), account.did, since_rev);
584
+
if (car.len == 0) return error.EmptyRepoCar;
585
+
total_bytes += car.len;
586
+
}
587
+
return .{
588
+
.name = "getRepo since",
589
+
.ops = iterations,
590
+
.bytes = total_bytes,
591
+
.elapsed_ns = nowNs() - start,
592
+
};
593
+
}
594
+
595
+
fn benchConcurrentRepoCar(
596
+
allocator: std.mem.Allocator,
597
+
account: zds.auth.tokens.Account,
598
+
level: ConcurrencyLevel,
599
+
) !ConcurrentResult {
600
+
const total_ops = level.callers * level.ops_per_caller;
601
+
const latencies = try allocator.alloc(u64, total_ops);
602
+
defer allocator.free(latencies);
603
+
const threads = try allocator.alloc(std.Thread, level.callers);
604
+
defer allocator.free(threads);
605
+
606
+
const start = nowNs();
607
+
for (threads, 0..) |*thread, i| {
608
+
thread.* = try std.Thread.spawn(.{}, repoCarWorker, .{
609
+
account.did,
610
+
latencies[i * level.ops_per_caller ..][0..level.ops_per_caller],
611
+
});
612
+
}
613
+
for (threads) |thread| thread.join();
614
+
615
+
return concurrentResult("getRepo full", level, total_ops, nowNs() - start, latencies);
616
+
}
617
+
547
618
fn benchSpace(allocator: std.mem.Allocator, options: Options) !void {
548
619
var state = try initBench(allocator);
549
620
defer state.deinit();
···
842
913
}
843
914
}
844
915
916
+
fn repoCarWorker(did: []const u8, latencies: []u64) !void {
917
+
var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
918
+
defer arena.deinit();
919
+
for (latencies) |*latency| {
920
+
_ = arena.reset(.retain_capacity);
921
+
const start = nowNs();
922
+
const car = try zds.storage.store.writeRepoCar(arena.allocator(), did);
923
+
if (car.len == 0) return error.EmptyRepoCar;
924
+
latency.* = nowNs() - start;
925
+
}
926
+
}
927
+
845
928
const CpuPath = enum { decode, render };
846
929
847
930
fn benchRecordBlockCpu(allocator: std.mem.Allocator, options: Options, path: CpuPath) !void {
···
911
994
} }});
912
995
if (result.records.len != 1) return error.UnexpectedWriteResult;
913
996
}
997
+
}
998
+
999
+
fn appendIndexedRecordRev(allocator: std.mem.Allocator, account: zds.auth.tokens.Account, index: usize) ![]const u8 {
1000
+
var arena = std.heap.ArenaAllocator.init(allocator);
1001
+
defer arena.deinit();
1002
+
const value = try benchRecordValue(arena.allocator(), index);
1003
+
const result = try zds.storage.store.applyWrites(arena.allocator(), account, &.{.{ .create = .{
1004
+
.collection = bench_collection,
1005
+
.rkey = try std.fmt.allocPrint(arena.allocator(), "rec{d:0>8}", .{index}),
1006
+
.value = value,
1007
+
} }});
1008
+
if (result.records.len != 1) return error.UnexpectedWriteResult;
1009
+
return try allocator.dupe(u8, result.commit.rev);
914
1010
}
915
1011
916
1012
fn concurrentResult(name: []const u8, level: ConcurrencyLevel, total_ops: usize, elapsed_ns: u64, latencies: []u64) ConcurrentResult {
···
1027
1123
if (std.mem.eql(u8, value, "get-record")) return .get_record;
1028
1124
if (std.mem.eql(u8, value, "list-records")) return .list_records;
1029
1125
if (std.mem.eql(u8, value, "repo-size")) return .repo_size;
1126
+
if (std.mem.eql(u8, value, "get-repo")) return .get_repo;
1030
1127
if (std.mem.eql(u8, value, "space")) return .space;
1031
1128
if (std.mem.eql(u8, value, "write-profile")) return .write_profile;
1032
1129
return error.UnknownScenario;
···
1034
1131
1035
1132
fn usage() void {
1036
1133
std.debug.print(
1037
-
\\usage: zds-bench [--scenario all|write|read|blob|repo|metastore|get-cid|get-block|decode-record|render-record|get-record|list-records|repo-size|space|write-profile] [--records N] [--blobs N] [--blob-size BYTES] [--callers N --ops-per-caller N]
1134
+
\\usage: zds-bench [--scenario all|write|read|blob|repo|metastore|get-cid|get-block|decode-record|render-record|get-record|list-records|repo-size|get-repo|space|write-profile] [--records N] [--blobs N] [--blob-size BYTES] [--callers N --ops-per-caller N]
1038
1135
\\
1039
1136
, .{});
1040
1137
}
+3
-1
docs/getrepo-notes.md
+3
-1
docs/getrepo-notes.md
···
25
25
3. Changed full export from whole-DID `repo_blocks` scan to current-root reachability.
26
26
4. Added lightweight route backpressure for full `getRepo`, configurable with `ZDS_MAX_CONCURRENT_REPO_EXPORTS`.
27
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.
28
+
6. Added `just bench get-repo` as the focused storage-pressure lane for this route. It measures full export construction, one-commit `since` export construction, and concurrent full export latency against one seeded repo.
28
29
29
30
Suggested next pass:
30
31
31
-
1. Add a larger deliberate `just bench repo-large 100000` run when we want stress numbers comparable to large public repos.
32
+
1. Add a larger deliberate `just bench get-repo 100000 4 1` run when we want stress numbers comparable to large public repos.
33
+
2. Add a cross-PDS HTTP/export comparison in `atproto-bench` once the ZDS-local storage seam is stable.