Streams

A stream is an append-only log. Each entry has a unique, increasing ID and a small set of field-value pairs. You can read a stream by ID range, follow it in real time with a blocking read, and split the work of processing it across a group of consumers that acknowledge what they have handled. Streams fit event sourcing, activity and audit logs, sensor readings, and work queues that must not lose a message when a worker crashes.

Compared with lists, a stream keeps entries after they are read, lets many independent readers each keep their own position, and tracks which entries each consumer has received but not yet acknowledged. Compared with Pub/Sub, a stream stores messages, so a reader that was offline can catch up.

Entry IDs

An ID has the form <milliseconds>-<sequence>, for example 1767225600000-0. Pass * to XADD and HotKV assigns the next ID from its clock; you can also give an explicit ID, which must be greater than the newest entry’s. Range commands accept - (the smallest possible ID) and + (the largest), and an ID without the sequence part covers every sequence number in that millisecond.

Create, read, update, delete

The examples on this page use *, so the IDs your server returns will differ.

127.0.0.1:6379> XADD orders * id 1001 total 42.50
"1767225600000-0"
127.0.0.1:6379> XADD orders * id 1002 total 18.00
"1767225600000-1"
127.0.0.1:6379> XLEN orders
(integer) 2
127.0.0.1:6379> XRANGE orders - +
1) 1) "1767225600000-0"
   2) 1) "id"
      2) "1001"
      3) "total"
      4) "42.50"
2) 1) "1767225600000-1"
   2) 1) "id"
      2) "1002"
      3) "total"
      4) "18.00"
127.0.0.1:6379> XREVRANGE orders + - COUNT 1
1) 1) "1767225600000-1"
   2) 1) "id"
      2) "1002"
      3) "total"
      4) "18.00"
127.0.0.1:6379> XDEL orders 1767225600000-0
(integer) 1

Entries cannot be edited: append a new one instead. XDEL removes single entries, and XTRIM removes the oldest ones in bulk. Unlike other types, a stream that becomes empty stays in place (with its last ID and consumer groups); use DEL to remove it.

Keeping a stream bounded

Trim as you append, either to a length or to a minimum ID (a point in time):

127.0.0.1:6379> XADD sensor:1 MAXLEN ~ 10000 * temp 21.5
"1767225600000-0"
127.0.0.1:6379> XTRIM sensor:1 MINID 1767225000000
(integer) 0

With ~, trimming is approximate: HotKV may keep slightly more entries than asked when that is cheaper, and LIMIT caps how many entries one call removes. With = (the default) the limit is exact. NOMKSTREAM makes XADD do nothing when the stream does not exist yet.

Reading new entries as they arrive

XREAD reads entries after a given ID from one or more streams. With BLOCK <milliseconds> it waits for new entries, and $ means “only entries added from now on”.

127.0.0.1:6379> XREAD COUNT 10 BLOCK 5000 STREAMS orders $
(nil)

The call above waited five seconds and nothing arrived. When an entry does arrive, the reply lists it under its stream’s name. To continue reading, pass the last ID you received instead of $.

Consumer groups

A consumer group shares a stream among several workers: each entry goes to only one consumer in the group, and HotKV remembers which entries each consumer has been given until it acknowledges them.

127.0.0.1:6379> XGROUP CREATE orders billing 0
OK
127.0.0.1:6379> XREADGROUP GROUP billing worker-1 COUNT 1 STREAMS orders >
1) 1) "orders"
   2) 1) 1) "1767225600000-1"
         2) 1) "id"
            2) "1002"
            3) "total"
            4) "18.00"
127.0.0.1:6379> XPENDING orders billing
1) (integer) 1
2) "1767225600000-1"
3) "1767225600000-1"
4) 1) 1) "worker-1"
      2) "1"
127.0.0.1:6379> XACK orders billing 1767225600000-1
(integer) 1

• XGROUP CREATE key group <id> starts the group after <id>: 0 to process the whole stream, $ for new entries only. Add MKSTREAM to create the stream if it does not exist.
• > in XREADGROUP asks for entries never delivered to this group. Passing an ID instead re-reads this consumer’s own pending entries, which is how a restarted worker resumes.
• XACK marks entries as done. NOACK on XREADGROUP skips the pending list for consumers that can afford to lose messages.
• XPENDING shows what is waiting for acknowledgement, overall or per entry with idle times and delivery counts.
• XCLAIM and XAUTOCLAIM hand entries that have been pending too long to another consumer, so the work of a crashed worker is not lost.

