document file: a clean stop marked, and `--keep-actor` writes on as the same actor

A node writes as a fresh Automerge actor at each start, so no change number of an actor
comes twice: one sent before a crash but not yet appended to the document file would
otherwise be made again, differently. An epoch thus gathers a writer per start, and
every transfer of its history carries a signature per writer.

The document file now says when it holds everything the node wrote:

* at a clean stop, after the last append, an empty `Closed` record, only where the
  file is up to the document's heads (ready, not rejoining);
* as a node starts on the file, before anything is written, an empty `Opened` record,
  so a crash from then on leaves a file without the mark;
* the header names the actor the node writes with (`current`).

With `--keep-actor` (off by default; needs `--document-file`), a node starting on a
file that ends with `Closed` writes on as that actor. A crash, a drop, a switch or a
rejoin still starts a fresh one, so an epoch's writers come to one per node plus one
per such event. lifecycle.md says so.

5 files changed, 156 insertions, 21 deletions

diff --git a/src/config/mod.rs b/src/config/mod.rs
index 7df68e7..c0abc09 100644
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@@ -264,6 +264,13 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[
         "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, and the file is rewritten after a switch, a rejoin or a restart, and when the appended changes outgrew it.",
     ),
     option(
+        "keep-actor",
+        Kind::Flag,
+        "BOOL",
+        Some("false"),
+        "Resume writing as the actor this node wrote with before it stopped, if it stopped cleanly (its document file holds everything it wrote then), instead of a fresh one: an epoch then gathers a writer per node and per crash or drop rather than per start, which every transfer of its history carries a signature for. Needs document-file.",
+    ),
+    option(
         "document-append",
         Kind::Duration,
         "DURATION",
@@ -654,6 +661,8 @@ pub(crate) struct Settings {
     pub(crate) genesis: bool,
     pub(crate) state_file: Option<PathBuf>,
     pub(crate) document_file: Option<PathBuf>,
+    #[serde(default)]
+    pub(crate) keep_actor: bool,
     #[serde(
         default = "default_document_append",
         with = "crate::units::duration_option"
@@ -982,6 +991,9 @@ impl Settings {
         if self.document_append.is_zero() || self.document_rewrite.is_zero() {
             bail!("document-append and document-rewrite must be longer than 0");
         }
+        if self.keep_actor && self.document_file.is_none() {
+            bail!("keep-actor needs document-file: it tells whether this node stopped cleanly");
+        }
         if self.keepalive < Duration::from_millis(100) {
             bail!("keepalive must be at least 100ms");
         }
diff --git a/src/document/file.rs b/src/document/file.rs
index 2d588ee..56114f5 100644
--- a/src/document/file.rs
+++ b/src/document/file.rs
@@ -2,12 +2,14 @@
 //! resumes with it instead of starting empty.
 //!
 //! Format: the line `continuum-document/2`, 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, the first of them its kind
-//! (`Record`): first a full save of the document, then increments (the changes after
-//! the heads written before), and the signatures of its changes
-//! (`document::signatures`), those the file holds none of yet, as
-//! `signatures::pack` packs them. A record cut short by a crash is dropped when
+//! node wrote with, so a later rejoin replays those writes as its own, and the one it
+//! writes with, `current`), then records, each a little-endian `u32` length and that
+//! many bytes, the first of them its kind (`Record`): first a full save of the document,
+//! then increments (the changes after the heads written before), and the signatures of
+//! its changes (`document::signatures`), those the file holds none of yet, as
+//! `signatures::pack` packs them; last, at a clean stop, an empty `Closed` record: the
+//! file holds everything the node wrote, so it may resume as `current`
+//! (`--keep-actor`). A record cut short by a crash is dropped when
 //! reading. A file of `continuum-document/1` has records of the document alone,
 //! without their kind: a full save, then increments.
 //!
@@ -44,6 +46,11 @@ enum Record {
     Save = 0,
     Increment = 1,
     Signatures = 2,
+    /// Empty, last: the node stopped with everything it wrote in the file.
+    Closed = 3,
+    /// Empty: the node started on the file, so a `Closed` before it says nothing of
+    /// what it writes now.
+    Opened = 4,
 }
 
 #[derive(Debug, Default, Serialize, Deserialize)]
@@ -51,6 +58,9 @@ struct Header {
     /// Hex actor IDs.
     #[serde(default)]
     actors: Vec<String>,
+    /// Hex: the actor the node wrote with when the file was rewritten.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    current: Option<String>,
 }
 
 /// A document file as read at start.
@@ -61,6 +71,9 @@ pub(crate) struct Loaded {
     pub(crate) actors: Vec<ActorId>,
     /// The signatures of its changes, in the order they were written.
     pub(crate) signatures: Vec<(ChangeHash, [u8; 64])>,
+    /// The actor it wrote with, where the file ends with `Closed`: nothing it wrote is
+    /// missing, so it may write on as that actor.
+    pub(crate) resumable: Option<ActorId>,
 }
 
 /// Reads the document file; None if there is none. An error if it can't be read or
@@ -109,9 +122,18 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> {
         })
         .collect::<Result<Vec<_>>>()?;
 
+    let current = header
+        .current
+        .as_deref()
+        .map(|hex| {
+            ActorId::try_from(hex).map_err(|_| anyhow!("invalid actor {hex:?} in the header"))
+        })
+        .transpose()?;
+
     let mut records = &rest[line_end + 1..];
     let mut doc: Option<Document> = None;
     let mut signatures = Vec::new();
+    let mut closed = false;
     while !records.is_empty() {
         let Some((length, after)) = records.split_first_chunk::<4>() else {
             break;
@@ -129,6 +151,8 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> {
                 0 => Record::Save,
                 1 => Record::Increment,
                 2 => Record::Signatures,
+                3 => Record::Closed,
+                4 => Record::Opened,
                 other => bail!("a record of an unknown kind ({other})"),
             };
             (kind, record)
@@ -137,6 +161,7 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> {
         } else {
             (Record::Increment, record)
         };
+        closed = kind == Record::Closed;
         match (kind, &mut doc) {
             (Record::Save, None) => doc = Some(super::load(record).context("the full save")?),
             (Record::Save, Some(_)) => bail!("a second full save"),
@@ -147,6 +172,7 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> {
                 super::signatures::unpack(record)
                     .ok_or_else(|| anyhow!("signatures of {} bytes", record.len()))?,
             ),
+            (Record::Closed | Record::Opened, Some(_)) => {}
             (_, None) => bail!("a record before the full save"),
         }
         records = after;
@@ -158,6 +184,7 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> {
         doc,
         actors,
         signatures,
+        resumable: current.filter(|_| closed && records.is_empty()),
     };
     Ok((loaded, records.len()))
 }
@@ -181,11 +208,13 @@ fn signatures_record(signatures: &[(ChangeHash, [u8; 64])]) -> Vec<u8> {
 /// A whole file: header, one full save, and the signatures.
 fn whole(
     actors: &[ActorId],
+    current: &ActorId,
     save: &[u8],
     signatures: &[(ChangeHash, [u8; 64])],
 ) -> Result<Vec<u8>> {
     let header = Header {
         actors: actors.iter().map(ActorId::to_hex_string).collect(),
+        current: Some(current.to_hex_string()),
     };
     let mut out = MAGIC.to_vec();
     out.extend(serde_json::to_vec(&header)?);
@@ -336,7 +365,7 @@ impl DocumentFile {
                 .filter(|actor| *actor == current || present.contains(actor))
                 .collect();
             let save = super::save(&copy);
-            let bytes = whole(&actors, &save, &signatures)?;
+            let bytes = whole(&actors, &current, &save, &signatures)?;
             writer
                 .files
                 .write_atomic(FileKind::Document, &writer.path, &bytes)?;
@@ -368,6 +397,44 @@ impl DocumentFile {
         }
     }
 
+    /// Marks the file as opened (`Record::Opened`), at start, before the node writes:
+    /// a `Closed` before it no longer says the file holds everything (a crash from here
+    /// on leaves it without one). Nothing where there is no file yet.
+    pub(crate) fn open(&self) {
+        if !self.path.exists() {
+            return;
+        }
+        if let Err(error) =
+            self.files
+                .append(FileKind::Document, &self.path, &record(Record::Opened, &[]))
+        {
+            error!(path = %self.path.display(), error = %error, "Couldn't mark the document file as opened");
+        }
+    }
+
+    /// Marks the file as holding everything the node wrote (`Record::Closed`), at a
+    /// clean stop: only where it was written up to `heads`, the document's now (after a
+    /// last append). Any record written after makes it not so.
+    pub(crate) fn close(&self, heads: &[ChangeHash]) {
+        let up_to_date = self
+            .lock()
+            .as_ref()
+            .is_some_and(|written| written.heads == heads);
+        if !up_to_date {
+            debug!("The document file isn't up to date at the stop: not marked as closed");
+            return;
+        }
+        match self
+            .files
+            .append(FileKind::Document, &self.path, &record(Record::Closed, &[]))
+        {
+            Ok(()) => debug!("Marked the document file as closed"),
+            Err(error) => {
+                error!(path = %self.path.display(), error = %error, "Couldn't mark the document file as closed");
+            }
+        }
+    }
+
     /// Appends `increment`: what changed from `since` (the heads written last) to
     /// `heads`; and `signatures`, those of the document the file holds none of yet
     /// (`Due::Append`).
@@ -430,7 +497,7 @@ mod tests {
         let actor = doc.get_actor().clone();
         let heads = doc.get_heads();
         let first = (heads[0], [1; 64]);
-        let mut bytes = whole(std::slice::from_ref(&actor), &doc.save(), &[first]).unwrap();
+        let mut bytes = whole(std::slice::from_ref(&actor), &actor, &doc.save(), &[first]).unwrap();
 
         change(&mut doc, |doc| doc.put(ROOT, "b", 2_i64).unwrap());
         bytes.extend(record(Record::Increment, &doc.save_after(&heads)));
@@ -443,6 +510,17 @@ mod tests {
         assert_ne!(loaded.doc.get_actor(), &actor);
         assert!(loaded.doc.get(ROOT, "b").unwrap().is_some());
         assert_eq!(loaded.signatures, [first, second]);
+        // Not closed: no actor to resume as.
+        assert!(loaded.resumable.is_none());
+        // Closed last: the actor it wrote with; a record after, not anymore.
+        let mut closed = bytes.clone();
+        closed.extend(record(Record::Closed, &[]));
+        assert_eq!(parse(&closed).unwrap().0.resumable, Some(actor.clone()));
+        let mut opened = closed.clone();
+        opened.extend(record(Record::Opened, &[]));
+        assert!(parse(&opened).unwrap().0.resumable.is_none());
+        closed.extend(signatures_record(&[(doc.get_heads()[0], [3; 64])]));
+        assert!(parse(&closed).unwrap().0.resumable.is_none());
 
         // A record cut short by a crash is dropped.
         let mut cut = bytes.clone();
diff --git a/src/main.rs b/src/main.rs
index 880d793..7c819e5 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -165,18 +165,37 @@ async fn main() -> Result<()> {
         &files,
         &force,
     )?;
-    let (initial_doc, starting_epoch, starting_swarm, earlier_actors, initial_signatures) =
-        match starting {
-            Some(starting) => (
-                starting.doc,
-                Some(starting.epoch),
-                Some((starting.swarm_id, starting.origin)),
-                starting.earlier_actors,
-                starting.signatures,
-            ),
-            None => (Document::new(), None, None, Vec::new(), Vec::new()),
-        };
+    let (
+        initial_doc,
+        starting_epoch,
+        starting_swarm,
+        earlier_actors,
+        initial_signatures,
+        resumable,
+    ) = match starting {
+        Some(starting) => (
+            starting.doc,
+            Some(starting.epoch),
+            Some((starting.swarm_id, starting.origin)),
+            starting.earlier_actors,
+            starting.signatures,
+            starting.resumable,
+        ),
+        None => (Document::new(), None, None, Vec::new(), Vec::new(), None),
+    };
     let mut initial_doc = initial_doc;
+    // Writing on as the actor of before, where this node stopped cleanly: everything it
+    // wrote as that one is in the document, so no change of it comes twice.
+    match resumable.filter(|_| args.keep_actor) {
+        Some(actor) => {
+            info!(actor = %actor, "Writing on as the actor of before (--keep-actor)");
+            initial_doc.set_actor(actor);
+        }
+        None if args.keep_actor && starting_epoch.is_some() => {
+            info!("Writing as a fresh actor: this node didn't stop cleanly (--keep-actor)");
+        }
+        None => {}
+    }
     if let Some(spec) = &args.remake {
         remake::remake_genesis(spec, &files, &mut initial_doc)
             .with_context(|| format!("Making the genesis document (--remake {spec})"))?;
@@ -228,6 +247,11 @@ async fn main() -> Result<()> {
             args.document_rewrite,
         ))
     });
+    // A clean stop's mark undone before anything is written: a crash from here on
+    // leaves a file that doesn't say so.
+    if let Some(file) = &document_file {
+        file.open();
+    }
     let node = node::Node::new(node::NodeConfig {
         transport,
         identity: Some(transport::identity::IdentityConfig {
@@ -367,6 +391,7 @@ async fn main() -> Result<()> {
     swarm::peer::farewell(&state).await;
     swarm::state_file::update(&state, true).await;
     swarm::document_file::flush(&state).await;
+    swarm::document_file::close(&state).await;
     Ok(())
 }
 
diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index d2a593b..5c3b5ef 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -266,6 +266,9 @@ pub(crate) struct StartingDocument {
     pub(crate) earlier_actors: Vec<ActorId>,
     /// The signatures of its changes (from the document file).
     pub(crate) signatures: Vec<(crate::document::ChangeHash, [u8; 64])>,
+    /// The actor this node wrote with, where it stopped cleanly (the document file's
+    /// `resumable`): it may write on as that one (`--keep-actor`).
+    pub(crate) resumable: Option<ActorId>,
 }
 
 /// What `--recover-from` and `--document-file` say about where to start.
@@ -314,7 +317,7 @@ pub(crate) fn starting_document(
         Some(path) => Some(load_backup(path, files)?),
     };
 
-    let (doc, origin, earlier_actors, signatures) = match (backup, from_file) {
+    let (doc, origin, earlier_actors, signatures, resumable) = 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"
@@ -326,13 +329,14 @@ pub(crate) fn starting_document(
                     "Recovering the backup instead of the document file's document (--force replaceDocumentFile)"
                 );
             }
-            (backup, Origin::Backup, Vec::new(), Vec::new())
+            (backup, Origin::Backup, Vec::new(), Vec::new(), None)
         }
         (None, Some(loaded)) => (
             loaded.doc,
             Origin::DocumentFile,
             loaded.actors,
             loaded.signatures,
+            loaded.resumable,
         ),
         (None, None) => return Ok(None),
     };
@@ -343,6 +347,7 @@ pub(crate) fn starting_document(
     }
     let mut starting = identify(doc, origin, earlier_actors)?;
     starting.signatures = signatures;
+    starting.resumable = resumable;
 
     if let Some(remembered) = options.remembered.filter(|id| *id != starting.swarm_id) {
         let what = origin.describe();
@@ -379,6 +384,7 @@ fn identify(
         origin,
         earlier_actors,
         signatures: Vec::new(),
+        resumable: None,
     })
 }
 
diff --git a/src/swarm/document_file.rs b/src/swarm/document_file.rs
index e066e27..4c7843d 100644
--- a/src/swarm/document_file.rs
+++ b/src/swarm/document_file.rs
@@ -25,6 +25,20 @@ pub(crate) async fn run(state: AppState) {
     }
 }
 
+/// At a clean stop, after the last `flush`: the file marked as holding everything this
+/// node wrote (`DocumentFile::close`), where it does (ready, not rejoining, up to the
+/// document's heads), so a restart may write on as the same actor (`--keep-actor`).
+pub(crate) async fn close(state: &AppState) {
+    let Some(file) = state.document_file.clone() else {
+        return;
+    };
+    if state.remake.is_rejoining() || !state.is_ready().await {
+        return;
+    }
+    let heads = state.doc.read().await.doc.get_heads();
+    file.close(&heads);
+}
+
 /// 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 {