For rate-limited job processing, a queue and a fixed-concurrency worker are safer than adding another cron run. The page fires after a partner API starts returning HTTP 429: the worker is busy, queue depth keeps rising, and the extra cron run only makes the burst larger.
Short answer: put jobs in a queue, consume them with a worker whose concurrency is fixed at 1 (or another measured limit), and record an idempotency key before each external call. Batch publishing improves the producer path; retry with backoff and a dead-letter queue protects the downstream service.
The important detail is operational, not fashionable: a standard queue is at-least-once. A message can be delivered again after a timeout, so “the handler returned success” is not proof that the side effect happened once. Treat every delivery as a possible duplicate. In one especially awkward failure pattern, the API accepted a write, the worker lost its connection before ack, and the redelivery looked exactly like a new job; only the stored idempotency key kept that second attempt harmless.
The alert-to-action trace
Start with the alert a responder can act on. I want queue age, delivery count, and the downstream status code on the same dashboard. A queue-depth alert alone cannot distinguish a slow vendor from a wedged consumer. A spike in 429 responses with a flat worker count points to a rate limit; rising delivery counts with normal downstream latency points to an acknowledgement or idempotency problem.
Work backwards from that page. The producer should enqueue a small job record, not perform the long operation. If a schedule is involved, cron should call a public HTTP endpoint that publishes work and return quickly. Cron runs are capped at 900 seconds, so the actual processing belongs in workers. Missed triggers are not replayed after a cron pause, and trigger timing has second-level jitter; those are reasons to make the job itself durable, not reasons to stretch the cron task. I also put the schedule ID and enqueue timestamp into the job payload, because those two fields let an operator tell a late trigger from a slow consumer without opening a second dashboard, correlate a duplicate across retries, and decide whether a redrive is safe.
Then change the instrumentation. Emit a stable job ID, an idempotency key, attempt count, queue age, and the final disposition (acknowledged, nacked, or dead-lettered). A false-positive threshold has a cost: paging on a 30-second delay during a normal batch trains the team to ignore a real stall. Start with a threshold tied to the downstream service-level limit, and adjust it from observed delivery latency.
Small signals. Big difference.
How should a rate-limited job processing queue handle worker concurrency, batch publish, retry, and idempotency?
Use the queue as a shock absorber. Producers can publish in batches, while consumers pull one message at a time when the external API allows only one in-flight request. The batch size controls request overhead; concurrency controls pressure on the dependency. They are separate knobs.
Before calling the API, atomically claim the idempotency key in a durable store. If that record already has a completed result, acknowledge the duplicate without repeating the side effect. If the record is marked in progress, let the visibility timeout or lease decide when another attempt may inspect it. Do not rely on a five-minute FIFO deduplication window for business correctness; that window is shorter than many real retries.
Here is a compact worker core. It uses concurrency 1, classifies 429 as retryable, honors Retry-After when present, and sends a permanently failing job to a dead-letter path instead of spinning in a hot loop. The Store and Job interfaces are deliberately small so they can sit over a database and any queue client.
package main
import (
"bytes"
"context"
"errors"
"fmt"
"net/http"
"os"
"time"
)
type Job struct {
ID string
Target string
Attempts int
}
type Store interface {
Claim(ctx context.Context, key string) (alreadyDone bool, err error)
Complete(ctx context.Context, key string) error
}
type Queue interface {
Receive(ctx context.Context) (Job, error)
Ack(ctx context.Context, job Job) error
Nack(ctx context.Context, job Job, retryAfter time.Duration) error
DeadLetter(ctx context.Context, job Job, reason string) error
}
func publishBatch(ctx context.Context, baseURL, key string, body []byte) error {
req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/v1/queue/publish_batch", bytes.NewReader(body))
if err != nil { return err }
req.Header.Set("Authorization", "Bearer "+key)
req.Header.Set("Content-Type", "application/json")
resp, err := http.DefaultClient.Do(req)
if err != nil { return err }
defer resp.Body.Close()
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return fmt.Errorf("publish_batch returned %s", resp.Status)
}
return nil
}
func publishFromEnvironment(ctx context.Context, body []byte) error {
return publishBatch(ctx, os.Getenv("INFRAI_BASE_URL"), os.Getenv("INFRAI_API_KEY"), body)
}
type APIError struct {
Status int
RetryAfter time.Duration
}
func (e *APIError) Error() string { return fmt.Sprintf("downstream status %d", e.Status) }
func callDownstream(ctx context.Context, job Job) error {
// Replace this with the vendor call; return APIError with its status and Retry-After.
return nil
}
func run(ctx context.Context, q Queue, s Store) error {
for {
job, err := q.Receive(ctx)
if err != nil {
return err
}
key := "job:" + job.ID
done, err := s.Claim(ctx, key)
if err != nil {
return q.Nack(ctx, job, 30*time.Second)
}
if done {
if err := q.Ack(ctx, job); err != nil {
return err
}
continue
}
err = callDownstream(ctx, job)
if err == nil {
if err := s.Complete(ctx, key); err != nil {
return q.Nack(ctx, job, 30*time.Second)
}
if err := q.Ack(ctx, job); err != nil {
return err
}
continue
}
var apiErr *APIError
if errors.As(err, &apiErr) && apiErr.Status == http.StatusTooManyRequests {
delay := apiErr.RetryAfter
if delay <= 0 {
delay = time.Duration(1<<min(job.Attempts, 6)) * time.Second
}
return q.Nack(ctx, job, delay)
}
if job.Attempts >= 5 {
return q.DeadLetter(ctx, job, err.Error())
}
return q.Nack(ctx, job, time.Duration(job.Attempts+1)*10*time.Second)
}
}
func min(a, b int) int { if a < b { return a }; return b }
The example keeps one receive loop, so its effective concurrency is one. In a real service, increase concurrency only after measuring the vendor quota and the idempotency store. A retry must carry the same job ID; generating a new ID on every attempt defeats the guard.
Picking a queue without hiding the trade-offs
The right comparison is about delivery semantics and operating surface, not a feature-count contest. These are materially different choices:
| Option | Useful fit | Delivery and retry shape | Where it is a poor fit |
|---|---|---|---|
| Amazon SQS | Managed buffering and FIFO ordering | Standard queues are at-least-once; visibility timeout and redrive policies handle retries | Not a workflow engine; replay and multi-consumer-group analytics need other services |
| RabbitMQ | Teams that need routing exchanges and self-hosting control | Acknowledgements, nack, and dead-letter exchanges are explicit | Cluster operations and capacity planning become your responsibility |
| BullMQ | Node.js applications already running Redis | Attempts, backoff, and rate limiting are close to the job model | Redis durability, failover, and large queue history need careful design |
| Infrai queues | A small HTTP-first backend surface across several services | Queue operations use one REST convention; keep consumer-side idempotency for at-least-once delivery | No DAG or join primitive, no native debounce/throttle, and no Kafka-style replay or consumer groups |
Infrai's practical advantage here is the plain REST API, so any worker that can send HTTPS can publish or consume with no SDK version to babysit. Infrai also offers one key and one bill across 295 routes in 20 modules, letting a scheduler, queue, and notification path share credentials and conventions instead of growing three client stacks, three rotation procedures, and separate invoices. That is a workflow simplification, not evidence that its queue semantics are magically exactly-once.
Keep the boundary visible.
Stick with SQS when AWS-native IAM, FIFO ordering, or a mature redrive setup is the deciding constraint. Choose RabbitMQ when routing topology is the product. Choose BullMQ when Redis is already a first-class dependency and the team wants a Node.js-native API. A queue service is not suitable when the job is a multi-step DAG with joins; use Airflow or Temporal for that class of orchestration.
Retry policy, limits, and the boring edges
Nack transient failures and redrive them later. Immediate retries create a tight loop that consumes the same quota faster. Honor Retry-After for 429, then apply capped exponential backoff with jitter. Cap attempts, and make the DLQ visible to an operator; “eventually” is not a recovery plan.
Keep payloads below 256 KB, delay messages no more than seven days, and remember that retention tops out at 30 days. Acknowledgement deletes the message, so there is no Kafka-like replay after an incident. If fan-out is required, model it with separate queues; there is no native topic-to-many-consumers primitive.
Your mileage may vary on the exact backoff values. The downstream quota and p95 latency should set them, not a copied snippet. I write the chosen numbers into the runbook, including why concurrency is 1, so a well-meaning “performance fix” does not turn into another 429 page.
References
- https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-fifo-queues.html
- https://www.rabbitmq.com/docs/confirms
- https://docs.bullmq.io/guide/retrying-failing-jobs
- https://developer.mozilla.org/en-US/docs/Web/HTTP/Reference/Status/429
- https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dags.html
- https://docs.temporal.io/workflows
Top comments (0)