ipfs: Replace dag/stat with files/stat (#3597)

* ipfs: Switch from `dag/stat` to `files/stat`

* ipfs: Double check file size restriction

* docker, ci: Downgrade IPFS to 0.10.0

Author: leoyvens

Diff for: .github/workflows/ci.yml

@@ -21,7 +21,7 @@ jobs:
     runs-on: ubuntu-latest
     services:
       ipfs:
-        image: ipfs/go-ipfs:v0.12.2
+        image: ipfs/go-ipfs:v0.10.0
         ports:
           - 5001:5001
       postgres:
@@ -75,7 +75,7 @@ jobs:
     runs-on: ubuntu-latest
     services:
       ipfs:
-        image: ipfs/go-ipfs:v0.12.2
+        image: ipfs/go-ipfs:v0.10.0
         ports:
           - 5001:5001
       postgres:

Diff for: .travis.yml

@@ -23,7 +23,7 @@ before_install:
   # Install Node.js 11.x
   - nvm install 11 && nvm use 11
   # Install IPFS
-  - wget "https://dist.ipfs.io/go-ipfs/v0.12.2/go-ipfs_v0.12.2_linux-amd64.tar.gz" -O /tmp/ipfs.tar.gz
+  - wget "https://dist.ipfs.io/go-ipfs/v0.10.0/go-ipfs_v0.10.0_linux-amd64.tar.gz" -O /tmp/ipfs.tar.gz
   - pushd . && cd $HOME/bin && tar -xzvf /tmp/ipfs.tar.gz && popd
   - export PATH="$HOME/bin/go-ipfs:$PATH"
   - ipfs init

Diff for: core/src/link_resolver.rs

@@ -12,7 +12,7 @@ use lru_time_cache::LruCache;
 use serde_json::Value;
 
 use graph::{
-    ipfs_client::{IpfsClient, StatApi, StatResponse},
+    ipfs_client::{IpfsClient, StatApi},
     prelude::{LinkResolver as LinkResolverTrait, *},
 };
 
@@ -55,7 +55,7 @@ async fn select_fastest_client_with_stat(
     path: String,
     timeout: Duration,
     do_retry: bool,
-) -> Result<(StatResponse, Arc<IpfsClient>), Error> {
+) -> Result<(u64, Arc<IpfsClient>), Error> {
     let mut err: Option<Error> = None;
 
     let mut stats: FuturesUnordered<_> = clients
@@ -67,7 +67,11 @@ async fn select_fastest_client_with_stat(
             retry_policy(do_retry, "IPFS stat", &logger).run(move || {
                 let path = path.clone();
                 let c = c.cheap_clone();
-                async move { c.stat(api, path, timeout).map_ok(move |s| (s, i)).await }
+                async move {
+                    c.stat_size(api, path, timeout)
+                        .map_ok(move |s| (s, i))
+                        .await
+                }
             })
         })
         .collect();
@@ -90,18 +94,14 @@ async fn select_fastest_client_with_stat(
 }
 
 // Returns an error if the stat is bigger than `max_file_bytes`
-fn restrict_file_size(
-    path: &str,
-    stat: &StatResponse,
-    max_file_bytes: &Option<u64>,
-) -> Result<(), Error> {
+fn restrict_file_size(path: &str, size: u64, max_file_bytes: &Option<u64>) -> Result<(), Error> {
     if let Some(max_file_bytes) = max_file_bytes {
-        if stat.size > *max_file_bytes {
+        if size > *max_file_bytes {
             return Err(anyhow!(
                 "IPFS file {} is too large. It can be at most {} bytes but is {} bytes",
                 path,
                 max_file_bytes,
-                stat.size
+                size
             ));
         }
     }
@@ -172,10 +172,10 @@ impl LinkResolverTrait for LinkResolver {
         }
         trace!(logger, "IPFS cache miss"; "hash" => &path);
 
-        let (stat, client) = select_fastest_client_with_stat(
+        let (size, client) = select_fastest_client_with_stat(
             self.clients.cheap_clone(),
             logger.cheap_clone(),
-            StatApi::Dag,
+            StatApi::Files,
             path.clone(),
             self.timeout,
             self.retry,
@@ -184,44 +184,40 @@ impl LinkResolverTrait for LinkResolver {
 
         let max_cache_file_size = self.env_vars.mappings.max_ipfs_cache_file_size;
         let max_file_size = self.env_vars.mappings.max_ipfs_file_bytes.map(|n| n as u64);
-        restrict_file_size(&path, &stat, &max_file_size)?;
+        restrict_file_size(&path, size, &max_file_size)?;
 
-        let path = path.clone();
-        let this = self.clone();
+        let req_path = path.clone();
         let timeout = self.timeout;
-        let logger = logger.clone();
         let data = retry_policy(self.retry, "ipfs.cat", &logger)
             .run(move || {
-                let path = path.clone();
+                let path = req_path.clone();
                 let client = client.clone();
-                let this = this.clone();
-                let logger = logger.clone();
-                async move {
-                    let data = client.cat_all(path.clone(), timeout).await?.to_vec();
-
-                    // Only cache files if they are not too large
-                    if data.len() <= max_cache_file_size {
-                        let mut cache = this.cache.lock().unwrap();
-                        if !cache.contains_key(&path) {
-                            cache.insert(path.to_owned(), data.clone());
-                        }
-                    } else {
-                        debug!(logger, "File too large for cache";
-                                    "path" => path,
-                                    "size" => data.len()
-                        );
-                    }
-                    Result::<Vec<u8>, reqwest::Error>::Ok(data)
-                }
+                async move { Ok(client.cat_all(path.clone(), timeout).await?.to_vec()) }
             })
             .await?;
 
+        // The size reported by `files/stat` is not guaranteed to be exact, so check the limit again.
+        restrict_file_size(&path, data.len() as u64, &max_file_size)?;
+
+        // Only cache files if they are not too large
+        if data.len() <= max_cache_file_size {
+            let mut cache = self.cache.lock().unwrap();
+            if !cache.contains_key(&path) {
+                cache.insert(path.to_owned(), data.clone());
+            }
+        } else {
+            debug!(logger, "File too large for cache";
+                        "path" => path,
+                        "size" => data.len()
+            );
+        }
+
         Ok(data)
     }
 
     async fn get_block(&self, logger: &Logger, link: &Link) -> Result<Vec<u8>, Error> {
         trace!(logger, "IPFS block get"; "hash" => &link.link);
-        let (stat, client) = select_fastest_client_with_stat(
+        let (size, client) = select_fastest_client_with_stat(
             self.clients.cheap_clone(),
             logger.cheap_clone(),
             StatApi::Block,
@@ -232,7 +228,7 @@ impl LinkResolverTrait for LinkResolver {
         .await?;
 
         let max_file_size = self.env_vars.mappings.max_ipfs_file_bytes.map(|n| n as u64);
-        restrict_file_size(&link.link, &stat, &max_file_size)?;
+        restrict_file_size(&link.link, size, &max_file_size)?;
 
         let link = link.link.clone();
         let data = retry_policy(self.retry, "ipfs.getBlock", &logger)
@@ -253,18 +249,18 @@ impl LinkResolverTrait for LinkResolver {
         // Discard the `/ipfs/` prefix (if present) to get the hash.
         let path = link.link.trim_start_matches("/ipfs/");
 
-        let (stat, client) = select_fastest_client_with_stat(
+        let (size, client) = select_fastest_client_with_stat(
             self.clients.cheap_clone(),
             logger.cheap_clone(),
-            StatApi::Dag,
+            StatApi::Files,
             path.to_string(),
             self.timeout,
             self.retry,
         )
         .await?;
 
         let max_file_size = Some(self.env_vars.mappings.max_ipfs_map_file_size as u64);
-        restrict_file_size(path, &stat, &max_file_size)?;
+        restrict_file_size(path, size, &max_file_size)?;
 
         let mut stream = client.cat(path.to_string()).await?.fuse().boxed().compat();
 

Diff for: docker/docker-compose.yml

@@ -22,7 +22,7 @@ services:
       ethereum: 'mainnet:http://host.docker.internal:8545'
       GRAPH_LOG: info
   ipfs:
-    image: ipfs/go-ipfs:v0.12.2
+    image: ipfs/go-ipfs:v0.10.0
     ports:
       - '5001:5001'
     volumes:

Diff for: graph/src/ipfs_client.rs

@@ -9,25 +9,31 @@ use serde::Deserialize;
 use std::time::Duration;
 use std::{str::FromStr, sync::Arc};
 
-#[derive(Clone, Copy)]
+#[derive(Clone, Copy, PartialEq, Eq)]
 pub enum StatApi {
     Block,
-    Dag,
+    Files,
 }
 
 impl StatApi {
     fn route(&self) -> &'static str {
         match self {
             Self::Block => "block",
-            Self::Dag => "dag",
+            Self::Files => "files",
         }
     }
 }
 
 #[derive(Debug, Deserialize)]
 #[serde(rename_all = "PascalCase")]
-pub struct StatResponse {
-    pub size: u64,
+struct BlockStatResponse {
+    size: u64,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+struct FilesStatResponse {
+    cumulative_size: u64,
 }
 
 #[derive(Debug, Deserialize)]
@@ -68,18 +74,24 @@ impl IpfsClient {
         }
     }
 
-    /// Calls stat for the given API route.
-    pub async fn stat(
+    /// Calls stat for the given API route, and returns the total size of the object.
+    pub async fn stat_size(
         &self,
         api: StatApi,
-        path: String,
+        mut cid: String,
         timeout: Duration,
-    ) -> Result<StatResponse, reqwest::Error> {
+    ) -> Result<u64, reqwest::Error> {
         let route = format!("{}/stat", api.route());
-        self.call(self.url(&route, path), None, Some(timeout))
-            .await?
-            .json()
-            .await
+        if api == StatApi::Files {
+            // files/stat requires a leading `/ipfs/`.
+            cid = format!("/ipfs/{}", cid);
+        }
+        let url = self.url(&route, cid);
+        let res = self.call(url, None, Some(timeout)).await?;
+        match api {
+            StatApi::Files => Ok(res.json::<FilesStatResponse>().await?.cumulative_size),
+            StatApi::Block => Ok(res.json::<BlockStatResponse>().await?.size),
+        }
     }
 
     /// Download the entire contents.

Diff for: store/test-store/devel/docker-compose.yml

@@ -1,7 +1,7 @@
 version: '3'
 services:
   ipfs:
-    image: ipfs/go-ipfs:v0.12.2
+    image: ipfs/go-ipfs:v0.10.0
     ports:
       - '5001:5001'
     volumes:

Diff for: tests/tests/common/docker.rs

@@ -7,7 +7,7 @@ use tokio::time::{sleep, Duration};
 use tokio_stream::StreamExt;
 
 const POSTGRES_IMAGE: &'static str = "postgres:latest";
-const IPFS_IMAGE: &'static str = "ipfs/go-ipfs:v0.12.2";
+const IPFS_IMAGE: &'static str = "ipfs/go-ipfs:v0.10.0";
 const GANACHE_IMAGE: &'static str = "trufflesuite/ganache-cli:latest";
 type DockerError = bollard::errors::Error;