atproto pds in zig
pds.zat.dev
pds
atproto
Decouple sync waiting from store polling
+59
-8
+7
-8
src/atproto/sync.zig
+7
-8
src/atproto/sync.zig
···
2
2
const config = @import("../core/config.zig");
3
3
const log = @import("../core/log.zig");
4
4
const http_api = @import("../http/api.zig");
5
+
const eventlog = @import("../storage/eventlog.zig");
5
6
const store = @import("../storage/store.zig");
6
7
const zat = @import("zat");
7
8
8
9
const http = std.http;
9
10
10
11
const max_subscribe_repos_connections = 32;
11
-
const subscribe_repos_idle_limit = 30;
12
12
var subscribe_repos_connections: usize = 0;
13
13
14
14
pub fn getBlob(request: *http.Server.Request) !void {
···
132
132
else
133
133
0;
134
134
135
-
var idle_seconds: usize = 0;
135
+
var observed = eventlog.snapshot().generation;
136
136
while (true) {
137
137
var sent = false;
138
138
{
···
146
146
}
147
147
}
148
148
if (sent) {
149
-
idle_seconds = 0;
149
+
observed = eventlog.snapshot().generation;
150
150
continue;
151
151
}
152
-
idle_seconds += 1;
153
-
if (idle_seconds >= subscribe_repos_idle_limit) {
154
-
log.info("sync subscribeRepos closing idle cursor={d} idle_seconds={d}\n", .{ cursor, idle_seconds });
155
-
return;
152
+
const next = try eventlog.waitForChange(observed);
153
+
if (next.generation != observed or next.latest_seq > cursor) {
154
+
observed = next.generation;
155
+
continue;
156
156
}
157
-
store.currentIo().sleep(.fromSeconds(1), .awake) catch {};
158
157
}
159
158
}
160
159
+1
src/main.zig
+1
src/main.zig
+1
src/root.zig
+1
src/root.zig
+44
src/storage/eventlog.zig
+44
src/storage/eventlog.zig
···
1
+
const std = @import("std");
2
+
3
+
const Io = std.Io;
4
+
5
+
var store_io: Io = undefined;
6
+
var initialized = false;
7
+
var mutex: Io.Mutex = .init;
8
+
var changed: Io.Condition = .init;
9
+
var generation: usize = 0;
10
+
var latest_seq: u64 = 0;
11
+
12
+
pub fn init(io: Io) void {
13
+
store_io = io;
14
+
initialized = true;
15
+
}
16
+
17
+
pub fn publish(seq: u64) void {
18
+
if (!initialized) return;
19
+
_ = @atomicRmw(usize, &generation, .Add, 1, .release);
20
+
_ = @atomicRmw(u64, &latest_seq, .Max, seq, .release);
21
+
changed.broadcast(store_io);
22
+
}
23
+
24
+
pub fn snapshot() Snapshot {
25
+
return .{
26
+
.generation = @atomicLoad(usize, &generation, .acquire),
27
+
.latest_seq = @atomicLoad(u64, &latest_seq, .acquire),
28
+
};
29
+
}
30
+
31
+
pub fn waitForChange(observed: usize) !Snapshot {
32
+
try mutex.lock(store_io);
33
+
defer mutex.unlock(store_io);
34
+
35
+
while (@atomicLoad(usize, &generation, .acquire) == observed) {
36
+
try changed.wait(store_io, &mutex);
37
+
}
38
+
return snapshot();
39
+
}
40
+
41
+
pub const Snapshot = struct {
42
+
generation: usize,
43
+
latest_seq: u64,
44
+
};
+6
src/storage/store.zig
+6
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 eventlog = @import("eventlog.zig");
5
6
const zat = @import("zat");
6
7
const zqlite = @import("zqlite");
7
8
const Io = std.Io;
···
335
336
const seq = try nextSeqLocked();
336
337
const frame = try accountEventFrame(allocator, seq, did, active);
337
338
try insertSeqEventLocked(seq, did, "", frame);
339
+
eventlog.publish(seq);
338
340
}
339
341
340
342
pub fn sequenceIdentityEvent(allocator: std.mem.Allocator, did: []const u8, handle: []const u8) !void {
···
344
346
const seq = try nextSeqLocked();
345
347
const frame = try identityEventFrame(allocator, seq, did, handle);
346
348
try insertSeqEventLocked(seq, did, "", frame);
349
+
eventlog.publish(seq);
347
350
}
348
351
349
352
pub fn sequenceSyncEvent(allocator: std.mem.Allocator, did: []const u8) !void {
···
354
357
const commit = try latestCommitRawLocked(allocator, did) orelse return Error.RepoNotFound;
355
358
const frame = try syncEventFrame(allocator, seq, did, commit.commit_cid_text, commit.data_cid_raw, commit.rev, commit.commit_data);
356
359
try insertSeqEventLocked(seq, did, commit.commit_cid_text, frame);
360
+
eventlog.publish(seq);
357
361
}
358
362
359
363
pub fn isAccountActive(did: []const u8) bool {
···
655
659
\\ evt = excluded.evt
656
660
, .{ @as(i64, @intCast(seq)), account.did, commit.cid, zqlite.blob(event_frame) });
657
661
try conn.commit();
662
+
eventlog.publish(seq);
658
663
659
664
return .{
660
665
.commit = .{ .cid = commit.cid, .rev = rev },
···
1144
1149
\\ evt = excluded.evt
1145
1150
, .{ @as(i64, @intCast(seq)), account.did, commit_cid, zqlite.blob(event_frame) });
1146
1151
try conn.commit();
1152
+
eventlog.publish(seq);
1147
1153
}
1148
1154
1149
1155
pub fn writeRepoCar(allocator: std.mem.Allocator, did: []const u8) ![]const u8 {