From da8340c7c1da0115189afe159f013557461b6a91 Mon Sep 17 00:00:00 2001 From: tv Date: Mon, 26 Oct 2026 04:32:36 +0100 Subject: split peer module --- src/swarm/peer.rs | 915 ------------------------------------------------------ 1 file changed, 915 deletions(-) delete mode 100644 src/swarm/peer.rs (limited to 'src/swarm/peer.rs') diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs deleted file mode 100644 index 79e958b..0000000 --- a/src/swarm/peer.rs +++ /dev/null @@ -1,915 +0,0 @@ -//! Peers: handshake (HELLO / HELLO_REPLY), liveness and keepalives, self-healing, -//! 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 tokio::sync::Mutex; -use tokio::time::{Instant, MissedTickBehavior}; -use tracing::{debug, info, instrument, trace, warn}; - -use crate::document::meta; -use crate::document::replica::{Epoch, EpochRelation}; -use crate::document::store::{self, StoreError}; -use crate::locks::MutexExt; -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 PEER_TICK: Duration = Duration::from_millis(500); - -const FORGET_CHECK_INTERVAL: Duration = Duration::from_secs(5); - -// --- forget records ---------------------------------------------------------- - -fn forgotten_dir() -> Vec { - meta::path(&[meta::FORGOTTEN]) -} - -pub(crate) fn forgotten_path(pk: &str) -> Vec { - let mut path = forgotten_dir(); - path.push(pk.to_owned()); - path -} - -fn forgotten_peers(doc: &AutoCommit) -> Vec { - store::keys_at(doc, &forgotten_dir()).unwrap_or_default() -} - -async fn apply_forget_records(state: &AppState) { - // Before the initial sync we can't judge the records, and we have nothing to delete. - if !state.is_ready().await { - return; - } - - let forgotten = forgotten_peers(&state.doc.read().await.doc); - if forgotten.is_empty() { - return; - } - - let now = Instant::now(); - let mut cancel = Vec::new(); - - for pk in forgotten { - if pk == state.local_id() { - cancel.push(pk); // we are evidently alive - continue; - } - - let Some(peer_state_mutex) = state.get_peer_state(&pk).await else { - continue; // already forgotten here - }; - - let alive = peer_state_mutex.lock().await.is_alive(&state.timing, now); - - if alive { - cancel.push(pk); - } else if state.peer_states.write().await.remove(&pk).is_some() { - info!(peer = %pk, "Forgetting peer per swarm-wide record"); - } - } - - if cancel.is_empty() { - return; - } - - // Individual failures are logged inside; NotFound means another node already - // cancelled the record. - let _ = state - .mutate_doc(move |doc| { - for pk in &cancel { - match store::delete_path(&mut *doc, &forgotten_path(pk), false) { - Ok(()) => info!(peer = %pk, "Peer is alive; cancelling swarm-wide forget"), - Err(StoreError::NotFound) => {} - Err(error) => { - warn!(peer = %pk, error = %error, "Failed to cancel forget record") - } - } - } - Ok::<(), StoreError>(()) - }) - .await; -} - -// --- keepalive and self-healing ---------------------------------------------- - -/// 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 -} - -/// Our keepalive for `peer`. -pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::Keepalive { - // Heads and epoch come from the same locked replica, so they always match. The heads - // are cached (`Replica`), so the read lock is enough. - let (heads, epoch, epoch_seq) = { - let replica = state.doc.read().await; - ( - transport::HeadsSummary::of(replica.heads().to_vec()), - replica.epoch, - replica.epoch_seq, - ) - }; - 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, - epoch, - epoch_seq, - 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( - state: &AppState, - peer: &str, - local: &transport::Keepalive, - remote: &transport::Keepalive, -) { - // 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; - } - - let Some(peer_state_mutex) = state.get_peer_state(peer).await else { - return; - }; - - let heal = { - let mut peer_state = peer_state_mutex.lock().await; - 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.reply_settled(); - } - // 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 { - warn!(peer = %peer, "Documents differ and sync is idle; re-handshaking to restart sync"); - ensure_handshake(state, peer).await; - } -} - -// --- peer discovery and HELLO ------------------------------------------------ - -pub(crate) async fn remember_peer(state: &AppState, peer: &str) { - if peer.is_empty() { - warn!("Ignoring empty peer public key"); - return; - } - - if peer == state.local_id() { - debug!("Ignoring own public key"); - return; - } - - let mut peer_states = state.peer_states.write().await; - - if !peer_states.contains_key(peer) { - peer_states.insert(peer.to_owned(), state.new_peer_state(peer)); - info!( - peer = %peer, - peer_count = peer_states.len(), - "Discovered peer" - ); - } else { - trace!(peer = %peer, "Peer already registered"); - } -} - -/// Starts a new handshake with `peer` even though one is complete: its reply resets -/// sync on both sides and brings its current peer list (`swarm::source`, after giving -/// it up as the catch-up source). -pub(crate) async fn rehandshake(state: &AppState, peer: &str) { - if let Some(peer_state) = state.get_peer_state(peer).await { - let mut peer_state = peer_state.lock().await; - if peer_state.hello_state == HelloState::Established { - peer_state.hello_state = HelloState::NotSent; - } - } - ensure_handshake(state, peer).await; -} - -pub(crate) async fn send_hello_once(state: &AppState, peer: &str) -> Result<()> { - let Some(peer_state_mutex) = state.get_peer_state(peer).await else { - trace!(peer = %peer, "Peer was forgotten; not sending HELLO"); - return Ok(()); - }; - - let claimed = { - let mut peer_state = peer_state_mutex.lock().await; - if peer_state.hello_state == HelloState::NotSent { - // 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"); - false - } - }; - - if !claimed { - return Ok(()); - } - - debug!(to = %peer, "Sending HELLO"); - transmit_hello(state, peer).await -} - -async fn retry_hello_if_due(state: &AppState, peer: &str) -> Result<()> { - let Some(peer_state_mutex) = state.get_peer_state(peer).await else { - return Ok(()); - }; - - { - let mut peer_state = peer_state_mutex.lock().await; - match peer_state.hello_state { - HelloState::AwaitingReply { next_attempt } if Instant::now() >= next_attempt => { - peer_state.hello_state = HelloState::Sending; - } - _ => return Ok(()), - } - } - - 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) -> 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(); - let (epoch, epoch_seq) = state.epoch_stamp().await; - state.lock_rtt().hello_sent(peer, Instant::now()); - let result = state - .transport - .send_message( - peer, - &ParsedMessage::Hello(transport::HelloRequest { - swarm_id, - epoch, - epoch_seq, - auth, - ready: state.is_ready().await, - run_id: state.run_id, - }), - ) - .await; - - if let Some(peer_state_mutex) = state.get_peer_state(peer).await { - let mut peer_state = peer_state_mutex.lock().await; - - if peer_state.hello_state == HelloState::Sending { - let was_unreachable = peer_state.unreachable_since.is_some(); - peer_state.hello_sent(Instant::now()); - if !was_unreachable && peer_state.unreachable_since.is_some() { - state.topology_changed.notify_one(); - } - 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, - }; - } - } - - result.context("Sending HELLO") -} - -async fn ensure_handshake(state: &AppState, peer: &str) { - if let Err(error) = send_hello_once(state, peer).await { - warn!(peer = %peer, error = %format!("{error:#}"), "HELLO failed; will retry"); - } -} - -/// 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, -) -> Result<()> { - let swarm_id = state - .swarm_id() - .await - .ok_or_else(|| anyhow!("Cannot send HELLO_REPLY without an established swarm ID"))?; - - let candidates: Vec<(String, Arc>)> = state - .peers_snapshot() - .await - .into_iter() - .filter(|(known, _)| known.as_str() != peer) - .collect(); - - let mut peers = Vec::new(); - for (known, peer_state_mutex) in candidates { - 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); - } - } - - let auth = state - .swarm_key - .as_ref() - .map(|key| key.hello_reply_tag(&swarm_id, state.local_id(), peer)) - .unwrap_or_default(); - let ready = state.is_ready().await; - let (epoch, epoch_seq) = state.epoch_stamp().await; - let reply = transport::HelloReply { - swarm_id, - peers, - epoch, - epoch_seq, - auth, - ready, - run_id, - away_ms: away.map(|away| away.as_millis() as u64), - sender_run_id: state.run_id, - }; - - debug!(to = %peer, ready, "Sending HELLO_REPLY"); - - state - .transport - .send_message(peer, &ParsedMessage::HelloReply(reply)) - .await - .context("Sending HELLO_REPLY")?; - - trace!(to = %peer, "HELLO_REPLY sent"); - Ok(()) -} - -async fn discover_peers(state: &AppState, from_peer: &str, advertised_peers: &[String]) { - // TODO here we could add a trace t oshwo advertised_peers - for discovered_peer in advertised_peers { - if discovered_peer == from_peer { - continue; - } - - 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, - error = %format!("{error:#}"), - "Failed to initiate HELLO with discovered peer" - ); - } - } -} - -/// New neighbours get our HELLO (we may not have known them before: the member list is -/// how a partial mesh discovers them) and a sync opening. The opening matters for a -/// neighbour whose handshake is already complete, e.g. one that returned from dormancy -/// while it wasn't our neighbour: it was only answered then, and with no HELLO to send, -/// nothing else would open sync until the keepalive heal (`resync_after`). 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. -pub(crate) 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"); - } - // Skipped there unless the handshake is complete; otherwise its reply restarts - // sync anyway. - let _ = state - .sync_events - .send(SyncEvent::PeerNeedsSync(peer.clone())); - } - 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) { - let mut tick = tokio::time::interval(PEER_TICK); - tick.set_missed_tick_behavior(MissedTickBehavior::Delay); - - info!("Starting peer maintenance worker"); - - let mut last_forget_check: Option = None; - let mut was_ready = false; - - loop { - tick.tick().await; - - // While joining, keep handshakes fast; dormancy would slow probes to minutes. - let ready = state.is_ready().await; - if ready && !was_ready { - // Just became ready: register in the member list (and do the other periodic - // checks) now rather than at the next check. Until it is registered, other - // members don't count this node as a neighbour. - last_forget_check = None; - } - was_ready = ready; - - let peers = state.peers_snapshot().await; - - 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 && 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) - }; - - if let Some(changes) = unanswered { - debug!(peer = %peer, changes, "No answer to our sync message; sending again"); - crate::swarm::source::reply_timed_out(&state, peer); - let _ = state - .sync_events - .send(SyncEvent::PeerNeedsSync(peer.clone())); - } - - if became_dormant { - state.topology_changed.notify_one(); - warn!( - peer = %peer, - silent_secs = state.timing.dormant_after.as_secs(), - probe_secs = state.timing.dormant_probe.as_secs(), - "Peer went dormant" - ); - } - - 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"); - } - } - - if let Err(error) = retry_hello_if_due(&state, peer).await { - warn!(peer = %peer, error = %format!("{error:#}"), "HELLO retry failed"); - } - } - - 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); - crate::swarm::source::warn_if_stalled(&state); - crate::swarm::state_file::update(&state, false).await; - { - let known = state.peer_states.read().await; - state.traffic.retain(|peer| known.contains_key(peer)); - state - .throughput - .lock_or_recover() - .retain(|peer| known.contains_key(peer)); - } - // Fallback for what doesn't notify (`topology::run_worker`). - state.topology_changed.notify_one(); - last_forget_check = Some(Instant::now()); - } - } -} - -// --- 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, - from_hello: bool, - ready: bool, -) -> Result<()> { - let expected_swarm = state.swarm_id().await; - - if let Some(expected) = &expected_swarm { - if expected != &swarm_id { - bail!( - "Incompatible swarm at {peer}: expected {}, received {}", - expected, - swarm_id - ); - } - } - - let peer_state_mutex = state.get_or_create_peer_state(peer).await; - let mut peer_state = peer_state_mutex.lock().await; - - if let Standing::Member { swarm_id: existing } = &peer_state.standing { - if existing != &swarm_id { - bail!( - "Peer {peer} changed swarm identity from {} to {}", - existing, - swarm_id - ); - } - } - - // Lock order: peer mutex -> lifecycle. - if expected_swarm.is_none() { - state.set_expected_swarm_id(swarm_id.clone()).await?; - } - - if !matches!(peer_state.standing, Standing::Member { .. }) { - peer_state.standing = Standing::Member { - swarm_id: swarm_id.clone(), - }; - } - - // 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, from_hello, "Handshake complete; sync state reset"); - } - - 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 { - let peer_state_mutex = state.get_peer_state(peer).await?; - let peer_state = peer_state_mutex.lock().await; - peer_state.standing.swarm_id().map(str::to_owned) -} - -// --- epochs ------------------------------------------------------------------ - -/// Records the epoch a peer advertised and compares it with ours. Returns true when the -/// message's epoch is ours, the only case in which its documents may be synced. -/// -/// `seq` is the peer's `epoch_seq` with it. A message with a lower one than we have was -/// sent before the peer's last epoch change and delivered late: its epoch isn't -/// recorded, so it can't flip our view of the peer back. The count starts over when the -/// peer restarts, which its HELLO or HELLO_REPLY tells by a new run (`run`). -async fn observe_peer_epoch( - state: &AppState, - peer: &str, - remote: Epoch, - seq: u32, - run: Option, -) -> bool { - let local = state.local_epoch(); - let relation = local.relation_to(&remote); - - let Some(peer_state_mutex) = state.get_peer_state(peer).await else { - return relation == EpochRelation::Same; - }; - - let previous = { - let mut peer_state = peer_state_mutex.lock().await; - if let Some(run) = run.filter(|run| peer_state.run_id != Some(*run)) { - peer_state.run_id = Some(run); - peer_state.epoch_seq = None; - } - if peer_state.epoch_seq.is_some_and(|newest| seq < newest) { - debug!( - peer = %peer, - epoch = %remote, - seq, - newest = ?peer_state.epoch_seq, - "Ignoring a late message's epoch" - ); - return relation == EpochRelation::Same; - } - peer_state.epoch_seq = Some(seq); - let previous = peer_state.epoch.replace(remote); - if previous.is_some_and(|previous| previous != remote) { - // The peer switched documents; our sync state describes its old one. Count - // the restart as sync activity, so the keepalive heal doesn't re-handshake - // (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 - }; - - // Log transitions only, not every keepalive. - if previous != Some(remote) { - match relation { - EpochRelation::Same => { - if previous.is_some() { - info!(peer = %peer, epoch = %remote, "Peer is on our epoch again; resuming sync"); - let _ = state - .sync_events - .send(SyncEvent::PeerNeedsSync(peer.to_owned())); - } - } - // Expected after a remake: either side switches or catches up. - EpochRelation::PeerAhead | EpochRelation::PeerBehind => { - info!( - peer = %peer, - local_epoch = %local, - peer_epoch = %remote, - relation = ?relation, - "Peer is on another epoch; not syncing with it until both are on the same" - ); - } - EpochRelation::Diverged => { - warn!( - peer = %peer, - local_epoch = %local, - peer_epoch = %remote, - "Peer is on a different epoch of the same generation; not syncing with it" - ); - } - } - } - - relation == EpochRelation::Same -} - -// --- swarm key ----------------------------------------------------------------- - -/// With a swarm key, only peers that proved it get past this point. A HELLO or -/// HELLO_REPLY must carry a valid tag, and it admits its sender; anything else is only -/// accepted from a peer admitted that way. Everything else is ignored without a reply, -/// so strangers learn nothing, not even whether we know the key. -async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool { - let Some(key) = &state.swarm_key else { - return true; - }; - let local = state.local_id(); - - let proof = match message { - 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)) - } - _ => None, - }; - - match proof { - Some(true) => { - 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.standing == Standing::Stranger { - peer_state.standing = Standing::Admitted; - debug!(peer = %peer, "Peer proved the swarm key"); - } - } - true - } - Some(false) => { - // A known peer with a wrong key (e.g. one that was given another key) loses - // its admission until it proves the right one, and with it its membership: - // 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.standing.is_admitted() { - warn!(peer = %peer, "Peer no longer proves the swarm key; dropping its membership"); - } - peer_state.standing = Standing::Stranger; - peer_state.hello_state = HelloState::NotSent; - peer_state.reset_sync(); - } - // The one place an operator can see why a node can't join: rate limited, - // since anyone can send us handshakes. - if let Some(suppressed) = key.note_rejection() { - info!( - peer = %peer, - suppressed, - "Rejected handshake without a valid swarm key proof" - ); - } - false - } - None => { - let admitted = match state.get_peer_state(peer).await { - Some(peer_state) => peer_state.lock().await.standing.is_admitted(), - None => false, - }; - if !admitted { - trace!(peer = %peer, "Ignoring message from a peer that hasn't proven the swarm key"); - } - admitted - } - } -} - -// --- inbound dispatch -------------------------------------------------------- - -pub(crate) async fn handle_message( - state: &AppState, - peer: String, - message: transport::ParsedMessage, -) -> Result<()> { - if peer == state.local_id() { - debug!("Ignoring message from self"); - 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. 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; - if peer_state.unreachable_since.is_some() { - // Reachable again (`mark_seen` clears it). - state.topology_changed.notify_one(); - } - (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"); - } - - let run = match &message { - transport::ParsedMessage::Hello(hello) => Some(hello.run_id), - transport::ParsedMessage::HelloReply(reply) => Some(reply.sender_run_id), - _ => None, - }; - let same_epoch = - observe_peer_epoch(state, &peer, message.epoch(), message.epoch_seq(), run).await; - - //debug!(from = %peer, message = ?message, "Received message"); - - match message { - transport::ParsedMessage::Hello(hello) => { - // A HELLO means the peer is (re)starting its handshake with us, e.g. after a - // 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?; - - 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"); - } - - let _ = state - .sync_events - .send(SyncEvent::PeerValidated(peer.clone())); - } - transport::ParsedMessage::HelloReply(reply) => { - 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; - - state - .sync_events - .send(SyncEvent::PeerValidated(peer.clone())) - .map_err(|error| anyhow!("Failed to notify sync worker: {error}"))?; - } - transport::ParsedMessage::Sync(envelope) => { - if reactivated { - ensure_handshake(state, &peer).await; - } - if !same_epoch { - debug!(peer = %peer, epoch = %envelope.epoch, "Dropping sync message from another epoch"); - return Ok(()); - } - crate::swarm::source::synced_from(state, &peer); - handle_sync(state, &peer, envelope).await?; - } - transport::ParsedMessage::Ping(remote) => { - if reactivated || peer_swarm(state, &peer).await.is_none() { - ensure_handshake(state, &peer).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 - .send_message(&peer, &ParsedMessage::Pong(local)) - .await - { - 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; - } - note_keepalive(state, &peer, &remote).await; - let local = local_keepalive(state, &peer).await; - heal_if_diverged(state, &peer, &local, &remote).await; - } - } - - Ok(()) -} -- cgit v1.3.1