EVM: Add pending transactions filter (#2774)

* Shift create filters

* Better refactor, add method documentations

* Fmt rs

* Refactor transaction cache out of core service

* Fmt rs

* Better rename

lib/ain-cpp-imports/src/bridge.rs

@@ -20,6 +20,7 @@ pub mod ffi {
         pub tx_type: u8,
         pub data: String,
         pub direction: u8,
+        pub entry_time: i64,
     }
 
     unsafe extern "C++" {

lib/ain-cpp-imports/src/lib.rs

@@ -26,6 +26,7 @@ mod ffi {
         pub tx_type: u8,
         pub data: String,
         pub direction: u8,
+        pub entry_time: i64,
     }
 
     const UNIMPL_MSG: &str = "This cannot be used on a test path";

lib/ain-evm/src/core.rs

@@ -1,6 +1,5 @@
 use std::{
     collections::{BTreeSet, HashMap},
-    num::NonZeroUsize,
     path::PathBuf,
     sync::Arc,
 };
@@ -11,10 +10,7 @@ use ain_contracts::{
     FixedContract,
 };
 use anyhow::format_err;
-use ethereum::{
-    AccessList, Account, Block, EnvelopedEncodable, Log, PartialHeader, TransactionAction,
-    TransactionV2,
-};
+use ethereum::{AccessList, Account, Block, Log, PartialHeader, TransactionAction, TransactionV2};
 use ethereum_types::{Bloom, BloomInput, H160, H256, U256};
 use evm::{
     executor::stack::{MemoryStackState, StackExecutor, StackSubstateMetadata},
@@ -23,7 +19,6 @@ use evm::{
 };
 use evm_runtime::tracing::using as runtime_using;
 use log::{debug, trace};
-use lru::LruCache;
 use parking_lot::Mutex;
 use vsdb_core::vsdb_set_base_dir;
 
@@ -37,93 +32,17 @@ use crate::{
     precompiles::MetachainPrecompiles,
     receipt::ReceiptService,
     storage::{traits::BlockStorage, Storage},
-    transaction::SignedTx,
+    transaction::{
+        cache::{TransactionCache, ValidateTxInfo},
+        SignedTx,
+    },
     trie::TrieDBStore,
     weiamount::{try_from_satoshi, WeiAmount},
     EVMError, Result,
 };
 
 pub type XHash = String;
 
-pub struct SignedTxCache {
-    inner: spin::Mutex<LruCache<String, SignedTx>>,
-}
-
-impl Default for SignedTxCache {
-    fn default() -> Self {
-        Self::new(ain_cpp_imports::get_ecc_lru_cache_count())
-    }
-}
-
-impl SignedTxCache {
-    pub fn new(capacity: usize) -> Self {
-        Self {
-            inner: spin::Mutex::new(LruCache::new(NonZeroUsize::new(capacity).unwrap())),
-        }
-    }
-
-    pub fn try_get_or_create(&self, key: &str) -> Result<SignedTx> {
-        let mut guard = self.inner.lock();
-        debug!("[signed-tx-cache]::get: {}", key);
-        let res = guard.try_get_or_insert(key.to_string(), || {
-            debug!("[signed-tx-cache]::create {}", key);
-            SignedTx::try_from(key)
-        })?;
-        Ok(res.clone())
-    }
-
-    pub fn pre_populate(&self, key: &str, signed_tx: SignedTx) -> Result<()> {
-        let mut guard = self.inner.lock();
-        debug!("[signed-tx-cache]::pre_populate: {}", key);
-        let _ = guard.get_or_insert(key.to_string(), move || {
-            debug!("[signed-tx-cache]::pre_populate:: create {}", key);
-            signed_tx
-        });
-
-        Ok(())
-    }
-
-    pub fn try_get_or_create_from_tx(&self, tx: &TransactionV2) -> Result<SignedTx> {
-        let data = EnvelopedEncodable::encode(tx);
-        let key = hex::encode(&data);
-        let mut guard = self.inner.lock();
-        debug!("[signed-tx-cache]::get from tx: {}", &key);
-        let res = guard.try_get_or_insert(key.clone(), || {
-            debug!("[signed-tx-cache]::create from tx {}", &key);
-            SignedTx::try_from(key.as_str())
-        })?;
-        Ok(res.clone())
-    }
-}
-
-struct TxValidationCache {
-    stateless: spin::Mutex<LruCache<String, ValidateTxInfo>>,
-}
-
-impl Default for TxValidationCache {
-    fn default() -> Self {
-        Self::new(ain_cpp_imports::get_evmv_lru_cache_count())
-    }
-}
-
-impl TxValidationCache {
-    pub fn new(capacity: usize) -> Self {
-        Self {
-            stateless: spin::Mutex::new(LruCache::new(NonZeroUsize::new(capacity).unwrap())),
-        }
-    }
-
-    pub fn get_stateless(&self, key: &str) -> Option<ValidateTxInfo> {
-        self.stateless.lock().get(key).cloned()
-    }
-
-    pub fn set_stateless(&self, key: String, value: ValidateTxInfo) -> ValidateTxInfo {
-        let mut cache = self.stateless.lock();
-        cache.put(key, value.clone());
-        value
-    }
-}
-
 #[derive(Clone, Debug)]
 pub struct ExecutionStep {
     pub pc: usize,
@@ -136,10 +55,9 @@ pub struct ExecutionStep {
 
 pub struct EVMCoreService {
     pub trie_store: Arc<TrieDBStore>,
-    pub signed_tx_cache: SignedTxCache,
     storage: Arc<Storage>,
+    pub tx_cache: Arc<TransactionCache>,
     nonce_store: Mutex<HashMap<H160, BTreeSet<U256>>>,
-    tx_validation_cache: TxValidationCache,
 }
 pub struct EthCallArgs<'a> {
     pub caller: H160,
@@ -152,12 +70,6 @@ pub struct EthCallArgs<'a> {
     pub block_number: U256,
 }
 
-#[derive(Clone, Debug)]
-pub struct ValidateTxInfo {
-    pub signed_tx: SignedTx,
-    pub max_prepay_fee: U256,
-}
-
 pub struct TransferDomainTxInfo {
     pub from: String,
     pub to: String,
@@ -175,20 +87,20 @@ fn init_vsdb(path: PathBuf) {
 }
 
 impl EVMCoreService {
-    pub fn restore(storage: Arc<Storage>, path: PathBuf) -> Self {
+    pub fn restore(storage: Arc<Storage>, tx_cache: Arc<TransactionCache>, path: PathBuf) -> Self {
         init_vsdb(path);
 
         Self {
             trie_store: Arc::new(TrieDBStore::restore()),
-            signed_tx_cache: SignedTxCache::default(),
             storage,
+            tx_cache,
             nonce_store: Mutex::new(HashMap::new()),
-            tx_validation_cache: TxValidationCache::default(),
         }
     }
 
     pub fn new_from_json(
         storage: Arc<Storage>,
+        tx_cache: Arc<TransactionCache>,
         genesis_path: PathBuf,
         evm_datadir: PathBuf,
     ) -> Result<Self> {
@@ -197,10 +109,9 @@ impl EVMCoreService {
 
         let handler = Self {
             trie_store: Arc::new(TrieDBStore::new()),
-            signed_tx_cache: SignedTxCache::default(),
             storage: Arc::clone(&storage),
+            tx_cache: Arc::clone(&tx_cache),
             nonce_store: Mutex::new(HashMap::new()),
-            tx_validation_cache: TxValidationCache::default(),
         };
         let (state_root, genesis) = TrieDBStore::genesis_state_root_from_json(
             &handler.trie_store,
@@ -324,11 +235,11 @@ impl EVMCoreService {
         let ValidateTxInfo {
             signed_tx,
             max_prepay_fee,
-        } = if let Some(validate_info) = self.tx_validation_cache.get_stateless(tx) {
+        } = if let Some(validate_info) = self.tx_cache.get_stateless(tx) {
             validate_info
         } else {
             let signed_tx = self
-                .signed_tx_cache
+                .tx_cache
                 .try_get_or_create(tx)
                 .map_err(|_| format_err!("Error: decoding raw tx to TransactionV2"))?;
             debug!("[validate_raw_tx] signed_tx : {:#?}", signed_tx);
@@ -359,7 +270,7 @@ impl EVMCoreService {
             let max_prepay_fee = calculate_max_prepay_gas_fee(&signed_tx)?;
             debug!("[validate_raw_tx] max_prepay_fee : {:x?}", max_prepay_fee);
 
-            self.tx_validation_cache.set_stateless(
+            self.tx_cache.set_stateless(
                 String::from(tx),
                 ValidateTxInfo {
                     signed_tx,
@@ -463,11 +374,11 @@ impl EVMCoreService {
         let ValidateTxInfo {
             signed_tx,
             max_prepay_fee,
-        } = if let Some(validate_info) = self.tx_validation_cache.get_stateless(tx) {
+        } = if let Some(validate_info) = self.tx_cache.get_stateless(tx) {
             validate_info
         } else {
             let signed_tx = self
-                .signed_tx_cache
+                .tx_cache
                 .try_get_or_create(tx)
                 .map_err(|_| format_err!("Error: decoding raw tx to TransactionV2"))?;
             debug!(

lib/ain-evm/src/evm.rs

@@ -30,7 +30,7 @@ use crate::{
     log::{LogService, Notification},
     receipt::ReceiptService,
     storage::{traits::BlockStorage, Storage},
-    transaction::SignedTx,
+    transaction::{cache::TransactionCache, SignedTx},
     trie::GENESIS_STATE_ROOT,
     Result,
 };
@@ -47,6 +47,7 @@ pub struct EVMServices {
     pub logs: LogService,
     pub filters: FilterService,
     pub storage: Arc<Storage>,
+    pub tx_cache: Arc<TransactionCache>,
     pub channel: NotificationChannel<Notification>,
 }
 
@@ -104,31 +105,36 @@ impl EVMServices {
                 .into());
             }
             let storage = Arc::new(Storage::new(&path)?);
+            let tx_cache = Arc::new(TransactionCache::new());
             Ok(Self {
                 core: EVMCoreService::new_from_json(
                     Arc::clone(&storage),
+                    Arc::clone(&tx_cache),
                     PathBuf::from(state_input_path),
                     path,
                 )?,
                 block: BlockService::new(Arc::clone(&storage))?,
                 receipt: ReceiptService::new(Arc::clone(&storage)),
                 logs: LogService::new(Arc::clone(&storage)),
-                filters: FilterService::new(Arc::clone(&storage)),
+                filters: FilterService::new(Arc::clone(&storage), Arc::clone(&tx_cache)),
                 storage,
+                tx_cache,
                 channel: NotificationChannel {
                     sender,
                     receiver: RwLock::new(receiver),
                 },
             })
         } else {
             let storage = Arc::new(Storage::restore(&path)?);
+            let tx_cache = Arc::new(TransactionCache::new());
             Ok(Self {
-                core: EVMCoreService::restore(Arc::clone(&storage), path),
+                core: EVMCoreService::restore(Arc::clone(&storage), Arc::clone(&tx_cache), path),
                 block: BlockService::new(Arc::clone(&storage))?,
                 receipt: ReceiptService::new(Arc::clone(&storage)),
                 logs: LogService::new(Arc::clone(&storage)),
-                filters: FilterService::new(Arc::clone(&storage)),
+                filters: FilterService::new(Arc::clone(&storage), Arc::clone(&tx_cache)),
                 storage,
+                tx_cache,
                 channel: NotificationChannel {
                     sender,
                     receiver: RwLock::new(receiver),