fix: don't write blob to disk if it already exists (#7338)

crates/transaction-pool/src/blobstore/disk.rs

@@ -81,11 +81,12 @@ impl BlobStore for DiskFileBlobStore {
         debug!(target:"txpool::blob", num_blobs=%txs_to_delete.len(), "Removing blobs from disk");
         for tx in txs_to_delete {
             let path = self.inner.blob_disk_file(tx);
-            let _ = fs::metadata(&path).map(|meta| {
-                subsize += meta.len();
-            });
+            let filesize = fs::metadata(&path).map_or(0, |meta| meta.len());
             match fs::remove_file(&path) {
-                Ok(_) => stat.delete_succeed += 1,
+                Ok(_) => {
+                    stat.delete_succeed += 1;
+                    subsize += filesize;
+                }
                 Err(e) => {
                     stat.delete_failed += 1;
                     let err = DiskFileBlobStoreError::DeleteFile(tx, path, e);
@@ -152,6 +153,7 @@ impl DiskFileBlobStoreInner {
         }
     }
 
+    /// Creates the directory where blobs will be stored on disk.
     fn create_blob_dir(&self) -> Result<(), DiskFileBlobStoreError> {
         debug!(target:"txpool::blob", blob_dir = ?self.blob_dir, "Creating blob store");
         fs::create_dir_all(&self.blob_dir)
@@ -170,6 +172,7 @@ impl DiskFileBlobStoreInner {
         Ok(())
     }
 
+    /// Ensures blob is in the blob cache and written to the disk.
     fn insert_one(&self, tx: B256, data: BlobTransactionSidecar) -> Result<(), BlobStoreError> {
         let mut buf = Vec::with_capacity(data.fields_len());
         data.encode(&mut buf);
@@ -181,6 +184,7 @@ impl DiskFileBlobStoreInner {
         Ok(())
     }
 
+    /// Ensures blobs are in the blob cache and written to the disk.
     fn insert_many(&self, txs: Vec<(B256, BlobTransactionSidecar)>) -> Result<(), BlobStoreError> {
         let raw = txs
             .iter()
@@ -202,7 +206,9 @@ impl DiskFileBlobStoreInner {
         {
             let _lock = self.file_lock.write();
             for (path, data) in raw {
-                if let Err(err) = fs::write(&path, &data) {
+                if path.exists() {
+                    debug!(target:"txpool::blob", ?path, "Blob already exists");
+                } else if let Err(err) = fs::write(&path, &data) {
                     debug!(target:"txpool::blob", %err, ?path, "Failed to write blob file");
                 } else {
                     add += data.len();
@@ -243,7 +249,7 @@ impl DiskFileBlobStoreInner {
         self.blob_dir.join(format!("{tx:x}"))
     }
 
-    /// Retries the blob data for the given transaction hash.
+    /// Retrieves the blob data for the given transaction hash.
     #[inline]
     fn read_one(&self, tx: B256) -> Result<Option<BlobTransactionSidecar>, BlobStoreError> {
         let path = self.blob_disk_file(tx);
@@ -276,7 +282,7 @@ impl DiskFileBlobStoreInner {
             .collect()
     }
 
-    /// Retries the raw blob data for the given transaction hashes.
+    /// Retrieves the raw blob data for the given transaction hashes.
     ///
     /// Only returns the blobs that were found on file.
     #[inline]
@@ -297,18 +303,27 @@ impl DiskFileBlobStoreInner {
         res
     }
 
-    /// Retries the blob data for the given transaction hash.
+    /// Writes the blob data for the given transaction hash to the disk.
     #[inline]
     fn write_one_encoded(&self, tx: B256, data: &[u8]) -> Result<usize, DiskFileBlobStoreError> {
         trace!(target:"txpool::blob", "[{:?}] writing blob file", tx);
+        let mut add = 0;
         let path = self.blob_disk_file(tx);
         {
             let _lock = self.file_lock.write();
-            fs::write(&path, data).map_err(|e| DiskFileBlobStoreError::WriteFile(tx, path, e))?;
+            if !path.exists() {
+                fs::write(&path, data)
+                    .map_err(|e| DiskFileBlobStoreError::WriteFile(tx, path, e))?;
+                add = data.len();
+            }
         }
-        Ok(data.len())
+        Ok(add)
     }
 
+    /// Retrieves blobs for the given transaction hashes from the blob cache or disk.
+    ///
+    /// This will not return an error if there are missing blobs. Therefore, the result may be a
+    /// subset of the request or an empty vector if none of the blobs were found.
     #[inline]
     fn get_all(
         &self,
@@ -342,6 +357,9 @@ impl DiskFileBlobStoreInner {
         Ok(res)
     }
 
+    /// Retrieves blobs for the given transaction hashes from the blob cache or disk.
+    ///
+    /// Returns an error if there are any missing blobs.
     #[inline]
     fn get_exact(&self, txs: Vec<B256>) -> Result<Vec<BlobTransactionSidecar>, BlobStoreError> {
         let mut res = Vec::with_capacity(txs.len());

crates/transaction-pool/src/blobstore/mem.rs

@@ -128,7 +128,7 @@ fn remove_size(store: &mut HashMap<B256, BlobTransactionSidecar>, tx: &B256) ->
     store.remove(tx).map(|rem| rem.size()).unwrap_or_default()
 }
 
-/// Inserts the given blob into the store and returns the size of the blob that was added
+/// Inserts the given blob into the store and returns the size of the blob that was added.
 ///
 /// We don't need to handle the size updates for replacements because transactions are unique.
 #[inline]
@@ -138,6 +138,6 @@ fn insert_size(
     blob: BlobTransactionSidecar,
 ) -> usize {
     let add = blob.size();
-    store.insert(tx, blob).map(|rem| rem.size());
+    store.insert(tx, blob);
     add
 }