Error handling

Carbon uses one error type across the framework, and the pipeline treats errors differently depending on where they come from. This page explains both, and shows how to add context, retry and stop on failure. Each error variant is described in the errors reference.

CarbonError and CarbonResult#

Carbon's error type is carbon_core::error::Error, usually imported as CarbonError, and CarbonResult<T> is Result<T, Error>. Processors, datasources, exporters and the builder all return CarbonResult.

pub enum Error {
    MissingUpdateTypeInDatasource(UpdateType),
    FailedToReceiveUpdates(String),
    MissingFeePayer,
    MissingInnerInstructions,
    MissingAccountInTransaction,
    MissingInstructionData,
    FailedToConsumeDatasource(String),
    Custom(String),
}

In 2.0.0 Carbon itself produces only a few of these:

• MissingFeePayer when a transaction has no account keys;
• FailedToConsumeDatasource from datasources that can't start, such as a failed getProgramAccounts call or an invalid gRPC endpoint;
• Custom from datasources and the Postgres layer, wrapping errors from other crates;
• MissingAccountInTransaction from the public transformers::extract_account_metas helper.

Use Custom(String) for your own errors. Error implements std::error::Error and Display through thiserror, so ? works in a main that returns CarbonResult<()>.

What the pipeline does with errors#

Nothing is retried, and in particular a failed update isn't re-delivered. "Skips the remaining pipes" means pipes registered after the failing one don't see that update. For a transaction, the failing instruction pipe also stops at that instruction, so later instructions, later instruction pipes and all transaction pipes are skipped for that transaction. See Architecture.

Network errors inside datasources usually never reach the pipeline. Datasources retry, back off and reconnect on their own, logging warnings as they go; the RPC transaction crawler, for example, retries each request three times with exponential backoff by default. Check each datasource's page for its retry settings.

Patterns#

Add context to errors#

The pipeline's log line includes the whole update in Debug form, which for a transaction is very long, and your error message is at the end. Put the identifiers you need in the error itself:

let label = std::str::from_utf8(&input.decoded_account.data).map_err(|e| {
    CarbonError::Custom(format!(
        "account {} at slot {}: {e}",
        input.metadata.pubkey, input.metadata.slot
    ))
})?;

For instruction processors, include input.metadata.transaction_metadata.signature and input.metadata.absolute_path. If you'd rather keep the log short, log the error yourself with that context and return Ok(()); the update then counts as successful in the metrics.

Retry transient failures#

The pipeline won't retry for you, so retry inside process when an operation can succeed on a later attempt, such as a database write during a failover. This helper retries an async operation three times with a doubling delay:

pub async fn with_retries<T, E, F, Fut>(mut operation: F) -> CarbonResult<T>
where
    E: std::fmt::Display,
    F: FnMut() -> Fut,
    Fut: std::future::Future<Output = Result<T, E>>,
{
    let mut delay = std::time::Duration::from_millis(200);
    let mut attempt = 1;
    loop {
        match operation().await {
            Ok(value) => return Ok(value),
            Err(e) if attempt < 3 => {
                log::warn!("attempt {attempt} failed, retrying in {delay:?}: {e}");
                tokio::time::sleep(delay).await;
                delay *= 2;
                attempt += 1;
            }
            Err(e) => return Err(CarbonError::Custom(format!("gave up after {attempt} attempts: {e}"))),
        }
    }
}

While process retries, the pipeline waits: no other update is processed. Keep retry budgets short, or hand the work to a task you own (see Processors).

Stop the pipeline on an error#

By default a pipeline keeps going after errors. When a failure means the indexed data is no longer trustworthy, stop instead: give the processor a clone of the pipeline's cancellation token and cancel it.

pub struct StopOnError {
    shutdown: tokio_util::sync::CancellationToken,
}

impl Processor<InstructionProcessorInputType<'_, MeteoraDlmmInstruction>> for StopOnError {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, MeteoraDlmmInstruction>,
    ) -> CarbonResult<()> {
        let signature = input.metadata.transaction_metadata.signature;
        let result = with_retries(|| async { Ok::<_, std::io::Error>(()) }).await;
        if let Err(error) = result {
            log::error!("tx {signature} path {:?}: {error}", input.metadata.absolute_path);
            self.shutdown.cancel(); // the pipeline stops after this update
            return Err(error);
        }
        Ok(())
    }
}

