Task Queue with PostgreSQL SKIP LOCKED

This repository demonstrates how to build an efficient task queue using PostgreSQL's FOR UPDATE SKIP LOCKED feature, providing row-level locking for parallel task processing without contention. This is the companion code for the blog post Simplifying Task Queues with PostgreSQL.

Features

• Task scheduling with priorities and delays
• Automatic retries with configurable intervals
• Unique task constraints
• Scheduled (recurring) tasks
• Graceful shutdown handling
• Parallel task processing with SKIP LOCKED

Example Tasks

• SendEmail: Simple email task simulation
• ProcessData: Data processing task simulation
• DailyReport: Daily scheduled task (unique, runs every 24 hours)
• HealthCheck: High-priority health check (unique, runs every 5 seconds, with retries)

Getting Started

1. Install dependencies:

npm install

2. Start PostgreSQL:

docker-compose up -d

3. Initialize the database:

npm run prisma:generate # Generate Prisma client
npm run prisma:migrate # Create database tables

4. Start the task system:

# Start both runner and scheduler
npm run start:all
# Or start them separately:
npm run start:runner # Terminal 1
npm run start:scheduler # Terminal 2

Architecture

• src/types.ts: Core task system types and registry
• src/tasks.ts: Task implementations using decorators
• src/taskQueue.ts: PostgreSQL-based queue operations
• src/runner.ts: Task execution and retry logic
• src/scheduler.ts: Scheduled task management

Database Schema

The system uses a simple PostgreSQL table:

CREATETABLE "Task" (
id String @id @default(uuid())
type String
priority Int @default(100)
payload Json
maturityAt DateTime @default(now())
createdAt DateTime @default(now())
updatedAt DateTime @updatedAt
)

Environment

Required environment variables (see .env.example):

DATABASE_URL="postgresql://taskuser:taskpass@localhost:6000/taskdb"

Development

1. Create a new task:

@RegisterTask("MyTask",frequency?,{unique?: boolean})exportclassMyTaskextendsTask{constructor(){super("MyTask",priority);}asyncrun(): Promise<TASK_STATUS>{// Implementation}}

2. Add retry logic:

nextTry(): number {returnthis.tries<3 ? 60*1000 : -1;// Retry 3 times with 1-minute delay}