Saavo

背景工作

了解 Reaction 非同步命令、非同步記錄與流量統計三條佇列的職責、重試方式及排查入口。

Saavo 使用 Cloudflare Queues，處理不適合阻塞目前請求的工作。專案內建三條用途明確的佇列，分別處理業務事件產生的非同步 Command、非同步記錄，以及第一方流量統計事件。

Reaction 不是一條「萬用非同步佇列」。Event 代表已發生的業務事實，Command 代表系統接下來要執行的動作。每個 Command 自行決定採用同步還是非同步執行。

警告

不要將自訂事件放入這些預先定義的佇列，它們各有自己的功能與處理邏輯。如果業務需要非同步執行，可以使用 Reaction 的 Event / Command 機制，或另建一條非同步業務處理佇列。

內建佇列

佇列Binding預設批次與並行數用途
async-policy-taskASYNC_POLICY_TASK_QUEUE每批 1 筆，並行數 1執行 Reaction 非同步 Command
async-loggerASYNC_LOGGER_QUEUE每批最多 50 筆，並行數 1將記錄寫入 KV
analytics-eventsANALYTICS_QUEUE每批最多 8 筆，並行數 1將第一方統計事件寫入 ANALYTICS_DB

三個取用者的 max_retries 目前都是 3，平台設定允許首次投遞後，最多再投遞 3 次，也就是最多嘗試取用 4 次。但取用者可以提前確認訊息並停止重試。Reaction 取用者依最多 4 次投遞處理，統計與記錄取用者則會在第 3 次處理失敗時，主動確認訊息並記錄錯誤或警示。

wrangler.jsonc 同時設定了佇列名稱、Binding 與 *_QUEUE_NAME。入口會依實際佇列名稱精確選擇取用者。名稱不一致時，訊息不會自動改由其他佇列處理。

Reaction 命令如何執行

業務程式碼透過 reaction.processor.emit() 發出 Event。處理器會先將 Event 與 Command 執行紀錄寫入儲存體，再依序執行：

  • mode: 'sync'：在目前請求、Webhook 或排程入口中，立即嘗試執行。
  • mode: 'async'：將 Event 執行 ID 傳送至 async-policy-task，由取用者繼續執行。

目前同步 Command 主要用於授予或撤回角色與使用權益。非同步 Command 則包括寄送郵件、傳送通知、建立站內通知、同步付款客戶的電子郵件地址、處理到期權益週期，以及清理過期資料。

同步只表示「目前呼叫流程會執行它」，不代表失敗一定會變成例外。Command 回傳的一般失敗會寫入執行紀錄，emit() 仍可能回傳 accepted。如果呼叫端必須確認某項 Command 已成功，除了檢查 Event 是否已接收，也要檢查回傳的 Command 狀態，或讀取後台執行紀錄。

管理後台提供兩個排查入口：

/dashboard/reaction/events
/dashboard/reaction/commands

重試、去除重複與失敗處理

Event 使用穩定的 idempotencyKey 去除重複。再次發出相同 Event 類型與等冪鍵時，處理器會回傳 duplicate，不會建立第二套執行紀錄。

每個 Command 也有自己的 maxAttempts，預設通常為 3。它控制 Command 處理函式的業務重試，Cloudflare 的 max_retries 則控制佇列訊息重新投遞。這是不同層級的重試，排查時不要混淆。

佇列採用至少一次投遞，訊息可能重複送達。Reaction 透過已儲存的狀態、穩定的執行 ID 與租用機制，減少重複執行。但若涉及寄信、呼叫第三方 API 等外部副作用，Command 仍應使用穩定的業務鍵實作等冪性。

目前 async-policy-task 沒有設定死信佇列。訊息在最後一次投遞仍失敗時，取用者會確認訊息，並發出高優先順序警示，之後需要人工檢查執行紀錄。如果外部副作用已完成，結果狀態卻無法寫入儲存體，取用者也會停止重新投遞並發出警示，避免重複執行該副作用。

成功加入佇列，不代表工作已完成

emit() 回傳 accepted，表示 Event 與 Command 已被接受並開始分派，非同步 Command 此時可能仍在排隊。回傳 abandoned 則表示多次嘗試發出 Event 後，仍未可靠地完成接收，呼叫端不能將它視為成功。

在業務程式碼中觸發事件

業務功能應使用 Reaction，不要直接向 ASYNC_POLICY_TASK_QUEUE 傳送自訂 JSON：

import { reaction } from '@/core/reaction';

const result = await reaction.processor.emit(
    workerCtx,
    SomeEvent.create({
        idempotencyKey: 'some-event:stable-business-id',
        payload: { /* 業務資料 */ },
    }),
);

emit() 可能回傳下列結果：

結果意義
accepted新 Event 已寫入儲存體，並開始執行或加入佇列
duplicate相同的等冪事件已存在，回傳原本的執行紀錄
ignoredEvent 沒有產生任何 Command
abandoned發出流程最終失敗，已寫入記錄與警示

呼叫端是否需要檢查這些結果，取決於業務以什麼條件判定完成。註冊授權、付款權益等重要流程，不能只確認函式已回傳，也需要確認對應同步 Command 的狀態。單純通知類的非同步工作，則可以在可靠地加入佇列後，結束目前請求。

HTTP 路由使用 resolveFetchWorkerCtx(c) 取得執行脈絡。Cron 與佇列入口，則使用各自執行環境已建立的 WorkerCtx。應用程式碼不要直接呼叫 createWorkerCtx。

選擇適合的處理方式

  • 業務事實與後續動作：定義 Reaction Event / Command。
  • 頁面造訪與產品事件追蹤：使用流量統計 API，由 analytics-events 處理。
  • 應用程式記錄：使用既有 logger，由 async-logger 處理。
  • 與既有業務無關的長時間運算或通用工作：個別評估執行平台，不要直接放入 async-policy-task。

config/deploy.ts 中的 logger.enableAsyncLoggerQueue 預設啟用。關閉後，記錄不再進入非同步記錄佇列，取用者收到遺留訊息時，會確認並捨棄。完全停用流量統計後，統計取用者也會確認並捨棄已排隊的事件。切換這些開關前，應先評估是否允許捨棄累積的訊息。

沒有量測資料時，不要任意提高批次大小或並行數。async-policy-task 維持每批一筆、並行數一，是為了依執行紀錄順序與租用機制，安全地推進業務 Command。

上線檢查

  • 三條佇列的名稱、Binding 與 *_QUEUE_NAME 完全一致。
  • 產生者與取用者已部署至同一套環境設定。
  • 註冊、測試購買與訂閱結束後，Reaction 執行紀錄符合預期。
  • 新 Event 使用穩定的業務等冪鍵。
  • 外部副作用可以安全重試，不依賴「佇列只會投遞一次」的假設。
  • 重要呼叫端會辨識 abandoned，並依業務要求檢查同步 Command 狀態。
  • 關閉記錄或統計開關前，已確認可以捨棄佇列中的遺留訊息。
  • 記錄與警示管道可以收到最終取用失敗、結果儲存失敗的通知。

常見問題

接下來