From 4bec02902247f0f48df00ec5523ab3459a72bc6b Mon Sep 17 00:00:00 2001 From: rejifald Date: Fri, 7 Aug 2026 18:49:53 +0300 Subject: [PATCH] fix(core): an abandoned stream cancels its reader instead of leaking the socket (#686) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A consumer that `break`s out of a `.stream()` loop leaked the connection on the DEFAULT decoder. `break` calls the generator's `.return()`, which runs the `finally` — and the `finally` only did `reader.releaseLock()`. Releasing a lock does not cancel the body, so the underlying response stream was never torn down: the vendor kept writing (and billing) into a connection nobody was reading. Two decoders were affected — `'bytes'` (stream.ts, the default, so this is reachable through the most natural consumer idiom without opting into anything) and `'json'` (json-stream.ts). `'lines'`/`'ndjson'` were already correct: they share `lineReader`, whose `finally` has done `await reader.cancel().catch(…)` on every exit path since the sse teardown work. So this is not a new pattern, it is bringing the other two decoders into line with the third. The three `finally` bodies are now identical, comment included, deliberately: three decoders doing the same job should not each carry their own idea of how a reader is let go. Cancelling unconditionally is safe — `cancel()` on an already-closed stream is a spec no-op — and the `.catch` swallows a reject from a body an abort already tore down. Tested with a cancel-recording endless body, which is what makes the leak observable at all: a real HTTP body cannot report its own cancellation back to a test, and an endless one is only ever ended by the consumer. An early `break` now cancels on the `bytes` default and on `json`, asserted at the surface level in stream.spec.ts and directly against the tokenizer in json-stream.spec.ts; all three of those fail on the pre-fix tree. A normal drain still delivers every chunk on both decoders, pinning that the unconditional cancel costs nothing. Scope is §2 of the issue only. §1 (nothing bounds a live stream) and §3 (no `done` event on abandon) are untouched and stay open, which is why this is `Refs` and not `Fixes`. Refs #686 Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 5 ++ packages/core/src/json-stream.ts | 4 ++ packages/core/src/stream.ts | 4 ++ packages/core/test/json-stream.spec.ts | 37 ++++++++++++ packages/core/test/stream.spec.ts | 83 ++++++++++++++++++++++++++ 5 files changed, 133 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index cbc0b623..bfb4804d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -566,6 +566,11 @@ 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. + - **Adding `cache: { ttl }` no longer turns a handled vendor failure into a process exit.** ([#670](https://github.com/rejifald/StitchAPI/issues/670)) A cached stitch whose vendor returned `503` emitted an **unhandled promise rejection**, which under Node's default 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({