diff --git a/runtime-tests/node/index.test.ts b/runtime-tests/node/index.test.ts index 558101bd29..50b782908e 100644 --- a/runtime-tests/node/index.test.ts +++ b/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 => { + 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) => { diff --git a/src/helper/streaming/sse.test.tsx b/src/helper/streaming/sse.test.tsx index f5052721f8..9506978a47 100644 --- a/src/helper/streaming/sse.test.tsx +++ b/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]) + }) }) diff --git a/src/helper/streaming/sse.ts b/src/helper/streaming/sse.ts index be658ec616..2e26bb70c7 100644 --- a/src/helper/streaming/sse.ts +++ b/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 ): 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()) { diff --git a/src/utils/stream.test.ts b/src/utils/stream.test.ts index 6f6209b9fe..7b0c7241c8 100644 --- a/src/utils/stream.test.ts +++ b/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) + }) }) diff --git a/src/utils/stream.ts b/src/utils/stream.ts index 3999c61c37..c7dac6d655 100644 --- a/src/utils/stream.ts +++ b/src/utils/stream.ts @@ -46,6 +46,11 @@ export class StreamingApi { } async write(input: Uint8Array | string): Promise { + 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 {