Use multiple datasources

A pipeline can read from several datasources at once. Use it to load current state and then follow changes, to backfill history while streaming new data, or to run two streams for redundancy. This guide builds the first case and runs it on the public endpoint, then shows redundancy with DeduplicationFilter.

You need: Rust 1.96.1 or newer. The first two examples run on https://api.mainnet-beta.solana.com. The production redundancy setup needs two Yellowstone gRPC endpoints.

How several datasources run#

Each datasource runs in its own task, and all of them send into the pipeline's single channel, so their updates interleave in arrival order. Every update carries the DatasourceId of the datasource that produced it, which filters can read. Processors don't see the ID; route with a filter instead.

The pipeline runs until every datasource has finished or you press Ctrl+C. A finite datasource, such as rpc-gpa or a bounded crawler, ends and the others keep going. A datasource that fails logs error consuming datasource: … and stops without affecting the others.

Load state, then follow changes#

This program loads the current state of every SOL/USDC Meteora DLMM pool with getProgramAccounts, and follows their changes with programSubscribe. Two pipes share one decoder type, each limited to one datasource with a DatasourceFilter, and a shared map applies an update only if it's at least as new as what's stored, so a late snapshot row can't overwrite a newer live one.

cargo new dlmm-sync
cd dlmm-sync

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

[dependencies]
carbon-core = "2.0.0"
carbon-meteora-dlmm-decoder = "2.0.0"
carbon-rpc-gpa-datasource = "2.0.0"
carbon-rpc-program-subscribe-datasource = "2.0.0"
env_logger = "0.11.11"
log = "0.4.35"
solana-client = "=4.2.2"
solana-pubkey = "4.2.1"
tokio = { version = "1.54.1", features = ["full"] }

use {
    carbon_core::{
        account::AccountProcessorInputType,
        datasource::DatasourceId,
        error::CarbonResult,
        filter::DatasourceFilter,
        pipeline::Pipeline,
        processor::Processor,
    },
    carbon_meteora_dlmm_decoder::{accounts::MeteoraDlmmAccount, MeteoraDlmmDecoder, PROGRAM_ID},
    carbon_rpc_gpa_datasource::GpaDatasource,
    carbon_rpc_program_subscribe_datasource::{Filters, RpcProgramSubscribe},
    solana_client::{
        rpc_config::{
            CommitmentConfig, RpcAccountInfoConfig, RpcProgramAccountsConfig, UiAccountEncoding,
        },
        rpc_filter::{Memcmp, RpcFilterType},
    },
    solana_pubkey::{pubkey, Pubkey},
    std::{
        collections::HashMap,
        sync::{Arc, Mutex},
    },
};

const LB_PAIR_DISCRIMINATOR: [u8; 8] = [33, 11, 49, 98, 181, 101, 177, 13];
const WSOL: Pubkey = pubkey!("So11111111111111111111111111111111111111112");
const USDC: Pubkey = pubkey!("EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v");

/// SOL/USDC `LbPair` accounts: discriminator, then token X mint at byte 88
/// and token Y mint at byte 120.
fn sol_usdc_pairs() -> RpcProgramAccountsConfig {
    RpcProgramAccountsConfig {
        filters: Some(vec![
            RpcFilterType::Memcmp(Memcmp::new_raw_bytes(0, LB_PAIR_DISCRIMINATOR.to_vec())),
            RpcFilterType::Memcmp(Memcmp::new_raw_bytes(88, WSOL.to_bytes().to_vec())),
            RpcFilterType::Memcmp(Memcmp::new_raw_bytes(120, USDC.to_bytes().to_vec())),
        ]),
        account_config: RpcAccountInfoConfig {
            encoding: Some(UiAccountEncoding::Base64),
            commitment: Some(CommitmentConfig::confirmed()),
            ..Default::default()
        },
        ..Default::default()
    }
}

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

    let snapshot_id = DatasourceId::new_named("snapshot");
    let live_id = DatasourceId::new_named("live");

    let snapshot = GpaDatasource::new_with_config(
        "https://api.mainnet-beta.solana.com".to_string(),
        PROGRAM_ID,
        sol_usdc_pairs(),
    );
    let live = RpcProgramSubscribe::new(
        "wss://api.mainnet-beta.solana.com".to_string(),
        Filters::new(PROGRAM_ID, Some(sol_usdc_pairs())),
    );

    // Shared by both processors: the newest slot applied for each pair.
    let state = Arc::new(Mutex::new(HashMap::new()));

    Pipeline::builder()
        .datasource_with_id(snapshot, snapshot_id.clone())
        .datasource_with_id(live, live_id.clone())
        .account_with_filters(
            MeteoraDlmmDecoder,
            PairWriter { source: "snapshot", state: state.clone() },
            vec![Box::new(DatasourceFilter::new(snapshot_id))],
        )
        .account_with_filters(
            MeteoraDlmmDecoder,
            PairWriter { source: "live", state },
            vec![Box::new(DatasourceFilter::new(live_id))],
        )
        .build()?
        .run()
        .await
}

