sync: a message carries the signatures of its changes

A sync message carried the signatures its peer wasn't sent yet, by a
cursor per peer. Changes sent again, as after a message taken as
delivered and lost on the way, went without them: the peer was past
them. Under a policy requiring signatures they waited until the
writer's next signed change, or the next session.

A message now carries the signatures of the changes it carries
(Signatures::for_changes), whatever the peer was sent before: changes
sent again arrive vouched for as the first time. Past 256 (a message
carrying the batches of many writers), each writer's latest instead,
as a delta carries. The cursor goes, and with it resetting it after a
failed send or a new session.

4 files changed, 57 insertions, 63 deletions

diff --git a/src/document/signatures.rs b/src/document/signatures.rs
index f2809fe..3001690 100644
--- a/src/document/signatures.rs
+++ b/src/document/signatures.rs
@@ -15,8 +15,8 @@
 //! `--sign-every` says (`SignEvery`), is its latest change, signed as it leaves in a
 //! sync message (`sent`), or every change at once (`all`).
 //!
-//! 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
+//! Signatures travel with the changes they are of, in the sync messages that carry them
+//! (`SyncEnvelope::signatures`, `for_changes`). 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: the oldest go).
@@ -147,7 +147,7 @@ impl<'de> serde::Deserialize<'de> for SignEvery {
     }
 }
 
-/// Signatures a sync message carries at most (`newer_than`).
+/// Signatures a sync message carries at most (`for_changes`, `latest_each`).
 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).
