signatures: a transfer carries one signature per writer in it

Under a policy requiring signatures, a node takes a change only once a signature
vouches for it: one of the change itself, or of a later change of the same writer. A
writer is a node's run, each start, rejoin or drop making a fresh actor, so an epoch
gathers writers for as long as it lasts, and a transfer of its whole history (a joiner
pulling the document, a delta) must carry a signature for each of them, or the joiner
waits for the rest for good.

A message's signatures are now chosen by what it carries:

* changes, up to 256 signed among them: each one's signature;
* more (a whole document pulled): each writer's latest signed change among them, found
  by the document's change metadata (`Signatures::for_changes`), vouching for the rest;
* a delta: each writer with changes in it (`get_changes_meta(from)`), its latest
  verified signature (`for_delta`).

A message carries up to 65536 (6 MiB), and as many wait at the receiver for their
changes (a delta's come after it). So a transfer leaves writers unvouched only in an
epoch with more than 65536 runs writing since it began; a remake starts the count again.
Wire version 48.

1 files changed, 97 insertions, 40 deletions

diff --git a/src/document/signatures.rs b/src/document/signatures.rs
index e5c9e69..b62e545 100644
--- a/src/document/signatures.rs
+++ b/src/document/signatures.rs
@@ -21,7 +21,7 @@
 //! more of its actor's changes than any before; one of a change not there yet waits
 //! (`pending`, a bounded number: the oldest go).
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
 use super::{ChangeHash, Document};
 
@@ -148,11 +148,17 @@ impl<'de> serde::Deserialize<'de> for SignEvery {
     }
 }
 
-/// Signatures a sync message carries at most (`for_changes`, `latest_each`), as many
-/// as wait for their changes (`MAX_PENDING`): 384 KiB of them. A whole document pulled,
-/// or a delta, carries each actor's latest, and an actor is a node's run: an epoch with
-/// more runs writing than this leaves some unvouched for in such a transfer.
-pub(crate) const MAX_PER_MESSAGE: usize = MAX_PENDING;
+/// Signatures a sync message carries at most, and as many wait for their changes
+/// (`MAX_PENDING`): 6 MiB of them. A message carries one per writer it holds changes of
+/// at most (`for_changes`, `for_delta`), and a writer is a node's run (each start, rejoin
+/// or drop makes a fresh actor) since the epoch began: a transfer of an epoch with more
+/// runs writing than this, a whole document pulled or a delta, leaves the rest unvouched
+/// for, and a node taking it under a policy requiring signatures waits for them. A remake
+/// starts the count again.
+pub(crate) const MAX_PER_MESSAGE: usize = 1 << 16;
+/// Past this many signed changes among those a message carries, it carries each
+/// writer's latest among them instead (`for_changes`): they vouch for the rest.
+const EACH_WRITER_PAST: 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;
@@ -186,7 +192,8 @@ pub(crate) fn unpack(bytes: &[u8]) -> Option<Vec<(ChangeHash, [u8; 64])>> {
     )
 }
 /// Received signatures of changes not there yet, kept at most (`offer`): the oldest go.
-const MAX_PENDING: usize = 4096;
+/// As many as a message carries, so a delta's (whose changes come after it) all wait.
+const MAX_PENDING: usize = MAX_PER_MESSAGE;
 
 /// Received signatures of changes the document doesn't hold yet, `MAX_PENDING` at most:
 /// the oldest go, so junk a peer sent once (signatures of changes that never come)
