diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/relay.rs | 9 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 17 | ||||
| -rw-r--r-- | src/swarm/tree.rs | 20 |
3 files changed, 40 insertions, 6 deletions
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, }) } |