Adding quic geyser source and making it work with all raydium pools (#5)

Adding quic geyser source and making it work with all raydium pools

Cargo.toml

@@ -17,6 +17,7 @@ solana-rpc-client-api = { version = "1.17" }
 mango-feeds-connector = { git = "https://github.com/blockworks-foundation/mango-feeds.git", tag = "connector-v0.4.8" }
 yellowstone-grpc-client = { version = "1.15.0", git = "https://github.com/blockworks-foundation/yellowstone-grpc.git", tag = "v1.15.0+solana.1.17" }
 yellowstone-grpc-proto = { version = "1.14.0", git = "https://github.com/blockworks-foundation/yellowstone-grpc.git", tag = "v1.15.0+solana.1.17" }
+
 reqwest = { version = "0.11.27", features = ["json"] }
 whirlpools-client = { git = "https://github.com/blockworks-foundation/whirlpools-client/", features = ["no-entrypoint"] }
 openbook-v2 = { git = "https://github.com/openbook-dex/openbook-v2", tag = "v0.2.7", features = ["no-entrypoint", "client"] }
@@ -25,6 +26,8 @@ stable-swap = { version = "1.8.1", features = ["no-entrypoint", "client"] }
 stable-swap-client = { version = "1.8.1" }
 stable-swap-math = { version = "1.8.1" }
 uint = { version = "0.9.1" }
+quic-geyser-client = { git = "https://github.com/blockworks-foundation/quic_geyser_plugin.git", branch = "router_v1.17.29" }
+quic-geyser-common = { git = "https://github.com/blockworks-foundation/quic_geyser_plugin.git", branch = "router_v1.17.29" }
 
 [profile.release]
 overflow-checks = true

bin/autobahn-router/Cargo.toml

@@ -87,6 +87,10 @@ yellowstone-grpc-client = { workspace = true }
 yellowstone-grpc-proto = { workspace = true }
 tonic = { version = "0.10.2", features = ["gzip"] }
 
+# quic
+quic-geyser-client = { workspace = true }
+quic-geyser-common = { workspace = true }
+
 # compressed snapshots
 lz4 = "1.24.0"
 

bin/autobahn-router/examples/grpc_source_tester.rs

@@ -22,13 +22,13 @@ pub async fn main() {
     let rpc_http_addr = env::var("RPC_HTTP_ADDR").expect("need rpc http url");
     let snapshot_config = AccountDataSourceConfig {
         region: None,
-        use_quic: None,
-        quic_address: None,
+        quic_sources: None,
         rpc_http_url: rpc_http_addr.clone(),
         rpc_support_compression: Some(false), /* no compression */
         re_snapshot_interval_secs: None,
-        grpc_sources: vec![],
+        grpc_sources: Some(vec![]),
         dedup_queue_size: 0,
+        request_timeout_in_seconds: None,
     };
 
     // Raydium

bin/autobahn-router/src/edge_updater.rs

@@ -138,7 +138,8 @@ pub fn spawn_updater_job(
         ),
     };
 
-    let snapshot_timeout = Instant::now() + Duration::from_secs(60 * 5);
+    let snapshot_timeout_in_seconds = config.snapshot_timeout_in_seconds.unwrap_or(60 * 5);
+    let snapshot_timeout = Instant::now() + Duration::from_secs(snapshot_timeout_in_seconds);
     let listener_job = tokio_spawn(format!("edge_updater_{}", dex.name).as_str(), async move {
         let mut updater = EdgeUpdater {
             dex,

bin/autobahn-router/src/main.rs

@@ -175,19 +175,28 @@ async fn main() -> anyhow::Result<()> {
         }
     });
 
-    if source_config.grpc_sources.len() > 1 {
-        error!("only one grpc source is supported ATM");
-        exit(-1);
+    if let Some(quic_sources) = &source_config.quic_sources {
+        info!(
+            "quic sources: {}",
+            quic_sources
+                .iter()
+                .map(|c| c.connection_string.clone())
+                .collect::<String>()
+        );
     }
-
-    info!(
-        "grpc sources: {}",
-        source_config
-            .grpc_sources
-            .iter()
-            .map(|c| c.connection_string.clone())
-            .collect::<String>()
-    );
+    if let Some(grpc_sources) = source_config.grpc_sources.clone() {
+        info!(
+            "grpc sources: {}",
+            grpc_sources
+                .iter()
+                .map(|c| c.connection_string.clone())
+                .collect::<String>()
+        );
+    } else {
+        // current grpc source is needed for transaction watcher even if there is quic
+        error!("No grpc geyser sources specified");
+        exit(-1);
+    };
 
     if config.metrics.output_http {
         let prom_bind_addr = config
@@ -443,8 +452,8 @@ async fn main() -> anyhow::Result<()> {
     let ef = exit_sender.subscribe();
     let sc = source_config.clone();
     let account_update_job = tokio_spawn("geyser", async move {
-        if sc.use_quic.unwrap_or(false) {
-            error!("not supported yet");
+        if sc.grpc_sources.is_none() && sc.quic_sources.is_none() {
+            error!("No quic or grpc plugin setup");
         } else {
             geyser::spawn_geyser_source(
                 &sc,
@@ -528,7 +537,7 @@ fn build_price_feed(
 fn build_rpc(source_config: &AccountDataSourceConfig) -> RpcClient {
     RpcClient::new_with_timeouts_and_commitment(
         string_or_env(source_config.rpc_http_url.clone()),
-        Duration::from_secs(60), // request timeout
+        Duration::from_secs(source_config.request_timeout_in_seconds.unwrap_or(60)), // request timeout
         CommitmentConfig::confirmed(),
         Duration::from_secs(60), // confirmation timeout
     )
@@ -537,7 +546,7 @@ fn build_rpc(source_config: &AccountDataSourceConfig) -> RpcClient {
 fn build_blocking_rpc(source_config: &AccountDataSourceConfig) -> BlockingRpcClient {
     BlockingRpcClient::new_with_timeouts_and_commitment(
         string_or_env(source_config.rpc_http_url.clone()),
-        Duration::from_secs(60), // request timeout
+        Duration::from_secs(source_config.request_timeout_in_seconds.unwrap_or(60)), // request timeout
         CommitmentConfig::confirmed(),
         Duration::from_secs(60), // confirmation timeout
     )

bin/autobahn-router/src/metrics.rs

@@ -27,6 +27,23 @@ lazy_static::lazy_static! {
     pub static ref GRPC_TO_EDGE_SLOT_LAG: IntGaugeVec =
         register_int_gauge_vec!(opts!("router_grpc_to_edge_slot_lag", "RPC Slot vs last slot used to update edges"), &["dex_name"]).unwrap();
 
+    pub static ref QUIC_ACCOUNT_WRITES: IntCounter =
+        register_int_counter!("quic_account_writes", "Number of account updates via Geyser gRPC").unwrap();
+    pub static ref QUIC_ACCOUNT_WRITE_QUEUE: IntGauge =
+        register_int_gauge!("quic_account_write_queue", "Items in account write queue via Geyser gPRC").unwrap();
+    pub static ref QUIC_DEDUP_QUEUE: GenericGauge<prometheus::core::AtomicI64> =
+        register_int_gauge!("quic_dedup_queue", "Items in dedup queue via Geyser gPRC").unwrap();
+    pub static ref QUIC_SLOT_UPDATE_QUEUE: GenericGauge<prometheus::core::AtomicI64> =
+        register_int_gauge!("quic_slot_update_queue", "Items in slot update queue via Geyser gPRC").unwrap();
+    pub static ref QUIC_SLOT_UPDATES: IntCounter =
+        register_int_counter!("quic_slot_updates", "Number of slot updates via Geyser gPRC").unwrap();
+    pub static ref QUIC_SNAPSHOT_ACCOUNT_WRITES: IntCounter =
+        register_int_counter!("quic_snapshot_account_writes", "Number of account writes from snapshot").unwrap();
+     pub static ref QUIC_SOURCE_CONNECTION_RETRIES: IntCounterVec =
+        register_int_counter_vec!(opts!("quic_source_connection_retries", "gRPC source connection retries"), &["source_name"]).unwrap();
+    pub static ref QUIC_NO_MESSAGE_FOR_DURATION_MS: IntGauge =
+        register_int_gauge!("quic_no_update_for_duration_ms", "Did not get any message from Geyser gPRC for this duration").unwrap();
+
     pub static ref HTTP_REQUEST_TIMING: HistogramVec =
         register_histogram_vec!(
             histogram_opts!("router_http_request_timing", "Endpoint timing in seconds",

bin/autobahn-router/src/source/geyser.rs

@@ -9,6 +9,8 @@ use router_feed_lib::get_program_account::FeedMetadata;
 
 use crate::source::grpc_plugin_source;
 
+use super::quic_plugin_source;
+
 pub async fn spawn_geyser_source(
     config: &AccountDataSourceConfig,
     exit_receiver: tokio::sync::broadcast::Receiver<()>,
@@ -20,16 +22,31 @@ pub async fn spawn_geyser_source(
     subscribed_token_accounts: &HashSet<Pubkey>,
     filters: &HashSet<Pubkey>,
 ) {
-    grpc_plugin_source::process_events(
-        config.clone(),
-        subscribed_accounts.clone(),
-        subscribed_programs.clone(),
-        subscribed_token_accounts.clone(),
-        filters.clone(),
-        account_write_sender,
-        Some(metadata_write_sender),
-        slot_sender,
-        exit_receiver,
-    )
-    .await;
+    if config.quic_sources.is_some() {
+        quic_plugin_source::process_events(
+            config.clone(),
+            subscribed_accounts.clone(),
+            subscribed_programs.clone(),
+            subscribed_token_accounts.clone(),
+            filters.clone(),
+            account_write_sender,
+            Some(metadata_write_sender),
+            slot_sender,
+            exit_receiver,
+        )
+        .await;
+    } else if config.grpc_sources.is_some() {
+        grpc_plugin_source::process_events(
+            config.clone(),
+            subscribed_accounts.clone(),
+            subscribed_programs.clone(),
+            subscribed_token_accounts.clone(),
+            filters.clone(),
+            account_write_sender,
+            Some(metadata_write_sender),
+            slot_sender,
+            exit_receiver,
+        )
+        .await;
+    }
 }