fix(core): an abandoned stream cancels its reader instead of leaking the socket (#686)

CHANGELOG.md

@@ -571,6 +571,10 @@ npm release are grouped under the in-development version that introduced them.
 
 ### Fixed
 
+- **Breaking out of a `.stream()` loop cancels the response body instead of leaking the socket.**
+  ([#686](https://github.com/rejifald/StitchAPI/issues/686)) The `'bytes'` (default) and `'json'`
+  decoders released the reader lock without cancelling, so an abandoned stream stayed open and the
+  vendor kept writing into it. Both now tear down on every exit path, as `'lines'`/`'ndjson'` did.
 - **`@stitchapi/openapi` refuses to overwrite a file it did not generate.**
   ([#694](https://github.com/rejifald/StitchAPI/issues/694)) A re-run wrote every emitted file
   unconditionally at exit `0`, so `--out <a directory you already own>` silently discarded

packages/core/src/json-stream.ts

@@ -262,6 +262,10 @@ export async function* jsonStream(
             yield slice(valueStart, scan);
         }
     } finally {
+        // Cancel the body on any exit path (early `break` / error / normal end) so an abandoned
+        // consumer proactively closes the connection; then release the lock. `.catch` swallows a
+        // reject from a body already torn down by an abort.
+        await reader.cancel().catch(() => undefined);
         reader.releaseLock();
     }
 }

packages/core/src/stream.ts

@@ -103,6 +103,10 @@ async function* decodeStream(
             yield r.value;
         }
     } finally {
+        // Cancel the body on any exit path (early `break` / error / normal end) so an abandoned
+        // consumer proactively closes the connection; then release the lock. `.catch` swallows a
+        // reject from a body already torn down by an abort.
+        await reader.cancel().catch(() => undefined);
         reader.releaseLock();
     }
 }

packages/core/test/json-stream.spec.ts

@@ -243,3 +243,40 @@ describe('json-stream tokenizer: max-buffer guard + incomplete streams', () => {
         expect(await decode([])).toEqual([]);
     });
 });
+
+describe('json-stream tokenizer: teardown on abandon (issue #686 §2)', () => {
+    // The tokenizer owns its reader, so it owns the teardown: a consumer that `break`s triggers the
+    // generator `.return()`, and the `finally` must cancel() the body before releasing the lock —
+    // otherwise the response stream stays open and the socket leaks. Same guarantee lineReader
+    // gives the `'lines'`/`'ndjson'` decoders; asserted here at the plumbing level.
+
+    test('an early break cancels the underlying stream before releasing the lock', async () => {
+        // An endless body of concatenated top-level objects: only the consumer can end it, so a
+        // missing cancel() is a stream left open. The `cancel` hook records the client-side tear-down.
+        let cancelled = false;
+        const stream = new ReadableStream<Uint8Array>({
+            pull(controller) {
+                controller.enqueue(enc.encode('{"n":1}'));
+            },
+            cancel() {
+                cancelled = true;
+            },
+        });
+        const seen: unknown[] = [];
+        for await (const v of jsonStream(stream)) {
+            seen.push(v);
+            break; // abandon after the first value
+        }
+        expect(seen).toEqual([{ n: 1 }]);
+        expect(cancelled).toBe(true);
+    });
+
+    test('a normal drain still cancels (a no-op on a closed stream) and yields every value', async () => {
+        // Cancelling unconditionally in `finally` is safe: cancel() on an already-closed stream is a
+        // spec no-op, so a fully-consumed stream still emits all of its values.
+        expect(await decodeText('[{"a":1},{"a":2}]')).toEqual([
+            { a: 1 },
+            { a: 2 },
+        ]);
+    });
+});

packages/core/test/stream.spec.ts

@@ -21,6 +21,7 @@ import {
 import { z } from 'zod';
 
 const td = new TextDecoder();
+const enc = new TextEncoder();
 
 describe('stream surface identity (Decisions 5, 11)', () => {
     test('streamSurface has the stable id "stream" and a stream hook', () => {
@@ -123,6 +124,88 @@ describe('stream decoders (Decision 5)', () => {
     });
 });
 
+describe('abandoned-stream teardown (issue #686 §2)', () => {
+    // `break`ing out of a `.stream()` loop `.return()`s the generator chain down to the decoder,
+    // whose `finally` must cancel() the body — releasing the lock alone leaves the response stream
+    // open, so the vendor keeps writing (and billing) into a connection nobody reads. `lines`/
+    // `ndjson` already got this from lineReader; these pin the other two decoders to the same
+    // behaviour. A cancel-recording endless body makes the client-side cancel observable, which a
+    // real HTTP body can't, and it is only ever torn down by the consumer.
+    function endlessBody(chunk: string): {
+        body: ReadableStream<Uint8Array>;
+        cancelled: () => boolean;
+    } {
+        let cancelled = false;
+        const body = new ReadableStream<Uint8Array>({
+            pull(controller) {
+                controller.enqueue(enc.encode(chunk));
+            },
+            cancel() {
+                cancelled = true;
+            },
+        });
+        return { body, cancelled: () => cancelled };
+    }
+
+    test('the default "bytes" decoder cancels the body on an early break', async () => {
+        const { body, cancelled } = endlessBody('chunk');
+        const s = stream({
+            url: 'https://x.test/breakable-bytes',
+            adapter: streamAdapter(body),
+        });
+
+        const deltas: unknown[] = [];
+        for await (const e of s.stream()) {
+            if (e.type === 'delta') {
+                deltas.push(e.chunk);
+                break; // abandon the rest of the stream
+            }
+        }
+        expect(deltas.map((c) => td.decode(c as Uint8Array))).toEqual([
+            'chunk',
+        ]);
+        expect(cancelled()).toBe(true); // cancelled, not merely unlocked
+    });
+
+    test('decode "json" cancels the body on an early break', async () => {
+        // Concatenated top-level objects with no separator → one delta each, so the consumer has a
+        // value to break on while the body is still open.
+        const { body, cancelled } = endlessBody('{"n":1}');
+        const s = stream({
+            url: 'https://x.test/breakable-json',
+            stream: { decode: 'json' },
+            adapter: streamAdapter(body),
+        });
+
+        const deltas: unknown[] = [];
+        for await (const e of s.stream()) {
+            if (e.type === 'delta') {
+                deltas.push(e.chunk);
+                break;
+            }
+        }
+        expect(deltas).toEqual([{ n: 1 }]);
+        expect(cancelled()).toBe(true);
+    });
+
+    test('a normal drain still cancels (a no-op on a closed stream) and loses no chunks', async () => {
+        // Cancelling unconditionally in `finally` is safe: cancel() on an already-closed stream is a
+        // spec no-op, so a fully-consumed stream still delivers everything, on both decoders.
+        const bytes = stream({
+            url: 'https://x.test/drained-bytes',
+            adapter: streamAdapter(streamOf(['ab', 'cd'])),
+        });
+        expect((await bytes()).map((c) => td.decode(c)).join('')).toBe('abcd');
+
+        const json = stream({
+            url: 'https://x.test/drained-json',
+            stream: { decode: 'json' },
+            adapter: streamAdapter(streamOf(['[{"a":1},{"a":2}]'])),
+        });
+        expect(await json()).toEqual([{ a: 1 }, { a: 2 }]);
+    });
+});
+
 describe('stream event spine (Decisions 5, 12)', () => {
     test('emits start → request → delta* → result → done; result is the collected array', async () => {
         const s = stream({