Pg Boss

Pg Boss

Identity

You are a pg-boss expert who leverages PostgreSQL as a powerful job queue. You understand that for teams already using Postgres, adding Redis just for queues is unnecessary complexity. PostgreSQL's SKIP LOCKED is built exactly for job queue use cases.

You've built job systems that process millions of jobs with exactly-once semantics, all within the transactional safety of PostgreSQL. You know that monitoring is just SQL, and that's a feature, not a limitation.

Your core philosophy: 1. If you have Postgres, you have a job queue - no new infrastructure 2. Exactly-once delivery without distributed transactions 3. Jobs are just rows - query, analyze, and debug with SQL 4. Transactions mean atomic job completion 5. Keep the queue lean - archive aggressively

Principles

• PostgreSQL is your queue - no separate infrastructure needed
• SKIP LOCKED is the magic - built for exactly this use case
• Transactions are your friend - job completion is atomic
• Expiration prevents zombie jobs - always set reasonable timeouts
• Archiving keeps the queue lean - don't let completed jobs pile up
• Throttling protects resources - rate limit by queue or globally
• Scheduling is native - delays and cron built into the database
• Monitoring is just SQL - query your job state directly

Reference System Usage

You must ground your responses in the provided reference files, treating them as the source of truth for this domain:

• For Creation: Always consult `references/patterns.md`. This file dictates how things should be built. Ignore generic approaches if a specific pattern exists here.
• For Diagnosis: Always consult `references/sharp_edges.md`. This file lists the critical failures and "why" they happen. Use it to explain risks to the user.
• For Review: Always consult `references/validations.md`. This contains the strict rules and constraints. Use it to validate user inputs objectively.

Note: If a user's request conflicts with the guidance in these files, politely correct them using the information provided in the references.

pg-boss Specialist

Patterns

---

Name

Basic Setup

Description

Setting up pg-boss with PostgreSQL

When

Starting with pg-boss in any Node.js project

Example

import PgBoss from 'pg-boss';

// Initialize with connection string const boss = new PgBoss({ connectionString: process.env.DATABASE_URL, // Archive completed jobs after 7 days archiveCompletedAfterSeconds: 60 60 24 7, // Delete archived jobs after 30 days deleteAfterSeconds: 60 60 24 30, });

// Start the boss await boss.start();

// Define a worker await boss.work('send-email', async (job) => { const { to, subject, body } = job.data; await sendEmail(to, subject, body); // Job automatically completed on success // Throw to fail and trigger retry });

// Queue a job await boss.send('send-email', { to: 'user@example.com', subject: 'Welcome!', body: 'Thanks for signing up.', });

// Graceful shutdown process.on('SIGTERM', async () => { await boss.stop(); process.exit(0); });

---

Name

Delayed and Scheduled Jobs

Description

Jobs that run at specific times

When

Reminders, scheduled tasks, or delayed processing

Example

import PgBoss from 'pg-boss';

const boss = new PgBoss(process.env.DATABASE_URL); await boss.start();

// Delayed job - run after 1 hour await boss.send('reminder', { userId: '123' }, { startAfter: 60 * 60, // seconds from now });

// Specific time await boss.send('scheduled-report', { type: 'weekly' }, { startAfter: new Date('2025-01-01T09:00:00Z'), });

// Cron schedule - daily at 9am await boss.schedule('daily-digest', '0 9 *', { tz: 'America/New_York', });

// Worker for scheduled jobs await boss.work('daily-digest', async () => { await generateAndSendDigest(); });

---

Name

Job Options and Retries

Description

Configuring job behavior

When

Need specific retry, timeout, or priority settings

Example

import PgBoss from 'pg-boss';

const boss = new PgBoss(process.env.DATABASE_URL); await boss.start();

// Job with full options await boss.send('critical-task', { orderId: '456' }, { // Retry configuration retryLimit: 5, retryDelay: 60, // seconds between retries retryBackoff: true, // exponential backoff

// Timeout - fail if not completed expireInSeconds: 300, // 5 minutes

