move all automerge references to document

15 files changed, 95 insertions, 86 deletions

diff --git a/src/api/document.rs b/src/api/document.rs
index 20bbc7b..08fd21f 100644
--- a/src/api/document.rs
+++ b/src/api/document.rs
@@ -50,12 +50,11 @@ pub(crate) async fn get_document(State(state): State<AppState>) -> Response {
     // Both steps take time proportional to the document: off the async threads, and
     // the lock only for the copy.
     let saved = tokio::task::spawn_blocking(move || {
-        let (mut copy, epoch) = {
+        let (mut copy, epoch, heads) = {
             let replica = doc.blocking_read();
-            (replica.doc.clone(), replica.epoch)
+            (replica.doc.clone(), replica.epoch, replica.heads().to_vec())
         };
-        let heads = copy.get_heads();
-        (copy.save(), heads, epoch)
+        (crate::document::save(&mut copy), heads, epoch)
     })
     .await;
     let (bytes, heads, epoch) = match saved {
diff --git a/src/document/changes.rs b/src/document/changes.rs
index d369c79..655063b 100644
--- a/src/document/changes.rs
+++ b/src/document/changes.rs
@@ -8,6 +8,26 @@ use anyhow::{Context, Result, anyhow, bail};
 
 use super::{Change, ChangeHash, Document};
 
+/// Those of `heads` that `doc` doesn't have. (Looking a change up needs the document
+/// mutably.)
+pub(crate) fn missing(doc: &mut Document, heads: &[ChangeHash]) -> Vec<ChangeHash> {
+    heads
+        .iter()
+        .copied()
+        .filter(|head| doc.get_change_by_hash(head).is_none())
+        .collect()
+}
+
+/// When the changes after `since` were committed, in Unix milliseconds by their
+/// writers' clocks (`Replica::commit`); changes without a time (snapshots) left out.
+pub(crate) fn commit_times(doc: &mut Document, since: &[ChangeHash]) -> Vec<i64> {
+    doc.get_changes(since)
+        .iter()
+        .map(|change| change.timestamp())
+        .filter(|&time| time > 0)
+        .collect()
+}
+
 /// 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)
@@ -33,7 +53,7 @@ pub(crate) fn delta_bytes(
     from: &[ChangeHash],
     to: &[ChangeHash],
 ) -> Option<Vec<u8>> {
-    if to.iter().any(|head| doc.get_change_by_hash(head).is_none()) {
+    if !missing(doc, to).is_empty() {
         return None;
     }
     // Heads of `from` we don't have can't hide anything we have.
diff --git a/src/document/file.rs b/src/document/file.rs
index 47d7310..e03c2e6 100644
--- a/src/document/file.rs
+++ b/src/document/file.rs
@@ -99,7 +99,7 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> {
         }
         let (record, after) = after.split_at(length);
         match &mut doc {
-            None => doc = Some(AutoCommit::load(record).context("the full save")?),
+            None => doc = Some(super::load(record).context("the full save")?),
             Some(doc) => {
                 doc.load_incremental(record).context("an increment")?;
             }
@@ -109,10 +109,7 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> {
     let Some(doc) = doc else {
         bail!("no full save");
     };
-    let loaded = Loaded {
-        doc: doc.with_actor(ActorId::random()),
-        actors,
-    };
+    let loaded = Loaded { doc, actors };
     Ok((loaded, records.len()))
 }
 
@@ -250,7 +247,7 @@ impl DocumentFile {
                 .into_iter()
                 .filter(|actor| *actor == current || present.contains(actor))
                 .collect();
-            let save = copy.save();
+            let save = super::save(&mut copy);
             let bytes = whole(&actors, &save)?;
             writer
                 .files
diff --git a/src/document/mod.rs b/src/document/mod.rs
index 1612145..2230e47 100644
--- a/src/document/mod.rs
+++ b/src/document/mod.rs
@@ -6,6 +6,30 @@
 /// Only `document` and `remake` use Automerge itself (`architecture.md`, The document).
 pub(crate) use automerge::{ActorId, AutoCommit as Document, Change, ChangeHash};
 
+/// The whole document as bytes, as the engine saves it: a backup, the document
+/// file's full save, the size a joining node downloads.
+pub(crate) fn save(doc: &mut Document) -> Vec<u8> {
+    doc.save()
+}
+
+/// The changes after `since` as bytes that, appended to a save, load with it (the
+/// document file's appends).
+pub(crate) fn save_since(doc: &mut Document, since: &[ChangeHash]) -> Vec<u8> {
+    doc.save_after(since)
+}
+
+/// A saved document (`save`), loaded to write to under a writer ID of its own: one ID
+/// must never write in two places, and a save may be loaded on several nodes, or again
+/// after the node that saved it wrote on.
+pub(crate) fn load(bytes: &[u8]) -> anyhow::Result<Document> {
+    Ok(Document::load(bytes)?.with_actor(ActorId::random()))
+}
+
+/// The actor (writer ID) `doc` writes under.
+pub(crate) fn actor(doc: &Document) -> ActorId {
+    doc.get_actor().clone()
+}
+
 pub(crate) mod changes;
 pub(crate) mod file;
 pub(crate) mod lock;
diff --git a/src/document/replica.rs b/src/document/replica.rs
index 5d3d7a7..5a87016 100644
--- a/src/document/replica.rs
+++ b/src/document/replica.rs
@@ -101,7 +101,7 @@ impl fmt::Debug for Epoch {
 /// The heads are cached: `AutoCommit::get_heads` takes `&mut self` (it closes a pending
 /// transaction), so without the cache every keepalive and status read would need the
 /// write lock. Every path that changes the document under the write lock calls
-/// [`Replica::refresh_heads`] before releasing it (`AppState::write_doc`, `mutate_doc`,
+/// [`Replica::refresh_heads`] before releasing it (`AppState::mutate_doc`,
 /// `switch_epoch`, the sync apply); readers take the read lock and use
 /// [`Replica::heads`].
 ///
diff --git a/src/node.rs b/src/node.rs
index f09de5b..8224c7a 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -72,7 +72,7 @@ impl Node {
         let (epoch_tx, _) = watch::channel(config.initial_epoch);
         let firewall = Arc::new(Firewall::default());
         let traffic = Arc::new(Traffic::default());
-        let initial_actor = config.initial_doc.get_actor().clone();
+        let initial_actor = crate::document::actor(&config.initial_doc);
         let transport = match config.identity {
             Some(identity) => Arc::new(IdentityTransport::new(
                 config.transport,
diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs
index 9642f28..aa30e7c 100644
--- a/src/simtest/mod.rs
+++ b/src/simtest/mod.rs
@@ -253,11 +253,11 @@ impl Sim {
                     ago(p.last_seen),
                     ago(p.last_sync_rx),
                     p.is_unreachable(),
-                    p.sync.automerge_view().0.map(|heads| heads
+                    p.sync.sync_view().0.map(|heads| heads
                         .iter()
                         .map(|head| head.to_string()[..8].to_owned())
                         .collect::<Vec<_>>()),
-                    p.sync.automerge_view().1,
+                    p.sync.sync_view().1,
                     p.sync.sending.is_some(),
                     p.sync
                         .reply
diff --git a/src/state/mod.rs b/src/state/mod.rs
index 673bd56..8ac1124 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -394,25 +394,6 @@ impl AppState {
         .expect("document read task panicked")
     }
 
-    /// Mutates the document on the blocking pool.
-    pub(crate) async fn write_doc<T>(
-        &self,
-        f: impl FnOnce(&mut Document) -> T + Send + 'static,
-    ) -> T
-    where
-        T: Send + 'static,
-    {
-        let doc = Arc::clone(&self.doc);
-        tokio::task::spawn_blocking(move || {
-            let mut replica = doc.blocking_write();
-            let out = f(&mut replica.doc);
-            replica.refresh_heads();
-            out
-        })
-        .await
-        .expect("document write task panicked")
-    }
-
     /// Mutates the document on the blocking pool as one change, then
     /// publishes it. Write and notification run in the same blocking task, so a
     /// cancelled request can't leave a committed write unannounced.
diff --git a/src/state/session.rs b/src/state/session.rs
index bf5c83f..7305c58 100644
--- a/src/state/session.rs
+++ b/src/state/session.rs
@@ -245,7 +245,7 @@ impl SyncSession {
     /// For the simulator's dump: the peer's heads as Automerge knows them, and whether
     /// it thinks a message is in flight.
     #[cfg(test)]
-    pub(crate) fn automerge_view(&self) -> (Option<&[crate::document::ChangeHash]>, bool) {
+    pub(crate) fn sync_view(&self) -> (Option<&[crate::document::ChangeHash]>, bool) {
         (self.state.their_heads.as_deref(), self.state.in_flight)
     }
 
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");
         }