Implement the filter argument in parity_pendingTransactions (#295)

* Add filters for pending transactions to RPC API

Allow filtering results in the parity_pendingTransaction endpoint as described in the API docs.

* Make arguments in parity_pendingTransactions work together

filter and limit

* fmt

* Requested changes

- filter in ethcore to avoid unneccessary copying
- rename gas_price to gasPrice
- implement requesting contract creation txs with "action"

* Some beautifying

Remove missing import and unneccessary dependency entry, add a comment
and set right lint level on new module

* fixed broken build after merge

* fmt

* fixing CI errors: type conversion

Co-authored-by: Karim Agha <karim.dev@gmail.com>

Cargo.lock

@@ -1188,6 +1188,9 @@ dependencies = [
  "price-info",
  "rlp",
  "rustc-hex 1.0.0",
+ "serde",
+ "serde_derive",
+ "serde_json",
  "trace-time",
  "transaction-pool",
  "url 2.1.0",

crates/concensus/miner/Cargo.toml

@@ -32,6 +32,9 @@ parity-util-mem = "0.7"
 parking_lot = "0.7"
 price-info = { path = "./price-info", optional = true }
 rlp = { version = "0.4.6" }
+serde = { version = "1.0", features = ["derive"] }
+serde_derive = "1.0"
+serde_json = "1.0"
 trace-time = "0.1"
 transaction-pool = "2.0.1"
 

crates/concensus/miner/src/lib.rs

@@ -44,6 +44,8 @@ extern crate error_chain;
 #[macro_use]
 extern crate log;
 #[macro_use]
+extern crate serde_derive;
+#[macro_use]
 extern crate trace_time;
 
 #[cfg(test)]

crates/concensus/miner/src/pool/mod.rs

@@ -29,6 +29,7 @@ pub mod client;
 pub mod local_transactions;
 pub mod replace;
 pub mod scoring;
+pub mod transaction_filter;
 pub mod verifier;
 
 #[cfg(test)]

crates/concensus/miner/src/pool/queue.rs

@@ -32,7 +32,10 @@ use txpool::{self, Verifier};
 use types::transaction;
 
 use pool::{
-    self, client, listener, local_transactions::LocalTransactionsList, ready, replace, scoring,
+    self, client, listener,
+    local_transactions::LocalTransactionsList,
+    ready, replace, scoring,
+    transaction_filter::{match_filter, TransactionFilter},
     verifier, PendingOrdering, PendingSettings, PrioritizationStrategy,
 };
 
@@ -120,6 +123,17 @@ impl CachedPending {
         current_timestamp: u64,
         nonce_cap: Option<&U256>,
         max_len: usize,
+    ) -> Option<Vec<Arc<pool::VerifiedTransaction>>> {
+        self.pending_filtered(block_number, current_timestamp, nonce_cap, max_len, None)
+    }
+
+    pub fn pending_filtered(
+        &self,
+        block_number: u64,
+        current_timestamp: u64,
+        nonce_cap: Option<&U256>,
+        max_len: usize,
+        filter: Option<TransactionFilter>,
     ) -> Option<Vec<Arc<pool::VerifiedTransaction>>> {
         // First check if we have anything in cache.
         let pending = self.pending.as_ref()?;
@@ -149,7 +163,14 @@ impl CachedPending {
             return None;
         }
 
-        Some(pending.iter().take(max_len).cloned().collect())
+        Some(
+            pending
+                .iter()
+                .filter(|tx| match_filter(&filter, tx))
+                .take(max_len)
+                .cloned()
+                .collect(),
+        )
     }
 }
 
@@ -426,6 +447,31 @@ impl TransactionQueue {
         pending
     }
 
+    /// Returns current pending transactions filtered.
+    ///
+    /// Different to the pending() method, this one does not cache.
+    pub fn pending_filtered<C>(
+        &self,
+        client: C,
+        settings: PendingSettings,
+        filter: &TransactionFilter,
+    ) -> Vec<Arc<pool::VerifiedTransaction>>
+    where
+        C: client::NonceClient,
+    {
+        self.collect_pending(
+            client,
+            settings.block_number,
+            settings.current_timestamp,
+            settings.nonce_cap,
+            |i| {
+                i.filter(|tx| filter.matches(tx))
+                    .take(settings.max_len)
+                    .collect()
+            },
+        )
+    }
+
     /// Collect pending transactions.
     ///
     /// NOTE This is re-computing the pending set and it might be expensive to do so.

crates/concensus/miner/src/pool/transaction_filter.rs

@@ -0,0 +1,129 @@
+// Copyright 2015-2021 Parity Technologies (UK) Ltd.
+// This file is part of OpenEthereum.
+
+// OpenEthereum is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// OpenEthereum is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with OpenEthereum.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Filter options available in the parity_pendingTransaction endpoint of the JSONRPC API.
+
+#![allow(missing_docs)]
+
+use ethereum_types::{Address, U256};
+
+use pool::VerifiedTransaction;
+use types::transaction::Action;
+
+#[allow(non_camel_case_types)]
+#[derive(Debug, Deserialize, Serialize)]
+#[serde()]
+pub enum SenderArgument {
+    eq(Address),
+    None,
+}
+
+impl Default for SenderArgument {
+    fn default() -> Self {
+        Self::None
+    }
+}
+
+impl SenderArgument {
+    fn matches(&self, value: &Address) -> bool {
+        match self {
+            Self::eq(expected) => value == expected,
+            Self::None => true,
+        }
+    }
+}
+
+#[allow(non_camel_case_types)]
+#[derive(Debug, Deserialize, Serialize)]
+#[serde()]
+pub enum ActionArgument {
+    eq(Address),
+    action(String),
+    None,
+}
+
+impl Default for ActionArgument {
+    fn default() -> Self {
+        Self::None
+    }
+}
+
+impl ActionArgument {
+    fn matches(&self, value: &Action) -> bool {
+        match self {
+            Self::eq(expected) => *value == Action::Call(*expected),
+            Self::action(name) => *value == Action::Create && name == "contract_creation",
+            Self::None => true,
+        }
+    }
+}
+
+#[allow(non_camel_case_types)]
+#[derive(Debug, Deserialize, Serialize)]
+#[serde()]
+pub enum ValueFilterArgument {
+    eq(U256),
+    lt(U256),
+    gt(U256),
+    None,
+}
+
+impl Default for ValueFilterArgument {
+    fn default() -> Self {
+        Self::None
+    }
+}
+
+impl ValueFilterArgument {
+    fn matches(&self, value: &U256) -> bool {
+        match self {
+            ValueFilterArgument::eq(expected) => value == expected,
+            ValueFilterArgument::lt(threshold) => value < threshold,
+            ValueFilterArgument::gt(threshold) => value > threshold,
+            ValueFilterArgument::None => true,
+        }
+    }
+}
+
+#[derive(Debug, Default, Deserialize, Serialize)]
+#[serde(default, rename_all = "camelCase")]
+pub struct TransactionFilter {
+    from: SenderArgument,
+    to: ActionArgument,
+    gas: ValueFilterArgument,
+    gas_price: ValueFilterArgument,
+    value: ValueFilterArgument,
+    nonce: ValueFilterArgument,
+}
+
+impl TransactionFilter {
+    pub fn matches(&self, transaction: &VerifiedTransaction) -> bool {
+        let tx = transaction.signed().tx();
+        self.from.matches(&transaction.sender)
+            && self.to.matches(&tx.action)
+            && self.gas.matches(&tx.gas)
+            && self.gas_price.matches(&tx.gas_price)
+            && self.nonce.matches(&tx.nonce)
+            && self.value.matches(&tx.value)
+    }
+}
+
+pub fn match_filter(filter: &Option<TransactionFilter>, transaction: &VerifiedTransaction) -> bool {
+    match filter {
+        Some(f) => f.matches(transaction),
+        None => true,
+    }
+}

crates/ethcore/src/miner/miner.rs

@@ -29,7 +29,11 @@ use ethcore_miner::work_notify::NotifyWork;
 use ethcore_miner::{
     gas_pricer::GasPricer,
     local_accounts::LocalAccounts,
-    pool::{self, PrioritizationStrategy, QueueStatus, TransactionQueue, VerifiedTransaction},
+    pool::{
+        self,
+        transaction_filter::{match_filter, TransactionFilter},
+        PrioritizationStrategy, QueueStatus, TransactionQueue, VerifiedTransaction,
+    },
     service_transaction_checker::ServiceTransactionChecker,
 };
 use ethereum_types::{Address, H256, U256};
@@ -1099,10 +1103,11 @@ impl miner::MinerService for Miner {
         }
     }
 
