Transactions

An instruction processor sees one instruction at a time. A transaction processor sees a whole transaction: its metadata and every instruction in it, decoded across several programs at once. This page explains transaction pipes and the decoder collections they need.

When to use a transaction processor#

Both can run in the same pipeline. For each transaction, every instruction pipe runs before any transaction pipe (see Architecture).

Transaction pipes#

Register a transaction processor with the collection type it decodes with:

Pipeline::builder()
    .datasource(datasource)
    .transaction::<AllInstructions, _>(SwapSummary)

The processor implements Processor<TransactionProcessorInputType<'_, AllInstructions>> and receives:

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

instructions holds every instruction the collection decoded, top-level and nested, in execution order, each with its InstructionMetadata. Instructions the collection doesn't recognize are left out. The processor runs once for every transaction that passes the pipe's filters, including transactions where nothing decoded: check for an empty slice if that matters to you.

transaction_with_filters takes filters too. Their filter_transaction method receives the transaction metadata and the full NestedInstructions tree before anything is decoded.

Transaction metadata#

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>,
}

fee_payer is the first account key of the message. meta has the execution results: status, fee, pre_balances and post_balances, pre_token_balances and post_token_balances, log_messages, inner_instructions, loaded_addresses and compute_units_consumed. index, block_time and block_hash are set only when the datasource provides them (see Updates). Instruction processors reach the same struct through input.metadata.transaction_metadata.

Decoder collections#

A collection is an enum with one variant per program, which knows how to decode an instruction from any of them. It implements InstructionDecoderCollection:

pub trait InstructionDecoderCollection:
    Clone + std::fmt::Debug + Send + Sync + Eq + std::hash::Hash + Serialize + 'static
{
    type InstructionType: Clone + std::fmt::Debug + PartialEq + Eq + Send + Sync + 'static;

    fn parse_instruction(instruction: &solana_instruction::Instruction) -> Option<Self>;

    fn get_type(&self) -> Self::InstructionType;
}

• parse_instruction returns the decoded instruction wrapped in the right variant, or None if no program matches.
• get_type returns a lightweight tag, such as AllInstructionTypes::Dlmm(DlmmType::Swap), which you can compare without matching on the payload.

The transaction pipe calls carbon_core::transaction::parse_instructions_flat::<T>(instructions), which runs parse_instruction over the transaction's flat instruction list and keeps the matches with their metadata. It's public, so you can call it yourself in tests.

The instruction_decoder_collection! macro#

Writing the enum and the trait by hand is repetitive, so carbon_core provides a macro (with the default macros feature):

instruction_decoder_collection!(
    AllInstructions, AllInstructionTypes, AllPrograms,
    Dlmm => carbon_meteora_dlmm_decoder::PROGRAM_ID => DlmmDecoder => Dlmm,
    Token => carbon_token_program_decoder::PROGRAM_ID => TokenDecoder => Token,
);

The first three names are the enums it generates: the collection (AllInstructions), its type tags (AllInstructionTypes) and a plain list of programs (AllPrograms). Each following entry is Variant => PROGRAM_ID => decoder expression => instruction type. parse_instruction matches instruction.program_id against each PROGRAM_ID and calls that decoder. You can omit the program ID (Variant => decoder => type); those entries are tried in order, for instructions whose program ID matched no other entry.

Each instruction type you list must:

• derive Debug, Clone, PartialEq, Eq, Hash, serde::Serialize and serde::Deserialize;
• derive InstructionType (from carbon_core), which generates a sibling enum named <Type>Type with one unit variant per variant, plus a get_instruction_type() method;
• be in scope by its bare name, together with its <Type>Type enum, because the macro refers to them by their last path segment.

The code the macro generates also needs the InstructionDecoder trait in scope and serde (with derive) and solana-instruction as direct dependencies of your crate.

Example#

This program decodes Meteora DLMM and SPL Token instructions together. For every transaction with a DLMM swap, it prints the fee payer and the amounts of the token transfers in the same transaction.

cargo add carbon-core@2.0.0 carbon-meteora-dlmm-decoder@2.0.0 carbon-token-program-decoder@2.0.0
cargo add carbon-rpc-transaction-crawler-datasource@2.0.0
cargo add serde --features derive
cargo add solana-instruction@3.4.1
cargo add tokio --features full

use {
    carbon_core::{
        collection::InstructionDecoderCollection,
        error::CarbonResult,
        instruction::InstructionDecoder,
        instruction_decoder_collection,
        pipeline::Pipeline,
        processor::Processor,
        transaction::TransactionProcessorInputType,
        InstructionType,
    },
    carbon_meteora_dlmm_decoder::{instructions::MeteoraDlmmInstruction, MeteoraDlmmDecoder},
    carbon_rpc_transaction_crawler_datasource::{ConnectionConfig, Filters, RpcTransactionCrawler},
    carbon_token_program_decoder::{instructions::TokenProgramInstruction, TokenProgramDecoder},
    serde::{Deserialize, Serialize},
};

