1 files changed, 113 insertions, 53 deletions

diff --git a/src/state/mod.rs b/src/state/mod.rs
index fec164d..aebc59a 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -2,7 +2,6 @@
 //!
 //! * `peer`: what a node knows about each peer (`PeerState`): standing, liveness,
 //!   handshake, incidents.
-//! * `session`: one Automerge sync with a peer (`SyncSession`).
 //! * `timing`: the intervals the protocol runs on (`Timing`).
 //! * `read`: serving reads, live or from the pre-rejoin copy.
 //! * `host`: the node as the remake module sees it (`remake::Host`).
@@ -21,13 +20,11 @@
 mod host;
 mod peer;
 mod read;
-mod session;
 mod timing;
 
 pub(crate) use peer::{HelloState, Incidents, Keepalive, Liveness, PeerState, Standing};
 pub(crate) use read::{ReadSource, StaleView};
-pub(crate) use session::SyncSession;
-pub(crate) use timing::Timing;
+pub(crate) use timing::{SYNC_REPLY_TIMEOUT, Timing};
 
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -35,16 +32,17 @@ use std::sync::atomic::AtomicBool;
 use std::time::Duration;
 
 use anyhow::{Result, bail};
-use automerge::AutoCommit;
 use tokio::sync::{Mutex, RwLock, broadcast, mpsc, watch};
 use tokio::time::Instant;
 use tracing::{debug, error, warn};
 
 use crate::document::lock::DocumentLock;
 use crate::document::replica::{Epoch, Stamp};
-use crate::locks::MutexExt;
+use crate::document::store::{ChangeSet, StoreError};
+use crate::document::{Document, Transaction};
+use crate::locks::{MutexExt, RwLockExt};
 use crate::remake::{Host, RemakeState};
-use crate::swarm::sync::{ChangeSet, SyncEvent};
+use crate::swarm::sync::SyncEvent;
 use crate::swarm::throughput::ThroughputTable;
 use crate::transport;
 use crate::transport::firewall::Firewall;
