atproto pds in zig
pds.zat.dev
pds
atproto
Fix firehose commit since rev
+99
-11
+99
-11
src/storage/store.zig
+99
-11
src/storage/store.zig
···
1909
1909
account.did,
1910
1910
commit.cid,
1911
1911
rev,
1912
-
if (current) |root| root.commit_cid_text else null,
1912
+
if (current) |root| root.rev else null,
1913
1913
if (current) |root| root.data_cid_raw else null,
1914
1914
commit.data,
1915
1915
record_blocks.items,
···
3651
3651
try migrateRecordsTable();
3652
3652
try validateRecordBlocksPresent();
3653
3653
try migrateSeqEventsSync11();
3654
+
try migrateSeqEventsSinceRev();
3654
3655
}
3655
3656
3656
3657
fn migratePermissionedDataTables() !void {
···
3793
3794
errdefer conn.rollback();
3794
3795
try rebuildSeqEventsSync11Locked(allocator);
3795
3796
try conn.exec("INSERT INTO zds_migrations (name) VALUES (?)", .{"sync11-seq-events"});
3797
+
try conn.commit();
3798
+
}
3799
+
3800
+
fn migrateSeqEventsSinceRev() !void {
3801
+
try conn.execNoArgs(
3802
+
\\CREATE TABLE IF NOT EXISTS zds_migrations (
3803
+
\\ name TEXT PRIMARY KEY,
3804
+
\\ applied_at INTEGER NOT NULL DEFAULT (unixepoch())
3805
+
\\)
3806
+
);
3807
+
const row = try conn.row("SELECT 1 FROM zds_migrations WHERE name = 'sync-since-rev-events'", .{});
3808
+
if (row != null) {
3809
+
defer row.?.deinit();
3810
+
return;
3811
+
}
3812
+
3813
+
var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
3814
+
defer arena.deinit();
3815
+
const allocator = arena.allocator();
3816
+
3817
+
db_mutex.lockUncancelable(store_io);
3818
+
defer db_mutex.unlock(store_io);
3819
+
try conn.exclusiveTransaction();
3820
+
errdefer conn.rollback();
3821
+
try rebuildSeqEventsSync11Locked(allocator);
3822
+
try conn.exec("INSERT INTO zds_migrations (name) VALUES (?)", .{"sync-since-rev-events"});
3796
3823
try conn.commit();
3797
3824
}
3798
3825
···
4691
4718
4692
4719
fn backfillSeqEventsLocked(allocator: std.mem.Allocator) !void {
4693
4720
var rows = try conn.rows(
4694
-
\\SELECT c.seq, c.did, c.cid, c.rev, c.prev
4721
+
\\SELECT c.seq, c.did, c.cid, c.rev, c.prev, p.rev
4695
4722
\\FROM commits c
4723
+
\\LEFT JOIN commits p ON p.did = c.did AND p.cid = c.prev
4696
4724
\\LEFT JOIN seq_events e ON e.seq = c.seq
4697
4725
\\WHERE e.seq IS NULL
4698
4726
\\ORDER BY c.seq ASC
···
4705
4733
cid: []const u8,
4706
4734
rev: []const u8,
4707
4735
prev: ?[]const u8,
4736
+
since_rev: ?[]const u8,
4708
4737
}) = .empty;
4709
4738
while (rows.next()) |row| {
4710
4739
try missing.append(allocator, .{
···
4713
4742
.cid = try allocator.dupe(u8, row.text(2)),
4714
4743
.rev = try allocator.dupe(u8, row.text(3)),
4715
4744
.prev = if (row.nullableText(4)) |prev| try allocator.dupe(u8, prev) else null,
4745
+
.since_rev = if (row.nullableText(5)) |rev| try allocator.dupe(u8, rev) else null,
4716
4746
});
4717
4747
}
4718
4748
if (rows.err) |err| return err;
4719
4749
4720
4750
for (missing.items) |commit| {
4721
4751
const repo_car = writeRepoCarFromLocked(allocator, commit.did, commit.cid) catch continue;
4722
-
const frame = commitEventFrameFromCar(allocator, commit.seq, commit.did, commit.cid, commit.rev, commit.prev, null, repo_car, &.{}) catch continue;
4752
+
const frame = commitEventFrameFromCar(allocator, commit.seq, commit.did, commit.cid, commit.rev, commit.since_rev, null, repo_car, &.{}) catch continue;
4723
4753
try conn.exec(
4724
4754
\\INSERT INTO seq_events (seq, did, commit_cid, evt)
4725
4755
\\VALUES (?, ?, ?, ?)
···
4730
4760
4731
4761
fn rebuildSeqEventsSync11Locked(allocator: std.mem.Allocator) !void {
4732
4762
var rows = try conn.rows(
4733
-
\\SELECT seq, did, cid, rev, prev
4734
-
\\FROM commits
4735
-
\\ORDER BY seq ASC
4763
+
\\SELECT c.seq, c.did, c.cid, c.rev, c.prev, p.rev
4764
+
\\FROM commits c
4765
+
\\LEFT JOIN commits p ON p.did = c.did AND p.cid = c.prev
4766
+
\\ORDER BY c.seq ASC
4736
4767
, .{});
4737
4768
defer rows.deinit();
4738
4769
···
4742
4773
cid: []const u8,
4743
4774
rev: []const u8,
4744
4775
prev: ?[]const u8,
4776
+
since_rev: ?[]const u8,
4745
4777
}) = .empty;
4746
4778
while (rows.next()) |row| {
4747
4779
try commits.append(allocator, .{
···
4750
4782
.cid = try allocator.dupe(u8, row.text(2)),
4751
4783
.rev = try allocator.dupe(u8, row.text(3)),
4752
4784
.prev = if (row.nullableText(4)) |prev| try allocator.dupe(u8, prev) else null,
4785
+
.since_rev = if (row.nullableText(5)) |rev| try allocator.dupe(u8, rev) else null,
4753
4786
});
4754
4787
}
4755
4788
if (rows.err) |err| return err;
···
4760
4793
commitDataCidRawLocked(allocator, commit.did, prev_commit) catch null
4761
4794
else
4762
4795
null;
4763
-
const frame = commitEventFrameFromCar(allocator, commit.seq, commit.did, commit.cid, commit.rev, commit.prev, prev_data_raw, repo_car, &.{}) catch continue;
4796
+
const frame = commitEventFrameFromCar(allocator, commit.seq, commit.did, commit.cid, commit.rev, commit.since_rev, prev_data_raw, repo_car, &.{}) catch continue;
4764
4797
try conn.exec(
4765
4798
\\INSERT INTO seq_events (seq, did, commit_cid, evt)
4766
4799
\\VALUES (?, ?, ?, ?)
···
4822
4855
did: []const u8,
4823
4856
commit_cid: []const u8,
4824
4857
rev: []const u8,
4825
-
prev: ?[]const u8,
4858
+
since_rev: ?[]const u8,
4826
4859
prev_data_raw: ?[]const u8,
4827
4860
commit_data: []const u8,
4828
4861
record_blocks: []const ImportedBlock,
···
4863
4896
op_values[idx] = try firehoseOp(allocator, "delete", delete_op.collection, delete_op.rkey, null, prev_record_cids[idx]);
4864
4897
},
4865
4898
};
4866
-
return commitEventFrameFromCar(allocator, seq, did, commit_cid, rev, prev, prev_data_raw, car_bytes, op_values);
4899
+
return commitEventFrameFromCar(allocator, seq, did, commit_cid, rev, since_rev, prev_data_raw, car_bytes, op_values);
4867
4900
}
4868
4901
4869
4902
fn importedCommitEventFrame(
···
4951
4984
did: []const u8,
4952
4985
commit_cid: []const u8,
4953
4986
rev: []const u8,
4954
-
prev: ?[]const u8,
4987
+
since_rev: ?[]const u8,
4955
4988
prev_data_raw: ?[]const u8,
4956
4989
car_bytes: []const u8,
4957
4990
ops: []const zat.cbor.Value,
···
4968
5001
.{ .key = "repo", .value = .{ .text = did } },
4969
5002
.{ .key = "commit", .value = .{ .cid = .{ .raw = commit_raw } } },
4970
5003
.{ .key = "rev", .value = .{ .text = rev } },
4971
-
.{ .key = "since", .value = if (prev) |p| .{ .text = p } else .null },
5004
+
.{ .key = "since", .value = if (since_rev) |since| .{ .text = since } else .null },
4972
5005
.{ .key = "blocks", .value = .{ .bytes = car_bytes } },
4973
5006
.{ .key = "ops", .value = .{ .array = ops } },
4974
5007
.{ .key = "blobs", .value = .{ .array = &.{} } },
···
5651
5684
try std.testing.expect(tree.get("app.bsky.feed.post/3zlazy0012") == null);
5652
5685
const retained = tree.get("app.bsky.feed.post/3zlazy0159") orelse return error.MissingRecord;
5653
5686
try std.testing.expectEqualStrings((get(account.did, "app.bsky.feed.post", "3zlazy0159") orelse return error.MissingRecord).cid, try cidText(allocator, retained.raw));
5687
+
}
5688
+
5689
+
test "commit firehose since uses previous repo rev" {
5690
+
var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
5691
+
defer arena.deinit();
5692
+
const allocator = arena.allocator();
5693
+
5694
+
try init(std.Options.debug_io, ":memory:");
5695
+
defer close();
5696
+
5697
+
const account = try createAccount(
5698
+
allocator,
5699
+
"firehose-since.test",
5700
+
"firehose-since@test.com",
5701
+
"password",
5702
+
"did:plc:firehosesince",
5703
+
true,
5704
+
);
5705
+
5706
+
const first = try std.json.parseFromSlice(std.json.Value, allocator, "{\"$type\":\"app.bsky.feed.post\",\"text\":\"first\"}", .{});
5707
+
defer first.deinit();
5708
+
const first_result = try applyWritesWithOptions(allocator, account, &.{.{ .create = .{
5709
+
.collection = "app.bsky.feed.post",
5710
+
.rkey = "3zsinceone",
5711
+
.value = first.value,
5712
+
} }}, .{});
5713
+
5714
+
const second = try std.json.parseFromSlice(std.json.Value, allocator, "{\"$type\":\"app.bsky.feed.post\",\"text\":\"second\"}", .{});
5715
+
defer second.deinit();
5716
+
const second_result = try applyWritesWithOptions(allocator, account, &.{.{ .create = .{
5717
+
.collection = "app.bsky.feed.post",
5718
+
.rkey = "3zsincetwo",
5719
+
.value = second.value,
5720
+
} }}, .{});
5721
+
5722
+
try conn.exec("DELETE FROM seq_events WHERE did = ?", .{account.did});
5723
+
try rebuildSeqEventsSync11Locked(allocator);
5724
+
5725
+
const events = try listSeqEvents(allocator, 0, 20);
5726
+
var found_second = false;
5727
+
for (events) |event| {
5728
+
const decoded = zat.firehose.decodeFrame(allocator, event.frame) catch continue;
5729
+
switch (decoded) {
5730
+
.commit => |commit| {
5731
+
if (!std.mem.eql(u8, commit.repo, account.did)) continue;
5732
+
if (!std.mem.eql(u8, commit.rev, second_result.commit.rev)) continue;
5733
+
found_second = true;
5734
+
try std.testing.expect(commit.since != null);
5735
+
try std.testing.expectEqualStrings(first_result.commit.rev, commit.since.?);
5736
+
try std.testing.expect(!std.mem.eql(u8, first_result.commit.cid, commit.since.?));
5737
+
},
5738
+
else => {},
5739
+
}
5740
+
}
5741
+
try std.testing.expect(found_second);
5654
5742
}
5655
5743
5656
5744
test "getRepo since filters repo blocks by revision" {