Stream with Yellowstone gRPC

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.

cargo new dlmm-stream
cd dlmm-stream

[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 }

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<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!("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(())
    }
}

GEYSER_URL="https://your-yellowstone-endpoint:443" X_TOKEN="your-x-token" cargo run

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
…

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:

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.

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:

"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). To follow several programs with one subscription, list them in account_include: see Multiple programs.

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:

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 can drop repeats. An update with zero lamports reaches your pipeline as an account deletion, 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:

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#

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:

use {carbon_core::datasource::DatasourceDisconnection, tokio::sync::mpsc};

let (disconnect_tx, mut disconnect_rx) = mpsc::channel::<DatasourceDisconnection>(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.

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:

[dependencies]
yellowstone-grpc-client = "=13.5.0"

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:

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:

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#