17152995555

Committed 22 Aug 2025 10:35AM UTC coverage: 59.322% (+3.4%) from 55.905%

Build # 17152995555

Build Type

push

github

Committed by

web-flow

Commit Message

feat!: update c29 fork block (#7434)

Description
---

Updated the c29 fork block to be at height `95180` and approximately
`target_time = Tue, 16 Aug 2025 10:00:00 +0000`. This has been
calculated as follows:

- Get block timings from the base node for the last 10,000 blocks.

```
11:24 v5.0.0-pre.4 mainnet State: Listening Tip: 77442 (Fri, 22 Aug 2025 09:22:28 +0000) ...
>>
>> block-timing 10000
Timing for blocks #77442 - #67443
Max block time: 1443
Min block time: 1
Avg block time: 121.8988898889889
>>
>>
>> watch status
11:24 v5.0.0-pre.4 mainnet State: Listening Tip: 77442 (Fri, 22 Aug 2025 09:22:28 +0000) ...
```

Our start time is the time published by the base node:
- `start_time = Fri, 22 Aug 2025 09:22:28 +0000`

Our start block is the block published by the base node:
- `start_block = 77442`

Approximate time difference in seconds: `target_time - start_time =
2162147 seconds`. Using average block time, we get `77442 + 2162147 /
121.899 = 95179.2` then rounded up to `95180`.

Motivation and Context
---
The fork block height was wrong.

How Has This Been Tested?
---
See the calculation above.

What process can a PR reviewer use to test or verify this change?
---
Code review and review the calculation in the PR description.

<!-- Checklist -->
<!-- 1. Is the title of your PR in the form that would make nice release
notes? The title, excluding the conventional commit
tag, will be included exactly as is in the CHANGELOG, so please think
about it carefully. -->


Breaking Changes
---

- [x] None
- [ ] Requires data directory on base node to be deleted
- [ ] Requires hard fork
- [ ] Other - Please specify

<!-- Does this include a breaking change? If so, include this line as a
footer -->
BREAKING CHANGE: Consensus constant change to
`con_5.effective_from_height = 95_180;`.


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Chores**
* Postponed the mainnet V2 network upgrade to block height 95... (continued)

Coverage Stats

1 of 1 new or added line in 1 file covered. (100.0%)

5077 existing lines in 113 files now uncovered.

70233 of 118392 relevant lines covered (59.32%)

538661.6 hits per line

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

0.0

/base_layer/core/src/base_node/sync/block_sync/synchronizer.rs

//  Copyright 2020, 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::{
    convert::{TryFrom, TryInto},
    sync::Arc,
    time::{Duration, Instant},
};

use futures::StreamExt;
use log::*;
use tari_comms::{connectivity::ConnectivityRequester, peer_manager::NodeId, protocol::rpc::RpcClient, PeerConnection};
use tari_transaction_components::{aggregated_body::AggregateBody, BanPeriod};
use tari_utilities::hex::Hex;

use super::error::BlockSyncError;
use crate::{
    base_node::{
        sync::{ban::PeerBanManager, hooks::Hooks, rpc, SyncPeer},
        BlockchainSyncConfig,
    },
    blocks::{Block, ChainBlock},
    chain_storage::{async_db::AsyncBlockchainDb, BlockchainBackend},
    common::rolling_avg::RollingAverageTime,
    proto::base_node::SyncBlocksRequest,
    validation::{BlockBodyValidator, ValidationError},
};
const LOG_TARGET: &str = "c::bn::block_sync";

const MAX_LATENCY_INCREASES: usize = 5;

pub struct BlockSynchronizer<'a, B> {
    config: BlockchainSyncConfig,
    db: AsyncBlockchainDb<B>,
    connectivity: ConnectivityRequester,
    sync_peers: &'a mut Vec<SyncPeer>,
    block_validator: Arc<dyn BlockBodyValidator<B>>,
    hooks: Hooks,
    peer_ban_manager: PeerBanManager,
}

