Reliable Job Queue

A lightweight, PostgreSQL-backed background job queue for Node.js with durable storage, retries, delayed execution, worker leasing, crash recovery, and Dead Letter Queue (DLQ) support.

Jobs are persisted in PostgreSQL, allowing them to survive worker crashes and application restarts.

---

Features

• Durable PostgreSQL-backed job storage
• Background workers
• Configurable worker concurrency
• Delayed jobs
• Scheduled execution
• Priority-based scheduling
• Automatic retries with exponential backoff
• Dead Letter Queue (DLQ)
• Replay failed jobs
• Lease-based job processing (visibility timeout)
• Automatic lease renewal (heartbeats)
• Recovery of abandoned jobs
• Graceful worker shutdown
• Queue statistics
• Job execution history

---

Current Status

This package has not yet been published to npm.

The current implementation relies on an internal Prisma schema and generated Prisma Client. Before the first npm release, the installation experience will be simplified so users can initialize the queue with minimal setup.

Until then, the project can be used directly from the source repository.

---

Installation

Clone the repository.

git clone https://github.com/<your-username>/reliable-job-queue.git
cd reliable-job-queue

Install dependencies.

npm install

---

Prerequisites

• Node.js 20+
• PostgreSQL

Create a .env file in the project root.

DATABASE_URL="postgresql://postgres:password@localhost:5432/reliable_job_queue"

---

Database Setup

Generate the Prisma Client.

npm run prisma:generate

Apply the database migrations.

npm run prisma:migrate

or

npx prisma migrate dev

Build the project.

npm run build

The queue is now ready to use.

---

Quick Start

Create a Queue

import { Queue } from "@reliable-job-queue/core";

const queue = new Queue();

---

Register a Job Handler

queue.register("send-email", async (job) => {
  console.log(job.payload);

  // Send email...
});

---

Start a Worker

await queue.startWorker();

---

Enqueue a Job

await queue.enqueue({
  queue: "emails",
  type: "send-email",
  payload: {
    to: "john@example.com",
    subject: "Welcome",
    body: "Hello!"
  }
});

---

Delayed Jobs

Run a job after a delay.

await queue.enqueue({
  queue: "emails",
  type: "send-email",
  payload: {...},
  options: {
    delay: 30000
  }
});

---

Scheduled Jobs

Run a job at a specific time.

await queue.enqueue({
  queue: "emails",
  type: "send-email",
  payload: {...},
  options: {
    runAt: new Date(Date.now() + 60000)
  }
});

---

Job Priority

Higher priority jobs execute before lower priority jobs.

await queue.enqueue({
  queue: "emails",
  type: "send-email",
  payload: {...},
  options: {
    priority: 100
  }
});

---

Retry Configuration

await queue.enqueue({
  queue: "emails",
  type: "send-email",
  payload: {...},
  options: {
    maxAttempts: 5
  }
});

Retries use exponential backoff by default.

---

Worker Configuration

const queue = new Queue({
  worker: {
    concurrency: 5,
    pollingInterval: 1000,
    heartbeatInterval: 3000,
    leaseDuration: 30000
  }
});

Option	Default	Description
concurrency	1	Maximum number of jobs processed simultaneously
pollingInterval	1000ms	Database polling interval
heartbeatInterval	10000ms	Lease renewal interval
leaseDuration	30000ms	Visibility timeout (lease duration)

---

Dead Letter Queue

Retrieve failed jobs.

const jobs = await queue.getDLQ();

Replay a failed job.

await queue.replay(jobId);

---

Queue Statistics

const stats = await queue.stats();

Example:

{
  "pending": 10,
  "processing": 3,
  "completed": 250,
  "failed": 0,
  "dlq": 2
}

---

Scheduler

Run a dedicated scheduler process to recover abandoned jobs.

import { Scheduler } from "@reliable-job-queue/core";

const scheduler = new Scheduler();

await scheduler.start();

The scheduler periodically:

• Detects expired leases
• Releases abandoned jobs
• Returns them to the queue

---

Graceful Shutdown

process.on("SIGINT", async () => {
  await queue.stopWorker();
});

The worker:

• Stops polling
• Waits for running jobs
• Stops renewing leases
• Marks itself offline

---

Reliability

Durable Storage

Jobs are persisted in PostgreSQL and survive application restarts.

---

Atomic Job Claiming

Workers claim jobs using an atomic database update.

Even if multiple workers attempt to claim the same job simultaneously, only one succeeds.

---

Visibility Timeout

When a worker claims a job:

• Status becomes PROCESSING
• A lease is assigned
• Other workers cannot process it until the lease expires.

---

Heartbeats

Long-running jobs periodically renew their lease to prevent premature recovery.

---

Crash Recovery

If a worker crashes while processing a job:

• The lease expires
• The scheduler detects the expired lease
• The job returns to the queue
• Another worker can continue processing it

---

Dead Letter Queue

Jobs exceeding the configured retry limit are automatically moved to the Dead Letter Queue for inspection and replay.

---

Architecture

                Producer

                    │

                    ▼

              PostgreSQL

           ┌────────┴────────┐
           │                 │

           ▼                 ▼

      Worker Process   Scheduler Process

           │                 │

      Job Executor    Recovery Manager

           │

      User Handler

---

Current Limitations

• Uses PostgreSQL polling for job discovery (LISTEN/NOTIFY is not yet implemented)
• Provides at-least-once delivery semantics; handlers should be idempotent
• Cron-based recurring jobs are not yet supported

API

Queue

const queue = new Queue(options);

Methods

queue.register(type, handler);

await queue.enqueue(job);

await queue.startWorker();

await queue.stopWorker();

await queue.stats();

await queue.getDLQ();

await queue.replay(jobId);

---

Scheduler

const scheduler = new Scheduler();

Methods

await scheduler.start();

await scheduler.stop();

---

License

MIT