# Stream with Yellowstone gRPC (Carbon 2.0) > Index transactions and account changes in real time from a Yellowstone (Dragon's Mouth) Geyser gRPC endpoint, with filters, commitment and reconnects. Yellowstone gRPC streams transactions and account writes from a validator's Geyser plugin as they happen. It's the lowest-latency way to feed a Carbon pipeline. This guide builds an indexer that decodes Meteora DLMM swaps from a Yellowstone stream, then covers filters, commitment and reconnect handling. **You need:** - Rust 1.96.1 or newer. - A Yellowstone gRPC endpoint URL and, for most providers, an access token (sent as the `x-token` header). Most Solana RPC providers sell Yellowstone (also called Dragon's Mouth) access; you can also run the plugin on your own node. The public Solana endpoint doesn't offer it. ### Step 1: Create a project ```sh cargo new dlmm-stream cd dlmm-stream ``` ### Step 2: Add the dependencies `Cargo.toml`: ```toml [package] name = "dlmm-stream" version = "0.1.0" edition = "2024" [dependencies] carbon-core = "2.0.0" carbon-log-metrics = "2.0.0" carbon-meteora-dlmm-decoder = "2.0.0" carbon-yellowstone-grpc-datasource = "2.0.0" env_logger = "0.11.11" log = "0.4.35" rustls = { version = "0.23", default-features = false, features = ["std", "ring"] } tokio = { version = "1.54.1", features = ["full"] } yellowstone-grpc-proto = { version = "=12.7.0", default-features = false } ``` You build the subscription filters with types from `yellowstone-grpc-proto`, so pin it to `=12.7.0`, the version `carbon-yellowstone-grpc-datasource` 2.0.0 uses. A different version compiles a second copy of the types and the constructor rejects them. `rustls` is there to install a TLS crypto provider; see the next step. ### Step 3: Write the pipeline `src/main.rs`: ```rust use { carbon_core::{ error::CarbonResult, instruction::InstructionProcessorInputType, pipeline::Pipeline, processor::Processor, }, carbon_log_metrics::LogMetrics, carbon_meteora_dlmm_decoder::{ instructions::MeteoraDlmmInstruction, MeteoraDlmmDecoder, 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(); // The gRPC client uses rustls; pick its crypto provider before connecting. rustls::crypto::ring::default_provider() .install_default() .expect("failed to install the rustls crypto provider"); // Successful, non-vote transactions that invoke Meteora DLMM. 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() }, ); 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 (default 30 s) ); Pipeline::builder() .datasource(datasource) .metrics(Arc::new(LogMetrics::new())) .instruction(MeteoraDlmmDecoder, SwapLogger) .build()? .run() .await } struct SwapLogger; impl Processor> 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!("slot {} swap {}: amount_in={}", meta.slot, meta.signature, data.amount_in) } MeteoraDlmmInstruction::Swap2 { data, .. } => { println!("slot {} swap2 {}: amount_in={}", meta.slot, meta.signature, data.amount_in) } _ => {} } Ok(()) } } ``` The constructor's arguments, in order: endpoint, token, commitment, account filters, transaction filters, block filters, client config, a channel for disconnect events, and the stream timeout. ### Step 4: Run it Pass your provider's endpoint and token. Use the full URL with its scheme and port, as your provider gives it: ```sh GEYSER_URL="https://your-yellowstone-endpoint:443" X_TOKEN="your-x-token" cargo run ``` Each decoded swap prints one line, a second or so after the transaction lands: Output: ```text slot 467110307 swap ue82HW7rk3DcWiCTzRrnMzMeKDYDLWUiyeMBtMi4tLtqiKaRGgbr7EYigU9RbpftsFsgEC7P6kgxkNWJgZhvKZs: amount_in=10298583 slot 467110307 swap 48kbDBkFMweb1bVYk9o3uGSR9CQS5pkgm8YCGB9Qg9ymui5dXmwjjoiSAHfF4eQ8mt6j688pkrG8n33GnZX4cffd: amount_in=1880866 slot 467110307 swap2 2d6GNQzSzxgV85mnzD4ed6JyxdLYdwBtyCH8LujdYaxySSKfD8x96yJLtxmuVYt6XGRjy8m3AutxyxqiumscEb4W: amount_in=381756 slot 467110308 swap2 2miT4n9yFXZMdUTVF4My1RVvHrbbZMwD64vYyppzdkxNg5qVtYJgsm4dL8MPYtBRrJm2V3EzP4ZfYJKjEc61Wzr1: amount_in=28328938705735 … ``` The stream runs until you press `Ctrl+C`. ## Verify it's streaming Run with `RUST_LOG=info`. Every five seconds `LogMetrics` logs the pipeline counters, and the datasource's own counter, `yellowstone_grpc_transaction_updates_received_total`, should keep growing: ```sh RUST_LOG=info GEYSER_URL="https://your-yellowstone-endpoint:443" X_TOKEN="your-x-token" cargo run ``` If the counter stays at 0, the subscription is accepted but nothing matches your filters. If you see `Geyser stream error` lines instead, see [Common problems](#common-problems). ## Filter transactions The transaction filter map's keys are labels of your choice; the server sends a transaction if it matches any entry. Within one `SubscribeRequestFilterTransactions`, all the fields you set must match: | Field | Matches | | ------------------ | --------------------------------------------------------------------------------------------------- | | `vote` | `Some(false)` excludes vote transactions, `Some(true)` sends only votes, `None` sends both. | | `failed` | `Some(false)` excludes failed transactions, `Some(true)` sends only failed ones, `None` sends both. | | `account_include` | Transactions that mention **any** of these accounts or programs. | | `account_required` | Transactions that mention **all** of these accounts or programs. | | `account_exclude` | Drops transactions that mention any of these. | | `signature` | One transaction by signature. | "Mention" includes accounts loaded through address lookup tables, so it catches a program called through a CPI, such as a Jupiter route through a DLMM pool. Carbon then decodes nested instructions too (see [Instructions](https://carbon.sevenlabs.org/concepts/instructions.md)). To follow several programs with one subscription, list them in `account_include`: see [Multiple programs](https://carbon.sevenlabs.org/guides/multiple-programs.md#choose-a-datasource-filter). Transaction updates carry the slot and the transaction's index in the block, but no block time: `transaction_metadata.block_time` is `None` for them. ## Stream account changes Account filters send an account's new state every time a transaction writes to it. This filter streams every Meteora DLMM `LbPair` account (the pools), selected by owner and by the account's 8-byte discriminator: ```rust use yellowstone_grpc_proto::geyser::{ subscribe_request_filter_accounts_filter::Filter as AccountsFilter, subscribe_request_filter_accounts_filter_memcmp::Data as MemcmpData, SubscribeRequestFilterAccounts, SubscribeRequestFilterAccountsFilter, SubscribeRequestFilterAccountsFilterMemcmp, }; let mut account_filters = HashMap::new(); account_filters.insert( "dlmm_pairs".to_string(), SubscribeRequestFilterAccounts { owner: vec![PROGRAM_ID.to_string()], filters: vec![SubscribeRequestFilterAccountsFilter { filter: Some(AccountsFilter::Memcmp(SubscribeRequestFilterAccountsFilterMemcmp { offset: 0, data: Some(MemcmpData::Bytes(vec![33, 11, 49, 98, 181, 101, 177, 13])), })), }], ..Default::default() }, ); ``` Pass `account_filters` as the fourth argument and register an account pipe, `.account(MeteoraDlmmDecoder, processor)`, whose processor takes `AccountProcessorInputType<'_, MeteoraDlmmAccount>`. `account` selects accounts by address, `owner` by owning program, and `filters` narrows further with `Memcmp`, `Datasize`, `Lamports` or `TokenAccountState`. Set `nonempty_txn_signature: Some(true)` to skip writes that don't come from a transaction. Account updates written by a transaction carry its signature, so a [`DeduplicationFilter`](https://carbon.sevenlabs.org/concepts/filters.md#deduplicationfilter) can drop repeats. An update with zero lamports reaches your pipeline as an [account deletion](https://carbon.sevenlabs.org/concepts/updates.md#account-deletions), so add `.account_deletions(processor)` if you track closures. ## Stream whole blocks `BlockFilters` subscribes to blocks instead. Each block's transactions and accounts become ordinary transaction and account updates, with the block's time set: ```rust use yellowstone_grpc_proto::geyser::SubscribeRequestFilterBlocks; let mut filters = HashMap::new(); filters.insert( "dlmm_blocks".to_string(), SubscribeRequestFilterBlocks { account_include: vec![PROGRAM_ID.to_string()], include_transactions: Some(true), ..Default::default() }, ); let block_filters = BlockFilters { filters, failed_transactions: Some(false) }; ``` `failed_transactions: Some(false)` drops failed transactions; `None` or `Some(true)` keeps them. Not every provider enables block subscriptions. ## Choose a commitment | `CommitmentLevel` | When updates arrive | Trade-off | | ----------------- | ------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------- | | `Processed` | As soon as the node executes the transaction | Fastest. The slot can still be skipped, so you may see transactions that never finalize. | | `Confirmed` | After a supermajority of stake voted on the block, usually under a second later | Rarely rolled back. A good default for indexers. | | `Finalized` | When 31 more blocks are confirmed on top, roughly 13 seconds later | Never rolled back. | `None` uses the server's default. ## Handle disconnects By default, when the stream ends, errors or goes silent for `stream_timeout` (30 seconds unless you pass another `Duration`), the datasource subscribes again with the same filters. A failed subscribe attempt is retried after 100 ms, doubling up to 3 s. Updates produced while disconnected are lost. To find out about gaps, pass a channel as the disconnect notifier. After each reconnect the datasource sends a `DatasourceDisconnection` with the last slot before and the first slot after the outage: ```rust use {carbon_core::datasource::DatasourceDisconnection, tokio::sync::mpsc}; let (disconnect_tx, mut disconnect_rx) = mpsc::channel::(16); tokio::spawn(async move { while let Some(gap) = disconnect_rx.recv().await { log::warn!( "missed slots {}..{} ({} slots)", gap.last_slot_before_disconnect, gap.first_slot_after_reconnect, gap.missed_slots ); } }); // pass Some(disconnect_tx) as the eighth argument ``` Backfill the reported range with the [block crawler](https://carbon.sevenlabs.org/guides/backfill-transactions.md#crawl-a-slot-range). ### Recover missed data with the client's reconnect Carbon 2.0 can also turn on the Yellowstone client's own reconnect, which is off by default. With `ReconnectionPolicy::RecoverMissedData`, the client reconnects inside the stream, asks the server to replay from the last slot it saw, and drops the duplicates the replay produces. The server must support replay (`from_slot`) and still hold those slots. Add the client crate at the version the datasource uses: `Cargo.toml`: ```toml [dependencies] yellowstone-grpc-client = "=13.5.0" ``` ```rust use { std::time::Duration, yellowstone_grpc_client::{Backoff, ReconnectConfig, ReconnectionPolicy}, }; let config = YellowstoneGrpcClientConfig::default().with_reconnect(ReconnectConfig { backoff: Backoff::new(Duration::from_millis(100), 2.0, 5), // 5 attempts per outage policy: ReconnectionPolicy::RecoverMissedData { slot_retention: 250 }, }); // pass config as the seventh argument ``` `ReconnectConfig::default()` uses a 10 ms backoff with 3 retries and `RecoverMissedData` with 250 slots of deduplication state. `ReconnectionPolicy::SkipMissedData` reconnects at the tip without replay. If the client gives up, the stream ends and the datasource's own resubscribe loop takes over. ## Other client settings `YellowstoneGrpcClientConfig::default()` sets 15 s connect and request timeouts, TLS with the system's root certificates, and no compression. Its public fields: | Field | Default | Description | | ---------------------------- | ----------------------- | -------------------------------------------------------------------------------------------------- | | `compression` | `None` | `Some(CompressionEncoding::Gzip)` or `Zstd` compresses both directions, if the server supports it. | | `connect_timeout`, `timeout` | 15 s | Connection and request timeouts. | | `max_decoding_message_size` | tonic's default (4 MiB) | Raise it if large blocks or accounts fail with a message-size error. | | `tls_config` | system roots | A custom `ClientTlsConfig`. | | `tcp_nodelay` | `None` | Sets `TCP_NODELAY`. | | `reconnect` | `None` | See above. | ## Keep up with the stream The datasource never waits for the pipeline: if the pipeline's channel is full, the update is dropped and logged as `Failed to send transaction update …` or `Failed to send account event …`. The channel holds 1,000 updates by default. If you see those lines, make processors faster (move slow I/O to a background task) and raise the buffer: ```rust Pipeline::builder() .datasource(datasource) .channel_buffer_size(10_000) ``` ## Common problems ### Panic: `Could not automatically determine the process-level CryptoProvider from Rustls crate features` The gRPC client uses rustls, and no crypto provider was selected: either none is enabled, or both `ring` and `aws-lc-rs` are. Add the `rustls` dependency from the steps above and call `rustls::crypto::ring::default_provider().install_default()` at the top of `main`. ### Pipeline exits immediately with `error consuming datasource: FailedToConsumeDatasource("gRPC transport error: transport error")` The client couldn't connect: the host or port is wrong, the URL has no `https://` scheme, or a firewall blocks the port. The pipeline stops when its only datasource fails, and the process exits with status 0. Check the URL with your provider. ### `Geyser stream error: Status { code: Unauthenticated, … }` repeated The token is missing or wrong. Set `X_TOKEN`. Some providers put the token in the URL instead; then leave `X_TOKEN` unset. ### Many `Geyser stream error` lines per second The connection works but the server rejects the subscription, for example `405 Method Not Allowed` when the URL points at an HTTP RPC endpoint instead of the gRPC one, or `PermissionDenied` when your plan doesn't include a filter you used. The datasource resubscribes immediately after a stream error, so it repeats quickly. Fix the endpoint or the filter. ### `mismatched types` errors on `SubscribeRequestFilterTransactions` or `CommitmentLevel` Your `yellowstone-grpc-proto` isn't `=12.7.0`, so two copies of the types are in the build. Pin the version shown in the steps and run `cargo update`. ### Nothing prints No transaction matched. `account_required` needs every listed key in the transaction; use `account_include` for "any of". `failed: Some(false)` and `vote: Some(false)` drop those transactions. Also check that the pipe's decoder matches the program you subscribed to. ## Next steps - [Stream with Helius LaserStream](https://carbon.sevenlabs.org/guides/stream-with-laserstream.md): The same stream with replay from a past slot. - [Multiple datasources](https://carbon.sevenlabs.org/guides/multiple-datasources.md): Two streams for redundancy, deduplicated. - [Backfill transactions](https://carbon.sevenlabs.org/guides/backfill-transactions.md): Fill history and disconnect gaps. - [Yellowstone gRPC datasource](https://carbon.sevenlabs.org/datasources/yellowstone-grpc.md): The datasource's full reference. --- Carbon 2.0.0 documentation · Web: https://carbon.sevenlabs.org/guides/stream-with-yellowstone · Last updated 2026-11-09 · Index of all pages: https://carbon.sevenlabs.org/llms.txt