feat: delete discarded blob txs from blob store (#4408)

Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>

crates/transaction-pool/src/pool/mod.rs

@@ -336,6 +336,9 @@ where
             changed_senders,
         );
 
+        // This will discard outdated transactions based on the account's nonce
+        self.delete_discarded_blobs(outcome.discarded.iter());
+
         // notify listeners about updates
         self.notify_on_new_state(outcome);
     }
@@ -351,6 +354,10 @@ where
 
         promoted.iter().for_each(|tx| listener.pending(tx.hash(), None));
         discarded.iter().for_each(|tx| listener.discarded(tx.hash()));
+
+        // This deletes outdated blob txs from the blob store, based on the account's nonce. This is
+        // called during txpool maintenance when the pool drifted.
+        self.delete_discarded_blobs(discarded.iter());
     }
 
     /// Add a single validated transaction into the pool.
@@ -402,6 +409,7 @@ where
                     // store the sidecar in the blob store
                     self.insert_blob(hash, sidecar);
                 }
+
                 if let Some(replaced) = added.replaced_blob_transaction() {
                     // delete the replaced transaction from the blob store
                     self.delete_blob(replaced);
@@ -415,6 +423,10 @@ where
                 // Notify tx event listeners
                 self.notify_event_listeners(&added);
 
+                if let Some(discarded) = added.discarded_transactions() {
+                    self.delete_discarded_blobs(discarded.iter());
+                }
+
                 // Notify listeners for _all_ transactions
                 self.on_new_transaction(added.into_new_transaction_event());
 
@@ -735,6 +747,19 @@ where
         }
         self.blob_store_metrics.blobstore_entries.set(self.blob_store.blobs_len() as f64);
     }
+
+    /// Deletes all blob transactions that were discarded.
+    fn delete_discarded_blobs<'a>(
+        &'a self,
+        transactions: impl IntoIterator<Item = &'a Arc<ValidPoolTransaction<T::Transaction>>>,
+    ) {
+        let blob_txs = transactions
+            .into_iter()
+            .filter(|tx| tx.transaction.is_eip4844())
+            .map(|tx| *tx.hash())
+            .collect();
+        self.delete_blobs(blob_txs);
+    }
 }
 
 impl<V, T: TransactionOrdering, S> fmt::Debug for PoolInner<V, T, S> {
@@ -849,6 +874,14 @@ impl<T: PoolTransaction> AddedTransaction<T> {
         }
     }
 
+    /// Returns the discarded transactions if there were any
+    pub(crate) fn discarded_transactions(&self) -> Option<&[Arc<ValidPoolTransaction<T>>]> {
+        match self {
+            AddedTransaction::Pending(tx) => Some(&tx.discarded),
+            AddedTransaction::Parked { .. } => None,
+        }
+    }
+
     /// Returns the hash of the replaced transaction if it is a blob transaction.
     pub(crate) fn replaced_blob_transaction(&self) -> Option<H256> {
         self.replaced().filter(|tx| tx.transaction.is_eip4844()).map(|tx| *tx.transaction.hash())