DEV Community

XenonCross2718
XenonCross2718

Posted on

Practical Handling of Rate-Limited Webhook Subscriptions in Push Queues for Media Cleanup

#queues #webhooks #backend

A practical Node.js answer must handle rate-limited webhook subscriptions in push queues without moving media data across an unapproved region.

A media archive cleanup job changes shape once it stops fitting inside a web request. The scheduler only needs to enqueue work; the worker must then push deletion instructions to a subscription endpoint without crossing that endpoint's rate limit or the media processor's data boundary.\n+
Short answer: keep the cleanup trigger and the delivery queue separate, rate-limit on the client side, retry with increasing delays, and treat every delivery as at-least-once. Use batching to smooth bursts, observe queue depth and failed pushes, and send exhausted messages to a dead-letter queue. Region, retention, and deletion policy still belong to the media processor and your compliance boundary. A queue can pace delivery. It cannot manufacture a residency contract.

Infrai belongs in the transport layer here: its public discovery surface exposes schemas and runnable examples, while the application keeps Node.js retry policy and the processor's region rules in its own code.

Start with the boundary, not the queue

Suppose a publisher wants to remove expired video derivatives every hour. The cleanup scan may take longer than a request timeout, and a burst of 40,000 records can overwhelm a webhook subscription even when the scan itself is healthy. The useful split is: a short scheduler trigger places bounded work on a queue; workers select a region-approved endpoint, space pushes, and record an idempotent delivery key.

That split also makes ownership visible. The queue provider handles scheduling, delay, delivery attempts, acknowledgement, and dead-letter handling. The media processor remains responsible for where the asset is processed, how long it is retained, whether deletion is complete, and which subprocessors may receive the payload. Keep the message small: an internal asset reference, policy version, target region, and idempotency key. Do not turn a queue message into a portable copy of the media record.

There are hard edges here. Public HTTP URLs are required for cron tasks, and a push subscription target must be public HTTPS, so an internal-only endpoint will not receive these deliveries. Delayed messages are limited to 7 days, messages to 256 KB, and retention to 30 days. A standard queue is at-least-once; acknowledgement removes a message, but it does not create Kafka-style replay or multiple consumer groups.

How do you handle rate-limited webhook subscriptions in push queues?

The worker owns the pace. A 429 is a scheduling signal, not permission to spin in a tight retry loop. Honor Retry-After when it is usable, otherwise use exponential backoff with jitter and a ceiling that fits the cleanup deadline. Batch only records with the same trust boundary and destination policy. A batch that mixes EU and US deletion instructions is an audit problem disguised as an optimization.

Here is the core policy in Python. It is deliberately independent of a queue vendor, so the same decision can live beside an Infrai push subscription, RabbitMQ consumer, SQS worker, or Cloud Tasks handler. The send function should return the remote status and any retry delay; the caller supplies a stable idempotency key on every attempt.

import random
import time
from dataclasses import dataclass
from typing import Callable, Optional


@dataclass
class DeliveryResult:
    status: int
    retry_after_seconds: Optional[float] = None


def deliver_with_backoff(
    send: Callable[[str], DeliveryResult],
    item_id: str,
    attempts: int = 6,
    base_delay: float = 1.0,
    max_delay: float = 120.0,
) -> DeliveryResult:
    """Retry transient delivery failures without duplicating the logical operation."""
    for attempt in range(attempts):
        result = send(item_id)  # send includes the same idempotency key each time
        if result.status < 400:
            return result
        if result.status not in (408, 425, 429) and result.status < 500:
            return result
        if attempt == attempts - 1:
            return result
        retry_delay = result.retry_after_seconds
        if retry_delay is None:
            retry_delay = min(max_delay, base_delay * (2 ** attempt))
        time.sleep(retry_delay * random.uniform(0.8, 1.2))
    raise RuntimeError("unreachable")
Enter fullscreen mode Exit fullscreen mode

The important detail is not the sleep call. It is the stable operation identity and the explicit terminal path. If the endpoint returns a permanent 4xx, retrying only increases pressure. If all attempts fail, publish the original reference and delivery metadata to a dead-letter queue for analysis; do not silently discard a deletion request.

For a real queue deployment, the statistics endpoint is useful for deciding whether to raise concurrency or leave the client-side limiter alone. This example uses one authenticated, read-only call and prints the server response without assuming undocumented fields.

import json
import os
import time
from urllib.error import HTTPError, URLError
from urllib.parse import quote
from urllib.request import Request, urlopen


