atproto pds in zig
pds.zat.dev
pds
atproto
Bound subscribeRepos connections
+22
-2
+22
-2
src/atproto/sync.zig
+22
-2
src/atproto/sync.zig
···
7
7
8
8
const http = std.http;
9
9
10
+
const max_subscribe_repos_connections = 4;
11
+
const subscribe_repos_idle_limit = 30;
12
+
var subscribe_repos_connections: usize = 0;
13
+
10
14
pub fn getBlob(request: *http.Server.Request) !void {
11
15
var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
12
16
defer arena.deinit();
···
111
115
else => return http_api.xrpcError(request, .upgrade_required, "InvalidRequest", "Expected WebSocket upgrade"),
112
116
};
113
117
118
+
const active = @atomicRmw(usize, &subscribe_repos_connections, .Add, 1, .monotonic);
119
+
if (active >= max_subscribe_repos_connections) {
120
+
_ = @atomicRmw(usize, &subscribe_repos_connections, .Sub, 1, .monotonic);
121
+
log.err("sync subscribeRepos rejected too_many_connections active={d} max={d}\n", .{ active + 1, max_subscribe_repos_connections });
122
+
return http_api.xrpcError(request, .too_many_requests, "RateLimitExceeded", "too many subscribeRepos connections");
123
+
}
124
+
defer _ = @atomicRmw(usize, &subscribe_repos_connections, .Sub, 1, .monotonic);
125
+
114
126
var ws = try request.respondWebSocket(.{ .key = key });
115
127
try ws.flush();
116
128
···
120
132
else
121
133
0;
122
134
135
+
var idle_seconds: usize = 0;
123
136
while (true) {
124
137
var sent = false;
125
138
{
···
132
145
sent = true;
133
146
}
134
147
}
135
-
if (!sent) {
136
-
store.currentIo().sleep(.fromSeconds(1), .awake) catch {};
148
+
if (sent) {
149
+
idle_seconds = 0;
150
+
continue;
137
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;
156
+
}
157
+
store.currentIo().sleep(.fromSeconds(1), .awake) catch {};
138
158
}
139
159
}
140
160