diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/peer/keepalive.rs | 4 | ||||
| -rw-r--r-- | src/swarm/relay.rs | 30 |
2 files changed, 20 insertions, 14 deletions
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs index 3b78b85..d698585 100644 --- a/src/swarm/peer/keepalive.rs +++ b/src/swarm/peer/keepalive.rs @@ -250,12 +250,14 @@ pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport ), } } + // One grace wait per peer: a STATUS meanwhile only becomes the one it ends with. + let waiting = peer_state.pull_pending.replace(remote.clone()).is_some(); drop(peer_state); if left { info!(peer = %peer, "Peer said farewell; dormant until it is heard from again"); } // Off the consumer: telling which heads we lack takes the document's lock. - { + if !waiting { let (state, peer, remote) = (state.clone(), peer.to_owned(), remote.clone()); tokio::spawn(async move { crate::swarm::relay::pull_after_grace(&state, &peer, &remote).await; diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index 31df0d3..0f8a697 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -325,21 +325,25 @@ pub(crate) async fn lacks(state: &AppState, heads: &HeadsSummary) -> bool { /// most; at once if we know none). While writes go on, the newest of them are usually /// on their way to us along their writers' trees, as fast as the STATUS: they come /// meanwhile, and pulling them would only fetch them twice. What is still missing then -/// was missed. +/// was missed. One wait per peer: a STATUS that comes meanwhile is the one judged at the +/// end (`PeerState::pull_pending`). pub(crate) async fn pull_after_grace(state: &AppState, peer: &str, remote: &transport::Status) { - if HeadsSummary::of(state.stamp.borrow().heads.to_vec()) == remote.heads { - return; - } - let grace = state - .lock_rtt() - .effective(peer, Instant::now(), state.timing.rtt_fresh_for()) - .map(|rtt| rtt.min(PULL_GRACE_MAX)); - if let Some(grace) = grace - && lacks(state, &remote.heads).await - { - tokio::time::sleep(grace).await; + if HeadsSummary::of(state.stamp.borrow().heads.to_vec()) != remote.heads { + let grace = state + .lock_rtt() + .effective(peer, Instant::now(), state.timing.rtt_fresh_for()) + .map(|rtt| rtt.min(PULL_GRACE_MAX)); + if let Some(grace) = grace + && lacks(state, &remote.heads).await + { + tokio::time::sleep(grace).await; + } } - pull_if_behind(state, peer, remote).await; + let latest = match state.get_peer_state(peer).await { + Some(peer_state) => peer_state.lock().await.pull_pending.take(), + None => None, + }; + pull_if_behind(state, peer, latest.as_ref().unwrap_or(remote)).await; } /// If `peer` has heads we lack (as its STATUS `remote` says), pull them (one peer at a |