aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--docs/peers.md13
-rw-r--r--src/document/sync.rs194
-rw-r--r--src/swarm/relay.rs9
-rw-r--r--src/swarm/sync.rs17
-rw-r--r--src/swarm/tree.rs20
-rw-r--r--src/transport/wire.rs2
6 files changed, 245 insertions, 10 deletions
diff --git a/docs/peers.md b/docs/peers.md
index cacceb5..051db4e 100644
--- a/docs/peers.md
+++ b/docs/peers.md
@@ -333,9 +333,16 @@ children in each of their writers' trees, however they came (along a tree, with
answer to one of its own messages, pulled). A sync message along a tree also says
whose; one that had those changes already passes them on too (they may have come
another way first, when it wasn't yet anyone's turn to pass them on), so a child that
-lacks them still gets them. Its announcement round then goes only to the
-neighbours the tree leaves out (no member they reach comes before them, or the member
-list's graph doesn't have them); all others are told in a slow round 2 s later, which
+lacks them still gets them. A message along a tree to a child that had nothing the
+parent lacks needs no answer: the parent takes it as delivered (the child's heads are
+its own now), and the child, level with it, says nothing; a child with changes the
+parent lacks answers with them, as before. That leaves out about half the sync
+messages, Automerge's answers that only say "I have your heads". Should the message
+be lost, the child's other neighbours tell it in their slow rounds (below), and it
+pulls; a child that reaches no other member is told in its parent's. The announcement
+round of a node that passed changes on goes only to the neighbours the tree leaves out
+(no member they reach comes before them, or the member list's graph doesn't have
+them); all others are told in a slow round 2 s later, which
catches what a tree missed (a parent that went away, nodes whose views of the tree
differed) and covers everything that came meanwhile: while changes keep coming, each
neighbour gets an announcement about every 2 s, not with every change.
diff --git a/src/document/sync.rs b/src/document/sync.rs
index 4a3fe5e..c432173 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -46,6 +46,12 @@ pub(crate) struct Sending {
/// 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,
+ /// Taken as delivered (`assume_delivered`): the peer's heads and the shared heads as
+ /// they were before, put back if the send fails.
+ pub(crate) assumed: Option<(
+ Option<Vec<crate::document::ChangeHash>>,
+ Vec<crate::document::ChangeHash>,
+ )>,
}
/// Session IDs, unique within the process.
@@ -165,6 +171,7 @@ impl SyncSession {
self.sending = Some(Sending {
sent,
regenerate: false,
+ assumed: None,
});
}
payload
@@ -227,6 +234,50 @@ impl SyncSession {
})
}
+ /// The message just generated passes a change on along a writer's tree, to a peer
+ /// that had nothing we lack (`have_all_theirs`), and no answer was awaited before it:
+ /// once it has the message, it holds exactly our `heads`, and answers nothing
+ /// (`answered_along_tree`). So it is taken as delivered: the peer's heads are ours,
+ /// and no answer is awaited (about half the sync messages, otherwise, are those
+ /// answers). Should it be lost, the announcements make the peer pull: the slow round
+ /// of its other neighbours, or ours where it reaches no other member
+ /// (`swarm::tree::Plan::alone`); should the peer have changes we lack after all, it
+ /// answers with them, as its heads differ from ours. Undone if our transport refuses
+ /// the message (`send_failed`).
+ pub(crate) fn assume_delivered(&mut self, heads: &[crate::document::ChangeHash]) {
+ let Some(sending) = &mut self.sending else {
+ return;
+ };
+ sending.assumed = Some((
+ self.state.their_heads.clone(),
+ self.state.shared_heads.clone(),
+ ));
+ self.state.their_heads = Some(heads.to_vec());
+ self.state.shared_heads = heads.to_vec();
+ self.state.in_flight = false;
+ self.reply = None;
+ }
+
+ /// We applied a message passing a change on along a writer's tree: if we hold
+ /// exactly its sender's heads now, its sender took it as delivered
+ /// (`assume_delivered`), and Automerge's answer, our heads, would tell it nothing.
+ /// So none goes: we count as having said our `heads`. Returns whether so.
+ pub(crate) fn answered_along_tree(&mut self, heads: &[crate::document::ChangeHash]) -> bool {
+ let mut ours = heads.to_vec();
+ ours.sort_unstable();
+ let theirs = self.state.their_heads.as_ref().map(|theirs| {
+ let mut theirs = theirs.clone();
+ theirs.sort_unstable();
+ theirs
+ });
+ if theirs.as_ref() != Some(&ours) {
+ return false;
+ }
+ self.state.last_sent_heads = heads.to_vec();
+ self.state.have_responded = true;
+ true
+ }
+
/// The peer has heads we lack (its keepalive said so): ask for what they need,
/// even though Automerge thinks a message is still in flight.
pub(crate) fn pull_from(&mut self, heads: Vec<crate::document::ChangeHash>) {
@@ -318,6 +369,10 @@ impl SyncSession {
for hash in &sending.sent {
self.state.sent_hashes.remove(hash);
}
+ if let Some((their_heads, shared_heads)) = sending.assumed {
+ self.state.their_heads = their_heads;
+ self.state.shared_heads = shared_heads;
+ }
if let Some(wait) = &mut self.reply {
wait.sent.retain(|hash| !sending.sent.contains(hash));
}
@@ -471,6 +526,7 @@ mod tests {
Some(Sending {
sent: Vec::new(),
regenerate: false,
+ assumed: None,
})
}
@@ -540,6 +596,144 @@ mod tests {
assert!(carried_again > 4_000, "sent again in {carried_again} bytes");
}
+ /// Two sessions exchanging messages until neither has anything to say.
+ fn in_step(ours: &mut Document, theirs: &mut Document) -> (SyncSession, SyncSession) {
+ let (mut a, mut b) = (session(), session());
+ let (now, bytes) = (Instant::now(), SyncBytes::default());
+ for _ in 0..10 {
+ let to_b = a.generate(ours, now, bytes);
+ a.send_succeeded(a.id());
+ let to_a = b.generate(theirs, now, bytes);
+ b.send_succeeded(b.id());
+ if to_b.is_none() && to_a.is_none() {
+ break;
+ }
+ if let Some(m) = to_b {
+ b.receive(theirs, decode(&m).unwrap()).unwrap();
+ }
+ if let Some(m) = to_a {
+ a.receive(ours, decode(&m).unwrap()).unwrap();
+ }
+ }
+ assert_eq!(ours.get_heads(), theirs.get_heads());
+ (a, b)
+ }
+
+ /// A message along a tree to a peer that had everything else: taken as delivered,
+ /// and not answered; neither side has more to say.
+ #[test]
+ fn tree_messages_need_no_answer() {
+ use crate::document::testing::change;
+ use automerge::{ROOT, transaction::Transactable};
+ let (mut ours, mut theirs) = (Document::new(), Document::new());
+ change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
+ let (mut a, mut b) = in_step(&mut ours, &mut theirs);
+ let (now, bytes) = (Instant::now(), SyncBytes::default());
+
+ change(&mut ours, |doc| doc.put(ROOT, "news", 2).unwrap());
+ assert!(a.have_all_theirs(&ours));
+ let message = a.generate(&ours, now, bytes).unwrap();
+ a.assume_delivered(&ours.get_heads());
+ a.send_succeeded(a.id());
+ assert!(a.reply.is_none(), "no answer awaited");
+
+ b.receive(&mut theirs, decode(&message).unwrap()).unwrap();
+ assert!(b.answered_along_tree(&theirs.get_heads()));
+ assert!(
+ b.generate(&theirs, now, bytes).is_none(),
+ "nothing to answer"
+ );
+ assert!(
+ a.generate(&ours, now, bytes).is_none(),
+ "nothing more to send"
+ );
+ assert_eq!(ours.get_heads(), theirs.get_heads());
+ }
+
+ /// The peer wrote meanwhile: its heads differ from ours after our message, so it
+ /// answers, with its change.
+ #[test]
+ fn tree_messages_are_answered_with_changes_we_lack() {
+ use crate::document::testing::change;
+ use automerge::{ROOT, transaction::Transactable};
+ let (mut ours, mut theirs) = (Document::new(), Document::new());
+ change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
+ let (mut a, mut b) = in_step(&mut ours, &mut theirs);
+ let (now, bytes) = (Instant::now(), SyncBytes::default());
+
+ change(&mut ours, |doc| doc.put(ROOT, "ours", 2).unwrap());
+ change(&mut theirs, |doc| doc.put(ROOT, "theirs", 3).unwrap());
+ let message = a.generate(&ours, now, bytes).unwrap();
+ a.assume_delivered(&ours.get_heads());
+ a.send_succeeded(a.id());
+
+ b.receive(&mut theirs, decode(&message).unwrap()).unwrap();
+ assert!(!b.answered_along_tree(&theirs.get_heads()));
+ let answer = b.generate(&theirs, now, bytes).expect("b answers");
+ b.send_succeeded(b.id());
+ a.receive(&mut ours, decode(&answer).unwrap()).unwrap();
+ assert_eq!(ours.get_heads(), theirs.get_heads());
+ }
+
+ /// A message along a tree is lost, and the next one arrives: the peer can't apply
+ /// it, answers what it needs, and gets the lost change then.
+ #[test]
+ fn a_lost_tree_message_is_asked_for() {
+ use crate::document::testing::change;
+ use automerge::{ROOT, transaction::Transactable};
+ let (mut ours, mut theirs) = (Document::new(), Document::new());
+ change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
+ let (mut a, mut b) = in_step(&mut ours, &mut theirs);
+ let (now, bytes) = (Instant::now(), SyncBytes::default());
+
+ change(&mut ours, |doc| doc.put(ROOT, "lost", 2).unwrap());
+ let _lost = a.generate(&ours, now, bytes).unwrap();
+ a.assume_delivered(&ours.get_heads());
+ a.send_succeeded(a.id());
+ change(&mut ours, |doc| doc.put(ROOT, "next", 3).unwrap());
+ assert!(
+ a.have_all_theirs(&ours),
+ "a takes the lost one as delivered"
+ );
+ let next = a.generate(&ours, now, bytes).unwrap();
+ a.assume_delivered(&ours.get_heads());
+ a.send_succeeded(a.id());
+
+ b.receive(&mut theirs, decode(&next).unwrap()).unwrap();
+ assert!(!b.answered_along_tree(&theirs.get_heads()));
+ let ask = b
+ .generate(&theirs, now, bytes)
+ .expect("b asks for what it lacks");
+ b.send_succeeded(b.id());
+ a.receive(&mut ours, decode(&ask).unwrap()).unwrap();
+ a.resend_if_needed("b");
+ let answer = a.generate(&ours, now, bytes).expect("a sends the lost one");
+ a.send_succeeded(a.id());
+ b.receive(&mut theirs, decode(&answer).unwrap()).unwrap();
+ assert_eq!(ours.get_heads(), theirs.get_heads());
+ }
+
+ /// Our transport refused a message taken as delivered: the peer's heads are as
+ /// before, so the next message carries the change again.
+ #[test]
+ fn a_refused_tree_message_goes_again() {
+ use crate::document::testing::change;
+ use automerge::{ROOT, transaction::Transactable};
+ let (mut ours, mut theirs) = (Document::new(), Document::new());
+ change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
+ let (mut a, mut b) = in_step(&mut ours, &mut theirs);
+ let (now, bytes) = (Instant::now(), SyncBytes::default());
+
+ change(&mut ours, |doc| doc.put(ROOT, "news", 2).unwrap());
+ let _refused = a.generate(&ours, now, bytes).unwrap();
+ a.assume_delivered(&ours.get_heads());
+ assert!(a.send_failed(a.id(), now).is_some());
+ let again = a.generate(&ours, now, bytes).expect("sent again");
+ a.send_succeeded(a.id());
+ b.receive(&mut theirs, decode(&again).unwrap()).unwrap();
+ assert_eq!(ours.get_heads(), theirs.get_heads());
+ }
+
/// A joining node opens, the answer is too large and becomes a delta, the node gets
/// the delta and says so: the answer to that carries our heads, so the node knows it
/// has everything (`have_all_theirs`).
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index f141dbb..98d4ca0 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -126,8 +126,15 @@ pub(crate) async fn announce(state: &AppState, from: &str, writers: &[u64], unau
let slow = slow.get_or_insert_with(Pending::default);
slow.senders.insert(from.to_owned());
slow.all = true;
+ // Our children were sent the changes, but those that reach no member but us
+ // hear of a lost message from nobody else (`Plan::alone`).
for plan in &plans {
- slow.sent.extend(plan.children.iter().cloned());
+ slow.sent.extend(
+ plan.children
+ .iter()
+ .filter(|child| !plan.alone.contains(child))
+ .cloned(),
+ );
}
schedule
};
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index b20f7ef..02c6379 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -148,6 +148,13 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
let replica = doc.blocking_read();
let mut peer_state = psm.blocking_lock();
let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
+ // Whose change it passes on, if any: this message carries it.
+ let writer = peer_state.tree_writer.take();
+ // Along a tree, to a peer that had nothing we lack, with nothing
+ // awaited: taken as delivered (`SyncSession::assume_delivered`).
+ let level = writer.is_some()
+ && peer_state.sync.reply.is_none()
+ && peer_state.sync.have_all_theirs(&replica.doc);
// Under the same locks as the epoch label: the message belongs to
// this document.
let message =
@@ -171,8 +178,9 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
);
transport::SyncBody::Delta(spec)
});
- // Whose change it passes on, if any: this message carries it.
- let writer = peer_state.tree_writer.take();
+ if level && matches!(body, Some(transport::SyncBody::Message(_))) {
+ peer_state.sync.assume_delivered(replica.heads());
+ }
(peer_state.sync.id(), epoch, epoch_seq, body, writer)
};
let Some(body) = body else {
@@ -411,6 +419,11 @@ pub(crate) async fn handle_sync(
peer_state.sync.resend_if_needed(&peer);
replica.refresh_heads();
let heads_after = replica.heads().to_vec();
+ if writer.is_some() {
+ // Level with the sender now: it takes the message as delivered, so
+ // there is nothing to answer (`SyncSession::answered_along_tree`).
+ peer_state.sync.answered_along_tree(&heads_after);
+ }
stats.heads_seen(heads_after.len());
let changed = heads_after != heads_before;
// Whose they are, to pass them on along their writers' trees, however they
diff --git a/src/swarm/tree.rs b/src/swarm/tree.rs
index b21908e..c474754 100644
--- a/src/swarm/tree.rs
+++ b/src/swarm/tree.rs
@@ -16,9 +16,10 @@
//! pair key here): whoever applies changes passes them on to its own children in each of
//! their writers' trees (`forward`), however they came; so does the writer, to all its
//! neighbours, with its own writes. A sync message along a tree also says whose
-//! (`SyncEnvelope::writer`), for one that had them already (`swarm::sync`). A member no
-//! member it reaches comes before (or whose parent didn't get the change) gets it from
-//! the announcements (`swarm::relay`), as before.
+//! (`SyncEnvelope::writer`), for one that had them already (`swarm::sync`), and needs no
+//! answer where the child had nothing the parent lacks (`SyncSession::assume_delivered`).
+//! A member no member it reaches comes before (or whose parent didn't get the change)
+//! gets it from the announcements (`swarm::relay`), as before.
//!
//! The graph is computed once per member list (tens of milliseconds for thousands of
//! members), the distances once per writer and graph (`MAX_WRITERS` kept).
@@ -176,6 +177,10 @@ pub(crate) fn whole(
/// out (`uncovered`).
pub(crate) struct Plan {
pub(crate) children: Vec<String>,
+ /// Those of our children that reach no member but us, as they said: nobody else's
+ /// slow round tells them of a change, should our message to them be lost (it is
+ /// taken as delivered, `SyncSession::assume_delivered`), so ours does.
+ pub(crate) alone: Vec<String>,
/// Our neighbours that no member they reach comes before (the tree reaches them only
/// by announcement), and those not in the member list's graph (whom no tree has).
pub(crate) uncovered: Vec<String>,
@@ -205,7 +210,9 @@ pub(crate) async fn plan(state: &AppState, writer: u64) -> Option<Plan> {
.is_none_or(|keys| keys.contains(&topology::pair_key(member)))
};
let my_rank = rank(&graph, &distance, &reached, me)?;
+ let my_key = topology::pair_key(me);
let mut children = Vec::new();
+ let mut alone = Vec::new();
let mut uncovered = Vec::new();
for member in graph.get(me).into_iter().flatten() {
let Some(its_rank) = rank(&graph, &distance, &reached, member) else {
@@ -219,6 +226,12 @@ pub(crate) async fn plan(state: &AppState, writer: u64) -> Option<Plan> {
Some(parent)
if parent == me && its_rank > my_rank && topology::reaches(state, member).await =>
{
+ if reaches
+ .get(member.as_str())
+ .is_some_and(|keys| keys.iter().all(|key| *key == my_key))
+ {
+ alone.push(member.clone());
+ }
children.push(member.clone());
}
Some(_) => {}
@@ -235,6 +248,7 @@ pub(crate) async fn plan(state: &AppState, writer: u64) -> Option<Plan> {
}
Some(Plan {
children,
+ alone,
uncovered,
})
}
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 67f6de0..899454c 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage};
/// The wire format's version, the first byte of every payload. Bump it with every change
/// to a message type or a codec (`golden_bytes` fails then).
-pub(crate) const VERSION: u8 = 23;
+pub(crate) const VERSION: u8 = 24;
/// What the second byte of a payload says.
fn tag(kind: MessageKind) -> u8 {