TypeScriptでSSEの再接続を実装する：Last-Event-IDでイベント欠損を埋める

リアルタイム文字起こしやLLMのストリーミングでは、接続が切れること自体は珍しくありません。問題は、再接続できたかではなく、その間に届かなかったイベントをどう埋めるかです。

SSEでは各イベントに単調増加する id を付けると、ブラウザは次回接続で Last-Event-ID を送ります。これをイベントログの再生位置として扱えば、切断中の差分を復旧できます。

この記事では、Node.js標準ライブラリだけで次を実装します。

• id 付きSSEフレーム
• Last-Event-ID 以降の再生
• 再生とライブ配信の間に穴を作らない購読順
• ハートビートと切断時の後始末

再接続だけではイベントが消える

EventSource は接続が落ちると自動で再接続します。ただし、サーバーが「今この瞬間から」だけを配信する実装なら、切断していた数秒間のイベントは失われます。

SSEの id: フィールドを送った場合、ブラウザは最後に受け取った値を保持し、再接続リクエストに Last-Event-ID ヘッダーとして載せます。仕様上の詳細は HTML Living Standard の Server-Sent Events にあります。

送るフレームは小さくて構いません。

id: 43
event: transcript
data: {"text":"SSEの復旧を説明します"}

ここで重要なのは、時刻ではなく順序です。複数のイベントが同じミリ秒に発生し得るので、イベントIDはストリームごとに単調増加する整数にします。

まず購読し、その後に再生する

一見すると「過去ログを送り、次に購読する」が自然に見えます。しかし、その順序には穴があります。

1. サーバーがID 43まで再生する
2. ID 44が発生する
3. まだ購読していないのでID 44を見逃す
4. ライブ配信を開始する

解決策は逆です。先に購読し、同じ lastSent を使って過去ログを流します。購読直後にライブイベントが来ても、lastSent 以下を捨てれば重複もしません。

最小実装

下の EventLog は説明用にメモリへ直近1,000件を持ちます。本番ではRedis StreamsやDBの連番付きテーブルなど、全プロセスから読めるログへ置き換えてください。

import { createServer } from "node:http";
import { EventEmitter } from "node:events";

type Entry = {
  id: number;
  event: string;
  data: unknown;
};

class EventLog extends EventEmitter {
  #nextId = 1;
  #entries: Entry[] = [];

  publish(event: string, data: unknown) {
    const entry = { id: this.#nextId++, event, data };
    this.#entries.push(entry);
    this.#entries = this.#entries.slice(-1000);
    this.emit("entry", entry);
    return entry;
  }

  after(id: number) {
    return this.#entries.filter((entry) => entry.id > id);
  }
}

const log = new EventLog();

const format = (entry: Entry) =>
  `id: ${entry.id}\nevent: ${entry.event}\ndata: ${JSON.stringify(entry.data)}\n\n`;

createServer((req, res) => {
  if (req.url !== "/events") {
    res.writeHead(404).end();
    return;
  }

  const parsed = Number(req.headers["last-event-id"] ?? 0);
  let lastSent =
    Number.isSafeInteger(parsed) && parsed >= 0 ? parsed : 0;

  res.writeHead(200, {
    "content-type": "text/event-stream; charset=utf-8",
    "cache-control": "no-cache, no-transform",
    connection: "keep-alive",
    "x-accel-buffering": "no",
  });
  res.write("retry: 2000\n: connected\n\n");

  const send = (entry: Entry) => {
    if (entry.id <= lastSent) return;
    lastSent = entry.id;
    res.write(format(entry));
  };

  // 先に購読する。以降は同じlastSentで重複を除外する。
  const onEntry = (entry: Entry) => send(entry);
  log.on("entry", onEntry);
  for (const entry of log.after(lastSent)) send(entry);

  const heartbeat = setInterval(
    () => res.write(": heartbeat\n\n"),
    15_000,
  );

  req.on("close", () => {
    clearInterval(heartbeat);
    log.off("entry", onEntry);
  });
}).listen(3000);

setInterval(() => {
  log.publish("transcript", { text: new Date().toISOString() });
}, 1_000);

data: はJSON文字列にしています。改行を含む文字列をそのまま複数行で出すとSSEフレームが壊れやすいためです。イベント名も固定の文字列にしておくと、クライアント側で用途ごとに処理を分けられます。

クライアント側はイベント名で受ける

ブラウザ側は接続の再作成を普段は意識しません。サーバーが id を返し続けていれば、EventSource が再接続時に位置情報を渡します。

const source = new EventSource("/events");

source.addEventListener("transcript", (event) => {
  const { text } = JSON.parse((event as MessageEvent).data);
  console.log(text);
});

source.onerror = () => {
  // EventSourceが再接続を試みるので、ここで即closeしない。
  console.warn("SSE connection interrupted");
};

手動で再生範囲を確かめるなら、サーバーを起動して別の端末から次のように叩けます。

curl -N -H 'Last-Event-ID: 42' http://localhost:3000/events

最初の出力がID 43以降になり、その後は新規イベントが続けば意図どおりです。

本番で決めておく境界

この最小実装だけで「再接続可能」とは言い切れません。少なくとも次の境界は設計時に決めます。

• 保持期限を越えた場合：クライアントのIDが最古の保持IDより古ければ、差分では復旧できません。event: reset を返し、HTTP APIでスナップショットを取り直す経路を用意します。
• 複数プロセスの場合：メモリ上の EventLog はプロセス内だけです。ロードバランサー配下では、イベントIDとログをRedisやDBへ共有します。
• 認可とチャンネル分離：接続を長く保持する前に認可し、ユーザーや会話IDごとに読めるイベントを絞ります。全イベントを接続後にフィルタする設計は避けます。
• プロキシのタイムアウト：CDNやリバースプロキシが無通信接続を閉じることがあります。コメント形式のハートビートは、その検知と維持のためです。
• バックプレッシャー：遅いクライアントに無制限で書き込まないことです。res.write() が false を返したら drain を待つ、または切断してスナップショット復旧へ切り替える判断が必要です。

設計面接なら、「SSEはサーバーからクライアントへの一方向通知に向く。差分復旧には順序付きID、保持期限を越えたらスナップショット」という順で説明すると、再接続だけを話すより障害時の設計まで伝わります。