Reduce RPC hot-path requests

magicblock-api/src/magic_validator.rs

@@ -1,7 +1,7 @@
 use std::{
     path::{Path, PathBuf},
     sync::{
-        atomic::{AtomicBool, Ordering},
+        atomic::{AtomicBool, AtomicU64, Ordering},
         Arc,
     },
     thread,
@@ -211,11 +211,19 @@ impl MagicValidator {
         };
         let accountsdb = Arc::new(accountsdb);
         let (mut dispatch, validator_channels) = link();
+        let shared_chain_slot =
+            (!Self::replication_mode_uses_disabled_chainlink(
+                &config.validator.replication_mode,
+            ))
+            .then(Arc::<AtomicU64>::default);
 
         let step_start = Instant::now();
-        let committor_service =
-            Self::init_committor_service(&config, ledger.latest_block())
-                .await?;
+        let committor_service = Self::init_committor_service(
+            &config,
+            ledger.latest_block(),
+            shared_chain_slot.clone(),
+        )
+        .await?;
         log_timing("startup", "committor_service_init", step_start);
         init_magic_sys(Arc::new(MagicSysAdapter::new(
             committor_service.clone(),
@@ -228,6 +236,7 @@ impl MagicValidator {
                 &dispatch.transaction_scheduler,
                 &ledger.latest_block().clone(),
                 &accountsdb,
+                shared_chain_slot.clone(),
             )
             .await?,
         );
@@ -443,6 +452,7 @@ impl MagicValidator {
     async fn init_committor_service(
         config: &ValidatorParams,
         latest_block: &LatestBlock,
+        chain_slot: Option<Arc<AtomicU64>>,
     ) -> ApiResult<Option<Arc<CommittorService>>> {
         let committor_persist_path =
             config.storage.join("committor_service.sqlite");
@@ -464,6 +474,7 @@ impl MagicValidator {
                 ),
                 actions_timeout: DEFAULT_ACTIONS_TIMEOUT,
             },
+            chain_slot,
             actions_callback_executor,
         )?));
 
@@ -477,6 +488,7 @@ impl MagicValidator {
         transaction_scheduler: &TransactionSchedulerHandle,
         latest_block: &LatestBlock,
         accountsdb: &Arc<AccountsDb>,
+        chain_slot: Option<Arc<AtomicU64>>,
     ) -> ApiResult<ChainlinkImpl> {
         if Self::replication_mode_uses_disabled_chainlink(
             &config.validator.replication_mode,
@@ -536,6 +548,7 @@ impl MagicValidator {
             chainlink_config,
             &config.chainlink,
             config.storage.as_path(),
+            chain_slot.unwrap_or_default(),
         )
         .await?;
 

magicblock-chainlink/src/chainlink/mod.rs

@@ -1,4 +1,8 @@
-use std::{collections::HashSet, path::Path, sync::Arc};
+use std::{
+    collections::HashSet,
+    path::Path,
+    sync::{atomic::AtomicU64, Arc},
+};
 
 use dlp_api::pda::ephemeral_balance_pda_from_payer;
 use errors::{ChainlinkError, ChainlinkResult};
@@ -320,6 +324,7 @@ impl<T: ChainRpcClient, U: ChainPubsubClient, V: AccountsBank, C: Cloner>
         config: ChainlinkConfig,
         chainlink_config: &ChainLinkConfig,
         ledger_path: &Path,
+        chain_slot: Arc<AtomicU64>,
     ) -> ChainlinkResult<
         InnerChainlink<
             ChainRpcClientImpl,
@@ -337,6 +342,7 @@ impl<T: ChainRpcClient, U: ChainPubsubClient, V: AccountsBank, C: Cloner>
             commitment,
             tx,
             &config.remote_account_provider,
+            Some(chain_slot),
         )
         .await?;
         let fetch_cloner = if let Some(provider) = account_provider {

magicblock-chainlink/src/remote_account_provider/mod.rs

@@ -494,6 +494,7 @@ impl
         commitment: CommitmentConfig,
         subscription_forwarder: mpsc::Sender<ForwardedSubscriptionUpdate>,
         config: &RemoteAccountProviderConfig,
+        chain_slot: Option<Arc<AtomicU64>>,
     ) -> ChainlinkResult<
         Option<
             RemoteAccountProvider<
@@ -513,6 +514,7 @@ impl
                 commitment,
                 subscription_forwarder,
                 config,
+                chain_slot.unwrap_or_default(),
             )
             .await?;
             Ok(Some(provider))
@@ -651,6 +653,7 @@ impl<T: ChainRpcClient, U: ChainPubsubClient> RemoteAccountProvider<T, U> {
         commitment: CommitmentConfig,
         subscription_forwarder: mpsc::Sender<ForwardedSubscriptionUpdate>,
         config: &RemoteAccountProviderConfig,
+        chain_slot: Arc<AtomicU64>,
     ) -> RemoteAccountProviderResult<
         RemoteAccountProvider<
             ChainRpcClientImpl,
