implement recursive auto sync for grafts

create subgraph before trying to create the deployment

add config value to flip on/off auto syncing grafts

add auto_sync_grafts value and plumb max depth when enabled

restore manifest validation comment

implement awaiting for auto graft sync

Author: dwerner

core/src/subgraph/registrar.rs

@@ -28,6 +28,8 @@ use graph::tokio_retry::Retry;
 use graph::util::futures::retry_strategy;
 use graph::util::futures::RETRY_DEFAULT_LIMIT;
 
+const MAX_AUTO_GRAFT_SYNC_DEPTH: u32 = 42;
+
 pub struct SubgraphRegistrar<P, S, SM> {
     logger: Logger,
     logger_factory: LoggerFactory,
@@ -303,20 +305,8 @@ 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 raw: serde_yaml::Mapping =
+            resolve_raw_manifest(&self.resolver, &self.logger, &hash).await?;
 
         let kind = BlockchainKind::from_manifest(&raw).map_err(|e| {
             SubgraphRegistrarError::ResolveError(SubgraphManifestResolveError::ResolveError(e))
@@ -326,6 +316,12 @@ where
         let history_blocks =
             history_blocks.or(self.settings.for_name(&name).map(|c| c.history_blocks));
 
+        let auto_graft_sync_depth = if self.store.auto_sync_grafts() {
+            Some(0)
+        } else {
+            None
+        };
+
         let deployment_locator = match kind {
             BlockchainKind::Arweave => {
                 create_subgraph_version::<graph_chain_arweave::Chain, _>(
@@ -342,6 +338,7 @@ where
                     self.version_switching_mode,
                     &self.resolver,
                     history_blocks,
+                    auto_graft_sync_depth,
                 )
                 .await?
             }
@@ -360,6 +357,7 @@ where
                     self.version_switching_mode,
                     &self.resolver,
                     history_blocks,
+                    auto_graft_sync_depth,
                 )
                 .await?
             }
@@ -378,6 +376,7 @@ where
                     self.version_switching_mode,
                     &self.resolver,
                     history_blocks,
+                    auto_graft_sync_depth,
                 )
                 .await?
             }
@@ -396,6 +395,7 @@ where
                     self.version_switching_mode,
                     &self.resolver,
                     history_blocks,
+                    auto_graft_sync_depth,
                 )
                 .await?
             }
@@ -414,6 +414,7 @@ where
                     self.version_switching_mode,
                     &self.resolver,
                     history_blocks,
+                    auto_graft_sync_depth,
                 )
                 .await?
             }
@@ -432,6 +433,7 @@ where
                     self.version_switching_mode,
                     &self.resolver,
                     history_blocks,
+                    auto_graft_sync_depth,
                 )
                 .await?
             }
@@ -555,9 +557,9 @@ async fn start_subgraph(
 }
 
 /// Resolves the subgraph's earliest block