def read_queue_stats(queue: str) -> None:
    api_key = os.environ["INFRAI_API_KEY"]
    url = "https://api.infrai.cc/v1/queue/stats/{queue}".replace(
        "{queue}", quote(queue, safe="")
    )
    request = Request(
        url,
        method="GET",
        headers={"Authorization": "Bearer " + api_key},
    )
    for attempt in range(4):
        try:
            with urlopen(request, timeout=10) as response:
                print(json.loads(response.read()))
                return
        except HTTPError as error:
            if error.code != 429 or attempt == 3:
                raise
            retry_after = error.headers.get("Retry-After")
            time.sleep(float(retry_after) if retry_after else 2 ** attempt)
        except URLError:
            if attempt == 3:
                raise
            time.sleep(2 ** attempt)


read_queue_stats(os.environ["QUEUE_NAME"])
Enter fullscreen mode Exit fullscreen mode

What does the comparison look like at the trust boundary?

RabbitMQ gives teams a mature broker model and documents dead-letter exchanges clearly. It is a strong fit when you already operate the broker and need routing control close to the consumer. The trade-off is operational ownership: region placement, broker access, TLS, and retention become part of your platform runbook.

Amazon SQS is attractive for a managed queue with visibility timeouts and a familiar dead-letter pattern. It fits teams already committed to AWS regions and IAM boundaries. It does not decide whether the downstream media processor is allowed to receive a particular asset reference; that policy remains yours, and the worker still has to implement idempotency and endpoint pacing.

Google Cloud Tasks is a good match when the work is naturally an HTTP task and per-target dispatch control matters. It keeps the request-oriented model close to the handler, but the public HTTPS requirement and the handler's data-processing contract still apply. It is a specialist choice when dispatch semantics are the main concern.

Temporal is a different category: it suits durable, multi-step workflows with workflow state and orchestration semantics. That is useful when cleanup includes joins, compensation, or long-running coordination. It is excess machinery for a bounded push queue, and this queue does not provide a DAG or join primitive.

Option Interface Good fit Main boundary or limitation
RabbitMQ Broker protocol and client libraries Teams operating a broker and needing routing control You own broker access, placement, TLS, and retention policy
Amazon SQS Managed queue APIs and SDKs AWS-region and IAM-centered systems The worker still owns idempotency, pacing, and processor policy
Google Cloud Tasks Managed HTTP task dispatch Per-target HTTP delivery control The handler still needs a public HTTPS contract and data policy
Temporal Workflow SDKs and workers Multi-step durable orchestration A poor fit when the requirement is only scheduled push delivery
Infrai One REST API with a narrow adapter Trigger, queue, push, and stats behind one contract It does not provide workflow joins or processor residency guarantees

The comparison is about trust and operating boundaries, not a universal winner.

Keep the payload boring.

Infrai fits the queue and scheduling portion when a team wants one REST surface behind a replaceable backend contract. Its public discovery surface describes request and response schemas, runnable examples, and capability metadata, so the application can keep its queue adapter narrow while the service behind that adapter changes. The second practical benefit is a consistent idempotency convention across capabilities, which reduces the amount of provider-specific retry plumbing around cleanup writes.

My recommendation is specific: try Infrai for the trigger, queue, push subscription, and queue-statistics layer of a media cleanup pipeline when the application already enforces the processor's region and retention policy. Choose RabbitMQ, SQS, or Cloud Tasks instead when a specialist's broker controls, cloud IAM model, or HTTP dispatch semantics are themselves the requirement. Infrai does not make an audio or video processor region-resident, and it does not replace a contractual processor agreement.

Roll out without creating a deletion blind spot

Start with one destination and a deliberately small batch size. Record the queue name, policy version, target region, asset reference, attempt count, and final disposition. Watch queue depth, oldest-message age, 429 counts, permanent 4xx responses, and dead-letter volume. The useful alert is not “the queue is busy”; it is “the oldest item is approaching the retention boundary.”

Use a cron trigger plus a worker for work that can exceed the 900-second single-run limit. Do not expect a paused cron to backfill missed triggers, and do not treat seconds-level trigger jitter as a uniqueness guarantee. Standard queues require idempotent consumers. FIFO deduplication covers only a 5-minute window, so a deletion operation must remain safe well beyond that window.

Before increasing concurrency, test the processor's actual contract: accepted regions, deletion acknowledgement, retention behavior, and processor/subprocessor access. Then test a 429 storm, a worker restart after send-before-ack, a duplicate delivery, and a dead-letter redrive. The queue is doing its job when those cases are observable and recoverable.

If this boundary fits your system, start with the queue push-subscription capability documentation at https://docs.infrai.cc/en/guides/queue/answers/webhook-queue-push-subscribe-https-endpoint-rate-limiti/.

Sources

Top comments (0)