move changes as bytes to document

2 files changed, 159 insertions, 0 deletions

diff --git a/src/document/changes.rs b/src/document/changes.rs
new file mode 100644
index 0000000..d369c79
--- /dev/null
+++ b/src/document/changes.rs
@@ -0,0 +1,158 @@
+//! Changes as bytes, to send and to load (`swarm::blob` moves them): one change by its
+//! hash, checked against it when it arrives, and a delta, the changes between two sets
+//! of heads, the same bytes on every node holding them.
+
+use std::collections::{HashMap, HashSet};
+
+use anyhow::{Context, Result, anyhow, bail};
+
+use super::{Change, ChangeHash, Document};
+
+/// The change `hash` as the document encodes it; None if `doc` doesn't have it.
+pub(crate) fn change_bytes(doc: &mut Document, hash: &ChangeHash) -> Option<Vec<u8>> {
+    doc.get_change_by_hash(hash)
+        .map(|change| change.raw_bytes().to_vec())
+}
+
+/// The change in `bytes`, if it is the change `hash`.
+pub(crate) fn change_from_bytes(bytes: Vec<u8>, hash: ChangeHash) -> Result<Change> {
+    let change = Change::from_bytes(bytes)
+        .map_err(|error| anyhow!("change {hash} doesn't load: {error}"))?;
+    if change.hash() != hash {
+        bail!("change has hash {}, not {hash}", change.hash());
+    }
+    Ok(change)
+}
+
+/// The bytes of the delta from `from` to `to` in `doc`: the changes in the history of
+/// `to` but not of `from`, each as encoded, sorted by hash and concatenated
+/// (`load_delta` reads that). The same on every node holding `to`, whatever else it
+/// has. None if `doc` lacks part of `to`.
+pub(crate) fn delta_bytes(
+    doc: &mut Document,
+    from: &[ChangeHash],
+    to: &[ChangeHash],
+) -> Option<Vec<u8>> {
+    if to.iter().any(|head| doc.get_change_by_hash(head).is_none()) {
+        return None;
+    }
+    // Heads of `from` we don't have can't hide anything we have.
+    let from: Vec<ChangeHash> = from
+        .iter()
+        .copied()
+        .filter(|head| doc.get_change_by_hash(head).is_some())
+        .collect();
+    let since: HashMap<ChangeHash, (Vec<ChangeHash>, Vec<u8>)> = doc
+        .get_changes(&from)
+        .iter()
+        .map(|change| {
+            (
+                change.hash(),
+                (change.deps().to_vec(), change.raw_bytes().to_vec()),
+            )
+        })
+        .collect();
+    // Of those, the ones `to` depends on: what we have beyond `to` stays out.
+    let mut included: HashSet<ChangeHash> = HashSet::new();
+    let mut pending: Vec<ChangeHash> = to.to_vec();
+    while let Some(hash) = pending.pop() {
+        if let Some((deps, _)) = since.get(&hash)
+            && included.insert(hash)
+        {
+            pending.extend(deps.iter().copied());
+        }
+    }
+    let mut hashes: Vec<ChangeHash> = included.into_iter().collect();
+    hashes.sort_unstable();
+    let mut bytes = Vec::new();
+    for hash in hashes {
+        bytes.extend_from_slice(&since[&hash].1);
+    }
+    Some(bytes)
+}
+
+/// Loads a delta (`delta_bytes`) into `doc`.
+pub(crate) fn load_delta(doc: &mut Document, bytes: &[u8]) -> Result<()> {
+    doc.load_incremental(bytes)
+        .context("Loading the downloaded changes")?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use automerge::{ActorId, ROOT, transaction::Transactable};
+
+    fn random_write(rng: &mut crate::testrand::Rng, doc: &mut Document) {
+        let key = format!("k{}", rng.below(6));
+        doc.put(ROOT, key, rng.below(100) as i64).unwrap();
+        doc.commit();
+    }
+
+    fn sorted(mut heads: Vec<ChangeHash>) -> Vec<ChangeHash> {
+        heads.sort_unstable();
+        heads
+    }
+
+    /// Over random histories: two nodes that reached the same heads through different
+    /// merge orders, one with more on top, produce the same bytes for a delta to those
+    /// heads; a document at `from` that loads them is at `to`.
+    #[test]
+    fn delta_bytes_are_the_same_everywhere_and_complete() {
+        crate::testrand::for_seeds(200, |rng| {
+            let actor = |i: u8| ActorId::from(&[i; 16][..]);
+            let mut base = Document::new().with_actor(actor(0));
+            for _ in 0..rng.between(0, 5) {
+                random_write(rng, &mut base);
+            }
+            let from = base.get_heads();
+            let mut writers: Vec<Document> = (1..=rng.between(1, 3))
+                .map(|i| base.fork().with_actor(actor(i as u8)))
+                .collect();
+            for _ in 0..rng.between(1, 15) {
+                let writer = rng.below(writers.len());
+                random_write(rng, &mut writers[writer]);
+            }
+            let mut one = base.fork().with_actor(actor(10));
+            let mut two = base.fork().with_actor(actor(11));
+            for writer in writers.iter_mut() {
+                one.merge(writer).unwrap();
+            }
+            for writer in writers.iter_mut().rev() {
+                two.merge(writer).unwrap();
+            }
+            let to = one.get_heads();
+            assert_eq!(sorted(to.clone()), sorted(two.get_heads()));
+            // Beyond `to`: must stay out of the delta.
+            random_write(rng, &mut two);
+
+            let bytes = delta_bytes(&mut one, &from, &to).unwrap();
+            assert_eq!(delta_bytes(&mut two, &from, &to).unwrap(), bytes);
+
+            let mut behind = base.fork();
+            load_delta(&mut behind, &bytes).unwrap();
+            assert_eq!(sorted(behind.get_heads()), sorted(to));
+        });
+    }
+
+    #[test]
+    fn no_delta_without_its_heads() {
+        let mut doc = Document::new();
+        doc.put(ROOT, "k", 1_i64).unwrap();
+        doc.commit();
+        assert!(delta_bytes(&mut doc, &[], &[ChangeHash([7; 32])]).is_none());
+    }
+
+    /// A change loads from its bytes only as the change it was asked for.
+    #[test]
+    fn a_change_is_checked_against_its_hash() {
+        let mut doc = Document::new();
+        doc.put(ROOT, "k", 1_i64).unwrap();
+        doc.commit();
+        let hash = doc.get_heads()[0];
+        let bytes = change_bytes(&mut doc, &hash).unwrap();
+        assert_eq!(change_from_bytes(bytes.clone(), hash).unwrap().hash(), hash);
+        assert!(change_from_bytes(bytes, ChangeHash([7; 32])).is_err());
+        assert!(change_bytes(&mut doc, &ChangeHash([7; 32])).is_none());
+    }
+}
diff --git a/src/document/mod.rs b/src/document/mod.rs
index f5faa11..d7d3e88 100644
--- a/src/document/mod.rs
+++ b/src/document/mod.rs
@@ -6,6 +6,7 @@
 /// Only `document` and `remake` use Automerge itself (`architecture.md`, The document).
 pub(crate) use automerge::{ActorId, AutoCommit as Document, Change, ChangeHash};
 
+pub(crate) mod changes;
 pub(crate) mod file;
 pub(crate) mod lock;
 pub(crate) mod meta;