Postgres Job Queue and Worker Pool

Almost every backend needs to do something later and reliably: send the email, render the report, call the slow partner API. This project builds that machinery on the database you already run. You will learn the parts that make queues hard: exactly-once delivery does not exist, so you build at-least-once delivery plus idempotent handlers; a crashed worker must not strand a job; a poison job must not retry forever; and a deploy must not kill work in flight. The key trick is one SQL statement, SELECT ... FOR UPDATE SKIP LOCKED, which lets many workers claim different rows without blocking each other.

Target behaviour

curl -s -X POST localhost:8080/api/jobs -H 'Content-Type: application/json' \
  -d '{"kind":"email.welcome","payload":{"to":"a@b.com"},"idempotencyKey":"welcome-42"}'
# 202 {"id":17,"status":"QUEUED"}

# same key again: 200 with the original job, no duplicate

curl -s -X POST localhost:8080/api/jobs -H 'Content-Type: application/json' \
  -d '{"kind":"report.build","payload":{},"delaySeconds":300}'

curl -s localhost:8080/api/jobs/17
# {"id":17,"status":"DONE","attempts":1}

curl -s 'localhost:8080/api/jobs?status=DEAD'      # inspect the dead-letter set
curl -s -X POST localhost:8080/api/jobs/9/retry    # status back to QUEUED
curl -s localhost:8080/actuator/prometheus | grep jobq
# jobq_jobs{status="queued"} 3.0
# jobq_processed_total{kind="email.welcome",outcome="success"} 12.0

Architecture

  HTTP API ──INSERT──▶ ┌───────────────── PostgreSQL: job table ─────────────────┐
  @Scheduled enqueuer  │ QUEUED > RUNNING > DONE | QUEUED(retry) | DEAD     │
                       └──▲──────────────▲───────────────────────▲───────────────┘
        claim (SKIP LOCKED)│   complete/fail (fenced)             │ reclaim expired
                    ┌──────┴───────┐                       ┌──────┴──────┐
   instance A       │ poller thread │──▶ worker pool (N)   │   reaper    │
   instance B       │ poller thread │──▶ worker pool (N)   │ (@Scheduled)│
                    └───────────────┘                      └─────────────┘

The row is the truth. A claim flips QUEUED to RUNNING and stamps locked_until; the handler then runs outside any transaction, so no connection or row lock is held for a slow job. That stamp is the lease (visibility timeout): if the worker dies, the reaper makes the job claimable again. Retries are just run_at moved forward, so delayed jobs, backoff and scheduling share one mechanism.

The core of it

The schema. The partial index keeps claims fast as finished rows pile up.

