2 files changed, 36 insertions, 14 deletions

diff --git a/src/state/host.rs b/src/state/host.rs
index a878381..24a989e 100644
--- a/src/state/host.rs
+++ b/src/state/host.rs
@@ -13,7 +13,7 @@ use tracing::debug;
 use super::{AppState, HelloState, NodeLifecycle, StaleView, Standing};
 use crate::document::replica::Epoch;
 use crate::document::store::ChangeSet;
-use crate::document::{ChangeHash, Document};
+use crate::document::{ChangeHash, Document, ReadDoc, Transaction};
 use crate::remake::{Host, PeerSummary, RemakeState};
 
 impl AppState {
@@ -86,7 +86,7 @@ impl Host for AppState {
         crate::swarm::skew::measure(self).await
     }
 
-    fn electorate(&self, doc: &Document, now: crate::clock::UnixMillis) -> Vec<String> {
+    fn electorate(&self, doc: &impl ReadDoc, now: crate::clock::UnixMillis) -> Vec<String> {
         crate::swarm::members::electorate(doc, now.as_u64(), self.electorate_window)
     }
 
@@ -218,7 +218,7 @@ impl Host for AppState {
 
     fn commit_doc<T, E, F>(&self, f: F) -> impl Future<Output = Result<T, E>> + Send
     where
-        F: FnOnce(&mut Document) -> Result<T, E> + Send + 'static,
+        F: FnOnce(&mut Transaction<'_>) -> Result<T, E> + Send + 'static,
         T: Send + 'static,
         E: Send + 'static,
     {
@@ -226,13 +226,39 @@ impl Host for AppState {
         async move { state.mutate_doc(f).await }
     }
 
+    fn apply_change(
+        &self,
+        change: crate::document::Change,
+    ) -> impl Future<Output = Result<(), String>> + Send {
+        let state = self.clone();
+        async move {
+            tokio::task::spawn_blocking(move || {
+                let result = {
+                    let mut replica = state.doc.blocking_write();
+                    let result = replica
+                        .apply_changes([change])
+                        .map_err(|error| error.to_string());
+                    replica.refresh_heads();
+                    result
+                }; // lock released before notifying
+
+                if result.is_ok() {
+                    state.publish_change();
+                }
+                result
+            })
+            .await
+            .expect("document write task panicked")
+        }
+    }
+
     fn switch_epoch<E, F>(
         &self,
         epoch: Epoch,
         f: F,
     ) -> impl Future<Output = Result<Document, E>> + Send
     where
-        F: FnOnce(&mut Document) -> Result<Document, E> + Send + 'static,
+        F: FnOnce(&Document) -> Result<Document, E> + Send + 'static,
         E: Send + 'static,
     {
         let state = self.clone();
@@ -241,7 +267,7 @@ impl Host for AppState {
                 let state = state.clone();
                 tokio::task::spawn_blocking(move || -> Result<Document, E> {
                     let mut replica = state.doc.blocking_write();
-                    let new_doc = f(&mut replica.doc)?;
+                    let new_doc = f(&replica.doc)?;
                     let old_doc = std::mem::replace(&mut replica.doc, new_doc);
                     replica.epoch = epoch;
                     replica.epoch_seq += 1;
diff --git a/src/state/mod.rs b/src/state/mod.rs
index 75c3780..d8bdac5 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -36,10 +36,10 @@ use tokio::sync::{Mutex, RwLock, broadcast, mpsc, watch};
 use tokio::time::Instant;
 use tracing::{debug, error, warn};
 
-use crate::document::Document;
 use crate::document::lock::DocumentLock;
 use crate::document::replica::{Epoch, Stamp};
 use crate::document::store::ChangeSet;
+use crate::document::{Document, Transaction};
 use crate::locks::MutexExt;
 use crate::remake::{Host, RemakeState};
 use crate::swarm::sync::SyncEvent;
@@ -396,7 +396,7 @@ impl AppState {
     /// cancelled request can't leave a committed write unannounced.
     pub(crate) async fn mutate_doc<T, E>(
         &self,
-        f: impl FnOnce(&mut Document) -> Result<T, E> + Send + 'static,
+        f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E> + Send + 'static,
     ) -> Result<T, E>
     where
         T: Send + 'static,
@@ -407,13 +407,9 @@ impl AppState {
         tokio::task::spawn_blocking(move || {
             let result = {
                 let mut replica = state.doc.blocking_write();
-                let result = f(&mut replica.doc);
-                match &result {
-                    // One API write = one change.
-                    Ok(_) => replica.commit(state.clock.now_millis()),
-                    // Discard partial ops (e.g. old value deleted, new one not written).
-                    Err(_) => replica.rollback(),
-                }
+                // One API write = one change; on error, its partial writes (e.g. the old
+                // value deleted, the new one not written) are dropped.
+                let result = replica.transact(state.clock.now_millis(), f);
                 replica.refresh_heads();
                 result
             }; // lock released before notifying