summaryrefslogtreecommitdiffstats
path: root/src/swarm/sync.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm/sync.rs')
-rw-r--r--src/swarm/sync.rs479
1 files changed, 243 insertions, 236 deletions
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 15b4ced..58a08db 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -2,30 +2,27 @@
//! messages, and the change publisher that feeds incremental streams.
use std::collections::HashSet;
-use std::sync::atomic::Ordering;
use std::sync::Arc;
-use std::time::Duration;
+use std::sync::atomic::Ordering;
-use anyhow::{anyhow, bail, Context, Result};
+use anyhow::{Context, Result, anyhow, bail};
use automerge::sync::SyncDoc;
use tokio::sync::mpsc;
use tokio::time::Instant;
use tracing::{debug, info, trace, warn};
-use crate::state::{AppState, Liveness, PeerMembership, PeerState, PendingOutbound};
+use crate::state::{AppState, Liveness, Sending, Standing};
use crate::swarm::peer::{peer_swarm, send_hello_once};
+use crate::swarm::source;
+use crate::swarm::throughput::SyncBytes;
use crate::transport::{self, ParsedMessage};
-const INITIAL_SYNC_RETRY: Duration = Duration::from_millis(500);
-const MAX_SYNC_RETRY: Duration = Duration::from_secs(30);
-
#[derive(Debug)]
pub(crate) enum SyncEvent {
PeerValidated(String),
DocumentChanged,
MessageReceived(String),
PeerNeedsSync(String),
- RetryDelivery(String),
}
/// Store paths touched by one batch of document changes.
@@ -34,12 +31,20 @@ pub(crate) struct ChangeSet {
pub(crate) paths: Vec<Vec<String>>,
}
-async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool) -> Result<()> {
+async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
+ if state.remake.fetching_snapshot() {
+ // Rejoining: the snapshot comes as a download, not in a sync message. Sync
+ // starts once it is here (`remake::catchup`).
+ trace!("Waiting for the snapshot download; not syncing yet");
+ return Ok(());
+ }
let Some(peer_state_mutex) = state.get_peer_state(peer).await else {
trace!("Peer was forgotten; skipping synchronization");
return Ok(());
};
+ let we_ready = state.is_ready().await;
+
let validated_swarm = {
let peer_state = peer_state_mutex.lock().await;
@@ -58,25 +63,41 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
return Ok(());
}
- let PeerMembership::Member { swarm_id } = &peer_state.membership else {
+ let Standing::Member { swarm_id } = &peer_state.standing else {
return Ok(());
};
- if !retry_delivery {
- let we_open = state.local_id() < peer;
- if let Some(wait) = peer_state.opening_wait(we_open, Instant::now()) {
- trace!(
- wait_ms = wait.as_millis(),
- "Leaving the opening sync message to the peer"
- );
- let tx = state.sync_events.clone();
- let peer = peer.to_owned();
- tokio::spawn(async move {
- tokio::time::sleep(wait).await;
- let _ = tx.send(SyncEvent::PeerNeedsSync(peer));
- });
- return Ok(());
- }
+ if let Some(wait) = source::defer(state, peer, &peer_state) {
+ trace!(
+ wait_ms = wait.as_millis(),
+ "Catching up from another peer first; not syncing with this one yet"
+ );
+ // Done earlier when the initial sync completes (every peer is enqueued
+ // then); this only matters if the source goes quiet.
+ let tx = state.sync_events.clone();
+ let peer = peer.to_owned();
+ tokio::spawn(async move {
+ tokio::time::sleep(wait).await;
+ let _ = tx.send(SyncEvent::PeerNeedsSync(peer));
+ });
+ return Ok(());
+ }
+
+ if peer_state
+ .sync
+ .leave_opening(we_ready, peer_state.reports_ready)
+ {
+ // Its own opening (or anything else from it) brings us back here.
+ trace!("Leaving the opening sync message to the peer, which isn't ready");
+ return Ok(());
+ }
+
+ if peer_state.sync.state.their_heads.is_none()
+ && !crate::swarm::topology::is_neighbour(state, peer)
+ {
+ // We only answer peers that aren't our neighbours.
+ trace!("Not opening sync with a peer that isn't our neighbour");
+ return Ok(());
}
swarm_id.clone()
@@ -91,104 +112,113 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
bail!("Peer {peer} belongs to a different swarm");
}
- let existing = {
- let mut guard = peer_state_mutex.lock().await;
- let peer_state = &mut *guard;
- match &peer_state.pending_outbound_sync {
- Some(entry) if retry_delivery => Some((entry.message.clone(), entry.next_attempt)),
- Some(_) => {
- peer_state.needs_generate = true;
- return Ok(());
- }
- None => None,
+ {
+ let mut peer_state = peer_state_mutex.lock().await;
+ if let Some(sending) = &mut peer_state.sync.sending {
+ sending.regenerate = true;
+ return Ok(());
}
- };
-
- let encoded = match existing {
- Some((message, next_attempt)) => {
- if Instant::now() < next_attempt {
- return Ok(());
- }
- message
+ if peer_state
+ .sync
+ .retry_at
+ .is_some_and(|at| Instant::now() < at)
+ {
+ // The retry timer enqueues the peer once it's due.
+ return Ok(());
}
- None if retry_delivery => return Ok(()),
- None => {
- let doc = Arc::clone(&state.doc);
- let psm = Arc::clone(&peer_state_mutex);
- let transport = Arc::clone(&state.transport);
- let peer_owned = peer.to_owned();
+ }
- let generated = tokio::task::spawn_blocking(
- move || -> Result<Option<transport::EncodedMessage>> {
- // Label the message with the epoch of the document it was generated
- // from, read under the same lock.
- let (epoch, message) = {
- let mut replica = doc.blocking_write();
- let mut peer_state = psm.blocking_lock();
- peer_state.needs_generate = false;
- let epoch = replica.epoch;
- let sent_before = peer_state.sync_state.sent_hashes.clone();
- let message = {
- let sync_doc = replica.doc.sync();
- sync_doc.generate_sync_message(&mut peer_state.sync_state)
- };
- // A peer whose heads are ours already has everything; Automerge
- // answers only when it has something to say, so it will stay
- // silent. Waiting for an answer would only end in a pointless
- // resend.
- let in_step = peer_state.sync_state.their_heads.as_ref()
- == Some(&replica.doc.get_heads());
- if in_step {
- peer_state.sync_in_step();
- } else if message.is_some() {
- let sent = peer_state
- .sync_state
- .sent_hashes
- .difference(&sent_before)
- .copied()
- .collect();
- peer_state.expect_sync_reply(sent, Instant::now());
- }
- (epoch, message)
- };
- let Some(message) = message else {
- return Ok(None);
- };
- let envelope = transport::SyncEnvelope {
- swarm_id,
- epoch,
- message,
+ let encoded = {
+ let doc = Arc::clone(&state.doc);
+ let psm = Arc::clone(&peer_state_mutex);
+ let transport = Arc::clone(&state.transport);
+ let peer_owned = peer.to_owned();
+ let ready = state.is_ready().await;
+ // Before this message is counted.
+ let bytes_before = SyncBytes::of(&state.traffic, peer);
+
+ let generated =
+ tokio::task::spawn_blocking(move || -> Result<Option<transport::EncodedMessage>> {
+ // Label the message with the epoch of the document it was generated
+ // from, read under the same lock.
+ let (epoch, message) = {
+ let mut replica = doc.blocking_write();
+ let mut peer_state = psm.blocking_lock();
+ let epoch = replica.epoch;
+ let sent_before = peer_state.sync.state.sent_hashes.clone();
+ let message = {
+ let sync_doc = replica.doc.sync();
+ sync_doc.generate_sync_message(&mut peer_state.sync.state)
};
- transport
- .encode(&peer_owned, &ParsedMessage::Sync(envelope))
- .map(Some)
- },
- )
+ // A peer whose heads are ours already has everything; Automerge
+ // answers only when it has something to say, so it will stay
+ // silent. Waiting for an answer would only end in a pointless
+ // resend.
+ let in_step = peer_state.sync.state.their_heads.as_ref()
+ == Some(&replica.doc.get_heads());
+ let sent: Vec<_> = peer_state
+ .sync
+ .state
+ .sent_hashes
+ .difference(&sent_before)
+ .copied()
+ .collect();
+ if in_step {
+ peer_state.sync.reply_settled();
+ } else if message.is_some() {
+ peer_state
+ .sync
+ .expect_reply(sent.clone(), Instant::now(), bytes_before);
+ }
+ if message.is_some() {
+ // Marked under the same lock as the generation, so no second
+ // message is generated while this one is handed over.
+ peer_state.sync.sending = Some(Sending {
+ sent,
+ regenerate: false,
+ });
+ }
+ (epoch, message)
+ };
+ let Some(message) = message else {
+ return Ok(None);
+ };
+ let envelope = transport::SyncEnvelope {
+ swarm_id,
+ epoch,
+ message,
+ ready,
+ };
+ transport
+ .encode(&peer_owned, &ParsedMessage::Sync(envelope))
+ .map(Some)
+ })
.await
- .context("Sync generation task panicked")??;
-
- let Some(encoded) = generated else {
- return Ok(());
- };
+ .context("Sync generation task panicked")?;
+ let generated = match generated {
+ Ok(generated) => generated,
+ Err(error) => {
+ // Not encodable, so nothing went out: take back what it would have carried.
+ let _ = peer_state_mutex
+ .lock()
+ .await
+ .sync
+ .send_failed(Instant::now());
+ return Err(error);
+ }
+ };
- peer_state_mutex.lock().await.pending_outbound_sync = Some(PendingOutbound {
- message: encoded.clone(),
- retry_delay: INITIAL_SYNC_RETRY,
- next_attempt: Instant::now(),
- });
+ let Some(encoded) = generated else {
+ return Ok(());
+ };
- encoded
- }
+ encoded
};
// No state locks are held across the send.
match state.transport.send_encoded(&encoded).await {
Ok(()) => {
- let regenerate = {
- let mut peer_state = peer_state_mutex.lock().await;
- peer_state.pending_outbound_sync = None;
- std::mem::take(&mut peer_state.needs_generate)
- };
+ let regenerate = peer_state_mutex.lock().await.sync.send_succeeded();
trace!(regenerate, "Outbound sync message accepted by Mycelium");
@@ -202,30 +232,27 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
Ok(())
}
Err(error) => {
- let delay = {
- let mut peer_state = peer_state_mutex.lock().await;
-
- let Some(ref mut entry) = peer_state.pending_outbound_sync else {
- debug!("Pending envelope discarded by a sync reset during delivery");
- return Ok(());
- };
-
- entry.retry_delay = entry.retry_delay.saturating_mul(2).min(MAX_SYNC_RETRY);
- entry.next_attempt = Instant::now() + entry.retry_delay;
- entry.retry_delay
+ let Some(delay) = peer_state_mutex
+ .lock()
+ .await
+ .sync
+ .send_failed(Instant::now())
+ else {
+ debug!("Sync message refused, but a reset already discarded it");
+ return Ok(());
};
warn!(
retry_ms = delay.as_millis(),
error = %format!("{error:#}"),
- "Sync delivery failed; retaining the same outbound envelope"
+ "Sync delivery failed; generating a fresh message later"
);
let tx = state.sync_events.clone();
let peer = peer.to_owned();
tokio::spawn(async move {
tokio::time::sleep(delay).await;
- let _ = tx.send(SyncEvent::RetryDelivery(peer));
+ let _ = tx.send(SyncEvent::PeerNeedsSync(peer));
});
Ok(())
@@ -239,38 +266,28 @@ pub(crate) async fn handle_sync(
envelope: transport::SyncEnvelope,
) -> Result<()> {
debug!(peer = %peer, "Processing sync envelope");
+ // This message is counted already (`transport::traffic`).
+ let arrived = (Instant::now(), SyncBytes::of(&state.traffic, peer));
let peer_state_mutex = state.get_or_create_peer_state(peer).await;
let needs_hello = {
- let mut peer_state = peer_state_mutex.lock().await;
+ let peer_state = peer_state_mutex.lock().await;
- match &peer_state.membership {
- PeerMembership::Unknown => {
- let duplicate = peer_state.pending_inbound_sync.iter().any(|pending| {
- pending.message == envelope.message
- && pending.swarm_id == envelope.swarm_id
- && pending.epoch == envelope.epoch
- });
- if !duplicate {
- peer_state.pending_inbound_sync.push(envelope.clone());
- info!(
- peer = %peer,
- pending_count = peer_state.pending_inbound_sync.len(),
- "Queued sync envelope pending peer membership"
- );
- } else {
- trace!(peer = %peer, "Ignoring duplicate pending sync envelope");
- }
+ match &peer_state.standing {
+ Standing::Stranger | Standing::Admitted => {
+ // Not processed: our handshake reply resets the session anyway, and the
+ // sender's reply timeout resends what still matters.
+ debug!(peer = %peer, "Dropping sync message from a peer that isn't a member yet");
true
}
- PeerMembership::Member { swarm_id } if swarm_id != &envelope.swarm_id => {
+ Standing::Member { swarm_id } if swarm_id != &envelope.swarm_id => {
bail!(
"Sync swarm mismatch for peer {peer}: validated_swarm={swarm_id}, envelope_swarm={}",
envelope.swarm_id
);
}
- PeerMembership::Member { .. } => false,
+ Standing::Member { .. } => false,
}
};
@@ -296,17 +313,26 @@ pub(crate) async fn handle_sync(
}
let epoch_label = envelope.epoch;
+ let from_ready = envelope.ready;
let applied = {
let doc = Arc::clone(&state.doc);
let peer_state_mutex = Arc::clone(&peer_state_mutex);
+ let stats = Arc::clone(&state.stats);
+ let throughput = Arc::clone(&state.throughput);
+ let rtt = state
+ .lock_rtt()
+ .effective(peer, Instant::now(), state.timing.rtt_fresh_for());
let check_caught_up = !state.initial_sync_done.load(Ordering::Acquire);
+ // While catching up, everything arrives late; that says nothing about propagation.
+ let measure_propagation = !check_caught_up;
let epoch = envelope.epoch;
let message = envelope.message;
let peer = peer.to_owned();
- // Some(caught_up) when applied, None when the message belongs to another epoch.
- tokio::task::spawn_blocking(move || -> Result<Option<bool>> {
+ // Some((caught_up, heads changed)) when applied, None when the message belongs
+ // to another epoch.
+ tokio::task::spawn_blocking(move || -> Result<Option<(bool, bool)>> {
let mut replica = doc.blocking_write();
// Authoritative check, under the same lock as the apply: the document may
@@ -316,28 +342,49 @@ pub(crate) async fn handle_sync(
}
let mut peer_state = peer_state_mutex.blocking_lock();
-
+ peer_state.reports_ready = Some(from_ready);
let heads_before = replica.doc.get_heads();
- let their_heads_before = peer_state.sync_state.their_heads.clone();
replica
.doc
.sync()
- .receive_sync_message(&mut peer_state.sync_state, message)
+ .receive_sync_message(&mut peer_state.sync.state, message)
.context("Automerge synchronization failed")?;
peer_state.last_sync_rx = Some(Instant::now());
- peer_state.sync_reply_received();
+ if let Some(wait) = &peer_state.sync.reply {
+ throughput
+ .lock()
+ .unwrap_or_else(|poisoned| poisoned.into_inner())
+ .exchange(&peer, wait.started, arrived, rtt);
+ }
+ peer_state.sync.reply_settled();
+ resend_if_needed(&peer, &mut peer_state.sync.state);
let heads_after = replica.doc.get_heads();
- let unproductive = heads_before == heads_after
- && peer_state.sync_state.their_heads == their_heads_before
- && peer_state.sync_state.their_heads.as_ref() != Some(&heads_after);
- break_sync_stall(&peer, &mut peer_state, unproductive);
+ stats.heads_seen(heads_after.len());
+ let changed = heads_after != heads_before;
+ if changed && measure_propagation {
+ let now_ms = crate::state::unix_millis() as i64;
+ // Time 0: not stamped (snapshots, older nodes).
+ let delays: Vec<i64> = replica
+ .doc
+ .get_changes(&heads_before)
+ .iter()
+ .map(|change| change.timestamp())
+ .filter(|&time| time > 0)
+ .map(|time| now_ms - time)
+ .collect();
+ stats.propagated(delays);
+ }
- Ok(Some(
+ Ok(Some((
+ // Only from a ready peer: a peer still catching up has empty heads, which
+ // every document "contains".
check_caught_up
+ && from_ready
&& peer_state
- .sync_state
+ .sync
+ .state
.their_heads
.as_ref()
.is_some_and(|heads| {
@@ -345,23 +392,35 @@ pub(crate) async fn handle_sync(
.iter()
.all(|h| replica.doc.get_change_by_hash(h).is_some())
}),
- ))
+ changed,
+ )))
})
.await
.context("Sync task panicked")??
};
- let Some(caught_up) = applied else {
+ let Some((caught_up, changed)) = applied else {
debug!(peer = %peer, epoch = %epoch_label, "Dropped sync message from another epoch");
return Ok(());
};
+ source::reported(state, peer, from_ready, None);
+
if caught_up && !state.initial_sync_done.swap(true, Ordering::AcqRel) {
info!(peer = %peer, "Initial sync complete");
+ state
+ .stats
+ .caught_up(state.traffic.total().received.sync.bytes);
+ source::finished(state, peer);
}
state.notify_store_changed();
+ crate::swarm::relay::pulled(state, peer);
+ if changed {
+ crate::swarm::relay::announce(state, peer).await;
+ }
+
state
.sync_events
.send(SyncEvent::MessageReceived(peer.to_owned()))
@@ -370,75 +429,38 @@ pub(crate) async fn handle_sync(
Ok(())
}
-/// Exchanges in a row that changed nothing on either side before `sent_hashes` is
-/// cleared, and before a full two-sided reset (re-handshake).
-const STALL_RESEND_NEEDED: u32 = 2;
-const STALL_CLEAR_SENT: u32 = 8;
-const STALL_RESET: u32 = 32;
-
-/// Breaks sync livelocks.
-///
/// Automerge never resends a change it believes the peer has received (`sent_hashes`).
-/// If that change was lost on the way (a dropped or undeliverable message, an epoch
-/// check, a reset of only one side's state), the two sides keep exchanging messages
-/// with differing heads and no changes, as fast as the transport allows. The keepalive
-/// heal never fires then, because sync traffic keeps flowing.
-///
-/// * A change the peer keeps asking for is sent again.
-/// * After a few unproductive exchanges, forget what we sent: the next message carries
-/// everything the peer's bloom filter lacks.
-/// * If that doesn't help either, reset both sides through a fresh handshake.
-fn break_sync_stall(peer: &str, peer_state: &mut PeerState, unproductive: bool) {
- if !unproductive {
- peer_state.stalled_exchanges = 0;
+/// If the message carrying it was lost while later ones arrive, the peer keeps asking for
+/// it (`need`, from the heads we advertise) and we keep not sending it: a livelock the
+/// reply timeout can't see (answers keep coming) and the keepalive heal can't either
+/// (sync never goes idle). So when the peer needs a change we think it has, what we
+/// think it has is wrong: forget all of it, and the next message carries everything its
+/// bloom filter lacks. (A reordered message can make that send a change twice; cheap.)
+fn resend_if_needed(peer: &str, sync_state: &mut automerge::sync::State) {
+ let Some(need) = &sync_state.their_need else {
return;
- }
- peer_state.stalled_exchanges += 1;
-
- let sync_state = &mut peer_state.sync_state;
-
- // Not on the first unproductive exchange: when messages cross, the peer asks for a
- // change that is merely still in flight, and resending it would only duplicate it.
- if peer_state.stalled_exchanges >= STALL_RESEND_NEEDED {
- if let Some(need) = &sync_state.their_need {
- let before = sync_state.sent_hashes.len();
- sync_state.sent_hashes.retain(|hash| !need.contains(hash));
- let resend = before - sync_state.sent_hashes.len();
- if resend > 0 {
- debug!(peer = %peer, resend, "Peer still needs changes we had sent; sending them again");
- }
- }
- }
-
- match peer_state.stalled_exchanges {
- STALL_CLEAR_SENT => {
- warn!(
- peer = %peer,
- sent = sync_state.sent_hashes.len(),
- their_heads = ?sync_state.their_heads,
- shared_heads = ?sync_state.shared_heads,
- "Sync is stalled; forgetting which changes were sent"
- );
- sync_state.sent_hashes.clear();
- }
- STALL_RESET => {
- warn!(peer = %peer, "Sync is still stalled; resetting it on both sides");
- peer_state.stalled_exchanges = 0;
- peer_state.request_resync();
- }
- _ => {}
+ };
+ if need
+ .iter()
+ .any(|hash| sync_state.sent_hashes.contains(hash))
+ {
+ debug!(
+ peer = %peer,
+ sent = sync_state.sent_hashes.len(),
+ "Peer needs changes we had sent; forgetting what was sent"
+ );
+ sync_state.sent_hashes.clear();
}
}
pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedReceiver<SyncEvent>) {
let mut dirty_peers = HashSet::<String>::new();
- let mut retry_peers = HashSet::<String>::new();
let mut document_dirty = false;
info!("Starting synchronization worker");
loop {
- if dirty_peers.is_empty() && retry_peers.is_empty() && !document_dirty {
+ if dirty_peers.is_empty() && !document_dirty {
match rx.recv().await {
Some(event) => match event {
SyncEvent::PeerValidated(peer)
@@ -446,9 +468,6 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
| SyncEvent::PeerNeedsSync(peer) => {
dirty_peers.insert(peer);
}
- SyncEvent::RetryDelivery(peer) => {
- retry_peers.insert(peer);
- }
SyncEvent::DocumentChanged => {
document_dirty = true;
}
@@ -468,9 +487,6 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
| SyncEvent::PeerNeedsSync(peer) => {
dirty_peers.insert(peer);
}
- SyncEvent::RetryDelivery(peer) => {
- retry_peers.insert(peer);
- }
SyncEvent::DocumentChanged => {
document_dirty = true;
}
@@ -489,30 +505,21 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece
peers_guard.keys().cloned().collect()
};
+ // Our own changes go to our neighbours; they pass them on. Others only get
+ // answers to what they send us.
for peer in peer_keys {
- if peer_swarm(&state, &peer).await.is_some() {
+ if peer_swarm(&state, &peer).await.is_some()
+ && crate::swarm::topology::is_neighbour(&state, &peer)
+ {
dirty_peers.insert(peer);
}
}
}
- // Retry existing envelopes separately from generating new ones.
- let retries: Vec<String> = retry_peers.drain().collect();
-
- for peer in retries {
- if let Err(error) = deliver_sync_to_peer(&state, &peer, true).await {
- warn!(
- peer = %peer,
- error = %format!("{error:#}"),
- "Sync delivery retry failed"
- );
- }
- }
-
let peers: Vec<String> = dirty_peers.drain().collect();
for peer in peers {
- if let Err(error) = deliver_sync_to_peer(&state, &peer, false).await {
+ if let Err(error) = deliver_sync_to_peer(&state, &peer).await {
warn!(
peer = %peer,
error = %format!("{error:#}"),
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-24 18:28:57 +0000