Add block-centric Storage API (#774)

* blocks: Add storage method

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* Add support for runtime API calls and expose it to the blocks API

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* storage: Add storage type for block centric API

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* Adjust subxt to the new Storage interface

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

* Fix clippy

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

examples/examples/concurrent_storage_requests.rs

@@ -32,8 +32,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     // For storage requests, we can join futures together to
     // await multiple futures concurrently:
-    let a_fut = api.storage().fetch(&staking_bonded, None);
+    let a_fut = api.storage().at(None).await?.fetch(&staking_bonded);
-    let b_fut = api.storage().fetch(&staking_ledger, None);
+    let b_fut = api.storage().at(None).await?.fetch(&staking_ledger);
     let (a, b) = join!(a_fut, b_fut);
 
     println!("{a:?}, {b:?}");

examples/examples/dynamic_queries.rs

@@ -66,7 +66,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     );
     let account = api
         .storage()
-        .fetch_or_default(&storage_address, None)
+        .at(None)
+        .await?
+        .fetch_or_default(&storage_address)
         .await?
         .to_value()?;
     println!("Bob's account details: {account}");
@@ -74,7 +76,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // 4. Dynamic storage iteration (the dynamic equivalent to the fetch_all_accounts example).
 
     let storage_address = subxt::dynamic::storage_root("System", "Account");
-    let mut iter = api.storage().iter(storage_address, 10, None).await?;
+    let mut iter = api
+        .storage()
+        .at(None)
+        .await?
+        .iter(storage_address, 10)
+        .await?;
     while let Some((key, account)) = iter.next().await? {
         println!("{}: {}", hex::encode(key), account.to_value()?);
     }

examples/examples/fetch_all_accounts.rs

@@ -26,7 +26,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let address = polkadot::storage().system().account_root();
 
