move changes as bytes to document

2 files changed, 13 insertions, 134 deletions

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