← Back to the section

Almost every application has this task: accumulate rows in a table and process them in the background. Payments, notifications, reports — anything. The first solution is quick to write: a schedule, fetch a batch, loop. It even works — until the first deploy with two replicas, or the first error in an external system.

This article explains what exactly breaks and how to make background processing reliable.

The Typical First Approach

Here's the code you'll find in most projects:

@Injectable()
export class PaymentBatchJob {

  constructor(
    @InjectRepository(Payment) private readonly repo: Repository<Payment>,
    private readonly svc: PaymentService,
  ) {}

  @Interval(300_000)
  async run(): Promise<void> {
    const batch = await this.repo.find({
      where: { status: PaymentStatus.NEW },
      order: { createdAt: 'ASC' },
      take: 100,
    });
    for (const payment of batch) {
      try {
        await this.svc.process(payment.id);
      } catch (ex) {
        this.logger.warn(`Failed: ${payment.id}`, ex);
      }
    }
  }
}

Looks reasonable: every 5 minutes we take 100 rows with status NEW and process them. But this code has several serious problems.

What Breaks Here

One Bad Record Jams the Queue

The query is sorted by creation date: ORDER BY created_at ASC. This means that on the next run the worker will pick up the very same rows it did last time.

If one payment fails with an error every time, it never goes away. It stays at the front of the queue and, together with a hundred equally "stuck" records, blocks all the new ones. New payments pile up but never get processed.

There is no attempt counter. There is no delayed retry. There is no terminal "gave up" status.

Stuck Records After a Restart

The worker takes a record and immediately sets its status to PROCESSING. Then it calls the external system. If the service crashes at that moment, the status update back to NEW never happens. The record stays in PROCESSING forever.

The worker selects only NEW, so it will never pick up such a "stuck" record again. After every restart under load, a few of these records accumulate — and nobody knows about them until a customer complains.

Double Processing With Two Replicas

You start two instances of the service — both workers read the same 100 rows at the same time. One payment gets sent twice. For notifications that's an annoyance; for payments it's a serious incident.

There is no locking on the query, and several parallel workers have no way to coordinate with each other.

One Hung Call Stops Everything

Processing is sequential: one payment → a call to the external system → the next payment. If the external system "thinks" about one request, the whole loop stalls and waits. There are no timeouts. The other 99 payments sit in the queue.

How to Do It Right: a Task Queue on PostgreSQL

For most cases you don't need a separate message broker — the database is already there, and transactions already work. You just need the right mechanics on top of it.

Step 1: Add the Necessary Columns

ALTER TABLE payment
    ADD COLUMN attempts        int         NOT NULL DEFAULT 0,
    ADD COLUMN next_attempt_at timestamptz NOT NULL DEFAULT now(),
    ADD COLUMN claimed_until   timestamptz;
  • attempts — how many times we've tried to process it.
  • next_attempt_at — when it can be tried again (for retries with backoff).
  • claimed_until — until when the record is "held" by a worker.

Step 2: Atomically Claim a Batch

The key idea is that claiming and processing must be separate transactions. Claiming is a short operation that atomically marks records as "taken for work" right inside the UPDATE statement:

async claimBatch(limit: number): Promise<string[]> {
  return this.dataSource.transaction(async (manager) => {
    const rows: { id: string }[] = await manager.query(
      `UPDATE payment SET
           status        = 'PROCESSING',
           attempts      = attempts + 1,
           claimed_until = now() + interval '5 minutes'
       WHERE id IN (
           SELECT id FROM payment
           WHERE (status = 'NEW' AND next_attempt_at <= now())
              OR (status = 'PROCESSING' AND claimed_until < now())
           ORDER BY created_at
           LIMIT $1
           FOR UPDATE SKIP LOCKED
       )
       RETURNING id`,
      [limit],
    );
    return rows.map((row) => row.id);
  });
}

The key line is FOR UPDATE SKIP LOCKED. It's a PostgreSQL construct: several workers run this query at the same time, but each one skips rows that are already locked and takes the next free ones. There will be no double processing — the database distributes the work between workers itself, without any coordinator.

