rpc v2: backpressure chainHead_v1_storage

substrate/client/rpc-servers/src/lib.rs

@@ -256,8 +256,9 @@ where
 							),
 						};
 
-						let rpc_middleware =
-							RpcServiceBuilder::new().option_layer(middleware_layer.clone());
+						let rpc_middleware = RpcServiceBuilder::new()
+							.rpc_logger(1024)
+							.option_layer(middleware_layer.clone());
 						let mut svc = service_builder
 							.set_rpc_middleware(rpc_middleware)
 							.build(methods, stop_handle);

substrate/client/rpc-spec-v2/Cargo.toml

@@ -28,7 +28,6 @@ sp-rpc = { workspace = true, default-features = true }
 sp-blockchain = { workspace = true, default-features = true }
 sp-version = { workspace = true, default-features = true }
 sc-client-api = { workspace = true, default-features = true }
-sc-utils = { workspace = true, default-features = true }
 sc-rpc = { workspace = true, default-features = true }
 codec = { workspace = true, default-features = true }
 thiserror = { workspace = true }
@@ -59,3 +58,4 @@ sc-service = { features = ["test-helpers"], workspace = true, default-features =
 assert_matches = { workspace = true }
 pretty_assertions = { workspace = true }
 sc-transaction-pool = { workspace = true, default-features = true }
+sc-utils = { workspace = true, default-features = true }

substrate/client/rpc-spec-v2/src/archive/archive.rs

@@ -30,6 +30,7 @@ use sc_client_api::{
 	Backend, BlockBackend, BlockchainEvents, CallExecutor, ChildInfo, ExecutorProvider, StorageKey,
 	StorageProvider,
 };
+use sc_rpc::SubscriptionTaskExecutor;
 use sp_api::{CallApiAt, CallContext};
 use sp_blockchain::{
 	Backend as BlockChainBackend, Error as BlockChainError, HeaderBackend, HeaderMetadata,
@@ -88,6 +89,8 @@ pub struct Archive<BE: Backend<Block>, Block: BlockT, Client> {
 	storage_max_queried_items: usize,
 	/// Phantom member to pin the block type.
 	_phantom: PhantomData<Block>,
+	/// Subscription task executor.
+	executor: SubscriptionTaskExecutor,
 }
 
 impl<BE: Backend<Block>, Block: BlockT, Client> Archive<BE, Block, Client> {
@@ -97,6 +100,7 @@ impl<BE: Backend<Block>, Block: BlockT, Client> Archive<BE, Block, Client> {
 		backend: Arc<BE>,
 		genesis_hash: GenesisHash,
 		config: ArchiveConfig,
+		executor: SubscriptionTaskExecutor,
 	) -> Self {
 		let genesis_hash = hex_string(&genesis_hash.as_ref());
 		Self {
@@ -106,6 +110,7 @@ impl<BE: Backend<Block>, Block: BlockT, Client> Archive<BE, Block, Client> {
 			storage_max_descendant_responses: config.max_descendant_responses,
 			storage_max_queried_items: config.max_queried_items,
 			_phantom: PhantomData,
+			executor,
 		}
 	}
 }
@@ -136,6 +141,9 @@ where
 		+ CallApiAt<Block>
 		+ StorageProvider<Block, BE>
 		+ 'static,
+	<<BE as sc_client_api::Backend<Block>>::State as sc_client_api::StateBackend<
+		<<Block as BlockT>::Header as HeaderT>::Hashing,
+	>>::RawIter: Send,
 {
 	fn archive_unstable_body(&self, hash: Block::Hash) -> RpcResult<Option<Vec<String>>> {
 		let Ok(Some(signed_block)) = self.client.block(hash) else { return Ok(None) };
@@ -274,7 +282,9 @@ where
 			self.client.clone(),
 			self.storage_max_descendant_responses,
 			self.storage_max_queried_items,
+			self.executor.clone(),
 		);
+
 		Ok(storage_client.handle_query(hash, items, child_trie))
 	}
 }

substrate/client/rpc-spec-v2/src/archive/archive_storage.rs

@@ -21,7 +21,8 @@
 use std::sync::Arc;
 
 use sc_client_api::{Backend, ChildInfo, StorageKey, StorageProvider};
-use sp_runtime::traits::Block as BlockT;
+use sc_rpc::SubscriptionTaskExecutor;
+use sp_runtime::traits::{Block as BlockT, Header as HeaderT};
 
 use crate::common::{
 	events::{ArchiveStorageResult, PaginatedStorageQuery, StorageQueryType},
@@ -44,9 +45,10 @@ impl<Client, Block, BE> ArchiveStorage<Client, Block, BE> {
 		client: Arc<Client>,
 		storage_max_descendant_responses: usize,
 		storage_max_queried_items: usize,
+		executor: SubscriptionTaskExecutor,
 	) -> Self {
 		Self {
-			client: Storage::new(client),
+			client: Storage::new(client, executor),
 			storage_max_descendant_responses,
 			storage_max_queried_items,
 		}
@@ -55,9 +57,12 @@ impl<Client, Block, BE> ArchiveStorage<Client, Block, BE> {
 
 impl<Client, Block, BE> ArchiveStorage<Client, Block, BE>
 where
-	Block: BlockT + 'static,
-	BE: Backend<Block> + 'static,
-	Client: StorageProvider<Block, BE> + 'static,
+	Block: BlockT + Send + 'static,
+	BE: Backend<Block> + Send + 'static,
+	Client: StorageProvider<Block, BE> + Send + Sync + 'static,
+	<<BE as sc_client_api::Backend<Block>>::State as sc_client_api::StateBackend<
+		<<Block as BlockT>::Header as HeaderT>::Hashing,
+	>>::RawIter: Send,
 {
 	/// Generate the response of the `archive_storage` method.
 	pub fn handle_query(
@@ -70,6 +75,8 @@ where
 		items.truncate(self.storage_max_queried_items);
 
 		let mut storage_results = Vec::with_capacity(items.len());
+		let mut query_iter = Vec::new();
+
 		for item in items {
 			match item.query_type {
 				StorageQueryType::Value => {
@@ -92,38 +99,39 @@ where
 						Err(error) => return ArchiveStorageResult::err(error),
 					},
 				StorageQueryType::DescendantsValues => {
-					match self.client.query_iter_pagination(
-						QueryIter {
-							query_key: item.key,
-							ty: IterQueryType::Value,
-							pagination_start_key: item.pagination_start_key,
-						},
-						hash,
-						child_key.as_ref(),
-						self.storage_max_descendant_responses,
-					) {
-						Ok((results, _)) => storage_results.extend(results),
-						Err(error) => return ArchiveStorageResult::err(error),
-					}
+					query_iter.push(QueryIter {
+						query_key: item.key,
+						ty: IterQueryType::Value,
+						pagination_start_key: item.pagination_start_key,
+					});
 				},
 				StorageQueryType::DescendantsHashes => {
-					match self.client.query_iter_pagination(
-						QueryIter {
-							query_key: item.key,
-							ty: IterQueryType::Hash,
-							pagination_start_key: item.pagination_start_key,
-						},
-						hash,
-						child_key.as_ref(),
-						self.storage_max_descendant_responses,
-					) {
-						Ok((results, _)) => storage_results.extend(results),
-						Err(error) => return ArchiveStorageResult::err(error),
-					}
+					query_iter.push(QueryIter {
+						query_key: item.key,
+						ty: IterQueryType::Hash,
+						pagination_start_key: item.pagination_start_key,
+					});
 				},
 			};
 		}
 
+		if !query_iter.is_empty() {
+			let mut rx = self.client.query_iter_pagination(
+				query_iter,
+				hash,
+				child_key,
+				Some(self.storage_max_descendant_responses),
+			);
+
+			while let Some(val) = rx.blocking_recv() {
+				match val {
+					Ok(Some(value)) => storage_results.push(value),
+					Ok(None) => continue,
+					Err(error) => return ArchiveStorageResult::err(error),
+				}
+			}
+		}
+
 		ArchiveStorageResult::ok(storage_results, discarded_items)
 	}
 }

substrate/client/rpc-spec-v2/src/archive/tests.rs

@@ -38,7 +38,7 @@ use sc_block_builder::BlockBuilderBuilder;
 use sc_client_api::ChildInfo;
 use sp_blockchain::HeaderBackend;
 use sp_consensus::BlockOrigin;
-use sp_core::{Blake2Hasher, Hasher};
+use sp_core::{testing::TaskExecutor, Blake2Hasher, Hasher};
 use sp_runtime::{
 	traits::{Block as BlockT, Header as HeaderT},
 	SaturatedConversion,
@@ -79,6 +79,7 @@ fn setup_api(
 		backend,
 		CHAIN_GENESIS,
 		ArchiveConfig { max_descendant_responses, max_queried_items },
+		Arc::new(TaskExecutor::new()),
 	)
 	.into_rpc();