diff options
| -rw-r--r-- | src/document/changes.rs | 61 | ||||
| -rw-r--r-- | src/document/sync.rs | 17 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 20 |
3 files changed, 61 insertions, 37 deletions
diff --git a/src/document/changes.rs b/src/document/changes.rs index 9a07fc0..7ee1a4b 100644 --- a/src/document/changes.rs +++ b/src/document/changes.rs @@ -43,39 +43,58 @@ pub(crate) fn missing(doc: &impl ReadDoc, heads: &[ChangeHash]) -> Vec<ChangeHas .collect() } -/// When the changes after `since` were committed, in Unix milliseconds by their -/// writers' clocks (`Replica::transact`); changes without a time (snapshots) left out. -pub(crate) fn commit_times(doc: &Document, since: &[ChangeHash]) -> Vec<i64> { - doc.get_changes(since) - .iter() - .map(|change| change.timestamp()) - .filter(|&time| time > 0) - .collect() +/// The changes after `since`, walked once: their hashes; who wrote them (the authors +/// their writers set, a node's ID, `replica::authored`), each once, with the hashes of +/// their changes, and whether any has none (a snapshot, or a change of a node that set +/// none); and when they were committed, in Unix milliseconds by their writers' clocks +/// (`Replica::transact`), changes without a time (snapshots) left out. +pub(crate) struct Since { + pub(crate) hashes: Vec<ChangeHash>, + pub(crate) authors: Vec<(String, Vec<ChangeHash>)>, + pub(crate) unauthored: bool, + pub(crate) times: Vec<i64>, } -/// Who wrote the changes after `since`: the authors their writers set (a node's ID, -/// `replica::authored`), each once, with the hashes of their changes; and whether any -/// change has none (a snapshot, or a change of a node that set none). -pub(crate) fn authors_since( - doc: &Document, - since: &[ChangeHash], -) -> (Vec<(String, Vec<ChangeHash>)>, bool) { - let mut authors: Vec<(String, Vec<ChangeHash>)> = Vec::new(); - let mut unknown = false; +pub(crate) fn since(doc: &Document, since: &[ChangeHash]) -> Since { + let mut found = Since { + hashes: Vec::new(), + authors: Vec::new(), + unauthored: false, + times: Vec::new(), + }; for change in doc.get_changes_meta(since) { + found.hashes.push(change.hash); + if change.timestamp > 0 { + found.times.push(change.timestamp); + } match change .author .as_ref() .and_then(|author| std::str::from_utf8(author.as_bytes()).ok()) { - Some(author) => match authors.iter_mut().find(|(known, _)| known == author) { + Some(author) => match found.authors.iter_mut().find(|(known, _)| known == author) { Some((_, hashes)) => hashes.push(change.hash), - None => authors.push((author.to_owned(), vec![change.hash])), + None => found.authors.push((author.to_owned(), vec![change.hash])), }, - None => unknown = true, + None => found.unauthored = true, } } - (authors, unknown) + found +} + +/// When the changes after `since` were committed (`Since::times`). +pub(crate) fn commit_times(doc: &Document, since: &[ChangeHash]) -> Vec<i64> { + self::since(doc, since).times +} + +/// Who wrote the changes after `since` (`Since::authors`, `Since::unauthored`). +#[cfg(test)] +pub(crate) fn authors_since( + doc: &Document, + since: &[ChangeHash], +) -> (Vec<(String, Vec<ChangeHash>)>, bool) { + let found = self::since(doc, since); + (found.authors, found.unauthored) } /// The change `hash` as the document encodes it; None if `doc` doesn't have it. diff --git a/src/document/sync.rs b/src/document/sync.rs index bfa0fef..688d455 100644 --- a/src/document/sync.rs +++ b/src/document/sync.rs @@ -564,11 +564,18 @@ impl SyncSession { doc: &Document, before: &[crate::document::ChangeHash], ) -> usize { - let changes = doc.get_changes_meta(before); - for change in &changes { - self.state.sent_hashes.insert(change.hash); - } - changes.len() + let hashes: Vec<_> = doc + .get_changes_meta(before) + .iter() + .map(|c| c.hash) + .collect(); + self.these_came_from_peer(&hashes) + } + + /// `came_from_peer`, for the changes `hashes` (found already). + pub(crate) fn these_came_from_peer(&mut self, hashes: &[crate::document::ChangeHash]) -> usize { + self.state.sent_hashes.extend(hashes.iter().copied()); + hashes.len() } /// Of the changes in `doc` beyond `from`, how many the peer has too, as far as we diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index a6467a7..c969678 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -610,9 +610,12 @@ pub(crate) async fn handle_sync( } stats.heads_seen(heads_after.len()); let changed = heads_after != heads_before; - if changed { + // What it brought, walked once. + let since = + changed.then(|| crate::document::changes::since(&replica.doc, &heads_before)); + if let Some(since) = &since { // They came from it: none goes back (`SyncSession::came_from_peer`). - let new = peer_state.sync.came_from_peer(&replica.doc, &heads_before); + let new = peer_state.sync.these_came_from_peer(&since.hashes); stats.new_changes(new); } if carried && !changed { @@ -626,20 +629,15 @@ pub(crate) async fn handle_sync( } // Whose they are, to pass them on along their writers' trees, however they // came (along a tree, with an answer, pulled). - let authors = if changed { - crate::document::changes::authors_since(&replica.doc, &heads_before) - } else { - (Vec::new(), false) + let (authors, times) = match since { + Some(since) => ((since.authors, since.unauthored), since.times), + None => ((Vec::new(), false), Vec::new()), }; if changed && measure_propagation { // Now by the peers' clocks (their median): this node's own clock skew // doesn't count; the writer's still does. let now_ms = clock.now_millis().as_u64() as i64 - ahead_ms; - let delays: Vec<i64> = - crate::document::changes::commit_times(&replica.doc, &heads_before) - .into_iter() - .map(|time| now_ms - time) - .collect(); + let delays: Vec<i64> = times.into_iter().map(|time| now_ms - time).collect(); if writer.is_none() { stats.off_tree(&delays); } |