signatures: kept in the document file

A node's signatures are saved with its document (--document-file) and
loaded with it.

Auth plan M5.

1 files changed, 167 insertions, 34 deletions

diff --git a/src/document/file.rs b/src/document/file.rs
index 3ea3afd..33f7934 100644
--- a/src/document/file.rs
+++ b/src/document/file.rs
@@ -1,11 +1,15 @@
 //! 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
+//! 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: 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.
+//! 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, each a change hash
+//! (32 bytes) and its signature (64). 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.
 //!
 //! 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):
@@ -30,7 +34,20 @@ use crate::document::replica::Epoch;
 use crate::files::{FileKind, Files};
 use crate::locks::MutexExt;
 
-const MAGIC: &[u8] = b"continuum-document/1\n";
+const MAGIC: &[u8] = b"continuum-document/2\n";
+/// The format before signatures, still read.
+const MAGIC_1: &[u8] = b"continuum-document/1\n";
+
+/// What a record holds: its first byte.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Record {
+    Save = 0,
+    Increment = 1,
+    Signatures = 2,
+}
+
+/// A signature as a record holds it: the change's hash, then the signature.
+const SIGNATURE_BYTES: usize = 32 + 64;
 
 #[derive(Debug, Default, Serialize, Deserialize)]
 struct Header {
@@ -45,6 +62,8 @@ pub(crate) struct Loaded {
     pub(crate) doc: Document,
     /// The actors this node wrote with before.
     pub(crate) actors: Vec<ActorId>,
+    /// The signatures of its changes, in the order they were written.
+    pub(crate) signatures: Vec<(ChangeHash, [u8; 64])>,
 }
 
 /// Reads the document file; None if there is none. An error if it can't be read or
@@ -70,9 +89,15 @@ pub(crate) fn load(path: &Path, files: &Files) -> Result<Option<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 (rest, kinds) = match bytes.strip_prefix(MAGIC) {
+        Some(rest) => (rest, true),
+        None => (
+            bytes
+                .strip_prefix(MAGIC_1)
+                .ok_or_else(|| anyhow!("not a continuum document file"))?,
+            false,
+        ),
+    };
     let line_end = rest
         .iter()
         .position(|byte| *byte == b'\n')
@@ -89,6 +114,7 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> {
 
     let mut records = &rest[line_end + 1..];
     let mut doc: Option<Document> = None;
+    let mut signatures = Vec::new();
     while !records.is_empty() {
         let Some((length, after)) = records.split_first_chunk::<4>() else {
             break;
@@ -98,37 +124,88 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> {
             break;
         }
         let (record, after) = after.split_at(length);
-        match &mut doc {
-            None => doc = Some(super::load(record).context("the full save")?),
-            Some(doc) => {
+        let (kind, record) = if kinds {
+            let (kind, record) = record
+                .split_first()
+                .ok_or_else(|| anyhow!("an empty record"))?;
+            let kind = match kind {
+                0 => Record::Save,
+                1 => Record::Increment,
+                2 => Record::Signatures,
+                other => bail!("a record of an unknown kind ({other})"),
+            };
+            (kind, record)
+        } else if doc.is_none() {
+            (Record::Save, record)
+        } else {
+            (Record::Increment, record)
+        };
+        match (kind, &mut doc) {
+            (Record::Save, None) => doc = Some(super::load(record).context("the full save")?),
+            (Record::Save, Some(_)) => bail!("a second full save"),
+            (Record::Increment, Some(doc)) => {
                 doc.load_incremental(record).context("an increment")?;
             }
+            (Record::Signatures, Some(_)) => {
+                if record.len() % SIGNATURE_BYTES != 0 {
+                    bail!("signatures of {} bytes", record.len());
+                }
+                for entry in record.chunks_exact(SIGNATURE_BYTES) {
+                    let (hash, signature) = entry.split_at(32);
+                    let hash = ChangeHash(hash.try_into().expect("32 bytes"));
+                    signatures.push((hash, signature.try_into().expect("64 bytes")));
+                }
+            }
+            (_, None) => bail!("a record before the full save"),
         }
         records = after;
     }
     let Some(doc) = doc else {
         bail!("no full save");
     };
-    let loaded = Loaded { doc, actors };
+    let loaded = Loaded {
+        doc,
+        actors,
+        signatures,
+    };
     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());
+fn record(kind: Record, bytes: &[u8]) -> Vec<u8> {
+    let mut out = Vec::with_capacity(bytes.len() + 5);
+    out.extend_from_slice(&(bytes.len() as u32 + 1).to_le_bytes());
+    out.push(kind as u8);
     out.extend_from_slice(bytes);
     out
 }
 
-/// A whole file: header and one full save.
-fn whole(actors: &[ActorId], save: &[u8]) -> Result<Vec<u8>> {
+/// A record of `signatures`; none without any.
+fn signatures_record(signatures: &[(ChangeHash, [u8; 64])]) -> Vec<u8> {
+    if signatures.is_empty() {
+        return Vec::new();
+    }
+    let mut bytes = Vec::with_capacity(signatures.len() * SIGNATURE_BYTES);
+    for (hash, signature) in signatures {
+        bytes.extend_from_slice(&hash.0);
+        bytes.extend_from_slice(signature);
+    }
+    record(Record::Signatures, &bytes)
+}
+
+/// A whole file: header, one full save, and the signatures.
+fn whole(
+    actors: &[ActorId],
+    save: &[u8],
+    signatures: &[(ChangeHash, [u8; 64])],
+) -> 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));
+    out.extend(record(Record::Save, save));
+    out.extend(signatures_record(signatures));
     Ok(out)
 }
 
@@ -148,6 +225,8 @@ struct Written {
     /// This node's actors when the file was last rewritten (`RemakeState::own_actors`).
     actors: usize,
     heads: Vec<ChangeHash>,
+    /// How many of the document's signatures (`Signatures::count`).
+    signatures: usize,
     /// The full save.
     base_bytes: u64,
     /// Appended since.
@@ -180,11 +259,13 @@ impl DocumentFile {
 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).
+    /// What changed since `since` appended, and the signatures after the first
+    /// `signatures`, 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<ChangeHash>,
+        signatures: usize,
     },
     Nothing,
 }
@@ -202,7 +283,15 @@ impl DocumentFile {
         self.lock().as_ref().map(|written| written.actors)
     }
 
-    pub(crate) fn due(&self, epoch: Epoch, actors: usize, heads: &[ChangeHash]) -> Due {
+    /// With `signatures`, how many the document has (`Signatures::count`); None for a
+    /// document whose signatures aren't kept (the one before a rejoin).
+    pub(crate) fn due(
+        &self,
+        epoch: Epoch,
+        actors: usize,
+        heads: &[ChangeHash],
+        signatures: Option<usize>,
+    ) -> Due {
         let written = self.lock().clone();
         let reason = match &written {
             None => Some("first write"),
@@ -218,23 +307,31 @@ impl DocumentFile {
         };
         match (reason, written) {
             (Some(reason), _) => Due::Rewrite(reason),
-            (None, Some(written)) if written.heads != heads => Due::Append {
-                epoch: written.epoch,
-                since: written.heads,
-            },
+            (None, Some(written))
+                if written.heads != heads
+                    || signatures.is_some_and(|count| count != written.signatures) =>
+            {
+                Due::Append {
+                    epoch: written.epoch,
+                    since: written.heads,
+                    signatures: written.signatures,
+                }
+            }
             (None, _) => Due::Nothing,
         }
     }
 
-    /// Writes the whole file, atomically: `copy` (a copy of the document, of `epoch`)
-    /// and this node's actors `own` that wrote in it.
+    /// Writes the whole file, atomically: `copy` (a copy of the document, of `epoch`),
+    /// its `signatures`, and this node's actors `own` that wrote in it.
     pub(crate) async fn rewrite(
         self: &Arc<Self>,
         epoch: Epoch,
         copy: Document,
+        signatures: Vec<(ChangeHash, [u8; 64])>,
         own: Vec<ActorId>,
         reason: &str,
     ) {
+        let signature_count = signatures.len();
         let start = Instant::now();
         let own_count = own.len();
         let writer = Arc::clone(self);
@@ -253,7 +350,7 @@ impl DocumentFile {
                 .filter(|actor| *actor == current || present.contains(actor))
                 .collect();
             let save = super::save(&copy);
-            let bytes = whole(&actors, &save)?;
+            let bytes = whole(&actors, &save, &signatures)?;
             writer
                 .files
                 .write_atomic(FileKind::Document, &writer.path, &bytes)?;
@@ -272,6 +369,7 @@ impl DocumentFile {
                     epoch,
                     actors: own_count,
                     heads,
+                    signatures: signature_count,
                     base_bytes,
                     tail_bytes: 0,
                     rewritten_at: Instant::now(),
@@ -285,14 +383,22 @@ impl DocumentFile {
     }
 
     /// Appends `increment`: what changed from `since` (the heads written last) to
-    /// `heads`.
+    /// `heads`; and `signatures`, those of the document the file holds none of yet
+    /// (`Due::Append`).
     pub(crate) async fn append(
         self: &Arc<Self>,
         since: Vec<ChangeHash>,
         heads: Vec<ChangeHash>,
         increment: Vec<u8>,
+        signatures: Vec<(ChangeHash, [u8; 64])>,
     ) {
-        let bytes = record(&increment);
+        let mut bytes = if heads == since {
+            Vec::new()
+        } else {
+            record(Record::Increment, &increment)
+        };
+        bytes.extend(signatures_record(&signatures));
+        let length = bytes.len() as u64;
         let writer = Arc::clone(self);
         let appended = tokio::task::spawn_blocking(move || {
             writer
@@ -305,7 +411,8 @@ impl DocumentFile {
                 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;
+                    current.signatures += signatures.len();
+                    current.tail_bytes += length;
                     debug!(
                         bytes = increment.len(),
                         tail_bytes = current.tail_bytes,
@@ -336,20 +443,24 @@ mod tests {
         change(&mut doc, |doc| doc.put(ROOT, "a", 1_i64).unwrap());
         let actor = doc.get_actor().clone();
         let heads = doc.get_heads();
-        let mut bytes = whole(std::slice::from_ref(&actor), &doc.save()).unwrap();
+        let first = (heads[0], [1; 64]);
+        let mut bytes = whole(std::slice::from_ref(&actor), &doc.save(), &[first]).unwrap();
 
         change(&mut doc, |doc| doc.put(ROOT, "b", 2_i64).unwrap());
-        bytes.extend(record(&doc.save_after(&heads)));
+        bytes.extend(record(Record::Increment, &doc.save_after(&heads)));
+        let second = (doc.get_heads()[0], [2; 64]);
+        bytes.extend(signatures_record(&[second]));
 
         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());
+        assert_eq!(loaded.signatures, [first, second]);
 
         // A record cut short by a crash is dropped.
         let mut cut = bytes.clone();
-        cut.extend(record(b"whatever")[..6].iter());
+        cut.extend(record(Record::Increment, b"whatever")[..6].iter());
         let (loaded, dropped) = parse(&cut).unwrap();
         assert_eq!(dropped, 6);
         assert!(loaded.doc.get(ROOT, "b").unwrap().is_some());
@@ -357,4 +468,26 @@ mod tests {
         assert!(parse(b"something else").is_err());
         assert!(parse(&bytes[..MAGIC.len() + 3]).is_err());
     }
+
+    /// A file of the format before signatures: records of the document alone.
+    #[test]
+    fn the_format_before_signatures() {
+        let mut doc = Document::new();
+        change(&mut doc, |doc| doc.put(ROOT, "a", 1_i64).unwrap());
+        let heads = doc.get_heads();
+        let untagged = |bytes: &[u8]| {
+            let mut out = (bytes.len() as u32).to_le_bytes().to_vec();
+            out.extend_from_slice(bytes);
+            out
+        };
+        let mut bytes = MAGIC_1.to_vec();
+        bytes.extend(b"{\"actors\":[]}\n");
+        bytes.extend(untagged(&doc.save()));
+        change(&mut doc, |doc| doc.put(ROOT, "b", 2_i64).unwrap());
+        bytes.extend(untagged(&doc.save_after(&heads)));
+        let (loaded, dropped) = parse(&bytes).unwrap();
+        assert_eq!(dropped, 0);
+        assert!(loaded.doc.get(ROOT, "b").unwrap().is_some());
+        assert!(loaded.signatures.is_empty());
+    }
 }