@@ -674,9 +677,6 @@ impl<T: ChainRpcClient, U: ChainPubsubClient> RemoteAccountProvider<T, U> {
         let rpc_client =
             ChainRpcClientImpl::new_from_url(rpc_url.as_str(), commitment);
 
-        // Create chain_slot to be shared with all pubsub clients
-        let chain_slot = Arc::<AtomicU64>::default();
-
         // Build startup pubsub clients and wrap them into a SubMuxClient.
         // gRPC clients are cheap to create and backfill subscriptions, so
         // when present we let slower WebSocket clients attach after startup.

magicblock-committor-service/src/committor_processor.rs

@@ -1,7 +1,7 @@
 use std::{
     collections::{HashMap, HashSet},
     path::Path,
-    sync::Arc,
+    sync::{atomic::AtomicU64, Arc},
 };
 
 use magicblock_core::traits::ActionsCallbackScheduler;
@@ -48,6 +48,7 @@ impl CommittorProcessor {
         authority: Keypair,
         persist_file: P,
         chain_config: ChainConfig,
+        chain_slot: Option<Arc<AtomicU64>>,
         actions_callback_executor: A,
     ) -> CommittorServiceResult<Self>
     where
@@ -59,7 +60,11 @@ impl CommittorProcessor {
             chain_config.commitment,
         );
         let rpc_client = Arc::new(rpc_client);
-        let magic_block_rpc_client = MagicblockRpcClient::new(rpc_client);
+        let magic_block_rpc_client = if let Some(chain_slot) = chain_slot {
+            MagicblockRpcClient::new_with_chain_slot(rpc_client, chain_slot)
+        } else {
+            MagicblockRpcClient::new(rpc_client)
+        };
 
         // Create TableMania
         let gc_config = GarbageCollectorConfig::default();

magicblock-committor-service/src/intent_executor/intent_execution_client.rs

@@ -37,6 +37,10 @@ impl IntentExecutionClient {
         IntentExecutionClient { rpc_client }
     }
 
+    pub(in crate::intent_executor) async fn invalidate_cached_blockhash(&self) {
+        self.rpc_client.invalidate_cached_blockhash().await;
+    }
+
     pub(in crate::intent_executor) async fn execute_message_with_retries(
         &self,
         authority: &Keypair,

magicblock-committor-service/src/intent_executor/single_stage_executor.rs

@@ -114,6 +114,7 @@ where
                     break Err(execution_err);
                 }
             };
+            self.intent_client.invalidate_cached_blockhash().await;
             self.execution_report.dispose(cleanup);
 
             if self.current_attempt >= RECURSION_CEILING {

magicblock-committor-service/src/intent_executor/two_stage_executor.rs

@@ -141,6 +141,7 @@ where
                     break Err(execution_err);
                 }
             };
+            self.intent_client.invalidate_cached_blockhash().await;
             self.execution_report.dispose(cleanup);
 
             if self.state.current_attempt >= Self::RECURSION_CEILING {
@@ -401,6 +402,7 @@ where
                     break Err(execution_err);
                 }
             };
+            self.intent_client.invalidate_cached_blockhash().await;
             self.execution_report.dispose(cleanup);
 
             if self.state.current_attempt >= Self::RECURSION_CEILING {

magicblock-committor-service/src/service.rs

@@ -1,4 +1,9 @@
-use std::{collections::HashMap, path::Path, sync::Arc, time::Instant};
+use std::{
+    collections::HashMap,
+    path::Path,
+    sync::{atomic::AtomicU64, Arc},
+    time::Instant,
+};
 
 use magicblock_core::traits::ActionsCallbackScheduler;
 use magicblock_program::magic_scheduled_base_intent::ScheduledIntentBundle;
@@ -113,6 +118,7 @@ impl CommittorActor {
         authority: Keypair,
         persist_file: P,
         chain_config: ChainConfig,
+        chain_slot: Option<Arc<AtomicU64>>,
         actions_callback_executor: A,
     ) -> CommittorServiceResult<Self>
     where
@@ -123,6 +129,7 @@ impl CommittorActor {
             authority,
             persist_file,
             chain_config,
+            chain_slot,
             actions_callback_executor,
         )?);
 
@@ -316,6 +323,7 @@ impl CommittorService {
         authority: Keypair,
         persist_file: P,
         chain_config: ChainConfig,
+        chain_slot: Option<Arc<AtomicU64>>,
         actions_callback_executor: A,
     ) -> CommittorServiceResult<Self>
     where
@@ -332,6 +340,7 @@ impl CommittorService {
                 authority,
                 persist_file,
                 chain_config,
+                chain_slot,
                 actions_callback_executor,
             )?;
             tokio::spawn(async move {

magicblock-committor-service/src/transaction_preparator/delivery_preparator.rs

@@ -193,6 +193,7 @@ impl DeliveryPreparator {
         let cleanup_task = preparation_task.cleanup_task();
 
         self.cleanup(authority, &[cleanup_task], &[]).await?;
+        self.rpc_client.invalidate_cached_blockhash().await;
 
         // Restore preparation stage for retry
         *stage = CommitBufferStage::Preparation(preparation_task);

magicblock-rpc-client/Cargo.toml

@@ -29,6 +29,7 @@ solana-address-lookup-table-interface = { workspace = true, features = [
   "bytemuck",
 ] }
 futures-util = { workspace = true }
+serde_json = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }