ipfs: optimize retries

Author: isum

core/src/polling_monitor/ipfs_service.rs

@@ -1,12 +1,13 @@
 use std::sync::Arc;
 use std::time::Duration;
 
-use anyhow::{anyhow, Error};
+use anyhow::anyhow;
+use anyhow::Error;
 use bytes::Bytes;
 use graph::futures03::future::BoxFuture;
 use graph::ipfs::ContentPath;
 use graph::ipfs::IpfsClient;
-use graph::ipfs::IpfsError;
+use graph::ipfs::RetryPolicy;
 use graph::{derive::CheapClone, prelude::CheapClone};
 use tower::{buffer::Buffer, ServiceBuilder, ServiceExt};
 
@@ -50,12 +51,17 @@ impl IpfsServiceInner {
 
         let res = self
             .client
-            .cat(&path, self.max_file_size, Some(self.timeout))
+            .cat(
+                &path,
+                self.max_file_size,
+                Some(self.timeout),
+                RetryPolicy::None,
+            )
             .await;
 
         match res {
             Ok(file_bytes) => Ok(Some(file_bytes)),
-            Err(IpfsError::RequestFailed(err)) if err.is_timeout() => {
+            Err(err) if err.is_timeout() => {
                 // Timeouts in IPFS mean that the content is not available, so we return `None`.
                 Ok(None)
             }
@@ -114,10 +120,9 @@ mod test {
 
         let client =
             IpfsRpcClient::new_unchecked(ServerAddress::local_rpc_api(), &graph::log::discard())
-                .unwrap()
-                .into_boxed();
+                .unwrap();
 
-        let svc = ipfs_service(client.into(), 100000, Duration::from_secs(30), 10);
+        let svc = ipfs_service(Arc::new(client), 100000, Duration::from_secs(30), 10);
 
         let path = ContentPath::new(format!("{dir_cid}/file.txt")).unwrap();
         let content = svc.oneshot(path).await.unwrap().unwrap();

graph/src/components/link_resolver/ipfs.rs

@@ -21,6 +21,7 @@ use crate::futures01::Async;
 use crate::futures01::Poll;
 use crate::ipfs::ContentPath;
 use crate::ipfs::IpfsClient;
+use crate::ipfs::RetryPolicy;
 use crate::prelude::{LinkResolver as LinkResolverTrait, *};
 
 #[derive(Clone, CheapClone, Derivative)]
@@ -36,6 +37,9 @@ pub struct IpfsResolver {
     max_file_size: usize,
     max_map_file_size: usize,
     max_cache_file_size: usize,
+
+    /// When set to `true`, it means infinite retries, ignoring the timeout setting.
+    retry: bool,
 }
 
 impl IpfsResolver {
@@ -51,6 +55,7 @@ impl IpfsResolver {
             max_file_size: env.max_ipfs_file_bytes,
             max_map_file_size: env.max_ipfs_map_file_size,
             max_cache_file_size: env.max_ipfs_cache_file_size,
+            retry: false,
         }
     }
 }
@@ -64,8 +69,9 @@ impl LinkResolverTrait for IpfsResolver {
     }
 
     fn with_retries(&self) -> Box<dyn LinkResolverTrait> {
-        // IPFS clients have internal retries enabled by default.
-        Box::new(self.cheap_clone())
+        let mut s = self.cheap_clone();
+        s.retry = true;
+        Box::new(s)
     }
 
     async fn cat(&self, logger: &Logger, link: &Link) -> Result<Vec<u8>, Error> {
@@ -81,9 +87,16 @@ impl LinkResolverTrait for IpfsResolver {
 
         trace!(logger, "IPFS cat cache miss"; "hash" => path.to_string());
 
+        let (timeout, retry_policy) = if self.retry {
+            (None, RetryPolicy::NonDeterministic)
+        } else {
+            (Some(timeout), RetryPolicy::Networking)
+        };
+
         let data = self
             .client
-            .cat(&path, max_file_size, Some(timeout))
+            .clone()
+            .cat(&path, max_file_size, timeout, retry_policy)
             .await?
             .to_vec();
 
@@ -111,20 +124,39 @@ impl LinkResolverTrait for IpfsResolver {
 
         trace!(logger, "IPFS block get"; "hash" => path.to_string());
 
-        let data = self.client.get_block(&path, Some(timeout)).await?.to_vec();
+        let (timeout, retry_policy) = if self.retry {
+            (None, RetryPolicy::NonDeterministic)
+        } else {
+            (Some(timeout), RetryPolicy::Networking)
+        };
+
+        let data = self
+            .client
+            .clone()
+            .get_block(&path, timeout, retry_policy)
+            .await?
+            .to_vec();
 
         Ok(data)
     }
 
     async fn json_stream(&self, logger: &Logger, link: &Link) -> Result<JsonValueStream, Error> {
         let path = ContentPath::new(&link.link)?;
         let max_map_file_size = self.max_map_file_size;
+        let timeout = self.timeout;
 
         trace!(logger, "IPFS JSON stream"; "hash" => path.to_string());
 
+        let (timeout, retry_policy) = if self.retry {
+            (None, RetryPolicy::NonDeterministic)
+        } else {
+            (Some(timeout), RetryPolicy::Networking)
+        };
+
         let mut stream = self
             .client
-            .cat_stream(&path, None)
+            .clone()
+            .cat_stream(&path, timeout, retry_policy)
             .await?
             .fuse()
             .boxed()
@@ -228,11 +260,8 @@ mod tests {
 
         let logger = crate::log::discard();
 
-        let client = IpfsRpcClient::new_unchecked(ServerAddress::local_rpc_api(), &logger)
-            .unwrap()
-            .into_boxed();
-
-        let resolver = IpfsResolver::new(client.into(), Arc::new(env_vars));
+        let client = IpfsRpcClient::new_unchecked(ServerAddress::local_rpc_api(), &logger).unwrap();
+        let resolver = IpfsResolver::new(Arc::new(client), Arc::new(env_vars));
 
         let err = IpfsResolver::cat(&resolver, &logger, &Link { link: cid.clone() })
             .await
@@ -250,9 +279,8 @@ mod tests {
             .to_owned();
 
         let logger = crate::log::discard();
-        let client =
-            IpfsRpcClient::new_unchecked(ServerAddress::local_rpc_api(), &logger)?.into_boxed();
-        let resolver = IpfsResolver::new(client.into(), Arc::new(env_vars));
+        let client = IpfsRpcClient::new_unchecked(ServerAddress::local_rpc_api(), &logger)?;
+        let resolver = IpfsResolver::new(Arc::new(client), Arc::new(env_vars));
 
         let stream = IpfsResolver::json_stream(&resolver, &logger, &Link { link: cid }).await?;
         stream.map_ok(|sv| sv.value).try_collect().await

graph/src/ipfs/client.rs

@@ -0,0 +1,186 @@
+use std::future::Future;
+use std::sync::Arc;
+use std::time::Duration;
+
+use async_trait::async_trait;
+use bytes::Bytes;
+use bytes::BytesMut;
+use futures03::stream::BoxStream;
+use futures03::StreamExt;
+use futures03::TryStreamExt;
+use slog::Logger;
+
+use crate::ipfs::ContentPath;
+use crate::ipfs::IpfsError;
+use crate::ipfs::IpfsResult;
+use crate::ipfs::RetryPolicy;
+
+/// A read-only connection to an IPFS server.
+#[async_trait]
+pub trait IpfsClient: Send + Sync + 'static {
+    /// Returns the logger associated with the client.
+    fn logger(&self) -> &Logger;
+
+    /// Sends a request to the IPFS server and returns a raw response.
+    async fn call(self: Arc<Self>, req: IpfsRequest) -> IpfsResult<IpfsResponse>;
+
+    /// Streams data from the specified content path.
+    ///
+    /// If a timeout is specified, the execution will be aborted if the IPFS server
+    /// does not return a response within the specified amount of time.
+    ///
+    /// The timeout is not propagated to the resulting stream.
+    async fn cat_stream(
+        self: Arc<Self>,
+        path: &ContentPath,
+        timeout: Option<Duration>,
+        retry_policy: RetryPolicy,
+    ) -> IpfsResult<BoxStream<'static, IpfsResult<Bytes>>> {
+        let fut = retry_policy.create("IPFS.cat_stream", self.logger()).run({
+            let path = path.to_owned();
+
+            move || {
+                let path = path.clone();
+                let client = self.clone();
+
+                async move { client.call(IpfsRequest::Cat(path)).await }
+            }
+        });
+
+        let resp = run_with_optional_timeout(path, fut, timeout).await?;
+
+        Ok(resp.bytes_stream())
+    }
+
+    /// Downloads data from the specified content path.
+    ///
+    /// If a timeout is specified, the execution will be aborted if the IPFS server
+    /// does not return a response within the specified amount of time.
+    async fn cat(
+        self: Arc<Self>,
+        path: &ContentPath,
+        max_size: usize,
+        timeout: Option<Duration>,
+        retry_policy: RetryPolicy,
+    ) -> IpfsResult<Bytes> {
+        let fut = retry_policy.create("IPFS.cat", self.logger()).run({
+            let path = path.to_owned();
+
+            move || {
+                let path = path.clone();
+                let client = self.clone();
+
+                async move {
+                    client
+                        .call(IpfsRequest::Cat(path))
+                        .await?
+                        .bytes(Some(max_size))
+                        .await
+                }
+            }
+        });
+
+        run_with_optional_timeout(path, fut, timeout).await
+    }
+
+    /// Downloads an IPFS block in raw format.
+    ///
+    /// If a timeout is specified, the execution will be aborted if the IPFS server
+    /// does not return a response within the specified amount of time.
+    async fn get_block(
+        self: Arc<Self>,
+        path: &ContentPath,
+        timeout: Option<Duration>,
+        retry_policy: RetryPolicy,
+    ) -> IpfsResult<Bytes> {
+        let fut = retry_policy.create("IPFS.get_block", self.logger()).run({
+            let path = path.to_owned();
+
+            move || {
+                let path = path.clone();
+                let client = self.clone();
+
+                async move {
+                    client
+                        .call(IpfsRequest::GetBlock(path))
+                        .await?
+                        .bytes(None)
+                        .await
+                }
+            }
+        });
+
+        run_with_optional_timeout(path, fut, timeout).await
+    }
+}
+
+/// Describes a request to an IPFS server.
+#[derive(Clone, Debug)]
+pub enum IpfsRequest {
+    Cat(ContentPath),
+    GetBlock(ContentPath),
+}
+
+/// Contains a raw, successful IPFS response.
+#[derive(Debug)]
+pub struct IpfsResponse {
+    pub(super) path: ContentPath,
+    pub(super) response: reqwest::Response,
+}
+
+impl IpfsResponse {
+    /// Reads and returns the response body.
+    ///
+    /// If the max size is specified and the response body is larger than the max size,
+    /// execution will result in an error.
+    pub async fn bytes(self, max_size: Option<usize>) -> IpfsResult<Bytes> {
+        let Some(max_size) = max_size else {
+            return self.response.bytes().await.map_err(Into::into);
+        };
+
+        let bytes = self
+            .response
+            .bytes_stream()
+            .err_into()
+            .try_fold(BytesMut::new(), |mut acc, chunk| async {
+                acc.extend(chunk);
+
+                if acc.len() > max_size {
+                    return Err(IpfsError::ContentTooLarge {
+                        path: self.path.clone(),
+                        max_size,
+                    });
+                }
+
+                Ok(acc)
+            })
+            .await?;
+
+        Ok(bytes.into())
+    }
+
+    /// Converts the response into a stream of bytes from the body.
+    pub fn bytes_stream(self) -> BoxStream<'static, IpfsResult<Bytes>> {
+        self.response.bytes_stream().err_into().boxed()
+    }
+}
+
+async fn run_with_optional_timeout<F, O>(
+    path: &ContentPath,
+    fut: F,
+    timeout: Option<Duration>,
+) -> IpfsResult<O>
+where
+    F: Future<Output = IpfsResult<O>>,
+{
+    match timeout {
+        Some(timeout) => {
+            tokio::time::timeout(timeout, fut)
+                .await
+                .map_err(|_| IpfsError::RequestTimeout {
+                    path: path.to_owned(),
+                })?
+        }
+        None => fut.await,
+    }
+}

graph/src/ipfs/content_path.rs

@@ -4,8 +4,8 @@ use cid::Cid;
 use crate::ipfs::IpfsError;
 use crate::ipfs::IpfsResult;
 
-#[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
 /// Represents a path to some data on IPFS.
+#[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub struct ContentPath {
     cid: Cid,
     path: Option<String>,