summaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/state.rs24
-rw-r--r--src/swarm/sync.rs24
2 files changed, 45 insertions, 3 deletions
diff --git a/src/state.rs b/src/state.rs
index 9c3b569..3fa01c2 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -110,6 +110,9 @@ pub(crate) struct PeerState {
pub(crate) sync_reply: Option<SyncReplyWait>,
/// How long to wait for it; doubles while answers stay out.
pub(crate) sync_reply_timeout: Duration,
+
+ /// When sync with this peer last started from scratch.
+ pub(crate) sync_reset_at: Instant,
}
/// Sync messages we sent and expect an answer to. Automerge assumes reliable delivery:
@@ -128,6 +131,10 @@ const SYNC_REPLY_TIMEOUT: Duration = Duration::from_secs(10);
/// Upper bound (Mycelium gives up on a message after five minutes).
const SYNC_REPLY_TIMEOUT_MAX: Duration = Duration::from_secs(300);
+/// How long the side that doesn't open a fresh sync waits for the other's opening
+/// before opening itself (the other one may not have reset, or its message got lost).
+const SYNC_OPENING_GRACE: Duration = Duration::from_secs(3);
+
impl Default for PeerState {
fn default() -> Self {
Self {
@@ -148,6 +155,7 @@ impl Default for PeerState {
authenticated: false,
sync_reply: None,
sync_reply_timeout: SYNC_REPLY_TIMEOUT,
+ sync_reset_at: Instant::now(),
epoch: None,
}
}
@@ -162,6 +170,22 @@ impl PeerState {
self.needs_generate = false;
self.sync_reply = None;
self.sync_reply_timeout = SYNC_REPLY_TIMEOUT;
+ self.sync_reset_at = Instant::now();
+ }
+
+ /// After a reset, usually both sides do it and both open the sync at once. The
+ /// openings cross, and each side answers the other's: harmless between peers that
+ /// share history, but an empty peer answers with empty heads, which Automerge takes
+ /// for "lost everything", so the whole document goes out a second time. Only the
+ /// side with the smaller key opens; the other waits for its opening, up to
+ /// `SYNC_OPENING_GRACE`. Returns how much longer to wait, if at all.
+ pub(crate) fn opening_wait(&self, we_open: bool, now: Instant) -> Option<Duration> {
+ if we_open || self.sync_state.their_heads.is_some() {
+ return None;
+ }
+ SYNC_OPENING_GRACE
+ .checked_sub(now.saturating_duration_since(self.sync_reset_at))
+ .filter(|wait| !wait.is_zero())
}
/// A sync message went out; `sent` are the changes it carried for the first time.
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 47c3b7f..d82238e 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -58,10 +58,28 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
return Ok(());
}
- match &peer_state.membership {
- PeerMembership::Unknown => return Ok(()),
- PeerMembership::Member { swarm_id } => swarm_id.clone(),
+ let PeerMembership::Member { swarm_id } = &peer_state.membership 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(());
+ }
}
+
+ swarm_id.clone()
};
let swarm_id = state