From 720a349d04a7a267461b21679ee2f748b8d8e957 Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 31 Oct 2026 04:58:46 +0100 Subject: document: move knowledge about node to swarm --- src/swarm/document_file.rs | 56 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 56 insertions(+) create mode 100644 src/swarm/document_file.rs (limited to 'src/swarm/document_file.rs') 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 => {} + } +} -- cgit v1.3.1