inngest-durable-functions

Inngest Durable Functions

障害耐性のある長時間実行ワークフローを構築するための、Inngest の耐久実行モデルをマスターしましょう。このスキルでは、トリガーからエラー処理までの完全なライフサイクルを扱います。

これらのスキルは TypeScript に焦点を当てています。 Python または Go については、言語固有のガイダンスとして Inngest ドキュメント を参照してください。コアコンセプトはすべての言語に共通です。

知っておくべきコアコンセプト

耐久実行モデル

• 各ステップは、副作用と非決定的なコードをカプセル化する必要があります
• メモ化により、完了したステップの再実行を防ぎます
• 状態の永続化により、インフラストラクチャの障害を乗り越えます
• 自動リトライと、設定可能なリトライ回数

ステップ実行フロー

// ❌ BAD: Non-deterministic logic outside steps
async ({ event, step }) => {
  const timestamp = Date.now(); // This runs multiple times!

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

// ✅ GOOD: All non-deterministic logic in steps
async ({ event, step }) => {
  const result = await step.run("process-with-timestamp", () => {
    const timestamp = Date.now(); // Only runs once
    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", // Unique, never change this
    triggers: [{ event: "order/created" }],
    retries: 4, // Default: 4 retries per step
    concurrency: 10 // Max concurrent executions
  },
  async ({ event, step }) => {
    // Your durable workflow
  }
);

ステップ ID とメモ化

// Step IDs can be reused - Inngest handles counters automatically
const data = await step.run("fetch-data", () => fetchUserData());
const more = await step.run("fetch-data", () => fetchOrderData()); // Different execution

// Use descriptive IDs for clarity
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 配列で定義します。

// Single event trigger
inngest.createFunction(
  { id: "my-fn", triggers: [{ event: "user/signup" }] },
  async ({ event }) => { /* ... */ }
);

// Event with conditional filter
inngest.createFunction(
  { id: "my-fn", triggers: [{ event: "user/action", if: 'event.data.action == "purchase" && event.data.amount > 100' }] },
  async ({ event }) => { /* ... */ }
);

// Multiple triggers (up to 10)
inngest.createFunction(
  {
    id: "my-fn",
    triggers: [
      { event: "user/signup" },
      { event: "user/login", if: 'event.data.firstLogin == true' },
      { cron: "0 9 * * *" } // Daily at 9 AM
    ]
  },
  async ({ event }) => { /* ... */ }
);

Cron トリガー

// Basic cron
inngest.createFunction(
  { id: "my-fn", triggers: [{ cron: "0 */6 * * *" }] }, // Every 6 hours
  async ({ step }) => { /* ... */ }
);

// With timezone
inngest.createFunction(
  { id: "my-fn", triggers: [{ cron: "TZ=Europe/Paris 0 12 * * 5" }] }, // Fridays at noon Paris time
  async ({ step }) => { /* ... */ }
);

// Combine with events
inngest.createFunction(
  {
    id: "my-fn",
    triggers: [
      { event: "manual/report.requested" },
      { cron: "0 0 * * 0" } // Weekly on Sunday
    ]
  },
  async ({ event, step }) => { /* ... */ }
);

関数の呼び出し

// Invoke another function as a step
const result = await step.invoke("generate-report", {
  function: generateReportFunction,
  data: { userId: event.data.userId }
});

// Use returned data
await step.run("process-report", () => {
  return processReport(result);
});

冪等性の戦略

イベントレベルの冪等性 (プロデューサー側)

// Prevent duplicate events with custom ID
await inngest.send({
  id: `checkout-completed-${cartId}`, // 24-hour deduplication
  name: "cart/checkout.completed",
  data: { cartId, email: "[email protected]" }
});

関数レベルの冪等性 (コンシューマー側)

const sendEmail = inngest.createFunction(
  {
    id: "send-checkout-email",
    triggers: [{ event: "cart/checkout.completed" }],
    // Only run once per cartId per 24 hours
    idempotency: "event.data.cartId"
  },
  async ({ event, step }) => {
    // This function won't run twice for same cartId
  }
);

// Complex idempotency keys
const processUserAction = inngest.createFunction(
  {
    id: "process-user-action",
    triggers: [{ event: "user/action.performed" }],
    // Unique per user + organization combination
    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);
    // Will be cancelled if order/cancelled event received
    await step.run("charge-payment", () => processPayment(event.data));
  }
);

タイムアウトによるキャンセル

const processWithTimeout = inngest.createFunction(
  {
    id: "process-with-timeout",
    triggers: [{ event: "long/process.requested" }],
    timeouts: {
      start: "5m", // Cancel if not started within 5 minutes
      finish: "30m" // Cancel if not finished within 30 minutes
    }
  },
  async ({ event, step }) => {
    /* ... */
  }
);

キャンセル時のクリーンアップ処理

// Listen for cancellation events
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 // Up to 10 retries per step
  },
  async ({ event, step, attempt }) => {
    // `attempt` is the function-level attempt counter (0-indexed)
    // It tracks retries for the currently executing step, not the overall function
    if (attempt > 5) {
      // Different logic for later attempts of the current step
    }
  }
);

リトライ不可のエラー

リトライしても成功しないコードに対して、リトライを防ぎます。

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) {
        // Don't retry - user doesn't exist
        throw new NonRetriableError("User not found, stopping execution");
      }

      return user;
    });

    // Continue processing...
  }
);

カスタムリトライのタイミング

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) {
        // Retry after specific time from API
        const retryAfter = response.headers["retry-after"];
        throw new RetryAfterError("Rate limited", `${retryAfter}s`);
      }

      return response.data;
    });
  }
);

ログ出力のベストプラクティス

適切なログ設定

import winston from "winston";

// Configure logger
const logger = winston.createLogger({
  level: "info",
  format: winston.format.json(),
  transports: [new winston.transports.Console()]
});

const inngest = new Inngest({
  id: "my-app",
  logger // Pass logger to client
});

// Or use the built-in ConsoleLogger for simple log level control
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 }) => {
    // ✅ GOOD: Log inside steps to avoid duplicates
    const result = await step.run("fetch-data", async () => {
      logger.info("Fetching data for user", { userId: event.data.userId });
      return await fetchUserData(event.data.userId);
    });

    // ❌ AVOID: Logging outside steps can duplicate
    // logger.info("Processing complete"); // This could run multiple times!

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

パフォーマンスの最適化

チェックポイント

チェックポイントは v4 ではデフォルトで有効です。これにより、関数は実行中に定期的に状態を永続化でき、ステップ間のレイテンシーを削減します。

// Checkpointing is enabled by default in v4
// Configure maxRuntime for serverless platforms (set to 60-80% of platform timeout)
const realTimeFunction = inngest.createFunction(
  {
    id: "real-time-function",
    triggers: [{ event: "realtime/process" }],
    checkpointing: {
      maxRuntime: "50s", // For serverless with 60s timeout
    }
  },
  async ({ event, step }) => {
    // Steps execute immediately with periodic checkpointing
    const result1 = await step.run("step-1", () => process1(event.data));
    const result2 = await step.run("step-2", () => process2(result1));
    return { result2 };
  }
);

// Disable checkpointing if needed
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);
    });

    // Conditional step execution
    if (userData.isPremium) {
      await step.run("premium-processing", () => {
        return processPremiumFeatures(userData);
      });
    }

    // Always runs
    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) {
      // Fallback to secondary service
      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 スキルを参照してください。