Implement BlocksClient for working with blocks

subxt/src/events/events_client.rs

@@ -11,15 +11,10 @@ use crate::{
         Events,
         FinalizedEventSub,
     },
+    rpc::Rpc,
     Config,
 };
 use derivative::Derivative;
-use futures::{
-    future::Either,
-    stream,
-    Stream,
-    StreamExt,
-};
 use sp_core::{
     storage::StorageKey,
     twox_128,
@@ -181,66 +176,15 @@ where
     let sub = client.rpc().subscribe_finalized_blocks().await?;
 
     // Fill in any gaps between the block above and the finalized blocks reported.
-    let block_subscription = subscribe_to_block_headers_filling_in_gaps(
-        client.clone(),
+    let block_subscription = Rpc::subscribe_to_block_headers_filling_in_gaps(
+        client.rpc().clone(),
         last_finalized_block_number,
         sub,
     );
 
     Ok(EventSubscription::new(client, Box::pin(block_subscription)))
 }
 
-/// Note: This is exposed for testing but is not considered stable and may change
-/// without notice in a patch release.
-#[doc(hidden)]
-pub fn subscribe_to_block_headers_filling_in_gaps<T, Client, S, E>(
-    client: Client,
-    mut last_block_num: Option<u64>,
-    sub: S,
-) -> impl Stream<Item = Result<T::Header, Error>> + Send
-where
-    T: Config,
-    Client: OnlineClientT<T> + Send + Sync,
-    S: Stream<Item = Result<T::Header, E>> + Send,
-    E: Into<Error> + Send + 'static,
-{
-    sub.flat_map(move |s| {
-        let client = client.clone();
-
-        // Get the header, or return a stream containing just the error. Our EventSubscription
-        // stream will return `None` as soon as it hits an error like this.
-        let header = match s {
-            Ok(header) => header,
-            Err(e) => return Either::Left(stream::once(async { Err(e.into()) })),
-        };
-
-        // We want all previous details up to, but not including this current block num.
-        let end_block_num = (*header.number()).into();
-
-        // This is one after the last block we returned details for last time.
-        let start_block_num = last_block_num.map(|n| n + 1).unwrap_or(end_block_num);
-
-        // Iterate over all of the previous blocks we need headers for, ignoring the current block
-        // (which we already have the header info for):
-        let previous_headers = stream::iter(start_block_num..end_block_num)
-            .then(move |n| {
-                let client = client.clone();
-                async move {
-                    let hash = client.rpc().block_hash(Some(n.into())).await?;
-                    let header = client.rpc().header(hash).await?;
-                    Ok::<_, Error>(header)
-                }
-            })
-            .filter_map(|h| async { h.transpose() });
-
-        // On the next iteration, we'll get details starting just after this end block.
-        last_block_num = Some(end_block_num);
-
-        // Return a combination of any previous headers plus the new header.
-        Either::Right(previous_headers.chain(stream::once(async { Ok(header) })))
-    })
-}
-
 // The storage key needed to access events.
 fn system_events_key() -> StorageKey {
     let mut storage_key = twox_128(b"System").to_vec();

subxt/src/events/mod.rs

@@ -16,11 +16,7 @@ pub use event_subscription::{
     EventSubscription,
     FinalizedEventSub,
 };
-pub use events_client::{
-    // Exposed only for testing:
-    subscribe_to_block_headers_filling_in_gaps,
-    EventsClient,
-};
+pub use events_client::EventsClient;
 pub use events_type::{
     EventDetails,
     Events,

subxt/src/rpc/rpc.rs

@@ -56,6 +56,12 @@ use codec::{
     Encode,
 };
 use frame_metadata::RuntimeMetadataPrefixed;
+use futures::{
+    future::Either,
+    stream,
+    Stream,
+    StreamExt,
+};
 use serde::{
     Deserialize,
     Serialize,
@@ -74,6 +80,7 @@ use sp_runtime::{
         Block,
         SignedBlock,
     },
+    traits::Header,
     ApplyExtrinsicResult,
 };
 use std::collections::HashMap;
@@ -569,6 +576,97 @@ impl<T: Config> Rpc<T> {
         Ok(subscription)
     }
 
+    /// Subscribe to finalized blocks and ensure that all missing blocks are filled.
+    ///
+    /// The Substrate's RPC does not guarantee that all finalized blocks are provided.
+    /// This function ensures that all missed blocks are fetched and returned in order.
+    ///
+    /// # Example
+    ///
+    /// The following is a possible scenario where Substrate does not generate an event
+    /// for the second block.
+    ///
+    /// ```sh
+    /// substrate      | Block A | --- | Block B | --- | Block C |
+    /// client         | Block A | ---             --- | Block C |
+    /// ```
+    ///
+    /// This function mitigates this by filling the gaps of missing blocks.
+    ///
+    /// ```sh
+    /// substrate      | Block A | --- | Block B | --- | Block C |
+    /// client         | Block A | --- | Block B | --- | Block C |
+    ///                                ^^^ filled
+    /// ```
+    pub async fn subscribe_finalized_blocks_filled(
+        &self,
+    ) -> Result<impl Stream<Item = Result<T::Header, Error>> + Send + '_, Error> {
+        // Fetch the last finalised block details immediately, so that we'll get
+        // all blocks after this one.
+        let last_finalized_block_hash = self.finalized_head().await?;
+        let last_finalized_block_num = self
+            .header(Some(last_finalized_block_hash))
+            .await?
+            .map(|h| (*h.number()).into());
+
+        let sub = self.subscribe_finalized_blocks().await?;
+
+        // Adjust the subscription stream to fill in any missing blocks.
+        Ok(Rpc::subscribe_to_block_headers_filling_in_gaps(
+            self.clone(),
+            last_finalized_block_num,
+            sub,
+        ))
+    }
+
+    /// Note: This is exposed for testing but is not considered stable and may change
+    /// without notice in a patch release.
+    #[doc(hidden)]
+    pub fn subscribe_to_block_headers_filling_in_gaps<S, E>(
+        client: Rpc<T>,
+        mut last_block_num: Option<u64>,
+        sub: S,
+    ) -> impl Stream<Item = Result<T::Header, Error>> + Send
+    where
+        S: Stream<Item = Result<T::Header, E>> + Send,
+        E: Into<Error> + Send + 'static,
+    {
+        sub.flat_map(move |s| {
+            let rpc = client.clone();
+
+            // Get the header, or return a stream containing just the error.
+            let header = match s {
+                Ok(header) => header,
+                Err(e) => return Either::Left(stream::once(async { Err(e.into()) })),
+            };
+
+            // We want all previous details up to, but not including this current block num.
+            let end_block_num = (*header.number()).into();
+
+            // This is one after the last block we returned details for last time.
+            let start_block_num = last_block_num.map(|n| n + 1).unwrap_or(end_block_num);
+
+            // Iterate over all of the previous blocks we need headers for, ignoring the current block
+            // (which we already have the header info for):
+            let previous_headers = stream::iter(start_block_num..end_block_num)
+                .then(move |n| {
+                    let rpc = rpc.clone();
+                    async move {
+                        let hash = rpc.block_hash(Some(n.into())).await?;
+                        let header = rpc.header(hash).await?;
+                        Ok::<_, Error>(header)
+                    }
+                })
+                .filter_map(|h| async { h.transpose() });
+
+            // On the next iteration, we'll get details starting just after this end block.
+            last_block_num = Some(end_block_num);
+
+            // Return a combination of any previous headers plus the new header.
+            Either::Right(previous_headers.chain(stream::once(async { Ok(header) })))
+        })
+    }
+
     /// Subscribe to runtime version updates that produce changes in the metadata.
     pub async fn subscribe_runtime_version(
         &self,

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

@@ -11,13 +11,17 @@ use crate::{
         wait_for_blocks,
     },
 };
+use futures::StreamExt;
 use sp_core::{
     sr25519::Pair as Sr25519Pair,
     storage::well_known_keys,
     Pair,
 };
 use sp_keyring::AccountKeyring;
-use subxt::error::DispatchError;
+use subxt::{
+    error::DispatchError,
+    rpc::Rpc,
+};
 
 #[tokio::test]
 async fn insert_key() {
@@ -91,6 +95,51 @@ async fn chain_subscribe_finalized_blocks() {
     blocks.next().await.unwrap().unwrap();
 }
 
+#[tokio::test]
+async fn missing_block_headers_will_be_filled_in() -> Result<(), subxt::Error> {
+    let ctx = test_context().await;
+    let api = ctx.client();
+
+    // Manually subscribe to the next 6 finalized block headers, but deliberately
+    // filter out some in the middle so we get back b _ _ b _ b. This guarantees
+    // that there will be some gaps, even if there aren't any from the subscription.
+    let some_finalized_blocks = api
+        .rpc()
+        .subscribe_finalized_blocks()
+        .await?
+        .enumerate()
+        .take(6)
+        .filter(|(n, _)| {
+            let n = *n;
+            async move { n == 0 || n == 3 || n == 5 }
+        })
+        .map(|(_, h)| h);
+
+    // This should spot any gaps in the middle and fill them back in.
+    let all_finalized_blocks = Rpc::subscribe_to_block_headers_filling_in_gaps(
+        ctx.client().rpc().clone(),
+        None,
+        some_finalized_blocks,
+    );
+    futures::pin_mut!(all_finalized_blocks);
+
+    // Iterate the block headers, making sure we get them all in order.
+    let mut last_block_number = None;
+    while let Some(header) = all_finalized_blocks.next().await {
+        let header = header?;
+
+        use sp_runtime::traits::Header;
+        let block_number: u128 = (*header.number()).into();
+
+        if let Some(last) = last_block_number {
+            assert_eq!(last + 1, block_number);
+        }
+        last_block_number = Some(block_number);
+    }
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn fetch_keys() {
     let ctx = test_context().await;

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

@@ -169,57 +169,6 @@ async fn balance_transfer_subscription() -> Result<(), subxt::Error> {
     Ok(())
 }
 
-#[tokio::test]
-async fn missing_block_headers_will_be_filled_in() -> Result<(), subxt::Error> {
-    let ctx = test_context().await;
-    let api = ctx.client();
-
-    // This function is not publically available to use, but contains
-    // the key logic for filling in missing blocks, so we want to test it.
-    // This is used in `subscribe_finalized` to ensure no block headers are
-    // missed.
-    use subxt::events::subscribe_to_block_headers_filling_in_gaps;
-
-    // Manually subscribe to the next 6 finalized block headers, but deliberately
-    // filter out some in the middle so we get back b _ _ b _ b. This guarantees
-    // that there will be some gaps, even if there aren't any from the subscription.
-    let some_finalized_blocks = api
-        .rpc()
-        .subscribe_finalized_blocks()
-        .await?
-        .enumerate()
-        .take(6)
-        .filter(|(n, _)| {
-            let n = *n;
-            async move { n == 0 || n == 3 || n == 5 }
-        })
-        .map(|(_, h)| h);
-
-    // This should spot any gaps in the middle and fill them back in.
-    let all_finalized_blocks = subscribe_to_block_headers_filling_in_gaps(
-        ctx.client(),
-        None,
-        some_finalized_blocks,
-    );
-    futures::pin_mut!(all_finalized_blocks);
-
-    // Iterate the block headers, making sure we get them all in order.
-    let mut last_block_number = None;
-    while let Some(header) = all_finalized_blocks.next().await {
-        let header = header?;
-
-        use sp_runtime::traits::Header;
-        let block_number: u128 = (*header.number()).into();
-
-        if let Some(last) = last_block_number {
-            assert_eq!(last + 1, block_number);
-        }
-        last_block_number = Some(block_number);
-    }
-
-    Ok(())
-}
-
 // This is just a compile-time check that we can subscribe to events in
 // a context that requires the event subscription/filtering to be Send-able.
 // We test a typical use of EventSubscription and FilterEvents. We don't need