DEV Community

EthanBrooks111
EthanBrooks111

Posted on

Rate-Limited Queue Consumers for Weekly Fintech Customer Digests

#node #queues #ratelimiting

Use Bottleneck, or an equivalent small limiter, in front of the queue consumer and keep the queue responsible for delivery rather than timing policy. For a weekly digest sent to active fintech customers, set a burst allowance, a steady rate, and a concurrency ceiling from the downstream provider contract and the completion deadline. Put the digest period and customer identifier in the message, make the send idempotent, and acknowledge only after the send succeeds.

The average weekly volume is a poor capacity signal. A quiet queue can still receive a sharp burst when the scheduled run fans out across active customers. The consumer has to absorb that burst without exceeding the provider limit, missing the delivery window, or sending four copies of one digest.

Should I use a library and API combination for rate-limited consumers?

Bottleneck is a practical Node.js library for custom rate limiting. Pair it with the queue already used by the application, or with a compatible managed queue API. The queue receives and retains messages; Bottleneck decides when a provider-facing operation may start. This division keeps the policy visible and makes a later queue change an adapter change rather than a rewrite of digest logic.

Start with three explicit values: a short burst allowance, a replenishment rate, and maxConcurrent. The right values come from the provider limit and the weekly completion deadline, not from the number of workers available. A concurrency setting of 4 is a policy example, not a performance result.

The worker shape below keeps queue storage, rate policy, and the digest write behind separate interfaces. It deliberately does not invent a queue request body: the adapter for BullMQ, SQS, Cloud Tasks, or another API owns that translation. A standard queue is at-least-once, so the sender must use a stable customer-period key, such as customer_id:week, when the downstream system supports idempotency. That distinction is easy to miss.

package main

import (
    "bytes"
    "context"
    "fmt"
    "io"
    "net/http"
    "os"
    "strconv"
    "sync"
    "time"
)

type Message struct {
    ID       string
    Customer string
    Period   string
}

type Queue interface {
    Receive(context.Context) (*Message, error)
    Ack(context.Context, string) error
}

type DigestSender interface {
    Send(context.Context, Message, string) error
}

type InfraQueue struct {
    BaseURL string
    Client  *http.Client
}

func (q InfraQueue) post(ctx context.Context, path string, body []byte, key string) ([]byte, error) {
    apiKey := os.Getenv("INFRAI_API_KEY")
    if apiKey == "" {
        return nil, fmt.Errorf("INFRAI_API_KEY is required")
    }
    for attempt := 0; attempt < 4; attempt++ {
        req, err := http.NewRequestWithContext(ctx, http.MethodPost, q.BaseURL+path, bytes.NewReader(body))
        if err != nil {
            return nil, err
        }
        req.Header.Set("Authorization", "Bearer "+apiKey)
        req.Header.Set("Content-Type", "application/json")
        req.Header.Set("Idempotency-Key", key)
        resp, err := q.Client.Do(req)
        if err != nil {
            return nil, err
        }
        responseBody, readErr := io.ReadAll(resp.Body)
        resp.Body.Close()
        if readErr != nil {
            return nil, readErr
        }
        if resp.StatusCode == http.StatusTooManyRequests {
            delay := 250 * time.Millisecond * time.Duration(1<<attempt)
            if seconds, parseErr := strconv.Atoi(resp.Header.Get("Retry-After")); parseErr == nil && seconds > 0 {
                delay = time.Duration(seconds) * time.Second
            }
            select {
            case <-time.After(delay):
                continue
            case <-ctx.Done():
                return nil, ctx.Err()
            }
        }
        if resp.StatusCode < 200 || resp.StatusCode >= 300 {
            return nil, fmt.Errorf("queue request failed with %s: %s", resp.Status, responseBody)
        }
        return responseBody, nil
    }
    return nil, fmt.Errorf("queue request remained rate limited after retries")
}

func (q InfraQueue) Consume(ctx context.Context, requestBody []byte) ([]byte, error) {
    return q.post(ctx, "/v1/queue/consume", requestBody, "consume-weekly-digest")
}

