gnd: Support multiple subgraphs, grafting, subgraph composition in dev mode (#6000)

* graph: Add clone_for_deployment to FileLinkResolver to create  FileLinkResolver with the right base dir for a subgraph

* graph: Add for_deployment to LinkResolverTrait

* core, graph: use for_deployment to get properly scoped resolver

* graph: Implement aliases for file link resolver

* node: Make gnd work with multiple subgraphs

* node: Support subgraph datasource in gnd

* node: correct the default value for manfiest

* core, node, graph: Ignore graft base in dev mode

* node: Allow providing a postgres url for gnd

* node: Do not use pgtemp in windows

* store: enable `vendored` feature for openssl crate

* chain/ethereum: Return error when ipc is used in non unix platform

* node: Refactor launcher

* node/dev : Better error message when database directory doesn't exist

* node: refactor watcher

* core, node, graph: Manipulate raw manifest instead of passing
ignore_graft_base

This reverts commit b5bbf93.

* node: Correct comments on `redeploy_all_subgraphs`

* node/gnd: Deploy all subgraphs first before wathcing files

* core, graph : Refactor LinkResolver trait

Author: incrypto32

Cargo.lock

@@ -32,6 +32,11 @@ impl Transport {
             .expect("Failed to connect to Ethereum IPC")
     }
 
+    #[cfg(not(unix))]
+    pub async fn new_ipc(_ipc: &str) -> Self {
+        panic!("IPC connections are not supported on non-Unix platforms")
+    }
+
     /// Creates a WebSocket transport.
     pub async fn new_ws(ws: &str) -> Self {
         ws::WebSocket::new(ws)

chain/ethereum/src/transport.rs

@@ -705,6 +705,10 @@ mod test {
             unimplemented!()
         }
 
+        fn for_manifest(&self, _manifest_path: &str) -> Result<Box<dyn LinkResolver>, Error> {
+            unimplemented!()
+        }
+
         async fn cat(&self, _logger: &Logger, _link: &Link) -> Result<Vec<u8>, Error> {
             Ok(gen_package().encode_to_vec())
         }

chain/substreams/src/data_source.rs

@@ -272,7 +272,12 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
         let manifest = UnresolvedSubgraphManifest::parse(deployment.hash.cheap_clone(), manifest)?;
 
         // Allow for infinite retries for subgraph definition files.
-        let link_resolver = Arc::from(self.link_resolver.with_retries());
+        let link_resolver = Arc::from(
+            self.link_resolver
+                .for_manifest(&deployment.hash.to_string())
+                .map_err(SubgraphRegistrarError::Unknown)?
+                .with_retries(),
+        );
 
         // Make sure the `raw_yaml` is present on both this subgraph and the graft base.
         self.subgraph_store

core/src/subgraph/instance_manager.rs

@@ -86,8 +86,12 @@ impl<I: SubgraphInstanceManager> SubgraphAssignmentProviderTrait for SubgraphAss
             ));
         }
 
-        let file_bytes = self
+        let link_resolver = self
             .link_resolver
+            .for_manifest(&loc.hash.to_string())
+            .map_err(SubgraphAssignmentProviderError::ResolveError)?;
+
+        let file_bytes = link_resolver
             .cat(&logger, &loc.hash.to_ipfs_link())
             .await
             .map_err(SubgraphAssignmentProviderError::ResolveError)?;

core/src/subgraph/provider.rs

@@ -278,6 +278,7 @@ where
         start_block_override: Option<BlockPtr>,
         graft_block_override: Option<BlockPtr>,
         history_blocks: Option<i32>,
