rpc: add full support reconnecting rpc client (#1505)

* add simple reconnecting rpc client * initial retryable calls * add reconnecting backend * add reconnecting example for unstable backend * add todo what isn't working * FollowStream: restart on reconn * naive fix: fetch sub_id in stream_headers * cleanup * remove resubscribe APIs * cleanup and remove many wrapper streams * remove retry backend * legacy rpc: make it retryable * unstable rpc: make it retryable * fix nits * support wasm as well * remove deadcode * address grumbles * revert rpc methods * don't create a subscription per block * get rid off retry logic in subxt rpc * Update subxt/Cargo.toml * Update subxt/src/backend/legacy/mod.rs * Update subxt/src/backend/legacy/mod.rs * remove outdated comments * fix bad merge * Fix reconnecting RPC client and update dependencies * add back retry logic and remove `finito` * fix nits * cleanup * add hack for race when reconnecting * backend: emit Stop event DisconnectWillRecoonect * merge reconnecting client examples * add fn retry_stream * cleanup * add all features from reconnecting-rpc-client * fix build * remove needless retry for fetch_storage * StorageFetchDescendantKeysStream handle disconnect err * dont retry transactions * fetch subscription ID from FollowStreamMsg * fix nits * Update subxt/src/backend/legacy/mod.rs * Update subxt/src/backend/legacy/mod.rs * add reconn to StorageItems stream * StorageFetchDescendantKeysStreamchore: retry storage call * RetryStream: emit DisconnectWillReconnect msg * runtime subscriptions ignore DisconnectWillReconn * Update subxt/examples/setup_reconnecting_rpc_client.rs * Update subxt/src/client/online_client.rs * Update subxt/src/client/online_client.rs * Add custom stream wrapper for finalized blocks * add missing retry block * clippy * clippy again * cleanup * remove duplicate logic * fix more grumbles * Update subxt/examples/setup_reconnecting_rpc_client.rs Co-authored-by: James Wilson <james@jsdw.me> * simplify the example * remove pin-project dep * remove duplicate retry logic * remove extra code * specify trait bounds for retry api * simplify the example * fix weird Poll::Pending return * fix nit in poll impl * remove needless paths * make retry_stream pub and add doc examples * Update subxt/src/backend/utils.rs --------- Co-authored-by: James Wilson <james@jsdw.me>
2026-07-03 00:37:23 +00:00 · 2024-05-08 15:12:54 +02:00
parent f034ac486c
commit bec896d91a
19 changed files with 1231 additions and 363 deletions
@@ -148,6 +148,12 @@ impl<Hash> Stream for FollowStream<Hash> {
                            continue;
                        }
                        Poll::Ready(Err(e)) => {
+                            // Re-start if a reconnecting backend was enabled.
+                            if e.is_disconnected_will_reconnect() {
+                                this.stream = InnerStreamState::Stopped;
+                                continue;
+                            }
+
                            // Finish forever if there's an error, passing it on.
                            this.stream = InnerStreamState::Finished;
                            return Poll::Ready(Some(Err(e)));
@@ -182,6 +188,12 @@ impl<Hash> Stream for FollowStream<Hash> {
                            return Poll::Ready(Some(Ok(FollowStreamMsg::Event(ev))));
                        }
                        Poll::Ready(Some(Err(e))) => {
+                            // Re-start if a reconnecting backend was enabled.
+                            if e.is_disconnected_will_reconnect() {
+                                this.stream = InnerStreamState::Stopped;
+                                continue;
+                            }
+
                            // Finish forever if there's an error, passing it on.
                            this.stream = InnerStreamState::Finished;
                            return Poll::Ready(Some(Err(e)));
@@ -5,7 +5,7 @@
 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 std::collections::{HashMap, HashSet, VecDeque};
 use std::ops::DerefMut;
@@ -380,6 +380,103 @@ struct SubscriberDetails<Hash: BlockHash> {
    waker: Option<Waker>,
 }

+/// A stream that subscribes to finalized blocks
+/// and indicates whether a block was missed if was restarted.
+#[derive(Debug)]
+pub struct FollowStreamFinalizedHeads<Hash: BlockHash, F> {
+    stream: FollowStreamDriverSubscription<Hash>,
+    sub_id: Option<String>,
+    last_seen_block: Option<BlockRef<Hash>>,
+    f: F,
+    is_done: bool,
+}
+
+impl<Hash: BlockHash, F> Unpin for FollowStreamFinalizedHeads<Hash, F> {}
+
+impl<Hash, F> FollowStreamFinalizedHeads<Hash, F>
+where
+    Hash: BlockHash,
+    F: Fn(FollowEvent<BlockRef<Hash>>) -> Vec<BlockRef<Hash>>,
+{
+    pub fn new(stream: FollowStreamDriverSubscription<Hash>, f: F) -> Self {
+        Self {
+            stream,
+            sub_id: None,
+            last_seen_block: None,
+            f,
+            is_done: false,
+        }
+    }
+}
+
+impl<Hash, F> Stream for FollowStreamFinalizedHeads<Hash, F>
+where
+    Hash: BlockHash,
+    F: Fn(FollowEvent<BlockRef<Hash>>) -> Vec<BlockRef<Hash>>,
+{
+    type Item = Result<(String, Vec<BlockRef<Hash>>), Error>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        if self.is_done {
+            return Poll::Ready(None);
+        }
+
+        loop {
+            let Some(ev) = futures::ready!(self.stream.poll_next_unpin(cx)) else {
+                self.is_done = true;
+                return Poll::Ready(None);
+            };
+
+            let block_refs = match ev {
+                FollowStreamMsg::Ready(sub_id) => {
+                    self.sub_id = Some(sub_id);
+                    continue;
+                }
+                FollowStreamMsg::Event(FollowEvent::Finalized(finalized)) => {
+                    self.last_seen_block = finalized.finalized_block_hashes.last().cloned();
+
+                    (self.f)(FollowEvent::Finalized(finalized))
+                }
+                FollowStreamMsg::Event(FollowEvent::Initialized(mut init)) => {
+                    let prev = self.last_seen_block.take();
+                    self.last_seen_block = init.finalized_block_hashes.last().cloned();
+
+                    if let Some(p) = prev {
+                        let Some(pos) = init
+                            .finalized_block_hashes
+                            .iter()
+                            .position(|b| b.hash() == p.hash())
+                        else {
+                            return Poll::Ready(Some(Err(RpcError::DisconnectedWillReconnect(
+                                "Missed at least one block when the connection was lost".to_owned(),
+                            )
+                            .into())));
+                        };
+
+                        // If we got older blocks than `prev`, we need to remove them
+                        // because they should already have been sent at this point.
+                        init.finalized_block_hashes.drain(0..=pos);
+                    }
+
+                    (self.f)(FollowEvent::Initialized(init))
+                }
+                FollowStreamMsg::Event(ev) => (self.f)(ev),
+            };
+
+            if block_refs.is_empty() {
+                continue;
+            }
+
+            let sub_id = self
+                .sub_id
+                .clone()
+                .expect("Ready is always emitted before any other event");
+
+            return Poll::Ready(Some(Ok((sub_id, block_refs))));
+        }
+    }
+}
+
 #[cfg(test)]
 mod test_utils {
    use super::super::follow_stream_unpin::test_utils::test_unpin_stream_getter;
@@ -402,6 +499,9 @@ mod test_utils {

 #[cfg(test)]
 mod test {
+    use futures::TryStreamExt;
+    use sp_core::H256;
+
    use super::super::follow_stream::test_utils::{
        ev_best_block, ev_finalized, ev_initialized, ev_new_block,
    };
@@ -545,4 +645,101 @@ mod test {
        ];
        assert_eq!(evs, expected);
    }
+
+    #[tokio::test]
+    async fn subscribe_finalized_blocks_restart_works() {
+        let mut driver = test_follow_stream_driver_getter(
+            || {
+                [
+                    Ok(ev_initialized(0)),
+                    Ok(ev_new_block(0, 1)),
+                    Ok(ev_best_block(1)),
+                    Ok(ev_finalized([1], [])),
+                    Ok(FollowEvent::Stop),
+                    Ok(ev_initialized(1)),
+                    Ok(ev_finalized([2], [])),
+                    Err(Error::Other("ended".to_owned())),
+                ]
+            },
+            10,
+        );
+
+        let handle = driver.handle();
+
+        tokio::spawn(async move { while driver.next().await.is_some() {} });
+
+        let f = |ev| match ev {
+            FollowEvent::Finalized(ev) => ev.finalized_block_hashes,
+            FollowEvent::Initialized(ev) => ev.finalized_block_hashes,
+            _ => vec![],
+        };
+
+        let stream = FollowStreamFinalizedHeads::new(handle.subscribe(), f);
+        let evs: Vec<_> = stream.try_collect().await.unwrap();
+
+        let expected = vec![
+            (
+                "sub_id_0".to_string(),
+                vec![BlockRef::new(H256::from_low_u64_le(0))],
+            ),
+            (
+                "sub_id_0".to_string(),
+                vec![BlockRef::new(H256::from_low_u64_le(1))],
+            ),
+            (
+                "sub_id_5".to_string(),
+                vec![BlockRef::new(H256::from_low_u64_le(2))],
+            ),
+        ];
+        assert_eq!(evs, expected);
+    }
+
+    #[tokio::test]
+    async fn subscribe_finalized_blocks_restart_with_missed_blocks() {
+        let mut driver = test_follow_stream_driver_getter(
+            || {
+                [
+                    Ok(ev_initialized(0)),
+                    Ok(FollowEvent::Stop),
+                    // Emulate that we missed some blocks.
+                    Ok(ev_initialized(13)),
+                    Ok(ev_finalized([14], [])),
+                    Err(Error::Other("ended".to_owned())),
+                ]
+            },
+            10,
+        );
+
+        let handle = driver.handle();
+
+        tokio::spawn(async move { while driver.next().await.is_some() {} });
+
+        let f = |ev| match ev {
+            FollowEvent::Finalized(ev) => ev.finalized_block_hashes,
+            FollowEvent::Initialized(ev) => ev.finalized_block_hashes,
+            _ => vec![],
+        };
+
+        let evs: Vec<_> = FollowStreamFinalizedHeads::new(handle.subscribe(), f)
+            .collect()
+            .await;
+
+        assert_eq!(
+            evs[0].as_ref().unwrap(),
+            &(
+                "sub_id_0".to_string(),
+                vec![BlockRef::new(H256::from_low_u64_le(0))]
+            )
+        );
+        assert!(
+            matches!(&evs[1], Err(Error::Rpc(RpcError::DisconnectedWillReconnect(e))) if e.contains("Missed at least one block when the connection was lost"))
+        );
+        assert_eq!(
+            evs[2].as_ref().unwrap(),
+            &(
+                "sub_id_2".to_string(),
+                vec![BlockRef::new(H256::from_low_u64_le(14))]
+            )
+        );
+    }
 }
@@ -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,
@@ -18,21 +18,22 @@ mod storage_items;

 pub mod rpc_methods;

+use self::follow_stream_driver::FollowStreamFinalizedHeads;
 use self::rpc_methods::{
    FollowEvent, MethodResponse, RuntimeEvent, StorageQuery, StorageQueryType, StorageResultType,
 };
 use crate::backend::{
-    rpc::RpcClient, Backend, BlockRef, BlockRefT, RuntimeVersion, StorageResponse, StreamOf,
-    StreamOfResults, TransactionStatus,
+    rpc::RpcClient, utils::retry, Backend, BlockRef, BlockRefT, RuntimeVersion, StorageResponse,
+    StreamOf, StreamOfResults, TransactionStatus,
 };
 use crate::config::BlockHash;
 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::sync::Arc;
 use std::task::Poll;
 use storage_items::StorageItems;

@@ -136,43 +137,50 @@ impl<T: Config> UnstableBackend<T> {
    }

    /// Stream block headers based on the provided filter fn
-    async fn stream_headers<F, I>(
+    async fn stream_headers<F>(
        &self,
        f: F,
    ) -> Result<StreamOfResults<(T::Header, BlockRef<T::Hash>)>, Error>
    where
-        F: Fn(FollowEvent<follow_stream_unpin::BlockRef<T::Hash>>) -> I + Copy + Send + 'static,
-        I: IntoIterator<Item = follow_stream_unpin::BlockRef<T::Hash>> + Send + 'static,
-        <I as IntoIterator>::IntoIter: Send,
+        F: Fn(
+                FollowEvent<follow_stream_unpin::BlockRef<T::Hash>>,
+            ) -> Vec<follow_stream_unpin::BlockRef<T::Hash>>
+            + Send
+            + Sync
+            + 'static,
    {
-        let sub_id = get_subscription_id(&self.follow_handle).await?;
-        let sub_id = Arc::new(sub_id);
        let methods = self.methods.clone();
-        let headers = self.follow_handle.subscribe().events().flat_map(move |ev| {
-            let sub_id = sub_id.clone();
-            let methods = methods.clone();

-            let block_refs = f(ev).into_iter();
-
-            futures::stream::iter(block_refs).filter_map(move |block_ref| {
-                let sub_id = sub_id.clone();
+        let headers =
+            FollowStreamFinalizedHeads::new(self.follow_handle.subscribe(), f).flat_map(move |r| {
                let methods = methods.clone();

-                async move {
-                    let res = methods
-                        .chainhead_v1_header(&sub_id, block_ref.hash())
-                        .await
-                        .transpose()?;
+                let (sub_id, block_refs) = match r {
+                    Ok(ev) => ev,
+                    Err(e) => return Either::Left(futures::stream::once(async { Err(e) })),
+                };

-                    let header = match res {
-                        Ok(header) => header,
-                        Err(e) => return Some(Err(e)),
-                    };
+                Either::Right(
+                    futures::stream::iter(block_refs).filter_map(move |block_ref| {
+                        let methods = methods.clone();
+                        let sub_id = sub_id.clone();

-                    Some(Ok((header, block_ref.into())))
-                }
-            })
-        });
+                        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(Box::pin(headers)))
    }
@@ -194,31 +202,34 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for UnstableBackend<T> {
        keys: Vec<Vec<u8>>,
        at: T::Hash,
    ) -> Result<StreamOfResults<StorageResponse>, Error> {
-        let queries = keys.iter().map(|key| StorageQuery {
-            key: &**key,
-            query_type: StorageQueryType::Value,
-        });
+        retry(|| async {
+            let queries = keys.iter().map(|key| StorageQuery {
+                key: &**key,
+                query_type: StorageQueryType::Value,
+            });

-        let storage_items =
-            StorageItems::from_methods(queries, at, &self.follow_handle, self.methods.clone())
-                .await?;
+            let storage_items =
+                StorageItems::from_methods(queries, at, &self.follow_handle, self.methods.clone())
+                    .await?;

-        let storage_result_stream = storage_items.filter_map(|val| async move {
-            let val = match val {
-                Ok(val) => val,
-                Err(e) => return Some(Err(e)),
-            };
+            let stream = storage_items.filter_map(|val| async move {
+                let val = match val {
+                    Ok(val) => val,
+                    Err(e) => return Some(Err(e)),
+                };

-            let StorageResultType::Value(result) = val.result else {
-                return None;
-            };
-            Some(Ok(StorageResponse {
-                key: val.key.0,
-                value: result.0,
-            }))
-        });
+                let StorageResultType::Value(result) = val.result else {
+                    return None;
+                };
+                Some(Ok(StorageResponse {
+                    key: val.key.0,
+                    value: result.0,
+                }))
+            });

-        Ok(StreamOf(Box::pin(storage_result_stream)))
+            Ok(StreamOf(Box::pin(stream)))
+        })
+        .await
    }

    async fn storage_fetch_descendant_keys(
@@ -226,22 +237,25 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for UnstableBackend<T> {
        key: Vec<u8>,
        at: T::Hash,
    ) -> Result<StreamOfResults<Vec<u8>>, Error> {
-        // Ask for hashes, and then just ignore them and return the keys that come back.
-        let query = StorageQuery {
-            key: &*key,
-            query_type: StorageQueryType::DescendantsHashes,
-        };
+        retry(|| async {
+            // Ask for hashes, and then just ignore them and return the keys that come back.
+            let query = StorageQuery {
+                key: &*key,
+                query_type: StorageQueryType::DescendantsHashes,
+            };

-        let storage_items = StorageItems::from_methods(
-            std::iter::once(query),
-            at,
-            &self.follow_handle,
-            self.methods.clone(),
-        )
-        .await?;
+            let storage_items = StorageItems::from_methods(
+                std::iter::once(query),
+                at,
+                &self.follow_handle,
+                self.methods.clone(),
+            )
+            .await?;

-        let storage_result_stream = storage_items.map(|val| val.map(|v| v.key.0));
-        Ok(StreamOf(Box::pin(storage_result_stream)))
+            let storage_result_stream = storage_items.map(|val| val.map(|v| v.key.0));
+            Ok(StreamOf(Box::pin(storage_result_stream)))
+        })
+        .await
    }

    async fn storage_fetch_descendant_values(
@@ -249,72 +263,81 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for UnstableBackend<T> {
        key: Vec<u8>,
        at: T::Hash,
    ) -> Result<StreamOfResults<StorageResponse>, Error> {
-        let query = StorageQuery {
-            key: &*key,
-            query_type: StorageQueryType::DescendantsValues,
-        };
-
-        let storage_items = StorageItems::from_methods(
-            std::iter::once(query),
-            at,
-            &self.follow_handle,
-            self.methods.clone(),
-        )
-        .await?;
-
-        let storage_result_stream = storage_items.filter_map(|val| async move {
-            let val = match val {
-                Ok(val) => val,
-                Err(e) => return Some(Err(e)),
+        retry(|| async {
+            let query = StorageQuery {
+                key: &*key,
+                query_type: StorageQueryType::DescendantsValues,
            };

-            let StorageResultType::Value(result) = val.result else {
-                return None;
-            };
-            Some(Ok(StorageResponse {
-                key: val.key.0,
-                value: result.0,
-            }))
-        });
+            let storage_items = StorageItems::from_methods(
+                std::iter::once(query),
+                at,
+                &self.follow_handle,
+                self.methods.clone(),
+            )
+            .await?;

-        Ok(StreamOf(Box::pin(storage_result_stream)))
+            let storage_result_stream = storage_items.filter_map(|val| async move {
+                let val = match val {
+                    Ok(val) => val,
+                    Err(e) => return Some(Err(e)),
+                };
+
+                let StorageResultType::Value(result) = val.result else {
+                    return None;
+                };
+                Some(Ok(StorageResponse {
+                    key: val.key.0,
+                    value: result.0,
+                }))
+            });
+
+            Ok(StreamOf(Box::pin(storage_result_stream)))
+        })
+        .await
    }

    async fn genesis_hash(&self) -> Result<T::Hash, Error> {
-        self.methods.chainspec_v1_genesis_hash().await
+        retry(|| self.methods.chainspec_v1_genesis_hash()).await
    }

    async fn block_header(&self, at: T::Hash) -> Result<Option<T::Header>, Error> {
-        let sub_id = get_subscription_id(&self.follow_handle).await?;
-        self.methods.chainhead_v1_header(&sub_id, at).await
+        retry(|| async {
+            let sub_id = get_subscription_id(&self.follow_handle).await?;
+            self.methods.chainhead_v1_header(&sub_id, at).await
+        })
+        .await
    }

    async fn block_body(&self, at: T::Hash) -> Result<Option<Vec<Vec<u8>>>, Error> {
-        let sub_id = get_subscription_id(&self.follow_handle).await?;
+        retry(|| async {
+            let sub_id = get_subscription_id(&self.follow_handle).await?;

-        // Subscribe to the body response and get our operationId back.
-        let follow_events = self.follow_handle.subscribe().events();
-        let status = self.methods.chainhead_v1_body(&sub_id, at).await?;
-        let operation_id = match status {
-            MethodResponse::LimitReached => {
-                return Err(RpcError::request_rejected("limit reached").into())
-            }
-            MethodResponse::Started(s) => s.operation_id,
-        };
-
-        // Wait for the response to come back with the correct operationId.
-        let mut exts_stream = follow_events.filter_map(|ev| {
-            let FollowEvent::OperationBodyDone(body) = ev else {
-                return std::future::ready(None);
+            // Subscribe to the body response and get our operationId back.
+            let follow_events = self.follow_handle.subscribe().events();
+            let status = self.methods.chainhead_v1_body(&sub_id, at).await?;
+            let operation_id = match status {
+                MethodResponse::LimitReached => {
+                    return Err(RpcError::request_rejected("limit reached").into())
+                }
+                MethodResponse::Started(s) => s.operation_id,
            };
-            if body.operation_id != operation_id {
-                return std::future::ready(None);
-            }
-            let exts: Vec<_> = body.value.into_iter().map(|ext| ext.0).collect();
-            std::future::ready(Some(exts))
-        });

-        Ok(exts_stream.next().await)
+            // Wait for the response to come back with the correct operationId.
+            let mut exts_stream = follow_events.filter_map(|ev| {
+                let FollowEvent::OperationBodyDone(body) = ev else {
+                    return std::future::ready(None);
+                };
+                if body.operation_id != operation_id {
+                    return std::future::ready(None);
+                }
+                let exts: Vec<_> = body.value.into_iter().map(|ext| ext.0).collect();
+                std::future::ready(Some(exts))
+            });
+
+            Ok(exts_stream.next().await)
+        })
+        .await
    }

    async fn latest_finalized_block_ref(&self) -> Result<BlockRef<T::Hash>, Error> {
@@ -423,12 +446,16 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for UnstableBackend<T> {
                std::future::ready(Some(Ok(runtime_version)))
            });

-        Ok(StreamOf(Box::pin(runtime_stream)))
+        Ok(StreamOf::new(Box::pin(runtime_stream)))
    }

    async fn stream_all_block_headers(
        &self,
    ) -> Result<StreamOfResults<(T::Header, BlockRef<T::Hash>)>, Error> {
+        // TODO: https://github.com/paritytech/subxt/issues/1568
+        //
+        // It's possible that blocks may be silently missed if
+        // a reconnection occurs because it's restarted by the unstable backend.
        self.stream_headers(|ev| match ev {
            FollowEvent::Initialized(init) => init.finalized_block_hashes,
            FollowEvent::NewBlock(ev) => {
@@ -442,6 +469,10 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for UnstableBackend<T> {
    async fn stream_best_block_headers(
        &self,
    ) -> Result<StreamOfResults<(T::Header, BlockRef<T::Hash>)>, Error> {
+        // TODO: https://github.com/paritytech/subxt/issues/1568
+        //
+        // It's possible that blocks may be silently missed if
+        // a reconnection occurs because it's restarted by the unstable backend.
        self.stream_headers(|ev| match ev {
            FollowEvent::Initialized(init) => init.finalized_block_hashes,
            FollowEvent::BestBlockChanged(ev) => vec![ev.best_block_hash],
@@ -638,37 +669,40 @@ impl<T: Config + Send + Sync + 'static> Backend<T> for UnstableBackend<T> {
        call_parameters: Option<&[u8]>,
        at: T::Hash,
    ) -> Result<Vec<u8>, Error> {
-        let sub_id = get_subscription_id(&self.follow_handle).await?;
+        retry(|| async {
+            let sub_id = get_subscription_id(&self.follow_handle).await?;

-        // Subscribe to the body response and get our operationId back.
-        let follow_events = self.follow_handle.subscribe().events();
-        let call_parameters = call_parameters.unwrap_or(&[]);
-        let status = self
-            .methods
-            .chainhead_v1_call(&sub_id, at, method, call_parameters)
-            .await?;
-        let operation_id = match status {
-            MethodResponse::LimitReached => {
-                return Err(RpcError::request_rejected("limit reached").into())
-            }
-            MethodResponse::Started(s) => s.operation_id,
-        };
-
-        // Wait for the response to come back with the correct operationId.
-        let mut call_data_stream = follow_events.filter_map(|ev| {
-            let FollowEvent::OperationCallDone(body) = ev else {
-                return std::future::ready(None);
+            // Subscribe to the body response and get our operationId back.
+            let follow_events = self.follow_handle.subscribe().events();
+            let call_parameters = call_parameters.unwrap_or(&[]);
+            let status = self
+                .methods
+                .chainhead_v1_call(&sub_id, at, method, call_parameters)
+                .await?;
+            let operation_id = match status {
+                MethodResponse::LimitReached => {
+                    return Err(RpcError::request_rejected("limit reached").into())
+                }
+                MethodResponse::Started(s) => s.operation_id,
            };
-            if body.operation_id != operation_id {
-                return std::future::ready(None);
-            }
-            std::future::ready(Some(body.output.0))
-        });

-        call_data_stream
-            .next()
-            .await
-            .ok_or_else(|| RpcError::SubscriptionDropped.into())
+            // Wait for the response to come back with the correct operationId.
+            let mut call_data_stream = follow_events.filter_map(|ev| {
+                let FollowEvent::OperationCallDone(body) = ev else {
+                    return std::future::ready(None);
+                };
+                if body.operation_id != operation_id {
+                    return std::future::ready(None);
+                }
+                std::future::ready(Some(body.output.0))
+            });
+
+            call_data_stream
+                .next()
+                .await
+                .ok_or_else(|| RpcError::SubscriptionDropped.into())
+        })
+        .await
    }
 }

@@ -111,6 +111,11 @@ impl<T: Config> Stream for StorageItems<T> {
                        return Poll::Pending;
                    }
                    Poll::Ready(Err(e)) => {
+                        if e.is_disconnected_will_reconnect() {
+                            self.continue_fut = Some((self.continue_call)());
+                            continue;
+                        }
+
                        self.done = true;
                        return Poll::Ready(Some(Err(e)));
                    }