zzstoatzz.io
bsky feeds about music
music-atmosphere-feed.plyr.fm/
bsky
feed
zig
fix: add TCP keepalive and clamp future-dated post timestamps
TCP keepalive detects dead peers in ~20s instead of blocking forever
on half-open connections (was causing multi-day stalls). Clamp
indexed_at to current time so posts with future TID timestamps
from clock-skewed clients don't float to the top of the feed.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
+23
-1
+21
src/jetstream.zig
+21
src/jetstream.zig
···
77
77
std.debug.print("failed to set read timeout: {}\n", .{err});
78
78
};
79
79
80
+
// TCP keepalive: detect dead peers in ~20s instead of blocking forever
81
+
// on half-open connections where SO_RCVTIMEO won't fire
82
+
configureKeepalive(&client);
83
+
80
84
std.debug.print("turbostream connected! (read timeout={}ms)\n", .{READ_TIMEOUT_MS});
81
85
stats.get().recordConnected();
82
86
···
210
214
std.debug.print("added: {s}\n", .{at_uri.string});
211
215
}
212
216
};
217
+
218
+
/// enable TCP keepalive so reads don't block forever when a peer
219
+
/// disappears without FIN/RST (network partition, crash, power loss).
220
+
/// detection time: 10s idle + 5s × 2 probes = 20s.
221
+
fn configureKeepalive(client: *websocket.Client) void {
222
+
const fd = client.stream.stream.handle;
223
+
const builtin = @import("builtin");
224
+
posix.setsockopt(fd, posix.SOL.SOCKET, posix.SO.KEEPALIVE, &std.mem.toBytes(@as(i32, 1))) catch return;
225
+
const tcp: i32 = @intCast(posix.IPPROTO.TCP);
226
+
if (builtin.os.tag == .linux) {
227
+
posix.setsockopt(fd, tcp, posix.TCP.KEEPIDLE, &std.mem.toBytes(@as(i32, 10))) catch return;
228
+
} else if (builtin.os.tag == .macos) {
229
+
posix.setsockopt(fd, tcp, posix.TCP.KEEPALIVE, &std.mem.toBytes(@as(i32, 10))) catch return;
230
+
}
231
+
posix.setsockopt(fd, tcp, posix.TCP.KEEPINTVL, &std.mem.toBytes(@as(i32, 5))) catch return;
232
+
posix.setsockopt(fd, tcp, posix.TCP.KEEPCNT, &std.mem.toBytes(@as(i32, 2))) catch return;
233
+
}
+2
-1
src/store/posts.zig
+2
-1
src/store/posts.zig
···
12
12
13
13
const conn = db.getConn() orelse return error.NotInitialized;
14
14
15
-
const ts = timestamp orelse parseTimestampFromUri(uri) orelse std.time.milliTimestamp();
15
+
const now = std.time.milliTimestamp();
16
+
const ts = @min(timestamp orelse parseTimestampFromUri(uri) orelse now, now);
16
17
const author_did = extractDidFromUri(uri);
17
18
18
19
conn.exec(