feat: support helius laser GRPC client for cloning

Cargo.toml

@@ -77,6 +77,7 @@ fs_extra = "1.3.0"
 futures = "0.3"
 futures-util = "0.3.30"
 git-version = "0.3.9"
+helius-laserstream = { git = "https://github.com/magicblock-labs/laserstream-sdk", branch = "v0.2.2-magicblock" }
 hostname = "0.4.0"
 http-body-util = "0.1.3"
 hyper = "1.6.0"

magicblock-accounts/src/scheduled_commits_processor.rs

@@ -9,8 +9,8 @@ use magicblock_account_cloner::ChainlinkCloner;
 use magicblock_accounts_db::AccountsDb;
 use magicblock_chainlink::{
     remote_account_provider::{
-        chain_pubsub_client::ChainPubsubClientImpl,
         chain_rpc_client::ChainRpcClientImpl,
+        chain_updates_client::ChainUpdatesClient,
     },
     submux::SubMuxClient,
     Chainlink,
@@ -48,7 +48,7 @@ const POISONED_MUTEX_MSG: &str =
 
 pub type ChainlinkImpl = Chainlink<
     ChainRpcClientImpl,
-    SubMuxClient<ChainPubsubClientImpl>,
+    SubMuxClient<ChainUpdatesClient>,
     AccountsDb,
     ChainlinkCloner,
 >;

magicblock-aperture/src/state/mod.rs

@@ -6,8 +6,8 @@ use magicblock_account_cloner::ChainlinkCloner;
 use magicblock_accounts_db::AccountsDb;
 use magicblock_chainlink::{
     remote_account_provider::{
-        chain_pubsub_client::ChainPubsubClientImpl,
         chain_rpc_client::ChainRpcClientImpl,
+        chain_updates_client::ChainUpdatesClient,
     },
     submux::SubMuxClient,
     Chainlink,
@@ -21,7 +21,7 @@ use transactions::TransactionsCache;
 
 pub type ChainlinkImpl = Chainlink<
     ChainRpcClientImpl,
-    SubMuxClient<ChainPubsubClientImpl>,
+    SubMuxClient<ChainUpdatesClient>,
     AccountsDb,
     ChainlinkCloner,
 >;

magicblock-api/src/external_config.rs

@@ -1,6 +1,7 @@
 use std::collections::HashSet;
 
 use magicblock_accounts::{AccountsConfig, RemoteCluster};
+use magicblock_chainlink::remote_account_provider::chain_laser_client::is_helius_laser_url;
 use magicblock_config::{errors::ConfigResult, RemoteConfig};
 use solana_sdk::pubkey::Pubkey;
 
@@ -59,19 +60,29 @@ pub fn remote_cluster_from_remote(
                 .map(|ws_urls| ws_urls.iter().map(|x| x.to_string()).collect())
                 .unwrap_or_else(|| {
                     let mut ws_url = rpc_url.clone();
-                    ws_url
-                        .set_scheme(if rpc_url.scheme() == "https" {
-                            "wss"
-                        } else {
-                            "ws"
-                        })
-                        .expect("valid scheme");
-                    if let Some(port) = ws_url.port() {
+                    // We only multiplex if the ws urls are actually websockets and only
+                    // then do we adjust the protocol.
+                    // We do not need to do this if we subscribe via GRPC, i.e. helius
+                    // laser which is more stable.
+                    let is_grpc = is_grpc_url(ws_url.as_ref());
+                    if !is_grpc {
                         ws_url
-                            .set_port(Some(port + 1))
-                            .expect("valid url with port");
+                            .set_scheme(if rpc_url.scheme() == "https" {
+                                "wss"
+                            } else {
+                                "ws"
+                            })
+                            .expect("valid scheme");
+                        if let Some(port) = ws_url.port() {
+                            ws_url
+                                .set_port(Some(port + 1))
+                                .expect("valid url with port");
+                        }
                     }
-                    vec![ws_url.to_string(); WS_MULTIPLEX_COUNT]
+                    vec![
+                        ws_url.to_string();
+                        if is_grpc { 1 } else { WS_MULTIPLEX_COUNT }
+                    ]
                 });
             (rpc_url.to_string(), ws_urls)
         }
@@ -88,10 +99,14 @@ pub fn remote_cluster_from_remote(
                 .first()
                 .expect("at least one ws url must be set for CustomWithWs")
                 .to_string();
-            let ws_urls = vec![ws_url; 3];
+            let is_grpc = is_grpc_url(&ws_url.to_string());
+            let ws_urls =
+                vec![ws_url; if is_grpc { 1 } else { WS_MULTIPLEX_COUNT }];
             (rpc_url, ws_urls)
         }
         CustomWithMultipleWs => {
+            // NOTE: we assume that if multiple ws urls are provided the user wants
+            //       to multiplex no matter if any is a GRPC based pubsub.
             let rpc_url = remote_config
                 .url
                 .as_ref()
@@ -138,3 +153,7 @@ fn allowed_program_ids_from_allowed_programs(
         None
     }
 }
+
+fn is_grpc_url(url: &str) -> bool {
+    is_helius_laser_url(url)
+}

magicblock-api/src/magic_validator.rs

@@ -23,8 +23,8 @@ use magicblock_aperture::{
 use magicblock_chainlink::{
     config::ChainlinkConfig,
     remote_account_provider::{
-        chain_pubsub_client::ChainPubsubClientImpl,
         chain_rpc_client::ChainRpcClientImpl,
+        chain_updates_client::ChainUpdatesClient,
     },
     submux::SubMuxClient,
     Chainlink,
@@ -98,7 +98,7 @@ use crate::{
 
 type ChainlinkImpl = Chainlink<
     ChainRpcClientImpl,
-    SubMuxClient<ChainPubsubClientImpl>,
+    SubMuxClient<ChainUpdatesClient>,
     AccountsDb,
     ChainlinkCloner,
 >;
@@ -392,9 +392,8 @@ impl MagicValidator {
         let endpoints = remote_cluster
             .ws_urls
             .iter()
-            .map(|pubsub_url| Endpoint {
-                rpc_url: rpc_url.clone(),
-                pubsub_url: pubsub_url.clone(),
+            .map(|pubsub_url| {
+                Endpoint::new(rpc_url.clone(), pubsub_url.to_string())
             })
             .collect::<Vec<_>>();
 

magicblock-chainlink/Cargo.toml

@@ -8,6 +8,7 @@ async-trait = { workspace = true }
 bincode = { workspace = true }
 env_logger = { workspace = true }
 futures-util = { workspace = true }
+helius-laserstream = { workspace = true }
 log = { workspace = true }
 lru = { workspace = true }
 magicblock-core = { workspace = true }

magicblock-chainlink/src/chainlink/mod.rs

@@ -17,8 +17,8 @@ use crate::{
     config::ChainlinkConfig,
     fetch_cloner::FetchAndCloneResult,
     remote_account_provider::{
-        ChainPubsubClient, ChainPubsubClientImpl, ChainRpcClient,
-        ChainRpcClientImpl, Endpoint, RemoteAccountProvider,
+        chain_updates_client::ChainUpdatesClient, ChainPubsubClient,
+        ChainRpcClient, ChainRpcClientImpl, Endpoint, RemoteAccountProvider,
     },
     submux::SubMuxClient,
 };
@@ -89,12 +89,7 @@ impl<T: ChainRpcClient, U: ChainPubsubClient, V: AccountsBank, C: Cloner>
         faucet_pubkey: Pubkey,
         config: ChainlinkConfig,
     ) -> ChainlinkResult<
-        Chainlink<
-            ChainRpcClientImpl,
-            SubMuxClient<ChainPubsubClientImpl>,
-            V,
-            C,
-        >,
+        Chainlink<ChainRpcClientImpl, SubMuxClient<ChainUpdatesClient>, V, C>,
     > {
         // Extract accounts provider and create fetch cloner while connecting
         // the subscription channel