graph: Use ObjectStore for IPFS cache when configured

Author: lutter

graph/src/components/link_resolver/ipfs.rs

@@ -1,3 +1,4 @@
+use std::path::PathBuf;
 use std::sync::Arc;
 use std::sync::Mutex;
 use std::time::Duration;
@@ -10,6 +11,9 @@ use futures03::compat::Stream01CompatExt;
 use futures03::stream::StreamExt;
 use futures03::stream::TryStreamExt;
 use lru_time_cache::LruCache;
+use object_store::local::LocalFileSystem;
+use object_store::path::Path;
+use object_store::ObjectStore;
 use serde_json::Value;
 
 use crate::derive::CheapClone;
@@ -29,22 +33,60 @@ enum Cache {
     Memory {
         cache: Arc<Mutex<LruCache<ContentPath, Vec<u8>>>>,
     },
+    Disk {
+        store: Arc<dyn ObjectStore>,
+    },
+}
+
+fn log_err(logger: &Logger, e: &object_store::Error, log_not_found: bool) {
+    if log_not_found || !matches!(e, object_store::Error::NotFound { .. }) {
+        warn!(
+            logger,
+            "Failed to get IPFS object from disk cache; fetching from IPFS";
+            "error" => e.to_string(),
+        );
+    }
 }
 
 impl Cache {
-    fn new(capacity: usize) -> Self {
-        Self::Memory {
-            cache: Arc::new(Mutex::new(LruCache::with_capacity(capacity))),
+    fn new(capacity: usize, path: Option<PathBuf>) -> Self {
+        match path {
+            Some(path) => {
+                let fs = match LocalFileSystem::new_with_prefix(&path) {
+                    Err(e) => {
+                        panic!(
+                            "Failed to create IPFS file based cache at {}: {}",
+                            path.display(),
+                            e
+                        );
+                    }
+                    Ok(fs) => fs,
+                };
+                Cache::Disk {
+                    store: Arc::new(fs),
+                }
+            }
+            None => Self::Memory {
+                cache: Arc::new(Mutex::new(LruCache::with_capacity(capacity))),
+            },
         }
     }
 
-    async fn find(&self, path: &ContentPath) -> Option<Vec<u8>> {
+    async fn find(&self, logger: &Logger, path: &ContentPath) -> Option<Vec<u8>> {
         match self {
             Cache::Memory { cache } => cache.lock().unwrap().get(path).cloned(),
+            Cache::Disk { store } => {
+                let log_err = |e: &object_store::Error| log_err(logger, e, false);
+
+                let path = Path::from(path.cid().to_string());
+                let object = store.get(&path).await.inspect_err(log_err).ok()?;
+                let data = object.bytes().await.inspect_err(log_err).ok()?;
+                Some(data.to_vec())
+            }
         }
     }
 
-    async fn insert(&self, path: ContentPath, data: Vec<u8>) {
+    async fn insert(&self, logger: &Logger, path: ContentPath, data: Vec<u8>) {
         match self {
             Cache::Memory { cache } => {
                 let mut cache = cache.lock().unwrap();
@@ -53,6 +95,15 @@ impl Cache {
                     cache.insert(path.clone(), data.clone());
                 }
             }
+            Cache::Disk { store } => {
+                let log_err = |e: &object_store::Error| log_err(logger, e, true);
+                let path = Path::from(path.cid().to_string());
+                store
+                    .put(&path, data.into())
+                    .await
+                    .inspect_err(log_err)
+                    .ok();
+            }
         }
     }
 }
@@ -81,7 +132,10 @@ impl IpfsResolver {
 
         Self {
             client,
-            cache: Cache::new(env.max_ipfs_cache_size as usize),
+            cache: Cache::new(
+                env.max_ipfs_cache_size as usize,
+                env.ipfs_cache_location.clone(),
+            ),
             timeout: env.ipfs_timeout,
             max_file_size: env.max_ipfs_file_bytes,
             max_map_file_size: env.max_ipfs_map_file_size,
@@ -111,7 +165,7 @@ impl LinkResolverTrait for IpfsResolver {
         let max_file_size = self.max_file_size;
         let max_cache_file_size = self.max_cache_file_size;
 
-        if let Some(data) = self.cache.find(&path).await {
+        if let Some(data) = self.cache.find(&logger, &path).await {
             trace!(logger, "IPFS cat cache hit"; "hash" => path.to_string());
             return Ok(data.to_owned());
         }
@@ -132,7 +186,7 @@ impl LinkResolverTrait for IpfsResolver {
             .to_vec();
 
         if data.len() <= max_cache_file_size {
-            self.cache.insert(path.clone(), data.clone()).await;
+            self.cache.insert(&logger, path.clone(), data.clone()).await;
         } else {
             debug!(
                 logger,