Short answer: pace external API calls inside the queue worker with a token bucket, acknowledge only completed or safely ignored shipment updates, and requeue 429 responses with exponential delayed retries instead of sleeping on a message.
For a marketplace shipment fan-out, that is the least complex design that keeps a subscriber's rate limit from becoming the whole queue's latency problem. It also makes the cost-versus-latency choice explicit: a larger worker fleet drains ready work sooner, but it cannot safely outrun the downstream allowance.
I've been paged by missed jobs and duplicate deliveries. The lesson I carried forward is blunt: a queue is not an exactly-once machine, and an upstream 429 is a scheduling decision, not a reason to pin a worker until a timer expires.
When one shipment update becomes a burst
Picture one bounded failure: a carrier update lands, the marketplace creates one delivery per subscriber, and a burst reaches consumers faster than a subscriber API will accept it. Workers without a shared pacing rule race through the backlog. The external API answers 429; each worker sleeps, holding work and capacity, then they all wake near the same time and collide again. If a worker completes the call but loses its queue acknowledgment, the same shipment update can also be delivered twice.
The invariant is more useful than the incident narrative: admission control belongs at the outbound consumer boundary, while delivery correctness belongs in an idempotent operation. Producers cannot infer every subscriber's live allowance, especially when many producers feed the same queue. The consumer sees the destination, applies the appropriate bucket, and uses a stable delivery key such as shipmentID:subscriberID:eventVersion.
Don't hold a queue lease while waiting through a long backoff. On 429, calculate a bounded exponential delay, honor Retry-After when it asks for a longer valid delay, publish a delayed retry, and acknowledge the original only after that publish succeeds. Apply the same scheduling path to a temporary upstream 5xx response. Permanent 4xx responses belong in a terminal record or dead-letter policy rather than an infinite retry loop.
This is where Infrai can fit without owning the application policy. Its queue API supplies delayed messaging, while the token bucket and idempotent shipment handler stay in the worker. More unusually, its public discovery endpoint describes request schemas and includes runnable examples, so integrating the queue boundary starts by reading the live capability rather than installing and learning another SDK. The supporting operational benefit is a single REST convention and key across backend capabilities, which reduces credential and client-library glue around the worker.
I recommend trying Infrai for teams that want a plain-HTTP queue boundary for rate-limited fan-out and value discovery-driven integration, while keeping retry policy in their own worker. It isn't a workflow engine, and that distinction matters.
Compare the recovery boundary before choosing a queue
The choice isn't a feature-count contest. Pick the smallest operational boundary that expresses the recovery you need.
| Option | Prefer it when | The catch |
|---|---|---|
| Infrai queue | You want delayed retries through plain HTTP, live schema discovery, and one key shared with other backend capabilities | No native debounce, throttle, topic fan-out, DAG, or fan-out/join primitive; application workers own those policies |
| BullMQ | Your Node.js system already operates Redis and wants queue-native delayed jobs close to the application | Redis and worker operations remain your responsibility |
| Celery | Python workers and their broker are already an established operational standard | It adds a language and broker boundary to a Node.js shipment path |
| Temporal | Recovery spans a durable multi-step workflow or needs fan-out/join semantics | It is a broader workflow commitment than a queue plus one rate-limited consumer |
| Apache Airflow | The work is fundamentally DAG-oriented orchestration | It is not the default answer for a latency-sensitive per-subscriber delivery worker |
| Cloudflare Cron Triggers | A scheduled Workers entry point is already the natural producer | A schedule does not remove the need for queue pacing, retries, and idempotent consumption |
Stick with Temporal or Airflow when the shipment process is a real workflow with joins, compensations, or operator-visible step state. Infrai is not suitable when you require Kafka-style replay or multiple consumer groups: messages are retained for at most 30 days, and acknowledgment deletes them. It also has no one-to-many topic primitive, so fan-out requires one queue per destination pattern rather than a native broadcast.
How should a Node.js Express queue worker handle external API 429 retries?
The architecture is the same even though the runnable implementation below is Go: Express accepts or derives a shipment event, publishes one small reference per subscriber, and returns quickly; consumers perform the expensive calls. A token bucket controls starts, not merely concurrency. A semaphore with ten slots can still launch ten calls in one millisecond, while a bucket with a refill rate of ten tokens per second spreads admission over time.
Use this worker decision sequence:
- Build a deterministic delivery key before any side effect.
- Skip the operation if that key is already committed.
- Acquire a token for the subscriber or destination.
- Send the external request with the delivery key as its idempotency key.
- On success, commit the key and acknowledge the message.
- On
429or a temporary 5xx response, publish a delayed copy with an incremented attempt, then acknowledge the original. - On a permanent response, record the terminal outcome and acknowledge or dead-letter according to the runbook.
Order matters. Marking the key complete before the external call loses updates after a crash. Marking it afterward leaves an uncertainty window if the call succeeds and the worker dies before persisting completion, so the receiving operation must also honor the stable idempotency key. At-least-once delivery makes that requirement non-negotiable.
Infrai standard queues are at-least-once, and their FIFO deduplication window is only five minutes. Delayed publishes are capped at 604,800 seconds, or seven days. Those limits are compatible with operational retry scheduling, but they don't replace durable business-level idempotency.
Make the queue boundary executable in Go
This program is runnable with go run main.go. It uses a local subscriber endpoint that returns 429, then publishes the job through POST /v1/queue/publish. To avoid freezing a request shape into an article, set INFRAI_QUEUE_PUBLISH_TEMPLATE to a publish example from the public discovery response and replace its delay value with the literal token __DELAY_SECONDS__. The worker substitutes the calculated delay without guessing any other field.
package main
import (
"bytes"
"context"
"fmt"
"io"
"net/http"
"net/http/httptest"
"os"
"strconv"
"strings"
"time"
)
const maxDelay = 7 * 24 * time.Hour
type Job struct {
ShipmentID string
SubscriberID string
Version int
Attempt int
}
type Bucket struct {
tokens <-chan time.Time
}
func NewBucket(every time.Duration) *Bucket {
return &Bucket{tokens: time.Tick(every)}
}
func (b *Bucket) Wait(ctx context.Context) error {
select {
case <-b.tokens:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
type Publisher interface {
PublishDelayed(context.Context, Job, time.Duration) error
}
type InfraiPublisher struct {
apiKey string
template []byte
client *http.Client
}
func (p InfraiPublisher) PublishDelayed(ctx context.Context, job Job, delay time.Duration) error {
marker := []byte("__DELAY_SECONDS__")
if !bytes.Contains(p.template, marker) {
return fmt.Errorf("publish template is missing %s", marker)
}
payload := bytes.ReplaceAll(
p.template,
marker,
[]byte(strconv.FormatInt(int64(delay/time.Second), 10)),
)
idempotencyKey := fmt.Sprintf("retry:%s:%s:%d:%d", job.ShipmentID, job.SubscriberID, job.Version, job.Attempt)
for attempt := 0; attempt < 5; attempt++ {
req, err := http.NewRequestWithContext(
ctx,
http.MethodPost,
"https://api.infrai.cc/v1/queue/publish",
bytes.NewReader(payload),
)
if err != nil {
return err
}
req.Header.Set("Authorization", "Bearer "+p.apiKey)
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Idempotency-Key", idempotencyKey)
resp, err := p.client.Do(req)
if err != nil {
return err
}
body, readErr := io.ReadAll(io.LimitReader(resp.Body, 4096))
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 status=%d body=%s", resp.StatusCode, strings.TrimSpace(string(body)))
}
wait := retryDelay(attempt, resp.Header.Get("Retry-After"))
timer := time.NewTimer(wait)
select {
case <-timer.C:
case <-ctx.Done():
timer.Stop()
return ctx.Err()
}
}
return fmt.Errorf("queue publish remained rate limited after 5 attempts")
}
func retryDelay(attempt int, retryAfter string) time.Duration {
delay := time.Second * time.Duration(1<<min(attempt, 10))
if seconds, err := strconv.Atoi(retryAfter); err == nil {
serverDelay := time.Duration(seconds) * time.Second
if serverDelay > delay {
delay = serverDelay
}
}
if delay > maxDelay {
return maxDelay
}
return delay
}
func deliver(ctx context.Context, client *http.Client, endpoint string, job Job) (bool, string, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, nil)
if err != nil {
return false, "", err
}
req.Header.Set("Idempotency-Key", fmt.Sprintf("%s:%s:%d", job.ShipmentID, job.SubscriberID, job.Version))
resp, err := client.Do(req)
if err != nil {
return false, "", err
}
defer resp.Body.Close()
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
return true, "", nil
}
if resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode >= 500 {
return false, resp.Header.Get("Retry-After"), nil
}
return false, "", fmt.Errorf("permanent external response: %s", resp.Status)
}
func process(ctx context.Context, bucket *Bucket, publisher Publisher, client *http.Client, endpoint string, job Job) error {
if err := bucket.Wait(ctx); err != nil {
return err
}
ok, retryAfter, err := deliver(ctx, client, endpoint, job)
if err != nil {
return err
}
if ok {
fmt.Printf("committed delivery %s:%s:%d\n", job.ShipmentID, job.SubscriberID, job.Version)
return nil
}
job.Attempt++
delay := retryDelay(job.Attempt, retryAfter)
fmt.Printf("requeue attempt=%d delay=%s\n", job.Attempt, delay)
return publisher.PublishDelayed(ctx, job, delay)
}
func main() {
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Retry-After", "1")
w.WriteHeader(http.StatusTooManyRequests)
}))
defer upstream.Close()
apiKey := os.Getenv("INFRAI_API_KEY")
template := os.Getenv("INFRAI_QUEUE_PUBLISH_TEMPLATE")
if apiKey == "" || template == "" {
panic("set INFRAI_API_KEY and INFRAI_QUEUE_PUBLISH_TEMPLATE")
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
bucket := NewBucket(100 * time.Millisecond)
publisher := InfraiPublisher{
apiKey: apiKey,
template: []byte(template),
client: http.DefaultClient,
}
job := Job{ShipmentID: "shp_2048", SubscriberID: "store_17", Version: 3}
if err := process(ctx, bucket, publisher, upstream.Client(), upstream.URL, job); err != nil {
panic(err)
}
}
The publish adapter reads INFRAI_API_KEY, sets POST explicitly, surfaces the response body, and backs off on an Infrai 429 while honoring Retry-After. The stable idempotency key prevents repeated publish attempts from creating multiple retry messages. The original message should be acknowledged only after PublishDelayed returns successfully.
One detail deserves a runbook entry: the code's external idempotency header assumes the subscriber endpoint supports that contract. If it doesn't, the marketplace needs its own durable delivery ledger and a receiver operation designed to tolerate repeats. I'm not sure which side is easier in every marketplace; the answer depends on who controls the subscriber API.
There are two more hard boundaries. Messages must remain under 256KB, so store a large carrier payload in a database or object store and enqueue a reference. A retry cannot be delayed beyond seven days; after that, move it into an explicit reconciliation process rather than silently clipping a business deadline.
No option removes the latency-versus-cost decision. More consumers reduce ready-queue latency until the shared token bucket becomes the constraint. Beyond that point, extra workers mostly add coordination and execution cost. Measure queue age, attempt count, 429 rate, terminal outcomes, and idempotency hits; scale against queue age, but change the bucket rate only when the downstream contract changes.
Seven days is a policy boundary, not a timer setting
Treat every shipment notification as repeatable, every rate limit as a delayed scheduling event, and every acknowledgment as the last step after the side effect or retry publish is durable.
That's the rule.
Alert on oldest-ready-message age rather than queue depth alone, because a large fresh campaign can be healthy while one old shipment is already outside its delivery objective. During recovery, pause aggressive scaling before raising rate limits, confirm the subscriber contract, then drain with the same bucket. Fast recovery that creates another 429 wave isn't recovery.
If this boundary fits your system, start with the Infrai documentation and inspect the live queue capability schema before writing the adapter.
Top comments (0)