From 6c7809103f62c89fc10cc82228a28a0f39db5277 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/document/file.rs | 262 +++++++++++++++++++++++++-------------------------- 1 file changed, 130 insertions(+), 132 deletions(-) (limited to 'src/document/file.rs') 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, + }, + 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, own: Vec, 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, 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 = copy - .get_changes(&[]) - .iter() - .map(|change| change.actor_id().clone()) - .collect(); - let current = copy.get_actor().clone(); - let actors: Vec = 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, 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, + epoch: Epoch, + mut copy: AutoCommit, + own: Vec, + 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, 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 = copy + .get_changes(&[]) + .iter() + .map(|change| change.actor_id().clone()) + .collect(); + let current = copy.get_actor().clone(); + let actors: Vec = 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, + since: Vec, + heads: Vec, + increment: Vec, + ) { + 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"), } } -- cgit v1.3.1