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

# Description

`readFrom` — used by `parseSseStream` for both the JSON-RPC and REST
client
transports — only calls `reader.releaseLock()` in its `finally`:

```ts
async function* readFrom(stream: ReadableStream<string>) {
  const reader = stream.getReader();
  try {
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      yield value;
    }
  } finally {
    reader.releaseLock();   // detaches the reader, but never cancels the stream
  }
}
```

When a consumer stops iterating early — a `break`, a `throw`, or the
REST
transport throwing on an `error` event — the async generator's
`return()`/`throw()` runs that `finally`. `releaseLock()` detaches the
reader
but does **not** cancel the underlying `ReadableStream`, so the fetch
body (and
its socket) is left open. Repeated early terminations leak connections.

## Fix

Cancel the reader on teardown so cancellation propagates to the response
body:

```ts
} finally {
  await reader.cancel().catch(() => {});
  reader.releaseLock();
}
```

Semantics (verified against the WHATWG Streams behaviour):

- **early break / throw** → `cancel()` propagates to the source, closing
the
  connection (the fix).
- **normal completion** → the stream is already closed; `cancel()` is a
no-op.
- **errored source** → `cancel()` rejects with the stored error; it is
ignored
  so the original error still surfaces to the caller.

## Tests

Adds regression tests asserting the underlying stream is canceled when
the
consumer breaks early and when it throws mid-iteration. Both fail
without the
fix (the source `cancel` callback is never invoked) and pass with it.

Full suite green (1370 tests).

- [x] Follows the `CONTRIBUTING` guide
- [x] PR title uses Conventional Commits (`fix:`)
- [x] Tests and linter pass
- [ ] Docs updated (not necessary)

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,33 @@ 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 source = new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.enqueue(new Uint8Array([1]));
+    },
+    cancel() {
+      cancelled = true;
+    },
+  });
+  // Mirror parseSseStream: pipe the response body through a TextDecoderStream.
+  const body = new Response(source).body;
+  if (!body) return false;
+  const reader = body.pipeThrough(new TextDecoderStream()).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 +204,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' };