rpc: expose missing nonces in future pool (#1973)

# Goal
The goal of this PR is to expose missing nonces in future pool which
causes problems when submitting a lot of transactions from the same
account.

Closes #1974 

## Discussions
- I had to remove node RPC test for events, since the setup
fundamentally didn't work with the Pool and I added the removed test
into e2e side.

# Checklist
- [X] Chain spec updated
- [X] Custom RPC OR Runtime API added/changed? Updated js/api-augment.
- [X] Tests added

Author: aramikm

e2e/miscellaneous/frequency.test.ts

@@ -0,0 +1,83 @@
+import '@frequency-chain/api-augment';
+import assert from 'assert';
+import { DOLLARS, createAndFundKeypair, getBlockNumber, getNonce } from '../scaffolding/helpers';
+import { KeyringPair } from '@polkadot/keyring/types';
+import { Extrinsic, ExtrinsicHelper } from '../scaffolding/extrinsicHelpers';
+import { getFundingSource } from '../scaffolding/funding';
+import { u8, Option } from '@polkadot/types';
+import { u8aToHex } from '@polkadot/util/u8a/toHex';
+
+const fundingSource: KeyringPair = getFundingSource('frequency-misc');
+
+describe('Frequency', function () {
+  describe('setup', function () {
+    let keypairA: KeyringPair;
+    let keypairB: KeyringPair;
+
+    before(async function () {
+      keypairA = await createAndFundKeypair(fundingSource, 100n * DOLLARS);
+      keypairB = await createAndFundKeypair(fundingSource, 100n * DOLLARS);
+    });
+
+    it('Get events successfully', async function () {
+      const balance_pallet = new u8(ExtrinsicHelper.api.registry, 10);
+      const transfer_event = new u8(ExtrinsicHelper.api.registry, 2);
+      const dest_account = u8aToHex(keypairB.publicKey).slice(2);
+      const beforeBlockNumber = await getBlockNumber();
+
+      const extrinsic = new Extrinsic(
+        () => ExtrinsicHelper.api.tx.balances.transferKeepAlive(keypairB.address, 1n * DOLLARS),
+        keypairA,
+        ExtrinsicHelper.api.events.balances.Transfer
+      );
+      const { target } = await extrinsic.signAndSend();
+      assert.notEqual(target, undefined, 'should have returned Transfer event');
+
+      const afterBlockNumber = await getBlockNumber();
+      let found = false;
+
+      for (let i = beforeBlockNumber + 1; i <= afterBlockNumber; i++) {
+        const block = await ExtrinsicHelper.apiPromise.rpc.chain.getBlockHash(i);
+        const events = await ExtrinsicHelper.getFrequencyEvents(block);
+        if (
+          events.find(
+            (e) => e.pallet.eq(balance_pallet) && e.event.eq(transfer_event) && e.data.toHex().includes(dest_account)
+          )
+        ) {
+          found = true;
+          break;
+        }
+      }
+
+      assert(found, 'Could not find the desired event');
+    });
+
+    it('Get missing nonce successfully', async function () {
+      const nonce = await getNonce(keypairB);
+      for (let i = 0; i < 10; i += 2) {
+        const extrinsic = new Extrinsic(
+          () => ExtrinsicHelper.api.tx.balances.transferKeepAlive(keypairA.address, 1n * DOLLARS),
+          keypairB,
+          ExtrinsicHelper.api.events.balances.Transfer
+        );
+        // intentionally we don't want an await here
+        extrinsic.signAndSend(nonce + i);
+      }
+      // wait a little for all of the above transactions to get queued
+      await new Promise((resolve) => setTimeout(resolve, 1000));
+      const missingNonce = await ExtrinsicHelper.getMissingNonceValues(keypairB.publicKey);
+      assert.equal(missingNonce.length, 4, 'Could not get missing nonce values');
+
+      // applying the missing nonce values to next transactions to unblock the stuck ones
+      for (const missing of missingNonce) {
+        const extrinsic = new Extrinsic(
+          () => ExtrinsicHelper.api.tx.balances.transferKeepAlive(keypairA.address, 1n * DOLLARS),
+          keypairB,
+          ExtrinsicHelper.api.events.balances.Transfer
+        );
+        const { target } = await extrinsic.signAndSend(missing.toNumber());
+        assert.notEqual(target, undefined, 'should have returned Transfer event');
+      }
+    });
+  });
+});

e2e/scaffolding/extrinsicHelpers.ts

@@ -9,19 +9,20 @@ import { firstValueFrom, filter, map, pipe, tap } from 'rxjs';
 import { getBlockNumber, getExistentialDeposit, log, Sr25519Signature } from './helpers';
 import autoNonce, { AutoNonce } from './autoNonce';
 import { connect, connectPromise } from './apiConnection';
-import { DispatchError, Event, SignedBlock } from '@polkadot/types/interfaces';
+import { DispatchError, Event, Index, SignedBlock } from '@polkadot/types/interfaces';
 import { IsEvent } from '@polkadot/types/metadata/decorate/types';
 import {
   HandleResponse,
   ItemizedStoragePageResponse,
   MessageSourceId,
   PaginatedStorageResponse,
   PresumptiveSuffixesResponse,
+  RpcEvent,
   SchemaResponse,
 } from '@frequency-chain/api-augment/interfaces';
 import { u8aToHex } from '@polkadot/util/u8a/toHex';
 import { u8aWrapBytes } from '@polkadot/util';
-import type { Call } from '@polkadot/types/interfaces/runtime';
+import type { AccountId32, Call, H256 } from '@polkadot/types/interfaces/runtime';
 import { hasRelayChain } from './env';
 
 export interface ReleaseSchedule {
@@ -728,6 +729,14 @@ export class ExtrinsicHelper {
     return ExtrinsicHelper.apiPromise.rpc.handles.validateHandle(base_handle);
   }
 
+  public static getFrequencyEvents(at: H256 | string): Promise<Vec<RpcEvent>> {
+    return ExtrinsicHelper.apiPromise.rpc.frequency.getEvents(at);
+  }
+
+  public static getMissingNonceValues(accountId: AccountId32 | string | Uint8Array): Promise<Vec<Index>> {
+    return ExtrinsicHelper.apiPromise.rpc.frequency.getMissingNonceValues(accountId);
+  }
+
   public static addOnChainMessage(keys: KeyringPair, schemaId: any, payload: string) {
     return new Extrinsic(
       () => ExtrinsicHelper.api.tx.messages.addOnchainMessage(null, schemaId, payload),

e2e/scaffolding/funding.ts

@@ -25,6 +25,7 @@ export const fundingSources = [
   'msa-key-management',
   'stateful-storage-handle-paginated',
   'stateful-storage-handle-itemized',
+  'frequency-misc',
 ] as const;
 
 // Get the correct key for this Funding Source

js/api-augment/definitions/frequency.ts

@@ -10,6 +10,16 @@ export default {
       ],
       type: 'Vec<RpcEvent>',
     },
+    getMissingNonceValues: {
+      description: 'Get missing nonce values for an account',
+      params: [
+        {
+          name: 'account',
+          type: 'AccountId32',
+        },
+      ],
+      type: 'Vec<Index>',
+    },
   },
   types: {
     RpcEvent: {

node/service/src/rpc/frequency_rpc.rs

@@ -11,43 +11,130 @@ use common_primitives::rpc::RpcEvent;
 use jsonrpsee::{
 	core::{async_trait, RpcResult},
 	proc_macros::rpc,
+	types::error::{CallError, ErrorObject},
 };
+use parity_scale_codec::{Codec, Decode, Encode};
+use sc_transaction_pool_api::{InPoolTransaction, TransactionPool};
 use sp_api::ProvideRuntimeApi;
-use sp_runtime::traits::Block as BlockT;
+use sp_blockchain::HeaderBackend;
+use sp_runtime::traits::{AtLeast32Bit, Block as BlockT, One};
 use std::sync::Arc;
+use substrate_frame_rpc_system::AccountNonceApi;
 use system_runtime_api::AdditionalRuntimeApi;
 
+/// This is an upper limit to restrict the number of returned nonce holes to eliminate a potential
+/// attack vector
+const MAX_RETURNED_MISSING_NONCE_SIZE: usize = 1000;
 /// Frequency MSA Custom RPC API
 #[rpc(client, server)]
-pub trait FrequencyRpcApi<B: BlockT> {
+pub trait FrequencyRpcApi<B: BlockT, AccountId, Nonce> {
 	/// gets the events for a block hash
 	#[method(name = "frequency_getEvents")]
 	fn get_events(&self, at: B::Hash) -> RpcResult<Vec<RpcEvent>>;
+	/// returns a list of missing nonce values from Future transaction pool.
+	#[method(name = "frequency_getMissingNonceValues")]
+	fn get_missing_nonce_values(&self, account: AccountId) -> RpcResult<Vec<Nonce>>;
 }
 
 /// The client handler for the API used by Frequency Service RPC with `jsonrpsee`
-pub struct FrequencyRpcHandler<C, M> {
+pub struct FrequencyRpcHandler<P: TransactionPool, C, M> {
 	client: Arc<C>,
+	pool: Arc<P>,
 	_marker: std::marker::PhantomData<M>,
 }
 
-impl<C, M> FrequencyRpcHandler<C, M> {
+impl<P: TransactionPool, C, M> FrequencyRpcHandler<P, C, M> {
 	/// Create new instance with the given reference to the client.
-	pub fn new(client: Arc<C>) -> Self {
-		Self { client, _marker: Default::default() }
+	pub fn new(client: Arc<C>, pool: Arc<P>) -> Self {
+		Self { client, pool, _marker: Default::default() }
 	}
 }
 
 #[async_trait]
-impl<C, Block> FrequencyRpcApiServer<Block> for FrequencyRpcHandler<C, Block>
+impl<P, C, Block, AccountId, Nonce> FrequencyRpcApiServer<Block, AccountId, Nonce>
+	for FrequencyRpcHandler<P, C, Block>
 where
 	Block: BlockT,
+	C: HeaderBackend<Block>,
 	C: Send + Sync + 'static,
 	C: ProvideRuntimeApi<Block>,
 	C::Api: AdditionalRuntimeApi<Block>,
+	C::Api: AccountNonceApi<Block, AccountId, Nonce>,
+	P: TransactionPool + 'static,
+	AccountId: Clone + Codec,
+	Nonce: Clone + Encode + Decode + AtLeast32Bit + 'static,
 {
 	fn get_events(&self, at: <Block as BlockT>::Hash) -> RpcResult<Vec<RpcEvent>> {
 		let api = self.client.runtime_api();
 		return map_rpc_result(api.get_events(at))
 	}
+
+	fn get_missing_nonce_values(&self, account: AccountId) -> RpcResult<Vec<Nonce>> {
+		let api = self.client.runtime_api();
+		let best = self.client.info().best_hash;
+
+		let nonce = api.account_nonce(best, account.clone()).map_err(|e| {
+			CallError::Custom(ErrorObject::owned(1, "Unable to query nonce.", Some(e.to_string())))
+		})?;
+		Ok(get_missing_nonces(&*self.pool, account, nonce))
+	}
+}
+
+/// Finds any missing nonce values inside Future pool and return them as result
+fn get_missing_nonces<P, AccountId, Nonce>(pool: &P, account: AccountId, nonce: Nonce) -> Vec<Nonce>
+where
+	P: TransactionPool,
+	AccountId: Clone + Encode,
+	Nonce: Clone + Encode + Decode + AtLeast32Bit + 'static,
+{
+	// Now we need to query the transaction pool
+	// and find transactions originating from the same sender.
+	// Since extrinsics are opaque to us, we look for them using
+	// `provides` tag. And increment the nonce if we find a transaction
+	// that matches the current one.
+	let mut current_nonce = nonce.clone();
+	let encoded_account = account.clone().encode();
+	let mut current_tag = (account.clone(), nonce).encode();
+	for tx in pool.ready() {
+		// since transactions in `ready()` need to be ordered by nonce
+		// it's fine to continue with current iterator.
+		if tx.provides().get(0) == Some(&current_tag) {
+			current_nonce += One::one();
+			current_tag = (account.clone(), current_nonce.clone()).encode();
+		}
+	}
+
+	let mut result = vec![];
+	let mut my_in_future: Vec<_> = pool
+		.futures()
+		.into_iter()
+		.filter_map(|x| match x.provides().get(0) {
+			// filtering transactions by account
+			Some(tag) if tag.starts_with(&encoded_account) => {
+				if let Ok(nonce) = Nonce::decode(&mut &tag[encoded_account.len()..]) {
+					return Some(nonce)
+				}
+				None
+			},
+			_ => None,
+		})
+		.collect();
+	my_in_future.sort();
+
+	for future_nonce in my_in_future {
+		while current_nonce < future_nonce {
+			result.push(current_nonce.clone());
+			current_nonce += One::one();
+
+			// short circuit if we reached the limit
+			if result.len() == MAX_RETURNED_MISSING_NONCE_SIZE {
+				return result
+			}
+		}
+
+		// progress the current_nonce
+		current_nonce += One::one();
+	}
+
+	result
 }

node/service/src/rpc/mod.rs

@@ -20,9 +20,6 @@ use sp_blockchain::{Error as BlockChainError, HeaderBackend, HeaderMetadata};
 
 mod frequency_rpc;
 
-#[cfg(test)]
-mod tests;
-
 /// A type representing all RPC extensions.
 pub type RpcExtension = jsonrpsee::RpcModule<()>;
 
@@ -80,15 +77,15 @@ where
 	let mut module = RpcExtension::new(());
 	let FullDeps { client, pool, deny_unsafe, command_sink } = deps;
 
-	module.merge(System::new(client.clone(), pool, deny_unsafe).into_rpc())?;
+	module.merge(System::new(client.clone(), pool.clone(), deny_unsafe).into_rpc())?;
 	module.merge(TransactionPayment::new(client.clone()).into_rpc())?;
 	module.merge(MessagesHandler::new(client.clone()).into_rpc())?;
 	module.merge(SchemasHandler::new(client.clone()).into_rpc())?;
 	module.merge(MsaHandler::new(client.clone(), offchain).into_rpc())?;
 	module.merge(StatefulStorageHandler::new(client.clone()).into_rpc())?;
 	module.merge(HandlesHandler::new(client.clone()).into_rpc())?;
 	module.merge(CapacityPaymentHandler::new(client.clone()).into_rpc())?;
-	module.merge(FrequencyRpcHandler::new(client).into_rpc())?;
+	module.merge(FrequencyRpcHandler::new(client, pool).into_rpc())?;
 	if let Some(command_sink) = command_sink {
 		module.merge(
 			// We provide the rpc handler with the sending end of the channel to allow the rpc