Using config

Author: rodoufu

src/agent/services/oracle.rs

@@ -67,11 +67,11 @@ where
     )));
 
     if config.oracle.subscriber_enabled {
-        let number_of_workers = 100;
-        let channel_size = 1000;
-        let (sender, receiver) = tokio::sync::mpsc::channel(channel_size);
-        let max_elapsed_time = Duration::from_secs(30);
-        let sleep_time = Duration::from_secs(1);
+        let number_of_workers = config.oracle.handle_price_account_update_worker_poll_size;
+        let (sender, receiver) =
+            tokio::sync::mpsc::channel(config.oracle.handle_price_account_update_channel_size);
+        let min_elapsed_time = config.oracle.subscriber_finished_min_time;
+        let sleep_time = config.oracle.subscriber_finished_sleep_time;
 
         handles.push(tokio::spawn(async move {
             loop {
@@ -86,7 +86,7 @@ where
                 .await
                 {
                     tracing::error!(?err, "Subscriber exited unexpectedly");
-                    if current_time.elapsed() < max_elapsed_time {
+                    if current_time.elapsed() < min_elapsed_time {
                         tracing::warn!(?sleep_time, "Subscriber restarting too quickly. Sleeping");
                         tokio::time::sleep(sleep_time).await;
                     }

src/agent/state/oracle.rs

@@ -137,6 +137,22 @@ pub struct Data {
     pub publisher_buffer_key:  Option<Pubkey>,
 }
 
+fn default_handle_price_account_update_channel_size() -> usize {
+    1000
+}
+
+fn default_handle_price_account_update_worker_poll_size() -> usize {
+    50
+}
+
+fn default_subscriber_finished_min_time() -> Duration {
+    Duration::from_secs(30)
+}
+
+fn default_subscriber_finished_sleep_time() -> Duration {
+    Duration::from_secs(1)
+}
+
 #[derive(Clone, Serialize, Deserialize, Debug)]
 #[serde(default)]
 pub struct Config {
@@ -158,6 +174,19 @@ pub struct Config {
     /// socket count at bay, the batches are looked up sequentially,
     /// trading off overall time it takes to fetch all symbols.
     pub max_lookup_batch_size: usize,
+
+    /// Number of workers used to wait for the handle_price_account_update
+    #[serde(default = "default_handle_price_account_update_worker_poll_size")]
+    pub handle_price_account_update_worker_poll_size: usize,
+    /// Channel size used to wait for the handle_price_account_update
+    #[serde(default = "default_handle_price_account_update_channel_size")]
+    pub handle_price_account_update_channel_size: usize,
+    /// Minimum time for a subscriber to run
+    #[serde(default = "default_subscriber_finished_min_time")]
+    pub subscriber_finished_min_time: Duration,
+    /// Time to sleep if the subscriber do not run for more than the minimum time
+    #[serde(default = "default_subscriber_finished_sleep_time")]
+    pub subscriber_finished_sleep_time: Duration,
 }
 
 impl Default for Config {
@@ -169,6 +198,12 @@ impl Default for Config {
             updates_channel_capacity: 10000,
             data_channel_capacity:    10000,
             max_lookup_batch_size:    100,
+            handle_price_account_update_worker_poll_size:
+                default_handle_price_account_update_worker_poll_size(),
+            handle_price_account_update_channel_size:
+                default_handle_price_account_update_channel_size(),
+            subscriber_finished_min_time: default_subscriber_finished_min_time(),
+            subscriber_finished_sleep_time: default_subscriber_finished_sleep_time(),
         }
     }
 }