use blockhash_watcher (#41)

madninja · web-flow · commit 067a3da072c1 · 2025-05-08T13:13:12.000-07:00
diff --git a/solana-transaction-utils/src/blockhash_watcher.rs b/solana-transaction-utils/src/blockhash_watcher.rs
@@ -0,0 +1,95 @@
+use crate::Error;
+use futures::{future, TryFutureExt};
+use solana_client::nonblocking::rpc_client::RpcClient;
+use solana_sdk::hash::Hash;
+use std::{sync::Arc, time::Duration};
+use tokio::{sync::watch, time};
+use tokio_graceful_shutdown::SubsystemHandle;
+use tracing::{info, warn};
+
+pub type MessageSender = watch::Sender<BlockHashData>;
+pub type MessageReceiver = watch::Receiver<BlockHashData>;
+pub const BLOCKHASH_REFRESH_INTERVAL: Duration = Duration::from_secs(30);
+
+pub fn last_valid<T>(receiver: &watch::Receiver<T>) -> watch::Ref<'_, T>
+where
+    T: Clone,
+{
+    receiver.borrow()
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct BlockHashData {
+    pub last_valid_block_height: u64,
+    pub last_valid_blockhash: Hash,
+    pub current_block_height: u64,
+}
+
+#[derive(Clone)]
+pub struct BlockhashWatcher {
+    watch: MessageSender,
+    interval: Duration,
+    client: Arc<RpcClient>,
+}
+
+impl BlockhashWatcher {
+    pub fn new(interval: Duration, client: Arc<RpcClient>) -> Self {
+        let (watch, _) = watch::channel(Default::default());
+        Self {
+            watch,
+            interval,
+            client,
+        }
+    }
+
+    pub fn watcher(&mut self) -> MessageReceiver {
+        self.watch.subscribe()
+    }
+
+    pub async fn run(mut self, shutdown: SubsystemHandle) -> Result<(), Error> {
+        info!("starting");
+        let mut interval = time::interval(self.interval);
+        loop {
+            tokio::select! {
+                _ = shutdown.on_shutdown_requested() => {
+                    info!("shutting down");
+                    return Ok(());
+                }
+                _ = interval.tick() => {
+                        match self.fetch_data(&shutdown).await {
+                            Ok(Some(new_data)) => {
+                                let _ = self.watch.send_replace(new_data);
+                            }
+                            Ok(None) => (),
+                            Err(err) => warn!(?err, "failed to get block hash data"),
+                        };
+                }
+            }
+        }
+    }
+
+    pub async fn fetch_data(
+        &mut self,
+        shutdown: &SubsystemHandle,
+    ) -> Result<Option<BlockHashData>, Error> {
+        let fetch_fut = future::try_join(
+            self.client
+                .get_latest_blockhash_with_commitment(self.client.commitment()),
+            self.client.get_block_height(),
+        )
+        .map_err(Error::from)
+        .map_ok(
+            |((last_valid_blockhash, last_valid_block_height), current_block_height)| {
+                BlockHashData {
+                    last_valid_block_height,
+                    last_valid_blockhash,
+                    current_block_height,
+                }
+            },
+        );
+        tokio::select! {
+            result = fetch_fut => result.map(Some),
+            _ = shutdown.on_shutdown_requested() => Ok(None)
+        }
+    }
+}
diff --git a/solana-transaction-utils/src/error.rs b/solana-transaction-utils/src/error.rs
@@ -29,10 +29,10 @@ pub enum Error {
     FeeTooHigh,
     #[error("Transaction has failed too many retries and gone stale")]
     StaleTransaction,
-    #[error("Channel send error: {0}")]
-    ChannelSendError(String),
     #[error("System time error: {0}")]
     SystemTimeError(String),
+    #[error("message channel closed")]
+    ChannelClosed,
 }
 
 impl From<solana_client::client_error::ClientError> for Error {
@@ -48,8 +48,8 @@ impl From<TpuSenderError> for Error {
 }
 
 impl<T> From<tokio::sync::mpsc::error::SendError<T>> for Error {
-    fn from(value: tokio::sync::mpsc::error::SendError<T>) -> Self {
-        Self::ChannelSendError(value.to_string())
+    fn from(_value: tokio::sync::mpsc::error::SendError<T>) -> Self {
+        Self::ChannelClosed
     }
 }
 
@@ -61,4 +61,8 @@ impl Error {
     pub fn serialization<S: ToString>(str: S) -> Self {
         Self::SerializationError(str.to_string())
     }
+
+    pub fn channel_closed() -> Error {
+        Error::ChannelClosed
+    }
 }
diff --git a/solana-transaction-utils/src/lib.rs b/solana-transaction-utils/src/lib.rs
@@ -1,6 +1,10 @@
+pub mod blockhash_watcher;
 pub mod error;
 pub mod pack;
 pub mod priority_fee;
 pub mod queue;
 pub mod send_and_confirm_transactions_in_parallel;
 pub mod sender;
+pub mod sync;
+
+pub use error::Error;
diff --git a/solana-transaction-utils/src/sender.rs b/solana-transaction-utils/src/sender.rs
@@ -1,11 +1,9 @@
-use std::{
-    sync::{
-        atomic::{AtomicU64, Ordering},
-        Arc,
-    },
-    time::Duration,
+use crate::{
+    blockhash_watcher,
+    error::Error,
+    queue::{CompletedTransactionTask, TransactionTask},
+    send_and_confirm_transactions_in_parallel::QuicTpuClient,
 };
-
 use dashmap::DashMap;
 use futures::{stream, StreamExt, TryStreamExt};
 use itertools::Itertools;
@@ -14,29 +12,19 @@ use solana_client::{
     tpu_client::TpuClientConfig,
 };
 use solana_sdk::{
-    hash::Hash,
     instruction::Instruction,
     message::{v0, AddressLookupTableAccount, VersionedMessage},
     signature::{Keypair, Signature},
     signer::Signer,
     transaction::VersionedTransaction,
 };
-use tokio::sync::{
-    mpsc::{Receiver, Sender},
-    RwLock,
-};
+use std::{sync::Arc, time::Duration};
+use tokio::sync::mpsc::{Receiver, Sender};
 use tokio_graceful_shutdown::{SubsystemBuilder, SubsystemHandle};
 use tracing::{error, info};
 
-use crate::{
-    error::Error,
-    queue::{CompletedTransactionTask, TransactionTask},
-    send_and_confirm_transactions_in_parallel::QuicTpuClient,
-};
-
 const SEND_TIMEOUT: Duration = Duration::from_secs(5);
 const CONFIRMATION_CHECK_INTERVAL: Duration = Duration::from_secs(5);
-const BLOCKHASH_REFRESH_INTERVAL: Duration = Duration::from_secs(30);
 const MAX_GET_SIGNATURE_STATUSES_QUERY_ITEMS: usize = 100;
 
 #[derive(Clone, Debug)]
@@ -71,16 +59,8 @@ struct TransactionData<T: Send + Clone> {
     sent_to_rpc: bool,
 }
 
-#[derive(Clone, Debug, Copy)]
-pub struct BlockHashData {
-    last_valid_block_height: u64,
-    blockhash: Hash,
-}
-
 pub struct TransactionSender<T: Send + Clone + Sync> {
     unconfirmed_txs: Arc<DashMap<Signature, TransactionData<T>>>,
-    blockhash_data: Arc<RwLock<BlockHashData>>,
-    current_block_height: Arc<AtomicU64>,
     rpc_client: Arc<RpcClient>,
     tpu_client: Option<QuicTpuClient>,
     ws_url: String,
@@ -90,37 +70,6 @@ pub struct TransactionSender<T: Send + Clone + Sync> {
     last_tpu_use: u64,
 }
 
-pub async fn spawn_background_tasks(
-    handle: SubsystemHandle,
-    blockhash_data: Arc<RwLock<BlockHashData>>,
-    current_block_height: Arc<AtomicU64>,
-    rpc_client: Arc<RpcClient>,
-) -> Result<(), Error> {
-    // Spawn blockhash updater
-    let mut interval = tokio::time::interval(BLOCKHASH_REFRESH_INTERVAL);
-    loop {
-        tokio::select! {
-            _ = handle.on_shutdown_requested() => {
-                return Ok(());
-            }
-            _ = interval.tick() => {
-                if let Ok((blockhash, last_valid_block_height)) = rpc_client
-                    .get_latest_blockhash_with_commitment(rpc_client.commitment())
-                    .await
-                {
-                    *blockhash_data.write().await = BlockHashData {
-                        last_valid_block_height,
-                        blockhash,
-                    };
-                }
-                if let Ok(block_height) = rpc_client.get_block_height().await {
-                    current_block_height.store(block_height, Ordering::Relaxed);
-                }
-            }
-        }
-    }
-}
-
 impl<T: Send + Clone + Sync> TransactionSender<T> {
     pub async fn new(
         rpc_client: Arc<RpcClient>,
@@ -129,24 +78,8 @@ impl<T: Send + Clone + Sync> TransactionSender<T> {
         result_tx: Sender<CompletedTransactionTask<T>>,
         max_re_sign_count: u32,
     ) -> Result<Self, Error> {
-        // Initialize blockhash data
-        let (blockhash, last_valid_block_height) = rpc_client
-            .get_latest_blockhash_with_commitment(rpc_client.commitment())
-            .await?;
-
-        let blockhash_data = Arc::new(RwLock::new(BlockHashData {
-            last_valid_block_height,
-            blockhash,
-        }));
-
-        // Initialize block height
-        let block_height = rpc_client.get_block_height().await?;
-        let current_block_height = Arc::new(AtomicU64::new(block_height));
-
         Ok(Self {
             unconfirmed_txs: Arc::new(DashMap::new()),
-            blockhash_data,
-            current_block_height,
             rpc_client,
             tpu_client: None,
             ws_url,
@@ -193,8 +126,9 @@ impl<T: Send + Clone + Sync> TransactionSender<T> {
     pub async fn handle_packed_tx(
         &self,
         packed_tx: PackedTransactionWithTasks<T>,
+        blockhash_rx: &blockhash_watcher::MessageReceiver,
     ) -> Result<(), Error> {
-        if let Err(e) = self.process_packed_tx(&packed_tx).await {
+        if let Err(e) = self.process_packed_tx(&packed_tx, blockhash_rx).await {
             // Handle processing error by notifying all tasks
             stream::iter(packed_tx.tasks)
                 .map(|task| Ok((task, Some(e.clone()))))
@@ -212,13 +146,14 @@ impl<T: Send + Clone + Sync> TransactionSender<T> {
     pub async fn process_packed_tx(
         &self,
         packed: &PackedTransactionWithTasks<T>,
+        blockhash_rx: &blockhash_watcher::MessageReceiver,
     ) -> Result<(), Error> {
         // Check if transaction has been resigned too many times
         if packed.re_sign_count >= self.max_re_sign_count {
             return Err(Error::StaleTransaction);
         }
 
-        let blockhash = self.blockhash_data.read().await.blockhash;
+        let blockhash = blockhash_rx.borrow().last_valid_blockhash;
         let message = v0::Message::try_compile(
             &self.payer.pubkey(),
             &packed.instructions,
@@ -240,7 +175,7 @@ impl<T: Send + Clone + Sync> TransactionSender<T> {
             signature,
             TransactionData {
                 packed_tx: packed.clone(),
-                last_valid_block_height: self.blockhash_data.read().await.last_valid_block_height,
+                last_valid_block_height: blockhash_rx.borrow().last_valid_block_height,
                 serialized_tx: serialized,
                 sent_to_rpc: false,
             },
@@ -262,8 +197,11 @@ impl<T: Send + Clone + Sync> TransactionSender<T> {
         }
     }
 
-    async fn check_and_retry(&mut self) -> Result<(), Error> {
-        let current_height = self.current_block_height.load(Ordering::Relaxed);
+    async fn check_and_retry(
+        &mut self,
+        blockhash_rx: &blockhash_watcher::MessageReceiver,
+    ) -> Result<(), Error> {
+        let current_height = blockhash_rx.borrow().current_block_height;
         let current_time = match std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)
         {
             Ok(duration) => duration.as_secs(),
@@ -337,7 +275,7 @@ impl<T: Send + Clone + Sync> TransactionSender<T> {
                                 self.unconfirmed_txs.remove(&signature);
                                 // Create new packed transaction with incremented resign count
                                 let new_packed = data.packed_tx.with_incremented_re_sign_count();
-                                self.handle_packed_tx(new_packed).await?;
+                                self.handle_packed_tx(new_packed, blockhash_rx).await?;
                             }
                         }
                     }
@@ -346,7 +284,7 @@ impl<T: Send + Clone + Sync> TransactionSender<T> {
                     self.unconfirmed_txs.remove(&signature);
                     // Create new packed transaction with incremented resign count
                     let new_packed = data.packed_tx.with_incremented_re_sign_count();
-                    self.handle_packed_tx(new_packed).await?;
+                    self.handle_packed_tx(new_packed, blockhash_rx).await?;
                 }
             }
         }
@@ -359,16 +297,17 @@ impl<T: Send + Clone + Sync> TransactionSender<T> {
         mut rx: Receiver<PackedTransactionWithTasks<T>>,
         handle: SubsystemHandle,
     ) -> Result<(), Error> {
+        let mut blockhash_watcher = blockhash_watcher::BlockhashWatcher::new(
+            blockhash_watcher::BLOCKHASH_REFRESH_INTERVAL,
+            self.rpc_client.clone(),
+        );
         handle.start(SubsystemBuilder::new("blockhash-updater", {
-            let blockhash_data = self.blockhash_data.clone();
-            let current_block_height = self.current_block_height.clone();
-            let rpc_client = self.rpc_client.clone();
-            move |handle| {
-                spawn_background_tasks(handle, blockhash_data, current_block_height, rpc_client)
-            }
+            let watcher = blockhash_watcher.clone();
+            move |handle| watcher.run(handle)
         }));
 
         let mut check_interval = tokio::time::interval(CONFIRMATION_CHECK_INTERVAL);
+        let blockchain_rx = blockhash_watcher.watcher();
 
         loop {
             tokio::select! {
@@ -379,10 +318,10 @@ impl<T: Send + Clone + Sync> TransactionSender<T> {
                     return Ok(());
                 }
                 Some(packed_tx) = rx.recv() => {
-                    self.handle_packed_tx(packed_tx).await?;
+                    self.handle_packed_tx(packed_tx, &blockchain_rx).await?;
                 }
                 _ = check_interval.tick() => {
-                    self.check_and_retry().await?;
+                    self.check_and_retry(&blockchain_rx).await?;
                 }
             }
         }
diff --git a/solana-transaction-utils/src/sync.rs b/solana-transaction-utils/src/sync.rs
diff --git a/tuktuk-crank-turner/src/task_processor.rs b/tuktuk-crank-turner/src/task_processor.rs

Original file line number	Diff line number	Diff line change
`@@ -29,10 +29,10 @@ pub enum Error {`
`29`	`29`	`FeeTooHigh,`
`30`	`30`	`#[error("Transaction has failed too many retries and gone stale")]`
`31`	`31`	`StaleTransaction,`
`32`		`- #[error("Channel send error: {0}")]`
`33`		`- ChannelSendError(String),`
`34`	`32`	`#[error("System time error: {0}")]`
`35`	`33`	`SystemTimeError(String),`
	`34`	`+ #[error("message channel closed")]`
	`35`	`+ ChannelClosed,`
`36`	`36`	`}`
`37`	`37`
`38`	`38`	`impl From<solana_client::client_error::ClientError> for Error {`
`@@ -48,8 +48,8 @@ impl From<TpuSenderError> for Error {`
`48`	`48`	`}`
`49`	`49`
`50`	`50`	`impl<T> From<tokio::sync::mpsc::error::SendError<T>> for Error {`
`51`		`- fn from(value: tokio::sync::mpsc::error::SendError<T>) -> Self {`
`52`		`- Self::ChannelSendError(value.to_string())`
	`51`	`+ fn from(_value: tokio::sync::mpsc::error::SendError<T>) -> Self {`
	`52`	`+ Self::ChannelClosed`
`53`	`53`	`}`
`54`	`54`	`}`
`55`	`55`
`@@ -61,4 +61,8 @@ impl Error {`
`61`	`61`	`pub fn serialization<S: ToString>(str: S) -> Self {`
`62`	`62`	`Self::SerializationError(str.to_string())`
`63`	`63`	`}`
	`64`	`+`
	`65`	`+ pub fn channel_closed() -> Error {`
	`66`	`+ Error::ChannelClosed`
	`67`	`+ }`
`64`	`68`	`}`