fix(streaming): expose Last-Event-ID and make write-after-abort a no-op

runtime-tests/node/index.test.ts

@@ -206,6 +206,103 @@ describe('streamSSE', () => {
   })
 })
 
+describe('streamSSE lifecycle (Last-Event-ID + write-after-abort)', () => {
+  const events = ['alpha', 'beta', 'gamma', 'delta']
+  const app = new Hono()
+
+  let handlerDone = false
+  let writesAfterAbort = 0
+
+  // Replays only the events after the Last-Event-ID cursor, as an EventSource
+  // reconnect would expect.
+  app.get('/feed', (c) =>
+    streamSSE(c, async (stream) => {
+      const cursor = Number(stream.lastEventId ?? 0)
+      for (let i = cursor; i < events.length; i++) {
+        await stream.writeSSE({ data: events[i], id: String(i + 1) })
+      }
+    })
+  )
+
+  app.get('/abrupt', (c) =>
+    streamSSE(c, async (stream) => {
+      await stream.writeSSE({ data: 'one', id: '1' })
+      // The client disconnects during this window; keep producing.
+      await stream.sleep(20)
+      for (let i = 2; i <= 4; i++) {
+        await stream.writeSSE({ data: `dropped-${i}`, id: String(i) })
+        writesAfterAbort++
+      }
+      handlerDone = true
+    })
+  )
+
+  const agent = createAgent(app)
+
+  beforeEach(() => {
+    handlerDone = false
+    writesAfterAbort = 0
+  })
+
+  const readEvents = async (res: Response, count: number): Promise<string[]> => {
+    const reader = res.body!.getReader()
+    const decoder = new TextDecoder()
+    const out: string[] = []
+    let buffer = ''
+    while (out.length < count) {
+      const { value, done } = await reader.read()
+      if (done) {
+        break
+      }
+      buffer += decoder.decode(value, { stream: true })
+      const frames = buffer.split('\n\n')
+      buffer = frames.pop() ?? ''
+      for (const frame of frames) {
+        const data = /^data: (.*)$/m.exec(frame)?.[1]
+        if (data !== undefined) {
+          out.push(data)
+        }
+      }
+    }
+    reader.releaseLock()
+    return out
+  }
+
+  it('Should resume from the Last-Event-ID header after a reconnect', async () => {
+    // First connection: the client reads two events, then drops.
+    const first = await agent.get('/feed')
+    expect(await readEvents(first, 2)).toEqual(['alpha', 'beta'])
+    await first.body!.cancel()
+
+    // Reconnect with the id of the last received event: only later events replay.
+    const second = await agent.get('/feed', { headers: { 'Last-Event-ID': '2' } })
+    expect(await readEvents(second, 2)).toEqual(['gamma', 'delta'])
+    await second.body!.cancel()
+  })
+
+  it('Should resume from the beginning without a Last-Event-ID header', async () => {
+    const res = await agent.get('/feed')
+    expect(await readEvents(res, 4)).toEqual(['alpha', 'beta', 'gamma', 'delta'])
+    await res.body!.cancel()
+  })
+
+  it('Should let the handler finish cleanly when the client disconnects mid-stream', async () => {
+    const controller = new AbortController()
+    const res = await agent.get('/abrupt', { signal: controller.signal })
+    expect(await readEvents(res, 1)).toEqual(['one'])
+    controller.abort()
+    await res.body!.cancel().catch(() => {})
+
+    // Writes after the disconnect are no-ops; the handler must still complete.
+    const deadline = Date.now() + 2000
+    while (!handlerDone && Date.now() < deadline) {
+      await new Promise((resolve) => setTimeout(resolve, 10))
+    }
+    expect(handlerDone).toBe(true)
+    expect(writesAfterAbort).toBe(3)
+  })
+})
+
 describe('compress', async () => {
   const cssContent = Array.from({ length: 60 }, () => 'body { color: red; }').join('\n')
   const [externalServer, serverInfo] = await new Promise<[Server, AddressInfo]>((resolve) => {

src/helper/streaming/sse.test.tsx

@@ -428,4 +428,47 @@ describe('SSE Streaming helper', () => {
     // Two \r should produce an empty line in between
     expect(decodedValue).toBe('event: test-double-cr\ndata: Left\ndata: \ndata: Right\n\n')
   })
+
+  it('Exposes the Last-Event-ID request header as stream.lastEventId', async () => {
+    const req = new Request('http://localhost/', { headers: { 'Last-Event-ID': '41' } })
+    const c = new Context(req)
+    const res = streamSSE(c, async (stream) => {
+      await stream.writeSSE({ data: `resumed from ${stream.lastEventId}` })
+    })
+    expect(await res.text()).toBe('data: resumed from 41\n\n')
+  })
+
+  it('stream.lastEventId is undefined when the header is not sent', async () => {
+    const req = new Request('http://localhost/')
+    const c = new Context(req)
+    const res = streamSSE(c, async (stream) => {
+      await stream.writeSSE({ data: `lastEventId=${stream.lastEventId}` })
+    })
+    expect(await res.text()).toBe('data: lastEventId=undefined\n\n')
+  })
+
+  it('writeSSE() after the client disconnects resolves as a no-op', async () => {
+    let handlerDone = false
+    const wroteWhileAborted: boolean[] = []
+    const res = streamSSE(c, async (stream) => {
+      await stream.writeSSE({ data: 'one', id: '1' })
+      await stream.sleep(10) // the test cancels the reader during this window
+      for (let i = 2; i <= 4; i++) {
+        // Must neither throw nor hang even though the client is gone.
+        await stream.writeSSE({ data: `after-abort-${i}`, id: String(i) })
+        wroteWhileAborted.push(stream.aborted)
+      }
+      handlerDone = true
+    })
+    if (!res.body) {
+      throw new Error('Body is null')
+    }
+    const reader = res.body.getReader()
+    const { value } = await reader.read()
+    expect(new TextDecoder().decode(value)).toBe('data: one\nid: 1\n\n')
+    await reader.cancel()
+    await new Promise((resolve) => setTimeout(resolve, 50))
+    expect(handlerDone).toBe(true)
+    expect(wroteWhileAborted).toEqual([true, true, true])
+  })
 })

src/helper/streaming/sse.ts

@@ -11,8 +11,15 @@ export interface SSEMessage {
 }
 
 export class SSEStreamingApi extends StreamingApi {
-  constructor(writable: WritableStream, readable: ReadableStream) {
+  /**
+   * The `Last-Event-ID` request header sent by an `EventSource` on reconnect
+   * (the id of the last event it received), or `undefined` when absent.
+   */
+  lastEventId?: string
+
+  constructor(writable: WritableStream, readable: ReadableStream, lastEventId?: string) {
     super(writable, readable)
+    this.lastEventId = lastEventId
   }
 
   async writeSSE(message: SSEMessage) {
@@ -76,7 +83,7 @@ export const streamSSE = (
   onError?: (e: Error, stream: SSEStreamingApi) => Promise<void>
 ): Response => {
   const { readable, writable } = new TransformStream()
-  const stream = new SSEStreamingApi(writable, readable)
+  const stream = new SSEStreamingApi(writable, readable, c.req.header('Last-Event-ID'))
 
   // Until Bun v1.1.27, Bun didn't call cancel() on the ReadableStream for Response objects from Bun.serve()
   if (isOldBunVersion()) {

src/utils/stream.test.ts

@@ -209,4 +209,52 @@ describe('StreamingApi', () => {
       process.off('unhandledRejection', unhandled)
     }
   })
+
+  it('write() is a no-op after abort() and does not touch the writer', async () => {
+    const { readable, writable } = new TransformStream()
+    const api = new StreamingApi(writable, readable)
+    const reader = api.responseReadable.getReader()
+    api.write('before')
+    expect((await reader.read()).value).toEqual(new TextEncoder().encode('before'))
+
+    api.abort()
+    expect(api.aborted).toBe(true)
+
+    // Must resolve immediately without enqueuing anything or throwing.
+    await expect(api.write('after')).resolves.toBe(api)
+    await expect(api.writeln('after')).resolves.toBe(api)
+    const { value, done } = await Promise.race([
+      reader.read(),
+      new Promise<{ value: undefined; done: true }>((resolve) =>
+        setTimeout(() => resolve({ value: undefined, done: true }), 100)
+      ),
+    ])
+    expect(done).toBe(true)
+    expect(value).toBeUndefined()
+  })
+
+  it('write() is a no-op after the client cancels the response stream', async () => {
+    const { readable, writable } = new TransformStream()
+    const api = new StreamingApi(writable, readable)
+    const reader = api.responseReadable.getReader()
+    api.write('first')
+    await reader.read()
+
+    await reader.cancel()
+    expect(api.aborted).toBe(true)
+
+    // Writing after the disconnect neither throws nor hangs.
+    const write = api.write('after-abort')
+    await expect(
+      Promise.race([write, new Promise((resolve) => setTimeout(() => resolve('hung'), 500))])
+    ).resolves.not.toBe('hung')
+  })
+
+  it('close() after abort() is a no-op and does not throw', async () => {
+    const { readable, writable } = new TransformStream()
+    const api = new StreamingApi(writable, readable)
+    api.abort()
+    await expect(api.close()).resolves.toBeUndefined()
+    expect(api.closed).toBe(true)
+  })
 })

src/utils/stream.ts

@@ -46,6 +46,11 @@ export class StreamingApi {
   }
 
   async write(input: Uint8Array | string): Promise<StreamingApi> {
+    if (this.aborted) {
+      // The client is gone and the writable may be errored; depending on the
+      // runtime, writing can throw or hang, so writes after abort are no-ops.
+      return this
+    }
     try {
       if (typeof input === 'string') {
         input = this.encoder.encode(input)
@@ -68,6 +73,10 @@ export class StreamingApi {
 
   async close() {
     this.closed = true
+    if (this.aborted) {
+      // The writable was torn down by abort(); there is nothing to close.
+      return
+    }
     try {
       await this.writer.close()
     } catch {