signatures: with sync messages

Sync messages carry, for each actor's writes, the latest signature the
sender checked; the receiver checks it once it holds the change.

Auth plan M5.

2 files changed, 249 insertions, 38 deletions

diff --git a/src/document/file.rs b/src/document/file.rs
index 33f7934..2d588ee 100644
--- a/src/document/file.rs
+++ b/src/document/file.rs
@@ -6,8 +6,8 @@
 //! 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
+//! (`document::signatures`), those the file holds none of yet, as
+//! `signatures::pack` packs them. 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.
 //!
@@ -46,9 +46,6 @@ enum Record {
     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 {
     /// Hex actor IDs.
@@ -146,16 +143,10 @@ fn parse(bytes: &[u8]) -> Result<(Loaded, usize)> {
             (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")));
-                }
-            }
+            (Record::Signatures, Some(_)) => signatures.extend(
+                super::signatures::unpack(record)
+                    .ok_or_else(|| anyhow!("signatures of {} bytes", record.len()))?,
+            ),
             (_, None) => bail!("a record before the full save"),
         }
         records = after;
@@ -184,12 +175,7 @@ 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)
+    record(Record::Signatures, &super::signatures::pack(signatures))
 }
 
 /// A whole file: header, one full save, and the signatures.
diff --git a/src/document/signatures.rs b/src/document/signatures.rs
index c8af89a..a6a30f7 100644
--- a/src/document/signatures.rs
+++ b/src/document/signatures.rs
@@ -11,6 +11,12 @@
 //! change left unsigned is vouched for by the node's next signed one, which, as
 //! `--sign-every` says, is its latest change, signed a while after it if nothing signed
 //! it before (`SignEvery`), or every change at once.
+//!
+//! Signatures travel with sync messages (`SyncEnvelope::signatures`): each actor's
+//! latest one a node verified, those a peer wasn't sent yet (`newer_than`). A received
+//! signature is checked once its change is there (`offer`), and kept if it vouches for
+//! more of its actor's changes than any before; one of a change not there yet waits
+//! (`pending`, a bounded number).
 
 use std::collections::HashMap;
 
@@ -150,6 +156,43 @@ impl<'de> serde::Deserialize<'de> for SignEvery {
     }
 }
 
