Streams

A stream is an append-only log: every entry has an identifier the server assigns and a set of fields. It differs from a list in that reading does not take the entry away, and consumer groups make Redis itself keep the books: who took what, and what has not been acknowledged.

What a stream is

An entry looks like this: the identifier 1755600000123-0 (milliseconds of the server clock and a sequence number within that millisecond) and the fields {"type": "signup", "user": "42"}. Identifiers increase monotonically, which makes “read everything after this one” an ordinary operation rather than a search.

The properties everything else follows from:

• Reading takes nothing away. An entry lives until it is deleted or trimmed away. Any number of independent readers can see the same one.
• Order is guaranteed and matches the order entries were added in.
• The log grows forever unless bounded. That is the price of keeping history.
• Groups are the server’s business. It remembers which entry went to which consumer and whether it was acknowledged. A list cannot do that — there the bookkeeping is yours.

How this differs from a list

If you need a queue and no history, take a list: simpler and cheaper. Streams are for history, several independent readers, or built-in delivery tracking.

Where streams belong

An event bus. One event, several independent handlers: an email, a metric, a webhook. Each reads through its own group without disturbing the others.

A queue with a guarantee. A worker took an entry and died — it stays in the group’s unfinished list and goes to another worker. Exactly what has to be assembled from moveTo() and remove() with lists.

A change log. An audit trail, an order’s history, an activity feed: re-read it, rewind it, reconstruct an incident from it.

A buffer with catch-up reads. A handler writes, a consumer reads at its own pace and after a restart continues where it stopped.

Where a stream is the wrong choice:

The handle

$events = $store->stream('events');    // the server sees 'session:events'

stream() executes nothing — it is a view, not a request. The prefix is applied here, once. The handle holds no connection either, with one exception: follow() opens one of its own, because a blocking read occupies its connection for the whole wait.

Entries

Reading methods return StreamEntry:

foreach ($events->range(count: 100) as $entry) {
  $entry->id;                  // '1755600000123-0'
  $entry->fields;              // ['type' => 'signup', 'user' => '42']
  $entry->get('type');         // 'signup'
  $entry->get('absent', '—');  // with a default
  $entry->has('user');         // bool
  $entry->timestamp();         // 1755600000123 — milliseconds of the server clock
}

The driver hands entries over as ['1755600000123-0' => ['type' => 'signup']] — an array whose single key is the identifier. Iterating that means reaching for key()/current(), and the identifier is needed constantly: it is what acknowledges work and what a reader remembers as its position.

RedisStream reference

Writing

add()

public function add(array $fields, ?int $cap = null, bool $exact = false, string $id = '*'): string

Appends an entry to the end of the log.

Returns the identifier the entry was given. Throws RedisCommandException when the key holds a structure of another type.

$events->add(['type' => 'signup', 'user' => '42']);
// '1755600000123-0'

$events->add(['level' => 'warn', 'msg' => $text], cap: 10_000);

Your own identifiers are almost never needed: the server’s are monotonic and carry time. Assigning them by hand only makes sense when importing from another log.

trim()

public function trim(int $cap, bool $exact = false): int

Keeps roughly the newest $cap entries.

Returns how many entries were removed.

$events->trim(10_000);                 // cheap, the length is "about right"
$events->trim(10_000, exact: true);    // exactly that many, more expensive

trimBefore()

public function trimBefore(string $id, bool $exact = false): int

Removes everything older than the given identifier.

Returns how many entries were removed.

This is how a log is bounded by time rather than by count: an identifier begins with the milliseconds of the server clock, so “older than a day” is an identifier:

$events->trimBefore((string) ((time() - 86400) * 1000));

delete()

public function delete(string ...$ids): int

Removes entries by identifier. Returns how many existed. With no arguments, 0.

Deleting from the middle of a log is allowed but exceptional: the ordinary way to bound a stream is trimming. delete() is for the entry that must not be readable again.

Reading

count()

public function count(): int

The number of entries; 0 for a missing key. No arguments.

range()

public function range(?string $from = null, ?string $to = null, ?int $count = null): array

Entries in the order they were added, taking nothing out.

Returns list<StreamEntry>; an empty array for an empty log.

$events->range();                    // everything
$events->range(count: 100);          // the first hundred
$events->range($since, count: 50);   // from a known position

reverse()

public function reverse(?int $count = null): array

The same, newest first — the cheap way to look at what just happened.

$events->reverse(count: 10);   // the last ten, freshest first

after()

public function after(string $id, int $count = 10): array

Entries added after the given identifier, without waiting.

Returns list<StreamEntry>; an empty array when there is nothing new.

This is the building block of a reader that keeps its own position: remember the identifier of the last entry handled, pass it back next time. Two such readers do not disturb each other — a stream is not consumed by reading.

$position = $this->positions->get('events') ?? '0';

foreach ($events->after($position, count: 100) as $entry) {
  $this->handle($entry);
  $position = $entry->id;
}

$this->positions->set('events', $position);

follow()

public function follow(float $timeout = 0.0, int $count = 10, ?string $from = null): array

Waits for entries newer than the ones this handle has already seen.

Returns list<StreamEntry>; an empty array when nothing arrived in time.

$events = $store->stream('events');

while (!$this->stopping) {
  foreach ($events->follow(timeout: 5) as $entry) {
      $this->handle($entry);
  }
}

$events->close();

Observation

info()

public function info(): array

The server’s summary of the log: length, first-entry, last-entry, last-generated-id, the number of groups. No arguments.

$events->info()['length'];              // 1204
$events->info()['last-generated-id'];   // '1755600000123-0'

