feat: add TransactionSigned::recover_signers with the same order (#4120)

Cargo.lock

@@ -5817,6 +5817,7 @@ dependencies = [
  "proptest",
  "proptest-derive",
  "rand 0.8.5",
+ "rayon",
  "reth-codecs",
  "reth-rlp",
  "reth-rlp-derive",

crates/consensus/auto-seal/src/lib.rs

@@ -349,10 +349,8 @@ impl StorageInner {
 
         let block = Block { header, body: transactions, ommers: vec![], withdrawals: None };
 
-        let senders =
-            block.body.iter().map(|tx| tx.recover_signer()).collect::<Option<Vec<_>>>().ok_or(
-                BlockExecutionError::Validation(BlockValidationError::SenderRecoveryError),
-            )?;
+        let senders = TransactionSigned::recover_signers(&block.body, block.body.len())
+            .ok_or(BlockExecutionError::Validation(BlockValidationError::SenderRecoveryError))?;
 
         trace!(target: "consensus::auto", transactions=?&block.body, "executing transactions");
 

crates/primitives/Cargo.toml

@@ -60,6 +60,7 @@ impl-serde = "0.4.0"
 once_cell = "1.17.0"
 zstd = { version = "0.12", features = ["experimental"] }
 paste = "1.0"
+rayon = "1.7"
 tempfile = "3.3"
 sha2 = "0.10.7"
 

crates/primitives/src/block.rs

@@ -163,7 +163,7 @@ impl SealedBlock {
 
     /// Expensive operation that recovers transaction signer. See [SealedBlockWithSenders].
     pub fn senders(&self) -> Option<Vec<Address>> {
-        self.body.iter().map(|tx| tx.recover_signer()).collect::<Option<Vec<Address>>>()
+        TransactionSigned::recover_signers(&self.body, self.body.len())
     }
 
     /// Seal sealed block with recovered transaction senders.

crates/primitives/src/transaction/mod.rs

@@ -7,6 +7,8 @@ use bytes::{Buf, BytesMut};
 use derive_more::{AsRef, Deref};
 pub use error::InvalidTransactionError;
 pub use meta::TransactionMeta;
+use once_cell::sync::Lazy;
+use rayon::prelude::{IntoParallelIterator, ParallelIterator};
 use reth_codecs::{add_arbitrary_tests, derive_arbitrary, Compact};
 use reth_rlp::{
     length_of_length, Decodable, DecodeError, Encodable, Header, EMPTY_LIST_CODE, EMPTY_STRING_CODE,
@@ -34,6 +36,15 @@ mod signature;
 mod tx_type;
 pub(crate) mod util;
 
+// Expected number of transactions where we can expect a speed-up by recovering the senders in
+// parallel.
+pub(crate) static PARALLEL_SENDER_RECOVERY_THRESHOLD: Lazy<usize> =
+    Lazy::new(|| match rayon::current_num_threads() {
+        0..=1 => usize::MAX,
+        2..=8 => 10,
+        _ => 5,
+    });
+
 /// A raw transaction.
 ///
 /// Transaction types were introduced in [EIP-2718](https://eips.ethereum.org/EIPS/eip-2718).
@@ -939,6 +950,21 @@ impl TransactionSigned {
         self.signature.recover_signer(signature_hash)
     }
 
+    /// Recovers a list of signers from a transaction list iterator
+    ///
+    /// Returns `None`, if some transaction's signature is invalid, see also
+    /// [Self::recover_signer].
+    pub fn recover_signers<'a, T>(txes: T, num_txes: usize) -> Option<Vec<Address>>
+    where
+        T: IntoParallelIterator<Item = &'a Self> + IntoIterator<Item = &'a Self> + Send,
+    {
+        if num_txes < *PARALLEL_SENDER_RECOVERY_THRESHOLD {
+            txes.into_iter().map(|tx| tx.recover_signer()).collect()
+        } else {
+            txes.into_par_iter().map(|tx| tx.recover_signer()).collect()
+        }
+    }
+
     /// Consumes the type, recover signer and return [`TransactionSignedEcRecovered`]
     ///
     /// Returns `None` if the transaction's signature is invalid, see also [Self::recover_signer].
@@ -1323,12 +1349,17 @@ impl IntoRecoveredTransaction for TransactionSignedEcRecovered {
 #[cfg(test)]
 mod tests {
     use crate::{
-        transaction::{signature::Signature, TransactionKind, TxEip1559, TxLegacy},
+        sign_message,
+        transaction::{
+            signature::Signature, TransactionKind, TxEip1559, TxLegacy,
+            PARALLEL_SENDER_RECOVERY_THRESHOLD,
+        },
         Address, Bytes, Transaction, TransactionSigned, TransactionSignedEcRecovered, H256, U256,
     };
     use bytes::BytesMut;
     use ethers_core::utils::hex;
     use reth_rlp::{Decodable, DecodeError, Encodable};
+    use secp256k1::{KeyPair, Secp256k1};
     use std::str::FromStr;
 
     #[test]
@@ -1564,4 +1595,32 @@ mod tests {
         tx.encode(&mut b);
         assert_eq!(s, hex::encode(&b));
     }
+
+    proptest::proptest! {
+        #![proptest_config(proptest::prelude::ProptestConfig::with_cases(1))]
+
+        #[test]
+        fn test_parallel_recovery_order(txes in proptest::collection::vec(proptest::prelude::any::<Transaction>(), *PARALLEL_SENDER_RECOVERY_THRESHOLD * 5)) {
+            let mut rng =rand::thread_rng();
+            let secp = Secp256k1::new();
+            let txes: Vec<TransactionSigned> = txes.into_iter().map(|mut tx| {
+                 if let Some(chain_id) = tx.chain_id() {
+                    // Otherwise we might overflow when calculating `v` on `recalculate_hash`
+                    tx.set_chain_id(chain_id % (u64::MAX / 2 - 36));
+                }
+
+                let key_pair = KeyPair::new(&secp, &mut rng);
+
+                let signature =
+                    sign_message(H256::from_slice(&key_pair.secret_bytes()[..]), tx.signature_hash()).unwrap();
+
+                TransactionSigned::from_transaction_and_signature(tx, signature)
+            }).collect();
+
+            let parallel_senders = TransactionSigned::recover_signers(&txes, txes.len()).unwrap();
+            let seq_senders = txes.iter().map(|tx| tx.recover_signer()).collect::<Option<Vec<_>>>().unwrap();
+
+            assert_eq!(parallel_senders, seq_senders);
+        }
+    }
 }

crates/revm/src/executor.rs

@@ -84,11 +84,8 @@ where
                 Err(BlockValidationError::SenderRecoveryError.into())
             }
         } else {
-            body.iter()
-                .map(|tx| {
-                    tx.recover_signer().ok_or(BlockValidationError::SenderRecoveryError.into())
-                })
-                .collect()
+            TransactionSigned::recover_signers(body, body.len())
+                .ok_or(BlockValidationError::SenderRecoveryError.into())
         }
     }
 

crates/storage/provider/src/providers/database/provider.rs

@@ -22,7 +22,10 @@ use reth_db::{
     transaction::{DbTx, DbTxMut},
     BlockNumberList, DatabaseError,
 };
-use reth_interfaces::Result;
+use reth_interfaces::{
+    executor::{BlockExecutionError, BlockValidationError},
+    Result,
+};
 use reth_primitives::{
     keccak256,
     stage::{StageCheckpoint, StageId},
@@ -1908,14 +1911,11 @@ impl<'this, TX: DbTxMut<'this> + DbTx<'this>> BlockWriter for DatabaseProvider<'
         let tx_iter = if Some(block.body.len()) == senders_len {
             block.body.into_iter().zip(senders.unwrap()).collect::<Vec<(_, _)>>()
         } else {
-            block
-                .body
-                .into_iter()
-                .map(|tx| {
-                    let signer = tx.recover_signer();
-                    (tx, signer.unwrap_or_default())
-                })
-                .collect::<Vec<(_, _)>>()
+            let senders = TransactionSigned::recover_signers(&block.body, block.body.len()).ok_or(
+                BlockExecutionError::Validation(BlockValidationError::SenderRecoveryError),
+            )?;
+            debug_assert_eq!(senders.len(), block.body.len(), "missing one or more senders");
+            block.body.into_iter().zip(senders).collect()
         };
 
         for (transaction, sender) in tx_iter {