# Processors (Carbon 2.0) > The Processor trait, the input each pipe passes to it, keeping state with &mut self, sharing state through Arc and channels, and what happens on errors. A processor is your code: Carbon calls it with each decoded update that reaches its pipe. This page covers the trait, what each kind of pipe passes in, and how to keep and share state. The [processor inputs reference](https://carbon.sevenlabs.org/reference/processor-inputs.md) lists every field. ## The `Processor` trait ```rust pub trait Processor where T: Sync, { fn process(&mut self, data: &T) -> impl Future> + Send; } ``` Implement it with a plain `async fn`; you don't need `#[async_trait]`: ```rust impl Processor for MyProcessor { async fn process(&mut self, deletion: &AccountDeletion) -> CarbonResult<()> { println!("closed {}", deletion.pubkey); Ok(()) } } ``` The builder requires processors to be `Send + Sync + 'static`, and the future `process` returns must be `Send`. In practice: don't hold a `std::sync::MutexGuard`, an `Rc` or another non-`Send` value across an `.await` inside `process`. The pipeline awaits each `process` call before it moves on, and never calls two processors at once (see [ordering guarantees](https://carbon.sevenlabs.org/concepts/architecture.md#ordering-and-concurrency-guarantees)). A slow `process` slows the whole pipeline. ## What each pipe passes in The input type depends on the builder method you register the processor with. One type can implement `Processor` for several inputs, and you register an instance for each. | Builder method | Your processor implements | | ---------------------------------- | ------------------------------------------------------------------------------------------------- | | `.account(decoder, processor)` | `Processor>`, where `T` is the decoder's `AccountType` | | `.instruction(decoder, processor)` | `Processor>`, where `T` is the decoder's `InstructionType` | | `.transaction::(processor)` | `Processor>`, where `C` is an `InstructionDecoderCollection` | | `.account_deletions(processor)` | `Processor` | | `.block_details(processor)` | `Processor` | The `*_with_filters` variants take the same processors. ### Account pipes ```rust pub struct AccountProcessorInputType<'a, T> { pub metadata: &'a AccountMetadata, // slot, pubkey, transaction_signature pub decoded_account: &'a DecodedAccount, // data: T, plus lamports, owner, executable, rent_epoch pub raw_account: &'a solana_account::Account, } ``` The processor runs once per account update the decoder recognizes. `raw_account` gives you the original bytes if you need fields the decoder doesn't expose. ### Instruction pipes ```rust pub struct InstructionProcessorInputType<'a, T> { pub metadata: &'a InstructionMetadata, pub decoded_instruction: &'a T, pub nested_instructions: &'a NestedInstructions, pub raw_instruction: &'a solana_instruction::Instruction, } ``` The processor runs once per decoded instruction, including instructions called through CPI. `metadata` holds the instruction's position (`stack_height`, `index`, `absolute_path`) and an `Arc` with the transaction's signature, slot, fee payer, message and execution meta. `nested_instructions` are the instructions this one called. See [Instructions](https://carbon.sevenlabs.org/concepts/instructions.md). ### Transaction pipes ```rust pub struct TransactionProcessorInputType<'a, T> { pub metadata: &'a Arc, pub instructions: &'a [(InstructionMetadata, T)], } ``` The processor runs once per transaction that passes the pipe's filters, with every instruction its collection decoded, top-level and nested, in execution order. It runs even when nothing decoded, with an empty slice. See [Transactions](https://carbon.sevenlabs.org/concepts/transactions.md). ### Account deletion and block details pipes These processors receive the update itself: an [`AccountDeletion`](https://carbon.sevenlabs.org/concepts/updates.md#account-deletions) (pubkey, final account, slot, transaction signature) or a [`BlockDetails`](https://carbon.sevenlabs.org/concepts/updates.md#block-details). There's no decoder, so an `account_deletions` processor sees closures for every program the datasource covers; check `deletion.account.owner` if you only care about one. ## Keep state in the processor `process` takes `&mut self`, and each registered processor is owned by its pipe. Keep counters, caches and batches in plain fields: nothing else can access them while `process` runs, so they need no locks. Each registration owns a separate instance. If you register the same processor type twice, each copy has its own state. ## Share state with the rest of your program To read a processor's state from elsewhere, or to share it between processors, put it behind an `Arc`: - **`Arc>` or atomics** for small shared values. A `std::sync::Mutex` is fine if you never `.await` while holding the guard; use `tokio::sync::Mutex` if you must. - **A channel to a task you own** for I/O. The processor sends a message and returns; a separate task writes to the database or calls an API. This keeps slow I/O out of the pipeline loop. With a bounded channel, a slow writer still slows the pipeline down instead of using unbounded memory. This processor forwards Meteora DLMM swaps to a writer task. It goes with a pipeline that registers `.instruction(MeteoraDlmmDecoder, Forward { tx })`: ```rust pub struct SwapRow { pub signature: String, pub amount_in: u64, } pub struct Forward { tx: tokio::sync::mpsc::Sender, } impl Processor> for Forward { async fn process( &mut self, input: &InstructionProcessorInputType<'_, MeteoraDlmmInstruction>, ) -> CarbonResult<()> { if let MeteoraDlmmInstruction::Swap { data, .. } = input.decoded_instruction { let row = SwapRow { signature: input.metadata.transaction_metadata.signature.to_string(), amount_in: data.amount_in, }; self.tx .send(row) .await .map_err(|_| CarbonError::Custom("writer task stopped".to_string()))?; } Ok(()) } } ``` `CarbonError` is `carbon_core::error::Error`. Once work leaves the processor, the pipeline no longer sees its errors or its order: the writer task has to handle both. ## Errors If `process` returns `Err`, the pipeline logs the error with the update, counts the update in `carbon_updates_failed_total`, skips the remaining pipes for that update and continues with the next one. It doesn't retry and doesn't stop. Retry inside `process` if the operation can succeed on a second attempt. See [Error handling](https://carbon.sevenlabs.org/concepts/error-handling.md). ## Example This processor counts updates in a private field and adds lamports per owner to a shared map, which `main` reads after the pipeline finishes. Three hand-built account updates stand in for a datasource. ```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 ``` `src/main.rs`: ```rust use { carbon_core::{ account::{AccountDecoder, AccountProcessorInputType, DecodedAccount}, datasource::AccountUpdate, error::CarbonResult, pipeline::Pipeline, processor::Processor, }, carbon_stream_message_datasource::{StreamMessageClient, UnifiedMessage}, solana_account::Account, solana_pubkey::Pubkey, std::{ collections::HashMap, sync::{Arc, Mutex}, }, }; struct AnyAccount; impl<'a> AccountDecoder<'a> for AnyAccount { type AccountType = (); fn decode_account(&self, account: &'a Account) -> Option> { Some(DecodedAccount { lamports: account.lamports, data: (), owner: account.owner, executable: account.executable, rent_epoch: account.rent_epoch, }) } } /// Keeps private state in `self` and publishes totals to shared state. struct LamportsByOwner { seen: u64, totals: Arc>>, } impl Processor> for LamportsByOwner { async fn process(&mut self, input: &AccountProcessorInputType<'_, ()>) -> CarbonResult<()> { self.seen += 1; // `&mut self`: no lock needed for state only this processor uses let mut totals = self.totals.lock().unwrap(); // never hold a std lock across `.await` *totals.entry(input.decoded_account.owner).or_default() += input.decoded_account.lamports; println!("update #{} at slot {}", self.seen, input.metadata.slot); Ok(()) } } #[tokio::main] async fn main() -> CarbonResult<()> { let owner_a = Pubkey::new_from_array([1; 32]); let owner_b = Pubkey::new_from_array([2; 32]); let (tx, rx) = tokio::sync::mpsc::channel(16); for (slot, owner, lamports) in [(1, owner_a, 10), (2, owner_b, 5), (3, owner_a, 7)] { let update = AccountUpdate { pubkey: Pubkey::new_unique(), account: Account { lamports, owner, ..Account::default() }, slot, transaction_signature: None, }; tx.send(UnifiedMessage::Account(update)).await.unwrap(); } drop(tx); let totals = Arc::new(Mutex::new(HashMap::new())); Pipeline::builder() .datasource(StreamMessageClient::new(rx)) .account(AnyAccount, LamportsByOwner { seen: 0, totals: totals.clone() }) .build()? .run() .await?; let totals = totals.lock().unwrap(); println!("owner A: {} lamports, owner B: {} lamports", totals[&owner_a], totals[&owner_b]); Ok(()) } ``` ```sh cargo run ``` Output: ```text update #1 at slot 1 update #2 at slot 2 update #3 at slot 3 owner A: 17 lamports, owner B: 5 lamports ``` ## Next steps - [Instructions](https://carbon.sevenlabs.org/concepts/instructions.md): Instruction metadata, nested instructions and events. - [Transactions](https://carbon.sevenlabs.org/concepts/transactions.md): Process every instruction of a transaction together. - [Persist to Postgres](https://carbon.sevenlabs.org/guides/postgres.md): Write decoded data to typed tables. --- Carbon 2.0.0 documentation · Web: https://carbon.sevenlabs.org/concepts/processors · Last updated 2026-11-09 · Index of all pages: https://carbon.sevenlabs.org/llms.txt