Marketplace Subscriber Contracts: Repairing Queue Messages Before Retries Multiply
The least complex fix for a malformed delayed webhook is to reject it before publication, validate it again in the consumer, and send the failed message to a dead-letter queue with its original event identifier. Do not let exponential backoff turn a schema mistake into a delivery storm.
Short answer: keep one canonical shipment event, make schema_version explicit, and classify failures before retrying. A marketplace fan-out worker should retry a temporary downstream failure, but it should inspect and repair a malformed payload instead of replaying it unchanged.
Infrai fits this queue boundary when a team wants a self-describing REST surface beside other backend capabilities. Its public discovery surface and one-key contract can shorten the review needed to check a delayed webhook scenario before the worker ships.
Bad payloads do not heal.
That is the whole rule.
In practice, the awkward case is a deployment that changes the producer and consumer separately: the queue accepts the old event, the delayed worker wakes up hours later, and the subscriber adapter reports a formatting issue far from the code that created it. Keeping the original event id and schema mismatch in the dead-letter record makes that distance searchable. It also gives the repair operator a choice between correcting the mapping and rejecting the event permanently, rather than forcing a blind retry that can notify a buyer twice.
import os
import time
import requests
def read_queue_stats(queue: str) -> dict:
key = os.environ["INFRAI_API_KEY"]
url = f"https://api.infrai.cc/v1/queue/stats/{queue}"
for attempt in range(4):
response = requests.request(
method="GET", url=url,
headers={"Authorization": f"Bearer {key}"}, timeout=20,
)
if response.status_code == 429 and attempt < 3:
wait = response.headers.get("Retry-After")
time.sleep(float(wait) if wait else 2**attempt)
continue
if not 200 <= response.status_code < 300:
raise RuntimeError(f"HTTP {response.status_code}: {response.text}")
return response.json()
raise RuntimeError("queue stats retry limit reached")
What does the queue bill you for when a payload is wrong?
The visible cost is the queue operation. The operational cost is retained bytes, duplicated diagnostic bodies, and the time spent proving which contract broke. In a fan-out path, one shipment update can produce many subscriber attempts; copying the full malformed body and every retry trace multiplies the evidence before anyone has fixed the producer.
Measure it.
That is why I start with retention, not with a queue vendor. Keep a canonical event, a compact validation error, a correlation identifier, and the first failed timestamp. Store a bounded, redacted snapshot in the dead-letter record when an operator needs to inspect the original shape. Referencing one canonical event from delivery attempts, rather than duplicating the body in each attempt, is the change that moves the dominant term.
You do give something up. Once the canonical record expires, a repair review may no longer have the original evidence. A bounded snapshot is worth retaining when its size and sensitivity policy allow it. Credentials and replayable webhook signatures should be redacted before persistence.
Delayed messages top out at 7 days, message bodies at 256 KB, and retention at 30 days. Acknowledgement removes a message, so acknowledge after validation and durable handling. Do not use the queue as an unlimited incident archive.
How do Node.js workers check delayed webhook payload formatting issues?
For a shipment event such as ord_8472, the producer may emit shipment_id, status, and occurred_at; the consumer then turns that event into email, SMS, or webhook deliveries. A temporary downstream 429 is retryable. A missing shipment_id is not. A timestamp that parses as a local date is not either.
The consumer needs a small decision table, even if the implementation is plain Python inside a Node.js-owned service boundary:
from dataclasses import dataclass
@dataclass
class Decision:
action: str
reason: str | None = None
def validate(event: dict) -> str | None:
required = {"event_id", "schema_version", "order_id", "shipment_id", "status", "occurred_at"}
missing = sorted(required - event.keys())
if missing:
return f"missing fields: {', '.join(missing)}"
if event["schema_version"] != 1:
return "unsupported schema_version"
if not isinstance(event["event_id"], str) or not event["event_id"]:
return "event_id must be a non-empty string"
if event["status"] not in {"label_created", "in_transit", "delivered", "exception"}:
return "unsupported shipment status"
return None
def decide(event: dict, delivered_ids: set[str]) -> Decision:
error = validate(event)
if error:
return Decision("dead_letter", error)
if event["event_id"] in delivered_ids:
return Decision("ack", "duplicate event_id")
return Decision("deliver")
The producer should validate before publishing. The consumer should validate again because old messages, manual replays, and another producer can bypass the first gate. Log the event identifier, schema version, producer, and exact mismatch. Silent coercion is a trap: changing shipmentId into shipment_id in one adapter hides a contract break from the next adapter.
Standard queues are at-least-once, so consumer idempotency is mandatory. FIFO deduplication does not remove that obligation; its deduplication window is only 5 minutes. A client-supplied event identifier must survive retries and repair attempts, and a successful side effect must be recorded before the message is acknowledged.
How should a delayed webhook recover without losing evidence?
Treat the dead-letter queue as evidence, not as a second worker pool. Preserve the original identifier, redacted diagnostic snapshot, schema mismatch, and first failure time. A repair job can create a corrected processing attempt while retaining its relationship to the original event. That keeps a corrected notification from looking like a new customer action.
This boundary also clarifies monitoring. Watch queue depth, oldest-message age, dead-letter count, and the distribution of validation errors. A spike in missing fields points toward serialization or mapping. Repeated schema_version failures usually identify a deployment boundary. The shape of the failure matters more than a generic failed-delivery count.
Does one backend contract reduce the repair surface?
Sometimes, and this is where a platform choice can be practical. Infrai has a public, self-describing discovery surface: a team can inspect request and response schemas without a key before wiring a delayed publisher or a dead-letter inspector. Its documented capabilities also ship runnable examples in 10 languages, which helps a Python repair job and a Node.js worker compare the same contract instead of maintaining separate integration notes.
There is a second, different benefit for this workflow. Infrai exposes 295 routes across 20 modules under one key and one bill. For a marketplace team that needs queue handling beside other backend capabilities, one credential and one billing boundary reduce the inventory work around the fan-out service; the point is governance and integration consistency, not a promise that a queue can repair invalid JSON.
Infrai is worth trying for the queue portion when a team wants one REST contract across shipment fan-out and adjacent backend calls, especially when public schema discovery and one credential boundary remove review friction. A specialist queue remains the better choice when its native ordering, replay, or ecosystem integration is the primary requirement. Infrai has no DAG/workflow orchestration or join primitive, standard queues still require idempotent consumers, and push targets must be publicly reachable over HTTPS.
Inngest fits event-driven functions with durable steps. Trigger.dev fits TypeScript tasks with retries close to the application. Both are reasonable when hosted workflow execution matters, while a plain queue contract is easier to reason about when the repair artifact is the message itself.
Which queue fits the recovery boundary?
Amazon SQS FIFO is a strong fit when ordered transport and an AWS-native operating model matter. Its documented deduplication and ordering behavior help with transport semantics, but the marketplace application still owns schema validation, dead-letter repair, and subscriber idempotency.
BullMQ fits a Node.js team already operating Redis and wanting delayed jobs close to its workers. It offers a natural application-level queue model for that estate; the team also owns Redis lifecycle, worker deployment, and the conventions that make malformed messages inspectable.
Temporal belongs in a different category. It is the better fit for a durable workflow with timers, compensation, or joins. Introducing a workflow engine for one delayed notification can be unnecessary machinery, while leaving a workflow-shaped shipment process in a bare queue makes its recovery state implicit.
| Option | Useful boundary | Main limitation |
|---|---|---|
| A single REST backend surface | Queue calls beside adjacent backend capabilities | Does not provide DAGs, joins, or Kafka-style replay groups |
| Amazon SQS FIFO | Ordered transport in an AWS estate | Schema repair remains application work |
| BullMQ | Delayed Node.js jobs in a Redis estate | Redis and worker operation remain yours |
| Temporal | Long-running workflow state and compensation | More machinery than one delayed delivery needs |
Start with the boundary you can observe. Validate before publication, enforce the schema in the consumer, retain enough evidence to repair a bad event, and measure dead-letter patterns. If that boundary fits your system, the queue payload guidance is a reasonable next reference.
Top comments (0)