inngest-durable-functions

Inngest 持久化函式

掌握 Inngest 的持久化執行模型，建構容錯、長時間執行的工作流程。本技能涵蓋從觸發器到錯誤處理的完整生命週期。

這些技能專注於 TypeScript。 若使用 Python 或 Go，請參閱 Inngest 文件 以取得語言特定指引。核心概念適用於所有語言。

你需要了解的核心概念

持久化執行模型

• 每個步驟應封裝副作用與非確定性程式碼
• 記憶化防止已完成的步驟重複執行
• 狀態持久化在基礎設施故障時存活
• 自動重試搭配可設定的重試次數

步驟執行流程

// ❌ 錯誤：非確定性邏輯放在步驟外
async ({ event, step }) => {
  const timestamp = Date.now(); // 這會執行多次！

  const result = await step.run("process-data", () => {
    return processData(event.data);
  });
};

// ✅ 正確：所有非確定性邏輯放在步驟內
async ({ event, step }) => {
  const result = await step.run("process-with-timestamp", () => {
    const timestamp = Date.now(); // 只執行一次
    return processData(event.data, timestamp);
  });
};

函式限制

每個 Inngest 函式都有以下硬性限制：

• 每次函式執行最多 1,000 個步驟
• 每個步驟回傳資料最多 4MB
• 函式執行狀態（包含事件資料、步驟輸出與函式輸出）總和最多 32MB
• 每個步驟 = 一次獨立的 HTTP 請求（約 50-100ms 開銷）

若達到這些限制，請將函式拆分為較小的函式，並透過 step.invoke() 或 step.sendEvent() 連接。

何時使用步驟

一律包在 step.run() 中：

• API 呼叫與網路請求
• 資料庫讀寫
• 檔案 I/O 操作
• 任何非確定性操作
• 任何你希望在失敗時獨立重試的操作

永遠不要包在 step.run() 中：

• 純計算與資料轉換
• 簡單的驗證邏輯
• 無副作用的確定性操作
• 日誌記錄（放在步驟外）

建立函式

基本函式結構

const processOrder = inngest.createFunction(
  {
    id: "process-order", // 唯一識別碼，建立後請勿變更
    triggers: [{ event: "order/created" }],
    retries: 4, // 預設：每個步驟重試 4 次
    concurrency: 10 // 最大並行執行數
  },
  async ({ event, step }) => {
    // 你的持久化工作流程
  }
);

步驟 ID 與記憶化

// 步驟 ID 可以重複使用 - Inngest 會自動處理計數器
const data = await step.run("fetch-data", () => fetchUserData());
const more = await step.run("fetch-data", () => fetchOrderData()); // 不同的執行

// 使用描述性 ID 以提升可讀性
await step.run("validate-payment", () => validatePayment(event.data.paymentId));
await step.run("charge-customer", () => chargeCustomer(event.data));
await step.run("send-confirmation", () => sendEmail(event.data.email));

觸發器與事件

事件觸發器

觸發器定義在 createFunction 第一個參數的 triggers 陣列中：

// 單一事件觸發器
inngest.createFunction(
  { id: "my-fn", triggers: [{ event: "user/signup" }] },
  async ({ event }) => { /* ... */ }
);

// 事件加上條件過濾
inngest.createFunction(
  { id: "my-fn", triggers: [{ event: "user/action", if: 'event.data.action == "purchase" && event.data.amount > 100' }] },
  async ({ event }) => { /* ... */ }
);

// 多個觸發器（最多 10 個）
inngest.createFunction(
  {
    id: "my-fn",
    triggers: [
      { event: "user/signup" },
      { event: "user/login", if: 'event.data.firstLogin == true' },
      { cron: "0 9 * * *" } // 每天上午 9 點
    ]
  },
  async ({ event }) => { /* ... */ }
);

Cron 觸發器

// 基本 Cron
inngest.createFunction(
  { id: "my-fn", triggers: [{ cron: "0 */6 * * *" }] }, // 每 6 小時
  async ({ step }) => { /* ... */ }
);

// 指定時區
inngest.createFunction(
  { id: "my-fn", triggers: [{ cron: "TZ=Europe/Paris 0 12 * * 5" }] }, // 巴黎時間每週五中午 12 點
  async ({ step }) => { /* ... */ }
);

// 與事件結合
inngest.createFunction(
  {
    id: "my-fn",
    triggers: [
      { event: "manual/report.requested" },
      { cron: "0 0 * * 0" } // 每週日
    ]
  },
  async ({ event, step }) => { /* ... */ }
);