CREATE TABLE job (
    id              BIGSERIAL PRIMARY KEY,
    kind            TEXT        NOT NULL,
    payload         JSONB       NOT NULL DEFAULT '{}',
    status          TEXT        NOT NULL DEFAULT 'QUEUED',
    priority        INT         NOT NULL DEFAULT 0,
    attempts        INT         NOT NULL DEFAULT 0,
    max_attempts    INT         NOT NULL DEFAULT 5,
    run_at          TIMESTAMPTZ NOT NULL DEFAULT now(),
    locked_by       TEXT,
    locked_until    TIMESTAMPTZ,
    last_error      TEXT,
    idempotency_key TEXT UNIQUE,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX job_claim_idx ON job (priority DESC, run_at) WHERE status = 'QUEUED';

Enqueue is a plain INSERT ... ON CONFLICT (idempotency_key) DO NOTHING RETURNING ...; an empty result means the key existed, so fetch and return the original row. Because it is plain SQL it can join the caller's transaction, so "order saved but job never queued" cannot happen.

The claim. The CTE locks a batch of rows, skipping any a concurrent worker already holds, and the outer UPDATE stamps the lease in the same statement. Because it is a single statement it needs no explicit transaction.

public List<Job> claim(String workerId, int limit, Duration lease) {
    return jdbc.sql("""
            WITH next AS (
                SELECT id FROM job
                WHERE status = 'QUEUED' AND run_at <= now()
                ORDER BY priority DESC, run_at, id
                FOR UPDATE SKIP LOCKED
                LIMIT :limit)
            UPDATE job j SET status = 'RUNNING', attempts = j.attempts + 1,
                   locked_by = :worker, locked_until = now() + make_interval(secs => :lease)
            FROM next WHERE j.id = next.id
            RETURNING j.id, j.kind, j.payload::text AS payload, j.attempts, j.max_attempts, j.status""")
        .param("limit", limit).param("worker", workerId)
        .param("lease", (double) lease.toSeconds())
        .query(JOB_MAPPER).list();
}

complete is the same statement with SET status = 'DONE'. Completion and failure are fenced with locked_by and status = 'RUNNING'. A worker whose lease expired and whose job was handed to someone else updates zero rows and must not overwrite the new owner's result.

public boolean fail(Job job, String workerId, String error, Instant nextRun) {
    boolean dead = job.attempts() >= job.maxAttempts();
    return jdbc.sql("""
            UPDATE job SET status = :status, run_at = :runAt, last_error = :err,
                   locked_by = NULL, locked_until = NULL
            WHERE id = :id AND locked_by = :w AND status = 'RUNNING'""")
        .param("status", dead ? "DEAD" : "QUEUED").param("runAt", Timestamp.from(nextRun))
        .param("err", error).param("id", job.id()).param("w", workerId).update() == 1;
}

Backoff uses exponential growth with full jitter, so a thousand jobs that failed together do not all retry together. Capping the exponent avoids overflow.

static Instant nextRun(int attempts, Duration base, Duration cap) {
    long ceilingMs = Math.min(cap.toMillis(), base.toMillis() << Math.min(attempts, 20));
    return Instant.now().plusMillis(ThreadLocalRandom.current().nextLong(1, ceilingMs + 1));
}

The worker pool: one poller claims only as many jobs as there are free slots (a Semaphore), so claimed jobs never wait in memory burning their lease.

public interface JobHandler { String kind(); void handle(Job job) throws Exception; }

private void pollLoop() {
    while (running) {
        try {
            int free = Math.min(slots.availablePermits(), batchSize);
            List<Job> jobs = free == 0 ? List.of() : store.claim(workerId, free, lease);
            jobs.forEach(j -> { slots.acquireUninterruptibly(); pool.execute(() -> runOne(j)); });
            if (jobs.isEmpty()) Thread.sleep(pollInterval);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

runOne turns any Throwable from a handler, including an unknown kind, into a retry, never a dead worker thread.

private void runOne(Job job) {
    try {
        handlers.getOrDefault(job.kind(), UNKNOWN).handle(job);
        store.complete(job.id(), workerId);
        count(job, "success");
    } catch (Throwable e) {
        store.fail(job, workerId, e.toString(), Backoff.nextRun(job.attempts(), base, cap));
        count(job, job.attempts() >= job.maxAttempts() ? "dead" : "retry");
    } finally {
        slots.release();
    }
}

The reaper reclaims jobs whose lease expired. An expired lease still counts as an attempt, so a job that always kills the JVM ends up dead.

UPDATE job SET
    status = CASE WHEN attempts >= max_attempts THEN 'DEAD' ELSE 'QUEUED' END,
    last_error = 'lease expired', locked_by = NULL, locked_until = NULL
WHERE status = 'RUNNING' AND locked_until < now();

Scheduled and recurring jobs: use @Scheduled only to enqueue, with the idempotency key set to the time bucket (for example report.hourly: plus the hour truncated). Every instance fires but only one insert wins, so you get one run per interval with no leader election.

Graceful shutdown: implement SmartLifecycle, stop claiming first, let in-flight jobs finish within a budget, then give up. Anything still running is rescued by the reaper after the lease expires.

@Override
public void stop() {
    running = false;
    poller.interrupt();
    pool.shutdown();
    try {
        if (!pool.awaitTermination(30, TimeUnit.SECONDS)) pool.shutdownNow();
    } catch (InterruptedException e) {
        pool.shutdownNow();
        Thread.currentThread().interrupt();
    }
}

Set server.shutdown=graceful and spring.lifecycle.timeout-per-shutdown-phase=40s, above the 30 second drain.

Metrics: register a gauge per status (SELECT count(*) FROM job WHERE status = ?) for queue depth and dead count, plus a counter by kind and outcome and a timer for duration. Alert on dead count and on the age of the oldest queued job.

Compared with the alternatives

• Redis (lists, Streams): higher throughput and lower latency, but durability depends on persistence and you cannot enqueue atomically with a database write. Postgres suffices until tens of thousands of jobs per second.
• RabbitMQ or SQS: real acknowledgements, dead-letter exchanges and push delivery without polling. Use them when producers and consumers are separate services; you give up transactional enqueue and gain another system to operate.
• @Scheduled: runs in one JVM, has no persistence, no retry and fires on every instance. Fine for the enqueuer, wrong as a queue.
• Quartz (JDBC job store): durable cron-style scheduling with clustering. It is a scheduler, not a work queue: no per-job payload backoff, and its table locking is coarser than SKIP LOCKED. Use it for "run X every night", a queue for "do this thing for this user".

Testing

Run against real PostgreSQL, because SKIP LOCKED behaviour cannot be faked. This test proves that eight concurrent claimers never receive the same job.

@SpringBootTest
@Testcontainers
class JobStoreIT {
    @Container @ServiceConnection
    static PostgreSQLContainer<?> pg = new PostgreSQLContainer<>("postgres:16");
    @Autowired JobStore store;

    @Test
    void concurrentClaimsNeverOverlap() throws Exception {
        for (int i = 0; i < 200; i++) store.enqueue("t", "{}", "k" + i, Duration.ZERO);
        ExecutorService ex = Executors.newFixedThreadPool(8);
        List<Future<List<Job>>> fs = new ArrayList<>();
        for (int w = 0; w < 8; w++) {
            String id = "w" + w;
            fs.add(ex.submit(() -> store.claim(id, 50, Duration.ofSeconds(30))));
        }
        List<Long> ids = new ArrayList<>();
        for (var f : fs) f.get().forEach(j -> ids.add(j.id()));
        assertThat(ids).hasSize(200).doesNotHaveDuplicates();
    }
}

Also test the lease: claim with a one-second lease, wait, run the reaper, and assert the job is QUEUED with attempts == 1. Use Awaitility, never Thread.sleep.

Bugs you will hit

• Holding the claim transaction open while the handler runs. Wrapping claim and handler in one @Transactional pins a connection and row lock for the job's duration and exhausts the Hikari pool. Claim in its own statement, run the handler outside, and finish with a separate update.
• Unfenced completion. A slow worker finishes after its lease expired and the job was re-run elsewhere, then overwrites the newer result. Always update WHERE id = ? AND locked_by = ? AND status = 'RUNNING' and treat zero rows as "I lost the lease".
• Lease shorter than the slowest job. The reaper duplicates work that was still progressing. Set the lease to a multiple of p99 duration, or extend it with a heartbeat update while the handler runs.
• Retrying forever or in lockstep. Without max_attempts a poison job loops; without jitter a recovered dependency is hit by every failed job at once. Send exhausted jobs to DEAD, alert on the gauge, and keep jitter.
• Table bloat and a slow claim. Finished rows and dead tuples slow the planner. Keep the partial index, delete or partition DONE rows after a retention window, and tune autovacuum on the table.

Interview angle

• Why is exactly-once delivery impossible, and what do you do? A worker can finish the side effect and crash before recording success, so the broker cannot tell. You deliver at-least-once and make handlers idempotent with a unique constraint on the effect or an idempotency key passed to the downstream API.
• How does SKIP LOCKED differ from plain FOR UPDATE and from advisory locks? Plain FOR UPDATE makes workers queue behind one another on the same row; SKIP LOCKED ignores locked rows so workers fan out across the backlog. Advisory locks need you to manage keys and release manually. The trade-off of SKIP LOCKED is that it gives up strict ordering.
• How does this scale, and where does it stop? Add workers and instances freely; contention is only on the index head. Limits are write amplification and vacuum on a hot table, and polling latency, which you can reduce with LISTEN/NOTIFY as a wake-up hint. Past that, move to a broker.