reth_rpc/eth/helpers/
sync_listener.rs

1//! A utility Future to asynchronously wait until a node has finished syncing.
2
3use futures::Stream;
4use pin_project::pin_project;
5use reth_network_api::NetworkInfo;
6use std::{
7    future::Future,
8    pin::Pin,
9    task::{ready, Context, Poll},
10};
11
12/// This future resolves once the node is no longer syncing: [`NetworkInfo::is_syncing`].
13#[must_use = "futures do nothing unless polled"]
14#[pin_project]
15#[derive(Debug)]
16pub struct SyncListener<N, St> {
17    #[pin]
18    tick: St,
19    network_info: N,
20}
21
22impl<N, St> SyncListener<N, St> {
23    /// Create a new [`SyncListener`] using the given tick stream.
24    pub const fn new(network_info: N, tick: St) -> Self {
25        Self { tick, network_info }
26    }
27}
28
29impl<N, St> Future for SyncListener<N, St>
30where
31    N: NetworkInfo,
32    St: Stream + Unpin,
33{
34    type Output = ();
35
36    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
37        let mut this = self.project();
38
39        if !this.network_info.is_syncing() {
40            return Poll::Ready(());
41        }
42
43        loop {
44            let tick_event = ready!(this.tick.as_mut().poll_next(cx));
45
46            match tick_event {
47                Some(_) => {
48                    if !this.network_info.is_syncing() {
49                        return Poll::Ready(());
50                    }
51                }
52                None => return Poll::Ready(()),
53            }
54        }
55    }
56}
57
58#[cfg(test)]
59mod tests {
60    use super::*;
61    use alloy_rpc_types_admin::EthProtocolInfo;
62    use futures::stream;
63    use reth_network_api::{CellCustody, NetworkError, NetworkStatus};
64    use std::{
65        net::{IpAddr, SocketAddr},
66        sync::{
67            atomic::{AtomicBool, Ordering},
68            Arc,
69        },
70    };
71
72    #[derive(Clone)]
73    struct TestNetwork {
74        syncing: Arc<AtomicBool>,
75    }
76
77    impl NetworkInfo for TestNetwork {
78        fn local_addr(&self) -> SocketAddr {
79            (IpAddr::from([0, 0, 0, 0]), 0).into()
80        }
81
82        async fn network_status(&self) -> Result<NetworkStatus, NetworkError> {
83            #[allow(deprecated)]
84            Ok(NetworkStatus {
85                client_version: "test".to_string(),
86                protocol_version: 5,
87                eth_protocol_info: EthProtocolInfo {
88                    network: 1,
89                    difficulty: None,
90                    genesis: Default::default(),
91                    config: Default::default(),
92                    head: Default::default(),
93                },
94                capabilities: vec![],
95            })
96        }
97
98        fn chain_id(&self) -> u64 {
99            1
100        }
101
102        fn cell_custody(&self) -> &CellCustody {
103            static CELL_CUSTODY: std::sync::OnceLock<CellCustody> = std::sync::OnceLock::new();
104            CELL_CUSTODY.get_or_init(CellCustody::default)
105        }
106
107        fn is_syncing(&self) -> bool {
108            self.syncing.load(Ordering::SeqCst)
109        }
110
111        fn is_initially_syncing(&self) -> bool {
112            self.is_syncing()
113        }
114    }
115
116    #[tokio::test]
117    async fn completes_immediately_if_not_syncing() {
118        let network = TestNetwork { syncing: Arc::new(AtomicBool::new(false)) };
119        let fut = SyncListener::new(network, stream::pending::<()>());
120        fut.await;
121    }
122
123    #[tokio::test]
124    async fn resolves_when_syncing_stops() {
125        use tokio::sync::mpsc::unbounded_channel;
126        use tokio_stream::wrappers::UnboundedReceiverStream;
127
128        let syncing = Arc::new(AtomicBool::new(true));
129        let network = TestNetwork { syncing: syncing.clone() };
130        let (tx, rx) = unbounded_channel();
131        let listener = SyncListener::new(network, UnboundedReceiverStream::new(rx));
132        let handle = tokio::spawn(listener);
133
134        syncing.store(false, Ordering::Relaxed);
135        let _ = tx.send(());
136
137        handle.await.unwrap();
138    }
139}
reth_rpc/eth/helpers/sync_listener.rs

reth_rpc/eth/helpers/
sync_listener.rs
