statement-store: use many workers for network statements processing (#10617)

# Description

Fixes #10814

Adds --statement-network-workers CLI parameter to enable concurrent
statement validation from the network. Previously, statements were
validated sequentially by a single worker. This change allows multiple
workers to process statements in parallel, improving throughput when
statement store is enabled

## Integration

Non-breaking change with opt-in performance improvement. Default
behavior unchanged (single worker).

Author: AndreiEres

cumulus/polkadot-omni-node/lib/src/cli.rs

@@ -223,6 +223,12 @@ pub struct Cli<Config: CliConfig> {
 	#[arg(long)]
 	pub enable_statement_store: bool,
 
+	/// Number of concurrent workers for statement validation from the network.
+	///
+	/// Only relevant when `--enable-statement-store` is used.
+	#[arg(long, default_value_t = 1)]
+	pub statement_network_workers: usize,
+
 	#[arg(skip)]
 	pub(crate) _phantom: PhantomData<Config>,
 }
@@ -268,6 +274,7 @@ impl<Config: CliConfig> Cli<Config> {
 			export_pov: self.export_pov_to_path.clone(),
 			max_pov_percentage: self.run.experimental_max_pov_percentage,
 			enable_statement_store: self.enable_statement_store,
+			statement_network_workers: self.statement_network_workers,
 			storage_monitor: self.storage_monitor.clone(),
 		}
 	}

cumulus/polkadot-omni-node/lib/src/common/mod.rs

@@ -130,6 +130,9 @@ pub struct NodeExtraArgs {
 	/// If true then the statement store will be enabled.
 	pub enable_statement_store: bool,
 
+	/// Number of concurrent workers for statement validation from the network.
+	pub statement_network_workers: usize,
+
 	/// Parameters for storage monitoring.
 	pub storage_monitor: sc_storage_monitor::StorageMonitorParams,
 }

cumulus/polkadot-omni-node/lib/src/common/spec.rs

@@ -412,6 +412,7 @@ pub(crate) trait NodeSpec: BaseNodeSpec {
 						sync_service.clone(),
 						params.keystore_container.local_keystore(),
 						statement_handler_proto,
+						node_extra_args.statement_network_workers,
 					)
 				})
 				.transpose()?;

cumulus/polkadot-omni-node/lib/src/common/statement_store.rs

@@ -63,6 +63,7 @@ pub(crate) fn build_statement_store<
 	sync_service: Arc<sc_network_sync::service::syncing_service::SyncingService<Block>>,
 	local_keystore: Arc<sc_keystore::LocalKeystore>,
 	statement_handler_proto: sc_network_statement::StatementHandlerPrototype,
+	statement_network_workers: usize,
 ) -> sc_service::error::Result<Arc<Store>> {
 	let statement_store = sc_statement_store::Store::new_shared(
 		&parachain_config.data_path,
@@ -85,6 +86,7 @@ pub(crate) fn build_statement_store<
 		statement_store.clone(),
 		parachain_config.prometheus_registry(),
 		statement_protocol_executor,
+		statement_network_workers,
 	)?;
 	task_manager.spawn_handle().spawn(
 		"network-statement-handler",

prdoc/pr_10617.prdoc

@@ -0,0 +1,14 @@
+title: "statement-store: use many workers for network statements processing"
+
+doc:
+  - audience: [Node Dev, Node Operator]
+    description: |
+      Adds --statement-network-workers CLI parameter to enable concurrent statement validation from the network.
+      Previously, statements were validated sequentially by a single worker. This change allows multiple workers
+      o process statements in parallel, improving throughput when statement store is enabled
+
+crates:
+  - name: sc-network-statement
+    bump: major
+  - name: polkadot-omni-node-lib
+    bump: major

substrate/bin/node/cli/benches/block_production.rs

@@ -117,6 +117,7 @@ fn new_node(tokio_handle: Handle) -> node_cli::service::NewFullBase {
 		config,
 		None,
 		false,
+		1,
 		|_, _| (),
 	)
 	.expect("creating a full node doesn't fail")

substrate/bin/node/cli/benches/transaction_pool.rs

@@ -104,6 +104,7 @@ fn new_node(tokio_handle: Handle) -> node_cli::service::NewFullBase {
 			config,
 			None,
 			false,
+			1,
 			|_, _| (),
 		)
 		.expect("Creates node")

substrate/bin/node/cli/src/chain_spec.rs

@@ -468,7 +468,13 @@ pub(crate) mod tests {
 
 		sc_service_test::connectivity(integration_test_config_with_two_authorities(), |config| {
 			let NewFullBase { task_manager, client, network, sync, transaction_pool, .. } =
-				new_full_base::<sc_network::NetworkWorker<_, _>>(config, None, false, |_, _| ())?;
+				new_full_base::<sc_network::NetworkWorker<_, _>>(
+					config,
+					None,
+					false,
+					1,
+					|_, _| (),
+				)?;
 			Ok(sc_service_test::TestNetComponents::new(
 				task_manager,
 				client,

substrate/bin/node/cli/src/cli.rs

@@ -43,6 +43,12 @@ pub struct Cli {
 	#[arg(long)]
 	pub no_hardware_benchmarks: bool,
 
+	/// Number of concurrent workers for statement validation from the network.
+	///
+	/// Only relevant when `--enable-statement-store` is used.
+	#[arg(long, default_value_t = 1)]
+	pub statement_network_workers: usize,
+
 	#[allow(missing_docs)]
 	#[clap(flatten)]
 	pub storage_monitor: sc_storage_monitor::StorageMonitorParams,

substrate/bin/node/cli/src/service.rs

@@ -409,6 +409,7 @@ pub fn new_full_base<N: NetworkBackend<Block, <Block as BlockT>::Hash>>(
 	config: Configuration,
 	mixnet_config: Option<sc_mixnet::Config>,
 	disable_hardware_benchmarks: bool,
+	statement_network_workers: usize,
 	with_startup_data: impl FnOnce(
 		&sc_consensus_babe::BabeBlockImport<
 			Block,
@@ -789,6 +790,7 @@ pub fn new_full_base<N: NetworkBackend<Block, <Block as BlockT>::Hash>>(
 		statement_store.clone(),
 		prometheus_registry.as_ref(),
 		statement_protocol_executor,
+		statement_network_workers,
 	)?;
 	task_manager.spawn_handle().spawn(
 		"network-statement-handler",
@@ -840,6 +842,7 @@ pub fn new_full(config: Configuration, cli: Cli) -> Result<TaskManager, ServiceE
 				config,
 				mixnet_config,
 				cli.no_hardware_benchmarks,
+				cli.statement_network_workers,
 				|_, _| (),
 			)
 			.map(|NewFullBase { task_manager, .. }| task_manager)?;
@@ -850,6 +853,7 @@ pub fn new_full(config: Configuration, cli: Cli) -> Result<TaskManager, ServiceE
 				config,
 				mixnet_config,
 				cli.no_hardware_benchmarks,
+				cli.statement_network_workers,
 				|_, _| (),
 			)
 			.map(|NewFullBase { task_manager, .. }| task_manager)?;
@@ -937,6 +941,7 @@ mod tests {
 						config,
 						None,
 						false,
+						1,
 						|block_import: &sc_consensus_babe::BabeBlockImport<Block, _, _, _, _>,
 						 babe_link: &sc_consensus_babe::BabeLink<Block>| {
 							setup_handles = Some((block_import.clone(), babe_link.clone()));
@@ -1152,6 +1157,7 @@ mod tests {
 						config,
 						None,
 						false,
+						1,
 						|_, _| (),
 					)?;
 				Ok(sc_service_test::TestNetComponents::new(