diff options
Diffstat (limited to 'src/swarm/peer.rs')
| -rw-r--r-- | src/swarm/peer.rs | 96 |
1 files changed, 82 insertions, 14 deletions
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index 3d7da32..e8248bf 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -110,11 +110,12 @@ pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> { 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) = { + 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 { @@ -129,6 +130,7 @@ pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport:: transport::Keepalive { heads, epoch, + epoch_seq, ready: state.is_ready().await, syncs_with_you, } @@ -276,6 +278,7 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> { .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 @@ -283,7 +286,8 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> { peer, &ParsedMessage::Hello(transport::HelloRequest { swarm_id, - epoch: state.local_epoch(), + epoch, + epoch_seq, auth, ready: state.is_ready().await, run_id: state.run_id, @@ -295,7 +299,11 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> { 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(), @@ -354,14 +362,17 @@ async fn send_hello_reply( .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: state.local_epoch(), + 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"); @@ -399,15 +410,24 @@ async fn discover_peers(state: &AppState, from_peer: &str, advertised_peers: &[S } /// New neighbours get our HELLO (we may not have known them before: the member list is -/// how a partial mesh discovers them). Former neighbours get no more HELLOs, keepalives -/// or sync openings from us; their session is reset, so we don't keep pushing changes -/// to them either. They are still answered if they contact us. -async fn apply_neighbour_change(state: &AppState, change: topology::NeighbourChange) { +/// 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 { @@ -427,12 +447,20 @@ pub(crate) async fn run_peer_worker(state: AppState) { 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: Vec<(String, Arc<Mutex<PeerState>>)> = state .peer_states @@ -461,12 +489,14 @@ pub(crate) async fn run_peer_worker(state: AppState) { 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(), @@ -493,6 +523,8 @@ pub(crate) async fn run_peer_worker(state: AppState) { 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)); @@ -502,9 +534,8 @@ pub(crate) async fn run_peer_worker(state: AppState) { .unwrap_or_else(|poisoned| poisoned.into_inner()) .retain(|peer| known.contains_key(peer)); } - if let Some(change) = topology::refresh(&state).await { - apply_neighbour_change(&state, change).await; - } + // Fallback for what doesn't notify (`topology::run_worker`). + state.topology_changed.notify_one(); last_forget_check = Some(Instant::now()); } } @@ -581,9 +612,20 @@ pub(crate) async fn peer_swarm(state: &AppState, peer: &str) -> Option<String> { // --- 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 { +/// 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<transport::RunId>, +) -> bool { let local = state.local_epoch(); let relation = local.relation_to(&remote); @@ -593,6 +635,21 @@ async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool 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 @@ -749,6 +806,10 @@ pub(crate) async fn handle_message( 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); @@ -757,7 +818,13 @@ pub(crate) async fn handle_message( info!(peer = %peer, "Dormant peer is active again; re-handshaking"); } - let same_epoch = observe_peer_epoch(state, &peer, message.epoch()).await; + 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"); @@ -807,6 +874,7 @@ pub(crate) async fn handle_message( 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) => { |