fix(server): SSE with keepalive crashes server on client disconnect (#464)

Fixes #463

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

- **Bug Fixes**
- Improved event stream handling to ensure that no additional events or
errors are processed after a stream is cancelled, providing more
reliable cancellation behavior.

- **Tests**
- Enhanced test coverage for event stream cancellation, including
scenarios where errors are thrown and ensuring cleanup logic executes
correctly.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: unnoq

packages/standard-server-fetch/src/event-iterator.test.ts

@@ -233,9 +233,10 @@ describe('toEventStream', () => {
 
     async function* gen() {
       try {
+        await new Promise(resolve => setTimeout(resolve, 10))
         yield 1
-        yield undefined
-        return { value: true }
+        await new Promise(resolve => setTimeout(resolve, 10))
+        yield 2
       }
       catch (err) {
         hasError = err
@@ -257,6 +258,32 @@ describe('toEventStream', () => {
     })
   })
 
+  it('when canceled from client without region - throw', async () => {
+    let hasFinally = false
+
+    async function* gen() {
+      try {
+        await new Promise(resolve => setTimeout(resolve, 10))
+        yield 1
+        await new Promise(resolve => setTimeout(resolve, 10))
+        throw new Error('something')
+      }
+      finally {
+        hasFinally = true
+      }
+    }
+
+    const stream = toEventStream(gen(), {})
+
+    const reader = stream.getReader()
+    await reader.read()
+    await reader.cancel()
+
+    await vi.waitFor(() => {
+      expect(hasFinally).toBe(true)
+    })
+  })
+
   it('when canceled from client with region', async () => {
     let hasError: any
     let hasFinally = false

packages/standard-server-fetch/src/event-iterator.ts

@@ -98,6 +98,7 @@ export function toEventStream(
   const keepAliveInterval = options.eventIteratorKeepAliveInterval ?? 5000
   const keepAliveComment = options.eventIteratorKeepAliveComment ?? ''
 
+  let cancelled = false
   let timeout: ReturnType<typeof setInterval> | undefined
 
   const stream = new ReadableStream<string>({
@@ -115,6 +116,10 @@ export function toEventStream(
 
         clearInterval(timeout)
 
+        if (cancelled) {
+          return
+        }
+
         const meta = getEventMeta(value.value)
 
         if (!value.done || value.value !== undefined || meta !== undefined) {
@@ -132,6 +137,10 @@ export function toEventStream(
       catch (err) {
         clearInterval(timeout)
 
+        if (cancelled) {
+          return
+        }
+
         controller.enqueue(encodeEventMessage({
           ...getEventMeta(err),
           event: 'error',
@@ -142,6 +151,9 @@ export function toEventStream(
       }
     },
     async cancel(reason) {
+      cancelled = true
+      clearInterval(timeout)
+
       if (reason) {
         await iterator.throw?.(reason)
       }

packages/standard-server-node/src/event-iterator.test.ts

@@ -233,9 +233,10 @@ describe('toEventStream', () => {
 
     async function* gen() {
       try {
+        await new Promise(resolve => setTimeout(resolve, 100))
         yield 1
-        yield undefined
-        return { value: true }
+        await new Promise(resolve => setTimeout(resolve, 100))
+        yield 2
       }
       catch (err) {
         hasError = err
@@ -249,6 +250,7 @@ describe('toEventStream', () => {
 
     const reader = Readable.toWeb(stream).getReader()
     await reader.read()
+    await new Promise(resolve => setTimeout(resolve, 1))
     await stream.destroy() // use stream.destroy() instead of reader.cancel() to improve node compatibility
 
     await vi.waitFor(() => {
@@ -257,6 +259,33 @@ describe('toEventStream', () => {
     })
   })
 
+  it('when canceled from client without region - throw', async () => {
+    let hasFinally = false
+
+    async function* gen() {
+      try {
+        await new Promise(resolve => setTimeout(resolve, 10))
+        yield 1
+        await new Promise(resolve => setTimeout(resolve, 10))
+        throw new Error('something')
+      }
+      finally {
+        hasFinally = true
+      }
+    }
+
+    const stream = toEventStream(gen(), {})
+
+    const reader = Readable.toWeb(stream).getReader()
+    await reader.read()
+    await new Promise(resolve => setTimeout(resolve, 1))
+    await stream.destroy() // use stream.destroy() instead of reader.cancel() to improve node compatibility
+
+    await vi.waitFor(() => {
+      expect(hasFinally).toBe(true)
+    })
+  })
+
   it('when canceled from client with region', async () => {
     let hasError: any
     let hasFinally = false

packages/standard-server-node/src/event-iterator.ts

@@ -99,6 +99,7 @@ export function toEventStream(
   const keepAliveInterval = options.eventIteratorKeepAliveInterval ?? 5000
   const keepAliveComment = options.eventIteratorKeepAliveComment ?? ''
 
+  let cancelled = false
   let timeout: ReturnType<typeof setInterval> | undefined
 
   const stream = new ReadableStream<string>({
@@ -116,6 +117,10 @@ export function toEventStream(
 
         clearInterval(timeout)
 
+        if (cancelled) {
+          return
+        }
+
         const meta = getEventMeta(value.value)
 
         if (!value.done || value.value !== undefined || meta !== undefined) {
@@ -133,6 +138,10 @@ export function toEventStream(
       catch (err) {
         clearInterval(timeout)
 
+        if (cancelled) {
+          return
+        }
+
         controller.enqueue(encodeEventMessage({
           ...getEventMeta(err),
           event: 'error',
@@ -143,6 +152,9 @@ export function toEventStream(
       }
     },
     async cancel(reason) {
+      cancelled = true
+      clearInterval(timeout)
+
       if (reason) {
         await iterator.throw?.(reason)
       }