From aec62d941ac383ffe183bb602e9945780958dabc Mon Sep 17 00:00:00 2001 From: Peter Dave Hello <3691490+PeterDaveHello@users.noreply.github.com> Date: Mon, 14 Sep 2026 01:39:58 +0800 Subject: [PATCH] Bound retained SSE parser data --- src/utils/eventsource-parser.mjs | 121 +++- src/utils/fetch-sse.mjs | 197 +++++- .../eventsource-parser-buffer-limit.test.mjs | 142 +++++ tests/unit/utils/eventsource-parser.test.mjs | 50 ++ .../utils/fetch-sse-reader-cleanup.test.mjs | 603 ++++++++++++++++++ 5 files changed, 1094 insertions(+), 19 deletions(-) create mode 100644 tests/unit/utils/eventsource-parser-buffer-limit.test.mjs create mode 100644 tests/unit/utils/fetch-sse-reader-cleanup.test.mjs diff --git a/src/utils/eventsource-parser.mjs b/src/utils/eventsource-parser.mjs index ff7033872..a5eb7214e 100644 --- a/src/utils/eventsource-parser.mjs +++ b/src/utils/eventsource-parser.mjs @@ -1,6 +1,18 @@ // https://www.npmjs.com/package/eventsource-parser/v/1.1.1 -function createParser(onParse) { +// Bytes per decode, independent of the retained-state limit. +const MAX_DECODE_CHUNK_SIZE = 64 * 1024 +const MAX_KNOWN_FIELD_NAME_LENGTH = 'event'.length + +// maxBufferSize counts retained UTF-16 code units, not bytes or total heap usage. +function createParser(onParse, { maxBufferSize } = {}) { + if ( + maxBufferSize !== undefined && + (!Number.isSafeInteger(maxBufferSize) || maxBufferSize < 0) + ) { + throw new TypeError('maxBufferSize must be a non-negative safe integer') + } + let isFirstChunk let decoder let buffer @@ -10,7 +22,9 @@ function createParser(onParse) { let eventName let data let extra + let extraLength let discardTrailingNewline + let terminated reset() return { feed, @@ -26,11 +40,45 @@ function createParser(onParse) { eventName = void 0 data = '' extra = void 0 + extraLength = 0 discardTrailingNewline = false + terminated = false } function feed(chunk) { - buffer += decoder.decode(chunk, { stream: true }) + if (terminated) { + const err = new RangeError( + 'Cannot feed parser after exceeding max buffer size; call reset() to resume parsing', + ) + err.code = 'SSE_BUFFER_LIMIT_EXCEEDED' + throw err + } + + if (maxBufferSize === undefined) { + processDecodedChunk(decoder.decode(chunk, { stream: true })) + return + } + + const bytes = ArrayBuffer.isView(chunk) + ? new Uint8Array(chunk.buffer, chunk.byteOffset, chunk.byteLength) + : new Uint8Array(chunk) + + if (bytes.byteLength === 0) { + processDecodedChunk(decoder.decode(bytes, { stream: true })) + return + } + + let offset = 0 + while (offset < bytes.byteLength) { + const end = Math.min(bytes.byteLength, offset + MAX_DECODE_CHUNK_SIZE) + const slice = offset === 0 && end === bytes.byteLength ? bytes : bytes.subarray(offset, end) + processDecodedChunk(decoder.decode(slice, { stream: true }), end === bytes.byteLength) + offset = end + } + } + + function processDecodedChunk(decodedChunk, isLastSlice = true) { + buffer += decodedChunk if (isFirstChunk && hasBom(buffer)) { buffer = buffer.slice(BOM.length) } @@ -74,6 +122,63 @@ function createParser(onParse) { } else if (position > 0) { buffer = buffer.slice(position) } + if (isLastSlice) { + checkBufferSize(buffer.length) + } else { + checkTransientLineSize() + } + } + + function getRetainedSize(pendingBufferLength = buffer.length, additionalSize = 0) { + return ( + pendingBufferLength + + data.length + + (eventId?.length ?? 0) + + (eventName?.length ?? 0) + + (extra ? extraLength : 0) + + additionalSize + ) + } + + function getTransientLineSize() { + if (buffer.length === 0 || startingFieldLength < 0) return buffer.length + if (startingFieldLength > MAX_KNOWN_FIELD_NAME_LENGTH) return buffer.length + + const field = buffer.slice(0, startingFieldLength) + if (!['', 'data', 'event', 'id', 'retry', 'meta'].includes(field)) return buffer.length + + let valuePosition = startingFieldLength + 1 + if (buffer[valuePosition] === ' ') ++valuePosition + return Math.max(0, buffer.length - valuePosition) + } + + function checkTransientLineSize() { + if (maxBufferSize === undefined) return + + // Bound live parser-retained text, not only the eventual logical event state. + // Until an event/id replacement line is complete, its partial value and the + // currently retained eventName/eventId coexist and are intentionally both counted. + if (getRetainedSize(getTransientLineSize()) <= maxBufferSize) return + + throwBufferLimitError() + } + + function checkBufferSize(pendingBufferLength = buffer.length, additionalSize = 0) { + if (maxBufferSize === undefined) return + if (getRetainedSize(pendingBufferLength, additionalSize) <= maxBufferSize) return + + throwBufferLimitError() + } + + function throwBufferLimitError() { + reset() + terminated = true + + const err = new RangeError( + `SSE parser retained state exceeded ${maxBufferSize} UTF-16 code units`, + ) + err.code = 'SSE_BUFFER_LIMIT_EXCEEDED' + throw err } function parseEventStreamLine(lineBuffer, index, fieldLength, lineLength) { @@ -91,6 +196,7 @@ function createParser(onParse) { data = '' eventId = void 0 extra = void 0 + extraLength = 0 } eventName = void 0 return @@ -109,10 +215,14 @@ function createParser(onParse) { const valueLength = lineLength - step const value = lineBuffer.slice(position, position + valueLength).toString() if (field === 'data') { + const addedLength = value ? value.length + 1 : 1 + checkBufferSize(0, addedLength) data += value ? ''.concat(value, '\n') : '\n' } else if (field === 'event') { + checkBufferSize(0, value.length) eventName = value } else if (field === 'id' && !value.includes('\0')) { + checkBufferSize(0, value.length) eventId = value } else if (field === 'retry') { const retry = parseInt(value, 10) @@ -123,9 +233,14 @@ function createParser(onParse) { }) } } else if (field === 'meta') { + checkBufferSize(0, lineLength) const str = `{"${field}":${value}}` - extra = extra ?? [] + if (!extra) { + extra = [] + extraLength = 0 + } extra.push(JSON.parse(str)) + extraLength += lineLength } } } diff --git a/src/utils/fetch-sse.mjs b/src/utils/fetch-sse.mjs index b5c2baca6..038d1b8ae 100644 --- a/src/utils/fetch-sse.mjs +++ b/src/utils/fetch-sse.mjs @@ -3,8 +3,16 @@ import { isAbortError } from './abort-error.mjs' export const FETCH_REQUEST_FAILED = 'FETCH_REQUEST_FAILED' export const FETCH_RESPONSE_STREAM_FAILED = 'FETCH_RESPONSE_STREAM_FAILED' +export const FETCH_JSON_RESPONSE_TOO_LARGE = 'FETCH_JSON_RESPONSE_TOO_LARGE' export const INVALID_API_ENDPOINT = 'INVALID_API_ENDPOINT' +// Decoded UTF-16 code units; this is not a byte-accurate heap limit. +const MAX_SSE_BUFFER_SIZE = 8 * 1024 * 1024 +// Byte budget for the plain-JSON fallback; format probes are capped separately below. +const MAX_SSE_START_JSON_SIZE = 8 * 1024 * 1024 +const MAX_SSE_START_PREVIEW_SIZE = 64 * 1024 +const UTF8_BOM = [0xef, 0xbb, 0xbf] + function setErrorProperty(err, key, value) { try { err[key] = value @@ -31,6 +39,14 @@ function createInvalidApiEndpointError() { return err } +function createJsonResponseTooLargeError() { + const err = new RangeError( + `JSON response exceeded the ${MAX_SSE_START_JSON_SIZE}-byte fallback inspection limit`, + ) + setErrorProperty(err, 'code', FETCH_JSON_RESPONSE_TOO_LARGE) + return err +} + function classifyTransportError(err, code, requestOrigin) { const hasCode = setErrorProperty(err, 'code', code) const hasRequestOrigin = @@ -60,6 +76,48 @@ function annotateResponseStreamError(resource, err) { return classifyTransportError(err, FETCH_RESPONSE_STREAM_FAILED, url?.origin) } +function hasJsonContentType(resp) { + const contentType = resp.headers?.get?.('content-type') + if (typeof contentType !== 'string') return false + + const mimeType = contentType.split(';', 1)[0].trim().toLowerCase() + return mimeType === 'application/json' || mimeType === 'text/json' || mimeType.endsWith('+json') +} + +function getJsonRootProbeKind(chunk) { + let index = 0 + if (chunk.byteLength < UTF8_BOM.length && chunk.byteLength > 0) { + const isPartialBom = UTF8_BOM.slice(0, chunk.byteLength).every( + (byte, i) => chunk[i] === byte, + ) + if (isPartialBom) return 'partial-bom' + } + if (UTF8_BOM.every((byte, i) => chunk[i] === byte)) index = UTF8_BOM.length + + while ( + index < chunk.byteLength && + (chunk[index] === 0x20 || + chunk[index] === 0x09 || + chunk[index] === 0x0a || + chunk[index] === 0x0d) + ) { + ++index + } + if (index === chunk.byteLength) return 'blank' + + const byte = chunk[index] + const isJsonRoot = + byte === 0x7b || + byte === 0x5b || + byte === 0x22 || + byte === 0x2d || + (byte >= 0x30 && byte <= 0x39) || + byte === 0x74 || + byte === 0x66 || + byte === 0x6e + return isJsonRoot ? 'json-root' : 'other' +} + export async function fetchSSE(resource, options) { const { onMessage, onStart, onEnd, onError, ...fetchOptions } = options if (!getHttpRequestUrl(resource)) { @@ -85,13 +143,22 @@ export async function fetchSSE(resource, options) { await onError(resp) return } - const parser = createParser((event) => { - if (event.type === 'event') { - onMessage(event.data) - } - }) + let hasSseEvent = false + const parser = createParser( + (event) => { + if (event.type === 'event') { + hasSseEvent = true + onMessage(event.data) + } + }, + { maxBufferSize: MAX_SSE_BUFFER_SIZE }, + ) const handleCallbackError = async (err) => { - await onError(err) + try { + await onError(err) + } catch (onErrorError) { + console.warn('[fetch-sse] onError threw while handling processing failure:', onErrorError) + } throw err } const handleResponseStreamError = async (err) => { @@ -106,6 +173,10 @@ export async function fetchSSE(resource, options) { await onError(annotateResponseStreamError(resource, err)) } let hasStarted = false + let oversizedJsonCandidate = hasJsonContentType(resp) + let jsonShapeDetectionPending = !oversizedJsonCandidate + let jsonShapeProbePrefix = new Uint8Array() + let responseBytes = 0 let reader try { reader = resp.body.getReader() @@ -113,12 +184,44 @@ export async function fetchSSE(resource, options) { await handleResponseStreamError(err) return } + let readerReleased = false + const cleanupReader = async (cancel, waitForCancel = true) => { + if (readerReleased) return + readerReleased = true + + let cancellation + if (cancel) { + try { + cancellation = Promise.resolve(reader.cancel?.()) + } catch (err) { + console.warn('[fetch-sse] reader cancellation failed:', err) + } + } + try { + reader.releaseLock?.() + } catch (err) { + console.warn('[fetch-sse] reader lock release failed:', err) + } + if (!cancellation) return + if (!waitForCancel) { + cancellation.catch((err) => { + console.warn('[fetch-sse] reader cancellation failed:', err) + }) + return + } + try { + await cancellation + } catch (err) { + console.warn('[fetch-sse] reader cancellation failed:', err) + } + } let result let done = false while (!done) { try { result = await reader.read() } catch (err) { + await cleanupReader(false) await handleResponseStreamError(err) return } @@ -127,36 +230,98 @@ export async function fetchSSE(resource, options) { if (done) break const chunk = result.value + if (!hasStarted && chunk.byteLength === 0) continue + + responseBytes = Math.min( + MAX_SSE_START_JSON_SIZE + 1, + responseBytes + chunk.byteLength, + ) + + if (jsonShapeDetectionPending && chunk.byteLength > 0) { + const rawProbeChunk = + chunk.byteLength > MAX_SSE_START_PREVIEW_SIZE + ? chunk.subarray(0, MAX_SSE_START_PREVIEW_SIZE) + : chunk + let probeChunk = rawProbeChunk + if (jsonShapeProbePrefix.byteLength > 0) { + probeChunk = new Uint8Array( + jsonShapeProbePrefix.byteLength + rawProbeChunk.byteLength, + ) + probeChunk.set(jsonShapeProbePrefix) + probeChunk.set(rawProbeChunk, jsonShapeProbePrefix.byteLength) + } + + const probeKind = getJsonRootProbeKind(probeChunk) + if (probeKind === 'partial-bom') { + jsonShapeProbePrefix = probeChunk + } else { + jsonShapeProbePrefix = new Uint8Array() + if (probeKind === 'json-root') { + oversizedJsonCandidate = true + jsonShapeDetectionPending = false + } else if (probeKind === 'other') { + jsonShapeDetectionPending = false + } else if (chunk.byteLength > rawProbeChunk.byteLength) { + // The bounded probe is entirely framing but the chunk continues. Treat the + // unseen remainder as ambiguous instead of scanning attacker-sized input. + // A later parsed SSE event still wins at EOF via hasSseEvent. + oversizedJsonCandidate = true + jsonShapeDetectionPending = false + } + } + } + if (!hasStarted) { - const str = new TextDecoder().decode(chunk) + const startChunk = + chunk.byteLength > MAX_SSE_START_JSON_SIZE + ? chunk.subarray(0, MAX_SSE_START_PREVIEW_SIZE) + : chunk + const str = new TextDecoder().decode(startChunk) hasStarted = true try { await onStart(str) } catch (err) { + await cleanupReader(true, false) await handleCallbackError(err) } - let fakeSseData - try { - const commonResponse = JSON.parse(str) - fakeSseData = 'data: ' + JSON.stringify(commonResponse) + '\n\ndata: [DONE]\n\n' - } catch (error) { - console.debug('not common response', error) + let commonResponse + let isCommonResponse = false + if (chunk.byteLength <= MAX_SSE_START_JSON_SIZE) { + try { + commonResponse = JSON.parse(str) + isCommonResponse = true + } catch (error) { + console.debug('not common response', error) + } } - if (fakeSseData) { + if (isCommonResponse) { try { - parser.feed(new TextEncoder().encode(fakeSseData)) + onMessage(JSON.stringify(commonResponse)) + onMessage('[DONE]') } catch (err) { + await cleanupReader(true, false) await handleCallbackError(err) } - break + await cleanupReader(true, false) + await onEnd() + return } } try { parser.feed(chunk) } catch (err) { + await cleanupReader(true, false) await handleCallbackError(err) } } + await cleanupReader(false) + if ( + oversizedJsonCandidate && + responseBytes > MAX_SSE_START_JSON_SIZE && + !hasSseEvent + ) { + await handleCallbackError(createJsonResponseTooLargeError()) + } await onEnd() } diff --git a/tests/unit/utils/eventsource-parser-buffer-limit.test.mjs b/tests/unit/utils/eventsource-parser-buffer-limit.test.mjs new file mode 100644 index 000000000..03cfe4b8b --- /dev/null +++ b/tests/unit/utils/eventsource-parser-buffer-limit.test.mjs @@ -0,0 +1,142 @@ +import assert from 'node:assert/strict' +import { test } from 'node:test' +import { createParser } from '../../../src/utils/eventsource-parser.mjs' + +const encoder = new TextEncoder() +const toBytes = (text) => encoder.encode(text) +const isBufferLimitError = (err) => + err instanceof RangeError && err.code === 'SSE_BUFFER_LIMIT_EXCEEDED' + +const assertBufferLimit = (input, maxBufferSize) => { + const parser = createParser(() => {}, { maxBufferSize }) + assert.throws(() => parser.feed(toBytes(input)), isBufferLimitError) +} + +const assertLiveReplacementLimit = (field) => { + const limit = 100 * 1024 + const parser = createParser(() => {}, { maxBufferSize: limit }) + + const oldValue = 'a'.repeat(90 * 1024) + parser.feed(toBytes(`${field}: ${oldValue}\n`)) + const slicedReplacement = toBytes(`${field}: ${'b'.repeat(90 * 1024)}\n`) + assert.ok(slicedReplacement.byteLength > 64 * 1024) + assert.throws(() => parser.feed(slicedReplacement), isBufferLimitError) + + parser.reset() + parser.feed(toBytes(`${field}: ${'a'.repeat(60 * 1024)}\n`)) + const finalSliceReplacement = toBytes(`${field}: ${'b'.repeat(50 * 1024)}\n`) + assert.ok(finalSliceReplacement.byteLength < 64 * 1024) + assert.throws(() => parser.feed(finalSliceReplacement), isBufferLimitError) +} + +test('createParser counts retained event, id, and meta state toward the limit', () => { + assertBufferLimit('event: 123456789\n', 8) + assertBufferLimit('id: 123456789\n', 8) + assertBufferLimit('meta: {"source":"test"}\n', 16) +}) + +test('createParser rejects an oversized complete event before dispatch', () => { + const parsed = [] + const parser = createParser((event) => parsed.push(event), { maxBufferSize: 10 }) + + assert.throws(() => parser.feed(toBytes('data: 1234567890\n\n')), isBufferLimitError) + assert.deepEqual(parsed, []) +}) + +test('createParser reset discards metadata and its buffer accounting', () => { + const parsed = [] + const parser = createParser((event) => parsed.push(event), { maxBufferSize: 13 }) + + parser.feed(toBytes('meta: {"a":1}\n')) + parser.reset() + parser.feed(toBytes('meta: {"a":1}\n')) + parser.reset() + parser.feed(toBytes('data: ok\n\n')) + + assert.deepEqual(parsed.map((event) => [event.data, event.extra]), [['ok', undefined]]) +}) + +test('createParser processes a large transport chunk containing small events', () => { + const parsed = [] + const parser = createParser((event) => parsed.push(event), { maxBufferSize: 8 }) + const eventCount = 20000 + + parser.feed(toBytes('data: a\n\n'.repeat(eventCount))) + + assert.equal(parsed.length, eventCount) + assert.equal(parsed.every((event) => event.data === 'a'), true) +}) + +test('createParser preserves UTF-8 and line framing across internal decode slices', () => { + const limit = 64 * 1024 + const answer = '台'.repeat(limit - 3) + '🙂' + const parsed = [] + const parser = createParser((event) => parsed.push(event), { maxBufferSize: limit }) + const bytes = toBytes(`data: ${answer}\r\n\r\n`) + + assert.ok(bytes.byteLength > 64 * 1024) + parser.feed(bytes) + assert.deepEqual(parsed.map((event) => event.data), [answer]) + assertBufferLimit(`data: ${answer}x\n\n`, limit) +}) + +test('createParser counts partial slices together with already retained event state', (t) => { + const limit = 128 * 1024 + const parser = createParser(() => {}, { maxBufferSize: limit }) + parser.feed(toBytes(`data: ${'a'.repeat(96 * 1024)}\n`)) + + const decode = TextDecoder.prototype.decode + let decodedBytes = 0 + t.mock.method(TextDecoder.prototype, 'decode', function (input, options) { + decodedBytes += input.byteLength + return decode.call(this, input, options) + }) + + assert.throws( + () => parser.feed(toBytes(`data: ${'b'.repeat(96 * 1024)}`)), + isBufferLimitError, + ) + assert.equal(decodedBytes, 64 * 1024) +}) + +test('createParser counts a live event replacement across decode slices and final slices', () => { + assertLiveReplacementLimit('event') +}) + +test('createParser counts a live id replacement across decode slices and final slices', () => { + assertLiveReplacementLimit('id') +}) + +test('createParser counts unknown field names instead of treating them as free syntax', () => { + const parser = createParser(() => {}, { maxBufferSize: 16 }) + assert.throws(() => parser.feed(toBytes(`${'x'.repeat(100)}:`)), isBufferLimitError) +}) + +test('createParser keeps BufferSource compatibility when the limit is enabled', () => { + const parsed = [] + const parser = createParser((event) => parsed.push(event), { maxBufferSize: 20 }) + const first = toBytes('data: one\n\n').slice() + const second = toBytes('data: two\n\n').slice() + + parser.feed(first.buffer) + parser.feed(new DataView(second.buffer, second.byteOffset, second.byteLength)) + + assert.deepEqual(parsed.map((event) => event.data), ['one', 'two']) +}) + +test('createParser bounds each decode and stops a huge unfinished line early', (t) => { + const decode = TextDecoder.prototype.decode + let decodedBytes = 0 + let maxDecodeBytes = 0 + t.mock.method(TextDecoder.prototype, 'decode', function (input, options) { + decodedBytes += input.byteLength + maxDecodeBytes = Math.max(maxDecodeBytes, input.byteLength) + return decode.call(this, input, options) + }) + + const bytes = toBytes('x'.repeat(1024 * 1024)) + const parser = createParser(() => {}, { maxBufferSize: 16 }) + assert.throws(() => parser.feed(bytes), isBufferLimitError) + assert.ok(maxDecodeBytes <= 64 * 1024) + assert.ok(decodedBytes < bytes.byteLength) +}) diff --git a/tests/unit/utils/eventsource-parser.test.mjs b/tests/unit/utils/eventsource-parser.test.mjs index 6fd691d80..d899035ad 100644 --- a/tests/unit/utils/eventsource-parser.test.mjs +++ b/tests/unit/utils/eventsource-parser.test.mjs @@ -212,6 +212,56 @@ test('createParser reset discards pending event metadata', () => { ]) }) +test('createParser rejects invalid maxBufferSize values', () => { + assert.throws( + () => createParser(() => {}, { maxBufferSize: -1 }), + /maxBufferSize must be a non-negative safe integer/, + ) + assert.throws( + () => createParser(() => {}, { maxBufferSize: 1.5 }), + /maxBufferSize must be a non-negative safe integer/, + ) +}) + +test('createParser limits an unfinished line buffered across chunks', () => { + const parser = createParser(() => {}, { maxBufferSize: 12 }) + + parser.feed(toBytes('data: 12345')) + assert.throws( + () => parser.feed(toBytes('67')), + (err) => err instanceof RangeError && err.code === 'SSE_BUFFER_LIMIT_EXCEEDED', + ) +}) + +test('createParser limits accumulated multiline event data', () => { + const parser = createParser(() => {}, { maxBufferSize: 10 }) + + parser.feed(toBytes('data: 12345\n')) + assert.throws( + () => parser.feed(toBytes('data: 67890\n')), + (err) => err instanceof RangeError && err.code === 'SSE_BUFFER_LIMIT_EXCEEDED', + ) +}) + +test('createParser reset recovers after exceeding maxBufferSize', () => { + const parsed = [] + const parser = createParser((event) => parsed.push(event), { maxBufferSize: 8 }) + + assert.throws( + () => parser.feed(toBytes('data: 123')), + (err) => err instanceof RangeError && err.code === 'SSE_BUFFER_LIMIT_EXCEEDED', + ) + assert.throws(() => parser.feed(toBytes('data: stale\n\n')), /Cannot feed parser/) + + parser.reset() + parser.feed(toBytes('data: ok\n\n')) + + assert.deepEqual( + parsed.map((event) => event.data), + ['ok'], + ) +}) + test('createParser handles \\r only line endings', () => { const parsed = [] const parser = createParser((event) => parsed.push(event)) diff --git a/tests/unit/utils/fetch-sse-reader-cleanup.test.mjs b/tests/unit/utils/fetch-sse-reader-cleanup.test.mjs new file mode 100644 index 000000000..5b1f1adfa --- /dev/null +++ b/tests/unit/utils/fetch-sse-reader-cleanup.test.mjs @@ -0,0 +1,603 @@ +import assert from 'node:assert/strict' +import { setImmediate } from 'node:timers' +import { test } from 'node:test' +import { + FETCH_JSON_RESPONSE_TOO_LARGE, + fetchSSE, +} from '../../../src/utils/fetch-sse.mjs' + +const encoder = new TextEncoder() + +test('fetchSSE cancels and releases the reader when parser processing throws', async (t) => { + t.mock.method(console, 'debug', () => {}) + const consoleWarn = t.mock.method(console, 'warn', () => {}) + const callbackError = new Error('message failed') + let readCount = 0 + let cancelCount = 0 + let releaseCount = 0 + let endCount = 0 + const errors = [] + + const reader = { + async read() { + if (readCount++ === 0) { + return { done: false, value: encoder.encode('data: hello\n\n') } + } + return { done: true, value: undefined } + }, + async cancel() { + cancelCount += 1 + throw new Error('reader cancellation failed') + }, + releaseLock() { + releaseCount += 1 + }, + } + + t.mock.method(globalThis, 'fetch', async () => ({ + ok: true, + body: { + getReader() { + return reader + }, + }, + })) + + await assert.rejects( + fetchSSE('https://example.com/sse', { + onStart: async () => {}, + onMessage: () => { + throw callbackError + }, + onEnd: async () => { + endCount += 1 + }, + onError: async (error) => { + errors.push(error) + }, + }), + callbackError, + ) + + await new Promise((resolve) => setImmediate(resolve)) + assert.deepEqual(errors, [callbackError]) + assert.equal(cancelCount, 1) + assert.equal(releaseCount, 1) + assert.equal(endCount, 0) + assert.equal(consoleWarn.mock.callCount(), 1) +}) + +test('fetchSSE preserves callback errors when onError and reader cleanup also throw', async (t) => { + for (const callback of ['onStart', 'onMessage']) { + await t.test(callback, async (t) => { + t.mock.method(console, 'debug', () => {}) + const warn = t.mock.method(console, 'warn', () => {}) + const original = new Error('original processing failure') + const cleanup = [] + const reader = { + async read() { + return { done: false, value: encoder.encode('data: hello\n\n') } + }, + async cancel() { + cleanup.push('cancel') + throw new Error('cancel failed') + }, + releaseLock() { + cleanup.push('release') + throw new Error('release failed') + }, + } + t.mock.method(globalThis, 'fetch', async () => ({ + ok: true, + body: { getReader: () => reader }, + })) + await assert.rejects( + fetchSSE('https://example.com/sse', { + onStart: () => {}, + onMessage: () => {}, + onEnd: () => assert.fail('must not complete after a processing failure'), + onError: async (err) => { + assert.equal(err, original) + throw new Error('onError failed') + }, + [callback]: () => { + throw original + }, + }), + (err) => err === original, + ) + await new Promise((resolve) => setImmediate(resolve)) + assert.deepEqual(cleanup, ['cancel', 'release']) + assert.equal(warn.mock.callCount(), 3) + }) + } +}) + +test( + 'fetchSSE bounds huge first-chunk decoding and preserves parser overflow errors', + async (t) => { + t.mock.method(console, 'debug', () => {}) + t.mock.method(console, 'warn', () => {}) + const decode = TextDecoder.prototype.decode + let maxDecodeBytes = 0 + t.mock.method(TextDecoder.prototype, 'decode', function (input, options) { + maxDecodeBytes = Math.max(maxDecodeBytes, input.byteLength) + return decode.call(this, input, options) + }) + const parse = t.mock.method(JSON, 'parse') + let cancellations = 0 + let reported + let previewLength + const body = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode('data: ' + 'x'.repeat(9 * 1024 * 1024))) + }, + cancel() { + cancellations += 1 + }, + }) + t.mock.method(globalThis, 'fetch', async () => ({ ok: true, body })) + await assert.rejects( + fetchSSE('https://example.com/sse', { + onStart: (preview) => { + previewLength = preview.length + }, + onMessage: () => assert.fail('must not dispatch an oversized event'), + onEnd: () => assert.fail('must not complete an oversized event'), + onError: async (err) => { + reported = err + throw new Error('secondary error') + }, + }), + (err) => err === reported && err.code === 'SSE_BUFFER_LIMIT_EXCEEDED', + ) + assert.ok(maxDecodeBytes <= 64 * 1024) + assert.equal(previewLength, 64 * 1024) + assert.equal(parse.mock.callCount(), 0) + assert.equal(cancellations, 1) + assert.equal(body.locked, false) + }, +) + +test( + 'fetchSSE accepts a huge first chunk with leading SSE framing despite JSON content type', + async (t) => { + t.mock.method(console, 'debug', () => {}) + const data = 'x'.repeat(1024) + const eventCount = 8192 + const leadingBlankLines = '\n'.repeat(64 * 1024 + 1) + const chunk = encoder.encode( + `${leadingBlankLines}: keepalive\n\n${`data: ${data}\n\n`.repeat(eventCount)}`, + ) + assert.ok(chunk.byteLength > 8 * 1024 * 1024) + let messageCount = 0 + let endCount = 0 + const body = new ReadableStream({ + start(controller) { + controller.enqueue(chunk) + controller.close() + }, + cancel: () => assert.fail('a normally completed stream must not be cancelled'), + }) + t.mock.method(globalThis, 'fetch', async () => ({ + ok: true, + headers: new Headers({ 'content-type': 'application/json' }), + body, + })) + await fetchSSE('https://example.com/sse', { + onStart: (preview) => assert.equal(preview, '\n'.repeat(64 * 1024)), + onMessage: (message) => { + assert.equal(message, data) + messageCount += 1 + }, + onEnd: () => { + assert.equal(body.locked, false) + endCount += 1 + }, + onError: (err) => { + throw err + }, + }) + assert.equal(messageCount, eventCount) + assert.equal(endCount, 1) + }, +) + +test( + 'fetchSSE lets the parser disambiguate object-like oversized SSE despite JSON content type', + async (t) => { + t.mock.method(console, 'debug', () => {}) + const paddingLine = `ignored: ${'x'.repeat(64 * 1024 - 16)}\n` + const chunk = encoder.encode(`{ignored: value\n${paddingLine.repeat(129)}data: ok\n\n`) + assert.ok(chunk.byteLength > 8 * 1024 * 1024) + + const messages = [] + let endCount = 0 + const body = new ReadableStream({ + start(controller) { + controller.enqueue(chunk) + controller.close() + }, + }) + t.mock.method(globalThis, 'fetch', async () => ({ + ok: true, + headers: new Headers({ 'content-type': 'application/json' }), + body, + })) + + await fetchSSE('https://example.com/sse', { + onStart: (preview) => assert.equal(preview.startsWith('{ignored: value\n'), true), + onMessage: (message) => messages.push(message), + onEnd: () => { + endCount += 1 + }, + onError: (err) => { + throw err + }, + }) + + assert.deepEqual(messages, ['ok']) + assert.equal(endCount, 1) + assert.equal(body.locked, false) + }, +) + +test( + 'fetchSSE lets the parser disambiguate scalar-like oversized SSE without content type', + async (t) => { + t.mock.method(console, 'debug', () => {}) + const paddingLine = `ignored: ${'x'.repeat(64 * 1024 - 16)}\n` + const chunk = encoder.encode(`"ignored: value\n${paddingLine.repeat(129)}data: ok\n\n`) + assert.ok(chunk.byteLength > 8 * 1024 * 1024) + + const messages = [] + let endCount = 0 + const body = new ReadableStream({ + start(controller) { + controller.enqueue(chunk) + controller.close() + }, + }) + t.mock.method(globalThis, 'fetch', async () => ({ ok: true, body })) + + await fetchSSE('https://example.com/sse', { + onStart: (preview) => assert.equal(preview.startsWith('"ignored: value\n'), true), + onMessage: (message) => messages.push(message), + onEnd: () => { + endCount += 1 + }, + onError: (err) => { + throw err + }, + }) + + assert.deepEqual(messages, ['ok']) + assert.equal(endCount, 1) + assert.equal(body.locked, false) + }, +) + +test( + 'fetchSSE finishes plain JSON fallback after an empty chunk before cancellation settles', + async (t) => { + const json = '{"answer":"hello"}' + const messages = [] + let cancellations = 0 + let endCount = 0 + let resolveCancellation + const cancellation = new Promise((resolve) => { + resolveCancellation = resolve + }) + t.after(() => resolveCancellation()) + + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array()) + controller.enqueue(encoder.encode(json)) + }, + cancel() { + cancellations += 1 + return cancellation + }, + }) + t.mock.method(globalThis, 'fetch', async () => ({ ok: true, body })) + + let outcome + const request = fetchSSE('https://example.com/sse', { + onStart: (preview) => assert.equal(preview, json), + onMessage: (message) => { + messages.push(message) + }, + onEnd: () => { + assert.equal(body.locked, false) + endCount += 1 + }, + onError: (err) => { + throw err + }, + }).then( + () => { + outcome = { status: 'fulfilled' } + }, + (err) => { + outcome = { status: 'rejected', err } + }, + ) + + await new Promise((resolve) => setImmediate(resolve)) + assert.deepEqual(messages, [json, '[DONE]']) + assert.equal(cancellations, 1) + assert.equal(endCount, 1) + assert.equal(body.locked, false) + assert.equal(outcome?.status, 'fulfilled') + + resolveCancellation() + await request + }, +) + +test('fetchSSE delivers plain JSON exactly at the first-chunk inspection limit', async (t) => { + const maxJsonBytes = 8 * 1024 * 1024 + const prefix = '{"value":"' + const suffix = '"}' + const payloadLength = maxJsonBytes - encoder.encode(prefix + suffix).byteLength + const json = prefix + 'x'.repeat(payloadLength) + suffix + const chunk = encoder.encode(json) + assert.equal(chunk.byteLength, maxJsonBytes) + + const messages = [] + let cancellations = 0 + let endCount = 0 + const body = new ReadableStream({ + start(controller) { + controller.enqueue(chunk) + }, + cancel() { + cancellations += 1 + }, + }) + t.mock.method(globalThis, 'fetch', async () => ({ ok: true, body })) + + await fetchSSE('https://example.com/sse', { + onStart: (preview) => assert.equal(preview.length, json.length), + onMessage: (message) => messages.push(message), + onEnd: () => { + endCount += 1 + }, + onError: (err) => { + throw err + }, + }) + + assert.deepEqual(messages, [json, '[DONE]']) + assert.equal(cancellations, 1) + assert.equal(endCount, 1) + assert.equal(body.locked, false) +}) + +test('fetchSSE rejects oversized JSON instead of silently completing', async (t) => { + const cases = [ + { + name: 'multibyte JSON without content type', + createJson: () => + JSON.stringify({ + choices: [{ message: { content: '台'.repeat(3 * 1024 * 1024) } }], + }), + contentType: null, + expectedCancellations: 0, + }, + { + name: 'multibyte JSON string without content type', + createJson: () => `"${'台'.repeat(3 * 1024 * 1024)}"`, + contentType: null, + expectedCancellations: 0, + }, + { + name: 'JSON with large trailing whitespace and JSON content type', + createJson: () => + '{"choices":[{"message":{"content":"ok"}}]}' + + '\n'.repeat(8 * 1024 * 1024), + contentType: 'application/json', + expectedCancellations: 0, + }, + { + name: 'JSON object after a full blank preview without content type', + createJson: () => + '\n'.repeat(64 * 1024 + 1) + + '{"choices":[{"message":{"content":"ok"}}]}' + + '\n'.repeat(8 * 1024 * 1024), + contentType: null, + expectedCancellations: 0, + }, + ] + + for (const { name, createJson, contentType, expectedCancellations } of cases) { + await t.test(name, async (t) => { + const json = createJson() + const chunk = encoder.encode(json) + assert.ok(chunk.byteLength > 8 * 1024 * 1024) + const expectedPreview = new TextDecoder().decode(chunk.subarray(0, 64 * 1024)) + + const messages = [] + const errors = [] + let endCount = 0 + let cancellations = 0 + const body = new ReadableStream({ + start(controller) { + controller.enqueue(chunk) + controller.close() + }, + cancel() { + cancellations += 1 + }, + }) + t.mock.method(globalThis, 'fetch', async () => ({ + ok: true, + headers: contentType ? new Headers({ 'content-type': contentType }) : undefined, + body, + })) + + await assert.rejects( + fetchSSE('https://example.com/sse', { + onStart: (preview) => assert.equal(preview, expectedPreview), + onMessage: (message) => messages.push(message), + onEnd: () => { + endCount += 1 + }, + onError: (err) => { + errors.push(err) + }, + }), + (err) => err.code === FETCH_JSON_RESPONSE_TOO_LARGE, + ) + + assert.deepEqual(messages, []) + assert.equal(errors.length, 1) + assert.equal(errors[0].code, FETCH_JSON_RESPONSE_TOO_LARGE) + assert.equal(endCount, 0) + assert.equal(cancellations, expectedCancellations) + assert.equal(body.locked, false) + }) + } +}) + +test('fetchSSE rejects oversized multibyte JSON after a split BOM and blank chunk', async (t) => { + const json = JSON.stringify({ + choices: [{ message: { content: '台'.repeat(3 * 1024 * 1024) } }], + }) + const bytes = encoder.encode(json) + assert.ok(bytes.byteLength > 8 * 1024 * 1024) + + const splitAt = 4 * 1024 * 1024 + const firstChunk = bytes.subarray(0, splitAt) + const secondChunk = bytes.subarray(splitAt) + const leadingBlankChunk = encoder.encode('\n\n') + assert.ok(firstChunk.byteLength <= 8 * 1024 * 1024) + assert.ok(secondChunk.byteLength <= 8 * 1024 * 1024) + + const messages = [] + const errors = [] + let endCount = 0 + const body = new ReadableStream({ + start(controller) { + controller.enqueue(Uint8Array.of(0xef)) + controller.enqueue(Uint8Array.of(0xbb)) + controller.enqueue(Uint8Array.of(0xbf)) + controller.enqueue(leadingBlankChunk) + controller.enqueue(firstChunk) + controller.enqueue(secondChunk) + controller.close() + }, + }) + t.mock.method(globalThis, 'fetch', async () => ({ ok: true, body })) + + await assert.rejects( + fetchSSE('https://example.com/sse', { + onStart: () => {}, + onMessage: (message) => messages.push(message), + onEnd: () => { + endCount += 1 + }, + onError: (err) => { + errors.push(err) + }, + }), + (err) => err.code === FETCH_JSON_RESPONSE_TOO_LARGE, + ) + + assert.deepEqual(messages, []) + assert.equal(errors.length, 1) + assert.equal(errors[0].code, FETCH_JSON_RESPONSE_TOO_LARGE) + assert.equal(endCount, 0) + assert.equal(body.locked, false) +}) + +test('fetchSSE reports processing errors before pending cancellation settles', async (t) => { + t.mock.method(console, 'debug', () => {}) + const original = new Error('message failed') + let resolveCancellation + const cancellation = new Promise((resolve) => { + resolveCancellation = resolve + }) + t.after(() => resolveCancellation()) + + let cancelStarted = false + let errorNotified = false + const body = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode('data: hello\n\n')) + }, + cancel() { + cancelStarted = true + return cancellation + }, + }) + t.mock.method(globalThis, 'fetch', async () => ({ ok: true, body })) + + let outcome + const request = fetchSSE('https://example.com/sse', { + onStart: () => {}, + onMessage: () => { + throw original + }, + onEnd: () => assert.fail('must not complete after a processing failure'), + onError: (err) => { + assert.equal(err, original) + errorNotified = true + }, + }).then( + () => { + outcome = { status: 'fulfilled' } + }, + (err) => { + outcome = { status: 'rejected', err } + }, + ) + + await new Promise((resolve) => setImmediate(resolve)) + assert.equal(cancelStarted, true) + assert.equal(errorNotified, true) + assert.equal(body.locked, false) + assert.equal(outcome?.status, 'rejected') + assert.equal(outcome?.err, original) + + resolveCancellation() + await request +}) + +test('fetchSSE releases the reader on stream errors and cancellation', async (t) => { + for (const abort of [false, true]) { + await t.test(abort ? 'abort' : 'read failure', async (t) => { + const original = abort + ? new DOMException('cancelled', 'AbortError') + : new Error('read failed') + const errors = [] + const endings = [] + const body = new ReadableStream({ + start(controller) { + controller.error(original) + }, + }) + t.mock.method(globalThis, 'fetch', async () => ({ ok: true, body })) + await fetchSSE('https://example.com/sse', { + onStart: () => assert.fail('must not start an errored stream'), + onMessage: () => assert.fail('must not dispatch from an errored stream'), + onError: (err) => { + errors.push(err) + }, + onEnd: (aborted) => { + endings.push(aborted) + }, + }) + assert.equal(body.locked, false) + if (abort) { + assert.deepEqual(errors, []) + assert.deepEqual(endings, [true]) + } else { + assert.equal(errors.length, 1) + assert.equal(errors[0].code, 'FETCH_RESPONSE_STREAM_FAILED') + assert.deepEqual(endings, []) + } + }) + } +}) \ No newline at end of file