@@ -58,13 +56,12 @@ pub(crate) enum NodeLifecycle {
 
 #[derive(Clone)]
 pub(crate) struct AppState {
-    /// The document and its epoch. Derefs to `AutoCommit`.
+    /// The document and its epoch. Derefs to `Document`.
     pub(crate) doc: Arc<DocumentLock>,
 
-    /// Copy of `doc.epoch` for cheap, lock-free reads (handshakes, keepalives, /peers).
-    /// Only ever changed while holding `doc`'s write lock, together with `doc.epoch`.
-    pub(crate) epoch_tx: watch::Sender<Epoch>,
-    /// `doc`'s epoch label and heads, readable without its lock (`Replica::stamps`).
+    /// `doc`'s epoch label and heads, readable without its lock (`Replica::stamps`):
+    /// for cheap reads (handshakes, keepalives, /peers). Only ever changed while holding
+    /// `doc`'s write lock, together with `doc.epoch`.
     pub(crate) stamp: watch::Receiver<Stamp>,
 
     pub(crate) lifecycle: Arc<RwLock<NodeLifecycle>>,
@@ -84,8 +81,10 @@ pub(crate) struct AppState {
     pub(crate) change_tx: watch::Sender<u64>,
 
     pub(crate) initial_sync_done: Arc<AtomicBool>,
-    /// The node is shutting down: its keepalives say it isn't ready (`peer::farewell`).
-    pub(crate) stopping: Arc<AtomicBool>,
+    /// True once the node shuts down (a signal, a failed join, a transport gone): the
+    /// HTTP API drains and its long-lived responses end, and keepalives say the node
+    /// isn't ready and is leaving (`peer::farewell`).
+    pub(crate) closing: Arc<watch::Sender<bool>>,
 
     pub(crate) change_events: broadcast::Sender<Arc<ChangeSet>>,
 
@@ -110,8 +109,20 @@ pub(crate) struct AppState {
     pub(crate) alias: Option<String>,
     /// Members alive within this window vote on remakes (`--electorate-window`).
     pub(crate) electorate_window: Duration,
+    /// This node's clock is off when it differs from most of its peers' by more than
+    /// this (`--clock-skew-warn`, `swarm::skew`); 0: never.
+    pub(crate) clock_skew_warn: Duration,
+    /// What a replay does with writes a newer one superseded (`--superseded-writes`).
+    pub(crate) superseded_writes: crate::remake::Superseded,
+    /// How far this node's clock is ahead of its peers' (their median, `swarm::skew`), as
+    /// last measured; 0 before. The propagation figures take it off.
+    pub(crate) clock_ahead_ms: Arc<std::sync::atomic::AtomicI64>,
+    /// What the clock checks remember between them (`swarm::skew`).
+    pub(crate) clock_skew: Arc<std::sync::Mutex<crate::swarm::skew::Watch>>,
     /// Chunked transfers served and fetched (`swarm::blob`).
     pub(crate) blobs: Arc<crate::swarm::blob::Blobs>,
+    /// LINKS asked and answered, for a map of the swarm (`swarm::links`).
+    pub(crate) links: Arc<crate::swarm::links::Links>,
     /// HTTP API users who may write protocol data through the store API
     /// (`--protocol-writers`).
     pub(crate) protocol_writers: Arc<std::collections::HashSet<String>>,
@@ -138,8 +149,16 @@ pub(crate) struct AppState {
     /// Our neighbours, once known (`swarm::topology`).
     pub(crate) neighbours: Arc<std::sync::RwLock<Option<crate::swarm::topology::NeighbourView>>>,
     /// Something the neighbours are computed from may have changed: the member list or a
-    /// peer's reachability (`swarm::topology::run_worker`).
+    /// peer's state (`swarm::topology::run_worker`, `swarm::topology::change_peer`).
     pub(crate) topology_changed: Arc<tokio::sync::Notify>,
+    /// Something a joining node waits for may have happened: the swarm ID learned, the
+    /// initial sync done, a rejoin ended, a seed found (`swarm::bootstrap`).
+    pub(crate) join_progress: Arc<tokio::sync::Notify>,
+    /// The forget records may have changed (`swarm::changes`): the peer worker acts on
+    /// them at once instead of at its next periodic check.
+    pub(crate) forget_records_changed: Arc<AtomicBool>,
+    /// When the peer worker is next due (`swarm::peer::Due`).
+    pub(crate) peers_due: Arc<crate::swarm::peer::Due>,
 
     /// While rejoining a newer epoch, the previous document, served read-only (and
     /// marked stale) instead of answering 503.
@@ -175,6 +194,29 @@ pub(crate) struct CatchUpSource {
 }
 
 impl AppState {
+    /// Asks the sync worker to look at `peer` (`SyncEvent::PeerNeedsSync`): whether to
+    /// send it a sync message now. Nothing if the worker is gone (the node stops).
+    pub(crate) fn request_sync(&self, peer: &str) {
+        let _ = self
+            .sync_events
+            .send(SyncEvent::PeerNeedsSync(peer.to_owned()));
+    }
+
+    /// `request_sync`, after `delay`.
+    pub(crate) fn request_sync_after(&self, peer: &str, delay: std::time::Duration) {
+        let events = self.sync_events.clone();
+        let peer = peer.to_owned();
+        tokio::spawn(async move {
+            tokio::time::sleep(delay).await;
+            let _ = events.send(SyncEvent::PeerNeedsSync(peer));
+        });
+    }
+
+    /// Whether the node shuts down (`closing`).
+    pub(crate) fn is_closing(&self) -> bool {
+        *self.closing.borrow()
+    }
+
     pub(crate) fn local_id(&self) -> &str {
         self.transport.local_id()
     }
@@ -198,7 +240,7 @@ impl AppState {
     /// The local epoch. Code that labels or applies sync messages must use the epoch
     /// of the locked `Replica` instead, so label and document can't disagree.
     pub(crate) fn local_epoch(&self) -> Epoch {
-        *self.epoch_tx.borrow()
+        self.stamp.borrow().epoch
     }
 
     /// The local epoch and `Replica::epoch_seq`, read together, without waiting for the
@@ -242,6 +284,7 @@ impl AppState {
                     discovered_swarm_id: Some(swarm_id),
                 };
                 debug!("Node transitioned from {old_lifecycle:?} {lifecycle:?}");
+                self.join_progress.notify_one();
                 Ok(())
             }
             NodeLifecycle::Joining {
@@ -281,6 +324,8 @@ impl AppState {
                 let old_lifecycle = lifecycle.clone();
                 *lifecycle = NodeLifecycle::Ready { swarm_id };
                 debug!("Node transitioned from {old_lifecycle:?} {lifecycle:?}");
+                // Dormancy is judged once ready, and the periodic checks run at once.
+                self.peers_due.now();
             }
             NodeLifecycle::Ready { swarm_id: existing } if existing == &swarm_id => {
                 debug!("Node already ready with the same identity");
@@ -328,6 +373,7 @@ impl AppState {
     /// State for a peer we haven't known so far (not inserted).
     pub(crate) fn new_peer_state(&self, pk: &str) -> Arc<Mutex<PeerState>> {
         let leads = crate::swarm::pair::leads(self.local_id(), pk);
+        self.peers_due.now();
         Arc::new(Mutex::new(PeerState::new(leads, &self.timing)))
     }
 
@@ -340,7 +386,7 @@ impl AppState {
     /// Reads the document on the blocking pool; `finish` runs after the lock is released.
     pub(crate) async fn render<T, R>(
         &self,
-        read: impl FnOnce(&AutoCommit) -> T + Send + 'static,
+        read: impl FnOnce(&Document) -> T + Send + 'static,
         finish: impl FnOnce(T) -> R + Send + 'static,
     ) -> R
     where
@@ -358,59 +404,73 @@ impl AppState {
         .expect("document read task panicked")
     }
 
-    /// Mutates the document on the blocking pool.
-    pub(crate) async fn write_doc<T>(
+    /// 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. While a rejoin runs,
+    /// the write goes into the document before it (`StaleView`), replayed when the
+    /// rejoin ends; on a node that keeps none, it is refused (`StoreError::Rejoining`,
+    /// `f` not called).
+    pub(crate) async fn mutate_doc<T, E>(
         &self,
-        f: impl FnOnce(&mut AutoCommit) -> T + Send + 'static,
-    ) -> T
+        f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E> + Send + 'static,
+    ) -> Result<T, E>
     where
         T: Send + 'static,
+        E: From<StoreError> + 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")
+        self.mutate_doc_for(None, f).await.map(|(out, _)| out)
     }
 
-    /// Mutates the document on the blocking pool as one Automerge change, then
-    /// publishes it. Write and notification run in the same blocking task, so a
-    /// cancelled request can't leave a committed write unannounced.
-    pub(crate) async fn mutate_doc<T, E>(
+    /// [`mutate_doc`](Self::mutate_doc) for `user`, who wrote through the API: the
+    /// change names them (`replica::Writer`). With the heads of the document written to
+    /// after the write (sorted): what a read must have to include it.
+    pub(crate) async fn mutate_doc_for<T, E>(
         &self,
-        f: impl FnOnce(&mut AutoCommit) -> Result<T, E> + Send + 'static,
-    ) -> Result<T, E>
+        user: Option<String>,
+        f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E> + Send + 'static,
+    ) -> Result<(T, Vec<crate::document::ChangeHash>), E>
     where
         T: Send + 'static,
-        E: Send + 'static,
+        E: From<StoreError> + Send + 'static,
     {
         let state = self.clone();
 
         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 Automerge change.
-                    Ok(_) => {
-                        // Stamped with our clock (milliseconds), so receivers can tell
-                        // how long the change took to reach them (`swarm::stats`).
-                        replica.doc.commit_with(
-                            automerge::transaction::CommitOptions::default()
-                                .with_time(state.clock.now_millis().as_u64() as i64),
-                        );
-                    }
-                    // Discard partial ops (e.g. old value deleted, new one not written).
-                    Err(_) => {
-                        replica.doc.rollback();
-                    }
+                // Checked here, under the lock: the caller's readiness check (if any)
+                // was before a rejoin could swap in a fresh document.
+                if !replica.writable {
+                    // Rejoining: into the document before the rejoin, which the rejoin
+                    // replays when it ends (`StaleView`); refused by a node that keeps
+                    // none (it wasn't ready).
+                    let Some((_, stale)) = state.stale_document() else {
+                        return Err(E::from(StoreError::Rejoining));
+                    };
+                    let mut doc = stale.write_or_recover();
+                    let writer = crate::document::replica::Writer {
+                        node: state.local_id(),
+                        user: user.as_deref(),
+                    };
+                    let out = crate::document::replica::transact(
+                        &mut doc,
+                        state.clock.now_millis(),
+                        &writer,
+                        f,
+                    )?;
+                    let mut heads = doc.get_heads();
+                    heads.sort_unstable();
+                    return Ok((out, heads));
                 }
+                // One API write = one change; on error, its partial writes (e.g. the old
+                // value deleted, the new one not written) are dropped.
+                let writer = crate::document::replica::Writer {
+                    node: state.local_id(),
+                    user: user.as_deref(),
+                };
+                let result = replica.transact(state.clock.now_millis(), &writer, f);
                 replica.refresh_heads();
-                result
+                result.map(|out| (out, replica.heads().to_vec()))
             }; // lock released before notifying
 
             if result.is_ok() {