Dead-Letter Queue (DLQ)

When a pipeline component fails to process a message, that message lands in the pipeline’s dead-letter queue (DLQ) instead of being dropped. Each entry records the component that failed, the error, and the original payload. The Enterprise list endpoint also assigns each message a stable message_id.

Open Source GlassFlow lets you inspect the DLQ and purge it. The Enterprise Edition adds a non-destructive paginated read, reprocessing (replaying failed messages through the pipeline), and selective discard.

Examples are shown for the Python SDK and the REST API. A Web UI for DLQ management is on the way; once it ships, the examples below will gain a Web UI tab with step-by-step screenshots.

All endpoints live under /api/v1/pipeline/{pipeline_id}/dlq. The examples use http://localhost:30180 as the host, the default for a local CLI install, and the Python SDK client:

# Open Source client
from glassflow.etl import Client
# Enterprise client (adds list, reprocess, discard)
from glassflow.ee import Client
 
client = Client(host="http://localhost:30180")
pipeline = client.get_pipeline(pipeline_id="my-pipeline")

DLQ state

Get a summary of the queue: totals and the last received and consumed timestamps.

print(pipeline.dlq.state())

curl "http://localhost:30180/api/v1/pipeline/my-pipeline/dlq/state"

Response:

{
  "last_consumed_at": "2026-05-20T10:32:38.113427621Z",
  "last_received_at": "2026-05-20T10:23:26.530466989Z",
  "total_messages": 2,
  "unconsumed_messages": 1
}

Reading messages

GlassFlow provides two endpoints for reading the DLQ: consume in Open Source, and list in the Enterprise Edition, which is non-destructive and returns stable message IDs.

Consume

consume reads a batch of messages from the DLQ. It is available in Open Source.

messages = pipeline.dlq.consume(batch_size=50)

curl "http://localhost:30180/api/v1/pipeline/my-pipeline/dlq/consume?batch_size=50"

Response:

[
  {
    "component": "ingestor",
    "error": "failed to validate data",
    "original_message": "<original message>"
  }
]

List

list is a non-destructive, paginated read. It leaves messages in place and assigns each a stable message_id (the value you pass to reprocess and discard), alongside component, error, original_message, and received_at.

The response is a page object: messages plus has_more, and a next_cursor when more pages remain. Pass next_cursor back as the cursor parameter to fetch the next page. Filter to a single component with the component parameter, one of ingestor, join, sink, dedup, otlp-receiver.

# One page, optionally filtered by component
page = pipeline.dlq.list(batch_size=50, component="sink")
for message in page["messages"]:
    print(message["message_id"], message["component"], message["error"])
 
# Or iterate every message, paging through the cursor for you
for message in pipeline.dlq.list_iter(component="sink"):
    print(message["message_id"], message["error"])

# First page (component filter is optional)
curl "http://localhost:30180/api/v1/pipeline/my-pipeline/dlq/list?batch_size=50&component=sink"
 
# Next page: pass the previous response's next_cursor as cursor
curl "http://localhost:30180/api/v1/pipeline/my-pipeline/dlq/list?batch_size=50&cursor=101"

Response:

{
  "messages": [
    {
      "message_id": "101",
      "component": "sink",
      "error": "connection refused",
      "original_message": "{...}",
      "received_at": "2026-05-29T14:00:00Z"
    }
  ],
  "next_cursor": "101",
  "has_more": true
}

Reprocessing

Reprocess moves messages from the DLQ back to the start of the pipeline (the source input), so they flow through the whole pipeline again. Use it after fixing a transient failure, for example a ClickHouse outage, or after fixing a transform bug. It returns 202 Accepted; the replay happens asynchronously.

# Reprocess everything currently in the queue
pipeline.dlq.reprocess_all()
 
# Reprocess specific messages by message_id
pipeline.dlq.reprocess(["101", "102"])

# Reprocess everything currently in the queue
curl -X POST "http://localhost:30180/api/v1/pipeline/my-pipeline/dlq/reprocess" \
  -H 'Content-Type: application/json' \
  -d '{"mode": "all"}'
 
# Reprocess specific messages by message_id
curl -X POST "http://localhost:30180/api/v1/pipeline/my-pipeline/dlq/reprocess" \
  -H 'Content-Type: application/json' \
  -d '{"mode": "selected", "message_ids": ["101", "102"]}'

Response:

{
  "request_id": "dlq-reprocess-a1b2c3d4-5e6f-7890-abcd-ef0123456789",
  "status": "accepted"
}

status is accepted when every source component acknowledges the request, or partial when only some do.

Discarding

Discard permanently removes messages without reprocessing them. Use it for genuinely malformed events you do not want to retry.

# Discard specific messages
pipeline.dlq.discard(["200"])
 
# Discard everything currently in the queue
pipeline.dlq.discard_all()

# Discard specific messages
curl -X POST "http://localhost:30180/api/v1/pipeline/my-pipeline/dlq/discard" \
  -H 'Content-Type: application/json' \
  -d '{"mode": "selected", "message_ids": ["200"]}'
 
# Discard everything currently in the queue
curl -X POST "http://localhost:30180/api/v1/pipeline/my-pipeline/dlq/discard" \
  -H 'Content-Type: application/json' \
  -d '{"mode": "all"}'

Response:

{
  "request_id": "dlq-discard-b2c3d4e5-6f70-8901-bcde-f01234567890",
  "discarded_count": 1
}

Purging

purge removes every message from the DLQ. It is available in Open Source and returns no body. In the Enterprise Edition it is superseded by discard_all (discard with mode: all), which reports how many messages were removed.

pipeline.dlq.purge()

curl -X POST "http://localhost:30180/api/v1/pipeline/my-pipeline/dlq/purge"

Modes, limits, and behavior

Reprocess and discard both support two modes:

• all acts on every message currently in the DLQ, up to the queue head at the time of the request. Messages that arrive after the request are left for the next call.

• selected acts only on the message_id values you pass.

• A selected request accepts at most 1000 message_id values.

• mode: all processes up to 10,000 messages per request with a 60 second timeout. For larger queues, page through the messages and use selected requests.

• Reprocessing a message_id that no longer exists is not an error. The missing ID is reported back and the remaining IDs are still processed.

• mode: all on an empty DLQ returns a count of zero rather than an error.

Related

• Enterprise Edition
• Python SDK
• Metrics and monitoring