@loro-dev/streams-crdt

Transport layer for synchronizing CRDT state over Loro Streams (Durable Streams).

Native durable Effect API

@loro-dev/streams-crdt/effect provides makeDurableStream, native DurableAdapter/DurablePersistence capabilities, StreamsAuth, and typed DurableFailure, DurableClosed, DurableReleaseFailure errors. Effect stays exactly 4.0.2. This durable API is available from CRDT 0.18.0, which depends on client 0.11.0 exactly. CRDT 0.17.0 provides native ephemeral sync only; client 0.10.0 does not expose the native HTTP entry.

import { Effect } from "effect";
import { LoroDoc } from "loro-crdt";
import { createLoroDocAdapter } from "@loro-dev/streams-crdt/loro";
import {
  makeDurableStream,
  durableAdapterLegacy,
  StreamsAuth,
} from "@loro-dev/streams-crdt/effect";

const program = Effect.scoped(Effect.gen(function* () {
  const doc = new LoroDoc();
  const stream = yield* makeDurableStream({
    streamUrl: "https://example.com/ds/bucket/room",
    adapter: durableAdapterLegacy(createLoroDocAdapter(doc)),
  });
  const subscription = yield* stream.join;
  yield* subscription.awaitReady;
  doc.getMap("data").set("answer", 42);
  doc.commit();
  const result = yield* subscription.sync;
  if (result.kind === "confirmed") {
    // Covered local updates have actual append ACKs and a new read-to-tail.
  }
})).pipe(Effect.provide(StreamsAuth.anonymous));

// Application entrypoint; the library creates no Runtime.
await Effect.runPromise(program);

Construction and each join require a Scope. Joining admits a shared subscription; awaitReady waits for initial bootstrap/catch-up, independently of upload ACKs. health.status === "joined" describes the live connection. Use sync for a new protocol confirmation, rather than mapping readiness or local persistence to confirmation. Concurrent sync observers share one transaction when it covers each caller's target revision and SDK export barrier. A later caller whose target exceeds a frozen transaction waits for a following shared transaction. Cancelling an observer keeps its producer and other observers alive.

createStream performs an explicit owned PUT without subscribing or starting bootstrap/read/write loops. It returns true for creation (201), false for an existing stream. createStream followed by join initializes only after PUT completion; a creation failure does not poison a future ready receipt. A successful explicit create after an unsuccessful initial join retires that failed generation before the next join. createStreamIfMissing remains an initialization capability: on a joined bootstrap/read 404 it creates, then bootstraps again. Control requests use the normal HTTP deadline; automatic creation shares the initial hard deadline.

disconnect retains the subscribed local FIFO. A sync called after disconnect returns { kind: "offline" }; a sync already waiting when disconnect starts fails with typed DurableClosed. Cancelling an individual sync waiter still interrupts only that waiter. reconnect retires the preceding network generation before bootstrapping another. close stops admission and subscriptions; concurrent calls share release. DurableReleaseFailure retains its complete private Cause, owner and retryRelease. Actual fetch/auth/reader/export work that ignores cancellation fences replacement until it settles and release is recovered. Scope finalizers expose release failures as defects; their Cause contains the same recovery error. A retired construction Scope cannot join again. Retryable initialization failures retry within the configured policy. Once initial readiness fails terminally, native initialization does not restart in the background. A later join, reconnect, or online sync retries after retiring its actual resources. Live reads retry separately; repeated 410/incomplete recovery is paced and bounded by maxAttempts, resetting after a successful live result. A pending healthy initialization remains shared. During incomplete recovery, ACKed FIFO heads leave the upload queue so later local edits can append, while confirmed sync, cursor/checkpoint and snapshots stay gated.

Production integrations should implement DurableAdapter directly, using their controlled native imports and persistence. durableAdapterLegacy is a registered SDK bridge for existing Loro/Flock adapters, not a Promise transport wrapper. Imports report global completeness, preserve accepted prefixes on a later input failure, and return only applied remote versions. Snapshot imports must return the server snapshot version, excluding pre-existing local edits. Provide an independent createIsolatedAdapter for bootstrap/410 inference; both SDK bridges provide it. Snapshot compression and payload protection use the same hooks for initial, incremental and recovered state. Providers own logical room identity and snapshot continuation-offset AAD; URLs remain opaque request targets.

Each bootstrap observation uses one isolated replica across the initial snapshot and all continuation pages. Recovery computes missing uploads from its full applied version, including cross-page pending dependencies. Live imports retain accepted prefixes, but bootstrap checkpoint occurs only after a complete read-to-tail and the isolated/live completeness gates agree. Cancellation or a later page failure keeps the observation incomplete; it cannot authorize cursor or snapshot repair. Each page is decoded/opened once, with independent input copies for isolated imports. Recovery refetches after actual append phases, then checkpoints only its final complete observation. Successful Loro pages use native batch import; failed batches replay in order to retain accepted prefixes before rethrowing.

Supply a native persistence capability when local state and replay progress must survive restart. Its checkpoint({ cursor, source }, io) must persist the covered replica data before the cursor and enforce generation/completeness. For a native loro-repo lease, acquire yield* lease.checkpoint, bind load to that checkpoint's load(streamUrl), and delegate the hook to save(cursor). Use the lease's controlled importUpdates; do not bypass pending-prefix retention with raw imports or substitute a global repo flush. checkpointed in a confirmed result identifies the configured checkpoint capability; without it, cursors are in-memory and remote confirmation remains separate from local durability. Persisting only a cursor is insufficient.

healthChanges owns each observation in its collecting Scope. Borrower observations end with their subscription Scope. Health and lastError.context are safe diagnostics; cursors, full errors, failureCause and recovery owners are private protocol/diagnostic data. Native auth composes Effects directly; shared refresh producers belong to the provider Scope. Only third-party I/O uses the supplied tracked io.fromPromise boundary. Adapters delivering local callbacks/exports asynchronously must provide flushLocalExports; each invocation covers work admitted before that invocation. Callback revision alone cannot describe export work whose callback has not arrived.

The sections below describe the retained Promise implementation, explicitly exported as StreamsCrdtLegacy and compatibly named StreamsCrdt. Ownership and release ordering: implementation note.

Upload-only receipts

stream.awaitUploaded waits for initial readiness, then captures the local export barrier and waits for the covered append ACKs. Calls admitted before readiness include edits made during bootstrap. stream.uploadThrough(version) instead accepts a version captured by a caller before waiting for a route. Both return acked or offline; errors retain the native typed Cause. They may need initial bootstrap to establish server history, but never wait for subsequent downlink catch-up or checkpointing. Later queue entries do not extend a frozen receipt after that capture. Use uploadThrough(version) when the call-time version must be fixed before readiness. Generation replacement returns offline; final construction-Scope retirement and released subscriptions still fail with DurableClosed. Cancelling a waiter detaches it from the generation-owned producer; close/recovery still owns unfinished HTTP, auth and adapter work. No local flush or peer handshake is an upload ACK.

