Run Carbon in production

This page collects what matters when a Carbon indexer runs unattended: where data comes from, what happens when a connection drops or the processors fall behind, how the process stops, and what to monitor. Every behaviour described here is that of Carbon 2.0.0; links point to the code.

Checklist#

A production-shaped indexer#

The rest of the page explains each part of this program. It streams Meteora DLMM transactions from one or more Yellowstone endpoints, deduplicates them, reports gaps, serves Prometheus metrics, stops on SIGTERM and exits non-zero if its datasources die.

[dependencies]
carbon-core = "2.0.0"
carbon-meteora-dlmm-decoder = "2.0.0"
carbon-prometheus-metrics = "2.0.0"
carbon-yellowstone-grpc-datasource = "2.0.0"
env_logger = "0.11.8"
log = "0.4.28"
rustls = { version = "0.23.35", default-features = false, features = ["std", "aws_lc_rs"] }
tokio = { version = "1.48.0", features = ["full"] }
tokio-util = "0.7.17"
yellowstone-grpc-proto = { version = "=12.7.0", default-features = false }

[profile.release]
codegen-units = 1
lto = "thin"

use {
    carbon_core::{
        datasource::{DatasourceDisconnection, DatasourceId},
        error::{CarbonResult, Error},
        filter::DeduplicationFilter,
        instruction::InstructionProcessorInputType,
        metrics::{Counter, MetricsRegistry},
        pipeline::{Pipeline, ShutdownStrategy},
        processor::Processor,
    },
    carbon_meteora_dlmm_decoder::{
        instructions::MeteoraDlmmInstruction, MeteoraDlmmDecoder, PROGRAM_ID,
    },
    carbon_prometheus_metrics::{PrometheusMetrics, PrometheusServerConfig},
    carbon_yellowstone_grpc_datasource::{
        BlockFilters, YellowstoneGrpcClientConfig, YellowstoneGrpcGeyserClient,
    },
    std::{collections::HashMap, env, sync::Arc, time::Duration},
    tokio::signal::unix::{signal, SignalKind},
    tokio_util::sync::CancellationToken,
    yellowstone_grpc_proto::geyser::{CommitmentLevel, SubscribeRequestFilterTransactions},
};

static MISSED_SLOTS: Counter = Counter::new(
    "indexer_missed_slots_total",
    "Slots skipped while a datasource was disconnected",
);

#[tokio::main]
async fn main() -> CarbonResult<()> {
    env_logger::init();
    rustls::crypto::aws_lc_rs::default_provider()
        .install_default()
        .map_err(|_| Error::Custom("install rustls crypto provider".into()))?;
    MetricsRegistry::global().register_counter(&MISSED_SLOTS);

    // One or more Yellowstone endpoints, ideally from different providers.
    let endpoints = env::var("GEYSER_URLS")
        .map_err(|_| Error::Custom("GEYSER_URLS not set".into()))?;
    let x_token = env::var("X_TOKEN").ok();

    // Gap reports: one message per reconnect, with the slots that were missed.
    let (disconnect_tx, mut disconnect_rx) =
        tokio::sync::mpsc::channel::<DatasourceDisconnection>(64);
    tokio::spawn(async move {
        while let Some(gap) = disconnect_rx.recv().await {
            MISSED_SLOTS.inc_by(gap.missed_slots);
            log::error!(
                "{} reconnected: slots {}..{} missed ({} slots); backfill them",
                gap.source,
                gap.last_slot_before_disconnect,
                gap.first_slot_after_reconnect,
                gap.missed_slots,
            );
        }
    });

    let mut builder = Pipeline::builder();
    for (n, endpoint) in endpoints.split(',').enumerate() {
        let transaction_filters = HashMap::from([(
            "dlmm".to_string(),
            SubscribeRequestFilterTransactions {
                vote: Some(false),
                failed: Some(false),
                account_include: vec![PROGRAM_ID.to_string()],
                ..Default::default()
            },
        )]);
        let datasource = YellowstoneGrpcGeyserClient::new(
            endpoint.trim().to_string(),
            x_token.clone(),
            Some(CommitmentLevel::Confirmed),
            HashMap::new(),
            transaction_filters,
            BlockFilters::default(),
            YellowstoneGrpcClientConfig::default(),
            Some(disconnect_tx.clone()),
            Some(Duration::from_secs(30)),
        );
        builder = builder.datasource_with_id(datasource, DatasourceId::new_named(&format!("geyser-{n}")));
    }
    drop(disconnect_tx);

    // Stop on SIGTERM (systemd, Docker, Kubernetes) and ctrl-C. Listening in a task of its
    // own means a signal that arrives while a processor is running isn't missed.
    let shutdown = CancellationToken::new();
    let on_signal = shutdown.clone();
    let mut sigterm = signal(SignalKind::terminate()).map_err(|e| Error::Custom(e.to_string()))?;
    tokio::spawn(async move {
        tokio::select! {
            _ = sigterm.recv() => log::info!("SIGTERM received, stopping"),
            _ = tokio::signal::ctrl_c() => log::info!("ctrl-C received, stopping"),
        }
        on_signal.cancel();
    });

    builder
        .channel_buffer_size(10_000)
        .metrics(Arc::new(PrometheusMetrics::with_server(
            PrometheusServerConfig::default(),
        )))
        .instruction_with_filters(
            MeteoraDlmmDecoder,
            SwapWriter,
            vec![Box::new(DeduplicationFilter::new(Duration::from_secs(120)))],
        )
        .datasource_cancellation_token(shutdown.clone())
        .shutdown_strategy(ShutdownStrategy::ProcessPending)
        .build()?
        .run()
        .await?;

    if shutdown.is_cancelled() {
        Ok(())
    } else {
        // Every datasource ended on its own: exit non-zero so the supervisor restarts us.
        Err(Error::Custom("all datasources stopped".into()))
    }
}

