Processor inputs

A processor implements Processor<T> for the input type of the pipe it is registered on. This page lists each input type and every field in it, so you know what data your process method can read. For how to write a processor, see Processors.

Every input is passed by reference and borrows data the pipeline owns for the duration of one process call. Clone what you need to keep.

Example#

This program prints the instruction input for each Meteora DLMM Swap2 the transaction crawler finds on the public RPC endpoint.

[dependencies]
carbon-core = "2.0.0"
carbon-meteora-dlmm-decoder = "2.0.0"
carbon-rpc-transaction-crawler-datasource = "2.0.0"
tokio = { version = "1", features = ["full"] }
tokio-util = "0.7"

use {
    carbon_core::{
        error::CarbonResult, instruction::InstructionProcessorInputType, pipeline::Pipeline,
        processor::Processor,
    },
    carbon_meteora_dlmm_decoder::{
        instructions::MeteoraDlmmInstruction, MeteoraDlmmDecoder, PROGRAM_ID,
    },
    carbon_rpc_transaction_crawler_datasource::{ConnectionConfig, Filters, RpcTransactionCrawler},
    std::time::Duration,
    tokio_util::sync::CancellationToken,
};

struct InputPrinter;

impl Processor<InstructionProcessorInputType<'_, MeteoraDlmmInstruction>> for InputPrinter {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, MeteoraDlmmInstruction>,
    ) -> CarbonResult<()> {
        let MeteoraDlmmInstruction::Swap2 { data, accounts, .. } = input.decoded_instruction else {
            return Ok(());
        };
        let meta = input.metadata;
        let tx = &meta.transaction_metadata;
        println!("signature:      {}", tx.signature);
        println!("slot:           {} (block_time {:?})", tx.slot, tx.block_time);
        println!("fee payer:      {}", tx.fee_payer);
        println!("stack height:   {}", meta.stack_height);
        println!("index:          {}", meta.index);
        println!("absolute path:  {:?}", meta.absolute_path);
        println!("inner ixs:      {}", input.nested_instructions.len());
        println!("raw accounts:   {}", input.raw_instruction.accounts.len());
        println!("lb_pair:        {}", accounts.lb_pair);
        println!("amount_in:      {}", data.amount_in);
        println!();
        Ok(())
    }
}

#[tokio::main]
async fn main() -> CarbonResult<()> {
    let datasource = RpcTransactionCrawler::new(
        "https://api.mainnet-beta.solana.com".to_string(),
        PROGRAM_ID,
        ConnectionConfig::default(),
        Filters::new(None, None, None),
        None,
    );

    let token = CancellationToken::new();
    let stop = token.clone();
    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_secs(90)).await;
        stop.cancel();
    });

    Pipeline::builder()
        .datasource(datasource)
        .instruction(MeteoraDlmmDecoder, InputPrinter)
        .datasource_cancellation_token(token)
        .build()?
        .run()
        .await
}

cargo run

The first swap is a top-level instruction; the second is a CPI from another program, so its stack_height is 2 and its index is the index of the top-level instruction that called it:

signature:      43eGCzcyUFYtJsuoWwmFsirBBu51tm6BRs7WiCWYRfR7txoz1DATD5iSQNFfS1LXuGJvYG2AXTA3t38N1DG9eYJp
slot:           467120934 (block_time Some(1794185094))
fee payer:      6TWMHcW3gGXRnzRoe6vTgENjJhCbTFApcA5yLEJJaBDh
stack height:   1
index:          2
absolute path:  [2]
inner ixs:      4
raw accounts:   17
lb_pair:        GeUkx21Vc6yg63YZ1BdXY95ZATm9LeBYCgSN1uJ3o18S
amount_in:      404287484

signature:      2W5fitdRrbLo2pNF4ChwDPPcmRNC3EfR1u7viGGyPXdnHZM8HBu2dFnJbVxktupvJKbXcxheZaadphndEMkCfNTi
slot:           467120933 (block_time Some(1794185094))
fee payer:      AgmLJBMDCqWynYnQiPCuj9ewsNNsBJXyzoUhD9LJzN51
stack height:   2
index:          1
absolute path:  [1, 1]
inner ixs:      4
raw accounts:   19
lb_pair:        CcG3fyDZn6uXTV5p2TV9cceCA2YhE3bUMLeXGn6vZSNr
amount_in:      680804450

AccountProcessorInputType#

What an account processor receives for each account its decoder accepts.

#[derive(Debug)]
pub struct AccountProcessorInputType<'a, T> {
    pub metadata: &'a AccountMetadata,
    pub decoded_account: &'a DecodedAccount<T>,
    pub raw_account: &'a solana_account::Account,
}

