aboutsummaryrefslogtreecommitdiffstats
path: root/src/document/sync.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/document/sync.rs')
-rw-r--r--src/document/sync.rs23
1 files changed, 16 insertions, 7 deletions
diff --git a/src/document/sync.rs b/src/document/sync.rs
index 2a24007..2119eda 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -95,6 +95,9 @@ pub(crate) struct SyncReplyWait {
pub(crate) started: (Instant, SyncBytes),
/// Changes first sent in the unanswered messages.
pub(crate) sent: Vec<crate::document::ChangeHash>,
+ /// The largest of them, in bytes: if they go unanswered, messages that large may
+ /// not make it to the peer whole (`PeerState::whole_failed`).
+ pub(crate) largest: usize,
}
impl SyncSession {
@@ -143,12 +146,13 @@ impl SyncSession {
.difference(&sent_before)
.copied()
.collect();
+ let payload = message.map(|message| SyncPayload(message.encode().into()));
if in_step {
self.reply_settled();
- } else if message.is_some() {
- self.expect_reply(sent.clone(), now, bytes);
+ } else if let Some(payload) = &payload {
+ self.expect_reply(sent.clone(), now, bytes, payload.0.len());
}
- if message.is_some() {
+ if payload.is_some() {
// Marked together with the generation, so no second message is generated
// while this one is handed over.
self.sending = Some(Sending {
@@ -156,7 +160,7 @@ impl SyncSession {
regenerate: false,
});
}
- message.map(|message| SyncPayload(message.encode().into()))
+ payload
}
/// Applies a sync message from the peer to `doc`. Doesn't settle the awaited
@@ -303,22 +307,27 @@ impl SyncSession {
Some(delay)
}
- /// A sync message went out; `sent` are the changes it carried for the first time.
- /// `bytes` are the sync bytes exchanged with the peer before it.
+ /// A sync message of `len` bytes went out; `sent` are the changes it carried for the
+ /// first time. `bytes` are the sync bytes exchanged with the peer before it.
pub(crate) fn expect_reply(
&mut self,
sent: Vec<crate::document::ChangeHash>,
now: Instant,
bytes: SyncBytes,
+ len: usize,
) {
match &mut self.reply {
// Still waiting for an earlier one: keep its deadline, remember these too.
- Some(wait) => wait.sent.extend(sent),
+ Some(wait) => {
+ wait.sent.extend(sent);
+ wait.largest = wait.largest.max(len);
+ }
None => {
self.reply = Some(SyncReplyWait {
deadline: now + self.reply_backoff.current(),
started: (now, bytes),
sent,
+ largest: len,
});
}
}