path: root/src/swarm/peer.rs

//! 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::replica::{Epoch, EpochRelation};
use crate::document::store::{self, StoreError};
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 META_ROOT: &str = "_continuum";
const FORGET_CHECK_INTERVAL: Duration = Duration::from_secs(5);

// --- forget records ----------------------------------------------------------

fn forgotten_dir() -> Vec<String> {
    vec![META_ROOT.to_owned(), "forgotten".to_owned()]
}

pub(crate) fn forgotten_path(pk: &str) -> Vec<String> {
    let mut path = forgotten_dir();
    path.push(pk.to_owned());
    path
}

fn forgotten_peers(doc: &AutoCommit) -> Vec<String> {
    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 {
    // AutoCommit::get_heads takes &mut self (it closes the pending transaction).
    // 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,
        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(
    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");
    }
}

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();
    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,
                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 {
            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,
            };
        }
    }

    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<Duration>,
) -> 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<Mutex<PeerState>>)> = state
        .peer_states
        .read()
        .await
        .iter()
        .filter(|(known, _)| known.as_str() != peer)
        .map(|(known, m)| (known.clone(), Arc::clone(m)))
        .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 reply = transport::HelloReply {
        swarm_id,
        peers,
        epoch: state.local_epoch(),
        auth,
        ready,
        run_id,
        away_ms: away.map(|away| away.as_millis() as u64),
    };

    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). 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) {
    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<Instant> = None;

    loop {
        tick.tick().await;

        // While joining, keep handshakes fast; dormancy would slow probes to minutes.
        let ready = state.is_ready().await;

        let peers: Vec<(String, Arc<Mutex<PeerState>>)> = state
            .peer_states
            .read()
            .await
            .iter()
            .map(|(pk, m)| (pk.clone(), Arc::clone(m)))
            .collect();

        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");
                let _ = state
                    .sync_events
                    .send(SyncEvent::PeerNeedsSync(peer.clone()));
            }

            if became_dormant {
                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;
            if let Some(change) = topology::refresh(&state).await {
                apply_neighbour_change(&state, change).await;
            }
            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<String> {
    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 both
/// are on the same epoch, the only case in which the documents may be synced.
async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> 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;
        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 {
            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;
        (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 same_epoch = observe_peer_epoch(state, &peer, message.epoch()).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(());
            }
            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::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(())
}