pg-boss

PG Boss ジョブキュー

PG Boss は Node.js 向けの PostgreSQL ベースのジョブキューです。ジョブをデータベースに保存することで、永続性と信頼性を提供します。

クイックセットアップ

import { PgBoss } from 'pg-boss';

const boss = new PgBoss({
  connectionString: process.env.DATABASE_URL,
  schema: 'pgboss',
  monitorIntervalSeconds: 30,
});

await boss.start();

よく使われるパターン

1. スケジュールされたジョブ（Cron のような処理）

// v12: queue must exist before schedule/work
await boss.createQueue('my-job', {
  retryLimit: 3,
  retryDelay: 60,
  expireInSeconds: 600,
  deleteAfterSeconds: 7 * 24 * 60 * 60,
});

// Schedule job to run every minute
await boss.schedule('my-job', '* * * * *', {}, {
  tz: 'UTC',
  singletonKey: 'my-job-schedule',  // Prevents duplicates on restart
});

// Register handler
await boss.work('my-job', { pollingIntervalSeconds: 10 }, async (jobs) => {
  for (const job of jobs) {
    // Process job
  }
});

2. ジョブのキューイング

// Queue a single job
await boss.send('process-user', { userId: '123' });

// Queue with options
await boss.send('send-email', { to: '[email protected]' }, {
  retryLimit: 5,
  expireInSeconds: 300,
});

3. 同時実行数を指定したバッチ処理

await boss.work(
  'batch-job',
  { batchSize: 10 },  // Process 10 jobs at once
  async (jobs) => {
    await Promise.all(jobs.map(job => processJob(job)));
  }
);

信頼性を高めるパターン

シングルトンパターン（重複の防止）

// Use getPgBoss() singleton instead of creating new instances
let boss: PgBoss | null = null;

export async function getPgBoss(): Promise<PgBoss> {
  if (boss) return boss;
  
  boss = new PgBoss({ connectionString: process.env.DATABASE_URL });
  await boss.start();
  return boss;
}

自動復旧のためのウォッチドッグ

PG Boss は予期せず停止することがあります（接続切断、クラッシュなど）。ウォッチドッグを追加してください。

let boss: PgBoss | null = null;
let watchdogInterval: NodeJS.Timeout | null = null;

// Listen for unexpected stops
boss.on('stopped', () => {
  console.error('[PG Boss] Stopped unexpectedly');
  boss = null;
  // Watchdog will attempt recovery
});

// Watchdog checks every 60s
watchdogInterval = setInterval(async () => {
  if (!boss && workerRegistrationFn) {
    console.log('[PG Boss] Attempting recovery...');
    await workerRegistrationFn();
  }
}, 60000);

ヘルスチェックとの連携

PG Boss が停止しているときにヘルスチェックを失敗させます（コンテナの再起動をトリガーします）。

app.get('/health', (c) => {
  const pgBossRunning = boss !== null;
  
  if (!pgBossRunning) {
    return c.json({ status: 'degraded', pgBossRunning: false }, 503);
  }
  
  return c.json({ status: 'ok', pgBossRunning: true });
});

デバッグ

データベースで PG Boss の状態を確認する

-- Check scheduled jobs
SELECT name, cron FROM pgboss.schedule;

-- Check recent jobs
SELECT name, state, created_on, completed_on 
FROM pgboss.job 
ORDER BY created_on DESC 
LIMIT 20;

-- Check job queue size
SELECT name, COUNT(*) 
FROM pgboss.job 
WHERE state = 'created' 
GROUP BY name;

Node.js 経由

const boss = await getPgBoss();

// Get queue size
const pending = await boss.getQueueSize('my-job');

// Get job by ID
const job = await boss.getJobById(jobId);

Railway 固有のデバッグ

# Check environment variables
railway variables --service api

# Get public DATABASE_URL for direct access
railway variables --service Postgres | grep DATABASE_PUBLIC_URL

# Query database directly
DATABASE_URL="postgresql://..." node -e "
const { Pool } = require('pg');
const pool = new Pool({ connectionString: process.env.DATABASE_URL, ssl: { rejectUnauthorized: false } });
// Run queries
"

よくある問題

問題	原因	解決方法
ジョブが実行されない	PG Boss が起動していない	ヘルスエンドポイントを確認し、DATABASE_URL が正しいことを確かめてください
ジョブが重複する	singletonKey が設定されていない	schedule() の呼び出しに singletonKey を追加してください
キューが見つからない	createQueue() の前に schedule() が呼び出されている	最初に boss.createQueue(name) を呼び出してください
ジョブが 'created' のまま停止する	ワーカーが登録されていない	キューに対して boss.work() を呼び出してください
接続エラー	データベース URL が誤っている／期限切れ	DATABASE_URL と SSL の設定を確認してください
ジョブが永続化されない	スキーマが誤っている	schema: 'pgboss' オプションを確認してください
ワーカーが過剰にポーリングする	オプション名が誤っている	pollingIntervalSeconds を使用してください（newJobCheckIntervalSeconds ではありません）
Next.js のビルドが失敗する（stream/net/tls）	instrumentation がサーバー専用の依存関係をバンドルしている	instrumentation の動的インポートに /* webpackIgnore: true */ を使用してください

ベストプラクティス

1. スケジュールされたジョブには、再起動時の重複を防ぐため、必ず singletonKey を使用してください
2. PG Boss v10 以降では、キューを明示的に作成してください：await boss.createQueue('my-job')
3. 予期しない停止を検知できるよう、'stopped' イベントを処理してください
4. クラッシュから自動復旧するためにウォッチドッグを使用してください（新しいウォッチドッグを開始する前に古いインターバルをクリアしてください）
5. PG Boss が停止しているときは、ヘルスチェックで 503 を返してください（コンテナの再起動をトリガーします）
6. ジョブのエラーをログに記録し、握りつぶさないでください。PG Boss に再試行させます
7. ジョブの種類（rollover、email など）ごとに別のワーカーを使用してください
8. グレースフルシャットダウン：SIGTERM／SIGINT の受信時に boss.stop({ graceful: true, timeout: 30000 }) を呼び出してください
9. Next.js の instrumentation：サーバー専用のジョブモジュールを import(/* webpackIgnore: true */ ...) で読み込んでください

ファイル構成の例

apps/api/src/
├── lib/
│   └── pgboss.ts          # Singleton, getPgBoss(), stopPgBoss()
├── workers/
│   ├── index.ts           # registerAllWorkers()
│   ├── rollover/
│   │   ├── index.ts       # registerRolloverWorkers()
│   │   ├── timezone-check.ts
│   │   └── batch-processor.ts
│   └── email/
│       ├── index.ts       # registerEmailWorkers()
│       └── send-email.ts
└── index.ts               # Calls registerAllWorkers() on startup

タイムゾーンを考慮したスケジューリング

深夜の切り替え処理など、特定の現地時刻に実行する必要があるジョブの場合：

import { toZonedTime } from 'date-fns-tz';

// Check if it's midnight in a timezone
const zonedNow = toZonedTime(new Date(), userTimezone);
const hour = zonedNow.getHours();
const minute = zonedNow.getMinutes();

// 10-minute window for reliability
const isMidnightWindow = hour === 0 && minute <= 10;

冪等性

重複処理を防ぐためにログテーブルを使用します。

// Check if already processed
const existing = await db.query.logs.findFirst({
  where: and(
    eq(logs.timezone, timezone),
    eq(logs.date, targetDate)
  ),
});

if (existing) return; // Already processed

// Process and log
await processJobs();
await db.insert(logs).values({ timezone, date: targetDate, status: 'completed' });