document: move knowledge about node to swarm

5 files changed, 189 insertions, 134 deletions

diff --git a/src/document/file.rs b/src/document/file.rs
index ed29e9b..47d7310 100644
--- a/src/document/file.rs
+++ b/src/document/file.rs
@@ -7,8 +7,8 @@
 //! document, then increments (the changes after the heads written before). A record cut
 //! short by a crash is dropped when reading.
 //!
-//! Written by a worker (`run`), only while the node is ready and not rejoining (a rejoin
-//! keeps the old file until it is done):
+//! Written by a worker (`swarm::document_file`), only while the node is ready and not
+//! rejoining (a rejoin keeps the old file until it is done):
 //! * every `--document-append`, what changed is appended;
 //! * the whole file is rewritten, atomically, when the epoch or this node's actors
 //!   changed (a switch, a rejoin, a restart), and when the appended increments outgrew
@@ -29,7 +29,6 @@ use tracing::{debug, error, warn};
 use crate::document::replica::Epoch;
 use crate::files::{FileKind, Files};
 use crate::locks::MutexExt;
-use crate::state::AppState;
 
 const MAGIC: &[u8] = b"continuum-document/1\n";
 
@@ -180,147 +179,146 @@ impl DocumentFile {
     }
 }
 
-/// Keeps the document file up to date. Runs for as long as the node does.
-pub(crate) async fn run(state: AppState) {
-    let Some(file) = state.document_file.clone() else {
-        return;
-    };
-    let mut tick = tokio::time::interval(file.append_every);
-    tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
-    loop {
-        tick.tick().await;
-        flush(&state).await;
-    }
+/// What the document file needs now (`DocumentFile::due`).
+pub(crate) enum Due {
+    /// Written whole: for this reason.
+    Rewrite(&'static str),
+    /// What changed since `since` appended, as long as the document is still of `epoch`
+    /// (an increment of another epoch's document must never be appended).
+    Append {
+        epoch: Epoch,
+        since: Vec<ChangeHash>,
+    },
+    Nothing,
 }
 
-/// Writes what is due now (see the module docs). Also called at shutdown.
-pub(crate) async fn flush(state: &AppState) {
-    let Some(file) = state.document_file.clone() else {
-        return;
-    };
-    if !state.is_ready().await || state.remake.is_rejoining() {
-        return;
-    }
-    let own = state.remake.own_actors();
-    let (epoch, heads) = {
-        let replica = state.doc.read().await;
-        (replica.epoch, replica.heads().to_vec())
-    };
-    let written = file.lock().clone();
-    let reason = match &written {
-        None => Some("first write"),
-        Some(w) if w.epoch != epoch => Some("new epoch"),
-        Some(w) if w.actors != own.len() => Some("new actor"),
-        Some(w)
-            if w.tail_bytes > w.base_bytes && w.rewritten_at.elapsed() >= file.rewrite_every =>
-        {
-            Some("increments outgrew the full save")
-        }
-        Some(_) => None,
-    };
-    match (reason, written) {
-        (Some(reason), _) => rewrite(state, &file, own, reason).await,
-        (None, Some(written)) if written.heads != heads => append(state, &file, written).await,
-        (None, _) => {}
+impl DocumentFile {
+    /// How often what changed is appended (`--document-append`).
+    pub(crate) fn append_every(&self) -> Duration {
+        self.append_every
     }
-}
 
-async fn rewrite(state: &AppState, file: &Arc<DocumentFile>, own: Vec<ActorId>, reason: &str) {
-    let start = Instant::now();
-    let (epoch, mut copy) = {
-        let replica = state.doc.read().await;
-        (replica.epoch, replica.doc.clone())
-    };
-    let own_count = own.len();
-    let writer = Arc::clone(file);
-    let written = tokio::task::spawn_blocking(move || -> Result<(Vec<ChangeHash>, u64)> {
-        let heads = copy.get_heads();
-        // Only the actors with changes in this document matter for a later rejoin;
-        // the current one in any case.
-        let present: HashSet<ActorId> = copy
-            .get_changes(&[])
-            .iter()
-            .map(|change| change.actor_id().clone())
-            .collect();
-        let current = copy.get_actor().clone();
-        let actors: Vec<ActorId> = own
-            .into_iter()
-            .filter(|actor| *actor == current || present.contains(actor))
-            .collect();
-        let save = copy.save();
-        let bytes = whole(&actors, &save)?;
-        writer
-            .files
-            .write_atomic(FileKind::Document, &writer.path, &bytes)?;
-        Ok((heads, save.len() as u64))
-    })
-    .await;
-    match written {
-        Ok(Ok((heads, base_bytes))) => {
-            debug!(
-                reason,
-                bytes = base_bytes,
-                took_ms = start.elapsed().as_millis() as u64,
-                "Rewrote the document file"
-            );
-            *file.lock() = Some(Written {
-                epoch,
-                actors: own_count,
-                heads,
-                base_bytes,
-                tail_bytes: 0,
-                rewritten_at: Instant::now(),
-            });
-        }
-        Ok(Err(error)) => {
-            error!(path = %file.path.display(), error = %format!("{error:#}"), "Couldn't rewrite the document file");
+    /// What is due for a document of `epoch` at `heads`, written by this node's `actors`
+    /// (see the module docs).
+    pub(crate) fn due(&self, epoch: Epoch, actors: usize, heads: &[ChangeHash]) -> Due {
+        let written = self.lock().clone();
+        let reason = match &written {
+            None => Some("first write"),
+            Some(w) if w.epoch != epoch => Some("new epoch"),
+            Some(w) if w.actors != actors => Some("new actor"),
+            Some(w)
+                if w.tail_bytes > w.base_bytes
+                    && w.rewritten_at.elapsed() >= self.rewrite_every =>
+            {
+                Some("increments outgrew the full save")
+            }
+            Some(_) => None,
+        };
+        match (reason, written) {
+            (Some(reason), _) => Due::Rewrite(reason),
+            (None, Some(written)) if written.heads != heads => Due::Append {
+                epoch: written.epoch,
+                since: written.heads,
+            },
+            (None, _) => Due::Nothing,
         }
-        Err(error) => error!(error = %error, "Document file rewrite task failed"),
     }
-}
 
-async fn append(state: &AppState, file: &Arc<DocumentFile>, written: Written) {
-    let (heads, increment) = {
-        let mut replica = state.doc.write().await;
-        // Switched since `flush` looked: the next one rewrites the file instead. An
-        // increment of another epoch's document must never be appended.
-        if replica.epoch != written.epoch {
-            return;
-        }
-        let heads = replica.heads().to_vec();
-        (heads, replica.doc.save_after(&written.heads))
-    };
-    let bytes = record(&increment);
-    let writer = Arc::clone(file);
-    let appended = tokio::task::spawn_blocking(move || {
-        writer
-            .files
-            .append(FileKind::Document, &writer.path, &bytes)
-    })
-    .await;
-    match appended {
-        Ok(Ok(())) => {
-            let mut guard = file.lock();
-            if let Some(current) = guard
-                .as_mut()
-                .filter(|current| current.heads == written.heads)
-            {
-                current.heads = heads;
-                current.tail_bytes += increment.len() as u64 + 4;
+    /// Writes the whole file, atomically: `copy` (a copy of the document, of `epoch`)
+    /// and this node's actors `own` that wrote in it.
+    pub(crate) async fn rewrite(
+        self: &Arc<Self>,
+        epoch: Epoch,
+        mut copy: AutoCommit,
+        own: Vec<ActorId>,
+        reason: &str,
+    ) {
+        let start = Instant::now();
+        let own_count = own.len();
+        let writer = Arc::clone(self);
+        let written = tokio::task::spawn_blocking(move || -> Result<(Vec<ChangeHash>, u64)> {
+            let heads = copy.get_heads();
+            // Only the actors with changes in this document matter for a later rejoin;
+            // the current one in any case.
+            let present: HashSet<ActorId> = copy
+                .get_changes(&[])
+                .iter()
+                .map(|change| change.actor_id().clone())
+                .collect();
+            let current = copy.get_actor().clone();
+            let actors: Vec<ActorId> = own
+                .into_iter()
+                .filter(|actor| *actor == current || present.contains(actor))
+                .collect();
+            let save = copy.save();
+            let bytes = whole(&actors, &save)?;
+            writer
+                .files
+                .write_atomic(FileKind::Document, &writer.path, &bytes)?;
+            Ok((heads, save.len() as u64))
+        })
+        .await;
+        match written {
+            Ok(Ok((heads, base_bytes))) => {
                 debug!(
-                    bytes = increment.len(),
-                    tail_bytes = current.tail_bytes,
-                    base_bytes = current.base_bytes,
-                    "Appended to the document file"
+                    reason,
+                    bytes = base_bytes,
+                    took_ms = start.elapsed().as_millis() as u64,
+                    "Rewrote the document file"
                 );
+                *self.lock() = Some(Written {
+                    epoch,
+                    actors: own_count,
+                    heads,
+                    base_bytes,
+                    tail_bytes: 0,
+                    rewritten_at: Instant::now(),
+                });
             }
+            Ok(Err(error)) => {
+                error!(path = %self.path.display(), error = %format!("{error:#}"), "Couldn't rewrite the document file");
+            }
+            Err(error) => error!(error = %error, "Document file rewrite task failed"),
         }
-        Ok(Err(error)) => {
-            // The file may end in a partial record now; the next rewrite replaces it.
-            error!(path = %file.path.display(), error = %error, "Couldn't append to the document file; rewriting it next");
-            *file.lock() = None;
+    }
+
+    /// Appends `increment`: what changed from `since` (the heads written last) to
+    /// `heads`.
+    pub(crate) async fn append(
+        self: &Arc<Self>,
+        since: Vec<ChangeHash>,
+        heads: Vec<ChangeHash>,
+        increment: Vec<u8>,
+    ) {
+        let bytes = record(&increment);
+        let writer = Arc::clone(self);
+        let appended = tokio::task::spawn_blocking(move || {
+            writer
+                .files
+                .append(FileKind::Document, &writer.path, &bytes)
+        })
+        .await;
+        match appended {
+            Ok(Ok(())) => {
+                let mut guard = self.lock();
+                if let Some(current) = guard.as_mut().filter(|current| current.heads == since) {
+                    current.heads = heads;
+                    current.tail_bytes += increment.len() as u64 + 4;
+                    debug!(
+                        bytes = increment.len(),
+                        tail_bytes = current.tail_bytes,
+                        base_bytes = current.base_bytes,
+                        "Appended to the document file"
+                    );
+                }
+            }
+            Ok(Err(error)) => {
+                // The file may end in a partial record now; the next rewrite replaces it.
+                error!(path = %self.path.display(), error = %error, "Couldn't append to the document file; rewriting it next");
+                *self.lock() = None;
+            }
+            Err(error) => error!(error = %error, "Document file append task failed"),
         }
-        Err(error) => error!(error = %error, "Document file append task failed"),
     }
 }
 
diff --git a/src/main.rs b/src/main.rs
index 8d4ce73..d7c28a2 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -315,7 +315,7 @@ async fn main() -> Result<()> {
     }
     swarm::peer::farewell(&stopping).await;
     swarm::state_file::update(&stopping, true).await;
-    document::file::flush(&stopping).await;
+    swarm::document_file::flush(&stopping).await;
     Ok(())
 }
 
diff --git a/src/node.rs b/src/node.rs
index acf1dc5..fda317c 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -195,5 +195,5 @@ impl Node {
 pub(crate) fn spawn_ready_workers(state: &AppState) {
     let publisher_state = state.clone();
     tokio::spawn(async move { changes::run_change_publisher(publisher_state).await });
-    tokio::spawn(crate::document::file::run(state.clone()));
+    tokio::spawn(crate::swarm::document_file::run(state.clone()));
 }
diff --git a/src/swarm/document_file.rs b/src/swarm/document_file.rs
new file mode 100644
index 0000000..188b2b6
--- /dev/null
+++ b/src/swarm/document_file.rs
@@ -0,0 +1,56 @@
+//! The worker that keeps the document file (`document::file`) up to date: it looks at
+//! the node (ready, not rejoining, the document's epoch and heads, this node's actors),
+//! and the file says what is due.
+
+use crate::document::file::Due;
+use crate::state::AppState;
+
+/// Keeps the document file up to date. Runs for as long as the node does.
+pub(crate) async fn run(state: AppState) {
+    let Some(file) = state.document_file.clone() else {
+        return;
+    };
+    let mut tick = tokio::time::interval(file.append_every());
+    tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+    loop {
+        tick.tick().await;
+        flush(&state).await;
+    }
+}
+
+/// Writes what is due now (`DocumentFile::due`). Also called at shutdown.
+pub(crate) async fn flush(state: &AppState) {
+    let Some(file) = state.document_file.clone() else {
+        return;
+    };
+    if !state.is_ready().await || state.remake.is_rejoining() {
+        return;
+    }
+    let own = state.remake.own_actors();
+    let (epoch, heads) = {
+        let replica = state.doc.read().await;
+        (replica.epoch, replica.heads().to_vec())
+    };
+    match file.due(epoch, own.len(), &heads) {
+        Due::Rewrite(reason) => {
+            let (epoch, copy) = {
+                let replica = state.doc.read().await;
+                (replica.epoch, replica.doc.clone())
+            };
+            file.rewrite(epoch, copy, own, reason).await;
+        }
+        Due::Append { epoch, since } => {
+            let (heads, increment) = {
+                let mut replica = state.doc.write().await;
+                // Switched since: the next flush rewrites the file instead.
+                if replica.epoch != epoch {
+                    return;
+                }
+                let heads = replica.heads().to_vec();
+                (heads, replica.doc.save_after(&since))
+            };
+            file.append(since, heads, increment).await;
+        }
+        Due::Nothing => {}
+    }
+}
diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs
index 9cdf154..f36af9c 100644
--- a/src/swarm/mod.rs
+++ b/src/swarm/mod.rs
@@ -5,6 +5,7 @@ pub(crate) mod auth;
 pub(crate) mod blob;
 pub(crate) mod bootstrap;
 pub(crate) mod changes;
+pub(crate) mod document_file;
 pub(crate) mod links;
 pub(crate) mod members;
 pub(crate) mod pair;