document: move knowledge about node to swarm

1 files changed, 56 insertions, 0 deletions

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 => {}
+    }
+}