fatxpool: limits handling optimizations and fixes

substrate/client/basic-authorship/src/basic_authorship.rs

@@ -530,7 +530,9 @@ where
 			);
 		}
 
-		self.transaction_pool.report_invalid(Some(self.parent_hash), unqueue_invalid);
+		self.transaction_pool
+			.report_invalid(Some(self.parent_hash), unqueue_invalid)
+			.await;
 		Ok(end_reason)
 	}
 

substrate/client/network/src/litep2p/shim/notification/config.rs

@@ -108,6 +108,7 @@ impl NotificationProtocolConfig {
 			.with_handshake(handshake.map_or(vec![1], |handshake| (*handshake).to_vec()))
 			.with_max_size(max_notification_size as usize)
 			.with_auto_accept_inbound(true)
+			.with_sync_channel_size(20 * 2048)
 			.with_fallback_names(fallback_names.into_iter().map(From::from).collect())
 			.build();
 

substrate/client/network/src/protocol/notifications/handler.rs

@@ -98,7 +98,7 @@ const LOG_TARGET: &str = "sub-libp2p::notification::handler";
 pub(crate) const ASYNC_NOTIFICATIONS_BUFFER_SIZE: usize = 8;
 
 /// Number of pending notifications in synchronous contexts.
-const SYNC_NOTIFICATIONS_BUFFER_SIZE: usize = 2048;
+const SYNC_NOTIFICATIONS_BUFFER_SIZE: usize = 20 * 2048;
 
 /// Maximum duration to open a substream and receive the handshake message. After that, we
 /// consider that we failed to open the substream.

substrate/client/network/transactions/src/config.rs

@@ -30,13 +30,13 @@ pub(crate) const PROPAGATE_TIMEOUT: time::Duration = time::Duration::from_millis
 /// Maximum number of known transaction hashes to keep for a peer.
 ///
 /// This should be approx. 2 blocks full of transactions for the network to function properly.
-pub(crate) const MAX_KNOWN_TRANSACTIONS: usize = 10240; // ~300kb per peer + overhead.
+pub(crate) const MAX_KNOWN_TRANSACTIONS: usize = 8 * 10240; // ~300kb per peer + overhead.
 
 /// Maximum allowed size for a transactions notification.
 pub(crate) const MAX_TRANSACTIONS_SIZE: u64 = MAX_RESPONSE_SIZE;
 
 /// Maximum number of transaction validation request we keep at any moment.
-pub(crate) const MAX_PENDING_TRANSACTIONS: usize = 8192;
+pub(crate) const MAX_PENDING_TRANSACTIONS: usize = 8 * 8192;
 
 /// Result of the transaction import.
 #[derive(Clone, Copy, Debug)]

substrate/client/rpc-api/src/author/mod.rs

@@ -61,7 +61,7 @@ pub trait AuthorApi<Hash, BlockHash> {
 
 	/// Remove given extrinsic from the pool and temporarily ban it to prevent reimporting.
 	#[method(name = "author_removeExtrinsic", with_extensions)]
-	fn remove_extrinsic(
+	async fn remove_extrinsic(
 		&self,
 		bytes_or_hash: Vec<hash::ExtrinsicOrHash<Hash>>,
 	) -> Result<Vec<Hash>, Error>;

substrate/client/rpc-spec-v2/src/transaction/transaction_broadcast.rs

@@ -226,20 +226,22 @@ where
 		let pool = self.pool.clone();
 		// The future expected by the executor must be `Future<Output = ()>` instead of
 		// `Future<Output = Result<(), Aborted>>`.
-		let fut = fut.map(move |result| {
-			// Connection space is cleaned when this object is dropped.
-			drop(reserved_identifier);
+		let fut = fut.then(move |result| {
+			async move {
+				// Connection space is cleaned when this object is dropped.
+				drop(reserved_identifier);
 
-			// Remove the entry from the broadcast IDs map.
-			let Some(broadcast_state) = broadcast_ids.write().remove(&drop_id) else { return };
+				// Remove the entry from the broadcast IDs map.
+				let Some(broadcast_state) = broadcast_ids.write().remove(&drop_id) else { return };
 
-			// The broadcast was not stopped.
-			if result.is_ok() {
-				return
-			}
+				// The broadcast was not stopped.
+				if result.is_ok() {
+					return
+				}
 
-			// Best effort pool removal (tx can already be finalized).
-			pool.report_invalid(None, [(broadcast_state.tx_hash, None)].into());
+				// Best effort pool removal (tx can already be finalized).
+				pool.report_invalid(None, [(broadcast_state.tx_hash, None)].into()).await;
+			}
 		});
 
 		// Keep track of this entry and the abortable handle.

substrate/client/rpc/src/author/mod.rs

@@ -153,7 +153,7 @@ where
 		Ok(self.pool.ready().map(|tx| tx.data().encode().into()).collect())
 	}
 
-	fn remove_extrinsic(
+	async fn remove_extrinsic(
 		&self,
 		ext: &Extensions,
 		bytes_or_hash: Vec<hash::ExtrinsicOrHash<TxHash<P>>>,
@@ -173,6 +173,7 @@ where
 		Ok(self
 			.pool
 			.report_invalid(None, hashes)
+			.await
 			.into_iter()
 			.map(|tx| tx.hash().clone())
 			.collect())

substrate/client/transaction-pool/Cargo.toml

@@ -40,6 +40,7 @@ sp-crypto-hashing = { workspace = true, default-features = true }
 sp-runtime = { workspace = true, default-features = true }
 sp-tracing = { workspace = true, default-features = true }
 sp-transaction-pool = { workspace = true, default-features = true }
+test-case = "3.3.1"
 thiserror = { workspace = true }
 tokio = { workspace = true, default-features = true, features = ["macros", "time"] }
 tokio-stream = { workspace = true }

substrate/client/transaction-pool/api/src/lib.rs

@@ -311,7 +311,7 @@ pub trait TransactionPool: Send + Sync {
 	/// occurred.
 	///
 	/// Function returns the transactions actually removed from the pool.
-	fn report_invalid(
+	async fn report_invalid(
 		&self,
 		at: Option<<Self::Block as BlockT>::Hash>,
 		invalid_tx_errors: TxInvalidityReportMap<TxHash<Self>>,

substrate/client/transaction-pool/src/common/api.rs

@@ -19,15 +19,9 @@
 //! Chain api required for the transaction pool.
 
 use crate::LOG_TARGET;
+use async_trait::async_trait;
 use codec::Encode;
-use futures::{
-	channel::{mpsc, oneshot},
-	future::{ready, Future, FutureExt, Ready},
-	lock::Mutex,
-	SinkExt, StreamExt,
-};
-use std::{marker::PhantomData, pin::Pin, sync::Arc};
-
+use futures::future::{ready, Future, FutureExt, Ready};
 use prometheus_endpoint::Registry as PrometheusRegistry;
 use sc_client_api::{blockchain::HeaderBackend, BlockBackend};
 use sp_api::{ApiExt, ProvideRuntimeApi};
@@ -39,13 +33,15 @@ use sp_runtime::{
 	transaction_validity::{TransactionSource, TransactionValidity},
 };
 use sp_transaction_pool::runtime_api::TaggedTransactionQueue;
+use std::{marker::PhantomData, pin::Pin, sync::Arc, time::Instant};
+use tokio::sync::{mpsc, oneshot, Mutex};
 
 use super::{
 	error::{self, Error},
 	metrics::{ApiMetrics, ApiMetricsExt},
 };
 use crate::graph;
-use tracing::{trace, warn};
+use tracing::{instrument, trace, warn, Level};
 
 /// The transaction pool logic for full client.
 pub struct FullChainApi<Client, Block> {
@@ -66,7 +62,7 @@ fn spawn_validation_pool_task(
 		Some("transaction-pool"),
 		async move {
 			loop {
-				let task = receiver.lock().await.next().await;
+				let task = receiver.lock().await.recv().await;
 				match task {
 					None => return,
 					Some(task) => task.await,
@@ -96,7 +92,7 @@ impl<Client, Block> FullChainApi<Client, Block> {
 			Ok(api) => Some(Arc::new(api)),
 		});
 
-		let (sender, receiver) = mpsc::channel(0);
+		let (sender, receiver) = mpsc::channel(1);
 
 		let receiver = Arc::new(Mutex::new(receiver));
 		spawn_validation_pool_task("transaction-pool-task-0", receiver.clone(), spawner);
@@ -106,6 +102,7 @@ impl<Client, Block> FullChainApi<Client, Block> {
 	}
 }
 
+#[async_trait]
 impl<Client, Block> graph::ChainApi for FullChainApi<Client, Block>
 where
 	Block: BlockT,
@@ -127,40 +124,38 @@ where
 		ready(self.client.block_body(hash).map_err(error::Error::from))
 	}
 
-	fn validate_transaction(
+	#[instrument(level = Level::TRACE, skip_all, target = "txpool", name = "api::validate_transaction")]
+	async fn validate_transaction(
 		&self,
 		at: <Self::Block as BlockT>::Hash,
 		source: TransactionSource,
 		uxt: graph::ExtrinsicFor<Self>,
-	) -> Self::ValidationFuture {
+	) -> Result<TransactionValidity, Self::Error> {
 		let (tx, rx) = oneshot::channel();
 		let client = self.client.clone();
-		let mut validation_pool = self.validation_pool.clone();
+		let validation_pool = self.validation_pool.clone();
 		let metrics = self.metrics.clone();
 
-		async move {
-			metrics.report(|m| m.validations_scheduled.inc());
+		metrics.report(|m| m.validations_scheduled.inc());
 
-			{
-				validation_pool
-					.send(
-						async move {
-							let res = validate_transaction_blocking(&*client, at, source, uxt);
-							let _ = tx.send(res);
-							metrics.report(|m| m.validations_finished.inc());
-						}
-						.boxed(),
-					)
-					.await
-					.map_err(|e| Error::RuntimeApi(format!("Validation pool down: {:?}", e)))?;
-			}
+		{
+			validation_pool
+				.send(
+					async move {
+						let res = validate_transaction_blocking(&*client, at, source, uxt);
+						let _ = tx.send(res);
+						metrics.report(|m| m.validations_finished.inc());
+					}
+					.boxed(),
+				)
+				.await
+				.map_err(|e| Error::RuntimeApi(format!("Validation pool down: {:?}", e)))?;
+		}
 
-			match rx.await {
-				Ok(r) => r,
-				Err(_) => Err(Error::RuntimeApi("Validation was canceled".into())),
-			}
+		match rx.await {
+			Ok(r) => r,
+			Err(_) => Err(Error::RuntimeApi("Validation was canceled".into())),
 		}
-		.boxed()
 	}
 
 	/// Validates a transaction by calling into the runtime.

substrate/client/transaction-pool/src/common/tests.rs

@@ -19,6 +19,7 @@
 //! Testing related primitives for internal usage in this crate.
 
 use crate::graph::{BlockHash, ChainApi, ExtrinsicFor, NumberFor, RawExtrinsicFor};
+use async_trait::async_trait;
 use codec::Encode;
 use parking_lot::Mutex;
 use sc_transaction_pool_api::error;
@@ -66,19 +67,20 @@ impl TestApi {
 	}
 }
 
+#[async_trait]
 impl ChainApi for TestApi {
 	type Block = Block;
 	type Error = error::Error;
 	type ValidationFuture = futures::future::Ready<error::Result<TransactionValidity>>;
 	type BodyFuture = futures::future::Ready<error::Result<Option<Vec<Extrinsic>>>>;
 
 	/// Verify extrinsic at given block.
-	fn validate_transaction(
+	async fn validate_transaction(
 		&self,
 		at: <Self::Block as BlockT>::Hash,
 		_source: TransactionSource,
 		uxt: ExtrinsicFor<Self>,
-	) -> Self::ValidationFuture {
+	) -> Result<TransactionValidity, error::Error> {
 		let uxt = (*uxt).clone();
 		self.validation_requests.lock().push(uxt.clone());
 		let hash = self.hash_and_length(&uxt).0;
@@ -155,7 +157,7 @@ impl ChainApi for TestApi {
 			_ => unimplemented!(),
 		};
 
-		futures::future::ready(Ok(res))
+		Ok(res)
 	}
 
 	fn validate_transaction_blocking(