ValorD33

Posted on Oct 29

Subscriber Shipment Fan-Out: 4 Recovery Boundaries for Rate-Limited Queue Consumption

#queues #ratelimiting #backend #reliability

The expensive part of a shipment-update fan-out is rarely the queue write. It is the work that remains retained while a downstream provider says “slow down,” plus the duplicate sends and manual investigation created by a poor recovery path. For a media SaaS application notifying 80,000 subscribers about a physical shipment, I would choose a queue by how precisely it lets workers stop, retry, quarantine, and resume—not by its peak publish rate.

Short answer: use a queue with an explicit visibility or acknowledgment boundary, durable delayed retry behavior, and a dead-letter path you can inspect. Put the dynamic limiter in the consumer layer, keyed by the destination or provider policy. Keep the shipment event immutable, make each notification idempotent, and measure the age of the oldest ready message. Those decisions matter more than a queue brand.

What is the bill actually paying for?

Suppose a fulfillment event fans out to 80,000 subscriber notifications. The first-order cost is the number of delivery attempts. The operational cost is the number of messages kept while the delivery rate changes. A bad design turns one provider limit into three queues of work: ready messages, delayed retries, and duplicates that were published again because the worker could not tell whether the previous attempt had committed.

That is a retention problem before it is a throughput problem. A worker consuming 200 messages per second against a provider that has temporarily accepted only 40 creates backlog at 160 messages per second. In five minutes, 48,000 messages are waiting. Increasing concurrency makes the arithmetic worse and can trigger a longer provider cooldown.

The queue choice shows up during recovery.

In practice, I would chart that backlog against the retention policy rather than celebrate a high consumer count. If retries remain visible for 30 minutes, the storage footprint is not just 48,000 payloads: it includes delivery metadata, attempt history, delayed scheduling state, and any copy held for reconciliation. If the same shipment update is fanned out to email and SMS, the unit of work is also different from the source event, so a single “shipment sent” counter hides the duplicate and missing-recipient cases that create support work later. This is why I quantify the retained work before changing concurrency. It tells me whether the fix is a limiter, a smaller payload, a shorter history, or a different recovery boundary.

I would keep three identifiers on every delivery task:

shipment_event_id, stable for the source event
subscriber_id, stable for the recipient
attempt, incremented only when a delivery attempt is actually made

The task can then be deduplicated as (shipment_event_id, subscriber_id), while the attempt count remains useful for policy. A queue that offers fast redelivery but no inspectable delay state may look attractive in a benchmark and still make this incident expensive to explain.

There is a deliberate loss here. I would not retain every successful provider response forever in the queue. The durable record belongs in an event or delivery ledger with a bounded retention policy; the queue should hold actionable work. When a team discards that ledger as “duplicate storage,” it gives up the ability to answer a subscriber who asks why an update arrived twice.

How should a message queue handle dynamic rate limiting?

Dynamic rate limiting has two clocks. The queue has an availability clock: when a task may be consumed again. The provider has a policy clock: when another request is allowed. Mixing them creates a familiar failure: a worker receives a task, gets a rate-limit response, and immediately requeues it. The same hot task wins the race repeatedly.

The consumer should own a small control loop. It reads a bounded batch, applies the current permit budget, acknowledges only work whose outcome is durable, and moves a rejected task into a retry state with a calculated delay. A token bucket is a useful mental model: refill permits from observed capacity, spend one per external request, and cap the bucket so a long pause does not produce a burst.

from dataclasses import dataclass
from time import monotonic


@dataclass
class TokenBucket:
    capacity: float
    refill_per_second: float
    tokens: float
    checked_at: float

    def take(self, now: float | None = None) -> bool:
        current = monotonic() if now is None else now
        elapsed = max(0.0, current - self.checked_at)
        self.tokens = min(
            self.capacity,
            self.tokens + elapsed * self.refill_per_second,
        )
        self.checked_at = current
        if self.tokens < 1.0:
            return False
        self.tokens -= 1.0
        return True

This bucket is local to one process, so it is not a distributed limiter by itself. That limitation is useful to state plainly. With ten workers, ten local buckets can still exceed a provider limit. A shared counter, lease, or coordinator is required when the policy is global; the queue should not be asked to impersonate that coordinator.

For retries, use exponential backoff with jitter and a maximum delay. A Retry-After value from an upstream HTTP response should take precedence when it is valid and within an agreed ceiling. A malformed or absurd value is not a reason to park all 80,000 tasks indefinitely. Record the raw response separately, clamp the scheduling decision, and preserve enough metadata to investigate it.

