# Datasources (Carbon 2.0) > The Datasource trait, datasource IDs and declared update types, and how the Geyser, RPC, crawler, snapshot, provider and adapter datasources compare. 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](https://carbon.sevenlabs.org/datasources.md). ## The `Datasource` trait ```rust #[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; } ``` `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`](https://carbon.sevenlabs.org/concepts/architecture.md#what-run-does) starts. The contract: - **Send each update with its ID.** Wrap your data in an [`Update`](https://carbon.sevenlabs.org/concepts/updates.md) 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](https://carbon.sevenlabs.org/concepts/architecture.md#when-processing-falls-behind). - **Stop when cancelled.** Watch `cancellation_token` and return when it's cancelled. See [Shutdown](https://carbon.sevenlabs.org/concepts/shutdown.md). - **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 [`UpdateType`](https://carbon.sevenlabs.org/concepts/updates.md#updatetype)s 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](https://carbon.sevenlabs.org/concepts/filters.md) read it from `FilterContext` to route updates by source. | Constructor | Gives | Use it when | | ------------------------------------ | --------------------------------------------------------------- | ---------------------------------------------------------- | | `DatasourceId::new_unique()` | A random UUID v4. `.datasource(ds)` calls this for you. | You never need to refer to the source. | | `DatasourceId::new_named("primary")` | The ID `"primary"`. Pass it with `.datasource_with_id(ds, id)`. | You filter by source, for example with `DatasourceFilter`. | 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. ```rust 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](https://carbon.sevenlabs.org/guides/multiple-datasources.md) 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 | Datasource | Family | Latency | History | You need | Update types | | ---------------------------------------------------------------------------------------------- | ------------- | ------------------------------------ | -------------------------------------------------------------- | -------------------------------------------------- | ------------------------------------- | | [Yellowstone gRPC](https://carbon.sevenlabs.org/datasources/yellowstone-grpc.md) | Geyser gRPC | Real time | From now on | A Yellowstone gRPC endpoint | Account, Transaction, AccountDeletion | | [Helius LaserStream](https://carbon.sevenlabs.org/datasources/helius-laserstream.md) | Geyser gRPC | Real time | Optional replay from a recent `from_slot` | A Helius plan with LaserStream | Account, Transaction, AccountDeletion | | [RPC block subscribe](https://carbon.sevenlabs.org/datasources/rpc-block-subscribe.md) | RPC websocket | One block at a time | From now on | An RPC websocket with `blockSubscribe` enabled | Transaction, BlockDetails | | [RPC program subscribe](https://carbon.sevenlabs.org/datasources/rpc-program-subscribe.md) | RPC websocket | Real time | From now on | Any RPC websocket, including the public one | Account, AccountDeletion | | [Helius Atlas websockets](https://carbon.sevenlabs.org/datasources/helius-atlas-ws.md) | Provider API | Real time | From now on | A Helius API key | Account, Transaction, AccountDeletion | | [RPC transaction crawler](https://carbon.sevenlabs.org/datasources/rpc-transaction-crawler.md) | RPC crawler | Seconds (polls every 5 s by default) | An address's full history, newest first, then new transactions | Any RPC, including the public one | Transaction | | [RPC block crawler](https://carbon.sevenlabs.org/datasources/rpc-block-crawler.md) | RPC crawler | Seconds | A slot range, or from a slot onward | An RPC node that serves `getBlock` for those slots | Transaction | | [Helius GTFA](https://carbon.sevenlabs.org/datasources/helius-gtfa.md) | Provider API | One pass, then stops | An address's history, with filters and sorting | A Helius API key | Transaction | | [RPC getProgramAccounts](https://carbon.sevenlabs.org/datasources/rpc-gpa.md) | State loader | One-shot | Current state only | Any RPC; the public one with filters | Account | | [Helius GPA v2](https://carbon.sevenlabs.org/datasources/helius-gpa-v2.md) | Provider API | One-shot | Current state, or changes since a slot | A Helius API key | Account | | [Validator snapshot](https://carbon.sevenlabs.org/datasources/validator-snapshot.md) | State loader | One-shot | State at the snapshot's slot | A local snapshot or a download URL, and disk space | Account | | [Stream message](https://carbon.sevenlabs.org/datasources/stream-message.md) | Adapter | Whatever you feed it | Whatever you feed it | Your own producer | Account, Transaction, AccountDeletion | Common combinations: - **Real-time indexer**: a Geyser stream, plus a second one with a [`DatasourceFilter`](https://carbon.sevenlabs.org/concepts/filters.md#datasourcefilter) or [`DeduplicationFilter`](https://carbon.sevenlabs.org/concepts/filters.md#deduplicationfilter) for redundancy. - **Backfill then follow**: the transaction or block crawler for history, then a stream for new data. See [Backfill transactions](https://carbon.sevenlabs.org/guides/backfill-transactions.md). - **Current state then follow**: `getProgramAccounts` or a snapshot to load accounts, then program subscribe or a Geyser stream for changes. See [Load account state](https://carbon.sevenlabs.org/guides/load-account-state.md). The [datasource index](https://carbon.sevenlabs.org/datasources.md) helps you choose. If none fits, [write your own](https://carbon.sevenlabs.org/guides/custom-datasource.md). ## 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. ```sh 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 ``` `src/main.rs`: ```rust 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 { vec![UpdateType::AccountUpdate, UpdateType::AccountDeletion] } } struct AnyAccount; impl<'a> AccountDecoder<'a> for AnyAccount { type AccountType = (); fn decode_account(&self, account: &'a Account) -> Option> { Some(DecodedAccount { lamports: account.lamports, data: (), owner: account.owner, executable: account.executable, rent_epoch: account.rent_epoch, }) } } struct PrintAccount; impl Processor> 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 } ``` ```sh cargo run ``` Output: ```text 4vJ9JU1bJJE96FWSJKvHsmmFADCg4gpZQff4P3bkLKi has 1000000 lamports 8qbHbw2BbbTHBW1sbeqakYXVKRQM8Ne7pLK7m6CVfeR has 2000000 lamports ``` [Write a custom datasource](https://carbon.sevenlabs.org/guides/custom-datasource.md) builds a networked one with reconnects. ## Next steps - [Choose a datasource](https://carbon.sevenlabs.org/datasources.md): Configuration for every published datasource. - [Decoders](https://carbon.sevenlabs.org/concepts/decoders.md): How raw accounts and instructions become Rust types. - [Use multiple datasources](https://carbon.sevenlabs.org/guides/multiple-datasources.md): Run redundant or complementary sources in one pipeline. --- Carbon 2.0.0 documentation · Web: https://carbon.sevenlabs.org/concepts/datasources · Last updated 2026-11-09 · Index of all pages: https://carbon.sevenlabs.org/llms.txt