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.

1 files changed, 15 insertions, 33 deletions

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.