ストリームの信頼性: レート制限、バックプレッシャー、イベント監視

1. なぜストリームは負荷に特に敏感なのか

これまでの回で、GiftGenius における MCP イベント、job.progress/job.completed ステータス、async‑job、ストリーミングチャネル（SSE/HTTP-stream）の仕組みを見てきました。次は、このアーキテクチャが実際の負荷下でどう振る舞うかを確認することが重要です。

ユーザーが1人で、たまにギフト検索を走らせる程度ならすべて順調に見えます。しかし GiftGenius が本番に出て、同時に「社員全員分のギフトを選ぶ」リクエストが何百件も来た瞬間、次のような事態に気づくことになります。

• サーバー上に何百もの長寿命な SSE 接続が存在する;
• ワーカーが job.progress をやたらと高頻度で送る;
• ログが1日にギガバイト単位で肥大化する;
• 「サーバーは落ちていないはず」なのに、ユーザーの UI がカクつき始める。

古典的な HTTP リクエストは数ミリ秒から数秒で完了します。SSE や HTTP‑stream のストリームは数分、時には数時間生き続けます。接続、メモリ、ファイルディスクリプタを保持し続けます。送られる各イベントは JSON のシリアライズ、ネットワークコピー、GC の仕事を伴います。これを「バックエンドの console.log を1つ増やしただけ」と見なしていると、システムはあっという間にヒーターと化します。

MCP イベントにはもう1つの特徴があります。1つのタスクに対して何度も繰り返し生成されがちだという点です。例えば、進捗を 0.1% ごとに更新するワーカーは、1つのジョブ当たりのイベント数を爆発的に増やします。結果として「ノイズ」、すなわち次のような大量の小さなメッセージが生まれます。

• ネットワークと CPU を圧迫する;
• キューやバッファを埋め尽くす;
• デバッグやログ分析を苦痛にする。

したがって、ストリームや MCP イベントは、データベースへのクエリやモデル呼び出しと同じくらい慎重に扱うべき対象です。高コストな資源であり、規制・制御・監視が必要です。

これに対処するために、次の3つの大きなテーマを念頭に置きます。

1. レート制限 — どれだけ頻繁に、どれだけ多くのイベント/ストリームを生成・送信してよいかを制限します。
2. バックプレッシャー — コンシューマがプロデューサに追いつけない状況に対応します。
3. 監視とメトリクス — 何が起きているかを測定し、沸騰の兆候を早期に検知します。

2. ストリームとイベントのレート制限

まずは分かりやすいレート制限から始めましょう。

ストリーミングのシナリオでは、「危険な違反者」はしばしばクライアントではなくサーバーである、という点を理解することが重要です。通常の REST API では、ユーザーが DDoS 化しないようにサーバーへのリクエスト数を制限します。しかし MCP とストリームの世界では、ワーカーや MCP サーバーがクライアントへ毎秒何千件ものイベントを爆撃してしまう「逆 DDoS」を容易に引き起こせます。

必要な制限の種類

一般的に、次の3つの観点で考えます。

第一に、ユーザーまたはセッション単位の制限。1人のユーザーが GiftGenius のウィザードを20個同時に開き、それぞれが SSE ストリームを持つ、といったことは許容できません。妥当なのは、1セッション当たりのアクティブなストリーム数を数本に制限し、ユーザーまたはテナントごとに running 状態のジョブ数を制限することです。

第二に、1つのジョブ単位の制限。ここで重要なのはイベントの頻度です。job.progress は N ミリ秒に1回以下、あるいは目に見える変化があった場合のみ、例えば 5% ごとに送るだけで十分です。カタログ内の各商品を処理するたびにメッセージを送る必要はありません。ペイロードサイズの制限も有効です。進捗イベントがメガバイト級のテキストを含むべきではありません。

第三に、IP や組織単位の制限。これは濫用対策です。誰かがタスクをスパムするスクリプトを走らせたり、あなたの App が予想外に人気になったりする場合です。ここでは API ゲートウェイやプロキシの仕組みが活躍します。

イベント頻度の簡易実装

長い受取人リストに対してバックグラウンドでギフト選定を行い、定期的に MCP 通知 event/progress で進捗を送る GiftGenius のワーカーを考えます。イベントは 500 ミリ秒に1回以下、かつパーセンテージが少なくとも 5 ポイント変化した時だけ送信したいとします。