struct PairWriter {
    source: &'static str,
    state: Arc<Mutex<HashMap<Pubkey, u64>>>,
}

impl Processor<AccountProcessorInputType<'_, MeteoraDlmmAccount>> for PairWriter {
    async fn process(
        &mut self,
        input: &AccountProcessorInputType<'_, MeteoraDlmmAccount>,
    ) -> CarbonResult<()> {
        let MeteoraDlmmAccount::LbPair(pair) = &input.decoded_account.data else {
            return Ok(());
        };
        let (pubkey, slot) = (input.metadata.pubkey, input.metadata.slot);

        let mut state = self.state.lock().unwrap();
        if state.get(&pubkey).is_some_and(|&seen| seen > slot) {
            return Ok(()); // older than what we already have
        }
        state.insert(pubkey, slot);

        println!("{:<8} slot {slot} pair {pubkey}: active_id={}", self.source, pair.active_id);
        Ok(())
    }
}

RUST_LOG=info cargo run

[2026-11-09T00:13:01Z INFO  carbon_core::pipeline] starting pipeline. num_datasources: 2, num_exporters: 0, num_account_pipes: 2, num_account_deletion_pipes: 0, num_instruction_pipes: 0, num_transaction_pipes: 0
[2026-11-09T00:13:03Z INFO  carbon_rpc_gpa_datasource] Fetched 113 accounts for program LBUZKhRxPF3XUpBCjp4YzTKgLccjZhTSDM9YuVaPwxo (slot: 467112022)
snapshot slot 467112022 pair 1jw5fDodwGEGBVqNXsx2eqiLgNmgMDEeXWSbrTreLCM: active_id=-2056
snapshot slot 467112022 pair 5rCf1DM8LjKTw4YqhnoLcngyZYeNnQqztScTogYHAS6: active_id=-6112
snapshot slot 467112022 pair BGm1tav58oGcsQJehL9WXBFXF7D27vZsKefj4xJKD5Y: active_id=-2446
…
live     slot 467112036 pair 2L47gbUzGsnJZqRzfny4b3VWvKBnMRTMeNH3uzFrj6ko: active_id=-307
live     slot 467112051 pair 5rCf1DM8LjKTw4YqhnoLcngyZYeNnQqztScTogYHAS6: active_id=-6112
live     slot 467112059 pair HTvjzsfX3yU6BUodCjZ5vZkUrAxMDTrBs3CJaq43ashR: active_id=-24438
live     slot 467112059 pair 5rCf1DM8LjKTw4YqhnoLcngyZYeNnQqztScTogYHAS6: active_id=-6111
…

The subscription starts at the same time as the load, so changes made while the load is in flight still arrive, and the slot check discards any that the load already covers. To follow changes with a gRPC stream instead, replace the live datasource:

let live = RpcProgramSubscribe::new(
    "wss://api.mainnet-beta.solana.com".to_string(),
    Filters::new(PROGRAM_ID, Some(sol_usdc_pairs())),
);

// account_filters: a SubscribeRequestFilterAccounts with owner = PROGRAM_ID and
// the three memcmp filters, as in /guides/stream-with-yellowstone#stream-account-changes
let live = YellowstoneGrpcGeyserClient::new(
    env::var("GEYSER_URL").expect("GEYSER_URL must be set"),
    env::var("X_TOKEN").ok(),
    Some(CommitmentLevel::Confirmed),
    account_filters,
    HashMap::new(),
    BlockFilters::default(),
    YellowstoneGrpcClientConfig::default(),
    None,
    None,
);