groups()

public function groups(): array

Every consumer group on this stream as the server sees it: name, consumers, pending, last-delivered-id, lag.

foreach ($events->groups() as $group) {
  if ($group['pending'] > 1000) {
      $this->alert("group {$group['name']} is falling behind");
  }
}

lag is how many entries the group has not seen yet; together with pending these are two different kinds of falling behind: “not read” and “read but not acknowledged”.

The key as a whole

$events->expireKey(86400);   // a lifetime for the whole log
$events->keyTtl();           // 86400 | null
$events->deleteKey();        // delete the log together with its groups
$events->name();             // 'session:events'
$events->close();            // close the connection follow() opened

deleteKey() takes the entries and the groups with their bookkeeping. Entries have no lifetimes of their own — trimming bounds them.

Consumer groups

What they are for

Without a group every reader sees all entries and tracks its own position. A group changes both halves: an entry goes to one consumer of the group, and the server holds it in an unfinished list until it is acknowledged.

Hence the guarantee they are taken for: a worker took an entry and died — it is not lost. It is on record against that worker, visible in pending(), and another worker can claim it.

$events->ensureGroup('mailers');                          // once, at startup
$group = $events->group('mailers', consumer: 'worker-1');

ensureGroup()

public function ensureGroup(string $name, string $from = '0'): bool

Returns true when the group was created and false when it already existed — the second is not an error. The stream is created too if needed, so a group can be declared at startup before the first entry is written.

group()

public function group(string $name, string $consumer = 'default'): RedisStreamGroup

A handle on the group, acting as one consumer. Creates nothing and sends nothing.

RedisStreamGroup reference

Consuming

consume()

public function consume(int $count = 10, float $timeout = 0.0): array

Takes entries nobody in the group has been given yet.

Returns list<StreamEntry>. Throws RedisCommandException with NOGROUP when the group does not exist.

Each entry goes to exactly one consumer of the group and enters that consumer’s unfinished list until it is acknowledged.

$group->consume(count: 10, timeout: 5);   // waits, on its own connection
$group->consume(timeout: -1);             // look and return, on a pooled connection

A negative timeout is the only mode that does not take a dedicated connection; it is for checks and one-off looks.

backlog()

public function backlog(int $count = 10): array

Returns what this consumer has already taken but not acknowledged.

Returns list<StreamEntry>, without waiting.

This is the first thing a restarted worker should do: entries it took before the restart are on record against it and will not arrive through consume() again. Reading them back is how the work resumes instead of waiting for someone to claim them on idle.

$group = $events->group('mailers', consumer: $this->workerName());

// 1. finish your own first
foreach ($group->backlog() as $entry) {
  $this->handle($entry);
  $group->ack($entry);
}

// 2. and only then take new work
while (!$this->stopping) {
  foreach ($group->consume(count: 10, timeout: 5) as $entry) {
      $this->handle($entry);
      $group->ack($entry);
  }
}

$group->close();

ack()

public function ack(StreamEntry|string ...$entries): int

Marks entries as handled, removing them from the unfinished list.

Returns how many were unfinished and are now acknowledged. With no arguments, 0.

$group->ack($entry);              // the object
$group->ack($entry->id);          // or the identifier
$group->ack(...$entries);         // or a batch

The package will not acknowledge automatically on delivery: that would destroy the single guarantee a group exists for — that an entry outlives the worker that took it and never finished.

Recovery

pending()

public function pending(int $count = 100, ?string $consumer = null): array

Entries the group delivered and nobody acknowledged.

Returns list<PendingEntry>:

foreach ($group->pending() as $stuck) {
  $stuck->id;           // '1755600000123-0'
  $stuck->consumer;     // 'worker-1'
  $stuck->idleMs;       // 61240 — milliseconds since it was delivered
  $stuck->deliveries;   // 3 — how many times it has been delivered
}

The two numbers answer different questions. A large idleMs with a single delivery means the consumer died holding it. A climbing deliveries means the entry breaks whoever takes it.

What to do about the second — retry, alert, bury — is the application’s decision: only the author of the work knows what a retry costs.

pendingCount()

public function pendingCount(): int

How many entries the group has delivered without getting an acknowledgement. No arguments.

claimStale()

public function claimStale(int $idle, int $count = 10, string $from = '0-0'): array

Takes over entries that have been idle too long in someone else’s hands.

Returns list<StreamEntry> — the entries now claimed, ready to be handled.

foreach ($group->claimStale(idle: 60_000) as $entry) {
  $this->handle($entry);
  $group->ack($entry);
}

Pick the threshold from the longest honest piece of work: if sending an email takes up to thirty seconds, idle: 60_000 will not steal work from a living worker.

consumers()

public function consumers(): array

Every consumer of the group as the server sees it: name, pending, idle. No arguments.

foreach ($group->consumers() as $consumer) {
  if ($consumer['idle'] > 300_000 && $consumer['pending'] > 0) {
      $this->alert("{$consumer['name']} has been silent for five minutes holding work");
  }
}

destroy()

public function destroy(): bool

Removes the group with its bookkeeping and its consumers. The log’s entries stay.

Utility

$group->as('worker-2');    // the same group, as another consumer
$group->name();            // 'mailers'
$group->consumer();        // 'worker-1'
$group->close();           // close the connection a blocking consume() opened

Mapping to Redis commands

What is not here — XCLAIM one by one, XSETID, XGROUP CREATECONSUMER — is available through raw() together with name().

Next

• Lists — a queue when history is not needed
• Hashes — an object’s state alongside a log of events
• Connection pool — why blocking reads stay away from the pool