RPC block crawler

RpcBlockCrawler calls getBlock for every slot from start_slot onwards and emits each successful transaction in the block. With an end_slot it stops after that slot; without one it follows the chain tip. Use it to backfill a slot range for any number of programs at once, or to follow the chain from a plain HTTP RPC.

Quick example#

This program crawls five blocks starting at START_SLOT, prints the Orca Whirlpool swapV2 and swap instructions in them, and exits.

cargo new block-crawler && cd block-crawler
cargo add carbon-core@2.0.0 carbon-log-metrics@2.0.0 carbon-orca-whirlpool-decoder@2.0.0 carbon-rpc-block-crawler-datasource@2.0.0
cargo add solana-client@=4.2.2
cargo add tokio@1 --features full
cargo add env_logger@0.11 log@0.4

use {
    carbon_core::{
        error::CarbonResult, instruction::InstructionProcessorInputType, pipeline::Pipeline,
        processor::Processor,
    },
    carbon_log_metrics::LogMetrics,
    carbon_orca_whirlpool_decoder::{instructions::OrcaWhirlpoolInstruction, OrcaWhirlpoolDecoder},
    carbon_rpc_block_crawler_datasource::{RpcBlockConfig, RpcBlockCrawler},
    solana_client::rpc_config::{TransactionDetails, UiTransactionEncoding},
    std::{env, sync::Arc},
};

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

    let start_slot: u64 = env::var("START_SLOT").expect("START_SLOT").parse().unwrap();

    let block_config = RpcBlockConfig {
        encoding: Some(UiTransactionEncoding::Base64), // required: Carbon can't decode JSON payloads
        transaction_details: Some(TransactionDetails::Full),
        rewards: Some(false),
        commitment: None, // defaults to confirmed
        max_supported_transaction_version: Some(1),
    };

    let datasource = RpcBlockCrawler::new(
        env::var("RPC_URL").unwrap_or("https://api.mainnet-beta.solana.com".to_string()),
        start_slot,
        Some(start_slot + 4), // end_slot is inclusive: 5 blocks, then the pipeline exits
        None,                 // block_interval: 100 ms
        block_config,
        Some(2), // max_concurrent_requests
        None,    // channel_buffer_size: 1000
    );

    Pipeline::builder()
        .datasource(datasource)
        .metrics(Arc::new(LogMetrics::new()))
        .instruction(OrcaWhirlpoolDecoder, SwapLogger)
        .build()?
        .run()
        .await
}

struct SwapLogger;

impl Processor<InstructionProcessorInputType<'_, OrcaWhirlpoolInstruction>> for SwapLogger {
    async fn process(
        &mut self,
        input: &InstructionProcessorInputType<'_, OrcaWhirlpoolInstruction>,
    ) -> CarbonResult<()> {
        let tx = &input.metadata.transaction_metadata;
        match input.decoded_instruction {
            OrcaWhirlpoolInstruction::Swap { data, accounts, .. } => {
                println!("slot {} tx #{:?} swap pool={} amount={}", tx.slot, tx.index, accounts.whirlpool, data.amount);
            }
            OrcaWhirlpoolInstruction::SwapV2 { data, accounts, .. } => {
                println!("slot {} tx #{:?} swapV2 pool={} amount={}", tx.slot, tx.index, accounts.whirlpool, data.amount);
            }
            _ => {}
        }
        Ok(())
    }
}

Pick a recent finalized slot, then run:

START_SLOT=467116060 RUST_LOG=info cargo run

slot 467116060 tx #Some(188) swapV2 pool=FAqh648xeeaTqL7du49sztp9nfj5PjRQrfvaMccyd9cz amount=14868973271
slot 467116060 tx #Some(193) swapV2 pool=B5EwJVDuAauzUEEdwvbuXzbFFgEYnUqqS37TUM1c4PQA amount=1813589446
slot 467116060 tx #Some(199) swapV2 pool=3ndjN1nJVUKGrJBc1hhVpER6kWTZKHdyDrPyCJyX3CXK amount=3112157408
...
[INFO  carbon_core::pipeline] update_receiver closed, shutting down.
[INFO  carbon_log_metrics]   block_crawler_blocks_fetched_total: 5
[INFO  carbon_log_metrics]   block_crawler_blocks_skipped_total: 0
[INFO  carbon_log_metrics]   block_crawler_transactions_processed_total: 1347
[INFO  carbon_core::pipeline] pipeline shutdown complete.

The five blocks held 1,347 successful transactions (votes included) and 63 Orca swaps; the run took about 6 seconds. The public endpoint only keeps recent blocks, so very old slots need an archival RPC.

Configuration#

RpcBlockCrawler::new#

pub fn new(
    rpc_url: String,
    start_slot: u64,
    end_slot: Option<u64>,
    block_interval: Option<Duration>,
    block_config: RpcBlockConfig,
    max_concurrent_requests: Option<usize>,
    channel_buffer_size: Option<usize>,
) -> Self

RpcBlockConfig#

Re-exported from solana_client::rpc_config. TransactionDetails and UiTransactionEncoding are in solana_client::rpc_config too.

Behaviour#

• Fetching. Slots are requested in increasing order, up to max_concurrent_requests at a time, and processed in the order the responses arrive. Blocks can therefore be emitted out of order; transactions inside a block keep their order.
• Following the tip. Without end_slot, the crawler calls getSlot whenever it reaches the last known slot and waits block_interval when there is nothing new.
• Skipped slots. Errors -32004 (block not available), -32007 and -32009 (slot skipped or missing) are counted in block_crawler_blocks_skipped_total and the slot is not retried. Other errors are logged as Error fetching block at slot … and the slot is also not retried. When following the tip, a block requested before the node has it is lost; use confirmed or finalized commitment and expect occasional gaps.
• Updates. One Transaction update per successful transaction, with index set to its position in the block, the block's block_time and block_hash. Failed transactions are skipped. Vote transactions are included and all updates have is_vote: false. No BlockDetails updates are emitted.
• Backpressure. Updates are sent to the pipeline with try_send. When the pipeline channel is full, the datasource logs Error sending transaction update and drops the rest of that block. For large backfills, raise the pipeline's channel_buffer_size and lower max_concurrent_requests.
• Finishing. With end_slot, the pipeline shuts down after the last block when this is its only datasource. Without it, it runs until cancelled.
• Metrics. Counters block_crawler_blocks_fetched_total, block_crawler_blocks_skipped_total, block_crawler_blocks_received_total, block_crawler_blocks_processed_total, block_crawler_transactions_processed_total; histograms block_crawler_blocks_fetch_times_milliseconds, block_crawler_block_process_time_nanoseconds, block_crawler_transaction_process_time_nanoseconds.

Troubleshooting#

Every transaction logs Failed to decode transaction#

block_config.encoding is None or a JSON encoding. Set encoding: Some(UiTransactionEncoding::Base64).

Error fetching block at slot … with "Transaction version (0) is not supported"#

Set max_supported_transaction_version: Some(1). A lower value fails on any block that contains a newer transaction version.

block_crawler_blocks_skipped_total grows quickly#

Normal mainnet skip rate is a few percent of slots. If almost every slot is skipped, your RPC doesn't keep blocks that old (-32009, missing in long-term storage); use an archival RPC.

Error sending transaction update and missing transactions#

The pipeline channel overflowed. Raise Pipeline::builder().channel_buffer_size(…) and reduce max_concurrent_requests.

See also#

• Backfill transactions
• Multiple programs
• RPC blockSubscribe: the same data pushed over a WebSocket.
• Datasources concept and update types