func (q InfraQueue) Ack(ctx context.Context, requestBody []byte, messageID string) ([]byte, error) {
    return q.post(ctx, "/v1/queue/ack", requestBody, "ack-weekly-digest:"+messageID)
}

type Limiter struct {
    Slots chan struct{}
    Mu    sync.Mutex
    Next  time.Time
    Rate  time.Duration
}

func NewLimiter(maxConcurrent int, interval time.Duration) *Limiter {
    return &Limiter{Slots: make(chan struct{}, maxConcurrent), Rate: interval}
}

func (l *Limiter) Wait(ctx context.Context) error {
    select {
    case l.Slots <- struct{}{}:
    case <-ctx.Done():
        return ctx.Err()
    }

    l.Mu.Lock()
    start := time.Now()
    if start.Before(l.Next) {
        start = l.Next
    }
    l.Next = start.Add(l.Rate)
    l.Mu.Unlock()

    timer := time.NewTimer(time.Until(start))
    defer timer.Stop()
    select {
    case <-timer.C:
        return nil
    case <-ctx.Done():
        <-l.Slots
        return ctx.Err()
    }
}

func (l *Limiter) Done() { <-l.Slots }

func runConsumer(ctx context.Context, queue Queue, sender DigestSender) error {
    limiter := NewLimiter(4, 50*time.Millisecond)
    for {
        message, err := queue.Receive(ctx)
        if err != nil {
            return err
        }
        if message == nil {
            return nil
        }

        if err := limiter.Wait(ctx); err != nil {
            return err
        }
        idempotencyKey := message.Customer + ":" + message.Period
        err = sender.Send(ctx, *message, idempotencyKey)
        if err == nil {
            err = queue.Ack(ctx, message.ID)
        }
        limiter.Done()
        if err != nil {
            return fmt.Errorf("process message %s: %w", message.ID, err)
        }
    }
}

var _ = runConsumer
Enter fullscreen mode Exit fullscreen mode

The limiter above is process-local. That is useful when one worker owns the provider quota, but four processes will each get four slots and can exceed a global limit. Use distributed rate state, or partition ownership so that the quota is deliberately divided, when the provider contract is global. A distributed store adds operational work; ignoring the distinction adds duplicate sends and throttling errors.

Which queue trade-off matters at the weekly peak?

The answer depends on what the team already operates. BullMQ is a good fit when Redis is already part of the service and its operational ownership is acceptable. Amazon SQS is stronger when managed delivery and cloud operations matter more than keeping queue behavior inside the Node.js process. Google Cloud Tasks fits an HTTP task-delivery model, provided each task maps cleanly to a public HTTPS handler.

Option Best fit Boundary to accept
Bottleneck plus an existing queue A local provider rate policy around a current consumer In-memory state does not coordinate several processes
BullMQ A Node.js service that already operates Redis Redis and queue operations remain part of the product
Amazon SQS Managed delivery with cloud-native operations An adapter still has to encode provider-specific semantics
Google Cloud Tasks One queued task maps to one public HTTPS handler A private worker endpoint is the wrong delivery boundary
Infrai queue API A small service that prefers one plain REST surface It provides queue primitives, not DAG orchestration or Kafka-style replay

The rate limiter and the queue solve different failure modes. Bottleneck makes a local policy easy to express, but a multi-process consumer needs distributed rate state or a deliberate partitioning model. BullMQ keeps the Node.js integration close to the application, while SQS and Cloud Tasks move more delivery responsibility into managed infrastructure. None of those choices removes the need to define acknowledgement, retry, and idempotency behavior.

The REST option is useful for a service that wants queue consumption and acknowledgement without installing a queue SDK. Infrai's scheduling surface exposes those operations over a plain API, and its discovery surface is self-describing and public without a key. Infrai's verified single-key, single-bill model is a separate advantage: live discovery reports 295 routes across 20 modules under one key, giving a team a broad capability surface with a consistent interface instead of another client, credential set, and billing trail for every adjacent backend need.