struct SwapWriter;

impl Processor<InstructionProcessorInputType<'_, MeteoraDlmmInstruction>> for SwapWriter {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, MeteoraDlmmInstruction>,
    ) -> CarbonResult<()> {
        if let MeteoraDlmmInstruction::Swap2 { data, accounts, .. } = input.decoded_instruction {
            let tx = &input.metadata.transaction_metadata;
            if tx.meta.status.is_err() {
                return Ok(()); // the stream filter drops failed transactions; backfills may not
            }
            // Idempotent key: the same instruction always maps to the same row.
            let key = (tx.signature, input.metadata.absolute_path.clone());
            log::info!("{key:?} pool={} amount_in={}", accounts.lb_pair, data.amount_in);
        }
        Ok(())
    }
}

You need a Yellowstone gRPC endpoint (from a provider such as Triton, Helius or QuickNode) to run it; see Stream with Yellowstone. Without a reachable endpoint the datasource fails at startup and the program exits with an error, which is what a supervisor needs to see:

GEYSER_URLS=http://127.0.0.1:1 RUST_LOG=info ./target/release/prod-indexer

[… INFO  carbon_core::pipeline] starting pipeline. num_datasources: 1, num_exporters: 1, num_account_pipes: 0, num_account_deletion_pipes: 0, num_instruction_pipes: 1, num_transaction_pipes: 0
[… INFO  carbon_prometheus_metrics] Prometheus metrics server starting on http://0.0.0.0:9464/metrics
[… ERROR carbon_core::pipeline] error consuming datasource: FailedToConsumeDatasource("gRPC transport error: transport error")
[… INFO  carbon_core::pipeline] update_receiver closed, shutting down.
[… INFO  carbon_core::pipeline] pipeline shutdown complete.
Error: Custom("all datasources stopped")

The exit status is 1.

Choose datasources for latency and completeness#

A common setup runs a stream for live data and a crawler (or block crawler) for history and gaps, writing to the same idempotent tables. See Backfill transactions and Use multiple datasources.

Run redundant streams and deduplicate#

Give each datasource a stable ID with datasource_with_id(.., DatasourceId::new_named(..)) and put a DeduplicationFilter on every pipe:

• For instructions it keys on (signature, absolute_path), for accounts on (transaction_signature, pubkey), and keeps keys for the TTL you pass (entries are purged about once a minute).
• Account updates without a transaction_signature are always accepted. Yellowstone and LaserStream set it; program subscribe and GPA datasources don't, so their duplicates pass through.
• It doesn't implement filter_transaction, so .transaction(..) pipes see every copy. Deduplicate those in your processor or with your own filter.
• Each pipe owns its own filter instance and its own key set. That's what you want: each pipe should see each instruction once.
• Pick a TTL longer than the time between two sources delivering the same transaction: 60 seconds covers two healthy live streams; raise it to a few minutes if a source can lag further, for example while it reconnects. Memory grows with the number of distinct keys in the window.

Detect gaps after reconnects#

What each streaming datasource does when its connection drops:

DatasourceDisconnection has source, disconnect_time, last_slot_before_disconnect, first_slot_after_reconnect and missed_slots (their difference). The program above counts missed slots in a Prometheus counter and logs them. To close the gap, crawl the range with RpcBlockCrawler from last_slot_before_disconnect (inclusive, because the stream may have dropped mid-slot) to first_slot_after_reconnect, through the same processors. Because writes are idempotent, the overlap is harmless.

To use client-side replay instead, add yellowstone-grpc-client = "=13.5.0" and:

use yellowstone_grpc_client::ReconnectConfig;

let config = YellowstoneGrpcClientConfig::default().with_reconnect(ReconnectConfig::default());

Replay depends on your provider keeping enough history.

Backpressure and the channel#

Datasources and the processing loop are connected by one bounded channel, channel_buffer_size long (1,000 by default). The loop takes one update at a time and runs it through every matching pipe before taking the next, so throughput is about one update per average processing time.

When processors fall behind, the channel fills, and what happens next depends on the datasource:

• Yellowstone, LaserStream, RPC block subscribe, RPC block crawler, RPC transaction crawler (default), Helius Atlas WS: try_send fails and the update is dropped, with an error such as Failed to send transaction update with signature … at slot …: … (Yellowstone).
• RPC program subscribe: the send failure (Error sending account update) ends the subscription, which reconnects after 3 s.
• RPC transaction crawler with blocking_send: true, RPC GPA, Helius GPA v2, Helius GTFA, stream message, validator snapshot, and datasources you write with send(..).await: the datasource waits. Nothing is lost, the source slows down.

So size the buffer to absorb bursts (.channel_buffer_size(10_000) above), alert when carbon_updates_queued stays high, and keep processors fast: batch database writes, or hand work to a task of your own through your own channel. Memory used by the buffer grows with its length times the size of an update; a full transaction with metadata is typically several kilobytes, so measure your process's resident memory at peak rather than guessing.

Stop cleanly#

What the 2.0.0 pipeline does on shutdown, measured with 2,000 queued updates and a 2 ms processor:

• SIGTERM isn't handled. Docker, Kubernetes and systemd stop processes with SIGTERM, which kills the process immediately unless you handle it.
• SIGINT (ctrl-C) is only noticed while the loop is waiting for the next update: the loop creates a fresh tokio::signal::ctrl_c() listener on each iteration, and a signal that arrives while no listener exists (while an update is being processed) is lost. Tokio has already replaced the default handler, so the process doesn't die either: in the test, a SIGINT sent while the backlog was being processed was ignored, and the pipeline went on to process all 2,000 updates before a second SIGINT stopped it.
• Cancelling your own token (datasource_cancellation_token) is reliable: the pipeline stops after the update in progress. Updates still in the channel are dropped; in the test, 128 of 2,000 were processed.
• ShutdownStrategy::ProcessPending, the default, is meant to drain the channel after SIGINT, but the loop exits as soon as it sees the cancelled token, so it doesn't drain. See Shutdown.

So: install your own handlers for SIGINT and SIGTERM, cancel the token from them in a task of their own, as the program above does, and design for at-least-once delivery. Updates queued at shutdown are lost, so on restart resume from a checkpoint a bit before the last slot you wrote (a stream's from_slot, a crawler's until signature or start slot) and let idempotent writes absorb the overlap.

To finish queued updates before exiting, stop the source instead of the pipeline. A datasource you write can watch its own token (not the one passed to datasource_cancellation_token), stop sending and return. Once every datasource has returned and dropped its sender, the pipeline processes what's left in the channel and run() returns. The built-in datasources only watch the pipeline's token, so this works only for your own.

run() also returns Ok(()) when every datasource has stopped on its own, for example after LaserStream gives up reconnecting. The program above checks whether shutdown was requested and otherwise returns an error, so the process exits with status 1 and your supervisor restarts it (Restart=on-failure in systemd, restartPolicy: Always in Kubernetes).

Commitment, forks and finality#

Streams let you choose a commitment level (CommitmentLevel for Yellowstone and LaserStream); the RPC datasources default to confirmed.

• Processed is fastest, but the block can still be skipped. Carbon 2.0.0 emits no event when a slot is dropped, so data from an abandoned fork stays in your database unless you reconcile it.
• Confirmed has been voted on by a supermajority. Rollbacks are very rare. This is the usual choice.
• Finalized can't be rolled back but arrives roughly 32 slots later.

If you must index at Processed, store the slot with every row and reconcile later, for example by re-checking recent slots against a Finalized source and deleting rows whose transactions aren't there.

Write idempotently#

The same data arrives more than once after reconnects, from redundant sources, from replay and from backfills. Make every write an upsert on a natural key:

• Instructions: (signature, absolute_path). absolute_path (for example [4, 1]) identifies the instruction's position in the CPI tree. Don't key on InstructionMetadata::index and stack_height alone: inner instructions share their top-level instruction's index, so two CPIs under one top-level instruction collide. The generated Postgres tables have this problem; see Store decoded data in Postgres.
• Accounts: pubkey, and only overwrite when the incoming slot is not older: ON CONFLICT (pubkey) DO UPDATE SET … WHERE excluded.slot >= accounts.slot. Updates from two sources can arrive out of order. Carbon's generated account upserts overwrite unconditionally.
• Failed transactions: the pipeline decodes instructions of failed transactions too. Check metadata.transaction_metadata.meta.status.is_err() and skip them unless you want them; some datasource filters (failed: Some(false) for Yellowstone) drop them upstream.

Install a TLS crypto provider#

The Yellowstone and LaserStream crates use rustls through tonic, and Carbon's dependency graph doesn't select a rustls crypto backend by itself. Without one, the first TLS connection panics inside the datasource task:

thread 'tokio-rt-worker' panicked at …/rustls-0.23.46/src/crypto/mod.rs:249:14:
Could not automatically determine the process-level CryptoProvider from Rustls crate features.
Call CryptoProvider::install_default() before this point to select a provider manually, or make sure exactly one of the 'aws-lc-rs' and 'ring' features is enabled.

The panic kills only that task, so the pipeline logs update_receiver closed, shutting down. and run() returns Ok(()): without the exit check below, the process exits with status 0 and looks like a clean stop. Add rustls with the aws_lc_rs feature and install the provider before building any datasource, as main above does:

rustls::crypto::aws_lc_rs::default_provider()
    .install_default()
    .map_err(|_| Error::Custom("install rustls crypto provider".into()))?;

install_default fails only if a provider is already installed, which you can ignore if another library installs the same one.

Build for release#

cargo build --release

Always deploy release builds. With the profile above, building this program from scratch took under a minute on an 8-core machine with a peak of 680 MB of memory and produced a 12.6 MB binary. lto = "thin" and codegen-units = 1 make the binary a little faster at the cost of build time; they're optional. Pin your dependencies with a committed Cargo.lock, and build on the same Rust version you test with (1.96.1 or newer).

Logs#

Carbon logs through the log crate, so the binary decides the format and level; with env_logger, set RUST_LOG:

RUST_LOG=info,carbon_log_metrics=warn ./target/release/prod-indexer

info shows startup, reconnects and shutdown. LogMetrics prints every metric every 5 seconds at info; silence it with its target as above if you also export to Prometheus, or use LogMetrics::new_with_flush_interval(60). Messages worth alerting on in your log platform:

Log as JSON for your log platform by swapping env_logger for a structured logger; Carbon's messages are plain strings, so the fields above are in the message text.

Metrics and alerts#

Serve metrics with carbon-prometheus-metrics and alert on at least: the endpoint being down, no updates received for longer than your program's quiet periods, a failure ratio above zero, carbon_updates_queued near channel_buffer_size, p99 processing time, and any increase of your own missed-slots counter. Monitor with Prometheus has rules for all of these.

Size resources#

• CPU. Processing runs on one task, so one fast core does the decoding and processor work; datasource tasks (gRPC decoding, RPC calls) run on other Tokio worker threads. More cores don't speed up the processing loop. If one core isn't enough, run several indexers, each on a subset of programs.
• Memory. Dominated by the channel buffer, DeduplicationFilter key sets and your own state. Measure resident memory at peak load and leave headroom for bursts and reconnect replays.
• Network. A Yellowstone subscription delivers only what its filters match, but a busy program's full transactions still add up to a lot of bandwidth. Filter as narrowly as you can upstream (account_include, failed: Some(false), vote: Some(false)).
• Database. Each processed instruction is one round trip with Carbon's Postgres processors, so database latency caps throughput. Keep the database in the same region, or batch.

Next steps#