1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
|
//! One Automerge sync with a peer, from a reset to the next.
use std::time::Duration;
use automerge::sync;
use tokio::time::Instant;
use crate::backoff::Backoff;
use crate::swarm::throughput::SyncBytes;
/// A sync message on its way to our Mycelium (the HTTP request is in progress).
#[derive(Clone)]
pub(crate) struct Sending {
/// Changes it carries for the first time; taken back if the send fails.
pub(crate) sent: Vec<automerge::ChangeHash>,
/// A sync pass was requested meanwhile. Once this one is handed over, the peer must
/// be enqueued again so the deferred changes go out.
pub(crate) regenerate: bool,
}
/// First wait before generating again after our Mycelium refused a message; doubles
/// while it keeps refusing.
const SEND_RETRY: Duration = Duration::from_millis(500);
const SEND_RETRY_MAX: Duration = Duration::from_secs(30);
/// One Automerge sync with a peer, from a reset to the next. Everything a reset must
/// forget lives here, so a reset replaces it whole and can't miss a field.
#[derive(Clone)]
pub(crate) struct SyncSession {
pub(crate) state: sync::State,
/// A message being handed to our Mycelium.
pub(crate) sending: Option<Sending>,
/// After our Mycelium refused a message: no new one before this. The refused one is
/// not kept; the next one is generated afresh and carries everything pending.
pub(crate) retry_at: Option<Instant>,
/// The wait after the next refusal.
pub(crate) send_backoff: Backoff,
/// The answer we expect to our last sync message(s), if any.
pub(crate) reply: Option<SyncReplyWait>,
/// How long to wait for it; doubles while answers stay out.
pub(crate) reply_backoff: Backoff,
}
/// Sync messages we sent and expect an answer to. Automerge assumes reliable delivery:
/// a lost message leaves both sides waiting (`in_flight`), and changes it carried are
/// never sent again (`sent_hashes`). We resend after a timeout instead.
#[derive(Clone, Debug)]
pub(crate) struct SyncReplyWait {
pub(crate) deadline: Instant,
/// When the first of the unanswered messages was generated, and the sync bytes
/// exchanged with the peer by then (`swarm::throughput`).
pub(crate) started: (Instant, SyncBytes),
/// Changes first sent in the unanswered messages.
pub(crate) sent: Vec<automerge::ChangeHash>,
}
impl SyncSession {
/// `reply_backoff`: waits for sync answers (`Timing::sync_reply_backoff`).
pub(crate) fn new(reply_backoff: Backoff) -> Self {
Self {
state: sync::State::new(),
sending: None,
retry_at: None,
send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX),
reply: None,
reply_backoff,
}
}
/// Whether we leave the first message of this session to the peer.
///
/// After a reset, both sides usually open at once, and the openings cross. Between
/// peers that share history that costs nothing. But a peer with an empty document
/// answers our opening with empty heads, which Automerge takes for "lost
/// everything", so the whole document would go out a second time. So a ready node
/// never opens towards a peer that isn't ready (one that is still catching up); that
/// peer opens itself.
pub(crate) fn leave_opening(&self, we_ready: bool, peer_ready: Option<bool>) -> bool {
self.state.their_heads.is_none() && we_ready && peer_ready == Some(false)
}
/// Our Mycelium took the message.
pub(crate) fn send_succeeded(&mut self) -> bool {
self.retry_at = None;
self.send_backoff.reset();
self.sending
.take()
.is_some_and(|sending| sending.regenerate)
}
/// Our Mycelium refused the message: as far as the peer is concerned, it was never
/// sent. Takes back its changes (they go out with the next message), clears
/// `in_flight`, and returns how long to wait before generating again. None if a reset
/// already discarded what was being sent.
pub(crate) fn send_failed(&mut self, now: Instant) -> Option<Duration> {
let sending = self.sending.take()?;
for hash in &sending.sent {
self.state.sent_hashes.remove(hash);
}
if let Some(wait) = &mut self.reply {
wait.sent.retain(|hash| !sending.sent.contains(hash));
}
self.state.in_flight = false;
let delay = self.send_backoff.next();
self.retry_at = Some(now + delay);
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.
pub(crate) fn expect_reply(
&mut self,
sent: Vec<automerge::ChangeHash>,
now: Instant,
bytes: SyncBytes,
) {
match &mut self.reply {
// Still waiting for an earlier one: keep its deadline, remember these too.
Some(wait) => wait.sent.extend(sent),
None => {
self.reply = Some(SyncReplyWait {
deadline: now + self.reply_backoff.current(),
started: (now, bytes),
sent,
});
}
}
}
/// No answer is due any more: the peer answered, or keepalives show both documents
/// are the same.
pub(crate) fn reply_settled(&mut self) {
self.reply = None;
self.reply_backoff.reset();
}
/// 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.
pub(super) fn take_overdue_reply(&mut self, now: Instant) -> Option<usize> {
if self.reply.as_ref().is_none_or(|wait| now < wait.deadline) {
return None;
}
let wait = self.reply.take()?;
for hash in &wait.sent {
self.state.sent_hashes.remove(hash);
}
self.state.in_flight = false;
self.reply_backoff.grow();
Some(wait.sent.len())
}
}
|
