@loro-dev/streams-client

TypeScript client for a single Durable Stream URL.

Use this package when you already know the stream URL and want to:

• create or delete the stream
• append bytes or close the stream
• read from an offset
• tail new data with SSE or long-poll
• bootstrap from a snapshot, then continue from the returned offset

This package is protocol-level on purpose. It does not encode CRDT updates for you or manage CRDT state in memory.

Native HTTP capability

The additive @loro-dev/streams-client/effect entry exports makeStreamsHttp, StreamsAuth, streamsAuthLegacy, and retained release errors. It consumes Effect 4.0.2; the peer is optional for the existing Promise entry. This entry is available from @loro-dev/[email protected].

makeStreamsHttp belongs to a Scope and consumes a native StreamsAuth Layer. http.use(url, init, consume, options) owns one authorized request, its response and actual locked reader. consume composes Effects with bytes, nextEvent and SSE activity (including comment keepalives). authTimeoutMs separately bounds each token waiter. Per-use timeoutMs starts with each actual send and covers fetch plus the consumed body; headersOnly bounds the wait for headers while leaving streaming body supervision to its owner. Native fetch exposes no independent TCP-only deadline. There is one 401/403 refresh retry and no durable protocol retry; the CRDT kernel owns retry/backoff.

For durable bodies, use phase: "upload" with an independent finite total timeoutMs; the native CRDT entry supplies its upload budget. The default fetch transport reports uploadMode: "total-only", without invented byte counts or upload stages. An optional uploadTransport may report actual transmitted bytes and body completion through the shared UploadTransport contract. Per-use uploadIdleTimeoutMs then bounds lack of progress, including the initial wait and the wait for response headers after body transmission. Genuine increments renew only this idle budget; the finite total cap still wins. Local body reads, encoding and timer heartbeats do not qualify. No built-in browser or Node fetch adapter claims this capability. Enabling idle without one, or a progress transport without a positive finite total cap, fails explicitly before I/O.

Opt-in onDiagnostic uses the shared safe request event schema and generated per-attempt X-Request-ID. Auth events do not assert an HTTP send. Request success means headers returned, while response-body failure and durability remain separate. Caller signals cancel their current auth/request waiter and stop owned I/O; raw abort-insensitive I/O remains retained and fences replacement until actual settlement and release recovery.

Cancellation detaches the waiter and stops its request resources. Shared refresh producers remain owned separately; provider-wide refresh belongs to the provider Scope. A failed or timed-out release preserves full private Cause, actual work and retryRelease, preventing replacement requests until recovery. close shares its result. Scope finalizers preserve that release error object as a defect, including its private Cause and usable retryRelease Effect. A cancellation signal alone never counts as completed raw work. After headers arrive, the response lease owns cancellation; cleanup releases its body/reader without first aborting fetch. SSE has a separate reader-cancellation signal. releaseCause also retains private history of producers that actually settled during cancellation; a settled producer rejection alone is not failed resource release. Actual cleanup failures and unfinished work still fence admission.

use, close, retryRelease and token acquisition expose fixed-name spans with safe outcome attributes. Automatic HTTP URL/header tracing and automatic logging of private Causes remain disabled. The original full Cause stays with the caller.

Each shared refresh has a new owner registered before its producer starts. Failure publication waits for bounded observation of actual release; unresolved raw auth work fences replacement until retryRelease observes settlement. Recovery also awaits the refresh coordinator before clearing its receipt. An open HTTP capability can refresh again after successful recovery; a closed capability remains closed. POST 403 responses carrying Producer-Epoch bypass token refresh so the durable protocol can repair that producer epoch with its frozen body.

Promise openSseSession also accepts strictRelease and onRelease when an owner needs the original reader cancellation completion. Its strict cancel waits for that actual reader/iterator, including failures. Default Promise readers retain their existing behavior. The native capability uses the same SSE parser directly with strict ownership; it does not call the Promise client's auth, retry or timeout state machines.

See ownership and release ordering.

Install

pnpm add @loro-dev/streams-client

Runtime requirements:

