fix(ipfs): make IPFS accept a reference to the context

Author: isum

chain/ethereum/src/data_source.rs

@@ -1391,7 +1391,7 @@ impl UnresolvedMapping {
                 .try_collect::<Vec<_>>(),
             async {
                 let module_bytes = resolver
-                    .cat(LinkResolverContext::new(deployment_hash, logger), &link)
+                    .cat(&LinkResolverContext::new(deployment_hash, logger), &link)
                     .await?;
                 Ok(Arc::new(module_bytes))
             },

chain/near/src/data_source.rs

@@ -451,7 +451,7 @@ impl UnresolvedMapping {
         let api_version = semver::Version::parse(&api_version)?;
 
         let module_bytes = resolver
-            .cat(LinkResolverContext::new(deployment_hash, logger), &link)
+            .cat(&LinkResolverContext::new(deployment_hash, logger), &link)
             .await
             .with_context(|| format!("failed to resolve mapping {}", link.link))?;
 

chain/substreams/src/data_source.rs

@@ -195,7 +195,7 @@ impl blockchain::UnresolvedDataSource<Chain> for UnresolvedDataSource {
     ) -> Result<DataSource, Error> {
         let content = resolver
             .cat(
-                LinkResolverContext::new(deployment_hash, logger),
+                &LinkResolverContext::new(deployment_hash, logger),
                 &self.source.package.file,
             )
             .await?;
@@ -244,7 +244,7 @@ impl blockchain::UnresolvedDataSource<Chain> for UnresolvedDataSource {
         let handler = match (self.mapping.handler, self.mapping.file) {
             (Some(handler), Some(file)) => {
                 let module_bytes = resolver
-                    .cat(LinkResolverContext::new(deployment_hash, logger), &file)
+                    .cat(&LinkResolverContext::new(deployment_hash, logger), &file)
                     .await
                     .with_context(|| format!("failed to resolve mapping {}", file.link))?;
 
@@ -722,21 +722,21 @@ mod test {
             unimplemented!()
         }
 
-        async fn cat(&self, _ctx: LinkResolverContext, _link: &Link) -> Result<Vec<u8>, Error> {
+        async fn cat(&self, _ctx: &LinkResolverContext, _link: &Link) -> Result<Vec<u8>, Error> {
             Ok(gen_package().encode_to_vec())
         }
 
         async fn get_block(
             &self,
-            _ctx: LinkResolverContext,
+            _ctx: &LinkResolverContext,
             _link: &Link,
         ) -> Result<Vec<u8>, Error> {
             unimplemented!()
         }
 
         async fn json_stream(
             &self,
-            _ctx: LinkResolverContext,
+            _ctx: &LinkResolverContext,
             _link: &Link,
         ) -> Result<JsonValueStream, Error> {
             unimplemented!()

core/src/polling_monitor/ipfs_service.rs

@@ -59,7 +59,7 @@ impl IpfsServiceInner {
         let res = self
             .client
             .cat(
-                ctx,
+                &ctx,
                 &path,
                 self.max_file_size,
                 Some(self.timeout),

core/src/subgraph/instance_manager.rs

@@ -284,7 +284,7 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
                 let file_bytes = self
                     .link_resolver
                     .cat(
-                        LinkResolverContext::new(&deployment.hash, &logger),
+                        &LinkResolverContext::new(&deployment.hash, &logger),
                         &graft.base.to_ipfs_link(),
                     )
                     .await?;

core/src/subgraph/provider.rs

@@ -90,7 +90,7 @@ impl<I: SubgraphInstanceManager> SubgraphAssignmentProviderTrait for SubgraphAss
         let file_bytes = self
             .link_resolver
             .cat(
-                LinkResolverContext::new(&loc.hash, &logger),
+                &LinkResolverContext::new(&loc.hash, &logger),
                 &loc.hash.to_ipfs_link(),
             )
             .await

core/src/subgraph/registrar.rs

@@ -291,7 +291,7 @@ where
             let file_bytes = self
                 .resolver
                 .cat(
-                    LinkResolverContext::new(&hash, &logger),
+                    &LinkResolverContext::new(&hash, &logger),
                     &hash.to_ipfs_link(),
                 )
                 .await

graph/src/components/link_resolver/ipfs.rs

@@ -64,7 +64,7 @@ impl LinkResolver for IpfsResolver {
         Box::new(s)
     }
 
-    async fn cat(&self, ctx: LinkResolverContext, link: &Link) -> Result<Vec<u8>, Error> {
+    async fn cat(&self, ctx: &LinkResolverContext, link: &Link) -> Result<Vec<u8>, Error> {
         let LinkResolverContext {
             deployment_hash,
             logger,
@@ -81,20 +81,20 @@ impl LinkResolver for IpfsResolver {
         };
 
         let ctx = IpfsContext {
-            deployment_hash,
-            logger,
+            deployment_hash: deployment_hash.cheap_clone(),
+            logger: logger.cheap_clone(),
         };
         let data = self
             .client
             .clone()
-            .cat(ctx, &path, max_file_size, timeout, retry_policy)
+            .cat(&ctx, &path, max_file_size, timeout, retry_policy)
             .await?
             .to_vec();
 
         Ok(data)
     }
 
-    async fn get_block(&self, ctx: LinkResolverContext, link: &Link) -> Result<Vec<u8>, Error> {
+    async fn get_block(&self, ctx: &LinkResolverContext, link: &Link) -> Result<Vec<u8>, Error> {
         let LinkResolverContext {
             deployment_hash,
             logger,
@@ -112,13 +112,13 @@ impl LinkResolver for IpfsResolver {
         };
 
         let ctx = IpfsContext {
-            deployment_hash,
-            logger,
+            deployment_hash: deployment_hash.cheap_clone(),
+            logger: logger.cheap_clone(),
         };
         let data = self
             .client
             .clone()
-            .get_block(ctx, &path, timeout, retry_policy)
+            .get_block(&ctx, &path, timeout, retry_policy)
             .await?
             .to_vec();
 
@@ -127,7 +127,7 @@ impl LinkResolver for IpfsResolver {
 
     async fn json_stream(
         &self,
-        ctx: LinkResolverContext,
+        ctx: &LinkResolverContext,
         link: &Link,
     ) -> Result<JsonValueStream, Error> {
         let LinkResolverContext {
@@ -148,13 +148,13 @@ impl LinkResolver for IpfsResolver {
         };
 
         let ctx = IpfsContext {
-            deployment_hash,
-            logger,
+            deployment_hash: deployment_hash.cheap_clone(),
+            logger: logger.cheap_clone(),
         };
         let mut stream = self
             .client
             .clone()
-            .cat_stream(ctx, &path, timeout, retry_policy)
+            .cat_stream(&ctx, &path, timeout, retry_policy)
             .await?
             .fuse()
             .boxed()
@@ -266,7 +266,7 @@ mod tests {
         .unwrap();
         let resolver = IpfsResolver::new(Arc::new(client), Arc::new(env_vars));
 
-        let err = IpfsResolver::cat(&resolver, Default::default(), &Link { link: cid.clone() })
+        let err = IpfsResolver::cat(&resolver, &Default::default(), &Link { link: cid.clone() })
             .await
             .unwrap_err();
 
@@ -290,7 +290,7 @@ mod tests {
         let resolver = IpfsResolver::new(Arc::new(client), Arc::new(env_vars));
 
         let stream =
-            IpfsResolver::json_stream(&resolver, Default::default(), &Link { link: cid }).await?;
+            IpfsResolver::json_stream(&resolver, &Default::default(), &Link { link: cid }).await?;
         stream.map_ok(|sv| sv.value).try_collect().await
     }
 

graph/src/components/link_resolver/mod.rs

@@ -26,18 +26,18 @@ pub trait LinkResolver: Send + Sync + 'static + Debug {
     fn with_retries(&self) -> Box<dyn LinkResolver>;
 
     /// Fetches the link contents as bytes.
-    async fn cat(&self, ctx: LinkResolverContext, link: &Link) -> Result<Vec<u8>, Error>;
+    async fn cat(&self, ctx: &LinkResolverContext, link: &Link) -> Result<Vec<u8>, Error>;
 
     /// Fetches the IPLD block contents as bytes.
-    async fn get_block(&self, ctx: LinkResolverContext, link: &Link) -> Result<Vec<u8>, Error>;
+    async fn get_block(&self, ctx: &LinkResolverContext, link: &Link) -> Result<Vec<u8>, Error>;
 
     /// Read the contents of `link` and deserialize them into a stream of JSON
     /// values. The values must each be on a single line; newlines are significant
     /// as they are used to split the file contents and each line is deserialized
     /// separately.
     async fn json_stream(
         &self,
-        ctx: LinkResolverContext,
+        ctx: &LinkResolverContext,
         link: &Link,
     ) -> Result<JsonValueStream, Error>;
 }

graph/src/data/subgraph/mod.rs

@@ -427,7 +427,7 @@ impl UnresolvedSchema {
     ) -> Result<InputSchema, anyhow::Error> {
         let schema_bytes = resolver
             .cat(
-                LinkResolverContext::new(deployment_hash, logger),
+                &LinkResolverContext::new(deployment_hash, logger),
                 &self.file,
             )
             .await