inngest-flow-control

Inngest フロー制御

Inngest のフロー制御メカニズムをマスターして、リソースを管理し、システムの過負荷を防ぎ、アプリケーションの信頼性を確保しましょう。このスキルでは、すべてのフロー制御オプションと、それぞれをいつどのように使用すべきかについての規範的なガイダンスを扱います。

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

クイック判断ガイド

• 「同時に実行する数を制限したい」 → 並行性
• 「実行を時間的に分散させたい」 → スロットリング
• 「一定期間内の N 回の実行後にブロックしたい」 → レート制限
• 「アクティビティが止まるのを待ってから 1 回だけ実行したい」 → デバウンス
• 「このキーに対しては同時に 1 回だけ実行したい」 → シングルトン
• 「イベントをグループ単位で処理したい」 → バッチ処理
• 「一部の実行はより重要である」 → 優先度

並行性

使用する場面: コンピューティングリソースを管理し、システムの過負荷を防ぐために、実行されるステップ数(関数の実行数ではない)を制限します。

重要なポイント: 並行性は、関数の実行数ではなく、アクティブなコード実行を制限します。step.sleep() や step.waitForEvent() で待機している関数は、制限にカウントされません。

基本的な並行性

inngest.createFunction(
  {
    id: "process-images",
    concurrency: 5,
    triggers: [{ event: "media/image.uploaded" }]
  },
  async ({ event, step }) => {
    // Only 5 steps can execute simultaneously
    await step.run("resize", () => resizeImage(event.data.imageUrl));
  }
);

キー付きの並行性(マルチテナント)

key パラメーターを使用して、キーの一意の値ごとに制限を適用します。

inngest.createFunction(
  {
    id: "user-sync",
    concurrency: [
      {
        key: "event.data.user_id",
        limit: 1
      }
    ],
    triggers: [{ event: "user/profile.updated" }]
  },
  async ({ event, step }) => {
    // Only 1 step per user can execute at once
    // Prevents race conditions in user-specific operations
  }
);

アカウントレベルの共有制限

inngest.createFunction(
  {
    id: "ai-summary",
    concurrency: [
      {
        scope: "account",
        key: `"openai"`,
        limit: 60
      }
    ],
    triggers: [{ event: "ai/summary.requested" }]
  },
  async ({ event, step }) => {
    // Share 60 concurrent OpenAI calls across all functions
  }
);

それぞれの使用場面:

• 基本: データベースを保護する、または一般的な容量を制限する
• キー付き: マルチテナントの公平性、「noisy neighbor」問題を防ぐ
• アカウントレベル: 複数の関数間でクォータを共有する(API 制限)

スロットリング

使用する場面: API のレート制限を回避したり、トラフィックのスパイクを平準化したりするために、時間あたりの関数開始のレートを制御します。

並行性との主な違い: スロットリングは関数の実行開始を制限し、並行性はステップの実行を制限します。

inngest.createFunction(
  {
    id: "sync-crm-data",
    throttle: {
      limit: 10, // 10 function starts
      period: "60s", // per minute
      burst: 5, // plus 5 immediate bursts
      key: "event.data.customer_id" // per customer
    },
    triggers: [{ event: "crm/contact.updated" }]
  },
  async ({ event, step }) => {
    // Respects CRM API rate limits: 10 calls/min per customer
    await step.run("sync", () => crmApi.updateContact(event.data));
  }
);

設定:

• limit: 期間ごとに開始できる関数の数
• period: 時間枠(1s 〜 7d)
• burst: 追加で即時開始できる数
• key: キーの一意の値ごとに制限を適用する

レート制限

使用する場面: 悪用を防ぐ、または過剰な重複イベントをスキップするためのハードリミットです。

スロットリングとの主な違い: レート制限はイベントを破棄し、スロットリングはイベントを遅延させます。

inngest.createFunction(
  {
    id: "webhook-processor",
    rateLimit: {
      limit: 1,
      period: "4h",
      key: "event.data.webhook_id"
    },
    triggers: [{ event: "webhook/data.received" }]
  },
  async ({ event, step }) => {
    // Process each webhook only once per 4 hours
    // Prevents duplicate webhook spam
  }
);

ユースケース:

• Webhook の重複を防ぐ
• ユーザーごとに高コストな操作を制限する
• 悪用に対する保護

