Index multiple programs

One pipeline can decode any number of programs. Register one pipe per decoder when each program's instructions stand on their own, or one transaction pipe with a decoder collection when you need to see several programs' instructions together. Either way, the datasource must deliver every program's transactions. This guide runs both approaches on the public endpoint with the Meteora DLMM and Pump.fun AMM decoders.

You need: Rust 1.96.1 or newer. The examples read two mainnet blocks from https://api.mainnet-beta.solana.com; no account is needed.

One pipe per decoder#

Each .instruction(decoder, processor) call adds a pipe. For every transaction, the pipeline runs each pipe over every instruction, top-level and nested. A decoder returns None for other programs' instructions, so each processor only sees its own program, with the decoder's full types.

cargo new dex-indexer
cd dex-indexer

[package]
name = "dex-indexer"
version = "0.1.0"
edition = "2024"

[dependencies]
carbon-core = "2.0.0"
carbon-meteora-dlmm-decoder = "2.0.0"
carbon-pump-swap-decoder = "2.0.0"
carbon-rpc-block-crawler-datasource = "2.0.0"
env_logger = "0.11.11"
log = "0.4.35"
solana-client = "=4.2.2"
tokio = { version = "1.54.1", features = ["full"] }

use {
    carbon_core::{
        error::CarbonResult, instruction::InstructionProcessorInputType, pipeline::Pipeline,
        processor::Processor,
    },
    carbon_meteora_dlmm_decoder::{instructions::MeteoraDlmmInstruction, MeteoraDlmmDecoder},
    carbon_pump_swap_decoder::{instructions::PumpSwapInstruction, PumpSwapDecoder},
    carbon_rpc_block_crawler_datasource::{RpcBlockConfig, RpcBlockCrawler},
    solana_client::rpc_config::{CommitmentConfig, TransactionDetails, UiTransactionEncoding},
};

#[tokio::main]
async fn main() -> CarbonResult<()> {
    env_logger::init();

    // Every transaction in two blocks, whichever programs it calls.
    let datasource = RpcBlockCrawler::new(
        "https://api.mainnet-beta.solana.com".to_string(),
        467_110_307,
        Some(467_110_308),
        None,
        RpcBlockConfig {
            encoding: Some(UiTransactionEncoding::Base64),
            transaction_details: Some(TransactionDetails::Full),
            rewards: Some(false),
            commitment: Some(CommitmentConfig::finalized()),
            max_supported_transaction_version: Some(1),
        },
        Some(1),
        None,
    );

    Pipeline::builder()
        .datasource(datasource)
        .instruction(MeteoraDlmmDecoder, DlmmLogger)
        .instruction(PumpSwapDecoder, PumpSwapLogger)
        .channel_buffer_size(10_000)
        .build()?
        .run()
        .await
}

struct DlmmLogger;

impl Processor<InstructionProcessorInputType<'_, MeteoraDlmmInstruction>> for DlmmLogger {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, MeteoraDlmmInstruction>,
    ) -> CarbonResult<()> {
        let signature = input.metadata.transaction_metadata.signature;
        match input.decoded_instruction {
            MeteoraDlmmInstruction::Swap { data, .. } => {
                println!("dlmm swap       {signature}: amount_in={}", data.amount_in)
            }
            MeteoraDlmmInstruction::Swap2 { data, .. } => {
                println!("dlmm swap2      {signature}: amount_in={}", data.amount_in)
            }
            _ => {}
        }
        Ok(())
    }
}

struct PumpSwapLogger;

impl Processor<InstructionProcessorInputType<'_, PumpSwapInstruction>> for PumpSwapLogger {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, PumpSwapInstruction>,
    ) -> CarbonResult<()> {
        let signature = input.metadata.transaction_metadata.signature;
        match input.decoded_instruction {
            PumpSwapInstruction::Buy { data, .. } => {
                println!("pump_swap buy   {signature}: base_amount_out={}", data.base_amount_out)
            }
            PumpSwapInstruction::Sell { data, .. } => {
                println!("pump_swap sell  {signature}: base_amount_in={}", data.base_amount_in)
            }
            _ => {}
        }
        Ok(())
    }
}