health.readEpoch counts completed initial/rejoin baselines, independently of coalesced health notifications or local edits. health.readStatus reports the read path separately from the combined status (which may also report a write failure). Local edits and append retries do not establish a new read baseline. Incomplete/tentative initial imports cannot increment readEpoch. Bootstrap/recovery and target backlog exports share the backlog gate. The complete implementation and acceptance are in the native durable ownership note.

Core Concepts

Two-class model

| Class | Responsibility | | --- | --- | | StreamsCrdt | Transport runtime bound to one stream URL. Handles auth, reconnect, bootstrap, SSE/long-poll, binary framing, and remote cursor persistence. | | CrdtAdapter | Binding for one local CRDT instance. Handles snapshot export/import, version tracking, local update export, remote update application, and optional isolated 410 recovery inference. |

Operations

| Method | Purpose | | --- | --- | | createStream() | Creates the stream on the server. Returns { created: true } on first call, { created: false } if it already exists. | | deleteStream() | Deletes the stream. Closes any active join() first. | | sync() | One-shot bootstrap/catch-up cycle. Syncs remote state into local CRDT and pushes any local changes to the server. Does not enter live mode. | | catchup() | Pull-only remote catch-up. Applies remote stream updates to the local CRDT but does not intentionally append new local updates. If live reads from join() are already healthy, returns immediately unless { force: true } is passed. | | join() | Initial sync plus live subscription. Internally performs the same sync as sync(), then enters a persistent SSE/long-poll read loop and starts forwarding local writes. No need to call sync() before join(). | | close() | Closes any active join and releases resources. If this instance was using appendWriteOnly(), that write-only path is also torn down and cannot be used again on the same instance. |

join() includes an initial sync. You do not need to call sync() before join(). When join() resolves, the local CRDT is already caught up with the server and the live subscription is active.

Error handling — Result, not exceptions

All transport operations return a Result<T, TransportError> instead of throwing. Callers must check .ok before accessing the value:

const result = await transport.join();
if (!result.ok) {
  // result.error is a TransportError with a discriminated `code` field.
  switch (result.error.code) {
    case "stream_not_found":
      // 404 — the stream doesn't exist yet. Create it first.
      break;
    case "auth_failed":
      // 401/403 from the server — refresh credentials and retry.
      break;
    case "auth_provider_error":
      // Your auth callback threw. The live loop will retry automatically.
      // For permanent logout, return undefined from the callback instead.
      break;
    case "server_error":
      // 5xx — transient backend failure. The live loop retries automatically;
      // for one-shot calls (sync/createStream/etc.) you decide whether to
      // retry. `retryAfterMs` honors the server's Retry-After hint.
      break;
    case "network_error":
    case "timeout":
    case "initial_sync_timeout":
      // Transient — safe to retry.
      break;
    case "protocol_error":
    case "internal_error":
      // Non-retryable: bad server response or local runtime/adapter failure.
      break;
    default:
      console.error("transport error:", result.error.code, result.error.context);
  }
  return;
}

const sub = result.value; // TransportSubscription

Every error carries a retryable flag so callers can decide whether to retry without hard-coding error codes:

if (!result.ok && result.error.retryable) {
  // safe to retry after a delay
}

Surfacing errors to end users

For wrappers such as Loro Repo and diagnostic sinks, forward error.context, error.code and the explicit error.retryable. TransportErrorContext is an additive optional safe projection: source, CRDT operation, failing stage, original client discriminator, request operation and available timeout budget, elapsed duration, HTTP status, bounded request ID and allowlisted network code. Do not forward the whole error as a diagnostic object: legacy message, body, cause, headers and URL metadata may contain application secrets.

Source/stage are recorded at the actual fetch, cursor load/save, durability hook, CRDT apply/export and payload encode/decode boundaries. Local TypeError is internal_error with retryable: false; fetch-thrown ordinary Error is network_error. This identifies the observed boundary, not the root cause of a bug inside custom fetch. Unobserved provenance is explicitly unknown. Initial-sync hard deadlines retain initial_sync_timeout, their own budget and source: "deadline". See the implemented safe error contract for exact semantics, security bounds and the Loro Repo integration still required downstream.

HTTP-derived variants (server_error, auth_failed, protocol_error, stream_not_found, gone) retain legacy response details. Prefer numeric status and the safe correlation ID for UI/diagnostics; response bodies and arbitrary messages need application-specific handling:

const err = result.error;
if (err.code === "server_error") {
  const requestId = err.context?.requestId;
  toast.error(`Server error ${err.status}${requestId ? ` (req: ${requestId})` : ""}`);
  console.error("Streams failure:", err.context);
  // err.retryAfterMs may suggest a backoff if the server set Retry-After.
}

| Code | status | message | body | Other | | ---------------------- | -------- | --------- | ------ | ------------------------------ | | stream_not_found | 404 | yes | yes | — | | auth_failed | 401/403 | yes | yes | — | | protocol_error | yes | yes | yes | — | | gone | 410 | yes | yes | — | | server_error | 5xx | yes | yes | requestId?, retryAfterMs? | | timeout | — | yes | — | phase: "connect" \| "poll" | | initial_sync_timeout | — | yes | — | operation, timeoutMs, elapsedMs | | network_error | — | yes | — | — | | auth_provider_error | — | yes | — | cause? (original thrown value) | | internal_error | — | yes | — | — | | apply_incomplete | — | yes | — | reason: "loro_pending_updates" |

Operation-level retries vs. live-mode retries

• sync(), catchup(), createStream(), deleteStream(), appendWriteOnly() return after a single attempt. The application is responsible for retrying retryable errors.
• After join() succeeds, the live read/write loops automatically reconnect on every retryable: true error using exponential backoff ([0, 0.5s, 1s, 2s, 4s, 8s, 15s, 30s] with ±20% jitter; cap 30s). When a 5xx response includes Retry-After, the runtime waits at least that long. Non-retryable errors transition the subscription to status: "error"; call transport.rejoin() after fixing the cause (e.g., refreshing the auth token) to reset retry state and reconnect.

Initial sync deadlines

Initial bootstrap/catch-up work is allowed to run longer than ordinary request connect and poll deadlines. The transport uses reconnectConfig for this:

const transport = new StreamsCrdt({
  streamUrl,
  adapter,
  auth,
  reconnectConfig: {
    // Logs a debug diagnostic when initial sync is still pending; does not abort.
    initialSyncSlowAfterMs: 30_000,
    // Aborts initial bootstrap/catch-up with `initial_sync_timeout`.
    initialSyncHardTimeoutMs: 120_000,
  },
});

join(), sync(), pull-only catchup(), and 410 bootstrap recovery all use the initial-sync deadline path when they need to bootstrap or catch up from a stored cursor. In StreamsCrdt constructor options, set initialSyncHardTimeoutMs to 0 or Infinity to disable the hard cap. Because constructor reconnectConfig is merged over defaults, omitting the field keeps the default hard cap. Set initialSyncSlowAfterMs the same way to disable the slow diagnostic.

Why Result instead of throw? Transport failures (network errors, auth expiry, missing streams) are expected at runtime, not exceptional. Wrapping them in Result makes the error path explicit in the type system so callers cannot accidentally ignore failures with a missing try/catch.

