15484013348

Committed 06 Jun 2025 06:08AM UTC coverage: 72.04% (+0.3%) from 71.789%

Build # 15484013348

Build Type

push

github

Committed by

web-flow

Commit Message

fix(network-discovery): add back idle event handling (#7194)

Description
---
fix(network-discovery): add back idle event handling

Motivation and Context
---
network discovery was spinning at full speed because the Idle event
transition was removed. Network logs would rotate < 1s.

```
[comms::dht::network_discovery::ready] [Thread:123190302967360] DEBUG NetworkDiscovery::Ready: Peer list contains 759 entries. Current discovery rounds in this cycle: 0.
[comms::dht::network_discovery::ready] [Thread:123190302967360] DEBUG First active round (current_num_rounds = 0) and num_peers (759) >= min_desired_peers (16). Forcing DHT discovery.
 [comms::dht::network_discovery::ready] [Thread:123190302967360] DEBUG Selecting 5 random peers for discovery (last round info available: false, new peers in last round: false).
[comms::dht::network_discovery::ready] [Thread:123190302967360] DEBUG No suitable peers found for the forced DHT discovery round (current_num_rounds = 0 path). Transitioning to Idle.
 [comms::dht::network_discovery] [Thread:123190302967360] DEBUG Transition triggered from current state `Ready` by event `Idle`
comms::dht::network_discovery] [Thread:123190302967360] DEBUG No state transition for event `Idle`. The current state is `Ready`

...instant rinse and repeat...
```

This PR adds the idle state transition back. Note that idle will idle
for 30 minutes so should only transition when all work is done and we
have downloaded sufficient peers.

How Has This Been Tested?
---
Manually - console wallet with empty peer db

What process can a PR reviewer use to test or verify this change?
---

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

Run Details

3 of 4 new or added lines in 2 files covered. (75.0%)

412 existing lines in 30 files now uncovered.

80882 of 112274 relevant lines covered (72.04%)

242938.65 hits per line

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

74.4

/base_layer/core/src/base_node/sync/sync_peer.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::{
    cmp::Ordering,
    fmt::{Display, Formatter},
    time::Duration,
};

use primitive_types::U512;
use tari_common_types::chain_metadata::ChainMetadata;
use tari_comms::peer_manager::NodeId;

use crate::{base_node::chain_metadata_service::PeerChainMetadata, common::rolling_avg::RollingAverageTime};

#[derive(Debug, Clone)]
pub struct SyncPeer {
    peer_metadata: PeerChainMetadata,
    avg_latency: RollingAverageTime,
}

impl SyncPeer {
    pub fn node_id(&self) -> &NodeId {
        self.peer_metadata.node_id()
    }

    pub fn claimed_chain_metadata(&self) -> &ChainMetadata {
        self.peer_metadata.claimed_chain_metadata()
    }

    pub fn claimed_difficulty(&self) -> U512 {
        self.peer_metadata.claimed_chain_metadata().accumulated_difficulty()
    }

    pub fn latency(&self) -> Option<Duration> {
        self.peer_metadata.latency()
    }

    pub(super) fn set_latency(&mut self, latency: Duration) -> &mut Self {
        self.peer_metadata.set_latency(latency);
        self
    }

    pub fn items_per_second(&self) -> Option<f64> {
        self.avg_latency.calc_samples_per_second()
    }

    pub(super) fn add_sample(&mut self, time: Duration) -> &mut Self {
        self.avg_latency.add_sample(time);
        self
    }

    pub fn calc_avg_latency(&self) -> Option<Duration> {
        self.avg_latency.calculate_average()
    }
}

impl From<PeerChainMetadata> for SyncPeer {
    fn from(peer_metadata: PeerChainMetadata) -> Self {
        Self {
            peer_metadata,
            avg_latency: RollingAverageTime::new(20),
        }
    }
}

impl Display for SyncPeer {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "Node ID: {}, Chain metadata: {}, Latency: {}",
            self.node_id(),
            self.claimed_chain_metadata(),
            self.latency()
                .map(|d| format!("{:.2?}", d))
                .unwrap_or_else(|| "--".to_string())
        )
    }
}