+/// Signatures a sync message carries at most (`newer_than`).
+pub(crate) const MAX_PER_MESSAGE: usize = 256;
+/// A signature as messages and the document file carry it (`pack`): the change's
+/// hash (32 bytes), then the signature (64).
+pub(crate) const PACKED: usize = 32 + 64;
+
+/// `signatures` as bytes, one after another (`PACKED`).
+pub(crate) fn pack(signatures: &[(ChangeHash, [u8; 64])]) -> Vec<u8> {
+    let mut bytes = Vec::with_capacity(signatures.len() * PACKED);
+    for (hash, signature) in signatures {
+        bytes.extend_from_slice(&hash.0);
+        bytes.extend_from_slice(signature);
+    }
+    bytes
+}
+
+/// What `pack` made; None if `bytes` aren't a whole number of signatures.
+pub(crate) fn unpack(bytes: &[u8]) -> Option<Vec<(ChangeHash, [u8; 64])>> {
+    if !bytes.len().is_multiple_of(PACKED) {
+        return None;
+    }
+    Some(
+        bytes
+            .chunks_exact(PACKED)
+            .map(|entry| {
+                let (hash, signature) = entry.split_at(32);
+                (
+                    ChangeHash(hash.try_into().expect("32 bytes")),
+                    signature.try_into().expect("64 bytes"),
+                )
+            })
+            .collect(),
+    )
+}
+/// Received signatures of changes not there yet, kept at most (`offer`).
+const MAX_PENDING: usize = 4096;
+
 /// A document's signatures, by change.
 #[derive(Debug, Default)]
 pub(crate) struct Signatures {
@@ -159,15 +202,48 @@ pub(crate) struct Signatures {
     order: Vec<ChangeHash>,
     /// This node's latest change, if no signature vouches for it yet (`SignEvery`).
     unsigned: Option<ChangeHash>,
+    /// By actor: its latest change a signature verified, so its changes up to it.
+    latest: HashMap<super::ActorId, Latest>,
+    /// Counts each change of `latest`: a peer is sent those after the last it was
+    /// (`newer_than`).
+    stamp: u64,
+    /// Received signatures of changes the document doesn't hold yet.
+    pending: HashMap<ChangeHash, [u8; 64]>,
+}
+
+/// An actor's latest change a signature verified.
+#[derive(Debug, Clone, Copy)]
+struct Latest {
+    seq: u64,
+    hash: ChangeHash,
+    /// `Signatures::stamp` when it became the latest.
+    stamp: u64,
 }
 
 impl Signatures {
-    /// Signatures kept before, as `since` gave them (the document file).
-    pub(crate) fn from_kept(kept: Vec<(ChangeHash, [u8; 64])>) -> Self {
+    /// Signatures kept before, as `since` gave them (the document file), of `doc`'s
+    /// changes: each actor's latest checked (`check`), as it vouches for the others.
+    pub(crate) fn from_kept(doc: &Document, kept: Vec<(ChangeHash, [u8; 64])>) -> Self {
         let mut signatures = Self::default();
+        let mut by_actor: HashMap<super::ActorId, (u64, ChangeHash)> = HashMap::new();
         for (hash, signature) in kept {
+            if let Some(meta) = doc.get_change_meta_by_hash(&hash) {
+                let entry = by_actor
+                    .entry(meta.actor.into_owned())
+                    .or_insert((meta.seq, hash));
+                if meta.seq > entry.0 {
+                    *entry = (meta.seq, hash);
+                }
+            }
             signatures.insert(hash, signature);
         }
+        for (hash, signature) in by_actor
+            .into_values()
+            .map(|(_, hash)| (hash, signatures.by_change[&hash]))
+            .collect::<Vec<_>>()
+        {
+            signatures.vouch(doc, hash, &signature);
+        }
         signatures
     }
 
@@ -177,6 +253,101 @@ impl Signatures {
         }
     }
 
+    /// `signature` of `hash`, if valid (`check`), as its actor's latest verified change
+    /// where it is later than the one before; whether it is.
+    fn vouch(&mut self, doc: &Document, hash: ChangeHash, signature: &[u8; 64]) -> bool {
+        // Not checked where it couldn't vouch for more: a signature costs about 30 µs.
+        let Some(meta) = doc.get_change_meta_by_hash(&hash) else {
+            return false;
+        };
+        if self
+            .latest
+            .get(meta.actor.as_ref())
+            .is_some_and(|latest| latest.seq >= meta.seq)
+        {
+            return false;
+        }
+        let Some((actor, seq)) = check(doc, &hash, signature) else {
+            return false;
+        };
+        if self
+            .latest
+            .get(&actor)
+            .is_some_and(|latest| latest.seq >= seq)
+        {
+            return false;
+        }
+        self.stamp += 1;
+        self.latest.insert(
+            actor,
+            Latest {
+                seq,
+                hash,
+                stamp: self.stamp,
+            },
+        );
+        true
+    }
+
+    /// Received signatures, from a peer: each checked once its change is in `doc`, kept
+    /// if it vouches for more of its actor's changes than any before; those of changes
+    /// not there yet wait, a bounded number. Then those waiting whose changes came
+    /// meanwhile. How many were kept.
+    pub(crate) fn offer(&mut self, doc: &Document, received: Vec<(ChangeHash, [u8; 64])>) -> usize {
+        let mut kept = 0;
+        for (hash, signature) in received {
+            if doc.get_change_meta_by_hash(&hash).is_none() {
+                if self.pending.len() < MAX_PENDING {
+                    self.pending.insert(hash, signature);
+                }
+            } else if self.vouch(doc, hash, &signature) {
+                self.insert(hash, signature);
+                kept += 1;
+            }
+        }
+        let arrived: Vec<ChangeHash> = self
+            .pending
+            .keys()
+            .filter(|hash| doc.get_change_meta_by_hash(hash).is_some())
+            .copied()
+            .collect();
+        for hash in arrived {
+            if let Some(signature) = self.pending.remove(&hash)
+                && self.vouch(doc, hash, &signature)
+            {
+                self.insert(hash, signature);
+                kept += 1;
+            }
+        }
+        kept
+    }
+
+    /// Each actor's latest verified signature that became so after `stamp`, oldest first,
+    /// `MAX_PER_MESSAGE` at most; with the stamp to take up from next time.
+    pub(crate) fn newer_than(&self, stamp: u64) -> (Vec<(ChangeHash, [u8; 64])>, u64) {
+        let mut newer: Vec<&Latest> = self
+            .latest
+            .values()
+            .filter(|latest| latest.stamp > stamp)
+            .collect();
+        newer.sort_unstable_by_key(|latest| latest.stamp);
+        newer.truncate(MAX_PER_MESSAGE);
+        let next = newer.last().map_or(stamp, |latest| latest.stamp);
+        let signatures = newer
+            .into_iter()
+            .map(|latest| (latest.hash, self.by_change[&latest.hash]))
+            .collect();
+        (signatures, next)
+    }
+
+    /// Whether a signature vouches for `actor`'s change `seq`.
+    #[cfg_attr(not(test), allow(dead_code))]
+    pub(crate) fn verified(&self, actor: &super::ActorId, seq: u64) -> bool {
+        self.latest
+            .get(actor)
+            .is_some_and(|latest| latest.seq >= seq)
+    }
+
     /// How many signatures there are.
     pub(crate) fn count(&self) -> usize {
         self.order.len()
@@ -192,10 +363,17 @@ impl Signatures {
             .collect()
     }
 
-    /// Signs `hash`, this node's latest change, with `key` and keeps the signature: it
-    /// vouches for every change of the node's before it too.
-    pub(crate) fn sign(&mut self, hash: ChangeHash, key: &crate::identity::NodeKey) {
-        self.insert(hash, key.sign(&message(&hash)));
+    /// Signs `hash`, this node's latest change in `doc`, with `key` and keeps the
+    /// signature: it vouches for every change of the node's before it too.
+    pub(crate) fn sign(
+        &mut self,
+        doc: &Document,
+        hash: ChangeHash,
+        key: &crate::identity::NodeKey,
+    ) {
+        let signature = key.sign(&message(&hash));
+        self.insert(hash, signature);
+        self.vouch(doc, hash, &signature);
         self.unsigned = None;
     }
 
@@ -209,12 +387,12 @@ impl Signatures {
         self.unsigned = Some(hash);
     }
 
-    /// Signs this node's latest change if no signature vouches for it yet; whether it
-    /// did.
-    pub(crate) fn sign_latest(&mut self, key: &crate::identity::NodeKey) -> bool {
+    /// Signs this node's latest change in `doc` if no signature vouches for it yet;
+    /// whether it did.
+    pub(crate) fn sign_latest(&mut self, doc: &Document, key: &crate::identity::NodeKey) -> bool {
         match self.unsigned {
             Some(hash) => {
-                self.sign(hash, key);
+                self.sign(doc, hash, key);
                 true
             }
             None => false,
@@ -226,7 +404,6 @@ impl Signatures {
 /// seq, committed by the node its author names (its node ID, whose key it is). None if
 /// `doc` doesn't hold the change, its author isn't a node ID, or the signature isn't
 /// that node's over it.
-#[cfg_attr(not(test), allow(dead_code))]
 pub(crate) fn check(
     doc: &Document,
     hash: &ChangeHash,
@@ -304,25 +481,73 @@ mod tests {
     #[test]
     fn the_latest_change_is_signed_later() {
         let key = NodeKey::from_seed([1; 32]);
+        let doc = Document::new();
         let mut signatures = Signatures::default();
-        assert!(!signatures.sign_latest(&key));
+        assert!(!signatures.sign_latest(&doc, &key));
         let a = ChangeHash([1; 32]);
         let b = ChangeHash([2; 32]);
         signatures.left_unsigned(a);
-        assert!(signatures.sign_latest(&key));
+        assert!(signatures.sign_latest(&doc, &key));
         assert!(signatures.by_change.contains_key(&a));
-        assert!(!signatures.sign_latest(&key));
+        assert!(!signatures.sign_latest(&doc, &key));
         signatures.left_unsigned(a);
-        signatures.sign(b, &key);
-        assert!(!signatures.sign_latest(&key));
+        signatures.sign(&doc, b, &key);
+        assert!(!signatures.sign_latest(&doc, &key));
         // Kept in the order they came, each once.
         assert_eq!(signatures.count(), 2);
         assert_eq!(signatures.since(1)[0].0, b);
         assert!(signatures.since(5).is_empty());
-        let again = Signatures::from_kept(signatures.since(0));
+        let again = Signatures::from_kept(&doc, signatures.since(0));
         assert_eq!(again.since(0), signatures.since(0));
     }
 
+    /// Received signatures: one vouches for its actor's changes up to it, an earlier one
+    /// adds nothing, one of a change not there yet waits for it, a forged one is
+    /// dropped; each actor's latest goes on to peers once (`newer_than`).
+    #[test]
+    fn received_signatures_vouch_for_their_actors_changes() {
+        let key = NodeKey::from_seed([1; 32]);
+        let other = NodeKey::from_seed([2; 32]);
+        let mut theirs = Document::new();
+        write(&mut theirs, &key.id(), &["a"], json!(1));
+        let first = theirs.get_heads()[0];
+        write(&mut theirs, &key.id(), &["b"], json!(2));
+        let second = theirs.get_heads()[0];
+        let actor = theirs.get_actor().clone();
+        let signed = |hash: ChangeHash| (hash, key.sign(&message(&hash)));
+
+        let mut ours = Document::new();
+        ours.load_incremental(&theirs.save_after(&[])).unwrap();
+        let mut signatures = Signatures::default();
+        assert!(!signatures.verified(&actor, 1));
+        assert_eq!(signatures.offer(&ours, vec![signed(second)]), 1);
+        assert!(signatures.verified(&actor, 1) && signatures.verified(&actor, 2));
+        // An earlier one vouches for nothing more; a forged one for nothing.
+        assert_eq!(signatures.offer(&ours, vec![signed(first)]), 0);
+        assert_eq!(
+            signatures.offer(&ours, vec![(second, other.sign(&message(&second)))]),
+            0
+        );
+        let (newer, stamp) = signatures.newer_than(0);
+        assert_eq!(newer, vec![signed(second)]);
+        assert!(signatures.newer_than(stamp).0.is_empty());
+
+        // A change not there yet: its signature waits for it.
+        write(&mut theirs, &key.id(), &["c"], json!(3));
+        let third = theirs.get_heads()[0];
+        assert_eq!(signatures.offer(&ours, vec![signed(third)]), 0);
+        assert!(!signatures.verified(&actor, 3));
+        ours.load_incremental(&theirs.save_after(&[second]))
+            .unwrap();
+        assert_eq!(signatures.offer(&ours, Vec::new()), 1);
+        assert!(signatures.verified(&actor, 3));
+        assert_eq!(signatures.newer_than(stamp).0, vec![signed(third)]);
+
+        // Kept in a file and read again: verified as before.
+        let again = Signatures::from_kept(&ours, signatures.since(0));
+        assert!(again.verified(&actor, 3));
+    }
+
     /// Protocol data, listed paths (within or above them) and asked-for writes are
     /// signed; others aren't.
     #[test]