Merge pull request #4346 from eval-exec/exec/rc/v0.114.x-4345

Backport to `v0.114.x` , `IndexerService::apply_init_tip` should stop after received exit signal.

util/indexer/Cargo.toml

@@ -22,6 +22,7 @@ ckb-notify = { path = "../../notify", version = "= 0.114.0-rc1" }
 ckb-store = { path = "../../store", version = "= 0.114.0-rc1" }
 ckb-stop-handler = { path = "../stop-handler", version = "= 0.114.0-rc1" }
 ckb-async-runtime = { path = "../runtime", version = "= 0.114.0-rc1" }
+ckb-channel = { path = "../channel", version = "= 0.114.0-rc1" }
 rhai = { version = "1.10.0", features = ["no_function", "no_float", "no_module", "sync"]}
 serde_json = "1.0"
 numext-fixed-uint = "0.1"

util/indexer/src/service.rs

@@ -18,7 +18,7 @@ use ckb_jsonrpc_types::{
 };
 use ckb_logger::{error, info};
 use ckb_notify::NotifyController;
-use ckb_stop_handler::{new_tokio_exit_rx, CancellationToken};
+use ckb_stop_handler::{new_crossbeam_exit_rx, new_tokio_exit_rx, CancellationToken};
 use ckb_store::ChainStore;
 use ckb_types::{
     core::{self, BlockNumber},
@@ -153,7 +153,16 @@ impl IndexerService {
                     }
                 }
             }
+            let stop_rx = new_crossbeam_exit_rx();
             loop {
+                ckb_channel::select! {
+                    recv(stop_rx) -> _ =>{
+                        info!("apply_init_tip received exit signal, exit now");
+                        break;
+                    },
+                    default() => {},
+                }
+
                 if let Err(e) = self.secondary_db.try_catch_up_with_primary() {
                     error!("secondary_db try_catch_up_with_primary error {}", e);
                 }
@@ -228,6 +237,12 @@ impl IndexerService {
         let poll_service = self.clone();
         self.async_handle.spawn(async move {
             let _initial_finished = initial_syncing.await;
+            info!("initial_syncing finished");
+            if stop.is_cancelled() {
+                info!("Indexer received exit signal, cancel new_block_watcher task, exit now");
+                return;
+            }
+
             let mut new_block_watcher = notify_controller
                 .watch_new_block(SUBSCRIBER_NAME.to_string())
                 .await;