curiositech/windags-skills

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

APIレスポンスをブロックせずに長時間実行タスクを処理する、本番グレードのバックグラウンドジョブシステムの設計と実装のエキスパート。

使用するタイミング

✅ 使用するケース:

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

❌ 使用しないケース:

• リアルタイム双方向通信（WebSocketを使用）
• サブミリ秒のレイテンシ要件（インメモリキャッシュを使用）
• 単純な遅延（setTimeoutで5秒未満は十分）
• 同期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 });
});

なぜ間違っているか: タイムアウト、ユーザー体験の悪化、サーバーリソースの浪費

正しいアプローチ:

// ✅ 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などからの「Rate limit exceeded」エラー

正しいアプローチ:

// 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: ワーカースケーリングを忘れる

問題: 単一ワーカーがキュー深度に追いつけない

症状: キューがバックログし、ジョブが数時間/日遅延する

正しいアプローチ:

// 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

---

本番環境チェックリスト

□ デッドレターキューが設定されている
□ 指数バックオフ付きのリトライ戦略
□ ジョブタイムアウト制限が設定されている
□ サードパーティAPIのレート制限
□ 重要な操作のべき等性キー
□ ワーカー並行性の調整（CPUコア * 2）
□ 水平スケーリングが設定されている（複数ワーカー）
□ アラート付きキュー深度モニタリング
□ 失敗ジョブ検査ワークフロー
□ ログにジョブデータにPIIが含まれない
□ Redisの永続性が有効（AOFまたはRDB）
□ グレースフルシャットダウン処理（SIGTERM）

---

使用すべきか回避すべきかの判断

シナリオ	バックグラウンドジョブを使用するか？
サインアップ時にウェルカムメールを送信	✅ はい - 2〜5秒かかる可能性あり
クレジットカードに請求	⚠️ 場合による - 支払いプロバイダーのレイテンシに依存
PDFレポートを生成（30秒）	✅ はい - 間違いなくバックグラウンド
DBからユーザープロファイルを取得	❌ いいえ - ミリ秒、同期のまま
動画アップロードを処理（5分）	✅ はい - 常にバックグラウンド
フォーム入力を検証	❌ いいえ - 同期バリデーション
日次cronジョブ	✅ はい - 繰り返しジョブを使用
リアルタイムチャットメッセージ	❌ いいえ - WebSocketを使用

---

技術比較

機能	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 - BullMQをRedisで初期化
• scripts/queue_health_check.ts - キューメトリクスダッシュボード
• scripts/retry_failed_jobs.ts - 失敗したジョブを一括リトライ

---

このスキルのガイド: バックグラウンドジョブ実装 | キューアーキテクチャ | リトライ戦略 | ワーカースケーリング | ジョブ可観測性