Details.

Example.

impl Processor<AccountProcessorInputType<'_, MeteoraDlmmAccount>> for PoolLogger {
    async fn process(
        &mut self,
        input: &AccountProcessorInputType<'_, MeteoraDlmmAccount>,
    ) -> CarbonResult<()> {
        if let MeteoraDlmmAccount::LbPair(pool) = &input.decoded_account.data {
            println!(
                "{} active_id={} ({} bytes, slot {})",
                input.metadata.pubkey,
                pool.active_id,
                input.raw_account.data.len(),
                input.metadata.slot
            );
        }
        Ok(())
    }
}

See also. account, docs.rs.

AccountMetadata#

Where and when an account update was observed.

#[derive(Debug, Clone)]
pub struct AccountMetadata {
    pub slot: u64,
    pub pubkey: Pubkey,
    pub transaction_signature: Option<Signature>,
}

Details. Copied from the AccountUpdate.

Example.

let key = (input.metadata.pubkey, input.metadata.slot);

See also. docs.rs.

DecodedAccount#

An account decoder's output: the typed data plus the account header.

#[derive(Debug, Clone)]
pub struct DecodedAccount<T> {
    pub lamports: u64,
    pub data: T,
    pub owner: Pubkey,
    pub executable: bool,
    pub rent_epoch: u64,
}

Details.

Generated decoders copy the four header fields from the raw account. A hand-written AccountDecoder builds this struct itself.

Example.

let decoded = &input.decoded_account;
println!("owner={} lamports={}", decoded.owner, decoded.lamports);

See also. docs.rs.

InstructionProcessorInputType#

What an instruction processor receives for each instruction its decoder accepts.

#[derive(Debug)]
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,
}

Details.

The pipe runs once per matching instruction, top-level or inner. If a program calls your program through CPI, you receive that inner instruction with stack_height > 1.

Example.

for child in input.nested_instructions.iter() {
    println!("CPI to {}", child.instruction.program_id);
}

See also. instruction, Instructions, docs.rs.

InstructionMetadata#

Where an instruction sits in its transaction.

#[derive(Debug, Clone)]
pub struct InstructionMetadata {
    pub transaction_metadata: Arc<TransactionMetadata>,
    pub stack_height: u32,
    pub index: u32,
    pub absolute_path: Vec<u8>,
}

Details.

Use (transaction_metadata.signature, absolute_path) as a unique key for an instruction; (signature, index, stack_height) is not unique when one instruction makes several CPIs at the same depth.

Carbon reads inner instructions from meta.inner_instructions and their depth from each inner instruction's stack_height. When the source reports a missing or out-of-range stack_height (outside 2 to 5), Carbon logs invalid inner instruction stack height (…), dropping the remaining inner instructions of this group at warn level and skips the rest of that group.

Example.

let key = (
    input.metadata.transaction_metadata.signature,
    input.metadata.absolute_path.clone(),
);

See also. docs.rs, source.

InstructionMetadata::decode_log_events#

Decodes the program log events emitted by this instruction.

pub fn decode_log_events<T: CarbonDeserialize>(&self) -> Vec<T>

Details. Walks transaction_metadata.meta.log_messages, tracks Program … invoke [n] and Program … success / failed lines to follow the call stack, and keeps the log lines emitted while this instruction (matched by absolute_path) was executing. For each such line, it base64-decodes the last whitespace-separated token and calls T::deserialize; lines that fail to decode or deserialize are skipped. This targets Anchor emit! events, logged as Program data: <base64>. Logs emitted by this instruction's own CPIs are not included.

Precompile instructions (Ed25519, Secp256k1, Secp256r1) log nothing, so Carbon subtracts the number of precompiles before this instruction from the top-level index before matching. Returns an empty Vec when the transaction has no log messages. If the runtime truncated the logs (Log truncated), events after the cut are missing.

Generated decoders in 2.0.0 do not implement CarbonDeserialize for their event types; they expose an inherent decode method instead, and Anchor events emitted through self-CPI (emit_cpi!) arrive as a decoded CpiEvent instruction. Implement CarbonDeserialize yourself to use this method; see CarbonDeserialize.

Example.

for event in input.metadata.decode_log_events::<MyEvent>() {
    println!("{} moved {}", event.pool, event.amount);
}

See also. Instructions, docs.rs.

NestedInstruction#

One node in a transaction's instruction tree.

#[derive(Debug, Clone)]
pub struct NestedInstruction {
    pub metadata: InstructionMetadata,
    pub instruction: solana_instruction::Instruction,
    pub inner_instructions: NestedInstructions,
}

