RPC transaction crawler

RpcTransactionCrawler pages through getSignaturesForAddress for one address, fetches each transaction with getTransaction, and emits it as a transaction update. It walks history backwards from the newest signature (or from a signature you choose), then keeps polling for new ones. Use it to backfill a program's or wallet's history, or to index a low-traffic program from any RPC endpoint, including the free public one.

Quick example#

This program crawls the Orca Whirlpool program on the public RPC and prints every decoded swap.

cargo new crawler && cd crawler
cargo add carbon-core@2.0.0 carbon-log-metrics@2.0.0 carbon-orca-whirlpool-decoder@2.0.0 carbon-rpc-transaction-crawler-datasource@2.0.0
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, PROGRAM_ID,
    },
    carbon_rpc_transaction_crawler_datasource::{ConnectionConfig, Filters, RpcTransactionCrawler},
    std::sync::Arc,
};

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

    let datasource = RpcTransactionCrawler::new(
        "https://api.mainnet-beta.solana.com".to_string(),
        PROGRAM_ID,                  // crawl signatures that mention the Orca Whirlpool program
        ConnectionConfig::default(), // 100 signatures per page, 5 concurrent getTransaction calls
        Filters::new(None, None, None),
        None, // commitment: defaults to confirmed
    );

    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 {} swap   pool={} amount={}", tx.slot, accounts.whirlpool, data.amount);
            }
            OrcaWhirlpoolInstruction::SwapV2 { data, accounts, .. } => {
                println!("slot {} swapV2 pool={} amount={}", tx.slot, accounts.whirlpool, data.amount);
            }
            _ => {}
        }
        Ok(())
    }
}

cargo run

slot 467115758 swap   pool=4v8ufj8Hj7UvFgtofQJAtzUud5xomwZfEqfCTHZ4wM72 amount=19999
slot 467115758 swapV2 pool=FpCMFDFGYotvufJ7HrFHsWEiiQCGbkLCtwHiDnh7o28Q amount=7639572
slot 467115758 swap   pool=Czfq3xZZDmsdGdUyrNLtRhGc47cXcZtLG4crryfu44zE amount=2476330084
slot 467115758 swap   pool=FpCMFDFGYotvufJ7HrFHsWEiiQCGbkLCtwHiDnh7o28Q amount=168756526

The public endpoint rate-limits getTransaction per IP (its x-ratelimit-method-limit header reported 10 when this page was written), so expect a few transactions per second at most. Run with RUST_LOG=info to see the 429 retries and the pipeline counters. The program runs until you press Ctrl-C.

Configuration#

RpcTransactionCrawler::new#

pub const fn new(
    rpc_url: String,
    account: Pubkey,
    connection_config: ConnectionConfig,
    filters: Filters,
    commitment: Option<CommitmentConfig>,
) -> Self

All fields of RpcTransactionCrawler are public, so you can also build it with a struct literal.

Filters#

pub const fn new(
    accounts: Option<Vec<Pubkey>>,
    before_signature: Option<Signature>,
    until_signature: Option<Signature>,
) -> Self

ConnectionConfig#

ConnectionConfig::default() is an inherent const fn, not the Default trait. Override single fields with ConnectionConfig { batch_limit: 1000, ..ConnectionConfig::default() }. The positional constructor takes the arguments in a different order from the fields:

pub const fn new(
    batch_limit: usize,
    polling_interval: Duration,
    max_concurrent_requests: usize,
    retry_config: RetryConfig,
    max_signature_channel_size: Option<usize>,
    max_transaction_channel_size: Option<usize>,
    blocking_send: bool,
) -> Self

RetryConfig#

pub const fn new(max_retries: u32, initial_backoff_ms: u64, max_backoff_ms: u64, backoff_multiplier: f64) -> Self
pub const fn default() -> Self   // 3, 1000, 10000, 2.0
pub const fn no_retry() -> Self  // 0, 0, 0, 0.0

Behaviour#

• Pipeline. Three tasks run in sequence: a signature fetcher, a transaction fetcher that runs up to max_concurrent_requests getTransaction calls (Base64 encoding, maxSupportedTransactionVersion: 1), and a processor that builds the update and sends it to the pipeline. consume returns as soon as the tasks are spawned.
• Lifecycle. The first pass pages backwards from before_signature (or the newest signature) until the RPC returns an empty page or until_signature is reached. The crawler then sleeps polling_interval and polls from the tip down to the newest signature it has seen, forever. It never finishes, so a pipeline that only has this datasource runs until Ctrl-C or until you cancel it (see shutdown). After the first pass it also picks up everything newer than before_signature.
• Ordering. Signatures arrive newest first, and transactions are emitted in the order their getTransaction calls complete. Updates are not in slot order. Sort by slot downstream if order matters.
• What is skipped. Failed transactions (non-Ok status) and transactions without metadata are dropped. Every update has is_vote: false, index: None and block_hash: None; block_time comes from the RPC.
• Retries. Each call retries with exponential backoff from RetryConfig. A signature page that still fails is retried from scratch on the next loop, so the crawler never skips a page. A transaction that still fails is logged with Failed to fetch transaction … after N retries and skipped. On top of that, Solana's RpcClient retries HTTP 429 responses up to 5 times, honouring Retry-After, and only logs at debug level.
• Backpressure. With blocking_send: false, updates are dropped with Failed to send update when the pipeline channel (channel_buffer_size, default 1000) is full.
• Metrics. transaction_crawler_signatures_fetched_total, transaction_crawler_transactions_fetched_total (counters), transaction_crawler_signatures_fetch_times_milliseconds, transaction_crawler_transaction_fetch_times_milliseconds, transaction_crawler_transaction_process_time_milliseconds (histograms). See metrics.

Troubleshooting#

Signatures are fetched but nothing is processed#

transaction_crawler_signatures_fetched_total grows while transaction_crawler_transactions_fetched_total stays at 0. At first nothing is logged, because RpcClient retries 429 responses silently; once those retries run out, RUST_LOG=info shows Failed to fetch transaction … Status(429, None). The RPC is rate limiting getTransaction. On the public endpoint the limit is shared by everyone on your IP. Lower max_concurrent_requests to 1 or 2, or use a dedicated RPC provider.

Method does not support commitment below confirmed#

You passed CommitmentConfig::processed(). getSignaturesForAddress and getTransaction only accept confirmed and finalized; the signature fetcher retries the failing call forever.

Transactions are missing from a backfill#

Check the logs for Failed to send update (pipeline channel full) and Failed to fetch transaction … after N retries (RPC errors). Set blocking_send: true, raise retry_config.max_retries, and raise the pipeline's channel_buffer_size if your processors are slow. Failed transactions are skipped by design.

Transactions arrive out of order#

Expected: getTransaction calls complete in any order. Use slot (and your own ordering) in the processor, or set max_concurrent_requests: 1 for strictly newest-first order.

The program never exits after the backfill#

Expected: the crawler switches to polling. To stop at the end of a range, cancel the pipeline from your code with datasource_cancellation_token, or use RpcBlockCrawler with an end_slot.

See also#

• Backfill transactions
• Stream with RPC
• Helius getTransactionsForAddress: the same job in fewer calls, on Helius.
• Datasources concept and update types