# Error handling (Carbon 2.0) > CarbonError and CarbonResult, what the pipeline does with processor, datasource and exporter errors, and patterns for context, retries and failing fast. 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](https://carbon.sevenlabs.org/reference/errors.md). ## `CarbonError` and `CarbonResult` Carbon's error type is `carbon_core::error::Error`, usually imported as `CarbonError`, and `CarbonResult` is `Result`. Processors, datasources, exporters and the builder all return `CarbonResult`. ```rust 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 | Where the error happens | What happens | | ------------------------------------------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | A processor returns `Err` | The pipeline logs `error processing update ({update:?}): {error:?}` at `error` level, increments `carbon_updates_failed_total` and skips the remaining pipes for that update. It then continues with the next update. | | Converting a transaction fails (`MissingFeePayer`) | Same as a processor error, before any pipe runs. | | A datasource's `consume` returns `Err` | The pipeline logs `error consuming datasource: {error:?}`. That datasource stops; the others continue. If it was the last one, the pipeline stops. | | An exporter's `initialize` returns `Err` | `run` returns the error before any datasource starts. | | An exporter's `export` or `shutdown` returns `Err` during shutdown | `run` returns the error. | | A processor panics | The panic isn't caught. It unwinds out of `run` and, in a `#[tokio::main]` program, ends the process. | 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](https://carbon.sevenlabs.org/concepts/architecture.md#errors). 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: ```rust 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: ```rust pub async fn with_retries(mut operation: F) -> CarbonResult where E: std::fmt::Display, F: FnMut() -> Fut, Fut: std::future::Future>, { 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](https://carbon.sevenlabs.org/concepts/processors.md#share-state-with-the-rest-of-your-program)). ### 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](https://carbon.sevenlabs.org/concepts/shutdown.md#stop-with-a-cancellation-token) and cancel it. ```rust pub struct StopOnError { shutdown: tokio_util::sync::CancellationToken, } impl Processor> 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](https://carbon.sevenlabs.org/concepts/metrics.md). ## 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. ```sh 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 ``` `src/main.rs`: ```rust 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; fn decode_account(&self, account: &'a Account) -> Option>> { 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>> for ParseLabel { async fn process(&mut self, input: &AccountProcessorInputType<'_, Vec>) -> 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>> for Audit { async fn process(&mut self, input: &AccountProcessorInputType<'_, Vec>) -> 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 } ``` ```sh RUST_LOG=error cargo run ``` Output: ```text 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 - [Errors reference](https://carbon.sevenlabs.org/reference/errors.md): Every error variant: message, cause and fix. - [Troubleshooting](https://carbon.sevenlabs.org/troubleshooting.md): Fixes for common problems, by symptom. - [Shutdown](https://carbon.sevenlabs.org/concepts/shutdown.md): Stop a pipeline cleanly, on a signal or on failure. --- Carbon 2.0.0 documentation · Web: https://carbon.sevenlabs.org/concepts/error-handling · Last updated 2026-11-09 · Index of all pages: https://carbon.sevenlabs.org/llms.txt