background-job-orchestrator

バックグラウンドジョブオーケストレーター

API レスポンスをブロックせずに長時間実行されるタスクを処理する、本番環境向けのバックグラウンドジョブシステムの設計と実装に精通しています。

使用する場面

✅ 次の場合に使用します：

• 長時間実行されるタスク（メール送信、レポート生成、画像処理）
• バッチ処理（一括インポート、エクスポート、データ移行）
• スケジュールされたタスク（毎日のダイジェスト、クリーンアップジョブ、定期レポート）
• 再試行ロジックが必要なタスク（外部 API 呼び出し、不安定な操作）
• 優先度に基づく処理（プレミアムユーザーを優先、重大なアラート）
• レート制限のある操作（API クォータ、サードパーティサービスの制限）

❌ 次の場合には使用しません：

• リアルタイムの双方向通信（WebSockets を使用）
• 1 秒未満のレイテンシが必要な場合（インメモリキャッシュを使用）
• 単純な遅延（<5 秒なら setTimeout で十分です）
• 同期 API レスポンス（リクエストハンドラー内にロジックを保持します）

すばやく判断するためのフローチャート

Does this task:
├── Take >5 seconds? → Background job
├── Need to retry on failure? → Background job
├── Run on a schedule? → Background job (cron pattern)
├── Block user interaction? → Background job
├── Process in batches? → Background job
└── Return immediately? → Keep synchronous

---

テクノロジーの選択

Node.js：BullMQ（2024 年以降の推奨）

使用する場面：

• TypeScript プロジェクト
• Redis をすでにスタックで使用している
• 高度な機能（レート制限、優先度、繰り返しジョブ）が必要

Bull より BullMQ を選ぶ理由：

• Bull（v3）→ BullMQ（v4+）：TypeScript で全面的に書き直し
• Redis 接続の処理が向上
• 同時実行性とパフォーマンスが改善
• 積極的にメンテナンスされている（Bull はメンテナンスモード）

Python：Celery

使用する場面：

• Python／Django プロジェクト
• 分散タスクの実行が必要
• 複雑なワークフロー（チェーン、グループ、コード）

代替手段：

• RQ（Redis Queue）：よりシンプルで、機能が少ない
• Dramatiq：モダンで、エコシステムは小規模
• Huey：軽量で、小規模プロジェクトに適している

クラウドネイティブ：AWS SQS、Google Cloud Tasks

使用する場面：

• サーバーレスアーキテクチャ
• Redis／RabbitMQ を管理したくない
• 配信保証とデッドレターキューが必要

---

よくあるアンチパターン

アンチパターン 1：デッドレターキューがない

初心者の考え方：「3 回再試行して、その後は何も知らせずに失敗させる」

問題：失敗したジョブが、可視性も復旧手段もないまま消えてしまいます。

正しい方法：

// BullMQ with dead letter queue
const queue = new Queue('email-queue', {
  connection: redis,
  defaultJobOptions: {
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 2000
    },
    removeOnComplete: 100, // Keep last 100 successful
    removeOnFail: false     // Keep all failed for inspection
  }
});

// Monitor failed jobs
const failedJobs = await queue.getFailed();

変遷：

• 2020 年以前：再試行して忘れる
• 2020 年以降：デッドレターキューが標準に
• 2024 年以降：ジョブ失敗の可観測性が必須に

---

アンチパターン 2：同期的なジョブ処理

症状：API エンドポイントがジョブの完了を待つ

問題：

// ❌ WRONG - Blocks API response
app.post('/send-email', async (req, res) => {
  await sendEmail(req.body.to, req.body.subject);
  res.json({ success: true });
});

誤りである理由：タイムアウト、低い UX、サーバーリソースの浪費

正しい方法：

// ✅ RIGHT - Queue and return immediately
app.post('/send-email', async (req, res) => {
  const job = await emailQueue.add('send', {
    to: req.body.to,
    subject: req.body.subject
  });

  res.json({
    success: true,
    jobId: job.id,
    status: 'queued'
  });
});

// Separate worker processes the job
worker.process('send', async (job) => {
  await sendEmail(job.data.to, job.data.subject);
});

---

アンチパターン 3：冪等性がない

問題：ジョブが 2 回実行される → 二重請求、メールの二重送信

発生する理由：