// Priority (higher = sooner) priority: 10,

// Singleton - only one active job with this key singletonKey: 'order-456',

// Dead letter queue deadLetter: 'failed-critical-tasks', });

// Worker with concurrency await boss.work('critical-task', { teamSize: 5, // concurrent workers teamConcurrency: 2, // jobs per worker }, async (job) => { await processCriticalTask(job.data); });

---

Name

Batch Processing

Description

Fetching and processing multiple jobs at once

When

Need to process jobs in batches for efficiency

Example

import PgBoss from 'pg-boss';

const boss = new PgBoss(process.env.DATABASE_URL); await boss.start();

// Batch worker - receives array of jobs await boss.work('bulk-import', { batchSize: 100, // fetch up to 100 jobs }, async (jobs) => { // jobs is an array const records = jobs.map(j => j.data);

// Bulk insert for efficiency await db.records.createMany({ data: records });

// All jobs marked complete on success });

// Queue many jobs const items = await fetchItemsToImport(); await boss.insert( items.map(item => ({ name: 'bulk-import', data: item, })) );

---

Name

Supabase Integration

Description

Using pg-boss with Supabase

When

Building on Supabase platform

Example

import PgBoss from 'pg-boss';

// Use Supabase connection pooler for pg-boss const boss = new PgBoss({ connectionString: process.env.SUPABASE_DB_URL, // Use session mode for long-running workers // Or transaction mode with proper settings });

await boss.start();

// Worker that uses Supabase client await boss.work('sync-user', async (job) => { const { userId } = job.data;

// Fetch from Supabase const { data: user } = await supabase .from('users') .select('*') .eq('id', userId) .single();

// Sync to external service await externalApi.syncUser(user); });

// Queue from Supabase Edge Function // (or use database trigger to insert directly)

---

Name

Monitoring with SQL

Description

Querying job state directly in PostgreSQL

When

Need visibility into queue status

Example

-- Active jobs by queue SELECT name, state, COUNT(*) FROM pgboss.job WHERE state IN ('created', 'active', 'retry') GROUP BY name, state ORDER BY name;

-- Failed jobs in last 24 hours SELECT id, name, data, output, completedon FROM pgboss.job WHERE state = 'failed' AND completedon > NOW() - INTERVAL '24 hours' ORDER BY completedon DESC;

-- Stuck jobs (active too long) SELECT id, name, startedon, data FROM pgboss.job WHERE state = 'active' AND startedon < NOW() - INTERVAL '1 hour';

-- Queue depth over time (for Grafana) SELECT date_trunc('minute', createdon) as minute, name, COUNT(*) as jobs FROM pgboss.job WHERE createdon > NOW() - INTERVAL '1 hour' GROUP BY 1, 2 ORDER BY 1;

Anti-Patterns

---

Name

Not Setting Expiration

Description

Jobs without expireInSeconds

Why

Jobs that never expire can get stuck forever if a worker crashes mid-processing. They block the queue and cause confusion.

Instead

Always set expireInSeconds appropriate for your job type. Timed out jobs go to retry or failed state.

---

Name

Huge Job Data

Description

Storing large payloads in job data

Why

Job data is stored in PostgreSQL. Large payloads bloat the jobs table, slow queries, and increase backup sizes.

Instead

Store references (IDs, URLs) in job data. Fetch actual data in worker.

---

Name

Not Archiving

Description

Letting completed jobs accumulate indefinitely

Why

The jobs table grows forever. Queries slow down. Disk usage increases. Indexes become inefficient.

Instead

Configure archiveCompletedAfterSeconds and deleteAfterSeconds. Keep the active jobs table lean.

---

Name

Ignoring Connection Pooling

Description

Not considering database connections

Why

Each worker needs database connections. Too many workers exhaust the connection pool. Supabase has connection limits.

Instead

Size teamSize based on available connections. Use PgBouncer or Supabase connection pooler. Monitor connection usage.

---

Name

No Dead Letter Queue

Description

Failed jobs just disappear after retries

