split peer module

8 files changed, 995 insertions, 0 deletions

diff --git a/src/swarm/peer/admit.rs b/src/swarm/peer/admit.rs
new file mode 100644
index 0000000..1800775
--- /dev/null
+++ b/src/swarm/peer/admit.rs
@@ -0,0 +1,79 @@
+//! Admission with a swarm key: only peers that proved it get past `admit`.
+
+use tracing::{debug, info, trace, warn};
+
+use super::handshake::remember_peer;
+use crate::state::{AppState, HelloState, Standing};
+use crate::transport::ParsedMessage;
+
+/// 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.
+pub(super) 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
+        }
+    }
+}
diff --git a/src/swarm/peer/epoch.rs b/src/swarm/peer/epoch.rs
new file mode 100644
index 0000000..56711b1
--- /dev/null
+++ b/src/swarm/peer/epoch.rs
@@ -0,0 +1,96 @@
+//! The epochs peers advertise: recorded per peer, late messages aside, and compared
+//! with ours.
+
+use tokio::time::Instant;
+use tracing::{debug, info, warn};
+
+use crate::document::replica::{Epoch, EpochRelation};
+use crate::state::AppState;
+use crate::swarm::sync::SyncEvent;
+use crate::transport;
+
+/// 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`).
+pub(super) async fn observe_peer_epoch(
+    state: &AppState,
+    peer: &str,
+    remote: Epoch,
+    seq: u32,
+    run: Option<transport::RunId>,
+) -> 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
+}
diff --git a/src/swarm/peer/forget.rs b/src/swarm/peer/forget.rs
new file mode 100644
index 0000000..624ca97
--- /dev/null
+++ b/src/swarm/peer/forget.rs
@@ -0,0 +1,79 @@
+//! Swarm-wide forget records (`_continuum/forgotten/<pk>`): a peer forgotten on one
+//! node is forgotten by all; a forgotten node that is alive cancels its record.
+
+use automerge::AutoCommit;
+use tokio::time::Instant;
+use tracing::{info, warn};
+
+use crate::document::meta;
+use crate::document::store::{self, StoreError};
+use crate::state::AppState;
+
+fn forgotten_dir() -> Vec<String> {
+    meta::path(&[meta::FORGOTTEN])
+}
+
+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()
+}
+
+pub(super) 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;
+}
diff --git a/src/swarm/peer/handshake.rs b/src/swarm/peer/handshake.rs
new file mode 100644
index 0000000..c46df9c
--- /dev/null
+++ b/src/swarm/peer/handshake.rs
@@ -0,0 +1,336 @@
+//! Discovering peers, the handshake (HELLO / HELLO_REPLY), and membership.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use anyhow::{Context, Result, anyhow, bail};
+use tokio::sync::Mutex;
+use tokio::time::Instant;
+use tracing::{debug, info, instrument, trace, warn};
+
+use crate::state::{AppState, HelloState, Liveness, PeerState, Standing};
+use crate::swarm::sync::SyncEvent;
+use crate::swarm::topology;
+use crate::transport::{self, ParsedMessage};
+
+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
+}
+
+pub(super) 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")
+}
+
+pub(super) 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.
+pub(super) 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
+        .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(())
+}
+
+pub(super) 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 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.
+pub(super) 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)
+}
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
new file mode 100644
index 0000000..e81c71d
--- /dev/null
+++ b/src/swarm/peer/inbound.rs
@@ -0,0 +1,152 @@
+//! Dispatch of inbound messages.
+
+use std::time::Duration;
+
+use anyhow::{Result, anyhow, bail};
+use tokio::time::Instant;
+use tracing::{debug, info, warn};
+
+use super::admit::admit;
+use super::epoch::observe_peer_epoch;
+use super::handshake::{
+    discover_peers, ensure_handshake, establish_peer, peer_swarm, remember_peer, send_hello_reply,
+};
+use super::keepalive::{heal_if_diverged, local_keepalive, note_keepalive};
+use crate::state::AppState;
+use crate::swarm::sync::{SyncEvent, handle_sync};
+use crate::transport::{self, ParsedMessage};
+
+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(())
+}
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
new file mode 100644
index 0000000..4fcad7e
--- /dev/null
+++ b/src/swarm/peer/keepalive.rs
@@ -0,0 +1,112 @@
+//! Keepalives (PING / PONG) and the self-healing re-handshake when they show our
+//! documents differ while sync is idle.
+
+use std::sync::atomic::Ordering;
+
+use anyhow::Result;
+use tokio::time::Instant;
+use tracing::warn;
+
+use super::handshake::ensure_handshake;
+use crate::state::{AppState, Liveness, Standing};
+use crate::swarm::sync::SyncEvent;
+use crate::swarm::topology;
+use crate::transport::{self, ParsedMessage};
+
+/// 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.
+pub(super) 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));
+}
+
+pub(super) 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;
+    }
+}
diff --git a/src/swarm/peer/mod.rs b/src/swarm/peer/mod.rs
new file mode 100644
index 0000000..086f5d2
--- /dev/null
+++ b/src/swarm/peer/mod.rs
@@ -0,0 +1,26 @@
+//! Peers: handshake (HELLO / HELLO_REPLY), liveness and keepalives, self-healing,
+//! swarm-wide forget records, and dispatch of inbound messages.
+//!
+//! * `handshake`: discovering peers, HELLO / HELLO_REPLY, membership.
+//! * `keepalive`: PING / PONG and the self-healing re-handshake.
+//! * `epoch`: the epochs peers advertise.
+//! * `admit`: admission with a swarm key.
+//! * `forget`: swarm-wide forget records.
+//! * `worker`: the periodic peer worker.
+//! * `inbound`: dispatch of inbound messages.
+
+mod admit;
+mod epoch;
+mod forget;
+mod handshake;
+mod inbound;
+mod keepalive;
+mod worker;
+
+pub(crate) use forget::forgotten_path;
+pub(crate) use handshake::{
+    apply_neighbour_change, peer_swarm, rehandshake, remember_peer, send_hello_once,
+};
+pub(crate) use inbound::handle_message;
+pub(crate) use keepalive::send_ping;
+pub(crate) use worker::run_peer_worker;
diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs
new file mode 100644
index 0000000..a27f144
--- /dev/null
+++ b/src/swarm/peer/worker.rs
@@ -0,0 +1,115 @@
+//! The peer worker: every `PEER_TICK`, keepalives due, dormancy, HELLO retries and
+//! overdue sync replies; every `FORGET_CHECK_INTERVAL` (and at once when the node
+//! becomes ready), forget records, the member registry, stability, measurements and
+//! the state file.
+
+use std::time::Duration;
+
+use tokio::time::{Instant, MissedTickBehavior};
+use tracing::{debug, info, warn};
+
+use super::forget::apply_forget_records;
+use super::handshake::retry_hello_if_due;
+use super::keepalive::send_ping;
+use crate::locks::MutexExt;
+use crate::state::AppState;
+use crate::swarm::sync::SyncEvent;
+use crate::swarm::topology;
+
+const PEER_TICK: Duration = Duration::from_millis(500);
+
+const FORGET_CHECK_INTERVAL: Duration = Duration::from_secs(5);
+
+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;
+    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());
+        }
+    }
+}