diff options
Diffstat (limited to 'src/swarm/peer.rs')
| -rw-r--r-- | src/swarm/peer.rs | 403 |
1 files changed, 235 insertions, 168 deletions
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index b6c9700..dd70bbb 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -2,24 +2,22 @@ //! swarm-wide forget records, and dispatch of inbound messages. use std::sync::Arc; +use std::sync::atomic::Ordering; use std::time::Duration; use anyhow::{Context, Result, anyhow, bail}; use automerge::AutoCommit; -use sha2::{Digest, Sha256}; use tokio::sync::Mutex; use tokio::time::{Instant, MissedTickBehavior}; use tracing::{debug, info, instrument, trace, warn}; use crate::document::replica::{Epoch, EpochRelation}; use crate::document::store::{self, StoreError}; -use crate::state::{AppState, HelloState, Liveness, PeerMembership, PeerState}; +use crate::state::{AppState, HelloState, Liveness, PeerState, Standing}; use crate::swarm::sync::{SyncEvent, handle_sync}; +use crate::swarm::topology; use crate::transport::{self, ParsedMessage}; -const HELLO_RETRY_INITIAL: Duration = Duration::from_secs(1); -const HELLO_RETRY_MAX: Duration = Duration::from_secs(30); - const PEER_TICK: Duration = Duration::from_millis(500); const META_ROOT: &str = "_continuum"; @@ -98,24 +96,63 @@ async fn apply_forget_records(state: &AppState) { // --- keepalive and self-healing ---------------------------------------------- -fn heads_digest(doc: &mut AutoCommit) -> transport::HeadsDigest { - let mut heads = doc.get_heads(); - heads.sort_unstable(); - let mut hasher = Sha256::new(); - for head in &heads { - hasher.update(head.0); - } - hasher.finalize().into() +/// Sends `peer` a PING with our keepalive, noting the time for its round trip. +pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> { + let keepalive = local_keepalive(state, peer).await; + state.lock_rtt().ping_sent(peer, Instant::now()); + state + .transport + .send_message(peer, &ParsedMessage::Ping(keepalive)) + .await } -pub(crate) async fn local_keepalive(state: &AppState) -> transport::Keepalive { +/// Our keepalive for `peer`. +pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::Keepalive { // AutoCommit::get_heads takes &mut self (it closes the pending transaction). - // Digest and epoch come from the same locked replica, so they always match. - let mut replica = state.doc.write().await; + // Heads and epoch come from the same locked replica, so they always match. + let (heads, epoch) = { + let mut replica = state.doc.write().await; + ( + transport::HeadsSummary::of(replica.doc.get_heads()), + replica.epoch, + ) + }; + let syncs_with_you = match state.get_peer_state(peer).await { + Some(peer_state) => { + let peer_state = peer_state.lock().await; + peer_state.liveness == Liveness::Active + && matches!(peer_state.standing, Standing::Member { .. }) + && peer_state.epoch == Some(epoch) + } + None => false, + }; transport::Keepalive { - heads_digest: heads_digest(&mut replica.doc), - epoch: replica.epoch, + heads, + epoch, + ready: state.is_ready().await, + syncs_with_you, + } +} + +/// What a peer's keepalive says about it. +async fn note_keepalive(state: &AppState, peer: &str, remote: &transport::Keepalive) { + if let Some(peer_state) = state.get_peer_state(peer).await { + let mut peer_state = peer_state.lock().await; + let became_ready = remote.ready && peer_state.reports_ready != Some(true); + peer_state.reports_ready = Some(remote.ready); + peer_state.syncs_with_us = Some(remote.syncs_with_you); + peer_state.reported_heads = remote.heads.heads().map(<[_]>::len); + drop(peer_state); + crate::swarm::relay::pull_if_behind(state, peer, remote).await; + if became_ready { + // We may have left the opening to it while it wasn't ready, and nothing else + // would make us look again. + let _ = state + .sync_events + .send(SyncEvent::PeerNeedsSync(peer.to_owned())); + } } + crate::swarm::source::reported(state, peer, remote.ready, Some(remote.syncs_with_you)); } async fn heal_if_diverged( @@ -124,8 +161,9 @@ async fn heal_if_diverged( local: &transport::Keepalive, remote: &transport::Keepalive, ) { - // Digests of documents from different epochs are never equal and never converge. - if local.epoch != remote.epoch { + // Heads of documents from different epochs are never equal and never converge. + // And a peer that isn't our neighbour isn't ours to keep in step. + if local.epoch != remote.epoch || !topology::is_neighbour(state, peer) { return; } @@ -135,18 +173,17 @@ async fn heal_if_diverged( let heal = { let mut peer_state = peer_state_mutex.lock().await; - if local.heads_digest == remote.heads_digest { + if local.heads == remote.heads { // Same document: whatever answer we were waiting for, there is nothing left // to send. (Automerge answers only when it has something to say, so a sync // message to a peer that already has everything gets no reply at all.) - peer_state.sync_in_step(); + peer_state.sync.reply_settled(); } - peer_state.needs_heal( - &local.heads_digest, - &remote.heads_digest, - &state.timing, - Instant::now(), - ) + // While we catch up, we hold back sync with everyone but the source on purpose: + // differing heads and idle sync are expected then, not a reason to heal. + let catching_up = !state.initial_sync_done.load(Ordering::Acquire); + !catching_up + && peer_state.needs_heal(&local.heads, &remote.heads, &state.timing, Instant::now()) }; if heal { @@ -171,7 +208,7 @@ pub(crate) async fn remember_peer(state: &AppState, peer: &str) { let mut peer_states = state.peer_states.write().await; if !peer_states.contains_key(peer) { - peer_states.entry(peer.to_owned()).or_default(); + peer_states.insert(peer.to_owned(), state.new_peer_state(peer)); info!( peer = %peer, peer_count = peer_states.len(), @@ -191,9 +228,10 @@ pub(crate) async fn send_hello_once(state: &AppState, peer: &str) -> Result<()> let claimed = { let mut peer_state = peer_state_mutex.lock().await; if peer_state.hello_state == HelloState::NotSent { - peer_state.hello_state = HelloState::Sending { - next_delay: HELLO_RETRY_INITIAL, - }; + // A new handshake: retries start over. + peer_state.hello_state = HelloState::Sending; + peer_state.hello_backoff.reset(); + peer_state.hello_attempts = 0; true } else { trace!(peer = %peer, hello_state = ?peer_state.hello_state, "not sending"); @@ -206,7 +244,7 @@ pub(crate) async fn send_hello_once(state: &AppState, peer: &str) -> Result<()> } debug!(to = %peer, "Sending HELLO"); - transmit_hello(state, peer, HELLO_RETRY_INITIAL).await + transmit_hello(state, peer).await } async fn retry_hello_if_due(state: &AppState, peer: &str) -> Result<()> { @@ -214,43 +252,41 @@ async fn retry_hello_if_due(state: &AppState, peer: &str) -> Result<()> { return Ok(()); }; - let delay = { + { let mut peer_state = peer_state_mutex.lock().await; match peer_state.hello_state { - HelloState::AwaitingReply { - next_attempt, - next_delay, - } if Instant::now() >= next_attempt => { - peer_state.hello_state = HelloState::Sending { next_delay }; - next_delay + HelloState::AwaitingReply { next_attempt } if Instant::now() >= next_attempt => { + peer_state.hello_state = HelloState::Sending; } _ => return Ok(()), } - }; + } - debug!( - peer = %peer, - next_retry_ms = delay.as_millis(), - "No HELLO_REPLY yet; re-sending HELLO" - ); - transmit_hello(state, peer, delay).await + debug!(peer = %peer, "No HELLO_REPLY yet; re-sending HELLO"); + transmit_hello(state, peer).await } /// Sends a HELLO without holding any lock, then schedules the next retry. /// Success and failure both end in AwaitingReply: Mycelium accepting the message /// says nothing about whether the peer received it. -async fn transmit_hello(state: &AppState, peer: &str, delay: Duration) -> Result<()> { +async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> { + let swarm_id = state.swarm_id().await; + let auth = state + .swarm_key + .as_ref() + .map(|key| key.hello_tag(swarm_id.as_deref().unwrap_or(""), state.local_id(), peer)) + .unwrap_or_default(); + state.lock_rtt().hello_sent(peer, Instant::now()); let result = state .transport .send_message( peer, &ParsedMessage::Hello(transport::HelloRequest { + swarm_id, epoch: state.local_epoch(), - auth: state - .swarm_key - .as_ref() - .map(|key| key.hello_tag(state.local_id(), peer)) - .unwrap_or_default(), + auth, + ready: state.is_ready().await, + run_id: state.run_id, }), ) .await; @@ -258,16 +294,14 @@ async fn transmit_hello(state: &AppState, peer: &str, delay: Duration) -> Result if let Some(peer_state_mutex) = state.get_peer_state(peer).await { let mut peer_state = peer_state_mutex.lock().await; - if matches!(peer_state.hello_state, HelloState::Sending { .. }) { - let (wait, next_delay) = match peer_state.liveness { - Liveness::Dormant { .. } => { - (state.timing.dormant_probe, state.timing.dormant_probe) - } - Liveness::Active => (delay, delay.saturating_mul(2).min(HELLO_RETRY_MAX)), + if peer_state.hello_state == HelloState::Sending { + peer_state.hello_sent(Instant::now()); + let wait = match peer_state.liveness { + Liveness::Dormant { .. } => state.timing.dormant_probe, + Liveness::Active => peer_state.hello_backoff.next(), }; peer_state.hello_state = HelloState::AwaitingReply { next_attempt: Instant::now() + wait, - next_delay, }; } } @@ -281,36 +315,14 @@ async fn ensure_handshake(state: &AppState, peer: &str) { } } -/// Sends a HELLO_REPLY if one is owed and our swarm ID is known. -/// Otherwise it stays owed and the peer worker tries again. -async fn send_owed_hello_reply(state: &AppState, peer: &str) { - if state.swarm_id().await.is_none() { - return; - } - - let Some(peer_state_mutex) = state.get_peer_state(peer).await else { - return; - }; - - let owed = std::mem::take(&mut peer_state_mutex.lock().await.reply_owed); - if !owed { - return; - } - - if let Err(error) = send_hello_reply(state, peer).await { - warn!(peer = %peer, error = %format!("{error:#}"), "HELLO_REPLY failed; will retry"); - peer_state_mutex.lock().await.reply_owed = true; - } -} - -async fn flush_owed_hello_replies(state: &AppState) { - let peers: Vec<String> = state.peer_states.read().await.keys().cloned().collect(); - for peer in &peers { - send_owed_hello_reply(state, peer).await; - } -} - -async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> { +/// Answers a HELLO that carried `run_id`; `away` is how long before it we had last heard +/// from the peer. +async fn send_hello_reply( + state: &AppState, + peer: &str, + run_id: transport::RunId, + away: Option<Duration>, +) -> Result<()> { let swarm_id = state .swarm_id() .await @@ -330,6 +342,7 @@ async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> { let peer_state = peer_state_mutex.lock().await; if peer_state.liveness == Liveness::Active && peer_state.hello_state == HelloState::Established + && topology::is_neighbour(state, &known) { peers.push(known); } @@ -340,15 +353,17 @@ async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> { .as_ref() .map(|key| key.hello_reply_tag(&swarm_id, state.local_id(), peer)) .unwrap_or_default(); + let ready = state.is_ready().await; let reply = transport::HelloReply { swarm_id, peers, epoch: state.local_epoch(), auth, + ready, + run_id, + away_ms: away.map(|away| away.as_millis() as u64), }; - let ready = state.is_ready().await; - debug!(to = %peer, ready, "Sending HELLO_REPLY"); state @@ -370,6 +385,9 @@ async fn discover_peers(state: &AppState, from_peer: &str, advertised_peers: &[S remember_peer(state, discovered_peer).await; + if !topology::is_neighbour(state, discovered_peer) { + continue; + } if let Err(error) = send_hello_once(state, discovered_peer).await { warn!( peer = %discovered_peer, @@ -380,6 +398,26 @@ async fn discover_peers(state: &AppState, from_peer: &str, advertised_peers: &[S } } +/// New neighbours get our HELLO (we may not have known them before: the member list is +/// how a partial mesh discovers them). Former neighbours get no more HELLOs, keepalives +/// or sync openings from us; their session is reset, so we don't keep pushing changes +/// to them either. They are still answered if they contact us. +async fn apply_neighbour_change(state: &AppState, change: topology::NeighbourChange) { + for peer in &change.added { + remember_peer(state, peer).await; + if let Err(error) = send_hello_once(state, peer).await { + warn!(peer = %peer, error = %format!("{error:#}"), "HELLO to new neighbour failed; will retry"); + } + } + for peer in &change.removed { + if let Some(peer_state) = state.get_peer_state(peer).await { + let mut peer_state = peer_state.lock().await; + peer_state.stop_hello_retries(); + peer_state.reset_sync(); + } + } +} + // --- peer worker ------------------------------------------------------------- pub(crate) async fn run_peer_worker(state: AppState) { @@ -406,12 +444,17 @@ pub(crate) async fn run_peer_worker(state: AppState) { for (peer, peer_state_mutex) in &peers { let now = Instant::now(); + // Only neighbours are kept alive and judged; others we only answer. + let neighbour = topology::is_neighbour(&state, peer); let (became_dormant, ping_due, unanswered) = { let mut peer_state = peer_state_mutex.lock().await; - let became_dormant = ready && peer_state.check_dormancy(&state.timing, now); - let we_ping = state.local_id() < peer.as_str(); - let ping_due = peer_state.take_keepalive_due(&state.timing, we_ping, now); + let became_dormant = + ready && neighbour && peer_state.check_dormancy(&state.timing, now); + let ping_due = neighbour && peer_state.take_keepalive_due(&state.timing, now); + if !neighbour { + peer_state.stop_hello_retries(); + } let unanswered = peer_state.take_overdue_sync_reply(now); (became_dormant, ping_due, unanswered) }; @@ -432,13 +475,10 @@ pub(crate) async fn run_peer_worker(state: AppState) { ); } - if ping_due { - let local = local_keepalive(&state).await; - if let Err(error) = state - .transport - .send_message(peer, &ParsedMessage::Ping(local)) - .await - { + let probe_due = + !ping_due && crate::swarm::source::probe_due(&state, peer, Instant::now()); + if ping_due || probe_due { + if let Err(error) = send_ping(&state, peer).await { debug!(peer = %peer, error = %format!("{error:#}"), "Keepalive PING failed"); } } @@ -446,12 +486,25 @@ pub(crate) async fn run_peer_worker(state: AppState) { if let Err(error) = retry_hello_if_due(&state, peer).await { warn!(peer = %peer, error = %format!("{error:#}"), "HELLO retry failed"); } - - send_owed_hello_reply(&state, peer).await; } if last_forget_check.is_none_or(|t| t.elapsed() >= FORGET_CHECK_INTERVAL) { apply_forget_records(&state).await; + crate::swarm::members::ensure_registered(&state).await; + crate::swarm::stability::tick(&state).await; + crate::swarm::stats::measure_saved_size(&state); + { + let known = state.peer_states.read().await; + state.traffic.retain(|peer| known.contains_key(peer)); + state + .throughput + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .retain(|peer| known.contains_key(peer)); + } + if let Some(change) = topology::refresh(&state).await { + apply_neighbour_change(&state, change).await; + } last_forget_check = Some(Instant::now()); } } @@ -459,11 +512,17 @@ pub(crate) async fn run_peer_worker(state: AppState) { // --- peer membership --------------------------------------------------------- +/// Makes `peer` a member of `swarm_id`, after its valid HELLO (`from_hello`) or +/// HELLO_REPLY. Either completes the handshake: one round trip proves both sides to +/// each other (the HELLO the sender's key, the reply the swarm ID and the responder's +/// key), so neither side needs a HELLO of its own. `ready` is what the message said. async fn establish_peer( state: &AppState, peer: &str, swarm_id: String, -) -> Result<Vec<transport::SyncEnvelope>> { + from_hello: bool, + ready: bool, +) -> Result<()> { let expected_swarm = state.swarm_id().await; if let Some(expected) = &expected_swarm { @@ -479,7 +538,7 @@ async fn establish_peer( let peer_state_mutex = state.get_or_create_peer_state(peer).await; let mut peer_state = peer_state_mutex.lock().await; - if let PeerMembership::Member { swarm_id: existing } = &peer_state.membership { + if let Standing::Member { swarm_id: existing } = &peer_state.standing { if existing != &swarm_id { bail!( "Peer {peer} changed swarm identity from {} to {}", @@ -494,34 +553,30 @@ async fn establish_peer( state.set_expected_swarm_id(swarm_id.clone()).await?; } - if matches!(peer_state.membership, PeerMembership::Unknown) { - peer_state.membership = PeerMembership::Member { + if !matches!(peer_state.standing, Standing::Member { .. }) { + peer_state.standing = Standing::Member { swarm_id: swarm_id.clone(), }; } - // First reply completes the handshake: stop retries and sync from a clean slate. - // Duplicate replies (caused by our own retries) don't reset again. - if peer_state.hello_state != HelloState::Established { + // Stop our retries and sync from a clean slate. A HELLO always resets: the peer + // starts over with us. Duplicate replies (caused by our own retries) don't. + peer_state.reports_ready = Some(ready); + if from_hello || peer_state.hello_state != HelloState::Established { peer_state.hello_state = HelloState::Established; peer_state.reset_sync(); - info!(peer = %peer, "Handshake complete; sync state reset"); + info!(peer = %peer, from_hello, "Handshake complete; sync state reset"); } - let pending = std::mem::take(&mut peer_state.pending_inbound_sync); - - info!(peer = %peer, swarm_id = %swarm_id, pending_messages = pending.len(), "Peer membership established"); - Ok(pending) + info!(peer = %peer, swarm_id = %swarm_id, "Peer membership established"); + Ok(()) } #[instrument(skip(state), fields(peer = %peer))] pub(crate) async fn peer_swarm(state: &AppState, peer: &str) -> Option<String> { let peer_state_mutex = state.get_peer_state(peer).await?; let peer_state = peer_state_mutex.lock().await; - match &peer_state.membership { - PeerMembership::Unknown => None, - PeerMembership::Member { swarm_id } => Some(swarm_id.clone()), - } + peer_state.standing.swarm_id().map(str::to_owned) } // --- epochs ------------------------------------------------------------------ @@ -545,6 +600,8 @@ async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool // (resetting both sides again) while this fresh sync is under way. peer_state.reset_sync(); peer_state.last_sync_rx = Some(Instant::now()); + // Whether it would sync with us was said about its old epoch. + peer_state.syncs_with_us = None; } previous }; @@ -597,7 +654,12 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool { let local = state.local_id(); let proof = match message { - ParsedMessage::Hello(hello) => Some(key.verify_hello(&hello.auth, peer, local)), + ParsedMessage::Hello(hello) => Some(key.verify_hello( + &hello.auth, + hello.swarm_id.as_deref().unwrap_or(""), + peer, + local, + )), ParsedMessage::HelloReply(reply) => { Some(key.verify_hello_reply(&reply.auth, &reply.swarm_id, peer, local)) } @@ -609,8 +671,8 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool { remember_peer(state, peer).await; if let Some(peer_state) = state.get_peer_state(peer).await { let mut peer_state = peer_state.lock().await; - if !peer_state.authenticated { - peer_state.authenticated = true; + if peer_state.standing == Standing::Stranger { + peer_state.standing = Standing::Admitted; debug!(peer = %peer, "Peer proved the swarm key"); } } @@ -622,11 +684,10 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool { // we stop syncing our document to it, not just listening to it. if let Some(peer_state) = state.get_peer_state(peer).await { let mut peer_state = peer_state.lock().await; - if peer_state.authenticated { + if peer_state.standing.is_admitted() { warn!(peer = %peer, "Peer no longer proves the swarm key; dropping its membership"); } - peer_state.authenticated = false; - peer_state.membership = PeerMembership::Unknown; + peer_state.standing = Standing::Stranger; peer_state.hello_state = HelloState::NotSent; peer_state.reset_sync(); } @@ -643,7 +704,7 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool { } None => { let admitted = match state.get_peer_state(peer).await { - Some(peer_state) => peer_state.lock().await.authenticated, + Some(peer_state) => peer_state.lock().await.standing.is_admitted(), None => false, }; if !admitted { @@ -666,19 +727,31 @@ pub(crate) async fn handle_message( return Ok(()); } + // A reply to a HELLO from an earlier run of ours (late, or queued in Mycelium while + // we were down) says nothing about now; the peer answers our current HELLO. + if let transport::ParsedMessage::HelloReply(reply) = &message { + if reply.run_id != state.run_id { + state.stats.stale_hello_reply(); + debug!(peer = %peer, "Ignoring a HELLO_REPLY to a HELLO from an earlier run"); + return Ok(()); + } + } + if !admit(state, &peer, &message).await { return Ok(()); } remember_peer(state, &peer).await; - // Any inbound message proves liveness. - let reactivated = state - .get_or_create_peer_state(&peer) - .await - .lock() - .await - .mark_seen(Instant::now()); + // Any inbound message proves liveness. When we heard from it before tells a + // restarted peer how long it was away (in our HELLO_REPLY). + let (reactivated, previously_seen) = { + let peer_state_mutex = state.get_or_create_peer_state(&peer).await; + let mut peer_state = peer_state_mutex.lock().await; + let previously_seen = peer_state.last_seen; + (peer_state.mark_seen(Instant::now()), previously_seen) + }; + crate::swarm::source::heard_from(state, &peer); if reactivated { info!(peer = %peer, "Dormant peer is active again; re-handshaking"); @@ -689,50 +762,38 @@ pub(crate) async fn handle_message( //debug!(from = %peer, message = ?message, "Received message"); match message { - transport::ParsedMessage::Hello(_) => { + transport::ParsedMessage::Hello(hello) => { // A HELLO means the peer is (re)starting its handshake with us, e.g. after a - // restart. Whatever our sync state believed about it is stale. - { - let peer_state_mutex = state.get_or_create_peer_state(&peer).await; - let mut peer_state = peer_state_mutex.lock().await; - peer_state.reset_sync(); - peer_state.reply_owed = true; + // restart. It becomes a member right away; our reply makes us one for it. + let Some(swarm_id) = state.swarm_id().await else { + // We are joining ourselves; the peer retries until we know our swarm. + debug!(peer = %peer, "HELLO before we know our swarm; not answering yet"); + return Ok(()); + }; + if let Some(theirs) = &hello.swarm_id { + if theirs != &swarm_id { + bail!("Incompatible swarm at {peer}: ours is {swarm_id}, it claims {theirs}"); + } } + establish_peer(state, &peer, swarm_id, true, hello.ready).await?; - // Sends now if we know our swarm ID; otherwise stays owed. - send_owed_hello_reply(state, &peer).await; - - // Our own HELLO goes out regardless of whether the reply could be sent. - if let Err(error) = send_hello_once(state, &peer).await { - warn!(peer = %peer, error = %format!("{error:#}"), "HELLO failed; will retry"); + let away = previously_seen.map(|seen| seen.elapsed()); + if let Err(error) = send_hello_reply(state, &peer, hello.run_id, away).await { + // The peer retries its HELLO. + warn!(peer = %peer, error = %format!("{error:#}"), "HELLO_REPLY failed"); } - // Restart sync with the fresh state (no-op until the peer is a validated member). let _ = state .sync_events - .send(SyncEvent::PeerNeedsSync(peer.clone())); + .send(SyncEvent::PeerValidated(peer.clone())); } transport::ParsedMessage::HelloReply(reply) => { - let pending = establish_peer(state, &peer, reply.swarm_id.clone()).await?; + state.lock_rtt().hello_reply_received(&peer, Instant::now()); + crate::swarm::stability::note_away(state, reply.away_ms.map(Duration::from_millis)); + establish_peer(state, &peer, reply.swarm_id.clone(), false, reply.ready).await?; discover_peers(state, &peer, &reply.peers).await; - // Our swarm ID may have just become known: pay any replies we owe. - flush_owed_hello_replies(state).await; - - if !pending.is_empty() { - debug!( - to = %peer, - pending_messages = pending.len(), - "Replaying pending sync messages" - ); - for envelope in pending { - if let Err(error) = handle_sync(state, &peer, envelope).await { - warn!(peer = %peer, error = %format!("{error:#}"), "Discarding invalid pending sync envelope"); - } - } - } - state .sync_events .send(SyncEvent::PeerValidated(peer.clone())) @@ -752,7 +813,8 @@ pub(crate) async fn handle_message( if reactivated || peer_swarm(state, &peer).await.is_none() { ensure_handshake(state, &peer).await; } - let local = local_keepalive(state).await; + note_keepalive(state, &peer, &remote).await; + let local = local_keepalive(state, &peer).await; heal_if_diverged(state, &peer, &local, &remote).await; if let Err(error) = state .transport @@ -762,11 +824,16 @@ pub(crate) async fn handle_message( debug!(peer = %peer, error = %format!("{error:#}"), "PONG failed"); } } + transport::ParsedMessage::Blob(blob) => { + crate::swarm::blob::handle(state, &peer, blob).await; + } transport::ParsedMessage::Pong(remote) => { + state.lock_rtt().pong_received(&peer, Instant::now()); if reactivated || peer_swarm(state, &peer).await.is_none() { ensure_handshake(state, &peer).await; } - let local = local_keepalive(state).await; + note_keepalive(state, &peer, &remote).await; + let local = local_keepalive(state, &peer).await; heal_if_diverged(state, &peer, &local, &remote).await; } } |