# Run Carbon in production (Carbon 2.0) > Choose datasources, handle reconnects and gaps, size the channel, stop cleanly on SIGTERM, write idempotently, and monitor a Carbon 2.0 indexer 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 | Area | Do this | Why | | ------------- | ----------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------- | | Datasource | Stream with Yellowstone or LaserStream; backfill with an RPC crawler | Streams are fast but can drop data; crawlers are complete but slow | | Redundancy | Two streams from different providers plus `DeduplicationFilter` on every instruction and account pipe | A single upstream is a single point of failure | | Gaps | Pass a `disconnect_notifier` and backfill the reported slots | Reconnects don't replay missed slots unless you enable replay | | Backpressure | Set `channel_buffer_size` from your peak rate; alert on `carbon_updates_queued` | Streaming datasources drop updates when the channel is full | | Shutdown | Handle SIGINT and SIGTERM yourself and cancel the token | The pipeline only handles SIGINT, and can miss it while busy | | Exit | Exit non-zero when `run()` returns without a shutdown request | `run()` returns `Ok` when all datasources have stopped | | Commitment | `Confirmed` or `Finalized` | Carbon 2.0 has no rollback events for `Processed` data | | Writes | Upsert on a natural key; skip failed transactions | Reconnects, backfills and redundant sources deliver the same data twice | | TLS | Install a rustls `CryptoProvider` first thing in `main` | gRPC datasources panic on their first TLS connection otherwise | | Build | `cargo build --release` | Debug builds are many times slower | | Observability | `RUST_LOG=info`, Prometheus endpoint, alerts | Processor errors are logged and skipped, not fatal | ## 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. `Cargo.toml`: ```toml [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" ``` `src/main.rs`: ```rust 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::(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> 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](https://carbon.sevenlabs.org/guides/stream-with-yellowstone.md). Without a reachable endpoint the datasource fails at startup and the program exits with an error, which is what a supervisor needs to see: ```sh GEYSER_URLS=http://127.0.0.1:1 RUST_LOG=info ./target/release/prod-indexer ``` Output: ```text [… 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 | Datasource | Latency | Completeness | Use for | | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------- | ------------------------------------- | | [Yellowstone gRPC](https://carbon.sevenlabs.org/datasources/yellowstone-grpc.md), [Helius LaserStream](https://carbon.sevenlabs.org/datasources/helius-laserstream.md) | Lowest: pushed as the validator processes them | Misses whatever arrives while disconnected; LaserStream replays from the last slot by default | Live indexing | | [RPC program subscribe](https://carbon.sevenlabs.org/datasources/rpc-program-subscribe.md), [RPC block subscribe](https://carbon.sevenlabs.org/datasources/rpc-block-subscribe.md), [Helius Atlas WS](https://carbon.sevenlabs.org/datasources/helius-atlas-ws.md) | Low, over websockets | Websocket drops lose data; block subscribe isn't available on many RPC nodes, including Solana's public endpoint | Live indexing without a gRPC provider | | [RPC transaction crawler](https://carbon.sevenlabs.org/datasources/rpc-transaction-crawler.md), [Helius GTFA](https://carbon.sevenlabs.org/datasources/helius-gtfa.md) | Polling interval plus one request per transaction | Complete for one address, from `getSignaturesForAddress` history | Backfills, low-volume programs | | [RPC block crawler](https://carbon.sevenlabs.org/datasources/rpc-block-crawler.md) | One `getBlock` per slot | Complete for a slot range | Filling known gaps | | [RPC GPA](https://carbon.sevenlabs.org/datasources/rpc-gpa.md), [Helius GPA v2](https://carbon.sevenlabs.org/datasources/helius-gpa-v2.md), [validator snapshot](https://carbon.sevenlabs.org/datasources/validator-snapshot.md) | One-off | Current state of every account | Bootstrapping account state | 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](https://carbon.sevenlabs.org/guides/backfill-transactions.md) and [Use multiple datasources](https://carbon.sevenlabs.org/guides/multiple-datasources.md). ## 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](https://carbon.sevenlabs.org/concepts/filters.md). - 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: | Datasource | Reconnect | Reports the gap | Gives up | | -------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------- | | Yellowstone gRPC | Resubscribes with backoff from 100 ms to 3 s; a stream silent for `stream_timeout` (30 s default) counts as dropped | `DatasourceDisconnection` on `disconnect_notifier` with the last slot before and first slot after | Never, once connected. A failed **first** connection ends the datasource. | | Yellowstone with `with_reconnect(ReconnectConfig)` | The gRPC client reconnects and replays missed data inside the stream, within its slot retention | No | Per the client's backoff | | Helius LaserStream | Resubscribes every 3 s; with `replay_enabled` (default) it resumes from the last slot it saw (31 slots earlier at `Processed`) | No | After 10 consecutive failures | | RPC block subscribe | Reconnects every 3 s | `DatasourceDisconnection` | After 10 consecutive failures | | RPC program subscribe | Reconnects every 3 s, also when the pipeline channel is full | No | After 10 consecutive failures | `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`](https://carbon.sevenlabs.org/datasources/rpc-block-crawler.md) 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: ```rust 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`](https://carbon.sevenlabs.org/reference/pipeline-builder.md#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](https://carbon.sevenlabs.org/concepts/shutdown.md). 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](https://carbon.sevenlabs.org/guides/postgres.md#two-swaps-in-one-transaction-but-only-one-row). - **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: Output: ```text 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: ```rust 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 ```sh 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`: ```sh 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: | Message | Meaning | | ------------------------------------------------------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `error consuming datasource: …` | A datasource stopped with an error | | `error processing update (…): …` | A decoder conversion or processor returned `Err`; the update was skipped for the remaining pipes. The message includes the full update with `Debug` formatting, so it can be long | | `Failed to send transaction update …`, `Failed to send account event …` | Yellowstone dropped an update because the channel was full | | `Stream timeout - no messages for …`, `Reconnected. Slots: … -> … (missed: …)` | Yellowstone reconnect | | `Max reconnection attempts reached: 10` | LaserStream gave up | | `Prometheus metrics server error: …` | The metrics endpoint didn't start | 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`](https://carbon.sevenlabs.org/guides/prometheus.md) 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](https://carbon.sevenlabs.org/guides/prometheus.md#alert-rules) 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 - [Monitor with Prometheus](https://carbon.sevenlabs.org/guides/prometheus.md): The endpoint, queries and alert rules. - [Shutdown](https://carbon.sevenlabs.org/concepts/shutdown.md): Cancellation tokens and shutdown strategies in detail. - [Use multiple datasources](https://carbon.sevenlabs.org/guides/multiple-datasources.md): Combine streams and crawlers in one pipeline. - [Troubleshooting](https://carbon.sevenlabs.org/troubleshooting.md): Symptoms and fixes for common failures. --- Carbon 2.0.0 documentation · Web: https://carbon.sevenlabs.org/guides/production · Last updated 2026-11-09 · Index of all pages: https://carbon.sevenlabs.org/llms.txt