• 処理中に Redis 接続が切断される
• ジョブが完了する前にワーカーがクラッシュする
• 実行中のままジョブのタイムアウトによって再試行が開始される

正しい方法：

// ✅ Idempotent job with deduplication key
await queue.add('charge-payment', {
  userId: 123,
  amount: 50.00
}, {
  jobId: `payment-${orderId}`, // Prevents duplicates
  attempts: 3
});

// In worker: Check if already processed
worker.process('charge-payment', async (job) => {
  const { userId, amount } = job.data;

  // Check idempotency
  const existing = await db.payments.findOne({
    jobId: job.id
  });
  if (existing) {
    return existing; // Already processed
  }

  // Process payment
  const result = await stripe.charges.create({...});

  // Store idempotency record
  await db.payments.create({
    jobId: job.id,
    result
  });

  return result;
});

---

アンチパターン 4：レート制限がない

問題：サードパーティ API に過剰な負荷をかける、またはクォータを使い切る

症状：Sendgrid、Stripe などから「レート制限を超過しました」というエラーが返る

正しい方法：

// BullMQ rate limiting
const queue = new Queue('api-calls', {
  limiter: {
    max: 100,        // Max 100 jobs
    duration: 60000  // Per 60 seconds
  }
});

// Or: Priority-based rate limits
await queue.add('send-email', data, {
  priority: user.isPremium ? 1 : 10,
  rateLimiter: {
    max: user.isPremium ? 1000 : 100,
    duration: 3600000 // Per hour
  }
});

---

アンチパターン 5：ワーカーのスケーリングを忘れる

問題：ワーカーが 1 つだけではキューの深さに対応できない

症状：キューが滞留し、ジョブが数時間／数日遅れる

正しい方法：

// Horizontal scaling with multiple workers
const worker = new Worker('email-queue', async (job) => {
  await processEmail(job.data);
}, {
  connection: redis,
  concurrency: 5  // Process 5 jobs concurrently per worker
});

// Run multiple worker processes (PM2, Kubernetes, etc.)
// Each worker processes concurrency * num_workers jobs

モニタリング：

// Set up alerts for queue depth
setInterval(async () => {
  const waiting = await queue.getWaitingCount();
  if (waiting > 1000) {
    alert('Queue depth exceeds 1000, scale workers!');
  }
}, 60000);

---

実装パターン

パターン 1：メールキャンペーン

// Queue setup
const emailQueue = new Queue('email-campaign', { connection: redis });

// Enqueue batch
async function sendCampaign(userIds: number[], template: string) {
  const jobs = userIds.map(userId => ({
    name: 'send',
    data: { userId, template },
    opts: {
      attempts: 3,
      backoff: { type: 'exponential', delay: 5000 }
    }
  }));

  await emailQueue.addBulk(jobs);
}

// Worker with retry logic
const worker = new Worker('email-campaign', async (job) => {
  const { userId, template } = job.data;

  const user = await db.users.findById(userId);
  const email = renderTemplate(template, user);

  try {
    await sendgrid.send({
      to: user.email,
      subject: email.subject,
      html: email.body
    });
  } catch (error) {
    if (error.code === 'ECONNREFUSED') {
      throw error; // Retry
    }
    // Invalid email, don't retry
    console.error(`Invalid email for user ${userId}`);
  }
}, {
  connection: redis,
  concurrency: 10
});

パターン 2：スケジュールされたレポート

// Daily report at 9 AM
await queue.add('daily-report', {
  type: 'sales',
  recipients: ['[email protected]']
}, {
  repeat: {
    pattern: '0 9 * * *', // Cron syntax
    tz: 'America/New_York'
  }
});

// Worker generates and emails report
worker.process('daily-report', async (job) => {
  const { type, recipients } = job.data;

  const data = await generateReport(type);
  const pdf = await createPDF(data);

  await emailQueue.add('send', {
    to: recipients,
    subject: `Daily ${type} Report`,
    attachments: [{ filename: 'report.pdf', content: pdf }]
  });
});

パターン 3：動画トランスコーディングパイプライン

// Multi-stage job with progress tracking
await videoQueue.add('transcode', {
  videoId: 123,
  formats: ['720p', '1080p', '4k']
}, {
  attempts: 2,
  timeout: 3600000 // 1 hour timeout
});