How do four recovery boundaries change the design?

The meaningful comparison is the recovery boundary, not a feature checklist.

An acknowledgment-based broker such as RabbitMQ can hold a message unacknowledged while a consumer works, and its dead-letter exchange documentation describes routing messages that are rejected, expired, or exceed a queue length limit. That gives the application a visible place to route poison work, but the team still has to design acknowledgment timing, retry delay, and dead-letter monitoring. Acknowledging before the delivery ledger commits creates loss; acknowledging only after a long remote call makes the consumer’s failure window larger.

Redis Streams consumer groups use explicit acknowledgment through XACK, and pending entries can be inspected and claimed by another consumer. That is a useful recovery shape when a team already operates Redis and wants ownership of pending work to be visible. It does not remove the need for an idempotency ledger or a policy for entries that are repeatedly claimed.

Amazon SQS uses a visibility timeout: a received message is hidden for a period and becomes visible again if it is not deleted. Its documentation also describes dead-letter queues and redrive controls. This is a straightforward model for retry recovery, but the visibility timeout must cover normal processing and be extended when processing can exceed it. Otherwise two workers can legitimately process the same shipment task after the timeout expires.

Kafka consumer groups track offsets rather than individual queue acknowledgments. That can suit an append-only shipment event stream where consumers replay partitions and maintain their own delivery state. It is a different operational contract from a task queue: a failed notification is not automatically an individually delayed item, so retry scheduling and per-subscriber progress belong in application state or a separate work queue.

These are not rankings. They are boundaries to test. The right choice is the one whose failure semantics the on-call team can observe and rehearse. If the requirement is “pause one provider for 30 seconds while unrelated providers continue,” a single global consumer pause is already the wrong abstraction, regardless of the queue’s advertised throughput.

What should the recovery path look like?

Partition the limiter by the thing that is actually constrained. For a media service, that might be provider_id, region, or notification channel. Do not partition only by queue if one provider’s limit is shared across email and SMS. Each partition needs its own permit state, retry schedule, and oldest-message metric.

The worker should classify outcomes before deciding what to do with the task:

Outcome	Queue action	Durable record
Accepted by provider	Acknowledge or delete	Mark delivered with an idempotency key
Temporary limit or network failure	Delay and retry	Record status and next attempt time
Permanent address or policy failure	Quarantine	Record final failure and reason
Unknown result after timeout	Reconcile before retry	Keep the key and request correlation ID

The unknown result is the uncomfortable row. If the connection drops after the provider accepts the request, a blind retry can duplicate the message. If the team treats every timeout as permanent, subscribers miss the update. A delivery ledger with a unique key cannot prove what a remote provider did, but it can prevent the application from creating its own duplicates and can direct uncertain cases to reconciliation.

I also want a separate control path for changing limits. Configuration should be versioned, validated, and applied with a timestamp. A value of zero should mean “do not send,” not “unbounded,” and an absent provider policy should fail closed for that partition. Those two edge cases have caused more dangerous bursts in real systems than sophisticated token-bucket math.

The queue metrics should tell a recovery story: ready depth, in-flight count, oldest message age, retry delay distribution, dead-letter count, delivery outcome by provider, and the number of tasks in an unknown state. A graph of total throughput alone can remain green while the oldest subscriber notification is six hours late.

What would I test before choosing?

I would run the same workload against each candidate with a fake provider that changes its limit during the shipment. Start at 200 accepted requests per second, drop to 40 for 90 seconds, return to 200, and inject timeouts after the provider has accepted a request. The exact numbers are test inputs, not promises about a real provider. The point is to force the queue and worker to reveal their recovery contract.

The test passes only if the system can answer these questions without reading application logs line by line:

How many tasks were attempted, delivered, delayed, quarantined, or left uncertain?
Can one constrained partition pause without starving other partitions?
After a worker disappears, where is its work and when can another worker take it?
Can an operator replay quarantined work without publishing a second shipment event?

I would also run a deployment during the pause, expire credentials, fill the retry store, and kill a worker after the remote call but before acknowledgment. The boring tests are valuable. A queue is operationally suitable when the answer survives those tests with bounded manual work.

The decision rule is simple: select the queue whose recovery boundary matches the team’s strongest guarantee, then put the dynamic limiter beside the consumer and keep delivery identity outside the queue. For shipment fan-out, retaining less history can reduce storage, but it also reduces the evidence available during a duplicate or missing-update investigation. That is the cost worth making explicit.

DEV Community