impl<'a, B: BlockchainBackend + 'static> BlockSynchronizer<'a, B> {
    pub fn new(
        config: BlockchainSyncConfig,
        db: AsyncBlockchainDb<B>,
        connectivity: ConnectivityRequester,
        sync_peers: &'a mut Vec<SyncPeer>,
        block_validator: Arc<dyn BlockBodyValidator<B>>,
    ) -> Self {
        let peer_ban_manager = PeerBanManager::new(config.clone(), connectivity.clone());
        Self {
            config,
            db,
            connectivity,
            sync_peers,
            block_validator,
            hooks: Default::default(),
            peer_ban_manager,
        }
    }

    pub fn on_starting<H>(&mut self, hook: H)
    where for<'r> H: FnOnce(&SyncPeer) + Send + Sync + 'static {
        self.hooks.add_on_starting_hook(hook);
    }

    pub fn on_progress<H>(&mut self, hook: H)
    where H: Fn(Arc<ChainBlock>, u64, &SyncPeer) + Send + Sync + 'static {
        self.hooks.add_on_progress_block_hook(hook);
    }

    pub fn on_complete<H>(&mut self, hook: H)
    where H: Fn(Arc<ChainBlock>, u64) + Send + Sync + 'static {
        self.hooks.add_on_complete_hook(hook);
    }

    pub async fn synchronize(&mut self) -> Result<(), BlockSyncError> {
        let mut max_latency = self.config.initial_max_sync_latency;
        let mut sync_round = 0;
        let mut latency_increases_counter = 0;
        loop {
            match self.attempt_block_sync(max_latency).await {
                Ok(_) => return Ok(()),
                Err(err @ BlockSyncError::AllSyncPeersExceedLatency) => {
                    warn!(target: LOG_TARGET, "{err}");
                    max_latency += self.config.max_latency_increase;
                    warn!(
                        target: LOG_TARGET,
                        "Retrying block sync with increased max latency {:.2?} with {} sync peers",
                        max_latency,
                        self.sync_peers.len()
                    );
                    latency_increases_counter += 1;
                    if latency_increases_counter > MAX_LATENCY_INCREASES {
                        return Err(err);
                    }
                    // Prohibit using a few slow sync peers only, rather get new sync peers assigned
                    if self.sync_peers.len() < 2 {
                        return Err(err);
                    } else {
                        continue;
                    }
                },
                Err(err @ BlockSyncError::SyncRoundFailed) => {
                    sync_round += 1;
                    warn!(target: LOG_TARGET, "{err} ({sync_round})");
                    continue;
                },
                Err(err) => {
                    return Err(err);
                },
            }
        }
    }

    async fn attempt_block_sync(&mut self, max_latency: Duration) -> Result<(), BlockSyncError> {
        let sync_peer_node_ids = self.sync_peers.iter().map(|p| p.node_id()).cloned().collect::<Vec<_>>();
        info!(
            target: LOG_TARGET,
            "Attempting to sync blocks({} sync peers)",
            sync_peer_node_ids.len()
        );
        let mut latency_counter = 0usize;
        for node_id in sync_peer_node_ids {
            let peer_index = self.get_sync_peer_index(&node_id).ok_or(BlockSyncError::PeerNotFound)?;
            let sync_peer = self.sync_peers.get(peer_index).expect("Already checked");
            self.hooks.call_on_starting_hook(sync_peer);
            let mut conn = match self.connect_to_sync_peer(node_id.clone()).await {
                Ok(val) => val,
                Err(e) => {
                    warn!(
                        target: LOG_TARGET,
                        "Failed to connect to sync peer `{node_id}`: {e}"
                    );
                    self.remove_sync_peer(&node_id);
                    continue;
                },
            };
            let config = RpcClient::builder()
                .with_deadline(self.config.rpc_deadline)
                .with_deadline_grace_period(Duration::from_secs(5));
            let mut client = match conn
                .connect_rpc_using_builder::<rpc::BaseNodeSyncRpcClient>(config)
                .await
            {
                Ok(val) => val,
                Err(e) => {
                    warn!(
                        target: LOG_TARGET,
                        "Failed to obtain RPC connection from sync peer `{node_id}`: {e}"
                    );
                    self.remove_sync_peer(&node_id);
                    continue;
                },
            };
            let latency = client
                .get_last_request_latency()
                .expect("unreachable panic: last request latency must be set after connect");
            self.sync_peers
                .get_mut(peer_index)
                .expect("Already checked")
                .set_latency(latency);
            let sync_peer = self.sync_peers.get(peer_index).expect("Already checked").clone();
            info!(
                target: LOG_TARGET,
                "Attempting to synchronize blocks with `{node_id}` latency: {latency:.2?}"
            );
            match self.synchronize_blocks(sync_peer, client, max_latency).await {
                Ok(_) => return Ok(()),
                Err(err) => {
                    warn!(target: LOG_TARGET, "{err}");
                    let ban_reason = BlockSyncError::get_ban_reason(&err);
                    if let Some(reason) = ban_reason {
                        let duration = match reason.ban_duration {
                            BanPeriod::Short => self.config.short_ban_period,
                            BanPeriod::Long => self.config.ban_period,
                        };
                        self.peer_ban_manager
                            .ban_peer_if_required(&node_id, reason.reason, duration)
                            .await;
                    }
                    if let BlockSyncError::MaxLatencyExceeded { .. } = err {
                        latency_counter += 1;
                    } else {
                        self.remove_sync_peer(&node_id);
                    }
                },
            }
        }

        if self.sync_peers.is_empty() {
            Err(BlockSyncError::NoMoreSyncPeers("Block sync failed".to_string()))
        } else if latency_counter >= self.sync_peers.len() {
            Err(BlockSyncError::AllSyncPeersExceedLatency)
        } else {
            Err(BlockSyncError::SyncRoundFailed)
        }
    }

    async fn connect_to_sync_peer(&self, peer: NodeId) -> Result<PeerConnection, BlockSyncError> {
        let connection = self.connectivity.dial_peer(peer).await?;
        Ok(connection)
    }

    #[allow(clippy::too_many_lines)]
    async fn synchronize_blocks(
        &mut self,
        mut sync_peer: SyncPeer,
        mut client: rpc::BaseNodeSyncRpcClient,
        max_latency: Duration,
    ) -> Result<(), BlockSyncError> {
        info!(target: LOG_TARGET, "Starting block sync from peer {}", sync_peer.node_id());

        let tip_header = self.db.fetch_last_header().await?;
        let local_metadata = self.db.get_chain_metadata().await?;

        if tip_header.height <= local_metadata.best_block_height() {
            debug!(
                target: LOG_TARGET,
                "Blocks already synchronized to height {}.", tip_header.height
            );
            return Ok(());
        }

        let tip_hash = tip_header.hash();
        let tip_height = tip_header.height;
        let best_height = local_metadata.best_block_height();
        let chain_header = self.db.fetch_chain_header(best_height).await?;

        let best_full_block_hash = chain_header.accumulated_data().hash;
        debug!(
            target: LOG_TARGET,
            "Starting block sync from peer `{}`. Current best block is #{} `{}`. Syncing to #{} ({}).",
            sync_peer,
            best_height,
            best_full_block_hash.to_hex(),
            tip_height,
            tip_hash.to_hex()
        );
        let request = SyncBlocksRequest {
            start_hash: best_full_block_hash.to_vec(),
            // To the tip!
            end_hash: tip_hash.to_vec(),
        };

        let mut block_stream = client.sync_blocks(request).await?;
        let mut prev_hash = best_full_block_hash;
        let mut current_block = None;
        let mut last_sync_timer = Instant::now();
        let mut avg_latency = RollingAverageTime::new(20);
        while let Some(block_result) = block_stream.next().await {
            let latency = last_sync_timer.elapsed();
            avg_latency.add_sample(latency);
            let block_body_response = block_result?;

            let header = self
                .db
                .fetch_chain_header_by_block_hash(block_body_response.hash.clone().try_into()?)
                .await?
                .ok_or_else(|| {
                    BlockSyncError::UnknownHeaderHash(format!(
                        "Peer sent hash ({}) for block header we do not have",
                        block_body_response.hash.to_hex()
                    ))
                })?;

            let current_height = header.height();
            let header_hash = *header.hash();
            let timestamp = header.timestamp();

            if header.header().prev_hash != prev_hash {
                return Err(BlockSyncError::BlockWithoutParent {
                    expected: prev_hash.to_hex(),
                    got: header.header().prev_hash.to_hex(),
                });
            }

            prev_hash = header_hash;

            let body = block_body_response
                .body
                .map(AggregateBody::try_from)
                .ok_or_else(|| BlockSyncError::InvalidBlockBody("Peer sent empty block".to_string()))?
                .map_err(BlockSyncError::InvalidBlockBody)?;

            debug!(
                target: LOG_TARGET,
                "Validating block body #{} (PoW = {}, {}, latency: {:.2?})",
                current_height,
                header.header().pow_algo(),
                body.to_counts_string(),
                latency
            );

            let timer = Instant::now();
            let (header, header_accum_data) = header.into_parts();
            let block = Block::new(header, body);

            // Validate the block inside a tokio task
            let task_block = block.clone();
            let db = self.db.inner().clone();
            let validator = self.block_validator.clone();
            let res = {
                let txn = db.db_read_access()?;
                validator.validate_body(&*txn, &task_block)
            };

            let block = match res {
                Ok(block) => block,
                Err(err @ ValidationError::BadBlockFound { .. }) | Err(err @ ValidationError::FatalStorageError(_)) => {
                    return Err(err.into());
                },
                Err(err) => {
                    // Add to bad blocks
                    if let Err(err) = self
                        .db
                        .write_transaction()
                        .delete_orphan(header_hash)
                        .insert_bad_block(header_hash, current_height, err.to_string())
                        .commit()
                        .await
                    {
                        error!(target: LOG_TARGET, "Failed to insert bad block: {err}");
                    }
                    return Err(err.into());
                },
            };

            let block = ChainBlock::try_construct(Arc::new(block), header_accum_data)
                .map(Arc::new)
                .ok_or(BlockSyncError::FailedToConstructChainBlock)?;

            debug!(
                target: LOG_TARGET,
                "Validated in {:.0?}. Storing block body #{} (PoW = {}, {})",
                timer.elapsed(),
                block.header().height,
                block.header().pow_algo(),
                block.block().body.to_counts_string(),
            );
            trace!(
                target: LOG_TARGET,
                "{block}"
            );

            self.db
                .write_transaction()
                .delete_orphan(header_hash)
                .insert_tip_block_body(block.clone())
                .set_best_block(
                    block.height(),
                    header_hash,
                    block.accumulated_data().total_accumulated_difficulty,
                    block.header().prev_hash,
                    timestamp,
                )
                .commit()
                .await?;

            // Average time between receiving blocks from the peer - used to detect a slow sync peer
            let last_avg_latency = avg_latency.calculate_average_with_min_samples(5);
            if let Some(latency) = last_avg_latency {
                sync_peer.set_latency(latency);
            }
            // Includes time to add block to database, used to show blocks/s on status line
            sync_peer.add_sample(last_sync_timer.elapsed());
            self.hooks
                .call_on_progress_block_hooks(block.clone(), tip_height, &sync_peer);

            if let Some(avg_latency) = last_avg_latency {
                if avg_latency > max_latency {
                    return Err(BlockSyncError::MaxLatencyExceeded {
                        peer: sync_peer.node_id().clone(),
                        latency: avg_latency,
                        max_latency,
                    });
                }
            }

            current_block = Some(block);
            last_sync_timer = Instant::now();
        }
        debug!(
            "Sync peer claim at start  - height: {}, accumulated difficulty: {}",
            sync_peer.claimed_chain_metadata().best_block_height(),
            sync_peer.claimed_chain_metadata().accumulated_difficulty(),
        );
        debug!(
            "Our best header at start  - height: {}, accumulated difficulty: {}",
            best_height,
            chain_header.accumulated_data().total_accumulated_difficulty,
        );
        let metadata_after_sync = self.db.get_chain_metadata().await?;
        debug!(
            "Our best block after sync - height: {}, accumulated difficulty: {}",
            metadata_after_sync.best_block_height(),
            metadata_after_sync.accumulated_difficulty(),
        );

        if metadata_after_sync.accumulated_difficulty() < sync_peer.claimed_chain_metadata().accumulated_difficulty() {
            return Err(BlockSyncError::PeerDidNotSupplyAllClaimedBlocks(format!(
                "Their claim - height: {}, accumulated difficulty: {}. Our status after block sync - height: {}, \
                 accumulated difficulty: {}",
                sync_peer.claimed_chain_metadata().best_block_height(),
                sync_peer.claimed_chain_metadata().accumulated_difficulty(),
                metadata_after_sync.best_block_height(),
                metadata_after_sync.accumulated_difficulty(),
            )));
        }

        if let Some(block) = current_block {
            self.hooks.call_on_complete_hooks(block, best_height);
        }

        debug!(target: LOG_TARGET, "Completed block sync with peer `{sync_peer}`");

        Ok(())
    }

    // Sync peers are also removed from the list of sync peers if the ban duration is longer than the short ban period.
    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);
        }
    }

    // Helper function to get the index to the node_id inside of the vec of peers
    fn get_sync_peer_index(&mut self, node_id: &NodeId) -> Option<usize> {
        self.sync_peers.iter().position(|p| p.node_id() == node_id)
    }
}

