Open source / Guides / SSE and AI streaming

Streaming systems guide

Build a bounded SSE client for AI streaming

Server-Sent Events looks like lines separated by blank lines. Production behavior also depends on arbitrary chunk boundaries, UTF-8 decoding, cancellation, resume IDs, retry policy, proxy buffering, and memory limits.

1. Parse a protocol, not a sequence of chunks

A network chunk is not an SSE message. One event can be split across many chunks, and one chunk can contain many events. UTF-8 characters can also span byte boundaries. Decode incrementally and dispatch only after a blank line commits an event block.

event: response.output_text.delta
id: 42
data: {"delta":"hel"}

event: response.output_text.delta
id: 43
data: {"delta":"lo"}

Multiple data: fields join with newline characters. Lines beginning with a colon are comments. An id: field updates reconnect state, including in a block that dispatches no message. A retry: field can update the server-requested delay.

import { decodeSSE } from '@stackline/sse';

const response = await fetch('/events');

for await (const event of decodeSSE(response)) {
  console.log(event.event, event.data, event.lastEventId);
}

The same decoder can consume a Response, ReadableStream, async iterable, or iterable of string and byte chunks. That makes parser tests independent of a live server.

2. Use Fetch when EventSource is too narrow

Native EventSource is excellent for a simple credential model and GET endpoint. Many AI endpoints need POST bodies, authorization headers, explicit timeouts, status inspection, and per-request cancellation. A Fetch-based async iterator fits those requirements.

import { fetchSSE } from '@stackline/sse';

const controller = new AbortController();

try {
  for await (const event of fetchSSE('/api/responses', {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${token}`,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({ input: prompt, stream: true }),
    signal: controller.signal,
    connectTimeout: 10_000,
    idleTimeout: 45_000,
    totalTimeout: 5 * 60_000,
    retry: { retries: 3, minDelay: 500, maxDelay: 10_000 }
  })) {
    if (event.data === '[DONE]') break;
    const payload = JSON.parse(event.data);
    renderProviderEvent(event.event, payload);
  }
} finally {
  controller.abort();
}

Connect, idle, and total timeouts answer different questions: did headers arrive, did a live stream stop making progress, and has the entire request exceeded its budget? Keep them separate so a long but healthy generation is not mistaken for a stalled connection.

3. Retry only requests that can be replayed

SSE defines resume through Last-Event-ID, but the application still owns replay safety. A GET subscription is usually replayable. A POST that starts billable work may need an idempotency key, a server-side session ID, or no automatic retry at all.

A production retry policy should distinguish:

• transient network failures and retryable HTTP statuses;
• permanent authentication, validation, and content-type failures;
• HTTP 204, which asks an SSE client to stop reconnecting;
• Retry-After and valid SSE retry: hints;
• a committed resume ID versus an event that was only partially received.

const options = {
  retry: {
    retries: 5,
    minDelay: 500,
    maxDelay: 30_000,
    factor: 2,
    jitter: 'full'
  },
  onRetry({ delay, reconnects, error }) {
    metrics.recordReconnect({ delay, reconnects, code: error.code });
  }
};

Never reuse an already-consumed streaming request body. Recreate it with a bodyFactory only when the operation is designed to be replayed.

4. Serve events with backpressure and proxy-safe headers

import { eventStreamResponse } from '@stackline/sse';

async function* tokens(signal) {
  yield { event: 'ready', data: 'connected', id: '1' };

  for await (const token of modelStream({ signal })) {
    yield { event: 'delta', data: JSON.stringify({ token }) };
  }

  yield { event: 'done', data: '{}' };
}

export function POST(request) {
  return eventStreamResponse(tokens(request.signal));
}

A streaming response should send Content-Type: text/event-stream, disable cache transformation, and prevent reverse-proxy buffering. Heartbeat comments can keep quiet connections visible to intermediaries, but they do not replace application-level idle policy.

Push APIs must expose backpressure. When a channel reports that a write cannot be accepted, wait until it becomes ready instead of accumulating an unbounded application queue.

import { createSSEChannel } from '@stackline/sse';

const channel = createSSEChannel({ heartbeatInterval: 15_000 });

if (!channel.sendJSON({ progress: 25 }, { event: 'progress' })) {
  await channel.ready;
}

channel.close();

5. Bound malformed and oversized streams

An SSE parser waits for delimiters, which makes an unterminated line or event a natural memory-exhaustion input. Set limits for line length, accumulated event size, queued callbacks, and work admitted per feed operation.

6. Test fragmentation, cancellation, and cleanup

A useful test matrix feeds the same document as one chunk, one byte at a time, awkward UTF-8 splits, mixed line endings, and many events per chunk. It also verifies malformed lines, size failures, ID-only blocks, cancellation during connect and read, retry exhaustion, and iterator early return.

import assert from 'node:assert/strict';
import { decodeSSE } from '@stackline/sse';

async function* oneByteAtATime(text) {
  const bytes = new TextEncoder().encode(text);
  for (const byte of bytes) yield Uint8Array.of(byte);
}

const events = [];
for await (const event of decodeSSE(
  oneByteAtATime('id: 7\ndata: hello\n\n')
)) {
  events.push(event);
}

assert.equal(events[0].data, 'hello');
assert.equal(events[0].lastEventId, '7');

Benchmark parser throughput separately from Fetch, retry, and server behavior. They are different contracts and should not be collapsed into one marketing number.