ERPC

Geyser gRPC - Примеры кода

Практические Rust-примеры для типовых сценариев использования Geyser gRPC. Подставьте свой endpoint Geyser gRPC. Solana Stream SDK поддерживает Geyser gRPC.

Базовая подписка: все non-vote, non-failed transactions

Эта подписка на transactions намеренно широкая. Исключение vote и failed transactions уменьшает трафик, но оставшийся stream всё ещё может быть значительным. Добавьте filters account, signature или transaction, которые действительно нужны вашему consumer, и держите обработку вне цикла приёма.
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 (~70% of traffic)
                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 с ограниченным переподключением после чистого закрытия, 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);
    }
}
Этот консервативный пример повторяет только сбои установления соединения и чистые завершения stream. При ошибке subscription или stream gRPC он останавливается после записи текущей диагностики и сохранения первой. До ручного повтора изучите отображённую коррекцию ERPC и retry-информацию сервера. Production code может явно повторять известные временные сбои, но должен соблюдать паузу сервера и не должен многократно посылать неизменённый отклонённый запрос.

Stream accounts для 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 клиента

  • Durable client с reconnection и gap recovery (from_slot)
  • Обработка ping/pong для поддержания живого stream
  • Bounded channels/backpressure между ingress и processing
  • Динамическое обновление subscriptions в том же bidi stream
  • Error handling, metrics и logging для slow consumers/dropped updates

Составной паттерн client (gap recovery, ping/pong и dynamic subscriptions)

Эта композиция показывает ограниченный ingress и dynamic subscriptions. Её ingress task ждёт и переподключается только после сбоя установления соединения или чистого завершения stream. Ошибка subscription или stream gRPC сообщается и останавливает task, чтобы можно было изучить коррекцию и retry-информацию; это не общая политика повторов.
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);
                }
            }
        }
    }
}

Как добавить свой пример

Нашли полезный паттерн? Рассмотрите возможность добавить его в эту документацию. Репозиторий Solana Stream SDK: https://github.com/ValidatorsDAO/solana-stream

Назад

Geyser gRPC - Best Practices

Далее

Solana Geyser gRPC - Сравнение скорости