remake coordination use document, not automerge

7 files changed, 50 insertions, 50 deletions

diff --git a/src/document/mod.rs b/src/document/mod.rs
index 671bf5d..f11fef0 100644
--- a/src/document/mod.rs
+++ b/src/document/mod.rs
@@ -54,13 +54,18 @@ mod tests {
         }
     }
 
-    /// Automerge is used only in `document` and `remake`; the rest of the node names
-    /// its types as `document` re-exports them and works through `document`
-    /// (`architecture.md`, The document).
+    /// Automerge is used only in `document` and in what remakes do to documents
+    /// (snapshots, replaying late writes, catching up, the kinds, quarantined writes);
+    /// the rest of the node, remakes' coordination included, names its types as
+    /// `document` re-exports them and works through `document` (`architecture.md`, The
+    /// document).
     #[test]
     fn automerge_stays_behind_document() {
         let src = Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
-        let allowed = [src.join("document"), src.join("remake")];
+        let mut allowed = vec![src.join("document")];
+        for engine in ["snapshot", "delta", "catchup", "kind", "quarantine"] {
+            allowed.push(src.join("remake").join(format!("{engine}.rs")));
+        }
         let mut files = Vec::new();
         sources(&src, &mut files);
         let mut found = Vec::new();
diff --git a/src/remake/api.rs b/src/remake/api.rs
index 75d0294..2cd55e5 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -54,7 +54,6 @@
 
 use std::time::Duration;
 
-use automerge::ChangeHash;
 use axum::{
     Json,
     body::Bytes,
@@ -80,6 +79,7 @@ use super::records::{
 };
 use super::{Host, kind};
 use crate::api::error_response;
+use crate::document::ChangeHash;
 use crate::document::store::StoreError;
 use crate::force::Force;
 
diff --git a/src/remake/local.rs b/src/remake/local.rs
index 7810690..e94657c 100644
--- a/src/remake/local.rs
+++ b/src/remake/local.rs
@@ -9,7 +9,6 @@ use std::collections::{HashMap, HashSet};
 use std::sync::{Arc, Mutex, MutexGuard};
 use std::time::Duration;
 
-use automerge::{ActorId, AutoCommit};
 use serde::Serialize;
 use tokio::time::Instant;
 
@@ -17,6 +16,7 @@ use super::records::Retention;
 use super::snapshot::Snapshot;
 use crate::clock::UnixMillis;
 use crate::document::replica::Epoch;
+use crate::document::{ActorId, Document};
 use crate::locks::MutexExt;
 
 pub(crate) struct RemakeState {
@@ -85,7 +85,7 @@ pub(crate) struct Rejoin {
     pub(crate) previous: Epoch,
     pub(crate) target: Epoch,
     /// The document of `previous`, until it is replayed, quarantined or restored.
-    pub(crate) old: AutoCommit,
+    pub(crate) old: Document,
     pub(crate) started: Instant,
     /// The node served requests before; it does again once the rejoin is over.
     pub(crate) was_ready: bool,
@@ -102,7 +102,7 @@ pub(crate) struct Quarantined {
     /// Store paths the changes touched (minimal, prefix-free).
     pub(crate) paths: Vec<Vec<String>>,
     /// The old document; `paths` are read from its merged view.
-    pub(crate) document: AutoCommit,
+    pub(crate) document: Document,
 }
 
 #[derive(Clone, Debug, Serialize)]
@@ -138,7 +138,7 @@ struct Archived {
     epoch: Epoch,
     // Kept for auditing and recovery by hand; not read by the protocol.
     #[allow(dead_code)]
-    doc: AutoCommit,
+    doc: Document,
     archived_at: UnixMillis,
     retention: Retention,
 }
@@ -249,7 +249,7 @@ impl RemakeState {
     pub(crate) fn archive(
         &self,
         epoch: Epoch,
-        doc: AutoCommit,
+        doc: Document,
         retention: Retention,
         current_generation: u64,
         now: UnixMillis,
@@ -431,7 +431,7 @@ mod tests {
         // Kept by epoch count only.
         remake.archive(
             epoch(0),
-            AutoCommit::new(),
+            Document::new(),
             Retention {
                 epochs: Some(1),
                 age_ms: None,
@@ -446,7 +446,7 @@ mod tests {
         // Kept by time while the epoch count has passed.
         remake.archive(
             epoch(1),
-            AutoCommit::new(),
+            Document::new(),
             Retention {
                 epochs: Some(0),
                 age_ms: Some(3_600_000),
@@ -459,7 +459,7 @@ mod tests {
         // No limits at all: not kept.
         remake.archive(
             epoch(2),
-            AutoCommit::new(),
+            Document::new(),
             Retention {
                 epochs: None,
                 age_ms: None,
diff --git a/src/remake/mod.rs b/src/remake/mod.rs
index c77d56d..9206bbf 100644
--- a/src/remake/mod.rs
+++ b/src/remake/mod.rs
@@ -45,7 +45,6 @@ use std::future::Future;
 use std::sync::Arc;
 use std::time::Duration;
 
-use automerge::AutoCommit;
 use axum::{
     Router,
     extract::DefaultBodyLimit,
@@ -53,6 +52,7 @@ use axum::{
 };
 use tokio::sync::broadcast;
 
+use crate::document::Document;
 use crate::document::replica::Epoch;
 use crate::document::store::ChangeSet;
 
@@ -74,7 +74,7 @@ pub(crate) struct Summary {
 }
 
 /// The `Summary` of `remake` with the records of `doc`, whose epoch is `epoch`.
-pub(crate) fn summary(remake: &RemakeState, doc: &AutoCommit, epoch: Epoch) -> Summary {
+pub(crate) fn summary(remake: &RemakeState, doc: &Document, epoch: Epoch) -> Summary {
     let local = records::EpochRecord::from(epoch);
     let mut summary = Summary {
         rejoining: remake.is_rejoining(),
@@ -129,14 +129,14 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
 
     /// The electorate of a remake proposed now (`vote`): the members that said they are
     /// alive recently enough, read from `doc`.
-    fn electorate(&self, doc: &AutoCommit, now: crate::clock::UnixMillis) -> Vec<String>;
+    fn electorate(&self, doc: &Document, now: crate::clock::UnixMillis) -> Vec<String>;
 
     /// Downloads a change from peers on the local epoch, in chunks from several at once,
     /// checked against its hash.
     fn fetch_change(
         &self,
-        hash: automerge::ChangeHash,
-    ) -> impl Future<Output = anyhow::Result<automerge::Change>> + Send;
+        hash: crate::document::ChangeHash,
+    ) -> impl Future<Output = anyhow::Result<crate::document::Change>> + Send;
 
     /// Downloads content of `size` bytes by its BLAKE3 hash from peers on the local
     /// epoch, checked against it.
@@ -163,7 +163,7 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
 
     /// Serves reads from this document (read-only, marked stale) while the node is not
     /// ready; `None` ends it.
-    fn set_stale_view(&self, view: Option<(Epoch, AutoCommit)>);
+    fn set_stale_view(&self, view: Option<(Epoch, Document)>);
 
     /// Whether the document caught up with a peer since the last reset.
     fn initial_sync_done(&self) -> bool;
@@ -177,13 +177,13 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
     /// Reads the document on the blocking pool.
     fn read_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send
     where
-        F: FnOnce(&AutoCommit) -> T + Send + 'static,
+        F: FnOnce(&Document) -> T + Send + 'static,
         T: Send + 'static;
 
     /// One committed change (rolled back on error), published and replicated.
     fn commit_doc<T, E, F>(&self, f: F) -> impl Future<Output = Result<T, E>> + Send
     where
-        F: FnOnce(&mut AutoCommit) -> Result<T, E> + Send + 'static,
+        F: FnOnce(&mut Document) -> Result<T, E> + Send + 'static,
         T: Send + 'static,
         E: Send + 'static;
 
@@ -194,9 +194,9 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
         &self,
         epoch: Epoch,
         f: F,
-    ) -> impl Future<Output = Result<AutoCommit, E>> + Send
+    ) -> impl Future<Output = Result<Document, E>> + Send
     where
-        F: FnOnce(&mut AutoCommit) -> Result<AutoCommit, E> + Send + 'static,
+        F: FnOnce(&mut Document) -> Result<Document, E> + Send + 'static,
         E: Send + 'static;
 }
 
diff --git a/src/remake/proposal.rs b/src/remake/proposal.rs
index ede25d2..d9cd893 100644
--- a/src/remake/proposal.rs
+++ b/src/remake/proposal.rs
@@ -5,13 +5,13 @@
 
 use std::collections::HashSet;
 
-use automerge::ChangeHash;
 use tracing::info;
 
 use super::coordinator;
 use super::records::{self, Decision, Phase, Proposal, Record, Vote};
 use super::{Host, kind, vote};
 use crate::clock::UnixMillis;
+use crate::document::ChangeHash;
 use crate::document::store::StoreError;
 use crate::force::Force;
 
@@ -220,9 +220,7 @@ pub(super) async fn propose<H: Host>(
             if !base_at_commit {
                 let base = match base {
                     Some(base) => {
-                        if let Some(unknown) = base
-                            .iter()
-                            .find(|head| doc.get_change_by_hash(head).is_none())
+                        if let Some(unknown) = crate::document::changes::missing(doc, &base).first()
                         {
                             return Err(StartError::UnknownBase(unknown.to_string()));
                         }
diff --git a/src/remake/records.rs b/src/remake/records.rs
index 463d085..f22b1e5 100644
--- a/src/remake/records.rs
+++ b/src/remake/records.rs
@@ -28,7 +28,6 @@
 
 use std::collections::BTreeMap;
 
-use automerge::{AutoCommit, ChangeHash};
 use serde::{Deserialize, Serialize};
 use serde_json::Value as JsonValue;
 use tracing::warn;
@@ -39,6 +38,7 @@ use crate::clock::UnixMillis;
 use crate::document::meta::{self, REMAKES, VOTES};
 use crate::document::replica::Epoch;
 use crate::document::store::{self, StoreError};
+use crate::document::{ChangeHash, Document};
 
 /// One day, for retention defaults.
 const DAY_MS: u64 = 24 * 60 * 60 * 1000;
@@ -337,13 +337,13 @@ pub(crate) fn record_path(id: &str, part: &str) -> Vec<String> {
 }
 
 /// The record with this ID, if it exists and is readable, with its votes.
-pub(crate) fn load(doc: &AutoCommit, id: &str) -> Option<Record> {
+pub(crate) fn load(doc: &Document, id: &str) -> Option<Record> {
     let record = load_record(doc, id)?;
     Some(with_votes(id, record, &load_register(doc)))
 }
 
 /// All readable records, with their votes.
-pub(crate) fn load_all(doc: &AutoCommit) -> Vec<(String, Record)> {
+pub(crate) fn load_all(doc: &Document) -> Vec<(String, Record)> {
     let register = load_register(doc);
     store::keys_at(doc, &records_dir())
         .unwrap_or_default()
@@ -366,7 +366,7 @@ fn with_votes(id: &str, mut record: Record, register: &Register) -> Record {
     record
 }
 
-fn load_record(doc: &AutoCommit, id: &str) -> Option<Record> {
+fn load_record(doc: &Document, id: &str) -> Option<Record> {
     let value = match store::read_path(doc, &record_path(id, "")) {
         Ok(Some(value)) => value,
         Ok(None) => return None,
@@ -390,7 +390,7 @@ fn load_record(doc: &AutoCommit, id: &str) -> Option<Record> {
 }
 
 /// A commit among concurrent decisions of remake `id` that disagree, if they do.
-fn contested_decision(doc: &AutoCommit, id: &str) -> Option<Decision> {
+fn contested_decision(doc: &Document, id: &str) -> Option<Decision> {
     let decisions: Vec<Decision> = store::read_candidates(doc, &record_path(id, "decision"))
         .ok()?
         .into_iter()
@@ -418,7 +418,7 @@ fn register_path(voter: &str) -> Vec<String> {
 }
 
 /// The votes register; unreadable entries are skipped with a warning.
-pub(crate) fn load_register(doc: &AutoCommit) -> Register {
+pub(crate) fn load_register(doc: &Document) -> Register {
     store::keys_at(doc, &meta::path(&[VOTES]))
         .unwrap_or_default()
         .into_iter()
@@ -446,7 +446,7 @@ pub(crate) fn load_register(doc: &AutoCommit) -> Register {
 }
 
 fn write_part<T: Serialize>(
-    doc: &mut AutoCommit,
+    doc: &mut Document,
     id: &str,
     part: &str,
     value: &T,
@@ -456,7 +456,7 @@ fn write_part<T: Serialize>(
 }
 
 pub(crate) fn write_proposal(
-    doc: &mut AutoCommit,
+    doc: &mut Document,
     id: &str,
     proposal: &Proposal,
 ) -> Result<(), StoreError> {
@@ -464,7 +464,7 @@ pub(crate) fn write_proposal(
 }
 
 pub(crate) fn write_decision(
-    doc: &mut AutoCommit,
+    doc: &mut Document,
     id: &str,
     decision: &Decision,
 ) -> Result<(), StoreError> {
@@ -472,7 +472,7 @@ pub(crate) fn write_decision(
 }
 
 pub(crate) fn write_abort_request(
-    doc: &mut AutoCommit,
+    doc: &mut Document,
     id: &str,
     request: &AbortRequest,
 ) -> Result<(), StoreError> {
@@ -480,7 +480,7 @@ pub(crate) fn write_abort_request(
 }
 
 /// Writes `voter`'s entry in the votes register (replacing its earlier one).
-pub(crate) fn write_vote(doc: &mut AutoCommit, voter: &str, vote: &Vote) -> Result<(), StoreError> {
+pub(crate) fn write_vote(doc: &mut Document, voter: &str, vote: &Vote) -> Result<(), StoreError> {
     let json = serde_json::to_value(vote).map_err(anyhow::Error::from)?;
     store::write_path(doc, &register_path(voter), &json)
 }
@@ -568,7 +568,7 @@ mod tests {
 
     #[test]
     fn round_trip_through_the_document() {
-        let mut doc = AutoCommit::new();
+        let mut doc = Document::new();
         let p = proposal(10_000);
         write_proposal(&mut doc, "c-1", &p).unwrap();
         write_abort_request(
@@ -612,7 +612,7 @@ mod tests {
 
     #[test]
     fn concurrent_votes_are_kept_apart() {
-        let mut a = AutoCommit::new();
+        let mut a = Document::new();
         let vote = |remake: &str, at: u64| Vote {
             remake: remake.into(),
             at: UnixMillis(at),
@@ -685,7 +685,7 @@ mod tests {
 
     #[test]
     fn disagreeing_decisions_are_contested() {
-        let mut base = AutoCommit::new();
+        let mut base = Document::new();
         write_proposal(&mut base, "c-1", &proposal(10_000)).unwrap();
         base.commit();
         let (mut a, mut b, mut c) = (base.fork(), base.fork(), base.fork());
@@ -725,7 +725,7 @@ mod tests {
 
     #[test]
     fn unreadable_records_are_skipped() {
-        let mut doc = AutoCommit::new();
+        let mut doc = Document::new();
         store::write_path(
             &mut doc,
             &record_path("broken", ""),
diff --git a/src/remake/switch.rs b/src/remake/switch.rs
index 1238499..9aae748 100644
--- a/src/remake/switch.rs
+++ b/src/remake/switch.rs
@@ -20,7 +20,6 @@
 
 use std::time::Duration;
 
-use automerge::{ActorId, AutoCommit, ChangeHash};
 use tokio::sync::broadcast;
 use tracing::{error, info, warn};
 
@@ -39,6 +38,7 @@ use crate::document::meta::{self, REMAKES};
 use crate::document::replica::Epoch;
 use crate::document::store;
 use crate::document::store::paths_related;
+use crate::document::{ActorId, ChangeHash, Document};
 use crate::locks::MutexExt;
 
 const TICK: Duration = Duration::from_secs(1);
@@ -218,7 +218,7 @@ async fn switch<H: Host>(
 
     // 2. Build (or reuse), verify and replay without holding any lock.
     let (new_doc, copy_heads, mut stats) = tokio::task::spawn_blocking(
-        move || -> Result<(AutoCommit, Vec<ChangeHash>, ReplayStats), String> {
+        move || -> Result<(Document, Vec<ChangeHash>, ReplayStats), String> {
             let mut copy = copy;
             let snapshot = match prebuilt {
                 Some(snapshot) => snapshot,
@@ -288,20 +288,17 @@ async fn switch<H: Host>(
 pub(super) async fn swap_in<H: Host>(
     host: &H,
     target: Epoch,
-    new_doc: AutoCommit,
+    new_doc: Document,
     copy_heads: Vec<ChangeHash>,
     translator: Translator,
-) -> Result<(AutoCommit, ReplayStats), String> {
+) -> Result<(Document, ReplayStats), String> {
     // The closure runs on the blocking pool; its replay numbers come back through this.
     let late_stats = std::sync::Arc::new(std::sync::Mutex::new(ReplayStats::default()));
     let late_out = std::sync::Arc::clone(&late_stats);
     let old = host
         .switch_epoch(target, move |live| {
             // Only one worker switches, but be sure the copy still describes the live document.
-            if !copy_heads
-                .iter()
-                .all(|head| live.get_change_by_hash(head).is_some())
-            {
+            if !crate::document::changes::missing(live, &copy_heads).is_empty() {
                 return Err("the document was replaced while preparing the switch".to_owned());
             }
             let mut new_doc = new_doc;