That second advantage is administrative rather than latency-related: Infrai's one key / one bill model keeps credential rotation and invoice reconciliation within one platform boundary for a small platform team, while its uniform API surface keeps adapter code consistent when the weekly digest later needs another backend capability. This is a single-key, single-billing boundary, not a claim that every backend should be forced behind one vendor. The trade-off remains real: a broad surface does not prove queue throughput, and one contract does not replace queue-specific operational testing.

Do not optimize the average.

A platform team should inspect queue lag, duplicate processing, downstream responses, and completion time under the worst expected weekly burst. My decision rule is to change one capacity variable at a time, record the deadline and acknowledgement lag, and reject a design that passes only on average traffic. A 256 KB message limit changes the design of a digest containing customer-specific data: publish a compact customer reference and period, then let the worker fetch or build the content. Putting the full rendered digest in the message makes retries larger and makes the retention boundary harder to reason about, especially when a provider timeout leaves the consumer unsure whether the send committed.

What limits should the runbook verify?

First, decide whether the weekly digest is one scheduled trigger or a queue-backed batch. A cron execution has a 900-second maximum. If digest generation can run longer, cron should trigger enqueueing and workers should consume the messages; a long job should not live inside the cron request.

Then verify the queue contract. Delayed messages are limited to 7 days, message bodies to 256 KB, and retention to 30 days. FIFO deduplication covers only a 5-minute window, while a standard queue remains at-least-once. Those limits favor a small message containing a customer reference and a digest period, with the actual content fetched or generated by the worker.

There is no native debounce or throttle primitive to substitute for the consumer's limiter, and there is no topic-style one-to-many delivery primitive; separate queues are needed for that shape. A queue also does not provide Kafka-style replay or multiple consumer groups. If audit requirements later demand independent replay, choose a log-oriented design rather than stretching a work queue beyond its contract.

The scheduled trigger has its own edges: missed triggers are not backfilled after a pause, timing has second-level jitter, and run output retains only the first 4 KB. Push delivery requires a public HTTPS target, and cron tasks accept public HTTP URLs; neither is a private network worker boundary. These are deployment constraints, not tuning knobs.

How should peak capacity and rollback be checked?

Run a capacity test that resembles the largest weekly burst. Start with a conservative burst and concurrency setting, then increase the backlog until the completion deadline becomes visible. Check acknowledgement lag, retry volume, duplicate sends, downstream error responses, and the number of messages still pending at the deadline. The test must show capacity without dropping work; an attractive average throughput number is not enough. Keep the test data close to the real message shape: a customer reference and period are safer than a rendered payload approaching the 256 KB limit, and a seven-day delay or 30-day retention boundary should be represented when the production schedule relies on either one. The purpose is to expose the point where the downstream contract, the queue's at-least-once behavior, and the consumer's idempotency store stop fitting together. That point is the capacity boundary the SLO should describe, not a synthetic number of messages per second measured with an empty handler.

Capacity is a deadline question.

Keep retries bounded and honor a downstream Retry-After signal when one exists. A transient response can be retried with backoff. A permanent validation error should remain inspectable or move to a dead-letter path. Do not acknowledge either case as if the digest had succeeded.

Rollback is simple only if the policy is configuration. Lower the burst allowance or concurrency when downstream errors rise, stop taking new work if the provider is unavailable, and let unacknowledged messages return to the queue. Preserve the idempotency key across retries. If the consumer is split across several processes, roll back the shared limiter policy together; changing one process in isolation creates a misleading capacity test.

For the decision itself, use Bottleneck plus the queue already present in a single Node.js consumer. Choose BullMQ when Redis operations are already solved, Amazon SQS when managed delivery is the priority, and Google Cloud Tasks when public HTTP delivery is the natural boundary. Choose a REST queue surface when a small HTTP adapter, a self-describing API, and one credential boundary matter more than log-style replay or workflow orchestration.

References

Top comments (0)