15280118615

// Copyright 2019. The Tari Project
//
// Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
// following conditions are met:
//
// 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
// disclaimer.
//
// 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
// following disclaimer in the documentation and/or other materials provided with the distribution.
//
// 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote
// products derived from this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
// INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
// WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
// USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

use std::{cmp::Ordering, time::Instant};

use log::*;
use tari_common_types::chain_metadata::ChainMetadata;
use tari_comms::peer_manager::NodeId;

#[cfg(feature = "metrics")]
use crate::base_node::metrics;
use crate::{
    base_node::{
        comms_interface::BlockEvent,
        state_machine_service::states::{BlockSyncInfo, StateEvent, StateInfo, StatusInfo},
        sync::{BlockHeaderSyncError, HeaderSynchronizer, SyncPeer},
        BaseNodeStateMachine,
    },
    chain_storage::BlockchainBackend,
};
const LOG_TARGET: &str = "c::bn::header_sync";

#[derive(Clone, Debug)]
pub struct HeaderSyncState {
    sync_peers: Vec<SyncPeer>,
    is_synced: bool,
    local_metadata: ChainMetadata,
}

impl HeaderSyncState {
    pub fn new(mut sync_peers: Vec<SyncPeer>, local_metadata: ChainMetadata) -> Self {
        // Sort by latency lowest to highest
        sync_peers.sort_by(|a, b| match a.claimed_difficulty().cmp(&b.claimed_difficulty()) {
            Ordering::Less => Ordering::Less,
            // No latency goes to the end
            Ordering::Greater => Ordering::Greater,
            Ordering::Equal => {
                match (a.latency(), b.latency()) {
                    (None, None) => Ordering::Equal,
                    // No latency goes to the end
                    (Some(_), None) => Ordering::Less,
                    (None, Some(_)) => Ordering::Greater,
                    (Some(la), Some(lb)) => la.cmp(&lb),
                }
            },
        });
        Self {
            sync_peers,
            is_synced: false,
            local_metadata,
        }
    }

    pub fn is_synced(&self) -> bool {
        self.is_synced
    }

    pub fn into_sync_peers(self) -> Vec<SyncPeer> {
        self.sync_peers
    }

    fn remove_sync_peer(&mut self, node_id: &NodeId) {
        if let Some(pos) = self.sync_peers.iter().position(|p| p.node_id() == node_id) {
            self.sync_peers.remove(pos);
        }
    }

