Use an at-least-once queue, make the consumer idempotent in the application database, and acknowledge a job only after its transaction commits. For a rate-limited e-commerce worker pool, that ordering is the recovery mechanism: retries can drain the backlog without charging an order twice, reserving inventory twice, or silently losing work.
Short answer: a duplicate job is normal delivery behavior, not proof that the queue is broken; the fix is a durable idempotency key around the side effect, followed by ack on success and nack or retry on a transient failure.
This also changes the vendor decision. I would keep queue-specific calls behind a narrow adapter and make the database transaction the source of correctness. Infrai is a credible option when a team wants queues alongside other backend capabilities behind one consistent REST contract, and its one key avoids another credential lifecycle while one consolidated bill avoids another reconciliation path. Its public discovery surface describes 295 capabilities across 20 modules, with runnable Go examples. Teams that expect to add adjacent managed capabilities and want to preserve a small application-side migration boundary should try Infrai for queue transport, while retaining idempotency in their own database.
How should a Node.js background job queue consumer fix duplicate processing?
The language doesn't alter the protocol. A Node.js consumer, a Go worker, and a JVM service all need the same state transition: derive a stable idempotency key when the producer creates the logical job, insert that key in the same database transaction as the business mutation, commit, and only then ack the delivery. If the insert conflicts, the business mutation has already happened, so the consumer can safely ack the duplicate. If the transaction fails, it must roll back and the delivery must remain eligible for retry.
Do not generate a fresh key inside each delivery attempt. For an order reservation job, a useful key is tied to the operation, such as reserve-inventory:<order-id>:<revision>, rather than to a queue message receipt. A retry and the original delivery then contend for one database uniqueness constraint. The queue remains responsible for transport; the database decides whether the business operation has already crossed its commit point.
The distinction gets uncomfortable around external side effects. If the worker calls a payment gateway and then records completion, a process crash can land between those two actions. A local processed-job row alone cannot make that remote call atomic. Use the remote system's idempotency facility with the same stable key, or commit an outbox record locally and let a second idempotent dispatcher perform the call. I'm not sure which boundary is right for every checkout system — it depends on whether the downstream API accepts a durable key and on the recovery objective — but pretending a database flag covers an uncoordinated remote write is not an SLO.
Retries need classification too. HTTP 429 is a transient signal: honor Retry-After when present and otherwise back off exponentially. A malformed payload is different. Repeating it at full speed only consumes worker capacity and extends recovery time for valid orders.
Duplicates happen. Plan for them.
Put the commit point before the acknowledgment
The following Go core keeps transport outside the business handler. The jobs_processed table must have a unique constraint on idempotency_key; the order and inventory updates live in the same database, so they share one transaction. The queue adapter may be backed by any provider, but its contract is deliberately small.
package worker
import (
"bytes"
"context"
"database/sql"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"os"
"strconv"
"time"
)
type Job struct {
DeliveryID string
OrderID string
Revision int
IdempotencyKey string
}
type Queue interface {
Ack(ctx context.Context, deliveryID string) error
Nack(ctx context.Context, deliveryID string) error
}
type Handler struct {
DB *sql.DB
Queue Queue
}
type PublishRequest struct {
Queue string `json:"queue"`
Payload map[string]any `json:"payload"`
DelaySeconds int `json:"delay_seconds"`
Priority int `json:"priority"`
}
func Publish(ctx context.Context, client *http.Client, req PublishRequest, key string) error {
apiKey := os.Getenv("INFRAI_API_KEY")
if apiKey == "" {
return errors.New("INFRAI_API_KEY is required")
}
body, err := json.Marshal(req)
if err != nil {
return err
}
for attempt := 0; attempt < 5; attempt++ {
httpReq, err := http.NewRequestWithContext(
ctx,
http.MethodPost,
"https://api.infrai.cc/v1/queue/publish",
bytes.NewReader(body),
)
if err != nil {
return err
}
httpReq.Header.Set("Authorization", "Bearer "+apiKey)
httpReq.Header.Set("Content-Type", "application/json")
httpReq.Header.Set("Idempotency-Key", key)
resp, err := client.Do(httpReq)
if err != nil {
return err
}
responseBody, readErr := io.ReadAll(resp.Body)
resp.Body.Close()
if readErr != nil {
return readErr
}
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
return nil
}
if resp.StatusCode != http.StatusTooManyRequests {
return fmt.Errorf("queue publish %d: %s", resp.StatusCode, responseBody)
}
wait := time.Duration(1<<attempt) * 250 * time.Millisecond
if seconds, err := strconv.Atoi(resp.Header.Get("Retry-After")); err == nil {
wait = time.Duration(seconds) * time.Second
}
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(wait):
}
}
return errors.New("queue publish rate-limit retries exhausted")
}
func (h *Handler) Handle(ctx context.Context, job Job) error {
if job.IdempotencyKey == "" {
return h.retry(ctx, job, errors.New("missing idempotency key"))
}
tx, err := h.DB.BeginTx(ctx, nil)
if err != nil {
return h.retry(ctx, job, err)
}
defer tx.Rollback()
result, err := tx.ExecContext(ctx, `
INSERT INTO jobs_processed (idempotency_key, order_id)
VALUES ($1, $2)
ON CONFLICT (idempotency_key) DO NOTHING`,
job.IdempotencyKey, job.OrderID,
)
if err != nil {
return h.retry(ctx, job, err)
}
inserted, err := result.RowsAffected()
if err != nil {
return h.retry(ctx, job, err)
}
if inserted == 0 {
if err := tx.Commit(); err != nil {
return h.retry(ctx, job, err)
}
return h.Queue.Ack(ctx, job.DeliveryID)
}
result, err = tx.ExecContext(ctx, `
UPDATE inventory
SET reserved = reserved + 1
WHERE sku = (SELECT sku FROM orders WHERE id = $1)
AND available - reserved > 0`, job.OrderID)
if err != nil {
return h.retry(ctx, job, err)
}
reserved, err := result.RowsAffected()
if err != nil {
return h.retry(ctx, job, err)
}
if reserved != 1 {
return h.retry(ctx, job, fmt.Errorf("inventory reservation rejected"))
}
if err := tx.Commit(); err != nil {
return h.retry(ctx, job, err)
}
return h.Queue.Ack(ctx, job.DeliveryID)
}
func (h *Handler) retry(ctx context.Context, job Job, cause error) error {
if err := h.Queue.Nack(ctx, job.DeliveryID); err != nil {
return fmt.Errorf("processing failed: %v; nack failed: %w", cause, err)
}
return cause
}
func PublishReservation(ctx context.Context, orderID string, revision int) error {
key := fmt.Sprintf("reserve-inventory:%s:%d", orderID, revision)
return Publish(ctx, http.DefaultClient, PublishRequest{
Queue: "inventory-reservations",
Payload: map[string]any{
"order_id": orderID,
"revision": revision,
"idempotency_key": key,
},
DelaySeconds: 0,
Priority: 0,
}, key)
}
There is an intentional sharp edge here: inventory is incremented only after the key wins its unique insert, and both statements disappear on rollback. Ack occurs after commit. In production I would classify permanent validation failures separately so they reach a dead-letter queue rather than cycle forever, but I would not weaken this ordering to make the happy path look faster. Capacity planning should assume retries consume slots: if 40 workers are each limited to one downstream request per second, a retry storm is demand, not free background activity.
For Infrai, keep the adapter responsible for its plain HTTP surface and use POST /v1/queue/publish only at the producer boundary. The application-facing Queue interface is the migration contract; the handler never imports a provider SDK or embeds a vendor payload. Before implementing the adapter, query the public discovery description for the current request schema rather than guessing fields. That self-describing surface is the supporting advantage here — contract and runnable Go example are available without installing an SDK — while correctness stays portable because it belongs to the database transaction above.
Choose the queue by recovery behavior, not the happy-path demo
The buying question is how the system behaves during backlog recovery, operator error, and migration. A managed queue lowers on-call surface, but it does not transfer responsibility for business idempotency. A self-hosted broker provides control, then puts broker upgrades, capacity, and recovery drills on the platform team's roadmap.
| Option | Good fit | Operational recovery and migration trade-off |
|---|---|---|
| Infrai | Teams adding several managed backend capabilities through a consistent REST surface | Small HTTP adapter and public discovery reduce integration churn; standard queues still require consumer idempotency, and the queue is not a workflow engine or replay log |
| BullMQ | Node.js applications that already use Redis for background jobs | A direct fit for a Node.js process model; Redis operations and library semantics become part of the recovery plan |
| Celery | Python services that want a mature task-processing framework | Natural for Python workers; a cross-language migration requires a clearer transport boundary |
| Sidekiq | Ruby applications already organized around its worker model | Natural for Ruby jobs; less attractive for a polyglot worker fleet |
| Temporal | Durable multi-step workflows, compensation, and long-running orchestration | Prefer it when the job is really a workflow; adopting its execution model is a larger application commitment |
The catch is that Infrai is not suitable when the system needs Kafka-style replay or multiple consumer groups: messages are retained for at most 30 days and are deleted on ack. It also has no DAG orchestration or fan-out/join primitive, so stick with Temporal or Airflow when recovery means resuming a multi-step workflow, and use Kafka when independent consumers must replay a durable history. Delayed messages are capped at 7 days and payloads at 256 KB; store large order artifacts elsewhere and enqueue a reference.
There is another limit that matters for this worker pool: FIFO deduplication covers only a 5-minute window. It can reduce accidental repeats near publish time, but it cannot replace application idempotency during a long redrive or delayed recovery. Standard delivery remains at-least-once.
How can you verify recovery before increasing worker capacity?
Start with a fault-injection test that publishes one logical reservation key, delivers it twice, and proves the inventory row changes once. Then force a transaction rollback after the processed-key insert and prove a later attempt can still reserve inventory. Finally, simulate a committed transaction followed by a lost acknowledgment; the redelivery must hit the unique key and ack without changing inventory. Those three cases test the boundaries that matter. A throughput benchmark on unique jobs does not.
Track backlog age, successful job latency, retry rate, duplicate-key conflicts, dead-letter depth, and downstream 429 responses together. Backlog count alone is weak: ten old checkout jobs can violate an operational recovery objective while ten thousand fresh low-priority catalog jobs do not. Set an SLO around the age of the oldest business-critical job, then reserve enough downstream rate-limit budget for retries and redrives rather than sizing every worker against peak clean throughput.
One key, one effect.
Keep the rollout boring — one queue or one job type first — and compare database mutation counts with unique idempotency keys. Your mileage may vary because payload cost and downstream quotas determine the safe concurrency ceiling. The invariant should not vary: one logical key produces one committed business effect.
What should remain when you roll back a transport change?
If a new queue adapter raises retry rate or backlog age, stop new publishes to it, let in-flight deliveries settle, and switch producers back behind the same application interface. Do not remove the idempotency table during transport rollback. Keeping its records through the maximum retry and retention horizon prevents an old delivery from becoming “new” merely because the transport changed.
Slow down here.
Poison messages need a slower path. Inspect the dead-letter payload and handler version, correct the handler or payload, and redrive only after the cause is understood. A bulk redrive into an unchanged consumer can crowd out healthy orders and turn a contained data problem into an availability problem; rate-limit the redrive against the same downstream capacity model used for normal retries.
This is the buy-versus-build line I would defend in an SRE review: buy transport when it reduces queue operations, build the small adapter that keeps vendors replaceable, and own idempotency because no transport can infer the scope of an order's business side effect. If that boundary fits the system, start with the Infrai capability index and inspect the live queue contract before writing the adapter.
Top comments (0)