Migration notes

For most callers nothing changes — if (!result.ok) { handle(result.error); } keeps working, and result.error.retryable is still the right gate for "safe to retry?".

Only adjust if any of these apply:

• Your auth callback throws on permanent failure (e.g., user is logged out). Throwing is now treated as a transient error and retried in live mode. Return undefined instead so the server returns 401 and you get the terminal auth_failed. See "What if the auth callback throws?" below.
• You construct TransportError literals (e.g., in tests or fixtures). stream_not_found now requires status: 404 + message; gone now requires status: 410. TypeScript will flag these.
• You exhaustively switch on code with a : never default. Add a case for auth_provider_error and initial_sync_timeout (and optionally server_error if you handled 5xx via unknown/internal_error before).
• You used "message" in error to defensively narrow. All variants now carry message; the guard is no longer needed.

Quick Start (Loro)

import { LoroDoc } from "loro-crdt";
import {
  createLoroDocAdapter,
  StreamsCrdt,
} from "@loro-dev/streams-crdt/loro";

const doc = new LoroDoc();
const transport = new StreamsCrdt({
  streamUrl: "https://streams-api.loro.dev/ds/my-bucket/doc-1",
  auth: async () => "<gateway-jwt>",
  adapter: createLoroDocAdapter(doc),
  // Optional TTL applied when the stream is created.
  streamTtlSeconds: 60 * 60 * 24 * 30,
  // Enable automatic snapshot upload so new readers bootstrap faster.
  snapshotUpload: { canUpload: async () => true },
});

// Create the stream (idempotent).
const created = await transport.createStream();
if (!created.ok) throw created.error;

// Join enters live collaborative editing mode.
// This performs an initial sync automatically before entering live mode.
const joined = await transport.join({
  onStatusChange(status) {
    console.log("room status:", status);
  },
});
if (!joined.ok) throw joined.error;

const sub = joined.value;

// The doc is now syncing in real time.
// Call sub.unsubscribe() or transport.close() to leave.

createLoroDocAdapter(doc) requires an attached LoroDoc. If the document was checked out or detached, call checkoutToLatest() or attach() before passing it to StreamsCrdt.

Request shard origins

Browsers can still show head-of-line blocking symptoms when large bootstrap/catch-up downloads and normal app requests contend on the same connection. shardUrls lets the transport replace only the request origin for selected operations while preserving the stream path and query:

const transport = new StreamsCrdt({
  streamUrl,
  auth: async () => "<gateway-jwt>",
  adapter: createLoroDocAdapter(doc),
  shardUrls: {
    bootstrap: [
      "https://control-a.streams.example.com",
      "https://control-b.streams.example.com",
    ],
    catchup: [
      "https://control-a.streams.example.com",
      "https://control-b.streams.example.com",
    ],
    largePost: [
      "https://write-a.streams.example.com",
      "https://write-b.streams.example.com",
    ],
    other: [
      "https://live-a.streams.example.com",
      "https://live-b.streams.example.com",
    ],
    largePostMinBytes: 64 * 1024,
  },
});

The hostnames above are placeholders. Configure your own origins to route to the same gateway; this transport does not provision DNS or deploy a proxy Worker.

Shard entries must be origin URLs with no path, query, or hash. Each transport instance rotates through the configured origins per operation, with a randomized starting point so multiple tabs are less likely to all begin on the same shard. Bootstrap applies to GET <stream>/bootstrap; catch-up applies to non-SSE offset reads, including long-poll fallback. largePost applies to non-empty append POST bodies at or above largePostMinBytes bytes. other applies when no operation-specific pool routes the request, covering live SSE, create/delete, head, snapshot upload, and append POSTs not routed by largePost. The default threshold is 0, so every non-empty append POST can use largePost when that pool is configured. Verify actual connection separation with browser NetLog: browsers may coalesce compatible HTTP/2 or HTTP/3 origins when DNS, certificate, and transport settings allow it.

Loro pending imports and cursor safety

Loro can accept an update whose dependencies are missing and keep it as a pending import. Pending imports are not represented in doc.version() and are not included in snapshots exported from that document. For that reason, the Loro adapter must not persist a RemoteCursor until all pending imports are resolved and the local document is complete at the cursor offset.

The transport may continue reading with an in-memory volatile cursor to find the missing dependencies. If the server reports up-to-date while Loro still has unresolved imports, the client must bootstrap again and only save the bootstrap cursor after the batch import is clean. Snapshot upload is skipped while Loro has unresolved imports.

See specs/loro-pending.md for the full behavior and implementation checklist.

Quick Start (Loro EphemeralStore)

Use a separate EphemeralStreamCrdt for presence-like Loro EphemeralStore data. It bootstraps through the first SSE event, posts local ephemeral updates, and never republishes state imported from other peers. If a client needs to keep its own presence alive, update that client's own keys with store.set() so the store emits a fresh local update.

Ephemeral live delivery is latest-state oriented. The server may keep only a tiny live fanout queue for current SSE subscribers and may disconnect slow subscribers instead of buffering every intermediate cursor or presence update. The client reconnects and receives a fresh bootstrap, so the local store converges to current state without requiring offset replay.

import { EphemeralStore } from "loro-crdt";
import {
  EphemeralStreamCrdt,
  EphemeralStoreAdaptor,
} from "@loro-dev/streams-crdt/loro";

const store = new EphemeralStore();
const docStreamUrl = "https://streams-api.loro.dev/ds/my-bucket/doc-1";

const presence = new EphemeralStreamCrdt({
  streamUrl: `${docStreamUrl}?ephemeral=presence`,
  auth: async () => "<gateway-jwt>",
  adaptor: EphemeralStoreAdaptor(store),
});

const joinedPresence = await presence.join();
if (!joinedPresence.ok) throw joinedPresence.error;

Ephemeral stream URL format

An ephemeral stream URL is the durable stream URL for the same document plus a channel query parameter:

{baseUrl}/ds/{bucketId}/{streamId}?ephemeral={channel}

For example, if the durable Flock document stream is:

https://streams-api.loro.dev/ds/my-bucket/flock-cm9vbS0x

then the matching presence stream is:

https://streams-api.loro.dev/ds/my-bucket/flock-cm9vbS0x?ephemeral=presence

Use the same bucketId and streamId as the durable StreamsCrdt room. The ephemeral value names a side-channel for that room, such as presence, cursor, or selection. Different channels under the same durable stream are separate in-memory rooms. The compatibility spelling ?awareness=presence is also accepted, but new code should use ?ephemeral=presence.

When an app exposes a room link such as /rooms/design-review-42, the link normally maps to one durable stream id first, and both durable and ephemeral sync derive from that same id:

const roomId = "design-review-42";
const durableStreamUrl = new URL(
  `ds/my-bucket/${encodeURIComponent(`flock-${roomId}`)}`,
  "https://streams-api.loro.dev/proxy/",
).toString();

const presenceStreamUrl = new URL(durableStreamUrl);
presenceStreamUrl.searchParams.set("ephemeral", "presence");

