diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/sync.rs | 31 | ||||
| -rw-r--r-- | src/swarm/tree.rs | 5 |
2 files changed, 23 insertions, 13 deletions
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. //! |