Improve logging and reliability (#45)

* Add a key for primary vs secondary in logs

* Add timeout for exporter

* Bump version

* Move timeout to network level

* Bump warp dependency

Suggested by dependabot

Author: ali-behjati

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "pyth-agent"
-version = "0.1.2"
+version = "0.1.3"
 edition = "2021"
 
 [[bin]]
@@ -11,7 +11,7 @@ path = "src/bin/agent.rs"
 anyhow = "1.0.55"
 serde = { version = "1.0.136", features = ["derive"] }
 async-trait = "0.1.52"
-warp = { version = "0.3.1", features = ["websocket"] }
+warp = { version = "0.3.3", features = ["websocket"] }
 tokio = { version = "1.0", features = ["full"] }
 tokio-stream = "0.1.1"
 futures-util = { version = "0.3", default-features = false, features = [

Cargo.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"
 

config/config.toml

@@ -103,7 +103,7 @@ impl Agent {
             self.config.primary_network.clone(),
             local_store_tx.clone(),
             primary_oracle_updates_tx,
-            logger.clone(),
+            logger.new(o!("primary" => true)),
         )?);
 
         // Spawn the secondary network, if needed
@@ -112,7 +112,7 @@ impl Agent {
                 config.clone(),
                 local_store_tx.clone(),
                 secondary_oracle_updates_tx,
-                logger.clone(),
+                logger.new(o!("primary" => false)),
             )?);
         }
 

src/agent.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.rs

@@ -130,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(),
@@ -150,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(),
     );
@@ -159,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,
@@ -206,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>,
@@ -214,7 +219,7 @@ impl Exporter {
     ) -> Self {
         let publish_interval = time::interval(config.publish_interval_duration);
         Exporter {
-            rpc_client: RpcClient::new(rpc_url.to_string()),
+            rpc_client: RpcClient::new_with_timeout(rpc_url.to_string(), rpc_timeout),
             config,
             publish_interval,
             key_store,
@@ -445,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,
@@ -558,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,