// account_filters: the same SubscribeRequestFilterAccounts as for Yellowstone
let live = LaserStreamGeyserClient::new(
    env::var("LASERSTREAM_URL").expect("LASERSTREAM_URL must be set"),
    Some(env::var("HELIUS_API_KEY").expect("HELIUS_API_KEY must be set")),
    Some(CommitmentLevel::Confirmed),
    account_filters,
    HashMap::new(),
    BlockFilters::default(),
    LaserStreamClientConfig::default(),
);

The gRPC datasources need the extra dependencies and the rustls provider setup from their guides (Yellowstone, LaserStream). For programs too large for getProgramAccounts, load the state from Helius gPA v2 or a validator snapshot instead; the pipeline is the same.

Deduplicate two redundant streams#

Two datasources that deliver the same transactions, such as two Yellowstone endpoints from different providers, keep the pipeline fed when one of them drops. Each transaction then arrives twice; a DeduplicationFilter on the pipe lets only the first copy through.

This program demonstrates it on the public endpoint, with two block crawlers reading the same block:

[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,
        filter::{DeduplicationFilter, Filter},
        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},
    std::{env, time::Duration},
};

/// Reads one block, 467110307.
fn crawler() -> RpcBlockCrawler {
    RpcBlockCrawler::new(
        "https://api.mainnet-beta.solana.com".to_string(),
        467_110_307,
        Some(467_110_307),
        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,
    )
}

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

    // Set NO_DEDUP=1 to see every duplicate.
    let filters: Vec<Box<dyn Filter>> = if env::var("NO_DEDUP").is_err() {
        vec![Box::new(DeduplicationFilter::new(Duration::from_secs(60)))]
    } else {
        vec![]
    };

    Pipeline::builder()
        .datasource(crawler())
        .datasource(crawler())
        .instruction_with_filters(MeteoraDlmmDecoder, SwapLogger, filters)
        .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!("swap  {}: amount_in={}", meta.signature, data.amount_in)
            }
            MeteoraDlmmInstruction::Swap2 { data, .. } => {
                println!("swap2 {}: amount_in={}", meta.signature, data.amount_in)
            }
            _ => {}
        }
        Ok(())
    }
}

cargo run

Each swap prints once, although both crawlers delivered it:

swap  ue82HW7rk3DcWiCTzRrnMzMeKDYDLWUiyeMBtMi4tLtqiKaRGgbr7EYigU9RbpftsFsgEC7P6kgxkNWJgZhvKZs: amount_in=10298583
swap  48kbDBkFMweb1bVYk9o3uGSR9CQS5pkgm8YCGB9Qg9ymui5dXmwjjoiSAHfF4eQ8mt6j688pkrG8n33GnZX4cffd: amount_in=1880866
swap2 2d6GNQzSzxgV85mnzD4ed6JyxdLYdwBtyCH8LujdYaxySSKfD8x96yJLtxmuVYt6XGRjy8m3AutxyxqiumscEb4W: amount_in=381756

Without the filter, every line appears twice:

NO_DEDUP=1 cargo run

swap  ue82HW7rk3DcWiCTzRrnMzMeKDYDLWUiyeMBtMi4tLtqiKaRGgbr7EYigU9RbpftsFsgEC7P6kgxkNWJgZhvKZs: amount_in=10298583
swap  ue82HW7rk3DcWiCTzRrnMzMeKDYDLWUiyeMBtMi4tLtqiKaRGgbr7EYigU9RbpftsFsgEC7P6kgxkNWJgZhvKZs: amount_in=10298583
swap  48kbDBkFMweb1bVYk9o3uGSR9CQS5pkgm8YCGB9Qg9ymui5dXmwjjoiSAHfF4eQ8mt6j688pkrG8n33GnZX4cffd: amount_in=1880866
swap  48kbDBkFMweb1bVYk9o3uGSR9CQS5pkgm8YCGB9Qg9ymui5dXmwjjoiSAHfF4eQ8mt6j688pkrG8n33GnZX4cffd: amount_in=1880866
swap2 2d6GNQzSzxgV85mnzD4ed6JyxdLYdwBtyCH8LujdYaxySSKfD8x96yJLtxmuVYt6XGRjy8m3AutxyxqiumscEb4W: amount_in=381756
swap2 2d6GNQzSzxgV85mnzD4ed6JyxdLYdwBtyCH8LujdYaxySSKfD8x96yJLtxmuVYt6XGRjy8m3AutxyxqiumscEb4W: amount_in=381756