• modern browsers and worker runtimes are supported
• Node.js >=18 is supported
• if your runtime does not expose standard Web APIs like fetch, Headers, ReadableStream, and DOMException, you need to provide compatible polyfills before using this package

Start Here

The main entry is:

import { StreamsClient } from "@loro-dev/streams-client";

Create one client per stream URL:

const client = new StreamsClient({
  url: "https://streams-api.loro.dev/ds/my-bucket/my-stream",
  auth: async () => accessToken,
});

Notes:

• url is the full stream URL, not just baseUrl or bucketId.
• auth should return the raw token string. The client adds Bearer for you.
• Import entry: @loro-dev/streams-client
• Repo source entry: src/index.ts
• Main client implementation in this repo: src/client.ts

30-Second Example

import { StreamsClient } from "@loro-dev/streams-client";

const client = new StreamsClient({
  url: "https://streams-api.loro.dev/ds/my-bucket/my-stream",
});

const created = await client.create({
  contentType: "application/json",
});
if (!created.ok) throw created.result;

const appended = await client.append({
  part: {
    contentType: "application/json",
    body: JSON.stringify({ type: "hello", value: 1 }),
  },
});
if (!appended.ok) throw appended.result;

const read = await client.read({ offset: "-1" });
if (!read.ok) throw read.result;
console.log(read.result.payload.json());

for await (const event of client.live({ offset: appended.result.nextOffset })) {
  if (event.type === "data") {
    console.log(event.payload.json());
  }
  if (event.type === "eof") {
    break;
  }
}

What You Usually Need

| Task | API | What it returns | Spec | | -------------------------------- | ---------------------------------------- | --------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------- | | Create or ensure a stream exists | client.create() | created, contentType, nextOffset, closed | docs/ds-protocol.md -> 5.1 Create Stream | | Append data | client.append() | nextOffset, closed | docs/ds-protocol.md -> 5.2 Append to Stream | | Close the stream | client.close() | same shape as append | docs/ds-protocol.md -> 5.3 Close Stream | | Read current stream metadata | client.head() | contentType, nextOffset, closed, optional snapshot info | docs/ds-protocol.md -> 5.5 Stream Metadata | | Read once from an offset | client.read() | payload plus nextOffset | docs/ds-protocol.md -> 5.6 Read Stream - Catch-up | | Do one long-poll round trip | client.readOnce({ live: "long-poll" }) | 200 with payload or 204 with no payload | docs/ds-protocol.md -> 5.7 Read Stream - Live (Long-poll) | | Tail the stream continuously | client.live() | async iterable of data, up_to_date, eof, reconnect events | docs/ds-protocol.md -> 5.7 and 5.8 | | Open a raw SSE session | client.openSseSession() | lower-level SSE session object | docs/ds-protocol.md -> 5.8 Read Stream - Live (SSE) | | Delete the stream | client.delete() | deleted | docs/ds-protocol.md -> 5.4 Delete Stream |

For most callers:

• use read() for one catch-up read
• use live() for long-running tailing
• use openSseSession() only if you need raw SSE control

Result Shape

Every network method returns:

type Result<T, E> = { ok: true; result: T } | { ok: false; result: E };

That means your normal flow is:

const result = await client.append(...);
if (!result.ok) {
  console.error(result.result);
  return;
}

console.log(result.result.nextOffset);

If you need request and response details for debugging, pass includeMeta: true.

For safe diagnostics across application wrappers, use result.result.context on failure. It contains fixed source/operation/discriminator/retry fields plus available timeout phase/budget, elapsed duration, HTTP status, constrained request ID and allowlisted network code. It excludes arbitrary error messages, causes, stacks, bodies, headers and URL queries. The legacy error and includeMeta fields are not safe diagnostic projections.

Fetch exceptions (including ordinary Error and browser/cross-realm TypeError) are classified at the actual fetch/response-reader boundary. An unmarked local TypeError remains unknown; an auth callback has source: "auth". A fetch classification does not prove a connection was attempted, especially with a custom fetch implementation. See the implemented safe error contract for exact fields and bounds. toStreamOperationError() preserves context; thrownToStreamError() also normalizes exceptions from low-level SSE readers without guessing sources from message text.

