streams

redis streams = append-only log + consumer groups. it's the right primitive when you want at-least-once delivery, horizontal worker scaling, and the ability to ack/replay individual messages.

the four operations that matter

XADD <key> * <field> <value> ... # append (auto-generated id) XGROUP CREATE <key> <group> 0 MKSTREAM # create a consumer group XREADGROUP GROUP <group> <consumer> # claim some messages COUNT N BLOCK ms STREAMS <key> > XACK <key> <group> <id> [id...] # finalize delivery

read group with > for "new messages since last_delivered_id". any other id (e.g. 0) means "replay this consumer's PEL from that id forward".

stream ids

{milliseconds_since_epoch}-{sequence}. monotonic across XADDs; if two XADDs land in the same ms, the seq increments. you compare them lex-order on (ms, seq).

the PEL is the load-balancing primitive

each consumer group has a Pending Entries List per consumer. when you XREADGROUP with >, messages move from "deliverable" to "in this consumer's PEL". XACK removes them. while in the PEL they're invisible to other XREADGROUP > calls in the same group — that's how multiple workers share a stream without stepping on each other.

worker A: XREADGROUP GROUP g cA COUNT 1 STREAMS s > -> [(1, fields)] worker B: XREADGROUP GROUP g cB COUNT 1 STREAMS s > -> [(2, fields)] worker A: XACK s g 1 -> 1 (removed from PEL)

if a worker dies between read and ack, its PEL entries sit there forever — that's the orphan-recovery problem.

XAUTOCLAIM for orphans

XAUTOCLAIM <key> <group> <consumer> <min-idle-ms> <start-id> [COUNT N]

walks every consumer's PEL in the group, picks entries idle for >= min-idle-ms AND id >= start-id, transfers them to <consumer>, returns (next-cursor, claimed-entries, deleted-ids).

if the underlying stream entry has been trimmed away (XDEL or XTRIM MAXLEN), it lands in deleted-ids — caller XACKs those to clean up the PEL. this matters: without that path, trimmed-but-pending entries can never be drained.

the next-cursor lets you paginate through the PEL on a clock — call it every minute with the previous cursor (start at 0-0).

XTRIM vs XDEL

• XTRIM MAXLEN N — keep only the most recent N entries; older ones get dropped. used for bounded retention.
• XTRIM MINID id — drop everything older than id. used for time-based retention.
• XDEL stream id [id...] — surgical, removes specific ids. note: doesn't touch the PEL — workers that haven't ack'd those ids still have them pending. XAUTOCLAIM will surface them as deleted-ids.

docket's specific use

docket (a distributed-task system built on streams + Lua) uses streams like this:

• {prefix}:stream — the work queue. one entry per ready-to-execute task.
• {prefix}:queue (sorted set) — scheduled-for-future tasks, score = when_micros.
• {prefix}:parked:<uuid> (hash) — fields for a parked task (field name → value).
• {prefix}:runs:<uuid> (hash) — terminal state observability.

three atomic Lua scripts compose them:

1. schedule.lua — if when ≤ now: XADD; else: HSET parked + ZADD queue
2. promote_due.lua — ZRANGEBYSCORE queue 0 now → HGETALL parked → XADD stream → DEL parked → HSET runs state=queued → ZREMRANGEBYSCORE
3. ack.lua — XACK + XDEL + HSET runs state=completed/failed

the dual "stream + ZSET + parked hashes" structure exists because XADD itself can't take a "deliver this at time T" parameter. so future-scheduled work lives in a ZSET keyed by score, and a periodic Lua script promotes due entries onto the stream.

reply parsing

XREADGROUP reply has 4 levels:

[[stream_name, [[entry_id, [field1, value1, field2, value2, ...]], [entry_id, [...]], ...]], [next_stream, [...]]]

every level is a RESP array, leaves are bulks. the parser has to walk the whole tree; the memory model has to either copy everything or hold the read buffer alive.