sync: signatures sent again after a failed send

A sync message whose send fails gives back the changes it carried: the
next message takes them again. The signatures it carried weren't
given back: the peer's cursor stayed past them, so the changes went
again without them, and under a policy requiring signatures waited
for them until the next session; SIGNATURES no longer covers that.
A failed send now sends the peer all signatures again with the next
changes (each actor's latest, a few hundred at most).

1 files changed, 16 insertions, 6 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index b7bf8ed..5ba12cb 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -369,12 +369,16 @@ 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.
-                let _ = peer_state_mutex
-                    .lock()
-                    .await
+                // 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
                     .sync
-                    .send_failed(session, Instant::now());
+                    .send_failed(session, Instant::now())
+                    .is_some()
+                {
+                    peer_state.signatures_sent = 0;
+                }
                 return Err(error);
             }
         };
@@ -440,7 +444,13 @@ 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);
-                peer_state.sync.send_failed(session, 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
             };
             let Some(delay) = delay else {
                 // The new session may not have opened yet.