串流可靠性：速率限制（rate‑limits）、回壓（backpressure）與事件監控

1. 為什麼串流對負載特別敏感

在前面的部分，我們已經解析了 MCP‑事件、job.progress/job.completed 狀態、async job 與 GiftGenius 的串流通道（SSE/HTTP-stream）如何運作。現在要看看在真實負載下，這套架構會發生什麼。

當只有一位使用者偶爾啟動一次禮物挑選時，一切看起來都很美好。但只要 GiftGenius 上線到生產環境，並同時湧入數百個「幫全體員工挑禮物」的請求，你就會突然發現：

• 伺服器上有數百個長壽的 SSE 連線；
• worker 動不動就把 job.progress 往外送；
• 日誌每天以 GB 為單位在增長；
• 使用者端的 UI 開始卡頓，雖然「伺服器看起來沒掛」。

傳統的 HTTP 請求存活的是毫秒或秒。SSE 或 HTTP‑stream 的串流可以活上數分鐘甚至數小時。它會佔住連線、記憶體、檔案描述符。每一次送出的 event 都意味著 JSON 序列化、跨網路複製、以及 GC 的工作。如果你把這一切當成「嗯，後端再多打一個 console.log 而已」，系統很快就會變成一台電暖器。

MCP‑events 還有另一個特性：它們常常會為同一個任務被多次產生。每 0.1% 就更新一次進度的 worker，會在單一 job 上產生驚人的事件數。結果你會得到「噪音」：大量細碎的訊息，它們：

• 拖累網路與 CPU；
• 塞滿佇列與緩衝；
• 讓除錯與日誌分析變得痛苦。

因此，對於串流與 MCP‑events，必須像看待資料庫查詢或模型呼叫一樣嚴肅：這是昂貴的資源，需要定額、控管與監控。

為了應對上述問題，記住三個大主題：

1. 速率限制（Rate‑limits）——限制我們能生成與發送事件/串流的數量與頻率。
2. 回壓（Backpressure）——當消費者跟不上生產者時的因應策略。
3. 監控與指標——量測實況，及早發現開始「沸騰」的跡象。

2. 串流與事件的速率限制

先從最直觀的限制說起。

要理解的是，在串流情境中，「危險的違規者」往往不是用戶，而是伺服器。在一般 REST‑API 中，你會限制到伺服器的請求數，避免使用者把你 DDoS 掉。在 MCP 與串流的世界裡，很容易反向 DDoS：worker 或 MCP 伺服器每秒狂轟客戶端成千上萬個事件。

需要哪些限制

一般從三個面向來考量。

首先，針對使用者或工作階段的限制。不該允許單一使用者開啟二十個 GiftGenius 的大型精靈視窗，每個都各自維持一條 SSE 串流。合理的做法是：每個 session 僅允許少量的活躍串流，並限制單一使用者或租戶在 running 狀態的 job 數量。

其次，針對單一 job 的限制。這裡我們關注事件頻率。通常僅需每隔 N 毫秒發送一次 job.progress，或只在可見的變化時（例如每 5% 進度）才發送。沒有必要為目錄裡每個處理過的商品都發一則訊息。同時也應限制 payload 的大小：進度事件不應攜帶數 MB 的文字。

最後，針對 IP 或組織的限制。這屬於防濫用：有人跑腳本瘋狂丟任務，或你的 App 突然爆紅。這時就該由熟悉的 API gateway 與 proxy 機制出場了。

簡單的事件頻率限制實作

來看一個 GiftGenius 的 worker，它在背景依據一長串收件人名單挑選禮物，並定期透過 MCP‑notification event/progress 傳送進度。我們希望事件發送頻率不超過每 500 毫秒一次，且只有在百分比至少變動 5 個百分點時才發送。

給 worker 的 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;

  // 僅在已過 >= 500 ms 或者增幅 >= 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 可以檢查該使用者目前開啟的連線數是否超過門檻（例如不超過三條並行串流）。若超過，就回應 429，並在 GPT 指令中告訴模型：遇到這種情況不要再開新的長任務，而是提示使用者「已有進行中的挑選，先等它完成」。

在小型系統中，這類檢查可以在程序的記憶體中實作。在更嚴謹的基礎設施裡，則會放到 MCP‑gateway 或獨立的 rate‑limit 服務中。

3. Backpressure：當消費者跟不上時該怎麼辦

速率限制約束的是我們想產生多少事件。但即便小心翼翼地限制，仍可能出現消費者「喝不過來」的情況：使用者用的是不穩定的行動網路、瀏覽器分頁卡住、ChatGPT 當下負載很高。

Backpressure 就是系統對「消費者跟不上」的反應。與其無限累積資料最終因 OOM 掛掉，我們有意識地：

• 放慢腳步；
• 聚合事件；
• 丟棄較不重要的事件。

壓力會在哪裡出現