const flockTransport = new StreamsCrdt({
  streamUrl: durableStreamUrl,
  auth,
  adapter: createFlockAdapter(flock),
});

const presenceTransport = new EphemeralStreamCrdt({
  streamUrl: presenceStreamUrl.toString(),
  auth,
  adaptor: EphemeralStoreAdaptor(store),
});

The ds/... path is relative so a gateway path prefix such as /proxy/ is preserved. Validate dynamic bucket and stream segments with isValidRillId() before constructing the URL; this rejects ./.. before URL normalization.

EphemeralStreamCrdt does not create or delete the durable stream. It posts updates to the ephemeral URL and opens SSE on the same URL with live=sse. Ephemeral state is not persisted with the Flock document stream; it is scoped by the URL path plus channel and is meant for current live state only.

Effect ephemeral API

Install [email protected] alongside this package. Native applications use @loro-dev/streams-crdt/effect; durable StreamsCrdt remains Promise-based.

import { Effect, Layer } from "effect";
import { EphemeralAuth, makeEphemeralStream } from "@loro-dev/streams-crdt/effect";
import { EphemeralStoreAdaptor } from "@loro-dev/streams-crdt/loro";

const authLayer = Layer.succeed(EphemeralAuth, {
  getToken: () => Effect.succeed(accessToken),
});
const program = Effect.scoped(Effect.gen(function* () {
  const room = yield* makeEphemeralStream({streamUrl, adaptor: EphemeralStoreAdaptor(store)});
  const subscription = yield* room.join();
  // The owning application keeps this Scope alive for the subscription lifetime.
  yield* subscription.waitUntilSynced;
})).pipe(Effect.provide(authLayer));

makeEphemeralStream requires Scope and EphemeralAuth. Use EphemeralAuth.anonymous for unauthenticated connections. Auth receives the existing request / unauthorized context; a 401/403 refresh is shared across requests and retried once. Native auth composes Effects. A third-party Promise provider uses getToken: (context, io) => io.fromPromise(signal => fetchToken(context, signal)); this tracks actual completion and forwards cancellation. ephemeralAuthLegacy adapts existing callbacks, adding an optional second AbortSignal argument. The /effect entry also exports StreamsAuthContext, TransportError and TransportErrorContext. getToken accepts Effect<string | undefined, unknown>: pass a native domain callback directly without mapError. A failed Effect prevents that request. The library separately projects a single expected failure into safe EphemeralFailure.transport: status 401/403 is terminal auth_failed, Effect's TimeoutError remains timeout, and other domain failures are auth_provider_error. The complete original Cause<unknown> and any provider recovery owner remain in EphemeralFailure.cause; mixed failures, defects and interruption remain complete Causes. These fields are never diagnostic data. Native terminal failures stay failed Effects; only a provider's explicit compatibility policy returns undefined. undefined omits Bearer and lets the server decide authentication (anonymous or terminal-auth compatibility), whereas failed Effects prevent that request. The unauthorized context carries the actual rejected request's previousToken. The transport does not mutate the provider's credentials.

A shared refresh producer belongs to the token provider's Scope. Room-owned auth fibers only wait for that capability; their cancellation must detach the local waiter without terminating a producer used by other rooms. The room's read/POST refresh record coalesces local waits, rather than owning cross-room credentials. The owning test exercises two rooms with a provider-scoped producer: closing one room leaves the other able to bootstrap and ACK a real EphemeralStore update.

A room owns one read fiber and one writer fiber; every join owns a Scope lease. Closing the construction Scope permanently retires new joins; recovery of retained release work remains available. Explicit close while that Scope is alive permits a fresh generation after successful cleanup. Bootstrap readiness is independent of local sending. waitUntilSynced captures chains when run; syncBarrier() captures them when called and returns the waiting Effect. Both barriers wait for successful POST acknowledgement and response cleanup; neither proves another peer received or applied the update. The caller owns any peer-receipt protocol. Cancelling either wait leaves the shared room alive. release awaits the last lease's cleanup. Explicit close closes all leases. releaseTimeoutMs defaults to 5,000 and must be positive/finite. Concurrent close shares its result. Uncooperative fetch/auth/body/adaptor work produces EphemeralReleaseFailure carrying owner and full cause. Keep that owner, resolve the underlying work or confirmed termination, then yield* owner.retryRelease; join/rejoin cannot replace unreleased resources. Request owners also await native auth task settlement and retain mixed interruption/cleanup Causes during close/rejoin. An auth Promise that rejects after cancellation but actually settles is retained privately as producer history; it does not by itself turn successful cleanup into a release failure. Internal request admission during shutdown is cancellation, not an EphemeralClosed release failure; public closed-handle failures remain observable. Raw native auth and join Exits are captured inside uninterruptibleMask before restoring caller interruptibility; Effect 4 recovery handlers must not strip a pending mixed Fail. Each request owner performs one bounded drain. Scope finalizers preserve this failure as a defect, so inspect the full Exit/Cause when body and release both fail.

subscription.stats / room.stats expose logical pending count, physical entries, queued bytes, last ACK time, upload error and outstanding release work. Readiness or a joined read side does not prove upload progress; observe pending count plus last ACK time/error and the post/ack/stall diagnostic stages. Auth cleanup failures remain EphemeralReleaseFailure, including after a 401/403 refresh. Upload health retains the original transport code (such as timeout) inside library-owned release receipts; its safe projection never replaces the full Cause/recovery owner. The default queue admission remains compatible: superseded bytes are released immediately and markers are compacted, but distinct/unknown keys and bytes have no new global cap. Publishers still own rate and size limits. Bodies are snapshotted at admission, including Buffer/subarray inputs, and are never merged or rewritten. This adds a copy per emission; throughput/allocation overhead is not benchmarked.

Diagnostics default to no-op. Provide EphemeralDiagnostics yourself or the ephemeralDiagnosticsLogging Layer with your application's Effect Logger Layer. Only fixed stages, safe context and counts are emitted; the full release Cause is not safe diagnostic data. debug no longer enables raw console errors.

EphemeralStreamCrdtLegacy is the only Promise facade. The existing EphemeralStreamCrdt spelling remains a deprecated alias. One runtime serves all methods and fixes its configuration bindings at construction. unsubscribe() initiates release; await close() to observe its outcome, or use terminal disposeLegacy() to close and dispose the facade runtime. close/retryRelease reject with EphemeralLegacyFailure.effectCause, and lastFailureCause retains background unsubscribe/rejoin failures. Delete the facade after downstream Promise consumers adopt their application's Effect runtime. See the draft lifecycle contract and decision.

Quick Start (Flock)

import { Flock } from "@loro-dev/flock-wasm";
import {
  createFlockAdapter,
  StreamsCrdt,
} from "@loro-dev/streams-crdt/flock";

const flock = new Flock();
const transport = new StreamsCrdt({
  streamUrl: "https://streams-api.loro.dev/ds/my-bucket/room-1",
  auth: async () => "<gateway-jwt>",
  adapter: createFlockAdapter(flock),
});

