Only free in replicate if we did not hold the reference in window stage

And then free when we are consuming blobs

src/ledger.rs

@@ -9,6 +9,7 @@ use rayon::prelude::*;
 use std::collections::VecDeque;
 use std::io::Cursor;
 use transaction::Transaction;
+use std::sync::Arc;
 
 // a Block is a slice of Entries
 
@@ -52,7 +53,11 @@ pub fn reconstruct_entries_from_blobs(
             let msg = blob.read().unwrap();
             deserialize(&msg.data()[..msg.meta.size])
         };
-        blob_recycler.recycle(blob);
+        // if erasure is enabled, the window may hold a reference to the blob
+        // to be able to perform erasure decoding for missing blobs
+        if Arc::strong_count(&blob) == 1 {
+            blob_recycler.recycle(blob);
+        }
 
         match entry {
             Ok(entry) => entries.push(entry),

src/streamer.rs

@@ -331,7 +331,12 @@ fn recv_window(
                         let block_start = *consumed - (*consumed % erasure::NUM_CODED);
                         let coding_end = block_start + erasure::NUM_CODED;
                         // We've received all this block's data blobs, go and null out the window now
-                        for j in block_start..coding_end {
+                        for j in block_start..*consumed {
+                            if let Some(b) = mem::replace(&mut window[j % WINDOW_SIZE], None) {
+                                recycler.recycle(b);
+                            }
+                        }
+                        for j in *consumed..coding_end {
                             window[j % WINDOW_SIZE] = None;
                         }