sync: a delta's signatures as a blob of their own

A sync message larger than the threshold for its peer (16 chunks, lower
on a slow link or after a message to it failed whole) goes as a delta:
the peer fetches its changes in chunks, from any node holding them. Its
signatures, each writer's latest with changes in it, went in the
envelope offering it: up to 65536 of them, 6 MiB, in one message, the
size the delta exists to avoid. Nor did an Automerge message count its
signatures (one per writer in a whole document pulled) towards the
threshold.

A delta's signatures are now a blob of their own: the offerer packs and
keeps them (Blobs::keep, the latest 64), and the spec names them by
BLAKE3 hash and size (DeltaSpec::signatures, None where no signature
vouches for its changes). They are served as content (BlobId::Content),
so their hash checks them. The receiver fetches them from the offerer
alone, as another node holding the changes holds signatures of its own,
before the changes, and offers them under the write lock just before
judging the changes, so they wait for them as before. The envelope of a
delta carries none. An Automerge message's signatures count towards its
size: past the threshold, it goes as a delta.

Wire version 49.

1 files changed, 52 insertions, 11 deletions

diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 5550c27..528fe47 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -217,7 +217,9 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                             )
                         })
                         .flatten();
-                    let body = if let Some((heads, changes)) = pushed {
+                    // A message's signatures, where finding them was part of deciding it.
+                    let mut carried_signatures = None;
+                    let mut body = if let Some((heads, changes)) = pushed {
                         Some(transport::SyncBody::Changes {
                             heads,
                             changes: changes
@@ -239,7 +241,16 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                                 .sync
                                 .generate(&replica.doc, Instant::now(), bytes_before);
                         message.and_then(|payload| {
-                            if payload.0.len() <= delta_threshold {
+                            // Its signatures count towards its size: a whole document
+                            // pulled carries one per writer.
+                            let signatures = replica
+                                .signatures
+                                .for_changes(&replica.doc, peer_state.sync.carried());
+                            if payload.0.len()
+                                + signatures.len() * crate::document::signatures::PACKED
+                                <= delta_threshold
+                            {
+                                carried_signatures = Some(signatures);
                                 return Some(transport::SyncBody::Message(payload));
                             }
                             // Too large for one message: all or nothing, and on a slow link
@@ -300,15 +311,25 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
                     // 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 changes sent again go
-                    // vouched for as the first time. A delta, of each writer with changes
-                    // in it, its latest. A message carrying no changes takes none.
-                    let signatures = match &body {
+                    // vouched for as the first time. A delta's, of each writer with
+                    // changes in it its latest, go as a blob of their own, from this node
+                    // (`Blobs::keep`): the peer fetches them with the delta. A message
+                    // carrying no changes takes none.
+                    let signatures = match &mut body {
                         Some(transport::SyncBody::Delta(spec)) => {
-                            replica.signatures.for_delta(&replica.doc, &spec.from)
+                            let signatures = replica.signatures.for_delta(&replica.doc, &spec.from);
+                            spec.signatures = (!signatures.is_empty()).then(|| {
+                                signer
+                                    .blobs
+                                    .keep(crate::document::signatures::pack(&signatures))
+                            });
+                            Vec::new()
                         }
-                        Some(_) => replica
-                            .signatures
-                            .for_changes(&replica.doc, peer_state.sync.carried()),
+                        Some(_) => carried_signatures.unwrap_or_else(|| {
+                            replica
+                                .signatures
+                                .for_changes(&replica.doc, peer_state.sync.carried())
+                        }),
                         None => Vec::new(),
                     };
                     let signatures = crate::document::signatures::pack(&signatures);
@@ -951,6 +972,24 @@ async fn download_and_apply_delta(
             return Ok(false);
         }
     }
+    // Its signatures first, from the offerer alone: another node holding the changes
+    // holds signatures of its own, not these.
+    let signatures = match spec.signatures {
+        Some(at) => {
+            let most = (crate::document::signatures::MAX_PER_MESSAGE
+                * crate::document::signatures::PACKED) as u64;
+            if at.size > most {
+                bail!("the offered changes' signatures are larger than any message's");
+            }
+            let bytes =
+                crate::swarm::blob::fetch_content(state, at.hash, at.size, vec![peer.to_owned()])
+                    .await
+                    .context("Fetching the offered changes' signatures")?;
+            crate::document::signatures::unpack(&bytes)
+                .context("the offered changes' signatures aren't whole")?
+        }
+        None => Vec::new(),
+    };
     info!(peer = %peer, to = spec.to.len(), "Fetching changes offered as a blob");
     let started = Instant::now();
     let bytes = crate::swarm::blob::fetch_delta(state, &spec, peers).await?;
@@ -977,10 +1016,12 @@ async fn download_and_apply_delta(
         }
         let heads_before = replica.heads().to_vec();
         {
-            // Judged first (`access::screen`), whether this document has a policy yet or
-            // not.
+            // Its signatures wait for its changes (`Signatures::offer`), each checked as
+            // they are judged (`access::screen`), whether this document has a policy yet
+            // or not.
             let epoch_seq = replica.epoch_seq;
             let replica = &mut *replica;
+            replica.signatures.offer(&replica.doc, signatures);
             let outcome = crate::access::screen::apply_bytes(
                 &mut replica.doc,
                 epoch,