RPC: query historical storage entries (#537)

* Query storage changes from-to block.

* Rewrite to use hashes instead of numbers. Add tests.

substrate/substrate/primitives/src/storage.rs

@@ -31,7 +31,7 @@ pub struct StorageKey(#[cfg_attr(feature = "std", serde(with="bytes"))] pub Vec<
 pub struct StorageData(#[cfg_attr(feature = "std", serde(with="bytes"))] pub Vec<u8>);
 
 /// Storage change set
-#[cfg_attr(feature = "std", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "std", derive(Serialize, Deserialize, Debug, PartialEq, Eq))]
 pub struct StorageChangeSet<Hash> {
 	/// Block hash
 	pub block: Hash,

substrate/substrate/rpc/src/state/error.rs

@@ -25,6 +25,11 @@ error_chain! {
 	}
 
 	errors {
+		/// Provided block range couldn't be resolved to a list of blocks.
+		InvalidBlockRange(from: String, to: String, details: String) {
+			description("Invalid block range"),
+			display("Cannot resolve a block range ['{:?}' ... '{:?}]. {}", from, to, details),
+		}
 		/// Not implemented yet
 		Unimplemented {
 			description("not implemented yet"),

substrate/substrate/rpc/src/state/mod.rs

@@ -16,7 +16,10 @@
 
 //! Polkadot state API.
 
-use std::sync::Arc;
+use std::{
+	collections::HashMap,
+	sync::Arc,
+};
 
 use client::{self, Client, CallExecutor, BlockchainEvents};
 use jsonrpc_macros::Trailing;
@@ -27,7 +30,7 @@ use primitives::storage::{StorageKey, StorageData, StorageChangeSet};
 use rpc::Result as RpcResult;
 use rpc::futures::{stream, Future, Sink, Stream};
 use runtime_primitives::generic::BlockId;
-use runtime_primitives::traits::Block as BlockT;
+use runtime_primitives::traits::{Block as BlockT, Header};
 use tokio::runtime::TaskExecutor;
 
 use subscriptions::Subscriptions;
@@ -59,6 +62,13 @@ build_rpc_trait! {
 		#[rpc(name = "state_getStorageSize", alias = ["state_getStorageSizeAt", ])]
 		fn storage_size(&self, StorageKey, Trailing<Hash>) -> Result<Option<u64>>;
 
+		/// Query historical storage entries (by key) starting from a block given as the second parameter.
+		///
+		/// NOTE This first returned result contains the initial state of storage for all keys.
+		/// Subsequent values in the vector represent changes to the previous state (diffs).
+		#[rpc(name = "state_queryStorage")]
+		fn query_storage(&self, Vec<StorageKey>, Hash, Trailing<Hash>) -> Result<Vec<StorageChangeSet<Hash>>>;
+
 		#[pubsub(name = "state_storage")] {
 			/// New storage subscription
 			#[rpc(name = "state_subscribeStorage")]
@@ -130,6 +140,74 @@ impl<B, E, Block> StateApi<Block::Hash> for State<B, E, Block> where
 		Ok(self.storage(key, block)?.map(|x| x.0.len() as u64))
 	}
 
+	fn query_storage(&self, keys: Vec<StorageKey>, from: Block::Hash, to: Trailing<Block::Hash>) -> Result<Vec<StorageChangeSet<Block::Hash>>> {
+		let to = self.unwrap_or_best(to)?;
+
+		let from_hdr = self.client.header(&BlockId::hash(from))?;
+		let to_hdr = self.client.header(&BlockId::hash(to))?;
+
+		match (from_hdr, to_hdr) {
+			(Some(ref from), Some(ref to)) if from.number() <= to.number() => {
+				let from = from.clone();
+				let to = to.clone();
+				// check if we can get from `to` to `from` by going through parent_hashes.
+				let blocks = {
+					let mut blocks = vec![to.hash()];
+					let mut last = to.clone();
+					while last.number() > from.number() {
+						if let Some(hdr) = self.client.header(&BlockId::hash(*last.parent_hash()))? {
+							blocks.push(hdr.hash());
+							last = hdr;
+						} else {
+							bail!(invalid_block_range(
+								Some(from),
+								Some(to),
+								format!("Parent of {} ({}) not found", last.number(), last.hash()),
+							))
+						}
+					}
+					if last.hash() != from.hash() {
+						bail!(invalid_block_range(
+							Some(from),
+							Some(to),
+							format!("Expected to reach `from`, got {} ({})", last.number(), last.hash()),
+						))
+					}
+					blocks.reverse();
+					blocks
+				};
+				let mut result = Vec::new();
+				let mut last_state: HashMap<_, Option<_>> = Default::default();
+				for block in blocks {
+					let mut changes = vec![];
+					let id = BlockId::hash(block.clone());
+
+					for key in &keys {
+						let (has_changed, data) = {
+							let curr_data = self.client.storage(&id, key)?;
+							let prev_data = last_state.get(key).and_then(|x| x.as_ref());
+
+							(curr_data.as_ref() != prev_data, curr_data)
+						};
+
+						if has_changed {
+							changes.push((key.clone(), data.clone()));
+						}
+
+						last_state.insert(key.clone(), data);
+					}
+
+					result.push(StorageChangeSet {
+						block,
+						changes,
+					});
+				}
+				Ok(result)
+			},
+			(from, to) => bail!(invalid_block_range(from, to, "Invalid range or unknown block".into())),
+		}
+	}
+
 	fn subscribe_storage(
 		&self,
 		_meta: Self::Metadata,
@@ -179,3 +257,12 @@ impl<B, E, Block> StateApi<Block::Hash> for State<B, E, Block> where
 		Ok(self.subscriptions.cancel(id))
 	}
 }
+
+fn invalid_block_range<H: Header>(from: Option<H>, to: Option<H>, reason: String) -> error::ErrorKind {
+	let to_string = |x: Option<H>| match x {
+		None => "unknown hash".into(),
+		Some(h) => format!("{} ({})", h.number(), h.hash()),
+	};
+
+	error::ErrorKind::InvalidBlockRange(to_string(from), to_string(to), reason)
+}

substrate/substrate/rpc/src/state/tests.rs

@@ -22,7 +22,6 @@ use jsonrpc_macros::pubsub;
 use rustc_hex::FromHex;
 use test_client::{self, runtime, keyring::Keyring, TestClient, BlockBuilderExt};
 
-
 #[test]
 fn should_return_storage() {
 	let core = ::tokio::runtime::Runtime::new().unwrap();
@@ -121,3 +120,67 @@ fn should_send_initial_storage_changes_and_notifications() {
 	// no more notifications on this channel
 	assert_eq!(core.block_on(next.into_future()).unwrap().0, None);
 }
+
+#[test]
+fn should_query_storage() {
+	let core = ::tokio::runtime::Runtime::new().unwrap();
+	let client = Arc::new(test_client::new());
+	let api = State::new(client.clone(), core.executor());
+
+	let add_block = |nonce| {
+		let mut builder = client.new_block().unwrap();
+		builder.push_transfer(runtime::Transfer {
+			from: Keyring::Alice.to_raw_public().into(),
+			to: Keyring::Ferdie.to_raw_public().into(),
+			amount: 42,
+			nonce,
+		}).unwrap();
+		let block = builder.bake().unwrap();
+		let hash = block.header.hash();
+		client.justify_and_import(BlockOrigin::Own, block).unwrap();
+		hash
+	};
+	let block1_hash = add_block(0);
+	let block2_hash = add_block(1);
+	let genesis_hash = client.genesis_hash();
+
+
+	let mut expected = vec![
+		StorageChangeSet {
+			block: genesis_hash,
+			changes: vec![
+				(StorageKey("a52da2b7c269da1366b3ed1cdb7299ce".from_hex().unwrap()), Some(StorageData(vec![232, 3, 0, 0, 0, 0, 0, 0]))),
+			],
+		},
+		StorageChangeSet {
+			block: block1_hash,
+			changes: vec![
+				(StorageKey("a52da2b7c269da1366b3ed1cdb7299ce".from_hex().unwrap()), Some(StorageData(vec![190, 3, 0, 0, 0, 0, 0, 0]))),
+			],
+		},
+	];
+
+	// Query changes only up to block1
+	let result = api.query_storage(
+		vec![StorageKey("a52da2b7c269da1366b3ed1cdb7299ce".from_hex().unwrap())],
+		genesis_hash,
+		Some(block1_hash).into(),
+	);
+
+	assert_eq!(result.unwrap(), expected);
+
+	// Query all changes
+	let result = api.query_storage(
+		vec![StorageKey("a52da2b7c269da1366b3ed1cdb7299ce".from_hex().unwrap())],
+		genesis_hash,
+		None.into(),
+	);
+
+	expected.push(StorageChangeSet {
+		block: block2_hash,
+		changes: vec![
+			(StorageKey("a52da2b7c269da1366b3ed1cdb7299ce".from_hex().unwrap()), Some(StorageData(vec![148, 3, 0, 0, 0, 0, 0, 0]))),
+		],
+	});
+	assert_eq!(result.unwrap(), expected);
+}