Exclude rejected txs (#2353)

Signed-off-by: Danil <[email protected]>

core/bin/zksync_core/src/state_keeper/mod.rs

@@ -335,13 +335,22 @@ impl ZkSyncStateKeeper {
         // Keeping in mind that we regularly clean the memmpool from executing txs, it's impossible when
         // tons of rejected txs will be returned from the database.
 
-        let executed_txs = self
+        let mut executed_txs: Vec<TxHash> = self
             .pending_block
             .success_operations
             .iter()
             .filter_map(|op| op.get_executed_tx().map(|tx| tx.signed_tx.hash()))
             .collect();
 
+        executed_txs.append(
+            &mut self
+                .pending_block
+                .failed_txs
+                .iter()
+                .map(|op| op.signed_tx.tx.hash())
+                .collect(),
+        );
+
         let mempool_req = MempoolBlocksRequest::GetBlock(GetBlockRequest {
             last_priority_op_number: self.pending_block.unprocessed_priority_op_current,
             block_timestamp,

core/lib/storage/src/chain/mempool/mod.rs

@@ -399,6 +399,24 @@ impl<'a, 'c> MempoolSchema<'a, 'c> {
 
         self.remove_txs(&tx_hashes_to_remove).await?;
 
+        let priority_ops = self.get_confirmed_priority_ops().await?;
+        let mut priority_ops_to_remove = Vec::new();
+        for op in priority_ops {
+            let should_remove = self
+                .0
+                .chain()
+                .operations_schema()
+                .get_executed_priority_operation(op.serial_id as u32)
+                .await?
+                .is_some();
+            if should_remove {
+                priority_ops_to_remove.push(op.serial_id);
+            }
+        }
+
+        self.remove_priority_ops_from_mempool(&priority_ops_to_remove)
+            .await?;
+
         metrics::histogram!("sql.chain.mempool.collect_garbage", start.elapsed());
         Ok(())
     }