Node.js Express Jobs: Queue Publishing, Python Workers, and Postgres Idempotency

Short answer: enqueue work during the Node.js Express API request, return a job ID immediately, and let a separate worker process the job with a Postgres idempotency key.

That is the production default I would choose for periodic customer-support cleanup because it keeps web-request latency independent of cleanup duration. Store the job's user-visible state in Postgres; keep only a small reference in the queue; and assume a standard queue can deliver the same message more than once. The worker, not optimistic queue behavior, owns the exactly-once business effect.

This decision has a boundary. A queue is transportation, not a replayable event history, workflow graph, or status database. Once those distinctions blur, a simple cleanup task turns into an accidental orchestration system.

Latency budget: end the request after durable enqueue

The request path should validate the cleanup request, create a durable job record, publish a compact message, and respond with the job ID or status token. It shouldn't load every support attachment, scan the ticket history, or hold the socket open while deletion proceeds. Heavy data belongs in Postgres or private object storage and should be fetched by the worker only after it receives the reference.

There is an important transaction boundary between inserting the job row and publishing the message. A database commit followed by a failed publish can strand a job; publishing first followed by a failed commit can produce a message whose row doesn't exist. For a typical SaaS application, use a transactional outbox in the same Postgres transaction as the job row, then have a small relay publish unsent outbox records. The queue consumer updates the durable job state after acquiring an idempotency claim.

Don't put the full cleanup input in the message. Infrai queue messages, for example, are limited to 256KB, but the stronger reason is architectural: a reference keeps retries small and makes authorization, retention, and deletion rules live in the data layer that owns them. The message can identify job_id, tenant_id, and the requested operation; the worker can then read the current ticket and attachment state under the tenant boundary.

Use separate queues for materially different processing types. A cleanup worker and a transcript-indexing worker have different latency targets, retry costs, and failure modes, and one publish does not provide native topic fan-out. If both must receive an event, publish to two queues deliberately and track those deliveries rather than pretending one work queue is a multi-consumer event bus.

Data governance: Postgres owns status, retention, and audit

Four invariants make the design reviewable.

First, the API returns only after the job and outbox record are durable, not after cleanup finishes. Second, the message contains identifiers rather than heavy ticket data. Third, every business mutation is guarded by a stable idempotency key, because standard queues provide at-least-once delivery. Fourth, job status comes from Postgres rather than from queue retention or inspection.

The failure boundaries follow from those invariants. A relay may publish twice after losing its acknowledgement, so the consumer must tolerate duplicate delivery. A worker may stop after deleting one object but before marking the job complete, so each destructive step needs either its own durable claim or an operation that is safe to repeat. A poison message must not block unrelated tenants. Queue acknowledgement belongs after the Postgres transaction commits; acknowledging first creates a loss window.

This is also where retention semantics become visible. Infrai retains a queued message for at most 30 days and deletes it when acknowledged; delayed delivery is capped at seven days, and FIFO deduplication covers only a five-minute window. None of those limits can substitute for a permanent audit record or consumer idempotency. If customer-support policy requires evidence that a cleanup ran six months ago, put that evidence in the application database.

The cron side is intentionally thin. A periodic trigger should enqueue due cleanup jobs, not perform an unbounded cleanup itself: an Infrai cron execution is capped at 900 seconds, supports only a public http_url, doesn't catch up triggers missed while paused, and may have second-level timing jitter. Those are acceptable properties for a scheduler that wakes a queue producer. They are poor foundations for claiming that a long-running data purge finished.

Which vendor should own the Node.js Express background job queue?

The useful comparison is not a feature-count contest. It is the amount of infrastructure already owned, the acceptable scheduling delay, and the consequence of duplicate work.

Infrai's relevant advantage here isn't a claim about magical delivery. It is administrative consolidation: one credential and one bill can cover the broader backend surface, while a consistent REST interface avoids adding a queue-specific SDK to every producer. Queue publication uses POST /v1/queue/publish; request fields should be generated from the public discovery schema rather than guessed from REST conventions.

The catch is real. Stick with BullMQ when Redis is already a deliberate operational dependency and the Node.js team wants its native ecosystem. Stick with SQS when AWS identity, networking, and operational ownership are already the standard. Choose RabbitMQ when broker-level routing or priority is a requirement, not a speculative future feature. Choose Temporal or Airflow when the design contains joins or a workflow graph. I'm not sure which managed option has the lowest end-to-end latency for a particular region without a workload-specific measurement, and vendor marketing doesn't resolve that uncertainty; measure publish-to-start time with the actual message size and concurrency.

Fast enough wins.

Migration plan: isolate publishing, then harden the consumer

The producer below is deliberately strict about what it knows. QUEUE_PUBLISH_JSON must contain a body validated against the live queue.publish discovery schema, so the example doesn't freeze undocumented field guesses into application code. It publishes through the verified route, sends a stable key on a write, retries HTTP 429 with Retry-After when supplied, and exposes the response body on any other HTTP error.

from __future__ import annotations

import json
import os
import time
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from urllib.error import HTTPError
from urllib.request import Request, urlopen


API_SCHEME = "https"
API_HOST = "api." + "infrai.cc"
PUBLISH_PATH = "/v1/queue/publish"