ワーカーのための仮の TS 風擬似コード:

// 仮に mcpClient.sendNotification(...) があるとする
let lastSentPercent = 0;
let lastSentAt = 0;

function reportProgress(jobId: string, percent: number, message: string) {
  const now = Date.now();
  const percentDelta = percent - lastSentPercent;
  const timeDelta = now - lastSentAt;

  // 500ms 経過した場合 または 5%以上増えた場合にのみ送信する
  if (percentDelta >= 5 || timeDelta >= 500) {
    mcpClient.sendNotification("event/progress", {
      jobId,
      percent,
      message,
    });
    lastSentPercent = percent;
    lastSentAt = now;
  }
}

このアプローチはスロットリング（throttling）と呼ばれます。時間と値の変化でイベントの流量を間引きます。

処理を段階に分けている場合（「ステップ1/3」「ステップ2/3」など）は、さらに単純で、段階が変わった時だけイベントを送れば十分です。

同時に開けるストリーム数の上限

MCP サーバー側には、おそらく SSE の HTTP ハンドラがあります。

// app/api/events/[userId]/route.ts (Next.js 16 App Router)
export async function GET(
  req: Request,
  { params }: { params: { userId: string } },
) {
  const userId = params.userId;

  if (!canOpenMoreStreams(userId)) {
    return new Response("Too many streams", { status: 429 });
  }

  const stream = new ReadableStream({
    start(controller) {
      registerSseClient(userId, controller);
    },
    cancel() {
      unregisterSseClient(userId);
    },
  });

  return new Response(stream, {
    headers: { "Content-Type": "text/event-stream" },
  });
}

canOpenMoreStreams 関数は、そのユーザーの現在の接続数を確認し、しきい値（例えば同時ストリーム3本まで）と比較できます。上限を超えたら 429 を返し、GPT の指示では「すでに進行中のウィザードがあるため、新しい長時間のウィザードを開始せず、完了まで待つように促す」といった振る舞いをモデルにさせます。

小規模なシステムでは、この種のチェックはプロセス内メモリで実装できます。より本格的なインフラでは、MCP ゲートウェイや専用のレート制限サービスに委譲します。

3. バックプレッシャー: コンシューマが追いつけない時はどうするか

レート制限は、私たちがどれだけイベントを「作りたいか」を縛ります。しかし、慎重な制限をかけても、コンシューマが処理に追いつけないことは十分にあり得ます。ユーザーのモバイル回線が遅い、ブラウザのタブが固まっている、ChatGPT が高負荷、など。

バックプレッシャーとは、コンシューマが追いつけない事実に対するシステムの反応です。データを無限に貯め込み、いつか OOM で落ちるのではなく、意図的に次を行います。

• スピードを落とす;
• イベントを集約する;
• 重要度の低いものを捨てる。

どこで圧力が生まれるか

GiftGenius の典型的なシナリオは次のようなものです。ワーカーがイベントをキュー（たとえば Redis Streams や DB テーブル）に書き込み、MCP サーバーがそれを読み出して SSE チャネルにプッシュします。もしクライアントが遅い（3G、古いノートPC、他のタブが大量に開いている）と、TCP バッファが埋まり始め、Node プロセスはキューを捌き切れず、最終的にメモリ内にイベントをため込みます。すると見慣れた次のような状況に出会います。

FATAL ERROR: Ineffective mark-compacts near heap limit

ネットワーク（TCP）レベルのバックプレッシャーは既にありますが、それはあなたのドメインの概念を知りません。ただ「バッファがいっぱいだから落ち着け」と伝えるだけです。私たちの仕事は、これを MCP イベントのレベルで解釈することです。

上限付きバッファリングとイベントのドロップ

進捗やステータスには都合のよい特性があります。すべてのイベントが同じ価値を持つわけではない、ということです。ユーザーに重要なのは最新の進捗率であって、「51%、52%、53%、54%」といった中間履歴ではありません。つまり、イベントの一部を思い切って捨て、最後のものだけを送ってよいのです。

ワーカーからの進捗イベントを受け取り、jobId ごとのバッファに入れる層があるとします。

type ProgressEvent = { jobId: string; percent: number; message: string };

const progressBuffers = new Map<string, ProgressEvent[]>();
const MAX_BUFFER = 10;

