# Use multiple datasources (Carbon 2.0)

> Combine datasources in one pipeline, such as a state load plus a live stream or two streams for redundancy, and route or deduplicate their updates.

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](https://carbon.sevenlabs.org/concepts/filters.md) 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.

### Step 1: Create a project

```sh
cargo new dlmm-sync
cd dlmm-sync
```

### Step 2: Add the dependencies

`Cargo.toml`:

```toml
[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"] }
```

### Step 3: Write the pipeline

`src/main.rs`:

```rust
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(())
    }
}
```

In a real indexer, the map is your database: upsert each account with its slot and skip writes whose slot is older than the stored one.

### Step 4: Run it

```sh
RUST_LOG=info cargo run
```

The snapshot's 113 pools print first, all at the same slot, then live updates follow as swaps move the pools:

Output:

```text
[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 `getProgramAccounts` datasource finishes after the load; the subscription keeps the pipeline running until `Ctrl+C`.

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:

**RPC**

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

**Yellowstone gRPC**

```rust
// 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,
);
```

**Helius LaserStream**

```rust
// 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](https://carbon.sevenlabs.org/guides/stream-with-yellowstone.md), [LaserStream](https://carbon.sevenlabs.org/guides/stream-with-laserstream.md)). For programs too large for `getProgramAccounts`, load the state from [Helius gPA v2](https://carbon.sevenlabs.org/guides/load-account-state.md#large-programs-with-helius-gpa-v2) or a [validator snapshot](https://carbon.sevenlabs.org/guides/load-a-snapshot.md) 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:

`Cargo.toml`:

```toml
[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"] }
```

`src/main.rs`:

```rust
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(())
    }
}
```

```sh
cargo run
```

Each swap prints once, although both crawlers delivered it:

Output:

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

Without the filter, every line appears twice:

```sh
NO_DEDUP=1 cargo run
```

Output:

```text
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](https://carbon.sevenlabs.org/guides/stream-with-yellowstone.md), the datasource part of the program becomes:

```rust
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](https://carbon.sevenlabs.org/guides/stream-with-laserstream.md) stream combine the same way.

### What `DeduplicationFilter` covers

| Pipe                                         | Considered a duplicate when                                          | Not deduplicated                                                                                  |
| -------------------------------------------- | -------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------- |
| Instruction                                  | Same transaction signature and same position in the instruction tree | —                                                                                                 |
| Account                                      | Same transaction signature and same account                          | Account updates without a signature: `rpc-program-subscribe`, `rpc-gpa`, Helius gPA v2, snapshots |
| Transaction, account deletion, block details | Never                                                                | Every update                                                                                      |

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:

```rust
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](https://carbon.sevenlabs.org/guides/backfill-transactions.md#crawl-a-slot-range) 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](https://carbon.sevenlabs.org/guides/prometheus.md), and alert when one of them stops growing.

## Next steps

- [Filters](https://carbon.sevenlabs.org/concepts/filters.md): DatasourceFilter, DeduplicationFilter, SlotRangeFilter and custom filters.
- [Multiple programs](https://carbon.sevenlabs.org/guides/multiple-programs.md): Several decoders in one pipeline.
- [Backfill transactions](https://carbon.sevenlabs.org/guides/backfill-transactions.md): History sources to combine with a stream.

---

Carbon 2.0.0 documentation · Web: https://carbon.sevenlabs.org/guides/multiple-datasources · Last updated 2026-11-09 · Index of all pages: https://carbon.sevenlabs.org/llms.txt