+        ignore_graft_base: bool,
     ) -> Result<DeploymentLocator, SubgraphRegistrarError> {
         // We don't have a location for the subgraph yet; that will be
         // assigned when we deploy for real. For logging purposes, make up a
@@ -286,19 +287,33 @@ where
             .logger_factory
             .subgraph_logger(&DeploymentLocator::new(DeploymentId(0), hash.clone()));
 
-        let raw: serde_yaml::Mapping = {
-            let file_bytes = self
-                .resolver
-                .cat(&logger, &hash.to_ipfs_link())
-                .await
-                .map_err(|e| {
-                    SubgraphRegistrarError::ResolveError(
-                        SubgraphManifestResolveError::ResolveError(e),
-                    )
-                })?;
-
-            serde_yaml::from_slice(&file_bytes)
-                .map_err(|e| SubgraphRegistrarError::ResolveError(e.into()))?
+        let resolver: Arc<dyn LinkResolver> = Arc::from(
+            self.resolver
+                .for_manifest(&hash.to_string())
+                .map_err(SubgraphRegistrarError::Unknown)?,
+        );
+
+        let raw = {
+            let mut raw: serde_yaml::Mapping = {
+                let file_bytes =
+                    resolver
+                        .cat(&logger, &hash.to_ipfs_link())
+                        .await
+                        .map_err(|e| {
+                            SubgraphRegistrarError::ResolveError(
+                                SubgraphManifestResolveError::ResolveError(e),
+                            )
+                        })?;
+
+                serde_yaml::from_slice(&file_bytes)
+                    .map_err(|e| SubgraphRegistrarError::ResolveError(e.into()))?
+            };
+
+            if ignore_graft_base {
+                raw.remove("graft");
+            }
+
+            raw
         };
 
         let kind = BlockchainKind::from_manifest(&raw).map_err(|e| {
@@ -323,7 +338,7 @@ where
                     node_id,
                     debug_fork,
                     self.version_switching_mode,
-                    &self.resolver,
+                    &resolver,
                     history_blocks,
                 )
                 .await?
@@ -341,7 +356,7 @@ where
                     node_id,
                     debug_fork,
                     self.version_switching_mode,
-                    &self.resolver,
+                    &resolver,
                     history_blocks,
                 )
                 .await?
@@ -359,7 +374,7 @@ where
                     node_id,
                     debug_fork,
                     self.version_switching_mode,
-                    &self.resolver,
+                    &resolver,
                     history_blocks,
                 )
                 .await?
@@ -549,10 +564,11 @@ async fn create_subgraph_version<C: Blockchain, S: SubgraphStore>(
     history_blocks_override: Option<i32>,
 ) -> Result<DeploymentLocator, SubgraphRegistrarError> {
     let raw_string = serde_yaml::to_string(&raw).unwrap();
+
     let unvalidated = UnvalidatedSubgraphManifest::<C>::resolve(
         deployment.clone(),
         raw,
-        resolver,
+        &resolver,
         logger,
         ENV_VARS.max_spec_version.clone(),
     )

core/src/subgraph/registrar.rs

@@ -1,3 +1,4 @@
+use std::collections::HashMap;
 use std::path::{Path, PathBuf};
 use std::time::Duration;
 
@@ -12,16 +13,29 @@ use crate::prelude::{Error, JsonValueStream, LinkResolver as LinkResolverTrait};
 pub struct FileLinkResolver {
     base_dir: Option<PathBuf>,
     timeout: Duration,
+    // This is a hashmap that maps the alias name to the path of the file that is aliased
+    aliases: HashMap<String, PathBuf>,
+}
+
+impl Default for FileLinkResolver {
+    fn default() -> Self {
+        Self {
+            base_dir: None,
+            timeout: Duration::from_secs(30),
+            aliases: HashMap::new(),
+        }
+    }
 }
 
 impl FileLinkResolver {
     /// Create a new FileLinkResolver
     ///
     /// All paths are treated as absolute paths.
-    pub fn new() -> Self {
+    pub fn new(base_dir: Option<PathBuf>, aliases: HashMap<String, PathBuf>) -> Self {
         Self {
-            base_dir: None,
+            base_dir: base_dir,
             timeout: Duration::from_secs(30),
+            aliases,
         }
     }
 
@@ -33,18 +47,59 @@ impl FileLinkResolver {
         Self {
             base_dir: Some(base_dir.as_ref().to_owned()),
             timeout: Duration::from_secs(30),
+            aliases: HashMap::new(),
         }
     }
 
     fn resolve_path(&self, link: &str) -> PathBuf {
         let path = Path::new(link);
 
+        // If the path is an alias, use the aliased path
+        if let Some(aliased) = self.aliases.get(link) {
+            return aliased.clone();
+        }
+
         // Return the path as is if base_dir is None, or join with base_dir if present.
         // if "link" is an absolute path, join will simply return that path.
         self.base_dir
             .as_ref()
             .map_or_else(|| path.to_owned(), |base_dir| base_dir.join(link))
     }
+
+    /// This method creates a new resolver that is scoped to a specific subgraph
+    /// It will set the base directory to the parent directory of the manifest path
+    /// This is required because paths mentioned in the subgraph manifest are relative paths
+    /// and we need a new resolver with the right base directory for the specific subgraph
+    fn clone_for_manifest(&self, manifest_path_str: &str) -> Result<Self, Error> {
+        let mut resolver = self.clone();
+
+        // Create a path to the manifest based on the current resolver's
+        // base directory or default to using the deployment string as path
+        // If the deployment string is an alias, use the aliased path
+        let manifest_path = if let Some(aliased) = self.aliases.get(&manifest_path_str.to_string())
+        {
+            aliased.clone()
+        } else {
+            match &resolver.base_dir {
+                Some(dir) => dir.join(&manifest_path_str),
+                None => PathBuf::from(manifest_path_str),
+            }
+        };
+
+        let canonical_manifest_path = manifest_path
+            .canonicalize()
+            .map_err(|e| Error::from(anyhow!("Failed to canonicalize manifest path: {}", e)))?;
+
+        // The manifest path is the path of the subgraph manifest file in the build directory
+        // We use the parent directory as the base directory for the new resolver
+        let base_dir = canonical_manifest_path
+            .parent()
+            .ok_or_else(|| Error::from(anyhow!("Manifest path has no parent directory")))?
+            .to_path_buf();
+
+        resolver.base_dir = Some(base_dir);
+        Ok(resolver)
+    }
 }
 
 pub fn remove_prefix(link: &str) -> &str {
@@ -86,6 +141,10 @@ impl LinkResolverTrait for FileLinkResolver {
         }
     }
 
+    fn for_manifest(&self, manifest_path: &str) -> Result<Box<dyn LinkResolverTrait>, Error> {
+        Ok(Box::new(self.clone_for_manifest(manifest_path)?))
+    }
+
     async fn get_block(&self, _logger: &Logger, _link: &Link) -> Result<Vec<u8>, Error> {
         Err(anyhow!("get_block is not implemented for FileLinkResolver").into())
     }
@@ -117,7 +176,7 @@ mod tests {
         file.write_all(test_content).unwrap();
 
         // Create a resolver without a base directory
-        let resolver = FileLinkResolver::new();
+        let resolver = FileLinkResolver::default();
         let logger = slog::Logger::root(slog::Discard, slog::o!());
 
         // Test valid path resolution
@@ -185,4 +244,64 @@ mod tests {
         let _ = fs::remove_file(test_file_path);
         let _ = fs::remove_dir(temp_dir);
     }
+
+    #[tokio::test]
+    async fn test_file_resolver_with_aliases() {
+        // Create a temporary directory for test files
+        let temp_dir = env::temp_dir().join("file_resolver_test_aliases");
+        let _ = fs::create_dir_all(&temp_dir);
+
+        // Create two test files with different content
+        let test_file1_path = temp_dir.join("file.txt");
+        let test_content1 = b"This is the file content";
+        let mut file1 = fs::File::create(&test_file1_path).unwrap();
+        file1.write_all(test_content1).unwrap();
+
+        let test_file2_path = temp_dir.join("another_file.txt");
+        let test_content2 = b"This is another file content";
+        let mut file2 = fs::File::create(&test_file2_path).unwrap();
+        file2.write_all(test_content2).unwrap();
+
+        // Create aliases mapping
+        let mut aliases = HashMap::new();
+        aliases.insert("alias1".to_string(), test_file1_path.clone());
+        aliases.insert("alias2".to_string(), test_file2_path.clone());
+        aliases.insert("deployment-id".to_string(), test_file1_path.clone());
+
+        // Create resolver with aliases
+        let resolver = FileLinkResolver::new(Some(temp_dir.clone()), aliases);
+        let logger = slog::Logger::root(slog::Discard, slog::o!());
+
+        // Test resolving by aliases
+        let link1 = Link {
+            link: "alias1".to_string(),
+        };
+        let result1 = resolver.cat(&logger, &link1).await.unwrap();
+        assert_eq!(result1, test_content1);
+
+        let link2 = Link {
+            link: "alias2".to_string(),
+        };
+        let result2 = resolver.cat(&logger, &link2).await.unwrap();
+        assert_eq!(result2, test_content2);
+
+        // Test that the alias works in for_deployment as well
+        let deployment_resolver = resolver.clone_for_manifest("deployment-id").unwrap();
+
+        let expected_dir = test_file1_path.parent().unwrap();
+        let deployment_base_dir = deployment_resolver.base_dir.clone().unwrap();
+
+        let canonical_expected_dir = expected_dir.canonicalize().unwrap();
+        let canonical_deployment_dir = deployment_base_dir.canonicalize().unwrap();
+
+        assert_eq!(
+            canonical_deployment_dir, canonical_expected_dir,
+            "Build directory paths don't match"
+        );
+
+        // Clean up
+        let _ = fs::remove_file(test_file1_path);
+        let _ = fs::remove_file(test_file2_path);
+        let _ = fs::remove_dir(temp_dir);
+    }
 }

graph/src/components/link_resolver/file.rs

@@ -64,6 +64,10 @@ impl LinkResolverTrait for IpfsResolver {
         Box::new(s)
     }
 
+    fn for_manifest(&self, _manifest_path: &str) -> Result<Box<dyn LinkResolverTrait>, Error> {
+        Ok(Box::new(self.cheap_clone()))
+    }
+
     async fn cat(&self, _logger: &Logger, link: &Link) -> Result<Vec<u8>, Error> {
         let path = ContentPath::new(&link.link)?;
         let timeout = self.timeout;