cuprated/blockchain/
syncer.rs

1use std::{
2    future::Future,
3    sync::{
4        atomic::{AtomicU64, Ordering},
5        Arc,
6    },
7};
8
9use futures::{FutureExt, StreamExt};
10use tokio::sync::{mpsc, Notify, OwnedSemaphorePermit, Semaphore};
11use tokio_util::sync::CancellationToken;
12use tower::{Service, ServiceExt};
13use tracing::instrument;
14
15use cuprate_consensus::{BlockChainContextRequest, BlockChainContextResponse, BlockchainContext};
16use cuprate_consensus_context::BlockchainContextService;
17use cuprate_helper::cast::usize_to_u64;
18use cuprate_p2p::{
19    block_downloader::{BlockBatch, BlockDownloaderConfig, ChainSvcRequest, ChainSvcResponse},
20    NetworkInterface, PeerSetRequest, PeerSetResponse,
21};
22use cuprate_p2p_core::{client::PeerSyncCallback, ClearNet, CoreSyncData, NetworkZone};
23
24use super::BlockchainManagerHandle;
25
26/// An error returned from the [`BlockchainSyncer`].
27#[derive(Debug, thiserror::Error)]
28pub enum SyncerError {
29    #[error("Incoming block channel closed.")]
30    IncomingBlockChannelClosed,
31    #[error("One of our services returned an error: {0}.")]
32    ServiceError(#[from] tower::BoxError),
33    #[error("Sync permit semaphore closed unexpectedly: {0}.")]
34    SemaphoreClosed(#[from] tokio::sync::AcquireError),
35}
36
37#[derive(Debug, PartialEq)]
38enum SyncStatus {
39    NoPeers,
40    BehindPeers,
41    Synced,
42}
43
44/// The syncer that makes sure we are fully synchronised with our connected peers.
45pub struct BlockchainSyncer {
46    notify_syncer: Arc<Notify>,
47    synced_tx: Option<futures::channel::oneshot::Sender<()>>,
48    target_height: Arc<AtomicU64>,
49}
50
51impl BlockchainSyncer {
52    /// Create a new [`BlockchainSyncer`] from its handle and the sender used to signal the node has synced.
53    pub fn new(
54        handle: &BlockchainSyncerHandle,
55        synced_tx: futures::channel::oneshot::Sender<()>,
56        offline: bool,
57    ) -> Self {
58        let synced_tx = if offline {
59            #[expect(clippy::let_underscore_must_use)]
60            let _ = synced_tx.send(());
61            None
62        } else {
63            Some(synced_tx)
64        };
65
66        Self {
67            notify_syncer: Arc::clone(&handle.notify_syncer),
68            synced_tx,
69            target_height: Arc::clone(&handle.target_height),
70        }
71    }
72
73    /// Run the syncer.
74    #[instrument(name = "syncer", level = "debug", skip_all)]
75    #[expect(clippy::significant_drop_tightening)]
76    #[expect(clippy::too_many_arguments)]
77    pub async fn run<CN>(
78        mut self,
79        mut context_svc: BlockchainContextService,
80        our_chain: CN,
81        mut clearnet_interface: NetworkInterface<ClearNet>,
82        incoming_block_batch_tx: mpsc::Sender<(BlockBatch, Arc<OwnedSemaphorePermit>)>,
83        stop_current_block_downloader: Arc<Notify>,
84        block_downloader_config: BlockDownloaderConfig,
85        shutdown_token: CancellationToken,
86    ) -> Result<(), SyncerError>
87    where
88        CN: Service<
89                ChainSvcRequest<ClearNet>,
90                Response = ChainSvcResponse<ClearNet>,
91                Error = tower::BoxError,
92            > + Clone
93            + Send
94            + 'static,
95        CN::Future: Send + 'static,
96    {
97        tracing::info!("Starting blockchain syncer");
98        tracing::debug!("Waiting for new sync info in top sync channel");
99
100        let semaphore = Arc::new(Semaphore::new(1));
101        let mut sync_permit = Arc::new(Arc::clone(&semaphore).acquire_owned().await?);
102
103        loop {
104            tokio::select! {
105                biased;
106                () = shutdown_token.cancelled() => {
107                    tracing::info!("Blockchain syncer shut down.");
108                    return Ok(());
109                }
110                () = self.notify_syncer.notified() => {}
111            }
112
113            tracing::trace!("Checking connected peers to see if we are behind",);
114
115            match self
116                .check_sync_status(&mut context_svc, &mut clearnet_interface)
117                .await?
118            {
119                SyncStatus::BehindPeers => {}
120                SyncStatus::NoPeers => continue,
121                SyncStatus::Synced => {
122                    if let Some(synced) = self.synced_tx.take() {
123                        tracing::info!("Synchronised with the network.");
124                        #[expect(clippy::let_underscore_must_use)]
125                        let _ = synced.send(());
126                    }
127                    continue;
128                }
129            }
130
131            tracing::debug!(
132                "We are behind peers claimed cumulative difficulty, starting block downloader"
133            );
134            let mut block_batch_stream =
135                clearnet_interface.block_downloader(our_chain.clone(), block_downloader_config);
136
137            loop {
138                tokio::select! {
139                    biased;
140                    () = shutdown_token.cancelled() => {
141                        tracing::info!("Blockchain syncer shut down.");
142                        return Ok(());
143                    }
144                    () = stop_current_block_downloader.notified() => {
145                        tracing::info!("Received stop signal, stopping block downloader");
146
147                        drop(sync_permit);
148                        sync_permit = Arc::new(Arc::clone(&semaphore).acquire_owned().await?);
149
150                        self.notify_syncer.notify_one();
151                        break;
152                    }
153                    batch = block_batch_stream.next() => {
154                        let Some(batch) = batch else {
155                            // Wait for all references to the permit have been dropped (which means all blocks in the queue
156                            // have been handled before checking if we are synced.
157                            drop(sync_permit);
158                            sync_permit = Arc::new(Arc::clone(&semaphore).acquire_owned().await?);
159
160                            if self.check_sync_status(&mut context_svc, &mut clearnet_interface).await? == SyncStatus::Synced {
161                                tracing::info!("Synchronised with the network.");
162                                if let Some(synced) = self.synced_tx.take() {
163                                    #[expect(clippy::let_underscore_must_use)]
164                                    let _ = synced.send(());
165                                }
166                            }
167
168                            break;
169                        };
170
171                        tracing::debug!("Got batch, len: {}", batch.blocks.len());
172                        if incoming_block_batch_tx.send((batch, Arc::clone(&sync_permit))).await.is_err() {
173                            if shutdown_token.is_cancelled() {
174                                return Ok(());
175                            }
176                            return Err(SyncerError::IncomingBlockChannelClosed);
177                        }
178                    }
179                }
180            }
181        }
182    }
183
184    /// Checks if we are behind the connected peers.
185    async fn check_sync_status(
186        &mut self,
187        context_svc: &mut BlockchainContextService,
188        clearnet_interface: &mut NetworkInterface<ClearNet>,
189    ) -> Result<SyncStatus, tower::BoxError> {
190        let PeerSetResponse::MostPoWSeen {
191            cumulative_difficulty,
192            height,
193            ..
194        } = clearnet_interface
195            .peer_set()
196            .ready()
197            .await?
198            .call(PeerSetRequest::MostPoWSeen)
199            .await?
200        else {
201            unreachable!();
202        };
203
204        if cumulative_difficulty == 0 {
205            self.target_height.store(0, Ordering::Relaxed);
206            return Ok(SyncStatus::NoPeers);
207        }
208
209        if cumulative_difficulty > context_svc.blockchain_context().cumulative_difficulty {
210            self.target_height
211                .store(usize_to_u64(height), Ordering::Relaxed);
212            return Ok(SyncStatus::BehindPeers);
213        }
214
215        self.target_height.store(0, Ordering::Relaxed);
216        Ok(SyncStatus::Synced)
217    }
218}
219
220/// Handle for the `BlockchainSyncer`.
221#[derive(Clone)]
222pub struct BlockchainSyncerHandle {
223    /// The syncer notify channel, used to wake the syncer.
224    notify_syncer: Arc<Notify>,
225    /// The synced notify channel, used to wake the tasks waiting on cuprate to be synced.
226    synced: futures::future::Shared<futures::channel::oneshot::Receiver<()>>,
227    /// The target height we are syncing to, 0 if not syncing.
228    target_height: Arc<AtomicU64>,
229}
230
231impl BlockchainSyncerHandle {
232    /// Create a new handle and the sender used to signal the node has synced.
233    pub(crate) fn new() -> (Self, futures::channel::oneshot::Sender<()>) {
234        let (synced_tx, synced_rx) = futures::channel::oneshot::channel();
235
236        (
237            Self {
238                notify_syncer: Arc::new(Notify::new()),
239                synced: synced_rx.shared(),
240                target_height: Arc::new(AtomicU64::new(0)),
241            },
242            synced_tx,
243        )
244    }
245
246    /// Returns the target sync height. 0 if not syncing.
247    pub fn target_height(&self) -> u64 {
248        self.target_height.load(Ordering::Relaxed)
249    }
250
251    /// A future that resolves when cuprate has synced with the network.
252    pub fn wait_for_synced(
253        &self,
254    ) -> impl Future<Output = Result<(), futures::channel::oneshot::Canceled>> + 'static {
255        self.synced.clone()
256    }
257
258    /// Creates a [`PeerSyncCallback`] that filters and wakes the syncer.
259    pub(crate) fn callback(
260        &self,
261        context_svc: BlockchainContextService,
262        blockchain_manager: BlockchainManagerHandle,
263    ) -> PeerSyncCallback {
264        let sync_handle = self.clone();
265        let disconnect_handle = self.clone();
266
267        let on_sync = move |peer_csd: &CoreSyncData| {
268            let ctx = context_svc.blockchain_context_snapshot();
269
270            // If we are synced and the syncer hasn't yet set the node to synced, wake the syncer.
271            if peer_csd.cumulative_difficulty() == ctx.cumulative_difficulty
272                && sync_handle.synced.peek().is_none()
273            {
274                sync_handle.notify_syncer.notify_one();
275            }
276
277            // If we are behind the peer, and we aren't just one block behind with the blockchain manager handling the block, wake the syncer.
278            if peer_csd.cumulative_difficulty() > ctx.cumulative_difficulty
279                && !(peer_csd.current_height.saturating_sub(1) == ctx.chain_height as u64
280                    && blockchain_manager.is_block_being_handled(&peer_csd.top_id))
281            {
282                sync_handle.notify_syncer.notify_one();
283            }
284        };
285
286        let on_disconnect = move || disconnect_handle.notify_syncer.notify_one();
287
288        PeerSyncCallback::new(on_sync, on_disconnect)
289    }
290}
cuprated/blockchain/syncer.rs

cuprated/blockchain/
syncer.rs
