atproto utils for zig
zat.dev
atproto
sdk
zig
release: v0.2.11
fix: enable TCP keepalive on websocket connections — detect dead peers
in ~20s instead of blocking forever when remote disappears without
FIN/RST.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
+41
-1
+4
CHANGELOG.md
+4
CHANGELOG.md
···
1
1
# changelog
2
2
3
+
## 0.2.11
4
+
5
+
- **fix**: enable TCP keepalive on websocket connections — detect dead peers in ~20s instead of blocking forever
6
+
3
7
## 0.2.10
4
8
5
9
- **deps**: bump websocket.zig to fork commit `9e6d732` — TCP split guard for HTTP body reads behind reverse proxies
+1
-1
build.zig.zon
+1
-1
build.zig.zon
+18
src/internal/streaming/firehose.zig
+18
src/internal/streaming/firehose.zig
···
485
485
const host_header = std.fmt.bufPrint(&host_header_buf, "Host: {s}\r\n", .{host}) catch host;
486
486
487
487
try client.handshake(path, .{ .headers = host_header });
488
+
configureKeepalive(&client);
488
489
489
490
log.info("firehose connected to {s}", .{host});
490
491
···
525
526
log.info("firehose connection closed", .{});
526
527
}
527
528
};
529
+
}
530
+
531
+
/// enable TCP keepalive so reads don't block forever when a peer
532
+
/// disappears without FIN/RST (network partition, crash, power loss).
533
+
/// detection time: 10s idle + 5s × 2 probes = 20s.
534
+
fn configureKeepalive(client: *websocket.Client) void {
535
+
const fd = client.stream.stream.handle;
536
+
const builtin = @import("builtin");
537
+
posix.setsockopt(fd, posix.SOL.SOCKET, posix.SO.KEEPALIVE, &std.mem.toBytes(@as(i32, 1))) catch return;
538
+
const tcp: i32 = @intCast(posix.IPPROTO.TCP);
539
+
if (builtin.os.tag == .linux) {
540
+
posix.setsockopt(fd, tcp, posix.TCP.KEEPIDLE, &std.mem.toBytes(@as(i32, 10))) catch return;
541
+
} else if (builtin.os.tag == .macos) {
542
+
posix.setsockopt(fd, tcp, posix.TCP.KEEPALIVE, &std.mem.toBytes(@as(i32, 10))) catch return;
543
+
}
544
+
posix.setsockopt(fd, tcp, posix.TCP.KEEPINTVL, &std.mem.toBytes(@as(i32, 5))) catch return;
545
+
posix.setsockopt(fd, tcp, posix.TCP.KEEPCNT, &std.mem.toBytes(@as(i32, 2))) catch return;
528
546
}
529
547
530
548
// === tests ===
+18
src/internal/streaming/jetstream.zig
+18
src/internal/streaming/jetstream.zig
···
202
202
const host_header = std.fmt.bufPrint(&host_header_buf, "Host: {s}\r\n", .{host}) catch host;
203
203
204
204
try client.handshake(path, .{ .headers = host_header });
205
+
configureKeepalive(&client);
205
206
206
207
log.info("jetstream connected to {s}", .{host});
207
208
···
268
269
log.info("jetstream connection closed", .{});
269
270
}
270
271
};
272
+
}
273
+
274
+
/// enable TCP keepalive so reads don't block forever when a peer
275
+
/// disappears without FIN/RST (network partition, crash, power loss).
276
+
/// detection time: 10s idle + 5s × 2 probes = 20s.
277
+
fn configureKeepalive(client: *websocket.Client) void {
278
+
const fd = client.stream.stream.handle;
279
+
const builtin = @import("builtin");
280
+
posix.setsockopt(fd, posix.SOL.SOCKET, posix.SO.KEEPALIVE, &std.mem.toBytes(@as(i32, 1))) catch return;
281
+
const tcp: i32 = @intCast(posix.IPPROTO.TCP);
282
+
if (builtin.os.tag == .linux) {
283
+
posix.setsockopt(fd, tcp, posix.TCP.KEEPIDLE, &std.mem.toBytes(@as(i32, 10))) catch return;
284
+
} else if (builtin.os.tag == .macos) {
285
+
posix.setsockopt(fd, tcp, posix.TCP.KEEPALIVE, &std.mem.toBytes(@as(i32, 10))) catch return;
286
+
}
287
+
posix.setsockopt(fd, tcp, posix.TCP.KEEPINTVL, &std.mem.toBytes(@as(i32, 5))) catch return;
288
+
posix.setsockopt(fd, tcp, posix.TCP.KEEPCNT, &std.mem.toBytes(@as(i32, 2))) catch return;
271
289
}
272
290
273
291
// === tests ===