await transport.createStream();
const joined = await transport.join();
if (!joined.ok) throw joined.error;

const sub = joined.value;

When to Use catchup(), sync(), and join()

Use catchup() when you only want to pull remote updates into the local CRDT:

// Pull remote changes without uploading local pending changes.
const caughtUp = await transport.catchup();
if (!caughtUp.ok) throw caughtUp.error;
// caughtUp.value.cursor contains the replay progress after the pull.

catchup() is safe to call while a room is live. When join() has a healthy SSE/long-poll read loop, catchup() returns immediately because the live loop is already catching up continuously. Pass { force: true } to issue one explicit non-SSE offset read anyway, for example after app foreground or device wake:

await transport.catchup({ force: true });

If the live read side of the room is reconnecting, disconnected, or in error, catchup() skips the live backoff path and tries one immediate non-SSE catch-up read from the room's current cursor. A successful call updates the local cursor and nudges the live loops to reconnect, but it does not by itself guarantee that local pending writes have reached the server.

Use sync() when you need a one-shot bidirectional sync without entering live mode:

// Offline-first: pull remote changes, upload local changes, then work offline.
const synced = await transport.sync();
if (!synced.ok) throw synced.error;
// synced.value.cursor contains the replay progress after sync.

Use sync() or subscription.waitUntilSynced() when the caller needs local pending changes uploaded to the server.

Use join() for real-time collaboration. It does everything sync() does and then keeps the connection alive for continuous bidirectional sync:

const joined = await transport.join();

Delete Stream

const deleted = await transport.deleteStream();
if (!deleted.ok) throw deleted.error;

console.log(deleted.value.deleted); // true when deleted now, false when already missing

deleteStream() closes any active join first. When the configured remoteCursorStore implements delete(streamUrl) (the built-in stores do), the transport also clears the stored replay cursor for that stream.

Auth

auth may be:

• a static string
• a sync callback returning a string
• an async callback returning a string

StreamsCrdt asks for a token before each request. If a request comes back with 401 or 403, it calls auth one more time with { reason: "unauthorized", status, previousToken } and retries that request once.

This lets the caller keep a cached token in the normal path, and only refresh after the server rejects it:

let token = readCachedGatewayJwt();

const transport = new StreamsCrdt({
  streamUrl,
  adapter,
  auth: async (context) => {
    if (context?.reason === "unauthorized") {
      token = await refreshGatewayJwt();
    }
    return token;
  },
});

If your callback already returns a fresh token every time, you can ignore the context argument.

What if the auth callback throws?

Throwing from the callback is treated as a transient failure. The runtime wraps the error as auth_provider_error (retryable: true) so a hiccupping auth backend doesn't terminate the live read/write loops — they retry under the same exponential backoff used for server_error and network_error.

For the "user is permanently logged out" case, return undefined (or null/empty string) instead of throwing. The request goes out without an Authorization header, the server returns 401, and you receive the non-retryable auth_failed instead — which the application typically routes to a sign-in page.

auth: async () => {
  try {
    return await fetchTokenFromAuthBackend(); // network → throw → retry
  } catch (e) {
    if (isPermanentLogoutError(e)) {
      return undefined; // → server 401 → auth_failed (terminal)
    }
    throw e; // → auth_provider_error (retryable)
  }
}

End-to-End Encryption (E2EE)

e2ee enables end-to-end encryption for CRDT update batches and snapshots before their bytes are appended to Durable Streams. Auth stays separate: gateway tokens authorize HTTP requests, while document keys protect document bytes from the server and unauthorized readers.

E2EE at a Glance

1. What the application provides

The application injects a PayloadProtectionProvider, not a raw key. At the streams-crdt API boundary, the provider supplies three things:

• seal() encrypts with the current write key and a fresh nonce, then returns an opaque authenticated header plus provider-defined sealed bytes.
• open() parses that header, resolves the corresponding historical read key, authenticates the ciphertext, and returns plaintext.
• maxSealOverheadBytes declares the maximum provider-added size so streams-crdt can batch updates without speculatively sealing or consuming a nonce.

The provider owns the audited AEAD implementation, nonce generation, sealed layout, opaque header schema, logical-room AAD, current write key/epoch, and historical key lookup. streams-crdt never receives a key directly and does not implement MLS, device identity, epoch state, key storage, or product policy.

2. What streams-crdt does

streams-crdt places that provider at the common durable payload boundary:

local CRDT payload
  -> batch and freeze
  -> provider.seal()
  -> streams-crdt envelope and transport framing
  -> Durable Streams

Durable Streams payload
  -> remove transport framing and parse the envelope
  -> provider.open()
  -> clear CRDT payload
  -> apply to Loro or Flock

The same boundary covers updates, snapshots, bootstrap, catch-up, live SSE/long-poll, 410 recovery, and write-only appends. Frozen update retries reuse the same ciphertext instead of calling seal() and generating a second nonce. Protected reads authenticate before adapter apply or cursor persistence and fail closed on any envelope or provider error.

streamUrl remains an opaque transport URL and is not cryptographic identity. The provider must bind a stable logical room identity itself when cross-room replay protection is required. This boundary covers only durable StreamsCrdt payloads: EphemeralStreamCrdt presence, awareness, cursor, selection, and display-name payloads are outside it.

3. What gets encoded

Every protected update batch and snapshot uses the same envelope:

protectedEnvelope =
  "LSCE"[4]
  || version[1]
  || payloadKind[1]
  || providerHeaderLength[2]
  || reserved[2]
  || providerHeader[H]
  || sealed[provider-defined]

The fixed envelope header is 10 bytes. providerHeader is 1–512 bytes, so the prefix before sealed is 10 + H, or 11–522 bytes. The provider header is visible but authenticated; streams-crdt does not parse it.

For example, a provider using XChaCha20-Poly1305 commonly chooses this sealed layout:

sealed = nonce[24] || ciphertext[plaintext.length] || authenticationTag[16]

The 24-byte nonce is the random 192-bit XChaCha nonce. The 16-byte tag is the 128-bit Poly1305 authentication tag. XChaCha ciphertext has the same length as its plaintext, so this provider layout adds 24 + 16 = 40 bytes. This layout is an example, not a format imposed by streams-crdt.

The two durable payload classes then differ only in their surrounding framing:

update append body = u32be(protectedEnvelope.length) || protectedEnvelope
snapshot PUT body  = protectedEnvelope

If H is the provider-header length, a direct XChaCha layout adds the following bytes relative to the plaintext passed to seal():

| Payload | Size calculation | Added bytes | | --- | --- | ---: | | Update batch | outer length 4 + envelope 10 + header H + nonce/tag 40 | 54 + H | | Snapshot | envelope 10 + header H + nonce/tag 40 | 50 + H |

For a 16-byte provider header, that is 70 bytes per encrypted update append and 66 bytes per encrypted snapshot. That provider's declared maxSealOverheadBytes must cover H + 40; the fixed 10-byte envelope and the update-only 4-byte outer length are accounted for separately by streams-crdt. Compression may independently change the snapshot plaintext size before sealing. The 40-byte streams-crdt AAD domain tag is reconstructed locally and authenticated, not serialized as another 40 bytes on the wire.

