diff --git a/CHANGELOG.md b/CHANGELOG.md index 9e76539..c0a652d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,8 +1,32 @@ # changelog +## 0.1.6 + +- round-robin host rotation for jetstream and firehose clients +- `Options.host` → `Options.hosts` with sensible defaults (bsky + community relays) +- backoff resets on host switch, jetstream rewinds cursor by 10s +- default jetstream hosts: 4 official bsky, waow.tech, fire.hose.cam, 6 firehose.stream regions +- default firehose hosts: bsky.network + 3 firehose.network regions + +## 0.1.5 + +- align firehose event types with AT Protocol sync spec + +## 0.1.4 + +- firehose support: DAG-CBOR codec, CAR codec, CID creation, firehose client +- encode and decode `com.atproto.sync.subscribeRepos` binary frames + +## 0.1.3 + +- jetstream WebSocket client with typed events, reconnection, and cursor tracking +- `extractAt` ignores unknown JSON fields by default +- HTTP I/O isolated behind `HttpTransport` for 0.16 prep +- websocket dependency pinned to specific commit + ## 0.1.2 -- `extractAt` now logs diagnostic info on parse failures (enable with `.zat` debug scope) +- `extractAt` logs diagnostic info on parse failures (enable with `.zat` debug scope) ## 0.1.1 diff --git a/src/internal/firehose.zig b/src/internal/firehose.zig index 9dc655f..a99138e 100644 --- a/src/internal/firehose.zig +++ b/src/internal/firehose.zig @@ -23,8 +23,15 @@ const log = std.log.scoped(.zat); pub const CommitAction = sync.CommitAction; pub const AccountStatus = sync.AccountStatus; +pub const default_hosts = [_][]const u8{ + "bsky.network", + "northamerica.firehose.network", + "europe.firehose.network", + "asia.firehose.network", +}; + pub const Options = struct { - host: []const u8 = "bsky.network", + hosts: []const []const u8 = &default_hosts, cursor: ?i64 = null, max_message_size: usize = 5 * 1024 * 1024, // 5MB — firehose frames can be large }; @@ -421,24 +428,40 @@ pub const FirehoseClient = struct { /// handler must implement: fn onEvent(*@TypeOf(handler), Event) void /// optional: fn onError(*@TypeOf(handler), anyerror) void /// blocks forever — reconnects with exponential backoff on disconnect. + /// rotates through hosts on each reconnect attempt. pub fn subscribe(self: *FirehoseClient, handler: anytype) void { var backoff: u64 = 1; + var host_index: usize = 0; const max_backoff: u64 = 60; + var prev_host_index: usize = 0; while (true) { - self.connectAndRead(handler) catch |err| { + const host = self.options.hosts[host_index % self.options.hosts.len]; + const effective_index = host_index % self.options.hosts.len; + + // reset backoff on host switch (fresh host deserves a fresh chance) + if (host_index > 0 and effective_index != prev_host_index) { + backoff = 1; + } + + log.info("connecting to host {d}/{d}: {s}", .{ effective_index + 1, self.options.hosts.len, host }); + + self.connectAndRead(host, handler) catch |err| { if (comptime @hasDecl(@TypeOf(handler.*), "onError")) { handler.onError(err); } else { log.err("firehose error: {s}, reconnecting in {d}s...", .{ @errorName(err), backoff }); } }; + + prev_host_index = effective_index; + host_index += 1; posix.nanosleep(backoff, 0); backoff = @min(backoff * 2, max_backoff); } } - fn connectAndRead(self: *FirehoseClient, handler: anytype) !void { + fn connectAndRead(self: *FirehoseClient, host: []const u8, handler: anytype) !void { var path_buf: [256]u8 = undefined; var stream = std.io.fixedBufferStream(&path_buf); const writer = stream.writer(); @@ -449,10 +472,10 @@ pub const FirehoseClient = struct { } const path = stream.getWritten(); - log.info("connecting to wss://{s}{s}", .{ self.options.host, path }); + log.info("connecting to wss://{s}{s}", .{ host, path }); var client = try websocket.Client.init(self.allocator, .{ - .host = self.options.host, + .host = host, .port = 443, .tls = true, .max_size = self.options.max_message_size, @@ -460,11 +483,11 @@ pub const FirehoseClient = struct { defer client.deinit(); var host_header_buf: [256]u8 = undefined; - const host_header = std.fmt.bufPrint(&host_header_buf, "Host: {s}\r\n", .{self.options.host}) catch self.options.host; + const host_header = std.fmt.bufPrint(&host_header_buf, "Host: {s}\r\n", .{host}) catch host; try client.handshake(path, .{ .headers = host_header }); - log.info("firehose connected", .{}); + log.info("firehose connected to {s}", .{host}); var ws_handler = WsHandler(@TypeOf(handler.*)){ .allocator = self.allocator, diff --git a/src/internal/jetstream.zig b/src/internal/jetstream.zig index 20d6bb9..a102c11 100644 --- a/src/internal/jetstream.zig +++ b/src/internal/jetstream.zig @@ -19,8 +19,23 @@ const log = std.log.scoped(.zat); pub const CommitAction = sync.CommitAction; pub const AccountStatus = sync.AccountStatus; +pub const default_hosts = [_][]const u8{ + "jetstream1.us-east.bsky.network", + "jetstream2.us-east.bsky.network", + "jetstream1.us-west.bsky.network", + "jetstream2.us-west.bsky.network", + "jetstream.waow.tech", + "jetstream.fire.hose.cam", + "jet.firehose.stream", + "sfo.firehose.stream", + "nyc.firehose.stream", + "london.firehose.stream", + "frankfurt.firehose.stream", + "chennai.firehose.stream", +}; + pub const Options = struct { - host: []const u8 = "jetstream2.us-east.bsky.network", + hosts: []const []const u8 = &default_hosts, wanted_collections: []const []const u8 = &.{}, wanted_dids: []const []const u8 = &.{}, cursor: ?i64 = null, @@ -133,31 +148,50 @@ pub const JetstreamClient = struct { /// handler must implement: fn onEvent(*@TypeOf(handler), Event) void /// optional: fn onError(*@TypeOf(handler), anyerror) void /// blocks forever — reconnects with exponential backoff on disconnect. + /// rotates through hosts on each reconnect attempt. pub fn subscribe(self: *JetstreamClient, handler: anytype) void { var backoff: u64 = 1; + var host_index: usize = 0; const max_backoff: u64 = 60; + var prev_host_index: usize = 0; while (true) { - self.connectAndRead(handler) catch |err| { + const host = self.options.hosts[host_index % self.options.hosts.len]; + const effective_index = host_index % self.options.hosts.len; + + // rewind cursor by 10s on host switch (different instances may lag) + if (host_index > 0 and effective_index != prev_host_index) { + if (self.last_time_us) |t| { + self.last_time_us = t - 10_000_000; + } + backoff = 1; + } + + log.info("connecting to host {d}/{d}: {s}", .{ effective_index + 1, self.options.hosts.len, host }); + + self.connectAndRead(host, handler) catch |err| { if (comptime @hasDecl(@TypeOf(handler.*), "onError")) { handler.onError(err); } else { log.err("jetstream error: {s}, reconnecting in {d}s...", .{ @errorName(err), backoff }); } }; + + prev_host_index = effective_index; + host_index += 1; posix.nanosleep(backoff, 0); backoff = @min(backoff * 2, max_backoff); } } - fn connectAndRead(self: *JetstreamClient, handler: anytype) !void { + fn connectAndRead(self: *JetstreamClient, host: []const u8, handler: anytype) !void { var path_buf: [2048]u8 = undefined; const path = try self.buildSubscribePath(&path_buf); - log.info("connecting to wss://{s}{s}", .{ self.options.host, path }); + log.info("connecting to wss://{s}{s}", .{ host, path }); var client = try websocket.Client.init(self.allocator, .{ - .host = self.options.host, + .host = host, .port = 443, .tls = true, .max_size = self.options.max_message_size, @@ -165,11 +199,11 @@ pub const JetstreamClient = struct { defer client.deinit(); var host_header_buf: [256]u8 = undefined; - const host_header = std.fmt.bufPrint(&host_header_buf, "Host: {s}\r\n", .{self.options.host}) catch self.options.host; + const host_header = std.fmt.bufPrint(&host_header_buf, "Host: {s}\r\n", .{host}) catch host; try client.handshake(path, .{ .headers = host_header }); - log.info("jetstream connected", .{}); + log.info("jetstream connected to {s}", .{host}); var ws_handler = WsHandler(@TypeOf(handler.*)){ .allocator = self.allocator, @@ -496,3 +530,42 @@ test "parse missing did returns error" { try std.testing.expectError(error.MissingDid, parseEvent(arena.allocator(), payload)); } + +test "default hosts contains known jetstream instances" { + try std.testing.expectEqual(@as(usize, 12), default_hosts.len); + try std.testing.expectEqualStrings("jetstream1.us-east.bsky.network", default_hosts[0]); + try std.testing.expectEqualStrings("jetstream2.us-east.bsky.network", default_hosts[1]); + try std.testing.expectEqualStrings("jetstream1.us-west.bsky.network", default_hosts[2]); + try std.testing.expectEqualStrings("jetstream2.us-west.bsky.network", default_hosts[3]); + try std.testing.expectEqualStrings("jetstream.waow.tech", default_hosts[4]); + try std.testing.expectEqualStrings("jetstream.fire.hose.cam", default_hosts[5]); + try std.testing.expectEqualStrings("jet.firehose.stream", default_hosts[6]); + try std.testing.expectEqualStrings("chennai.firehose.stream", default_hosts[11]); +} + +test "round-robin cycles through hosts" { + const hosts = [_][]const u8{ "host-a", "host-b", "host-c" }; + // simulate the index logic from subscribe() + for (0..9) |i| { + const host = hosts[i % hosts.len]; + const expected: []const u8 = switch (i % 3) { + 0 => "host-a", + 1 => "host-b", + 2 => "host-c", + else => unreachable, + }; + try std.testing.expectEqualStrings(expected, host); + } +} + +test "options default hosts are used" { + const opts = Options{}; + try std.testing.expectEqual(@as(usize, 12), opts.hosts.len); + try std.testing.expectEqualStrings("jetstream1.us-east.bsky.network", opts.hosts[0]); +} + +test "options custom single host" { + const opts = Options{ .hosts = &.{"my-custom-host.example.com"} }; + try std.testing.expectEqual(@as(usize, 1), opts.hosts.len); + try std.testing.expectEqualStrings("my-custom-host.example.com", opts.hosts[0]); +}