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.

1 files changed, 40 insertions, 1 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 2cbfa11..ed3679d 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -146,7 +146,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
             move || -> (u64, Result<Option<transport::EncodedMessage>>) {
                 // Label the message with the epoch of the document it was generated
                 // from, read under the same lock.
-                let (session, epoch, epoch_seq, body, writer, delivered) = {
+                let (session, epoch, epoch_seq, body, writer, delivered, signatures) = {
                     let replica = doc.blocking_read();
                     let mut peer_state = psm.blocking_lock();
                     let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
@@ -258,6 +258,15 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                             _ => {}
                         }
                     }
+                    // With what it carries, the signatures the peer wasn't sent yet.
+                    let signatures = if body.is_some() {
+                        let (signatures, sent) =
+                            replica.signatures.newer_than(peer_state.signatures_sent);
+                        peer_state.signatures_sent = sent;
+                        crate::document::signatures::pack(&signatures)
+                    } else {
+                        Vec::new()
+                    };
                     (
                         peer_state.sync.id(),
                         epoch,
@@ -265,6 +274,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                         body,
                         writer,
                         delivered,
+                        signatures,
                     )
                 };
                 let Some(body) = body else {
@@ -304,6 +314,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                     writer,
                     delivered,
                     sent: clock.now_millis(),
+                    signatures: serde_bytes::ByteBuf::from(signatures),
                 };
                 let encoded = transport
                     .encode(&peer_owned, &ParsedMessage::Sync(envelope))
@@ -464,6 +475,11 @@ pub(crate) async fn handle_sync(
         return Ok(());
     };
     let epoch_label = epoch;
+    // Signatures of changes (`document::signatures`): a malformed list, or more than any
+    // sender sends, counts for nothing.
+    let signatures = crate::document::signatures::unpack(&envelope.signatures)
+        .filter(|signatures| signatures.len() <= crate::document::signatures::MAX_PER_MESSAGE)
+        .unwrap_or_default();
     let from_ready = envelope.ready;
     let writer = envelope.writer;
     let delivered = envelope.delivered;
@@ -496,6 +512,19 @@ pub(crate) async fn handle_sync(
                 peer_state.last_sync_rx = Some(Instant::now());
                 peer_state.sync.answered_by_offer();
             }
+            // Of changes the blob brings: they wait for them (`Signatures::offer`).
+            if !signatures.is_empty() {
+                let doc = Arc::clone(&state.doc);
+                tokio::task::spawn_blocking(move || {
+                    let mut replica = doc.blocking_write();
+                    if replica.epoch == epoch {
+                        let replica = &mut *replica;
+                        replica.signatures.offer(&replica.doc, signatures);
+                    }
+                })
+                .await
+                .context("Signatures task panicked")?;
+            }
             fetch_delta(state, peer, epoch, spec);
             return Ok(());
         }
@@ -601,6 +630,11 @@ pub(crate) async fn handle_sync(
             stats.asked_known_elsewhere(asked);
             peer_state.asked_known += (asked.0 + asked.1) as u64;
             replica.refresh_heads();
+            // The signatures it sent, and those waiting for what it brought.
+            if !signatures.is_empty() || replica.heads() != heads_before.as_slice() {
+                let replica = &mut *replica;
+                replica.signatures.offer(&replica.doc, signatures);
+            }
             let heads_after = replica.heads().to_vec();
             if delivered {
                 // Level with the sender now, which takes the message as delivered:
@@ -831,6 +865,11 @@ async fn download_and_apply_delta(
         let heads_before = replica.heads().to_vec();
         crate::document::changes::load_delta(&mut replica.doc, &bytes)?;
         replica.refresh_heads();
+        // Signatures waiting for what it brought (`Signatures::offer`).
+        {
+            let replica = &mut *replica;
+            replica.signatures.offer(&replica.doc, Vec::new());
+        }
         let missing = crate::document::changes::missing(&replica.doc, &spec.to).len();
         if missing > 0 {
             bail!("{missing} of the offered heads are still missing after loading");