    // converting u64 to i64 is okay as the future time limit is the hundreds so way below u32 even
    #[allow(clippy::too_many_lines)]
    #[allow(clippy::cast_possible_wrap)]
    pub async fn next_event<B: BlockchainBackend + 'static>(
        &mut self,
        shared: &mut BaseNodeStateMachine<B>,
    ) -> StateEvent {
        // Only sync to peers with better claimed accumulated difficulty than the local chain: this may be possible
        // at this stage due to read-write lock race conditions in the database
        match shared.db.get_chain_metadata().await {
            Ok(best_block_metadata) => {
                let mut remove = Vec::new();
                for sync_peer in &self.sync_peers {
                    if sync_peer.claimed_chain_metadata().accumulated_difficulty() <=
                        best_block_metadata.accumulated_difficulty()
                    {
                        remove.push(sync_peer.node_id().clone());
                    }
                }
                for node_id in remove {
                    self.remove_sync_peer(&node_id);
                }
                if self.sync_peers.is_empty() {
                    // Go back to Listening state
                    return StateEvent::Continue;
                }
            },
            Err(e) => return StateEvent::FatalError(format!("{}", e)),
        }

        let mut synchronizer = HeaderSynchronizer::new(
            shared.config.blockchain_sync_config.clone(),
            shared.db.clone(),
            shared.consensus_rules.clone(),
            shared.connectivity.clone(),
            &mut self.sync_peers,
            shared.randomx_factory.clone(),
            &self.local_metadata,
        );

        let status_event_sender = shared.status_event_sender.clone();
        let bootstrapped = shared.is_bootstrapped();
        let randomx_vm_cnt = shared.get_randomx_vm_cnt();
        let randomx_vm_flags = shared.get_randomx_vm_flags();
        synchronizer.on_starting(move |sync_peer| {
            let _result = status_event_sender.send(StatusInfo {
                bootstrapped,
                state_info: StateInfo::Connecting(sync_peer.clone()),
                randomx_vm_cnt,
                randomx_vm_flags,
            });
        });

        let status_event_sender = shared.status_event_sender.clone();
        synchronizer.on_progress(move |current_height, remote_tip_height, sync_peer| {
            let details = BlockSyncInfo {
                tip_height: remote_tip_height,
                local_height: current_height,
                sync_peer: sync_peer.clone(),
            };
            let _result = status_event_sender.send(StatusInfo {
                bootstrapped,
                state_info: StateInfo::HeaderSync(Some(details)),
                randomx_vm_cnt,
                randomx_vm_flags,
            });
        });

        let local_nci = shared.local_node_interface.clone();
        synchronizer.on_rewind(move |removed| {
            #[cfg(feature = "metrics")]
            if let Some(fork_height) = removed.last().map(|b| b.height().saturating_sub(1)) {
                metrics::tip_height().set(fork_height as i64);
                metrics::reorg(fork_height, 0, removed.len()).inc();
            }

            local_nci.publish_block_event(BlockEvent::BlockSyncRewind(removed));
        });

        let timer = Instant::now();
        match synchronizer.synchronize().await {
            Ok((sync_peer, sync_result)) => {
                info!(
                    target: LOG_TARGET,
                    "Headers synchronized from peer {} in {:.0?}",
                    sync_peer,
                    timer.elapsed()
                );
                // Move the sync peer used in header sync to the front of the queue
                if let Some(pos) = self.sync_peers.iter().position(|p| *p == sync_peer) {
                    if pos > 0 {
                        let sync_peer = self.sync_peers.remove(pos);
                        self.sync_peers.insert(0, sync_peer);
                    }
                }
                self.is_synced = true;
                StateEvent::HeadersSynchronized(sync_peer, sync_result)
            },
            Err(err) => {
                let _ignore = shared.status_event_sender.send(StatusInfo {
                    bootstrapped,
                    state_info: StateInfo::SyncFailed("HeaderSyncFailed".to_string()),
                    randomx_vm_cnt,
                    randomx_vm_flags,
                });
                match err {
                    BlockHeaderSyncError::SyncFailedAllPeers => {
                        error!(target: LOG_TARGET, "Header sync failed with all peers. Error: {}", err);
                        warn!(target: LOG_TARGET, "{}. Continuing...", err);
                        StateEvent::Continue
                    },
                    _ => {
                        debug!(target: LOG_TARGET, "Header sync failed: {}", err);
                        StateEvent::HeaderSyncFailed(err.to_string())
                    },
                }
            },
        }
    }
}

