ERPC

Geyser gRPC - Code Examples

Practical Rust examples for common Geyser gRPC use cases. Fill in your Geyser gRPC endpoint. Solana Stream SDK supports Geyser gRPC.

Basic subscription: all non-vote, non-failed transactions

This is intentionally a broad transaction subscription. Excluding failed transactions reduces traffic, but the remaining stream can still be substantial. Add the account, signature, or transaction filters that match the work your consumer actually needs, and keep processing outside the receive loop.
rust
use anyhow::Result;
use futures::{sink::SinkExt, stream::StreamExt};
use solana_signature::Signature;
use std::collections::HashMap;
use solana_stream_sdk::{
    GeyserCommitmentLevel as CommitmentLevel, GeyserGrpcClient,
    GeyserSubscribeRequest as SubscribeRequest,
    GeyserSubscribeRequestFilterTransactions as SubscribeRequestFilterTransactions,
    GeyserUpdateOneof as UpdateOneof,
};

#[tokio::main]
async fn main() -> Result<()> {
    let endpoint = "http://your-erpc-grpc-endpoint"; // Set your ERPC gRPC URL

    let mut client = GeyserGrpcClient::build_from_shared(endpoint)?
        .connect()
        .await?;

    let (mut tx, mut stream) = client.subscribe().await?;

    tx.send(SubscribeRequest {
        transactions: HashMap::from([(
            "all_transactions".to_string(),
            SubscribeRequestFilterTransactions {
                vote: Some(false),   // Drop vote txs
                failed: Some(false), // Drop failed txs
                ..Default::default()
            },
        )]),
        commitment: Some(CommitmentLevel::Confirmed as i32),
        ..Default::default()
    })
    .await?;

    while let Some(Ok(msg)) = stream.next().await {
        if let Some(UpdateOneof::Transaction(tx)) = msg.update_oneof {
            let sig = tx
                .transaction
                .and_then(|t| Signature::try_from(t.signature.as_slice()).ok())
                .unwrap_or_default();
            println!("Slot: {} | Signature: {}", tx.slot, sig);
        }
    }
    Ok(())
}

Durable client with bounded reconnect after clean closure, gap recovery, ping/pong

rust
use anyhow::Result;
use futures::stream::StreamExt;
use std::{collections::HashMap, time::Duration};
use solana_stream_sdk::{
    GeyserCommitmentLevel as CommitmentLevel, GeyserGrpcClient,
    GeyserSlotStatus as SlotStatus, GeyserSubscribeRequest as SubscribeRequest,
    GeyserSubscribeRequestFilterSlots as SubscribeRequestFilterSlots,
    GeyserUpdateOneof as UpdateOneof,
};

#[tokio::main]
async fn main() -> Result<()> {
    let endpoint = "http://your-erpc-grpc-endpoint";

    let mut subscribe_request = SubscribeRequest {
        slots: HashMap::from([(
            "slots".to_string(),
            SubscribeRequestFilterSlots::default(),
        )]),
        commitment: Some(CommitmentLevel::Confirmed as i32),
        ..Default::default()
    };

    let mut tracked_slot: u64 = 0;
    let mut retry_delay = Duration::from_secs(1);
    let max_retry_delay = Duration::from_secs(30);
    let mut first_diagnostic: Option<String> = None;

    loop {
        let mut client = match GeyserGrpcClient::build_from_shared(endpoint)?.connect().await {
            Ok(client) => client,
            Err(error) => {
                let current_diagnostic = error.to_string();
                first_diagnostic.get_or_insert_with(|| current_diagnostic.clone());
                eprintln!("Connection error: {current_diagnostic}");
                tokio::time::sleep(retry_delay).await;
                retry_delay = (retry_delay * 2).min(max_retry_delay);
                continue;
            }
        };

        subscribe_request.from_slot = if tracked_slot > 0 {
            Some(tracked_slot)
        } else {
            None
        };

        let (mut _sink, mut stream) = match client
            .subscribe_with_request(Some(subscribe_request.clone()))
            .await
        {
            Ok(subscription) => subscription,
            Err(error) => {
                let current_diagnostic = error.to_string();
                first_diagnostic.get_or_insert_with(|| current_diagnostic.clone());
                eprintln!("Subscribe error: {current_diagnostic}");
                eprintln!("First diagnostic retained: {}", first_diagnostic.as_deref().unwrap());
                return Err(error.into());
            }
        };

        loop {
            match stream.next().await {
                Some(Ok(msg)) => {
                    retry_delay = Duration::from_secs(1);
                    if let Some(UpdateOneof::Slot(slot)) = msg.update_oneof {
                        tracked_slot = slot.slot;
                        println!(
                            "Slot: {} | Status: {:?}",
                            slot.slot,
                            SlotStatus::try_from(slot.status).unwrap()
                        );
                    }
                }
                Some(Err(error)) => {
                    let current_diagnostic = error.to_string();
                    first_diagnostic.get_or_insert_with(|| current_diagnostic.clone());
                    eprintln!("Stream error: {current_diagnostic}");
                    eprintln!("First diagnostic retained: {}", first_diagnostic.as_deref().unwrap());
                    return Err(error.into());
                }
                None => break,
            }
        }

        tokio::time::sleep(retry_delay).await;
        retry_delay = (retry_delay * 2).min(max_retry_delay);
    }
}
This conservative sample retries connection-establishment failures and clean stream closures only. It stops on a gRPC subscribe or stream error after logging the current diagnostic and retaining the first one. Examine the displayed ERPC correction and any server retry information before manually retrying. Production code may explicitly retry known transient failures, but it must honor a server pause and must not repeatedly send an unchanged rejected request.

