diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/sync.rs | 47 |
1 files changed, 46 insertions, 1 deletions
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 594e5de..5819bb7 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -839,9 +839,32 @@ async fn download_and_apply_delta( .context("Delta apply task panicked")? } +/// Our own changes since `seen_heads` (which become the heads now), if what changed +/// is a write of ours; None if none of it is ours (a change fetched as a blob), or the +/// document was replaced (an epoch switch, a rejoin: its heads are no longer ours). +async fn own_changes( + state: &AppState, + seen_heads: &mut Vec<crate::document::ChangeHash>, +) -> Option<Vec<crate::document::ChangeHash>> { + let replica = state.doc.read().await; + let before = std::mem::replace(seen_heads, replica.heads().to_vec()); + if !crate::document::changes::holds_all(&replica.doc, &before) { + return None; + } + let since = crate::document::changes::since(&replica.doc, &before); + since + .authors + .into_iter() + .find(|(author, _)| author == state.local_id()) + .map(|(_, hashes)| hashes) +} + pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedReceiver<SyncEvent>) { let mut dirty_peers = HashSet::<String>::new(); let mut document_dirty = false; + // The heads as of the last document change looked at: what came since, and which of + // it is ours (`own_changes`). + let mut seen_heads = state.doc.read().await.heads().to_vec(); info!("Starting the sync worker"); @@ -893,11 +916,33 @@ pub(crate) async fn run_sync_worker(state: AppState, mut rx: mpsc::UnboundedRece // Our own changes go to our neighbours, saying they're ours: they pass them // on along our tree (`swarm::tree`). Others only get answers to what they - // send us. + // send us. But neighbours that get them from another parent in our tree (a + // sibling they have a much faster link to: `tree::parent`) count as having + // them, as in `tree::forward`, and get nothing now. let own = crate::swarm::topology::pair_key(state.local_id()); + let ours = own_changes(&state, &mut seen_heads).await; + let covered = match &ours { + Some(ours) => match crate::swarm::tree::plan(&state, own).await { + Some(plan) => { + for neighbour in &plan.covered { + if let Some(peer_state) = state.get_peer_state(neighbour).await { + peer_state + .lock() + .await + .sync + .known_to_peer(ours, Instant::now()); + } + } + plan.covered + } + None => Vec::new(), + }, + None => Vec::new(), + }; for peer in peer_keys { if peer_swarm(&state, &peer).await.is_some() && crate::swarm::topology::is_neighbour(&state, &peer) + && !covered.contains(&peer) { if let Some(peer_state) = state.get_peer_state(&peer).await { peer_state.lock().await.tree_writer = Some(own); |