The server can still observe the stream URL, provider header, offsets, total payload sizes, timing, and auth metadata. It cannot read the CRDT payload or, within one encrypted update batch, its update count and individual update sizes.

Minimal Provider Setup

The cryptographic and header helpers in this example are application pseudocode, not exports from @loro-dev/streams-crdt. An application should implement them with an audited AEAD library and an unambiguous binary encoding. The only streams-crdt API being implemented here is PayloadProtectionProvider.

import {
  PayloadProtectionError,
  type PayloadProtectionProvider,
} from "@loro-dev/streams-crdt/loro";

// This provider belongs to one application room session and captures that
// session's finalized write epoch and stable room AAD. Header parsing remains
// provider-owned. frameAad() must encode its inputs unambiguously.
const roomAad = encodeApplicationRoomIdentity(roomId);
const provider: PayloadProtectionProvider = {
  // header + random 24-byte nonce + 16-byte tag, conservatively bounded.
  maxSealOverheadBytes: 256,

  async seal({ plaintext, additionalData }) {
    const header = encodeLodyHeader({
      epoch: roomSession.writeEpoch,
      keyId: roomSession.writeKeyId,
    });
    const nonce = crypto.getRandomValues(new Uint8Array(24));
    const ciphertextAndTag = await xchacha20poly1305Seal({
      key: roomSession.writeKey,
      nonce,
      plaintext,
      aad: frameAad(roomAad, additionalData(header)),
    });
    return { header, sealed: concatBytes(nonce, ciphertextAndTag) };
  },
  async open({ sealed, header, additionalData }) {
    const historical = decodeLodyHeader(header);
    // Resolve from logical room + provider-owned epoch/key id.
    const key = await roomKeys.readKey(roomId, historical);
    if (key == null) {
      throw new PayloadProtectionError("missing_read_key");
    }
    return await xchacha20poly1305Open({
      key,
      nonce: sealed.slice(0, 24),
      ciphertextAndTag: sealed.slice(24),
      aad: frameAad(roomAad, additionalData),
    });
  },
};

const transport = new StreamsCrdt({
  streamUrl: "https://streams-api.loro.dev/ds/my-bucket/doc-1",
  adapter: createLoroDocAdapter(doc),
  e2ee: {
    provider,
    // Defaults shown explicitly:
    readPolicy: "encrypted-only",
    writePolicy: "encrypt",
  },
});

Provider Contract

The package's exported TypeScript declarations are the source of truth. The relevant contract, defined in src/types.ts, is equivalent to this abbreviated definition:

type MaybePromise<T> = T | Promise<T>;
type PayloadProtectionKind = "update_batch" | "snapshot";

interface PayloadProtectionContext {
  readonly protocol: "loro-streams-crdt-payload-protection";
  readonly version: 2;
  readonly kind: PayloadProtectionKind;
}

interface PayloadProtectionProvider {
  readonly maxSealOverheadBytes: number;

  seal(input: {
    readonly plaintext: Uint8Array;
    readonly context: PayloadProtectionContext;
    readonly additionalData: (header: Uint8Array) => Uint8Array;
  }): MaybePromise<{
    readonly header: Uint8Array;
    readonly sealed: Uint8Array;
  }>;

  open(input: {
    readonly sealed: Uint8Array;
    readonly header: Uint8Array;
    readonly context: PayloadProtectionContext;
    readonly additionalData: Uint8Array;
  }): MaybePromise<Uint8Array>;
}

interface E2eeOptions {
  readonly provider?: PayloadProtectionProvider;
  readonly readPolicy?: "encrypted-only" | "allow-plaintext";
  readonly writePolicy?: "encrypt" | "plaintext";
}

The provider is optional only when writePolicy is "plaintext"; protected reads still need one to open encrypted history.

The inputs have these exact meanings:

| Field | Meaning | | --- | --- | | plaintext in seal() | For update_batch, an inner length-prefixed batch containing one or more CRDT update blobs. For snapshot, the adapter snapshot after optional snapshotCodec.compress. | | context | Stable streams-crdt protocol/version/payload-kind context. It contains no URL, bucket, stream, room, epoch, or key id. | | additionalData(header) in seal() | Builds the exact streams-crdt-owned AAD for the final provider header. It must be called exactly once. | | header | Provider-owned, plaintext metadata used to select an algorithm or historical key. It must be 1–512 bytes and must be authenticated. | | sealed | Provider-owned non-empty bytes, normally nonce plus ciphertext and authentication tag. | | additionalData in open() | The same streams-crdt-owned AAD bytes produced for that envelope at write time. |

For cross-room replay protection, use keys isolated per logical room or combine application room AAD and streams-crdt AAD without ambiguity. For example, use domain tags plus length-prefixed fields; do not use plain string concatenation where two input pairs can produce the same bytes. seal() must return the exact header passed to additionalData(header).

The sealed layout stays provider-owned. For example, it may be a random 192-bit XChaCha20-Poly1305 nonce followed by ciphertext and tag; streams-crdt does not parse the nonce or enforce an algorithm suite.

seal() runs exactly once after a logical append is frozen for retry, so a retry reuses the same ciphertext. It does not define an epoch transition barrier. For a Lody write-epoch change, drain pending appends with waitUntilSynced(), close the old room transport, finalize the epoch change, then create a replacement room transport/provider. Only that room session must be rebuilt, not the whole repo. The replacement provider may still resolve old headers dynamically in open().

Encrypted streams still use application/octet-stream. The server stores opaque bodies and never parses the protected envelope. Update appends retain their outer item-length framing. Snapshot PUT bodies contain the snapshot envelope directly; they do not use the update append's outer item frame.

Payload Lifecycle and Framing

E2EE wraps the existing transport payloads; it does not replace Durable Streams HTTP, CAS, offset, multipart, or live-event framing.

Update write path

One seal() call protects one frozen append batch, not one individual CRDT update. Before freezing an append, the transport may merge FIFO queued batches while the estimated encoded body remains within its 256 KiB append budget. It then performs this pipeline:

adapter export/subscription
  -> clearUpdates[]                         one or more opaque CRDT update blobs
  -> encodeItems(clearUpdates)              inner update-batch plaintext
  -> provider.seal(kind = "update_batch")  exactly once for this frozen append
  -> protectedEnvelope
  -> encodeItems([protectedEnvelope])       outer Durable Streams append frame
  -> POST/CAS body

encodeItems() is a sequence of unsigned 32-bit big-endian length prefixes:

innerUpdateBatch =
  u32be(update_1.length) || update_1
  || u32be(update_2.length) || update_2
  || ...

protectedEnvelope =
  fixedEnvelopeHeader || providerHeader || providerSealedBytes

encryptedAppendBody =
  u32be(protectedEnvelope.length) || protectedEnvelope

The provider receives innerUpdateBatch as seal().plaintext. The outer length prefix is added only after sealing so streams-crdt readers can recover the protected item from the stored append. The server still treats the whole append body as opaque and does not learn the number or individual sizes of the CRDT updates inside the batch. Once the protected append and producer sequence are frozen, retries reuse the same bytes and do not call seal() or generate another nonce.

