# Yellowstone gRPC (Carbon 2.0) > Stream accounts, transactions and blocks from any Yellowstone (Dragon's Mouth) Geyser gRPC endpoint with the lowest latency Carbon offers. `YellowstoneGrpcGeyserClient` opens one Yellowstone Geyser gRPC subscription and turns its account, transaction and block messages into Carbon updates. Use it for production real-time indexing when you have a Yellowstone endpoint: your own validator with the plugin, or a provider such as Triton, Helius, QuickNode or Shyft. For Helius LaserStream with slot replay, use [Helius LaserStream](https://carbon.sevenlabs.org/datasources/helius-laserstream.md) instead. | | | | -------- | -------------------------------------------------------------------------------------------------------------------------------------------------- | | Crate | `carbon-yellowstone-grpc-datasource` 2.0.0 | | Install | `cargo add carbon-yellowstone-grpc-datasource@2.0.0` | | Emits | `AccountUpdate`, `AccountDeletion`, `Transaction` | | Mode | Live. Runs until cancelled. | | Provider | A Yellowstone gRPC endpoint and, usually, an `x-token`. Paid with most providers. | | API docs | [docs.rs](https://docs.rs/carbon-yellowstone-grpc-datasource/2.0.0/carbon_yellowstone_grpc_datasource/) | | Source | [`datasources/yellowstone-grpc-datasource`](https://github.com/sevenlabs-hq/carbon/blob/v2.0.0/datasources/yellowstone-grpc-datasource/src/lib.rs) | ## Quick example This program subscribes to confirmed, successful, non-vote transactions that include the Orca Whirlpool program and prints each `swapV2`. It compiles as shown; to run it, set `GEYSER_URL` (for example `https://your-endpoint:443`) and `X_TOKEN`. ```sh cargo new yellowstone && cd yellowstone cargo add carbon-core@2.0.0 carbon-log-metrics@2.0.0 carbon-orca-whirlpool-decoder@2.0.0 carbon-yellowstone-grpc-datasource@2.0.0 cargo add yellowstone-grpc-proto@=12.7.0 cargo add tokio@1 --features full cargo add env_logger@0.11 log@0.4 ``` `src/main.rs`: ```rust use { carbon_core::{ error::CarbonResult, instruction::InstructionProcessorInputType, pipeline::Pipeline, processor::Processor, }, carbon_log_metrics::LogMetrics, carbon_orca_whirlpool_decoder::{ instructions::OrcaWhirlpoolInstruction, OrcaWhirlpoolDecoder, PROGRAM_ID, }, carbon_yellowstone_grpc_datasource::{ BlockFilters, YellowstoneGrpcClientConfig, YellowstoneGrpcGeyserClient, }, std::{collections::HashMap, env, sync::Arc}, yellowstone_grpc_proto::geyser::{CommitmentLevel, SubscribeRequestFilterTransactions}, }; #[tokio::main] async fn main() -> CarbonResult<()> { env_logger::init(); let mut transaction_filters = HashMap::new(); transaction_filters.insert( "orca_whirlpool".to_string(), SubscribeRequestFilterTransactions { vote: Some(false), failed: Some(false), account_include: vec![PROGRAM_ID.to_string()], ..Default::default() }, ); let datasource = YellowstoneGrpcGeyserClient::new( env::var("GEYSER_URL").expect("GEYSER_URL must be set"), env::var("X_TOKEN").ok(), Some(CommitmentLevel::Confirmed), HashMap::new(), // account filters transaction_filters, BlockFilters::default(), YellowstoneGrpcClientConfig::default(), None, // disconnect notifier None, // stream timeout: 30 s ); Pipeline::builder() .datasource(datasource) .metrics(Arc::new(LogMetrics::new())) .instruction(OrcaWhirlpoolDecoder, SwapLogger) .build()? .run() .await } struct SwapLogger; impl Processor> for SwapLogger { async fn process( &mut self, input: &InstructionProcessorInputType<'_, OrcaWhirlpoolInstruction>, ) -> CarbonResult<()> { let tx = &input.metadata.transaction_metadata; if let OrcaWhirlpoolInstruction::SwapV2 { data, accounts, .. } = input.decoded_instruction { println!("slot {} swapV2 pool={} amount={}", tx.slot, accounts.whirlpool, data.amount); } Ok(()) } } ``` ```sh GEYSER_URL=https://your-yellowstone-endpoint:443 X_TOKEN=your-x-token RUST_LOG=info cargo run ``` Each matching swap prints one `slot … swapV2 pool=… amount=…` line as its slot is confirmed. For a step-by-step walkthrough, see [Stream with Yellowstone](https://carbon.sevenlabs.org/guides/stream-with-yellowstone.md). ## Configuration ### `YellowstoneGrpcGeyserClient::new` ```rust pub fn new( endpoint: String, x_token: Option, commitment: Option, account_filters: HashMap, transaction_filters: HashMap, block_filters: BlockFilters, geyser_config: YellowstoneGrpcClientConfig, disconnect_notifier: Option>, stream_timeout: Option, ) -> Self ``` | Parameter | Type | Default | Meaning | | --------------------- | ----------------------------------------------------- | ------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `endpoint` | `String` | — | gRPC URL, including scheme and port. | | `x_token` | `Option` | — | Sent as the `x-token` header. `None` for endpoints without auth. | | `commitment` | `Option` | — | `Processed`, `Confirmed` or `Finalized`. `None` leaves it to the server (Yellowstone defaults to `Processed`, which can include transactions from forks that are later dropped). | | `account_filters` | `HashMap` | — | Named account filters. The key is a label of your choice. Empty: no account stream. | | `transaction_filters` | `HashMap` | — | Named transaction filters. Empty: no transaction stream. | | `block_filters` | `BlockFilters` | `BlockFilters::default()` | Block subscriptions. See below. | | `geyser_config` | `YellowstoneGrpcClientConfig` | `default()` | Connection settings. See below. | | `disconnect_notifier` | `Option>` | `None` | Receives one event per reconnect with the slot gap. | | `stream_timeout` | `Option` | 30 s | With no message for this long, the stream is treated as dead and reopened. | Filter types come from `yellowstone_grpc_proto::geyser` (`yellowstone-grpc-proto` 12.7.0). They are prost messages, so `..Default::default()` fills the fields you don't set. `SubscribeRequestFilterTransactions`: | Field | Type | Meaning | | ------------------------------------------ | ---------------- | ----------------------------------------------------------------------------------------------------------------- | | `vote` | `Option` | `Some(false)` excludes vote transactions, `Some(true)` keeps only votes. | | `failed` | `Option` | `Some(false)` excludes failed transactions. Carbon does not drop failed transactions from this stream on its own. | | `signature` | `Option` | A single transaction signature. | | `account_include` | `Vec` | Match transactions that use any of these addresses. | | `account_exclude` | `Vec` | Drop transactions that use any of these addresses. | | `account_required` | `Vec` | Match only transactions that use all of these addresses. | | `cuckoo_account_include`, `token_accounts` | | Provider extensions; see your provider's docs. | `SubscribeRequestFilterAccounts`: | Field | Type | Meaning | | ------------------------ | ------------------------------------------- | ------------------------------------------------------------------------------------------------- | | `account` | `Vec` | Specific account addresses. | | `owner` | `Vec` | Accounts owned by these programs. | | `filters` | `Vec` | `Filter::Memcmp`, `Filter::Datasize(u64)`, `Filter::TokenAccountState(bool)`, `Filter::Lamports`. | | `nonempty_txn_signature` | `Option` | Only updates caused by a transaction. | For example, to stream every Orca Whirlpool pool account: ```rust use yellowstone_grpc_proto::geyser::{ subscribe_request_filter_accounts_filter::Filter, SubscribeRequestFilterAccounts, SubscribeRequestFilterAccountsFilter, }; let mut account_filters = HashMap::new(); account_filters.insert( "whirlpools".to_string(), SubscribeRequestFilterAccounts { owner: vec![PROGRAM_ID.to_string()], filters: vec![SubscribeRequestFilterAccountsFilter { filter: Some(Filter::Datasize(653)), }], ..Default::default() }, ); ``` ### `BlockFilters` | Field | Type | Default | Meaning | | --------------------- | ----------------------------------------------- | ------------- | ------------------------------------------------------------------------------------------------------ | | `filters` | `HashMap` | empty | Named block filters: `account_include`, `include_transactions`, `include_accounts`, `include_entries`. | | `failed_transactions` | `Option` | `None` (keep) | `Some(false)` drops failed transactions from block messages. | ### `YellowstoneGrpcClientConfig` ```rust pub const fn new( compression: Option, connect_timeout: Option, timeout: Option, max_decoding_message_size: Option, tls_config: Option, tcp_nodelay: Option, ) -> Self pub fn with_reconnect(self, reconnect: ReconnectConfig) -> Self ``` | Field | Type | Default | Meaning | | --------------------------- | ----------------------------- | ------------------------------------------------ | -------------------------------------------------------------------------------------------------------------------------- | | `compression` | `Option` | `None` | Send and accept compressed messages (`tonic::codec::CompressionEncoding`, re-exported by `yellowstone_grpc_proto::tonic`). | | `connect_timeout` | `Option` | 15 s | Connection timeout. | | `timeout` | `Option` | 15 s | Request timeout. | | `max_decoding_message_size` | `Option`, bytes | `None`: tonic's limit of 4 MiB (4,194,304 bytes) | Largest message the client accepts. Raise it when large blocks or accounts fail; see below. | | `tls_config` | `Option` | `None` (system and webpki roots) | Custom TLS configuration. | | `tcp_nodelay` | `Option` | `None` | Set `TCP_NODELAY`. | | `reconnect` | `Option` | `None` | Reconnect inside the client library and replay missed data. Set with `with_reconnect`. | A message over `max_decoding_message_size` ends the stream with tonic's `OutOfRange` status, which the datasource logs as a stream error before resubscribing. The status message has this form, with your sizes: ```text Error, decoded message length too large: found 5242880 bytes, the limit is: 4194304 bytes ``` Block subscriptions and large program accounts are the usual causes. Raise the limit when you build the config: ```rust // Accept messages up to 64 MiB instead of tonic's 4 MiB default. let config = YellowstoneGrpcClientConfig::new(None, None, None, Some(64 * 1024 * 1024), None, None); ``` ### Reconnects Without any configuration, the datasource resubscribes on its own after a stream error and never gives up, but data sent during the gap is lost (see [Behaviour](#behaviour)). To have the client library also replay what you missed, opt in with `with_reconnect`. `ReconnectConfig` comes from `yellowstone-grpc-client`, which you add with `cargo add yellowstone-grpc-client@=13.5.0`. Its default retries 3 times with a 10 ms backoff and recovers missed messages for up to 250 slots: ```rust use yellowstone_grpc_client::ReconnectConfig; let geyser_config = YellowstoneGrpcClientConfig::default().with_reconnect(ReconnectConfig::default()); ``` ### Detect gaps with `DatasourceDisconnection` With a `disconnect_notifier`, the datasource sends `{ source: "yellowstone-grpc", disconnect_time, last_slot_before_disconnect, first_slot_after_reconnect, missed_slots }` on the first message after a reconnect. Read it in your own task: ```rust let (disconnect_tx, mut disconnect_rx) = tokio::sync::mpsc::channel::(16); tokio::spawn(async move { while let Some(gap) = disconnect_rx.recv().await { log::warn!( "missed {} slots ({}..{})", gap.missed_slots, gap.last_slot_before_disconnect, gap.first_slot_after_reconnect ); } }); // pass Some(disconnect_tx) as `disconnect_notifier` ``` ## Behaviour - **Connecting.** `consume` connects before spawning the stream task. If the first connection fails, it returns `FailedToConsumeDatasource("gRPC transport error: …")`; a pipeline with only this datasource then exits. - **Subscription.** One `SubscribeRequest` with your account, transaction and block filters and commitment. When the server sends a ping, the datasource answers by resending the request with a ping. - **Account messages** become `AccountUpdate` (or `AccountDeletion` at zero lamports) with the message's slot and the causing `transaction_signature`. - **Transaction messages** become `Transaction` updates with `is_vote` and `index` from Geyser, and `block_time`/`block_hash` set to `None`. Failed and vote transactions pass through unless your filter excludes them. - **Block messages** produce a `Transaction` update for each transaction (with the block's `block_time`) and an account update for each account in the block. No `BlockDetails` updates are emitted. - **Ordering.** As delivered by the server: slot order for a given commitment, but accounts and transactions from different filters interleave. - **Reconnects.** When the stream errors, closes or is silent for `stream_timeout`, the datasource subscribes again on the same client right away. If subscribing fails, it backs off from 100 ms, doubling up to 3 s, and never gives up. Without `reconnect`, data during the gap is lost; `disconnect_notifier` tells you which slots. - **Backpressure.** Every update uses `try_send`. When the pipeline channel is full the update is dropped with `Failed to send transaction update …` or `Failed to send account event …`. Size the pipeline's [`channel_buffer_size`](https://carbon.sevenlabs.org/reference/pipeline-builder.md) for your peak rate. - **Finishing.** Never finishes on its own. - **Metrics.** Counters `yellowstone_grpc_account_updates_received_total`, `yellowstone_grpc_account_deletions_received_total`, `yellowstone_grpc_transaction_updates_received_total`; histograms `yellowstone_grpc_account_process_time_nanoseconds`, `yellowstone_grpc_account_deletion_process_time_nanoseconds`, `yellowstone_grpc_transaction_process_time_nanoseconds`. ## Troubleshooting ### `error consuming datasource: FailedToConsumeDatasource("gRPC transport error: transport error")` The first connection failed and the pipeline exited. Check the URL (scheme and port, for example `https://host:443`), that the host is reachable, and the provider's IP allowlist. ### The log fills with `Geyser stream error: Status { … }` The server rejects the subscription; the `Status` code and message say why. A missing or wrong token is the most common cause: pass it as `x_token`. A filter the provider doesn't allow (too many accounts, an unsupported field) is the next. The datasource resubscribes immediately after each stream error, so a persistent problem produces many log lines per second. ### `Geyser stream error … invalid compression flag … status: 501 Not Implemented` The URL points at something that isn't a Yellowstone gRPC server, such as a JSON-RPC endpoint. Use the provider's gRPC URL. ### Panic: `Could not automatically determine the process-level CryptoProvider` Two `rustls` crypto backends ended up in your dependency tree. Add `rustls` (`cargo add rustls@0.23 --no-default-features --features ring`) and install one provider at the top of `main`, before building the pipeline: ```rust rustls::crypto::ring::default_provider() .install_default() .expect("failed to install rustls crypto provider"); ``` ### Updates are missing under load Look for `Failed to send transaction update` in the logs. Raise `Pipeline::builder().channel_buffer_size(…)`, make processors faster, or narrow your filters. ### Transactions that later disappear With `CommitmentLevel::Processed` (or `None`) you also receive transactions from forks that never confirm. Use `Confirmed` unless you handle rollbacks. ## See also - [Stream with Yellowstone](https://carbon.sevenlabs.org/guides/stream-with-yellowstone.md) - [Helius LaserStream](https://carbon.sevenlabs.org/datasources/helius-laserstream.md): same protocol, with slot replay. - [Production](https://carbon.sevenlabs.org/guides/production.md) and [Prometheus](https://carbon.sevenlabs.org/guides/prometheus.md) - [Datasources concept](https://carbon.sevenlabs.org/concepts/datasources.md) and [update types](https://carbon.sevenlabs.org/concepts/updates.md) --- Carbon 2.0.0 documentation · Web: https://carbon.sevenlabs.org/datasources/yellowstone-grpc · Last updated 2026-11-09 · Index of all pages: https://carbon.sevenlabs.org/llms.txt