fix(client): cancel the SSE stream on teardown to avoid leaking connections

`readFrom` (used by `parseSseStream` for both the JSON-RPC and REST client
transports) only called `reader.releaseLock()` in its `finally`. When a
consumer stops iterating early — a `break`, a `throw`, or a REST transport
that throws on an `error` event — the async generator's `return()`/`throw()`
runs that `finally`, which detaches the reader but never cancels the
underlying `ReadableStream`. The fetch body (and its socket) is left open,
so repeated early terminations leak connections.

Cancel the reader on teardown so cancellation propagates to the response
body. On normal completion the stream is already closed and `cancel()` is a
no-op; on an errored stream `cancel()` rejects with the same error, which is
ignored so the original error still surfaces.

Adds regression tests asserting the underlying stream is canceled when the
consumer breaks early and when it throws mid-iteration.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: RenKoya1

src/sse_utils.ts

@@ -144,6 +144,10 @@ async function* readFrom(stream: ReadableStream<string>): AsyncGenerator<string,
       yield value;
     }
   } finally {
+    // `releaseLock()` alone leaves the body un-cancelled, leaking the
+    // connection when a consumer breaks/throws early. `.catch()` swallows the
+    // rejection cancel() produces on an already-errored stream.
+    await reader.cancel().catch(() => {});
     reader.releaseLock();
   }
 }

test/sse_utils.spec.ts

@@ -51,6 +51,30 @@ function createMockResponseWithoutAsyncIterator(sseData: string): Response {
   });
 }
 
+// The teardown tests observe the leak fix by watching the source stream's
+// `cancel` callback fire through `parseSseStream`'s internal
+// `pipeThrough(TextDecoderStream)`. Some runtimes (notably the Cloudflare
+// Workers test pool / workerd) do not propagate a TextDecoderStream cancel to
+// the upstream source, so that signal is unobservable there even though the
+// fix runs. Probe the capability once and gate the assertions on it — the
+// leak matters most under Node/undici, where the probe passes.
+async function cancelPropagatesThroughTextDecoder(): Promise<boolean> {
+  let cancelled = false;
+  const decoded = new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.enqueue(new Uint8Array([1]));
+    },
+    cancel() {
+      cancelled = true;
+    },
+  }).pipeThrough(new TextDecoderStream());
+  const reader = decoded.getReader();
+  await reader.read();
+  await reader.cancel().catch(() => {});
+  return cancelled;
+}
+const CANCEL_PROPAGATES = await cancelPropagatesThroughTextDecoder();
+
 describe('SSE Utils', () => {
   describe('formatSSEEvent', () => {
     it('should format a data event', () => {
@@ -177,6 +201,66 @@ describe('SSE Utils', () => {
     });
   });
 
+  describe('parseSseStream teardown', () => {
+    it.runIf(CANCEL_PROPAGATES)(
+      'cancels the underlying stream when the consumer stops early',
+      async () => {
+        // An early break must cancel the response body, not just release the lock.
+        let sourceCancelled = false;
+        const stream = new ReadableStream<Uint8Array>({
+          start(controller) {
+            // One event, then stay open — a long-lived SSE connection.
+            controller.enqueue(new TextEncoder().encode('data: {"id":1}\n\n'));
+          },
+          cancel() {
+            sourceCancelled = true;
+          },
+        });
+        const response = new Response(stream, {
+          headers: { 'Content-Type': 'text/event-stream' },
+        });
+
+        const seen: SseEvent[] = [];
+        for await (const event of parseSseStream(response)) {
+          seen.push(event);
+          break;
+        }
+
+        expect(seen).toHaveLength(1);
+        expect(sourceCancelled).toBe(true);
+      }
+    );
+
+    it.runIf(CANCEL_PROPAGATES)(
+      'cancels the underlying stream when the consumer throws',
+      async () => {
+        let sourceCancelled = false;
+        const stream = new ReadableStream<Uint8Array>({
+          start(controller) {
+            controller.enqueue(new TextEncoder().encode('data: {"id":1}\n\n'));
+          },
+          cancel() {
+            sourceCancelled = true;
+          },
+        });
+        const response = new Response(stream, {
+          headers: { 'Content-Type': 'text/event-stream' },
+        });
+
+        await expect(
+          (async () => {
+            for await (const event of parseSseStream(response)) {
+              expect(event.data).toBe('{"id":1}');
+              throw new Error('consumer boom');
+            }
+          })()
+        ).rejects.toThrow('consumer boom');
+
+        expect(sourceCancelled).toBe(true);
+      }
+    );
+  });
+
   describe('Symmetry: parser understands formatter output', () => {
     it('should parse what formatSSEEvent produces', async () => {
       const originalData = { kind: 'task', id: '123', status: 'completed' };