Datasources

A datasource reads updates from somewhere (a Geyser stream, an RPC node, a snapshot file, your own queue) and sends them into the pipeline. This page explains the trait every datasource implements, how datasources are identified, and how the published datasources compare. Each datasource's configuration is on its own page under Datasources.

The Datasource trait#

#[async_trait]
pub trait Datasource: Send + Sync {
    async fn consume(
        &self,
        id: DatasourceId,
        sender: tokio::sync::mpsc::Sender<(Update, DatasourceId)>,
        cancellation_token: CancellationToken,
    ) -> CarbonResult<()>;

    fn update_types(&self) -> Vec<UpdateType>;
}

Datasource uses the async_trait macro, so implementations need the async-trait crate. The pipeline calls consume once, in a Tokio task of its own, when run starts. The contract:

• Send each update with its ID. Wrap your data in an Update and send (update, id.clone()) on sender. Build account updates with AccountUpdate { … }.into_update() so zero-lamport accounts become deletions.
• Choose how to handle a full channel. sender.send(…).await waits for room, which slows your source down to the pipeline's pace. sender.try_send(…) fails immediately, so you drop the update or buffer it yourself. Most streaming datasources use try_send; see When processing falls behind.
• Stop when cancelled. Watch cancellation_token and return when it's cancelled. See Shutdown.
• Drop the sender when you're done. The pipeline stops once every datasource has dropped its sender and the channel is empty. A finite source, such as a crawler with an end or a snapshot, returns from consume or ends its tasks; a stream keeps going until cancelled.
• Return errors only for fatal problems. If consume returns Err, the pipeline logs it and that datasource is gone; nothing retries it. Reconnect and retry inside consume if your source can recover.

consume may also spawn tasks that hold clones of sender and return Ok(()) straight away. Many published datasources do this. The pipeline keeps running for as long as any clone of the sender is alive.

update_types lists the UpdateTypes the datasource can send. In 2.0.0 the pipeline doesn't check it, but keep it accurate for readers of your code.

Datasource IDs#

Every datasource in a pipeline has a DatasourceId, which the pipeline passes along with each of its updates. Filters read it from FilterContext to route updates by source.

IDs compare by their string value, and the builder doesn't check that they're unique. Two datasources registered with the same name look like one source to every filter.

let primary = DatasourceId::new_named("primary");

Pipeline::builder()
    .datasource_with_id(yellowstone, primary.clone())
    .datasource_with_id(laserstream, DatasourceId::new_named("backup"))
    .instruction_with_filters(decoder, processor, vec![Box::new(DatasourceFilter::new(primary))])

Use multiple datasources shows a full redundant setup.

Kinds of datasources#

The published datasources fall into six families:

• Geyser gRPC streams push account writes and transactions as a validator processes them. Lowest latency; you need a Yellowstone-compatible endpoint from a provider, or your own node with the plugin.
• RPC websocket subscriptions use the standard Solana blockSubscribe and programSubscribe methods. They work with many RPC providers, but the public endpoint doesn't support blockSubscribe.
• RPC crawlers poll RPC methods to walk history: by address with getSignaturesForAddress and getTransaction, or by slot with getBlock. Use them for backfills; they work against the public endpoint, slowly.
• Account state loaders read every account a program owns at one point in time, with getProgramAccounts or from a validator snapshot.
• Hosted provider APIs use Helius-specific methods for websockets, history and paginated account reads.
• Adapters turn your own channel of updates into a datasource, for queues, files or a custom Geyser plugin.

Comparison#

Common combinations:

• Real-time indexer: a Geyser stream, plus a second one with a DatasourceFilter or DeduplicationFilter for redundancy.
• Backfill then follow: the transaction or block crawler for history, then a stream for new data. See Backfill transactions.
• Current state then follow: getProgramAccounts or a snapshot to load accounts, then program subscribe or a Geyser stream for changes. See Load account state.

The datasource index helps you choose. If none fits, write your own.

Example#

This custom datasource replays a fixed list of accounts and then finishes, which ends the pipeline. It sends with send().await, respects cancellation and uses into_update. The pipeline registers it under a named ID.

cargo add carbon-core@2.0.0 async-trait tokio-util
cargo add solana-account@4.3.2 solana-pubkey@4.2.1
cargo add tokio --features full

use {
    async_trait::async_trait,
    carbon_core::{
        account::{AccountDecoder, AccountProcessorInputType, DecodedAccount},
        datasource::{AccountUpdate, Datasource, DatasourceId, Update, UpdateType},
        error::CarbonResult,
        pipeline::Pipeline,
        processor::Processor,
    },
    solana_account::Account,
    solana_pubkey::Pubkey,
    tokio::sync::mpsc::Sender,
    tokio_util::sync::CancellationToken,
};

/// Replays a fixed list of accounts, then finishes.
struct Fixtures {
    accounts: Vec<(Pubkey, Account)>,
    slot: u64,
}

#[async_trait]
impl Datasource for Fixtures {
    async fn consume(
        &self,
        id: DatasourceId,
        sender: Sender<(Update, DatasourceId)>,
        cancellation_token: CancellationToken,
    ) -> CarbonResult<()> {
        for (pubkey, account) in &self.accounts {
            if cancellation_token.is_cancelled() {
                break;
            }
            let update = AccountUpdate {
                pubkey: *pubkey,
                account: account.clone(),
                slot: self.slot,
                transaction_signature: None,
            }
            .into_update(); // zero-lamport accounts become AccountDeletion

            // `send` waits when the pipeline's channel is full (back-pressure).
            if sender.send((update, id.clone())).await.is_err() {
                break; // the pipeline has stopped
            }
        }
        Ok(())
    }

    fn update_types(&self) -> Vec<UpdateType> {
        vec![UpdateType::AccountUpdate, UpdateType::AccountDeletion]
    }
}

struct AnyAccount;

impl<'a> AccountDecoder<'a> for AnyAccount {
    type AccountType = ();

    fn decode_account(&self, account: &'a Account) -> Option<DecodedAccount<()>> {
        Some(DecodedAccount {
            lamports: account.lamports,
            data: (),
            owner: account.owner,
            executable: account.executable,
            rent_epoch: account.rent_epoch,
        })
    }
}

struct PrintAccount;

impl Processor<AccountProcessorInputType<'_, ()>> for PrintAccount {
    async fn process(&mut self, input: &AccountProcessorInputType<'_, ()>) -> CarbonResult<()> {
        println!("{} has {} lamports", input.metadata.pubkey, input.raw_account.lamports);
        Ok(())
    }
}

#[tokio::main]
async fn main() -> CarbonResult<()> {
    let fixtures = Fixtures {
        accounts: vec![
            (Pubkey::new_from_array([1; 32]), Account { lamports: 1_000_000, ..Account::default() }),
            (Pubkey::new_from_array([2; 32]), Account { lamports: 2_000_000, ..Account::default() }),
        ],
        slot: 1,
    };

    Pipeline::builder()
        .datasource_with_id(fixtures, DatasourceId::new_named("fixtures"))
        .account(AnyAccount, PrintAccount)
        .build()?
        .run()
        .await
}

cargo run

4vJ9JU1bJJE96FWSJKvHsmmFADCg4gpZQff4P3bkLKi has 1000000 lamports
8qbHbw2BbbTHBW1sbeqakYXVKRQM8Ne7pLK7m6CVfeR has 2000000 lamports

Write a custom datasource builds a networked one with reconnects.

Next steps#