XGROUP also destroys groups, creates and deletes consumers, and moves a group’s position (SETID). XINFO reports on a stream, its groups, and their consumers.

Common patterns

Event log. XADD events * type signup user 42 for each event, with MAXLEN ~ or MINID to bound storage. Services read with XREAD BLOCK from their own last-seen ID.

Reliable work queue. One consumer group, one consumer name per worker process. Workers loop on XREADGROUP ... BLOCK, process, then XACK. A periodic XAUTOCLAIM moves entries idle for longer than your timeout to a live worker.

Fan-out to several services. Give each service its own consumer group on the same stream: every group sees every entry, and within a group the work is shared.

Expiry

A time to live applies to the whole stream. Entries do not expire on their own: use XADD ... MINID or XTRIM ... MINID with a timestamp to drop old entries.

Limits and performance

Field names and values are strings up to proto-max-bulk-len (512 MB by default). Appending is fast, and range reads cost a lookup plus the entries returned. Keep consumer groups’ pending lists short by acknowledging promptly; a long pending list makes XPENDING and claiming slower.

Compatibility

Stream commands, IDs, consumer groups and replies match Redis 7 and Valkey. HotKV does not have the newer Redis additions XACKDEL, XDELEX, XCFGSET, XIDMPRECORD, or XADD’s KEEPREF, DELREF, ACKED and idempotency options.

Commands

Command	What it does
XADD	Append an entry, optionally trimming
XLEN	Number of entries
XRANGE / XREVRANGE	Entries in an ID range
XREAD	Read from one or more streams, optionally blocking
XDEL	Delete entries
XTRIM	Trim to a length or minimum ID
XSETID	Set the stream’s last ID
XGROUP CREATE	Create a consumer group
XGROUP DESTROY	Delete a consumer group
XGROUP SETID	Move a group’s position
XGROUP CREATECONSUMER / XGROUP DELCONSUMER	Add or remove a consumer
XREADGROUP	Read as a member of a group
XACK	Acknowledge entries
XPENDING	Entries waiting for acknowledgement
XCLAIM / XAUTOCLAIM	Take over pending entries
XINFO STREAM / XINFO GROUPS / XINFO CONSUMERS	Inspect a stream

From your application

Python

import hotkv

client = hotkv.HotKV(host="localhost", port=6379, decode_responses=True)

client.xadd("orders", {"id": "1001", "total": "42.50"}, maxlen=10000, approx=True)
client.xgroup_create("orders", "billing", id="0", mkstream=True)

batch = client.xreadgroup("billing", "worker-1", {"orders": ">"}, count=10, block=5000)
print(batch)
# After processing each entry:
# client.xack("orders", "billing", entry_id)
client.close()

Node.js

import { HotKV } from "@hotkv/client";

const client = new HotKV({ url: "hotkv://127.0.0.1:6379" });

await client.xAdd("orders", { id: "1001", total: "42.50" });
await client.xGroupCreate("orders", "billing", "0", true);

const batch = await client.xReadGroup("billing", "worker-1", { orders: ">" }, { count: 10, blockMs: 5000 });
console.log(batch);
// After processing each entry:
// await client.xAck("orders", "billing", entryId);
await client.close();

Go

client, err := hotkv.NewClient(&hotkv.Options{Addr: "localhost:6379"})
if err != nil {
    log.Fatal(err)
}
defer client.Close()
ctx := context.Background()

id, _ := client.XAdd(ctx, "orders", "*", "id", "1001", "total", "42.50").Result()
fmt.Println(id)

entries, _ := client.XRange(ctx, "orders", "-", "+", 10).Result()
fmt.Println(entries)

In the Go client, send XGROUP, XREADGROUP and XACK through a pipeline’s Do.

Java

import com.hotkv.HotKV;
import java.util.Map;

try (HotKV client = HotKV.create("hotkv://localhost:6379")) {
    client.xadd("orders", Map.of("id", "1001", "total", "42.50"));
    client.xgroupCreate("orders", "billing", "0", true);

    var batch = client.xreadGroup("billing", "worker-1", Map.of("orders", ">"), 10L, 5000L, false);
    System.out.println(batch);
    // After processing each entry:
    // client.xack("orders", "billing", entryId);
}