summaryrefslogtreecommitdiffstats
path: root/src/swarm/peer.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm/peer.rs')
-rw-r--r--src/swarm/peer.rs311
1 files changed, 158 insertions, 153 deletions
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index d801b0f..13a980c 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -2,24 +2,21 @@
//! swarm-wide forget records, and dispatch of inbound messages.
use std::sync::Arc;
+use std::sync::atomic::Ordering;
use std::time::Duration;
-use anyhow::{anyhow, bail, Context, Result};
+use anyhow::{Context, Result, anyhow, bail};
use automerge::AutoCommit;
-use sha2::{Digest, Sha256};
use tokio::sync::Mutex;
use tokio::time::{Instant, MissedTickBehavior};
use tracing::{debug, info, instrument, trace, warn};
use crate::document::replica::{Epoch, EpochRelation};
use crate::document::store::{self, StoreError};
-use crate::state::{AppState, HelloState, Liveness, PeerMembership, PeerState};
-use crate::swarm::sync::{handle_sync, SyncEvent};
+use crate::state::{AppState, HelloState, Liveness, PeerState, Standing};
+use crate::swarm::sync::{SyncEvent, handle_sync};
use crate::transport::{self, ParsedMessage};
-const HELLO_RETRY_INITIAL: Duration = Duration::from_secs(1);
-const HELLO_RETRY_MAX: Duration = Duration::from_secs(30);
-
const PEER_TICK: Duration = Duration::from_millis(500);
const META_ROOT: &str = "_continuum";
@@ -98,24 +95,52 @@ async fn apply_forget_records(state: &AppState) {
// --- keepalive and self-healing ----------------------------------------------
-fn heads_digest(doc: &mut AutoCommit) -> transport::HeadsDigest {
- let mut heads = doc.get_heads();
- heads.sort_unstable();
- let mut hasher = Sha256::new();
- for head in &heads {
- hasher.update(head.0);
+/// 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,
}
- hasher.finalize().into()
}
-pub(crate) async fn local_keepalive(state: &AppState) -> transport::Keepalive {
- // AutoCommit::get_heads takes &mut self (it closes the pending transaction).
- // Digest and epoch come from the same locked replica, so they always match.
- let mut replica = state.doc.write().await;
- transport::Keepalive {
- heads_digest: heads_digest(&mut replica.doc),
- epoch: replica.epoch,
+/// 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);
+ if became_ready {
+ // We may have left the opening to it while it wasn't ready, and nothing else
+ // would make us look again.
+ let _ = state
+ .sync_events
+ .send(SyncEvent::PeerNeedsSync(peer.to_owned()));
+ }
}
+ crate::swarm::source::reported(state, peer, remote.ready, Some(remote.syncs_with_you));
}
async fn heal_if_diverged(
@@ -124,7 +149,7 @@ async fn heal_if_diverged(
local: &transport::Keepalive,
remote: &transport::Keepalive,
) {
- // Digests of documents from different epochs are never equal and never converge.
+ // Heads of documents from different epochs are never equal and never converge.
if local.epoch != remote.epoch {
return;
}
@@ -133,12 +158,20 @@ async fn heal_if_diverged(
return;
};
- let heal = peer_state_mutex.lock().await.needs_heal(
- &local.heads_digest,
- &remote.heads_digest,
- &state.timing,
- Instant::now(),
- );
+ 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");
@@ -162,7 +195,7 @@ pub(crate) async fn remember_peer(state: &AppState, peer: &str) {
let mut peer_states = state.peer_states.write().await;
if !peer_states.contains_key(peer) {
- peer_states.entry(peer.to_owned()).or_default();
+ peer_states.insert(peer.to_owned(), state.new_peer_state(peer));
info!(
peer = %peer,
peer_count = peer_states.len(),
@@ -182,9 +215,9 @@ pub(crate) async fn send_hello_once(state: &AppState, peer: &str) -> Result<()>
let claimed = {
let mut peer_state = peer_state_mutex.lock().await;
if peer_state.hello_state == HelloState::NotSent {
- peer_state.hello_state = HelloState::Sending {
- next_delay: HELLO_RETRY_INITIAL,
- };
+ // A new handshake: retries start over.
+ peer_state.hello_state = HelloState::Sending;
+ peer_state.hello_backoff.reset();
true
} else {
trace!(peer = %peer, hello_state = ?peer_state.hello_state, "not sending");
@@ -197,7 +230,7 @@ pub(crate) async fn send_hello_once(state: &AppState, peer: &str) -> Result<()>
}
debug!(to = %peer, "Sending HELLO");
- transmit_hello(state, peer, HELLO_RETRY_INITIAL).await
+ transmit_hello(state, peer).await
}
async fn retry_hello_if_due(state: &AppState, peer: &str) -> Result<()> {
@@ -205,43 +238,40 @@ async fn retry_hello_if_due(state: &AppState, peer: &str) -> Result<()> {
return Ok(());
};
- let delay = {
+ {
let mut peer_state = peer_state_mutex.lock().await;
match peer_state.hello_state {
- HelloState::AwaitingReply {
- next_attempt,
- next_delay,
- } if Instant::now() >= next_attempt => {
- peer_state.hello_state = HelloState::Sending { next_delay };
- next_delay
+ HelloState::AwaitingReply { next_attempt } if Instant::now() >= next_attempt => {
+ peer_state.hello_state = HelloState::Sending;
}
_ => return Ok(()),
}
- };
+ }
- debug!(
- peer = %peer,
- next_retry_ms = delay.as_millis(),
- "No HELLO_REPLY yet; re-sending HELLO"
- );
- transmit_hello(state, peer, delay).await
+ debug!(peer = %peer, "No HELLO_REPLY yet; re-sending HELLO");
+ transmit_hello(state, peer).await
}
/// Sends a HELLO without holding any lock, then schedules the next retry.
/// Success and failure both end in AwaitingReply: Mycelium accepting the message
/// says nothing about whether the peer received it.
-async fn transmit_hello(state: &AppState, peer: &str, delay: Duration) -> Result<()> {
+async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> {
+ let swarm_id = state.swarm_id().await;
+ let auth = state
+ .swarm_key
+ .as_ref()
+ .map(|key| key.hello_tag(swarm_id.as_deref().unwrap_or(""), state.local_id(), peer))
+ .unwrap_or_default();
let result = state
.transport
.send_message(
peer,
&ParsedMessage::Hello(transport::HelloRequest {
+ swarm_id,
epoch: state.local_epoch(),
- auth: state
- .swarm_key
- .as_ref()
- .map(|key| key.hello_tag(state.local_id(), peer))
- .unwrap_or_default(),
+ auth,
+ ready: state.is_ready().await,
+ run_id: state.run_id,
}),
)
.await;
@@ -249,16 +279,13 @@ async fn transmit_hello(state: &AppState, peer: &str, delay: Duration) -> Result
if let Some(peer_state_mutex) = state.get_peer_state(peer).await {
let mut peer_state = peer_state_mutex.lock().await;
- if matches!(peer_state.hello_state, HelloState::Sending { .. }) {
- let (wait, next_delay) = match peer_state.liveness {
- Liveness::Dormant { .. } => {
- (state.timing.dormant_probe, state.timing.dormant_probe)
- }
- Liveness::Active => (delay, delay.saturating_mul(2).min(HELLO_RETRY_MAX)),
+ if peer_state.hello_state == HelloState::Sending {
+ let wait = match peer_state.liveness {
+ Liveness::Dormant { .. } => state.timing.dormant_probe,
+ Liveness::Active => peer_state.hello_backoff.next(),
};
peer_state.hello_state = HelloState::AwaitingReply {
next_attempt: Instant::now() + wait,
- next_delay,
};
}
}
@@ -272,36 +299,8 @@ async fn ensure_handshake(state: &AppState, peer: &str) {
}
}
-/// Sends a HELLO_REPLY if one is owed and our swarm ID is known.
-/// Otherwise it stays owed and the peer worker tries again.
-async fn send_owed_hello_reply(state: &AppState, peer: &str) {
- if state.swarm_id().await.is_none() {
- return;
- }
-
- let Some(peer_state_mutex) = state.get_peer_state(peer).await else {
- return;
- };
-
- let owed = std::mem::take(&mut peer_state_mutex.lock().await.reply_owed);
- if !owed {
- return;
- }
-
- if let Err(error) = send_hello_reply(state, peer).await {
- warn!(peer = %peer, error = %format!("{error:#}"), "HELLO_REPLY failed; will retry");
- peer_state_mutex.lock().await.reply_owed = true;
- }
-}
-
-async fn flush_owed_hello_replies(state: &AppState) {
- let peers: Vec<String> = state.peer_states.read().await.keys().cloned().collect();
- for peer in &peers {
- send_owed_hello_reply(state, peer).await;
- }
-}
-
-async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> {
+/// Answers a HELLO that carried `run_id`.
+async fn send_hello_reply(state: &AppState, peer: &str, run_id: transport::RunId) -> Result<()> {
let swarm_id = state
.swarm_id()
.await
@@ -331,15 +330,16 @@ async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> {
.as_ref()
.map(|key| key.hello_reply_tag(&swarm_id, state.local_id(), peer))
.unwrap_or_default();
+ let ready = state.is_ready().await;
let reply = transport::HelloReply {
swarm_id,
peers,
epoch: state.local_epoch(),
auth,
+ ready,
+ run_id,
};
- let ready = state.is_ready().await;
-
debug!(to = %peer, ready, "Sending HELLO_REPLY");
state
@@ -422,8 +422,10 @@ pub(crate) async fn run_peer_worker(state: AppState) {
);
}
- if ping_due {
- let local = local_keepalive(&state).await;
+ let probe_due =
+ !ping_due && crate::swarm::source::probe_due(&state, peer, Instant::now());
+ if ping_due || probe_due {
+ let local = local_keepalive(&state, peer).await;
if let Err(error) = state
.transport
.send_message(peer, &ParsedMessage::Ping(local))
@@ -436,8 +438,6 @@ pub(crate) async fn run_peer_worker(state: AppState) {
if let Err(error) = retry_hello_if_due(&state, peer).await {
warn!(peer = %peer, error = %format!("{error:#}"), "HELLO retry failed");
}
-
- send_owed_hello_reply(&state, peer).await;
}
if last_forget_check.is_none_or(|t| t.elapsed() >= FORGET_CHECK_INTERVAL) {
@@ -449,11 +449,17 @@ pub(crate) async fn run_peer_worker(state: AppState) {
// --- peer membership ---------------------------------------------------------
+/// Makes `peer` a member of `swarm_id`, after its valid HELLO (`from_hello`) or
+/// HELLO_REPLY. Either completes the handshake: one round trip proves both sides to
+/// each other (the HELLO the sender's key, the reply the swarm ID and the responder's
+/// key), so neither side needs a HELLO of its own. `ready` is what the message said.
async fn establish_peer(
state: &AppState,
peer: &str,
swarm_id: String,
-) -> Result<Vec<transport::SyncEnvelope>> {
+ from_hello: bool,
+ ready: bool,
+) -> Result<()> {
let expected_swarm = state.swarm_id().await;
if let Some(expected) = &expected_swarm {
@@ -469,7 +475,7 @@ async fn establish_peer(
let peer_state_mutex = state.get_or_create_peer_state(peer).await;
let mut peer_state = peer_state_mutex.lock().await;
- if let PeerMembership::Member { swarm_id: existing } = &peer_state.membership {
+ if let Standing::Member { swarm_id: existing } = &peer_state.standing {
if existing != &swarm_id {
bail!(
"Peer {peer} changed swarm identity from {} to {}",
@@ -484,34 +490,30 @@ async fn establish_peer(
state.set_expected_swarm_id(swarm_id.clone()).await?;
}
- if matches!(peer_state.membership, PeerMembership::Unknown) {
- peer_state.membership = PeerMembership::Member {
+ if !matches!(peer_state.standing, Standing::Member { .. }) {
+ peer_state.standing = Standing::Member {
swarm_id: swarm_id.clone(),
};
}
- // First reply completes the handshake: stop retries and sync from a clean slate.
- // Duplicate replies (caused by our own retries) don't reset again.
- if peer_state.hello_state != HelloState::Established {
+ // Stop our retries and sync from a clean slate. A HELLO always resets: the peer
+ // starts over with us. Duplicate replies (caused by our own retries) don't.
+ peer_state.reports_ready = Some(ready);
+ if from_hello || peer_state.hello_state != HelloState::Established {
peer_state.hello_state = HelloState::Established;
peer_state.reset_sync();
- info!(peer = %peer, "Handshake complete; sync state reset");
+ info!(peer = %peer, from_hello, "Handshake complete; sync state reset");
}
- let pending = std::mem::take(&mut peer_state.pending_inbound_sync);
-
- info!(peer = %peer, swarm_id = %swarm_id, pending_messages = pending.len(), "Peer membership established");
- Ok(pending)
+ info!(peer = %peer, swarm_id = %swarm_id, "Peer membership established");
+ Ok(())
}
#[instrument(skip(state), fields(peer = %peer))]
pub(crate) async fn peer_swarm(state: &AppState, peer: &str) -> Option<String> {
let peer_state_mutex = state.get_peer_state(peer).await?;
let peer_state = peer_state_mutex.lock().await;
- match &peer_state.membership {
- PeerMembership::Unknown => None,
- PeerMembership::Member { swarm_id } => Some(swarm_id.clone()),
- }
+ peer_state.standing.swarm_id().map(str::to_owned)
}
// --- epochs ------------------------------------------------------------------
@@ -535,6 +537,8 @@ async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool
// (resetting both sides again) while this fresh sync is under way.
peer_state.reset_sync();
peer_state.last_sync_rx = Some(Instant::now());
+ // Whether it would sync with us was said about its old epoch.
+ peer_state.syncs_with_us = None;
}
previous
};
@@ -587,7 +591,12 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool {
let local = state.local_id();
let proof = match message {
- ParsedMessage::Hello(hello) => Some(key.verify_hello(&hello.auth, peer, local)),
+ ParsedMessage::Hello(hello) => Some(key.verify_hello(
+ &hello.auth,
+ hello.swarm_id.as_deref().unwrap_or(""),
+ peer,
+ local,
+ )),
ParsedMessage::HelloReply(reply) => {
Some(key.verify_hello_reply(&reply.auth, &reply.swarm_id, peer, local))
}
@@ -599,8 +608,8 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool {
remember_peer(state, peer).await;
if let Some(peer_state) = state.get_peer_state(peer).await {
let mut peer_state = peer_state.lock().await;
- if !peer_state.authenticated {
- peer_state.authenticated = true;
+ if peer_state.standing == Standing::Stranger {
+ peer_state.standing = Standing::Admitted;
debug!(peer = %peer, "Peer proved the swarm key");
}
}
@@ -612,11 +621,10 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool {
// we stop syncing our document to it, not just listening to it.
if let Some(peer_state) = state.get_peer_state(peer).await {
let mut peer_state = peer_state.lock().await;
- if peer_state.authenticated {
+ if peer_state.standing.is_admitted() {
warn!(peer = %peer, "Peer no longer proves the swarm key; dropping its membership");
}
- peer_state.authenticated = false;
- peer_state.membership = PeerMembership::Unknown;
+ peer_state.standing = Standing::Stranger;
peer_state.hello_state = HelloState::NotSent;
peer_state.reset_sync();
}
@@ -633,7 +641,7 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool {
}
None => {
let admitted = match state.get_peer_state(peer).await {
- Some(peer_state) => peer_state.lock().await.authenticated,
+ Some(peer_state) => peer_state.lock().await.standing.is_admitted(),
None => false,
};
if !admitted {
@@ -656,6 +664,15 @@ pub(crate) async fn handle_message(
return Ok(());
}
+ // A reply to a HELLO from an earlier run of ours (late, or queued in Mycelium while
+ // we were down) says nothing about now; the peer answers our current HELLO.
+ if let transport::ParsedMessage::HelloReply(reply) = &message {
+ if reply.run_id != state.run_id {
+ debug!(peer = %peer, "Ignoring a HELLO_REPLY to a HELLO from an earlier run");
+ return Ok(());
+ }
+ }
+
if !admit(state, &peer, &message).await {
return Ok(());
}
@@ -669,6 +686,7 @@ pub(crate) async fn handle_message(
.lock()
.await
.mark_seen(Instant::now());
+ crate::swarm::source::heard_from(state, &peer);
if reactivated {
info!(peer = %peer, "Dormant peer is active again; re-handshaking");
@@ -679,50 +697,35 @@ pub(crate) async fn handle_message(
//debug!(from = %peer, message = ?message, "Received message");
match message {
- transport::ParsedMessage::Hello(_) => {
+ transport::ParsedMessage::Hello(hello) => {
// A HELLO means the peer is (re)starting its handshake with us, e.g. after a
- // restart. Whatever our sync state believed about it is stale.
- {
- let peer_state_mutex = state.get_or_create_peer_state(&peer).await;
- let mut peer_state = peer_state_mutex.lock().await;
- peer_state.reset_sync();
- peer_state.reply_owed = true;
+ // restart. It becomes a member right away; our reply makes us one for it.
+ let Some(swarm_id) = state.swarm_id().await else {
+ // We are joining ourselves; the peer retries until we know our swarm.
+ debug!(peer = %peer, "HELLO before we know our swarm; not answering yet");
+ return Ok(());
+ };
+ if let Some(theirs) = &hello.swarm_id {
+ if theirs != &swarm_id {
+ bail!("Incompatible swarm at {peer}: ours is {swarm_id}, it claims {theirs}");
+ }
}
+ establish_peer(state, &peer, swarm_id, true, hello.ready).await?;
- // Sends now if we know our swarm ID; otherwise stays owed.
- send_owed_hello_reply(state, &peer).await;
-
- // Our own HELLO goes out regardless of whether the reply could be sent.
- if let Err(error) = send_hello_once(state, &peer).await {
- warn!(peer = %peer, error = %format!("{error:#}"), "HELLO failed; will retry");
+ if let Err(error) = send_hello_reply(state, &peer, hello.run_id).await {
+ // The peer retries its HELLO.
+ warn!(peer = %peer, error = %format!("{error:#}"), "HELLO_REPLY failed");
}
- // Restart sync with the fresh state (no-op until the peer is a validated member).
let _ = state
.sync_events
- .send(SyncEvent::PeerNeedsSync(peer.clone()));
+ .send(SyncEvent::PeerValidated(peer.clone()));
}
transport::ParsedMessage::HelloReply(reply) => {
- let pending = establish_peer(state, &peer, reply.swarm_id.clone()).await?;
+ establish_peer(state, &peer, reply.swarm_id.clone(), false, reply.ready).await?;
discover_peers(state, &peer, &reply.peers).await;
- // Our swarm ID may have just become known: pay any replies we owe.
- flush_owed_hello_replies(state).await;
-
- if !pending.is_empty() {
- debug!(
- to = %peer,
- pending_messages = pending.len(),
- "Replaying pending sync messages"
- );
- for envelope in pending {
- if let Err(error) = handle_sync(state, &peer, envelope).await {
- warn!(peer = %peer, error = %format!("{error:#}"), "Discarding invalid pending sync envelope");
- }
- }
- }
-
state
.sync_events
.send(SyncEvent::PeerValidated(peer.clone()))
@@ -742,7 +745,8 @@ pub(crate) async fn handle_message(
if reactivated || peer_swarm(state, &peer).await.is_none() {
ensure_handshake(state, &peer).await;
}
- let local = local_keepalive(state).await;
+ note_keepalive(state, &peer, &remote).await;
+ let local = local_keepalive(state, &peer).await;
heal_if_diverged(state, &peer, &local, &remote).await;
if let Err(error) = state
.transport
@@ -756,7 +760,8 @@ pub(crate) async fn handle_message(
if reactivated || peer_swarm(state, &peer).await.is_none() {
ensure_handshake(state, &peer).await;
}
- let local = local_keepalive(state).await;
+ note_keepalive(state, &peer, &remote).await;
+ let local = local_keepalive(state, &peer).await;
heal_if_diverged(state, &peer, &local, &remote).await;
}
}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-24 01:56:23 +0000