Processors

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 lists every field.

The Processor trait#

pub trait Processor<T>
where
    T: Sync,
{
    fn process(&mut self, data: &T) -> impl Future<Output = CarbonResult<()>> + Send;
}

Implement it with a plain async fn; you don't need #[async_trait]:

impl Processor<AccountDeletion> 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). 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.

The *_with_filters variants take the same processors.

Account pipes#

pub struct AccountProcessorInputType<'a, T> {
    pub metadata: &'a AccountMetadata,        // slot, pubkey, transaction_signature
    pub decoded_account: &'a DecodedAccount<T>, // 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#

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<TransactionMetadata> with the transaction's signature, slot, fee payer, message and execution meta. nested_instructions are the instructions this one called. See Instructions.

Transaction pipes#

pub struct TransactionProcessorInputType<'a, T> {
    pub metadata: &'a Arc<TransactionMetadata>,
    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.

Account deletion and block details pipes#

These processors receive the update itself: an AccountDeletion (pubkey, final account, slot, transaction signature) or a BlockDetails. 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<Mutex<T>> 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 }):

pub struct SwapRow {
    pub signature: String,
    pub amount_in: u64,
}

pub struct Forward {
    tx: tokio::sync::mpsc::Sender<SwapRow>,
}

impl Processor<InstructionProcessorInputType<'_, MeteoraDlmmInstruction>> 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.

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.

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

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<DecodedAccount<()>> {
        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<Mutex<HashMap<Pubkey, u64>>>,
}

impl Processor<AccountProcessorInputType<'_, ()>> 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(())
}

cargo run

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#