Durable Streams

A Durable Stream is an append-only log with a durable, replayable history. Writers append messages to the end, and readers read from any offset or follow the end live. Durable Streams is an open protocol, and Rivet serves it from Rivet Services, a prebuilt node Rivet ships: each stream is its own Actor, with its messages kept by the control plane like any other Actor’s data.

Take a chat room’s messages. Each room has a stream, keyed by the room’s name. The room appends every message, a page loads the history from the start, and a live view follows new messages as they arrive:

Messages stay in the order they were appended. A read returns them from an offset and says where to read next; a tail is caught up, and gets each new message as it's appended.

Run Rivet Services

Streams are served by a Services pool next to your own pools. Give it the same control plane:

src/cluster.ts

import * as RivetDocker from "@alchemy.run/rivet/Docker";
import * as Layer from "effect/Layer";

export const Cluster = RivetDocker.SwarmControlPlane("Rivet");

export const Default = RivetDocker.SwarmPool("Default").pipe(Layer.provide(Cluster));

export const Services = RivetDocker.ServicesPool("Services").pipe(Layer.provide(Cluster));

Every cloud has one: RivetAWS.ServicesPool, RivetGCP.ServicesPool, RivetHetzner.ServicesPool and the rest. The Services node connects out to the control plane like any other node. Callers reach streams through the gateway, with the token.

Declare a stream

A stream declaration names its messages’ Schema:

src/chat-log.ts

import * as Rivet from "@alchemy.run/rivet";
import * as Schema from "effect/Schema";

export const Message = Schema.Struct({ user: Schema.String, text: Schema.String });

export class ChatLog extends Rivet.DurableStream<ChatLog>()("ChatLog", {
  schema: Message,
}) {}

One declaration holds many streams, one per key: ChatLog’s stream for the lobby and its stream for the kitchen are separate logs. A stream is created by its first append.

Write and read it

yield* ChatLog in any runtime gives a typed client. Provide the Services pool and a format with it:

src/room.ts

import * as Rivet from "@alchemy.run/rivet";
import * as StreamFormat from "@alchemy.run/rivet/StreamFormat";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import { ChatLog } from "./chat-log.ts";
import { Default, Services } from "./cluster.ts";

export default class Room extends Rivet.DurableObject<Room>()(
  "Room",
  { main: import.meta.url },
  Effect.gen(function* () {
    const log = yield* ChatLog;
    return Effect.gen(function* () {
      const state = yield* Rivet.DurableObjectState;
      return {
        say: (user: string, text: string) => log.append(state.id.name, { user, text }),
        history: () => log.read(state.id.name),
      };
    });
  }).pipe(Effect.provide(Layer.mergeAll(Default, Services, StreamFormat.Json))),
) {}

append writes one message, and appendAll writes several in one request, in order. A message that doesn’t match the Schema fails before it’s sent.

read returns the messages from an offset, the start by default, and where to read next:

const { messages, next, upToDate } = yield* log.read("lobby");
const later = yield* log.read("lobby", { from: next });

from: "now" skips the history. A read of a stream nobody appended to fails with Rivet.DurableStreamNotFound.

Follow it live

tail is a Stream of every message from an offset, then each new one as it’s appended:

yield* log.tail("lobby", { from: next }).pipe(
  Stream.runForEach((message) => Console.log(`${message.user}: ${message.text}`)),
);

It reconnects on its own when the connection drops, from the last offset it saw, and ends when the stream is closed.

Choose a format

A stream’s wire format is a Layer, provided where the client is built.

StreamFormat.Json sends each message as one JSON value, as application/json. Rivet’s dashboard can read it.

StreamFormat.Ndjson writes one JSON value per line on a byte stream, as application/x-ndjson.

StreamFormat.SchemaBinary() writes Effect’s compact binary codec on a byte stream, as application/octet-stream.

A program only bundles the formats it provides. A stream whose format differs from the rest of the runtime’s takes its own Layer as format:

src/telemetry.ts

import * as Rivet from "@alchemy.run/rivet";
import * as StreamFormat from "@alchemy.run/rivet/StreamFormat";
import * as Schema from "effect/Schema";

export const Sample = Schema.Struct({ at: Schema.Number, value: Schema.Number });

export class Telemetry extends Rivet.DurableStream<Telemetry>()("Telemetry", {
  schema: Sample,
  format: StreamFormat.SchemaBinary(),
}) {}

The format is fixed when a stream is created, so keep a stream’s format once it holds messages.

Keep it, close it, delete it

A stream is kept until it’s deleted, unless the declaration sets its retention when it’s created:

src/chat-log.ts

export class ChatLog extends Rivet.DurableStream<ChatLog>()("ChatLog", {
  schema: Message,
  ttl: "7 days",
}) {}

ttl deletes a stream after that long with no reads or writes; expiresAt deletes it at a fixed time.

close makes a stream final: reads and tails keep working, and appends fail with Rivet.DurableStreamConflict. delete removes it and its history:

yield* log.close("lobby");
yield* log.delete("kitchen");

Write exactly once

A writer that retries can append the same message twice. Give each append a producer, and a repeated one is written once:

yield* log.append("lobby", message, { producer: { id: "room-lobby", epoch: 1, seq: 42 } });

seq goes up by one per append, and epoch goes up when the writer restarts. An append with an older epoch than the stream has seen fails with Rivet.DurableStreamStaleProducer.

Where streams are reached

Callers reach every stream through the control plane’s gateway, with its token, never through the Services node: that node’s own port checks no token, and Services pools publish no port. See Tokens and networks.

Services pools run rivetdev/services:0.2.5 by default. A newer release goes in the pool’s image, after you’ve checked the streams you have still read back.

Every option, with its default, is in the reference: DurableStream.