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, 38 insertions, 1 deletions

diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index 588868f..ec75bb5 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -100,8 +100,18 @@ pub(crate) struct Blobs {
     /// The peer whose offered delta is being downloaded (`swarm::sync`), and the peers
     /// whose offers came meanwhile (`start_delta`).
     deltas: Mutex<(Option<String>, HashSet<String>)>,
+    /// The signatures of deltas this node offered (`DeltaSpec::signatures`), served as
+    /// content by their hash: the latest `KEPT`.
+    kept: Mutex<VecDeque<Kept>>,
 }
 
+/// Bytes kept to serve as content, by their BLAKE3 hash (`Blobs::keep`).
+type Kept = ([u8; 32], Arc<Vec<u8>>);
+
+/// Deltas' signatures kept to serve (`Blobs::keep`): a peer fetches them as it takes
+/// the delta, one offered delta at a time.
+const KEPT: usize = 64;
+
 /// A sync message larger than this many chunks is sent as a delta blob instead
 /// (`swarm::sync`): below it, a blob costs more round trips than it saves; above it, a
 /// lost message costs too much to send again whole, and a slow link may not deliver it
@@ -131,7 +141,31 @@ impl Blobs {
             cache: Mutex::new(None),
             fetches: Mutex::new(HashMap::new()),
             deltas: Mutex::new((None, HashSet::new())),
+            kept: Mutex::new(VecDeque::new()),
+        }
+    }
+
+    /// Signatures for a delta this node offers, packed (`signatures::pack`): kept to
+    /// serve as content, and named for the spec.
+    pub(crate) fn keep(&self, bytes: Vec<u8>) -> crate::transport::DeltaSignatures {
+        let hash = *blake3::hash(&bytes).as_bytes();
+        let size = bytes.len() as u64;
+        let mut kept = self.kept.lock_or_recover();
+        if !kept.iter().any(|(kept, _)| *kept == hash) {
+            kept.push_back((hash, Arc::new(bytes)));
+            while kept.len() > KEPT {
+                kept.pop_front();
+            }
         }
+        crate::transport::DeltaSignatures { hash, size }
+    }
+
+    fn kept(&self, hash: &[u8; 32]) -> Option<Arc<Vec<u8>>> {
+        self.kept
+            .lock_or_recover()
+            .iter()
+            .find(|(kept, _)| kept == hash)
+            .map(|(_, bytes)| Arc::clone(bytes))
     }
 
     /// The largest sync message sent as it is (`DELTA_THRESHOLD_CHUNKS`).
@@ -293,7 +327,10 @@ async fn prepare(state: &AppState, id: BlobId, delta: Option<DeltaSpec>) -> Opti
             .flatten()
             .map(Arc::new)?
         }
-        BlobId::Content(hash) => state.remake.content(&hash)?,
+        BlobId::Content(hash) => state
+            .remake
+            .content(&hash)
+            .or_else(|| state.blobs.kept(&hash))?,
     };
     let chunk_size = state.blobs.chunk_size;
     let prepared = tokio::task::spawn_blocking(move || {