Backfill transactions

A backfill runs your decoders and processors over past transactions, to build an index from a program's launch or to fill a gap in a live stream. Carbon has three history datasources. This guide runs the two RPC ones against the public endpoint and shows how to bound, resume and speed up a backfill.

You need: Rust 1.96.1 or newer. The RPC crawlers work with any Solana RPC endpoint, including https://api.mainnet-beta.solana.com, which is enough for small ranges. For large backfills, use a paid RPC endpoint or Helius GTFA, which needs a Helius API key.

Choose a source#

All three skip failed transactions unless you configure GTFA to keep them.

Backfill an address's history#

This program crawls Meteora DLMM's transactions between two signatures and prints its swaps.

cargo new dlmm-backfill
cd dlmm-backfill

[package]
name = "dlmm-backfill"
version = "0.1.0"
edition = "2024"

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

use {
    carbon_core::{
        error::CarbonResult, filter::SlotRangeFilter, instruction::InstructionProcessorInputType,
        pipeline::Pipeline, processor::Processor,
    },
    carbon_meteora_dlmm_decoder::{
        instructions::MeteoraDlmmInstruction, MeteoraDlmmDecoder, PROGRAM_ID,
    },
    carbon_rpc_transaction_crawler_datasource::{ConnectionConfig, Filters, RpcTransactionCrawler},
    solana_client::rpc_config::CommitmentConfig,
    solana_signature::Signature,
    std::str::FromStr,
};

// The newest transaction to stop before (exclusive) and its slot.
const BEFORE: &str =
    "54hSq76xk8noTVuNQaEyCM9m7j9t7LWEbu2tRRBwAFmFPHXC54jYLJqUyexrGGBaDEC8PokZKN7jUQeZnBejP9fc";
const BEFORE_SLOT: u64 = 467_110_309;
// The oldest transaction to stop at (exclusive) and its slot.
const UNTIL: &str =
    "2d6GNQzSzxgV85mnzD4ed6JyxdLYdwBtyCH8LujdYaxySSKfD8x96yJLtxmuVYt6XGRjy8m3AutxyxqiumscEb4W";
const UNTIL_SLOT: u64 = 467_110_307;

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

    let filters = Filters::new(
        None,
        Some(Signature::from_str(BEFORE).expect("valid signature")),
        Some(Signature::from_str(UNTIL).expect("valid signature")),
    );

    let connection = ConnectionConfig {
        max_concurrent_requests: 2, // stay under the public endpoint's limits
        blocking_send: true,        // wait for the pipeline instead of dropping updates
        ..ConnectionConfig::default()
    };

    let datasource = RpcTransactionCrawler::new(
        "https://api.mainnet-beta.solana.com".to_string(),
        PROGRAM_ID,
        connection,
        filters,
        Some(CommitmentConfig::finalized()),
    );

    Pipeline::builder()
        .datasource(datasource)
        .instruction_with_filters(
            MeteoraDlmmDecoder,
            SwapLogger,
            vec![Box::new(SlotRangeFilter::between(UNTIL_SLOT, None, BEFORE_SLOT, None))],
        )
        .build()?
        .run()
        .await
}

struct SwapLogger;

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

cargo run

slot 467110309 swap2 cXStywP9DDr2Cgk4RBnCkhKTVwwRq9XaYYebcTUd1Uk97xZiVSVnc814B3G9J1xAJufVJaRGQiPeVZaX2tRKfdL: amount_in=516957831
slot 467110309 swap2 57o62AmoqKSKqznKJQ4yd5fwcKi7KezErbVS1tFDy5xexHmVeKmTgyGHtfsGY184W7zHvmjypBBXT4XDnyAympuZ: amount_in=512968
slot 467110309 swap2 2f5k2YKnopfhtGG9ro2Zd4KLXUznr1U2ege2GbrQdUbwDeNSX4cSDcMAadMk3TVjG9Ggc9TpbojRyeVDZhQHzoi3: amount_in=381964223
slot 467110309 swap2 4jvbKVzCJ1SYVDWh6pftBtUnrub7QDq3nGzQGVNAaybRV1jhjvoXhDkUSU6wK7BoNCNoERondJSHC9uxEhGgiTHC: amount_in=2002103
…

Verify the backfill#

Count what your processor handled and compare it with the source. For a signature range, call getSignaturesForAddress with the same before and until: the signatures with "err": null are the transactions the crawler delivers, since it skips failed ones. For a slot range, a block explorer shows each block's transactions for the program. Add a LogMetrics exporter and run with RUST_LOG=info to watch transaction_crawler_signatures_fetched_total and transaction_crawler_transactions_fetched_total (both count failed transactions too) while the crawl runs.

Bound the range#

Filters::new(
    None,                 // accounts: keep only transactions that also mention one of these
    Some(before),         // start just older than this signature; None = the newest
    Some(until),          // stop just before this signature; None = the address's first transaction
)

use carbon_helius_gtfa_datasource::{HeliusGtfaFilters, SlotFilter, TransactionStatusFilter};

