diff --git a/open-sse/utils/streamHandler.ts b/open-sse/utils/streamHandler.ts index 0de178772a..1e813a978f 100644 --- a/open-sse/utils/streamHandler.ts +++ b/open-sse/utils/streamHandler.ts @@ -345,6 +345,22 @@ function buildStreamErrorChunks( return encodeSseEvent(errorEvent, { includeDone: true }); } +/** + * Minimal `writable` half used by `pipeWithDisconnect`. The real writable is + * driven entirely by the upstream-piped readable, so the writer only needs an + * `abort()` hook for `createDisconnectAwareStream`'s `cancel()` path. + * + * `abort()` returns `Promise` to match the native + * `WritableStreamDefaultWriter.abort()` contract — `cancel()` (and any caller + * that awaits the writer) gets a real thenable instead of `undefined`, which + * keeps abort/error handling clean. Ported from decolua/9router@6b624af4. + */ +export function createNoopAbortWritable(): { + getWriter: () => { abort: () => Promise }; +} { + return { getWriter: () => ({ abort: () => Promise.resolve() }) }; +} + /** * Create transform stream with disconnect detection * Wraps existing transform stream and adds abort capability @@ -432,7 +448,7 @@ export function pipeWithDisconnect( if (!stallTimeoutMs || stallTimeoutMs <= 0) { const transformedBody = providerResponse.body.pipeThrough(transformStream); return createDisconnectAwareStream( - { readable: transformedBody, writable: { getWriter: () => ({ abort: () => {} }) } }, + { readable: transformedBody, writable: createNoopAbortWritable() }, streamController ); } @@ -526,7 +542,7 @@ export function pipeWithDisconnect( const transformedBody = providerResponse.body.pipeThrough(upstreamTap).pipeThrough(transformStream); return createDisconnectAwareStream( - { readable: transformedBody, writable: { getWriter: () => ({ abort: () => {} }) } }, + { readable: transformedBody, writable: createNoopAbortWritable() }, wrappedController ); } diff --git a/tests/unit/stream-handler.test.ts b/tests/unit/stream-handler.test.ts index a0f0c223a2..789d685d55 100644 --- a/tests/unit/stream-handler.test.ts +++ b/tests/unit/stream-handler.test.ts @@ -3,6 +3,7 @@ import assert from "node:assert/strict"; import { createDisconnectAwareStream, + createNoopAbortWritable, createStreamController, pipeWithDisconnect, } from "../../open-sse/utils/streamHandler.ts"; @@ -283,6 +284,40 @@ test("createDisconnectAwareStream cancel propagates disconnect reason and aborts assert.ok(disconnectEvent.duration >= 0); }); +test("createNoopAbortWritable: getWriter().abort() returns a resolved Promise (matches WritableStreamDefaultWriter contract)", async () => { + // The mock writable that pipeWithDisconnect hands to createDisconnectAwareStream + // is consumed only via its writer's abort() hook (in the cancel() path). The + // native WritableStreamDefaultWriter.abort() returns Promise; the mock + // must match that contract so cancel/error handling can await it instead of + // receiving `undefined`. Ported from decolua/9router@6b624af4. + const writable = createNoopAbortWritable(); + const writer = writable.getWriter(); + + const aborted = writer.abort(); + + assert.ok(aborted instanceof Promise, "abort() must return a Promise, not undefined"); + // Awaiting must resolve cleanly to undefined (Promise), never reject. + assert.equal(await aborted, undefined); +}); + +test("createNoopAbortWritable: cancelling a stream wired through it awaits the abort promise without throwing", async () => { + // End-to-end seam: the noop writable is what pipeWithDisconnect injects. Wire + // it into createDisconnectAwareStream exactly as production does and drive the + // cancel() path. With abort() returning undefined (the pre-fix shape) this + // still completes, but a thenable abort keeps the cancel/error path clean. + const transformStream = { + readable: new ReadableStream({ + pull() {}, + cancel() {}, + }), + writable: createNoopAbortWritable(), + }; + + const stream = createDisconnectAwareStream(transformStream, createStreamController()); + + await assert.doesNotReject(stream.cancel("client-gone")); +}); + test("createDisconnectAwareStream uses the default cancel reason when none is provided", async () => { let disconnectEvent = null;