function bufferProgress(event: ProgressEvent) {
  const buffer = progressBuffers.get(event.jobId) ?? [];
  buffer.push(event);

  // バッファサイズを制限する
  if (buffer.length > MAX_BUFFER) {
    // 直近のいくつかのイベントだけ残す
    progressBuffers.set(event.jobId, buffer.slice(-MAX_BUFFER));
  } else {
    progressBuffers.set(event.jobId, buffer);
  }
}

別のタイマーが例えば 500 ミリ秒ごとにバッファを見て、最後のイベントだけを送り、残りは無視します。

setInterval(() => {
  for (const [jobId, buffer] of progressBuffers.entries()) {
    if (!buffer.length) continue;

    const last = buffer[buffer.length - 1];
    sendProgressToClient(last); // SSE/MCP notification

    progressBuffers.set(jobId, []); // クリア
  }
}, 500);

これはコンフレート（conflation）の一例です。複数の更新を直近の1つにまとめます。進捗では黄金パターンです。

「ログ」や partial_result のようなイベントでは戦略が異なるかもしれません。そこではイベント損失が許容されないことが多いからです。ログテキストは重要で、失われた JSON チャンクはデータ構造を壊す可能性があります。こういった場合は次のようにします。

• メッセージを集約する（複数行のログを1つのパケットにまとめる）;
• または、ワーカーに「ログ生成を落として」といった制御シグナルを送る。

非同期システムでは後者は難易度が上がりますが、少なくとも検討はすべきです。

キューの深さ制限

バックプレッシャーは、送信直前のイベントバッファにとどまりません。システム内のすべてのキューに目を配る必要があります。

• ワーカー待ちのタスクキュー;
• ワーカーと MCP サーバーの間のイベントキュー;
• サーバー側のストリーミングライブラリ内部のバッファ。

各キューに対して妥当な深さの上限を設定することが重要です。キューがあふれたら、クライアントに「混雑のため後でお試しください」と返すか、重要度の低いジョブを捨てるか、一部のシナリオを「オフライン」へ切り替えます（例: レポートを生成して後でリンクを送る）。

興味深い手法としてイベントタイプの優先度付けがあります。過負荷時には job.completed と job.failed だけを送り、job.progress は優先度を下げるか、いっそ止めてしまうこともできます。

4. ストリームとイベントの監視

測定がなければ、レート制限やバックプレッシャーの工夫もただのまじないです。ストリームが不自然に増えていないか、イベントが遅延していないか、クライアントが大量に切断されていないかを可視化する必要があります。

ストリームは通常の HTTP リクエストと振る舞いが異なります。存続時間が分単位・時間単位に及ぶため、「毎秒のリクエスト数」や「平均レイテンシ」といった古典的なメトリクスだけでは十分な絵が描けません。

主要なメトリクス

SSE や HTTP/stream のストリームでは、いくつかの指標群を追うと有用です。

1. 接続メトリクス。 現在アクティブな SSE ストリームは何本か？1つの接続は平均どれくらい生きるか？エラーやタイムアウトで終わるストリームの割合は？アクティブ接続が急増したら、トラフィックの嵐やリソースリーク（クライアントが接続を閉じない）の兆候です。急減は一斉切断（ネットワーク障害やサーバー側の致命的なバグ）を示唆します。
2. イベントメトリクス。 すべてのストリームで1秒あたりいくつのイベントを送っているか（EPS — events per second）？イベントの平均サイズは？ペイロードのデシリアライズやバリデーションのエラーはどれくらいか？イベントサイズの増加が見えたなら、誰かが job.progress に短いテキストの代わりにレポート全文を載せ始めたのかもしれません。
3. ジョブのメトリクス。 ステータスの分布（pending、running、completed、failed、canceled）、タスク種別ごとの平均実行時間、再試行やデッドレター行きの割合。これにより、ネットワーク層だけでなくワーカー側の問題（外部 API の遅延、頻発するエラー）も把握できます。
4. バックプレッシャーのメトリクスとシステム指標。 ストリーミングシステムでは、コンポーネント間のバッファやキューの深さ、そして「コンシューマの空きを待ってストリームがブロックされていた時間の割合」を見るのが一般的です。キューが常に満杯に近いなら、システムが限界にある明確なシグナルです。また、ストリーミングを担うサーバーの CPU・メモリ、ネットワーク層のエラーやタイムアウトなど、システム指標も重要です。時に MCP サーバーと ChatGPT の間の帯域がボトルネックになります。

