Rework storage iterators (#13284)

* Rework storage iterators

* Make sure storage iteration is also accounted for when benchmarking

* Use `trie-db` from crates.io

* Appease clippy

* Bump `trie-bench` to 0.35.0

* Fix tests' compilation

* Update comment to clarify how `IterArgs::start_at` works

* Add extra tests

* Fix iterators on `Client` so that they behave as before

* Add extra `unwrap`s in tests

* More clippy fixes

* Come on clippy, give me a break already

* Rename `allow_missing` to `stop_on_incomplete_database`

* Add `#[inline]` to `with_recorder_and_cache`

* Use `with_recorder_and_cache` in `with_trie_db`; add doc comment

* Simplify code: use `with_trie_db` in `next_storage_key_from_root`

* Remove `expect`s in the benchmarking CLI

* Add extra doc comments

* Move `RawIter` before `TrieBackendEssence` (no code changes; just cut-paste)

* Remove a TODO in tests

* Update comment for `StorageIterator::was_complete`

* Update `trie-db` to 0.25.1

substrate/client/api/src/backend.rs

@@ -31,14 +31,13 @@ use sp_runtime::{
 	Justification, Justifications, StateVersion, Storage,
 };
 use sp_state_machine::{
-	backend::AsTrieBackend, ChildStorageCollection, IndexOperation, OffchainChangesCollection,
-	StorageCollection,
+	backend::AsTrieBackend, ChildStorageCollection, IndexOperation, IterArgs,
+	OffchainChangesCollection, StorageCollection, StorageIterator,
 };
 use sp_storage::{ChildInfo, StorageData, StorageKey};
 use std::collections::{HashMap, HashSet};
 
 pub use sp_state_machine::{Backend as StateBackend, KeyValueStates};
-use std::marker::PhantomData;
 
 /// Extracts the state backend type for the given backend.
 pub type StateBackendFor<B, Block> = <B as Backend<Block>>::State;
@@ -303,32 +302,61 @@ pub trait AuxStore {
 }
 
 /// An `Iterator` that iterates keys in a given block under a prefix.
-pub struct KeyIterator<State, Block> {
+pub struct KeysIter<State, Block>
+where
+	State: StateBackend<HashFor<Block>>,
+	Block: BlockT,
+{
+	inner: <State as StateBackend<HashFor<Block>>>::RawIter,
 	state: State,
-	child_storage: Option<ChildInfo>,
-	prefix: Option<StorageKey>,
-	current_key: Vec<u8>,
-	_phantom: PhantomData<Block>,
+	skip_if_first: Option<StorageKey>,
 }
 
-impl<State, Block> KeyIterator<State, Block> {
-	/// create a KeyIterator instance
-	pub fn new(state: State, prefix: Option<StorageKey>, current_key: Vec<u8>) -> Self {
-		Self { state, child_storage: None, prefix, current_key, _phantom: PhantomData }
+impl<State, Block> KeysIter<State, Block>
+where
+	State: StateBackend<HashFor<Block>>,
+	Block: BlockT,
+{
+	/// Create a new iterator over storage keys.
+	pub fn new(
+		state: State,
+		prefix: Option<&StorageKey>,
+		start_at: Option<&StorageKey>,
+	) -> Result<Self, State::Error> {
+		let mut args = IterArgs::default();
+		args.prefix = prefix.as_ref().map(|prefix| prefix.0.as_slice());
+		args.start_at = start_at.as_ref().map(|start_at| start_at.0.as_slice());
+
+		let start_at = args.start_at;
+		Ok(Self {
+			inner: state.raw_iter(args)?,
+			state,
+			skip_if_first: start_at.map(|key| StorageKey(key.to_vec())),
+		})
 	}
 
-	/// Create a `KeyIterator` instance for a child storage.
+	/// Create a new iterator over a child storage's keys.
 	pub fn new_child(
 		state: State,
 		child_info: ChildInfo,
-		prefix: Option<StorageKey>,
-		current_key: Vec<u8>,
-	) -> Self {
-		Self { state, child_storage: Some(child_info), prefix, current_key, _phantom: PhantomData }
+		prefix: Option<&StorageKey>,
+		start_at: Option<&StorageKey>,
+	) -> Result<Self, State::Error> {
+		let mut args = IterArgs::default();
+		args.prefix = prefix.as_ref().map(|prefix| prefix.0.as_slice());
+		args.start_at = start_at.as_ref().map(|start_at| start_at.0.as_slice());
+		args.child_info = Some(child_info);
+
+		let start_at = args.start_at;
+		Ok(Self {
+			inner: state.raw_iter(args)?,
+			state,
+			skip_if_first: start_at.map(|key| StorageKey(key.to_vec())),
+		})
 	}
 }
 
-impl<State, Block> Iterator for KeyIterator<State, Block>
+impl<State, Block> Iterator for KeysIter<State, Block>
 where
 	Block: BlockT,
 	State: StateBackend<HashFor<Block>>,
@@ -336,25 +364,78 @@ where
 	type Item = StorageKey;
 
 	fn next(&mut self) -> Option<Self::Item> {
-		let next_key = if let Some(child_info) = self.child_storage.as_ref() {
-			self.state.next_child_storage_key(child_info, &self.current_key)
-		} else {
-			self.state.next_storage_key(&self.current_key)
-		}
-		.ok()
-		.flatten()?;
-		// this terminates the iterator the first time it fails.
-		if let Some(ref prefix) = self.prefix {
-			if !next_key.starts_with(&prefix.0[..]) {
-				return None
+		let key = self.inner.next_key(&self.state)?.ok().map(StorageKey)?;
+
+		if let Some(skipped_key) = self.skip_if_first.take() {
+			if key == skipped_key {
+				return self.next()
 			}
 		}
-		self.current_key = next_key.clone();
-		Some(StorageKey(next_key))
+
+		Some(key)
 	}
 }
 
-/// Provides acess to storage primitives
+/// An `Iterator` that iterates keys and values in a given block under a prefix.
+pub struct PairsIter<State, Block>
+where
+	State: StateBackend<HashFor<Block>>,
+	Block: BlockT,
+{
+	inner: <State as StateBackend<HashFor<Block>>>::RawIter,
+	state: State,
+	skip_if_first: Option<StorageKey>,
+}
+
+impl<State, Block> Iterator for PairsIter<State, Block>
+where
+	Block: BlockT,
+	State: StateBackend<HashFor<Block>>,
+{
+	type Item = (StorageKey, StorageData);
+
+	fn next(&mut self) -> Option<Self::Item> {
+		let (key, value) = self
+			.inner
+			.next_pair(&self.state)?
+			.ok()
+			.map(|(key, value)| (StorageKey(key), StorageData(value)))?;
+
+		if let Some(skipped_key) = self.skip_if_first.take() {
+			if key == skipped_key {
+				return self.next()
+			}
+		}
+
+		Some((key, value))
+	}
+}
+
+impl<State, Block> PairsIter<State, Block>
+where
+	State: StateBackend<HashFor<Block>>,
+	Block: BlockT,
+{
+	/// Create a new iterator over storage key and value pairs.
+	pub fn new(
+		state: State,
+		prefix: Option<&StorageKey>,
+		start_at: Option<&StorageKey>,
+	) -> Result<Self, State::Error> {
+		let mut args = IterArgs::default();
+		args.prefix = prefix.as_ref().map(|prefix| prefix.0.as_slice());
+		args.start_at = start_at.as_ref().map(|start_at| start_at.0.as_slice());
+
+		let start_at = args.start_at;
+		Ok(Self {
+			inner: state.raw_iter(args)?,
+			state,
+			skip_if_first: start_at.map(|key| StorageKey(key.to_vec())),
+		})
+	}
+}
+
+/// Provides access to storage primitives
 pub trait StorageProvider<Block: BlockT, B: Backend<Block>> {
 	/// Given a block's `Hash` and a key, return the value under the key in that block.
 	fn storage(
@@ -363,13 +444,6 @@ pub trait StorageProvider<Block: BlockT, B: Backend<Block>> {
 		key: &StorageKey,
 	) -> sp_blockchain::Result<Option<StorageData>>;
 
-	/// Given a block's `Hash` and a key prefix, return the matching storage keys in that block.
-	fn storage_keys(
-		&self,
-		hash: Block::Hash,
-		key_prefix: &StorageKey,
-	) -> sp_blockchain::Result<Vec<StorageKey>>;
-
 	/// Given a block's `Hash` and a key, return the value under the hash in that block.
 	fn storage_hash(
 		&self,
@@ -377,22 +451,23 @@ pub trait StorageProvider<Block: BlockT, B: Backend<Block>> {
 		key: &StorageKey,
 	) -> sp_blockchain::Result<Option<Block::Hash>>;
 
-	/// Given a block's `Hash` and a key prefix, return the matching child storage keys and values
-	/// in that block.
-	fn storage_pairs(
-		&self,
-		hash: Block::Hash,
-		key_prefix: &StorageKey,
-	) -> sp_blockchain::Result<Vec<(StorageKey, StorageData)>>;
-
-	/// Given a block's `Hash` and a key prefix, return a `KeyIterator` iterates matching storage
+	/// Given a block's `Hash` and a key prefix, returns a `KeysIter` iterates matching storage
 	/// keys in that block.
-	fn storage_keys_iter(
+	fn storage_keys(
 		&self,
 		hash: Block::Hash,
 		prefix: Option<&StorageKey>,
 		start_key: Option<&StorageKey>,
-	) -> sp_blockchain::Result<KeyIterator<B::State, Block>>;
+	) -> sp_blockchain::Result<KeysIter<B::State, Block>>;
+
+	/// Given a block's `Hash` and a key prefix, returns an iterator over the storage keys and
+	/// values in that block.
+	fn storage_pairs(
+		&self,
+		hash: <Block as BlockT>::Hash,
+		prefix: Option<&StorageKey>,
+		start_key: Option<&StorageKey>,
+	) -> sp_blockchain::Result<PairsIter<B::State, Block>>;
 
 	/// Given a block's `Hash`, a key and a child storage key, return the value under the key in
 	/// that block.
@@ -403,24 +478,15 @@ pub trait StorageProvider<Block: BlockT, B: Backend<Block>> {
 		key: &StorageKey,
 	) -> sp_blockchain::Result<Option<StorageData>>;
 
-	/// Given a block's `Hash`, a key prefix, and a child storage key, return the matching child
-	/// storage keys.
-	fn child_storage_keys(
-		&self,
-		hash: Block::Hash,
-		child_info: &ChildInfo,
-		key_prefix: &StorageKey,
-	) -> sp_blockchain::Result<Vec<StorageKey>>;
-
 	/// Given a block's `Hash` and a key `prefix` and a child storage key,
-	/// return a `KeyIterator` that iterates matching storage keys in that block.
-	fn child_storage_keys_iter(
+	/// returns a `KeysIter` that iterates matching storage keys in that block.
+	fn child_storage_keys(
 		&self,
 		hash: Block::Hash,
 		child_info: ChildInfo,
 		prefix: Option<&StorageKey>,
 		start_key: Option<&StorageKey>,
-	) -> sp_blockchain::Result<KeyIterator<B::State, Block>>;
+	) -> sp_blockchain::Result<KeysIter<B::State, Block>>;
 
 	/// Given a block's `Hash`, a key and a child storage key, return the hash under the key in that
 	/// block.

substrate/client/db/Cargo.toml

@@ -44,6 +44,7 @@ quickcheck = { version = "1.0.3", default-features = false }
 kitchensink-runtime = { path = "../../bin/node/runtime" }
 sp-tracing = { version = "6.0.0", path = "../../primitives/tracing" }
 substrate-test-runtime-client = { version = "2.0.0", path = "../../test-utils/runtime/client" }
+array-bytes = "4.1"
 
 [features]
 default = []

substrate/client/db/src/bench.rs

@@ -22,6 +22,7 @@ use crate::{DbState, DbStateBuilder};
 use hash_db::{Hasher, Prefix};
 use kvdb::{DBTransaction, KeyValueDB};
 use linked_hash_map::LinkedHashMap;
+use parking_lot::Mutex;
 use sp_core::{
 	hexdisplay::HexDisplay,
 	storage::{ChildInfo, TrackedStorageKey},
@@ -31,7 +32,8 @@ use sp_runtime::{
 	StateVersion, Storage,
 };
 use sp_state_machine::{
-	backend::Backend as StateBackend, ChildStorageCollection, DBValue, StorageCollection,
+	backend::Backend as StateBackend, ChildStorageCollection, DBValue, IterArgs, StorageCollection,
+	StorageIterator, StorageKey, StorageValue,
 };
 use sp_trie::{
 	cache::{CacheSize, SharedTrieCache},
@@ -59,6 +61,19 @@ impl<Block: BlockT> sp_state_machine::Storage<HashFor<Block>> for StorageDb<Bloc
 	}
 }
 
+struct KeyTracker {
+	enable_tracking: bool,
+	/// Key tracker for keys in the main trie.
+	/// We track the total number of reads and writes to these keys,
+	/// not de-duplicated for repeats.
+	main_keys: LinkedHashMap<Vec<u8>, TrackedStorageKey>,
+	/// Key tracker for keys in a child trie.
+	/// Child trie are identified by their storage key (i.e. `ChildInfo::storage_key()`)
+	/// We track the total number of reads and writes to these keys,
+	/// not de-duplicated for repeats.
+	child_keys: LinkedHashMap<Vec<u8>, LinkedHashMap<Vec<u8>, TrackedStorageKey>>,
+}
+
 /// State that manages the backend database reference. Allows runtime to control the database.
 pub struct BenchmarkingState<B: BlockT> {
 	root: Cell<B::Hash>,
@@ -67,22 +82,52 @@ pub struct BenchmarkingState<B: BlockT> {
 	db: Cell<Option<Arc<dyn KeyValueDB>>>,
 	genesis: HashMap<Vec<u8>, (Vec<u8>, i32)>,
 	record: Cell<Vec<Vec<u8>>>,
-	/// Key tracker for keys in the main trie.
-	/// We track the total number of reads and writes to these keys,
-	/// not de-duplicated for repeats.
-	main_key_tracker: RefCell<LinkedHashMap<Vec<u8>, TrackedStorageKey>>,
-	/// Key tracker for keys in a child trie.
-	/// Child trie are identified by their storage key (i.e. `ChildInfo::storage_key()`)
-	/// We track the total number of reads and writes to these keys,
-	/// not de-duplicated for repeats.
-	child_key_tracker: RefCell<LinkedHashMap<Vec<u8>, LinkedHashMap<Vec<u8>, TrackedStorageKey>>>,
+	key_tracker: Arc<Mutex<KeyTracker>>,
 	whitelist: RefCell<Vec<TrackedStorageKey>>,
 	proof_recorder: Option<sp_trie::recorder::Recorder<HashFor<B>>>,
 	proof_recorder_root: Cell<B::Hash>,
-	enable_tracking: bool,
 	shared_trie_cache: SharedTrieCache<HashFor<B>>,
 }
 
+/// A raw iterator over the `BenchmarkingState`.
+pub struct RawIter<B: BlockT> {
+	inner: <DbState<B> as StateBackend<HashFor<B>>>::RawIter,
+	child_trie: Option<Vec<u8>>,
+	key_tracker: Arc<Mutex<KeyTracker>>,
+}
+
+impl<B: BlockT> StorageIterator<HashFor<B>> for RawIter<B> {
+	type Backend = BenchmarkingState<B>;
+	type Error = String;
+
+	fn next_key(&mut self, backend: &Self::Backend) -> Option<Result<StorageKey, Self::Error>> {
+		match self.inner.next_key(backend.state.borrow().as_ref()?) {
+			Some(Ok(key)) => {
+				self.key_tracker.lock().add_read_key(self.child_trie.as_deref(), &key);
+				Some(Ok(key))
+			},
+			result => result,
+		}
+	}
+
+	fn next_pair(
+		&mut self,
+		backend: &Self::Backend,
+	) -> Option<Result<(StorageKey, StorageValue), Self::Error>> {
+		match self.inner.next_pair(backend.state.borrow().as_ref()?) {
+			Some(Ok((key, value))) => {
+				self.key_tracker.lock().add_read_key(self.child_trie.as_deref(), &key);
+				Some(Ok((key, value)))
+			},
+			result => result,
+		}
+	}
+
+	fn was_complete(&self) -> bool {
+		self.inner.was_complete()
+	}
+}
+
 impl<B: BlockT> BenchmarkingState<B> {
 	/// Create a new instance that creates a database in a temporary dir.
 	pub fn new(
@@ -103,12 +148,14 @@ impl<B: BlockT> BenchmarkingState<B> {
 			genesis: Default::default(),
 			genesis_root: Default::default(),
 			record: Default::default(),
-			main_key_tracker: Default::default(),
-			child_key_tracker: Default::default(),
+			key_tracker: Arc::new(Mutex::new(KeyTracker {
+				main_keys: Default::default(),
+				child_keys: Default::default(),
+				enable_tracking,
+			})),
 			whitelist: Default::default(),
 			proof_recorder: record_proof.then(Default::default),
 			proof_recorder_root: Cell::new(root),
-			enable_tracking,
 			// Enable the cache, but do not sync anything to the shared state.
 			shared_trie_cache: SharedTrieCache::new(CacheSize::new(0)),
 		};
@@ -123,7 +170,7 @@ impl<B: BlockT> BenchmarkingState<B> {
 			)
 		});
 		let (root, transaction): (B::Hash, _) =
-			state.state.borrow_mut().as_mut().unwrap().full_storage_root(
+			state.state.borrow().as_ref().unwrap().full_storage_root(
 				genesis.top.iter().map(|(k, v)| (k.as_ref(), Some(v.as_ref()))),
 				child_delta,
 				state_version,
@@ -157,36 +204,51 @@ impl<B: BlockT> BenchmarkingState<B> {
 	}
 
 	fn add_whitelist_to_tracker(&self) {
-		let mut main_key_tracker = self.main_key_tracker.borrow_mut();
-
-		let whitelist = self.whitelist.borrow();
-
-		whitelist.iter().for_each(|key| {
-			let mut whitelisted = TrackedStorageKey::new(key.key.clone());
-			whitelisted.whitelist();
-			main_key_tracker.insert(key.key.clone(), whitelisted);
-		});
+		self.key_tracker.lock().add_whitelist(&self.whitelist.borrow());
 	}
 
 	fn wipe_tracker(&self) {
-		*self.main_key_tracker.borrow_mut() = LinkedHashMap::new();
-		*self.child_key_tracker.borrow_mut() = LinkedHashMap::new();
-		self.add_whitelist_to_tracker();
+		let mut key_tracker = self.key_tracker.lock();
+		key_tracker.main_keys = LinkedHashMap::new();
+		key_tracker.child_keys = LinkedHashMap::new();
+		key_tracker.add_whitelist(&self.whitelist.borrow());
+	}
+
+	fn add_read_key(&self, childtrie: Option<&[u8]>, key: &[u8]) {
+		self.key_tracker.lock().add_read_key(childtrie, key);
+	}
+
+	fn add_write_key(&self, childtrie: Option<&[u8]>, key: &[u8]) {
+		self.key_tracker.lock().add_write_key(childtrie, key);
+	}
+
+	fn all_trackers(&self) -> Vec<TrackedStorageKey> {
+		self.key_tracker.lock().all_trackers()
+	}
+}
+
+impl KeyTracker {
+	fn add_whitelist(&mut self, whitelist: &[TrackedStorageKey]) {
+		whitelist.iter().for_each(|key| {
+			let mut whitelisted = TrackedStorageKey::new(key.key.clone());
+			whitelisted.whitelist();
+			self.main_keys.insert(key.key.clone(), whitelisted);
+		});
 	}
 
 	// Childtrie is identified by its storage key (i.e. `ChildInfo::storage_key`)
-	fn add_read_key(&self, childtrie: Option<&[u8]>, key: &[u8]) {
+	fn add_read_key(&mut self, childtrie: Option<&[u8]>, key: &[u8]) {
 		if !self.enable_tracking {
 			return
 		}
 
-		let mut child_key_tracker = self.child_key_tracker.borrow_mut();
-		let mut main_key_tracker = self.main_key_tracker.borrow_mut();
+		let child_key_tracker = &mut self.child_keys;
+		let main_key_tracker = &mut self.main_keys;
 
 		let key_tracker = if let Some(childtrie) = childtrie {
 			child_key_tracker.entry(childtrie.to_vec()).or_insert_with(LinkedHashMap::new)
 		} else {
-			&mut main_key_tracker
+			main_key_tracker
 		};
 
 		let should_log = match key_tracker.get_mut(key) {
@@ -216,18 +278,18 @@ impl<B: BlockT> BenchmarkingState<B> {
 	}
 
 	// Childtrie is identified by its storage key (i.e. `ChildInfo::storage_key`)
-	fn add_write_key(&self, childtrie: Option<&[u8]>, key: &[u8]) {
+	fn add_write_key(&mut self, childtrie: Option<&[u8]>, key: &[u8]) {
 		if !self.enable_tracking {
 			return
 		}
 
-		let mut child_key_tracker = self.child_key_tracker.borrow_mut();
-		let mut main_key_tracker = self.main_key_tracker.borrow_mut();
+		let child_key_tracker = &mut self.child_keys;
+		let main_key_tracker = &mut self.main_keys;
 
 		let key_tracker = if let Some(childtrie) = childtrie {
 			child_key_tracker.entry(childtrie.to_vec()).or_insert_with(LinkedHashMap::new)
 		} else {
-			&mut main_key_tracker
+			main_key_tracker
 		};
 
 		// If we have written to the key, we also consider that we have read from it.
@@ -261,11 +323,11 @@ impl<B: BlockT> BenchmarkingState<B> {
 	fn all_trackers(&self) -> Vec<TrackedStorageKey> {
 		let mut all_trackers = Vec::new();
 
-		self.main_key_tracker.borrow().iter().for_each(|(_, tracker)| {
+		self.main_keys.iter().for_each(|(_, tracker)| {
 			all_trackers.push(tracker.clone());
 		});
 
-		self.child_key_tracker.borrow().iter().for_each(|(_, child_tracker)| {
+		self.child_keys.iter().for_each(|(_, child_tracker)| {
 			child_tracker.iter().for_each(|(_, tracker)| {
 				all_trackers.push(tracker.clone());
 			});
@@ -283,6 +345,7 @@ impl<B: BlockT> StateBackend<HashFor<B>> for BenchmarkingState<B> {
 	type Error = <DbState<B> as StateBackend<HashFor<B>>>::Error;
 	type Transaction = <DbState<B> as StateBackend<HashFor<B>>>::Transaction;
 	type TrieBackendStorage = <DbState<B> as StateBackend<HashFor<B>>>::TrieBackendStorage;
+	type RawIter = RawIter<B>;
 
 	fn storage(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {
 		self.add_read_key(None, key);
@@ -356,58 +419,6 @@ impl<B: BlockT> StateBackend<HashFor<B>> for BenchmarkingState<B> {
 			.next_child_storage_key(child_info, key)
 	}
 
-	fn for_keys_with_prefix<F: FnMut(&[u8])>(&self, prefix: &[u8], f: F) {
-		if let Some(ref state) = *self.state.borrow() {
-			state.for_keys_with_prefix(prefix, f)
-		}
-	}
-
-	fn for_key_values_with_prefix<F: FnMut(&[u8], &[u8])>(&self, prefix: &[u8], f: F) {
-		if let Some(ref state) = *self.state.borrow() {
-			state.for_key_values_with_prefix(prefix, f)
-		}
-	}
-
-	fn apply_to_key_values_while<F: FnMut(Vec<u8>, Vec<u8>) -> bool>(
-		&self,
-		child_info: Option<&ChildInfo>,
-		prefix: Option<&[u8]>,
-		start_at: Option<&[u8]>,
-		f: F,
-		allow_missing: bool,
-	) -> Result<bool, Self::Error> {
-		self.state.borrow().as_ref().ok_or_else(state_err)?.apply_to_key_values_while(
-			child_info,
-			prefix,
-			start_at,
-			f,
-			allow_missing,
-		)
-	}
-
-	fn apply_to_keys_while<F: FnMut(&[u8]) -> bool>(
-		&self,
-		child_info: Option<&ChildInfo>,
-		prefix: Option<&[u8]>,
-		start_at: Option<&[u8]>,
-		f: F,
-	) {
-		if let Some(ref state) = *self.state.borrow() {
-			state.apply_to_keys_while(child_info, prefix, start_at, f)
-		}
-	}
-
-	fn for_child_keys_with_prefix<F: FnMut(&[u8])>(
-		&self,
-		child_info: &ChildInfo,
-		prefix: &[u8],
-		f: F,
-	) {
-		if let Some(ref state) = *self.state.borrow() {
-			state.for_child_keys_with_prefix(child_info, prefix, f)
-		}
-	}
-
 	fn storage_root<'a>(
 		&self,
 		delta: impl Iterator<Item = (&'a [u8], Option<&'a [u8]>)>,
@@ -437,19 +448,19 @@ impl<B: BlockT> StateBackend<HashFor<B>> for BenchmarkingState<B> {
 			.map_or(Default::default(), |s| s.child_storage_root(child_info, delta, state_version))
 	}
 
-	fn pairs(&self) -> Vec<(Vec<u8>, Vec<u8>)> {
-		self.state.borrow().as_ref().map_or(Default::default(), |s| s.pairs())
-	}
-
-	fn keys(&self, prefix: &[u8]) -> Vec<Vec<u8>> {
-		self.state.borrow().as_ref().map_or(Default::default(), |s| s.keys(prefix))
-	}
-
-	fn child_keys(&self, child_info: &ChildInfo, prefix: &[u8]) -> Vec<Vec<u8>> {
+	fn raw_iter(&self, args: IterArgs) -> Result<Self::RawIter, Self::Error> {
+		let child_trie =
+			args.child_info.as_ref().map(|child_info| child_info.storage_key().to_vec());
 		self.state
 			.borrow()
 			.as_ref()
-			.map_or(Default::default(), |s| s.child_keys(child_info, prefix))
+			.map(|s| s.raw_iter(args))
+			.unwrap_or(Ok(Default::default()))
+			.map(|raw_iter| RawIter {
+				inner: raw_iter,
+				key_tracker: self.key_tracker.clone(),
+				child_trie,
+			})
 	}
 
 	fn commit(
@@ -587,7 +598,7 @@ impl<B: BlockT> StateBackend<HashFor<B>> for BenchmarkingState<B> {
 	}
 
 	fn register_overlay_stats(&self, stats: &sp_state_machine::StateMachineStats) {
-		self.state.borrow_mut().as_mut().map(|s| s.register_overlay_stats(stats));
+		self.state.borrow().as_ref().map(|s| s.register_overlay_stats(stats));
 	}
 
 	fn usage_info(&self) -> sp_state_machine::UsageInfo {
@@ -639,6 +650,29 @@ mod test {
 	use crate::bench::BenchmarkingState;
 	use sp_state_machine::backend::Backend as _;
 
+	fn hex(hex: &str) -> Vec<u8> {
+		array_bytes::hex2bytes(hex).unwrap()
+	}
+
+	#[test]
+	fn iteration_is_also_counted_in_rw_counts() {
+		let storage = sp_runtime::Storage {
+			top: vec![(
+				hex("ce6e1397e668c7fcf47744350dc59688455a2c2dbd2e2a649df4e55d93cd7158"),
+				hex("0102030405060708"),
+			)]
+			.into_iter()
+			.collect(),
+			..sp_runtime::Storage::default()
+		};
+		let bench_state =
+			BenchmarkingState::<crate::tests::Block>::new(storage, None, false, true).unwrap();
+
+		assert_eq!(bench_state.read_write_count(), (0, 0, 0, 0));
+		assert_eq!(bench_state.keys(Default::default()).unwrap().count(), 1);
+		assert_eq!(bench_state.read_write_count(), (1, 0, 0, 0));
+	}
+
 	#[test]
 	fn read_to_main_and_child_tries() {
 		let bench_state =

substrate/client/db/src/lib.rs

@@ -86,8 +86,9 @@ use sp_runtime::{
 };
 use sp_state_machine::{
 	backend::{AsTrieBackend, Backend as StateBackend},
-	ChildStorageCollection, DBValue, IndexOperation, OffchainChangesCollection, StateMachineStats,
-	StorageCollection, UsageInfo as StateUsageInfo,
+	ChildStorageCollection, DBValue, IndexOperation, IterArgs, OffchainChangesCollection,
+	StateMachineStats, StorageCollection, StorageIterator, StorageKey, StorageValue,
+	UsageInfo as StateUsageInfo,
 };
 use sp_trie::{cache::SharedTrieCache, prefixed_key, MemoryDB, PrefixedMemoryDB};
 
@@ -159,10 +160,36 @@ impl<Block: BlockT> std::fmt::Debug for RefTrackingState<Block> {
 	}
 }
 
+/// A raw iterator over the `RefTrackingState`.
+pub struct RawIter<B: BlockT> {
+	inner: <DbState<B> as StateBackend<HashFor<B>>>::RawIter,
+}
+
+impl<B: BlockT> StorageIterator<HashFor<B>> for RawIter<B> {
+	type Backend = RefTrackingState<B>;
+	type Error = <DbState<B> as StateBackend<HashFor<B>>>::Error;
+
+	fn next_key(&mut self, backend: &Self::Backend) -> Option<Result<StorageKey, Self::Error>> {
+		self.inner.next_key(&backend.state)
+	}
+
+	fn next_pair(
+		&mut self,
+		backend: &Self::Backend,
+	) -> Option<Result<(StorageKey, StorageValue), Self::Error>> {
+		self.inner.next_pair(&backend.state)
+	}
+
+	fn was_complete(&self) -> bool {
+		self.inner.was_complete()
+	}
+}
+
 impl<B: BlockT> StateBackend<HashFor<B>> for RefTrackingState<B> {
 	type Error = <DbState<B> as StateBackend<HashFor<B>>>::Error;
 	type Transaction = <DbState<B> as StateBackend<HashFor<B>>>::Transaction;
 	type TrieBackendStorage = <DbState<B> as StateBackend<HashFor<B>>>::TrieBackendStorage;
+	type RawIter = RawIter<B>;
 
 	fn storage(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {
 		self.state.storage(key)
@@ -212,45 +239,6 @@ impl<B: BlockT> StateBackend<HashFor<B>> for RefTrackingState<B> {
 		self.state.next_child_storage_key(child_info, key)
 	}
 
-	fn for_keys_with_prefix<F: FnMut(&[u8])>(&self, prefix: &[u8], f: F) {
-		self.state.for_keys_with_prefix(prefix, f)
-	}
-
-	fn for_key_values_with_prefix<F: FnMut(&[u8], &[u8])>(&self, prefix: &[u8], f: F) {
-		self.state.for_key_values_with_prefix(prefix, f)
-	}
-
-	fn apply_to_key_values_while<F: FnMut(Vec<u8>, Vec<u8>) -> bool>(
-		&self,
-		child_info: Option<&ChildInfo>,
-		prefix: Option<&[u8]>,
-		start_at: Option<&[u8]>,
-		f: F,
-		allow_missing: bool,
-	) -> Result<bool, Self::Error> {
-		self.state
-			.apply_to_key_values_while(child_info, prefix, start_at, f, allow_missing)
-	}
-
-	fn apply_to_keys_while<F: FnMut(&[u8]) -> bool>(
-		&self,
-		child_info: Option<&ChildInfo>,
-		prefix: Option<&[u8]>,
-		start_at: Option<&[u8]>,
-		f: F,
-	) {
-		self.state.apply_to_keys_while(child_info, prefix, start_at, f)
-	}
-
-	fn for_child_keys_with_prefix<F: FnMut(&[u8])>(
-		&self,
-		child_info: &ChildInfo,
-		prefix: &[u8],
-		f: F,
-	) {
-		self.state.for_child_keys_with_prefix(child_info, prefix, f)
-	}
-
 	fn storage_root<'a>(
 		&self,
 		delta: impl Iterator<Item = (&'a [u8], Option<&'a [u8]>)>,
@@ -274,16 +262,8 @@ impl<B: BlockT> StateBackend<HashFor<B>> for RefTrackingState<B> {
 		self.state.child_storage_root(child_info, delta, state_version)
 	}
 
-	fn pairs(&self) -> Vec<(Vec<u8>, Vec<u8>)> {
-		self.state.pairs()
-	}
-
-	fn keys(&self, prefix: &[u8]) -> Vec<Vec<u8>> {
-		self.state.keys(prefix)
-	}
-
-	fn child_keys(&self, child_info: &ChildInfo, prefix: &[u8]) -> Vec<Vec<u8>> {
-		self.state.child_keys(child_info, prefix)
+	fn raw_iter(&self, args: IterArgs) -> Result<Self::RawIter, Self::Error> {
+		self.state.raw_iter(args).map(|inner| RawIter { inner })
 	}
 
 	fn register_overlay_stats(&self, stats: &StateMachineStats) {

substrate/client/db/src/record_stats_state.rs

@@ -26,7 +26,7 @@ use sp_runtime::{
 };
 use sp_state_machine::{
 	backend::{AsTrieBackend, Backend as StateBackend},
-	TrieBackend,
+	IterArgs, StorageIterator, StorageKey, StorageValue, TrieBackend,
 };
 use std::sync::Arc;
 
@@ -73,10 +73,43 @@ impl<S: StateBackend<HashFor<B>>, B: BlockT> RecordStatsState<S, B> {
 	}
 }
 
+pub struct RawIter<S, B>
+where
+	S: StateBackend<HashFor<B>>,
+	B: BlockT,
+{
+	inner: <S as StateBackend<HashFor<B>>>::RawIter,
+}
+
+impl<S, B> StorageIterator<HashFor<B>> for RawIter<S, B>
+where
+	S: StateBackend<HashFor<B>>,
+	B: BlockT,
+{
+	type Backend = RecordStatsState<S, B>;
+	type Error = S::Error;
+
+	fn next_key(&mut self, backend: &Self::Backend) -> Option<Result<StorageKey, Self::Error>> {
+		self.inner.next_key(&backend.state)
+	}
+
+	fn next_pair(
+		&mut self,
+		backend: &Self::Backend,
+	) -> Option<Result<(StorageKey, StorageValue), Self::Error>> {
+		self.inner.next_pair(&backend.state)
+	}
+
+	fn was_complete(&self) -> bool {
+		self.inner.was_complete()
+	}
+}
+
 impl<S: StateBackend<HashFor<B>>, B: BlockT> StateBackend<HashFor<B>> for RecordStatsState<S, B> {
 	type Error = S::Error;
 	type Transaction = S::Transaction;
 	type TrieBackendStorage = S::TrieBackendStorage;
+	type RawIter = RawIter<S, B>;
 
 	fn storage(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {
 		let value = self.state.storage(key)?;
@@ -122,28 +155,6 @@ impl<S: StateBackend<HashFor<B>>, B: BlockT> StateBackend<HashFor<B>> for Record
 		self.state.exists_child_storage(child_info, key)
 	}
 
-	fn apply_to_key_values_while<F: FnMut(Vec<u8>, Vec<u8>) -> bool>(
-		&self,
-		child_info: Option<&ChildInfo>,
-		prefix: Option<&[u8]>,
-		start_at: Option<&[u8]>,
-		f: F,
-		allow_missing: bool,
-	) -> Result<bool, Self::Error> {
-		self.state
-			.apply_to_key_values_while(child_info, prefix, start_at, f, allow_missing)
-	}
-
-	fn apply_to_keys_while<F: FnMut(&[u8]) -> bool>(
-		&self,
-		child_info: Option<&ChildInfo>,
-		prefix: Option<&[u8]>,
-		start_at: Option<&[u8]>,
-		f: F,
-	) {
-		self.state.apply_to_keys_while(child_info, prefix, start_at, f)
-	}
-
 	fn next_storage_key(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {
 		self.state.next_storage_key(key)
 	}
@@ -156,23 +167,6 @@ impl<S: StateBackend<HashFor<B>>, B: BlockT> StateBackend<HashFor<B>> for Record
 		self.state.next_child_storage_key(child_info, key)
 	}
 
-	fn for_keys_with_prefix<F: FnMut(&[u8])>(&self, prefix: &[u8], f: F) {
-		self.state.for_keys_with_prefix(prefix, f)
-	}
-
-	fn for_key_values_with_prefix<F: FnMut(&[u8], &[u8])>(&self, prefix: &[u8], f: F) {
-		self.state.for_key_values_with_prefix(prefix, f)
-	}
-
-	fn for_child_keys_with_prefix<F: FnMut(&[u8])>(
-		&self,
-		child_info: &ChildInfo,
-		prefix: &[u8],
-		f: F,
-	) {
-		self.state.for_child_keys_with_prefix(child_info, prefix, f)
-	}
-
 	fn storage_root<'a>(
 		&self,
 		delta: impl Iterator<Item = (&'a [u8], Option<&'a [u8]>)>,
@@ -196,16 +190,8 @@ impl<S: StateBackend<HashFor<B>>, B: BlockT> StateBackend<HashFor<B>> for Record
 		self.state.child_storage_root(child_info, delta, state_version)
 	}
 
-	fn pairs(&self) -> Vec<(Vec<u8>, Vec<u8>)> {
-		self.state.pairs()
-	}
-
-	fn keys(&self, prefix: &[u8]) -> Vec<Vec<u8>> {
-		self.state.keys(prefix)
-	}
-
-	fn child_keys(&self, child_info: &ChildInfo, prefix: &[u8]) -> Vec<Vec<u8>> {
-		self.state.child_keys(child_info, prefix)
+	fn raw_iter(&self, args: IterArgs) -> Result<Self::RawIter, Self::Error> {
+		self.state.raw_iter(args).map(|inner| RawIter { inner })
 	}
 
 	fn register_overlay_stats(&self, stats: &sp_state_machine::StateMachineStats) {

substrate/client/rpc/src/state/state_full.rs

@@ -213,23 +213,29 @@ where
 			.map_err(client_err)
 	}
 
+	// TODO: This is horribly broken; either remove it, or make it streaming.
 	fn storage_keys(
 		&self,
 		block: Option<Block::Hash>,
 		prefix: StorageKey,
 	) -> std::result::Result<Vec<StorageKey>, Error> {
+		// TODO: Remove the `.collect`.
 		self.block_or_best(block)
-			.and_then(|block| self.client.storage_keys(block, &prefix))
+			.and_then(|block| self.client.storage_keys(block, Some(&prefix), None))
+			.map(|iter| iter.collect())
 			.map_err(client_err)
 	}
 
+	// TODO: This is horribly broken; either remove it, or make it streaming.
 	fn storage_pairs(
 		&self,
 		block: Option<Block::Hash>,
 		prefix: StorageKey,
 	) -> std::result::Result<Vec<(StorageKey, StorageData)>, Error> {
+		// TODO: Remove the `.collect`.
 		self.block_or_best(block)
-			.and_then(|block| self.client.storage_pairs(block, &prefix))
+			.and_then(|block| self.client.storage_pairs(block, Some(&prefix), None))
+			.map(|iter| iter.collect())
 			.map_err(client_err)
 	}
 
@@ -241,9 +247,7 @@ where
 		start_key: Option<StorageKey>,
 	) -> std::result::Result<Vec<StorageKey>, Error> {
 		self.block_or_best(block)
-			.and_then(|block| {
-				self.client.storage_keys_iter(block, prefix.as_ref(), start_key.as_ref())
-			})
+			.and_then(|block| self.client.storage_keys(block, prefix.as_ref(), start_key.as_ref()))
 			.map(|iter| iter.take(count as usize).collect())
 			.map_err(client_err)
 	}
@@ -284,7 +288,7 @@ where
 			}
 
 			// The key doesn't point to anything, so it's probably a prefix.
-			let iter = match client.storage_keys_iter(block, Some(&key), None).map_err(client_err) {
+			let iter = match client.storage_keys(block, Some(&key), None).map_err(client_err) {
 				Ok(iter) => iter,
 				Err(e) => return Ok(Err(e)),
 			};
@@ -531,6 +535,7 @@ where
 		storage_key: PrefixedStorageKey,
 		prefix: StorageKey,
 	) -> std::result::Result<Vec<StorageKey>, Error> {
+		// TODO: Remove the `.collect`.
 		self.block_or_best(block)
 			.and_then(|block| {
 				let child_info = match ChildType::from_prefixed_key(&storage_key) {
@@ -538,8 +543,9 @@ where
 						ChildInfo::new_default(storage_key),
 					None => return Err(sp_blockchain::Error::InvalidChildStorageKey),
 				};
-				self.client.child_storage_keys(block, &child_info, &prefix)
+				self.client.child_storage_keys(block, child_info, Some(&prefix), None)
 			})
+			.map(|iter| iter.collect())
 			.map_err(client_err)
 	}
 
@@ -558,7 +564,7 @@ where
 						ChildInfo::new_default(storage_key),
 					None => return Err(sp_blockchain::Error::InvalidChildStorageKey),
 				};
-				self.client.child_storage_keys_iter(
+				self.client.child_storage_keys(
 					block,
 					child_info,
 					prefix.as_ref(),

substrate/client/service/src/chain_ops/export_raw_state.rs

@@ -21,7 +21,10 @@ use sc_client_api::{StorageProvider, UsageProvider};
 use sp_core::storage::{well_known_keys, ChildInfo, Storage, StorageChild, StorageKey, StorageMap};
 use sp_runtime::traits::Block as BlockT;
 
-use std::{collections::HashMap, sync::Arc};
+use std::{
+	collections::{BTreeMap, HashMap},
+	sync::Arc,
+};
 
 /// Export the raw state at the given `block`. If `block` is `None`, the
 /// best block will be used.
@@ -31,35 +34,30 @@ where
 	B: BlockT,
 	BA: sc_client_api::backend::Backend<B>,
 {
-	let empty_key = StorageKey(Vec::new());
-	let mut top_storage = client.storage_pairs(hash, &empty_key)?;
+	let mut top = BTreeMap::new();
 	let mut children_default = HashMap::new();
 
-	// Remove all default child storage roots from the top storage and collect the child storage
-	// pairs.
-	while let Some(pos) = top_storage
-		.iter()
-		.position(|(k, _)| k.0.starts_with(well_known_keys::DEFAULT_CHILD_STORAGE_KEY_PREFIX))
-	{
-		let (key, _) = top_storage.swap_remove(pos);
-
-		let key =
-			StorageKey(key.0[well_known_keys::DEFAULT_CHILD_STORAGE_KEY_PREFIX.len()..].to_vec());
-		let child_info = ChildInfo::new_default(&key.0);
-
-		let keys = client.child_storage_keys(hash, &child_info, &empty_key)?;
-		let mut pairs = StorageMap::new();
-		keys.into_iter().try_for_each(|k| {
-			if let Some(value) = client.child_storage(hash, &child_info, &k)? {
-				pairs.insert(k.0, value.0);
+	for (key, value) in client.storage_pairs(hash, None, None)? {
+		// Remove all default child storage roots from the top storage and collect the child storage
+		// pairs.
+		if key.0.starts_with(well_known_keys::DEFAULT_CHILD_STORAGE_KEY_PREFIX) {
+			let child_root_key = StorageKey(
+				key.0[well_known_keys::DEFAULT_CHILD_STORAGE_KEY_PREFIX.len()..].to_vec(),
+			);
+			let child_info = ChildInfo::new_default(&child_root_key.0);
+			let mut pairs = StorageMap::new();
+			for child_key in client.child_storage_keys(hash, child_info.clone(), None, None)? {
+				if let Some(child_value) = client.child_storage(hash, &child_info, &child_key)? {
+					pairs.insert(child_key.0, child_value.0);
+				}
 			}
 
-			Ok::<_, Error>(())
-		})?;
+			children_default.insert(child_root_key.0, StorageChild { child_info, data: pairs });
+			continue
+		}
 
-		children_default.insert(key.0, StorageChild { child_info, data: pairs });
+		top.insert(key.0, value.0);
 	}
 
-	let top = top_storage.into_iter().map(|(k, v)| (k.0, v.0)).collect();
 	Ok(Storage { top, children_default })
 }

substrate/client/service/src/client/client.rs

@@ -40,8 +40,8 @@ use sc_client_api::{
 	},
 	execution_extensions::ExecutionExtensions,
 	notifications::{StorageEventStream, StorageNotifications},
-	CallExecutor, ExecutorProvider, KeyIterator, OnFinalityAction, OnImportAction, ProofProvider,
-	UsageProvider,
+	CallExecutor, ExecutorProvider, KeysIter, OnFinalityAction, OnImportAction, PairsIter,
+	ProofProvider, UsageProvider,
 };
 use sc_consensus::{
 	BlockCheckParams, BlockImportParams, ForkChoiceStrategy, ImportResult, StateAction,
@@ -1462,52 +1462,37 @@ where
 	Block: BlockT,
 {
 	fn storage_keys(
-		&self,
-		hash: Block::Hash,
-		key_prefix: &StorageKey,
-	) -> sp_blockchain::Result<Vec<StorageKey>> {
-		let keys = self.state_at(hash)?.keys(&key_prefix.0).into_iter().map(StorageKey).collect();
-		Ok(keys)
-	}
-
-	fn storage_pairs(
-		&self,
-		hash: <Block as BlockT>::Hash,
-		key_prefix: &StorageKey,
-	) -> sp_blockchain::Result<Vec<(StorageKey, StorageData)>> {
-		let state = self.state_at(hash)?;
-		let keys = state
-			.keys(&key_prefix.0)
-			.into_iter()
-			.map(|k| {
-				let d = state.storage(&k).ok().flatten().unwrap_or_default();
-				(StorageKey(k), StorageData(d))
-			})
-			.collect();
-		Ok(keys)
-	}
-
-	fn storage_keys_iter(
 		&self,
 		hash: <Block as BlockT>::Hash,
 		prefix: Option<&StorageKey>,
 		start_key: Option<&StorageKey>,
-	) -> sp_blockchain::Result<KeyIterator<B::State, Block>> {
+	) -> sp_blockchain::Result<KeysIter<B::State, Block>> {
 		let state = self.state_at(hash)?;
-		let start_key = start_key.or(prefix).map(|key| key.0.clone()).unwrap_or_else(Vec::new);
-		Ok(KeyIterator::new(state, prefix.cloned(), start_key))
+		KeysIter::new(state, prefix, start_key)
+			.map_err(|e| sp_blockchain::Error::from_state(Box::new(e)))
 	}
 
-	fn child_storage_keys_iter(
+	fn child_storage_keys(
 		&self,
 		hash: <Block as BlockT>::Hash,
 		child_info: ChildInfo,
 		prefix: Option<&StorageKey>,
 		start_key: Option<&StorageKey>,
-	) -> sp_blockchain::Result<KeyIterator<B::State, Block>> {
+	) -> sp_blockchain::Result<KeysIter<B::State, Block>> {
 		let state = self.state_at(hash)?;
-		let start_key = start_key.or(prefix).map(|key| key.0.clone()).unwrap_or_else(Vec::new);
-		Ok(KeyIterator::new_child(state, child_info, prefix.cloned(), start_key))
+		KeysIter::new_child(state, child_info, prefix, start_key)
+			.map_err(|e| sp_blockchain::Error::from_state(Box::new(e)))
+	}
+
+	fn storage_pairs(
+		&self,
+		hash: <Block as BlockT>::Hash,
+		prefix: Option<&StorageKey>,
+		start_key: Option<&StorageKey>,
+	) -> sp_blockchain::Result<PairsIter<B::State, Block>> {
+		let state = self.state_at(hash)?;
+		PairsIter::new(state, prefix, start_key)
+			.map_err(|e| sp_blockchain::Error::from_state(Box::new(e)))
 	}
 
 	fn storage(
@@ -1532,21 +1517,6 @@ where
 			.map_err(|e| sp_blockchain::Error::from_state(Box::new(e)))
 	}
 
-	fn child_storage_keys(
-		&self,
-		hash: <Block as BlockT>::Hash,
-		child_info: &ChildInfo,
-		key_prefix: &StorageKey,
-	) -> sp_blockchain::Result<Vec<StorageKey>> {
-		let keys = self
-			.state_at(hash)?
-			.child_keys(child_info, &key_prefix.0)
-			.into_iter()
-			.map(StorageKey)
-			.collect();
-		Ok(keys)
-	}
-
 	fn child_storage(
 		&self,
 		hash: <Block as BlockT>::Hash,

substrate/client/service/test/src/client/mod.rs

@@ -341,7 +341,20 @@ fn block_builder_works_with_transactions() {
 		.expect("block 1 was just imported. qed");
 
 	assert_eq!(client.chain_info().best_number, 1);
-	assert_ne!(client.state_at(hash1).unwrap().pairs(), client.state_at(hash0).unwrap().pairs());
+	assert_ne!(
+		client
+			.state_at(hash1)
+			.unwrap()
+			.pairs(Default::default())
+			.unwrap()
+			.collect::<Vec<_>>(),
+		client
+			.state_at(hash0)
+			.unwrap()
+			.pairs(Default::default())
+			.unwrap()
+			.collect::<Vec<_>>()
+	);
 	assert_eq!(
 		client
 			.runtime_api()
@@ -394,8 +407,18 @@ fn block_builder_does_not_include_invalid() {
 
 	assert_eq!(client.chain_info().best_number, 1);
 	assert_ne!(
-		client.state_at(hashof1).unwrap().pairs(),
-		client.state_at(hashof0).unwrap().pairs()
+		client
+			.state_at(hashof1)
+			.unwrap()
+			.pairs(Default::default())
+			.unwrap()
+			.collect::<Vec<_>>(),
+		client
+			.state_at(hashof0)
+			.unwrap()
+			.pairs(Default::default())
+			.unwrap()
+			.collect::<Vec<_>>()
 	);
 	assert_eq!(client.body(hashof1).unwrap().unwrap().len(), 1)
 }
@@ -1688,7 +1711,7 @@ fn returns_status_for_pruned_blocks() {
 }
 
 #[test]
-fn storage_keys_iter_prefix_and_start_key_works() {
+fn storage_keys_prefix_and_start_key_works() {
 	let child_info = ChildInfo::new_default(b"child");
 	let client = TestClientBuilder::new()
 		.add_extra_child_storage(&child_info, b"first".to_vec(), vec![0u8; 32])
@@ -1703,7 +1726,7 @@ fn storage_keys_iter_prefix_and_start_key_works() {
 	let child_prefix = StorageKey(b"sec".to_vec());
 
 	let res: Vec<_> = client
-		.storage_keys_iter(block_hash, Some(&prefix), None)
+		.storage_keys(block_hash, Some(&prefix), None)
 		.unwrap()
 		.map(|x| x.0)
 		.collect();
@@ -1717,7 +1740,7 @@ fn storage_keys_iter_prefix_and_start_key_works() {
 	);
 
 	let res: Vec<_> = client
-		.storage_keys_iter(
+		.storage_keys(
 			block_hash,
 			Some(&prefix),
 			Some(&StorageKey(array_bytes::hex2bytes_unchecked("3a636f6465"))),
@@ -1728,7 +1751,7 @@ fn storage_keys_iter_prefix_and_start_key_works() {
 	assert_eq!(res, [array_bytes::hex2bytes_unchecked("3a686561707061676573")]);
 
 	let res: Vec<_> = client
-		.storage_keys_iter(
+		.storage_keys(
 			block_hash,
 			Some(&prefix),
 			Some(&StorageKey(array_bytes::hex2bytes_unchecked("3a686561707061676573"))),
@@ -1739,19 +1762,14 @@ fn storage_keys_iter_prefix_and_start_key_works() {
 	assert_eq!(res, Vec::<Vec<u8>>::new());
 
 	let res: Vec<_> = client
-		.child_storage_keys_iter(block_hash, child_info.clone(), Some(&child_prefix), None)
+		.child_storage_keys(block_hash, child_info.clone(), Some(&child_prefix), None)
 		.unwrap()
 		.map(|x| x.0)
 		.collect();
 	assert_eq!(res, [b"second".to_vec()]);
 
 	let res: Vec<_> = client
-		.child_storage_keys_iter(
-			block_hash,
-			child_info,
-			None,
-			Some(&StorageKey(b"second".to_vec())),
-		)
+		.child_storage_keys(block_hash, child_info, None, Some(&StorageKey(b"second".to_vec())))
 		.unwrap()
 		.map(|x| x.0)
 		.collect();
@@ -1759,7 +1777,7 @@ fn storage_keys_iter_prefix_and_start_key_works() {
 }
 
 #[test]
-fn storage_keys_iter_works() {
+fn storage_keys_works() {
 	let client = substrate_test_runtime_client::new();
 
 	let block_hash = client.info().best_hash;
@@ -1767,7 +1785,7 @@ fn storage_keys_iter_works() {
 	let prefix = StorageKey(array_bytes::hex2bytes_unchecked(""));
 
 	let res: Vec<_> = client
-		.storage_keys_iter(block_hash, Some(&prefix), None)
+		.storage_keys(block_hash, Some(&prefix), None)
 		.unwrap()
 		.take(9)
 		.map(|x| array_bytes::bytes2hex("", &x.0))
@@ -1787,8 +1805,56 @@ fn storage_keys_iter_works() {
 		]
 	);
 
+	// Starting at an empty key nothing gets skipped.
 	let res: Vec<_> = client
-		.storage_keys_iter(
+		.storage_keys(block_hash, Some(&prefix), Some(&StorageKey("".into())))
+		.unwrap()
+		.take(9)
+		.map(|x| array_bytes::bytes2hex("", &x.0))
+		.collect();
+	assert_eq!(
+		res,
+		[
+			"00c232cf4e70a5e343317016dc805bf80a6a8cd8ad39958d56f99891b07851e0",
+			"085b2407916e53a86efeb8b72dbe338c4b341dab135252f96b6ed8022209b6cb",
+			"0befda6e1ca4ef40219d588a727f1271",
+			"1a560ecfd2a62c2b8521ef149d0804eb621050e3988ed97dca55f0d7c3e6aa34",
+			"1d66850d32002979d67dd29dc583af5b2ae2a1f71c1f35ad90fff122be7a3824",
+			"237498b98d8803334286e9f0483ef513098dd3c1c22ca21c4dc155b4ef6cc204",
+			"26aa394eea5630e07c48ae0c9558cef75e0621c4869aa60c02be9adcc98a0d1d",
+			"29b9db10ec5bf7907d8f74b5e60aa8140c4fbdd8127a1ee5600cb98e5ec01729",
+			"3a636f6465",
+		]
+	);
+
+	// Starting at an incomplete key nothing gets skipped.
+	let res: Vec<_> = client
+		.storage_keys(
+			block_hash,
+			Some(&prefix),
+			Some(&StorageKey(array_bytes::hex2bytes_unchecked("3a636f64"))),
+		)
+		.unwrap()
+		.take(8)
+		.map(|x| array_bytes::bytes2hex("", &x.0))
+		.collect();
+	assert_eq!(
+		res,
+		[
+			"3a636f6465",
+			"3a686561707061676573",
+			"52008686cc27f6e5ed83a216929942f8bcd32a396f09664a5698f81371934b56",
+			"5348d72ac6cc66e5d8cbecc27b0e0677503b845fe2382d819f83001781788fd5",
+			"5c2d5fda66373dabf970e4fb13d277ce91c5233473321129d32b5a8085fa8133",
+			"6644b9b8bc315888ac8e41a7968dc2b4141a5403c58acdf70b7e8f7e07bf5081",
+			"66484000ed3f75c95fc7b03f39c20ca1e1011e5999278247d3b2f5e3c3273808",
+			"7d5007603a7f5dd729d51d93cf695d6465789443bb967c0d1fe270e388c96eaa",
+		]
+	);
+
+	// Starting at a complete key the first key is skipped.
+	let res: Vec<_> = client
+		.storage_keys(
 			block_hash,
 			Some(&prefix),
 			Some(&StorageKey(array_bytes::hex2bytes_unchecked("3a636f6465"))),
@@ -1811,7 +1877,7 @@ fn storage_keys_iter_works() {
 	);
 
 	let res: Vec<_> = client
-		.storage_keys_iter(
+		.storage_keys(
 			block_hash,
 			Some(&prefix),
 			Some(&StorageKey(array_bytes::hex2bytes_unchecked(