@thru/replay

@thru/replay turns chain RPC backfill plus live streaming into a single ordered feed.

Install

npm install @thru/replay @thru/sdk

When To Use It

Choose this package when you need a durable ordered feed for analytics, ETL, or event processing and want to decide where the data lands yourself.

Choose a different package when:

• you need persistence, checkpoints, and Drizzle-backed stream definitions on top of the feed: use @thru/indexer
• you need a full app-facing RPC SDK instead of replay primitives: use @thru/sdk

Entry Point

The package is root-only. Import what you need from @thru/replay; there are no public subpath exports.

Main Exports

Export	Use it for
ChainClient	Connecting to Thru query and streaming services with one client wrapper.
createBlockReplay	Ordered block replay with optional filters, consensus floor, and block view settings.
createTransactionReplay	Ordered transaction replay, including optional event payloads.
createEventReplay	Ordered event replay with reconnect support.
createAccountReplay	Replaying one account’s state over time.
createAccountsByOwnerReplay	Replaying all accounts owned by a program with backfill plus live updates.
AccountSeqTracker and MultiAccountReplay	Tracking sequence numbers and managing multiple single-account replay streams.
ReplayStream	The async iterator that merges backfill and live data into one ordered stream.
PageAssembler	Reassembling multi-page account updates into complete account payloads.
ReplaySink and ConsoleSink	Writing replay output to a sink implementation or the console.
createConsoleLogger and NOOP_LOGGER	Structured logging for replay runs.
DEFAULT_RETRY_CONFIG, calculateBackoff, withTimeout, delay, TimeoutError	Retry and timeout helpers used by replay and available to consumers.

Common Workflows

• Use createBlockReplay, createTransactionReplay, or createEventReplay when you want a typed ordered feed for analytics, ETL, or event processing.
• Use createAccountsByOwnerReplay when you need to index all accounts owned by a program and keep them current.
• Use ReplayStream directly when you already have your own backfill fetcher and live subscriber.
• Use PageAssembler when you need to assemble multi-page account updates into complete payloads before processing them.

Replay Lifecycle

ReplayStream joins a historical backfill source and a live streaming source into one AsyncIterable.

1. backfill: fetches ordered pages from startSlot.
2. switching: drains live items buffered while backfill was catching up.
3. streaming: yields live items and reconnects when the stream errors or ends unexpectedly.

The live stream starts while backfill is running. safetyMargin controls how far behind the observed live tip the backfill must reach before the stream switches to live output. Items are deduplicated across the backfill/live overlap using each replay type’s key function.

Backfill pages must be ordered by ascending slot. If a custom ReplayStream backfill source returns pages out of order, replay throws rather than silently emitting an inconsistent feed.

Use signal to stop replay without reconnecting, and resubscribeOnEnd to control whether an ended live stream should be treated as reconnectable. The default is to resubscribe on end.

const abort = new AbortController();
const replay = createTransactionReplay({
  clientFactory: () => new ChainClient({ baseUrl: process.env.CHAIN_RPC_URL! }),
  startSlot: 1_000_000n,
  safetyMargin: 64n,
  signal: abort.signal,
});

for await (const tx of replay) {
  console.log(tx.slot?.toString());
  console.log(replay.getMetrics());
}

getMetrics() reports in-memory counters for buffered items, backfill/live emissions, reconnect emissions, and discarded duplicates. Persist durable progress in your own storage or use @thru/indexer when you want managed checkpoints.

Factory Options

Each replay factory is tuned to the source it reads:

Factory	Client shape	Notable options
createBlockReplay	client	filter, view, minConsensus, safetyMargin, pageSize, resubscribeOnEnd, signal
createTransactionReplay	client or clientFactory	filter, minConsensus, returnEvents, safetyMargin, pageSize, resubscribeOnEnd, signal
createEventReplay	client or clientFactory	filter, resumeAfter, safetyMargin, pageSize, resubscribeOnEnd, signal
createAccountReplay	client	address, view, filter, pageAssemblerOptions, cleanupInterval
createAccountsByOwnerReplay	client or clientFactory	owner, view, dataSizes, minUpdatedSlot, pageSize, maxRetries, retryConfig, onBackfillComplete, signal

For long-running services, prefer clientFactory whenever the selected factory supports it. Reconnects can then create a fresh transport and dispose stale clients that expose close().

Account Replay

Account replay is split into two shapes:

• createAccountReplay for one account address.
• createAccountsByOwnerReplay for owner-scoped indexing with backfill, live updates, and reconnect handling.

createAccountsByOwnerReplay accepts either client or clientFactory. Prefer clientFactory for long-running workers so reconnects can create a fresh transport. Owner-scoped replay also supports minUpdatedSlot, dataSizes, pageSize, maxRetries, pageAssemblerOptions, reconnectCleanupTimeoutMs, retryConfig, onBackfillComplete, logger, and signal.

Account replay yields AccountReplayEvent values:

for await (const event of replay) {
  if (event.type === "account") {
    console.log(event.account.addressHex);
    console.log(event.account.slot);
    console.log(event.account.seq);
    console.log(event.account.isDelete);
    console.log(event.account.source); // "backfill" or "stream"
  }
}

The replay package depends on generated protobuf types from @thru/sdk/proto internally.

ChainClient Configuration

ChainClient wraps Thru query and streaming services for replay workloads.

Option	What it does
baseUrl	gRPC endpoint used when ChainClient creates its own transport.
apiKey	Adds a bearer Authorization header to owned transports.
userAgent	Adds a User-Agent header to owned transports.
transport	Uses an application-owned Connect transport instead of creating one.
interceptors	Adds Connect interceptors to an owned transport.
callOptions	Passes Connect call options to every RPC made by the client.
useBinaryFormat	Controls binary protobuf format for owned transports. Defaults to true.

When you do not pass transport, baseUrl is required and the client owns the underlying HTTP/2 session. When you do pass transport, your application owns that transport’s lifecycle.

const client = new ChainClient({
  baseUrl: process.env.CHAIN_RPC_URL!,
  apiKey: process.env.CHAIN_API_KEY,
  userAgent: "my-indexer/1.0",
});

Client Lifecycle

ChainClient owns an HTTP/2 session when it creates its own transport. Call close() when you are done with a standalone client. If you pass a custom transport, that transport remains owned by your application.

Long-running replay factories that accept clientFactory close stale clients on reconnect when those clients expose close().

const client = new ChainClient({ baseUrl: process.env.CHAIN_RPC_URL! });

try {
  const replay = createBlockReplay({ client, startSlot: 0n });
  for await (const block of replay) {
    console.log(block.header?.slot?.toString());
  }
} finally {
  client.close();
}

Calling close() is idempotent. It aborts the owned HTTP/2 session, so make sure no in-flight RPCs or streams are still expected to complete on that client.

Data Model

Replay items are ordered by slot, deduplicated across the backfill/live overlap window, and exposed through an AsyncIterable. ReplaySinkContext tags each item with the replay phase, backfill or live, so downstream code can treat historical and realtime data differently if needed.

Minimal Example

import { ChainClient, createBlockReplay } from "@thru/replay";

const client = new ChainClient({ baseUrl: process.env.CHAIN_RPC_URL! });
const replay = createBlockReplay({ client, startSlot: 1_000_000n });

for await (const block of replay) {
  console.log(block.header?.slot?.toString());
}

For transactions, events, and owner-scoped account replay, prefer clientFactory in services that must reconnect cleanly:

const replay = createEventReplay({
  clientFactory: () => new ChainClient({ baseUrl: process.env.CHAIN_RPC_URL! }),
  startSlot: checkpoint.slot,
  resumeAfter: checkpoint.eventId
    ? { slot: checkpoint.slot, eventId: checkpoint.eventId }
    : undefined,
});

Related Guides

• Indexing Overview for package selection and the full indexing guide structure.
• Build an Indexer for a step-by-step guide using @thru/indexer on top of replay.
• @thru/indexer for persistence and checkpoints built on replay.
• @thru/sdk for the full app-facing RPC client.