archive: Implement archive_unstable_storage (paritytech#1846)

This PR implements the `archive_unstable_storage` method that offers
support for:
- fetching values
- fetching hashes
- iterating over keys and values
- iterating over keys and hashes
- fetching merkle values from the trie-db

A common component dedicated to RPC-V2 storage queries is created to
bridge the gap between `chainHead/storage` and `archive/storage`.
Query pagination is supported by `paginationStartKey`, similar to the
old APIs.
Similarly to the `chainHead/storage`, the `archive/storage` method
accepts a maximum number of queried items.

The design builds upon:
paritytech/json-rpc-interface-spec#94.
Closes paritytech#1512.

cc @paritytech/subxt-team

---------

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>
Co-authored-by: Niklas Adolfsson <niklasadolfsson1@gmail.com>

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

@@ -18,7 +18,10 @@
 
 //! API trait of the archive methods.
 
-use crate::MethodResult;
+use crate::{
+	common::events::{ArchiveStorageResult, PaginatedStorageQuery},
+	MethodResult,
+};
 use jsonrpsee::{core::RpcResult, proc_macros::rpc};
 
 #[rpc(client, server)]
@@ -88,4 +91,17 @@ pub trait ArchiveApi<Hash> {
 		function: String,
 		call_parameters: String,
 	) -> RpcResult<MethodResult>;
+
+	/// Returns storage entries at a specific block's state.
+	///
+	/// # Unstable
+	///
+	/// This method is unstable and subject to change in the future.
+	#[method(name = "archive_unstable_storage", blocking)]
+	fn archive_unstable_storage(
+		&self,
+		hash: Hash,
+		items: Vec<PaginatedStorageQuery<String>>,
+		child_trie: Option<String>,
+	) -> RpcResult<ArchiveStorageResult>;
 }

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

@@ -20,14 +20,15 @@
 
 use crate::{
 	archive::{error::Error as ArchiveError, ArchiveApiServer},
-	chain_head::hex_string,
-	MethodResult,
+	common::events::{ArchiveStorageResult, PaginatedStorageQuery},
+	hex_string, MethodResult,
 };
 
 use codec::Encode;
 use jsonrpsee::core::{async_trait, RpcResult};
 use sc_client_api::{
-	Backend, BlockBackend, BlockchainEvents, CallExecutor, ExecutorProvider, StorageProvider,
+	Backend, BlockBackend, BlockchainEvents, CallExecutor, ChildInfo, ExecutorProvider, StorageKey,
+	StorageProvider,
 };
 use sp_api::{CallApiAt, CallContext};
 use sp_blockchain::{
@@ -40,6 +41,8 @@ use sp_runtime::{
 };
 use std::{collections::HashSet, marker::PhantomData, sync::Arc};
 
+use super::archive_storage::ArchiveStorage;
+
 /// An API for archive RPC calls.
 pub struct Archive<BE: Backend<Block>, Block: BlockT, Client> {
 	/// Substrate client.
@@ -48,8 +51,12 @@ 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 reported items by the `archive_storage` at a time.
+	storage_max_reported_items: 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, BE)>,
+	_phantom: PhantomData<Block>,
 }
 
 impl<BE: Backend<Block>, Block: BlockT, Client> Archive<BE, Block, Client> {
@@ -58,9 +65,18 @@ impl<BE: Backend<Block>, Block: BlockT, Client> Archive<BE, Block, Client> {
 		client: Arc<Client>,
 		backend: Arc<BE>,
 		genesis_hash: GenesisHash,
+		storage_max_reported_items: usize,
+		storage_max_queried_items: usize,
 	) -> Self {
 		let genesis_hash = hex_string(&genesis_hash.as_ref());
-		Self { client, backend, genesis_hash, _phantom: PhantomData }
+		Self {
+			client,
+			backend,
+			genesis_hash,
+			storage_max_reported_items,
+			storage_max_queried_items,
+			_phantom: PhantomData,
+		}
 	}
 }
 
@@ -185,4 +201,48 @@ where
 			Err(error) => MethodResult::err(error.to_string()),
 		})
 	}
