ERPC

Geyser gRPC - 코드 예제

자주 쓰이는 Geyser gRPC 사용 사례를 위한 실용적인 Rust 예제입니다. 사용 중인 Geyser gRPC 엔드포인트를 입력하세요. Solana Stream SDK는 Geyser gRPC를 지원합니다.

기본 구독: 투표 트랜잭션과 실패한 트랜잭션을 제외한 모든 트랜잭션

이 예제는 의도적으로 범위를 넓게 잡은 트랜잭션 구독입니다. 실패한 트랜잭션을 제외하면 트래픽은 줄어들지만, 남은 스트림도 여전히 양이 많을 수 있습니다. 컨슈머가 실제로 필요로 하는 작업에 맞는 계정, 서명 또는 트랜잭션 필터를 추가하고, 처리는 수신 루프 밖에서 수행하세요.
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(())
}

정상 종료 후 제한된 재연결, gap 복구, ping/pong을 갖춘 durable 클라이언트

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);
    }
}
이 보수적인 샘플은 연결 수립 실패와 스트림의 정상 종료에 대해서만 재시도합니다. gRPC subscribe 오류나 스트림 오류가 발생하면 현재 진단 정보를 로그에 남기고 첫 번째 진단 정보를 보존한 뒤 중지합니다. 수동으로 재시도하기 전에 표시된 ERPC 수정 안내와 서버의 재시도 정보를 확인하세요. 프로덕션 코드에서는 알려진 일시적 실패를 명시적으로 재시도할 수 있지만, 서버가 지정한 일시 정지(pause)를 지켜야 하며 거부된 요청을 바꾸지 않은 채 반복해서 보내서는 안 됩니다.

특정 mint의 계정 스트리밍 (토큰 계정)

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

트랜잭션 스트리밍 (성공한 트랜잭션, 투표 제외)

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

프로덕션 수준 클라이언트의 핵심 사항

  • 재연결과 gap 복구(from_slot)를 갖춘 durable 클라이언트
  • 스트림을 유지하기 위한 Ping/pong 처리
  • 수신(ingress)과 처리 사이의 bounded 채널/backpressure
  • 같은 양방향(bidi) 스트림에서의 동적 구독 업데이트
  • 느린 컨슈머와 드롭된 업데이트를 위한 오류 처리, 메트릭, 로깅

조합형 클라이언트 패턴 (gap 복구, ping/pong, 동적 구독)

이 조합은 bounded ingress와 동적 구독을 보여 줍니다. ingress 태스크는 연결 수립 실패 또는 스트림의 정상 종료 후에만 대기했다가 재연결합니다. gRPC 구독 오류나 스트림 오류는 보고된 뒤 태스크를 중지시키므로, 해당 오류의 수정 안내와 재시도 정보를 확인할 수 있습니다. 이는 범용 재시도 정책이 아닙니다.
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
이전필터다음자주 묻는 질문