move changes as bytes to document

5 files changed, 173 insertions, 135 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;
diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index 57b02b2..a136217 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -27,12 +27,12 @@ use std::collections::{HashMap, HashSet, VecDeque};
 use std::sync::{Arc, Mutex, MutexGuard};
 use std::time::Duration;
 
-use anyhow::{Result, anyhow, bail};
+use anyhow::{Context, Result, bail};
 use tokio::sync::{Semaphore, mpsc};
 use tokio::time::Instant;
 use tracing::{debug, info, warn};
 
-use crate::document::{ChangeHash, Document};
+use crate::document::{ChangeHash, changes};
 use crate::locks::MutexExt;
 use crate::state::AppState;
 use crate::transport::{BlobBody, BlobId, BlobMessage, DeltaSpec, ParsedMessage};
@@ -216,21 +216,19 @@ async fn prepare(state: &AppState, id: BlobId, delta: Option<DeltaSpec>) -> Opti
             // The spec must be the one the ID names.
             let spec = delta.filter(|spec| spec.id() == delta_id)?;
             let doc = Arc::clone(&state.doc);
-            tokio::task::spawn_blocking(move || delta_bytes(&mut doc.blocking_write().doc, &spec))
-                .await
-                .ok()
-                .flatten()
-                .map(Arc::new)?
+            tokio::task::spawn_blocking(move || {
+                changes::delta_bytes(&mut doc.blocking_write().doc, &spec.from, &spec.to)
+            })
+            .await
+            .ok()
+            .flatten()
+            .map(Arc::new)?
         }
         BlobId::Change(hash) => {
             let doc = Arc::clone(&state.doc);
+            // Looking a change up needs the document mutably.
             tokio::task::spawn_blocking(move || {
-                // get_change_by_hash needs the document mutably.
-                let mut replica = doc.blocking_write();
-                replica
-                    .doc
-                    .get_change_by_hash(&hash)
-                    .map(|change| change.raw_bytes().to_vec())
+                changes::change_bytes(&mut doc.blocking_write().doc, &hash)
             })
             .await
             .ok()
@@ -588,12 +586,7 @@ pub(crate) async fn fetch_change(
     peers: Vec<String>,
 ) -> Result<crate::document::Change> {
     let bytes = fetch(state, BlobId::Change(hash), peers, None, None).await?;
-    let change = crate::document::Change::from_bytes(bytes)
-        .map_err(|error| anyhow!("downloaded change {hash} doesn't load: {error}"))?;
-    if change.hash() != hash {
-        bail!("downloaded change has hash {}, not {hash}", change.hash());
-    }
-    Ok(change)
+    changes::change_from_bytes(bytes, hash).context("downloaded")
 }
 
 /// Downloads the changes of `spec` from `peers` (any that hold its `to`). The caller
@@ -606,120 +599,9 @@ pub(crate) async fn fetch_delta(
     fetch(state, BlobId::Delta(spec.id()), peers, None, Some(spec)).await
 }
 
-/// The bytes of delta `spec` from `doc`: the changes in the history of `spec.to` but
-/// not of `spec.from`, each as Automerge encodes it, sorted by hash and concatenated
-/// (`Document::load_incremental` 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, spec: &DeltaSpec) -> Option<Vec<u8>> {
-    if spec
-        .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> = spec
-        .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> = spec.to.clone();
-    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)
-}
-
 #[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 spec = DeltaSpec::new(from.clone(), to.clone());
-            let bytes = delta_bytes(&mut one, &spec).unwrap();
-            assert_eq!(delta_bytes(&mut two, &spec).unwrap(), bytes);
-
-            let mut behind = base.fork();
-            behind.load_incremental(&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();
-        let spec = DeltaSpec::new(Vec::new(), vec![ChangeHash([7; 32])]);
-        assert!(delta_bytes(&mut doc, &spec).is_none());
-    }
 
     #[test]
     fn chunking() {
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 772932d..40e577a 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -503,10 +503,7 @@ async fn download_and_apply_delta(
             bail!("the epoch changed during the download");
         }
         let heads_before = replica.heads().to_vec();
-        replica
-            .doc
-            .load_incremental(&bytes)
-            .context("Loading the downloaded changes")?;
+        crate::document::changes::load_delta(&mut replica.doc, &bytes)?;
         replica.refresh_heads();
         let missing = spec
             .to
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index cef31ff..61d54f2 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -646,7 +646,7 @@ pub(crate) enum SyncBody {
 
 /// The changes in the history of `to` but not of `from`: what a peer at `from` lacks of
 /// a document at `to`. Every node holding `to` produces the same bytes for it
-/// (`swarm::blob::delta_bytes`).
+/// (`document::changes::delta_bytes`).
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub(crate) struct DeltaSpec {
     /// Sorted.