move all automerge references to document

6 files changed, 39 insertions, 51 deletions

diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index ec9a355..d2bbe24 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -300,15 +300,13 @@ fn identify(
     })
 }
 
-/// A backup to recover from, as `GET /admin/document` saves it, under a fresh actor.
+/// A backup to recover from, as `GET /admin/document` saves it (`document::load`).
 fn load_backup(path: &Path, files: &Files) -> Result<Document> {
     let bytes = files
         .read(FileKind::Backup, path)
         .with_context(|| format!("Reading the backup {}", path.display()))?
         .with_context(|| format!("The backup {} doesn't exist", path.display()))?;
-    Ok(Document::load(&bytes)
-        .with_context(|| format!("Loading the backup {}", path.display()))?
-        .with_actor(ActorId::random()))
+    crate::document::load(&bytes).with_context(|| format!("Loading the backup {}", path.display()))
 }
 
 /// Resolves the seeds in `unresolved`, says HELLO to each that resolves, and keeps the
diff --git a/src/swarm/changes.rs b/src/swarm/changes.rs
index 6831081..3a50769 100644
--- a/src/swarm/changes.rs
+++ b/src/swarm/changes.rs
@@ -6,42 +6,47 @@ use std::sync::Arc;
 
 use tracing::{info, warn};
 
+use crate::document::changes;
 use crate::document::store::{self, ChangeSet, InLists};
 use crate::state::AppState;
 
 pub(crate) async fn run_change_publisher(state: AppState) {
     let mut changed = state.change_tx.subscribe();
-    let mut last_heads = state.write_doc(|doc| doc.get_heads()).await;
+    // The replica's heads (sorted, as of its last refresh: every change refreshes them).
+    let mut last_heads = state.doc.read().await.heads().to_vec();
 
     info!("Starting change publisher");
 
     while changed.changed().await.is_ok() {
         let before = last_heads.clone();
 
-        let (heads, paths, conflicts) = state
-            .write_doc(move |doc| {
-                let heads = doc.get_heads();
-                if heads == before {
-                    return (heads, Vec::new(), Vec::new());
-                }
+        let doc = Arc::clone(&state.doc);
+        // Looking changes up needs the document mutably.
+        let (heads, paths, conflicts) = tokio::task::spawn_blocking(move || {
+            let mut replica = doc.blocking_write();
+            let heads = replica.heads().to_vec();
+            if heads == before {
+                return (heads, Vec::new(), Vec::new());
+            }
 
-                // After an epoch switch the previous heads belong to the old document:
-                // report the root as changed, which every stream is related to.
-                if !before.iter().all(|h| doc.get_change_by_hash(h).is_some()) {
-                    return (heads, vec![Vec::new()], Vec::new());
-                }
+            // After an epoch switch the previous heads belong to the old document:
+            // report the root as changed, which every stream is related to.
+            if !changes::missing(&mut replica.doc, &before).is_empty() {
+                return (heads, vec![Vec::new()], Vec::new());
+            }
 
-                let mut paths = Vec::new();
-                let mut conflicts = Vec::new();
-                for changed in store::changed(doc, &before, &heads, InLists::Index) {
-                    if changed.conflict {
-                        conflicts.push(changed.path.clone());
-                    }
-                    paths.push(changed.path);
+            let mut paths = Vec::new();
+            let mut conflicts = Vec::new();
+            for changed in store::changed(&mut replica.doc, &before, &heads, InLists::Index) {
+                if changed.conflict {
+                    conflicts.push(changed.path.clone());
                 }
-                (heads, paths, conflicts)
-            })
-            .await;
+                paths.push(changed.path);
+            }
+            (heads, paths, conflicts)
+        })
+        .await
+        .expect("change publisher task panicked");
 
         if heads == last_heads {
             continue;
diff --git a/src/swarm/document_file.rs b/src/swarm/document_file.rs
index 188b2b6..435ff32 100644
--- a/src/swarm/document_file.rs
+++ b/src/swarm/document_file.rs
@@ -47,7 +47,7 @@ pub(crate) async fn flush(state: &AppState) {
                     return;
                 }
                 let heads = replica.heads().to_vec();
-                (heads, replica.doc.save_after(&since))
+                (heads, crate::document::save_since(&mut replica.doc, &since))
             };
             file.append(since, heads, increment).await;
         }
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index c28c9f1..d11c3e0 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -130,16 +130,9 @@ pub(crate) async fn pull_if_behind(
     }
     // Those of the listed heads we lack. With a digest only a sample is listed; if we
     // have all of it, a later keepalive lists others.
-    let missing: Vec<_> = {
-        // Document's lookups take &mut self.
+    let missing = {
         let mut replica = state.doc.write().await;
-        remote
-            .heads
-            .listed()
-            .iter()
-            .filter(|head| replica.doc.get_change_by_hash(head).is_none())
-            .copied()
-            .collect()
+        crate::document::changes::missing(&mut replica.doc, remote.heads.listed())
     };
     if missing.is_empty() {
         return false;
diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
index b5548d5..2398b09 100644
--- a/src/swarm/stats.rs
+++ b/src/swarm/stats.rs
@@ -279,7 +279,7 @@ pub(crate) fn measure_saved_size(state: &AppState) {
         // Both steps take time proportional to the document: off the async threads.
         let measured = match tokio::task::spawn_blocking(move || {
             let mut copy = doc.blocking_read().doc.clone();
-            copy.save().len() as u64
+            crate::document::save(&mut copy).len() as u64
         })
         .await
         {
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 40e577a..3cbe5f3 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -389,15 +389,11 @@ pub(crate) async fn handle_sync(
                 // 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;
-                // Time 0: not stamped (snapshots, older nodes).
-                let delays: Vec<i64> = replica
-                    .doc
-                    .get_changes(&heads_before)
-                    .iter()
-                    .map(|change| change.timestamp())
-                    .filter(|&time| time > 0)
-                    .map(|time| now_ms - time)
-                    .collect();
+                let delays: Vec<i64> =
+                    crate::document::changes::commit_times(&mut replica.doc, &heads_before)
+                        .into_iter()
+                        .map(|time| now_ms - time)
+                        .collect();
                 stats.propagated(delays);
             }
 
@@ -505,11 +501,7 @@ async fn download_and_apply_delta(
         let heads_before = replica.heads().to_vec();
         crate::document::changes::load_delta(&mut replica.doc, &bytes)?;
         replica.refresh_heads();
-        let missing = spec
-            .to
-            .iter()
-            .filter(|head| replica.doc.get_change_by_hash(head).is_none())
-            .count();
+        let missing = crate::document::changes::missing(&mut replica.doc, &spec.to).len();
         if missing > 0 {
             bail!("{missing} of the offered heads are still missing after loading");
         }