From a7761ae3dc0ceb47fabccd7ad5ef44ba4cb736a7 Mon Sep 17 00:00:00 2001
From: zzstoatzz <thrast36@gmail.com>
Date: Sat, 3 Oct 2026 16:25:28 -0700
Subject: [PATCH] jetstream: return error.Canceled from subscribe instead of
 reconnecting

The reconnect loop treated every error out of connectAndRead as a dropped
connection: it reported it to onError (or logged it), slept the backoff and
connected again. That included error.Canceled from the connect, handshake,
read loop and idle watchdog. std.Io delivers a cancel request to only the
next cancellation point, so once the loop consumed it the backoff sleep ran
normally and the client kept reconnecting, leaving the canceller's blocking
cancel() waiting forever.

subscribe already returns Io.Cancelable!void, so the error set is unchanged.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 src/internal/streaming/jetstream.zig | 56 ++++++++++++++++++++++++++++
 1 file changed, 56 insertions(+)

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);
+}
-- 
2.54.0