claimed_until solves the "stuck records" problem: if a worker died before finishing, after 5 minutes the condition claimed_until < now() brings the record back into the selection.

Step 3: Process Outside the Transaction

We've claimed the records and committed — now we call the external system. The HTTP call happens between transactions, not inside one:

private readonly pool = pLimit(10);

@Interval(5_000)
async run(): Promise<void> {
  const claimed = await this.claimBatch(100);
  await Promise.allSettled(
    claimed.map((id) => this.pool(() => this.processOne(id))));
}

async processOne(paymentId: string): Promise<void> {
  try {
    const decision = await this.limitsClient.check(paymentId);
    await this.finishSuccessfully(paymentId, decision);
  } catch (ex) {
    await this.scheduleRetry(paymentId, ex as Error);
  }
}

Several important changes compared to the first approach:

  • Records are processed in parallel with bounded concurrency (p-limit), not sequentially. One slow call doesn't block the rest.
  • The schedule is now every 5 seconds instead of every 5 minutes — an empty poll is cheap (one fast query), and processing latency drops from minutes to seconds.
  • The limitsClient HTTP client must have timeouts. Without an explicit timeout, one hung call will hold a concurrency slot forever.

Step 4: Retries With Backoff and Terminal Failure

When processing fails, the record does not stay at the front of the queue. It goes to the back with a delay:

async scheduleRetry(paymentId: string, ex: Error): Promise<void> {
  await this.dataSource.transaction(async (manager) => {
    const payment = await manager.findOneByOrFail(Payment, { id: paymentId });
    if (payment.attempts >= MAX_ATTEMPTS) {
      payment.markFailed(ex.message); // terminal FAILED status
    } else {
      payment.scheduleNextAttempt(backoff(payment.attempts)); // back to NEW with a delay
    }
    await manager.save(payment);
  });
}

backoff is a function that increases the pause with each attempt: for example, 1 min → 2 min → 4 min → 8 min. After N attempts the record gets the FAILED status with the error reason — a signal for a human to look into it manually. The queue never jams.

When You Need a Batch Processing Framework

BullMQ, Agenda, Celery with beat — these are tools for finite jobs: export a monthly payment register to a file, migrate ten million rows, re-price all customers once a quarter.

Their key capability is restart from the point of failure: "it crashed on the 7 millionth row — restarted, continued from the 7 millionth." To do that, the framework stores the run's state and supports skip and retry policies at the level of data chunks.

For our task (a continuous queue of incoming payments) such a framework is overkill: the stream has no end, restart-from-a-point isn't needed, and the questions of parallel claiming and retries stay the same — the framework doesn't solve them.

A simple rule: a finite batch with restart → a framework; an endless stream of tasks → a task queue on PostgreSQL.

When to Add a Message Broker

A broker (RabbitMQ, Kafka) solves the same problems as a queue on PostgreSQL, but adds new capabilities: several independent types of consumers for a single event, scaling workers without loading the main database, built-in dead-letter queues.

Moving to a broker makes sense when volumes grow to tens of thousands of tasks per minute, or when a single event has to be processed by several different services. Before that, a broker is just extra infrastructure with operational costs.

In Short

  • A plain @Interval loop breaks with parallel replicas, stuck records, and sequential HTTP calls without timeouts.
  • The right way to claim tasks is an atomic UPDATE with FOR UPDATE SKIP LOCKED: the database distributes the work between workers itself.
  • claimed_until (the "task lease") protects against stuck records after a restart — an expired lease is picked up by workers automatically.
  • HTTP calls go between transactions, not inside them; processing is parallel with a concurrency limit, not sequential.
  • Retries with growing backoff and a terminal FAILED status keep the queue from jamming.
  • A batch processing framework is for finite jobs with restart; a broker is for scale and multiple consumers.
  • Resilience Patterns — timeouts, retries, and the circuit breaker for calling external systems.
  • Distributed Patterns — transactional outbox and idempotency when working with a broker.
  • @Transactional in Spring — why transaction boundaries matter and where declarative management fails to kick in.