rpc v2: backpressure chainHead_v1_storage

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

@@ -225,7 +225,7 @@ where
 						};
 
 						let rpc_middleware =
-							RpcServiceBuilder::new().option_layer(middleware_layer.clone());
+							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,
@@ -43,36 +44,6 @@ use std::{collections::HashSet, marker::PhantomData, sync::Arc};
 
 use super::archive_storage::ArchiveStorage;
 
-/// The configuration of [`Archive`].
-pub struct ArchiveConfig {
-	/// The maximum number of items the `archive_storage` can return for a descendant query before
-	/// pagination is required.
-	pub max_descendant_responses: usize,
-	/// The maximum number of queried items allowed for the `archive_storage` at a time.
-	pub max_queried_items: usize,
-}
-
-/// The maximum number of items the `archive_storage` can return for a descendant query before
-/// pagination is required.
-///
-/// Note: this is identical to the `chainHead` value.
-const MAX_DESCENDANT_RESPONSES: usize = 5;
-
-/// The maximum number of queried items allowed for the `archive_storage` at a time.
-///
-/// Note: A queried item can also be a descendant query which can return up to
-/// `MAX_DESCENDANT_RESPONSES`.
-const MAX_QUERIED_ITEMS: usize = 8;
-
-impl Default for ArchiveConfig {
-	fn default() -> Self {
-		Self {
-			max_descendant_responses: MAX_DESCENDANT_RESPONSES,
-			max_queried_items: MAX_QUERIED_ITEMS,
-		}
-	}
-}
-
 /// An API for archive RPC calls.
 pub struct Archive<BE: Backend<Block>, Block: BlockT, Client> {
 	/// Substrate client.
@@ -81,13 +52,10 @@ pub struct Archive<BE: Backend<Block>, Block: BlockT, Client> {
 	backend: Arc<BE>,
 	/// The hexadecimal encoded hash of the genesis block.
 	genesis_hash: String,
-	/// The maximum number of items the `archive_storage` can return for a descendant query before
-	/// pagination is required.
-	storage_max_descendant_responses: usize,
-	/// The maximum number of queried items allowed for the `archive_storage` at a time.
-	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> {
@@ -96,17 +64,10 @@ impl<BE: Backend<Block>, Block: BlockT, Client> Archive<BE, Block, Client> {
 		client: Arc<Client>,
 		backend: Arc<BE>,
 		genesis_hash: GenesisHash,
-		config: ArchiveConfig,
+		executor: SubscriptionTaskExecutor,
 	) -> Self {
 		let genesis_hash = hex_string(&genesis_hash.as_ref());
-		Self {
-			client,
-			backend,
-			genesis_hash,
-			storage_max_descendant_responses: config.max_descendant_responses,
-			storage_max_queried_items: config.max_queried_items,
-			_phantom: PhantomData,
-		}
+		Self { client, backend, genesis_hash, _phantom: PhantomData, executor }
 	}
 }
 
@@ -270,11 +231,8 @@ where
 			.transpose()?
 			.map(ChildInfo::new_default_from_vec);
 
-		let storage_client = ArchiveStorage::new(
-			self.client.clone(),
-			self.storage_max_descendant_responses,
-			self.storage_max_queried_items,
-		);
+		let storage_client = ArchiveStorage::new(self.client.clone(), self.executor.clone());
+
 		Ok(storage_client.handle_query(hash, items, child_trie))
 	}
 }

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

@@ -21,6 +21,7 @@
 use std::sync::Arc;
 
 use sc_client_api::{Backend, ChildInfo, StorageKey, StorageProvider};
+use sc_rpc::SubscriptionTaskExecutor;
 use sp_runtime::traits::Block as BlockT;
 
 use crate::common::{
@@ -32,44 +33,31 @@ use crate::common::{
 pub struct ArchiveStorage<Client, Block, BE> {
 	/// Storage client.
 	client: Storage<Client, Block, BE>,
-	/// The maximum number of responses the API can return for a descendant query at a time.
-	storage_max_descendant_responses: usize,
-	/// The maximum number of queried items allowed for the `archive_storage` at a time.
-	storage_max_queried_items: usize,
 }
 
 impl<Client, Block, BE> ArchiveStorage<Client, Block, BE> {
 	/// Constructs a new [`ArchiveStorage`].
-	pub fn new(
-		client: Arc<Client>,
-		storage_max_descendant_responses: usize,
-		storage_max_queried_items: usize,
-	) -> Self {
-		Self {
-			client: Storage::new(client),
-			storage_max_descendant_responses,
-			storage_max_queried_items,
-		}
+	pub fn new(client: Arc<Client>, executor: SubscriptionTaskExecutor) -> Self {
+		Self { client: Storage::new(client, executor) }
 	}
 }
 
 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,
 {
 	/// Generate the response of the `archive_storage` method.
 	pub fn handle_query(
 		&self,
 		hash: Block::Hash,
-		mut items: Vec<PaginatedStorageQuery<StorageKey>>,
+		items: Vec<PaginatedStorageQuery<StorageKey>>,
 		child_key: Option<ChildInfo>,
 	) -> ArchiveStorageResult {
-		let discarded_items = items.len().saturating_sub(self.storage_max_queried_items);
-		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 +80,34 @@ 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,
+					});
 				},
 			};
 		}
 
-		ArchiveStorageResult::ok(storage_results, discarded_items)
+		if !query_iter.is_empty() {
+			let mut rx = self.client.query_iter_pagination(query_iter, hash, child_key);
+
+			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, 0)
 	}
 }

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

@@ -32,4 +32,4 @@ pub mod archive;
 pub mod error;
 
 pub use api::ArchiveApiServer;
-pub use archive::{Archive, ArchiveConfig};
+pub use archive::Archive;