Skip to main content
Sign In
Concepts

Writing

Append to a stream, retry without duplicates, and close it when you are done.

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.

Next steps

Edit this page