これら4群を総合すると、「今何本のストリームが生きているか」「どれだけのデータを流しているか」「ジョブはどう振る舞っているか」「どこで詰まり始めているか」の3つの問いに答えが出ます。

ログに何を書くか

ログは可観測性の第二の柱です。後から特定のジョブの履歴を再構成できるよう、イベントと接続をロギングすることが大切です。

通常、各イベントやストリームのログには次を含めます。

• jobId および/または eventId;
• userId と sessionId（マルチテナントの場合）;
• イベントタイプ（progress、completed、failed、resource.updated）;
• チャネルタイプ（SSE または HTTP/stream）;
• 送信時刻、可能ならワーカー側でのイベント生成時刻。

こうして、ワーカーがイベントを生成した時刻とソケットへ出て行った時刻の差、すなわち遅延（lag）を算出できます。この lag の増加はバックプレッシャー問題の優れた指標です。

ログ自体が過負荷の原因にならないよう注意が必要です。job.progress のような高頻度イベントを逐一ログするのは得策でないことが多く、サンプリング（N件に1件だけログする）や統計の集計に切り替えるのが良いでしょう。

コードでは簡単なヘルパとして次のように書けます。

function logEvent(event: {
  type: string;
  jobId: string;
  userId?: string;
  channel: "sse" | "http-stream";
  payload: unknown;
}) {
  console.info({
    ...event,
    timestamp: new Date().toISOString(),
  });
}

実プロジェクトでは structured logging のライブラリで包みますが、発想は同じです。各レコードに最大限有用なコンテキストを載せます。

5. アラートとデグレードの方針

メトリクスとログがそろったら、次はアラート設定と「システムが苦しい時にどうデグレードするか」を考えます。突然落ちるより、正直に性能を落として動き続ける方が良い、という考え方です。

アラートの例

GiftGenius では次のような典型的状況を監視するのが妥当です。

第一に、アクティブなストリーム数の異常。通常は数十本の SSE 接続なのに、突然数千本になったら要調査です。人気化かもしれませんし、接続が閉じられないバグかもしれません。

第二に、ジョブの実完了とクライアントの job.completed 受信の間の遅延。この遅延が閾値（例えば 5〜10 秒）を超え始めたら、ワーカーからクライアントまでのどこかでイベントが滞留しているか、接続に問題がある兆候です。

第三に、job.failed や job.canceled の割合の高さ。原因はワーカー側（外部 API の崩れ、新しいバグ）かもしれませんし、ユーザーが遅延に敏感になり（ジョブのキャンセルが増える）、結果として失敗が膨らむこともあります。

最後に、接続エラーやストリーム切断の増加。異常な切断が増えているなら、ネットワークやクライアント側の問題が疑われ、フォールバックシナリオの検討が必要です。

デグレード・パターン

システムが過負荷なら「省資源モード」をオンにします。何にでも 500 を返すよりはるかに良い対応です。

最も一般的なのは適応的なイベント頻度です。event‑rate（1秒あたりのイベント数）が通常の10倍に跳ね上がり、キューの遅延が増え始めたら、進捗イベントの頻度を下げます。これまで 1% ごとだったのを 10% ごとに、500 ミリ秒ごとだったのを 2〜3 秒に。ユーザーは超精密な進捗なしでも困りませんが、UI が完全に固まるのは困ります。

重要度の低いイベント、例えば商品フィードのバックグラウンド更新での resource.updated は、負荷中は一時的に送信停止するのも手です。

もう1つは、ストリームから定期ポーリングへの切り替えです。SSE チャネルが崩れた場合、MCP サーバーはウィジェットに system.overloaded のようなシステムイベントを送り、ウィジェットは「N 秒ごとに REST エンドポイントでジョブのステータスを問い合わせる」戦略に切り替えます。

6. GiftGenius のための小さな実用断片

ここまでをつなげるため、すでに次があるとします。

• MCP ツール startGiftSearch（ジョブを作成して jobId を返す）;
• 検索を実行して event/progress と event/completed を送るワーカー;
• ウィジェット（Next.js）から接続される SSE エンドポイント /api/events/[userId]。

ここに「イベントストーム」対策の薄い層と、最小限の監視を加えます。

進捗のステップ/時間による制限