Stream accounts for a mint (token accounts)

rust
SubscribeRequest {
    accounts: HashMap::from([(
        "token_accounts_by_mint".to_string(),
        SubscribeRequestFilterAccounts {
            owner: vec!["TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA".to_string()],
            filters: vec![
                SubscribeRequestFilterAccountsFilter {
                    filter: Some(Filter::TokenAccountState(true)),
                },
                SubscribeRequestFilterAccountsFilter {
                    filter: Some(Filter::Memcmp(SubscribeRequestFilterAccountsFilterMemcmp {
                        offset: 0,
                        data: Some(Data::Base58(
                            "pumpCmXqMfrsAkQ5r49WcJnRayYRqmXz6ae8H7H9Dfn".to_string(), // replace with mint
                        )),
                    })),
                },
            ],
            ..Default::default()
        },
    )]),
    commitment: Some(CommitmentLevel::Confirmed as i32),
    ..Default::default()
}

Stream transactions (successful, non-vote)

rust
SubscribeRequest {
    transactions: HashMap::from([(
        "all_transactions".to_string(),
        SubscribeRequestFilterTransactions {
            vote: Some(false),
            failed: Some(false),
            ..Default::default()
        },
    )]),
    commitment: Some(CommitmentLevel::Confirmed as i32),
    ..Default::default()
}

Production-grade client highlights

  • Durable client with reconnection and gap recovery (from_slot)
  • Ping/pong handling to keep the stream alive
  • Bounded channels/backpressure between ingress and processing
  • Dynamic subscription updates on the same bidi stream
  • Error handling, metrics, and logging for slow consumers/dropped updates

Composed client pattern (gap recovery, ping/pong, dynamic subscriptions)

This composition shows bounded ingress and dynamic subscriptions. Its ingress task waits and reconnects only after a connection-establishment failure or a clean end of stream. A gRPC subscription or stream error is reported and stops the task so its correction and retry information can be examined; it is not a generic retry policy.
rust
use futures::{SinkExt, stream::StreamExt};
use solana_signature::Signature;
use std::{collections::HashMap, sync::Arc, time::Duration};
use tokio::sync::{RwLock, mpsc};
use solana_stream_sdk::{
    GeyserCommitmentLevel as CommitmentLevel, GeyserGrpcClient, GeyserSlotStatus,
    GeyserSubscribeRequest as SubscribeRequest,
    GeyserSubscribeRequestFilterSlots as SubscribeRequestFilterSlots,
    GeyserSubscribeRequestFilterTransactions as SubscribeRequestFilterTransactions,
    GeyserSubscribeUpdate as SubscribeUpdate, GeyserUpdateOneof as UpdateOneof,
};
use solana_stream_sdk::yellowstone_grpc_proto::geyser::SubscribeRequestPing;

const UPDATE_CHANNEL_CAPACITY: usize = 10_000;

#[tokio::main]
async fn main() {
    let endpoint = "http://your-erpc-grpc-endpoint".to_string();

    let subscribe_request = Arc::new(RwLock::new(SubscribeRequest {
        slots: HashMap::from([(
            "slots".to_string(),
            SubscribeRequestFilterSlots {
                ..Default::default()
            },
        )]),
        transactions: HashMap::from([(
            "all_transactions".to_string(),
            SubscribeRequestFilterTransactions {
                vote: Some(false),
                failed: Some(false),
                ..Default::default()
            },
        )]),
        commitment: Some(CommitmentLevel::Confirmed as i32),
        ..Default::default()
    }));

    let (updates_tx, updates_rx) = mpsc::channel::<SubscribeUpdate>(UPDATE_CHANNEL_CAPACITY);
    let (subscribe_tx, subscribe_rx) = mpsc::unbounded_channel::<SubscribeRequest>();

    let dispatcher_handle = tokio::spawn(async move {
        dispatcher_loop(updates_rx).await;
    });

    let ingress_handle = tokio::spawn(async move {
        ingress_loop(
            updates_tx,
            subscribe_rx,
            endpoint,
            subscribe_request,
        )
        .await;
    });

    if let Err(e) = tokio::join!(dispatcher_handle, ingress_handle).0 {
        println!("Error: {:?}", e);
    }
}