cargo run

pump_swap sell  5UzcVzyLK7uM1ttjjcN98zLV1s9b7xJ3e572bthzEorbiLCbcEXzDPNaPXnkv8AV1zw6Pbfs7cfddFaov54e3oaH: base_amount_in=28426915529
pump_swap sell  4wwzVEoKvEnqL98QXEEE9ftRUgYnLCDHLt5WiYiYdG9xzvYFFKPNEq9pHCvYURw7vsSHhYa5vBLvc8cU66oYJujT: base_amount_in=14949087
pump_swap sell  4e4qr8vA7t39w5TCMygkGk4DHC6RWo1BnwBgiQ6BfhaqDoxHuTqcNSmkWe8dszozPFeBrVbHyRapfAd9juR49S1P: base_amount_in=3680531
dlmm swap       ue82HW7rk3DcWiCTzRrnMzMeKDYDLWUiyeMBtMi4tLtqiKaRGgbr7EYigU9RbpftsFsgEC7P6kgxkNWJgZhvKZs: amount_in=10298583
pump_swap sell  36N7ynNwzHfkaa2gKNPRdj9AnFyLdx8U1qudgC2UKUziR4uL8sCRYkoK4jM8B9q3DoE8ebN8AaqmHiMPR6Xjiqn8: base_amount_in=208155812
…
dlmm swap2      2d6GNQzSzxgV85mnzD4ed6JyxdLYdwBtyCH8LujdYaxySSKfD8x96yJLtxmuVYt6XGRjy8m3AutxyxqiumscEb4W: amount_in=381756
pump_swap buy   2i5pWH6S4KZP6Q232XiMWPxS7MXwJ9JHqB38TvLtmJPkusy7AziWhbrVZXFEBF9pcdVFUScAQx5V5diptWaQc9H8: base_amount_out=1110999784
…

Within one transaction, the pipes run in the order you registered them: all of the DLMM pipe's instructions, then all of the Pump.fun AMM pipe's. Pipes don't share state; if two processors need the same data, share it with an Arc<Mutex<…>> or write to the same database.

Accounts work the same way: add .account(MeteoraDlmmDecoder, …) and .account(PumpSwapDecoder, …). Each generated account decoder checks the account's owner, so it ignores the other program's accounts.

One transaction processor for several programs#

A transaction processor sees a whole transaction at once: its metadata and every instruction that any of your programs' decoders understands, in execution order. Use it when one row depends on several programs, such as an aggregator route through two DEXes, or when you classify transactions by what they contain. See Transactions for how transaction pipes work.

A transaction pipe needs a type that implements InstructionDecoderCollection: an enum with one variant per program. In 2.0.0, the instruction_decoder_collection! macro can't take generated decoders' instruction enums directly, because they don't derive Eq, Hash or InstructionType. You can either wrap each decoder in a slim enum and use the macro, as shown in Transactions, or implement the trait by hand over the full generated types, as here.

Add serde and solana-instruction, and turn on the decoders' serde feature (the trait requires Serialize):

[dependencies]
carbon-core = "2.0.0"
carbon-meteora-dlmm-decoder = { version = "2.0.0", features = ["serde"] }
carbon-pump-swap-decoder = { version = "2.0.0", features = ["serde"] }
carbon-rpc-block-crawler-datasource = "2.0.0"
env_logger = "0.11.11"
log = "0.4.35"
serde = { version = "1.0.228", features = ["derive"] }
solana-client = "=4.2.2"
solana-instruction = "3.4.1"
tokio = { version = "1.54.1", features = ["full"] }