GiftGenius 的典型場景可能是這樣：worker 把事件寫進佇列（例如 Redis Streams 或資料庫中的一張表），MCP 伺服器讀取後推送到 SSE 通道。如果客戶端很慢（3G、舊筆電、很多其他分頁），TCP 緩衝會開始填滿，Node 程序來不及把佇列完全清空，最終在記憶體中堆積事件。接著你會看到熟悉的：

FATAL ERROR: Ineffective mark-compacts near heap limit

你已經擁有網路層（TCP）的 backpressure，但它並不了解你的網域實體。它只是在說：「喂，慢一點，緩衝滿了。」我們的任務是把這件事用 MCP 事件的角度來解讀。

有限緩衝與事件丟棄

對於進度與狀態，我們有個有利條件：不是所有事件都同等重要。使用者在乎的是最後的最新百分比，而不是所有中間歷史「51%、52%、53%、54%」。也就是說，我們可以放心丟掉部分事件，只送出最後一筆。

假設我們有一層，從 worker 收到進度事件後，為每個 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 ms，看一下緩衝並只發送最後一筆，忽略其他：

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 策略的例子：把多個更新合併為一個最新的。對進度而言是黃金模式。

對於「log」或 partial_result 類型的事件，策略可能不同。那裡通常不允許丟失事件：日誌文字很重要，而遺失的 JSON 片段可能破壞資料結構。這些情況你可以：

• 聚合訊息（把多行日誌黏成一個封包）；
• 或向 worker 發出控制訊號，請它「放慢日誌產生速度」。

在非同步系統中第二種做法較難，但至少要把它納入考量。

限制佇列深度

Backpressure 不只是在發送前的事件緩衝而已。你需要審視系統中的所有佇列：

• 等待 worker 的任務佇列；
• worker 與 MCP 伺服器之間的事件佇列；
• 伺服器端串流函式庫內部的緩衝。

對每個佇列都要設定合理的最大深度。如果佇列滿了，你要不是回覆客戶端「系統過載，稍後再試」，要不就是丟棄較不重要的 job，要不就把部分情境轉為「離線模式」（例如生成報告，稍後傳送連結）。

另一個有趣手法是事件類型的優先級。當過載時，你可以只發送 job.completed 與 job.failed，把 job.progress 降權，甚至暫時停用。

4. 串流與事件的監控

沒有量測，所有關於速率限制與 backpressure 的努力都只是巫術。你需要看見串流數量異常增多、事件出現延遲、客戶端成群斷線。

串流與一般 HTTP 請求的行為不同：它們的存活時間以分鐘、甚至小時計。傳統的「每秒請求數」與「平均延遲」無法給出完整圖像。

關鍵指標

對 SSE 或 HTTP/stream 的串流，幾個指標特別有用。

1. 連線指標。 目前有多少條活躍的 SSE 串流？平均一條連線存活多久？多少比例的串流以錯誤或逾時結束？活躍連線數的劇烈增加，可能表示流量風暴或資源外洩（客戶端不關連線）。劇烈下降則可能是大規模斷線（例如網路問題或伺服器上的關鍵臭蟲）。
2. 事件指標。 你每秒送出多少事件（EPS——events per second，每秒事件數）？事件的平均大小是多少？你觀察到多少序列化或 payload 驗證錯誤？如果事件大小突然增加——可能有人在 job.progress 裡塞了整份報告的長文，而不是一段短字串。
3. job 指標。 狀態分佈（pending、running、completed、failed、canceled）、依任務類型的平均耗時、進入重試或 dead‑letter 的比例。這能幫你看出問題不只是網路層，還可能在 worker：外部 API 變慢、或出現大量錯誤。
4. backpressure 與系統層指標。 串流系統常觀察元件間緩衝與佇列的深度，以及串流等待消費者釋放空間的時間比例。如果佇列幾乎總是滿到頂，這是系統處於極限的明確訊號。也要注意系統層指標：執行串流的伺服器 CPU 與記憶體，以及網路層的錯誤/逾時。有時 MCP 伺服器與 ChatGPT 之間的網路吞吐才是瓶頸。

綜合這四組指標，你能回答三個問題：現在有多少條串流在活著？你在傳多少資料？job 的狀況如何、以及系統從哪裡開始「噎住」。

該記錄哪些日誌

日誌是可觀測性的第二支柱。要以能還原單一 job 歷史的方式來記錄事件與連線。

通常會為每個事件與串流記錄：

• jobId 與/或 eventId；
• userId 與 sessionId（若是多租戶）；
• 事件類型（progress、completed、failed、resource.updated）；
• 通道類型（SSE 或 HTTP/stream）；
• 發送的 timestamp，以及可能的話，worker 產生事件的 timestamp。

這樣就能算出 lag：worker 產生事件的時間與它送入 socket 的時間差。這個 lag 時間的上升，是 backpressure 問題的良好指標。

要留意別讓日誌本身成為過載來源。對於像 job.progress 這樣的高頻事件，不一定要記錄每一則；可以啟用抽樣（sampling）——只記錄每第 N 則——或聚合統計。