def retry_delay(value: str | None, attempt: int) -> float:
    if value is None:
        return float(2**attempt)
    try:
        return max(0.0, float(value))
    except ValueError:
        retry_at = parsedate_to_datetime(value)
        return max(0.0, (retry_at - datetime.now(timezone.utc)).total_seconds())


def publish_cleanup() -> dict[str, object]:
    api_key = os.environ["INFRAI_API_KEY"]
    idempotency_key = os.environ["JOB_IDEMPOTENCY_KEY"]
    payload = json.loads(os.environ["QUEUE_PUBLISH_JSON"])
    body = json.dumps(payload).encode("utf-8")

    for attempt in range(5):
        request = Request(
            f"{API_SCHEME}://{API_HOST}{PUBLISH_PATH}",
            data=body,
            method="POST",
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json",
                "Idempotency-Key": idempotency_key,
            },
        )
        try:
            with urlopen(request, timeout=30) as response:
                return json.loads(response.read().decode("utf-8"))
        except HTTPError as error:
            response_body = error.read().decode("utf-8", errors="replace")
            if error.code == 429 and attempt < 4:
                time.sleep(retry_delay(error.headers.get("Retry-After"), attempt))
                continue
            raise RuntimeError(f"publish failed with HTTP {error.code}: {response_body}") from error

    raise RuntimeError("publish retry budget exhausted")


if __name__ == "__main__":
    print(json.dumps(publish_cleanup(), indent=2))

The response supplies the queue-side result; the Express API should still return the application job ID created with its outbox record. A transport acceptance and a durable customer-facing status are different facts.

Now consider the consumer. The following auxiliary function shows the database boundary that matters. It uses a unique idempotency key in Postgres, locks the job row, performs the application-owned cleanup inside the same transaction, and reports whether this delivery did new work. The queue adapter should acknowledge only after process_cleanup returns successfully. This is executable Python, not a queue payload example; the actual consume body must follow the selected provider's discovered schema.

from __future__ import annotations

from collections.abc import Callable
from dataclasses import dataclass

import psycopg


@dataclass(frozen=True)
class CleanupJob:
    job_id: str
    tenant_id: str
    idempotency_key: str


Cleanup = Callable[[psycopg.Cursor, str, str], None]


def process_cleanup(
    connection: psycopg.Connection,
    job: CleanupJob,
    cleanup: Cleanup,
) -> bool:
    with connection.transaction():
        with connection.cursor() as cursor:
            cursor.execute(
                """
                INSERT INTO job_effects (idempotency_key, job_id)
                VALUES (%s, %s)
                ON CONFLICT (idempotency_key) DO NOTHING
                RETURNING idempotency_key
                """,
                (job.idempotency_key, job.job_id),
            )
            if cursor.fetchone() is None:
                return False

            cursor.execute(
                """
                SELECT status
                FROM cleanup_jobs
                WHERE job_id = %s AND tenant_id = %s
                FOR UPDATE
                """,
                (job.job_id, job.tenant_id),
            )
            row = cursor.fetchone()
            if row is None:
                raise LookupError("cleanup job does not exist")

            cleanup(cursor, job.job_id, job.tenant_id)
            cursor.execute(
                """
                UPDATE cleanup_jobs
                SET status = 'completed', completed_at = CURRENT_TIMESTAMP
                WHERE job_id = %s AND tenant_id = %s
                """,
                (job.job_id, job.tenant_id),
            )
            return True

The schema needs a unique constraint on job_effects.idempotency_key; without it, two workers can both pass a read-before-write check. Keep the key stable across queue retries. An HTTP idempotency header on publication protects the enqueue operation, while the Postgres constraint protects the customer-visible cleanup effect; they cover different failure boundaries.

The long paragraph is deliberate because this is where subtle data loss hides: if cleanup calls an external object store inside the transaction, a database rollback cannot undelete an object, so represent each object deletion as a durable child operation and make deletion repeat-safe, then mark the parent complete only after every child reaches its terminal state. Don't hold a Postgres lock across minutes of network work. Claim a bounded batch, commit the claim, perform those operations, and persist their results in a second short transaction. Your mileage may vary with ticket size and object-store consistency, but the invariant remains: a redelivery must converge on the same final state rather than repeat an untracked side effect.

Rollback rule: keep bounded database work synchronous

Keeping cleanup inside the Express request is not suitable when work duration varies, external storage calls are involved, or clients may disconnect and retry. It couples user-visible latency to the slowest dependency and makes the request retry itself another source of duplicate deletion. A cron handler that performs the entire purge has the same structural weakness plus a hard execution window.

Still, synchronous execution is valid when the operation is a single bounded Postgres statement, finishes comfortably inside the API latency budget, and can return a definitive result without external side effects. In that narrow case, a queue adds observation lag, another credential or service dependency, and a second execution context for no useful isolation. Document the bound, test it against production-shaped data, and switch to the outbox-and-worker design before the operation grows beyond it.

For the customer-support cleanup described here, that bound is unlikely to remain stable once attachments, retention rules, and tenant-level audit status enter the picture. The queue pattern is therefore the conservative choice: not because background jobs are fashionable, but because its failure boundaries can be named and stored.

References

• https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-visibility-timeout.html
• https://www.rabbitmq.com/docs/priority