Route permanently failed messages to a dead-letter queue and manage them through the HTTP admin API.
An MQTT-to-SQS bridge processes telemetry data. Some messages fail permanently – malformed JSON, authentication errors, or expired TTLs. Instead of silently dropping these messages, you want to:
flowchart LR
subgraph Normal Path
MQTT["MQTT Broker\ntelemetry/#"] -->|subscribe| R[Receiver\nmqtt-in]
R --> Route[Route\ningest]
Route --> S[Sender\nsqs-out]
S -->|SendMessageBatch| SQS["SQS Queue\nevents"]
end
subgraph Failure Path
Route -->|permanent failure\nor expired| DLQ[(DLQ Store)]
end
subgraph HTTP Admin
OP[Operator] -->|GET /dlq| API[Admin API\n:8080]
OP -->|POST /dlq/redrive| API
API --> DLQ
API -->|re-inject| Route
end
style Route fill:#f96,stroke:#333
style DLQ fill:#f66,stroke:#333
style API fill:#6cf,stroke:#333
bridge:
id: dlq-managed-bridge
sessions:
- id: mqtt-conn
transport: mqtt
# direct_hold relies on the broker redelivering what a crashed process never
# acknowledged; only a persistent (or exclusive) session does that.
session_mode: persistent
options:
session:
broker_url: tcp://mqtt.example.com:1883
client_id: dlq-bridge-01
clean_start: false
session_expiry_interval: 3600
receivers:
- id: mqtt-in
session_id: mqtt-conn
topics:
- topic: "telemetry/#"
qos: 1
senders:
- id: sqs-out
transport: sqs
options:
queue_url: https://sqs.us-west-1.amazonaws.com/123456789/events
region: us-west-1
batch_size: 10
bindings:
- id: to-events
sender_id: sqs-out
# Naming the session on the binding is what makes the bridge manage it:
# connect, subscribe, reconcile. A session nobody manages never subscribes.
session_id: mqtt-conn
address: events
stores:
dlq:
type: memory
options:
acknowledge_volatile: true
# A persistent session keeps an exact record of the filters it installed on
# the broker (ADR 0003); seed the baseline once, before the first start:
# gobridge -config bridge.yaml -seed-managed-subscriptions mqtt-conn
managed_subscriptions:
type: sqlite
options:
path: /var/lib/gobridge/state/managed-subscriptions.db
routes:
- id: ingest
receiver_id: mqtt-in
delivery_mode: direct_hold
dispatch_mode: single
bindings: [to-events]
policy:
# Exactly one replica consumes this subscription; a second copy of this
# process would double-deliver. See Scenario 8 for fenced ownership.
allow_unfenced: true
max_in_flight: 100
max_replay_attempts: 3
on_permanent_failure: dlq
on_expired: dlq
backoff:
initial_interval: 1s
max_interval: 30s
multiplier: 2.0
http:
admin_addr: ":8080"
monitor_addr: ":8081"
admin_api_key: "change-me-to-a-real-secret-key"
stores.dlq: { type: memory }The DLQ store holds failed message entries. The memory type is suitable for development and single-instance deployments, but entries are lost on restart – which erases the only record that a message existed and was given up on. That loss must be acknowledged with acknowledge_volatile: true before the store will build. See Variations for persistent options.
policy.on_permanent_failure: dlqWhen a message encounters a non-recoverable error – invalid payload, authentication rejection, schema violation – the runtime routes it to the DLQ store. The alternative value drop silently discards the message.
policy.on_expired: dlqWhen a message exceeds its TTL, it is routed to the DLQ instead of being dropped. This catches messages that spent too long in retry loops or were delayed in transit.
policy.max_replay_attempts: 3Transient errors are retried up to 3 times with exponential backoff. After exhausting retries, the message follows the on_permanent_failure policy – in this case, DLQ.
admin_api_key – Must be at least 16 characters. The server refuses to start with a shorter key. Required for all admin endpoints.admin_addr: ":8080" – Control operations: bridge start/stop, DLQ management, message injection. All endpoints require authentication.monitor_addr: ":8081" – Health probes (unauthenticated) and topology inspection (authenticated).All admin endpoints require the API key via the X-API-Key header or Authorization: Bearer <key>. Key comparison uses SHA-256 constant-time hashing to prevent timing attacks. Failed auth returns HTTP 401 with a WWW-Authenticate: Bearer realm="gobridge-admin" header.
curl -s -H "X-API-Key: change-me-to-a-real-secret-key" \
"http://localhost:8080/api/v1/admin/dlq" | jq .
{ "configured": true, "count": 1 }
All DLQ endpoints return HTTP 404 with {"error": "no DLQ store configured"} when no DLQ store is present.
Paginated message listing with filtering. Supports route_id, category, since, before, limit (max 1000), and offset parameters.
curl -s -H "X-API-Key: change-me-to-a-real-secret-key" \
"http://localhost:8080/api/v1/admin/dlq/messages?route_id=ingest&limit=10" | jq .
{
"messages": [
{
"id": "dlq-001", "route_id": "ingest", "binding_id": "to-events",
"source_id": "mqtt-in", "correlation_id": "corr-abc-123",
"subject": "telemetry/temperature/sensor-42",
"reason": "invalid payload", "category": "rejected",
"error_code": "INVALID_PAYLOAD",
"last_error": "json: cannot unmarshal string into Go value of type int",
"failed_at": "2026-03-28T10:15:30Z", "attempts": 3
}
],
"limit": 10, "offset": 0, "has_more": false
}
Re-inject (redrive) entries back into their original route. Maximum 100 IDs per request. Successfully redriven entries are automatically deleted from the DLQ store.
curl -s -X POST -H "X-API-Key: change-me-to-a-real-secret-key" \
-H "Content-Type: application/json" \
-d '{"ids": ["dlq-001", "dlq-002"]}' \
"http://localhost:8080/api/v1/admin/dlq/redrive" | jq .
{ "redriven": 2, "failed": 0 }
Redrive claims each entry by delete-before-inject (no double-delivery) and
confines the replay to the entry’s original binding_id out-of-band, not via
a header. It emits a dlq.redrive.begin audit record (outcome pending) with the
requested IDs before the first claim. When the runtime lacks redrive-safe
injection the response adds a warning field — the replay reuses the original
envelope ID and may be silently deduplicated by the outbox on shared_outbox
routes, so verify delivery:
{ "redriven": 2, "failed": 0, "warning": "runtime lacks redrive-safe injection: replays reuse the original envelope id and may be silently deduplicated by the outbox on shared_outbox routes; verify delivery" }
See HTTP API Reference for the full contract (207
on partial failure, per-entry errors).
Permanently delete all expired entries up to the current time.
curl -s -X POST -H "X-API-Key: change-me-to-a-real-secret-key" \
"http://localhost:8080/api/v1/admin/dlq/purge" | jq .
{ "purged": 5 }
sequenceDiagram
participant M as MQTT Broker
participant B as GoBridge Route
participant D as DLQ Store
participant O as Operator (curl)
M->>B: Deliver message (QoS 1)
B->>B: Process (transform, send)
B--xB: Permanent failure (INVALID_PAYLOAD)
B->>D: Write DLQ entry (envelope + error context)
B->>M: ACK message (remove from broker)
Note over D: Entry stored with reason, category, error_code
O->>D: GET /api/v1/admin/dlq/messages
D-->>O: Full message bodies
Note over O: Diagnose root cause, deploy fix
O->>D: POST /api/v1/admin/dlq/redrive
D->>B: Re-inject envelope into route
B->>B: Reprocess (succeeds this time)
on_expired: dlq – TTL ExceededTriggers when now > envelope.ExpiresAt. Common causes: message sat too long in the source queue, transient retries consumed more time than TTL allows, or a circuit breaker held the message open. Error class: expired, code: MESSAGE_EXPIRED.
on_permanent_failure: dlq – Unrecoverable ErrorTriggers in two scenarios:
Immediate permanent failure – First attempt returns class permanent or rejected:
INVALID_PAYLOAD, NOT_AUTHORIZED, FORBIDDEN, PAYLOAD_TOO_LARGE, SCHEMA_VIOLATIONExhausted retries – Transient errors (TIMEOUT, CONNECTION_LOST, UNAVAILABLE, THROTTLED) persist beyond max_replay_attempts.
Count-less sources. Sources that carry a native receive-count header (SQS
ApproximateReceiveCount, JetStreamnum_delivered, etc.) drive themax_replay_attemptscap directly from that header. Sources WITHOUT one (MQTT, AMQP 0-9-1, HTTP) are capped by a bridge-owned attempt ledger keyed on a stable per-message identity (dedup key / envelope ID), so a deterministic transient failure is DLQ’d or dropped aftermax_replay_attemptsinstead of looping forever. The ledger is an in-process, bounded structure evicted on every terminal settle; its cap is per bridge instance and does not survive a restart or span instances. Full cross-instance / durable replay accounting requires the source adapter to stamp a stable idempotency key that survives redelivery (seeUBIQUITOUS.md– dedup key) and a durable ledger port – the upgrade path when a native receive count is unavailable.
flowchart TD
E[Error Occurs] --> C{Error Class?}
C -->|transient| R{Retries Left?}
R -->|yes| B[Backoff + Retry]
R -->|no| PF[on_permanent_failure policy]
C -->|permanent / rejected| PF
C -->|expired| EX[on_expired policy]
PF --> D{Policy Value?}
EX --> D2{Policy Value?}
D -->|dlq| DLQ[Write to DLQ Store]
D -->|drop| DROP[Discard silently]
D2 -->|dlq| DLQ
D2 -->|drop| DROP
The monitor server (:8081) exposes unauthenticated probes and authenticated observability endpoints. All health probes set Cache-Control: no-cache, max-age=0.
| Endpoint | Purpose | Response |
|---|---|---|
GET /api/v1/monitor/health |
Coarse health check | {"status":"ok"} (200) or {"status": ...} (503) |
GET /api/v1/monitor/live |
Liveness probe | {"status":"alive"} – 200 while the runtime is recoverable; {"status":"terminal"} with 503 once the runtime is terminal |
GET /api/v1/monitor/ready |
Readiness probe | {"status":"ready", "role":"standalone"} – 200 when processing |
The health endpoint returns only a coarse status: ok (200), or unhealthy, not_running, or unavailable (503) — never instance_id, route count, or component detail, which live behind auth on /deephealth. The role field on /ready and /deephealth reflects the deployment mode: standalone, active (lease holder), or standby (waiting for lease).
These require the monitor API key (or admin key as fallback):
| Endpoint | Purpose |
|---|---|
GET /api/v1/monitor/topology |
Instance identity, running state, compact route list |
GET /api/v1/monitor/routes |
Detailed routes with policy (max_in_flight, ack_after, on_expired) |
GET /api/v1/monitor/deephealth |
Session connectivity, lease status, subscription convergence, service levels |
The deep health endpoint returns 200 when ready for traffic, 503 otherwise, with a service_level field aggregating session health: full, degraded, or none.
reg := ports.NewRegistry()
_ = paho.Register(reg)
_ = sqs.Register(reg)
_ = nativestore.Register(reg)
cfg, _ := cfgparser.ParseFile("bridge.yaml", cfgparser.FormatAuto, reg)
rt, _ := bridge.NewBuilder(cfg, bridge.WithLogger(logger)).
RegisterTransportFactory("mqtt", paho.NewFactory(logger)).
RegisterTransportFactory("sqs", sqs.NewFactory(logger)).
RegisterStoreFactory("memory", nativestore.NewMemoryStoreFactory()).
Build(ctx)
rt.Start(ctx)
The memory store factory must be registered for stores.dlq to resolve. Without it, the builder returns "no store factory registered for dlq type \"memory\"".
http:
admin_addr: ":8080"
monitor_addr: ":8081"
admin_api_key: "admin-key-min-16-characters"
monitor_api_key: "monitor-key-min-16-chars"
Authenticated monitor endpoints (/topology, /routes) use the monitor key. Health probes remain unauthenticated.
For single-instance production where DLQ entries must survive restarts:
stores:
dlq:
type: sqlite
options:
path: /data/dlq.db
builder.RegisterStoreFactory("sqlite", nativestore.NewSQLiteStoreFactory())
For clustered deployments where multiple instances share a DLQ:
bridge:
id: dlq-managed-bridge
deployment_mode: clustered
stores:
dlq:
type: dynamodb
options:
table_name: gobridge-dlq
builder.RegisterStoreFactory("dynamodb", awsstore.NewDynamoDBStoreFactory(ddbClient))
Clustered mode requires all configured stores to be distributed. Memory and SQLite stores are rejected during validation.
http:
admin_addr: ":8080"
admin_api_key: "change-me-to-a-real-secret-key"
cors_origins: "https://dashboard.example.com,https://admin.example.com"
Wildcard * is explicitly rejected to prevent open CORS. List specific origins, separated by commas.