impl PartialEq for SyncPeer {
    fn eq(&self, other: &Self) -> bool {
        self.node_id() == other.node_id()
    }
}
impl Eq for SyncPeer {}

impl Ord for SyncPeer {
    fn cmp(&self, other: &Self) -> Ordering {
        let mut result = other
            .peer_metadata
            .claimed_chain_metadata()
            .accumulated_difficulty()
            .cmp(&self.peer_metadata.claimed_chain_metadata().accumulated_difficulty());
        if result == Ordering::Equal {
            match (self.latency(), other.latency()) {
                (None, None) => result = Ordering::Equal,
                // No latency goes to the end
                (Some(_), None) => result = Ordering::Less,
                (None, Some(_)) => result = Ordering::Greater,
                (Some(la), Some(lb)) => result = la.cmp(&lb),
            }
        }
        result
    }
}

impl PartialOrd for SyncPeer {
    fn partial_cmp(&self, other: &SyncPeer) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

#[cfg(test)]
mod test {
    use std::time::Duration;

    use rand::rngs::OsRng;
    use tari_common_types::chain_metadata::ChainMetadata;

    use super::*;

    mod sort_by_latency {
        use tari_common_types::types::FixedHash;
        use tari_comms::types::{CommsPublicKey, CommsSecretKey};
        use tari_crypto::keys::SecretKey;

        use super::*;

        // Helper function to generate a peer with a given latency
        fn generate_peer(latency: Option<usize>, accumulated_difficulty: Option<U512>) -> SyncPeer {
            let sk = CommsSecretKey::random(&mut OsRng);
            let pk = CommsPublicKey::from_secret_key(&sk);
            let node_id = NodeId::from_key(&pk);
            let latency_option = latency.map(|latency| Duration::from_millis(latency as u64));
            let peer_accumulated_difficulty = match accumulated_difficulty {
                Some(v) => v,
                None => 1.into(),
            };
            PeerChainMetadata::new(
                node_id,
                ChainMetadata::new(0, FixedHash::zero(), 0, 0, peer_accumulated_difficulty, 0).unwrap(),
                latency_option,
            )
            .into()
        }

        #[test]
        fn it_sorts_by_latency() {
            const DISTINCT_LATENCY: usize = 5;

            // Generate a list of peers with latency, adding duplicates
            let mut peers = (0..2 * DISTINCT_LATENCY)
                .map(|latency| generate_peer(Some(latency % DISTINCT_LATENCY), None))
                .collect::<Vec<SyncPeer>>();

            // Add peers with no latency in a few places
            peers.insert(0, generate_peer(None, None));
            peers.insert(DISTINCT_LATENCY, generate_peer(None, None));
            peers.push(generate_peer(None, None));

            // Sort the list; because difficulty is identical, it should sort by latency
            peers.sort();

            // Confirm that the sorted latency is correct: numerical ordering, then `None`
            for (i, peer) in peers[..2 * DISTINCT_LATENCY].iter().enumerate() {
                assert_eq!(peer.latency(), Some(Duration::from_millis((i as u64) / 2)));
            }
            for _ in 0..3 {
                assert_eq!(peers.pop().unwrap().latency(), None);
            }
        }