// Slim per-program enums with the derives a collection needs.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, InstructionType)]
pub enum Dlmm {
    Swap { amount_in: u64, min_amount_out: u64 },
    Other,
}

#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, InstructionType)]
pub enum Token {
    Transfer { amount: u64 },
    TransferChecked { amount: u64, decimals: u8 },
    Other,
}

pub struct DlmmDecoder;

impl<'a> InstructionDecoder<'a> for DlmmDecoder {
    type InstructionType = Dlmm;

    fn decode_instruction(&self, ix: &'a solana_instruction::Instruction) -> Option<Dlmm> {
        Some(match MeteoraDlmmDecoder.decode_instruction(ix)? {
            MeteoraDlmmInstruction::Swap { data, .. } => Dlmm::Swap {
                amount_in: data.amount_in,
                min_amount_out: data.min_amount_out,
            },
            _ => Dlmm::Other,
        })
    }
}

pub struct TokenDecoder;

impl<'a> InstructionDecoder<'a> for TokenDecoder {
    type InstructionType = Token;

    fn decode_instruction(&self, ix: &'a solana_instruction::Instruction) -> Option<Token> {
        Some(match TokenProgramDecoder.decode_instruction(ix)? {
            TokenProgramInstruction::Transfer { data, .. } => Token::Transfer { amount: data.amount },
            TokenProgramInstruction::TransferChecked { data, .. } => Token::TransferChecked {
                amount: data.amount,
                decimals: data.decimals,
            },
            _ => Token::Other,
        })
    }
}

instruction_decoder_collection!(
    AllInstructions, AllInstructionTypes, AllPrograms,
    Dlmm => carbon_meteora_dlmm_decoder::PROGRAM_ID => DlmmDecoder => Dlmm,
    Token => carbon_token_program_decoder::PROGRAM_ID => TokenDecoder => Token,
);

struct SwapSummary;

impl Processor<TransactionProcessorInputType<'_, AllInstructions>> for SwapSummary {
    async fn process(
        &mut self,
        input: &TransactionProcessorInputType<'_, AllInstructions>,
    ) -> CarbonResult<()> {
        let swaps = input
            .instructions
            .iter()
            .filter(|(_, ix)| ix.get_type() == AllInstructionTypes::Dlmm(DlmmType::Swap))
            .count();
        if swaps == 0 {
            return Ok(()); // the processor runs for every transaction, even with no matches
        }

        let transfers: Vec<u64> = input
            .instructions
            .iter()
            .filter_map(|(_, ix)| match ix {
                AllInstructions::Token(Token::Transfer { amount })
                | AllInstructions::Token(Token::TransferChecked { amount, .. }) => Some(*amount),
                _ => None,
            })
            .collect();

        println!(
            "{} fee_payer={} swaps={} token_transfers={:?}",
            input.metadata.signature, input.metadata.fee_payer, swaps, transfers
        );
        Ok(())
    }
}

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

    Pipeline::builder()
        .datasource(crawler)
        .transaction::<AllInstructions, _>(SwapSummary)
        .build()?
        .run()
        .await
}

cargo run

2HNnxMudUcYCYXcRuV2SejC4p1fcY2kiLqXfqgZeNgPZ6pZjLUkxCT4QzXJruDXp6rXeu7SHwYr2aa22j1YtmSTt fee_payer=HYe4vSaEGqQKnDrxWDrk3o5H2gznv7qtij5G6NNG8WHd swaps=1 token_transfers=[51427757, 115, 4468209]
hX9jqFeYAkVk7rnAYATuRBxsrDQ36TBgHHfPNhP1uHGjbcdt5YR9JXMJTkX4L5K7UZNAMkHUEAM2iLEHcuBtaxK fee_payer=9nXDunV8eNvYSVys7JSBkm2hz9Q79Q3EYeko19srVeMC swaps=1 token_transfers=[98400000, 345227222]
fAQWC4yoDc5QT542V8YqKf1CncpVeC2ST9Kvoi5F38vQKjgkxmdzG8g3v1rULNnPpvYpdT2Mae6BzzcGPJjC6Uw fee_payer=GoGoGo6N99mpyB7rfzhw2R4fXmaFctURXHaHMoGCyoLD swaps=1 token_transfers=[20000000, 230210607, 230210607, 20000688]
…

The crawler only fetches transactions that mention the DLMM program, but the collection decodes the token program's instructions in them too. Token-2022 transfers don't appear because the collection has no Token-2022 entry; add carbon-token-2022-decoder the same way to include them. The public endpoint is rate limited, so output can pause for a minute or more.

Next steps#