Yellowstone gRPC

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 instead.

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.

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

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<InstructionProcessorInputType<'_, OrcaWhirlpoolInstruction>> 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(())
    }
}

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.

Configuration#

YellowstoneGrpcGeyserClient::new#

pub fn new(
    endpoint: String,
    x_token: Option<String>,
    commitment: Option<CommitmentLevel>,
    account_filters: HashMap<String, SubscribeRequestFilterAccounts>,
    transaction_filters: HashMap<String, SubscribeRequestFilterTransactions>,
    block_filters: BlockFilters,
    geyser_config: YellowstoneGrpcClientConfig,
    disconnect_notifier: Option<mpsc::Sender<DatasourceDisconnection>>,
    stream_timeout: Option<Duration>,
) -> Self

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:

SubscribeRequestFilterAccounts:

For example, to stream every Orca Whirlpool pool account:

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#

YellowstoneGrpcClientConfig#

pub const fn new(
    compression: Option<CompressionEncoding>,
    connect_timeout: Option<Duration>,
    timeout: Option<Duration>,
    max_decoding_message_size: Option<usize>,
    tls_config: Option<ClientTlsConfig>,
    tcp_nodelay: Option<bool>,
) -> Self
pub fn with_reconnect(self, reconnect: ReconnectConfig) -> Self

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:

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:

// 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). 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:

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:

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

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
• Helius LaserStream: same protocol, with slot replay.
• Production and Prometheus
• Datasources concept and update types