-    let mut iter = api.storage().iter(address, 10, None).await?;
+    let mut iter = api.storage().at(None).await?.iter(address, 10).await?;
 
     while let Some((key, account)) = iter.next().await? {
         println!("{}: {}", hex::encode(key), account.data.free);

examples/examples/fetch_staking_details.rs

@@ -32,7 +32,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let api = OnlineClient::<PolkadotConfig>::new().await?;
 
     let active_era_addr = polkadot::storage().staking().active_era();
-    let era = api.storage().fetch(&active_era_addr, None).await?.unwrap();
+    let era = api
+        .storage()
+        .at(None)
+        .await?
+        .fetch(&active_era_addr)
+        .await?
+        .unwrap();
     println!(
         "Staking active era: index: {:?}, start: {:?}",
         era.index, era.start
@@ -52,13 +58,20 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let controller_acc_addr = polkadot::storage().staking().bonded(&alice_stash_id);
     let controller_acc = api
         .storage()
-        .fetch(&controller_acc_addr, None)
+        .at(None)
+        .await?
+        .fetch(&controller_acc_addr)
         .await?
         .unwrap();
     println!("    account controlled by: {:?}", controller_acc);
 
     let era_reward_addr = polkadot::storage().staking().eras_reward_points(era.index);
-    let era_result = api.storage().fetch(&era_reward_addr, None).await?;
+    let era_result = api
+        .storage()
+        .at(None)
+        .await?
+        .fetch(&era_reward_addr)
+        .await?;
     println!("Era reward points: {:?}", era_result);
 
     Ok(())

examples/examples/storage_iterating.rs

@@ -35,7 +35,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     {
         let key_addr = polkadot::storage().xcm_pallet().version_notifiers_root();
 
-        let mut iter = api.storage().iter(key_addr, 10, None).await?;
+        let mut iter = api.storage().at(None).await?.iter(key_addr, 10).await?;
 
         println!("\nExample 1. Obtained keys:");
         while let Some((key, value)) = iter.next().await? {
@@ -52,14 +52,18 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         // Fetch at most 10 keys from below the prefix XcmPallet' VersionNotifiers.
         let keys = api
             .storage()
-            .fetch_keys(&key_addr.to_root_bytes(), 10, None, None)
+            .at(None)
+            .await?
+            .fetch_keys(&key_addr.to_root_bytes(), 10, None)
             .await?;
 
         println!("Example 2. Obtained keys:");
         for key in keys.iter() {
             println!("Key: 0x{}", hex::encode(key));
 
-            if let Some(storage_data) = api.storage().fetch_raw(&key.0, None).await? {
+            if let Some(storage_data) =
+                api.storage().at(None).await?.fetch_raw(&key.0).await?
+            {
                 // We know the return value to be `QueryId` (`u64`) from inspecting either:
                 // - polkadot code
                 // - polkadot.rs generated file under `version_notifiers()` fn
@@ -86,13 +90,20 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         // `twox_128("XcmPallet") ++ twox_128("VersionNotifiers") ++ twox_64(2u32) ++ 2u32`
         println!("\nExample 3\nQuery key: 0x{}", hex::encode(&query_key));
 
-        let keys = api.storage().fetch_keys(&query_key, 10, None, None).await?;
+        let keys = api
+            .storage()
+            .at(None)
+            .await?
+            .fetch_keys(&query_key, 10, None)
+            .await?;
 
         println!("Obtained keys:");
         for key in keys.iter() {
             println!("Key: 0x{}", hex::encode(key));
 
-            if let Some(storage_data) = api.storage().fetch_raw(&key.0, None).await? {
+            if let Some(storage_data) =
+                api.storage().at(None).await?.fetch_raw(&key.0).await?
+            {
                 // We know the return value to be `QueryId` (`u64`) from inspecting either:
                 // - polkadot code
                 // - polkadot.rs generated file under `version_notifiers()` fn

subxt/src/blocks/block_types.rs

@@ -19,6 +19,7 @@ use crate::{
     events,
     rpc::types::ChainBlockResponse,
     runtime_api::RuntimeApi,
+    storage::Storage,
 };
 use derivative::Derivative;
 use futures::lock::Mutex as AsyncMutex;
@@ -91,6 +92,12 @@ where
         ))
     }
 
+    /// Work with storage.
+    pub fn storage(&self) -> Storage<T, C> {
+        let block_hash = self.hash();
+        Storage::new(self.client.clone(), block_hash)
+    }
+
     /// Execute a runtime API call at this block.
     pub async fn runtime_api(&self) -> Result<RuntimeApi<T, C>, Error> {
         Ok(RuntimeApi::new(self.client.clone(), self.hash()))

subxt/src/rpc/rpc.rs

@@ -173,6 +173,25 @@ impl<T: Config> Rpc<T> {
         Ok(metadata)
     }
 
+    /// Execute a runtime API call.
+    pub async fn call(
+        &self,
+        function: String,
+        call_parameters: Option<&[u8]>,
+        at: Option<T::Hash>,
+    ) -> Result<types::Bytes, Error> {
+        let call_parameters = call_parameters.unwrap_or_default();
+
+        let bytes: types::Bytes = self
+            .client
+            .request(
+                "state_call",
+                rpc_params![function, to_hex(call_parameters), at],
+            )
+            .await?;
+        Ok(bytes)
+    }
+
     /// Fetch system properties
     pub async fn system_properties(&self) -> Result<types::SystemProperties, Error> {
         self.client

subxt/src/storage/mod.rs

@@ -7,12 +7,15 @@
 mod storage_address;
 mod storage_client;
 mod storage_map_key;
+mod storage_type;
 
 pub mod utils;
 
-pub use storage_client::{
+pub use storage_client::StorageClient;
+
+pub use storage_type::{
     KeyIter,
-    StorageClient,
+    Storage,
 };
 
 // Re-export as this is used in the public API in this module:

subxt/src/storage/storage_client.rs

@@ -2,29 +2,23 @@
 // This file is dual-licensed as Apache-2.0 or GPL-3.0.
 // see LICENSE for license details.
 
-use super::storage_address::{
+use super::{
+    storage_type::{
+        validate_storage_address,
+        Storage,
+    },
     StorageAddress,
-    Yes,
 };
+
 use crate::{
     client::{
         OfflineClientT,
         OnlineClientT,
     },
     error::Error,
-    metadata::{
-        DecodeWithMetadata,
-        Metadata,
-    },
-    rpc::types::{
-        StorageData,
-        StorageKey,
-    },
     Config,
 };
 use derivative::Derivative;
-use frame_metadata::StorageEntryType;
-use scale_info::form::PortableForm;
 use std::{
     future::Future,
     marker::PhantomData,
@@ -61,15 +55,7 @@ where
         &self,
         address: &Address,
     ) -> Result<(), Error> {
-        if let Some(hash) = address.validation_hash() {
+        validate_storage_address(address, &self.client.metadata())
-            validate_storage(
-                address.pallet_name(),
-                address.entry_name(),
-                hash,
-                &self.client.metadata(),
-            )?;
-        }
-        Ok(())
     }
 }
 
@@ -78,339 +64,29 @@ where
     T: Config,
     Client: OnlineClientT<T>,
 {
-    /// Fetch the raw encoded value at the address/key given.
+    /// Obtain storage at some block hash.
-    pub fn fetch_raw<'a>(
+    pub fn at(
         &self,
-        key: &'a [u8],
+        block_hash: Option<T::Hash>,
-        hash: Option<T::Hash>,
+    ) -> impl Future<Output = Result<Storage<T, Client>, Error>> + Send + 'static {
-    ) -> impl Future<Output = Result<Option<Vec<u8>>, Error>> + 'a {
+        // Clone and pass the client in like this so that we can explicitly
-        let client = self.client.clone();
+        // return a Future that's Send + 'static, rather than tied to &self.
-        // Ensure that the returned future doesn't have a lifetime tied to api.storage(),
-        // which is a temporary thing we'll be throwing away quickly:
-        async move {
-            let data = client.rpc().storage(key, hash).await?;
-            Ok(data.map(|d| d.0))
-        }
-    }
-
-    /// Fetch a decoded value from storage at a given address and optional block hash.
-    ///
-    /// # Example
-    ///
-    /// ```no_run
-    /// use subxt::{ PolkadotConfig, OnlineClient };
-    ///
-    /// #[subxt::subxt(runtime_metadata_path = "../artifacts/polkadot_metadata.scale")]
-    /// pub mod polkadot {}
-    ///
-    /// # #[tokio::main]
-    /// # async fn main() {
-    /// let api = OnlineClient::<PolkadotConfig>::new().await.unwrap();
-    ///
-    /// // Address to a storage entry we'd like to access.
-    /// let address = polkadot::storage().xcm_pallet().queries(&12345);
-    ///
-    /// // Fetch just the keys, returning up to 10 keys.
-    /// let value = api
-    ///     .storage()
-    ///     .fetch(&address, None)
-    ///     .await
-    ///     .unwrap();
-    ///
-    /// println!("Value: {:?}", value);
-    /// # }
-    /// ```
-    pub fn fetch<'a, Address>(
-        &self,
-        address: &'a Address,
-        hash: Option<T::Hash>,
-    ) -> impl Future<
-        Output = Result<Option<<Address::Target as DecodeWithMetadata>::Target>, Error>,
-    > + 'a
-    where
-        Address: StorageAddress<IsFetchable = Yes> + 'a,
-    {
-        let client = self.clone();
-        async move {
-            // Metadata validation checks whether the static address given
-            // is likely to actually correspond to a real storage entry or not.
-            // if not, it means static codegen doesn't line up with runtime
-            // metadata.
-            client.validate(address)?;
-
-            // Look up the return type ID to enable DecodeWithMetadata:
-            let metadata = client.client.metadata();
-            let lookup_bytes = super::utils::storage_address_bytes(address, &metadata)?;
-            if let Some(data) = client
-                .client
-                .storage()
-                .fetch_raw(&lookup_bytes, hash)
-                .await?
-            {
-                let val = <Address::Target as DecodeWithMetadata>::decode_storage_with_metadata(
-                    &mut &*data,
-                    address.pallet_name(),
-                    address.entry_name(),
-                    &metadata,
-                )?;
-                Ok(Some(val))
-            } else {
-                Ok(None)
-            }
-        }
-    }
-
-    /// Fetch a StorageKey that has a default value with an optional block hash.
-    pub fn fetch_or_default<'a, Address>(
-        &self,
-        address: &'a Address,
-        hash: Option<T::Hash>,
-    ) -> impl Future<Output = Result<<Address::Target as DecodeWithMetadata>::Target, Error>>
-           + 'a
-    where
-        Address: StorageAddress<IsFetchable = Yes, IsDefaultable = Yes> + 'a,
-    {
         let client = self.client.clone();
         async move {
-            let pallet_name = address.pallet_name();
+            // If block hash is not provided, get the hash
-            let storage_name = address.entry_name();
+            // for the latest block and use that.
-            // Metadata validation happens via .fetch():
+            let block_hash = match block_hash {
-            if let Some(data) = client.storage().fetch(address, hash).await? {
+                Some(hash) => hash,
-                Ok(data)
+                None => {
-            } else {
+                    client
-                let metadata = client.metadata();
+                        .rpc()
-
+                        .block_hash(None)
-                // We have to dig into metadata already, so no point using the optimised `decode_storage_with_metadata` call.
+                        .await?
-                let pallet_metadata = metadata.pallet(pallet_name)?;
+                        .expect("didn't pass a block number; qed")
-                let storage_metadata = pallet_metadata.storage(storage_name)?;
+                }
-                let return_ty_id =
-                    return_type_from_storage_entry_type(&storage_metadata.ty);
-                let bytes = &mut &storage_metadata.default[..];
-
-                let val = <Address::Target as DecodeWithMetadata>::decode_with_metadata(
-                    bytes,
-                    return_ty_id,
-                    &metadata,
-                )?;
-                Ok(val)
-            }
-        }
-    }
-
-    /// Fetch up to `count` keys for a storage map in lexicographic order.
-    ///
-    /// Supports pagination by passing a value to `start_key`.
-    pub fn fetch_keys<'a>(
-        &self,
-        key: &'a [u8],
-        count: u32,
-        start_key: Option<&'a [u8]>,
-        hash: Option<T::Hash>,
-    ) -> impl Future<Output = Result<Vec<StorageKey>, Error>> + 'a {
-        let client = self.client.clone();
-        async move {
-            let keys = client
-                .rpc()
-                .storage_keys_paged(key, count, start_key, hash)
-                .await?;
-            Ok(keys)
-        }
-    }
-
-    /// Returns an iterator of key value pairs.
-    ///
-    /// ```no_run
-    /// use subxt::{ PolkadotConfig, OnlineClient };
-    ///
-    /// #[subxt::subxt(runtime_metadata_path = "../artifacts/polkadot_metadata.scale")]
-    /// pub mod polkadot {}
-    ///
-    /// # #[tokio::main]
-    /// # async fn main() {
-    /// let api = OnlineClient::<PolkadotConfig>::new().await.unwrap();
-    ///
-    /// // Address to the root of a storage entry that we'd like to iterate over.
-    /// let address = polkadot::storage().xcm_pallet().version_notifiers_root();
-    ///
-    /// // Iterate over keys and values at that address.
-    /// let mut iter = api
-    ///     .storage()
-    ///     .iter(address, 10, None)
-    ///     .await
-    ///     .unwrap();
-    ///
-    /// while let Some((key, value)) = iter.next().await.unwrap() {
-    ///     println!("Key: 0x{}", hex::encode(&key));
-    ///     println!("Value: {}", value);
-    /// }
-    /// # }
-    /// ```
-    pub fn iter<Address>(
-        &self,
-        address: Address,
-        page_size: u32,
-        hash: Option<T::Hash>,
-    ) -> impl Future<Output = Result<KeyIter<T, Client, Address::Target>, Error>> + 'static
-    where
-        Address: StorageAddress<IsIterable = Yes> + 'static,
-    {
-        let client = self.clone();
-        async move {
-            // Metadata validation checks whether the static address given
-            // is likely to actually correspond to a real storage entry or not.
-            // if not, it means static codegen doesn't line up with runtime
-            // metadata.
-            client.validate(&address)?;
-
-            // Fetch a concrete block hash to iterate over. We do this so that if new blocks
-            // are produced midway through iteration, we continue to iterate at the block
-            // we started with and not the new block.
-            let hash = if let Some(hash) = hash {
-                hash
-            } else {
-                client
-                    .client
-                    .rpc()
-                    .block_hash(None)
-                    .await?
-                    .expect("didn't pass a block number; qed")
             };
 
-            let metadata = client.client.metadata();
+            Ok(Storage::new(client, block_hash))
-
-            // Look up the return type for flexible decoding. Do this once here to avoid
-            // potentially doing it every iteration if we used `decode_storage_with_metadata`
-            // in the iterator.
-            let return_type_id = lookup_storage_return_type(
-                &metadata,
-                address.pallet_name(),
-                address.entry_name(),
-            )?;
-
-            // The root pallet/entry bytes for this storage entry:
-            let address_root_bytes = super::utils::storage_address_root_bytes(&address);
-
-            Ok(KeyIter {
-                client,
-                address_root_bytes,
-                metadata,
-                return_type_id,
-                block_hash: hash,
-                count: page_size,
-                start_key: None,
-                buffer: Default::default(),
-                _marker: std::marker::PhantomData,
-            })
         }
     }
 }
-
-/// Iterates over key value pairs in a map.
-pub struct KeyIter<T: Config, Client, ReturnTy> {
-    client: StorageClient<T, Client>,
-    address_root_bytes: Vec<u8>,
-    return_type_id: u32,
-    metadata: Metadata,
-    count: u32,
-    block_hash: T::Hash,
-    start_key: Option<StorageKey>,
-    buffer: Vec<(StorageKey, StorageData)>,
-    _marker: std::marker::PhantomData<ReturnTy>,
-}
-
-impl<'a, T, Client, ReturnTy> KeyIter<T, Client, ReturnTy>
-where
-    T: Config,
-    Client: OnlineClientT<T>,
-    ReturnTy: DecodeWithMetadata,
-{
-    /// Returns the next key value pair from a map.
-    pub async fn next(
-        &mut self,
-    ) -> Result<Option<(StorageKey, ReturnTy::Target)>, Error> {
-        loop {
-            if let Some((k, v)) = self.buffer.pop() {
-                let val = ReturnTy::decode_with_metadata(
-                    &mut &v.0[..],
-                    self.return_type_id,
-                    &self.metadata,
-                )?;
-                return Ok(Some((k, val)))
-            } else {
-                let start_key = self.start_key.take();
-                let keys = self
-                    .client
-                    .fetch_keys(
-                        &self.address_root_bytes,
-                        self.count,
-                        start_key.as_ref().map(|k| &*k.0),
-                        Some(self.block_hash),
-                    )
-                    .await?;
-
-                if keys.is_empty() {
-                    return Ok(None)
-                }
-
-                self.start_key = keys.last().cloned();
-
-                let change_sets = self
-                    .client
-                    .client
-                    .rpc()
-                    .query_storage_at(keys.iter().map(|k| &*k.0), Some(self.block_hash))
-                    .await?;
-                for change_set in change_sets {
-                    for (k, v) in change_set.changes {
-                        if let Some(v) = v {
-                            self.buffer.push((k, v));
-                        }
-                    }
-                }
-                debug_assert_eq!(self.buffer.len(), keys.len());
-            }
-        }
-    }
-}
-
-/// Validate a storage entry against the metadata.
-fn validate_storage(
-    pallet_name: &str,
-    storage_name: &str,
-    hash: [u8; 32],
-    metadata: &Metadata,
-) -> Result<(), Error> {
-    let expected_hash = match metadata.storage_hash(pallet_name, storage_name) {
-        Ok(hash) => hash,
-        Err(e) => return Err(e.into()),
-    };
-    match expected_hash == hash {
-        true => Ok(()),
-        false => {
-            Err(crate::error::MetadataError::IncompatibleStorageMetadata(
-                pallet_name.into(),
-                storage_name.into(),
-            )
-            .into())
-        }
-    }
-}
-
-/// look up a return type ID for some storage entry.
-fn lookup_storage_return_type(
-    metadata: &Metadata,
-    pallet: &str,
-    entry: &str,
-) -> Result<u32, Error> {
-    let storage_entry_type = &metadata.pallet(pallet)?.storage(entry)?.ty;
-
-    Ok(return_type_from_storage_entry_type(storage_entry_type))
-}
-
-/// Fetch the return type out of a [`StorageEntryType`].
-fn return_type_from_storage_entry_type(entry: &StorageEntryType<PortableForm>) -> u32 {
-    match entry {
-        StorageEntryType::Plain(ty) => ty.id(),
-        StorageEntryType::Map { value, .. } => value.id(),
-    }
-}

subxt/src/storage/storage_type.rs

@@ -0,0 +1,404 @@
+// Copyright 2019-2022 Parity Technologies (UK) Ltd.
+// This file is dual-licensed as Apache-2.0 or GPL-3.0.
+// see LICENSE for license details.
+
+use super::storage_address::{
+    StorageAddress,
+    Yes,
+};
+use crate::{
+    client::{
+        OfflineClientT,
+        OnlineClientT,
+    },
+    error::Error,
+    metadata::{
+        DecodeWithMetadata,
+        Metadata,
+    },
+    rpc::types::{
+        StorageData,
+        StorageKey,
+    },
+    Config,
+};
+use derivative::Derivative;
+use frame_metadata::StorageEntryType;
+use scale_info::form::PortableForm;
+use std::{
+    future::Future,
+    marker::PhantomData,
+};
+
+/// Query the runtime storage.
+#[derive(Derivative)]
+#[derivative(Clone(bound = "Client: Clone"))]
+pub struct Storage<T: Config, Client> {
+    client: Client,
+    block_hash: T::Hash,
+    _marker: PhantomData<T>,
+}
+
+impl<T: Config, Client> Storage<T, Client> {
+    /// Create a new [`Storage`]
+    pub(crate) fn new(client: Client, block_hash: T::Hash) -> Self {
+        Self {
+            client,
+            block_hash,
+            _marker: PhantomData,
+        }
+    }
+}
+
+impl<T, Client> Storage<T, Client>
+where
+    T: Config,
+    Client: OfflineClientT<T>,
+{
+    /// Run the validation logic against some storage address you'd like to access.
+    ///
+    /// Method has the same meaning as [`StorageClient::validate`](super::storage_client::StorageClient::validate).
+    pub fn validate<Address: StorageAddress>(
+        &self,
+        address: &Address,
+    ) -> Result<(), Error> {
+        validate_storage_address(address, &self.client.metadata())
+    }
+}
+
+impl<T, Client> Storage<T, Client>
+where
+    T: Config,
+    Client: OnlineClientT<T>,
+{
+    /// Fetch the raw encoded value at the address/key given.
+    pub fn fetch_raw<'a>(
+        &self,
+        key: &'a [u8],
+    ) -> impl Future<Output = Result<Option<Vec<u8>>, Error>> + 'a {
+        let client = self.client.clone();
+        let block_hash = self.block_hash;
+        // Ensure that the returned future doesn't have a lifetime tied to api.storage(),
+        // which is a temporary thing we'll be throwing away quickly:
+        async move {
+            let data = client.rpc().storage(key, Some(block_hash)).await?;
+            Ok(data.map(|d| d.0))
+        }
+    }
+
+    /// Fetch a decoded value from storage at a given address.
+    ///
+    /// # Example
+    ///
+    /// ```no_run
+    /// use subxt::{ PolkadotConfig, OnlineClient };
+    ///
+    /// #[subxt::subxt(runtime_metadata_path = "../artifacts/polkadot_metadata.scale")]
+    /// pub mod polkadot {}
+    ///
+    /// # #[tokio::main]
+    /// # async fn main() {
+    /// let api = OnlineClient::<PolkadotConfig>::new().await.unwrap();
+    ///
+    /// // Address to a storage entry we'd like to access.
+    /// let address = polkadot::storage().xcm_pallet().queries(&12345);
+    ///
+    /// // Fetch just the keys, returning up to 10 keys.
+    /// let value = api
+    ///     .storage()
+    ///     .at(None)
+    ///     .await
+    ///     .unwrap()
+    ///     .fetch(&address)
+    ///     .await
+    ///     .unwrap();
+    ///
+    /// println!("Value: {:?}", value);
+    /// # }
+    /// ```
+    pub fn fetch<'a, Address>(
+        &self,
+        address: &'a Address,
+    ) -> impl Future<
+        Output = Result<Option<<Address::Target as DecodeWithMetadata>::Target>, Error>,
+    > + 'a
+    where
+        Address: StorageAddress<IsFetchable = Yes> + 'a,
+    {
+        let client = self.clone();
+        async move {
+            // Metadata validation checks whether the static address given
+            // is likely to actually correspond to a real storage entry or not.
+            // if not, it means static codegen doesn't line up with runtime
+            // metadata.
+            client.validate(address)?;
+
+            // Look up the return type ID to enable DecodeWithMetadata:
+            let metadata = client.client.metadata();
+            let lookup_bytes = super::utils::storage_address_bytes(address, &metadata)?;
+            if let Some(data) = client.fetch_raw(&lookup_bytes).await? {
+                let val = <Address::Target as DecodeWithMetadata>::decode_storage_with_metadata(
+                    &mut &*data,
+                    address.pallet_name(),
+                    address.entry_name(),
+                    &metadata,
+                )?;
+                Ok(Some(val))
+            } else {
+                Ok(None)
+            }
+        }
+    }
+
+    /// Fetch a StorageKey that has a default value with an optional block hash.
+    pub fn fetch_or_default<'a, Address>(
+        &self,
+        address: &'a Address,
+    ) -> impl Future<Output = Result<<Address::Target as DecodeWithMetadata>::Target, Error>>
+           + 'a
+    where
+        Address: StorageAddress<IsFetchable = Yes, IsDefaultable = Yes> + 'a,
+    {
+        let client = self.clone();
+        async move {
+            let pallet_name = address.pallet_name();
+            let storage_name = address.entry_name();
+            // Metadata validation happens via .fetch():
+            if let Some(data) = client.fetch(address).await? {
+                Ok(data)
+            } else {
+                let metadata = client.client.metadata();
+
+                // We have to dig into metadata already, so no point using the optimised `decode_storage_with_metadata` call.
+                let pallet_metadata = metadata.pallet(pallet_name)?;
+                let storage_metadata = pallet_metadata.storage(storage_name)?;
+                let return_ty_id =
+                    return_type_from_storage_entry_type(&storage_metadata.ty);
+                let bytes = &mut &storage_metadata.default[..];
+
+                let val = <Address::Target as DecodeWithMetadata>::decode_with_metadata(
+                    bytes,
+                    return_ty_id,
+                    &metadata,
+                )?;
+                Ok(val)
+            }
+        }
+    }
+
+    /// Fetch up to `count` keys for a storage map in lexicographic order.
+    ///
+    /// Supports pagination by passing a value to `start_key`.
+    pub fn fetch_keys<'a>(
+        &self,
+        key: &'a [u8],
+        count: u32,
+        start_key: Option<&'a [u8]>,
+    ) -> impl Future<Output = Result<Vec<StorageKey>, Error>> + 'a {
+        let client = self.client.clone();
+        let block_hash = self.block_hash;
+        async move {
+            let keys = client
+                .rpc()
+                .storage_keys_paged(key, count, start_key, Some(block_hash))
+                .await?;
+            Ok(keys)
+        }
+    }
+
+    /// Returns an iterator of key value pairs.
+    ///
+    /// ```no_run
+    /// use subxt::{ PolkadotConfig, OnlineClient };
+    ///
+    /// #[subxt::subxt(runtime_metadata_path = "../artifacts/polkadot_metadata.scale")]
+    /// pub mod polkadot {}
+    ///
+    /// # #[tokio::main]
+    /// # async fn main() {
+    /// let api = OnlineClient::<PolkadotConfig>::new().await.unwrap();
+    ///
+    /// // Address to the root of a storage entry that we'd like to iterate over.
+    /// let address = polkadot::storage().xcm_pallet().version_notifiers_root();
+    ///
+    /// // Iterate over keys and values at that address.
+    /// let mut iter = api
+    ///     .storage()
+    ///     .at(None)
+    ///     .await
+    ///     .unwrap()
+    ///     .iter(address, 10)
+    ///     .await
+    ///     .unwrap();
+    ///
+    /// while let Some((key, value)) = iter.next().await.unwrap() {
+    ///     println!("Key: 0x{}", hex::encode(&key));
+    ///     println!("Value: {}", value);
+    /// }
+    /// # }
+    /// ```
+    pub fn iter<Address>(
+        &self,
+        address: Address,
+        page_size: u32,
+    ) -> impl Future<Output = Result<KeyIter<T, Client, Address::Target>, Error>> + 'static
+    where
+        Address: StorageAddress<IsIterable = Yes> + 'static,
+    {
+        let client = self.clone();
+        let block_hash = self.block_hash;
+        async move {
+            // Metadata validation checks whether the static address given
+            // is likely to actually correspond to a real storage entry or not.
+            // if not, it means static codegen doesn't line up with runtime
+            // metadata.
+            client.validate(&address)?;
+
+            let metadata = client.client.metadata();
+
+            // Look up the return type for flexible decoding. Do this once here to avoid
+            // potentially doing it every iteration if we used `decode_storage_with_metadata`
+            // in the iterator.
+            let return_type_id = lookup_storage_return_type(
+                &metadata,
+                address.pallet_name(),
+                address.entry_name(),
+            )?;
+
+            // The root pallet/entry bytes for this storage entry:
+            let address_root_bytes = super::utils::storage_address_root_bytes(&address);
+
+            Ok(KeyIter {
+                client,
+                address_root_bytes,
+                metadata,
+                return_type_id,
+                block_hash,
+                count: page_size,
+                start_key: None,
+                buffer: Default::default(),
+                _marker: std::marker::PhantomData,
+            })
+        }
+    }
+}
+
+/// Iterates over key value pairs in a map.
+pub struct KeyIter<T: Config, Client, ReturnTy> {
+    client: Storage<T, Client>,
+    address_root_bytes: Vec<u8>,
+    return_type_id: u32,
+    metadata: Metadata,
+    count: u32,
+    block_hash: T::Hash,
+    start_key: Option<StorageKey>,
+    buffer: Vec<(StorageKey, StorageData)>,
+    _marker: std::marker::PhantomData<ReturnTy>,
+}
+
+impl<'a, T, Client, ReturnTy> KeyIter<T, Client, ReturnTy>
+where
+    T: Config,
+    Client: OnlineClientT<T>,
+    ReturnTy: DecodeWithMetadata,
+{
+    /// Returns the next key value pair from a map.
+    pub async fn next(
+        &mut self,
+    ) -> Result<Option<(StorageKey, ReturnTy::Target)>, Error> {
+        loop {
+            if let Some((k, v)) = self.buffer.pop() {
+                let val = ReturnTy::decode_with_metadata(
+                    &mut &v.0[..],
+                    self.return_type_id,
+                    &self.metadata,
+                )?;
+                return Ok(Some((k, val)))
+            } else {
+                let start_key = self.start_key.take();
+                let keys = self
+                    .client
+                    .fetch_keys(
+                        &self.address_root_bytes,
+                        self.count,
+                        start_key.as_ref().map(|k| &*k.0),
+                    )
+                    .await?;
+
+                if keys.is_empty() {
+                    return Ok(None)
+                }
+
+                self.start_key = keys.last().cloned();
+
+                let change_sets = self
+                    .client
+                    .client
+                    .rpc()
+                    .query_storage_at(keys.iter().map(|k| &*k.0), Some(self.block_hash))
+                    .await?;
+                for change_set in change_sets {
+                    for (k, v) in change_set.changes {
+                        if let Some(v) = v {
+                            self.buffer.push((k, v));
+                        }
+                    }
+                }
+                debug_assert_eq!(self.buffer.len(), keys.len());
+            }
+        }
+    }
+}
+
+/// Validate a storage address against the metadata.
+pub(crate) fn validate_storage_address<Address: StorageAddress>(
+    address: &Address,
+    metadata: &Metadata,
+) -> Result<(), Error> {
+    if let Some(hash) = address.validation_hash() {
+        validate_storage(address.pallet_name(), address.entry_name(), hash, metadata)?;
+    }
+    Ok(())
+}
+
+/// Validate a storage entry against the metadata.
+fn validate_storage(
+    pallet_name: &str,
+    storage_name: &str,
+    hash: [u8; 32],
+    metadata: &Metadata,
+) -> Result<(), Error> {
+    let expected_hash = match metadata.storage_hash(pallet_name, storage_name) {
+        Ok(hash) => hash,
+        Err(e) => return Err(e.into()),
+    };
+    match expected_hash == hash {
+        true => Ok(()),
+        false => {
+            Err(crate::error::MetadataError::IncompatibleStorageMetadata(
+                pallet_name.into(),
+                storage_name.into(),
+            )
+            .into())
+        }
+    }
+}
+
+/// look up a return type ID for some storage entry.
+fn lookup_storage_return_type(
+    metadata: &Metadata,
+    pallet: &str,
+    entry: &str,
+) -> Result<u32, Error> {
+    let storage_entry_type = &metadata.pallet(pallet)?.storage(entry)?.ty;
+
+    Ok(return_type_from_storage_entry_type(storage_entry_type))
+}
+
+/// Fetch the return type out of a [`StorageEntryType`].
+fn return_type_from_storage_entry_type(entry: &StorageEntryType<PortableForm>) -> u32 {
+    match entry {
+        StorageEntryType::Plain(ty) => ty.id(),
+        StorageEntryType::Map { value, .. } => value.id(),
+    }
+}