async fn ingress_loop(
    updates_tx: mpsc::Sender<SubscribeUpdate>,
    mut subscribe_rx: mpsc::UnboundedReceiver<SubscribeRequest>,
    endpoint: String,
    subscribe_request: Arc<RwLock<SubscribeRequest>>,
) {
    let mut tracked_slot: u64 = 0;
    let mut retry_delay = Duration::from_secs(1);
    let max_retry_delay = Duration::from_secs(30);

    loop {
        let builder = match GeyserGrpcClient::build_from_shared(endpoint.clone()) {
            Ok(builder) => builder,
            Err(error) => {
                eprintln!("Client configuration error: {error}");
                return;
            }
        };

        let mut client = match builder.connect().await {
            Ok(client) => client,
            Err(error) => {
                eprintln!("Connection error: {error}");
                tokio::time::sleep(retry_delay).await;
                retry_delay = (retry_delay * 2).min(max_retry_delay);
                continue;
            }
        };

        subscribe_request.write().await.from_slot = if tracked_slot > 0 {
            Some(tracked_slot)
        } else {
            None
        };

        match client
            .subscribe_with_request(Some(subscribe_request.read().await.clone()))
            .await
        {
            Ok((mut subscribe_tx, mut stream)) => {
                loop {
                    tokio::select! {
                        subscription = subscribe_rx.recv() => {
                            match subscription {
                                Some(subscribe_request) => {
                                    if let Err(error) = subscribe_tx.send(subscribe_request).await {
                                        eprintln!("Subscription update error: {error}");
                                        return;
                                    }
                                }
                                None => return,
                            }
                        }
                        result = stream.next() => {
                            match result {
                                Some(Ok(update)) => {
                                    retry_delay = Duration::from_secs(1);
                                    if matches!(update.update_oneof, Some(UpdateOneof::Ping(_))) {
                                        if let Err(error) = subscribe_tx
                                            .send(SubscribeRequest {
                                                ping: Some(SubscribeRequestPing { id: 1 }),
                                                ..Default::default()
                                            })
                                            .await
                                        {
                                            eprintln!("Ping response error: {error}");
                                            return;
                                        }
                                    }
                                    if matches!(update.update_oneof, Some(UpdateOneof::Ping(_)) | Some(UpdateOneof::Pong(_))) {
                                        continue;
                                    }
                    if let Some(UpdateOneof::Slot(ref slot_update)) = update.update_oneof {
                        tracked_slot = slot_update.slot;
                        continue;
                    }
                                    if let Err(error) = updates_tx.send(update.clone()).await {
                                        eprintln!("Update consumer stopped: {error}");
                                        return;
                                    }
                                }
                                Some(Err(error)) => {
                                    eprintln!("Stream error: {error}");
                                    return;
                                }
                                None => {
                                    println!("Stream ended cleanly");
                                    break;
                                }
                            }
                        }
                    }
                }
            }
            Err(error) => {
                eprintln!("Subscribe error: {error}");
                return;
            }
        };

        println!("Disconnected; retrying after bounded backoff...");
        tokio::time::sleep(retry_delay).await;
        retry_delay = (retry_delay * 2).min(max_retry_delay);
    }
}

async fn dispatcher_loop(mut updates_rx: mpsc::Receiver<SubscribeUpdate>) {
    loop {
        if let Some(update) = updates_rx.recv().await {
            match update.update_oneof {
                Some(UpdateOneof::Transaction(tx)) => {
                    let sig = tx
                        .transaction
                        .and_then(|t| Signature::try_from(t.signature.as_slice()).ok())
                        .unwrap_or_default();
                    println!("Slot: {} | Signature: {}", tx.slot, sig);
                }
                _ => {
                    println!("Received update: {:?}", update);
                }
            }
        }
    }
}

Contributing examples

Found a useful pattern? Consider contributing it to these docs.
Solana Stream SDK repository: https://github.com/ValidatorsDAO/solana-stream
PreviousFiltersNextFAQ