use {
    carbon_core::{
        collection::InstructionDecoderCollection,
        error::CarbonResult,
        instruction::InstructionDecoder,
        pipeline::Pipeline,
        processor::Processor,
        transaction::TransactionProcessorInputType,
    },
    carbon_meteora_dlmm_decoder::{instructions::MeteoraDlmmInstruction, MeteoraDlmmDecoder},
    carbon_pump_swap_decoder::{instructions::PumpSwapInstruction, PumpSwapDecoder},
    carbon_rpc_block_crawler_datasource::{RpcBlockConfig, RpcBlockCrawler},
    solana_client::rpc_config::{CommitmentConfig, TransactionDetails, UiTransactionEncoding},
    std::hash::{Hash, Hasher},
};

/// One enum over every program the transaction processor cares about.
#[derive(Debug, Clone, PartialEq, serde::Serialize)]
pub enum AllInstructions {
    MeteoraDlmm(MeteoraDlmmInstruction),
    PumpSwap(PumpSwapInstruction),
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AllPrograms {
    MeteoraDlmm,
    PumpSwap,
}

// `InstructionDecoderCollection` requires `Eq + Hash`; the decoded
// instruction types only implement `PartialEq`, so implement both by hand.
impl Eq for AllInstructions {}

impl Hash for AllInstructions {
    fn hash<H: Hasher>(&self, state: &mut H) {
        std::mem::discriminant(self).hash(state);
    }
}

impl InstructionDecoderCollection for AllInstructions {
    type InstructionType = AllPrograms;

    fn parse_instruction(instruction: &solana_instruction::Instruction) -> Option<Self> {
        match instruction.program_id {
            carbon_meteora_dlmm_decoder::PROGRAM_ID => MeteoraDlmmDecoder
                .decode_instruction(instruction)
                .map(AllInstructions::MeteoraDlmm),
            carbon_pump_swap_decoder::PROGRAM_ID => PumpSwapDecoder
                .decode_instruction(instruction)
                .map(AllInstructions::PumpSwap),
            _ => None,
        }
    }

    fn get_type(&self) -> Self::InstructionType {
        match self {
            AllInstructions::MeteoraDlmm(_) => AllPrograms::MeteoraDlmm,
            AllInstructions::PumpSwap(_) => AllPrograms::PumpSwap,
        }
    }
}

#[tokio::main]
async fn main() -> CarbonResult<()> {
    env_logger::init();

    let datasource = RpcBlockCrawler::new(
        "https://api.mainnet-beta.solana.com".to_string(),
        467_110_307,
        Some(467_110_308),
        None,
        RpcBlockConfig {
            encoding: Some(UiTransactionEncoding::Base64),
            transaction_details: Some(TransactionDetails::Full),
            rewards: Some(false),
            commitment: Some(CommitmentConfig::finalized()),
            max_supported_transaction_version: Some(1),
        },
        Some(1),
        None,
    );

    Pipeline::builder()
        .datasource(datasource)
        .transaction::<AllInstructions, _>(RouteLogger)
        .channel_buffer_size(10_000)
        .build()?
        .run()
        .await
}

struct RouteLogger;

impl Processor<TransactionProcessorInputType<'_, AllInstructions>> for RouteLogger {
    async fn process(
        &mut self,
        input: &TransactionProcessorInputType<'_, AllInstructions>,
    ) -> CarbonResult<()> {
        let programs: Vec<AllPrograms> =
            input.instructions.iter().map(|(_, ix)| ix.get_type()).collect();

        if !programs.is_empty() {
            println!("{}: {programs:?}", input.metadata.signature);
        }
        Ok(())
    }
}

cargo run

Each transaction with at least one decoded instruction prints once, with every decoded instruction's program. A DLMM swap shows three entries, the swap and the two events Meteora emits through self-CPI:

5UzcVzyLK7uM1ttjjcN98zLV1s9b7xJ3e572bthzEorbiLCbcEXzDPNaPXnkv8AV1zw6Pbfs7cfddFaov54e3oaH: [PumpSwap, PumpSwap]
4wwzVEoKvEnqL98QXEEE9ftRUgYnLCDHLt5WiYiYdG9xzvYFFKPNEq9pHCvYURw7vsSHhYa5vBLvc8cU66oYJujT: [PumpSwap, PumpSwap]
4e4qr8vA7t39w5TCMygkGk4DHC6RWo1BnwBgiQ6BfhaqDoxHuTqcNSmkWe8dszozPFeBrVbHyRapfAd9juR49S1P: [PumpSwap, PumpSwap]
ue82HW7rk3DcWiCTzRrnMzMeKDYDLWUiyeMBtMi4tLtqiKaRGgbr7EYigU9RbpftsFsgEC7P6kgxkNWJgZhvKZs: [MeteoraDlmm, MeteoraDlmm, MeteoraDlmm]
36N7ynNwzHfkaa2gKNPRdj9AnFyLdx8U1qudgC2UKUziR4uL8sCRYkoK4jM8B9q3DoE8ebN8AaqmHiMPR6Xjiqn8: [PumpSwap, PumpSwap]
…

To work with the payloads, match on them: AllInstructions::MeteoraDlmm(MeteoraDlmmInstruction::Swap2 { data, .. }). Each entry of input.instructions also carries its InstructionMetadata, including its position in the instruction tree.

The transaction processor runs for every transaction the datasource delivers, including those where nothing decoded, so check for an empty list. It can run next to instruction pipes in the same pipeline; for each transaction, the instruction pipes run first.

Which approach to use#

Choose a datasource filter#

Every datasource must deliver the transactions of all your programs. How you widen its filter depends on the datasource:

Pipeline::builder()
    .datasource(RpcTransactionCrawler::new(rpc_url.clone(), carbon_meteora_dlmm_decoder::PROGRAM_ID, ConnectionConfig::default(), Filters::new(None, None, None), None))
    .datasource(RpcTransactionCrawler::new(rpc_url, carbon_pump_swap_decoder::PROGRAM_ID, ConnectionConfig::default(), Filters::new(None, None, None), None))
    .instruction_with_filters(MeteoraDlmmDecoder, DlmmLogger, vec![Box::new(DeduplicationFilter::new(Duration::from_secs(300)))])
    .instruction_with_filters(PumpSwapDecoder, PumpSwapLogger, vec![Box::new(DeduplicationFilter::new(Duration::from_secs(300)))])

SubscribeRequestFilterTransactions {
    vote: Some(false),
    failed: Some(false),
    account_include: vec![
        carbon_meteora_dlmm_decoder::PROGRAM_ID.to_string(),
        carbon_pump_swap_decoder::PROGRAM_ID.to_string(),
    ],
    ..Default::default()
}

SubscribeRequestFilterTransactions {
    vote: Some(false),
    failed: Some(false),
    account_include: vec![
        carbon_meteora_dlmm_decoder::PROGRAM_ID.to_string(),
        carbon_pump_swap_decoder::PROGRAM_ID.to_string(),
    ],
    ..Default::default()
}

Transaction pipes aren't deduplicated, so when two datasources can deliver the same transaction, prefer a single datasource that covers every program, or deduplicate by signature in your processor.

Common problems#

One program's processor never prints#

The datasource doesn't deliver that program's transactions: a Yellowstone filter with account_required instead of account_include, a crawler for only one address, or a MentionsAccountOrProgram filter for the other program. Widen the filter as shown above.

Instructions are processed twice#

Two datasources delivered the same transaction. Add a DeduplicationFilter to each instruction pipe, or use one datasource for all programs.

the trait bound `MeteoraDlmmInstruction: Eq` is not satisfied (or Hash, or cannot find type `MeteoraDlmmInstructionType` )#

You passed a generated instruction enum to instruction_decoder_collection!. Implement InstructionDecoderCollection by hand as above, or wrap the decoder as in Transactions.

the trait bound `AllInstructions: serde::Serialize` is not satisfied#

The collection must implement Serialize. Turn on each decoder's serde feature and derive serde::Serialize, with serde as a direct dependency.

The transaction processor prints nothing for transactions you expect#

parse_instruction returned None for all their instructions. Check the match on instruction.program_id lists each program's PROGRAM_ID, and that the datasource delivers those transactions.

Next steps#