From 89399763c53a093c8d86b1d477336a0ca055dcd4 Mon Sep 17 00:00:00 2001 From: Zeref Date: Tue, 16 Jun 2026 06:24:37 +0530 Subject: [PATCH 1/2] feat: adds a singleton blockhash cache updator --- Cargo.lock | 2 + magicblock-api/src/domain_registry_manager.rs | 43 ++++-- magicblock-api/src/magic_validator.rs | 34 ++++- .../src/committor_processor.rs | 22 +-- magicblock-committor-service/src/config.rs | 5 + magicblock-rpc-client/Cargo.toml | 2 + magicblock-rpc-client/src/lib.rs | 140 +++++++++++++++++- magicblock-validator-admin/src/claim_fees.rs | 32 +++- 8 files changed, 237 insertions(+), 43 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a13ea66c6..7ba7587a0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4225,6 +4225,7 @@ dependencies = [ name = "magicblock-rpc-client" version = "0.12.3" dependencies = [ + "arc-swap", "async-trait", "futures-util", "magicblock-metrics", @@ -4245,6 +4246,7 @@ dependencies = [ "solana-transaction-status-client-types", "thiserror 2.0.18", "tokio", + "tokio-util", "tracing", ] diff --git a/magicblock-api/src/domain_registry_manager.rs b/magicblock-api/src/domain_registry_manager.rs index 54deec76c..4d3c9be20 100644 --- a/magicblock-api/src/domain_registry_manager.rs +++ b/magicblock-api/src/domain_registry_manager.rs @@ -3,7 +3,7 @@ use std::{io, sync::Arc, thread, time::Duration}; use anyhow::Context; use borsh::BorshDeserialize; use magicblock_rpc_client::{ - MagicBlockSendTransactionConfig, MagicblockRpcClient, + BaseLayerBlockhash, MagicBlockSendTransactionConfig, MagicblockRpcClient, }; use mdp::{ consts::ER_RECORD_SEED, @@ -29,16 +29,25 @@ const UNREGISTER_CONFIRMATION_INTERVAL: Duration = Duration::from_millis(400); pub struct DomainRegistryManager { client: Arc, rpc_client: MagicblockRpcClient, + blockhash_registry: Option, } impl DomainRegistryManager { - pub fn new(url: impl ToString) -> Self { - Self::new_with_commitment(url, CommitmentConfig::confirmed()) + pub fn new( + url: impl ToString, + blockhash_registry: Option, + ) -> Self { + Self::new_with_commitment( + url, + CommitmentConfig::confirmed(), + blockhash_registry, + ) } pub fn new_with_commitment( url: impl ToString, commitment: CommitmentConfig, + blockhash_registry: Option, ) -> Self { let client = Arc::new(RpcClient::new_with_commitment( url.to_string(), @@ -47,6 +56,7 @@ impl DomainRegistryManager { Self { client: client.clone(), rpc_client: MagicblockRpcClient::new(client), + blockhash_registry, } } @@ -189,8 +199,9 @@ impl DomainRegistryManager { url: impl ToString, payer: &Keypair, validator_info: ErRecord, + blockhash_registry: Option, ) -> Result<(), Error> { - let manager = DomainRegistryManager::new(url); + let manager = DomainRegistryManager::new(url, blockhash_registry); manager.handle_registration(payer, validator_info).await } @@ -239,11 +250,19 @@ impl DomainRegistryManager { let instruction = Instruction::new_with_borsh(ID, &instruction, accounts); - let recent_blockhash = self - .client - .get_latest_blockhash() - .await - .context("Failed to get latest blockhash")?; + let recent_blockhash = match self + .blockhash_registry + .as_ref() + .and_then(|r| r.latest_blockhash()) + { + Some(cached) => cached, + None => self + .client + .get_latest_blockhash() + .await + .context("Failed to get latest blockhash")?, + }; + Ok(Transaction::new_signed_with_payer( &[instruction], Some(&payer.pubkey()), @@ -255,11 +274,13 @@ impl DomainRegistryManager { pub async fn handle_unregistration_static( url: impl ToString, payer: &Keypair, + blockhash_registry: Option, ) -> Result<(), Error> { info!("Unregistering validator from domain registry"); let manager = DomainRegistryManager::new_with_commitment( url, CommitmentConfig::confirmed(), + blockhash_registry, ); manager.unregister(payer).await } @@ -267,11 +288,13 @@ impl DomainRegistryManager { pub async fn send_unregistration_static( url: impl ToString, payer: &Keypair, + blockhash_registry: Option, ) -> Result { info!("Sending validator unregister transaction"); let manager = DomainRegistryManager::new_with_commitment( url, CommitmentConfig::confirmed(), + blockhash_registry, ); manager.send_unregister(payer).await } @@ -279,11 +302,13 @@ impl DomainRegistryManager { pub async fn send_unregistration_and_confirm_in_background_static( url: impl ToString, payer: &Keypair, + blockhash_registry: Option, ) -> Result<(Signature, thread::JoinHandle<()>), Error> { info!("Sending validator unregister transaction"); let manager = DomainRegistryManager::new_with_commitment( url, CommitmentConfig::confirmed(), + blockhash_registry, ); let signature = manager.send_unregister(payer).await?; let rpc_client = manager.rpc_client.clone(); diff --git a/magicblock-api/src/magic_validator.rs b/magicblock-api/src/magic_validator.rs index 7a3df2594..a6dfc6251 100644 --- a/magicblock-api/src/magic_validator.rs +++ b/magicblock-api/src/magic_validator.rs @@ -59,6 +59,7 @@ use magicblock_program::{ TransactionScheduler as ActionTransactionScheduler, }; use magicblock_replicator::{nats::Broker, ReplicationService}; +use magicblock_rpc_client::BaseLayerBlockhash; use magicblock_services::actions_callback_service::ActionsCallbackService; use magicblock_task_scheduler::{SchedulerDatabase, TaskSchedulerService}; use magicblock_validator_admin::claim_fees::{claim_fees, ClaimFeesTask}; @@ -127,6 +128,7 @@ pub struct MagicValidator { replication_tx: Sender, unregister_handle: Option>, is_standalone: bool, + blockhash_registry: BaseLayerBlockhash, } impl MagicValidator { @@ -220,10 +222,22 @@ impl MagicValidator { .then(Arc::::default); let step_start = Instant::now(); + let base_blockhash = BaseLayerBlockhash::default(); + base_blockhash + .spawn_blockhash_refresher( + Arc::new(RpcClient::new_with_commitment( + config.rpc_url().to_owned(), + CommitmentConfig::confirmed(), + )), + token.clone(), + ) + .await; + let committor_service = Self::init_committor_service( &config, ledger.latest_block(), shared_chain_slot.clone(), + base_blockhash.clone(), ) .await?; log_timing("startup", "committor_service_init", step_start); @@ -449,6 +463,7 @@ impl MagicValidator { replication_tx: validator_channels.replication_messages, unregister_handle: None, is_standalone, + blockhash_registry: base_blockhash, }) } @@ -457,6 +472,7 @@ impl MagicValidator { config: &ValidatorParams, latest_block: &LatestBlock, chain_slot: Option>, + blockhash_registry: BaseLayerBlockhash, ) -> ApiResult>> { let committor_persist_path = config.storage.join("committor_service.sqlite"); @@ -481,6 +497,7 @@ impl MagicValidator { config.commit.compute_unit_price, ), actions_timeout: DEFAULT_ACTIONS_TIMEOUT, + blockhash_registry: Some(blockhash_registry), }, chain_slot, actions_callback_executor, @@ -658,6 +675,7 @@ impl MagicValidator { config: &ChainOperationConfig, block_time_ms: u64, base_fee: u64, + blockhash_registry: Option, ) -> ApiResult<()> { let country_code = CountryCode::from(config.country_code.alpha3()); let validator_keypair = validator_authority(); @@ -676,6 +694,7 @@ impl MagicValidator { rpc_url, &validator_keypair, validator_info, + blockhash_registry, ) .await .map_err(|err| { @@ -703,7 +722,7 @@ impl MagicValidator { DomainRegistryManager::send_unregistration_and_confirm_in_background_static( rpc_url, &validator_keypair, - ) + Some(self.blockhash_registry.clone())) .await .map_err(|err| { ApiError::FailedToUnregisterValidatorOnChain(err.to_string()) @@ -851,6 +870,7 @@ impl MagicValidator { let chain_operation_config = self.config.chain_operation.clone(); let block_time_ms = self.config.ledger.block_time_ms(); let base_fee = self.config.validator.basefee; + let bh_registry = self.blockhash_registry.clone(); // Ephemeral mode does a non-blocking startup balance check. // Intentionally fire-and-forget: the task itself exits the process on failure. @@ -893,7 +913,9 @@ impl MagicValidator { if let Some(ref config) = chain_operation_config { if !config.claim_fees_frequency.is_zero() { let step_start = Instant::now(); - if let Err(err) = claim_fees(rpc_url.clone()).await { + if let Err(err) = + claim_fees(rpc_url.clone(), bh_registry.clone()).await + { error!( error = ?err, "Failed to claim validator fees on startup" @@ -913,6 +935,7 @@ impl MagicValidator { config, block_time_ms, base_fee, + Some(bh_registry.clone()), ) .await { @@ -1003,8 +1026,11 @@ impl MagicValidator { .map(|co| co.claim_fees_frequency) { let step_start = Instant::now(); - self.claim_fees_task - .start(frequency, self.config.rpc_url().to_owned()); + self.claim_fees_task.start( + frequency, + self.config.rpc_url().to_owned(), + self.blockhash_registry.clone(), + ); log_timing("startup", "claim_fees_task_start", step_start); } diff --git a/magicblock-committor-service/src/committor_processor.rs b/magicblock-committor-service/src/committor_processor.rs index 5061eb4b0..56f463ff6 100644 --- a/magicblock-committor-service/src/committor_processor.rs +++ b/magicblock-committor-service/src/committor_processor.rs @@ -71,22 +71,12 @@ impl CommittorProcessor { ); let rpc_client = Arc::new(rpc_client); let websocket_uri = chain_config.websocket_uri.clone(); - let magic_block_rpc_client = match (chain_slot, websocket_uri) { - (Some(chain_slot), websocket_uri) => { - MagicblockRpcClient::new_with_chain_slot_and_websocket( - rpc_client, - chain_slot, - websocket_uri, - ) - } - (None, Some(websocket_uri)) => { - MagicblockRpcClient::new_with_websocket( - rpc_client, - Some(websocket_uri), - ) - } - (None, None) => MagicblockRpcClient::new(rpc_client), - }; + let magic_block_rpc_client = MagicblockRpcClient::new_with_registry( + rpc_client, + chain_slot, + chain_config.websocket_uri.clone(), + chain_config.blockhash_registry.clone(), + ); // Create TableMania let gc_config = GarbageCollectorConfig::default(); diff --git a/magicblock-committor-service/src/config.rs b/magicblock-committor-service/src/config.rs index 7c1aee59e..d72a7e551 100644 --- a/magicblock-committor-service/src/config.rs +++ b/magicblock-committor-service/src/config.rs @@ -1,5 +1,6 @@ use std::time::Duration; +use magicblock_rpc_client::BaseLayerBlockhash; use solana_commitment_config::CommitmentConfig; use crate::compute_budget::ComputeBudgetConfig; @@ -13,6 +14,7 @@ pub struct ChainConfig { pub commitment: CommitmentConfig, pub compute_budget_config: ComputeBudgetConfig, pub actions_timeout: Duration, + pub blockhash_registry: Option, } impl ChainConfig { @@ -23,6 +25,7 @@ impl ChainConfig { commitment: CommitmentConfig::confirmed(), compute_budget_config, actions_timeout: DEFAULT_ACTIONS_TIMEOUT, + blockhash_registry: None, } } @@ -33,6 +36,7 @@ impl ChainConfig { commitment: CommitmentConfig::confirmed(), compute_budget_config, actions_timeout: DEFAULT_ACTIONS_TIMEOUT, + blockhash_registry: None, } } @@ -43,6 +47,7 @@ impl ChainConfig { commitment: CommitmentConfig::processed(), compute_budget_config, actions_timeout: DEFAULT_ACTIONS_TIMEOUT, + blockhash_registry: None, } } } diff --git a/magicblock-rpc-client/Cargo.toml b/magicblock-rpc-client/Cargo.toml index a713df830..e7fd21037 100644 --- a/magicblock-rpc-client/Cargo.toml +++ b/magicblock-rpc-client/Cargo.toml @@ -13,6 +13,8 @@ doctest = false [dependencies] tracing = { workspace = true } magicblock-metrics = { workspace = true } +arc-swap = { workspace = true } +tokio-util = { workspace = true } solana-pubsub-client = { workspace = true } solana-rpc-client = { workspace = true } solana-rpc-client-api = { workspace = true } diff --git a/magicblock-rpc-client/src/lib.rs b/magicblock-rpc-client/src/lib.rs index 900b4fb56..bb7bf9757 100644 --- a/magicblock-rpc-client/src/lib.rs +++ b/magicblock-rpc-client/src/lib.rs @@ -12,6 +12,7 @@ use std::{ time::{Duration, Instant}, }; +use arc_swap::ArcSwapOption; use futures_util::future::try_join_all; use serde_json::json; use signature_confirmer::{SignatureConfirmer, SignatureConfirmerConfig}; @@ -42,6 +43,7 @@ use solana_transaction_status_client_types::{ EncodedConfirmedTransactionWithStatusMeta, UiTransactionEncoding, }; use tokio::sync::Mutex as TMutex; +use tokio_util::sync::CancellationToken; use tracing::*; /// The encoding to use when sending transactions @@ -254,14 +256,108 @@ struct BlockhashCache { recent: HashMap, } -#[derive(Clone, Copy)] +#[derive(Clone, Copy, Debug)] struct CachedBlockhash { - blockhash: Hash, + pub blockhash: Hash, context_slot: Slot, last_valid_block_height: u64, fetched_at: Instant, } +/// Process-wide, shared source of truth for the latest base-layer +/// blockhash. A single background task writes it; all consumers read it +/// lock-free. Cloning is cheap (Arc bump) and all clones share state. +#[derive(Clone, Default, Debug)] +pub struct BaseLayerBlockhash { + inner: Arc>, +} + +impl BaseLayerBlockhash { + /// Lock-free read of the cached blockhash, if the refresher has + /// populated it at least once and it hasn't been invalidated. + pub fn get(&self) -> Option { + self.inner.load().as_deref().copied() + } + + /// Returns the cached base-layer blockhash, if available. + pub fn latest_blockhash(&self) -> Option { + self.get().map(|cached| cached.blockhash) + } + + fn store(&self, value: CachedBlockhash) { + self.inner.store(Some(Arc::new(value))); + } + + /// Drop the cached value so the next read falls back to a direct + /// fetch and the next refresh tick re-populates. Called by retry + /// paths after a transaction is rejected for a stale blockhash. + pub fn invalidate(&self) { + self.inner.store(None); + } + + /// Primes the cache synchronously (so the first reader never races an + /// empty cache) then spawns the single background refresher. + pub async fn spawn_blockhash_refresher( + &self, + rpc_client: Arc, + cancel: CancellationToken, + ) { + if let Err(err) = self.refresh_once(&rpc_client).await { + warn!(error = ?err, "Initial base-layer blockhash fetch failed"); + } + let cache = self.clone(); + tokio::spawn(async move { + let mut interval = tokio::time::interval(REFRESH_INTERVAL); + loop { + tokio::select! { + _ = interval.tick() => { + if let Err(err) = cache.refresh_once(&rpc_client).await { + warn!(error = ?err, "Failed to refresh base-layer blockhash"); + } + } + _ = cancel.cancelled() => break, + } + } + }); + } + + async fn refresh_once( + &self, + rpc_client: &RpcClient, + ) -> Result<(), Box> { + let resp: Response = rpc_client + .send( + RpcRequest::GetLatestBlockhash, + json!([rpc_client.commitment()]), + ) + .await + .map_err(|e| { + MagicBlockRpcClientError::GetLatestBlockhash(Box::new(e)) + })?; + let blockhash = resp.value.blockhash.parse().map_err(|_| { + MagicBlockRpcClientError::GetLatestBlockhash(Box::new( + RpcClientError::new_with_request( + RpcClientErrorKind::RpcError(RpcError::ParseError( + "Hash".to_string(), + )), + RpcRequest::GetLatestBlockhash, + ), + )) + })?; + + self.store(CachedBlockhash { + blockhash, + last_valid_block_height: resp.value.last_valid_block_height, + context_slot: resp.context.slot, + fetched_at: Instant::now(), + }); + + Ok(()) + } +} + +const REFRESH_INTERVAL: Duration = Duration::from_secs(10); + #[derive(Clone, Copy)] struct CachedSlot { slot: Slot, @@ -276,6 +372,7 @@ pub struct MagicblockRpcClient { cache: Arc, chain_slot: Option>, confirmer: Arc, + blockhash_registry: Option, } impl From for MagicblockRpcClient { @@ -287,21 +384,21 @@ impl From for MagicblockRpcClient { impl MagicblockRpcClient { /// Create a new [MagicBlockRpcClient] from an existing [RpcClient]. pub fn new(client: Arc) -> Self { - Self::new_with_options(client, None, None) + Self::new_with_options(client, None, None, None) } pub fn new_with_chain_slot( client: Arc, chain_slot: Arc, ) -> Self { - Self::new_with_options(client, Some(chain_slot), None) + Self::new_with_options(client, Some(chain_slot), None, None) } pub fn new_with_websocket( client: Arc, websocket_url: Option, ) -> Self { - Self::new_with_options(client, None, websocket_url) + Self::new_with_options(client, None, websocket_url, None) } pub fn new_with_chain_slot_and_websocket( @@ -309,13 +406,28 @@ impl MagicblockRpcClient { chain_slot: Arc, websocket_url: Option, ) -> Self { - Self::new_with_options(client, Some(chain_slot), websocket_url) + Self::new_with_options(client, Some(chain_slot), websocket_url, None) + } + + pub fn new_with_registry( + client: Arc, + chain_slot: Option>, + websocket_url: Option, + blockhash_registry: Option, + ) -> Self { + Self::new_with_options( + client, + chain_slot, + websocket_url, + blockhash_registry, + ) } fn new_with_options( client: Arc, chain_slot: Option>, websocket_url: Option, + blockhash_registry: Option, ) -> Self { let confirmer = Arc::new(SignatureConfirmer::new( client.clone(), @@ -326,12 +438,25 @@ impl MagicblockRpcClient { cache: Arc::new(RpcClientCache::default()), chain_slot, confirmer, + blockhash_registry, } } pub async fn get_latest_blockhash( &self, ) -> MagicBlockRpcClientResult { + // 1. Shared registry: lock-free, no RPC. + if let Some(ref bh_registry) = self.blockhash_registry { + if let Some(cached) = bh_registry.get() { + self.record_observed_slot(cached.context_slot).await; + + let mut local = self.cache.blockhash.lock().await; + Self::cache_blockhash(&mut local, cached); + return Ok(cached.blockhash); + } + } + + // 2. Legacy per-client cache + direct fetch let mut cached = self.cache.blockhash.lock().await; if let Some(blockhash) = Self::fresh_cached_blockhash(&cached) { return Ok(blockhash); @@ -412,6 +537,9 @@ impl MagicblockRpcClient { } pub async fn invalidate_cached_blockhash(&self) { + if let Some(registry) = &self.blockhash_registry { + registry.invalidate(); + } let mut cached = self.cache.blockhash.lock().await; cached.latest = None; } diff --git a/magicblock-validator-admin/src/claim_fees.rs b/magicblock-validator-admin/src/claim_fees.rs index 38bcb8571..55928a02e 100644 --- a/magicblock-validator-admin/src/claim_fees.rs +++ b/magicblock-validator-admin/src/claim_fees.rs @@ -2,7 +2,7 @@ use std::time::Duration; use dlp_api::instruction_builder::validator_claim_fees; use magicblock_program::validator::validator_authority; -use magicblock_rpc_client::MagicBlockRpcClientError; +use magicblock_rpc_client::{BaseLayerBlockhash, MagicBlockRpcClientError}; use solana_commitment_config::CommitmentConfig; use solana_rpc_client::nonblocking::rpc_client::RpcClient; use solana_signer::Signer; @@ -26,14 +26,24 @@ impl ClaimFeesTask { } } - pub fn start(&mut self, tick_period: Duration, url: String) { + pub fn start( + &mut self, + tick_period: Duration, + url: String, + blockhash_registry: BaseLayerBlockhash, + ) { if self.handle.is_some() { error!("Claim fees task already started"); return; } let token = self.token.clone(); - let handle = tokio::spawn(run_claim_fees_loop(token, tick_period, url)); + let handle = tokio::spawn(run_claim_fees_loop( + token, + tick_period, + url, + blockhash_registry, + )); self.handle = Some(handle); } @@ -63,6 +73,7 @@ async fn run_claim_fees_loop( token: CancellationToken, tick_period: Duration, url: String, + blockhash_registry: BaseLayerBlockhash, ) { info!("Starting claim fees task"); let start_time = Instant::now() + tick_period; @@ -70,7 +81,7 @@ async fn run_claim_fees_loop( loop { tokio::select! { _ = interval.tick() => { - if let Err(err) = claim_fees(url.clone()).await { + if let Err(err) = claim_fees(url.clone(), blockhash_registry.clone()).await { error!(error = ?err, "Failed to claim fees"); } }, @@ -81,7 +92,10 @@ async fn run_claim_fees_loop( } #[instrument(fields(validator = %validator_authority().pubkey()))] -pub async fn claim_fees(url: String) -> Result<(), MagicBlockRpcClientError> { +pub async fn claim_fees( + url: String, + blockhash_registry: BaseLayerBlockhash, +) -> Result<(), MagicBlockRpcClientError> { info!("Claiming validator fees"); let rpc_client = @@ -108,10 +122,12 @@ pub async fn claim_fees(url: String) -> Result<(), MagicBlockRpcClientError> { let ix = validator_claim_fees(validator, None); - let latest_blockhash = - rpc_client.get_latest_blockhash().await.map_err(|e| { + let latest_blockhash = match blockhash_registry.latest_blockhash() { + Some(cached_bh) => cached_bh, + None => rpc_client.get_latest_blockhash().await.map_err(|e| { MagicBlockRpcClientError::GetLatestBlockhash(Box::new(e)) - })?; + })?, + }; let tx = Transaction::new_signed_with_payer( &[ix], From 7e937273f1a92d5f32bc93142c53be9d2253c762 Mon Sep 17 00:00:00 2001 From: Zeref Date: Fri, 19 Jun 2026 09:35:40 +0530 Subject: [PATCH 2/2] chore: change BaseLayerBlockhash::get visibility --- magicblock-rpc-client/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/magicblock-rpc-client/src/lib.rs b/magicblock-rpc-client/src/lib.rs index bb7bf9757..78863e0b5 100644 --- a/magicblock-rpc-client/src/lib.rs +++ b/magicblock-rpc-client/src/lib.rs @@ -275,7 +275,7 @@ pub struct BaseLayerBlockhash { impl BaseLayerBlockhash { /// Lock-free read of the cached blockhash, if the refresher has /// populated it at least once and it hasn't been invalidated. - pub fn get(&self) -> Option { + fn get(&self) -> Option { self.inner.load().as_deref().copied() }