Running the Indexer

Use this page when the stream definitions are ready and you need to wire the runtime, database, migrations, and process shape.

Architecture

The typical Thru indexing stack looks like this:

Chain RPC -> @thru/replay ChainClient -> @thru/indexer runtime -> Postgres tables
                                                                     |
                                                                     v
                                                        app-owned queries/routes

Tech Stack Assumptions

The current @thru/indexer runtime assumes:

• PostgreSQL-backed tables generated through Drizzle
• a Drizzle database client passed as db
• a clientFactory that returns a ChainClient from @thru/replay
• Drizzle Kit or equivalent migration management
• a standalone Node service to run the indexer continuously

If you are not using Postgres plus Drizzle, start with the package reference for @thru/indexer and validate whether the current runtime fits your stack.

Runtime Setup

import { Indexer } from "@thru/indexer";
import { ChainClient } from "@thru/replay";
import { db } from "./db";
import tokenAccounts from "./account-streams/token-accounts";
import tokenTransfers from "./streams/token-transfers";

export function createIndexer() {
  return new Indexer({
    db,
    clientFactory: () => new ChainClient({ baseUrl: process.env.CHAIN_RPC_URL! }),
    eventStreams: [tokenTransfers],
    accountStreams: [tokenAccounts],
    defaultStartSlot: 0n,
    safetyMargin: 64,
    pageSize: 512,
    logLevel: "info",
  });
}

What Each Runtime Option Does

Option	What it controls
db	The Drizzle client used for inserts, updates, and checkpoints.
clientFactory	Fresh replay client creation for backfill and live streaming.
eventStreams	Append-only streams for event rows.
accountStreams	Current-state streams for account rows.
defaultStartSlot	Starting slot when no checkpoint exists yet.
safetyMargin	How far behind the live tip replay should stay during backfill-to-live switchover.
pageSize	How many records to request per backfill page.
logLevel	Runtime verbosity.
logger	Structured logger used by the runtime, stream processors, and replay layer.
endpointLabel	Human-readable endpoint label included in normalized stream errors.
supervisorInitialBackoffMs	First stream-supervisor restart delay after an unexpected stream failure.
supervisorMaxBackoffMs	Maximum stream-supervisor restart delay.
streamStaleMs	Marks a running stream stale after this much time without activity. Disabled by default.
validateParse	Validates parsed stream rows against generated Zod schemas. Useful in development.

Runtime Supervision

Indexer.start() supervises every configured stream. If a stream fails or ends unexpectedly, the supervisor restarts it with backoff unless shutdown has been requested. indexer.stop() asks streams to stop gracefully; calling stop() a second time forces process exit.

const indexer = createIndexer();

process.on("SIGINT", () => indexer.stop());
process.on("SIGTERM", () => indexer.stop());

await indexer.start();

Use supervisorInitialBackoffMs and supervisorMaxBackoffMs to tune restart behavior for your deployment.

Runtime Status

Use indexer.getStatus() for process health checks, admin endpoints, and debugging. Status is in-memory runtime state; checkpoints remain the durable resume source.

const status = indexer.getStatus();

console.log(status.running);
console.log(status.healthy);
console.log(status.streams.map((stream) => ({
  name: stream.name,
  kind: stream.kind,
  state: stream.state,
  checkpointSlot: stream.checkpointSlot,
  lastProcessedSlot: stream.lastProcessedSlot,
  stale: stream.stale,
  restartCount: stream.restartCount,
  counters: stream.counters,
  lastError: stream.lastError,
})));

When streamStaleMs is configured, a running stream is marked stale if it has not seen activity since lastEventAt or lastStartedAt. Staleness is a signal for monitoring, not a restart trigger by itself.

Checkpoints And Schema

Your Drizzle schema needs the checkpoint table plus every stream table.

export { checkpointTable } from "@thru/indexer";
export { tokenAccountsTable } from "./account-streams/token-accounts";
export { tokenTransferEvents } from "./streams/token-transfers";

Without checkpointTable, the runtime cannot resume safely after restarts.

Process Shape

In practice, most apps run the indexer as its own long-lived service:

1. load environment and connect to Postgres
2. run or verify migrations
3. build the Indexer
4. call await indexer.start()
5. stop gracefully on SIGTERM or SIGINT

const indexer = createIndexer();

process.on("SIGINT", () => indexer.stop());
process.on("SIGTERM", () => indexer.stop());

await indexer.start();

Operations Runbook

Run migrations before the worker starts. The schema must include checkpointTable and every stream table exported by your stream modules. If a new deploy changes a stream schema, deploy the migration before starting a worker that writes the new row shape.

Expose indexer.getStatus() from an internal-only health endpoint or process supervisor hook. Treat healthy: false, a stream state other than running, a non-null lastError, or a growing restartCount as signals for operator attention.

export function indexerHealthResponse(indexer: Indexer) {
  const status = indexer.getStatus();

  return {
    ok: status.healthy,
    uptimeMs: status.uptimeMs,
    streams: status.streams.map((stream) => ({
      name: stream.name,
      kind: stream.kind,
      state: stream.state,
      checkpointSlot: stream.checkpointSlot,
      lastProcessedSlot: stream.lastProcessedSlot,
      stale: stream.stale,
      restartCount: stream.restartCount,
      lastError: stream.lastError,
    })),
  };
}

Recommended alerts:

• healthy stays false after startup
• any stream is retrying for longer than the expected endpoint outage window
• restartCount increases repeatedly
• stale is true when streamStaleMs is configured
• lastError.phase is parse, filterBatch, or onCommit

Parser and hook errors are usually application issues. Backfill, live, commit, and supervisor errors are more likely to be endpoint, database, or process lifecycle issues.

Checkpoint Resets

Use checkpoint resets deliberately. A reset causes the affected stream to replay from defaultStartSlot or from the new checkpoint value you write manually. Make sure the target tables can tolerate replayed rows before resetting: event streams should have stable primary keys, and account streams should use slot-aware current-state rows.

import {
  deleteCheckpoint,
  getAllCheckpoints,
  updateCheckpoint,
} from "@thru/indexer";
import { db } from "./db";

console.table(await getAllCheckpoints(db));

// Replay one event stream from the configured defaultStartSlot.
await deleteCheckpoint(db, "token-transfers");

// Replay one account stream from the configured defaultStartSlot.
await deleteCheckpoint(db, "account:token-accounts");

// Or move a stream to a known-good slot after an operator decision.
await updateCheckpoint(db, "token-transfers", 1_250_000n, null);

If the table contents are not idempotent for the replay window, truncate or delete the affected rows in the same maintenance window as the checkpoint reset. For event streams, keep the last event ID with the checkpoint when resuming inside a slot. For account streams, the checkpoint name is prefixed with account: and records the last handled account update slot.

Development Validation

Enable validateParse in development or staging when changing stream schemas. It validates parse output against the generated schema before rows are committed, which catches missing fields and type mismatches earlier. Leave it off for hot production paths unless the extra validation cost is acceptable.

Reading The Results

The indexer writes standard Drizzle tables. Query those tables directly from your backend or expose routes owned by your application.

import { desc } from "drizzle-orm";
import { db } from "./db";
import { tokenTransferEvents } from "./schema";

export async function listRecentTransfers(limit = 50) {
  return await db
    .select()
    .from(tokenTransferEvents)
    .orderBy(desc(tokenTransferEvents.slot))
    .limit(limit);
}

Next Steps

• Open Querying Indexed Data once rows are landing in Postgres.
• See Build an Indexer for production guidance on worker/API separation, resumability, and live validation.
• See @thru/indexer and @thru/replay for full package references.