-async fn resolve_start_block(
-    manifest: &SubgraphManifest<impl Blockchain>,
-    chain: &impl Blockchain,
+async fn resolve_start_block<C: Blockchain>(
+    manifest: &SubgraphManifest<C>,
+    chain: &C,
     logger: &Logger,
 ) -> Result<Option<BlockPtr>, SubgraphRegistrarError> {
     // If the minimum start block is 0 (i.e. the genesis block),
@@ -618,9 +620,12 @@ async fn create_subgraph_version<C: Blockchain, S: SubgraphStore>(
     version_switching_mode: SubgraphVersionSwitchingMode,
     resolver: &Arc<dyn LinkResolver>,
     history_blocks_override: Option<i32>,
+    depth: Option<u32>,
 ) -> Result<DeploymentLocator, SubgraphRegistrarError> {
     let raw_string = serde_yaml::to_string(&raw).unwrap();
-    let unvalidated = UnvalidatedSubgraphManifest::<C>::resolve(
+
+    // We need to defer validation of the manifest until after we have synced the base subgraph.
+    let unvalidated_manifest = UnvalidatedSubgraphManifest::<C>::resolve(
         deployment.clone(),
         raw,
         resolver,
@@ -630,16 +635,37 @@ async fn create_subgraph_version<C: Blockchain, S: SubgraphStore>(
     .map_err(SubgraphRegistrarError::ResolveError)
     .await?;
 
-    // Determine if the graft_base should be validated.
-    // Validate the graft_base if there is a pending graft, ensuring its presence.
-    // If the subgraph is new (indicated by DeploymentNotFound), the graft_base should be validated.
-    // If the subgraph already exists and there is no pending graft, graft_base validation is not required.
+    if let (Some(depth), Some(graft)) = (depth, unvalidated_manifest.unvalidated_graft()) {
+        if depth < MAX_AUTO_GRAFT_SYNC_DEPTH {
+            Box::pin(auto_sync_graft::<C, S>(
+                graft,
+                resolver,
+                logger,
+                &store,
+                &chains,
+                &name,
+                &node_id,
+                &debug_fork,
+                version_switching_mode,
+                history_blocks_override,
+                depth,
+            ))
+            .await?;
+        } else {
+            warn!(
+                logger,
+                "Auto-syncing subgraph grafts depth limit reached";
+                "depth" => depth
+            );
+        }
+    }
+
     let should_validate = match store.graft_pending(&deployment) {
         Ok(graft_pending) => graft_pending,
         Err(StoreError::DeploymentNotFound(_)) => true,
         Err(e) => return Err(SubgraphRegistrarError::StoreError(e)),
     };
-    let manifest = unvalidated
+    let manifest = unvalidated_manifest
         .validate(store.cheap_clone(), should_validate)
         .await
         .map_err(SubgraphRegistrarError::ManifestValidationError)?;
@@ -732,3 +758,103 @@ async fn create_subgraph_version<C: Blockchain, S: SubgraphStore>(
         )
         .map_err(SubgraphRegistrarError::SubgraphDeploymentError)
 }
+
+/// Automatically syncs a subgraph graft from the base subgraph.
+/// This will await the syncing of the base subgraph before proceeding.
+/// Recursively calls `create_subgraph_version` to create any grafts of
+/// this graft up to `MAX_AUTO_GRAFT_SYNC_DEPTH`.`
+async fn auto_sync_graft<C: Blockchain, S: SubgraphStore>(
+    graft: &Graft,
+    resolver: &Arc<dyn LinkResolver>,
+    logger: &Logger,
+    store: &Arc<S>,
+    chains: &Arc<BlockchainMap>,
+    name: &SubgraphName,
+    node_id: &NodeId,
+    debug_fork: &Option<DeploymentHash>,
+    version_switching_mode: SubgraphVersionSwitchingMode,
+    history_blocks_override: Option<i32>,
+    depth: u32,
+) -> Result<DeploymentLocator, SubgraphRegistrarError> {
+    info!(
+        logger,
+        "Auto-syncing subgraph graft";
+        "subgraph" => name.to_string(),
+        "hash" => graft.base.to_string(),
+        "depth" => depth,
+        "block" => graft.block
+    );
+    let subgraft_raw_manifest = resolve_raw_manifest(resolver, logger, &graft.base).await?;
+
+    let id = graft.base.clone();
+    let name = &id[id.len().saturating_sub(10)..];
+    let name = format!("auto-graft-sync/{}", name);
+    let name =
+        SubgraphName::new(name.clone()).map_err(|_| SubgraphRegistrarError::NameNotValid(name))?;
+
+    info!(
+        logger,
+        "auto-syncing subgraph";
+        "subgraph" => name.to_string(),
+        "hash" => graft.base.to_string()
+    );
+
+    let _ = store.create_subgraph(name.clone())?;
+    info!(logger, "Created subgraph"; "subgraph_name" => name.to_string(), "id" => id.to_string());
+
+    let locator = create_subgraph_version::<C, S>(
+        logger,
+        store.clone(),
+        chains.clone(),
+        name.clone(),
+        graft.base.clone(),
+        None,
+        None,
+        subgraft_raw_manifest.clone(),
+        node_id.clone(),
+        debug_fork.clone(),
+        version_switching_mode,
+        resolver,
+        history_blocks_override,
+        Some(depth + 1),
+    )
+    .await?;
+
+    info!(
+        logger,
+        "Awaiting subgraph sync";
+        "subgraph" => name.to_string(),
+        "hash" => graft.base.to_string()
+    );
+
+    graft
+        .await_sync(store.clone(), Duration::from_secs(1))
+        .await?;
+
+    info!(
+        logger,
+        "auto-syncing subgraph complete";
+        "subgraph" => name.to_string(),
+        "hash" => graft.base.to_string()
+    );
+    Ok(locator)
+}
+
+async fn resolve_raw_manifest(
+    resolver: &Arc<dyn LinkResolver>,
+    logger: &Logger,
+    deployment_hash: &DeploymentHash,
+) -> Result<serde_yaml::Mapping, SubgraphRegistrarError> {
+    let subgraft_raw_manifest: serde_yaml::Mapping = {
+        let file_bytes = resolver
+            .cat(&logger, &deployment_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()))?
+    };
+    Ok(subgraft_raw_manifest)
+}

graph/src/components/store/traits.rs

@@ -215,6 +215,8 @@ pub trait SubgraphStore: Send + Sync + 'static {
     /// When this flag is set, indexing of the deployment should log
     /// additional diagnostic information
     fn instrument(&self, deployment: &DeploymentLocator) -> Result<bool, StoreError>;
+
+    fn auto_sync_grafts(&self) -> bool;
 }
 
 pub trait ReadStore: Send + Sync + 'static {

graph/src/data/subgraph/mod.rs

@@ -26,6 +26,8 @@ use stable_hash_legacy::SequenceNumber;
 use std::{
     collections::{BTreeSet, HashMap, HashSet},
     marker::PhantomData,
+    time::Duration,
+    time::Instant,
 };
 use thiserror::Error;
 use wasmparser;
@@ -284,6 +286,8 @@ pub enum SubgraphRegistrarError {
     NameExists(String),
     #[error("subgraph name not found: {0}")]
     NameNotFound(String),
+    #[error("subgraph name not valid: {0}")]
+    NameNotValid(String),
     #[error("network not supported by registrar: {0}")]
     NetworkNotSupported(Error),
     #[error("deployment not found: {0}")]
@@ -473,14 +477,7 @@ impl Graft {
     ) -> Result<(), SubgraphManifestValidationError> {
         use SubgraphManifestValidationError::*;
 
-        let last_processed_block = store
-            .least_block_ptr(&self.base)
-            .await
-            .map_err(|e| GraftBaseInvalid(e.to_string()))?;
-        let is_base_healthy = store
-            .is_healthy(&self.base)
-            .await
-            .map_err(|e| GraftBaseInvalid(e.to_string()))?;
+        let (last_processed_block, is_base_healthy) = self.graft_status(store).await?;
 
         // We are being defensive here: we don't know which specific
         // instance of a subgraph we will use as the base for the graft,
@@ -522,6 +519,60 @@ impl Graft {
             (Some(_), _) => Ok(()),
         }
     }
+
+    async fn graft_status<S: SubgraphStore>(
+        &self,
+        store: Arc<S>,
+    ) -> Result<(Option<BlockPtr>, bool), SubgraphManifestValidationError> {
+        use SubgraphManifestValidationError::*;
+        let last_processed_block = store
+            .least_block_ptr(&self.base)
+            .await
+            .map_err(|e| GraftBaseInvalid(e.to_string()))?;
+        let is_base_healthy = store
+            .is_healthy(&self.base)
+            .await
+            .map_err(|e| GraftBaseInvalid(e.to_string()))?;
+        Ok((last_processed_block, is_base_healthy))
+    }
+
+    /// Awaits the target block sync for the graft.
+    pub async fn await_sync<S: SubgraphStore>(
+        &self,
+        store: Arc<S>,
+        interval: Duration,
+    ) -> Result<(), SubgraphManifestValidationError> {
+        use SubgraphManifestValidationError::*;
+
+        const MAX_WAIT_NO_BLOCKS: Duration = Duration::from_secs(60 * 60);
+        let start = Instant::now();
+
+        loop {
+            let (ptr, healthy) = self.graft_status(store.clone()).await?;
+            match ptr {
+                Some(ptr) if ptr.number < self.block => {
+                    tokio::time::sleep(interval).await;
+                }
+                Some(_) => {
+                    if !healthy {
+                        return Err(GraftBaseInvalid(format!(
+                            "failed to graft onto `{}` at block {} since it's not healthy",
+                            self.base, self.block
+                        )));
+                    }
+                    break Ok(());
+                }
+                None => {
+                    if start.elapsed() > MAX_WAIT_NO_BLOCKS {
+                        return Err(GraftBaseInvalid(format!(
+                            "failed to graft onto `{}` at block {} since it has not processed any blocks",
+                            self.base, self.block
+                        )));
+                    }
+                }
+            }
+        }
+    }
 }
 
 #[derive(Clone, Debug)]
@@ -753,6 +804,11 @@ impl<C: Blockchain> UnvalidatedSubgraphManifest<C> {
     pub fn spec_version(&self) -> &Version {
         &self.0.spec_version
     }
+
+    /// Get the graft from this unvalidated manifest.
+    pub fn unvalidated_graft(&self) -> Option<&Graft> {
+        self.0.graft.as_ref()
+    }
 }
 
 impl<C: Blockchain> SubgraphManifest<C> {

node/resources/tests/full_config.toml

@@ -1,5 +1,6 @@
 [general]
 query = "query_node_.*"
+auto_sync_grafts = false
 
 [store]
 [store.primary]

node/src/chain.rs

@@ -706,6 +706,7 @@ mod test {
             ethereum_ws: vec![],
             ethereum_ipc: vec![],
             unsafe_config: false,
+            auto_sync_grafts: false,
         };
 
         let metrics = Arc::new(EndpointMetrics::mock());