worker.process('transcode', async (job) => {
  const { videoId, formats } = job.data;

  for (let i = 0; i < formats.length; i++) {
    const format = formats[i];

    // Update progress
    await job.updateProgress((i / formats.length) * 100);

    // Transcode
    await ffmpeg.transcode(videoId, format);
  }

  await job.updateProgress(100);
});

// Client polls for progress
app.get('/videos/:id/status', async (req, res) => {
  const job = await queue.getJob(req.params.jobId);
  res.json({
    state: await job.getState(),
    progress: job.progress
  });
});

---

モニタリングと可観測性

必須メトリクス

// Queue health dashboard
async function getQueueMetrics() {
  const [waiting, active, completed, failed, delayed] = await Promise.all([
    queue.getWaitingCount(),
    queue.getActiveCount(),
    queue.getCompletedCount(),
    queue.getFailedCount(),
    queue.getDelayedCount()
  ]);

  return {
    waiting,    // Jobs waiting to be processed
    active,     // Jobs currently processing
    completed,  // Successfully completed
    failed,     // Failed after retries
    delayed,    // Scheduled for future
    health: waiting < 1000 && failed < 100 ? 'healthy' : 'degraded'
  };
}

BullMQ Board（UI）

// Development: Monitor jobs visually
import { createBullBoard } from '@bull-board/api';
import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
import { ExpressAdapter } from '@bull-board/express';

const serverAdapter = new ExpressAdapter();

createBullBoard({
  queues: [
    new BullMQAdapter(emailQueue),
    new BullMQAdapter(videoQueue)
  ],
  serverAdapter
});

app.use('/admin/queues', serverAdapter.getRouter());
// Visit http://localhost:3000/admin/queues

---

本番環境向けチェックリスト

□ Dead letter queue configured
□ Retry strategy with exponential backoff
□ Job timeout limits set
□ Rate limiting for third-party APIs
□ Idempotency keys for critical operations
□ Worker concurrency tuned (CPU cores * 2)
□ Horizontal scaling configured (multiple workers)
□ Queue depth monitoring with alerts
□ Failed job inspection workflow
□ Job data doesn't contain PII in logs
□ Redis persistence enabled (AOF or RDB)
□ Graceful shutdown handling (SIGTERM)

---

使用する場面と避ける場面

シナリオ	バックグラウンドジョブを使用しますか？
サインアップ時にウェルカムメールを送信する	✅ はい - 2～5 秒かかる場合があります
クレジットカードに請求する	⚠️ 場合によります - 決済プロバイダーのレイテンシによります
PDF レポートを生成する（30 秒）	✅ はい - 必ずバックグラウンドで処理します
DB からユーザープロフィールを取得する	❌ いいえ - 数ミリ秒なので、同期処理のままにします
動画アップロードを処理する（5 分）	✅ はい - 常にバックグラウンドで処理します
フォーム入力を検証する	❌ いいえ - 同期的に検証します
毎日の cron ジョブ	✅ はい - 繰り返しジョブを使用します
リアルタイムのチャットメッセージ	❌ いいえ - WebSockets を使用します

---

技術の比較

機能	BullMQ	Celery	AWS SQS
言語	Node.js	Python	任意（HTTP API）
バックエンド	Redis	Redis/RabbitMQ/SQS	マネージド
優先度	✅	✅	✅
レート制限	✅	❌	✅（属性を使用）
繰り返し／Cron	✅	✅（celery-beat）	❌（EventBridge を使用）
UI ダッシュボード	Bull Board	Flower	CloudWatch
ワークフロー	❌	✅（チェーン、グループ）	❌
学習コスト	中	中	低
コスト	Redis のホスティング	Redis のホスティング	100 万リクエストあたり $0.40

---

参考資料

• /references/bullmq-patterns.md - BullMQ の高度なパターンと例
• /references/celery-workflows.md - Celery のチェーン、グループ、コード
• /references/job-observability.md - モニタリング、アラート、デバッグ

スクリプト

• scripts/setup_bullmq.sh - Redis を使用して BullMQ を初期化
• scripts/queue_health_check.ts - キューのメトリクスダッシュボード
• scripts/retry_failed_jobs.ts - 失敗したジョブを一括で再試行

---

このスキルで扱う内容：バックグラウンドジョブの実装 | キューアーキテクチャ | 再試行戦略 | ワーカーのスケーリング | ジョブの可観測性