@@ -237,8 +244,6 @@ pub(crate) struct Signatures {
     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`: which became so last (`latest_each`).
-    stamp: u64,
     /// Received signatures of changes the document doesn't hold yet.
     pending: Pending,
 }
@@ -248,8 +253,6 @@ pub(crate) struct Signatures {
 struct Latest {
     seq: u64,
     hash: ChangeHash,
-    /// `Signatures::stamp` when it became the latest.
-    stamp: u64,
 }
 
 impl Signatures {
@@ -325,15 +328,7 @@ impl Signatures {
         {
             return false;
         }
-        self.stamp += 1;
-        self.latest.insert(
-            actor,
-            Latest {
-                seq,
-                hash,
-                stamp: self.stamp,
-            },
-        );
+        self.latest.insert(actor, Latest { seq, hash });
         true
     }
 
@@ -368,24 +363,40 @@ impl Signatures {
         kept
     }
 
-    /// Each actor's latest verified signature, those that became so last first,
-    /// `MAX_PER_MESSAGE` at most: for changes of any actor, as a delta carries.
-    pub(crate) fn latest_each(&self) -> Vec<(ChangeHash, [u8; 64])> {
-        let mut latest: Vec<&Latest> = self.latest.values().collect();
-        latest.sort_unstable_by_key(|latest| std::cmp::Reverse(latest.stamp));
-        latest.truncate(MAX_PER_MESSAGE);
-        latest
+    /// The signatures to send with a delta from `from` (`SyncBody::Delta`): of each
+    /// writer with changes in it, its latest verified, vouching for its changes before
+    /// it; `MAX_PER_MESSAGE` at most. The delta's changes come after it, so these wait
+    /// for them at the peer (`offer`).
+    pub(crate) fn for_delta(
+        &self,
+        doc: &Document,
+        from: &[ChangeHash],
+    ) -> Vec<(ChangeHash, [u8; 64])> {
+        let writers: HashSet<super::ActorId> = doc
+            .get_changes_meta(from)
             .into_iter()
+            .map(|meta| meta.actor.into_owned())
+            .collect();
+        writers
+            .iter()
+            .filter_map(|writer| self.latest.get(writer))
             .map(|latest| (latest.hash, self.by_change[&latest.hash]))
+            .take(MAX_PER_MESSAGE)
             .collect()
     }
 
     /// The signatures to send with `changes`, those a message carries: of each of them
-    /// signed, each vouching for its actor's changes before it among them. Whatever the
+    /// signed, each vouching for its writer's changes before it among them. Whatever the
     /// peer was sent before, so changes sent again (a message lost on the way, then
-    /// repaired) arrive vouched for as the first time. Past `MAX_PER_MESSAGE` (a message
-    /// carrying the batches of many writers), each actor's latest instead.
-    pub(crate) fn for_changes(&self, changes: &[ChangeHash]) -> Vec<(ChangeHash, [u8; 64])> {
+    /// repaired) arrive vouched for as the first time. Past `EACH_WRITER_PAST` (a whole
+    /// document pulled, a message carrying the batches of many writers), each writer's
+    /// latest among them, which vouches for the rest: one per writer,
+    /// `MAX_PER_MESSAGE` at most.
+    pub(crate) fn for_changes(
+        &self,
+        doc: &Document,
+        changes: &[ChangeHash],
+    ) -> Vec<(ChangeHash, [u8; 64])> {
         let signed: Vec<(ChangeHash, [u8; 64])> = changes
             .iter()
             .filter_map(|hash| {
@@ -394,10 +405,28 @@ impl Signatures {
                     .map(|signature| (*hash, *signature))
             })
             .collect();
-        if signed.len() > MAX_PER_MESSAGE {
-            return self.latest_each();
+        if signed.len() <= EACH_WRITER_PAST {
+            return signed;
         }
-        signed
+        let signed: HashMap<ChangeHash, [u8; 64]> = signed.into_iter().collect();
+        // Each signed one's writer and place among its changes: the metadata of the
+        // document's changes, read once (a message this large carries most of them).
+        let mut latest: HashMap<super::ActorId, (u64, ChangeHash)> = HashMap::new();
+        for meta in doc.get_changes_meta(&[]) {
+            if signed.contains_key(&meta.hash) {
+                let at = latest
+                    .entry(meta.actor.into_owned())
+                    .or_insert((meta.seq, meta.hash));
+                if meta.seq > at.0 {
+                    *at = (meta.seq, meta.hash);
+                }
+            }
+        }
+        latest
+            .into_values()
+            .map(|(_, hash)| (hash, signed[&hash]))
+            .take(MAX_PER_MESSAGE)
+            .collect()
     }
 
     /// Received signatures waiting for their changes (`offer`) checked against
@@ -424,13 +453,11 @@ impl Signatures {
                 .get(actor)
                 .is_none_or(|latest| latest.seq < change.seq())
             {
-                self.stamp += 1;
                 self.latest.insert(
                     actor.clone(),
                     Latest {
                         seq: change.seq(),
                         hash,
-                        stamp: self.stamp,
                     },
                 );
             }
@@ -588,6 +615,35 @@ mod tests {
         .unwrap();
     }
 
+    /// Past `EACH_WRITER_PAST` signed changes, a message carries each writer's latest
+    /// among them; a delta, each writer's with changes in it.
+    #[test]
+    fn each_writers_latest_past_many() {
+        let key = NodeKey::from_seed([1; 32]);
+        let mut doc = Document::new();
+        let mut signatures = Signatures::default();
+        let mut hashes = Vec::new();
+        for i in 0..=EACH_WRITER_PAST {
+            write(&mut doc, &key.id(), &["n"], json!(i));
+            let hash = doc.get_heads()[0];
+            signatures.sign(&doc, hash, &key);
+            hashes.push(hash);
+        }
+        let last = *hashes.last().unwrap();
+        assert_eq!(signatures.for_changes(&doc, &hashes).len(), 1);
+        assert_eq!(signatures.for_changes(&doc, &hashes)[0].0, last);
+        // Fewer than that: each signed one.
+        assert_eq!(signatures.for_changes(&doc, &hashes[..10]).len(), 10);
+        // Another writer's changes alone in a delta: only its latest.
+        let mut theirs = doc.fork().with_actor(ActorId::random());
+        write(&mut theirs, &key.id(), &["m"], json!(1));
+        let other = theirs.get_heads()[0];
+        signatures.sign(&theirs, other, &key);
+        let delta = signatures.for_delta(&theirs, &[last]);
+        assert_eq!(delta.len(), 1);
+        assert_eq!(delta[0].0, other);
+    }
+
     /// A listed signed path is a pattern: `*` takes any one key, within and above.
     #[test]
     fn listed_paths_take_a_star() {
@@ -718,13 +774,14 @@ mod tests {
             signatures.offer(&ours, vec![(second, other.sign(&message(&second)))]),
             0
         );
-        assert_eq!(signatures.latest_each(), vec![signed(second)]);
+        assert_eq!(signatures.for_delta(&ours, &[]), vec![signed(second)]);
+        assert!(signatures.for_delta(&ours, &[second]).is_empty());
         // Sent with the changes they are of: the signed one's alone.
         assert_eq!(
-            signatures.for_changes(&[first, second]),
+            signatures.for_changes(&ours, &[first, second]),
             vec![signed(second)]
         );
-        assert!(signatures.for_changes(&[first]).is_empty());
+        assert!(signatures.for_changes(&ours, &[first]).is_empty());
 
         // A change not there yet: its signature waits for it.
         write(&mut theirs, &key.id(), &["c"], json!(3));
@@ -735,7 +792,7 @@ mod tests {
             .unwrap();
         assert_eq!(signatures.offer(&ours, Vec::new()), 1);
         assert!(signatures.verified(&actor, 3));
-        assert_eq!(signatures.latest_each(), vec![signed(third)]);
+        assert_eq!(signatures.for_delta(&ours, &[]), vec![signed(third)]);
 
         // Kept in a file and read again: verified as before.
         let again = Signatures::from_kept(&ours, signatures.since(0));