ワーカーでは、前述の通りスロットリングとコンフレートを加えます。これで、イベントは0.5秒に1回以下かつ 5% 以上の変化時のみ送信されます。

アクティブなストリームの計上

SSE エンドポイントではユーザーごとのカウンタを保持します。

const activeStreams = new Map<string, number>();
const STREAM_LIMIT = 3;

function canOpenMoreStreams(userId: string) {
  const current = activeStreams.get(userId) ?? 0;
  return current < STREAM_LIMIT;
}

function registerSseClient(userId: string, controller: ReadableStreamDefaultController) {
  const current = activeStreams.get(userId) ?? 0;
  activeStreams.set(userId, current + 1);

  // ここで controller をどこかの構造に保存して、
  // 後でこのストリームにイベントを書き込めるようにする
}

function unregisterSseClient(userId: string) {
  const current = activeStreams.get(userId) ?? 1;
  activeStreams.set(userId, Math.max(0, current - 1));
}

サーバーは activeStreams.size のメトリクスを Prometheus/Grafana などに送ることもできます。

最小の event‑rate メトリクス

まずは送信イベント数をざっくり数えます。

let eventsSentLastMinute = 0;

function sendProgressToClient(ev: ProgressEvent) {
  // ... シリアライズして SSE ストリームに書き込む
  eventsSentLastMinute++;
}

setInterval(() => {
  console.info({
    metric: "events_per_minute",
    value: eventsSentLastMinute,
    timestamp: new Date().toISOString(),
  });
  eventsSentLastMinute = 0;
}, 60_000);

将来的には正規のカウンタやアラートに置き換えられますが、出発点としては十分です。

ここまでの要素（レート制限、バックプレッシャー、メトリクス/アラート、適切な UX フォールバック）を組み合わせれば、GiftGenius は「デモのためのデモ」から脱し、実際のトラフィックの嵐にも耐えられるようになります。次のモジュールでは、ゲートウェイ、本番アーキテクチャ、フルスタックな可観測性について扱い、ここでのパターンが再び役立ちます。

7. ストリーム、レート制限、監視でよくあるミス

エラー №1: ストリーム数とイベント頻度の制限がない。
開発者は「見映えのため」に SSE を追加し、ワーカーは処理したオブジェクトごとに進捗を律儀に送ります。デモでは動いていたのに、最初の実ユーザーのスパイクでサーバー資源の大半が、無数の小イベントのシリアライズと転送に費やされ、ChatGPT の UI はスライドショー化します。

エラー №2: 無制限の「全部ため込む」バッファリング。
「未送出イベント」の無制限配列が登場し、クライアントが回復するまで増え続けます。ネタバレ: クライアントは回復せず、先にサーバーが死にます。どんなバッファにも固い上限が必要で、あふれ時の処理は明示されていなければなりません。

エラー №3: すべてのイベントタイプを同一視する。
進捗は集約・ドロップできます（最後のパーセンテージが履歴より重要）。ログや partial 結果ではそれはできません。1チャンクの欠落がデータ破損を意味することがあります。システム設計時に、イベントを重要度でグルーピングし、過負荷時の戦略をタイプごとに決めておきましょう。

エラー №4: 可観測性がない。
アクティブストリームのメトリクスなし、event‑rate の把握なし、ログには「何かがうまくいかなかった」だけ。この状況では、ユーザーの苦情や CPU 負荷グラフでしか問題を知れません。少なくとも jobId と eventId に基づく基本メトリクスとログを用意するのは、贅沢ではなく必須です。

エラー №5: デグレードを考慮しない硬直した UX。
ウィジェットと GPT の指示は、ストリームが常に利用可能で、進捗が「リアルタイム」で更新され、partial 結果が筋書き通りに届く前提です。最初のネットワーク問題で、ユーザーは「止まった」プログレスバーと無説明に直面します。ずっと良いのは正直なフォールバックを UX に組み込むことです。「ライブ更新に問題がありますが、選定は続け、完了したらお知らせします」と示し、更新頻度を落とすかポーリングに切り替えましょう。

エラー №6: 「うちのユーザーは同時に大量のタスクを作らない」という楽観。
並列ジョブ数やストリーム数を制限していないと、必ず誰かがタブを5つ開き、各タブで「最大限」にギフト選定を走らせてコーヒーを飲みに行きます。「運よく大丈夫だろう」は本番ではほぼ必ず、アラートの轟音とともに監視とのご対面で終わります。