With writePolicy: "plaintext", there is no inner encrypted batch or protected envelope: the append body is simply encodeItems(clearUpdates).

Update read path

Bootstrap tails, catch-up responses, long-poll responses, and SSE events all reuse the same reverse path:

Durable Streams append bytes
  -> decodeItems(outer append frame)
  -> detect and parse protectedEnvelope
  -> provider.open(kind = "update_batch")
  -> innerUpdateBatch plaintext
  -> decodeItems(innerUpdateBatch)
  -> clearUpdates[]
  -> adapter.applyRemoteUpdates(clearUpdates, remoteVersion)

An encrypted outer item must open to one valid inner encodeItems() payload. It may expand to several CRDT update blobs before the adapter is called. Under allow-plaintext, a non-envelope outer item is passed to the adapter as one clear update. Under encrypted-only, that same item fails before adapter apply or cursor persistence.

Snapshot write and read paths

Snapshots use the same provider envelope but not the update batch's inner or outer item framing:

adapter.exportSnapshot()
  -> snapshotCodec.compress?()
  -> provider.seal(kind = "snapshot")
  -> protectedEnvelope
  -> PUT snapshot body directly

bootstrap snapshot part
  -> parse protectedEnvelope
  -> provider.open(kind = "snapshot")
  -> snapshotCodec.decompress?()
  -> adapter.applySnapshot()

The provider therefore sees the optionally compressed snapshot bytes as seal().plaintext. A selected snapshot is exported, compressed, and sealed once before its PUT; an HTTP-level resend of that prepared request reuses the same body. A later scheduled snapshot may export and seal newer state again. Snapshot offset headers and bootstrap multipart boundaries remain outside the envelope because the service must read them.

Bootstrap, catch-up, live, and recovery

All durable paths converge on the two update hooks and two snapshot hooks described above:

| Path | Processing order | | --- | --- | | Initial bootstrap | Open/decompress/apply the snapshot when present, then open and apply the retained update tail. | | sync() / catchup() | Deframe and open every returned update batch before adapter apply. | | join() live SSE or long-poll | Use the same update read path for every event; there is no live-specific plaintext bypass. | | 410 recovery | Re-bootstrap through the same snapshot and update hooks before resuming live reads. | | Local join/sync writes | Merge, freeze, seal, and append through the same update write path. | | appendWriteOnly() | Uses the same frozen update write path but performs no remote read or snapshot upload. | | Snapshot upload | Compress, seal, and PUT through the same snapshot write path. |

Protected reads complete authentication and decoding before applySnapshot(), applyRemoteUpdates(), local durability hooks, or remote cursor persistence. Any failure therefore leaves the corresponding remote cursor unadvanced.

Provider Envelope and AAD

Provider envelope v2 is:

magic               4 bytes   "LSCE"
version             1 byte    0x02
kind                1 byte    0x01 update_batch, 0x02 snapshot
provider_header_len 2 bytes   unsigned big-endian; 1..512
reserved            2 bytes   both 0
provider_header     provider_header_len bytes, opaque
sealed              remaining bytes, provider-defined and non-empty

The common AAD is an unambiguous byte sequence:

UTF8("loro-streams-crdt-payload-protection/v2\0")
|| exact envelope bytes from magic through provider_header

This streams-crdt-owned AAD binds protocol/version, payload kind, and the provider's entire opaque header. Algorithm, epoch, key-id, application room, and stream identity schemas are not part of the streams-crdt protocol. A provider that needs cross-room replay protection must capture a stable logical room identity and combine it unambiguously with additionalData, as in the example above. Do not use a gateway origin as room identity unless moving that origin is intentionally a cryptographic migration.

The service does not parse the v2 envelope. It remains inside the application payload: append/CAS headers, offsets, multipart boundaries, and snapshot offsets stay outside so Durable Streams can route and retain bytes.

maxSealOverheadBytes must be a safe integer from 0 through 4096 and must bound header.length + sealed.length - plaintext.length. The transport uses that bound to merge queued CRDT batches without speculative sealing or nonce use; actual output above the declaration fails closed.

Draft Compatibility

Provider v2 is still unreleased. Ciphertext produced by earlier revisions of this draft branch used URL-derived AAD and is not compatible with the final caller-owned AAD contract; discard or recreate that draft data rather than adding a production dual-reader for an unpublished format.

Mode Matrix

e2ee supports a few distinct shapes. Choose one explicitly:

| Use case | readPolicy | writePolicy | Required protection config | Remote plaintext accepted? | Local writes sent as | | --- | --- | --- | --- | --- | --- | | Existing plaintext stream | omitted | omitted | none | Yes | plaintext | | Provider private reader/writer | encrypted-only (default) | encrypt (default) | streamUrl + provider | No | provider v2 | | Mixed provider reader/writer | allow-plaintext | encrypt | streamUrl + provider | Yes | provider v2 | | Public write-only writer | allow-plaintext | plaintext | none | Yes | plaintext via appendWriteOnly() | | Provider reader without writes | encrypted-only | plaintext | streamUrl + provider | No | plaintext if local writes are exported |

There is no separate transport-level read-only mode. The last row is only appropriate when the local CRDT is treated as read-only or the caller's auth rejects writes. writePolicy: "plaintext" avoids requiring a write key; it does not suppress local export attempts by itself.

Adapter Compatibility: Loro and Flock

e2ee is adapter-agnostic at the transport layer: the same options shape works with createLoroDocAdapter(doc) and createFlockAdapter(flock). Encryption wraps the CRDT payload bytes that the adapter already exports.

That also means one stream must use exactly one CRDT family:

• A stream written with the Loro adapter must be read with the Loro adapter.
• A stream written with the Flock adapter must be read with the Flock adapter.
• Do not mix Loro and Flock replicas against the same stream, whether payload protection is enabled or not. The decrypted inner update bytes and snapshots are CRDT-specific and are not cross-compatible.

The examples below use separate stream URLs and room-bound providers:

const streamUrl = "https://streams-api.loro.dev/ds/my-bucket/doc-1";
const flockStreamUrl = "https://streams-api.loro.dev/ds/my-bucket/flock-1";
const loroProvider = createRoomProvider("doc-1");
const flockProvider = createRoomProvider("flock-1");

Protected Loro room:

import { LoroDoc } from "loro-crdt";
import {
  createLoroDocAdapter,
  StreamsCrdt,
} from "@loro-dev/streams-crdt/loro";

const doc = new LoroDoc();
const loroTransport = new StreamsCrdt({
  streamUrl,
  adapter: createLoroDocAdapter(doc),
  e2ee: {
    provider: loroProvider,
  },
});

Protected Flock room:

import { Flock } from "@loro-dev/flock-wasm";
import {
  createFlockAdapter,
  StreamsCrdt,
} from "@loro-dev/streams-crdt/flock";

const flock = new Flock();
const flockTransport = new StreamsCrdt({
  streamUrl: flockStreamUrl,
  adapter: createFlockAdapter(flock),
  e2ee: {
    provider: flockProvider,
  },
});

