2023-12-15 01:20:41 -08:00
|
|
|
use async_stream::stream;
|
|
|
|
use futures::{Stream, StreamExt};
|
2024-01-17 23:45:55 -08:00
|
|
|
use log::{debug, info, log, trace, warn, Level, error};
|
2023-12-19 05:19:27 -08:00
|
|
|
use solana_sdk::commitment_config::CommitmentConfig;
|
2023-12-15 01:20:41 -08:00
|
|
|
use std::collections::HashMap;
|
2024-01-16 23:31:22 -08:00
|
|
|
use std::fmt::{Debug, Display};
|
2024-01-18 01:57:04 -08:00
|
|
|
use std::pin::Pin;
|
2023-12-19 04:03:07 -08:00
|
|
|
use std::time::Duration;
|
2024-01-17 23:45:55 -08:00
|
|
|
use futures::channel::mpsc;
|
2024-01-18 02:11:46 -08:00
|
|
|
use tokio::sync::broadcast::error::SendError;
|
2024-01-17 23:45:55 -08:00
|
|
|
use tokio::sync::broadcast::Receiver;
|
2023-12-15 01:20:41 -08:00
|
|
|
use tokio::task::JoinHandle;
|
2024-01-18 01:57:04 -08:00
|
|
|
use tokio::time::{sleep, timeout, Timeout};
|
|
|
|
use tokio::time::error::Elapsed;
|
2024-01-17 23:45:55 -08:00
|
|
|
use yellowstone_grpc_client::{GeyserGrpcClient, GeyserGrpcClientError, GeyserGrpcClientResult};
|
2023-12-22 08:05:48 -08:00
|
|
|
use yellowstone_grpc_proto::geyser::{
|
|
|
|
CommitmentLevel, SubscribeRequest, SubscribeRequestFilterBlocks, SubscribeUpdate,
|
|
|
|
};
|
2024-01-18 01:09:32 -08:00
|
|
|
use yellowstone_grpc_proto::geyser::subscribe_update::UpdateOneof;
|
2023-12-15 01:20:41 -08:00
|
|
|
use yellowstone_grpc_proto::prelude::SubscribeRequestFilterBlocksMeta;
|
2024-01-17 23:45:55 -08:00
|
|
|
use yellowstone_grpc_proto::tonic;
|
|
|
|
use yellowstone_grpc_proto::tonic::codegen::http::uri::InvalidUri;
|
|
|
|
use yellowstone_grpc_proto::tonic::metadata::errors::InvalidMetadataValue;
|
2024-01-18 01:57:04 -08:00
|
|
|
use yellowstone_grpc_proto::tonic::service::Interceptor;
|
2023-12-15 01:20:41 -08:00
|
|
|
use yellowstone_grpc_proto::tonic::transport::ClientTlsConfig;
|
2024-01-18 02:38:45 -08:00
|
|
|
use yellowstone_grpc_proto::tonic::{Code, Status};
|
2024-01-18 01:57:04 -08:00
|
|
|
use crate::grpc_subscription_autoreconnect::TheState::*;
|
2023-12-15 01:20:41 -08:00
|
|
|
|
2023-12-15 02:59:19 -08:00
|
|
|
#[derive(Clone, Debug)]
|
|
|
|
pub struct GrpcConnectionTimeouts {
|
|
|
|
pub connect_timeout: Duration,
|
|
|
|
pub request_timeout: Duration,
|
|
|
|
pub subscribe_timeout: Duration,
|
|
|
|
}
|
|
|
|
|
2024-01-16 23:31:22 -08:00
|
|
|
#[derive(Clone)]
|
2023-12-15 01:20:41 -08:00
|
|
|
pub struct GrpcSourceConfig {
|
|
|
|
grpc_addr: String,
|
|
|
|
grpc_x_token: Option<String>,
|
|
|
|
tls_config: Option<ClientTlsConfig>,
|
2023-12-15 02:59:19 -08:00
|
|
|
timeouts: Option<GrpcConnectionTimeouts>,
|
2023-12-15 01:20:41 -08:00
|
|
|
}
|
|
|
|
|
2023-12-19 03:44:49 -08:00
|
|
|
impl Display for GrpcSourceConfig {
|
|
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
2023-12-19 05:19:27 -08:00
|
|
|
write!(
|
|
|
|
f,
|
|
|
|
"grpc_addr {}",
|
|
|
|
crate::obfuscate::url_obfuscate_api_token(&self.grpc_addr)
|
|
|
|
)
|
2023-12-19 03:44:49 -08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2024-01-16 23:31:22 -08:00
|
|
|
impl Debug for GrpcSourceConfig {
|
|
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
|
|
std::fmt::Display::fmt(&self, f)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2023-12-15 01:20:41 -08:00
|
|
|
impl GrpcSourceConfig {
|
2023-12-19 02:27:42 -08:00
|
|
|
/// Create a grpc source without tls and timeouts
|
2023-12-19 02:57:23 -08:00
|
|
|
pub fn new_simple(grpc_addr: String) -> Self {
|
2023-12-15 01:20:41 -08:00
|
|
|
Self {
|
|
|
|
grpc_addr,
|
2023-12-19 02:27:42 -08:00
|
|
|
grpc_x_token: None,
|
2023-12-15 01:20:41 -08:00
|
|
|
tls_config: None,
|
2023-12-15 02:59:19 -08:00
|
|
|
timeouts: None,
|
2023-12-15 01:20:41 -08:00
|
|
|
}
|
|
|
|
}
|
2023-12-19 02:27:42 -08:00
|
|
|
pub fn new(
|
2023-12-15 07:28:01 -08:00
|
|
|
grpc_addr: String,
|
|
|
|
grpc_x_token: Option<String>,
|
2023-12-19 02:27:42 -08:00
|
|
|
tls_config: Option<ClientTlsConfig>,
|
2023-12-15 07:28:01 -08:00
|
|
|
timeouts: GrpcConnectionTimeouts,
|
|
|
|
) -> Self {
|
|
|
|
Self {
|
|
|
|
grpc_addr,
|
|
|
|
grpc_x_token,
|
2023-12-19 02:27:42 -08:00
|
|
|
tls_config,
|
2023-12-15 07:28:01 -08:00
|
|
|
timeouts: Some(timeouts),
|
|
|
|
}
|
|
|
|
}
|
2023-12-15 01:20:41 -08:00
|
|
|
}
|
|
|
|
|
2023-12-19 02:27:42 -08:00
|
|
|
type Attempt = u32;
|
|
|
|
|
|
|
|
// wraps payload and status messages
|
2024-01-18 01:09:32 -08:00
|
|
|
// clone is required by broacast channel
|
|
|
|
#[derive(Clone)]
|
2023-12-15 03:12:06 -08:00
|
|
|
pub enum Message {
|
2023-12-19 23:54:20 -08:00
|
|
|
GeyserSubscribeUpdate(Box<SubscribeUpdate>),
|
2023-12-19 02:27:42 -08:00
|
|
|
// connect (attempt=1) or reconnect(attempt=2..)
|
|
|
|
Connecting(Attempt),
|
2023-12-15 03:12:06 -08:00
|
|
|
}
|
|
|
|
|
2023-12-15 01:20:41 -08:00
|
|
|
enum ConnectionState<S: Stream<Item = Result<SubscribeUpdate, Status>>> {
|
2023-12-19 02:27:42 -08:00
|
|
|
NotConnected(Attempt),
|
|
|
|
Connecting(Attempt, JoinHandle<GeyserGrpcClientResult<S>>),
|
|
|
|
Ready(Attempt, S),
|
|
|
|
WaitReconnect(Attempt),
|
2023-12-15 01:20:41 -08:00
|
|
|
}
|
|
|
|
|
2023-12-15 08:02:42 -08:00
|
|
|
#[derive(Clone)]
|
2023-12-20 22:52:37 -08:00
|
|
|
pub struct GeyserFilter(pub CommitmentConfig);
|
2023-12-15 08:02:42 -08:00
|
|
|
|
|
|
|
impl GeyserFilter {
|
2023-12-20 22:52:37 -08:00
|
|
|
pub fn blocks_and_txs(&self) -> SubscribeRequest {
|
|
|
|
let mut blocks_subs = HashMap::new();
|
|
|
|
blocks_subs.insert(
|
|
|
|
"client".to_string(),
|
|
|
|
SubscribeRequestFilterBlocks {
|
|
|
|
account_include: Default::default(),
|
|
|
|
include_transactions: Some(true),
|
|
|
|
include_accounts: Some(false),
|
|
|
|
include_entries: Some(false),
|
|
|
|
},
|
|
|
|
);
|
|
|
|
|
|
|
|
SubscribeRequest {
|
|
|
|
slots: HashMap::new(),
|
|
|
|
accounts: Default::default(),
|
|
|
|
transactions: HashMap::new(),
|
|
|
|
entry: Default::default(),
|
|
|
|
blocks: blocks_subs,
|
|
|
|
blocks_meta: HashMap::new(),
|
|
|
|
commitment: Some(map_commitment_level(self.0) as i32),
|
|
|
|
accounts_data_slice: Default::default(),
|
|
|
|
ping: None,
|
|
|
|
}
|
2023-12-19 02:27:42 -08:00
|
|
|
}
|
2023-12-20 22:52:37 -08:00
|
|
|
|
|
|
|
pub fn blocks_meta(&self) -> SubscribeRequest {
|
|
|
|
let mut blocksmeta_subs = HashMap::new();
|
2023-12-22 08:05:48 -08:00
|
|
|
blocksmeta_subs.insert("client".to_string(), SubscribeRequestFilterBlocksMeta {});
|
2023-12-20 22:52:37 -08:00
|
|
|
|
|
|
|
SubscribeRequest {
|
|
|
|
slots: HashMap::new(),
|
|
|
|
accounts: Default::default(),
|
|
|
|
transactions: HashMap::new(),
|
|
|
|
entry: Default::default(),
|
|
|
|
blocks: HashMap::new(),
|
|
|
|
blocks_meta: blocksmeta_subs,
|
|
|
|
commitment: Some(map_commitment_level(self.0) as i32),
|
|
|
|
accounts_data_slice: Default::default(),
|
|
|
|
ping: None,
|
|
|
|
}
|
2023-12-15 08:02:42 -08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2023-12-20 22:52:37 -08:00
|
|
|
fn map_commitment_level(commitment_config: CommitmentConfig) -> CommitmentLevel {
|
2023-12-22 08:05:48 -08:00
|
|
|
// solana_sdk -> yellowstone
|
|
|
|
match commitment_config.commitment {
|
2023-12-19 04:03:07 -08:00
|
|
|
solana_sdk::commitment_config::CommitmentLevel::Processed => {
|
|
|
|
yellowstone_grpc_proto::prelude::CommitmentLevel::Processed
|
|
|
|
}
|
2023-12-15 01:25:21 -08:00
|
|
|
solana_sdk::commitment_config::CommitmentLevel::Confirmed => {
|
|
|
|
yellowstone_grpc_proto::prelude::CommitmentLevel::Confirmed
|
|
|
|
}
|
|
|
|
solana_sdk::commitment_config::CommitmentLevel::Finalized => {
|
|
|
|
yellowstone_grpc_proto::prelude::CommitmentLevel::Finalized
|
|
|
|
}
|
2023-12-19 04:03:07 -08:00
|
|
|
_ => {
|
2023-12-19 05:19:27 -08:00
|
|
|
panic!(
|
|
|
|
"unsupported commitment level {}",
|
|
|
|
commitment_config.commitment
|
|
|
|
)
|
2023-12-19 04:03:07 -08:00
|
|
|
}
|
2023-12-22 08:05:48 -08:00
|
|
|
}
|
2023-12-20 22:52:37 -08:00
|
|
|
}
|
|
|
|
|
|
|
|
// Take geyser filter, connect to Geyser and return a generic stream of SubscribeUpdate
|
|
|
|
// note: stream never terminates
|
|
|
|
pub fn create_geyser_reconnecting_stream(
|
|
|
|
grpc_source: GrpcSourceConfig,
|
|
|
|
subscribe_filter: SubscribeRequest,
|
|
|
|
) -> impl Stream<Item = Message> {
|
2023-12-15 07:38:19 -08:00
|
|
|
let mut state = ConnectionState::NotConnected(0);
|
2023-12-15 01:20:41 -08:00
|
|
|
|
|
|
|
// in case of cancellation, we restart from here:
|
|
|
|
// thus we want to keep the progression in a state object outside the stream! makro
|
2023-12-19 02:57:23 -08:00
|
|
|
let the_stream = stream! {
|
2023-12-15 02:43:00 -08:00
|
|
|
loop {
|
2023-12-15 01:20:41 -08:00
|
|
|
let yield_value;
|
2023-12-15 02:43:00 -08:00
|
|
|
|
2023-12-15 01:20:41 -08:00
|
|
|
(state, yield_value) = match state {
|
2023-12-15 02:43:00 -08:00
|
|
|
|
2023-12-15 07:38:19 -08:00
|
|
|
ConnectionState::NotConnected(mut attempt) => {
|
2023-12-19 23:54:20 -08:00
|
|
|
attempt += 1;
|
2023-12-15 01:20:41 -08:00
|
|
|
|
|
|
|
let connection_task = tokio::spawn({
|
|
|
|
let addr = grpc_source.grpc_addr.clone();
|
|
|
|
let token = grpc_source.grpc_x_token.clone();
|
|
|
|
let config = grpc_source.tls_config.clone();
|
2023-12-15 02:59:19 -08:00
|
|
|
let connect_timeout = grpc_source.timeouts.as_ref().map(|t| t.connect_timeout);
|
|
|
|
let request_timeout = grpc_source.timeouts.as_ref().map(|t| t.request_timeout);
|
|
|
|
let subscribe_timeout = grpc_source.timeouts.as_ref().map(|t| t.subscribe_timeout);
|
2023-12-20 22:52:37 -08:00
|
|
|
let subscribe_filter = subscribe_filter.clone();
|
2023-12-15 07:38:19 -08:00
|
|
|
log!(if attempt > 1 { Level::Warn } else { Level::Debug }, "Connecting attempt #{} to {}", attempt, addr);
|
2023-12-15 01:20:41 -08:00
|
|
|
async move {
|
|
|
|
|
|
|
|
let connect_result = GeyserGrpcClient::connect_with_timeout(
|
2023-12-15 02:59:19 -08:00
|
|
|
addr, token, config,
|
|
|
|
connect_timeout,
|
|
|
|
request_timeout,
|
|
|
|
false)
|
|
|
|
.await;
|
2023-12-15 01:20:41 -08:00
|
|
|
let mut client = connect_result?;
|
|
|
|
|
2023-12-20 22:52:37 -08:00
|
|
|
|
|
|
|
debug!("Subscribe with filter {:?}", subscribe_filter);
|
2023-12-20 22:40:18 -08:00
|
|
|
|
2023-12-15 08:02:42 -08:00
|
|
|
let subscribe_result = timeout(subscribe_timeout.unwrap_or(Duration::MAX),
|
2023-12-15 02:59:19 -08:00
|
|
|
client
|
2023-12-20 22:52:37 -08:00
|
|
|
.subscribe_once2(subscribe_filter))
|
2023-12-15 08:02:42 -08:00
|
|
|
.await;
|
|
|
|
|
2023-12-15 08:07:23 -08:00
|
|
|
// maybe not optimal
|
2023-12-19 04:03:07 -08:00
|
|
|
subscribe_result.map_err(|_| Status::unknown("unspecific subscribe timeout"))?
|
2023-12-15 01:20:41 -08:00
|
|
|
}
|
|
|
|
});
|
|
|
|
|
2023-12-15 08:02:42 -08:00
|
|
|
(ConnectionState::Connecting(attempt, connection_task), Message::Connecting(attempt))
|
2023-12-15 01:20:41 -08:00
|
|
|
}
|
2023-12-15 02:43:00 -08:00
|
|
|
|
2023-12-15 07:38:19 -08:00
|
|
|
ConnectionState::Connecting(attempt, connection_task) => {
|
2023-12-15 01:20:41 -08:00
|
|
|
let subscribe_result = connection_task.await;
|
|
|
|
|
|
|
|
match subscribe_result {
|
2023-12-15 08:02:42 -08:00
|
|
|
Ok(Ok(subscribed_stream)) => (ConnectionState::Ready(attempt, subscribed_stream), Message::Connecting(attempt)),
|
2023-12-15 01:20:41 -08:00
|
|
|
Ok(Err(geyser_error)) => {
|
2023-12-19 04:03:07 -08:00
|
|
|
// ATM we consider all errors recoverable
|
|
|
|
warn!("! subscribe failed on {} - retrying: {:?}", grpc_source, geyser_error);
|
2023-12-15 08:02:42 -08:00
|
|
|
(ConnectionState::WaitReconnect(attempt), Message::Connecting(attempt))
|
2023-12-15 01:20:41 -08:00
|
|
|
},
|
|
|
|
Err(geyser_grpc_task_error) => {
|
2023-12-19 04:03:07 -08:00
|
|
|
panic!("! task aborted - should not happen :{geyser_grpc_task_error}");
|
2023-12-15 01:20:41 -08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
2023-12-15 02:43:00 -08:00
|
|
|
|
2023-12-15 07:38:19 -08:00
|
|
|
ConnectionState::Ready(attempt, mut geyser_stream) => {
|
2023-12-15 01:20:41 -08:00
|
|
|
|
2023-12-15 01:34:05 -08:00
|
|
|
match geyser_stream.next().await {
|
|
|
|
Some(Ok(update_message)) => {
|
2023-12-19 03:44:49 -08:00
|
|
|
trace!("> recv update message from {}", grpc_source);
|
2023-12-19 23:54:20 -08:00
|
|
|
(ConnectionState::Ready(attempt, geyser_stream), Message::GeyserSubscribeUpdate(Box::new(update_message)))
|
2023-12-15 01:20:41 -08:00
|
|
|
}
|
2023-12-15 01:34:05 -08:00
|
|
|
Some(Err(tonic_status)) => {
|
2023-12-19 04:03:07 -08:00
|
|
|
// ATM we consider all errors recoverable
|
2023-12-22 08:05:48 -08:00
|
|
|
warn!("! error on {} - retrying: {:?}", grpc_source, tonic_status);
|
2023-12-15 08:02:42 -08:00
|
|
|
(ConnectionState::WaitReconnect(attempt), Message::Connecting(attempt))
|
2023-12-15 01:34:05 -08:00
|
|
|
}
|
|
|
|
None => {
|
2023-12-19 04:03:07 -08:00
|
|
|
// should not arrive here, Mean the stream close.
|
2024-01-07 10:59:34 -08:00
|
|
|
warn!("geyser stream closed on {} - retrying", grpc_source);
|
|
|
|
(ConnectionState::WaitReconnect(attempt), Message::Connecting(attempt))
|
2023-12-15 01:34:05 -08:00
|
|
|
}
|
|
|
|
}
|
2023-12-15 01:20:41 -08:00
|
|
|
|
|
|
|
}
|
2023-12-15 02:43:00 -08:00
|
|
|
|
2023-12-15 07:38:19 -08:00
|
|
|
ConnectionState::WaitReconnect(attempt) => {
|
|
|
|
let backoff_secs = 1.5_f32.powi(attempt as i32).min(15.0);
|
2023-12-19 04:03:07 -08:00
|
|
|
info!("! waiting {} seconds, then reconnect to {}", backoff_secs, grpc_source);
|
2023-12-15 07:38:19 -08:00
|
|
|
sleep(Duration::from_secs_f32(backoff_secs)).await;
|
2023-12-15 08:02:42 -08:00
|
|
|
(ConnectionState::NotConnected(attempt), Message::Connecting(attempt))
|
2023-12-15 01:20:41 -08:00
|
|
|
}
|
2023-12-15 02:43:00 -08:00
|
|
|
|
2023-12-15 01:20:41 -08:00
|
|
|
}; // -- match
|
2023-12-15 02:43:00 -08:00
|
|
|
|
2023-12-15 01:20:41 -08:00
|
|
|
yield yield_value
|
|
|
|
}
|
|
|
|
|
2023-12-19 02:57:23 -08:00
|
|
|
}; // -- stream!
|
|
|
|
|
|
|
|
the_stream
|
2023-12-15 01:20:41 -08:00
|
|
|
}
|
2024-01-16 23:31:22 -08:00
|
|
|
|
2024-01-17 23:45:55 -08:00
|
|
|
|
2024-01-18 01:57:04 -08:00
|
|
|
enum TheState<S: Stream<Item = Result<SubscribeUpdate, Status>>, F: Interceptor> {
|
2024-01-17 23:45:55 -08:00
|
|
|
NotConnected(Attempt),
|
2024-01-18 01:57:04 -08:00
|
|
|
// Connected(Attempt, Box<Pin<GeyserGrpcClient<F>>>),
|
|
|
|
Connected(Attempt, GeyserGrpcClient<F>),
|
|
|
|
Ready(Attempt, S),
|
|
|
|
// error states
|
2024-01-17 23:45:55 -08:00
|
|
|
RecoverableConnectionError(Attempt),
|
|
|
|
FatalError(Attempt),
|
2024-01-18 01:57:04 -08:00
|
|
|
WaitReconnect(Attempt),
|
2024-01-17 23:45:55 -08:00
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
pub fn create_geyser_reconnecting_task(
|
|
|
|
grpc_source: GrpcSourceConfig,
|
|
|
|
subscribe_filter: SubscribeRequest,
|
2024-01-18 01:09:32 -08:00
|
|
|
) -> (JoinHandle<()>, Receiver<Message>) {
|
2024-01-18 02:11:46 -08:00
|
|
|
let (sender, receiver_stream) = tokio::sync::broadcast::channel::<Message>(1000);
|
2024-01-17 23:45:55 -08:00
|
|
|
|
2024-01-18 01:09:32 -08:00
|
|
|
let jh_geyser_task = tokio::spawn(async move {
|
2024-01-17 23:45:55 -08:00
|
|
|
let mut state = NotConnected(0);
|
|
|
|
|
|
|
|
loop {
|
|
|
|
|
|
|
|
state = match state {
|
2024-01-18 01:09:32 -08:00
|
|
|
NotConnected(mut attempt) => {
|
|
|
|
attempt += 1;
|
|
|
|
|
2024-01-17 23:45:55 -08:00
|
|
|
let addr = grpc_source.grpc_addr.clone();
|
|
|
|
let token = grpc_source.grpc_x_token.clone();
|
|
|
|
let config = grpc_source.tls_config.clone();
|
|
|
|
let connect_timeout = grpc_source.timeouts.as_ref().map(|t| t.connect_timeout);
|
|
|
|
let request_timeout = grpc_source.timeouts.as_ref().map(|t| t.request_timeout);
|
|
|
|
let subscribe_timeout = grpc_source.timeouts.as_ref().map(|t| t.subscribe_timeout);
|
|
|
|
let subscribe_filter = subscribe_filter.clone();
|
|
|
|
log!(if attempt > 1 { Level::Warn } else { Level::Debug }, "Connecting attempt #{} to {}", attempt, addr);
|
|
|
|
let connect_result = GeyserGrpcClient::connect_with_timeout(
|
|
|
|
addr, token, config,
|
|
|
|
connect_timeout,
|
|
|
|
request_timeout,
|
|
|
|
false)
|
|
|
|
.await;
|
2024-01-18 01:09:32 -08:00
|
|
|
|
2024-01-17 23:45:55 -08:00
|
|
|
|
2024-01-18 01:57:04 -08:00
|
|
|
match connect_result {
|
|
|
|
Ok(client) => {
|
|
|
|
Connected(attempt, client)
|
|
|
|
}
|
|
|
|
Err(_) => {
|
|
|
|
todo!()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
Connected(attempt, mut client) => {
|
|
|
|
let subscribe_timeout = grpc_source.timeouts.as_ref().map(|t| t.subscribe_timeout);
|
|
|
|
let subscribe_filter = subscribe_filter.clone();
|
2024-01-17 23:45:55 -08:00
|
|
|
debug!("Subscribe with filter {:?}", subscribe_filter);
|
|
|
|
|
2024-01-18 01:57:04 -08:00
|
|
|
let subscribe_result_timeout =
|
|
|
|
timeout(subscribe_timeout.unwrap_or(Duration::MAX),
|
|
|
|
client.subscribe_once2(subscribe_filter))
|
2024-01-17 23:45:55 -08:00
|
|
|
.await;
|
|
|
|
|
2024-01-18 01:57:04 -08:00
|
|
|
let subscribe_result;
|
|
|
|
match subscribe_result_timeout.map_err(|_| Status::unknown("unspecific subscribe timeout")) {
|
|
|
|
Ok(fooo) => {
|
|
|
|
subscribe_result = fooo;
|
|
|
|
}
|
|
|
|
Err(_elapsed) => {
|
|
|
|
todo!()
|
|
|
|
}
|
|
|
|
}
|
2024-01-17 23:45:55 -08:00
|
|
|
// maybe not optimal
|
2024-01-18 01:57:04 -08:00
|
|
|
// let subscribe_result = subscribe_result_timeout.map_err(|_| Status::unknown("unspecific subscribe timeout"));
|
2024-01-17 23:45:55 -08:00
|
|
|
|
|
|
|
match subscribe_result {
|
|
|
|
Ok(geyser_stream) => {
|
2024-01-18 01:57:04 -08:00
|
|
|
Ready(attempt, geyser_stream)
|
2024-01-17 23:45:55 -08:00
|
|
|
}
|
|
|
|
Err(GeyserGrpcClientError::TonicError(_)) => {
|
|
|
|
warn!("! subscribe failed on {} - retrying", grpc_source);
|
|
|
|
RecoverableConnectionError(attempt)
|
|
|
|
}
|
|
|
|
Err(GeyserGrpcClientError::TonicStatus(_)) => {
|
|
|
|
warn!("! subscribe failed on {} - retrying", grpc_source);
|
|
|
|
RecoverableConnectionError(attempt)
|
|
|
|
}
|
|
|
|
// non-recoverable
|
|
|
|
Err(unrecoverable_error) => {
|
|
|
|
error!("! subscribe to {} failed with unrecoverable error: {}", grpc_source, unrecoverable_error);
|
|
|
|
FatalError(attempt)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
RecoverableConnectionError(attempt) => {
|
|
|
|
let backoff_secs = 1.5_f32.powi(attempt as i32).min(15.0);
|
|
|
|
info!("! waiting {} seconds, then reconnect to {}", backoff_secs, grpc_source);
|
|
|
|
sleep(Duration::from_secs_f32(backoff_secs)).await;
|
2024-01-18 01:09:32 -08:00
|
|
|
NotConnected(attempt)
|
2024-01-17 23:45:55 -08:00
|
|
|
}
|
|
|
|
FatalError(_) => {
|
|
|
|
// TOOD what to do
|
|
|
|
panic!("Fatal error")
|
|
|
|
}
|
2024-01-18 01:57:04 -08:00
|
|
|
TheState::WaitReconnect(attempt) => {
|
|
|
|
let backoff_secs = 1.5_f32.powi(attempt as i32).min(15.0);
|
|
|
|
info!("! waiting {} seconds, then reconnect to {}", backoff_secs, grpc_source);
|
|
|
|
sleep(Duration::from_secs_f32(backoff_secs)).await;
|
|
|
|
TheState::NotConnected(attempt)
|
|
|
|
}
|
|
|
|
Ready(attempt, mut geyser_stream) => {
|
2024-01-18 02:11:46 -08:00
|
|
|
'recv_loop: loop {
|
|
|
|
match geyser_stream.next().await {
|
|
|
|
Some(Ok(update_message)) => {
|
|
|
|
trace!("> recv update message from {}", grpc_source);
|
|
|
|
match sender.send(Message::GeyserSubscribeUpdate(Box::new(update_message))) {
|
|
|
|
Ok(n_subscribers) => {
|
|
|
|
trace!("sent update message to {} subscribers (buffer={})",
|
|
|
|
n_subscribers,
|
|
|
|
sender.len());
|
|
|
|
continue 'recv_loop;
|
|
|
|
}
|
|
|
|
Err(SendError(_)) => {
|
|
|
|
// note: error does not mean that future sends will also fail!
|
|
|
|
trace!("no subscribers for update message");
|
|
|
|
continue 'recv_loop;
|
|
|
|
}
|
|
|
|
};
|
|
|
|
}
|
|
|
|
Some(Err(tonic_status)) => {
|
2024-01-18 02:38:45 -08:00
|
|
|
// all tonic errors are recoverable
|
2024-01-18 02:11:46 -08:00
|
|
|
warn!("! error on {} - retrying: {:?}", grpc_source, tonic_status);
|
|
|
|
break 'recv_loop TheState::WaitReconnect(attempt);
|
|
|
|
}
|
|
|
|
None => {
|
|
|
|
warn!("geyser stream closed on {} - retrying", grpc_source);
|
|
|
|
break 'recv_loop TheState::WaitReconnect(attempt);
|
|
|
|
}
|
2024-01-18 01:09:32 -08:00
|
|
|
}
|
2024-01-18 02:38:45 -08:00
|
|
|
} // -- end loop
|
2024-01-18 01:09:32 -08:00
|
|
|
}
|
2024-01-17 23:45:55 -08:00
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
2024-01-18 02:11:46 -08:00
|
|
|
(jh_geyser_task, receiver_stream)
|
2024-01-17 23:45:55 -08:00
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
2024-01-16 23:31:22 -08:00
|
|
|
#[cfg(test)]
|
|
|
|
mod tests {
|
|
|
|
use super::*;
|
|
|
|
|
|
|
|
#[tokio::test]
|
|
|
|
async fn test_debug_no_secrets() {
|
|
|
|
let timeout_config = GrpcConnectionTimeouts {
|
|
|
|
connect_timeout: Duration::from_secs(1),
|
|
|
|
request_timeout: Duration::from_secs(2),
|
|
|
|
subscribe_timeout: Duration::from_secs(3),
|
|
|
|
};
|
|
|
|
assert_eq!(
|
|
|
|
format!(
|
|
|
|
"{:?}",
|
|
|
|
GrpcSourceConfig::new(
|
|
|
|
"http://localhost:1234".to_string(),
|
|
|
|
Some("my-secret".to_string()),
|
|
|
|
None,
|
|
|
|
timeout_config
|
|
|
|
)
|
|
|
|
),
|
|
|
|
"grpc_addr http://localhost:1234"
|
|
|
|
);
|
|
|
|
}
|
|
|
|
|
|
|
|
#[tokio::test]
|
|
|
|
async fn test_display_no_secrets() {
|
|
|
|
let timeout_config = GrpcConnectionTimeouts {
|
|
|
|
connect_timeout: Duration::from_secs(1),
|
|
|
|
request_timeout: Duration::from_secs(2),
|
|
|
|
subscribe_timeout: Duration::from_secs(3),
|
|
|
|
};
|
|
|
|
assert_eq!(
|
|
|
|
format!(
|
|
|
|
"{}",
|
|
|
|
GrpcSourceConfig::new(
|
|
|
|
"http://localhost:1234".to_string(),
|
|
|
|
Some("my-secret".to_string()),
|
|
|
|
None,
|
|
|
|
timeout_config
|
|
|
|
)
|
|
|
|
),
|
|
|
|
"grpc_addr http://localhost:1234"
|
|
|
|
);
|
|
|
|
}
|
|
|
|
}
|
2024-01-18 02:38:45 -08:00
|
|
|
|