absolutebi/serai
1
0
Fork 0
mirror of https://github.com/serai-dex/serai.git synced 2025-12-08 12:19:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

Parallelize requests within Validators::update

Browse Source
This commit is contained in:
Luke Parker
2025-01-05 00:17:05 -05:00
parent 479ca0410a
commit 2b8f481364
3 changed files with 60 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

86
coordinator/src/p2p/validators.rs
View File

@@ -4,6 +4,8 @@ use serai_client::{primitives::NetworkId, validator_sets::primitives::Session, S
use libp2p::PeerId;
use futures_util::stream::{StreamExt, FuturesUnordered};
use crate::p2p::peer_id_from_public;
pub(crate) struct Validators {
@@ -27,41 +29,63 @@ impl Validators {
let temporal_serai =
self.serai.as_of_latest_finalized_block().await.map_err(|e| format!("{e:?}"))?;
let temporal_serai = temporal_serai.validator_sets();
for network in serai_client::primitives::NETWORKS {
if network == NetworkId::Serai {
continue;
}
let Some(session) = temporal_serai.session(network).await.map_err(|e| format!("{e:?}"))?
else {
continue;
};
// If the session has changed, populate it with the current validators
if self.sessions.get(&network) != Some(&session) {
let new_validators =
temporal_serai.active_network_validators(network).await.map_err(|e| format!("{e:?}"))?;
let new_validators =
new_validators.into_iter().map(peer_id_from_public).collect::<HashSet<_>>();
// Remove the existing validators
for validator in self.by_network.remove(&network).unwrap_or_else(HashSet::new) {
let mut networks = self.validators.remove(&validator).unwrap();
networks.remove(&network);
if networks.is_empty() {
removed.insert(validator);
let mut session_changes = vec![];
{
// FuturesUnordered can be bad practice as it'll cause timeouts if infrequently polled, but
// we poll it till it yields all futures with the most minimal processing possible
let mut futures = FuturesUnordered::new();
for network in serai_client::primitives::NETWORKS {
if network == NetworkId::Serai {
continue;
}
let sessions = &self.sessions;
futures.push(async move {
let session = match temporal_serai.session(network).await {
Ok(Some(session)) => session,
Ok(None) => return Ok(None),
Err(e) => return Err(format!("{e:?}")),
};
if sessions.get(&network) == Some(&session) {
Ok(None)
} else {
self.validators.insert(validator, networks);
match temporal_serai.active_network_validators(network).await {
Ok(validators) => Ok(Some((network, session, validators))),
Err(e) => Err(format!("{e:?}")),
}
}
}
// Add the new validators
for validator in new_validators.iter().copied() {
self.validators.entry(validator).or_insert_with(HashSet::new).insert(network);
}
self.by_network.insert(network, new_validators);
// Update the session we have populated
self.sessions.insert(network, session);
});
}
while let Some(session_change) = futures.next().await {
if let Some(session_change) = session_change? {
session_changes.push(session_change);
}
}
}
for (network, session, validators) in session_changes {
let validators = validators.into_iter().map(peer_id_from_public).collect::<HashSet<_>>();
// Remove the existing validators
for validator in self.by_network.remove(&network).unwrap_or_else(HashSet::new) {
let mut networks = self.validators.remove(&validator).unwrap();
networks.remove(&network);
if networks.is_empty() {
removed.insert(validator);
} else {
self.validators.insert(validator, networks);
}
}
// Add the new validators
for validator in validators.iter().copied() {
self.validators.entry(validator).or_insert_with(HashSet::new).insert(network);
}
self.by_network.insert(network, validators);
// Update the session we have populated
self.sessions.insert(network, session);
}
Ok(removed)