testing/integration-tests/src/client/mod.rs

@@ -126,7 +126,10 @@ async fn fetch_keys() {
     let addr = node_runtime::storage().system().account_root();
     let keys = api
         .storage()
-        .fetch_keys(&addr.to_root_bytes(), 4, None, None)
+        .at(None)
+        .await
+        .unwrap()
+        .fetch_keys(&addr.to_root_bytes(), 4, None)
         .await
         .unwrap();
     assert_eq!(keys.len(), 4)
@@ -138,7 +141,14 @@ async fn test_iter() {
     let api = ctx.client();
 
     let addr = node_runtime::storage().system().account_root();
-    let mut iter = api.storage().iter(addr, 10, None).await.unwrap();
+    let mut iter = api
+        .storage()
+        .at(None)
+        .await
+        .unwrap()
+        .iter(addr, 10)
+        .await
+        .unwrap();
     let mut i = 0;
     while iter.next().await.unwrap().is_some() {
         i += 1;

testing/integration-tests/src/frame/balances.rs

@@ -42,11 +42,15 @@ async fn tx_basic_transfer() -> Result<(), subxt::Error> {
 
     let alice_pre = api
         .storage()
-        .fetch_or_default(&alice_account_addr, None)
+        .at(None)
+        .await?
+        .fetch_or_default(&alice_account_addr)
         .await?;
     let bob_pre = api
         .storage()
-        .fetch_or_default(&bob_account_addr, None)
+        .at(None)
+        .await?
+        .fetch_or_default(&bob_account_addr)
         .await?;
 
     let tx = node_runtime::tx().balances().transfer(bob_address, 10_000);
@@ -75,11 +79,15 @@ async fn tx_basic_transfer() -> Result<(), subxt::Error> {
 
     let alice_post = api
         .storage()
-        .fetch_or_default(&alice_account_addr, None)
+        .at(None)
+        .await?
+        .fetch_or_default(&alice_account_addr)
         .await?;
     let bob_post = api
         .storage()
-        .fetch_or_default(&bob_account_addr, None)
+        .at(None)
+        .await?
+        .fetch_or_default(&bob_account_addr)
         .await?;
 
     assert!(alice_pre.data.free - 10_000 >= alice_post.data.free);
@@ -113,11 +121,15 @@ async fn tx_dynamic_transfer() -> Result<(), subxt::Error> {
 
     let alice_pre = api
         .storage()
-        .fetch_or_default(&alice_account_addr, None)
+        .at(None)
+        .await?
+        .fetch_or_default(&alice_account_addr)
         .await?;
     let bob_pre = api
         .storage()
-        .fetch_or_default(&bob_account_addr, None)
+        .at(None)
+        .await?
+        .fetch_or_default(&bob_account_addr)
         .await?;
 
     let tx = subxt::dynamic::tx(
@@ -159,11 +171,15 @@ async fn tx_dynamic_transfer() -> Result<(), subxt::Error> {
 
     let alice_post = api
         .storage()
-        .fetch_or_default(&alice_account_addr, None)
+        .at(None)
+        .await?
+        .fetch_or_default(&alice_account_addr)
         .await?;
     let bob_post = api
         .storage()
-        .fetch_or_default(&bob_account_addr, None)
+        .at(None)
+        .await?
+        .fetch_or_default(&bob_account_addr)
         .await?;
 
     let alice_pre_free = alice_pre
@@ -214,7 +230,9 @@ async fn multiple_transfers_work_nonce_incremented() -> Result<(), subxt::Error>
 
     let bob_pre = api
         .storage()
-        .fetch_or_default(&bob_account_addr, None)
+        .at(None)
+        .await?
+        .fetch_or_default(&bob_account_addr)
         .await?;
 
     let tx = node_runtime::tx()
@@ -233,7 +251,9 @@ async fn multiple_transfers_work_nonce_incremented() -> Result<(), subxt::Error>
 
     let bob_post = api
         .storage()
-        .fetch_or_default(&bob_account_addr, None)
+        .at(None)
+        .await?
+        .fetch_or_default(&bob_account_addr)
         .await?;
 
     assert_eq!(bob_pre.data.free + 30_000, bob_post.data.free);
@@ -246,7 +266,14 @@ async fn storage_total_issuance() {
     let api = ctx.client();
 
     let addr = node_runtime::storage().balances().total_issuance();
-    let total_issuance = api.storage().fetch_or_default(&addr, None).await.unwrap();
+    let total_issuance = api
+        .storage()
+        .at(None)
+        .await
+        .unwrap()
+        .fetch_or_default(&addr)
+        .await
+        .unwrap();
     assert_ne!(total_issuance, 0);
 }
 
@@ -274,7 +301,12 @@ async fn storage_balance_lock() -> Result<(), subxt::Error> {
 
     let locks_addr = node_runtime::storage().balances().locks(bob);
 
-    let locks = api.storage().fetch_or_default(&locks_addr, None).await?;
+    let locks = api
+        .storage()
+        .at(None)
+        .await?
+        .fetch_or_default(&locks_addr)
+        .await?;
 
     assert_eq!(
         locks.0,

testing/integration-tests/src/frame/contracts.rs

@@ -222,13 +222,23 @@ async fn tx_call() {
         .contracts()
         .contract_info_of(&contract);
 
-    let contract_info = cxt.client().storage().fetch(&info_addr, None).await;
+    let contract_info = cxt
+        .client()
+        .storage()
+        .at(None)
+        .await
+        .unwrap()
+        .fetch(&info_addr)
+        .await;
     assert!(contract_info.is_ok());
 
     let keys = cxt
         .client()
         .storage()
-        .fetch_keys(&info_addr.to_bytes(), 10, None, None)
+        .at(None)
+        .await
+        .unwrap()
+        .fetch_keys(&info_addr.to_bytes(), 10, None)
         .await
         .unwrap()
         .iter()

testing/integration-tests/src/frame/staking.rs

@@ -150,7 +150,13 @@ async fn chill_works_for_controller_only() -> Result<(), Error> {
         .await?;
 
     let ledger_addr = node_runtime::storage().staking().ledger(alice.account_id());
-    let ledger = api.storage().fetch(&ledger_addr, None).await?.unwrap();
+    let ledger = api
+        .storage()
+        .at(None)
+        .await?
+        .fetch(&ledger_addr)
+        .await?
+        .unwrap();
     assert_eq!(alice_stash.account_id(), &ledger.stash);
 
     let chill_tx = node_runtime::tx().staking().chill();
@@ -232,7 +238,9 @@ async fn storage_current_era() -> Result<(), Error> {
     let current_era_addr = node_runtime::storage().staking().current_era();
     let _current_era = api
         .storage()
-        .fetch(&current_era_addr, None)
+        .at(None)
+        .await?
+        .fetch(&current_era_addr)
         .await?
         .expect("current era always exists");
     Ok(())
@@ -243,7 +251,12 @@ async fn storage_era_reward_points() -> Result<(), Error> {
     let ctx = test_context().await;
     let api = ctx.client();
     let reward_points_addr = node_runtime::storage().staking().eras_reward_points(0);
-    let current_era_result = api.storage().fetch(&reward_points_addr, None).await;
+    let current_era_result = api
+        .storage()
+        .at(None)
+        .await?
+        .fetch(&reward_points_addr)
+        .await;
     assert!(current_era_result.is_ok());
 
     Ok(())

testing/integration-tests/src/frame/system.rs

@@ -24,7 +24,9 @@ async fn storage_account() -> Result<(), subxt::Error> {
 
     let account_info = api
         .storage()
-        .fetch_or_default(&account_info_addr, None)
+        .at(None)
+        .await?
+        .fetch_or_default(&account_info_addr)
         .await;
 
     assert_matches!(account_info, Ok(_));

testing/integration-tests/src/frame/timestamp.rs

@@ -14,7 +14,10 @@ async fn storage_get_current_timestamp() {
 
     let timestamp = api
         .storage()
-        .fetch(&node_runtime::storage().timestamp().now(), None)
+        .at(None)
+        .await
+        .unwrap()
+        .fetch(&node_runtime::storage().timestamp().now())
         .await;
 
     assert!(timestamp.is_ok())

testing/integration-tests/src/storage/mod.rs

@@ -21,7 +21,12 @@ async fn storage_plain_lookup() -> Result<(), subxt::Error> {
     wait_for_blocks(&api).await;
 
     let addr = node_runtime::storage().timestamp().now();
-    let entry = api.storage().fetch_or_default(&addr, None).await?;
+    let entry = api
+        .storage()
+        .at(None)
+        .await?
+        .fetch_or_default(&addr)
+        .await?;
     assert!(entry > 0);
 
     Ok(())
@@ -45,7 +50,12 @@ async fn storage_map_lookup() -> Result<(), subxt::Error> {
 
     // Look up the nonce for the user (we expect it to be 1).
     let nonce_addr = node_runtime::storage().system().account(alice);
-    let entry = api.storage().fetch_or_default(&nonce_addr, None).await?;
+    let entry = api
+        .storage()
+        .at(None)
+        .await?
+        .fetch_or_default(&nonce_addr)
+        .await?;
     assert_eq!(entry.nonce, 1);
 
     Ok(())
@@ -113,7 +123,7 @@ async fn storage_n_map_storage_lookup() -> Result<(), subxt::Error> {
 
     // The actual test; look up this approval in storage:
     let addr = node_runtime::storage().assets().approvals(99, alice, bob);
-    let entry = api.storage().fetch(&addr, None).await?;
+    let entry = api.storage().at(None).await?.fetch(&addr).await?;
     assert_eq!(entry.map(|a| a.amount), Some(123));
     Ok(())
 }