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,