allow to persist document

9 files changed, 670 insertions, 84 deletions

diff --git a/src/api/stats.rs b/src/api/stats.rs
index 8899421..fa11b45 100644
--- a/src/api/stats.rs
+++ b/src/api/stats.rs
@@ -49,6 +49,8 @@ struct DiskStats {
     read_bytes: u64,
     writes: u64,
     written_bytes: u64,
+    appends: u64,
+    appended_bytes: u64,
     errors: u64,
 }
 
@@ -220,6 +222,8 @@ async fn collect(state: &AppState) -> StatsResponse {
                     read_bytes: counts.read_bytes,
                     writes: counts.writes,
                     written_bytes: counts.written_bytes,
+                    appends: counts.appends,
+                    appended_bytes: counts.appended_bytes,
                     errors: counts.errors,
                 };
                 (kind.as_str(), stats)
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index e267042..74c18e9 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -54,6 +54,45 @@ impl Out {
     }
 }
 
+const DISK_COUNTERS: usize = 7;
+
+/// Name, help and value of each counter of one kind of file.
+fn disk_counters(c: &super::DiskStats) -> [(&'static str, &'static str, u64); DISK_COUNTERS] {
+    [
+        (
+            "disk_reads_total",
+            "Reads of this node's own files.",
+            c.reads,
+        ),
+        (
+            "disk_read_bytes_total",
+            "Bytes read from them.",
+            c.read_bytes,
+        ),
+        ("disk_writes_total", "Whole files written.", c.writes),
+        (
+            "disk_written_bytes_total",
+            "Bytes of them.",
+            c.written_bytes,
+        ),
+        (
+            "disk_appends_total",
+            "Appends to this node's own files.",
+            c.appends,
+        ),
+        (
+            "disk_appended_bytes_total",
+            "Bytes appended.",
+            c.appended_bytes,
+        ),
+        (
+            "disk_errors_total",
+            "Failed reads, writes and appends.",
+            c.errors,
+        ),
+    ]
+}
+
 fn escape(value: &str) -> String {
     value
         .replace('\\', "\\\\")
@@ -297,32 +336,13 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String {
         stats.stale_hello_replies as f64,
     );
 
-    type Field = fn(&super::DiskStats) -> u64;
-    let disk: [(&str, &str, Field); 5] = [
-        ("disk_reads_total", "Reads of this node's own files.", |c| {
-            c.reads
-        }),
-        ("disk_read_bytes_total", "Bytes read from them.", |c| {
-            c.read_bytes
-        }),
-        (
-            "disk_writes_total",
-            "Writes of this node's own files.",
-            |c| c.writes,
-        ),
-        ("disk_written_bytes_total", "Bytes written to them.", |c| {
-            c.written_bytes
-        }),
-        (
-            "disk_errors_total",
-            "Failed reads and writes of them.",
-            |c| c.errors,
-        ),
-    ];
-    for (name, help, field) in disk {
-        out.family(name, "counter", help);
-        for (kind, counts) in &stats.disk {
-            out.sample(name, &[("file", *kind)], field(counts) as f64);
+    for counter in 0..DISK_COUNTERS {
+        for (i, (kind, counts)) in stats.disk.iter().enumerate() {
+            let (name, help, value) = disk_counters(counts)[counter];
+            if i == 0 {
+                out.family(name, "counter", help);
+            }
+            out.sample(name, &[("file", *kind)], value as f64);
         }
     }
 
diff --git a/src/config/mod.rs b/src/config/mod.rs
index c83ca70..c56afa6 100644
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@@ -66,7 +66,7 @@ const fn option(
 }
 
 /// What `--force` may override at start (`force`).
-pub(crate) const FORCE: &[&str] = &["swarmId"];
+pub(crate) const FORCE: &[&str] = &["swarmId", "badDocumentFile", "replaceDocumentFile"];
 
 /// Every option. Names are the long option names and the configuration file keys.
 pub(crate) const OPTIONS: &[OptionSpec] = &[
@@ -117,7 +117,7 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[
         Kind::List,
         "NAME,[NAME...]",
         None,
-        "Override checks that otherwise refuse to start: true for all, or names: swarmId (a backup of another swarm than the state file remembers).",
+        "Override checks that otherwise refuse to start: true for all, or names: swarmId (a document to start on is of another swarm than the state file remembers), badDocumentFile (an unusable document file: moved aside, start without it), replaceDocumentFile (recover-from while the document file holds a document).",
     ),
     option(
         "recover-from",
@@ -127,6 +127,27 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[
         "Start the swarm again from a backup (as GET /admin/document saves it): this node loads it, on the swarm and epoch it names, and is ready at once; other nodes join it. Members of the swarm still running merge with it, or it rejoins them if they moved on to a newer epoch (the backup is then dropped).",
     ),
     option(
+        "document-file",
+        Kind::Text,
+        "PATH",
+        None,
+        "This node's document on disk. Read at start: the node resumes with it, ready at once. Kept up to date while ready: changes are appended every --document-append-secs, and the file is rewritten after a switch, a rejoin or a restart, and when the appended changes outgrew it.",
+    ),
+    option(
+        "document-append-secs",
+        Kind::Integer,
+        "SECONDS",
+        Some("5"),
+        "How often changes are appended to the document file.",
+    ),
+    option(
+        "document-rewrite-secs",
+        Kind::Integer,
+        "SECONDS",
+        Some("3600"),
+        "Shortest time between rewrites of the document file because its appended changes outgrew its full save (a new epoch or a restart rewrites it at once).",
+    ),
+    option(
         "state-file",
         Kind::Text,
         "PATH",
@@ -247,6 +268,11 @@ pub(crate) struct Settings {
     #[serde(default)]
     pub(crate) genesis: bool,
     pub(crate) state_file: Option<PathBuf>,
+    pub(crate) document_file: Option<PathBuf>,
+    #[serde(default = "default_document_append_secs")]
+    pub(crate) document_append_secs: u64,
+    #[serde(default = "default_document_rewrite_secs")]
+    pub(crate) document_rewrite_secs: u64,
     pub(crate) recover_from: Option<PathBuf>,
     #[serde(default)]
     pub(crate) force: Vec<String>,
@@ -273,6 +299,12 @@ pub(crate) struct Settings {
 fn default_bind_addr() -> SocketAddr {
     SocketAddr::from(([127, 0, 0, 1], 3000))
 }
+fn default_document_append_secs() -> u64 {
+    5
+}
+fn default_document_rewrite_secs() -> u64 {
+    3600
+}
 fn default_keepalive_secs() -> u64 {
     60
 }
@@ -340,6 +372,9 @@ impl Settings {
         if self.neighbours < 2 {
             bail!("neighbours must be at least 2");
         }
+        if self.document_append_secs == 0 || self.document_rewrite_secs == 0 {
+            bail!("document-append-secs and document-rewrite-secs must be at least 1");
+        }
         if self.keepalive_secs == 0 {
             bail!("keepalive-secs must be at least 1");
         }
diff --git a/src/document/file.rs b/src/document/file.rs
new file mode 100644
index 0000000..a773502
--- /dev/null
+++ b/src/document/file.rs
@@ -0,0 +1,362 @@
+//! The document file (`--document-file`): this node's document on disk, so a restart
+//! resumes with it instead of starting empty.
+//!
+//! Format: the line `continuum-document/1`, a header line (JSON: the actor IDs this
+//! node wrote with, so a later rejoin replays those writes as its own), then records,
+//! each a little-endian `u32` length and that many bytes: first a full save of the
+//! 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):
+//! * every `--document-append-secs`, 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
+//!   the full save, at most every `--document-rewrite-secs`;
+//! * at shutdown, what changed is appended.
+
+use std::collections::HashSet;
+use std::path::{Path, PathBuf};
+use std::sync::{Arc, Mutex, MutexGuard};
+use std::time::Duration;
+
+use anyhow::{Context, Result, anyhow, bail};
+use automerge::{ActorId, AutoCommit, ChangeHash};
+use serde::{Deserialize, Serialize};
+use tokio::time::Instant;
+use tracing::{debug, error, warn};
+
+use crate::document::replica::Epoch;
+use crate::files::{FileKind, Files};
+use crate::state::AppState;
+
+const MAGIC: &[u8] = b"continuum-document/1\n";
+
+#[derive(Debug, Default, Serialize, Deserialize)]
+struct Header {
+    /// Hex actor IDs.
+    #[serde(default)]
+    actors: Vec<String>,
+}
+
+/// A document file as read at start.
+pub(crate) struct Loaded {
+    /// Under a fresh actor.
+    pub(crate) doc: AutoCommit,
+    /// The actors this node wrote with before.
+    pub(crate) actors: Vec<ActorId>,
+}
+
+/// Reads the document file; None if there is none. An error if it can't be read or
+/// isn't one: the caller decides (`--force badDocumentFile`).
+pub(crate) fn load(path: &Path, files: &Files) -> Result<Option<Loaded>> {
+    let Some(bytes) = files
+        .read(FileKind::Document, path)
+        .with_context(|| format!("Reading the document file {}", path.display()))?
+    else {
+        return Ok(None);
+    };
+    let (loaded, dropped) =
+        parse(&bytes).with_context(|| format!("The document file {}", path.display()))?;
+    if dropped > 0 {
+        warn!(
+            path = %path.display(),
+            bytes = dropped,
+            "Dropped a record cut short at the end of the document file (a crash while appending)"
+        );
+    }
+    Ok(Some(loaded))
+}
+
+/// The file's content, and how many bytes at its end were cut short.
+fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> {
+    let rest = bytes
+        .strip_prefix(MAGIC)
+        .ok_or_else(|| anyhow!("not a continuum document file"))?;
+    let line_end = rest
+        .iter()
+        .position(|byte| *byte == b'\n')
+        .ok_or_else(|| anyhow!("no header"))?;
+    let header: Header = serde_json::from_slice(&rest[..line_end]).context("header")?;
+    let actors = header
+        .actors
+        .iter()
+        .map(|hex| {
+            ActorId::try_from(hex.as_str())
+                .map_err(|_| anyhow!("invalid actor {hex:?} in the header"))
+        })
+        .collect::<Result<Vec<_>>>()?;
+
+    let mut records = &rest[line_end + 1..];
+    let mut doc: Option<AutoCommit> = None;
+    while !records.is_empty() {
+        let Some((length, after)) = records.split_first_chunk::<4>() else {
+            break;
+        };
+        let length = u32::from_le_bytes(*length) as usize;
+        if after.len() < length {
+            break;
+        }
+        let (record, after) = after.split_at(length);
+        match &mut doc {
+            None => doc = Some(AutoCommit::load(record).context("the full save")?),
+            Some(doc) => {
+                doc.load_incremental(record).context("an increment")?;
+            }
+        }
+        records = after;
+    }
+    let Some(doc) = doc else {
+        bail!("no full save");
+    };
+    let loaded = Loaded {
+        doc: doc.with_actor(ActorId::random()),
+        actors,
+    };
+    Ok((loaded, records.len()))
+}
+
+fn record(bytes: &[u8]) -> Vec<u8> {
+    let mut out = Vec::with_capacity(bytes.len() + 4);
+    out.extend_from_slice(&(bytes.len() as u32).to_le_bytes());
+    out.extend_from_slice(bytes);
+    out
+}
+
+/// A whole file: header and one full save.
+fn whole(actors: &[ActorId], save: &[u8]) -> Result<Vec<u8>> {
+    let header = Header {
+        actors: actors.iter().map(ActorId::to_hex_string).collect(),
+    };
+    let mut out = MAGIC.to_vec();
+    out.extend(serde_json::to_vec(&header)?);
+    out.push(b'\n');
+    out.extend(record(save));
+    Ok(out)
+}
+
+/// The document file and what was last written to it.
+pub(crate) struct DocumentFile {
+    path: PathBuf,
+    files: Arc<Files>,
+    append_every: Duration,
+    rewrite_every: Duration,
+    // A std mutex: never held across an await.
+    written: Mutex<Option<Written>>,
+}
+
+#[derive(Debug, Clone)]
+struct Written {
+    epoch: Epoch,
+    /// This node's actors when the file was last rewritten (`RemakeState::own_actors`).
+    actors: usize,
+    heads: Vec<ChangeHash>,
+    /// The full save.
+    base_bytes: u64,
+    /// Appended since.
+    tail_bytes: u64,
+    rewritten_at: Instant,
+}
+
+impl DocumentFile {
+    pub(crate) fn new(
+        path: PathBuf,
+        files: Arc<Files>,
+        append_every: Duration,
+        rewrite_every: Duration,
+    ) -> Self {
+        Self {
+            path,
+            files,
+            append_every,
+            rewrite_every,
+            written: Mutex::new(None),
+        }
+    }
+
+    fn lock(&self) -> MutexGuard<'_, Option<Written>> {
+        self.written
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+    }
+}
+
+/// 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 (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, _) => {}
+    }
+}
+
+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");
+        }
+        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;
+                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 = %file.path.display(), error = %error, "Couldn't append to the document file; rewriting it next");
+            *file.lock() = None;
+        }
+        Err(error) => error!(error = %error, "Document file append task failed"),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use automerge::{ROOT, ReadDoc, transaction::Transactable};
+
+    #[test]
+    fn whole_then_increments() {
+        let mut doc = AutoCommit::new();
+        doc.put(ROOT, "a", 1_i64).unwrap();
+        doc.commit();
+        let actor = doc.get_actor().clone();
+        let heads = doc.get_heads();
+        let mut bytes = whole(std::slice::from_ref(&actor), &doc.save()).unwrap();
+
+        doc.put(ROOT, "b", 2_i64).unwrap();
+        doc.commit();
+        bytes.extend(record(&doc.save_after(&heads)));
+
+        let (loaded, dropped) = parse(&bytes).unwrap();
+        assert_eq!(dropped, 0);
+        assert_eq!(loaded.actors, [actor.clone()]);
+        assert_ne!(loaded.doc.get_actor(), &actor);
+        assert!(loaded.doc.get(ROOT, "b").unwrap().is_some());
+
+        // A record cut short by a crash is dropped.
+        let mut cut = bytes.clone();
+        cut.extend(record(b"whatever")[..6].iter());
+        let (loaded, dropped) = parse(&cut).unwrap();
+        assert_eq!(dropped, 6);
+        assert!(loaded.doc.get(ROOT, "b").unwrap().is_some());
+
+        assert!(parse(b"something else").is_err());
+        assert!(parse(&bytes[..MAGIC.len() + 3]).is_err());
+    }
+}
diff --git a/src/document/mod.rs b/src/document/mod.rs
index ce24815..0da1d67 100644
--- a/src/document/mod.rs
+++ b/src/document/mod.rs
@@ -1,5 +1,6 @@
 //! The replicated document: the replica with its epoch, and the store semantics
 //! (merged-view paths over JSON) the HTTP API and remakes work with.
 
+pub(crate) mod file;
 pub mod replica;
 pub mod store;
diff --git a/src/files.rs b/src/files.rs
index 81c8b1e..7bf9117 100644
--- a/src/files.rs
+++ b/src/files.rs
@@ -1,9 +1,12 @@
 //! The files this node reads and writes for itself, and how much it did (`/admin/stats`,
-//! `disk`): the state file (`swarm::state_file`) and a backup to recover from
-//! (`swarm::bootstrap`). Not the Mycelium daemon's files, configuration or logs.
+//! `disk`): the state file (`swarm::state_file`), the document file
+//! (`document::file`) and a backup to recover from (`swarm::bootstrap`). Not the
+//! Mycelium daemon's files, configuration or logs.
 //!
 //! Writes replace a file atomically: a temporary file next to it, flushed to disk, then
-//! renamed over it, so a crash leaves the old content or the new one.
+//! renamed over it, so a crash leaves the old content or the new one. Appends add to the
+//! end and are flushed to disk before they count as done; a crash may leave the last one
+//! cut short, which the file's reader has to expect.
 
 use std::io::Write;
 use std::path::{Path, PathBuf};
@@ -13,15 +16,17 @@ use std::sync::{Mutex, MutexGuard};
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub(crate) enum FileKind {
     State,
+    Document,
     Backup,
 }
 
 impl FileKind {
-    pub(crate) const ALL: [FileKind; 2] = [FileKind::State, FileKind::Backup];
+    pub(crate) const ALL: [FileKind; 3] = [FileKind::State, FileKind::Document, FileKind::Backup];
 
     pub(crate) fn as_str(self) -> &'static str {
         match self {
             FileKind::State => "state",
+            FileKind::Document => "document",
             FileKind::Backup => "backup",
         }
     }
@@ -36,9 +41,12 @@ impl FileKind {
 pub(crate) struct Counts {
     pub(crate) reads: u64,
     pub(crate) read_bytes: u64,
+    /// Whole files written.
     pub(crate) writes: u64,
     pub(crate) written_bytes: u64,
-    /// Reads or writes that failed (a missing file read is no failure).
+    pub(crate) appends: u64,
+    pub(crate) appended_bytes: u64,
+    /// Reads, writes or appends that failed (a missing file read is no failure).
     pub(crate) errors: u64,
 }
 
@@ -96,6 +104,23 @@ impl Files {
         result
     }
 
+    /// Adds `bytes` to the end of the file (which must exist), flushed to disk.
+    pub(crate) fn append(&self, kind: FileKind, path: &Path, bytes: &[u8]) -> std::io::Result<()> {
+        let result = (|| {
+            let mut file = std::fs::OpenOptions::new().append(true).open(path)?;
+            file.write_all(bytes)?;
+            file.sync_data()
+        })();
+        self.count(kind, |counts| match &result {
+            Ok(()) => {
+                counts.appends += 1;
+                counts.appended_bytes += bytes.len() as u64;
+            }
+            Err(_) => counts.errors += 1,
+        });
+        result
+    }
+
     /// Moves an unusable file aside (`<path><suffix>`), returning where to.
     pub(crate) fn move_aside(&self, path: &Path, suffix: &str) -> std::io::Result<PathBuf> {
         let aside = with_suffix(path, suffix);
@@ -167,6 +192,12 @@ mod tests {
         assert_eq!(kind, FileKind::State);
         assert_eq!((state.writes, state.written_bytes), (2, 8));
         assert_eq!((state.reads, state.read_bytes, state.errors), (1, 5, 0));
+        files.append(FileKind::State, &path, b"!").unwrap();
+        assert_eq!(
+            files.read(FileKind::State, &path).unwrap().as_deref(),
+            Some(&b"three!"[..])
+        );
+        assert_eq!(files.counts()[0].1.appended_bytes, 1);
         assert_eq!(files.counts()[1].1.reads, 0);
 
         std::fs::remove_dir_all(&dir).unwrap();
diff --git a/src/main.rs b/src/main.rs
index 278a0f4..0a00f1b 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -9,7 +9,7 @@ use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
 use std::time::Duration;
 
-use anyhow::{Context, Result, bail};
+use anyhow::{Context, Result};
 use automerge::AutoCommit;
 use tokio::sync::{RwLock, broadcast, mpsc, watch};
 use tracing::{debug, error, info, trace, warn};
@@ -122,32 +122,27 @@ async fn main() -> Result<()> {
     let (change_tx, _) = watch::channel(0_u64);
     let (change_events, _) = broadcast::channel(256);
 
-    let recovered = match &args.recover_from {
-        Some(path) => Some(bootstrap::load_backup(path, &files)?),
-        None => None,
-    };
-    if let (Some(backup), Some(remembered)) = (&recovered, &remembered) {
-        if backup.swarm_id != remembered.swarm_id {
-            if !force.allows("swarmId") {
-                bail!(
-                    "The backup is of swarm {}, but the state file remembers swarm {}; add --force swarmId to recover the backup's swarm anyway",
-                    backup.swarm_id,
-                    remembered.swarm_id
-                );
-            }
-            warn!(
-                backup = %backup.swarm_id,
-                remembered = %remembered.swarm_id,
-                "The backup is of another swarm than the remembered one; recovering the backup's (--force swarmId)"
-            );
-        }
-    }
-    let (initial_doc, recovered_epoch, recovered_swarm) = match recovered {
-        Some(backup) => (backup.doc, Some(backup.epoch), Some(backup.swarm_id)),
-        None => (AutoCommit::new(), None, None),
+    let starting = bootstrap::starting_document(
+        &bootstrap::StartOptions {
+            recover_from: args.recover_from.as_deref(),
+            document_file: args.document_file.as_deref(),
+            genesis: args.genesis,
+            remembered: remembered.as_ref().map(|r| r.swarm_id.as_str()),
+        },
+        &files,
+        &force,
+    )?;
+    let (initial_doc, starting_epoch, starting_swarm, earlier_actors) = match starting {
+        Some(starting) => (
+            starting.doc,
+            Some(starting.epoch),
+            Some((starting.swarm_id, starting.origin)),
+            starting.earlier_actors,
+        ),
+        None => (AutoCommit::new(), None, None, Vec::new()),
     };
     let initial_actor = initial_doc.get_actor().clone();
-    let initial_epoch = match (recovered_epoch, args.debug_epoch) {
+    let initial_epoch = match (starting_epoch, args.debug_epoch) {
         (Some(epoch), _) => epoch,
         (None, Some(generation)) => Epoch {
             generation,
@@ -197,6 +192,14 @@ async fn main() -> Result<()> {
         blobs: Arc::new(swarm::blob::Blobs::new(args.blob_chunk_kib * 1024)),
         protocol_writers: Arc::new(args.protocol_writers.iter().cloned().collect()),
         state_file,
+        document_file: args.document_file.clone().map(|path| {
+            Arc::new(document::file::DocumentFile::new(
+                path,
+                Arc::clone(&files),
+                Duration::from_secs(args.document_append_secs),
+                Duration::from_secs(args.document_rewrite_secs),
+            ))
+        }),
         files,
         config: Arc::new(config.report()),
         neighbours: Default::default(),
@@ -211,14 +214,17 @@ async fn main() -> Result<()> {
     };
 
     trace!("Application state initialized");
+    for actor in earlier_actors {
+        state.remake.add_own_actor(actor);
+    }
     state.remake.add_own_actor(initial_actor);
 
-    // Join the remembered swarm and no other; HELLOs claim it. A backup to recover from
+    // Join the remembered swarm and no other; HELLOs claim it. A document to start on
     // names its swarm itself (checked against the remembered one above).
-    let start = match recovered_swarm {
-        Some(swarm_id) => {
+    let start = match starting_swarm {
+        Some((swarm_id, origin)) => {
             state.set_expected_swarm_id(swarm_id).await?;
-            bootstrap::Start::Recovered
+            bootstrap::Start::FromDocument(origin)
         }
         None => {
             if let Some(remembered) = remembered {
@@ -283,6 +289,7 @@ async fn main() -> Result<()> {
         "Bootstrap completed"
     );
     swarm::state_file::update(&state, true).await;
+    tokio::spawn(document::file::run(state.clone()));
 
     let auth_state = AuthState::new(&args.htpasswd);
     let app = api::router(state.clone(), auth_state, args.debug_api);
@@ -328,6 +335,7 @@ async fn main() -> Result<()> {
 
     info!("HTTP API server stopped");
     swarm::state_file::update(&stopping, true).await;
+    document::file::flush(&stopping).await;
     Ok(())
 }
 
diff --git a/src/state.rs b/src/state.rs
index 15652ab..ce47673 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -604,6 +604,8 @@ pub(crate) struct AppState {
     pub(crate) protocol_writers: Arc<std::collections::HashSet<String>>,
     /// This node's own files, and how much it read and wrote them.
     pub(crate) files: Arc<crate::files::Files>,
+    /// This node's document on disk (`--document-file`), if configured.
+    pub(crate) document_file: Option<Arc<crate::document::file::DocumentFile>>,
     /// What this node remembers across restarts (`--state-file`), if configured.
     pub(crate) state_file: Option<Arc<crate::swarm::state_file::StateFile>>,
     /// The effective configuration with sources, imported values left out
diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index f41a00b..bd15a34 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -24,9 +24,11 @@ use tokio::time::Instant;
 use tracing::{debug, info, warn};
 use uuid::Uuid;
 
+use crate::document;
 use crate::document::replica::Epoch;
 use crate::document::store;
 use crate::files::{FileKind, Files};
+use crate::force::Force;
 use crate::remake::snapshot::{self, META_ROOT, SWARM};
 use crate::state::AppState;
 use crate::swarm::peer::{remember_peer, send_hello_once};
@@ -45,16 +47,21 @@ pub(crate) enum Start {
     Join,
     /// Start a swarm, empty (`--genesis`).
     Genesis,
-    /// Start a swarm from a backup already loaded as the document (`--recover-from`).
-    Recovered,
+    /// Start on a document already loaded as the replica (`StartingDocument`).
+    FromDocument(Origin),
 }
 
 pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start) -> Result<()> {
-    if start == Start::Recovered {
+    if let Start::FromDocument(origin) = start {
         let Some(swarm_id) = state.swarm_id().await else {
-            bail!("recovering without a swarm ID");
+            bail!("starting on a document without a swarm ID");
         };
-        info!(swarm_id = %swarm_id, epoch = %state.local_epoch(), "Recovered the swarm from a backup");
+        info!(
+            swarm_id = %swarm_id,
+            epoch = %state.local_epoch(),
+            "Starting on {}, ready at once",
+            origin.describe()
+        );
         state.initial_sync_done.store(true, Ordering::Release);
         state.mark_ready(swarm_id).await?;
         // Members of the swarm still running, if any: they merge with this node, or it
@@ -143,38 +150,154 @@ fn read_swarm_id(doc: &AutoCommit) -> Option<String> {
     }
 }
 
-/// A backup to recover from (`--recover-from`).
-pub(crate) struct Backup {
+/// Where a document to start from came from.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum Origin {
+    /// A backup to recover from (`--recover-from`).
+    Backup,
+    /// This node's document file (`--document-file`).
+    DocumentFile,
+}
+
+impl Origin {
+    fn describe(self) -> &'static str {
+        match self {
+            Origin::Backup => "the backup",
+            Origin::DocumentFile => "the document file",
+        }
+    }
+}
+
+/// A document the node starts from, ready at once: on the swarm and epoch it names.
+pub(crate) struct StartingDocument {
     /// Under a fresh actor.
     pub(crate) doc: AutoCommit,
     pub(crate) epoch: Epoch,
     pub(crate) swarm_id: String,
+    pub(crate) origin: Origin,
+    /// The actors this node wrote with before (from the document file).
+    pub(crate) earlier_actors: Vec<ActorId>,
 }
 
-/// Loads a backup to recover from, as `GET /admin/document` saves it.
-pub(crate) fn load_backup(path: &Path, files: &Files) -> Result<Backup> {
-    let bytes = files
-        .read(FileKind::Backup, path)
-        .with_context(|| format!("Reading the backup {}", path.display()))?
-        .with_context(|| format!("The backup {} doesn't exist", path.display()))?;
-    let mut doc = AutoCommit::load(&bytes)
-        .with_context(|| format!("Loading the backup {}", path.display()))?
-        .with_actor(ActorId::random());
-    let Some(swarm_id) = read_swarm_id(&doc) else {
+/// What `--recover-from` and `--document-file` say about where to start.
+pub(crate) struct StartOptions<'a> {
+    pub(crate) recover_from: Option<&'a Path>,
+    pub(crate) document_file: Option<&'a Path>,
+    pub(crate) genesis: bool,
+    /// The swarm the state file remembers.
+    pub(crate) remembered: Option<&'a str>,
+}
+
+/// The document to start from, if any, checked against the other options. Conflicts
+/// are errors unless forced (`force`):
+/// * an unreadable document file (`badDocumentFile`: moved aside, start without it);
+/// * a backup to recover from while the document file holds a document
+///   (`replaceDocumentFile`: the backup wins, and the file is rewritten from it);
+/// * a document of another swarm than the state file remembers (`swarmId`).
+///
+/// A document file holding a document with `--genesis` is an error: move it away.
+pub(crate) fn starting_document(
+    options: &StartOptions<'_>,
+    files: &Files,
+    force: &Force,
+) -> Result<Option<StartingDocument>> {
+    let from_file = match options.document_file {
+        None => None,
+        Some(path) => match document::file::load(path, files) {
+            Ok(loaded) => loaded,
+            Err(error) if force.allows("badDocumentFile") => {
+                warn!(error = %format!("{error:#}"), "Unusable document file; moving it aside and starting without it (--force badDocumentFile)");
+                let aside = files.move_aside(path, ".bad").with_context(|| {
+                    format!("Moving the document file {} aside", path.display())
+                })?;
+                warn!(moved_to = %aside.display(), "Moved the document file aside");
+                None
+            }
+            Err(error) => {
+                bail!(
+                    "{error:#}; add --force badDocumentFile to move it aside and start without it"
+                )
+            }
+        },
+    };
+    let backup = match options.recover_from {
+        None => None,
+        Some(path) => Some(load_backup(path, files)?),
+    };
+
+    let (doc, origin, earlier_actors) = match (backup, from_file) {
+        (Some(_), Some(_)) if !force.allows("replaceDocumentFile") => {
+            bail!(
+                "recover-from would replace the document in the document file; add --force replaceDocumentFile to recover the backup anyway"
+            );
+        }
+        (Some(backup), from_file) => {
+            if from_file.is_some() {
+                warn!(
+                    "Recovering the backup instead of the document file's document (--force replaceDocumentFile)"
+                );
+            }
+            (backup, Origin::Backup, Vec::new())
+        }
+        (None, Some(loaded)) => (loaded.doc, Origin::DocumentFile, loaded.actors),
+        (None, None) => return Ok(None),
+    };
+    if options.genesis && origin == Origin::DocumentFile {
         bail!(
-            "The backup {} names no swarm (no _continuum/swarm); it predates swarm IDs in documents",
-            path.display()
+            "The document file holds a document of a swarm, and genesis would start a new one; move the file away to start a new swarm"
+        );
+    }
+    let starting = identify(doc, origin, earlier_actors)?;
+
+    if let Some(remembered) = options.remembered.filter(|id| *id != starting.swarm_id) {
+        let what = origin.describe();
+        if !force.allows("swarmId") {
+            bail!(
+                "{what} is of swarm {}, but the state file remembers swarm {remembered}; add --force swarmId to start on {what}'s swarm anyway",
+                starting.swarm_id
+            );
+        }
+        warn!(
+            document = %starting.swarm_id,
+            remembered = %remembered,
+            "Starting on {what}'s swarm, not the remembered one (--force swarmId)"
         );
+    }
+    Ok(Some(starting))
+}
+
+/// The swarm and epoch `doc` is on.
+fn identify(
+    mut doc: AutoCommit,
+    origin: Origin,
+    earlier_actors: Vec<ActorId>,
+) -> Result<StartingDocument> {
+    let what = origin.describe();
+    let Some(swarm_id) = read_swarm_id(&doc) else {
+        bail!("{what} names no swarm (no _continuum/swarm); it predates swarm IDs in documents");
     };
-    let epoch = snapshot::epoch_of(&mut doc)
-        .map_err(|reason| anyhow::anyhow!("The backup {}: {reason}", path.display()))?;
-    Ok(Backup {
+    let epoch =
+        snapshot::epoch_of(&mut doc).map_err(|reason| anyhow::anyhow!("{what}: {reason}"))?;
+    Ok(StartingDocument {
         doc,
         epoch,
         swarm_id,
+        origin,
+        earlier_actors,
     })
 }
 
+/// A backup to recover from, as `GET /admin/document` saves it, under a fresh actor.
+fn load_backup(path: &Path, files: &Files) -> Result<AutoCommit> {
+    let bytes = files
+        .read(FileKind::Backup, path)
+        .with_context(|| format!("Reading the backup {}", path.display()))?
+        .with_context(|| format!("The backup {} doesn't exist", path.display()))?;
+    Ok(AutoCommit::load(&bytes)
+        .with_context(|| format!("Loading the backup {}", path.display()))?
+        .with_actor(ActorId::random()))
+}
+
 /// Resolves the seeds in `unresolved`, says HELLO to each that resolves, and keeps the
 /// rest for later. Warns once per seed that fails (`failing` remembers which), and says
 /// when such a seed resolves after all.