rpc: add full support reconnecting rpc client

Cargo.toml

@@ -113,6 +113,7 @@ which = "5.0.0"
 strip-ansi-escapes = "0.2.0"
 proptest = "1.4.0"
 hex-literal = "0.4.1"
+pin-project-lite = "0.2.14"
 
 # Light client support:
 smoldot = { version = "0.16.0", default-features = false }

subxt/Cargo.toml

@@ -85,6 +85,7 @@ tracing = { workspace = true }
 frame-metadata = { workspace = true }
 either = { workspace = true }
 instant = { workspace = true }
+pin-project-lite = { workspace = true }
 
 # Provides some deserialization, types like U256/H256 and hashing impls like twox/blake256:
 impl-serde = { workspace = true }

subxt/src/backend/unstable/follow_stream_driver.rs

@@ -5,8 +5,9 @@
 use super::follow_stream_unpin::{BlockRef, FollowStreamMsg, FollowStreamUnpin};
 use crate::backend::unstable::rpc_methods::{FollowEvent, Initialized, RuntimeEvent};
 use crate::config::BlockHash;
-use crate::error::Error;
+use crate::error::{Error, RpcError};
 use futures::stream::{Stream, StreamExt};
+use pin_project_lite::pin_project;
 use std::collections::{HashMap, HashSet, VecDeque};
 use std::ops::DerefMut;
 use std::pin::Pin;
@@ -379,6 +380,77 @@ struct SubscriberDetails<Hash: BlockHash> {
     waker: Option<Waker>,
 }
 
+pin_project! {
+    /// A stream that subscribes to finalized blocks
+    /// and indicates whether a block was missed if the connection was lost.
+    #[derive(Debug)]
+    pub struct FollowStreamFinalizedHeads<Hash: BlockHash> {
+        #[pin]
+        stream: FollowStreamDriverSubscription<Hash>,
+        sub_id: Option<String>,
+        last_seen_block: Option<BlockRef<Hash>>,
+    }
+}
+
+impl<Hash: BlockHash> FollowStreamFinalizedHeads<Hash> {
+    pub fn new(stream: FollowStreamDriverSubscription<Hash>) -> Self {
+        Self {
+            stream,
+            sub_id: None,
+            last_seen_block: None,
+        }
+    }
+}
+
+impl<Hash: BlockHash> Stream for FollowStreamFinalizedHeads<Hash> {
+    type Item = Result<(String, Vec<BlockRef<Hash>>), Error>;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let mut this = self.project();
+
+        loop {
+            match futures::ready!(this.stream.poll_next_unpin(cx)) {
+                Some(FollowStreamMsg::Ready(sub_id)) => {
+                    *this.sub_id = Some(sub_id);
+                }
+                Some(FollowStreamMsg::Event(FollowEvent::Finalized(finalized))) => {
+                    *this.last_seen_block = finalized.finalized_block_hashes.last().cloned();
+                    let sub_id = this
+                        .sub_id
+                        .clone()
+                        .expect("Ready is always emitted before Finalized; qed");
+                    return Poll::Ready(Some(Ok((
+                        sub_id,
+                        finalized.finalized_block_hashes.clone(),
+                    ))));
+                }
+                Some(FollowStreamMsg::Event(FollowEvent::Initialized(init))) => {
+                    let prev = this.last_seen_block.take();
+                    *this.last_seen_block = init.finalized_block_hashes.last().cloned();
+
+                    if let Some(p) = prev {
+                        if !init.finalized_block_hashes.contains(&p) {
+                            return Poll::Ready(Some(Err(RpcError::DisconnectedWillReconnect(
+                                "Missed at least one block when the connection was lost".to_owned(),
+                            )
+                            .into())));
+                        }
+                    }
+
+                    let sub_id = this
+                        .sub_id
+                        .clone()
+                        .expect("Ready is always emitted before Initialized; qed");
+
+                    return Poll::Ready(Some(Ok((sub_id, init.finalized_block_hashes))));
+                }
+                // Ignore other events.
+                _ => (),
+            };
+        }
+    }
+}
+
 #[cfg(test)]
 mod test_utils {
     use super::super::follow_stream_unpin::test_utils::test_unpin_stream_getter;

subxt/src/backend/unstable/follow_stream_unpin.rs

@@ -474,7 +474,7 @@ pub(super) mod test_utils {
 
     pub type UnpinRx<Hash> = std::sync::mpsc::Receiver<(Hash, Arc<str>)>;
 
-    /// Get a `FolowStreamUnpin` from an iterator over events.
+    /// Get a [`FollowStreamUnpin`] from an iterator over events.
     pub fn test_unpin_stream_getter<Hash, F, I>(
         events: F,
         max_life: usize,

subxt/src/backend/unstable/mod.rs

@@ -18,6 +18,7 @@ mod storage_items;
 
 pub mod rpc_methods;
 
+use self::follow_stream_driver::FollowStreamFinalizedHeads;
 use self::follow_stream_unpin::FollowStreamMsg;
 use self::rpc_methods::{
     FollowEvent, MethodResponse, RuntimeEvent, StorageQuery, StorageQueryType, StorageResultType,
@@ -31,6 +32,7 @@ use crate::error::{Error, RpcError};
 use crate::Config;
 use async_trait::async_trait;
 use follow_stream_driver::{FollowStreamDriver, FollowStreamDriverHandle};
+use futures::future::Either;
 use futures::{Stream, StreamExt};
 use std::collections::HashMap;
 use std::task::Poll;
@@ -496,15 +498,40 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for UnstableBackend<T> {
     async fn stream_finalized_block_headers(
         &self,
     ) -> Result<StreamOfResults<(T::Header, BlockRef<T::Hash>)>, Error> {
-        let stream = self
-            .stream_headers(|ev| match ev {
-                FollowEvent::Initialized(init) => init.finalized_block_hashes,
-                FollowEvent::Finalized(ev) => ev.finalized_block_hashes,
-                _ => vec![],
-            })
-            .await?;
+        let methods = self.methods.clone();
 
-        Ok(StreamOf::new(Box::pin(stream)))
+        let headers =
+            FollowStreamFinalizedHeads::new(self.follow_handle.subscribe()).flat_map(move |r| {
+                let methods = methods.clone();
+
+                let (sub_id, block_refs) = match r {
+                    Ok(ev) => ev,
+                    Err(e) => return Either::Left(futures::stream::once(async { Err(e) })),
+                };
+
+                Either::Right(
+                    futures::stream::iter(block_refs).filter_map(move |block_ref| {
+                        let methods = methods.clone();
+                        let sub_id = sub_id.clone();
+
+                        async move {
+                            let res = methods
+                                .chainhead_v1_header(&sub_id, block_ref.hash())
+                                .await
+                                .transpose()?;
+
+                            let header = match res {
+                                Ok(header) => header,
+                                Err(e) => return Some(Err(e)),
+                            };
+
+                            Some(Ok((header, block_ref.into())))
+                        }
+                    }),
+                )
+            });
+
+        Ok(StreamOf::new(Box::pin(headers)))
     }
 
     async fn submit_transaction(