diff --git a/src/internal/streaming/jetstream.zig b/src/internal/streaming/jetstream.zig index e2496ff..d56654a 100644 --- a/src/internal/streaming/jetstream.zig +++ b/src/internal/streaming/jetstream.zig @@ -194,6 +194,7 @@ pub const JetstreamClient = struct { self.connectAndRead(host, handler) catch |err| { if (err == error.EndCursorReached or err == error.ReplayIdle) return; + if (err == error.Canceled) return error.Canceled; if (comptime @hasDecl(@TypeOf(handler.*), "onError")) { handler.onError(err); } else { @@ -738,3 +739,58 @@ test "options custom single host" { try std.testing.expectEqual(@as(usize, 1), opts.hosts.len); try std.testing.expectEqualStrings("my-custom-host.example.com", opts.hosts[0]); } + +test "subscribe returns error.Canceled instead of reconnecting" { + const Net = struct { + var real: Io = undefined; + var parked: Io.Event = .unset; + var lookups: usize = 0; + + fn lookup( + _: ?*anyopaque, + _: Io.net.HostName, + resolved: *Io.Queue(Io.net.HostName.LookupResult), + _: Io.net.HostName.LookupOptions, + ) Io.net.HostName.LookupError!void { + defer resolved.close(real); + lookups += 1; + parked.set(real); + try real.sleep(.fromSeconds(60), .awake); + return error.UnknownHostName; + } + + // only the reconnect backoff sleeps through this vtable: ending it + // makes a swallowed cancel fail the test instead of hanging it. + fn sleep(_: ?*anyopaque, _: Io.Timeout) Io.Cancelable!void { + return error.Canceled; + } + }; + const Handler = struct { + errors: usize = 0, + + pub fn onEvent(_: *@This(), _: Event) void {} + + pub fn onError(self: *@This(), _: anyerror) void { + self.errors += 1; + } + + fn run(client: *JetstreamClient, self: *@This()) Io.Cancelable!void { + return client.subscribe(self); + } + }; + + Net.real = std.testing.io; + var vtable = Net.real.vtable.*; + vtable.netLookup = Net.lookup; + vtable.sleep = Net.sleep; + const io: Io = .{ .userdata = Net.real.userdata, .vtable = &vtable }; + + var client = JetstreamClient.init(io, std.testing.allocator, .{ .hosts = &.{"jetstream.invalid"} }); + var handler = Handler{}; + var future = try Net.real.concurrent(Handler.run, .{ &client, &handler }); + Net.parked.waitUncancelable(Net.real); + + try std.testing.expectError(error.Canceled, future.cancel(Net.real)); + try std.testing.expectEqual(@as(usize, 0), handler.errors); + try std.testing.expectEqual(@as(usize, 1), Net.lookups); +}