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 || {