HeliusGtfaFilters {
    slot: Some(SlotFilter { gte: Some(467_110_307), lte: Some(467_110_309), gt: None, lt: None }),
    block_time: None, // BlockTimeFilter { gte, gt, lte, lt, eq } in Unix seconds
    signature: None,
    status: Some(TransactionStatusFilter::Succeeded),
}

Any datasource can also be bounded on the pipe with SlotRangeFilter: from(slot, None), to(slot, None) or between(from, None, to, None), inclusive at both ends.

Crawl a slot range#

The block crawler fetches every block from start_slot to end_slot (inclusive), turns each successful transaction into an update, and exits when it's done. Each block holds all programs' transactions, so the decoders pick what they understand. This program prints DLMM swaps from three blocks on the public endpoint.

[dependencies]
carbon-core = "2.0.0"
carbon-meteora-dlmm-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_rpc_block_crawler_datasource::{RpcBlockConfig, RpcBlockCrawler},
    solana_client::rpc_config::{CommitmentConfig, TransactionDetails, UiTransactionEncoding},
};

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

    let block_config = RpcBlockConfig {
        encoding: Some(UiTransactionEncoding::Base64),
        transaction_details: Some(TransactionDetails::Full),
        rewards: Some(false),
        commitment: Some(CommitmentConfig::finalized()),
        max_supported_transaction_version: Some(1),
    };

    let datasource = RpcBlockCrawler::new(
        "https://api.mainnet-beta.solana.com".to_string(),
        467_110_307,       // start slot (inclusive)
        Some(467_110_309), // end slot (inclusive); None keeps following the tip
        None,              // block interval (default 100 ms)
        block_config,
        Some(1), // concurrent getBlock requests (default 10)
        None,    // internal channel size (default 1000)
    );

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

struct SwapLogger;

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

RUST_LOG=info cargo run

[2026-11-09T00:10:31Z INFO  carbon_core::pipeline] starting pipeline. num_datasources: 1, num_exporters: 0, num_account_pipes: 0, num_account_deletion_pipes: 0, num_instruction_pipes: 1, num_transaction_pipes: 0
slot 467110307 tx 171 swap  ue82HW7rk3DcWiCTzRrnMzMeKDYDLWUiyeMBtMi4tLtqiKaRGgbr7EYigU9RbpftsFsgEC7P6kgxkNWJgZhvKZs: amount_in=10298583
slot 467110307 tx 475 swap  48kbDBkFMweb1bVYk9o3uGSR9CQS5pkgm8YCGB9Qg9ymui5dXmwjjoiSAHfF4eQ8mt6j688pkrG8n33GnZX4cffd: amount_in=1880866
slot 467110307 tx 623 swap2 2d6GNQzSzxgV85mnzD4ed6JyxdLYdwBtyCH8LujdYaxySSKfD8x96yJLtxmuVYt6XGRjy8m3AutxyxqiumscEb4W: amount_in=381756
slot 467110308 tx 192 swap2 2miT4n9yFXZMdUTVF4My1RVvHrbbZMwD64vYyppzdkxNg5qVtYJgsm4dL8MPYtBRrJm2V3EzP4ZfYJKjEc61Wzr1: amount_in=28328938705735
slot 467110308 tx 411 swap  36sp8szy2wPdGGFWXnqJLh2Y4yFhd5SRxE7PR1UYcbJf9Kjd7TuksxUDt26wWHTcfhqg3mU4UquJAGFeVEwvpdQM: amount_in=23315968
slot 467110309 tx 47 swap  5XMR5d7vnGdiNoicdAPXhcSrB62d6z4rmM3vqyG2TSxg9wzDfKUMp8PSmt7awJhbhAujdG4Pq8BuvhxHBioS31vf: amount_in=1231056
…
[2026-11-09T00:10:34Z INFO  carbon_core::pipeline] update_receiver closed, shutting down.
[2026-11-09T00:10:34Z INFO  carbon_core::pipeline] pipeline shutdown complete.

Unlike the transaction crawler, the block crawler sets index (the transaction's position in the block), block_time and block_hash. Slots without a block are skipped silently. With several concurrent requests, blocks finish out of order; transactions within a block stay in order.

RpcBlockConfig must request Base64 encoding, Full transaction details and max_supported_transaction_version: Some(1); with the default Json encoding, every transaction fails with Failed to decode transaction.

Backfill with Helius GTFA#

Helius's getTransactionsForAddress returns full transactions for an address in pages of up to 100, filtered and sorted on the server, so it needs one request per 100 transactions instead of one per transaction. It needs a Helius RPC URL with an API key, on a plan that includes the method.

[dependencies]
carbon-core = "2.0.0"
carbon-helius-gtfa-datasource = "2.0.0"
carbon-meteora-dlmm-decoder = "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_helius_gtfa_datasource::{
        HeliusGtfaConfig, HeliusGtfaDatasource, HeliusGtfaFilters, SlotFilter, SortOrder,
        TransactionStatusFilter,
    },
    carbon_meteora_dlmm_decoder::{
        instructions::MeteoraDlmmInstruction, MeteoraDlmmDecoder, PROGRAM_ID,
    },
    solana_client::rpc_config::CommitmentConfig,
    std::env,
};

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

    let filters = HeliusGtfaFilters {
        slot: Some(SlotFilter {
            gte: Some(467_110_307),
            lte: Some(467_110_309),
            gt: None,
            lt: None,
        }),
        block_time: None,
        signature: None,
        status: Some(TransactionStatusFilter::Succeeded),
    };

    let config = HeliusGtfaConfig::new(
        Some(SortOrder::Asc), // oldest first
        Some(100),            // page size (max 100)
        Some(CommitmentConfig::finalized()),
        Some(filters),
        None, // min_context_slot
    );

    let datasource = HeliusGtfaDatasource::new_with_config(
        env::var("HELIUS_RPC_URL").expect("HELIUS_RPC_URL must be set"),
        PROGRAM_ID,
        config,
    );

    Pipeline::builder()
        .datasource(datasource)
        .instruction(MeteoraDlmmDecoder, SwapLogger)
        .build()?
        .run()
        .await
}

