aboutsummaryrefslogtreecommitdiffstats
path: root/src/document/sync.rs
diff options
context:
space:
mode:
authortv <tv@krebsco.de>2026-11-04 23:09:04 +0100
committertv <tv@krebsco.de>2026-11-04 23:09:04 +0100
commit2ee5211c855a63d29ef85307ab925ac97de76b02 (patch)
tree9054051de6689adf1bfbe558f6492266c573b78e /src/document/sync.rs
parenta68339a2d47982b432ef09485d2a15478ccbeb5f (diff)
offered deltas lapse
Diffstat (limited to 'src/document/sync.rs')
-rw-r--r--src/document/sync.rs41
1 files changed, 38 insertions, 3 deletions
diff --git a/src/document/sync.rs b/src/document/sync.rs
index 6eb791e..e921b22 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -267,7 +267,15 @@ impl SyncSession {
/// message being handed over becomes the offer (it carries nothing); the answer is
/// still awaited, and if it doesn't come, the offer is made again. Returns the
/// peer's heads as far as we know them: where the delta starts.
- pub(crate) fn withdraw_for_delta(&mut self) -> Vec<crate::document::ChangeHash> {
+ ///
+ /// An answer is awaited even where none was (the peer's heads were ours: `generate`),
+ /// so the offer lapses at the reply timeout at the latest: nothing else ends it but a
+ /// message from the peer, and one that has everything has nothing to say. Without
+ /// that, no message would go to it again.
+ pub(crate) fn withdraw_for_delta(&mut self, now: Instant) -> Vec<crate::document::ChangeHash> {
+ if self.reply.is_none() {
+ self.expect_reply(Vec::new(), now, SyncBytes::default(), 0);
+ }
if let Some(sending) = &mut self.sending {
let sent = std::mem::take(&mut sending.sent);
for hash in &sent {
@@ -283,6 +291,15 @@ impl SyncSession {
self.state.their_heads.clone().unwrap_or_default()
}
+ /// The message just generated, withdrawn for a delta (`withdraw_for_delta`), turns
+ /// out to carry nothing the peer lacks (it has our heads): none goes, and nothing is
+ /// offered or awaited.
+ pub(crate) fn nothing_to_send(&mut self) {
+ self.sending = None;
+ self.delta_offered = false;
+ self.reply_settled();
+ }
+
/// Whether `doc` holds everything the peer had when it last told us its heads: they
/// are known, not empty (empty heads are "contained" in every document, so they
/// prove nothing), and all in `doc`.
@@ -712,6 +729,24 @@ mod tests {
assert!(new.sending.is_none());
}
+ /// A delta offered where no answer was awaited (the peer's heads were ours) still
+ /// lapses at the reply timeout: else nothing ends the wait for a peer with nothing
+ /// to say, and no message goes to it again.
+ #[test]
+ fn an_offered_delta_always_lapses() {
+ let mut a = session();
+ a.sending = sending();
+ assert!(a.reply.is_none());
+ let now = Instant::now();
+ a.withdraw_for_delta(now);
+ assert!(a.delta_offered);
+ assert!(
+ a.take_overdue_reply(now + Duration::from_secs(3600))
+ .is_some()
+ );
+ assert!(!a.delta_offered);
+ }
+
/// A message too large to send is taken back: the peer, after the offer and an
/// overdue answer, gets the same changes in the next message.
#[test]
@@ -743,7 +778,7 @@ mod tests {
let carried = carrying.0.len();
assert!(carried > 4_000, "the changes took {carried} bytes");
- let from = a.withdraw_for_delta();
+ let from = a.withdraw_for_delta(now);
assert!(from.is_empty());
assert!(a.delta_offered);
assert!(!a.send_succeeded(a.id()));
@@ -1045,7 +1080,7 @@ mod tests {
b.send_succeeded(b.id());
a.receive(&mut ours, decode(&opening).unwrap()).unwrap();
assert!(a.generate(&ours, now, bytes).is_some());
- a.withdraw_for_delta();
+ a.withdraw_for_delta(now);
a.send_succeeded(a.id());
// b loads the delta and tells a its heads, which are a's now.