chore: use Byte instead of Vec<u8> in some channels

src/db/car/many.rs

@@ -317,7 +317,7 @@ impl<T: Blockstore + SettingsStore> super::super::HeaviestTipsetKeyProvider for
 }
 
 impl<WriterT: BlockstoreWriteOpsSubscribable> BlockstoreWriteOpsSubscribable for ManyCar<WriterT> {
-    fn subscribe_write_ops(&self) -> tokio::sync::broadcast::Receiver<Vec<(Cid, Vec<u8>)>> {
+    fn subscribe_write_ops(&self) -> tokio::sync::broadcast::Receiver<Vec<(Cid, bytes::Bytes)>> {
         self.writer().subscribe_write_ops()
     }
 

src/db/gc/snapshot.rs

@@ -71,7 +71,7 @@ pub struct SnapshotGarbageCollector<DB> {
     blessed_lite_snapshot: RwLock<Option<PathBuf>>,
     chain_follower: Arc<ChainFollower<DB>>,
     // On mainnet, it takes ~50MiB-200MiB RAM, depending on the time cost of snapshot export
-    memory_db: RwLock<Option<HashMap<Cid, Vec<u8>>>>,
+    memory_db: RwLock<Option<HashMap<Cid, bytes::Bytes>>>,
     memory_db_head_key: RwLock<Option<TipsetKey>>,
     exported_head_key: RwLock<Option<TipsetKey>>,
     trigger_tx: flume::Sender<()>,

src/db/mod.rs

@@ -196,7 +196,7 @@ pub trait HeaviestTipsetKeyProvider {
 
 #[auto_impl::auto_impl(&, Arc)]
 pub trait BlockstoreWriteOpsSubscribable {
-    fn subscribe_write_ops(&self) -> tokio::sync::broadcast::Receiver<Vec<(Cid, Vec<u8>)>>;
+    fn subscribe_write_ops(&self) -> tokio::sync::broadcast::Receiver<Vec<(Cid, bytes::Bytes)>>;
 
     fn unsubscribe_write_ops(&self);
 }

src/db/parity_db.rs

@@ -83,7 +83,7 @@ impl DbColumn {
     }
 }
 
-type WriteOpsBroadcastTxSender = tokio::sync::broadcast::Sender<Vec<(Cid, Vec<u8>)>>;
+type WriteOpsBroadcastTxSender = tokio::sync::broadcast::Sender<Vec<(Cid, bytes::Bytes)>>;
 
 pub struct ParityDb {
     pub db: parity_db::Db,
@@ -242,7 +242,7 @@ impl Blockstore for ParityDb {
         self.write_to_column(k.to_bytes(), block, column)?;
         match &*self.write_ops_broadcast_tx.read() {
             Some(tx) if has_subscribers(tx) => {
-                let _ = tx.send(vec![(*k, block.to_vec())]);
+                let _ = tx.send(vec![(*k, bytes::Bytes::copy_from_slice(block))]);
             }
             _ => {}
         }
@@ -263,7 +263,7 @@ impl Blockstore for ParityDb {
             let column = Self::choose_column(&k);
             let v = v.as_ref().to_vec();
             if has_subscribers {
-                values_for_subscriber.push((k, v.clone()));
+                values_for_subscriber.push((k, bytes::Bytes::copy_from_slice(&v)));
             }
             (column, k.to_bytes(), v)
         });
@@ -372,7 +372,7 @@ impl ParityDb {
 }
 
 impl super::BlockstoreWriteOpsSubscribable for ParityDb {
-    fn subscribe_write_ops(&self) -> tokio::sync::broadcast::Receiver<Vec<(Cid, Vec<u8>)>> {
+    fn subscribe_write_ops(&self) -> tokio::sync::broadcast::Receiver<Vec<(Cid, bytes::Bytes)>> {
         let tx_lock = self.write_ops_broadcast_tx.read();
         if let Some(tx) = &*tx_lock {
             return tx.subscribe();
@@ -557,14 +557,9 @@ mod test {
         for (idx, cid) in cids.iter().enumerate() {
             let data_entry = &data[idx];
             db.put_keyed(cid, data_entry).unwrap();
-            assert_eq!(
-                rx1.blocking_recv().unwrap(),
-                vec![(*cid, data_entry.clone())]
-            );
-            assert_eq!(
-                rx2.blocking_recv().unwrap(),
-                vec![(*cid, data_entry.clone())]
-            );
+            let expected = vec![(*cid, bytes::Bytes::copy_from_slice(data_entry))];
+            assert_eq!(rx1.blocking_recv().unwrap(), expected);
+            assert_eq!(rx2.blocking_recv().unwrap(), expected);
         }
 
         drop(rx1);

src/db/parity_db/gc.rs

@@ -184,7 +184,7 @@ impl DBStatistics for GarbageCollectableParityDb {
 }
 
 impl BlockstoreWriteOpsSubscribable for GarbageCollectableParityDb {
-    fn subscribe_write_ops(&self) -> tokio::sync::broadcast::Receiver<Vec<(Cid, Vec<u8>)>> {
+    fn subscribe_write_ops(&self) -> tokio::sync::broadcast::Receiver<Vec<(Cid, bytes::Bytes)>> {
         BlockstoreWriteOpsSubscribable::subscribe_write_ops(&*self.db.read())
     }