src/sse-framer.js
Rama master · réplica de lectura del repositorio original.
// Shared SSE framer: one line-terminator rule for the three translators and the
// downstream commit gate. No dependencies.
//
// The canonical producer (chat.js) writes whole `data: <json>\n\n` strings, so the
// happy path is byte-identical. The framer hardens the seam against CRLF, split
// UTF-8, multi-line data, EOF without a blank line, and an unbounded buffer.
//
// The cap bounds retained logical frame bytes (event + data), not the wire
// prefix, so a network chunk boundary cannot change whether a legal frame is
// accepted. Data and the current event have separate budgets; the combined
// frame payload is checked when the frame dispatches, so a later event may
// shrink a frame that temporarily exceeded the cap. An event value over the cap
// fails at its completed or partial line even if a later event would replace
// it. Comment/unknown lines and id/retry values are bounded on their own and
// never enter either payload budget. A partial line's value bytes count
// against the frame cap as soon as its field prefix is decidable; the first
// MAX_FIELD_PREFIX_BYTES bytes of an undecided line are free.
//
// The partial line is kept as a list of chunks plus an 8-char head, never as a
// growing rope: classification and byte accounting stay O(1) per push and the
// whole line is joined once when its terminator arrives.
//
// Two budgets are enforced independently: `maxFrameBytes` (payload bytes of one
// frame) and `maxQueuedBytes` (the OBJECT overhead the framer retains while
// assembling it). The second exists because a flood of one-byte `data:` lines
// can stay far under the payload cap while retaining hundreds of MB of array
// slots and strings, so the queue budget charges a fixed per-line/per-segment
// overhead and NEVER the payload bytes — those are already bounded by the
// frame cap, and charging them twice would silently cap every frame at the
// smaller queue number.
//
// A `data:` payload carrying several concatenated JSON documents is repaired
// per line and dispatched as one frame per document; an unrepairable JSON
// payload becomes a typed `SSE_MALFORMED_JSON` error frame instead of a silent
// drop that would end the turn with zero content and a clean stop.
import { StringDecoder } from 'node:string_decoder';
export const DEFAULT_MAX_FRAME_BYTES = 64 * 1024 * 1024;
// A second, independent budget for the OBJECT overhead the framer retains while
// a frame is being assembled. The frame cap alone bounds payload bytes, but one
// retained line costs far more RSS than its value: it is an array slot plus a
// string object, so tens of millions of one-byte lines stay under the 64 MB
// payload cap while retaining hundreds of megabytes of JS objects. The queued
// budget therefore charges a fixed per-line (and per partial-line segment)
// overhead and nothing else: the value bytes are already bounded by the frame
// cap, so charging them here too would silently cap every frame at this
// smaller number and the documented 64 MB cap could never fire. Deliberately
// below the frame cap: it is the object-count bound, not the payload bound.
export const DEFAULT_MAX_QUEUED_BYTES = 8 * 1024 * 1024;
// Retained-cost estimate for one data/event line or one partial-line segment:
// array slot + string header, rounded up. Charged so a flood of tiny lines (or
// of tiny wire chunks) cannot hide behind its small payload byte count.
const QUEUED_LINE_OVERHEAD_BYTES = 16;
const TERMINATOR_RE = /(\r\n\r\n|\r\n\n|\r\n\r|\n\r\n|\n\n|\n\r|\r\r)/g;
const NEWLINE_RE = /[\r\n]/g;
const FIELD_RE = /^(event|data|id|retry)(?::(.*))?$/;
const FIELDS = ['event', 'data', 'id', 'retry'];
const MAX_FIELD_PREFIX_BYTES = 7; // 'event: ' / 'retry: '
const HEAD_CHARS = MAX_FIELD_PREFIX_BYTES + 1;
export function findFrameEnd(text, from = 0) {
TERMINATOR_RE.lastIndex = from;
const match = TERMINATOR_RE.exec(text);
if (!match) return null;
return { frameEnd: match.index, next: match.index + match[0].length };
}
// One line rule for the framer and the deferred commit gate. SSE consumes at
// most one ASCII space after the colon; a tab stays in the value. A recognized
// field with no colon is a field with an empty value; other colon-less lines
// name no field and are ignored.
export function parseSseLine(line) {
const match = FIELD_RE.exec(line);
if (!match) return null;
const raw = match[2] ?? '';
return { field: match[1], value: raw.startsWith(' ') ? raw.slice(1) : raw };
}
// ─── JSON document repair for `data:` payloads ───────────────────────────────
//
// An upstream may put several complete JSON documents on one `data:` line
// (`{"a":1}{"b":2}`) instead of one document per frame. `JSON.parse` rejects
// that payload as a whole, and a consumer that drops it ends the turn with zero
// content and a clean `stop` — the exact silent-empty failure this helper
// removes. The payload is split at document boundaries so every document
// reaches `onFrame` on its own; a JSON-looking payload that cannot be split
// into complete documents is reported as malformed, and the caller must fail
// typed instead of dropping it.
//
// Only object/array payloads are repair candidates. `[DONE]`, comments and
// plain-text data are not JSON frames and are returned untouched.
const JSON_WS = new Set([0x20, 0x09, 0x0a, 0x0d]);
function isJsonSpace(code) {
return JSON_WS.has(code);
}
/** Index one past the closing bracket of the JSON value at `start`, or -1. */
function scanJsonDocumentEnd(text, start) {
let depth = 0;
let inString = false;
let escaped = false;
for (let i = start; i < text.length; i += 1) {
const c = text.charCodeAt(i);
if (inString) {
if (escaped) { escaped = false; continue; }
if (c === 0x5c) { escaped = true; continue; }
if (c === 0x22) inString = false;
continue;
}
if (c === 0x22) { inString = true; continue; }
if (c === 0x7b || c === 0x5b) depth += 1;
else if (c === 0x7d || c === 0x5d) {
depth -= 1;
if (depth === 0) return i + 1;
}
}
return -1;
}
/**
* True when the payload is a JSON-object frame (its first non-whitespace
* character is `{`). `[DONE]`, comments and plain text are not.
*/
function isJsonObjectPayload(text) {
let start = 0;
while (start < text.length && isJsonSpace(text.charCodeAt(start))) start += 1;
return start < text.length && text.charCodeAt(start) === 0x7b;
}
function malformedDataFrame(event) {
return {
event,
data: JSON.stringify({
error: {
message: 'upstream SSE data is not valid JSON and could not be repaired',
type: 'api_error',
code: 'SSE_MALFORMED_JSON',
},
}),
};
}
/**
* Split concatenated JSON documents (`{"a":1}{"b":2}`) into parsed values.
* Returns `null` when the text is not a sequence of complete JSON values.
*/
export function splitJsonDocuments(text) {
const docs = [];
let i = 0;
while (i < text.length) {
while (i < text.length && isJsonSpace(text.charCodeAt(i))) i += 1;
if (i >= text.length) break;
const first = text.charCodeAt(i);
if (first !== 0x7b && first !== 0x5b) return null;
const end = scanJsonDocumentEnd(text, i);
if (end < 0) return null;
try {
docs.push(JSON.parse(text.slice(i, end)));
} catch {
return null;
}
i = end;
}
return docs.length ? docs : null;
}
/**
* Parse an SSE `data:` payload into every JSON document it carries.
*
* Returns `{ docs, malformed }`:
* * `docs: [...]` — one or more parsed documents (a plain JSON frame is one);
* * `docs: null` — the payload is not a JSON object payload (`[DONE]`,
* comments, plain text) and must be passed through as-is;
* * `malformed:true` — the payload starts like a JSON object but no complete
* document sequence can be parsed from it.
*
* Repair is per `data:` line, not per frame: a frame's data is the joined value
* of its `data:` lines, so the whole payload is tried first (a pretty-printed
* document spans lines) and each line on its own afterwards.
*/
export function parseSseDataJson(data) {
const text = typeof data === 'string' ? data : '';
const trimmed = text.trim();
if (!trimmed) return { docs: [], malformed: false };
try {
return { docs: [JSON.parse(trimmed)], malformed: false };
} catch { /* try to repair below */ }
if (trimmed.charCodeAt(0) !== 0x7b) return { docs: null, malformed: false };
const whole = splitJsonDocuments(trimmed);
if (whole && whole.length > 1) return { docs: whole, malformed: false };
const docs = [];
for (const line of trimmed.split('\n')) {
const value = line.trim();
if (!value) continue;
if (value.charCodeAt(0) !== 0x7b) return { docs: null, malformed: true };
try {
docs.push(JSON.parse(value));
continue;
} catch { /* concatenated documents on one line */ }
const split = splitJsonDocuments(value);
if (!split) return { docs: null, malformed: true };
docs.push(...split);
}
if (!docs.length) return { docs: null, malformed: true };
return { docs, malformed: false };
}
function classifyHead(head) {
if (head.charCodeAt(0) === 58) return { comment: true, prefixBytes: 0 };
for (const field of FIELDS) {
if (head.length <= field.length) continue;
if (!head.startsWith(field) || head.charCodeAt(field.length) !== 58) continue;
// Wait for the next character before deciding whether a space belongs to
// the prefix; the choice changes the charged value bytes by one.
if (head.length === field.length + 1) return undefined;
const prefixBytes = field.length + 1 + (head.charCodeAt(field.length + 1) === 32 ? 1 : 0);
return { field, prefixBytes };
}
return undefined;
}
export class SseFramer {
constructor({
maxFrameBytes = DEFAULT_MAX_FRAME_BYTES,
maxQueuedBytes = DEFAULT_MAX_QUEUED_BYTES,
onFrame,
onComment,
onOverflow,
} = {}) {
this.maxFrameBytes = maxFrameBytes;
this.maxQueuedBytes = maxQueuedBytes;
this.onFrame = onFrame;
this.onComment = onComment;
this.onOverflow = onOverflow;
this.decoder = new StringDecoder('utf8');
// A UTF-8 BOM is transport decoration for the first field, not payload;
// it is dropped once, before the first line is classified.
this.atStart = true;
this.parts = [];
this.partialBytes = 0;
this.lineHead = '';
this.partialClass = undefined;
this.pendingCr = false;
this.event = null;
this.data = [];
this.payloadBytes = 0;
this.dataBytes = 0;
this.eventBytes = 0;
// Retained-cost accounting for the queued-byte budget: one fixed charge
// per retained partial-line segment and per retained data/event line.
// `eventCharge` is what the current event line contributed, so a replacing
// event line can release it.
this.queuedBytes = 0;
this.eventCharge = 0;
this.failed = false;
this.ended = false;
}
get bufferedBytes() {
return this.payloadBytes + this.partialBytes;
}
push(chunk) {
if (this.failed || this.ended) return;
const text = typeof chunk === 'string' ? chunk : this.decoder.write(Buffer.from(chunk));
if (text) this.#feed(text);
const over = this.failed ? null : this.#overCapReason();
if (over) this.#overflow(over);
}
end() {
if (this.failed || this.ended) return;
this.ended = true;
const tail = this.decoder.end();
if (tail) this.#feed(tail);
if (!this.failed && this.pendingCr) {
this.pendingCr = false;
this.#completeLine();
}
if (!this.failed && this.parts.length) this.#completeLine();
if (!this.failed) this.#dispatch();
const over = this.failed ? null : this.#overCapReason();
if (over) this.#overflow(over);
}
#feed(text) {
if (this.atStart) {
this.atStart = false;
if (text.charCodeAt(0) === 0xfeff) text = text.slice(1);
if (!text) return;
}
let i = 0;
const n = text.length;
if (this.pendingCr) {
// The previous chunk ended with CR. LF makes it one CRLF terminator;
// anything else makes the CR a lone terminator.
if (text.charCodeAt(0) === 10) i = 1;
this.pendingCr = false;
this.#completeLine();
if (this.failed) return;
}
while (i < n) {
NEWLINE_RE.lastIndex = i;
const match = NEWLINE_RE.exec(text);
if (!match) {
this.#appendSegment(text.slice(i));
return;
}
const idx = match.index;
if (idx > i) this.#appendSegment(text.slice(i, idx));
if (text.charCodeAt(idx) === 13) {
if (idx + 1 === n) {
this.pendingCr = true;
return;
}
this.#completeLine();
i = idx + (text.charCodeAt(idx + 1) === 10 ? 2 : 1);
} else {
this.#completeLine();
i = idx + 1;
}
if (this.failed) return;
}
}
#appendSegment(segment) {
if (!segment) return;
const bytes = Buffer.byteLength(segment, 'utf8');
this.parts.push(segment);
this.partialBytes += bytes;
// The queue budget charges the retained OBJECT (one array slot plus one
// string per segment), never the payload bytes — those are the frame
// cap's business. A line arriving in many small chunks retains many slots,
// so the charge is per segment, not per byte.
this.queuedBytes += QUEUED_LINE_OVERHEAD_BYTES;
if (this.lineHead.length < HEAD_CHARS) {
this.lineHead += segment.slice(0, HEAD_CHARS - this.lineHead.length);
}
}
#partialInfo() {
if (this.partialClass !== undefined) return this.partialClass;
const info = classifyHead(this.lineHead);
if (info !== undefined) {
this.partialClass = info;
return info;
}
if (this.partialBytes > MAX_FIELD_PREFIX_BYTES) {
this.partialClass = null;
return null;
}
return undefined;
}
// Two independent budgets. The frame cap bounds payload bytes (data + event)
// and keeps its exact previous semantics. The queued budget bounds the OBJECT
// overhead the framer RETAINS: one fixed charge per partial-line segment and
// per retained data/event line, so a flood of tiny lines cannot grow the
// array/strings unbounded while staying under the payload cap — and a
// legitimate multi-megabyte line keeps its documented frame-cap headroom.
// Returns the tripped budget's name so the overflow error is typed correctly,
// or null.
#overCapReason() {
if (this.dataBytes > this.maxFrameBytes) return 'frame';
if (this.queuedBytes > this.maxQueuedBytes) return 'queue';
if (this.partialBytes === 0) return null;
const info = this.#partialInfo();
if (info === undefined) return null;
if (info === null || info.comment) {
return this.partialBytes > this.maxFrameBytes ? 'frame' : null;
}
const valueBytes = Math.max(0, this.partialBytes - info.prefixBytes);
if (info.field === 'data') {
return this.dataBytes + valueBytes > this.maxFrameBytes ? 'frame' : null;
}
return valueBytes > this.maxFrameBytes ? 'frame' : null;
}
#completeLine() {
const line = this.parts.length === 1 ? this.parts[0] : this.parts.join('');
// Release exactly the per-segment overhead charged while the line was
// accumulating; the completed line's own slot is charged again by #line.
this.queuedBytes = Math.max(0, this.queuedBytes - this.parts.length * QUEUED_LINE_OVERHEAD_BYTES);
this.parts = [];
this.partialBytes = 0;
this.lineHead = '';
this.partialClass = undefined;
this.#line(line);
}
#overflow(reason = 'frame') {
if (this.failed) return;
this.failed = true;
this.parts = [];
this.partialBytes = 0;
this.lineHead = '';
this.partialClass = undefined;
this.pendingCr = false;
this.event = null;
this.data = [];
this.payloadBytes = 0;
this.dataBytes = 0;
this.eventBytes = 0;
// Release the queued bytes with the buffers: a failed stream must not keep
// its flood reachable from the framer while the caller tears it down.
this.queuedBytes = 0;
this.eventCharge = 0;
const queue = reason === 'queue';
const error = Object.assign(
new Error(queue ? 'SSE queued bytes exceed cap' : 'SSE frame exceeds cap'),
{ code: queue ? 'SSE_QUEUE_OVERFLOW' : 'SSE_FRAME_TOO_LARGE' },
);
this.onOverflow?.(error);
// A queue overflow is a transport failure, not a partial frame: emit the
// typed upstream error every translator already maps to its protocol error
// event. Without this the consumer saw `failed` with no frame and finished
// the turn with zero content and a clean `stop`.
if (queue) {
this.onFrame?.({
event: null,
data: JSON.stringify({
error: {
message: 'upstream SSE exceeded the queued-byte budget',
type: 'api_error',
code: 'SSE_QUEUE_OVERFLOW',
},
}),
});
}
}
#line(line) {
if (line === '') {
this.#dispatch();
return;
}
if (line.charCodeAt(0) === 58) {
if (Buffer.byteLength(line, 'utf8') > this.maxFrameBytes) {
this.#overflow('frame');
return;
}
this.onComment?.(line);
return;
}
const parsed = parseSseLine(line);
if (!parsed) {
if (Buffer.byteLength(line, 'utf8') > this.maxFrameBytes) this.#overflow('frame');
return;
}
if (parsed.field === 'event') {
const bytes = Buffer.byteLength(parsed.value, 'utf8');
this.queuedBytes = Math.max(0, this.queuedBytes - this.eventCharge);
// One retained slot for the event string; its value bytes are already on
// the frame budget via eventBytes.
this.eventCharge = QUEUED_LINE_OVERHEAD_BYTES;
this.queuedBytes += this.eventCharge;
this.eventBytes = bytes;
this.payloadBytes = this.dataBytes + this.eventBytes;
this.event = parsed.value;
if (this.eventBytes > this.maxFrameBytes) { this.#overflow('frame'); return; }
if (this.queuedBytes > this.maxQueuedBytes) { this.#overflow('queue'); return; }
return;
}
if (parsed.field === 'data') {
const bytes = Buffer.byteLength(parsed.value, 'utf8');
const separator = this.data.length ? 1 : 0;
this.dataBytes += separator + bytes;
this.payloadBytes = this.dataBytes + this.eventBytes;
this.data.push(parsed.value);
// One retained slot in this.data per line; the value bytes themselves
// are already bounded by dataBytes on the frame budget.
this.queuedBytes += QUEUED_LINE_OVERHEAD_BYTES;
if (this.dataBytes > this.maxFrameBytes) { this.#overflow('frame'); return; }
if (this.queuedBytes > this.maxQueuedBytes) { this.#overflow('queue'); return; }
return;
}
// id/retry are not part of the emitted frame; bound the value so a partial
// field cannot retain unbounded wire data.
if (Buffer.byteLength(parsed.value, 'utf8') > this.maxFrameBytes) this.#overflow('frame');
}
#dispatch() {
if (this.failed) return;
if (this.event !== null || this.data.length) {
if (this.payloadBytes > this.maxFrameBytes) {
this.#overflow('frame');
return;
}
const data = this.data.join('\n');
if (isJsonObjectPayload(data)) {
// JSON-object frames are validated so an unrepairable payload becomes a
// typed error instead of a silent drop. A single document is forwarded
// byte-identically; concatenated documents are delivered one per frame.
const parsed = parseSseDataJson(data);
if (parsed.malformed) {
this.onFrame?.(malformedDataFrame(this.event));
} else if (parsed.docs && parsed.docs.length > 1) {
for (const doc of parsed.docs) {
this.onFrame?.({ event: this.event, data: JSON.stringify(doc) });
}
} else {
this.onFrame?.({ event: this.event, data });
}
} else {
// `[DONE]`, comments and plain-text data are not JSON frames.
this.onFrame?.({ event: this.event, data });
}
}
this.event = null;
this.data = [];
this.payloadBytes = 0;
this.dataBytes = 0;
this.eventBytes = 0;
this.queuedBytes = 0;
this.eventCharge = 0;
}
/**
* Cancel the stream and release everything it retains. Callers that tear a
* translation down (client gone, route abort) must not leave a flood's bytes
* reachable from the framer until GC.
*/
dispose() {
this.ended = true;
this.parts = [];
this.partialBytes = 0;
this.lineHead = '';
this.partialClass = undefined;
this.pendingCr = false;
this.event = null;
this.data = [];
this.payloadBytes = 0;
this.dataBytes = 0;
this.eventBytes = 0;
this.queuedBytes = 0;
this.eventCharge = 0;
}
}