+
+	fn archive_unstable_storage(
+		&self,
+		hash: Block::Hash,
+		items: Vec<PaginatedStorageQuery<String>>,
+		child_trie: Option<String>,
+	) -> RpcResult<ArchiveStorageResult> {
+		let items = items
+			.into_iter()
+			.map(|query| {
+				let key = StorageKey(parse_hex_param(query.key)?);
+				let pagination_start_key = query
+					.pagination_start_key
+					.map(|key| parse_hex_param(key).map(|key| StorageKey(key)))
+					.transpose()?;
+
+				// Paginated start key is only supported
+				if pagination_start_key.is_some() && !query.query_type.is_descendant_query() {
+					return Err(ArchiveError::InvalidParam(
+						"Pagination start key is only supported for descendants queries"
+							.to_string(),
+					))
+				}
+
+				Ok(PaginatedStorageQuery {
+					key,
+					query_type: query.query_type,
+					pagination_start_key,
+				})
+			})
+			.collect::<Result<Vec<_>, ArchiveError>>()?;
+
+		let child_trie = child_trie
+			.map(|child_trie| parse_hex_param(child_trie))
+			.transpose()?
+			.map(ChildInfo::new_default_from_vec);
+
+		let storage_client = ArchiveStorage::new(
+			self.client.clone(),
+			self.storage_max_reported_items,
+			self.storage_max_queried_items,
+		);
+		Ok(storage_client.handle_query(hash, items, child_trie))
+	}
 }

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

@@ -0,0 +1,125 @@
+// This file is part of Substrate.
+
+// Copyright (C) Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+//! Implementation of the `archive_storage` method.
+
+use std::sync::Arc;
+
+use sc_client_api::{Backend, ChildInfo, StorageKey, StorageProvider};
+use sp_runtime::traits::Block as BlockT;
+
+use crate::common::{
+	events::{ArchiveStorageResult, PaginatedStorageQuery, StorageQueryType},
+	storage::{IterQueryType, QueryIter, Storage},
+};
+
+/// Generates the events of the `chainHead_storage` method.
+pub struct ArchiveStorage<Client, Block, BE> {
+	/// Storage client.
+	client: Storage<Client, Block, BE>,
+	/// The maximum number of reported items by the `archive_storage` at a time.
+	storage_max_reported_items: 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_reported_items: usize,
+		storage_max_queried_items: usize,
+	) -> Self {
+		Self { client: Storage::new(client), storage_max_reported_items, storage_max_queried_items }
+	}
+}
+
+impl<Client, Block, BE> ArchiveStorage<Client, Block, BE>
+where
+	Block: BlockT + 'static,
+	BE: Backend<Block> + 'static,
+	Client: StorageProvider<Block, BE> + 'static,
+{
+	/// Generate the response of the `archive_storage` method.
+	pub fn handle_query(
+		&self,
+		hash: Block::Hash,
+		mut 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());
+		for item in items {
+			match item.query_type {
+				StorageQueryType::Value => {
+					match self.client.query_value(hash, &item.key, child_key.as_ref()) {
+						Ok(Some(value)) => storage_results.push(value),
+						Ok(None) => continue,
+						Err(error) => return ArchiveStorageResult::err(error),
+					}
+				},
+				StorageQueryType::Hash =>
+					match self.client.query_hash(hash, &item.key, child_key.as_ref()) {
+						Ok(Some(value)) => storage_results.push(value),
+						Ok(None) => continue,
+						Err(error) => return ArchiveStorageResult::err(error),
+					},
+				StorageQueryType::ClosestDescendantMerkleValue =>
+					match self.client.query_merkle_value(hash, &item.key, child_key.as_ref()) {
+						Ok(Some(value)) => storage_results.push(value),
+						Ok(None) => continue,
+						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_reported_items,
+					) {
+						Ok((results, _)) => storage_results.extend(results),
+						Err(error) => return ArchiveStorageResult::err(error),
+					}
+				},
+				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_reported_items,
+					) {
+						Ok((results, _)) => storage_results.extend(results),
+						Err(error) => return ArchiveStorageResult::err(error),
+					}
+				},
+			};
+		}
+
+		ArchiveStorageResult::ok(storage_results, discarded_items)
+	}
+}

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

@@ -25,6 +25,8 @@
 #[cfg(test)]
 mod tests;
 
+mod archive_storage;
+
 pub mod api;
 pub mod archive;
 pub mod error;