In production: two gRPC streams#

Swap the crawlers for two streams with the same filters. With the dependencies from Stream with Yellowstone gRPC, the datasource part of the program becomes:

fn geyser(endpoint: String, x_token: Option<String>) -> YellowstoneGrpcGeyserClient {
    let mut transaction_filters = HashMap::new();
    transaction_filters.insert(
        "meteora_dlmm".to_string(),
        SubscribeRequestFilterTransactions {
            vote: Some(false),
            failed: Some(false),
            account_required: vec![PROGRAM_ID.to_string()],
            ..Default::default()
        },
    );

    YellowstoneGrpcGeyserClient::new(
        endpoint,
        x_token,
        Some(CommitmentLevel::Confirmed),
        HashMap::new(),
        transaction_filters,
        BlockFilters::default(),
        YellowstoneGrpcClientConfig::default(),
        None,
        None,
    )
}

// in main, after installing the rustls provider:
let primary = geyser(env::var("GEYSER_URL_A").unwrap(), env::var("X_TOKEN_A").ok());
let backup = geyser(env::var("GEYSER_URL_B").unwrap(), env::var("X_TOKEN_B").ok());

Pipeline::builder()
    .datasource(primary)
    .datasource(backup)
    .instruction_with_filters(
        MeteoraDlmmDecoder,
        SwapLogger,
        vec![Box::new(DeduplicationFilter::new(Duration::from_secs(60)))],
    )

A Yellowstone stream and a LaserStream stream combine the same way.

What DeduplicationFilter covers#

Keep these limits in mind:

• One instance per pipe. Each DeduplicationFilter remembers only what passed through its own pipe. Give every pipe that needs it its own.
• The TTL is a window. It forgets an entry after its TTL (entries are cleaned up at most once a minute). Set it longer than the worst lag between your sources: 60 seconds covers two healthy live streams; raise it to a few minutes if a source can lag further, for example while it reconnects.
• In memory only. A restart forgets everything, so storage writes should still be idempotent.

For transaction processors and signature-less account sources, deduplicate in your processor or database, for example with a unique key on the signature.

Route updates by source#

datasource_with_id names a datasource; DatasourceFilter::new(id) (or new_many(vec![…])) limits a pipe to the named ones. Use it when sources need different processing:

let backfill = DatasourceId::new_named("backfill");
let live = DatasourceId::new_named("live");

Pipeline::builder()
    .datasource_with_id(block_crawler, backfill.clone())
    .datasource_with_id(yellowstone, live.clone())
    .instruction_with_filters(MeteoraDlmmDecoder, BulkWriter, vec![Box::new(DatasourceFilter::new(backfill))])
    .instruction_with_filters(MeteoraDlmmDecoder, LiveWriter, vec![Box::new(DatasourceFilter::new(live))])

Here a block crawler fills history from a stored slot while Yellowstone streams from now on: the backfill can batch its writes, and the live writer can publish notifications that make no sense for old data. DatasourceId::new_named compares by name, so build each ID once and clone it, and never give two datasources the same name. .datasource(…) assigns a random ID you can't filter on.

Common problems#

Every update appears twice#

Two datasources deliver the same data and the pipe has no DeduplicationFilter, or the updates have no signature to deduplicate on (see the table above). Add the filter, or make writes idempotent.

A pipe with a DatasourceFilter receives nothing#

The datasource was added with .datasource(…) instead of .datasource_with_id(…), or the IDs differ in spelling. Build the DatasourceId once and pass clones to both places.

Old values overwrite newer ones#

A finite source (a load or a backfill) delivered an account after the stream had already delivered a newer version. Compare input.metadata.slot with the stored slot before writing, as in the first example.

The pipeline keeps running after the load finished#

That's expected while a stream is still running. The pipeline stops when every datasource has finished, or on Ctrl+C.

One datasource failed and the pipeline kept going without it#

Each datasource fails independently and the pipeline only logs error consuming datasource: …. Watch the per-datasource counters with metrics, and alert when one of them stops growing.

Next steps#