feat: isolate and track forester worker concurrency

Author: sergeytimoshin

forester/src/epoch_manager.rs

@@ -440,21 +440,19 @@ pub async fn metrics_handler() -> Result<impl warp::Reply> {
     if let Err(e) = encoder.encode(&REGISTRY.gather(), &mut buffer) {
         error!("could not encode custom metrics: {}", e);
     };
-    let mut res = String::from_utf8(buffer.clone()).unwrap_or_else(|e| {
+    let mut res = String::from_utf8(buffer).unwrap_or_else(|e| {
         error!("custom metrics could not be from_utf8'd: {}", e);
         String::new()
     });
-    buffer.clear();
 
     let mut buffer = Vec::new();
     if let Err(e) = encoder.encode(&prometheus::gather(), &mut buffer) {
         error!("could not encode prometheus metrics: {}", e);
     };
-    let res_prometheus = String::from_utf8(buffer.clone()).unwrap_or_else(|e| {
+    let res_prometheus = String::from_utf8(buffer).unwrap_or_else(|e| {
         error!("prometheus metrics could not be from_utf8'd: {}", e);
         String::new()
     });
-    buffer.clear();
 
     res.push_str(&res_prometheus);
     Ok(res)

forester/src/metrics.rs

@@ -208,7 +208,7 @@ pub async fn request_priority_fee_estimate(
         .map_err(|error| PriorityFeeEstimateError::ClientBuild(error.clone()))?;
 
     let response = http_client
-        .post(url.clone())
+        .post(url.as_str())
         .header("Content-Type", "application/json")
         .json(&rpc_request)
         .send()

forester/src/priority_fee.rs

@@ -277,7 +277,7 @@ async fn e2e_test() {
             validator_args: vec![],
         }))
         .await;
-        spawn_prover().await;
+        spawn_prover().await.unwrap();
     }
 
     let mut rpc = setup_rpc_connection(&env.protocol.forester).await;
@@ -799,15 +799,22 @@ async fn setup_forester_pipeline(
     let (shutdown_bootstrap_sender, shutdown_bootstrap_receiver) = oneshot::channel();
     let (work_report_sender, work_report_receiver) = mpsc::channel(100);
 
-    let service_handle = tokio::spawn(run_pipeline::<LightClient>(
-        Arc::from(config.clone()),
-        None,
-        None,
-        shutdown_receiver,
-        Some(shutdown_compressible_receiver),
-        Some(shutdown_bootstrap_receiver),
-        work_report_sender,
-    ));
+    let config = Arc::new(config.clone());
+    let service_handle = tokio::task::spawn_blocking(move || {
+        let runtime = tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(2)
+            .enable_all()
+            .build()?;
+        runtime.block_on(run_pipeline::<LightClient>(
+            config,
+            None,
+            None,
+            shutdown_receiver,
+            Some(shutdown_compressible_receiver),
+            Some(shutdown_bootstrap_receiver),
+            work_report_sender,
+        ))
+    });
 
     (
         service_handle,

forester/tests/e2e_test.rs

@@ -87,7 +87,7 @@ async fn test_state_indexer_async_batched() {
         validator_args: vec![],
     }))
     .await;
-    spawn_prover().await;
+    spawn_prover().await.unwrap();
 
     let env = TestAccounts::get_local_test_validator_accounts();
     let mut config = forester_config();

forester/tests/legacy/batched_state_async_indexer_test.rs

@@ -26,7 +26,7 @@ pub async fn init(config: Option<LightValidatorConfig>) {
 #[allow(dead_code)]
 pub async fn spawn_test_validator(config: Option<LightValidatorConfig>) {
     let config = config.unwrap_or_default();
-    spawn_validator(config).await;
+    spawn_validator(config).await.unwrap();
 }
 
 #[allow(dead_code)]

forester/tests/legacy/test_utils.rs

@@ -328,15 +328,22 @@ async fn run_forester(config: &ForesterConfig, duration: Duration) {
         tokio::sync::broadcast::channel(1);
     let (work_report_sender, _) = mpsc::channel(100);
 
-    let service_handle = tokio::spawn(run_pipeline::<LightClient>(
-        Arc::from(config.clone()),
-        None,
-        None,
-        shutdown_receiver,
-        Some(shutdown_compressible_receiver),
-        None, // shutdown_bootstrap
-        work_report_sender,
-    ));
+    let config = Arc::new(config.clone());
+    let service_handle = tokio::task::spawn_blocking(move || {
+        let runtime = tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(2)
+            .enable_all()
+            .build()?;
+        runtime.block_on(run_pipeline::<LightClient>(
+            config,
+            None,
+            None,
+            shutdown_receiver,
+            Some(shutdown_compressible_receiver),
+            None, // shutdown_bootstrap
+            work_report_sender,
+        ))
+    });
 
     tokio::time::sleep(duration).await;
 

forester/tests/test_batch_append_spent.rs

@@ -65,7 +65,8 @@ async fn test_indexer_interface_scenarios() {
         validator_args: vec![],
         use_surfpool: true,
     })
-    .await;
+    .await
+    .unwrap();
 
     let mut rpc = LightClient::new(LightClientConfig::local())
         .await

forester/tests/test_indexer_interface.rs

@@ -36,7 +36,7 @@ pub async fn init(config: Option<LightValidatorConfig>) {
 #[allow(dead_code)]
 pub async fn spawn_test_validator(config: Option<LightValidatorConfig>) {
     let config = config.unwrap_or_default();
-    spawn_validator(config).await;
+    spawn_validator(config).await.unwrap();
 }
 
 #[allow(dead_code)]