Payloads and Offsets

Read APIs return a StreamPart:

const read = await client.read({ offset: "-1" });
if (!read.ok) throw read.result;

read.result.payload.body;
read.result.payload.text();
read.result.payload.json();

Offset rules:

• use -1 to read from the earliest retained data
• use now to start from the current tail
• otherwise, reuse the server-returned nextOffset
• treat offsets as opaque strings

Spec: docs/ds-protocol.md -> 6. Offsets

Live Reads

client.live() is the high-level live API:

for await (const event of client.live({ offset: "now" })) {
  if (event.type === "data") {
    console.log(event.payload.text());
  }
}

Defaults:

• mode defaults to "auto"
• "auto" tries SSE first. It falls back to long-poll only if the first SSE connection is rejected as unsupported.

Reconnect behavior:

• Network errors, timeouts, and HTTP 429/502/503/504 are retried, up to retry.maxAttempts consecutive failures. The backoff is never shorter than the server's Stream-Retry-After-Ms / Retry-After hint.
• An SSE connection that receives nothing, not even the server's 15s keepalives, for max(pollTimeoutMs + connectTimeoutMs, 45s) is treated as a poll timeout and reconnected.
• Long-poll requests send timeout=<pollTimeoutMs in seconds> so the server ends an idle poll with a 204. The client deadline is pollTimeoutMs + connectTimeoutMs.
• An SSE event cut off by a dropped connection is discarded and read again on reconnect.

Request deadlines:

• connectTimeoutMs retains its legacy name but bounds a whole control/read request, including pool wait, connection, upload and response body. It is not a TCP-only deadline; native fetch does not expose that boundary.
• Durable append, appendCas and putSnapshot use uploadTimeoutMs instead (default 60 seconds), including queueing and waiting for headers/body. This is a total budget, not an idle-progress timer: slow useful uploads may use it, while stalled uploads and lost responses still abort within it.
• Token acquisition and unauthorized refresh use authTimeoutMs, independently of request body size (default connectTimeoutMs). Timeout phases distinguish auth, upload, control connect, and poll. Each retry has a fresh budget.
• Long-poll uses pollTimeoutMs + connectTimeoutMs. Error bodies stay bounded; an error response whose body stalls still returns its observed HTTP status.
• Optional onDiagnostic receives safe phase observations using monotonic time. Opting in generates a separate opaque X-Request-ID for every fetch attempt, including an unauthorized retry; an auth event does not imply fetch started. No URL, token, payload, arbitrary message or provider cause is emitted. Native /effect executes the corresponding budgets and diagnostics through its own HTTP/IoOwner implementation; it does not wrap the Promise client.
• Optional uploadTransport can provide genuine transmitted-byte observations; uploadIdleTimeoutMs then renews a stall window while the independent finite upload hard cap still wins. It is rejected without the capability. Default fetch remains explicitly total-only, with no invented byte counts or idle default. See the native section above and the owning upload decision.
• SSE (openSseSession, live()) bounds only the wait for headers. The event body is long-lived and guarded by the idle watchdog instead.

Spec:

• docs/ds-protocol.md -> 5.7 Read Stream - Live (Long-poll)
• docs/ds-protocol.md -> 5.8 Read Stream - Live (SSE)

Snapshot and Bootstrap

If your stream supports snapshots, these are the key methods:

| Task | API | Spec | | ------------------------------------------- | -------------------------------------- | ------------------------------------------------------------------------------------------------------ | | Read latest visible snapshot | client.readLatestSnapshot() | docs/ds-extensions.md -> 2.3 Read Latest Snapshot | | Read a specific snapshot | client.readSnapshot(offset) | docs/ds-extensions.md -> 2.5 Read Snapshot | | Publish a snapshot | client.putSnapshot({ offset, part }) | docs/ds-extensions.md -> 2.2 Publish Snapshot | | Delete a snapshot | client.deleteSnapshot(offset) | docs/ds-extensions.md -> 2.6 Delete Snapshot | | Get snapshot + retained updates in one call | client.bootstrap() | docs/ds-extensions.md -> 3. Bootstrap |

