fix: rpc downloader shutdown (#1762)

+ chore: put hard-coded value in importer constant

Author: marcospb19-cw

src/bin/rpc_downloader.rs

@@ -3,8 +3,8 @@ use std::sync::Arc;
 
 use anyhow::anyhow;
 use anyhow::Context;
+use futures::stream;
 use futures::StreamExt;
-use futures::TryStreamExt;
 use itertools::Itertools;
 use serde::Deserialize;
 use stratus::config::RpcDownloaderConfig;
@@ -14,9 +14,9 @@ use stratus::eth::primitives::Hash;
 use stratus::eth::storage::ExternalRpcStorage;
 use stratus::ext::not;
 use stratus::infra::BlockchainClient;
-use stratus::log_and_err;
 use stratus::utils::DropTimer;
 use stratus::GlobalServices;
+use stratus::GlobalState;
 #[cfg(all(not(target_env = "msvc"), any(feature = "jemalloc", feature = "jeprof")))]
 use tikv_jemallocator::Jemalloc;
 
@@ -83,7 +83,8 @@ async fn download_balances(rpc_storage: Arc<dyn ExternalRpcStorage>, chain: &Blo
 }
 
 async fn download_blocks(rpc_storage: Arc<dyn ExternalRpcStorage>, chain: Arc<BlockchainClient>, paralellism: usize, end: BlockNumber) -> anyhow::Result<()> {
-    let _timer = DropTimer::start("rpc-downloader::download_blocks");
+    const TASK_NAME: &str = "rpc-downloader::download_blocks";
+    let _timer = DropTimer::start(TASK_NAME);
 
     // prepare download block tasks
     let mut start = BlockNumber::ZERO;
@@ -99,16 +100,20 @@ async fn download_blocks(rpc_storage: Arc<dyn ExternalRpcStorage>, chain: Arc<Bl
 
     // execute download block tasks
     tracing::info!(tasks = %tasks.len(), %paralellism, "executing block downloads");
-    let result = futures::stream::iter(tasks).buffer_unordered(paralellism).try_collect::<Vec<()>>().await;
-    match result {
-        Ok(_) => {
-            tracing::info!("tasks finished");
-            Ok(())
+
+    let mut stream = stream::iter(tasks).buffered(paralellism);
+    while let Some(result) = stream.next().await {
+        if let Err(e) = result {
+            tracing::error!(reason = ?e, "download task failed");
         }
-        Err(e) => {
-            log_and_err!(reason = e, "tasks failed")
+
+        if GlobalState::is_shutdown_warn(TASK_NAME) {
+            break;
         }
     }
+
+    tracing::info!("download finished");
+    Ok(())
 }
 
 async fn download(

src/eth/follower/importer/importer.rs

@@ -150,6 +150,8 @@ impl Importer {
     // Executor
     // -----------------------------------------------------------------------------
 
+    pub const TASKS_COUNT: usize = 3;
+
     // Executes external blocks and persist them to storage.
     async fn start_block_executor(
         executor: Arc<Executor>,

src/globals.rs

@@ -15,6 +15,7 @@ use crate::alias::JsonValue;
 use crate::config;
 use crate::config::StratusConfig;
 use crate::config::WithCommonConfig;
+use crate::eth::follower::importer::Importer;
 use crate::eth::rpc::RpcContext;
 use crate::ext::spawn_signal_handler;
 use crate::infra::tracing::warn_task_cancellation;
@@ -104,7 +105,7 @@ pub static STRATUS_SHUTDOWN_SIGNAL: Lazy<CancellationToken> = Lazy::new(Cancella
 static IMPORTER_SHUTDOWN: AtomicBool = AtomicBool::new(true);
 
 /// A guard that is taken when importer is running.
-pub static IMPORTER_ONLINE_TASKS_SEMAPHORE: Lazy<Semaphore> = Lazy::new(|| Semaphore::new(3));
+pub static IMPORTER_ONLINE_TASKS_SEMAPHORE: Lazy<Semaphore> = Lazy::new(|| Semaphore::new(Importer::TASKS_COUNT));
 
 /// Transaction should be accepted?
 static TRANSACTIONS_ENABLED: AtomicBool = AtomicBool::new(true);
@@ -178,7 +179,7 @@ impl GlobalState {
     /// Waits till importer is done.
     pub async fn wait_for_importer_to_finish() {
         // 3 permits will be available when all 3 tasks are finished
-        let result = IMPORTER_ONLINE_TASKS_SEMAPHORE.acquire_many(3).await;
+        let result = IMPORTER_ONLINE_TASKS_SEMAPHORE.acquire_many(Importer::TASKS_COUNT as u32).await;
 
         if let Err(e) = result {
             tracing::error!(reason = ?e, "error waiting for importer to finish");