1	// Copyright 2019. The Tari Project
2	//
3	// Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
4	// following conditions are met:
5	//
6	// 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
7	// disclaimer.
8	//
9	// 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
10	// following disclaimer in the documentation and/or other materials provided with the distribution.
11	//
12	// 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote
13	// products derived from this software without specific prior written permission.
14	//
15	// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
16	// INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
17	// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
18	// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
19	// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
20	// WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
21	// USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
22
23	use std::{cmp::Ordering, time::Instant};
24
25	use log::*;
26	use tari_common_types::chain_metadata::ChainMetadata;
27	use tari_comms::peer_manager::NodeId;
28
29	#[cfg(feature = "metrics")]
30	use crate::base_node::metrics;
31	use crate::{
32	base_node::{
33	comms_interface::BlockEvent,
34	state_machine_service::states::{BlockSyncInfo, StateEvent, StateInfo, StatusInfo},
35	sync::{BlockHeaderSyncError, HeaderSynchronizer, SyncPeer},
36	BaseNodeStateMachine,
37	},
38	chain_storage::BlockchainBackend,
39	};
40	const LOG_TARGET: &str = "c::bn::header_sync";
41
42	#[derive(Clone, Debug)]
43	pub struct HeaderSyncState {
44	sync_peers: Vec<SyncPeer>,
45	is_synced: bool,
46	local_metadata: ChainMetadata,
47	}
48
49	impl HeaderSyncState {
50	pub fn new(mut sync_peers: Vec<SyncPeer>, local_metadata: ChainMetadata) -> Self {	15✔
51	// Sort by latency lowest to highest	15✔
52	sync_peers.sort_by(\|a, b\| match a.claimed_difficulty().cmp(&b.claimed_difficulty()) {	15✔
53	Ordering::Less => Ordering::Less,	×
54	// No latency goes to the end
55	Ordering::Greater => Ordering::Greater,	×
56	Ordering::Equal => {
57	match (a.latency(), b.latency()) {	×
58	(None, None) => Ordering::Equal,	×
59	// No latency goes to the end
60	(Some(_), None) => Ordering::Less,	×
61	(None, Some(_)) => Ordering::Greater,	×
62	(Some(la), Some(lb)) => la.cmp(&lb),	×
63	}
64	},
65	});	15✔
66	Self {	15✔
67	sync_peers,	15✔
68	is_synced: false,	15✔
69	local_metadata,	15✔
70	}	15✔
71	}	15✔
72
UNCOV 73	pub fn is_synced(&self) -> bool {	×
UNCOV 74	self.is_synced	×
UNCOV 75	}	×
76
77	pub fn into_sync_peers(self) -> Vec<SyncPeer> {	2✔
78	self.sync_peers	2✔
79	}	2✔
80
81	fn remove_sync_peer(&mut self, node_id: &NodeId) {	×
82	if let Some(pos) = self.sync_peers.iter().position(\|p\| p.node_id() == node_id) {	×
83	self.sync_peers.remove(pos);	×
84	}	×
85	}	×
86
87	// converting u64 to i64 is okay as the future time limit is the hundreds so way below u32 even
88	#[allow(clippy::too_many_lines)]
89	#[allow(clippy::cast_possible_wrap)]
90	pub async fn next_event<B: BlockchainBackend + 'static>(	15✔
91	&mut self,	15✔
92	shared: &mut BaseNodeStateMachine<B>,	15✔
93	) -> StateEvent {	15✔
94	// Only sync to peers with better claimed accumulated difficulty than the local chain: this may be possible	15✔
95	// at this stage due to read-write lock race conditions in the database	15✔
96	match shared.db.get_chain_metadata().await {	15✔
97	Ok(best_block_metadata) => {	14✔
98	let mut remove = Vec::new();	14✔
99	for sync_peer in &self.sync_peers {	28✔
100	if sync_peer.claimed_chain_metadata().accumulated_difficulty() <=	14✔
101	best_block_metadata.accumulated_difficulty()	14✔
102	{	×
103	remove.push(sync_peer.node_id().clone());	×
104	}	14✔
105	}
106	for node_id in remove {	14✔
107	self.remove_sync_peer(&node_id);	×
108	}	×
109	if self.sync_peers.is_empty() {	14✔
110	// Go back to Listening state
111	return StateEvent::Continue;	×
112	}	14✔
113	},
114	Err(e) => return StateEvent::FatalError(format!("{}", e)),	×
115	}
116
117	let mut synchronizer = HeaderSynchronizer::new(	14✔
118	shared.config.blockchain_sync_config.clone(),	14✔
119	shared.db.clone(),	14✔
120	shared.consensus_rules.clone(),	14✔
121	shared.connectivity.clone(),	14✔
122	&mut self.sync_peers,	14✔
123	shared.randomx_factory.clone(),	14✔
124	&self.local_metadata,	14✔
125	);	14✔
126		14✔
127	let status_event_sender = shared.status_event_sender.clone();	14✔
128	let bootstrapped = shared.is_bootstrapped();	14✔
129	let randomx_vm_cnt = shared.get_randomx_vm_cnt();	14✔
130	let randomx_vm_flags = shared.get_randomx_vm_flags();	14✔
131	synchronizer.on_starting(move \|sync_peer\| {	14✔
132	let _result = status_event_sender.send(StatusInfo {	14✔
133	bootstrapped,	14✔
134	state_info: StateInfo::Connecting(sync_peer.clone()),	14✔
135	randomx_vm_cnt,	14✔
136	randomx_vm_flags,	14✔
137	});	14✔
138	});	14✔
139		14✔
140	let status_event_sender = shared.status_event_sender.clone();	14✔
141	synchronizer.on_progress(move \|current_height, remote_tip_height, sync_peer\| {	14✔
142	let details = BlockSyncInfo {	6✔
143	tip_height: remote_tip_height,	6✔
144	local_height: current_height,	6✔
145	sync_peer: sync_peer.clone(),	6✔
146	};	6✔
147	let _result = status_event_sender.send(StatusInfo {	6✔
148	bootstrapped,	6✔
149	state_info: StateInfo::HeaderSync(Some(details)),	6✔
150	randomx_vm_cnt,	6✔
151	randomx_vm_flags,	6✔
152	});	6✔
153	});	14✔
154		14✔
155	let local_nci = shared.local_node_interface.clone();	14✔
156	synchronizer.on_rewind(move \|removed\| {	14✔
157	#[cfg(feature = "metrics")]
158	if let Some(fork_height) = removed.last().map(\|b\| b.height().saturating_sub(1)) {	1✔
159	metrics::tip_height().set(fork_height as i64);	1✔
160	metrics::reorg(fork_height, 0, removed.len()).inc();	1✔
161	}	1✔
162
163	local_nci.publish_block_event(BlockEvent::BlockSyncRewind(removed));	1✔
164	});	14✔
165		14✔
166	let timer = Instant::now();	14✔
167	match synchronizer.synchronize().await {	14✔
168	Ok((sync_peer, sync_result)) => {	9✔
169	info!(	9✔
170	target: LOG_TARGET,	×
171	"Headers synchronized from peer {} in {:.0?}",	×
172	sync_peer,	×
173	timer.elapsed()	×
174	);
175	// Move the sync peer used in header sync to the front of the queue
176	if let Some(pos) = self.sync_peers.iter().position(\|p\| *p == sync_peer) {	9✔
177	if pos > 0 {	9✔
178	let sync_peer = self.sync_peers.remove(pos);	×
179	self.sync_peers.insert(0, sync_peer);	×
180	}	9✔
181	}	×
182	self.is_synced = true;	9✔
183	StateEvent::HeadersSynchronized(sync_peer, sync_result)	9✔
184	},
185	Err(err) => {	5✔
186	let _ignore = shared.status_event_sender.send(StatusInfo {	5✔
187	bootstrapped,	5✔
188	state_info: StateInfo::SyncFailed("HeaderSyncFailed".to_string()),	5✔
189	randomx_vm_cnt,	5✔
190	randomx_vm_flags,	5✔
191	});	5✔
192	match err {	5✔
193	BlockHeaderSyncError::SyncFailedAllPeers => {
194	error!(target: LOG_TARGET, "Header sync failed with all peers. Error: {}", err);	×
195	warn!(target: LOG_TARGET, "{}. Continuing...", err);	×
196	StateEvent::Continue	×
197	},
198	_ => {
199	debug!(target: LOG_TARGET, "Header sync failed: {}", err);	5✔
200	StateEvent::HeaderSyncFailed(err.to_string())	5✔
201	},
202	}
203	},
204	}
205	}	14✔
206	}

tari-project / tari / 15280118615

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous