bullmq

BullMQ

BullMQ のキュー、ワーカー、フロー、再試行／スケジューリング、NestJS BullMQ、PostgreSQL と Redis のバックエンド、または Pro と OSS の選択に関わる作業では、このスキルを使用してください。

スナップショット：[email protected]（2026-10-01）。インストールされているメジャー／マイナーバージョンが異なる場合は、source-map.md から最新情報を取得してください。

ワークフロー

1. コードを変更する前に、ローカルの BullMQ の構成を確認してください。
   • bullmq、オプションのピア依存関係（ioredis、redis、pg、bullmq-otel）、存在する場合は @nestjs/bullmq、Node/Bun、TypeScript のパッケージバージョン。
   • バックエンド：デフォルトの Redis（connection オプションまたはクライアント）と、createPostgresBackend／setDefaultBackendFactory のどちらを使用しているか。
   • 接続形式：オプションオブジェクト、共有クライアント、アダプター（createNodeRedisClient、createBunRedisClient、createValkeyGlideClient）のいずれか。Redis の prefix と Postgres の schema のどちらを使用しているか。
   • プロデューサー（Queue、FlowProducer）とコンシューマー（Worker）、プロセスの分離方法（インラインと、サンドボックス化されたファイル／ワーカースレッドのどちらか）。
   • デフォルトのジョブオプション、スケジューラー、レート制限、重複排除の使用状況、存在する場合は Nest モジュールの登録。
2. 最新 API、バージョンの差異、Pro 機能、アダプター、Postgres、Nest ラッパーが関係するタスクでは、最新の公式ドキュメントを確認してください。source-map.md から始めてください。
3. 作業内容に応じて、適切なリファレンスを参照してください。
   • セットアップ、Redis／Postgres バックエンド、Queue、Worker、QueueEvents：setup-connections.md。
   • ジョブのライフサイクルと JobsOptions：jobs-options.md。
   • 親子フロー／FlowProducer：flows.md。
   • 再試行、遅延、ジョブスケジューラー、レート制限、重複排除：retries-scheduling-limits.md。
   • 運用、v5→v6 のアップグレード、テスト、NestJS、Pro と OSS の比較、複数言語の利用：production-ops-integrations.md。
4. ユーザーから移行を明示的に求められない限り、リポジトリ既存の接続ファクトリー、Nest モジュール、デプロイ構成を維持してください。
5. まず最小限で適切な境界（プロセッサーの単体テスト）で動作を検証し、次にキュー／ワーカーの統合を検証してください。

基本的な判断