-    fn ready_transactions<C>(
+    fn ready_transactions_filtered<C>(
         &self,
         chain: &C,
         max_len: usize,
+        filter: Option<TransactionFilter>,
         ordering: miner::PendingOrdering,
     ) -> Vec<Arc<VerifiedTransaction>>
     where
@@ -1116,16 +1121,20 @@ impl miner::MinerService for Miner {
             // those transactions are valid and will just be ready to be included in next block.
             let nonce_cap = None;
 
-            self.transaction_queue.pending(
-                CachedNonceClient::new(chain, &self.nonce_cache),
-                pool::PendingSettings {
-                    block_number: chain_info.best_block_number,
-                    current_timestamp: chain_info.best_block_timestamp,
-                    nonce_cap,
-                    max_len,
-                    ordering,
-                },
-            )
+            let client = CachedNonceClient::new(chain, &self.nonce_cache);
+            let settings = pool::PendingSettings {
+                block_number: chain_info.best_block_number,
+                current_timestamp: chain_info.best_block_timestamp,
+                nonce_cap,
+                max_len,
+                ordering,
+            };
+
+            if let Some(ref f) = filter {
+                self.transaction_queue.pending_filtered(client, settings, f)
+            } else {
+                self.transaction_queue.pending(client, settings)
+            }
         };
 
         let from_pending = || {
@@ -1139,6 +1148,7 @@ impl miner::MinerService for Miner {
                                 signed.clone(),
                             )
                         })
