Make eth_getTransactionCount respect the pending block number (#861)

* Make `eth_getTransactionCount` respect the `pending` block number

When the block number is `pending`, we also count the transactions
in the mempool.

* Don't remove transactions from mempool until they are mined

zilliqa/src/api/eth.rs

@@ -191,25 +191,14 @@ fn get_storage_at(params: Params, node: &Arc<Mutex<Node>>) -> Result<String> {
 }
 
 fn get_transaction_count(params: Params, node: &Arc<Mutex<Node>>) -> Result<String> {
-    trace!("get_transaction_count: params: {:?}", params);
     let mut params = params.sequence();
     let address: H160 = params.next()?;
     let block_number: BlockNumber = params.next()?;
 
-    trace!(
-        "get_transaction_count resp: {:?}",
-        node.lock()
-            .unwrap()
-            .get_account(address, block_number)?
-            .nonce
-            .to_hex()
-    );
-
     Ok(node
         .lock()
         .unwrap()
-        .get_account(address, block_number)?
-        .nonce
+        .get_transaction_count(address, block_number)?
         .to_hex())
 }
 

zilliqa/src/consensus.rs

@@ -145,7 +145,7 @@ pub struct Consensus {
     /// The persistence database
     db: Arc<Db>,
     /// Actions that act on newly created blocks
-    transaction_pool: TransactionPool,
+    pub transaction_pool: TransactionPool,
     // PRNG - non-cryptographically secure, but we don't need that here
     rng: SmallRng,
     /// Flag indicating that block creation should be postponed due to empty mempool
@@ -773,7 +773,7 @@ impl Consensus {
     }
 
     pub fn get_txns_to_execute(&mut self) -> Vec<VerifiedTransaction> {
-        std::iter::from_fn(|| self.transaction_pool.best_transaction())
+        let txns: Vec<_> = std::iter::from_fn(|| self.transaction_pool.best_transaction())
             .filter(|txn| {
                 let account_nonce = self.state.must_get_account(txn.signer).nonce;
                 // Ignore this transaction if it is no longer valid.
@@ -784,7 +784,17 @@ impl Consensus {
                     .map(|tx_nonce| tx_nonce >= account_nonce)
                     .unwrap_or(true)
             })
-            .collect()
+            .collect();
+
+        // Reinsert these transactions into the pool. In case this proposal doesn't get mined, we want to retain the
+        // pending transactions for another try later.
+        for txn in txns.clone() {
+            let account_nonce = self.state.must_get_account(txn.signer).nonce;
+            // This insertion should always succeed, because we just took this transaction out of the pool.
+            assert!(self.transaction_pool.insert_transaction(txn, account_nonce));
+        }
+
+        txns
     }
 
     pub fn get_touched_transactions(&self, address: H160) -> Result<Vec<Hash>> {
@@ -1036,19 +1046,11 @@ impl Consensus {
         );
         // as a future improvement, process the proposal before broadcasting it
         trace!(proposal_hash = ?proposal.hash(), ?proposal.header.view, ?proposal.header.number, "######### vote successful, we are proposing block");
-        // intershard transactions are not meant to be broadcast
-        let (broadcasted_transactions, opaque_transactions): (Vec<_>, Vec<_>) =
-            applied_transactions
-                .into_iter()
-                .partition(|tx| !matches!(tx.tx, SignedTransaction::Intershard { .. }));
-        // however, for the transactions that we are NOT broadcasting, we re-insert
-        // them into the pool - this is because upon broadcasting the proposal, we will
-        // have to re-execute it ourselves (in order to vote on it) and thus will
-        // need those transactions again
-        for tx in opaque_transactions {
-            let account_nonce = self.state.get_account(tx.signer)?.nonce;
-            self.transaction_pool.insert_transaction(tx, account_nonce);
-        }
+        let broadcasted_transactions = applied_transactions
+            .into_iter()
+            // intershard transactions are not meant to be broadcast
+            .filter(|tx| !matches!(tx.tx, SignedTransaction::Intershard { .. }))
+            .collect();
         Ok(Some((proposal, broadcasted_transactions)))
     }
 

zilliqa/src/node.rs

@@ -463,6 +463,25 @@ impl Node {
             .get_account(address)
     }
 
+    /// Return the `nonce` of an account. If the `block_number` is [BlockNumber::Pending], any pending transactions in
+    /// our pool are also added to the nonce.
+    pub fn get_transaction_count(
+        &self,
+        address: Address,
+        block_number: BlockNumber,
+    ) -> Result<u64> {
+        let nonce = self.get_account(address, block_number)?.nonce;
+
+        if matches!(block_number, BlockNumber::Pending) {
+            Ok(self
+                .consensus
+                .transaction_pool
+                .pending_nonce(address, nonce))
+        } else {
+            Ok(nonce)
+        }
+    }
+
     pub fn get_account_storage(
         &self,
         address: Address,

zilliqa/src/pool.rs

@@ -192,6 +192,21 @@ impl TransactionPool {
         std::mem::take(&mut self.transactions).into_values()
     }
 
+    /// Calculate the effective nonce of the sender, if all the currently pending transactions in the pool were mined
+    /// in order. Effectively, this calculates the current nonce plus the number of transactions with sequential nonces
+    /// in the pool.
+    pub fn pending_nonce(&self, sender: Address, mut current_nonce: u64) -> u64 {
+        loop {
+            let index = TxIndex::Nonced(sender, current_nonce);
+            if self.transactions.get(&index).is_none() {
+                break;
+            }
+            current_nonce += 1;
+        }
+
+        current_nonce
+    }
+
     pub fn size(&self) -> usize {
         self.transactions.len()
     }

zilliqa/tests/it/eth.rs

@@ -18,7 +18,7 @@ use futures::{future::join_all, StreamExt};
 use primitive_types::{H160, H256};
 use serde::Serialize;
 
-use crate::{deploy_contract, LocalRpcClient, Network};
+use crate::{deploy_contract, LocalRpcClient, Network, Wallet};
 
 #[zilliqa_macros::test]
 async fn call_block_number(mut network: Network) {
@@ -1056,3 +1056,34 @@ async fn block_subscription(mut network: Network) {
 
     assert!(block_stream.unsubscribe().await.unwrap());
 }
+
+#[zilliqa_macros::test]
+async fn get_transaction_count_pending(mut network: Network) {
+    async fn pending_count(wallet: &Wallet) -> u64 {
+        wallet
+            .get_transaction_count(wallet.address(), Some(BlockNumber::Pending.into()))
+            .await
+            .unwrap()
+            .as_u64()
+    }
+
+    let wallet = network.genesis_wallet().await;
+
+    assert_eq!(pending_count(&wallet).await, 0);
+
+    wallet
+        .send_transaction(TransactionRequest::pay(H160::random(), 10), None)
+        .await
+        .unwrap()
+        .tx_hash();
+
+    assert_eq!(pending_count(&wallet).await, 1);
+
+    wallet
+        .send_transaction(TransactionRequest::pay(H160::random(), 10).nonce(1), None)
+        .await
+        .unwrap()
+        .tx_hash();
+
+    assert_eq!(pending_count(&wallet).await, 2);
+}