Move timeout to network level

Author: ali-behjati

config/config.toml

@@ -67,6 +67,9 @@ key_store.root_path = "/path/to/keystore"
 # This can be omitted when oracle.subscriber_enabled is set to false.
 # wss_url = "ws://api.devnet.solana.com"
 
+# Timeout for the requests to the RPC
+# rpc_timeout = "10s"
+
 # Path to the key store.
 # key_store.root_path = "/path/to/keystore"
 
@@ -89,9 +92,6 @@ key_store.root_path = "/path/to/keystore"
 # Age after which a price update is considered stale and not published
 # exporter.staleness_threshold = "5s"
 
-# RPC timeout for the requests to the RPC for sending price updates and monitoring them
-# exporter.rpc_timeout = "10s"
-
 # Maximum size of a batch
 # exporter.max_batch_size = 12
 

src/agent/solana.rs

@@ -24,6 +24,7 @@ pub mod network {
             Serialize,
         },
         slog::Logger,
+        std::time::Duration,
         tokio::{
             sync::{
                 mpsc,
@@ -38,25 +39,29 @@ pub mod network {
     #[serde(default)]
     pub struct Config {
         /// HTTP RPC endpoint
-        pub rpc_url:   String,
+        pub rpc_url:     String,
         /// WSS RPC endpoint
-        pub wss_url:   String,
+        pub wss_url:     String,
+        /// Timeout for the requests to the RPC
+        #[serde(with = "humantime_serde")]
+        pub rpc_timeout: Duration,
         /// Keystore
-        pub key_store: key_store::Config,
+        pub key_store:   key_store::Config,
         /// Configuration for the Oracle reading data from this network
-        pub oracle:    oracle::Config,
+        pub oracle:      oracle::Config,
         /// Configuration for the Exporter publishing data to this network
-        pub exporter:  exporter::Config,
+        pub exporter:    exporter::Config,
     }
 
     impl Default for Config {
         fn default() -> Self {
             Self {
-                rpc_url:   "http://localhost:8899".to_string(),
-                wss_url:   "ws://localhost:8900".to_string(),
-                key_store: Default::default(),
-                oracle:    Default::default(),
-                exporter:  Default::default(),
+                rpc_url:     "http://localhost:8899".to_string(),
+                wss_url:     "ws://localhost:8900".to_string(),
+                rpc_timeout: Duration::from_secs(10),
+                key_store:   Default::default(),
+                oracle:      Default::default(),
+                exporter:    Default::default(),
             }
         }
     }
@@ -72,6 +77,7 @@ pub mod network {
             config.oracle.clone(),
             &config.rpc_url,
             &config.wss_url,
+            config.rpc_timeout,
             KeyStore::new(config.key_store.clone())?,
             global_store_update_tx,
             logger.clone(),
@@ -81,6 +87,7 @@ pub mod network {
         let exporter_jhs = exporter::spawn_exporter(
             config.exporter,
             &config.rpc_url,
+            config.rpc_timeout,
             KeyStore::new(config.key_store.clone())?,
             local_store_tx,
             logger,

src/agent/solana/exporter.rs

@@ -97,9 +97,6 @@ pub struct Config {
     /// Age after which a price update is considered stale and not published
     #[serde(with = "humantime_serde")]
     pub staleness_threshold:                     Duration,
-    /// RPC timeout for the requests to the RPC for sending price updates and monitoring them
-    #[serde(with = "humantime_serde")]
-    pub rpc_timeout:                             Duration,
     /// Maximum size of a batch
     pub max_batch_size:                          usize,
     /// Capacity of the channel between the Exporter and the Transaction Monitor
@@ -120,7 +117,6 @@ impl Default for Config {
             refresh_network_state_interval_duration: Duration::from_millis(200),
             publish_interval_duration:               Duration::from_secs(1),
             staleness_threshold:                     Duration::from_secs(5),
-            rpc_timeout:                             Duration::from_secs(10),
             max_batch_size:                          12,
             inflight_transactions_channel_capacity:  10000,
             transaction_monitor:                     Default::default(),
@@ -134,14 +130,16 @@ impl Default for Config {
 pub fn spawn_exporter(
     config: Config,
     rpc_url: &str,
+    rpc_timeout: Duration,
     key_store: KeyStore,
     local_store_tx: Sender<store::local::Message>,
     logger: Logger,
 ) -> Result<Vec<JoinHandle<()>>> {
     // Create and spawn the network state querier
     let (network_state_tx, network_state_rx) = watch::channel(Default::default());
     let mut network_state_querier = NetworkStateQuerier::new(
-        &rpc_url,
+        rpc_url,
+        rpc_timeout,
         time::interval(config.refresh_network_state_interval_duration),
         network_state_tx,
         logger.clone(),
@@ -154,6 +152,7 @@ pub fn spawn_exporter(
     let mut transaction_monitor = TransactionMonitor::new(
         config.transaction_monitor.clone(),
         rpc_url,
+        rpc_timeout,
         transactions_rx,
         logger.clone(),
     );
@@ -163,6 +162,7 @@ pub fn spawn_exporter(
     let mut exporter = Exporter::new(
         config,
         rpc_url,
+        rpc_timeout,
         key_store,
         local_store_tx,
         network_state_rx,
@@ -210,6 +210,7 @@ impl Exporter {
     pub fn new(
         config: Config,
         rpc_url: &str,
+        rpc_timeout: Duration,
         key_store: KeyStore,
         local_store_tx: Sender<store::local::Message>,
         network_state_rx: watch::Receiver<NetworkState>,
@@ -218,7 +219,7 @@ impl Exporter {
     ) -> Self {
         let publish_interval = time::interval(config.publish_interval_duration);
         Exporter {
-            rpc_client: RpcClient::new_with_timeout(rpc_url.to_string(), config.rpc_timeout),
+            rpc_client: RpcClient::new_with_timeout(rpc_url.to_string(), rpc_timeout),
             config,
             publish_interval,
             key_store,
@@ -449,12 +450,13 @@ struct NetworkStateQuerier {
 impl NetworkStateQuerier {
     pub fn new(
         rpc_endpoint: &str,
+        rpc_timeout: Duration,
         query_interval: Interval,
         network_state_tx: watch::Sender<NetworkState>,
         logger: Logger,
     ) -> Self {
         NetworkStateQuerier {
-            rpc_client: RpcClient::new(rpc_endpoint.to_string()),
+            rpc_client: RpcClient::new_with_timeout(rpc_endpoint.to_string(), rpc_timeout),
             query_interval,
             network_state_tx,
             logger,
@@ -562,11 +564,12 @@ mod transaction_monitor {
         pub fn new(
             config: Config,
             rpc_url: &str,
+            rpc_timeout: Duration,
             transactions_rx: mpsc::Receiver<Signature>,
             logger: Logger,
         ) -> Self {
             let poll_interval = time::interval(config.poll_interval_duration);
-            let rpc_client = RpcClient::new(rpc_url.to_string());
+            let rpc_client = RpcClient::new_with_timeout(rpc_url.to_string(), rpc_timeout);
             TransactionMonitor {
                 config,
                 rpc_client,

src/agent/solana/oracle.rs

@@ -123,6 +123,7 @@ pub fn spawn_oracle(
     config: Config,
     rpc_url: &str,
     wss_url: &str,
+    rpc_timeout: Duration,
     key_store: KeyStore,
     global_store_update_tx: mpsc::Sender<global::Update>,
     logger: Logger,
@@ -135,6 +136,7 @@ pub fn spawn_oracle(
         let subscriber = Subscriber::new(
             rpc_url.to_string(),
             wss_url.to_string(),
+            rpc_timeout,
             config.commitment,
             key_store.program_key.clone(),
             updates_tx,
@@ -149,6 +151,7 @@ pub fn spawn_oracle(
         data_tx,
         key_store.mapping_key,
         rpc_url,
+        rpc_timeout,
         config.commitment,
         config.poll_interval_duration,
         logger.clone(),
@@ -337,12 +340,16 @@ impl Poller {
         data_tx: mpsc::Sender<Data>,
         mapping_account_key: Pubkey,
         rpc_url: &str,
+        rpc_timeout: Duration,
         commitment: CommitmentLevel,
         poll_interval_duration: Duration,
         logger: Logger,
     ) -> Self {
-        let rpc_client =
-            RpcClient::new_with_commitment(rpc_url.to_string(), CommitmentConfig { commitment });
+        let rpc_client = RpcClient::new_with_timeout_and_commitment(
+            rpc_url.to_string(),
+            rpc_timeout,
+            CommitmentConfig { commitment },
+        );
         let poll_interval = tokio::time::interval(poll_interval_duration);
 
         Poller {
@@ -513,6 +520,7 @@ mod subscriber {
             BlockchainShadow,
             SyncOptions,
         },
+        std::time::Duration,
         tokio::sync::{
             broadcast,
             mpsc,
@@ -527,6 +535,9 @@ mod subscriber {
         /// WSS RPC endpoint
         wss_url: String,
 
+        /// Timeout for RPC requests
+        rpc_timeout: Duration,
+
         /// Commitment level used to read account data
         commitment: CommitmentLevel,
 
@@ -544,6 +555,7 @@ mod subscriber {
         pub fn new(
             rpc_url: String,
             wss_url: String,
+            rpc_timeout: Duration,
             commitment: CommitmentLevel,
             account_key: Pubkey,
             updates_tx: mpsc::Sender<(Pubkey, solana_sdk::account::Account)>,
@@ -552,6 +564,7 @@ mod subscriber {
             Subscriber {
                 rpc_url,
                 wss_url,
+                rpc_timeout,
                 commitment,
                 account_key,
                 updates_tx,
@@ -597,6 +610,7 @@ mod subscriber {
                         self.wss_url.clone(),
                     ),
                     commitment: self.commitment,
+                    rpc_timeout: self.rpc_timeout,
                     max_lag: Some(10000),
                     ..SyncOptions::default()
                 },