Replace the placeholder operation with your own write. After the pipeline stops, run returns Ok(()); record the failure somewhere your main can check if the process should exit with an error.

Watch the failure rate#

carbon_updates_failed_total counts failed updates, and LogMetrics prints the failure percentage on every flush. Alert on it in production; see Metrics.

Example#

This pipeline parses each account's data as a UTF-8 label. The second account's data isn't valid UTF-8, so ParseLabel returns an error with context. The pipeline logs it, skips the Audit pipe for that update and carries on.

cargo add carbon-core@2.0.0 carbon-stream-message-datasource@2.0.0
cargo add solana-account@4.3.2 solana-pubkey@4.2.1
cargo add tokio --features full
cargo add env_logger

use {
    carbon_core::{
        account::{AccountDecoder, AccountProcessorInputType, DecodedAccount},
        datasource::AccountUpdate,
        error::{CarbonResult, Error as CarbonError},
        pipeline::Pipeline,
        processor::Processor,
    },
    carbon_stream_message_datasource::{StreamMessageClient, UnifiedMessage},
    solana_account::Account,
    solana_pubkey::Pubkey,
};

struct RawData;

impl<'a> AccountDecoder<'a> for RawData {
    type AccountType = Vec<u8>;

    fn decode_account(&self, account: &'a Account) -> Option<DecodedAccount<Vec<u8>>> {
        Some(DecodedAccount {
            lamports: account.lamports,
            data: account.data.clone(),
            owner: account.owner,
            executable: account.executable,
            rent_epoch: account.rent_epoch,
        })
    }
}

/// Parses the account data as UTF-8 and fails on anything else.
struct ParseLabel;

impl Processor<AccountProcessorInputType<'_, Vec<u8>>> for ParseLabel {
    async fn process(&mut self, input: &AccountProcessorInputType<'_, Vec<u8>>) -> CarbonResult<()> {
        let label = std::str::from_utf8(&input.decoded_account.data).map_err(|e| {
            CarbonError::Custom(format!(
                "account {} at slot {}: {e}",
                input.metadata.pubkey, input.metadata.slot
            ))
        })?;
        println!("slot {}: label {label:?}", input.metadata.slot);
        Ok(())
    }
}

struct Audit;

impl Processor<AccountProcessorInputType<'_, Vec<u8>>> for Audit {
    async fn process(&mut self, input: &AccountProcessorInputType<'_, Vec<u8>>) -> CarbonResult<()> {
        println!("slot {}: audited", input.metadata.slot);
        Ok(())
    }
}

#[tokio::main]
async fn main() -> CarbonResult<()> {
    env_logger::init();

    let (tx, rx) = tokio::sync::mpsc::channel(16);
    for (slot, data) in [(1, b"alpha".to_vec()), (2, vec![0xff, 0xfe]), (3, b"gamma".to_vec())] {
        let update = AccountUpdate {
            pubkey: Pubkey::new_from_array([slot as u8; 32]),
            account: Account { lamports: 1, data, ..Account::default() },
            slot,
            transaction_signature: None,
        };
        tx.send(UnifiedMessage::Account(update)).await.unwrap();
    }
    drop(tx);

    Pipeline::builder()
        .datasource(StreamMessageClient::new(rx))
        .account(RawData, ParseLabel) // runs first
        .account(RawData, Audit) // skipped for an update when ParseLabel fails
        .build()?
        .run()
        .await
}

RUST_LOG=error cargo run

slot 1: label "alpha"
slot 1: audited
[2026-11-09T00:08:56Z ERROR carbon_core::pipeline] error processing update (Account(AccountUpdate { pubkey: 8qbHbw2BbbTHBW1sbeqakYXVKRQM8Ne7pLK7m6CVfeR, account: Account { lamports: 1, data.len: 2, owner: 11111111111111111111111111111111, executable: false, rent_epoch: 0, data: fffe }, slot: 2, transaction_signature: None })): Custom("account 8qbHbw2BbbTHBW1sbeqakYXVKRQM8Ne7pLK7m6CVfeR at slot 2: invalid utf-8 sequence of 1 bytes from index 0")
slot 3: label "gamma"
slot 3: audited

Next steps#