15326284040

Committed 29 May 2025 02:33PM UTC coverage: 72.791% (-0.8%) from 73.545%

Build # 15326284040

Build Type

push

github

Committed by

web-flow

Commit Message

fix: peer retention and connections (#7123)

Description
---
- Fixed peer retention in the database by improving delete all stale
peers logic.
- Improved error responses when peers could not be found in the
database.
- Improved DHT connection pools sync with peer database.

Motivation and Context
---
DHT neighbour and random pools were not in sync with actual connections
and peers in the peer db.

Issue - Why `PeerManagerError(PeerNotFoundError)`
```
2025-05-28 04:49:55.101009900 [comms::dht::connectivity] ERROR Error refreshing neighbour peer pool: PeerManagerError(PeerNotFoundError)
2025-05-28 04:49:55.101120100 [comms::connectivity::manager] TRACE Request (14743808475136314793): GetAllowList(Sender { inner: Some(Inner { state: State { is_complete: false, is_closed: false, is_rx_task_set: true, is_tx_task_set: false } }) })
2025-05-28 04:49:55.101160300 [comms::connectivity::manager] TRACE Request (14743808475136314793) done
2025-05-28 04:49:55.104823200 [comms::dht::connectivity] ERROR Error refreshing random peer pool: PeerManagerError(PeerNotFoundError)
```
Issue  -  Why `0 connected` but `active DHT connections: 10/12`
```
2025-05-28 04:49:55.104929100 [comms::dht::connectivity] DEBUG DHT connectivity status: neighbour pool: 8/8 (0 connected), random pool: 4/4 (0 connected, last refreshed 12777s ago), active DHT connections: 10/12
```
Issue - Why `Inbound pipeline returned an error: 'The requested peer
does not exist'`
```
2025-05-28 10:42:21.447513700 [comms::pipeline::inbound] WARN  Inbound pipeline returned an error: 'The requested peer does not exist'
```
How Has This Been Tested?
---
- Adapted unit test for improved delete all stale peers logic.
- System-level testing [**TBD**]

What process can a PR reviewer use to test or verify this change?
---
- Code review.
- System-level testing

<!-- 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 ... (continued)

Run Details

163 of 194 new or added lines in 11 files covered. (84.02%)

1028 existing lines in 27 files now uncovered.

82035 of 112699 relevant lines covered (72.79%)

252963.2 hits per line

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

81.82

/base_layer/core/src/base_node/sync/config.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::time::Duration;

use serde::{Deserialize, Serialize};
use tari_common::configuration::serializers;
use tari_comms::peer_manager::NodeId;
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct BlockchainSyncConfig {
    /// The initial max sync latency. If a peer fails to stream a header/block within this deadline another sync peer
    /// will be selected. If there are no further peers the sync will be restarted with an increased by
    /// `max_latency_increase`.
    #[serde(with = "serializers::seconds")]
    pub initial_max_sync_latency: Duration,
    /// If all sync peers exceed latency, increase allowed latency by this value
    #[serde(with = "serializers::seconds")]
    pub max_latency_increase: Duration,
    /// Longer ban period for potentially malicious infractions (protocol violations etc.)
    #[serde(with = "serializers::seconds")]
    pub ban_period: Duration,
    /// Short ban period for infractions that are likely not malicious (slow to respond, spotty connections etc)
    #[serde(with = "serializers::seconds")]
    pub short_ban_period: Duration,
    /// An allowlist of sync peers from which to sync. No other peers will be selected for sync. If empty, sync peers
    /// are chosen based on their advertised chain metadata.
    pub forced_sync_peers: Vec<NodeId>,
    /// Number of threads to use for validation
    pub validation_concurrency: usize,
    /// The RPC deadline to set on sync clients. If this deadline is reached, a new sync peer will be selected for
    /// sync.
    #[serde(with = "serializers::seconds")]
    pub rpc_deadline: Duration,
    /// The number of initial rounds of seed peer based bootstrapping.
    #[serde(default = "default_num_initial_sync_rounds_seed_bootstrap")]
    pub num_initial_sync_rounds_seed_bootstrap: usize,
}

fn default_num_initial_sync_rounds_seed_bootstrap() -> usize {
    // This should ideally match or be related to DhtConfig.network_discovery.max_seed_peer_sync_count
    // For now, a sensible default. This will be overridden by DhtEvent if DhtNetworkDiscoveryRoundInfo provides
    // total_rounds.
    5
}

impl Default for BlockchainSyncConfig {
    fn default() -> Self {
        Self {
            initial_max_sync_latency: Duration::from_secs(240), // Syncing many full blocks over tor require this
            max_latency_increase: Duration::from_secs(10),      // Syncing many full blocks over tor require this
            ban_period: Duration::from_secs(60 * 60 * 2),       // 2 hours
            short_ban_period: Duration::from_secs(240),         // 4 mins
            forced_sync_peers: Default::default(),
            validation_concurrency: 6,
            rpc_deadline: Duration::from_secs(240), // Syncing many full blocks over tor require this
            num_initial_sync_rounds_seed_bootstrap: default_num_initial_sync_rounds_seed_bootstrap(),
        }
    }
}

impl BlockchainSyncConfig {
    // Add an accessor if one doesn't exist for the new field
    pub fn num_initial_sync_rounds_seed_bootstrap(&self) -> usize {
        self.num_initial_sync_rounds_seed_bootstrap
    }
}

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::time::Duration;
24
25	use serde::{Deserialize, Serialize};
26	use tari_common::configuration::serializers;
27	use tari_comms::peer_manager::NodeId;
UNCOV 28	#[derive(Debug, Clone, Serialize, Deserialize)]	×
29	#[serde(deny_unknown_fields)]
30	pub struct BlockchainSyncConfig {
31	/// The initial max sync latency. If a peer fails to stream a header/block within this deadline another sync peer
32	/// will be selected. If there are no further peers the sync will be restarted with an increased by
33	/// `max_latency_increase`.
34	#[serde(with = "serializers::seconds")]
35	pub initial_max_sync_latency: Duration,
36	/// If all sync peers exceed latency, increase allowed latency by this value
37	#[serde(with = "serializers::seconds")]
38	pub max_latency_increase: Duration,
39	/// Longer ban period for potentially malicious infractions (protocol violations etc.)
40	#[serde(with = "serializers::seconds")]
41	pub ban_period: Duration,
42	/// Short ban period for infractions that are likely not malicious (slow to respond, spotty connections etc)
43	#[serde(with = "serializers::seconds")]
44	pub short_ban_period: Duration,
45	/// An allowlist of sync peers from which to sync. No other peers will be selected for sync. If empty, sync peers
46	/// are chosen based on their advertised chain metadata.
47	pub forced_sync_peers: Vec<NodeId>,
48	/// Number of threads to use for validation
49	pub validation_concurrency: usize,
50	/// The RPC deadline to set on sync clients. If this deadline is reached, a new sync peer will be selected for
51	/// sync.
52	#[serde(with = "serializers::seconds")]
53	pub rpc_deadline: Duration,
54	/// The number of initial rounds of seed peer based bootstrapping.
55	#[serde(default = "default_num_initial_sync_rounds_seed_bootstrap")]
56	pub num_initial_sync_rounds_seed_bootstrap: usize,
57	}
58
59	fn default_num_initial_sync_rounds_seed_bootstrap() -> usize {	73✔
60	// This should ideally match or be related to DhtConfig.network_discovery.max_seed_peer_sync_count	73✔
61	// For now, a sensible default. This will be overridden by DhtEvent if DhtNetworkDiscoveryRoundInfo provides	73✔
62	// total_rounds.	73✔
63	5	73✔
64	}	73✔
65
66	impl Default for BlockchainSyncConfig {
67	fn default() -> Self {	73✔
68	Self {	73✔
69	initial_max_sync_latency: Duration::from_secs(240), // Syncing many full blocks over tor require this	73✔
70	max_latency_increase: Duration::from_secs(10), // Syncing many full blocks over tor require this	73✔
71	ban_period: Duration::from_secs(60 * 60 * 2), // 2 hours	73✔
72	short_ban_period: Duration::from_secs(240), // 4 mins	73✔
73	forced_sync_peers: Default::default(),	73✔
74	validation_concurrency: 6,	73✔
75	rpc_deadline: Duration::from_secs(240), // Syncing many full blocks over tor require this	73✔
76	num_initial_sync_rounds_seed_bootstrap: default_num_initial_sync_rounds_seed_bootstrap(),	73✔
77	}	73✔
78	}	73✔
79	}
80
81	impl BlockchainSyncConfig {
82	// Add an accessor if one doesn't exist for the new field
UNCOV 83	pub fn num_initial_sync_rounds_seed_bootstrap(&self) -> usize {	×
UNCOV 84	self.num_initial_sync_rounds_seed_bootstrap	×
UNCOV 85	}	×
86	}

tari-project / tari / 15326284040

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