Integrate a Custom Indexer
The sui-indexer-alt-framework crate handles checkpoint ingestion, batching, watermarks, and crash recovery, so your service only implements 2 traits: Processor extracts the data you care about from each checkpoint, and Handler writes it to your store.
The steps below cover the Sui-specific work of adding that pipeline to a service you already have. For the conceptual background, read Custom Indexing Framework and Indexer Pipeline Architecture first.
- Prerequisites
-
An existing Rust service, with Rust and Cargo installed.
-
A store to write to. The examples use PostgreSQL with Diesel, which the framework supports out of the box.
-
To write somewhere other than PostgreSQL, implement the framework's
StoreandConnectiontraits. See Bring Your Own Store.
Starting from nothing rather than an existing service? Build a Custom Indexer walks through the same framework from an empty directory, including project creation, database setup, and Diesel migrations.
To follow a complete runnable project instead of building by hand, clone the example:
$ git clone https://github.com/MystenLabs/sui.git
$ cd sui/examples/rust/basic-sui-indexer
That project contains the Cargo.toml, Diesel migrations, and every source file this guide imports code from. Its README covers database creation and migration setup. The framework source lives in the same repo.
Add the framework dependency
Add the following to your Cargo.toml:
[package]
name = "basic-sui-indexer"
version = "0.1.0"
edition = "2024"
[dependencies]
# Core framework dependencies
sui-indexer-alt-framework = { git = "https://github.com/MystenLabs/sui.git", branch = "testnet" }
# Async runtime
tokio = { version = "1.0", features = ["full"] }
# Error handling
anyhow = "1.0"
# Diesel PostgreSQL
diesel = { version = "2.0", features = ["postgres", "r2d2"] }
diesel-async = { version = "0.5", features = ["bb8", "postgres", "async-connection-wrapper"] }
diesel_migrations = "2.0"
# Async traits
async-trait = "0.1"
# URL parsing
url = "2.0"
# Use .env file
dotenvy = "0.15"
# Command line parsing
clap = { version = "4.0", features = ["derive"] }
Each dependency serves a distinct purpose:
sui-indexer-alt-framework: Core framework providing pipeline infrastructure.diesel/diesel-async: Type-safe database ORM with asynchronous support.tokio: Async runtime the framework requires.clap: Command-line argument parsing for configuration.anyhow: Error handling, and async-trait for trait implementations.dotenvy: Reads the.envfile that holds your PostgreSQL URL.
If your service already pulls in tokio, clap, or anyhow, reconcile the versions rather than adding duplicates.
Prepare the destination table
Your pipeline needs somewhere to write. The example indexer stores transaction digests in a transaction_digests table, created through a Diesel migration:
CREATE TABLE IF NOT EXISTS transaction_digests (
tx_digest TEXT PRIMARY KEY,
checkpoint_sequence_number BIGINT NOT NULL
);
This example uses the TEXT data type for tx_digest, but a production indexer should use BYTEA.
TEXT keeps the digest readable and directly usable with external tools. Digests are Base58 encoded, and because PostgreSQL cannot natively display BYTEA data in that format, storing it as TEXT lets you copy a digest from a query and paste it into an explorer like SuiScan.
BYTEA stores the raw byte representation, which is more compact and significantly faster to compare than a string. Refer to Binary data performance in PostgreSQL on the CYBERTEC website for more information.
Generate the matching Rust schema with diesel print-schema --database-url $DATABASE_URL > src/schema.rs. The example project's schema looks like this:
// @generated automatically by Diesel CLI.
diesel::table! {
transaction_digests (tx_digest) {
tx_digest -> Text,
checkpoint_sequence_number -> Int8,
}
}
Define your row type
Define a struct that represents 1 record in your destination table:
use diesel::prelude::*;
use sui_indexer_alt_framework::FieldCount;
use crate::schema::transaction_digests;
#[derive(Insertable, Debug, Clone, FieldCount)]
#[diesel(table_name = transaction_digests)]
pub struct StoredTransactionDigest {
pub tx_digest: String,
pub checkpoint_sequence_number: i64,
}
3 annotations matter here:
FieldCount: The framework requires this for memory optimization and batch efficiency. It caps batch size so a single SQL statement stays within the PostgreSQL bind parameter limit.diesel(table_name = transaction_digests): Maps the struct to the table whose schema you generated.Insertable: Lets Diesel insert the struct into the database.
Implement the Processor trait
Processor defines how your indexer extracts and transforms data from each checkpoint. Whatever it returns flows to Handler::commit.
Declare a concrete struct to hang both trait implementations on:
pub struct TransactionDigestHandler;
#[async_trait::async_trait]
impl Processor for TransactionDigestHandler {
const NAME: &'static str = "transaction_digest_handler";
type Value = StoredTransactionDigest;
async fn process(&self, checkpoint: &Arc<Checkpoint>) -> Result<Vec<Self::Value>> {
let checkpoint_seq = checkpoint.summary.sequence_number as i64;
let digests = checkpoint
.transactions
.iter()
.map(|tx| StoredTransactionDigest {
tx_digest: tx.transaction.digest().to_string(),
checkpoint_sequence_number: checkpoint_seq,
})
.collect();
Ok(digests)
}
}
Import the dependencies the trait needs:
use anyhow::Result;
use std::sync::Arc;
use sui_indexer_alt_framework::pipeline::Processor;
use sui_indexer_alt_framework::types::full_checkpoint_content::Checkpoint;
use crate::models::StoredTransactionDigest;
use crate::schema::transaction_digests::dsl::*;
Then implement the trait:
#[async_trait::async_trait]
impl Processor for TransactionDigestHandler {
const NAME: &'static str = "transaction_digest_handler";
type Value = StoredTransactionDigest;
async fn process(&self, checkpoint: &Arc<Checkpoint>) -> Result<Vec<Self::Value>> {
let checkpoint_seq = checkpoint.summary.sequence_number as i64;
let digests = checkpoint
.transactions
.iter()
.map(|tx| StoredTransactionDigest {
tx_digest: tx.transaction.digest().to_string(),
checkpoint_sequence_number: checkpoint_seq,
})
.collect();
Ok(digests)
}
}
3 members carry the configuration:
NAME: Unique identifier for this processor, used in monitoring and logging.type Value: The data that flows through the pipeline, which keeps the pipeline type safe.process(): The core logic that turns checkpoint data into your row type.
Processor trait definition
/// Implementors of this trait are responsible for transforming checkpoint into rows for their
/// table.
#[async_trait]
pub trait Processor: Send + Sync + 'static {
/// Used to identify the pipeline in logs and metrics.
const NAME: &'static str;
/// The type of value being inserted by the handler.
type Value: Send + Sync + 'static;
/// The processing logic for turning a checkpoint into rows of the table.
///
/// All errors returned from this method are treated as transient and will be retried
/// indefinitely with exponential backoff.
///
/// If you encounter a permanent error that will never succeed on retry (e.g., invalid data
/// format, unsupported protocol version), you should panic! This stops the indexer and alerts
/// operators that manual intervention is required. Do not return permanent errors as they will
/// cause infinite retries and block the pipeline.
///
/// For transient errors (e.g., network issues, rate limiting), simply return the error and
/// let the framework retry automatically.
async fn process(&self, checkpoint: &Arc<Checkpoint>) -> anyhow::Result<Vec<Self::Value>>;
}
Implement the Handler trait
Handler defines how your indexer commits data to your store. Append its dependencies to the imports you added previously:
use diesel_async::RunQueryDsl;
use sui_indexer_alt_framework::{
pipeline::sequential::Handler,
postgres::{Connection, Db},
};
Then implement the trait:
#[async_trait::async_trait]
impl Handler for TransactionDigestHandler {
type Store = Db;
type Batch = Vec<Self::Value>;
fn batch(&self, batch: &mut Self::Batch, values: std::vec::IntoIter<Self::Value>) {
batch.extend(values);
}
async fn commit<'a>(&self, batch: &Self::Batch, conn: &mut Connection<'a>) -> Result<usize> {
let inserted = diesel::insert_into(transaction_digests)
.values(batch)
.on_conflict(tx_digest)
.do_nothing()
.execute(conn)
.await?;
Ok(inserted)
}
}
Sequential batching moves through 3 stages:
process()returns values for each checkpoint.batch()accumulates values across multiple checkpoints.commit()writes the batch once the framework reaches its limits (H::MAX_BATCH_CHECKPOINTS).
while batch_checkpoints < max_batch_checkpoints {
let Some(entry) = pending.first_entry() else {
break;
};
match next_checkpoint.cmp(entry.key()) {
// Next pending checkpoint is from the future.
Ordering::Less => break,
// This is the next checkpoint -- include it.
Ordering::Equal => {
let indexed = entry.remove();
batch_rows += indexed.len();
batch_checkpoints += 1;
handler.batch(&mut batch, indexed.values.into_iter());
watermark = Some(indexed.watermark);
next_checkpoint += 1;
}
// Next pending checkpoint is in the past, ignore it to avoid double
// writes.
Ordering::Greater => {
metrics
.total_watermarks_out_of_order
.with_label_values(&[H::NAME])
.inc();
let indexed = entry.remove();
pending_rows -= indexed.len();
}
}
}
Override the default batch limits by implementing constants in your Handler.
Complete handlers.rs file
handlers.rs fileuse anyhow::Result;
use std::sync::Arc;
use sui_indexer_alt_framework::pipeline::Processor;
use sui_indexer_alt_framework::types::full_checkpoint_content::Checkpoint;
use crate::models::StoredTransactionDigest;
use crate::schema::transaction_digests::dsl::*;
use diesel_async::RunQueryDsl;
use sui_indexer_alt_framework::{
pipeline::sequential::Handler,
postgres::{Connection, Db},
};
pub struct TransactionDigestHandler;
#[async_trait::async_trait]
impl Processor for TransactionDigestHandler {
const NAME: &'static str = "transaction_digest_handler";
type Value = StoredTransactionDigest;
async fn process(&self, checkpoint: &Arc<Checkpoint>) -> Result<Vec<Self::Value>> {
let checkpoint_seq = checkpoint.summary.sequence_number as i64;
let digests = checkpoint
.transactions
.iter()
.map(|tx| StoredTransactionDigest {
tx_digest: tx.transaction.digest().to_string(),
checkpoint_sequence_number: checkpoint_seq,
})
.collect();
Ok(digests)
}
}
#[async_trait::async_trait]
impl Handler for TransactionDigestHandler {
type Store = Db;
type Batch = Vec<Self::Value>;
fn batch(&self, batch: &mut Self::Batch, values: std::vec::IntoIter<Self::Value>) {
batch.extend(values);
}
async fn commit<'a>(&self, batch: &Self::Batch, conn: &mut Connection<'a>) -> Result<usize> {
let inserted = diesel::insert_into(transaction_digests)
.values(batch)
.on_conflict(tx_digest)
.do_nothing()
.execute(conn)
.await?;
Ok(inserted)
}
}
Handler trait definition
/// Handlers implement the logic for a given indexing pipeline: How to process checkpoint data (by
/// implementing [Processor]) into rows for their table, how to combine multiple rows into a single
/// DB operation, and then how to write those rows atomically to the database.
///
/// The handler is also responsible for tuning the various parameters of the pipeline (provided as
/// associated values).
///
/// Sequential handlers can only be used in sequential pipelines, where checkpoint data is
/// processed out-of-order, but then gathered and written in order. If multiple checkpoints are
/// available, the pipeline will attempt to combine their writes taking advantage of batching to
/// avoid emitting redundant writes.
///
/// Back-pressure is handled by the bounded subscriber channel from the ingestion service, the
/// same as concurrent pipelines: the channel blocks broadcaster sends when full, and the adaptive
/// ingestion controller cuts fetch concurrency as the channel fills up.
#[async_trait]
pub trait Handler: Processor {
type Store: SequentialStore;
/// If at least this many rows are pending, the committer will commit them eagerly.
const MIN_EAGER_ROWS: usize = 50;
/// Soft cap: once this many rows are pending, the collector stops eagerly draining
/// its input channel and yields to the flush phase. Receive is never hard-gated — unlike
/// concurrent pipelines, a missing predecessor may be buried in the input channel, and
/// blocking receive would risk deadlock. The cap only bounds receive-to-flush latency in
/// the happy path.
const MAX_PENDING_ROWS: usize = 5000;
/// Maximum number of checkpoints to try and write in a single batch. The larger this number
/// is, the more chances the pipeline has to merge redundant writes, but the longer each write
/// transaction is likely to be.
const MAX_BATCH_CHECKPOINTS: usize = 5 * 60;
/// A type to combine multiple `Self::Value`-s into. This can be used to avoid redundant writes
/// by combining multiple rows into one (e.g. if one row supersedes another, the latter can be
/// omitted).
type Batch: Default + Send + Sync + 'static;
/// Add `values` from processing a checkpoint to the current `batch`. Checkpoints are
/// guaranteed to be presented to the batch in checkpoint order. The handler takes ownership
/// of the iterator and consumes all values.
///
/// Returns `BatchStatus::Ready` if the batch is full and should be committed,
/// or `BatchStatus::Pending` if the batch can accept more values.
///
/// Note: The handler can signal batch readiness via `BatchStatus::Ready`, but the framework
/// may also decide to commit a batch based on the trait parameters above.
fn batch(&self, batch: &mut Self::Batch, values: std::vec::IntoIter<Self::Value>);
/// Take a batch of values and commit them to the database, returning the number of rows
/// affected.
async fn commit<'a>(
&self,
batch: &Self::Batch,
conn: &mut <Self::Store as Store>::Connection<'a>,
) -> anyhow::Result<usize>;
}
Register the pipeline
Build an IndexerCluster and register your pipeline on it. The example project does this in main.rs, but the same calls work anywhere in your service that owns an async runtime:
mod handlers;
mod models;
use handlers::TransactionDigestHandler;
pub mod schema;
use anyhow::{Result, bail};
use clap::Parser;
use diesel_migrations::{EmbeddedMigrations, embed_migrations};
use sui_indexer_alt_framework::{
cluster::{Args, IndexerCluster},
pipeline::sequential::SequentialConfig,
service::Error,
};
use tokio;
use url::Url;
// Embed database migrations into the binary so they run automatically on startup
const MIGRATIONS: EmbeddedMigrations = embed_migrations!("migrations");
#[tokio::main]
async fn main() -> Result<()> {
// Load .env data
dotenvy::dotenv().ok();
// Local database URL created in step 3 above
let database_url = std::env::var("DATABASE_URL")
.expect("DATABASE_URL must be set in the environment")
.parse::<Url>()
.expect("Invalid database URL");
// Parse command-line arguments (checkpoint range, URLs, performance settings)
let args = Args::parse();
// Build and configure the indexer cluster
let mut cluster = IndexerCluster::builder()
.with_args(args) // Apply command-line configuration
.with_database_url(database_url) // Set up database URL
.with_migrations(&MIGRATIONS) // Enable automatic schema migrations
.build()
.await?;
// Register our custom sequential pipeline with the cluster
cluster
.sequential_pipeline(
TransactionDigestHandler, // Our processor/handler implementation
SequentialConfig::default(), // Use default batch sizes and checkpoint lag
)
.await?;
// Start the indexer and wait for completion
match cluster.run().await?.main().await {
Ok(()) | Err(Error::Terminated) => Ok(()),
Err(Error::Aborted) => {
bail!("Indexer aborted due to an unexpected error")
}
Err(Error::Task(e)) => {
bail!(e)
}
}
}
The key components:
embed_migrations!: Bundles your migration files into the binary so the indexer updates the database schema on startup.Args::parse(): Supplies command-line configuration such as--first-checkpointand--remote-store-url.IndexerCluster::builder(): Sets up database connections, checkpoint streaming, and monitoring.sequential_pipeline(): Registers a sequential pipeline that processes checkpoints in order with batching.SequentialConfig::default(): Uses framework defaults for batch sizes and checkpoint lag.cluster.run(): Starts processing checkpoints and blocks until completion.
The example reads DATABASE_URL from the environment and passes it to with_database_url. The builder takes the URL as an argument, so your service can source it however it already handles configuration.
Choose a checkpoint source
Sui serves checkpoint data from several sources, and production indexers use different ones for backfill and steady state. See Checkpoint Data Sources for the full comparison.
- GCS buckets (production backfill and recovery):
gs://mysten-testnet-checkpoints-use4for Testnet orgs://mysten-mainnet-checkpoints-use4for Mainnet. These buckets have full checkpoint retention and are Requester Pays enabled, so configure GCS credentials and a billing project before using them. - Fullnode gRPC (production steady state): Use a production-grade endpoint from an RPC provider for polling and streaming after the historical backfill reaches the network tip.
- Public-good fullnode gRPC (testing only):
https://fullnode.testnet.sui.io:443,https://fullnode.mainnet.sui.io:443, andhttps://fullnode.devnet.sui.io:443. Do not use these endpoints for production ingestion. - Public HTTPS checkpoint stores (testing only, most recent 30 days):
https://checkpoints.testnet.sui.iofor Testnet orhttps://checkpoints.mainnet.sui.iofor Mainnet. Do not use these public-good endpoints for production ingestion.
Before reading from a Requester Pays GCS bucket, configure credentials and a billing project, then pass the billing project in the x-goog-user-project header:
$ export GOOGLE_APPLICATION_CREDENTIALS=/path/to/gcp-credentials.json
$ export GOOGLE_CLOUD_PROJECT=your-billing-project
Phase 1, backfill from the full-retention GCS checkpoint bucket:
$ cargo run -- \
--remote-store-gcs mysten-testnet-checkpoints-use4 \
--remote-store-header "x-goog-user-project:${GOOGLE_CLOUD_PROJECT}"
After the indexer reaches the network tip, stop it and restart it against fullnode gRPC. Existing pipelines resume from their committed watermarks, so no data is lost across the switch.
Phase 2, fullnode gRPC polling and streaming for steady state:
$ export FULLNODE_GRPC_URL=https://your-rpc-provider.example:443
$ cargo run -- \
--rpc-api-url "$FULLNODE_GRPC_URL" \
--streaming-url "$FULLNODE_GRPC_URL"
For testing only, you can point at the public HTTPS checkpoint store instead:
$ cargo run -- \
--remote-store-url https://checkpoints.testnet.sui.io \
--first-checkpoint RECENT_CHECKPOINT
The public HTTPS checkpoint stores retain only the most recent 30 days of checkpoints. When you use one with a fresh database, set --first-checkpoint to a checkpoint still inside the retained range. Otherwise the indexer starts at checkpoint 0 and retries missing checkpoints indefinitely. For production backfills, use --remote-store-gcs. See Use Requester Pays and Running a Remote Store for more information.
Allow incoming network requests if your operating system prompts for them when the indexer starts.
Verify the pipeline
Connect to your database and confirm rows are landing:
SELECT COUNT(*) FROM transaction_digests;
SELECT * FROM transaction_digests LIMIT 5;
To confirm the data is accurate, copy any transaction digest from your database and look it up on SuiScan.
The shape stays the same for any custom indexer: define your row type, implement Processor and Handler, register the pipeline, and let the framework handle ingestion and recovery.