If one application stack uses Loro and another uses Flock, give them separate streams. Reusing the same key material across those streams is an application decision; it does not make the CRDT payloads interoperable.

Private E2EE Room

This is the default protected mode: all retained updates and snapshots are encrypted, and readers reject any plaintext payload that appears in the stream.

const privateReaderWriter = new StreamsCrdt({
  streamUrl,
  adapter: createLoroDocAdapter(doc),
  e2ee: {
    provider,
  },
});

Mixed Plaintext and Encrypted Updates

Some applications intentionally allow public writers to append CRDT updates but not read private history. Configure readers explicitly before accepting those plaintext updates:

const privateReaderWriter = new StreamsCrdt({
  streamUrl,
  adapter: createLoroDocAdapter(doc),
  e2ee: {
    readPolicy: "allow-plaintext",
    writePolicy: "encrypt",
    provider,
  },
});

Plaintext writers should opt in just as explicitly:

const publicWriter = new StreamsCrdt({
  streamUrl,
  adapter: createLoroDocAdapter(publicDoc),
  auth: async () => getWriteOnlyToken(),
  e2ee: {
    readPolicy: "allow-plaintext",
    writePolicy: "plaintext",
  },
});

publicDoc.getText("comments").insert(0, "public note");
const appended = await publicWriter.appendWriteOnly();

Behavior and Validation Rules

• If e2ee is present and no policies are specified, the defaults are readPolicy: "encrypted-only" and writePolicy: "encrypt".
• streamUrl is required, remains opaque, and is never parsed to invent a security identity.
• streams-crdt does not bind ciphertext to streamUrl. The provider/application must authenticate a stable logical room identity or use keys isolated per room. A provider configured for the wrong room must fail authentication.
• PayloadProtectionProviderConfig (alias E2eeProviderConfig) is a scope-free { provider, readPolicy?, writePolicy? } shape suitable for a repo to forward directly. Set payloadProtectionRequired: true once protection is enabled at repo level so a missing room config throws instead of becoming plaintext. This flag checks config presence only; it does not override an explicit allow-plaintext / plaintext policy such as public write-only mode.
• Lody rekey is an application-owned room-session barrier: drain, close, switch the finalized write epoch, and create a new room transport. Historical reads may resolve provider headers dynamically.
• appendWriteOnly() is the write-only path. It only posts local CRDT batches that this transport observed after construction. It does not bootstrap, catch up, join SSE, upload snapshots, apply remote updates, or persist a remote cursor.
• appendWriteOnly() must not be mixed with sync() or join() on the same transport instance.
• After close(), appendWriteOnly() is permanently unavailable on that instance because the write-only subscription has been released. Create a new StreamsCrdt instance for later write-only use.
• Snapshot uploads are encrypted when writePolicy is "encrypt". Snapshot upload is rejected when writePolicy is "plaintext"; a public writer that cannot read private state should not publish full-document snapshots.
• Protected read failures stop before applySnapshot(), applyRemoteUpdates(), local durability hooks, or remote cursor persistence.
• A recognized protected envelope never falls back to plaintext after an authentication failure, unknown version/key/epoch, wrong scope/kind, or tamper. Under encrypted-only, bytes without the envelope magic are also rejected as plaintext_forbidden. allow-plaintext intentionally accepts non-envelope bytes and is therefore not an encrypted-only room policy.
• Keep historical keys until retained protected updates and snapshots no longer reference them.
• Provider batching uses only declared bounded overhead. It never calls seal() while estimating a merge and validates the actual frozen output.

If e2ee is configured and readPolicy is left at the default "encrypted-only", remote plaintext updates fail with code: "payload_protection_error" and reason: "plaintext_forbidden".

E2EE Errors

E2EE-related transport failures surface as code: "payload_protection_error" with one of these reasons:

| Reason | Meaning | | --- | --- | | plaintext_forbidden | A reader configured as encrypted-only saw plaintext data in the stream. | | missing_read_key | The provider cannot resolve its opaque header to a historical read key. | | decrypt_failed | The provider AEAD open failed. Common causes are a wrong key, wrong scope, or tampered bytes. | | invalid_envelope | The payload was not a supported streams-crdt protected envelope/version. | | wrong_payload_kind | An encrypted snapshot was used where an update batch was expected, or the reverse. | | encrypt_failed | Local write-side encryption could not proceed because the key/config/input was invalid. |

Callers usually treat these as operator or configuration errors, not transient network failures. They are always returned as retryable: false. The public TransportError carries only the stable reason and a stable redacted message; it never copies provider causes, key ids, or epochs into the default error path.

E2EE does not hide stream URLs, the opaque provider header, offsets, total append/snapshot sizes, timing, or auth metadata. It hides the count and individual sizes of updates within one encrypted append batch, but it is not a traffic-analysis defense. It also does not by itself detect a malicious server rolling back or forking history. In mixed allow-plaintext mode, no format can distinguish deliberate plaintext from ciphertext whose magic was stripped; use encrypted-only for a no-downgrade room.

Remote Cursor Store

The RemoteCursorStore persists replay progress metadata only — it does not store any CRDT update data. A remote cursor contains:

• nextOffset — where the transport should resume reading from
• serverLowerBoundVersion — the CRDT version that the server data represents
• streamUrl — identity of the stream

Important: cursor-only persistence is unsafe

If RemoteCursorStore.load() returns a saved cursor, the transport resumes from nextOffset and assumes the caller has already restored the matching local CRDT state represented by serverLowerBoundVersion.

If the local CRDT was reset to empty but the cursor was persisted, the transport will do delta-only catch-up from nextOffset — applying those deltas against an empty document will fail or produce an invalid state.

Rule of thumb

| Local CRDT state | Cursor store | Safe? | | --- | --- | --- | | Not persisted (ephemeral) | InMemoryRemoteCursorStore (default) | Yes | | Persisted (IndexedDB, SQLite, etc.) | IndexedDbRemoteCursorStore + beforeRemoteCursorSave hook | Yes | | Not persisted | IndexedDbRemoteCursorStore | No — data loss risk |

If you do not persist the local CRDT payload across reloads, keep the remote cursor ephemeral too. The default InMemoryRemoteCursorStore is the safe choice.

Hooking Local Durability

Use beforeRemoteCursorSave to couple local payload persistence with remote cursor advancement. The transport advances the cursor only after:

1. Remote data was applied locally
2. The durability hook succeeded
3. The cursor store write succeeded

const transport = new StreamsCrdt({
  streamUrl: "https://streams-api.loro.dev/ds/my-bucket/doc-1",
  adapter: createLoroDocAdapter(doc),
  beforeRemoteCursorSave: async ({ cursor, source }) => {
    // Persist local CRDT state before cursor advances.
    // `source` is "local" | "remote" | "bootstrap".
    await persistLocalPayload(doc);
    console.log(source, cursor.nextOffset);
  },
});

persistLocalPayload(doc) must make the local durable state correspond to the cursor that is about to be stored. On the next app start, restore that local state before calling sync() or join().

Pers