diff options
| author | tv <tv@krebsco.de> | 2026-11-05 15:59:35 +0100 |
|---|---|---|
| committer | tv <tv@krebsco.de> | 2026-11-05 15:59:35 +0100 |
| commit | 12fae75f850c3ca63a68378b2684b5d89f667e10 (patch) | |
| tree | 11f229293dd21b7279234daa9f0cc86e3fc01b43 /src | |
| parent | 13476a8cf598770b501cf0b02a26725a6f5ce3d5 (diff) | |
offer answers
Diffstat (limited to 'src')
| -rw-r--r-- | src/document/sync.rs | 10 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 11 |
2 files changed, 21 insertions, 0 deletions
diff --git a/src/document/sync.rs b/src/document/sync.rs index 7695b82..c703300 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -870,6 +870,16 @@ impl SyncSession { self.reply_backoff.reset(); } + /// The peer answered our message with a delta offer (`SyncBody::Delta`): the wait + /// ends, and to Automerge no message of ours is in flight any more, as an answer of + /// its own would have told it. Else, with our heads unchanged (an offer we don't + /// fetch, or a download that fails), it would say nothing again, and the peer would + /// never hear what we have to offer afresh. + pub(crate) fn answered_by_offer(&mut self) { + self.reply_settled(); + self.state.in_flight = false; + } + /// If the answer is overdue: forgets that the unanswered changes were sent and that /// a message is in flight, so the next sync pass sends them again. Returns how many /// changes that concerns. diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index e551cb8..0edc277 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -487,6 +487,17 @@ pub(crate) async fn handle_sync( ) } transport::SyncBody::Delta(spec) => { + // An offer answers what we sent as much as a message would: the wait for an + // answer ends here, not when the download does (its reply timeout would + // otherwise count against a peer that answered, suspect its path and give it + // up as the source to catch up from). What it offers is what it has. + { + let mut peer_state = peer_state_mutex.lock().await; + peer_state.reports_ready = Some(from_ready); + peer_state.last_heads = Some((epoch, transport::HeadsSummary::of(spec.to.clone()))); + peer_state.last_sync_rx = Some(Instant::now()); + peer_state.sync.answered_by_offer(); + } fetch_delta(state, peer, epoch, spec); return Ok(()); } |