函式呼叫

// 以步驟形式呼叫另一個函式
const result = await step.invoke("generate-report", {
  function: generateReportFunction,
  data: { userId: event.data.userId }
});

// 使用回傳的資料
await step.run("process-report", () => {
  return processReport(result);
});

冪等性策略

事件層級冪等性（生產者端）

// 使用自訂 ID 防止重複事件
await inngest.send({
  id: `checkout-completed-${cartId}`, // 24 小時去重
  name: "cart/checkout.completed",
  data: { cartId, email: "user@example.com" }
});

函式層級冪等性（消費者端）

const sendEmail = inngest.createFunction(
  {
    id: "send-checkout-email",
    triggers: [{ event: "cart/checkout.completed" }],
    // 每個 cartId 每 24 小時只執行一次
    idempotency: "event.data.cartId"
  },
  async ({ event, step }) => {
    // 此函式不會對同一個 cartId 執行兩次
  }
);

// 複雜的冪等性鍵
const processUserAction = inngest.createFunction(
  {
    id: "process-user-action",
    triggers: [{ event: "user/action.performed" }],
    // 每個使用者 + 組織組合唯一
    idempotency: 'event.data.userId + "-" + event.data.organizationId'
  },
  async ({ event, step }) => {
    /* ... */
  }
);

取消模式

基於事件的取消

在表達式中，event 代表原始觸發事件，async 代表新匹配的事件。詳見表達式語法參考。

const processOrder = inngest.createFunction(
  {
    id: "process-order",
    triggers: [{ event: "order/created" }],
    cancelOn: [
      {
        event: "order/cancelled",
        if: "event.data.orderId == async.data.orderId"
      }
    ]
  },
  async ({ event, step }) => {
    await step.sleepUntil("wait-for-payment", event.data.paymentDue);
    // 若收到 order/cancelled 事件，將被取消
    await step.run("charge-payment", () => processPayment(event.data));
  }
);

超時取消

const processWithTimeout = inngest.createFunction(
  {
    id: "process-with-timeout",
    triggers: [{ event: "long/process.requested" }],
    timeouts: {
      start: "5m", // 若 5 分鐘內未開始則取消
      finish: "30m" // 若 30 分鐘內未完成則取消
    }
  },
  async ({ event, step }) => {
    /* ... */
  }
);

處理取消清理

// 監聽取消事件
const cleanupCancelled = inngest.createFunction(
  { id: "cleanup-cancelled-process", triggers: [{ event: "inngest/function.cancelled" }] },
  async ({ event, step }) => {
    if (event.data.function_id === "process-order") {
      await step.run("cleanup-resources", () => {
        return cleanupOrderResources(event.data.run_id);
      });
    }
  }
);

錯誤處理與重試

預設重試行為

• 每個步驟 總共 5 次嘗試（1 次初始 + 4 次重試）
• 指數退避加上抖動
• 每個步驟有獨立的重試計數器

自訂重試設定

const reliableFunction = inngest.createFunction(
  {
    id: "reliable-function",
    triggers: [{ event: "critical/task" }],
    retries: 10 // 每個步驟最多重試 10 次
  },
  async ({ event, step, attempt }) => {
    // `attempt` 是函式層級的嘗試計數器（從 0 開始）
    // 它追蹤目前執行步驟的重試次數，而非整個函式
    if (attempt > 5) {
      // 目前步驟後續嘗試的不同邏輯
    }
  }
);

不可重試錯誤

防止對重試也無法成功的程式碼進行重試。

import { NonRetriableError } from "inngest";

const processUser = inngest.createFunction(
  { id: "process-user", triggers: [{ event: "user/process.requested" }] },
  async ({ event, step }) => {
    const user = await step.run("fetch-user", async () => {
      const user = await db.users.findOne(event.data.userId);

      if (!user) {
        // 不要重試 - 使用者不存在
        throw new NonRetriableError("User not found, stopping execution");
      }

      return user;
    });

    // 繼續處理...
  }
);

自訂重試時機

import { RetryAfterError } from "inngest";

const respectRateLimit = inngest.createFunction(
  { id: "api-call", triggers: [{ event: "api/call.requested" }] },
  async ({ event, step }) => {
    await step.run("call-api", async () => {
      const response = await externalAPI.call(event.data);

      if (response.status === 429) {
        // 根據 API 指定的時間重試
        const retryAfter = response.headers["retry-after"];
        throw new RetryAfterError("Rate limited", `${retryAfter}s`);
      }

      return response.data;
    });
  }
);

