Durable Streams

Real-time streams with durable, replayable history on Rivet: create streams, write and read them, follow them live, fork them, and use the HTTP API.

Durable Streams is an open standard for real-time streams with durable, replayable history. Rivet provides an implementation backed by Rivet Actors.

Quickstart

npx @rivet-dev/services dev

npm install @durable-streams/client

import { DurableStream } from "@durable-streams/client";

const stream = await DurableStream.create({
	url: "http://127.0.0.1:8642/durable-streams/v1/stream/demo",
	contentType: "application/json",
});

await stream.append(JSON.stringify({ message: "hello" }));

// Read from the beginning, then keep receiving new messages live
const res = await stream.stream<{ message: string }>();
res.subscribeJson(async (batch) => {
	for (const item of batch.items) console.log(item.message);
});

curl -i -X PUT \
  -H 'content-type: application/json' \
  --data '[{"message":"hello"}]' \
  http://127.0.0.1:8642/durable-streams/v1/stream/demo

curl -i -X POST \
  -H 'content-type: application/json' \
  --data '{"message":"world"}' \
  http://127.0.0.1:8642/durable-streams/v1/stream/demo

curl 'http://127.0.0.1:8642/durable-streams/v1/stream/demo?offset=-1'

Streams

A stream is an append-only log at its own URL. Writers add to the end, and every reader sees the same messages in the same order.

The rest of this page follows one example: an AI agent streams its reply into chat/session-1, and a browser shows it as it arrives.

Create a stream

Pick any path. Slashes are fine, and there is nothing to register first.

import { DurableStream } from "@durable-streams/client";

// One stream per chat session
const stream = await DurableStream.create({
	url: "http://127.0.0.1:8642/durable-streams/v1/stream/chat/session-1",
	contentType: "application/json",
	// Delete the stream after a day with no reads or writes
	ttlSeconds: 86400,
});

console.log("created", stream.url);

Creating the same stream twice is safe. Creating it again with different settings returns 409 Conflict.

Choose JSON or bytes

The content type is set at creation and never changes.

• application/json: each append is one message, and reads return a JSON array. Use it for chat, events, and agent output.
• Anything else, such as text/plain or application/octet-stream: one continuous run of bytes. Frame messages yourself, for example one JSON object per line.

Clean up automatically

Set one of these when you create the stream, or neither to keep it until you delete it:

• ttlSeconds: delete after this many seconds with no reads or writes. An active stream never expires.
• expiresAt: delete at a fixed time, such as 2027-01-01T00:00:00Z.

To delete a stream now, call stream.delete().

On Rivet, each stream is its own Rivet Actor, so streams scale independently and sleep when no one is using them.

Writing

Writers add messages to the end of a stream. Once a write is acknowledged it is stored, and every reader sees it in the same place.

Append messages

In a JSON stream, each append is one message. Close the stream when the reply is complete, optionally with a last message in the same call.

import { DurableStream } from "@durable-streams/client";

const stream = new DurableStream({
	url: "http://127.0.0.1:8642/durable-streams/v1/stream/chat/session-1",
	contentType: "application/json",
});

// Each append is one message
await stream.append(JSON.stringify({ token: "It's" }));
await stream.append(JSON.stringify({ token: " sunny" }));

// Close with a final message: readers know the reply is complete
await stream.close({ body: JSON.stringify({ token: "." }) });

Retry without duplicates

If a response is lost, a plain append cannot tell whether it landed, and a retry may write it twice. An idempotent producer numbers every write so the server drops repeats.

import { DurableStream, IdempotentProducer } from "@durable-streams/client";

const stream = new DurableStream({
	url: "http://127.0.0.1:8642/durable-streams/v1/stream/chat/session-1",
	contentType: "application/json",
});

// Keep the producer ID the same across restarts. With autoClaim, a restarted
// agent takes over and the old one can no longer write.
const producer = new IdempotentProducer(stream, "agent-session-1", {
	autoClaim: true,
	onError: (error) => console.error("append failed", error),
});

for (const token of ["It's", " sunny", " today"]) {
	producer.append(JSON.stringify({ token }));
}

// Wait until every token is stored, then close the stream
await producer.flush();
await producer.close();

• Producer ID: a stable name for the writer. Reuse it after a restart.
• autoClaim: a restarted writer takes over, and the old one is rejected, so a stalled process cannot write over its replacement.

Rivet remembers each producer for seven days after its last write. A retry sent later than that may be written again.

Finish the reply

Closing tells readers no more data is coming. Close a stream when the work it records is done, such as a finished agent reply.

• Call stream.close(), or producer.close() when writing through a producer.
• A closed stream stays readable, but new appends are rejected.
• Closing is permanent.

Keep writes in your own order