デバウンス

使用する場面: 一連のイベントの到着が止まるのを待ってから、最新のものを処理します。

inngest.createFunction(
  {
    id: "save-document",
    debounce: {
      period: "5m", // Wait 5min after last edit
      key: "event.data.document_id",
      timeout: "30m" // Force save after 30min max
    },
    triggers: [{ event: "document/content.changed" }]
  },
  async ({ event, step }) => {
    // Saves document only after user stops editing
    // Uses the LAST event received
    await step.run("save", () => saveDocument(event.data));
  }
);

最適な用途:

• 急速に変化するユーザー入力(検索、ドキュメント編集)
• ノイズの多い Webhook イベント
• 最新のデータが処理されることの保証

優先度

使用する場面: 動的なデータに基づいて、一部の関数の実行を他の実行より先に処理します。

inngest.createFunction(
  {
    id: "process-order",
    priority: {
      // VIP users get priority up to 120 seconds ahead
      run: "event.data.user_tier == 'vip' ? 120 : 0"
    },
    triggers: [{ event: "order/placed" }]
  },
  async ({ event, step }) => {
    // VIP orders jump ahead in the queue
  }
);

高度な例:

inngest.createFunction(
  {
    id: "support-ticket",
    priority: {
      run: `
        event.data.severity == 'critical' ? 300 :
        event.data.severity == 'high' ? 120 :
        event.data.user_plan == 'enterprise' ? 60 : 0
      `
    },
    triggers: [{ event: "support/ticket.created" }]
  },
  async ({ event, step }) => {
    // Critical tickets get highest priority (300s ahead)
    // High severity: 120s ahead
    // Enterprise users: 60s ahead
    // Everyone else: normal priority
  }
);

シングルトン

使用する場面: 関数のインスタンスが同時に 1 つだけ実行されるようにします。

スキップモード(現在の実行を保持)

inngest.createFunction(
  {
    id: "data-backup",
    singleton: {
      key: "event.data.database_id",
      mode: "skip"
    },
    triggers: [{ event: "backup/requested" }]
  },
  async ({ event, step }) => {
    // Skip new backups if one is already running for this database
    await step.run("backup", () => performBackup(event.data.database_id));
  }
);

キャンセルモード(最新のイベントを使用)

inngest.createFunction(
  {
    id: "realtime-sync",
    singleton: {
      key: "event.data.user_id",
      mode: "cancel"
    },
    triggers: [{ event: "user/data.changed" }]
  },
  async ({ event, step }) => {
    // Cancel previous sync and start with latest data
    await step.run("sync", () => syncUserData(event.data));
  }
);

バッチ処理

使用する場面: 効率化のために複数のイベントをまとめて処理します。

inngest.createFunction(
  {
    id: "bulk-email-send",
    batchEvents: {
      maxSize: 100, // Up to 100 events
      timeout: "30s", // Or 30 seconds, whichever first
      // `key` groups events into separate batches per unique value
      // This is different from expressions `if` which filters events
      key: "event.data.campaign_id" // Batch per campaign
    },
    triggers: [{ event: "email/send.queued" }]
  },
  async ({ events, step }) => {
    // Process array of events together
    const emails = events.map((evt) => ({
      to: evt.data.email,
      subject: evt.data.subject,
      body: evt.data.body
    }));

    await step.run("send-batch", () => emailService.sendBulk(emails));
  }
);

フロー制御の組み合わせ

例: 公平な AI 処理

inngest.createFunction(
  {
    id: "ai-image-processing",
    // Global throttling for API limits
    throttle: {
      limit: 50,
      period: "60s",
      key: `"gpu-cluster"`
    },
    // Per-user concurrency for fairness
    concurrency: [
      {
        key: "event.data.user_id",
        limit: 3
      }
    ],
    // VIP users get priority
    priority: {
      run: "event.data.plan == 'pro' ? 60 : 0"
    },
    triggers: [{ event: "ai/image.generate" }]
  },
  async ({ event, step }) => {
    // Combines multiple flow controls for optimal resource usage
  }
);

プロのヒント: ほとんどの本番関数は、最適な信頼性とパフォーマンスのために、1〜3 個のフロー制御メカニズムを組み合わせることで恩恵を受けます。