atproto pds in zig
pds.zat.dev
pds
atproto
Add metastore-shaped benchmarks
+304
-20
+32
-19
bench/README.md
+32
-19
bench/README.md
···
12
12
just bench read
13
13
just bench repo
14
14
just bench blob
15
+
just bench metastore 10 1000
15
16
just bench run --scenario write --records 10000
16
17
```
17
18
···
24
25
- `read`: repeated `listRecords` queries against one seeded repo.
25
26
- `repo`: full repo CAR materialization through `writeRepoCar`.
26
27
- `blob`: blob write/read against disk blobstore plus SQLite metadata.
28
+
- `metastore`: Tranquil-shaped apply/get/list benchmark with caller counts and
29
+
latency percentiles.
27
30
28
31
Each run uses temporary SQLite and blobstore state.
29
32
···
47
50
SQLX_OFFLINE=true cargo bench -p tranquil-store --bench metastore --features tranquil-store/test-harness
48
51
```
49
52
50
-
The `metastore` bench is not an exact mirror of ZDS's current `write` scenario:
51
-
it measures Tranquil's storage commit path with generated commits and explicit
52
-
concurrency tiers, while ZDS currently measures sequential `applyWrites` through
53
-
the local store API. Treat this as an orientation pass, not a scoreboard.
53
+
ZDS now has a matching `metastore` scenario:
54
54
55
-
| operation | zds sequential | tranquil single caller |
56
-
|---|---:|---:|
57
-
| write/apply commit | 294 ops/s | 268 ops/s, p95 4.1 ms |
58
-
| list records | 30k ops/s | 27k ops/s, p95 53 us |
55
+
```sh
56
+
just bench metastore 1 1000
57
+
just bench metastore 10 1000
58
+
just bench metastore 100 200
59
+
just bench metastore 1000 50
60
+
```
59
61
60
-
Tranquil's same bench also reports concurrent storage pressure:
62
+
The operation classes match Tranquil's `metastore` bench: apply/write one
63
+
record, get one record CID/record, and list records from a seeded repo. The
64
+
implementations are not identical internally. Tranquil measures its metastore
65
+
handler pool and eventlog path; ZDS measures the local store API, including repo
66
+
commit construction and the current process-wide store mutex.
61
67
62
-
| operation | callers | tranquil |
63
-
|---|---:|---:|
64
-
| apply commit | 10 | 738 ops/s, p95 16.0 ms |
65
-
| apply commit | 100 | 888 ops/s, p95 123.9 ms |
66
-
| apply commit | 1000 | 1169 ops/s, then handler backpressure during reads |
67
-
| list records | 10 | 33k ops/s, p95 411 us |
68
-
| list records | 100 | 33k ops/s, p95 3.6 ms |
68
+
| operation | callers | ops | zds | tranquil |
69
+
|---|---:|---:|---:|---:|
70
+
| apply/write | 1 | 1000 | 259 ops/s, p95 8.7 ms | not measured at this count |
71
+
| apply/write | 1 | 5000 | did not finish in 150 s | 268 ops/s, p95 4.1 ms |
72
+
| apply/write | 10 | 10000 | 172 ops/s, p95 48.3 ms | 738 ops/s, p95 16.0 ms |
73
+
| apply/write | 100 | 20000 | 1531 ops/s, p95 32.1 ms | 888 ops/s, p95 123.9 ms |
74
+
| apply/write | 1000 | 50000 | did not finish in 150 s | 1169 ops/s, then read backpressure |
75
+
| get record | 1 | 1000 | 94k ops/s, p95 15 us | not measured at this count |
76
+
| get record | 10 | 10000 | 88k ops/s, p95 14 us | 339k ops/s, p95 49 us |
77
+
| get record | 100 | 20000 | 82k ops/s, p95 849 us | 425k ops/s, p95 349 us |
78
+
| list records | 1 | 1000 | 58k ops/s, p95 18 us | not measured at this count |
79
+
| list records | 10 | 10000 | 53k ops/s, p95 22 us | 33k ops/s, p95 411 us |
80
+
| list records | 100 | 20000 | 44k ops/s, p95 941 us | 33k ops/s, p95 3.6 ms |
69
81
70
-
The 1000-caller Tranquil run reached 1169 apply commits/sec, then the next read
71
-
phase failed with `metastore handler backpressure` on this machine. That is a
72
-
useful data point, not a completed clean run.
82
+
The ZDS 1-caller/5000-write tier and 1000-caller/50000-write tier were stopped
83
+
after roughly 150 seconds. The 1000-caller Tranquil run reached 1169 apply
84
+
commits/sec, then the next read phase failed with `metastore handler
85
+
backpressure` on this machine.
73
86
74
87
## comparison work
75
88
+4
bench/justfile
+4
bench/justfile
···
24
24
# benchmark full repo CAR materialization
25
25
repo:
26
26
{{zig}} build bench -Doptimize=ReleaseFast -- --scenario repo --records {{scale}}
27
+
28
+
# benchmark Tranquil metastore-shaped store operations
29
+
metastore callers="1" ops_per_caller="100":
30
+
{{zig}} build bench -Doptimize=ReleaseFast -- --scenario metastore --callers {{callers}} --ops-per-caller {{ops_per_caller}}
+268
-1
bench/main.zig
+268
-1
bench/main.zig
···
7
7
read,
8
8
blob,
9
9
repo,
10
+
metastore,
10
11
};
12
+
13
+
const bench_collection = "dev.zds.bench.record";
11
14
12
15
const Options = struct {
13
16
scenario: Scenario = .all,
14
17
records: usize = 1000,
15
18
blobs: usize = 128,
16
19
blob_size: usize = 64 * 1024,
20
+
callers: ?usize = null,
21
+
ops_per_caller: ?usize = null,
17
22
};
18
23
19
24
const BenchResult = struct {
···
87
92
(try benchRepoCar(allocator, state.account, options.records)).print();
88
93
},
89
94
.blob => (try benchBlob(allocator, state.account, options.blobs, options.blob_size)).print(),
95
+
.metastore => try benchMetastore(allocator, options),
90
96
}
91
97
}
92
98
···
184
190
return .{ .name = "blob put+get", .ops = blobs, .bytes = blobs * blob_size * 2, .elapsed_ns = nowNs() - start };
185
191
}
186
192
193
+
const ConcurrentStats = struct {
194
+
p50: u64,
195
+
p95: u64,
196
+
p99: u64,
197
+
max: u64,
198
+
mean: u64,
199
+
};
200
+
201
+
const ConcurrentResult = struct {
202
+
name: []const u8,
203
+
callers: usize,
204
+
ops: usize,
205
+
elapsed_ns: u64,
206
+
stats: ConcurrentStats,
207
+
208
+
fn print(self: ConcurrentResult) void {
209
+
const elapsed_s = @as(f64, @floatFromInt(self.elapsed_ns)) / std.time.ns_per_s;
210
+
const ops_per_s = @as(f64, @floatFromInt(self.ops)) / elapsed_s;
211
+
std.debug.print(
212
+
"{s: <16} callers={d: >4} ops={d: >6} {d: >10.1} ops/s p50={d:.3}ms p95={d:.3}ms p99={d:.3}ms max={d:.3}ms mean={d:.3}ms\n",
213
+
.{
214
+
self.name,
215
+
self.callers,
216
+
self.ops,
217
+
ops_per_s,
218
+
nsToMs(self.stats.p50),
219
+
nsToMs(self.stats.p95),
220
+
nsToMs(self.stats.p99),
221
+
nsToMs(self.stats.max),
222
+
nsToMs(self.stats.mean),
223
+
},
224
+
);
225
+
}
226
+
};
227
+
228
+
const ConcurrencyLevel = struct {
229
+
callers: usize,
230
+
ops_per_caller: usize,
231
+
};
232
+
233
+
fn benchMetastore(allocator: std.mem.Allocator, options: Options) !void {
234
+
const default_levels = [_]ConcurrencyLevel{
235
+
.{ .callers = 1, .ops_per_caller = 5000 },
236
+
.{ .callers = 10, .ops_per_caller = 1000 },
237
+
.{ .callers = 100, .ops_per_caller = 200 },
238
+
.{ .callers = 1000, .ops_per_caller = 50 },
239
+
};
240
+
241
+
std.debug.print("\n=== zds metastore-shaped benchmarks ===\n", .{});
242
+
if (options.callers) |callers| {
243
+
const level: ConcurrencyLevel = .{
244
+
.callers = callers,
245
+
.ops_per_caller = options.ops_per_caller orelse return error.MissingOpsPerCaller,
246
+
};
247
+
(try benchConcurrentWrites(allocator, level)).print();
248
+
(try benchConcurrentGets(allocator, level)).print();
249
+
(try benchConcurrentLists(allocator, level)).print();
250
+
return;
251
+
}
252
+
253
+
for (default_levels) |level| {
254
+
(try benchConcurrentWrites(allocator, level)).print();
255
+
(try benchConcurrentGets(allocator, level)).print();
256
+
(try benchConcurrentLists(allocator, level)).print();
257
+
}
258
+
}
259
+
260
+
fn benchConcurrentWrites(allocator: std.mem.Allocator, level: ConcurrencyLevel) !ConcurrentResult {
261
+
var state = try initBench(allocator);
262
+
defer state.deinit();
263
+
264
+
var arena = std.heap.ArenaAllocator.init(allocator);
265
+
defer arena.deinit();
266
+
const a = arena.allocator();
267
+
268
+
const accounts = try a.alloc(zds.auth.tokens.Account, level.callers);
269
+
for (accounts, 0..) |*account, i| {
270
+
account.* = try zds.storage.store.createAccount(
271
+
a,
272
+
try std.fmt.allocPrint(a, "bench-{d}.test", .{i}),
273
+
try std.fmt.allocPrint(a, "bench-{d}@test.com", .{i}),
274
+
"password",
275
+
try std.fmt.allocPrint(a, "did:plc:zdsbench{d}", .{i}),
276
+
true,
277
+
);
278
+
}
279
+
280
+
const total_ops = level.callers * level.ops_per_caller;
281
+
const latencies = try allocator.alloc(u64, total_ops);
282
+
defer allocator.free(latencies);
283
+
const threads = try allocator.alloc(std.Thread, level.callers);
284
+
defer allocator.free(threads);
285
+
286
+
const start = nowNs();
287
+
for (threads, 0..) |*thread, i| {
288
+
thread.* = try std.Thread.spawn(.{}, writeWorker, .{
289
+
accounts[i],
290
+
i * level.ops_per_caller,
291
+
latencies[i * level.ops_per_caller ..][0..level.ops_per_caller],
292
+
});
293
+
}
294
+
for (threads) |thread| thread.join();
295
+
296
+
return concurrentResult("applyWrites", level, total_ops, nowNs() - start, latencies);
297
+
}
298
+
299
+
fn benchConcurrentGets(allocator: std.mem.Allocator, level: ConcurrencyLevel) !ConcurrentResult {
300
+
var state = try initBench(allocator);
301
+
defer state.deinit();
302
+
try seedIndexedRecords(allocator, state.account, 1000);
303
+
304
+
const total_ops = level.callers * level.ops_per_caller;
305
+
const latencies = try allocator.alloc(u64, total_ops);
306
+
defer allocator.free(latencies);
307
+
const threads = try allocator.alloc(std.Thread, level.callers);
308
+
defer allocator.free(threads);
309
+
310
+
const start = nowNs();
311
+
for (threads, 0..) |*thread, i| {
312
+
thread.* = try std.Thread.spawn(.{}, getWorker, .{
313
+
state.account.did,
314
+
i,
315
+
latencies[i * level.ops_per_caller ..][0..level.ops_per_caller],
316
+
});
317
+
}
318
+
for (threads) |thread| thread.join();
319
+
320
+
return concurrentResult("getRecord", level, total_ops, nowNs() - start, latencies);
321
+
}
322
+
323
+
fn benchConcurrentLists(allocator: std.mem.Allocator, level: ConcurrencyLevel) !ConcurrentResult {
324
+
var state = try initBench(allocator);
325
+
defer state.deinit();
326
+
try seedIndexedRecords(allocator, state.account, 1000);
327
+
328
+
const total_ops = level.callers * level.ops_per_caller;
329
+
const latencies = try allocator.alloc(u64, total_ops);
330
+
defer allocator.free(latencies);
331
+
const threads = try allocator.alloc(std.Thread, level.callers);
332
+
defer allocator.free(threads);
333
+
334
+
const start = nowNs();
335
+
for (threads, 0..) |*thread, i| {
336
+
thread.* = try std.Thread.spawn(.{}, listWorker, .{
337
+
state.account.did,
338
+
latencies[i * level.ops_per_caller ..][0..level.ops_per_caller],
339
+
});
340
+
}
341
+
for (threads) |thread| thread.join();
342
+
343
+
return concurrentResult("listRecords", level, total_ops, nowNs() - start, latencies);
344
+
}
345
+
346
+
fn writeWorker(account: zds.auth.tokens.Account, offset: usize, latencies: []u64) !void {
347
+
var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
348
+
defer arena.deinit();
349
+
for (latencies, 0..) |*latency, i| {
350
+
_ = arena.reset(.retain_capacity);
351
+
const value = try benchRecordValue(arena.allocator(), offset + i);
352
+
const start = nowNs();
353
+
const result = try zds.storage.store.applyWrites(arena.allocator(), account, &.{.{ .create = .{
354
+
.collection = bench_collection,
355
+
.rkey = try std.fmt.allocPrint(arena.allocator(), "r{d:0>10}", .{offset + i}),
356
+
.value = value,
357
+
} }});
358
+
if (result.records.len != 1) return error.UnexpectedWriteResult;
359
+
latency.* = nowNs() - start;
360
+
}
361
+
}
362
+
363
+
fn getWorker(did: []const u8, task_id: usize, latencies: []u64) !void {
364
+
for (latencies, 0..) |*latency, i| {
365
+
const rec_idx = (task_id * 7 + i * 13) % 1000;
366
+
var rkey_buf: [32]u8 = undefined;
367
+
const rkey = try std.fmt.bufPrint(&rkey_buf, "rec{d:0>8}", .{rec_idx});
368
+
const start = nowNs();
369
+
const record = zds.storage.store.get(did, bench_collection, rkey) orelse return error.MissingRecord;
370
+
std.heap.page_allocator.free(record.did);
371
+
std.heap.page_allocator.free(record.collection);
372
+
std.heap.page_allocator.free(record.rkey);
373
+
std.heap.page_allocator.free(record.cid);
374
+
std.heap.page_allocator.free(record.value_json);
375
+
std.heap.page_allocator.free(record.validation_status);
376
+
std.heap.page_allocator.free(record.rev);
377
+
latency.* = nowNs() - start;
378
+
}
379
+
}
380
+
381
+
fn listWorker(did: []const u8, latencies: []u64) !void {
382
+
var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
383
+
defer arena.deinit();
384
+
for (latencies) |*latency| {
385
+
_ = arena.reset(.retain_capacity);
386
+
const start = nowNs();
387
+
const records = try zds.storage.store.listRecords(arena.allocator(), did, bench_collection, 50);
388
+
if (records.len == 0) return error.MissingRecords;
389
+
latency.* = nowNs() - start;
390
+
}
391
+
}
392
+
393
+
fn seedIndexedRecords(allocator: std.mem.Allocator, account: zds.auth.tokens.Account, target: usize) !void {
394
+
var arena = std.heap.ArenaAllocator.init(allocator);
395
+
defer arena.deinit();
396
+
for (0..target) |i| {
397
+
_ = arena.reset(.retain_capacity);
398
+
const value = try benchRecordValue(arena.allocator(), i);
399
+
const result = try zds.storage.store.applyWrites(arena.allocator(), account, &.{.{ .create = .{
400
+
.collection = bench_collection,
401
+
.rkey = try std.fmt.allocPrint(arena.allocator(), "rec{d:0>8}", .{i}),
402
+
.value = value,
403
+
} }});
404
+
if (result.records.len != 1) return error.UnexpectedWriteResult;
405
+
}
406
+
}
407
+
408
+
fn concurrentResult(name: []const u8, level: ConcurrencyLevel, total_ops: usize, elapsed_ns: u64, latencies: []u64) ConcurrentResult {
409
+
return .{
410
+
.name = name,
411
+
.callers = level.callers,
412
+
.ops = total_ops,
413
+
.elapsed_ns = elapsed_ns,
414
+
.stats = latencyStats(latencies),
415
+
};
416
+
}
417
+
418
+
fn latencyStats(latencies: []u64) ConcurrentStats {
419
+
std.mem.sort(u64, latencies, {}, std.sort.asc(u64));
420
+
var sum: u128 = 0;
421
+
for (latencies) |latency| sum += latency;
422
+
const last = latencies.len - 1;
423
+
return .{
424
+
.p50 = latencies[last * 50 / 100],
425
+
.p95 = latencies[last * 95 / 100],
426
+
.p99 = latencies[last * 99 / 100],
427
+
.max = latencies[last],
428
+
.mean = @intCast(sum / latencies.len),
429
+
};
430
+
}
431
+
432
+
fn nsToMs(ns: u64) f64 {
433
+
return @as(f64, @floatFromInt(ns)) / std.time.ns_per_ms;
434
+
}
435
+
187
436
fn seedRecords(allocator: std.mem.Allocator, account: zds.auth.tokens.Account, target: usize) !void {
188
437
const existing = try zds.storage.store.listRecords(allocator, account.did, "app.bsky.feed.post", 1);
189
438
if (existing.len > 0) return;
···
199
448
return try std.json.parseFromSliceLeaky(std.json.Value, allocator, json, .{});
200
449
}
201
450
451
+
fn benchRecordValue(allocator: std.mem.Allocator, index: usize) !std.json.Value {
452
+
const json = try std.fmt.allocPrint(
453
+
allocator,
454
+
"{{\"$type\":\"{s}\",\"text\":\"zds bench record {d}\",\"createdAt\":\"2026-05-23T00:{d:0>2}:00.000Z\"}}",
455
+
.{ bench_collection, index, index % 60 },
456
+
);
457
+
return try std.json.parseFromSliceLeaky(std.json.Value, allocator, json, .{});
458
+
}
459
+
202
460
fn parseOptions(init: std.process.Init) !Options {
203
461
var options: Options = .{};
204
462
var args = std.process.Args.Iterator.init(init.minimal.args);
···
220
478
options.blob_size = try std.fmt.parseInt(usize, args.next() orelse return error.MissingBlobSize, 10);
221
479
} else if (std.mem.startsWith(u8, arg, "--blob-size=")) {
222
480
options.blob_size = try std.fmt.parseInt(usize, arg["--blob-size=".len..], 10);
481
+
} else if (std.mem.eql(u8, arg, "--callers")) {
482
+
options.callers = try std.fmt.parseInt(usize, args.next() orelse return error.MissingCallers, 10);
483
+
} else if (std.mem.startsWith(u8, arg, "--callers=")) {
484
+
options.callers = try std.fmt.parseInt(usize, arg["--callers=".len..], 10);
485
+
} else if (std.mem.eql(u8, arg, "--ops-per-caller")) {
486
+
options.ops_per_caller = try std.fmt.parseInt(usize, args.next() orelse return error.MissingOpsPerCaller, 10);
487
+
} else if (std.mem.startsWith(u8, arg, "--ops-per-caller=")) {
488
+
options.ops_per_caller = try std.fmt.parseInt(usize, arg["--ops-per-caller=".len..], 10);
223
489
} else if (std.mem.eql(u8, arg, "--help") or std.mem.eql(u8, arg, "-h")) {
224
490
usage();
225
491
std.process.exit(0);
···
236
502
if (std.mem.eql(u8, value, "read")) return .read;
237
503
if (std.mem.eql(u8, value, "blob")) return .blob;
238
504
if (std.mem.eql(u8, value, "repo")) return .repo;
505
+
if (std.mem.eql(u8, value, "metastore")) return .metastore;
239
506
return error.UnknownScenario;
240
507
}
241
508
242
509
fn usage() void {
243
510
std.debug.print(
244
-
\\usage: zds-bench [--scenario all|write|read|blob|repo] [--records N] [--blobs N] [--blob-size BYTES]
511
+
\\usage: zds-bench [--scenario all|write|read|blob|repo|metastore] [--records N] [--blobs N] [--blob-size BYTES] [--callers N --ops-per-caller N]
245
512
\\
246
513
, .{});
247
514
}