If your writes carry their own sequence, such as database commit numbers, pass it as seq on each append. The server rejects any write whose seq is not greater than the last one. Values compare as strings, so pad numbers to a fixed width, such as 00000042.

Reading

Readers start at an offset and move forward. An offset is a position in the stream, and every read tells you the offset to continue from.

Load everything so far

import { stream } from "@durable-streams/client";

const url = "http://127.0.0.1:8642/durable-streams/v1/stream/chat/session-1";

// Everything written so far, then stop
const res = await stream<{ token: string }>({ url, offset: "-1", live: false });
const messages = await res.json();

console.log(messages);
console.log("read from here next:", res.offset);

res.offset is where the next read should start.

Pick up where you left off

Save the latest offset, for example in localStorage, and pass it back as offset next time. You get every message after it: nothing missed, nothing repeated.

• -1 starts at the beginning.
• now skips history and starts at the end.

Offsets are opaque strings. Store them and send them back as-is.

Know when a stream is done

A closed stream tells readers that nothing more will arrive, so they can stop waiting. Without closing, a reader cannot tell “nothing yet” from “nothing ever”.

Realtime streaming

A live read catches up from an offset, then keeps the connection open and delivers new messages as they are written. This is how the browser shows the agent’s reply token by token.

Follow a stream live

import { stream } from "@durable-streams/client";

const url = "http://127.0.0.1:8642/durable-streams/v1/stream/chat/session-1";

// "-1" starts at the beginning. Pass a saved offset to pick up where you left off.
const savedOffset: string = "-1";

const res = await stream<{ token: string }>({ url, offset: savedOffset, live: "sse" });
res.subscribeJson(async (batch) => {
	for (const item of batch.items) console.log(item.token);

	// Save this to resume from the same place after a refresh
	console.log("offset", batch.offset);
});

The callback runs for each batch, first with the history after your offset, then with every new write.

Reconnect without gaps

Connections drop: a laptop sleeps, a phone changes networks, a tab refreshes. Save batch.offset as you go and pass it back as offset when you reconnect. The read resumes right after the last message you saw.

The client also reconnects on its own while the page stays open.

Transport methods

• Server-Sent Events (live: "sse"): one open connection, and the server pushes each write as it lands. Lowest latency. Byte streams arrive base64-encoded, and the client decodes them.
• Long-polling (live: "long-poll" or live: true): each request waits for new data, then the client sends the next one. Use it when a proxy buffers streaming responses.
• Both deliver the same messages and offsets. See Live reads to use them without the client.

Stop when the stream closes

When the writer closes the stream, a live read stops once it reaches the end.

Forking

A fork is a new stream that starts with another stream’s history, then goes its own way. Use it to retry an agent from partway through a conversation, or to try two answers side by side.

Fork a stream

Create the new stream with Stream-Forked-From set to the source stream’s path:

curl -X PUT \
  -H 'Stream-Forked-From: /durable-streams/v1/stream/chat/session-1' \
  http://127.0.0.1:8642/durable-streams/v1/stream/chat/session-1-retry

The TypeScript client does not create forks yet. Once the fork exists, read and write it like any other stream.

Fork from an earlier point

By default a fork copies the whole history. Add Stream-Fork-Offset with an offset from a previous read to branch from that point instead, for example right before the agent’s last reply.

What a fork shares

• The fork keeps the source’s content type and history up to the fork point.
• History is shared, not copied, so forking a long stream is cheap.
• Writes to the fork never touch the source, and later writes to the source do not appear in the fork.

HTTP API

Every stream is a URL, and every operation is a plain HTTP request on it:

http://127.0.0.1:8642/durable-streams/v1/stream/<path>

Any HTTP client works, and existing Durable Streams clients work unchanged. The Durable Streams protocol covers every edge case.

Requests

STREAM=http://127.0.0.1:8642/durable-streams/v1/stream/chat/session-1

curl -X PUT -H 'content-type: application/json' $STREAM
curl -X POST -H 'content-type: application/json' --data '{"token":"hello"}' $STREAM
curl "$STREAM?offset=-1"
curl -I $STREAM
curl -X DELETE $STREAM

Creating

Appending

In a JSON stream, posting a JSON array stores each element as its own message, so one request can carry a batch.

Reading over HTTP

Live reads

• live=sse: Server-Sent Events. data events carry messages and control events carry streamNextOffset, upToDate, and closed status. The server ends connections periodically; reconnect from the last streamNextOffset. For byte streams, data is base64-encoded and the response carries stream-sse-data-encoding: base64.
• live=long-poll: the request waits until data arrives, then responds like a normal read. Returns 204 on timeout; retry with the same offset.

curl -N "$STREAM?offset=-1&live=sse"

Status codes