kabu_node_json_rpc/
wait_for_node_sync_actor.rs

1use alloy_network::Ethereum;
2use alloy_provider::Provider;
3use alloy_rpc_types::SyncStatus;
4use kabu_core_components::Component;
5
6use eyre::Result;
7use kabu_node_debug_provider::DebugProviderExt;
8use reth_tasks::TaskExecutor;
9use std::time::Duration;
10use tokio::time::timeout;
11use tracing::{error, info};
12
13const SYNC_CHECK_INTERVAL: Duration = Duration::from_secs(1);
14const CLIENT_TIMEOUT: Duration = Duration::from_secs(10);
15
16/// Wait for the node to sync. This works only for http/ipc/ws providers.
17async fn wait_for_node_sync_one_shot_worker<P>(client: P) -> Result<()>
18where
19    P: Provider<Ethereum> + DebugProviderExt<Ethereum> + Send + Sync + Clone + 'static,
20{
21    info!("Waiting for node to sync...");
22    let mut print_count = 0;
23    loop {
24        match timeout(CLIENT_TIMEOUT, client.syncing()).await {
25            Ok(result) => match result {
26                Ok(syncing_status) => match syncing_status {
27                    SyncStatus::None => {
28                        break;
29                    }
30                    SyncStatus::Info(sync_progress) => {
31                        if print_count == 0 {
32                            info!("Sync progress: {:?}", sync_progress);
33                        }
34                    }
35                },
36                Err(e) => {
37                    error!("Error retrieving syncing status: {:?}", e);
38                    break;
39                }
40            },
41            Err(elapsed) => {
42                error!("Timeout during get syncing status. Elapsed time: {:?}", elapsed);
43                break;
44            }
45        }
46        tokio::time::sleep(SYNC_CHECK_INTERVAL).await;
47        print_count = if print_count > 4 { 0 } else { print_count + 1 };
48    }
49    Ok(())
50}
51
52pub struct WaitForNodeSyncOneShotBlockingActor<P> {
53    client: P,
54}
55
56impl<P> WaitForNodeSyncOneShotBlockingActor<P>
57where
58    P: Provider<Ethereum> + DebugProviderExt<Ethereum> + Send + Sync + Clone + 'static,
59{
60    pub fn new(client: P) -> WaitForNodeSyncOneShotBlockingActor<P> {
61        WaitForNodeSyncOneShotBlockingActor { client }
62    }
63}
64
65impl<P> Component for WaitForNodeSyncOneShotBlockingActor<P>
66where
67    P: Provider<Ethereum> + DebugProviderExt<Ethereum> + Send + Sync + Clone + 'static,
68{
69    fn spawn(self, executor: TaskExecutor) -> Result<()> {
70        let client = self.client;
71        executor.spawn(async move {
72            if let Err(e) = wait_for_node_sync_one_shot_worker(client).await {
73                error!("Wait for node sync failed: {}", e);
74            }
75        });
76        Ok(())
77    }
78    fn name(&self) -> &'static str {
79        "WaitForNodeSyncOneShotBlockingActor"
80    }
81}
kabu_node_json_rpc/wait_for_node_sync_actor.rs

kabu_node_json_rpc/
wait_for_node_sync_actor.rs