程式上可以是一個簡單的 helper：

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

在真實專案裡你會包裝成結構化日誌的函式庫，但理念一樣：每條紀錄都包含盡可能多的有用上下文。

5. 警報與降級策略

當你已經有了指標與日誌，下一步就是設定警報與設計「系統不舒服時要怎麼優雅變慢」。理念是：誠實地變慢，比突然整個倒下要好。

警報範例

對 GiftGenius 而言，有幾種常見情境值得監控。

首先是異常的活躍串流數量。如果平時只有幾十條活躍 SSE 連線，卻突然變成上千，值得查明發生了什麼。也許你爆紅了，也許是臭蟲導致連線不關閉。

其次，job 實際完成與客戶端收到 job.completed 之間的延遲。如果這個延遲超過門檻（比如 5–10 秒），代表在 worker 與客戶端之間，事件正在累積或連線在打滑。

再者，job.failed 或 job.canceled 的比例相對成功的過高。原因可能在 worker（外部 API 掛了、新臭蟲），也可能是使用者對延遲更敏感（更常取消任務）。

最後，連線錯誤與串流中斷的提升：若非正常的 disconnect 增加，可能是網路或客戶端問題，值得考慮 fallback 情境。

降級模式

當系統過載，可以啟用「節省資源模式」。這總比到處回 500 要好。

最常見的模式是自適應事件頻率。如果你看到 event‑rate（每秒事件數）暴增到平常的十倍，佇列的延遲開始拉長，就降低進度事件的頻率。原本每 1% 一次——改成每 10%。原本每 500 ms 一次——改成每 2–3 秒一次。使用者不需要超高精度的進度，但完全卡死的 UI 卻很糟。

對於較不重要的事件——例如在背景更新商品 feed 的 resource.updated——可以在系統過載時暫停發送。

另一個手法是把部分情境從串流切換到週期性輪詢。如果 SSE 通道崩潰，MCP 伺服器可以送一個系統事件給元件，例如 system.overloaded，而元件則切換到「每隔 N 秒透過 REST endpoint 詢問 job 狀態」的策略。

6. GiftGenius 的小型實作片段

把所有內容串起來，假設我們已經有：

• MCP‑tool startGiftSearch，會建立 job 並回傳 jobId；
• worker 負責搜尋並發送 event/progress 與 event/completed；
• 元件會連到的 SSE endpoint：/api/events/[userId]（Next.js）。

加上簡單的「事件風暴」保護與最小化的監控。

以步進與時間限制進度

在 worker 端加入節流與合併（conflation），如同上面所說。現在事件發送頻率不超過每半秒一次，且變動至少 5% 才發。

記錄活躍串流

在 SSE endpoint 為每位使用者維持一個計數器：

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 fallback，能讓你的 GiftGenius 不再只是「展示用的 demo」，而是能撐過真實流量風暴。在接下來關於 gateway、生產架構與完整可觀測性的模組，這些模式還會派上用場。

7. 處理串流、速率限制與監控的常見錯誤

錯誤 1：沒有對串流數量與事件頻率設置限制。
開發者為了「好看」加了 SSE，worker 也老實地在每個處理過的物件上回報進度，demo 時一切正常。但第一次遇到真實用戶高峰時，伺服器把大部分資源都耗在序列化與傳送成千上萬個細小事件上，而 ChatGPT 的 UI 變成投影片。

錯誤 2：嘗試「無限緩衝全部東西」。
程式裡出現一個無上限的「尚未送出的事件」陣列，直到客戶端恢復為止一路長大。劇透：它不會等到恢復，伺服器先死。任何緩衝都必須有硬上限，溢出的處理邏輯也要明確。

錯誤 3：把所有事件類型一視同仁。
進度可以聚合與丟棄（最後的百分比比歷史變化更重要）。日誌與 partial 結果則不能——丟掉一個片段可能意味著資料損壞。設計系統時，請先按重要性對事件分群，並為每一群設計過載時的策略。

錯誤 4：缺乏可觀測性。
沒有活躍串流的指標、沒有 event‑rate 的統計，日誌只有一句「出了點狀況」。這種情況下，你只能從使用者回饋與 CPU 負載圖得知災情。至少配置基本的指標，以及按 jobId 與 eventId 的日誌，這不是奢侈，是必需。

錯誤 5：忽視降級的僵硬 UX。
元件與 GPT 指令假設串流永遠可用、進度「即時」更新、partial 結果嚴格按劇本來。一遇到網路問題，使用者只看到「卡住」的進度條，毫無解釋。更好的做法是預留誠實的 fallback：「即時更新暫時有問題，我會持續挑選並在完成時通知你」，同時改成較低頻更新或輪詢。

錯誤 6：相信「我們的使用者不會同時建立很多任務」。
實務證明，如果你不限制並行 job 與串流的數量，總會有人開五個分頁、在每個分頁都把挑選推到「最大化」，然後去喝咖啡。「應該不會吧」在生產環境幾乎總是以警報聲中學會監控而告終。