Redis Streams

An append-only log, inside your key/value store

Every data structure you've built so far (strings, lists, hashes, sets) is mutable in place — SET overwrites, LPUSH reorders. A stream is different: it's an append-only sequence of immutable entries, each with a unique, ordered ID. That single property — strictly increasing IDs, never rewritten — is what makes streams useful for event logs, activity feeds, and message queues: consumers can ask "give me everything after ID X" and get a stable, replayable answer, the same guarantee Kafka gives you, but living inside Redis.

Entry IDs: the ordering primitive

Real Redis IDs are <milliseconds>-<sequence> — the wall-clock time the entry was added, plus a per-millisecond sequence number to break ties when multiple entries land in the same tick. For test determinism, this exercise swaps wall-clock time for a simple monotonically increasing counter: 1-0, 2-0, 3-0, ...

next_id = 1

def xadd_auto(stream, fields):
    entry_id = f"{next_id}-0"
    next_id += 1
    streams[stream].append((entry_id, fields))
    return entry_id

Explicit IDs (XADD s 5-0 field val) are allowed too, but must be strictly greater than the stream's current last ID — streams only ever grow forward. Comparing IDs means comparing the (ms, seq) pair lexicographically, not the raw string (so "10-0" correctly sorts after "9-0").

def id_tuple(id_str):
    ms, seq = id_str.split("-")
    return (int(ms), int(seq))

Reject with the exact Redis error text when an explicit ID doesn't advance the stream — this is one of the few commands where the wire-level error message itself is part of the contract other tools parse.

Storage shape

A stream is just a list of (id, [field1, val1, field2, val2, ...]) tuples, kept in insertion order (which, because IDs are monotonic, is automatically ID order too):

streams = {}   # stream_name -> [(id, flat_fields), ...]

XRANGE: bounded scans

XRANGE key start end returns every entry whose ID falls in [start, end]. The special tokens - and + mean "smallest possible ID" and "largest possible ID" respectively — so XRANGE key - + is "give me the whole stream," the same convention -inf/+inf play in sorted-set range queries. Since entries are already stored in ID order, this is a straightforward linear filter (or binary search once your stream is large).

XREAD: tailing the log

XREAD COUNT n STREAMS key last_id is how a consumer follows a stream: "give me up to n entries with an ID greater than the last one I've already seen." This is the read pattern that turns an append-only log into a message queue — a consumer just remembers the last ID it processed and keeps asking for "what's new since then," which is exactly how a poll-based worker loop or a live activity feed stays caught up without re-reading history.

def xread(stream, last_id, count):
    last = id_tuple(last_id)
    matches = [(i, f) for i, f in streams[stream] if id_tuple(i) > last]
    return matches[:count]

If nothing new has arrived, the reply is a null bulk ($-1\r\n) rather than an empty array — callers use this to distinguish "no stream" / "nothing new yet" from "here's an empty-but-valid result."

Where this leads

This exercise stops short of consumer groups — the feature that lets multiple workers cooperatively split a stream's workload with acknowledgment and at-least-once delivery (Redis's answer to Kafka's consumer groups / SQS's visibility timeout). But XADD + XRANGE + XREAD is the storage and traversal foundation everything else is built on: get entry ordering and ID validation right here, and consumer groups are "just" bookkeeping on top of these same primitives.