@@ -233,8 +233,7 @@ 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`: a peer is sent those after the last it was
-    /// (`newer_than`).
+    /// 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,
@@ -365,22 +364,36 @@ impl Signatures {
         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
+    /// 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
             .into_iter()
             .map(|latest| (latest.hash, self.by_change[&latest.hash]))
+            .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
+    /// 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])> {
+        let signed: Vec<(ChangeHash, [u8; 64])> = changes
+            .iter()
+            .filter_map(|hash| {
+                self.by_change
+                    .get(hash)
+                    .map(|signature| (*hash, *signature))
+            })
             .collect();
-        (signatures, next)
+        if signed.len() > MAX_PER_MESSAGE {
+            return self.latest_each();
+        }
+        signed
     }
 
     /// Received signatures waiting for their changes (`offer`) checked against
@@ -679,9 +692,13 @@ mod tests {
             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());
+        assert_eq!(signatures.latest_each(), vec![signed(second)]);
+        // Sent with the changes they are of: the signed one's alone.
+        assert_eq!(
+            signatures.for_changes(&[first, second]),
+            vec![signed(second)]
+        );
+        assert!(signatures.for_changes(&[first]).is_empty());
 
         // A change not there yet: its signature waits for it.
         write(&mut theirs, &key.id(), &["c"], json!(3));
@@ -692,7 +709,7 @@ mod tests {
             .unwrap();
         assert_eq!(signatures.offer(&ours, Vec::new()), 1);
         assert!(signatures.verified(&actor, 3));
-        assert_eq!(signatures.newer_than(stamp).0, vec![signed(third)]);
+        assert_eq!(signatures.latest_each(), vec![signed(third)]);
 
         // Kept in a file and read again: verified as before.
         let again = Signatures::from_kept(&ours, signatures.since(0));
diff --git a/src/document/sync.rs b/src/document/sync.rs
index a7a466b..e9afe06 100644
--- a/src/document/sync.rs
+++ b/src/document/sync.rs
@@ -983,11 +983,11 @@ impl SyncSession {
         self.state.their_heads.is_none() && we_ready && peer_ready == Some(false)
     }
 
-    /// Whether the message just generated (`generate`) carries changes.
-    pub(crate) fn carrying(&self) -> bool {
+    /// The changes the message just generated (`generate`, `push`) carries.
+    pub(crate) fn carried(&self) -> &[crate::document::ChangeHash] {
         self.sending
             .as_ref()
-            .is_some_and(|sending| !sending.sent.is_empty())
+            .map_or(&[], |sending| sending.sent.as_slice())
     }
 
     /// Which session this is: a message generated from it reports its send's outcome
diff --git a/src/state/peer.rs b/src/state/peer.rs
index f6e4a62..a7c6cc9 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -98,9 +98,6 @@ pub(crate) struct PeerState {
 
     /// Everything about the current Automerge sync with this peer; a reset replaces it.
     pub(crate) sync: SyncSession,
-    /// The signatures of changes it was sent, up to this stamp
-    /// (`Signatures::newer_than`); a reset sends them again.
-    pub(crate) signatures_sent: u64,
 
     pub(crate) liveness: Liveness,
     pub(crate) added_at: Instant,
@@ -307,7 +304,6 @@ impl PeerState {
             hello_backoff: Backoff::new(HELLO_RETRY, HELLO_RETRY_MAX),
             probe_backoff: timing.probe_backoff(),
             sync: SyncSession::new(timing.sync_reply_backoff()),
-            signatures_sent: 0,
             liveness: Liveness::Unknown,
             added_at: Instant::now(),
             last_seen: None,
@@ -378,7 +374,6 @@ impl PeerState {
         let mut reply_backoff = self.sync.reply_backoff;
         reply_backoff.reset();
         self.sync = SyncSession::new(reply_backoff);
-        self.signatures_sent = 0;
         // Said over the session that ended.
         self.reaches = None;
         self.round_trips = None;
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 5ba12cb..dd3d804 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -287,25 +287,17 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                             _ => {}
                         }
                     }
-                    // With changes, the signatures the peer wasn't sent yet: they go where
+                    // With changes, the signatures of those it carries: they go where
                     // changes go, along the trees, so a peer that got these changes
-                    // elsewhere got the signatures with them, and a message carrying
-                    // none takes none.
-                    let carries = match &body {
-                        Some(
-                            transport::SyncBody::Changes { .. } | transport::SyncBody::Delta(_),
-                        ) => true,
-                        Some(_) => peer_state.sync.carrying(),
-                        None => false,
-                    };
-                    let signatures = if carries {
-                        let (signatures, sent) =
-                            replica.signatures.newer_than(peer_state.signatures_sent);
-                        peer_state.signatures_sent = sent;
-                        crate::document::signatures::pack(&signatures)
-                    } else {
-                        Vec::new()
+                    // elsewhere got the signatures with them; and changes sent again go
+                    // vouched for as the first time. A delta, of any writer's changes,
+                    // each writer's latest. A message carrying no changes takes none.
+                    let signatures = match &body {
+                        Some(transport::SyncBody::Delta(_)) => replica.signatures.latest_each(),
+                        Some(_) => replica.signatures.for_changes(peer_state.sync.carried()),
+                        None => Vec::new(),
                     };
+                    let signatures = crate::document::signatures::pack(&signatures);
                     (
                         peer_state.sync.id(),
                         epoch,
@@ -369,16 +361,12 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
         let generated = match generated {
             Ok(generated) => generated,
             Err(error) => {
-                // Not encodable, so nothing went out: take back what it would have carried,
-                // its signatures too (all again with the next changes: they are few).
-                let mut peer_state = peer_state_mutex.lock().await;
-                if peer_state
+                // Not encodable, so nothing went out: take back what it would have carried.
+                let _ = peer_state_mutex
+                    .lock()
+                    .await
                     .sync
-                    .send_failed(session, Instant::now())
-                    .is_some()
-                {
-                    peer_state.signatures_sent = 0;
-                }
+                    .send_failed(session, Instant::now());
                 return Err(error);
             }
         };
@@ -444,13 +432,7 @@ async fn send_generated(
                 let mut peer_state = peer_state_mutex.lock().await;
                 let now = Instant::now();
                 peer_state.whole_failed(encoded.payload_len, now);
-                let delay = peer_state.sync.send_failed(session, now);
-                // The signatures it carried go again, all, with the next changes (they are
-                // few): the changes it carried do, and may wait for them.
-                if delay.is_some() {
-                    peer_state.signatures_sent = 0;
-                }
-                delay
+                peer_state.sync.send_failed(session, now)
             };
             let Some(delay) = delay else {
                 // The new session may not have opened yet.