Skip to main content
Logo for PG Boss

PG Boss

The pg-boss runtime provides PostgreSQL-based job queue processing with robust persistence, transactions, and reliability features.

Live Example​

Overview​

pg-boss provides:

  • PostgreSQL-backed queues for ACID-compliant job processing
  • Built-in persistence with automatic table creation
  • Job retries with configurable limits
  • Batch processing for high-throughput scenarios
  • Graceful shutdown with proper cleanup
  • Job results tracking with full status monitoring

Quick Start​

1. Create a New Project​

Create a new Pikku project with pg-boss support:

npm create pikku@latest

Select pg-boss as your runtime option during setup.

Project Structure​

After creating your project, you'll have these key files:

Queue Worker Functions​

Define your job processing logic (same as other queue systems):

queue.functions.ts
loading...

Queue Worker Registration​

Register workers with specific queues:

queue.wiring.ts
loading...

pg-boss Runtime Server​

The main server that processes jobs using PostgreSQL:

start.ts
loading...

Setup Guide​

Complete setup instructions:

README.md
loading...

How It Works​

  1. Database Connection: Connects to PostgreSQL using DATABASE_URL
  2. Auto Schema: Creates necessary tables automatically on startup
  3. Job Processing: Polls PostgreSQL for new jobs and distributes to workers
  4. ACID Compliance: All job operations are transactional

Configuration​

Database Connection​

Set your PostgreSQL connection string:

export DATABASE_URL="postgres://user:password@localhost:5432/your_database"

Worker Configuration​

Configure job processing behavior:

  • batchSize: Number of jobs to process in parallel
  • pollInterval: How often to poll for new jobs (milliseconds)

Unsupported Options​

Some configurations are managed by pg-boss internally:

  • Worker names (uses queue names instead)
  • Lock duration (handled by PostgreSQL)
  • Visibility timeout (uses PostgreSQL locks)

Database Requirements​

  • PostgreSQL 9.5+ required
  • Automatic setup: pg-boss creates tables on first run
  • No migrations needed: Schema is managed automatically

Job Processing​

Queue workers are standard Pikku functions that:

  • Accept typed job data from PostgreSQL
  • Return typed results stored in the database
  • Handle errors with automatic retries
  • Support long-running operations with transaction safety

Monitoring​

pg-boss provides comprehensive monitoring:

  • Job status in PostgreSQL tables
  • Built-in job completion tracking
  • Failed job inspection and retry
  • Database-level monitoring with SQL queries

Error Handling​

Failed jobs are automatically retried with PostgreSQL-backed persistence:

  • Configurable retry limits
  • Exponential backoff support
  • Dead letter queue functionality
  • Full error logging and tracking

PgBossServiceFactory API​

The PgBossServiceFactory from @pikku/queue-pg-boss manages the lifecycle of all pg-boss components:

import { PgBossServiceFactory } from '@pikku/queue-pg-boss'

const pgBossFactory = new PgBossServiceFactory({
  connectionString: process.env.DATABASE_URL!,
})
await pgBossFactory.init()

Key Methods​

MethodReturnsDescription
getQueueService()QueueServicePublishes jobs to queues
getQueueWorkers()QueueWorkersProcesses jobs from queues
getSchedulerService()SchedulerServiceManages scheduled/recurring tasks
init()Promise<void>Initializes pg-boss and creates tables
close()Promise<void>Shuts down pg-boss gracefully

Worker Registration​

Get the workers from the factory and call registerQueues(). The job runner is wired automatically when you import the generated bootstrap, so no extra setup is needed (build singleton services before registering so the logger is available):

import './.pikku/pikku-bootstrap.gen.js'

const queueWorkers = pgBossFactory.getQueueWorkers()
await queueWorkers.registerQueues()

Full Setup with Workflows​

import { PgBossServiceFactory } from '@pikku/queue-pg-boss'
import { PikkuKysely, PgKyselyWorkflowService } from '@pikku/kysely-postgres'
import type { KyselyPikkuDB } from '@pikku/kysely-postgres'
import './.pikku/pikku-bootstrap.gen.js'

const pgBossFactory = new PgBossServiceFactory(process.env.DATABASE_URL!)
await pgBossFactory.init()

const pikkuKysely = new PikkuKysely<KyselyPikkuDB>(logger, process.env.DATABASE_URL!)
await pikkuKysely.init()

const workflowService = new PgKyselyWorkflowService(pikkuKysely.kysely)
await workflowService.init()

const schedulerService = pgBossFactory.getSchedulerService()

const singletonServices = await createSingletonServices(config, {
  queueService: pgBossFactory.getQueueService(),
  schedulerService,
  workflowService,
})

// Register queue workers (includes the workflow queues)
const queueWorkers = pgBossFactory.getQueueWorkers()
await queueWorkers.registerQueues()

// Start the scheduler
await schedulerService.start()

Graceful Shutdown​

process.on('SIGTERM', async () => {
  await schedulerService.stop()
  await pgBossFactory.close()
  process.exit(0)
})

Advantages​

  • ACID Compliance: Full transaction support
  • No External Dependencies: Uses existing PostgreSQL infrastructure
  • Reliability: Database-backed persistence
  • Monitoring: Standard SQL queries for job inspection
  • Scalability: Horizontal scaling with multiple workers