diff --git a/CHANGELOG.md b/CHANGELOG.md index 3d8edf93..97245137 100644 --- a/CHANGELOG.md +++ b/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 ` silently discarded diff --git a/packages/core/src/json-stream.ts b/packages/core/src/json-stream.ts index e39fc4aa..cc7ee17a 100644 --- a/packages/core/src/json-stream.ts +++ b/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(); } } diff --git a/packages/core/src/stream.ts b/packages/core/src/stream.ts index 7babc720..f1b7754a 100644 --- a/packages/core/src/stream.ts +++ b/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(); } } diff --git a/packages/core/test/json-stream.spec.ts b/packages/core/test/json-stream.spec.ts index 2e6d2808..6bd64469 100644 --- a/packages/core/test/json-stream.spec.ts +++ b/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({ + 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 }, + ]); + }); +}); diff --git a/packages/core/test/stream.spec.ts b/packages/core/test/stream.spec.ts index e6d3fc87..d8c2a21a 100644 --- a/packages/core/test/stream.spec.ts +++ b/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; + cancelled: () => boolean; + } { + let cancelled = false; + const body = new ReadableStream({ + 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({