Typical recovery flow:

1. Call client.bootstrap()
2. Apply snapshot if it exists
3. Apply updates in order
4. Continue with client.live({ offset: bootstrap.result.nextOffset })

If a normal read returns 410 Gone, you usually need bootstrap() before you continue. Spec: docs/ds-extensions.md -> 3.4 Compatibility

Producer Coordination

If you use idempotent writers, pass producer to append() or close():

await client.append({
  part: {
    contentType: "application/octet-stream",
    body: bytes,
  },
  producer: {
    producerId: "writer-1",
    epoch: 1,
    seq: 42,
  },
});

Spec: docs/ds-protocol.md -> 5.2.1 Idempotent Producers

Other Exports

These helpers are useful when you need them, but most users only need StreamsClient:

• createStreamPart(contentType, body): build a StreamPart from bytes
• normalizeStreamBody(body): normalize string, Uint8Array, ArrayBuffer, or view into Uint8Array
• DEFAULT_RETRY_CONFIG
• DEFAULT_TIMEOUT_CONFIG

Spec Map

This package ships the protocol docs in docs/, so after install you can read:

• node_modules/@loro-dev/streams-client/docs/ds-protocol.md
• node_modules/@loro-dev/streams-client/docs/ds-extensions.md

Use this map when you need exact wire semantics:

• stream URL shape and bucket rules: docs/ds-extensions.md -> 1. Buckets
• create, append, close, delete, head, read, long-poll, SSE: docs/ds-protocol.md
• offsets and JSON mode: docs/ds-protocol.md -> 6. Offsets, docs/ds-protocol.md -> 7.1 JSON Mode
• snapshots and bootstrap: docs/ds-extensions.md -> 2. Snapshots, docs/ds-extensions.md -> 3. Bootstrap

Backward history (experimental RR extension)

client.readBackwardMessages({ before?, pageBytes?, maxMessageBytes?, signal? }) returns complete messages, oldest first, plus startOffset, endOffset, atStart and atRetainedStart:

let before: string | undefined; // undefined = stream tail
const page = await client.readBackwardMessages({ before });
if (page.ok) {
  render(page.result.messages);
  before = page.result.startOffset; // message-aligned; safe to persist
  if (page.result.atRetainedStart) stopLoadingOlder(); // start or trim floor
}

• Each request fetches at most pageBytes (default DEFAULT_BACKWARD_PAGE_BYTES, 256 KiB). A larger message is fetched over several requests and buffered in memory until it is complete, up to maxMessageBytes (default DEFAULT_BACKWARD_MAX_MESSAGE_BYTES, 64 MiB); fragments are never returned. Returned messages are copies, so keeping one does not pin a whole page.
• startOffset is where the oldest returned message starts. Pass it as before to keep reading backward, or as a forward read({ offset }) to re-read from there. endOffset is where the newest one ends.
• atRetainedStart means nothing older is retained: the original start (atStart too) or a trim floor. A cursor that later falls below a trim floor also ends with atRetainedStart instead of an error. A message cut by the floor is dropped rather than returned partial.
• A before that is not on a message boundary, or not a canonical 20-digit offset as returned by the server, is rejected.

The lower-level client.readBackward({ before?, limitBytes }) returns one raw page as frames (data, continuesBefore, continuesAfter) with prevOffset, endOffset, atStart and atRetainedStart, for callers that reassemble themselves. See the protocol.

readSseEvents also accepts strictRelease and onRelease for resource owners. With strict release, abort cancels the locked reader, waits for that cancellation, propagates release failure and releases its lock. onRelease reports the original cancellation Promise even with default best-effort observation, which does not await an uncooperative cancellation. An owner must retain that original completion. safeRequestId and networkErrorCode expose the existing bounded diagnostic projections for native client adapters; they never return raw messages or causes.

isAbortError is exported for owning Effect adapters: only an AbortError paired with their actually cancelled signal becomes interruption; unrelated failures remain observable.