• データストアは共有インフラストラクチャとして扱ってください。接続オプションやファクトリーは意図的に再利用してください。ブロッキング待機のため、Worker と QueueEvents は引き続き接続を複製するため、Redis クライアント／アダプターは duplicate() をサポートしている必要があります。
• Redis のデフォルトパスには、オプションのピア依存関係 ioredis が必要です。bullmq に ioredis が引き続きバンドルされているとは限りません。ioredis 6.0（デフォルトで RESP3、Node 20 以降）は >=5.0.0 のピア依存関係の範囲を満たしますが、BullMQ は引き続き ioredis 5.11 に対してテストされており、ioredis 6 の RedisOptions を ConnectionOptions に割り当てられないという未解決の問題が報告されています。プロジェクトですでに ioredis 6 を使用しており、型チェックと統合テストが通る場合を除き、bun add bullmq ioredis@5 を推奨します。
• Redis 以外のバックエンド（Postgres またはカスタム）の場合は、ファクトリーを最後に渡してバックエンド型と接続型を TypeScript に推論させるか、ジョブ型の型引数も必要な場合は withBackend(createPostgresBackend)（6.3.9 以降）で一度バインドしてください。Queue<Data, Result, string, PostgresQueueBackend> のような不完全なジェネリクスは絶対に指定しないでください。Queue の第 4 スロットはバックエンドではありません。
• 高レベルクラスはデータストアに依存しません。削除された Queue#client、Queue#redisVersion、Queue#databaseType、Worker#blockingClient、FlowProducer#client は使用しないでください。回避策として Redis にアクセスする必要がある場合に限り、queue.getBackend().client（RedisQueueBackend として型付けされています）を使用してください。
• waitUntilReady() は void に解決されます。戻り値を Redis クライアントとして扱わないでください。await worker.resume() を使用してください。
• Queue、Worker、QueueEvents、FlowProducer に error リスナーを設定してください。処理されない接続エラーがあると Node がクラッシュします。
• プロセッサーは冪等になるよう設計してください。ジョブはリトライされたり、ロック喪失後に停止したり、複数回実行されたりする可能性があります。
• attempts と backoff は意図的に設定してください。致命的または恒久的な失敗には UnrecoverableError をスローしてください（Job#discard() は削除されています）。文字列ではなく Error オブジェクトをスローしてください。
• 親子グラフには FlowProducer を推奨します。Flow の親には deduplication を設定しないでください。Flow のジョブオプションでは repeat は省略します。opts.jobId がないノードには UUID が割り当てられます。子の失敗時オプション（failParentOnFailure、ignoreDependencyOnFailure、removeDependencyOnFailure、continueParentOnFailure）は意図的に選択してください。
• 定期実行の処理には Job Schedulers のみを使用します：upsertJobScheduler / getJobSchedulers / removeJobScheduler。BullMQ 6.x では Queue.add(..., { repeat })、Repeat、getRepeatableJobs、removeRepeatable、removeRepeatableByKey、debounce、Job#debounceId、debounced イベント、QueueScheduler、v3 より前の groupKey によるレート制限を復活させないでください。グループレート制限は Pro 限定です。
• 重複排除には安定した deduplication.id が必要です。シンプルな方式、スロットリング（ttl）、デバウンス（extend/replace + delay）、keepLastIfActive から選択してください。
• レート制限は、すべての Worker にわたってキュー単位でグローバルに適用されます。queue.rateLimit を推奨します（Worker の #rateLimit は非推奨です）。任意のキュー単位の setGlobalConcurrency / setGlobalRateLimit で全 Worker に上限を設けられます。Worker ごとの concurrency/limiter をこれらの上限より大きくすることはできません。
• 優先度：優先度が設定されたジョブでは、数値が小さいほど優先されます（1..2097151）。未設定／0 は優先度なしで、優先度が設定されたジョブより先に処理されます。一時停止中のキューもジョブを waiting 状態で保持します。getJobCounts() / JobType において、paused はジョブの状態ではありません。
• removeOnComplete / removeOnFail（保持期間または保持件数のポリシー）を調整してください。完了／失敗したジョブを無制限に保持すると、際限なく増え続けます。ioredis の keyPrefix は絶対に使用しないでください。BullMQ の prefix を使用してください（Cluster では波括弧のハッシュタグを使用します）。Postgres では Redis のプレフィックスではなく schema を使用します。
• シャットダウン時は Worker を適切に終了してください（await worker.close()）。運用上その時間帯を想定していない限り、ロック中にプロセスを強制終了すると、停止や重複処理が発生します。
• レガシーな @nestjs/bull より @nestjs/bullmq を推奨します。プロジェクトですでに Pro を使用している場合を除き、BullMQ Pro 限定の API（groups、グループレート制限、worker の job.getBatch()、observable）は推奨しないでください。
• プロセッサーは簡潔に保ち、イベントループをブロックしないようにしてください。CPU 負荷の高い処理やクラッシュ分離が必要な場合は、サンドボックス化されたプロセッサー／useWorkerThreads を使用してください。ジョブの data は平文です。シークレットを含めたり、フィールドを暗号化せずに保存したりしないでください。
• Redis は、引き続きデフォルトの実績あるバックエンドです。単一データベースモデルをプロジェクトで選択する場合に限り、PostgreSQL（createPostgresBackend、明示的な runMigrations）を使用してください。同じキューで Redis と Postgres を混在させないでください。

検証

リポジトリに用意されたコマンドを優先してください。BullMQ に影響する重要な変更では、関連する項目を選んで確認してください。

• Queue／Worker／FlowProducer のオプション（バックエンドのジェネリック型を含む）、ジョブの data／returnvalue／ProgressType の型、使用している場合は Nest のインジェクショントークンを型チェックします。
• プロセッサーの成功、再試行可能な失敗、UnrecoverableError、進捗、AbortSignal によるキャンセルを分離してユニットテストします。
• Redis または Postgres に対する結合テストを行います。ジョブを追加 → 処理 → 完了／失敗させ、ジョブの状態遷移を検証します。ジョブを追加する前に await worker.waitUntilReady() を実行してください（戻り値は void です）。
• 遅延／スケジューラのテストでは、短い間隔を使用します。再デプロイ後にスケジュールが重複しないことを検証します（upsertJobScheduler）。repeat／getRepeatableJobs が残っていないことも確認します。
• レート制限／重複排除のテストでは、無視／置換／延期の動作と、debounced ではなく deduplicated イベントが発生することを検証します。
• フローのテストでは、親より先に子が完了すること、親が getChildrenValues を読み取ること、子の失敗時のポリシーが製品の意図と一致すること、親ノードで deduplication が拒否されることを確認します。
• lockDuration、stalledInterval、maxStalledCount、または長時間実行されるプロセッサーを変更する場合は、スタール／ロックのスモークテストを行います。
• グレースフルシャットダウンのスモークテストでは、SIGTERM／close の処理経路で、進行中の作業が完了し、スタールウィンドウを超えて予期しない孤立ロックが残らないことを確認します。
• NestJS を使用している場合は、モジュールのスモークテストで、キューの登録、WorkerHost プロセッサー、FlowProducer のインジェクションを確認します。
• Postgres：runMigrations が適用されていること、起動時に SchemaVersionMismatchError／SchemaMigrationRequiredError が発生しないことを確認します。
• 保持設定（removeOnComplete／removeOnFail）を変更する場合、または高スループットのキューを追加する場合は、Redis の増加量を確認します。

実施したチェックと実施しなかったチェック、および残っている Pro／OSS、Redis と Postgres の違い、パッケージバージョンに関する前提を報告してください。