diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/document/sync.rs | 108 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 31 | ||||
| -rw-r--r-- | src/swarm/tree.rs | 5 | ||||
| -rw-r--r-- | src/transport/mod.rs | 5 | ||||
| -rw-r--r-- | src/transport/wire.rs | 3 |
5 files changed, 112 insertions, 40 deletions
diff --git a/src/document/sync.rs b/src/document/sync.rs index c432173..14adca7 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -234,16 +234,19 @@ 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`). + /// The message just generated goes 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_as_delivered`; the + /// envelope says so, `SyncEnvelope::delivered`). So it is taken as delivered: the + /// peer's heads are ours, and no answer is awaited (otherwise about half the sync + /// messages are those answers: along trees, to pulls, to what is sent again). Should + /// it be lost, the peer still lacks it: where it asked (a pull), it asks again after + /// its own reply timeout; otherwise the announcements make it pull (the slow round of + /// its other neighbours, or ours where it reaches no other member: + /// `swarm::tree::Plan::alone`), or our next message, which it can't apply, makes it + /// ask. 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; @@ -258,11 +261,11 @@ impl SyncSession { 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 { + /// We applied a message its sender took as delivered (`assume_delivered`, + /// `SyncEnvelope::delivered`): if we hold exactly its sender's heads now, 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_as_delivered(&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| { @@ -619,10 +622,10 @@ mod tests { (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. + /// A message 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() { + fn delivered_messages_need_no_answer() { use crate::document::testing::change; use automerge::{ROOT, transaction::Transactable}; let (mut ours, mut theirs) = (Document::new(), Document::new()); @@ -638,7 +641,7 @@ mod tests { 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.answered_as_delivered(&theirs.get_heads())); assert!( b.generate(&theirs, now, bytes).is_none(), "nothing to answer" @@ -653,7 +656,7 @@ mod tests { /// 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() { + fn delivered_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()); @@ -668,17 +671,17 @@ mod tests { a.send_succeeded(a.id()); b.receive(&mut theirs, decode(&message).unwrap()).unwrap(); - assert!(!b.answered_along_tree(&theirs.get_heads())); + assert!(!b.answered_as_delivered(&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. + /// A message taken as delivered 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() { + fn a_lost_delivered_message_is_asked_for() { use crate::document::testing::change; use automerge::{ROOT, transaction::Transactable}; let (mut ours, mut theirs) = (Document::new(), Document::new()); @@ -700,7 +703,7 @@ mod tests { a.send_succeeded(a.id()); b.receive(&mut theirs, decode(&next).unwrap()).unwrap(); - assert!(!b.answered_along_tree(&theirs.get_heads())); + assert!(!b.answered_as_delivered(&theirs.get_heads())); let ask = b .generate(&theirs, now, bytes) .expect("b asks for what it lacks"); @@ -713,10 +716,63 @@ mod tests { assert_eq!(ours.get_heads(), theirs.get_heads()); } + /// Two messages taken as delivered arrive in reverse order: the peer asks for the + /// first, which arrives meanwhile, and what is sent again is taken as delivered too. + /// Nobody waits for an answer that won't come. + #[test] + fn reordered_delivered_messages_leave_nothing_awaited() { + 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()); + let send = |a: &mut SyncSession, ours: &Document| { + assert!(a.reply.is_none() && a.have_all_theirs(ours)); + let message = a.generate(ours, now, bytes).unwrap(); + a.assume_delivered(&ours.get_heads()); + a.send_succeeded(a.id()); + message + }; + + change(&mut ours, |doc| doc.put(ROOT, "first", 2).unwrap()); + let first = send(&mut a, &ours); + change(&mut ours, |doc| doc.put(ROOT, "second", 3).unwrap()); + let second = send(&mut a, &ours); + + b.receive(&mut theirs, decode(&second).unwrap()).unwrap(); + assert!(!b.answered_as_delivered(&theirs.get_heads())); + let ask = b + .generate(&theirs, now, bytes) + .expect("b asks for the first"); + b.send_succeeded(b.id()); + b.receive(&mut theirs, decode(&first).unwrap()).unwrap(); + assert_eq!(theirs.get_heads(), ours.get_heads(), "b has both now"); + // The first says older heads than b's now: b tells its own, once. + assert!(!b.answered_as_delivered(&theirs.get_heads())); + let level = b.generate(&theirs, now, bytes); + b.send_succeeded(b.id()); + + a.receive(&mut ours, decode(&ask).unwrap()).unwrap(); + a.resend_if_needed("b"); + let again = send(&mut a, &ours); + if let Some(level) = level { + a.receive(&mut ours, decode(&level).unwrap()).unwrap(); + } + b.receive(&mut theirs, decode(&again).unwrap()).unwrap(); + assert!(b.answered_as_delivered(&theirs.get_heads())); + assert!( + b.generate(&theirs, now, bytes).is_none(), + "nothing to answer" + ); + assert!(a.reply.is_none(), "nothing awaited"); + 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() { + fn a_refused_delivered_message_goes_again() { use crate::document::testing::change; use automerge::{ROOT, transaction::Transactable}; let (mut ours, mut theirs) = (Document::new(), Document::new()); diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 02c6379..5be0754 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -144,16 +144,15 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { move || -> (u64, Result<Option<transport::EncodedMessage>>) { // Label the message with the epoch of the document it was generated // from, read under the same lock. - let (session, epoch, epoch_seq, body, writer) = { + let (session, epoch, epoch_seq, body, writer, delivered) = { 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() + // To a peer that had nothing we lack, with nothing awaited: taken + // as delivered (`SyncSession::assume_delivered`). + let level = 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. @@ -178,10 +177,18 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { ); transport::SyncBody::Delta(spec) }); - if level && matches!(body, Some(transport::SyncBody::Message(_))) { + let delivered = level && matches!(body, Some(transport::SyncBody::Message(_))); + if delivered { peer_state.sync.assume_delivered(replica.heads()); } - (peer_state.sync.id(), epoch, epoch_seq, body, writer) + ( + peer_state.sync.id(), + epoch, + epoch_seq, + body, + writer, + delivered, + ) }; let Some(body) = body else { return (session, Ok(None)); @@ -193,6 +200,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> { body, ready, writer, + delivered, }; let encoded = transport .encode(&peer_owned, &ParsedMessage::Sync(envelope)) @@ -353,6 +361,7 @@ pub(crate) async fn handle_sync( let epoch_label = envelope.epoch; let from_ready = envelope.ready; let writer = envelope.writer; + let delivered = envelope.delivered; let payload = match envelope.body { transport::SyncBody::Message(payload) => payload, transport::SyncBody::Delta(spec) => { @@ -419,10 +428,10 @@ 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); + if delivered { + // Level with the sender now, which takes the message as delivered: + // nothing to answer (`SyncSession::answered_as_delivered`). + peer_state.sync.answered_as_delivered(&heads_after); } stats.heads_seen(heads_after.len()); let changed = heads_after != heads_before; diff --git a/src/swarm/tree.rs b/src/swarm/tree.rs index c474754..c6baff9 100644 --- a/src/swarm/tree.rs +++ b/src/swarm/tree.rs @@ -16,8 +16,9 @@ //! 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`), and needs no -//! answer where the child had nothing the parent lacks (`SyncSession::assume_delivered`). +//! (`SyncEnvelope::writer`), for one that had them already (`swarm::sync`); like any +//! sync message, it 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. //! diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 17d6d4c..426ef72 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -666,6 +666,11 @@ pub(crate) struct SyncEnvelope { /// the writer's, on its own push and on every member's after; None otherwise /// (answers, pulls). pub(crate) writer: Option<u64>, + /// The sender takes the message as delivered (`SyncSession::assume_delivered`): it + /// had everything the receiver had, so the receiver will hold exactly the sender's + /// heads, and the sender awaits no answer. A receiver it leaves level with the + /// sender answers nothing; one that has changes the sender lacks answers with them. + pub(crate) delivered: bool, } /// What a sync envelope carries. diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 899454c..4199927 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 = 24; +pub(crate) const VERSION: u8 = 25; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -410,6 +410,7 @@ pub(crate) fn samples() -> Vec<ParsedMessage> { }), ready: true, writer: Some(7), + delivered: true, }), ParsedMessage::Status(status.clone()), ParsedMessage::StatusReply(status), |