日誌記錄最佳實踐

正確的日誌設定

import winston from "winston";

// 設定日誌記錄器
const logger = winston.createLogger({
  level: "info",
  format: winston.format.json(),
  transports: [new winston.transports.Console()]
});

const inngest = new Inngest({
  id: "my-app",
  logger // 將日誌記錄器傳入客戶端
});

// 或使用內建的 ConsoleLogger 進行簡單的日誌層級控制
import { ConsoleLogger, Inngest } from "inngest";

const inngest = new Inngest({
  id: "my-app",
  logger: new ConsoleLogger({ level: "debug" }) // "debug" | "info" | "warn" | "error"
});

⚠️ v4 重大變更： logLevel 選項已移除。請改用 logger 選項搭配 ConsoleLogger 或自訂日誌記錄器。

函式日誌記錄模式

const processData = inngest.createFunction(
  { id: "process-data", triggers: [{ event: "data/process.requested" }] },
  async ({ event, step, logger }) => {
    // ✅ 正確：在步驟內記錄日誌以避免重複
    const result = await step.run("fetch-data", async () => {
      logger.info("Fetching data for user", { userId: event.data.userId });
      return await fetchUserData(event.data.userId);
    });

    // ❌ 避免：在步驟外記錄日誌可能導致重複
    // logger.info("Processing complete"); // 這可能執行多次！

    await step.run("log-completion", async () => {
      logger.info("Processing complete", { resultCount: result.length });
    });
  }
);

效能最佳化

檢查點

檢查點在 v4 中預設啟用。它允許函式在執行期間定期持久化狀態，減少步驟之間的延遲。

// 檢查點在 v4 中預設啟用
// 為無伺服器平台設定 maxRuntime（設為平台超時時間的 60-80%）
const realTimeFunction = inngest.createFunction(
  {
    id: "real-time-function",
    triggers: [{ event: "realtime/process" }],
    checkpointing: {
      maxRuntime: "50s", // 用於超時 60 秒的無伺服器環境
    }
  },
  async ({ event, step }) => {
    // 步驟立即執行，並定期進行檢查點
    const result1 = await step.run("step-1", () => process1(event.data));
    const result2 = await step.run("step-2", () => process2(result1));
    return { result2 };
  }
);

// 如有需要可停用檢查點
const legacyFunction = inngest.createFunction(
  {
    id: "legacy-function",
    triggers: [{ event: "legacy/process" }],
    checkpointing: false
  },
  async ({ event, step }) => { /* ... */ }
);

進階模式

條件式步驟執行

const conditionalProcess = inngest.createFunction(
  { id: "conditional-process", triggers: [{ event: "process/conditional" }] },
  async ({ event, step }) => {
    const userData = await step.run("fetch-user", () => {
      return getUserData(event.data.userId);
    });

    // 條件式步驟執行
    if (userData.isPremium) {
      await step.run("premium-processing", () => {
        return processPremiumFeatures(userData);
      });
    }

    // 總是執行
    await step.run("standard-processing", () => {
      return processStandardFeatures(userData);
    });
  }
);

錯誤復原模式

const robustProcess = inngest.createFunction(
  { id: "robust-process", triggers: [{ event: "process/robust" }] },
  async ({ event, step }) => {
    let primaryResult;

    try {
      primaryResult = await step.run("primary-service", () => {
        return callPrimaryService(event.data);
      });
    } catch (error) {
      // 備援至次要服務
      primaryResult = await step.run("fallback-service", () => {
        return callSecondaryService(event.data);
      });
    }

    return { result: primaryResult };
  }
);

應避免的常見錯誤

1. ❌ 非確定性程式碼放在步驟外
2. ❌ 資料庫呼叫放在步驟外
3. ❌ 日誌記錄放在步驟外（導致重複）
4. ❌ 部署後變更步驟 ID
5. ❌ 未處理 NonRetriableError 情況
6. ❌ 對關鍵函式忽略冪等性

下一步

• 參閱 inngest-steps 以取得詳細的步驟方法參考
• 參閱 references/step-execution.md 以取得詳細的步驟模式
• 參閱 references/error-handling.md 以取得全面的錯誤處理策略
• 參閱 references/observability.md 以取得監控與追蹤設定
• 參閱 references/checkpointing.md 以取得效能最佳化細節

---

本技能涵蓋 Inngest 的持久化函式模式。關於事件發送與 Webhook 處理，請參閱 inngest-events 技能。