atproto pds in zig
pds.zat.dev
pds
atproto
Split record index from repo content
+253
-106
+17
-16
bench/README.md
+17
-16
bench/README.md
···
84
84
The operations are matched to Tranquil's metastore bench: write one record, look
85
85
up one record CID, and list records from a seeded repo.
86
86
87
-
Summary: Tranquil is faster on CID lookups, by about 1.6x at 10 callers and
88
-
1.6x at 100 callers. ZDS is slower on 10-caller writes, at about 0.7x
89
-
Tranquil's throughput, but faster on 100-caller writes, at about 1.6x
90
-
Tranquil's throughput. ZDS is faster on list-record reads in these runs, by
91
-
about 1.4x at both 10 and 100 callers. The next read-path question is why
92
-
Tranquil's CID lookup is still ahead while ZDS's list path is already ahead.
87
+
Summary: Tranquil is faster on the read rows after ZDS moved record materialize
88
+
paths to decode DAG-CBOR from `repo_blocks`: about 1.5-1.8x faster for CID
89
+
lookup and about 1.6x faster for list records. Tranquil is also faster on
90
+
10-caller writes, by about 2.8x. ZDS is faster on the 100-caller write row, by
91
+
about 1.3x, which is the main sign that the sharded write lane is helping under
92
+
contention.
93
93
94
94
| operation | callers | ops | zds | tranquil |
95
95
|---|---:|---:|---:|---:|
96
-
| apply/write | 10 | 10000 | 299 ops/s, p95 84 ms | 436 ops/s, p95 36.0 ms |
97
-
| apply/write | 100 | 20000 | 1502 ops/s, p95 56 ms | 944 ops/s, p95 119 ms |
98
-
| get record CID | 10 | 10000 | 355k ops/s, p95 5 us | 571k ops/s, p95 31 us |
99
-
| get record CID | 100 | 20000 | 349k ops/s, p95 6 us | 571k ops/s, p95 244 us |
100
-
| list records | 10 | 10000 | 52k ops/s, p95 20 us | 38k ops/s, p95 309 us |
101
-
| list records | 100 | 20000 | 51k ops/s, p95 22 us | 37k ops/s, p95 3.0 ms |
96
+
| apply/write | 10 | 10000 | 282 ops/s, p95 93.8 ms | 780 ops/s, p95 14.1 ms |
97
+
| apply/write | 100 | 20000 | 1357 ops/s, p95 66.7 ms | 1007 ops/s, p95 109 ms |
98
+
| get record CID | 10 | 10000 | 296k ops/s, p95 6 us | 455k ops/s, p95 43 us |
99
+
| get record CID | 100 | 20000 | 294k ops/s, p95 405 us | 537k ops/s, p95 233 us |
100
+
| list records | 10 | 10000 | 22k ops/s, p95 52 us | 36k ops/s, p95 310 us |
101
+
| list records | 100 | 20000 | 23k ops/s, p95 49 us | 36k ops/s, p95 2.9 ms |
102
102
103
103
Tranquil's read row is `get_record_cid`, so ZDS reports `get-cid` for that
104
104
comparison. Full `com.atproto.repo.getRecord` is a separate local probe because
105
-
it materializes record JSON and response fields that Tranquil's metastore bench
106
-
does not request.
105
+
it joins the record index to `repo_blocks`, decodes DAG-CBOR, and renders the
106
+
XRPC JSON response shape. Tranquil's metastore bench intentionally measures a
107
+
metadata lookup for the current CID, not record body materialization.
107
108
108
109
## local probes
109
110
···
111
112
112
113
| operation | callers | ops | zds |
113
114
|---|---:|---:|---:|
114
-
| full getRecord | 10 | 10000 | 56k ops/s, p95 65 us |
115
-
| full getRecord | 100 | 20000 | 52k ops/s, p95 7.2 ms |
115
+
| full getRecord | 10 | 10000 | 40k ops/s, p95 352 us |
116
+
| full getRecord | 100 | 20000 | 39k ops/s, p95 8.2 ms |
116
117
117
118
## write profile
118
119
-43
src/atproto/repo.zig
-43
src/atproto/repo.zig
···
461
461
for (node.entries.items) |entry| {
462
462
const block = zat.car.findBlock(repo_car, entry.value.raw) orelse return error.MissingRecordBlock;
463
463
const record_value = try zat.cbor.decodeAll(allocator, block);
464
-
const record_json = try cborToJson(allocator, record_value);
465
464
var blobs: std.ArrayList([]const u8) = .empty;
466
465
try collectBlobCids(allocator, record_value, &blobs);
467
466
const slash = std.mem.indexOfScalar(u8, entry.key, '/') orelse return error.InvalidRepoPath;
···
469
468
.collection = try allocator.dupe(u8, entry.key[0..slash]),
470
469
.rkey = try allocator.dupe(u8, entry.key[slash + 1 ..]),
471
470
.cid = try cidString(allocator, entry.value.raw),
472
-
.value_json = record_json,
473
471
.blob_cids = try blobs.toOwnedSlice(allocator),
474
472
});
475
473
try collectFromChild(allocator, repo_car, entry.right, out);
···
486
484
.none => {},
487
485
.stub => return error.PartialTree,
488
486
.node => |node| try collectImportedRecords(allocator, repo_car, node, out),
489
-
}
490
-
}
491
-
492
-
fn cborToJson(allocator: std.mem.Allocator, value: zat.cbor.Value) ![]const u8 {
493
-
var out: std.Io.Writer.Allocating = .init(allocator);
494
-
defer out.deinit();
495
-
try writeCborJson(allocator, &out.writer, value);
496
-
return out.toOwnedSlice();
497
-
}
498
-
499
-
fn writeCborJson(allocator: std.mem.Allocator, writer: anytype, value: zat.cbor.Value) !void {
500
-
switch (value) {
501
-
.unsigned => |v| try writer.print("{d}", .{v}),
502
-
.negative => |v| try writer.print("{d}", .{v}),
503
-
.bytes => |bytes| {
504
-
const encoded = try allocator.alloc(u8, std.base64.standard.Encoder.calcSize(bytes.len));
505
-
defer allocator.free(encoded);
506
-
_ = std.base64.standard.Encoder.encode(encoded, bytes);
507
-
try writer.print("{{\"$bytes\":{f}}}", .{std.json.fmt(encoded, .{})});
508
-
},
509
-
.text => |text| try writer.print("{f}", .{std.json.fmt(text, .{})}),
510
-
.array => |items| {
511
-
try writer.writeByte('[');
512
-
for (items, 0..) |item, idx| {
513
-
if (idx != 0) try writer.writeByte(',');
514
-
try writeCborJson(allocator, writer, item);
515
-
}
516
-
try writer.writeByte(']');
517
-
},
518
-
.map => |entries| {
519
-
try writer.writeByte('{');
520
-
for (entries, 0..) |entry, idx| {
521
-
if (idx != 0) try writer.writeByte(',');
522
-
try writer.print("{f}:", .{std.json.fmt(entry.key, .{})});
523
-
try writeCborJson(allocator, writer, entry.value);
524
-
}
525
-
try writer.writeByte('}');
526
-
},
527
-
.boolean => |v| try writer.print("{}", .{v}),
528
-
.null => try writer.writeAll("null"),
529
-
.cid => |cid| try writer.print("{{\"$link\":{f}}}", .{std.json.fmt(try cidString(allocator, cid.raw), .{})}),
530
487
}
531
488
}
532
489
+46
src/internal/cbor_json.zig
+46
src/internal/cbor_json.zig
···
1
+
const std = @import("std");
2
+
const zat = @import("zat");
3
+
4
+
pub fn writeAlloc(allocator: std.mem.Allocator, value: zat.cbor.Value) ![]const u8 {
5
+
var out: std.Io.Writer.Allocating = .init(allocator);
6
+
defer out.deinit();
7
+
try writeValue(allocator, &out.writer, value);
8
+
return out.toOwnedSlice();
9
+
}
10
+
11
+
fn writeValue(allocator: std.mem.Allocator, writer: anytype, value: zat.cbor.Value) !void {
12
+
switch (value) {
13
+
.unsigned => |v| try writer.print("{d}", .{v}),
14
+
.negative => |v| try writer.print("{d}", .{v}),
15
+
.bytes => |bytes| {
16
+
const encoded = try allocator.alloc(u8, std.base64.standard.Encoder.calcSize(bytes.len));
17
+
defer allocator.free(encoded);
18
+
_ = std.base64.standard.Encoder.encode(encoded, bytes);
19
+
try writer.print("{{\"$bytes\":{f}}}", .{std.json.fmt(encoded, .{})});
20
+
},
21
+
.text => |text| try writer.print("{f}", .{std.json.fmt(text, .{})}),
22
+
.array => |items| {
23
+
try writer.writeByte('[');
24
+
for (items, 0..) |item, idx| {
25
+
if (idx != 0) try writer.writeByte(',');
26
+
try writeValue(allocator, writer, item);
27
+
}
28
+
try writer.writeByte(']');
29
+
},
30
+
.map => |entries| {
31
+
try writer.writeByte('{');
32
+
for (entries, 0..) |entry, idx| {
33
+
if (idx != 0) try writer.writeByte(',');
34
+
try writer.print("{f}:", .{std.json.fmt(entry.key, .{})});
35
+
try writeValue(allocator, writer, entry.value);
36
+
}
37
+
try writer.writeByte('}');
38
+
},
39
+
.boolean => |v| try writer.print("{}", .{v}),
40
+
.null => try writer.writeAll("null"),
41
+
.cid => |cid| try writer.print(
42
+
"{{\"$link\":{f}}}",
43
+
.{std.json.fmt(try zat.multibase.base32lower.encode(allocator, cid.raw), .{})},
44
+
),
45
+
}
46
+
}
+190
-47
src/storage/store.zig
+190
-47
src/storage/store.zig
···
2
2
const atid = @import("../core/atid.zig");
3
3
const auth = @import("../auth/tokens.zig");
4
4
const blobstore = @import("blobstore.zig");
5
+
const cbor_json = @import("../internal/cbor_json.zig");
5
6
const eventlog = @import("eventlog.zig");
6
7
const sharded_locks = @import("../internal/sharded_locks.zig");
7
8
const zat = @import("zat");
···
13
14
InvalidRecordKey,
14
15
InvalidRecordType,
15
16
MissingRecord,
17
+
MissingRecordBlock,
16
18
RepoNotFound,
17
19
InvalidRepoPath,
18
20
InvalidDagCbor,
···
115
117
collection: []const u8,
116
118
rkey: []const u8,
117
119
cid: []const u8,
118
-
value_json: []const u8,
119
120
blob_cids: []const []const u8,
120
121
};
121
122
···
309
310
try requireInitialized();
310
311
311
312
const row = try conn.row(
312
-
\\SELECT value_json
313
-
\\FROM records
314
-
\\WHERE did = ? AND collection = 'app.bsky.actor.profile' AND rkey = 'self'
313
+
\\SELECT rb.data
314
+
\\FROM records r
315
+
\\JOIN repo_blocks rb ON rb.did = r.did AND rb.cid = r.cid
316
+
\\WHERE r.did = ? AND r.collection = 'app.bsky.actor.profile' AND r.rkey = 'self'
315
317
\\LIMIT 1
316
318
, .{did});
317
319
if (row == null) return null;
318
320
defer row.?.deinit();
319
321
320
-
var parsed = std.json.parseFromSlice(std.json.Value, allocator, row.?.text(0), .{}) catch return null;
322
+
const profile_json = recordJsonFromBlock(allocator, row.?.nullableBlob(0) orelse "") catch return null;
323
+
var parsed = std.json.parseFromSlice(std.json.Value, allocator, profile_json, .{}) catch return null;
321
324
defer parsed.deinit();
322
325
if (parsed.value != .object) return null;
323
326
const avatar = parsed.value.object.get("avatar") orelse return null;
···
823
826
for (records.items) |record| {
824
827
const uri = try record.uri(allocator);
825
828
try conn.exec(
826
-
\\INSERT INTO records (did, collection, rkey, uri, cid, value_json, rev, seq)
827
-
\\VALUES (?, ?, ?, ?, ?, ?, ?, ?)
829
+
\\INSERT INTO records (did, collection, rkey, uri, cid, rev, seq)
830
+
\\VALUES (?, ?, ?, ?, ?, ?, ?)
828
831
\\ON CONFLICT(did, collection, rkey) DO UPDATE SET
829
832
\\ uri = excluded.uri,
830
833
\\ cid = excluded.cid,
831
-
\\ value_json = excluded.value_json,
832
834
\\ rev = excluded.rev,
833
835
\\ seq = excluded.seq
834
-
, .{ record.did, record.collection, record.rkey, uri, record.cid, record.value_json, record.rev, @as(i64, @intCast(seq)) });
836
+
, .{ record.did, record.collection, record.rkey, uri, record.cid, record.rev, @as(i64, @intCast(seq)) });
835
837
try conn.exec("DELETE FROM expected_blobs WHERE record_uri = ?", .{uri});
836
838
}
837
839
for (blob_refs.items) |ref| {
···
945
947
requireInitialized() catch return null;
946
948
947
949
const row = conn.row(
948
-
\\SELECT did, collection, rkey, cid, value_json, rev, seq
949
-
\\FROM records
950
-
\\WHERE did = ? AND collection = ? AND rkey = ?
950
+
\\SELECT r.did, r.collection, r.rkey, r.cid, rb.data, r.rev, r.seq
951
+
\\FROM records r
952
+
\\JOIN repo_blocks rb ON rb.did = r.did AND rb.cid = r.cid
953
+
\\WHERE r.did = ? AND r.collection = ? AND r.rkey = ?
951
954
, .{ did, collection, rkey }) catch return null;
952
955
if (row == null) return null;
953
956
defer row.?.deinit();
···
982
985
try requireInitialized();
983
986
984
987
var rows = try conn.rows(
985
-
\\SELECT did, collection, rkey, cid, value_json, rev, seq
986
-
\\FROM records
987
-
\\ORDER BY seq DESC
988
+
\\SELECT r.did, r.collection, r.rkey, r.cid, rb.data, r.rev, r.seq
989
+
\\FROM records r
990
+
\\JOIN repo_blocks rb ON rb.did = r.did AND rb.cid = r.cid
991
+
\\ORDER BY r.seq DESC
988
992
\\LIMIT ?
989
993
, .{@as(i64, @intCast(if (limit == 0) 100 else limit))});
990
994
defer rows.deinit();
···
1008
1012
try requireInitialized();
1009
1013
1010
1014
var rows = try conn.rows(
1011
-
\\SELECT did, collection, rkey, cid, value_json, rev, seq
1012
-
\\FROM records
1013
-
\\WHERE did = ? AND collection = ?
1014
-
\\ORDER BY seq DESC
1015
+
\\SELECT r.did, r.collection, r.rkey, r.cid, rb.data, r.rev, r.seq
1016
+
\\FROM records r
1017
+
\\JOIN repo_blocks rb ON rb.did = r.did AND rb.cid = r.cid
1018
+
\\WHERE r.did = ? AND r.collection = ?
1019
+
\\ORDER BY r.seq DESC
1015
1020
\\LIMIT ?
1016
1021
, .{ did, collection, @as(i64, @intCast(if (limit == 0) 100 else limit)) });
1017
1022
defer rows.deinit();
···
1035
1040
try requireInitialized();
1036
1041
1037
1042
var rows = try conn.rows(
1038
-
\\SELECT did, collection, rkey, cid, value_json, rev, seq
1039
-
\\FROM records
1040
-
\\WHERE collection = ? AND instr(value_json, ?)
1041
-
\\ORDER BY seq DESC
1043
+
\\SELECT r.did, r.collection, r.rkey, r.cid, rb.data, r.rev, r.seq
1044
+
\\FROM records r
1045
+
\\JOIN repo_blocks rb ON rb.did = r.did AND rb.cid = r.cid
1046
+
\\WHERE r.collection = ?
1047
+
\\ORDER BY r.seq DESC
1042
1048
\\LIMIT ?
1043
-
, .{ collection, needle, @as(i64, @intCast(if (limit == 0) 100 else limit)) });
1049
+
, .{ collection, @as(i64, @intCast(if (limit == 0) 100 else limit)) });
1044
1050
defer rows.deinit();
1045
1051
1046
1052
var records: std.ArrayList(Record) = .empty;
1047
1053
while (rows.next()) |row| {
1048
-
try records.append(allocator, try recordFromRow(row, allocator));
1054
+
const record = try recordFromRow(row, allocator);
1055
+
if (std.mem.indexOf(u8, record.value_json, needle) != null) {
1056
+
try records.append(allocator, record);
1057
+
}
1049
1058
}
1050
1059
if (rows.err) |err| return err;
1051
1060
return records.toOwnedSlice(allocator);
···
1063
1072
try requireInitialized();
1064
1073
1065
1074
var rows = try conn.rows(
1066
-
\\SELECT did, collection, rkey, cid, value_json, rev, seq
1067
-
\\FROM records
1068
-
\\WHERE did = ? AND collection = ? AND instr(value_json, ?)
1069
-
\\ORDER BY seq DESC
1075
+
\\SELECT r.did, r.collection, r.rkey, r.cid, rb.data, r.rev, r.seq
1076
+
\\FROM records r
1077
+
\\JOIN repo_blocks rb ON rb.did = r.did AND rb.cid = r.cid
1078
+
\\WHERE r.did = ? AND r.collection = ?
1079
+
\\ORDER BY r.seq DESC
1070
1080
\\LIMIT ?
1071
-
, .{ did, collection, needle, @as(i64, @intCast(if (limit == 0) 100 else limit)) });
1081
+
, .{ did, collection, @as(i64, @intCast(if (limit == 0) 100 else limit)) });
1072
1082
defer rows.deinit();
1073
1083
1074
1084
var records: std.ArrayList(Record) = .empty;
1075
1085
while (rows.next()) |row| {
1076
-
try records.append(allocator, try recordFromRow(row, allocator));
1086
+
const record = try recordFromRow(row, allocator);
1087
+
if (std.mem.indexOf(u8, record.value_json, needle) != null) {
1088
+
try records.append(allocator, record);
1089
+
}
1077
1090
}
1078
1091
if (rows.err) |err| return err;
1079
1092
return records.toOwnedSlice(allocator);
···
1093
1106
1094
1107
fn getByStoredUriLocked(uri: []const u8) ?Record {
1095
1108
const row = conn.row(
1096
-
\\SELECT did, collection, rkey, cid, value_json, rev, seq
1097
-
\\FROM records
1098
-
\\WHERE uri = ?
1109
+
\\SELECT r.did, r.collection, r.rkey, r.cid, rb.data, r.rev, r.seq
1110
+
\\FROM records r
1111
+
\\JOIN repo_blocks rb ON rb.did = r.did AND rb.cid = r.cid
1112
+
\\WHERE r.uri = ?
1099
1113
, .{uri}) catch return null;
1100
1114
if (row == null) return null;
1101
1115
defer row.?.deinit();
···
1116
1130
db_mutex.lockUncancelable(store_io);
1117
1131
defer db_mutex.unlock(store_io);
1118
1132
requireInitialized() catch return 0;
1119
-
const row = conn.row("SELECT count(*) FROM records WHERE collection = ? AND instr(value_json, ?) > 0", .{ collection, subject_did }) catch return 0;
1120
-
if (row == null) return 0;
1121
-
defer row.?.deinit();
1122
-
return @intCast(row.?.int(0));
1133
+
var rows = conn.rows(
1134
+
\\SELECT rb.data
1135
+
\\FROM records r
1136
+
\\JOIN repo_blocks rb ON rb.did = r.did AND rb.cid = r.cid
1137
+
\\WHERE r.collection = ?
1138
+
, .{collection}) catch return 0;
1139
+
defer rows.deinit();
1140
+
1141
+
var count_result: usize = 0;
1142
+
var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
1143
+
defer arena.deinit();
1144
+
while (rows.next()) |row| {
1145
+
_ = arena.reset(.retain_capacity);
1146
+
const value = recordJsonFromBlock(arena.allocator(), row.nullableBlob(0) orelse "") catch continue;
1147
+
if (std.mem.indexOf(u8, value, subject_did) != null) count_result += 1;
1148
+
}
1149
+
return count_result;
1123
1150
}
1124
1151
1125
1152
pub fn listCollectionsJson(allocator: std.mem.Allocator, did: []const u8) ![]const u8 {
···
1390
1417
const uri = try std.fmt.allocPrint(std.heap.page_allocator, "at://{s}/{s}/{s}", .{ account.did, record.collection, record.rkey });
1391
1418
defer std.heap.page_allocator.free(uri);
1392
1419
try conn.exec(
1393
-
\\INSERT INTO records (did, collection, rkey, uri, cid, value_json, rev, seq)
1394
-
\\VALUES (?, ?, ?, ?, ?, ?, ?, ?)
1395
-
, .{ account.did, record.collection, record.rkey, uri, record.cid, record.value_json, rev, @as(i64, @intCast(seq)) });
1420
+
\\INSERT INTO records (did, collection, rkey, uri, cid, rev, seq)
1421
+
\\VALUES (?, ?, ?, ?, ?, ?, ?)
1422
+
, .{ account.did, record.collection, record.rkey, uri, record.cid, rev, @as(i64, @intCast(seq)) });
1396
1423
for (record.blob_cids) |blob_cid| {
1397
1424
try conn.exec(
1398
1425
\\INSERT INTO expected_blobs (blob_cid, record_uri)
···
1587
1614
try requireInitialized();
1588
1615
1589
1616
var rows = try conn.rows(
1590
-
\\SELECT did, collection, rkey, cid, value_json, rev, seq
1591
-
\\FROM records
1592
-
\\WHERE did = ? AND collection = ?
1593
-
\\ORDER BY seq DESC
1617
+
\\SELECT r.did, r.collection, r.rkey, r.cid, rb.data, r.rev, r.seq
1618
+
\\FROM records r
1619
+
\\JOIN repo_blocks rb ON rb.did = r.did AND rb.cid = r.cid
1620
+
\\WHERE r.did = ? AND r.collection = ?
1621
+
\\ORDER BY r.seq DESC
1594
1622
\\LIMIT ?
1595
1623
, .{ did, collection, @as(i64, @intCast(if (limit == 0) 100 else limit)) });
1596
1624
defer rows.deinit();
···
1631
1659
try migrateBlobTable();
1632
1660
try migrateAppPreferencesTable();
1633
1661
inline for (post_schema_statements) |sql| try conn.execNoArgs(sql);
1662
+
try migrateRecordsTable();
1663
+
try validateRecordBlocksPresent();
1664
+
}
1665
+
1666
+
fn migrateRecordsTable() !void {
1667
+
if (!try recordsTableHasValueJsonColumn()) return;
1668
+
try conn.execNoArgs("DROP TABLE IF EXISTS records_next");
1669
+
try conn.execNoArgs(
1670
+
\\CREATE TABLE IF NOT EXISTS records_next (
1671
+
\\ did TEXT NOT NULL REFERENCES accounts(did) ON DELETE CASCADE,
1672
+
\\ collection TEXT NOT NULL,
1673
+
\\ rkey TEXT NOT NULL,
1674
+
\\ uri TEXT NOT NULL UNIQUE,
1675
+
\\ cid TEXT NOT NULL,
1676
+
\\ rev TEXT NOT NULL,
1677
+
\\ seq INTEGER NOT NULL,
1678
+
\\ PRIMARY KEY (did, collection, rkey)
1679
+
\\)
1680
+
);
1681
+
try conn.execNoArgs(
1682
+
\\INSERT INTO records_next (did, collection, rkey, uri, cid, rev, seq)
1683
+
\\SELECT did, collection, rkey, uri, cid, rev, seq
1684
+
\\FROM records
1685
+
);
1686
+
try conn.execNoArgs("PRAGMA foreign_keys = OFF");
1687
+
errdefer conn.execNoArgs("PRAGMA foreign_keys = ON") catch {};
1688
+
try conn.execNoArgs("DROP TABLE records");
1689
+
try conn.execNoArgs("ALTER TABLE records_next RENAME TO records");
1690
+
try conn.execNoArgs("PRAGMA foreign_keys = ON");
1691
+
try conn.execNoArgs("CREATE INDEX IF NOT EXISTS records_collection_idx ON records (did, collection, seq DESC)");
1692
+
try conn.execNoArgs("CREATE INDEX IF NOT EXISTS records_cid_idx ON records (cid)");
1693
+
}
1694
+
1695
+
fn recordsTableHasValueJsonColumn() !bool {
1696
+
var rows = try conn.rows("PRAGMA table_info(records)", .{});
1697
+
defer rows.deinit();
1698
+
while (rows.next()) |row| {
1699
+
if (std.mem.eql(u8, row.text(1), "value_json")) return true;
1700
+
}
1701
+
if (rows.err) |err| return err;
1702
+
return false;
1703
+
}
1704
+
1705
+
fn validateRecordBlocksPresent() !void {
1706
+
const row = try conn.row(
1707
+
\\SELECT COUNT(*)
1708
+
\\FROM records r
1709
+
\\LEFT JOIN repo_blocks rb ON rb.did = r.did AND rb.cid = r.cid
1710
+
\\WHERE rb.cid IS NULL
1711
+
, .{});
1712
+
if (row == null) return;
1713
+
defer row.?.deinit();
1714
+
if (row.?.int(0) != 0) return Error.MissingRecordBlock;
1634
1715
}
1635
1716
1636
1717
fn migrateAppPreferencesTable() !void {
···
2316
2397
}
2317
2398
2318
2399
fn recordFromRow(row: zqlite.Row, allocator: std.mem.Allocator) !Record {
2400
+
const record_bytes = row.nullableBlob(4) orelse return Error.MissingRecordBlock;
2319
2401
return .{
2320
2402
.did = try allocator.dupe(u8, row.text(0)),
2321
2403
.collection = try allocator.dupe(u8, row.text(1)),
2322
2404
.rkey = try allocator.dupe(u8, row.text(2)),
2323
2405
.cid = try allocator.dupe(u8, row.text(3)),
2324
-
.value_json = try allocator.dupe(u8, row.text(4)),
2406
+
.value_json = try recordJsonFromBlock(allocator, record_bytes),
2325
2407
.validation_status = validationStatusForRecord(row.text(1)),
2326
2408
.rev = try allocator.dupe(u8, row.text(5)),
2327
2409
.seq = @intCast(row.int(6)),
2328
2410
};
2411
+
}
2412
+
2413
+
fn recordJsonFromBlock(allocator: std.mem.Allocator, data: []const u8) ![]const u8 {
2414
+
const value = zat.cbor.decodeAll(allocator, data) catch return Error.InvalidDagCbor;
2415
+
return cbor_json.writeAlloc(allocator, value);
2329
2416
}
2330
2417
2331
2418
fn oauthRequestFromRow(row: zqlite.Row, allocator: std.mem.Allocator) !OAuthRequest {
···
2712
2799
\\ rkey TEXT NOT NULL,
2713
2800
\\ uri TEXT NOT NULL UNIQUE,
2714
2801
\\ cid TEXT NOT NULL,
2715
-
\\ value_json BLOB NOT NULL,
2716
2802
\\ rev TEXT NOT NULL,
2717
2803
\\ seq INTEGER NOT NULL,
2718
2804
\\ PRIMARY KEY (did, collection, rkey)
···
2832
2918
2833
2919
const record = try create(allocator, account, "app.bsky.feed.post", "3ztest", parsed.value);
2834
2920
try std.testing.expectEqualStrings("3ztest", record.rkey);
2921
+
try std.testing.expect(!try recordsTableHasValueJsonColumn());
2922
+
2923
+
const block_row = try conn.row(
2924
+
"SELECT count(*) FROM repo_blocks WHERE did = ? AND cid = ?",
2925
+
.{ account.did, record.cid },
2926
+
);
2927
+
try std.testing.expect(block_row != null);
2928
+
defer block_row.?.deinit();
2929
+
try std.testing.expectEqual(@as(i64, 1), block_row.?.int(0));
2835
2930
2836
2931
const fetched = get(account.did, "app.bsky.feed.post", "3ztest").?;
2837
2932
try std.testing.expectEqualStrings(record.cid, fetched.cid);
2838
2933
try std.testing.expect(std.mem.indexOf(u8, fetched.value_json, "hello") != null);
2934
+
2935
+
const listed = try writeListJson(allocator, account.did, "app.bsky.feed.post", 10);
2936
+
try std.testing.expect(std.mem.indexOf(u8, listed, "hello") != null);
2839
2937
2840
2938
const repo_car = try writeRepoCar(allocator, account.did);
2841
2939
const loaded = try zat.loadCommitFromCAR(allocator, repo_car);
···
2846
2944
const tree = try zat.mst.Mst.loadFromBlocks(allocator, loaded.repo_car, loaded.commit.data_cid);
2847
2945
const found = tree.get("app.bsky.feed.post/3ztest") orelse return error.MissingRecord;
2848
2946
try std.testing.expectEqualStrings(record.cid, try cidText(allocator, found.raw));
2947
+
2948
+
try conn.exec(
2949
+
"DELETE FROM repo_blocks WHERE did = ? AND cid = ?",
2950
+
.{ account.did, record.cid },
2951
+
);
2952
+
try std.testing.expect(get(account.did, "app.bsky.feed.post", "3ztest") == null);
2953
+
}
2954
+
2955
+
test "put updates record index while content comes from repo blocks" {
2956
+
var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
2957
+
defer arena.deinit();
2958
+
const allocator = arena.allocator();
2959
+
2960
+
try init(std.Options.debug_io, ":memory:");
2961
+
defer close();
2962
+
2963
+
const account = try createAccount(
2964
+
allocator,
2965
+
"alice.test",
2966
+
"alice@test.com",
2967
+
"password",
2968
+
"did:plc:cmadossymmii3izkabdbp5en",
2969
+
true,
2970
+
);
2971
+
2972
+
var first = try std.json.parseFromSlice(std.json.Value, allocator, "{\"text\":\"before\"}", .{});
2973
+
defer first.deinit();
2974
+
_ = try create(allocator, account, "app.bsky.feed.post", "3zput", first.value);
2975
+
2976
+
var second = try std.json.parseFromSlice(std.json.Value, allocator, "{\"text\":\"after\"}", .{});
2977
+
defer second.deinit();
2978
+
const updated = try put(allocator, account, "app.bsky.feed.post", "3zput", second.value);
2979
+
2980
+
const fetched = get(account.did, "app.bsky.feed.post", "3zput").?;
2981
+
try std.testing.expectEqualStrings(updated.cid, fetched.cid);
2982
+
try std.testing.expect(std.mem.indexOf(u8, fetched.value_json, "after") != null);
2983
+
try std.testing.expect(std.mem.indexOf(u8, fetched.value_json, "before") == null);
2984
+
2985
+
const block_row = try conn.row(
2986
+
"SELECT count(*) FROM repo_blocks WHERE did = ? AND cid = ?",
2987
+
.{ account.did, updated.cid },
2988
+
);
2989
+
try std.testing.expect(block_row != null);
2990
+
defer block_row.?.deinit();
2991
+
try std.testing.expectEqual(@as(i64, 1), block_row.?.int(0));
2849
2992
}
2850
2993
2851
2994
test "stores blob metadata in sqlite and bytes in disk blobstore" {