# Streams

> Append-only logs of field-value entries with range reads, blocking reads, and consumer groups.

URL: https://docs.hotkv.com/data-types/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](https://docs.hotkv.com/data-types/lists/index.md), 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](https://docs.hotkv.com/develop/pubsub/index.md), 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](https://docs.hotkv.com/commands/xadd/index.md) 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](https://docs.hotkv.com/commands/xdel/index.md)
removes single entries, and [XTRIM](https://docs.hotkv.com/commands/xtrim/index.md) 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](https://docs.hotkv.com/commands/del/index.md) 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](https://docs.hotkv.com/commands/xread/index.md) 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](https://docs.hotkv.com/commands/xreadgroup/index.md) 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](https://docs.hotkv.com/commands/xack/index.md) marks entries as done. `NOACK` on `XREADGROUP`
  skips the pending list for consumers that can afford to lose messages.
- [XPENDING](https://docs.hotkv.com/commands/xpending/index.md) shows what is waiting for
  acknowledgement, overall or per entry with idle times and delivery counts.
- [XCLAIM](https://docs.hotkv.com/commands/xclaim/index.md) and [XAUTOCLAIM](https://docs.hotkv.com/commands/xautoclaim/index.md) hand
  entries that have been pending too long to another consumer, so the work
  of a crashed worker is not lost.

[XGROUP](https://docs.hotkv.com/commands/xgroup/index.md) also destroys groups, creates and deletes
consumers, and moves a group's position (`SETID`).
[XINFO](https://docs.hotkv.com/commands/xinfo/index.md) 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](https://docs.hotkv.com/commands/xadd/index.md) | Append an entry, optionally trimming |
| [XLEN](https://docs.hotkv.com/commands/xlen/index.md) | Number of entries |
| [XRANGE](https://docs.hotkv.com/commands/xrange/index.md) / [XREVRANGE](https://docs.hotkv.com/commands/xrevrange/index.md) | Entries in an ID range |
| [XREAD](https://docs.hotkv.com/commands/xread/index.md) | Read from one or more streams, optionally blocking |
| [XDEL](https://docs.hotkv.com/commands/xdel/index.md) | Delete entries |
| [XTRIM](https://docs.hotkv.com/commands/xtrim/index.md) | Trim to a length or minimum ID |
| [XSETID](https://docs.hotkv.com/commands/xsetid/index.md) | Set the stream's last ID |
| [XGROUP CREATE](https://docs.hotkv.com/commands/xgroup-create/index.md) | Create a consumer group |
| [XGROUP DESTROY](https://docs.hotkv.com/commands/xgroup-destroy/index.md) | Delete a consumer group |
| [XGROUP SETID](https://docs.hotkv.com/commands/xgroup-setid/index.md) | Move a group's position |
| [XGROUP CREATECONSUMER](https://docs.hotkv.com/commands/xgroup-createconsumer/index.md) / [XGROUP DELCONSUMER](https://docs.hotkv.com/commands/xgroup-delconsumer/index.md) | Add or remove a consumer |
| [XREADGROUP](https://docs.hotkv.com/commands/xreadgroup/index.md) | Read as a member of a group |
| [XACK](https://docs.hotkv.com/commands/xack/index.md) | Acknowledge entries |
| [XPENDING](https://docs.hotkv.com/commands/xpending/index.md) | Entries waiting for acknowledgement |
| [XCLAIM](https://docs.hotkv.com/commands/xclaim/index.md) / [XAUTOCLAIM](https://docs.hotkv.com/commands/xautoclaim/index.md) | Take over pending entries |
| [XINFO STREAM](https://docs.hotkv.com/commands/xinfo-stream/index.md) / [XINFO GROUPS](https://docs.hotkv.com/commands/xinfo-groups/index.md) / [XINFO CONSUMERS](https://docs.hotkv.com/commands/xinfo-consumers/index.md) | Inspect a stream |

## From your application

**Python**

```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**

```ts
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**

```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**

```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);
}
```