+                        .filter(|tx| match_filter(&filter, tx))
                         .map(Arc::new)
                         .take(max_len)
                         .collect()

crates/ethcore/src/miner/mod.rs

@@ -26,7 +26,10 @@ pub mod pool_client;
 pub mod stratum;
 
 pub use self::miner::{Author, AuthoringParams, Miner, MinerOptions, Penalization, PendingSet};
-pub use ethcore_miner::{local_accounts::LocalAccounts, pool::PendingOrdering};
+pub use ethcore_miner::{
+    local_accounts::LocalAccounts,
+    pool::{transaction_filter::TransactionFilter, PendingOrdering},
+};
 
 use std::{
     collections::{BTreeMap, BTreeSet},
@@ -209,11 +212,23 @@ pub trait MinerService: Send + Sync {
     where
         C: ChainInfo + Sync;
 
-    /// Get a list of all ready transactions either ordered by priority or unordered (cheaper).
+    /// Get a list of all ready transactions either ordered by priority or unordered (cheaper),
+    /// and optionally filtered by sender, recipient, gas, gas price, value and/or nonce.
     ///
     /// Depending on the settings may look in transaction pool or only in pending block.
     /// If you don't need a full set of transactions, you can add `max_len` and create only a limited set of
     /// transactions.
+    fn ready_transactions_filtered<C>(
+        &self,
+        chain: &C,
+        max_len: usize,
+        filter: Option<TransactionFilter>,
+        ordering: PendingOrdering,
+    ) -> Vec<Arc<VerifiedTransaction>>
+    where
+        C: ChainInfo + Nonce + Sync;
+
+    /// Get an unfiltered list of all ready transactions.
     fn ready_transactions<C>(
         &self,
         chain: &C,
@@ -221,7 +236,10 @@ pub trait MinerService: Send + Sync {
         ordering: PendingOrdering,
     ) -> Vec<Arc<VerifiedTransaction>>
     where
-        C: ChainInfo + Nonce + Sync;
+        C: ChainInfo + Nonce + Sync,
+    {
+        self.ready_transactions_filtered(chain, max_len, None, ordering)
+    }
 
     /// Get a list of all transactions in the pool (some of them might not be ready for inclusion yet).
     fn queued_transactions(&self) -> Vec<Arc<VerifiedTransaction>>;

crates/rpc/src/v1/impls/parity.rs

@@ -20,7 +20,7 @@ use std::{collections::BTreeMap, str::FromStr, sync::Arc};
 use crypto::{publickey::ecies, DEFAULT_MAC};
 use ethcore::{
     client::{BlockChainClient, Call, StateClient},
-    miner::{self, MinerService},
+    miner::{self, MinerService, TransactionFilter},
     snapshot::{RestorationStatus, SnapshotService},
     state::StateInfo,
 };
@@ -32,8 +32,6 @@ use jsonrpc_core::{futures::future, BoxFuture, Result};
 use stats::PrometheusMetrics;
 use sync::{ManageNetwork, SyncProvider};
 use types::ids::BlockId;
-use version::version_data;
-
 use v1::{
     helpers::{
         self,
@@ -50,6 +48,7 @@ use v1::{
         Transaction, TransactionStats,
     },
 };
+use version::version_data;
 use Host;
 
 /// Parity implementation.
@@ -265,10 +264,15 @@ where
             .map(Into::into)
     }
 
-    fn pending_transactions(&self, limit: Option<usize>) -> Result<Vec<Transaction>> {
-        let ready_transactions = self.miner.ready_transactions(
+    fn pending_transactions(
+        &self,
+        limit: Option<usize>,
+        filter: Option<TransactionFilter>,
+    ) -> Result<Vec<Transaction>> {
+        let ready_transactions = self.miner.ready_transactions_filtered(
             &*self.client,
             limit.unwrap_or_else(usize::max_value),
+            filter,
             miner::PendingOrdering::Priority,
         );
 

crates/rpc/src/v1/tests/helpers/miner_service.rs

@@ -30,7 +30,7 @@ use ethcore::{
     },
     engines::{signer::EngineSigner, EthEngine},
     error::Error,
-    miner::{self, AuthoringParams, MinerService},
+    miner::{self, AuthoringParams, MinerService, TransactionFilter},
 };
 use ethereum_types::{Address, H256, U256};
 use miner::pool::{
@@ -264,13 +264,21 @@ impl MinerService for TestMinerService {
             .collect()
     }
 
-    fn ready_transactions<C>(
+    fn ready_transactions_filtered<C>(
         &self,
         _chain: &C,
         _max_len: usize,
+        filter: Option<TransactionFilter>,
         _ordering: miner::PendingOrdering,
     ) -> Vec<Arc<VerifiedTransaction>> {
-        self.queued_transactions()
+        match filter {
+            Some(f) => self
+                .queued_transactions()
+                .into_iter()
+                .filter(|tx| f.matches(tx))
+                .collect(),
+            None => self.queued_transactions(),
+        }
     }
 
     fn pending_transaction_hashes<C>(&self, _chain: &C) -> BTreeSet<H256> {

crates/rpc/src/v1/tests/mocked/parity.rs

@@ -19,7 +19,7 @@ use ethcore::client::{Executed, TestBlockChainClient, TransactionId};
 use ethcore_logger::RotatingLogger;
 use ethereum_types::{Address, BigEndianHash, Bloom, H256, U256};
 use miner::pool::local_transactions::Status as LocalTransactionStatus;
-use std::sync::Arc;
+use std::{str::FromStr, sync::Arc};
 use sync::ManageNetwork;
 use types::{
     receipt::{LocalizedReceipt, TransactionOutcome},
@@ -290,6 +290,85 @@ fn rpc_parity_pending_transactions() {
     assert_eq!(io.handle_request_sync(request), Some(response.to_owned()));
 }
 
+fn assert_txs_filtered(io: &IoHandler<Metadata>, filter: &str, expected: Vec<u8>) {
+    let request = format!(
+        r#"{{"jsonrpc": "2.0", "method": "parity_pendingTransactions",
+        "params":[10, {}], "id": 1}}"#,
+        filter
+    );
+    let response_str = io.handle_request_sync(&request).unwrap();
+    let response = serde_json::Value::from_str(&response_str).unwrap();
+    assert_eq!(response["result"].as_array().unwrap().len(), expected.len());
+    for n in expected {
+        let expected_sender = format!("0x000000000000000000000000000000000000005{}", n);
+        assert!(response_str.contains(&expected_sender));
+    }
+}
+
+#[test]
+fn rpc_parity_pending_transactions_with_filter() {
+    use types::transaction::{Action, Transaction, TypedTransaction};
+    let deps = Dependencies::new();
+    let io = deps.default_client();
+
+    for i in 1..6 {
+        let tx = TypedTransaction::Legacy(Transaction {
+            value: i.into(),
+            gas: (i + 0x10).into(),
+            gas_price: (i + 0x20).into(),
+            nonce: (i + 0x30).into(),
+            action: Action::Call(Address::from_low_u64_be(i + 0x40)),
+            data: vec![],
+        })
+        .fake_sign(Address::from_low_u64_be(i + 0x50));
+        deps.miner
+            .pending_transactions
+            .lock()
+            .insert(H256::from_low_u64_be(i + 0x60), tx);
+    }
+
+    let tx = TypedTransaction::Legacy(Transaction {
+        value: 0.into(),
+        gas: 0x16.into(),
+        gas_price: 0x26.into(),
+        nonce: 0x36.into(),
+        action: Action::Create,
+        data: vec![0x01, 0x02, 0x03],
+    })
+    .fake_sign(Address::from_low_u64_be(0x56));
+    deps.miner
+        .pending_transactions
+        .lock()
+        .insert(H256::from_low_u64_be(0x66), tx);
+
+    assert_txs_filtered(
+        &io,
+        r#"{"from":{"eq":"0x0000000000000000000000000000000000000052"}}"#,
+        vec![2],
+    );
+    assert_txs_filtered(
+        &io,
+        r#"{"to":{"eq":"0x0000000000000000000000000000000000000041"}}"#,
+        vec![1],
+    );
+    assert_txs_filtered(&io, r#"{"to":{"action":"contract_creation"}}"#, vec![6]);
+    assert_txs_filtered(&io, r#"{"gas":{"gt":"0x12"}}"#, vec![3, 4, 5, 6]);
+    assert_txs_filtered(&io, r#"{"gasPrice":{"eq":"0x24"}}"#, vec![4]);
+    assert_txs_filtered(&io, r#"{"nonce":{"lt":"0x33"}}"#, vec![1, 2]);
+    assert_txs_filtered(&io, r#"{"value":{"lt":"0x2"}}"#, vec![1, 6]);
+    assert_txs_filtered(
+        &io,
+        r#"{"value":{"gt":"0x1"},"gas":{"lt":"0x14"}}"#,
+        vec![2, 3],
+    );
+    assert_txs_filtered(&io, r#"{"value":{"gt":"0x6"},"gas":{"gt":"0x1"}}"#, vec![]);
+    assert_txs_filtered(
+        &io,
+        r#"{"value":{"lt":"0x60"},"nonce":{"lt":"0x60"}}"#,
+        vec![1, 2, 3, 4, 5, 6],
+    );
+}
+
 #[test]
 fn rpc_parity_encrypt() {
     let deps = Dependencies::new();

crates/rpc/src/v1/traits/parity.rs

@@ -21,6 +21,8 @@ use std::collections::BTreeMap;
 use ethereum_types::{H160, H256, H512, H64, U256, U64};
 use jsonrpc_core::{BoxFuture, Result};
 use jsonrpc_derive::rpc;
+
+use ethcore::miner::TransactionFilter;
 use v1::types::{
     BlockNumber, Bytes, CallRequest, ChainStatus, Histogram, LocalTransactionStatus, Peers,
     Receipt, RecoveredAccount, RichHeader, RpcSettings, Transaction, TransactionStats,
@@ -132,7 +134,11 @@ pub trait Parity {
 
     /// Returns all pending transactions from transaction queue.
     #[rpc(name = "parity_pendingTransactions")]
-    fn pending_transactions(&self, _: Option<usize>) -> Result<Vec<Transaction>>;
+    fn pending_transactions(
+        &self,
+        _: Option<usize>,
+        _: Option<TransactionFilter>,
+    ) -> Result<Vec<Transaction>>;
 
     /// Returns all transactions from transaction queue.
     ///

crates/rpc/src/v1/types/mod.rs

@@ -16,37 +16,6 @@
 
 //! RPC types
 
-#[cfg(test)]
-mod eth_types;
-
-mod account_info;
-mod block;
-mod block_number;
-mod bytes;
-mod call_request;
-mod confirmations;
-mod derivation;
-mod eip191;
-mod filter;
-mod histogram;
-mod index;
-mod log;
-mod node_kind;
-mod provenance;
-mod receipt;
-mod rpc_settings;
-mod secretstore;
-mod sync;
-mod trace;
-mod trace_filter;
-mod transaction;
-mod transaction_access_list;
-mod transaction_condition;
-mod transaction_request;
-mod work;
-
-pub mod pubsub;
-
 pub use self::{
     account_info::{AccountInfo, EthAccount, ExtAccountInfo, RecoveredAccount, StorageProof},
     block::{Block, BlockTransactions, Header, Rich, RichBlock, RichHeader},
@@ -82,6 +51,37 @@ pub use self::{
     work::Work,
 };
 
+#[cfg(test)]
+mod eth_types;
+
+mod account_info;
+mod block;
+mod block_number;
+mod bytes;
+mod call_request;
+mod confirmations;
+mod derivation;
+mod eip191;
+mod filter;
+mod histogram;
+mod index;
+mod log;
+mod node_kind;
+mod provenance;
+mod receipt;
+mod rpc_settings;
+mod secretstore;
+mod sync;
+mod trace;
+mod trace_filter;
+mod transaction;
+mod transaction_access_list;
+mod transaction_condition;
+mod transaction_request;
+mod work;
+
+pub mod pubsub;
+
 // TODO [ToDr] Refactor to a proper type Vec of enums?
 /// Expected tracing type.
 pub type TraceOptions = Vec<String>;