        #[test]
        fn it_sorts_by_pow() {
            let mut peers = Vec::new();

            let mut pow = U512::from(1);
            let new_peer = generate_peer(Some(1), Some(pow));
            peers.push(new_peer);
            pow = U512::from(100);
            let new_peer = generate_peer(Some(100), Some(pow));
            peers.push(new_peer);
            pow = U512::from(1000);
            let new_peer = generate_peer(Some(1000), Some(pow));
            peers.push(new_peer);

            // Sort the list;
            peers.sort();

            assert_eq!(
                peers[0].peer_metadata.claimed_chain_metadata().accumulated_difficulty(),
                1000.into()
            );
            assert_eq!(
                peers[1].peer_metadata.claimed_chain_metadata().accumulated_difficulty(),
                100.into()
            );
            assert_eq!(
                peers[2].peer_metadata.claimed_chain_metadata().accumulated_difficulty(),
                1.into()
            );
        }
    }
}

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	cmp::Ordering,
25	fmt::{Display, Formatter},
26	time::Duration,
27	};
28
29	use primitive_types::U512;
30	use tari_common_types::chain_metadata::ChainMetadata;
31	use tari_comms::peer_manager::NodeId;
32
33	use crate::{base_node::chain_metadata_service::PeerChainMetadata, common::rolling_avg::RollingAverageTime};
34
35	#[derive(Debug, Clone)]
36	pub struct SyncPeer {
37	peer_metadata: PeerChainMetadata,
38	avg_latency: RollingAverageTime,
39	}
40
41	impl SyncPeer {
42	pub fn node_id(&self) -> &NodeId {	30✔
43	self.peer_metadata.node_id()	30✔
44	}	30✔
45
46	pub fn claimed_chain_metadata(&self) -> &ChainMetadata {	12✔
47	self.peer_metadata.claimed_chain_metadata()	12✔
48	}	12✔
49
50	pub fn claimed_difficulty(&self) -> U512 {	×
51	self.peer_metadata.claimed_chain_metadata().accumulated_difficulty()	×
52	}	×
53
54	pub fn latency(&self) -> Option<Duration> {	87✔
55	self.peer_metadata.latency()	87✔
56	}	87✔
57
UNCOV 58	pub(super) fn set_latency(&mut self, latency: Duration) -> &mut Self {	×
UNCOV 59	self.peer_metadata.set_latency(latency);	×
UNCOV 60	self	×
UNCOV 61	}	×
62
63	pub fn items_per_second(&self) -> Option<f64> {	×
64	self.avg_latency.calc_samples_per_second()	×
65	}	×
66
67	pub(super) fn add_sample(&mut self, time: Duration) -> &mut Self {	×
68	self.avg_latency.add_sample(time);	×
69	self	×
70	}	×
71
72	pub fn calc_avg_latency(&self) -> Option<Duration> {	×
73	self.avg_latency.calculate_average()	×
74	}	×
75	}
76
77	impl From<PeerChainMetadata> for SyncPeer {
78	fn from(peer_metadata: PeerChainMetadata) -> Self {	28✔
79	Self {	28✔
80	peer_metadata,	28✔
81	avg_latency: RollingAverageTime::new(20),	28✔
82	}	28✔
83	}	28✔
84	}
85
86	impl Display for SyncPeer {
87	fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {	×
88	write!(	×
89	f,	×
90	"Node ID: {}, Chain metadata: {}, Latency: {}",	×
91	self.node_id(),	×
92	self.claimed_chain_metadata(),	×
93	self.latency()	×
94	.map(\|d\| format!("{:.2?}", d))	×
95	.unwrap_or_else(\|\| "--".to_string())	×
96	)	×
97	}	×
98	}
99
100	impl PartialEq for SyncPeer {
UNCOV 101	fn eq(&self, other: &Self) -> bool {	×
UNCOV 102	self.node_id() == other.node_id()	×
UNCOV 103	}	×
104	}
105	impl Eq for SyncPeer {}
106
107	impl Ord for SyncPeer {
108	fn cmp(&self, other: &Self) -> Ordering {	40✔
109	let mut result = other	40✔
110	.peer_metadata	40✔
111	.claimed_chain_metadata()	40✔
112	.accumulated_difficulty()	40✔
113	.cmp(&self.peer_metadata.claimed_chain_metadata().accumulated_difficulty());	40✔
114	if result == Ordering::Equal {	40✔
115	match (self.latency(), other.latency()) {	37✔
116	(None, None) => result = Ordering::Equal,	2✔
117	// No latency goes to the end
118	(Some(_), None) => result = Ordering::Less,	16✔
119	(None, Some(_)) => result = Ordering::Greater,	×
120	(Some(la), Some(lb)) => result = la.cmp(&lb),	19✔
121	}
122	}	3✔
123	result	40✔
124	}	40✔
125	}
126
127	impl PartialOrd for SyncPeer {
128	fn partial_cmp(&self, other: &SyncPeer) -> Option<Ordering> {	40✔
129	Some(self.cmp(other))	40✔
130	}	40✔
131	}
132
133	#[cfg(test)]
134	mod test {
135	use std::time::Duration;
136
137	use rand::rngs::OsRng;
138	use tari_common_types::chain_metadata::ChainMetadata;
139
140	use super::*;
141
142	mod sort_by_latency {
143	use tari_common_types::types::FixedHash;
144	use tari_comms::types::{CommsPublicKey, CommsSecretKey};
145	use tari_crypto::keys::SecretKey;
146
147	use super::*;
148
149	// Helper function to generate a peer with a given latency
150	fn generate_peer(latency: Option<usize>, accumulated_difficulty: Option<U512>) -> SyncPeer {	16✔
151	let sk = CommsSecretKey::random(&mut OsRng);	16✔
152	let pk = CommsPublicKey::from_secret_key(&sk);	16✔
153	let node_id = NodeId::from_key(&pk);	16✔
154	let latency_option = latency.map(\|latency\| Duration::from_millis(latency as u64));	16✔
155	let peer_accumulated_difficulty = match accumulated_difficulty {	16✔
156	Some(v) => v,	3✔
157	None => 1.into(),	13✔
158	};
159	PeerChainMetadata::new(	16✔
160	node_id,	16✔
161	ChainMetadata::new(0, FixedHash::zero(), 0, 0, peer_accumulated_difficulty, 0).unwrap(),	16✔
162	latency_option,	16✔
163	)	16✔
164	.into()	16✔
165	}	16✔
166
167	#[test]
168	fn it_sorts_by_latency() {	1✔
169	const DISTINCT_LATENCY: usize = 5;
170
171	// Generate a list of peers with latency, adding duplicates
172	let mut peers = (0..2 * DISTINCT_LATENCY)	1✔
173	.map(\|latency\| generate_peer(Some(latency % DISTINCT_LATENCY), None))	10✔
174	.collect::<Vec<SyncPeer>>();	1✔
175		1✔
176	// Add peers with no latency in a few places	1✔
177	peers.insert(0, generate_peer(None, None));	1✔
178	peers.insert(DISTINCT_LATENCY, generate_peer(None, None));	1✔
179	peers.push(generate_peer(None, None));	1✔
180		1✔
181	// Sort the list; because difficulty is identical, it should sort by latency	1✔
182	peers.sort();	1✔
183
184	// Confirm that the sorted latency is correct: numerical ordering, then `None`
185	for (i, peer) in peers[..2 * DISTINCT_LATENCY].iter().enumerate() {	10✔
186	assert_eq!(peer.latency(), Some(Duration::from_millis((i as u64) / 2)));	10✔
187	}
188	for _ in 0..3 {	4✔
189	assert_eq!(peers.pop().unwrap().latency(), None);	3✔
190	}
191	}	1✔
192
193	#[test]
194	fn it_sorts_by_pow() {	1✔
195	let mut peers = Vec::new();	1✔
196		1✔
197	let mut pow = U512::from(1);	1✔
198	let new_peer = generate_peer(Some(1), Some(pow));	1✔
199	peers.push(new_peer);	1✔
200	pow = U512::from(100);	1✔
201	let new_peer = generate_peer(Some(100), Some(pow));	1✔
202	peers.push(new_peer);	1✔
203	pow = U512::from(1000);	1✔
204	let new_peer = generate_peer(Some(1000), Some(pow));	1✔
205	peers.push(new_peer);	1✔
206		1✔
207	// Sort the list;	1✔
208	peers.sort();	1✔
209		1✔
210	assert_eq!(	1✔
211	peers[0].peer_metadata.claimed_chain_metadata().accumulated_difficulty(),	1✔
212	1000.into()	1✔
213	);	1✔
214	assert_eq!(	1✔
215	peers[1].peer_metadata.claimed_chain_metadata().accumulated_difficulty(),	1✔
216	100.into()	1✔
217	);	1✔
218	assert_eq!(	1✔
219	peers[2].peer_metadata.claimed_chain_metadata().accumulated_difficulty(),	1✔
220	1.into()	1✔
221	);	1✔
222	}	1✔
223	}
224	}

tari-project / tari / 15484013348

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