1 files changed, 27 insertions, 22 deletions

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;