feat(fetch)!: throw when an event stream ends without a close event

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.
 

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,24 +87,38 @@ describe('toAsyncIteratorObject', () => {
       return true
     })
 
-    expect(await generator.next()).toSatisfy(({ done, value }) => {
-      expect(done).toEqual(true)
-      expect(value).toEqual(undefined)
-      expect(getEventMeta(value)).toEqual(undefined)
-
-      return true
-    })
+    // ending without a close event is treated like ending with an incomplete message
+    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('with empty stream', async () => {
+    const stream = new ReadableStream<string>({
+      start(controller) {
+        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('with extra blank lines', async () => {
     const stream = new ReadableStream<string>({
       async pull(controller) {
         controller.enqueue('\n: ping\n\n\n')
         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 +139,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 +169,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 () => {

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)

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())