struct SwapLogger;

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

HELIUS_RPC_URL="https://mainnet.helius-rpc.com/?api-key=your-helius-api-key" cargo run

It prints the same swaps as the crawlers, oldest first, and exits after Reached end of pagination.

• HeliusGtfaDatasource::new(url, address) uses the defaults: oldest first, 100 per page, confirmed, no filters.
• status: Succeeded (and None) skip failed transactions, Failed keeps only failed ones, Any keeps both.
• Each transaction carries its index in the block and its block_time.
• The datasource waits for the pipeline, so nothing is dropped. It doesn't retry: the first HTTP or RPC error stops it with FailedToConsumeDatasource. Resume as described below.

Resume an interrupted backfill#

Store your progress next to the data your processor writes, then restart from it:

• GTFA, oldest first: store the last slot you finished. Restart with slot: Some(SlotFilter { gt: Some(last_slot), .. }). Its transactions arrive in order, so the last slot is a clean checkpoint.
• Block crawler: store finished slots. Blocks complete out of order, so restart from the lowest slot you haven't finished, not the highest you have.
• Transaction crawler: store the oldest signature you processed and restart with it as before. Requests run concurrently, so a few transactions just newer than it may run again.

Every source can deliver a transaction twice across a restart. Make writes idempotent: key rows on (signature, instruction path) for instructions, or (pubkey, slot) for accounts, and upsert. See Persist to Postgres.

Rate limits and throughput#

Throughput depends on the RPC endpoint far more than on Carbon. The public endpoint limits each IP per method (the response headers x-ratelimit-method-limit and x-ratelimit-method-remaining show the current budget) and the total data you download. Solana's RPC client retries HTTP 429 responses on its own, so on a rate-limited endpoint the crawler slows down instead of failing. To see the retries:

RUST_LOG=info,solana_rpc_client=debug cargo run

[2026-11-09T00:14:13Z DEBUG solana_rpc_client::http_sender] Too many requests: server responded with Response { url: "https://api.mainnet-beta.solana.com/", status: 429, headers: {"retry-after": "10", …

The transaction crawler's ConnectionConfig controls its request rate. Its fields are public, so set the ones you need on top of ConnectionConfig::default():

For the block crawler, the max_concurrent_requests argument (default 10) sets parallel getBlock calls and block_interval (default 100 ms) how often it checks for new slots when following the tip.

Backfill, then follow live data#

Run a backfill datasource and a stream in the same pipeline: for example the block crawler from a stored slot to the current slot, plus Yellowstone gRPC from now on, with a DeduplicationFilter for the overlap. Multiple datasources shows how to combine and route them.

Common problems#

Long pauses, or nothing for minutes, on the public endpoint#

You're being rate-limited, possibly by other programs on the same IP. Run with RUST_LOG=info,solana_rpc_client=debug to see the 429 responses. Lower max_concurrent_requests, or use your own RPC endpoint.

Failed to fetch transaction … after 3 retries#

The endpoint kept failing for that transaction, and the crawler skipped it. Raise RetryConfig::max_retries or the backoff, or lower the request rate. Re-crawl the range later to pick up skipped transactions, with idempotent writes.

Failed to send update warnings and missing transactions#

blocking_send is false and the pipeline was busy, so updates were dropped. Set blocking_send: true.

Error sending transaction update from the block crawler#

The pipeline's channel was full and the rest of that block was dropped. Raise .channel_buffer_size(…) on the builder.

The backfill never finishes#

The transaction crawler polls for new transactions forever after its history; stop it with Ctrl+C or a cancellation token. The block crawler with end_slot: None follows the tip forever.

Failed to decode transaction for every transaction#

RpcBlockConfig doesn't request Base64 encoding. Set encoding: Some(UiTransactionEncoding::Base64).

Looking for transactions that failed on chain rather than ones Carbon couldn't decode? See Decode or skip failed transactions.

RPC error: … (code: …) and the GTFA backfill stops#

Helius rejected a request: a wrong API key, a plan without getTransactionsForAddress, or an invalid filter. The datasource doesn't retry; fix the cause and resume from your last slot.

Next steps#