Ecosystem/serai
Ecosystem/serai
1
0
Fork 0
mirror of https://github.com/serai-dex/serai.git synced 2025-01-26 12:36:12 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Spawn a task for Heartbeat responses, preventing it from holding up P2P handling

Browse source
This commit is contained in:
Luke Parker 2023-09-27 00:10:37 -04:00
parent 086458d041
commit db8dc1e864
No known key found for this signature in database
1 changed files with 49 additions and 51 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
coordinator/src/main.rs
View file

@ -383,7 +383,6 @@ pub async fn handle_p2p<D: Db, P: P2p>(
// TODO2: Rate limit this per timestamp // TODO2: Rate limit this per timestamp
// And/or slash on Heartbeat which justifies a response, since the node obviously // And/or slash on Heartbeat which justifies a response, since the node obviously
// was offline and we must now use our bandwidth to compensate for them? // was offline and we must now use our bandwidth to compensate for them?
// TODO: Dedicated task for heartbeats
P2pMessageKind::Heartbeat(msg_genesis) => { P2pMessageKind::Heartbeat(msg_genesis) => {
assert_eq!(msg_genesis, genesis); assert_eq!(msg_genesis, genesis);
if msg.msg.len() != 40 { if msg.msg.len() != 40 {
@ -391,8 +390,12 @@ pub async fn handle_p2p<D: Db, P: P2p>(
continue; continue;
} }
let tributary_read = &tributary.tributary; let p2p = p2p.clone();
let spec = tributary.spec.clone();
let reader = tributary.tributary.reader();
// Spawn a dedicated task as this may require loading large amounts of data from
// disk and take a notable amount of time
tokio::spawn(async move {
/* /*
// Have sqrt(n) nodes reply with the blocks // Have sqrt(n) nodes reply with the blocks
let mut responders = (tributary.spec.n() as f32).sqrt().floor() as u64; let mut responders = (tributary.spec.n() as f32).sqrt().floor() as u64;
@ -403,22 +406,20 @@ pub async fn handle_p2p<D: Db, P: P2p>(
*/ */
// Have up to three nodes respond // Have up to three nodes respond
let responders = u64::from(tributary.spec.n().min(3)); let responders = u64::from(spec.n().min(3));
// Decide which nodes will respond by using the latest block's hash as a mutually // Decide which nodes will respond by using the latest block's hash as a
// agreed upon entropy source // mutually agreed upon entropy source
// This isn't a secure source of entropy, yet it's fine for this // This isn't a secure source of entropy, yet it's fine for this
let entropy = let entropy = u64::from_le_bytes(reader.tip()[.. 8].try_into().unwrap());
u64::from_le_bytes(tributary_read.tip().await[.. 8].try_into().unwrap());
// If n = 10, responders = 3, we want `start` to be 0 ..= 7 // If n = 10, responders = 3, we want `start` to be 0 ..= 7
// (so the highest is 7, 8, 9) // (so the highest is 7, 8, 9)
// entropy % (10 + 1) - 3 = entropy % 8 = 0 ..= 7 // entropy % (10 + 1) - 3 = entropy % 8 = 0 ..= 7
let start = let start =
usize::try_from(entropy % (u64::from(tributary.spec.n() + 1) - responders)) usize::try_from(entropy % (u64::from(spec.n() + 1) - responders)).unwrap();
.unwrap();
let mut selected = false; let mut selected = false;
for validator in &tributary.spec.validators() for validator in
[start .. (start + usize::try_from(responders).unwrap())] &spec.validators()[start .. (start + usize::try_from(responders).unwrap())]
{ {
if our_key == validator.0 { if our_key == validator.0 {
selected = true; selected = true;
@ -427,24 +428,21 @@ pub async fn handle_p2p<D: Db, P: P2p>(
} }
if !selected { if !selected {
log::debug!("received heartbeat and not selected to respond"); log::debug!("received heartbeat and not selected to respond");
continue; return;
} }
log::debug!("received heartbeat and selected to respond"); log::debug!("received heartbeat and selected to respond");
let reader = tributary_read.reader();
let mut latest = msg.msg[.. 32].try_into().unwrap(); let mut latest = msg.msg[.. 32].try_into().unwrap();
while let Some(next) = reader.block_after(&latest) { while let Some(next) = reader.block_after(&latest) {
let mut res = reader.block(&next).unwrap().serialize(); let mut res = reader.block(&next).unwrap().serialize();
res.extend(reader.commit(&next).unwrap()); res.extend(reader.commit(&next).unwrap());
// Also include the timestamp used within the Heartbeat // Also include the timestamp used within the Heartbeat
res.extend(&msg.msg[32 .. 40]); res.extend(&msg.msg[32 .. 40]);
p2p p2p.send(msg.sender, P2pMessageKind::Block(spec.genesis()), res).await;
.send(msg.sender, P2pMessageKind::Block(tributary.spec.genesis()), res)
.await;
latest = next; latest = next;
} }
});
} }
P2pMessageKind::Block(msg_genesis) => { P2pMessageKind::Block(msg_genesis) => {