Details.

Example.

fn count(instructions: &NestedInstructions) -> usize {
    instructions
        .iter()
        .map(|ix| 1 + count(&ix.inner_instructions))
        .sum()
}

See also. docs.rs.

NestedInstructions#

An ordered list of NestedInstructions.

#[derive(Debug, Default)]
pub struct NestedInstructions(pub Vec<NestedInstruction>);

impl NestedInstructions {
    pub fn len(&self) -> usize;
    pub fn is_empty(&self) -> bool;
    pub fn push(&mut self, nested_instruction: NestedInstruction);
}

Details. Derefs to [NestedInstruction], so slice methods such as iter(), first() and indexing work. It also implements Clone, owned IntoIterator, and From<InstructionsWithMetadata>, which builds the tree from a flat Vec<(InstructionMetadata, Instruction)> ordered top-level-first as the runtime reports it. Filters receive the whole transaction's tree in filter_transaction; instruction processors receive one subtree in nested_instructions.

From<InstructionsWithMetadata> asserts that every stack_height is between 1 and MAX_INSTRUCTION_STACK_DEPTH (5) and panics otherwise. The pipeline only builds trees from validated data, but if you call it yourself, check heights first.

Example.

if let Some(first) = input.nested_instructions.first() {
    println!("first CPI: {:?}", first.metadata.absolute_path);
}

See also. docs.rs.

TransactionProcessorInputType#

What a transaction processor receives, once per transaction.

#[derive(Debug)]
pub struct TransactionProcessorInputType<'a, T> {
    pub metadata: &'a Arc<TransactionMetadata>,
    pub instructions: &'a [(InstructionMetadata, T)],
}

Details.

The processor runs for every transaction the datasource sends, including transactions where instructions is empty. There is no nested tree here; use InstructionMetadata::stack_height and absolute_path to rebuild parent-child relations.

Example.

impl Processor<TransactionProcessorInputType<'_, AllInstructions>> for TxLogger {
    async fn process(
        &mut self,
        input: &TransactionProcessorInputType<'_, AllInstructions>,
    ) -> CarbonResult<()> {
        if input.instructions.is_empty() {
            return Ok(());
        }
        println!(
            "{}: {} known instructions",
            input.metadata.signature,
            input.instructions.len()
        );
        Ok(())
    }
}

See also. transaction, Transactions, docs.rs.

TransactionMetadata#

Data shared by every instruction of one transaction.

#[derive(Debug, Clone, Default)]
pub struct TransactionMetadata {
    pub slot: u64,
    pub signature: Signature,
    pub fee_payer: Pubkey,
    pub meta: solana_transaction_status::TransactionStatusMeta,
    pub message: solana_message::VersionedMessage,
    pub index: Option<u64>,
    pub block_time: Option<i64>,
    pub block_hash: Option<Hash>,
}

Details. Built from the TransactionUpdate with TryFrom, which fails with MissingFeePayer when the message has no account keys.

Check meta.status.is_err() to detect failed transactions. Whether you receive failed transactions depends on the datasource: the RPC transaction crawler, RPC block crawler and RPC block subscribe drop them before they reach the pipeline; Yellowstone and LaserStream follow your subscription's failed filter; Helius GTFA follows its status filter.

Example.

let tx = &input.metadata.transaction_metadata;
if tx.meta.status.is_err() {
    return Ok(());
}
println!("{} paid {} lamports", tx.fee_payer, tx.meta.fee);

See also. docs.rs.

MAX_INSTRUCTION_STACK_DEPTH#

The deepest instruction nesting Carbon accepts.

pub const MAX_INSTRUCTION_STACK_DEPTH: usize = 5;

Details. Matches the Agave runtime's limit of a top-level instruction plus four levels of CPI. Inner instructions reported deeper than this are dropped with a warning.

Example.

assert!(input.metadata.stack_height as usize <= MAX_INSTRUCTION_STACK_DEPTH);

See also. docs.rs.

Deletions and block details#

Account-deletion and block-details processors receive the update itself, with no metadata wrapper and no decoder:

impl Processor<AccountDeletion> for DeletionLogger {
    async fn process(&mut self, deletion: &AccountDeletion) -> CarbonResult<()> {
        println!("closed {} (owner {})", deletion.pubkey, deletion.account.owner);
        Ok(())
    }
}

impl Processor<BlockDetails> for BlockLogger {
    async fn process(&mut self, block: &BlockDetails) -> CarbonResult<()> {
        println!("block {} at {:?}", block.slot, block.block_time);
        Ok(())
    }
}

See AccountDeletion and BlockDetails for their fields.