diff options
| -rw-r--r-- | src/api/peers.rs | 2 | ||||
| -rw-r--r-- | src/state.rs | 63 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 285 |
3 files changed, 180 insertions, 170 deletions
diff --git a/src/api/peers.rs b/src/api/peers.rs index 3317185..aeb4373 100644 --- a/src/api/peers.rs +++ b/src/api/peers.rs @@ -125,7 +125,7 @@ pub(crate) async fn list_peers( .last_seen .map(|t| now.saturating_duration_since(t).as_secs()), dormant_for_secs, - has_pending_delivery: ps.sync.outbound.is_some(), + has_pending_delivery: ps.sync.sending.is_some() || ps.sync.retry_at.is_some(), epoch: ps.epoch.map(EpochInfo::from), same_epoch: ps.epoch.map(|epoch| epoch == local_epoch), heads: ps.reported_heads, diff --git a/src/state.rs b/src/state.rs index 6d95bb5..49bc1a8 100644 --- a/src/state.rs +++ b/src/state.rs @@ -77,16 +77,21 @@ pub(crate) enum HelloState { Established, } +/// A sync message on its way to our Mycelium (the HTTP request is in progress). #[derive(Clone)] -pub(crate) struct PendingOutbound { - pub(crate) message: transport::EncodedMessage, - pub(crate) retry_delay: Duration, - pub(crate) next_attempt: Instant, - /// A sync pass was requested while this was pending. Once it is delivered, the peer - /// must be enqueued again so the deferred changes go out. +pub(crate) struct Sending { + /// Changes it carries for the first time; taken back if the send fails. + pub(crate) sent: Vec<automerge::ChangeHash>, + /// A sync pass was requested meanwhile. Once this one is handed over, the peer must + /// be enqueued again so the deferred changes go out. pub(crate) regenerate: bool, } +/// First wait before generating again after our Mycelium refused a message; doubles +/// while it keeps refusing. +const SEND_RETRY: Duration = Duration::from_millis(500); +const SEND_RETRY_MAX: Duration = Duration::from_secs(30); + #[derive(Clone)] pub(crate) struct PeerState { /// Whether we lead the pair (`swarm::pair`): we send the keepalive PINGs. Fixed for @@ -134,8 +139,14 @@ pub(crate) struct PeerState { pub(crate) struct SyncSession { pub(crate) state: sync::State, - /// Generated for this peer, waiting for delivery. - pub(crate) outbound: Option<PendingOutbound>, + /// A message being handed to our Mycelium. + pub(crate) sending: Option<Sending>, + + /// After our Mycelium refused a message: no new one before this. The refused one is + /// not kept; the next one is generated afresh and carries everything pending. + pub(crate) retry_at: Option<Instant>, + /// The wait after the next refusal. + pub(crate) retry_delay: Duration, /// The answer we expect to our last sync message(s), if any. pub(crate) reply: Option<SyncReplyWait>, @@ -163,7 +174,9 @@ impl SyncSession { pub(crate) fn new() -> Self { Self { state: sync::State::new(), - outbound: None, + sending: None, + retry_at: None, + retry_delay: SEND_RETRY, reply: None, reply_timeout: SYNC_REPLY_TIMEOUT, } @@ -181,6 +194,34 @@ impl SyncSession { self.state.their_heads.is_none() && we_ready && peer_ready == Some(false) } + /// Our Mycelium took the message. + pub(crate) fn send_succeeded(&mut self) -> bool { + self.retry_at = None; + self.retry_delay = SEND_RETRY; + self.sending + .take() + .is_some_and(|sending| sending.regenerate) + } + + /// Our Mycelium refused the message: as far as the peer is concerned, it was never + /// sent. Takes back its changes (they go out with the next message), clears + /// `in_flight`, and returns how long to wait before generating again. None if a reset + /// already discarded what was being sent. + pub(crate) fn send_failed(&mut self, now: Instant) -> Option<Duration> { + let sending = self.sending.take()?; + for hash in &sending.sent { + self.state.sent_hashes.remove(hash); + } + if let Some(wait) = &mut self.reply { + wait.sent.retain(|hash| !sending.sent.contains(hash)); + } + self.state.in_flight = false; + let delay = self.retry_delay; + self.retry_at = Some(now + delay); + self.retry_delay = (delay * 2).min(SEND_RETRY_MAX); + Some(delay) + } + /// A sync message went out; `sent` are the changes it carried for the first time. pub(crate) fn expect_reply(&mut self, sent: Vec<automerge::ChangeHash>, now: Instant) { match &mut self.reply { @@ -240,8 +281,8 @@ impl PeerState { } } - /// Start Automerge sync with this peer from scratch. Any envelope generated against - /// the old state is discarded; a stale RetryDelivery event is then a no-op. + /// Start Automerge sync with this peer from scratch. A message still being handed to + /// Mycelium was generated against the old state; its outcome is then ignored. pub(crate) fn reset_sync(&mut self) { self.sync = SyncSession::new(); } diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index be4b1e2..e05d275 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -4,7 +4,6 @@ use std::collections::HashSet; use std::sync::Arc; use std::sync::atomic::Ordering; -use std::time::Duration; use anyhow::{Context, Result, anyhow, bail}; use automerge::sync::SyncDoc; @@ -12,21 +11,17 @@ use tokio::sync::mpsc; use tokio::time::Instant; use tracing::{debug, info, trace, warn}; -use crate::state::{AppState, Liveness, PendingOutbound, Standing}; +use crate::state::{AppState, Liveness, Sending, Standing}; use crate::swarm::peer::{peer_swarm, send_hello_once}; use crate::swarm::source; 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. @@ -35,7 +30,7 @@ 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<()> { let Some(peer_state_mutex) = state.get_peer_state(peer).await else { trace!("Peer was forgotten; skipping synchronization"); return Ok(()); @@ -65,31 +60,29 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool return Ok(()); }; - if !retry_delivery { - 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 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 + .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(()); } swarm_id.clone() @@ -104,110 +97,109 @@ 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 &mut peer_state.sync.outbound { - Some(entry) if retry_delivery => Some((entry.message.clone(), entry.next_attempt)), - Some(entry) => { - entry.regenerate = 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 ready = state.is_ready().await; + } - 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) - }; - // 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.reply_settled(); - } else if message.is_some() { - let sent = peer_state - .sync - .state - .sent_hashes - .difference(&sent_before) - .copied() - .collect(); - peer_state.sync.expect_reply(sent, Instant::now()); - } - (epoch, message) - }; - let Some(message) = message else { - return Ok(None); - }; - let envelope = transport::SyncEnvelope { - swarm_id, - epoch, - message, - ready, + 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; + + 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()); + } + 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.sync.outbound = Some(PendingOutbound { - message: encoded.clone(), - retry_delay: INITIAL_SYNC_RETRY, - next_attempt: Instant::now(), - regenerate: false, - }); + 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 - .sync - .outbound - .take() - .is_some_and(|entry| entry.regenerate) - }; + let regenerate = peer_state_mutex.lock().await.sync.send_succeeded(); trace!(regenerate, "Outbound sync message accepted by Mycelium"); @@ -221,30 +213,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.sync.outbound 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(()) @@ -405,13 +394,12 @@ fn resend_if_needed(peer: &str, sync_state: &mut automerge::sync::State) { 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) @@ -419,9 +407,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; } @@ -441,9 +426,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; } @@ -469,23 +451,10 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece } } - // 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:#}"), |