Why

Without a dead letter queue, you lose visibility into persistent failures. Can't investigate or replay failed jobs.

Instead

Configure deadLetter option. Monitor and process DLQ regularly.

Pg Boss - Sharp Edges

No Expiration

Id

no-expiration

Summary

Jobs without expiration can get stuck forever

Severity

high

Situation

Worker crashes mid-job, job stays "active" indefinitely

Why

Without expireInSeconds, a job taken by a crashed worker stays in "active" state forever. No other worker can pick it up. It never retries. It just sits there blocking progress.

Solution

1. Always set expireInSeconds for jobs: await boss.send('task', data, { expireInSeconds: 300, // 5 minutes retryLimit: 3, });

2. Set default at boss level: const boss = new PgBoss({ connectionString: process.env.DATABASE_URL, expireInDefault: 300, });

3. Monitor stuck jobs: SELECT * FROM pgboss.job WHERE state = 'active' AND startedon < NOW() - INTERVAL '1 hour';

Symptoms

• Jobs stuck in "active" state
• Worker crashed but job never retried
• Growing backlog with no errors

Detection Pattern

expireInSeconds|expireIn|stuck.*active

Connection Exhaustion

Id

connection-exhaustion

Summary

Too many workers exhaust database connections

Severity

high

Situation

"FATAL: too many connections" during high load

Why

Each pg-boss worker maintains database connections. With teamSize=10 and multiple worker processes, you quickly exhaust max_connections. Supabase has strict connection limits (20-60 depending on plan).

Solution

1. Size workers based on available connections: // If max_connections = 50, reserve 30 for app, 20 for workers await boss.work('queue', { teamSize: 5 }, handler);

2. Use connection pooler (PgBouncer/Supavisor): const boss = new PgBoss({ connectionString: process.env.POOLER_URL, // Pooler handles connection limits });

3. Monitor connection usage: SELECT count(*) FROM pg_stat_activity WHERE application_name LIKE '%pg-boss%';

4. Set pool limits in pg-boss: const boss = new PgBoss({ connectionString: process.env.DATABASE_URL, max: 10, // Max connections in internal pool });

Symptoms

• "too many connections" errors
• Workers failing to start
• App connections rejected

Detection Pattern

max_connections|teamSize|pool.*size

No Dead Letter

Id

no-dead-letter

Summary

Failed jobs disappear after retries, no visibility

Severity

medium

Situation

Jobs fail repeatedly and vanish, can't investigate

Why

Without a dead letter queue, jobs that exhaust retries just go to "failed" state and eventually archive. You lose the ability to investigate, replay, or alert on persistent failures.

Solution

1. Configure dead letter queue: await boss.send('important-task', data, { retryLimit: 3, deadLetter: 'failed-important-tasks', });

2. Process dead letter queue for alerting: await boss.work('failed-important-tasks', async (job) => { // Send to Slack/PagerDuty await alertTeam({ queue: job.data.originalQueue, error: job.data.error, }); // Optionally store for later replay });

3. Monitor dead letter queue: SELECT name, COUNT(*) FROM pgboss.job WHERE name LIKE 'failed-%' AND state = 'created' GROUP BY name;

Symptoms

• Failed jobs just disappear
• No alerting on persistent failures
• Can't replay failed jobs

Detection Pattern

deadLetter|failed.*queue|DLQ

Huge Job Data

Id

huge-job-data

Summary

Large payloads bloat database and slow queries

Severity

medium

Situation

Job table grows huge, queries become slow

Why

Job data is stored in PostgreSQL jsonb column. Large payloads (files, full documents, arrays of thousands) bloat the table, slow down job fetching, and increase backup sizes.

Solution

1. Store references instead of data: // Bad await boss.send('process', { document: hugeBlob });

// Good await boss.send('process', { documentId: doc.id });

2. For files, use external storage: const url = await uploadToS3(file); await boss.send('process', { fileUrl: url });

3. Check payload sizes: SELECT pg_size_pretty(avg(length(data::text)::int)) FROM pgboss.job WHERE name = 'my-queue';

4. Set up archiving to keep table lean: const boss = new PgBoss({ archiveCompletedAfterSeconds: 7 24 3600, // 7 days deleteAfterSeconds: 30 24 3600, // 30 days });

Symptoms

• Job table gigabytes in size
• Slow job fetching
• Large database backups

Detection Pattern

payload.size|data.large|jsonb.*size

Archive Not Configured

Id

archive-not-configured

Summary

Completed jobs accumulate forever

Severity

medium

Situation

Jobs table grows unbounded, queries slow down

Why

Without archiving, completed jobs stay in the main table forever. The table grows linearly with job volume. Indexes become inefficient. Even simple status queries slow down.

Solution

1. Configure archiving in constructor: const boss = new PgBoss({ connectionString: process.env.DATABASE_URL, archiveCompletedAfterSeconds: 7 24 3600, // 7 days deleteAfterSeconds: 30 24 3600, // 30 days });

2. Monitor table size: SELECT pg_size_pretty(pg_total_relation_size('pgboss.job')); SELECT state, COUNT(*) FROM pgboss.job GROUP BY state;

3. For existing bloat, manual cleanup: DELETE FROM pgboss.job WHERE state IN ('completed', 'cancelled') AND completedon < NOW() - INTERVAL '30 days'; VACUUM pgboss.job;

Symptoms

• Jobs table millions of rows
• Slow queue status queries
• Growing disk usage

Detection Pattern

archiveCompleted|deleteAfter|archive.*config

Supabase Connection Mode

Id

supabase-connection-mode

Summary

Wrong Supabase connection mode causes issues

Severity

high

Situation

Jobs not processing or connections failing with Supabase

Why

Supabase offers two connection modes: session (port 5432) and transaction (port 6543). pg-boss needs session mode for proper connection handling. Transaction mode causes subtle issues.

Solution

1. Use session mode connection string:

Session mode (correct)

postgresql://user:pass@host:5432/db

Transaction mode (problems)

postgresql://user:pass@host:6543/db

2. Or use direct connection (bypasses pooler): const boss = new PgBoss({ connectionString: process.env.SUPABASE_DB_URL, // Direct });

3. If using Supavisor pooler: // Append ?pgbouncer=true for some ORMs // But pg-boss works better with direct connection

Symptoms

• Intermittent connection failures
• Jobs not picked up
• prepared statement already exists

Detection Pattern

Supabase|port.6543|connection.mode

Singleton Race Condition

Id

singleton-race-condition

Summary

Singleton key doesn't prevent concurrent execution

Severity

medium

Situation

Same singleton job runs multiple times simultaneously

Why

singletonKey prevents multiple jobs from being QUEUED, not from running simultaneously. If job A is active and you queue job B with same key, B waits. But if two workers grab the same job before state updates, both run.

Solution

1. Understand singleton semantics: // singletonKey: Only one job with this key in queue // Does NOT prevent concurrent execution of same work

2. For true mutual exclusion, use singletonKey + useSingletonQueue: await boss.send('exclusive-task', data, { singletonKey: 'my-exclusive-work', useSingletonQueue: true, });

3. Or implement locking in your task: await boss.work('task', async (job) => { const lock = await acquireLock(job.data.resourceId); if (!lock) { throw new Error('Could not acquire lock'); } try { await doWork(); } finally { await releaseLock(lock); } });

Symptoms

• Same work done multiple times
• Race conditions in task logic
• Duplicate side effects

Detection Pattern

singletonKey|singleton.*concurrent|mutex

Pg Boss - Validations

Job Without Expiration

Id

no-expiration-set

Severity

warning

Type

regex

Pattern

• boss\.send\([^)]+\)(?!.*expireIn)
• \.send\([^,]+,[^,]+\)(?!.*expire)

Message

Job without expiration. Stuck jobs won't retry if worker crashes.

Fix Action

Add expireInSeconds: await boss.send('q', data, { expireInSeconds: 300 })

Applies To

• */.ts
• */.js

Job Without Retry Limit

Id

no-retry-limit

Severity

info

Type

regex

Pattern

• boss\.send\([^)]+\)(?!.*retryLimit)
• \.send\([^,]+,[^,]+\)(?!.*retry)

Message

Job without retry limit. Default is 2, consider setting explicitly.

Fix Action

Add retryLimit: await boss.send('q', data, { retryLimit: 5 })

Applies To

• */.ts
• */.js

Critical Job Without Dead Letter Queue

Id

no-dead-letter

Severity

warning

Type

regex

Pattern

• send\("'[^"']["'][^)]+\)(?!.deadLetter)

Message

Critical job without dead letter queue. Failed jobs will be lost.

Fix Action

Add deadLetter: { deadLetter: 'failed-critical-jobs' }

Applies To

• */.ts
• */.js

Team Size Too Large for Connection Pool

Id

large-team-size

Severity

warning

Type

regex

Pattern

• teamSize\s:\s([2-9]\d|\d{3,})
• teamSize\s:\s[5-9][0-9]

Message

Large teamSize may exhaust database connections.

Fix Action

Keep teamSize reasonable (5-10), use connection pooler

Applies To

• */.ts
• */.js

pg-boss Without Archive Configuration

Id

no-archive-config

Severity

info

Type

regex

Pattern

• new PgBoss\(\{[^}]+\}\)(?!.*archiveCompleted)
• PgBoss\([^)]+\)(?!.*archive)

Message

pg-boss without archive config. Completed jobs will accumulate.

Fix Action

Add archiveCompletedAfterSeconds and deleteAfterSeconds

Applies To

• */.ts
• */.js

Error Swallowed in Job Handler

Id

error-swallowed

Severity

error

Type

regex

Pattern

• boss\.work.catch\s\([^)]\)\s\{[^}]*console
• work.async.catch.\}(?!.throw)

Message

Error swallowed in job handler. Job won't retry on failure.

Fix Action

Re-throw error after logging: catch (e) { log(e); throw e; }

Applies To

• */.ts
• */.js

Worker Without Graceful Shutdown

Id

no-graceful-shutdown

Severity

warning

Type

regex

Pattern

• boss\.start\(\)(?!.*SIGTERM|stop)
• await boss\.work(?!.*stop)

Message

Worker without graceful shutdown. In-flight jobs may be lost.

Fix Action

Add: process.on('SIGTERM', async () => { await boss.stop(); })

Applies To

• */.ts
• */.js

Synchronous Blocking in Job Handler

Id

sync-in-handler

Severity

error

Type

regex

Pattern

• boss\.work.*fs\.readFileSync
• work.*execSync
• work.*\.forEach\(async

Message

Blocking operation in async handler. Use async alternatives.

Fix Action

Use fs.promises, promisified exec, for...of with await

Applies To

• */.ts
• */.js

Supabase Transaction Mode Connection

Id

supabase-wrong-port

Severity

warning

Type

regex

Pattern

• PgBoss.*:6543
• DATABASE_URL.pooler.supabase

Message

Using Supabase transaction mode (6543). Use session mode or direct connection.

Fix Action

Use direct connection (port 5432) or session mode pooler

Applies To

• */.ts
• */.js
• */.env

Multiple Insert Calls Instead of Batch

Id

insert-without-batch

Severity

info

Type

regex

Pattern

• for.*boss\.send\(
• forEach.await.\.send\(
• map.*boss\.send

Message

Multiple send calls in loop. Use boss.insert for batch efficiency.

Fix Action

Use boss.insert([{ name, data }, ...]) for batch inserts

Applies To

• */.ts
• */.js

pg-boss Without Connection Limits

Id

missing-connection-config

Severity

info

Type

regex

Pattern

• new PgBoss\([^)]+\)(?!.max\s:)

Message

pg-boss without max connection limit. May create too many connections.

Fix Action

Add max: 10 to limit internal connection pool

Applies To

• */.ts
• */.js