From a595659000f59a162eb64943a1ac7266d7e950dd Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 18 Oct 2026 08:51:51 +0000 Subject: [PATCH 1/3] feat(fetch)!: throw when an event stream ends without a close event `toAsyncIteratorObject` now only finishes cleanly on a `close` event (or throws on an `error` event). A stream that ends without either, including a missing body, rejects with the same `EventStreamDecoderError('Event Stream ended before complete')` as a stream ending with an incomplete message, so a dropped connection is no longer mistaken for a finished stream. Since #125 `toEventStream` always sends a `close` event, so streams from standard-server senders are unaffected. This applies to `@standard-server/fetch`, `@standard-server/node` and `@standard-server/aws-lambda`, which share the fetch implementation. `@standard-server/peer` already rejects when its queue closes without a `close` event. Cancelling with `return()` still resolves as done. ## Breaking change - Reading an event stream that ends without a `close` event now throws instead of resolving `{ done: true, value: undefined }`. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01NYpqRELEp4zdwCYWeqRMAj --- packages/fetch/src/event-stream.test.ts | 64 +++++++++++++++++++++---- packages/fetch/src/event-stream.ts | 18 +++---- packages/node/src/event-stream.test.ts | 1 + 3 files changed, 66 insertions(+), 17 deletions(-) diff --git a/packages/fetch/src/event-stream.test.ts b/packages/fetch/src/event-stream.test.ts index 8602cbbc..e1cc71ab 100644 --- a/packages/fetch/src/event-stream.test.ts +++ b/packages/fetch/src/event-stream.test.ts @@ -1,4 +1,9 @@ -import { ErrorEvent, getEventMeta, withEventMeta } from '@standard-server/core' +import { + ErrorEvent, + EventStreamDecoderError, + getEventMeta, + withEventMeta, +} from '@standard-server/core' import { isAsyncIteratorObject, sleep } from '@standard-server/shared' import { toAsyncIteratorObject, toEventStream } from './event-stream' @@ -82,17 +87,56 @@ describe('toAsyncIteratorObject', () => { return true }) - expect(await generator.next()).toSatisfy(({ done, value }) => { - expect(done).toEqual(true) - expect(value).toEqual(undefined) - expect(getEventMeta(value)).toEqual(undefined) + // ending without a close event is treated like ending with an incomplete message + await expect(generator.next()).rejects.toSatisfy((error: any) => { + expect(error).toBeInstanceOf(EventStreamDecoderError) + expect(error.message).toBe('Event Stream ended before complete') return true }) + expect(await generator.next()).toEqual({ done: true, value: undefined }) await expect(stream.getReader().closed).resolves.toBe(undefined) }) + it.each([ + ['nothing', ''], + ['only comments', ': \n\n: ping\n\n'], + ['blank lines', '\n\n\r\n'], + ])('throws when the stream sends %s and ends without a close event', async (_, body) => { + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(body) + controller.close() + }, + }).pipeThrough(new TextEncoderStream()) + + const generator = toAsyncIteratorObject(stream) + + await expect(generator.next()).rejects.toThrowError( + new EventStreamDecoderError('Event Stream ended before complete'), + ) + expect(await generator.next()).toEqual({ done: true, value: undefined }) + }) + + it('throws when the stream ends with an incomplete message', async () => { + const stream = new ReadableStream({ + start(controller) { + controller.enqueue('event: message\ndata: {"order": 1}\n\n') + controller.enqueue('event: close\ndata: {"order": 2}\n') + controller.close() + }, + }).pipeThrough(new TextEncoderStream()) + + const generator = toAsyncIteratorObject(stream) + + expect(await generator.next()).toEqual({ done: false, value: { order: 1 } }) + await expect(generator.next()).rejects.toThrowError( + new EventStreamDecoderError('Event Stream ended before complete'), + ) + expect(await generator.next()).toEqual({ done: true, value: undefined }) + }) + it('with extra blank lines', async () => { const stream = new ReadableStream({ async pull(controller) { @@ -100,6 +144,7 @@ describe('toAsyncIteratorObject', () => { controller.enqueue('event: message\ndata: {"order": 1}\n\n\n') controller.enqueue('\r\n: ping\r\n\r\n\r\n') controller.enqueue('event: message\ndata: {"order": 2}\n\n\n') + controller.enqueue('\n\nevent: close\n\n\n') controller.close() }, }).pipeThrough(new TextEncoderStream()) @@ -120,6 +165,7 @@ describe('toAsyncIteratorObject', () => { controller.enqueue('event:\ndata: {"order": 3}\nid: id-3\n\n') controller.enqueue('data: 4\n\n') controller.enqueue('data:\n\n') + controller.enqueue('event: close\n\n') controller.close() }, }).pipeThrough(new TextEncoderStream()) @@ -149,11 +195,13 @@ describe('toAsyncIteratorObject', () => { expect(await generator.next()).toEqual({ done: true, value: undefined }) }) - it('with empty stream', async () => { + it('with null stream', async () => { const generator = toAsyncIteratorObject(null) expect(generator).toSatisfy(isAsyncIteratorObject) - expect(await generator.next()).toEqual({ done: true }) - expect(await generator.next()).toEqual({ done: true }) + await expect(generator.next()).rejects.toThrowError( + new EventStreamDecoderError('Event Stream ended before complete'), + ) + expect(await generator.next()).toEqual({ done: true, value: undefined }) }) it.skipIf(process.version.startsWith('v26.'))('with error event', async () => { diff --git a/packages/fetch/src/event-stream.ts b/packages/fetch/src/event-stream.ts index 94462f88..4834147f 100644 --- a/packages/fetch/src/event-stream.ts +++ b/packages/fetch/src/event-stream.ts @@ -1,6 +1,7 @@ import { encodeEventStreamMessage, ErrorEvent, + EventStreamDecoderError, EventStreamDecoderStream, getEventMeta, unwrapEvent, @@ -25,21 +26,20 @@ export function toAsyncIteratorObject( return new AsyncIteratorClass( async () => { while (true) { - if (reader === undefined) { - return { done: true, value: undefined } - } - - const { done, value } = await reader.read() + const result = await reader?.read() /** - * The sender closed the stream without sending a 'close' event. + * A stream only completes with a 'close' or 'error' event, so ending without + * one means it was cut off, like a stream ending with an incomplete message. * A read cancelled by `return()` also ends here, and AsyncIteratorClass - * resolves it as done either way. + * resolves it as done instead. */ - if (done) { - return { done: true, value: undefined } + if (result === undefined || result.done) { + throw new EventStreamDecoderError('Event Stream ended before complete') } + const { value } = result + switch (value.event) { case 'message': { let message = parseEmptyableJSON(value.data) diff --git a/packages/node/src/event-stream.test.ts b/packages/node/src/event-stream.test.ts index ad61a29a..2a576c6a 100644 --- a/packages/node/src/event-stream.test.ts +++ b/packages/node/src/event-stream.test.ts @@ -20,6 +20,7 @@ it('toAsyncIteratorObject', async () => { controller.enqueue('event: message\ndata: 1\n\n') controller.enqueue('event: message\ndata: 2\n\n') controller.enqueue('event: message\ndata: 3\n\n') + controller.enqueue('event: close\n\n') controller.close() }, }).pipeThrough(new TextEncoderStream()) From f068817cb304d779ee9b4f828e9611d4c69c8d6b Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 18 Oct 2026 08:55:05 +0000 Subject: [PATCH 2/3] test(fetch): trim redundant close-event tests Drop the cases that only re-test core decoder behavior, keep a plain empty-stream case, and assert the error the same way everywhere. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01NYpqRELEp4zdwCYWeqRMAj --- packages/fetch/src/event-stream.test.ts | 32 +++---------------------- 1 file changed, 3 insertions(+), 29 deletions(-) diff --git a/packages/fetch/src/event-stream.test.ts b/packages/fetch/src/event-stream.test.ts index e1cc71ab..ca0d7ca9 100644 --- a/packages/fetch/src/event-stream.test.ts +++ b/packages/fetch/src/event-stream.test.ts @@ -88,49 +88,23 @@ describe('toAsyncIteratorObject', () => { }) // ending without a close event is treated like ending with an incomplete message - await expect(generator.next()).rejects.toSatisfy((error: any) => { - expect(error).toBeInstanceOf(EventStreamDecoderError) - expect(error.message).toBe('Event Stream ended before complete') - - return true - }) - expect(await generator.next()).toEqual({ done: true, value: undefined }) - - await expect(stream.getReader().closed).resolves.toBe(undefined) - }) - - it.each([ - ['nothing', ''], - ['only comments', ': \n\n: ping\n\n'], - ['blank lines', '\n\n\r\n'], - ])('throws when the stream sends %s and ends without a close event', async (_, body) => { - const stream = new ReadableStream({ - start(controller) { - controller.enqueue(body) - controller.close() - }, - }).pipeThrough(new TextEncoderStream()) - - const generator = toAsyncIteratorObject(stream) - await expect(generator.next()).rejects.toThrowError( new EventStreamDecoderError('Event Stream ended before complete'), ) expect(await generator.next()).toEqual({ done: true, value: undefined }) + + await expect(stream.getReader().closed).resolves.toBe(undefined) }) - it('throws when the stream ends with an incomplete message', async () => { + it('with empty stream', async () => { const stream = new ReadableStream({ start(controller) { - controller.enqueue('event: message\ndata: {"order": 1}\n\n') - controller.enqueue('event: close\ndata: {"order": 2}\n') controller.close() }, }).pipeThrough(new TextEncoderStream()) const generator = toAsyncIteratorObject(stream) - expect(await generator.next()).toEqual({ done: false, value: { order: 1 } }) await expect(generator.next()).rejects.toThrowError( new EventStreamDecoderError('Event Stream ended before complete'), ) From 07cbb4e8e2871fe1c82d7ad49d37171e759a59c1 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 18 Oct 2026 09:06:18 +0000 Subject: [PATCH 3/3] docs(core): note that an event stream without a close event throws Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01NYpqRELEp4zdwCYWeqRMAj --- packages/core/README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/core/README.md b/packages/core/README.md index 8576312d..81a569d2 100644 --- a/packages/core/README.md +++ b/packages/core/README.md @@ -182,7 +182,7 @@ const response: StandardResponse = { } ``` -Events are interpreted as follows: `yield` emits a `message`, `throw` emits an `error`, and `return` emits a `close` event. Note that `close` does not cause [EventSource](https://developer.mozilla.org/en-US/docs/Web/API/EventSource) to close the connection because it is not part of the SSE specification. However, when using Standard Server for client-side streaming, `close` is treated as the end of the stream, so the connection is closed and no reconnection is attempted. +Events are interpreted as follows: `yield` emits a `message`, `throw` emits an `error`, and `return` emits a `close` event. Note that `close` does not cause [EventSource](https://developer.mozilla.org/en-US/docs/Web/API/EventSource) to close the connection because it is not part of the SSE specification. However, when using Standard Server for client-side streaming, `close` is treated as the end of the stream, so the connection is closed and no reconnection is attempted. A stream that ends without a `close` or `error` event is treated as cut off, so reading it throws an `EventStreamDecoderError`. For explicit SSE encoding, decoding, and metadata handling, see the [Event-Stream Helpers](#event-stream-helpers) below.