1	// Copyright 2020, 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::{
24	convert::{TryFrom, TryInto},
25	sync::Arc,
26	time::{Duration, Instant},
27	};
28
29	use futures::StreamExt;
30	use log::*;
31	use tari_comms::{connectivity::ConnectivityRequester, peer_manager::NodeId, protocol::rpc::RpcClient, PeerConnection};
32	use tari_transaction_components::{aggregated_body::AggregateBody, BanPeriod};
33	use tari_utilities::hex::Hex;
34
35	use super::error::BlockSyncError;
36	use crate::{
37	base_node::{
38	sync::{ban::PeerBanManager, hooks::Hooks, rpc, SyncPeer},
39	BlockchainSyncConfig,
40	},
41	blocks::{Block, ChainBlock},
42	chain_storage::{async_db::AsyncBlockchainDb, BlockchainBackend},
43	common::rolling_avg::RollingAverageTime,
44	proto::base_node::SyncBlocksRequest,
45	validation::{BlockBodyValidator, ValidationError},
46	};
47	const LOG_TARGET: &str = "c::bn::block_sync";
48
49	const MAX_LATENCY_INCREASES: usize = 5;
50
51	pub struct BlockSynchronizer<'a, B> {
52	config: BlockchainSyncConfig,
53	db: AsyncBlockchainDb<B>,
54	connectivity: ConnectivityRequester,
55	sync_peers: &'a mut Vec<SyncPeer>,
56	block_validator: Arc<dyn BlockBodyValidator<B>>,
57	hooks: Hooks,
58	peer_ban_manager: PeerBanManager,
59	}
60
61	impl<'a, B: BlockchainBackend + 'static> BlockSynchronizer<'a, B> {
UNCOV 62	pub fn new(	×
63	config: BlockchainSyncConfig,	×
64	db: AsyncBlockchainDb<B>,	×
65	connectivity: ConnectivityRequester,	×
66	sync_peers: &'a mut Vec<SyncPeer>,	×
67	block_validator: Arc<dyn BlockBodyValidator<B>>,	×
68	) -> Self {	×
69	let peer_ban_manager = PeerBanManager::new(config.clone(), connectivity.clone());	×
70	Self {	×
71	config,	×
72	db,	×
73	connectivity,	×
74	sync_peers,	×
75	block_validator,	×
76	hooks: Default::default(),	×
77	peer_ban_manager,	×
78	}	×
79	}	×
80
UNCOV 81	pub fn on_starting<H>(&mut self, hook: H)	×
82	where for<'r> H: FnOnce(&SyncPeer) + Send + Sync + 'static {	×
83	self.hooks.add_on_starting_hook(hook);	×
84	}	×
85
UNCOV 86	pub fn on_progress<H>(&mut self, hook: H)	×
87	where H: Fn(Arc<ChainBlock>, u64, &SyncPeer) + Send + Sync + 'static {	×
88	self.hooks.add_on_progress_block_hook(hook);	×
89	}	×
90
UNCOV 91	pub fn on_complete<H>(&mut self, hook: H)	×
92	where H: Fn(Arc<ChainBlock>, u64) + Send + Sync + 'static {	×
93	self.hooks.add_on_complete_hook(hook);	×
94	}	×
95
UNCOV 96	pub async fn synchronize(&mut self) -> Result<(), BlockSyncError> {	×
97	let mut max_latency = self.config.initial_max_sync_latency;	×
98	let mut sync_round = 0;	×
99	let mut latency_increases_counter = 0;	×
100	loop {
UNCOV 101	match self.attempt_block_sync(max_latency).await {	×
102	Ok(_) => return Ok(()),	×
103	Err(err @ BlockSyncError::AllSyncPeersExceedLatency) => {	×
104	warn!(target: LOG_TARGET, "{err}");	×
105	max_latency += self.config.max_latency_increase;	×
106	warn!(	×
107	target: LOG_TARGET,	×
108	"Retrying block sync with increased max latency {:.2?} with {} sync peers",	×
109	max_latency,	×
110	self.sync_peers.len()	×
111	);
UNCOV 112	latency_increases_counter += 1;	×
113	if latency_increases_counter > MAX_LATENCY_INCREASES {	×
114	return Err(err);	×
115	}	×
116	// Prohibit using a few slow sync peers only, rather get new sync peers assigned	×
117	if self.sync_peers.len() < 2 {	×
118	return Err(err);	×
119	} else {
UNCOV 120	continue;	×
121	}
122	},
UNCOV 123	Err(err @ BlockSyncError::SyncRoundFailed) => {	×
124	sync_round += 1;	×
125	warn!(target: LOG_TARGET, "{err} ({sync_round})");	×
126	continue;	×
127	},
UNCOV 128	Err(err) => {	×
129	return Err(err);	×
130	},
131	}
132	}
UNCOV 133	}	×
134
UNCOV 135	async fn attempt_block_sync(&mut self, max_latency: Duration) -> Result<(), BlockSyncError> {	×
136	let sync_peer_node_ids = self.sync_peers.iter().map(\|p\| p.node_id()).cloned().collect::<Vec<_>>();	×
137	info!(	×
138	target: LOG_TARGET,	×
139	"Attempting to sync blocks({} sync peers)",	×
140	sync_peer_node_ids.len()	×
141	);
UNCOV 142	let mut latency_counter = 0usize;	×
143	for node_id in sync_peer_node_ids {	×
144	let peer_index = self.get_sync_peer_index(&node_id).ok_or(BlockSyncError::PeerNotFound)?;	×
145	let sync_peer = self.sync_peers.get(peer_index).expect("Already checked");	×
146	self.hooks.call_on_starting_hook(sync_peer);	×
147	let mut conn = match self.connect_to_sync_peer(node_id.clone()).await {	×
148	Ok(val) => val,	×
149	Err(e) => {	×
150	warn!(	×
151	target: LOG_TARGET,	×
152	"Failed to connect to sync peer `{node_id}`: {e}"	×
153	);
UNCOV 154	self.remove_sync_peer(&node_id);	×
155	continue;	×
156	},
157	};
UNCOV 158	let config = RpcClient::builder()	×
159	.with_deadline(self.config.rpc_deadline)	×
160	.with_deadline_grace_period(Duration::from_secs(5));	×
161	let mut client = match conn	×
162	.connect_rpc_using_builder::<rpc::BaseNodeSyncRpcClient>(config)	×
163	.await	×
164	{
UNCOV 165	Ok(val) => val,	×
166	Err(e) => {	×
167	warn!(	×
168	target: LOG_TARGET,	×
169	"Failed to obtain RPC connection from sync peer `{node_id}`: {e}"	×
170	);
UNCOV 171	self.remove_sync_peer(&node_id);	×
172	continue;	×
173	},
174	};
UNCOV 175	let latency = client	×
176	.get_last_request_latency()	×
177	.expect("unreachable panic: last request latency must be set after connect");	×
178	self.sync_peers	×
179	.get_mut(peer_index)	×
180	.expect("Already checked")	×
181	.set_latency(latency);	×
182	let sync_peer = self.sync_peers.get(peer_index).expect("Already checked").clone();	×
183	info!(	×
184	target: LOG_TARGET,	×
185	"Attempting to synchronize blocks with `{node_id}` latency: {latency:.2?}"	×
186	);
UNCOV 187	match self.synchronize_blocks(sync_peer, client, max_latency).await {	×
188	Ok(_) => return Ok(()),	×
189	Err(err) => {	×
190	warn!(target: LOG_TARGET, "{err}");	×
191	let ban_reason = BlockSyncError::get_ban_reason(&err);	×
192	if let Some(reason) = ban_reason {	×
193	let duration = match reason.ban_duration {	×
194	BanPeriod::Short => self.config.short_ban_period,	×
195	BanPeriod::Long => self.config.ban_period,	×
196	};
UNCOV 197	self.peer_ban_manager	×
198	.ban_peer_if_required(&node_id, reason.reason, duration)	×
199	.await;	×
200	}	×
201	if let BlockSyncError::MaxLatencyExceeded { .. } = err {	×
202	latency_counter += 1;	×
203	} else {	×
204	self.remove_sync_peer(&node_id);	×
205	}	×
206	},
207	}
208	}
209
UNCOV 210	if self.sync_peers.is_empty() {	×
211	Err(BlockSyncError::NoMoreSyncPeers("Block sync failed".to_string()))	×
212	} else if latency_counter >= self.sync_peers.len() {	×
213	Err(BlockSyncError::AllSyncPeersExceedLatency)	×
214	} else {
UNCOV 215	Err(BlockSyncError::SyncRoundFailed)	×
216	}
UNCOV 217	}	×
218
UNCOV 219	async fn connect_to_sync_peer(&self, peer: NodeId) -> Result<PeerConnection, BlockSyncError> {	×
220	let connection = self.connectivity.dial_peer(peer).await?;	×
221	Ok(connection)	×
222	}	×
223
224	#[allow(clippy::too_many_lines)]
UNCOV 225	async fn synchronize_blocks(	×
226	&mut self,	×
227	mut sync_peer: SyncPeer,	×
228	mut client: rpc::BaseNodeSyncRpcClient,	×
229	max_latency: Duration,	×
230	) -> Result<(), BlockSyncError> {	×
231	info!(target: LOG_TARGET, "Starting block sync from peer {}", sync_peer.node_id());	×
232
UNCOV 233	let tip_header = self.db.fetch_last_header().await?;	×
234	let local_metadata = self.db.get_chain_metadata().await?;	×
235
UNCOV 236	if tip_header.height <= local_metadata.best_block_height() {	×
237	debug!(	×
238	target: LOG_TARGET,	×
239	"Blocks already synchronized to height {}.", tip_header.height	×
240	);
UNCOV 241	return Ok(());	×
242	}	×
243		×
244	let tip_hash = tip_header.hash();	×
245	let tip_height = tip_header.height;	×
246	let best_height = local_metadata.best_block_height();	×
247	let chain_header = self.db.fetch_chain_header(best_height).await?;	×
248
UNCOV 249	let best_full_block_hash = chain_header.accumulated_data().hash;	×
250	debug!(	×
251	target: LOG_TARGET,	×
252	"Starting block sync from peer `{}`. Current best block is #{} `{}`. Syncing to #{} ({}).",	×
253	sync_peer,	×
254	best_height,	×
255	best_full_block_hash.to_hex(),	×
256	tip_height,	×
257	tip_hash.to_hex()	×
258	);
UNCOV 259	let request = SyncBlocksRequest {	×
260	start_hash: best_full_block_hash.to_vec(),	×
261	// To the tip!	×
262	end_hash: tip_hash.to_vec(),	×
263	};	×
264
UNCOV 265	let mut block_stream = client.sync_blocks(request).await?;	×
266	let mut prev_hash = best_full_block_hash;	×
267	let mut current_block = None;	×
268	let mut last_sync_timer = Instant::now();	×
269	let mut avg_latency = RollingAverageTime::new(20);	×
270	while let Some(block_result) = block_stream.next().await {	×
271	let latency = last_sync_timer.elapsed();	×
272	avg_latency.add_sample(latency);	×
273	let block_body_response = block_result?;	×
274
UNCOV 275	let header = self	×
276	.db	×
277	.fetch_chain_header_by_block_hash(block_body_response.hash.clone().try_into()?)	×
278	.await?	×
279	.ok_or_else(\|\| {	×
280	BlockSyncError::UnknownHeaderHash(format!(	×
281	"Peer sent hash ({}) for block header we do not have",	×
282	block_body_response.hash.to_hex()	×
283	))	×
284	})?;	×
285
UNCOV 286	let current_height = header.height();	×
287	let header_hash = *header.hash();	×
288	let timestamp = header.timestamp();	×
289		×
290	if header.header().prev_hash != prev_hash {	×
291	return Err(BlockSyncError::BlockWithoutParent {	×
292	expected: prev_hash.to_hex(),	×
293	got: header.header().prev_hash.to_hex(),	×
294	});	×
295	}	×
296		×
297	prev_hash = header_hash;	×
298
UNCOV 299	let body = block_body_response	×
300	.body	×
301	.map(AggregateBody::try_from)	×
302	.ok_or_else(\|\| BlockSyncError::InvalidBlockBody("Peer sent empty block".to_string()))?	×
303	.map_err(BlockSyncError::InvalidBlockBody)?;	×
304
UNCOV 305	debug!(	×
306	target: LOG_TARGET,	×
307	"Validating block body #{} (PoW = {}, {}, latency: {:.2?})",	×
308	current_height,	×
309	header.header().pow_algo(),	×
310	body.to_counts_string(),	×
311	latency
312	);
313
UNCOV 314	let timer = Instant::now();	×
315	let (header, header_accum_data) = header.into_parts();	×
316	let block = Block::new(header, body);	×
317		×
318	// Validate the block inside a tokio task	×
319	let task_block = block.clone();	×
320	let db = self.db.inner().clone();	×
321	let validator = self.block_validator.clone();	×
322	let res = {	×
323	let txn = db.db_read_access()?;	×
324	validator.validate_body(&*txn, &task_block)	×
325	};
326
UNCOV 327	let block = match res {	×
328	Ok(block) => block,	×
329	Err(err @ ValidationError::BadBlockFound { .. }) \| Err(err @ ValidationError::FatalStorageError(_)) => {	×
330	return Err(err.into());	×
331	},
UNCOV 332	Err(err) => {	×
333	// Add to bad blocks
UNCOV 334	if let Err(err) = self	×
335	.db	×
336	.write_transaction()	×
337	.delete_orphan(header_hash)	×
338	.insert_bad_block(header_hash, current_height, err.to_string())	×
339	.commit()	×
340	.await	×
341	{
UNCOV 342	error!(target: LOG_TARGET, "Failed to insert bad block: {err}");	×
343	}	×
344	return Err(err.into());	×
345	},
346	};
347
UNCOV 348	let block = ChainBlock::try_construct(Arc::new(block), header_accum_data)	×
349	.map(Arc::new)	×
350	.ok_or(BlockSyncError::FailedToConstructChainBlock)?;	×
351
UNCOV 352	debug!(	×
353	target: LOG_TARGET,	×
354	"Validated in {:.0?}. Storing block body #{} (PoW = {}, {})",	×
355	timer.elapsed(),	×
356	block.header().height,	×
357	block.header().pow_algo(),	×
358	block.block().body.to_counts_string(),	×
359	);
UNCOV 360	trace!(	×
361	target: LOG_TARGET,	×
362	"{block}"	×
363	);
364
UNCOV 365	self.db	×
366	.write_transaction()	×
367	.delete_orphan(header_hash)	×
368	.insert_tip_block_body(block.clone())	×
369	.set_best_block(	×
370	block.height(),	×
371	header_hash,	×
372	block.accumulated_data().total_accumulated_difficulty,	×
373	block.header().prev_hash,	×
374	timestamp,	×
375	)	×
376	.commit()	×
377	.await?;	×
378
379	// Average time between receiving blocks from the peer - used to detect a slow sync peer
UNCOV 380	let last_avg_latency = avg_latency.calculate_average_with_min_samples(5);	×
381	if let Some(latency) = last_avg_latency {	×
382	sync_peer.set_latency(latency);	×
383	}	×
384	// Includes time to add block to database, used to show blocks/s on status line
UNCOV 385	sync_peer.add_sample(last_sync_timer.elapsed());	×
386	self.hooks	×
387	.call_on_progress_block_hooks(block.clone(), tip_height, &sync_peer);	×
388
UNCOV 389	if let Some(avg_latency) = last_avg_latency {	×
390	if avg_latency > max_latency {	×
391	return Err(BlockSyncError::MaxLatencyExceeded {	×
392	peer: sync_peer.node_id().clone(),	×
393	latency: avg_latency,	×
394	max_latency,	×
395	});	×
396	}	×
397	}	×
398
UNCOV 399	current_block = Some(block);	×
400	last_sync_timer = Instant::now();	×
401	}
UNCOV 402	debug!(	×
403	"Sync peer claim at start - height: {}, accumulated difficulty: {}",	×
404	sync_peer.claimed_chain_metadata().best_block_height(),	×
405	sync_peer.claimed_chain_metadata().accumulated_difficulty(),	×
406	);
UNCOV 407	debug!(	×
408	"Our best header at start - height: {}, accumulated difficulty: {}",	×
409	best_height,	×
410	chain_header.accumulated_data().total_accumulated_difficulty,	×
411	);
UNCOV 412	let metadata_after_sync = self.db.get_chain_metadata().await?;	×
413	debug!(	×
414	"Our best block after sync - height: {}, accumulated difficulty: {}",	×
415	metadata_after_sync.best_block_height(),	×
416	metadata_after_sync.accumulated_difficulty(),	×
417	);
418
UNCOV 419	if metadata_after_sync.accumulated_difficulty() < sync_peer.claimed_chain_metadata().accumulated_difficulty() {	×
420	return Err(BlockSyncError::PeerDidNotSupplyAllClaimedBlocks(format!(	×
421	"Their claim - height: {}, accumulated difficulty: {}. Our status after block sync - height: {}, \	×
422	accumulated difficulty: {}",	×
423	sync_peer.claimed_chain_metadata().best_block_height(),	×
424	sync_peer.claimed_chain_metadata().accumulated_difficulty(),	×
425	metadata_after_sync.best_block_height(),	×
426	metadata_after_sync.accumulated_difficulty(),	×
427	)));	×
428	}	×
429
UNCOV 430	if let Some(block) = current_block {	×
431	self.hooks.call_on_complete_hooks(block, best_height);	×
432	}	×
433
UNCOV 434	debug!(target: LOG_TARGET, "Completed block sync with peer `{sync_peer}`");	×
435
UNCOV 436	Ok(())	×
437	}	×
438
439	// Sync peers are also removed from the list of sync peers if the ban duration is longer than the short ban period.
UNCOV 440	fn remove_sync_peer(&mut self, node_id: &NodeId) {	×
441	if let Some(pos) = self.sync_peers.iter().position(\|p\| p.node_id() == node_id) {	×
442	self.sync_peers.remove(pos);	×
443	}	×
444	}	×
445
446	// Helper function to get the index to the node_id inside of the vec of peers
UNCOV 447	fn get_sync_peer_index(&mut self, node_id: &NodeId) -> Option<usize> {	×
448	self.sync_peers.iter().position(\|p\| p.node_id() == node_id)	×
449	}	×
450	}

tari-project / tari / 17152995555

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