A reliable SQS on Redis. At-least-once delivery, a visibility timeout, retries with exponential backoff, and a dead-letter queue — built on Redis Streams consumer groups. For projects that already run Redis and don't want to operate a heavy SQS, Kafka, or RabbitMQ setup to have a simple queue.
SQS / Kafka is a whole river system. Sometimes all you need is one stream: small, lightweight, and it never drops the load.
- 📦 One runtime dependency. Just
redis— Pydantic and the ops CLI are opt-in extras, so a plain install stays tiny. - 🔒 A real visibility timeout. One consumer owns a message for
retry_on_idle_ms. Miss the window and another worker picks it up — the SQS bargain, on the Redis you already run. - ⚛️ Atomic redelivery. Every move is one server-side Lua script: copy forward, ack behind, indivisibly. No crash window where a message sits in both places, or neither.
- ♻️ Retries that actually back off. Exponential escalation, a cap, and jitter applied after the cap — so a recovering cohort spreads out instead of stampeding your database together.
- 💀 A dead-letter queue that keeps things. Never trimmed, no TTL, no cap. Undecodable messages stay readable,
redrive_dlqputs them back atomically, andpurge_dlqis the only thing that ever deletes. - 🎯 Typed bodies with Pydantic.
schema=validates on publish and parses on consume, so your handler stops hand-rollingdictaccess — and it isn't hard-coupled: anydump/loadpair works. - 🧰 An ops CLI in the box.
xstreamq depth,peek,dlq ls / redrive / purge,compact— human-readable or--json. - 🛑 Graceful shutdown. Drains in-flight handlers on a rolling deploy, cancels the stragglers, and interrupts blocked reads rather than waiting them out.
- 🩺 Health you can wire to a probe.
healthy()checks the connection and every background loop — a consume loop that died silently is exactly what it exists to catch. - 🩹 Connection resilience, on by default. TCP keepalive and a health-check interval on every client xstreamq opens, so a silently dropped connection doesn't park a queue forever.
- 🧹 Safe compaction.
compacttrims only what every group is finished with — the no-loss alternative to a rawMAXLEN. - 🧱 Cluster-ready keys. Stream, retry, and DLQ share one hash tag, so the atomic move works on Redis Cluster too.
- 🤖 An Agent Skill in the repo that teaches a coding agent how to set up and operate xstreamq.
- 🧪 Tested against a real Redis, under an 85% coverage gate, with the Lua scripts tested as the correctness core they are.
And what you don't get — stated up front, because a reliability library that oversells is worse than one that doesn't exist.
- This is at-least-once, not exactly-once: handlers must be idempotent (ideally).
- Retries are opt-in — no
retry_on_idle_ms, no visibility timeout. - Handlers run on your event loop. A plain
defhandler that blocks blocks the whole queue, and a handler that never awaits can't be cancelled at shutdown. - It's built for small-to-medium workloads — a few thousand messages a day up to a few hundred a second on one Redis. Past that, buy a real broker.
Each of those has its own section below. None of them is a footnote.
If you already run Redis, a decent "queue" is four Redis primitives and one background loop away — a stream, a consumer group, the pending-entries list, and XACK. The primitives are the easy part; the reliability semantics around them are not, and that is what xstreamq owns so you don't stand up another piece of infrastructure to get them. xstreamq is built on the main premise that things will fail or go wrong in production, so resilience and safety are the core principles.
"Why not use arq/Celery/Dramatiq etc.?" The honest answer is that xstreamq is a simple message queue (decoupled producers/consumers, fan-out groups, a DLQ ops surface), not a job framework — and it's ~800 statements you can read in an afternoon versus a big framework you need to adopt. That readability is a legitimate selling point.
uv add xstreamq # or: pip install xstreamqNothing else comes with it: the core runtime dependency is redis and nothing more. Typed bodies and the ops CLI are extras you opt into:
uv add 'xstreamq[pydantic]' # typed message bodies
uv add 'xstreamq[cli]' # the `xstreamq` ops commandRequires Python ≥ 3.10 and Redis ≥ 6.2 (the reclaimer pages through the PEL with exclusive XPENDING ranges; lag reporting needs 7.0 and returns None below that).
import asyncio
from xstreamq import XStreamQ
dq = XStreamQ("redis://localhost:6379")
@dq.subscriber(
"orders",
group="order-service",
retry_on_idle_ms=30_000, # the visibility timeout — see below
max_retries=5, # then it goes to the DLQ
backoff_multiplier=2.0, # 30s → 60s → 120s → … between attempts
)
async def handle(msg):
await process_order(msg.body)
async def main():
async with dq: # start() on enter, stop() on exit
await dq.publish("orders", {"id": 1, "total": 4200})
await asyncio.Event().wait() # run until cancelled
asyncio.run(main())A handler that returns normally acks the message; a handler that raises leaves it in the Pending Entries List (PEL) to be retried — so handlers must be idempotent. await dq.subscribe("orders", handler, group=...) is the non-decorator form, and it brings the subscription live immediately if the queue is already started.
Handlers may be async def or plain def, but both run on your event loop. xstreamq does not hand a sync handler to a thread pool, so a def handler that does blocking I/O stalls every other consumer in the process. Make it async def, or push the blocking call into asyncio.to_thread yourself.
publish appends an entry to a Redis stream. Each subscription gets its own consumer group, so workers in one group share the work (add workers to scale out) and separate groups each get a full copy (fan-out).
The moment a worker reads a message it lands in the group's pending entries list — "delivered, not yet acknowledged". It leaves only on XACK:
- handler returns → acked → gone
- handler raises → deliberately not acked → stays pending
- worker crashes → never acked → stays pending
That is the whole basis of at-least-once delivery: a failed message stays put instead of vanishing. A background reclaimer then picks up anything pending longer than retry_on_idle_ms, bumps its retry count, and redelivers it — and after max_retries attempts, moves it to the dead-letter queue instead.
The reclaimer's redelivery is a single server-side Lua script, so the "copy to the retry stream" and "ack on the source" steps cannot be torn apart by a crash. There is no window in which a message exists in both places.
publish()
│
▼
┌───────────────────────────┐
│ {orders} │ the queue
└─────────────┬─────────────┘
│ XREADGROUP >
▼
┌───────────────────────────┐
│ PEL · group "svc" │ in flight — delivered to
│ delivered, un-acked │ exactly one consumer in
└───┬───────────────────┬───┘ the group
handler │ │ handler raises, or the
returns ▼ │ worker dies → no XACK,
XACK ✓ │ so it just stays put
done │
│ ...idle > retry_on_idle_ms...
▼
┌─────────────────────────┐
│ reclaim.lua │ XADD + XACK as
│ ONE indivisible step │ one atomic op
└────┬───────────────┬────┘
rc + 1 <= max_retries │ │ rc + 1 > max_retries
▼ ▼
┌─────────────────────────┐ ┌─────────────────────────┐
│ {orders}.svc.retry │ │ {orders}.svc.dlq │
│ rc + 1, oid stamped │ │ NEVER trimmed, no TTL │
└────────────┬────────────┘ └────────────┬────────────┘
│ XREADGROUP > │
▼ │ waits for a human:
┌─────────────────────────┐ │
│ PEL · "svc-retry" │ │ dlq_messages() inspect
│ the same handler │ │ redrive_dlq() put back
└────────────┬────────────┘ │ purge_dlq() delete
│
│ fails again → reclaim.lua → back to the SAME retry
└─ stream, rc + 1 each time, until rc > max_retries → DLQ
Everything below the pending list only exists if you set retry_on_idle_ms — see the warning just below. Without it a failed message stops at the PEL and stays there.
Three things the picture is meant to make obvious:
- Nothing is ever deleted on the way through. A message moves by being copied forward and acked behind, in one atomic step, so there is no moment where it is in neither place — or in both.
- The retry stream loops back to itself, not to a
.retry.retry. Every failed attempt bumpsrcin the same place, so the retry chain never forks. - The DLQ is a terminus, not a stage. Nothing consumes it and nothing trims it. It stays exactly as it is until a human calls
redrive_dlqorpurge_dlq.
Two shortcuts are not drawn, to keep the main path clear. An entry with no body at all — which only a foreign producer can create — skips the retry chain and goes straight to the DLQ, since no handler could ever process it. And a redrive puts a dead letter back on {orders} as a fresh publish: rc reset to 0, no oid, a brand-new id.
A subscription with no retry_on_idle_ms has no visibility timeout at all. There is no reclaimer for it, so a failed delivery sits in the pending list forever — not lost (you can still see it via pending()), but not retried either. This is deliberate (no retry config, no background cost), and it is the sharpest default in the library. If you want "never drops the load", you want retry_on_idle_ms.
A consumer group is durable — once created, it remembers its own position, so a restarted consumer never re-reads messages it already acked. The only question is where a brand-new group begins, and that's the start_from subscribe option:
# default — start at the tail: only messages published AFTER this group exists
@dq.subscriber("orders", group="g")
async def handle(msg): ...
# start at the head: also drain everything already in the stream
@dq.subscriber("orders", group="g", start_from="0")
async def handle(msg): ...The default is start_from="$" (tail). It's the cheap, no-replay default, but it's also the one footgun worth knowing: a group that subscribes after messages were published silently skips that backlog. If you want a queue that loses nothing published before the consumer came up, use start_from="0". Either way it only matters the first time the group is created — after that the group's durable position wins and start_from is ignored.
Every key for one subscription — main stream, retry, DLQ, dedupe markers — shares a hash tag ({orders}, {orders}.<group>.retry, {orders}.<group>.dlq) so they land in one slot and the atomic Lua move works on Redis Cluster too.
@dq.subscriber(
"orders",
group="order-service",
retry_on_idle_ms=30_000, # the visibility timeout — see below
max_retries=5, # then it goes to the DLQ (None = retry forever, no DLQ)
backoff_multiplier=2.0, # each attempt waits longer (1.0 = constant cadence)
backoff_max_ms=300_000, # capped, then jittered
backoff_jitter=0.2, # a random 0–20% added on top, upward only
on_dlq=alert, # called when a message is dead-lettered
)
async def handle(msg): ...retry_on_idle_ms is the visibility timeout: how long one consumer gets to finish a message before another may take it. Un-acked for longer than that and the message is redelivered — to any worker in the group, across every replica you run. Set it to how long you are willing to let a single attempt hold a message.
Two things follow from that, and they are the same trade-off SQS makes. A handler slower than the timeout will see its message picked up elsewhere while it is still working, so set the timeout above your slowest realistic attempt — or use dedupe_ttl_s so the second delivery is acked instead of re-run. And retries are opt-in: with no retry_on_idle_ms there is no visibility timeout at all, so a failed message sits un-acked forever, neither retried nor dead-lettered.
Retried messages move to {orders}.<group>.retry; dead letters land in {orders}.<group>.dlq. Both are per-(stream, group), so each subscription has its own DLQ to inspect and redrive.
The DLQ is never trimmed. A dead letter you can't read is worthless, so it grows without a length cap regardless of any other setting. The retry stream can be capped with XStreamQ(retry_max_len=...), but it defaults to None on purpose: trimming a retry entry that is still in flight would drop a message silently, so the default is to grow visibly under a failure storm instead.
Pass an on_dlq callback (sync or async) to the subscription — useful for alerting:
def alert(body, original_id, retry_count):
log.error("dead-lettered %s after %d tries: %r", original_id, retry_count, body)# triage — read-only, decoded envelopes
for m in await dq.dlq_messages("orders", "order-service", count=100):
print(m.id, m.original_message_id, m.retry_count, m.body)
# put one back (returns the new source id, or None if it was already gone)
new_id = await dq.redrive_dlq_message("orders", "order-service", m.id)
# or a batch, oldest first (returns how many moved)
n = await dq.redrive_dlq("orders", "order-service", count=100)
# give up on them
await dq.purge_dlq_message("orders", "order-service", m.id) # one (True if it existed)
await dq.purge_dlq("orders", "order-service") # all (returns count removed)Each redrive is a single atomic Lua move — re-publish to the source plus delete from the DLQ in one server-side step — so there's no crash window that could redrive a message twice. The redriven message is a fresh publish: retry count reset to 0, body and headers preserved. Redrive is idempotent; redriving an id that's already gone is a no-op (None).
Only purge ever deletes a dead letter. A poison dead letter — one with no body to republish — cannot be redriven, so it is left exactly where it is and reported (UnredrivableError for a single id, skipped-and-logged for a batch) rather than quietly dropped. Those are the entries a human most needs to see.
Redrive a message only once its failure cause is actually fixed, or it will just land back in the DLQ.
stop() shuts down cleanly, which matters most on the most common real-world event — a rolling deploy. It stops reading new messages immediately (the blocking read is interrupted, not waited out, so shutdown isn't delayed by block_ms), but a handler that is mid-flight is allowed to finish and ack within a timeout instead of being killed — so you don't get needless retries and duplicate work every time you restart.
await dq.stop() # default: drain in-flight handlers up to 30s
await dq.stop(drain_timeout=10) # give them 10s, then cancel any stragglers
await dq.stop(drain_timeout=0) # immediate, non-draining stopA handler that overruns drain_timeout is cancelled — but only at an await. That's the honest limitation: asyncio cancellation lands at suspension points, so a handler stuck in a blocking call (or in CPU-bound work with no awaits) keeps running to completion regardless of the timeout, and the process may not exit until it settles. Either way at-least-once still holds: the message stays in the PEL and gets redelivered.
Consumers drain concurrently, so a multi-subscriber queue is bounded by drain_timeout, not the sum across consumers. The connection is closed only after draining, so in-flight acks still land. async with dq: uses the default drain on exit.
If you have long-running handlers, read this section again before your first rolling deploy.
By default a body is any JSON-serialisable value and your handler gets it back as-is. Pass schema= to have xstreamq validate and parse each message into a typed model — handlers stop hand-rolling dict access and bad payloads are caught at the edge. Pydantic lives behind an extra so the core stays redis-only:
uv add 'xstreamq[pydantic]'from pydantic import BaseModel
class Order(BaseModel):
id: int
item: str
qty: int = 1
@dq.subscriber("orders", group="order-service", schema=Order)
async def handle(msg):
msg.body # an Order instance, already validated
msg.body.qty # typed attribute access
await dq.publish("orders", Order(id=1, item="book", qty=3)) # serialised for you
await dq.publish("orders", {"id": 2, "item": "pen"}) # dict works too- Validate on publish, parse on consume. Publishing a model instance serialises it automatically. Pass
schema=topublish(...)too if you want a raw dict validated at the call site (it raises there) rather than discovered as poison on the consumer. - A body that won't validate on consume is poison. It's left un-acked so the reclaimer routes it to the DLQ, exactly like an undecodable entry — it never crash-loops your handler. DLQ contents are read back as the raw stored JSON (not re-validated), so a bad payload stays readable for triage.
- Not hard-coupled to Pydantic.
schema=also accepts any object exposingdump(value)/load(raw), leaving room for dataclass / msgspec / attrs adapters without an API change.
Handlers should be idempotent — at-least-once means "at least". A handler can run more than once for the same message: a crash in the window between handling and acking, or a retry-stream copy of a delivery that actually succeeded. msg.original_message_id is stable across retries (msg.dedup_id falls back to msg.id on the first delivery) and is the natural dedupe key.
For a shortcut, opt into built-in dedupe with dedupe_ttl_s:
@dq.subscriber("orders", group="order-service", dedupe_ttl_s=3600)
async def handle(msg):
await charge_card(msg.body) # runs at most once per message id within the TTLxstreamq remembers each successfully processed message's dedup_id for that many seconds (a Redis key with that TTL) and skips — and acks — any redelivery of the same id.
- The marker is written only after the handler returns, so a failed handler is never recorded and is still retried. Dedupe never costs you a message.
- Not a distributed lock. Two genuinely concurrent deliveries of the same id (a slow live handler while the reclaimer hands a copy to the retry consumer) can still both run. This collapses the common duplicate sources, not every possible one.
⚠️ Set the TTL above your whole retry window (max_retries×retry_on_idle_ms× backoff). Too short and a late retry arrives after the marker expired and runs again.
await dq.stream_length("orders") # entries retained in the log (XLEN) — NOT backlog
await dq.pending("orders", "order-service") # delivered-but-not-acked (PEL) — in flight right now
await dq.lag("orders", "order-service") # published-but-not-yet-delivered (Redis ≥ 7)
await dq.peek("orders", count=10) # read without consuming
await dq.registered() # every (stream, group) registered against this Redis
await dq.health() # itemised liveness; healthy() is this, as a bool
await dq.dlq_messages("orders", "order-service") # triage the DLQ
await dq.redrive_dlq("orders", "order-service") # put a batch back
await dq.purge_dlq("orders", "order-service") # give up on them
await dq.compact("orders") # trim what every group has finished with
await dq.healthy() # is the connection up and are the loops alive?The three size numbers measure different things: stream_length never shrinks on ack (entries stay in the log), so it's retained size, not unprocessed work; pending is what's currently being worked on (or stuck); lag is "how far behind is this consumer?" and returns None if Redis can't compute it after a trim.
Because stream_length never shrinks on ack, a busy stream grows forever. compact is the safe way to shrink it:
removed = await dq.compact("orders") # entries trimmed; safe to call on a timer / cronIt enumerates every consumer group on the stream (via XINFO GROUPS) and trims (XTRIM MINID) only up to the oldest entry that any group still needs — the oldest pending entry of each group, or its delivered cursor if its PEL is empty, whichever is smaller across all groups. Everything below that floor has been delivered and acked by every group, so nothing live is lost.
- The most-behind group gates the trim. A group that hasn't consumed yet — including one freshly created with
start_from="0"— pins the floor at the head and blocks compaction until it catches up. That's deliberate: compaction never races ahead of a slow consumer. - It respects groups xstreamq didn't create (the enumeration is authoritative for the stream, not based on xstreamq's registry).
- It's a manual primitive — call it on whatever schedule fits. xstreamq does not auto-compact; retention stays an explicit decision.
One caveat: a brand-new
start_from="0"consumer created during acompact()call can race the trim. Create history-reading consumers before compacting.
healthy() is a bounded, end-to-end liveness check for a Kubernetes/ECS probe (or your own /health endpoint):
@app.get("/health")
async def health():
return {"ok": await dq.healthy()} # True only if Redis answers AND every loop is aliveIt returns True only when xstreamq is started, a PING round-trips to Redis within timeout (default 2s), and every background loop (each consumer plus the reclaimer) is still running. It never raises — any failure is False.
This is a different axis from the connection resilience defaults: those recover a dropped connection; healthy() reports status. It also catches what TCP keepalive can't — a Redis that's up but not serving (loading an RDB, BUSY on a script, mid-failover) keeps the socket alive yet fails the PING, and a consumer loop that has silently died is Redis-healthy but xstreamq-unhealthy. The PING is wrapped in a timeout so a wedged connection can't make the probe itself hang.
The same operations are available from the shell, behind the cli extra:
pip install 'xstreamq[cli]'xstreamq depth orders --group order-service # XLEN (+ pending/lag with --group)
xstreamq peek orders --count 5 # read the head of a stream (no consume)
xstreamq dlq ls orders --group order-service # list dead letters
xstreamq dlq redrive orders -g order-service --all # …or --id <id>
xstreamq dlq purge orders -g order-service --all --yes # irreversible; prompts without --yes
xstreamq compact orders # safe, no-loss trimConnection comes from --url or $XSTREAMQ_URL (default redis://localhost:6379). Note that --url is an option of the xstreamq command itself, so it goes before the subcommand: xstreamq --url redis://prod:6379 depth orders (a trailing xstreamq depth orders --url ... is rejected with "No such option"). Add --json (-j) to any subcommand for machine-readable output to pipe into jq.
xstreamq --help is the full reference. Every command opens a short-lived connection and runs no consumers, so inspecting never delivers or acks a message.
xstreamq's no-loss guarantees assume Redis itself doesn't lose data:
- AOF on,
appendfsync everysec - ideally a replica for failover
- idempotent handlers
A queue spends most of its life parked in a blocking read (XREADGROUP ... BLOCK, the reclaimer's PEL scans). If the connection is silently dropped — a managed-Redis failover, an idle-connection reaper, a NAT timeout — there's no FIN/RST, so that read can hang forever and the queue quietly stops. xstreamq guards against this out of the box: every client it opens (the main client and the reclaimer's) is built through one factory that enables TCP keepalive (a dead peer is detected in ~60s instead of the OS default of ~2h) and a health-check interval. You don't have to configure anything.
To tune or add connection options, pass connection_kwargs — forwarded to redis from_url for every client:
dq = XStreamQ(
"rediss://cache.example.com:6380",
connection_kwargs={
"ssl_cert_reqs": "required",
"retry_on_timeout": True,
"socket_timeout": 30.0, # ⚠️ must exceed your subscriber block_ms (default 2000ms)
},
)
⚠️ socket_timeoutandblock_ms.socket_timeoutapplies to the blockingXREADGROUPread. If you set it at or below a subscription'sblock_ms(default 2000ms), a perfectly healthy idle queue will raiseTimeoutErroron every read. Keepsocket_timeoutcomfortably aboveblock_ms. (This is why xstreamq does not set a defaultsocket_timeout— keepalive is the footgun-free default.)decode_responsesis pinned toTrueand cannot be overridden.
xstreamq uses the standard logging module under the xstreamq namespace and attaches no handlers of its own — your app stays in control of format and level. It's quiet by default (warnings and errors only): handler failures, DLQ routing, and recovery events. Raise the level to see more — at DEBUG you get a line per message received/acked plus reclaimer detail (verbose at high throughput).
import logging
logging.getLogger("xstreamq").setLevel(logging.DEBUG) # trace everything
logging.getLogger("xstreamq.consumer").setLevel(logging.WARNING) # …or quiet one modulexstreamq ships an Agent Skill under skills/xstreamq that teaches an AI coding agent (Claude Code, Copilot, Codex, …) how to set up, configure, and operate xstreamq — publishers/subscribers, retries/backoff/DLQ, dedupe, typed bodies, inspection, compaction, and the ops CLI.
Install it by copying the folder into your agent's skills directory, e.g. for Claude Code:
mkdir -p ~/.claude/skills && cp -R skills/xstreamq ~/.claude/skills/ # then run /skills to confirmSee skills/README.md for project-scope install, other agents, and usage examples.
make install # uv sync, including the dev group
make hooks # git pre-commit hooks (ruff + ty)
make redis-up # Redis on :6399 for the integration tests
make check # lint, types, format-check, and the suite under an 85% coverage gatemake check is the CI gate. Integration tests need a real Redis (set XSTREAMQ_TEST_REDIS_URL, default redis://localhost:6399); they are skipped if it is unreachable. make help lists everything else.
Design notes live in docs/DESIGN.md — the plan of record.
- v1 (this): publish / consume, retries + backoff, DLQ, atomic Lua move, the
xstreamq:registryset, opt-in typed bodies (xstreamq[pydantic]), connection resilience on by default, graceful shutdown,healthy(), atomic DLQ redrive / purge,start_from(consume from head or tail),compact(safe, no-loss stream trimming), opt-in idempotent dedupe (dedupe_ttl_s), and an ops CLI (xstreamq[cli]). - v2: a mountable web dashboard (queue depths, PEL, peek, DLQ contents) — its redrive / purge buttons call the atomic API that already ships in v1.
- later: scheduled / delayed delivery (a timestamp-scored ZSET + poller).
MIT