implement recursive auto sync for grafts

Author: dwerner

chain/ethereum/src/chain.rs

@@ -15,6 +15,7 @@ use graph::prelude::{
     EthereumCallCache, LightEthereumBlock, LightEthereumBlockExt, MetricsRegistry,
 };
 use graph::schema::InputSchema;
+use graph::slog::info;
 use graph::substreams::Clock;
 use graph::{
     blockchain::{
@@ -451,6 +452,17 @@ impl Blockchain for Chain {
         logger: &Logger,
         number: BlockNumber,
     ) -> Result<BlockPtr, IngestorError> {
+        // temporary hack until we can refactor this to use the block stream
+        if self.block_stream_builder.can_directly_resolve_blocks() {
+            info!(&logger, "block_pointer_from_number - can_directly_resolve_blocks"; "number" => number);
+            if let Some(block_ptr) = self
+                .block_stream_builder
+                .directly_resolve_block_from_number(number)
+            {
+                return Ok(block_ptr);
+            }
+        }
+        info!(&logger, "block_pointer_from_number"; "number" => number);
         match self.client.as_ref() {
             ChainClient::Firehose(endpoints) => endpoints
                 .endpoint()

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,9 +316,15 @@ 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_graft_sync() {
+            Some(0)
+        } else {
+            None
+        };
+
         let deployment_locator = match kind {
             BlockchainKind::Arweave => {
-                create_subgraph_version::<graph_chain_arweave::Chain, _>(
+                create_subgraph_version::<graph_chain_arweave::Chain, _, _>(
                     &logger,
                     self.store.clone(),
                     self.chains.cheap_clone(),
@@ -342,11 +338,13 @@ where
                     self.version_switching_mode,
                     &self.resolver,
                     history_blocks,
+                    auto_graft_sync_depth,
+                    self.provider.clone(),
                 )
                 .await?
             }
             BlockchainKind::Ethereum => {
-                create_subgraph_version::<graph_chain_ethereum::Chain, _>(
+                create_subgraph_version::<graph_chain_ethereum::Chain, _, _>(
                     &logger,
                     self.store.clone(),
                     self.chains.cheap_clone(),
@@ -360,11 +358,13 @@ where
                     self.version_switching_mode,
                     &self.resolver,
                     history_blocks,
+                    auto_graft_sync_depth,
+                    self.provider.clone(),
                 )
                 .await?
             }
             BlockchainKind::Near => {
-                create_subgraph_version::<graph_chain_near::Chain, _>(
+                create_subgraph_version::<graph_chain_near::Chain, _, _>(
                     &logger,
                     self.store.clone(),
                     self.chains.cheap_clone(),
@@ -378,11 +378,13 @@ where
                     self.version_switching_mode,
                     &self.resolver,
                     history_blocks,
+                    auto_graft_sync_depth,
+                    self.provider.clone(),
                 )
                 .await?
             }
             BlockchainKind::Cosmos => {
-                create_subgraph_version::<graph_chain_cosmos::Chain, _>(
+                create_subgraph_version::<graph_chain_cosmos::Chain, _, _>(
                     &logger,
                     self.store.clone(),
                     self.chains.cheap_clone(),
@@ -396,11 +398,13 @@ where
                     self.version_switching_mode,
                     &self.resolver,
                     history_blocks,
+                    auto_graft_sync_depth,
+                    self.provider.clone(),
                 )
                 .await?
             }
             BlockchainKind::Substreams => {
-                create_subgraph_version::<graph_chain_substreams::Chain, _>(
+                create_subgraph_version::<graph_chain_substreams::Chain, _, _>(
                     &logger,
                     self.store.clone(),
                     self.chains.cheap_clone(),
@@ -414,11 +418,13 @@ where
                     self.version_switching_mode,
                     &self.resolver,
                     history_blocks,
+                    auto_graft_sync_depth,
+                    self.provider.clone(),
                 )
                 .await?
             }
             BlockchainKind::Starknet => {
-                create_subgraph_version::<graph_chain_starknet::Chain, _>(
+                create_subgraph_version::<graph_chain_starknet::Chain, _, _>(
                     &logger,
                     self.store.clone(),
                     self.chains.cheap_clone(),
@@ -432,6 +438,8 @@ where
                     self.version_switching_mode,
                     &self.resolver,
                     history_blocks,
+                    auto_graft_sync_depth,
+                    self.provider.clone(),
                 )
                 .await?
             }
@@ -555,9 +563,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),
@@ -591,20 +599,26 @@ async fn resolve_graft_block(
     chain: &impl Blockchain,
     logger: &Logger,
 ) -> Result<BlockPtr, SubgraphRegistrarError> {
+    debug!(&logger, "Resolve graft block"; "base" => base.base.to_string(), "block" => base.block);
     chain
         .block_pointer_from_number(logger, base.block)
         .await
-        .map_err(|_| {
+        .map_err(|err| {
+            error!(&logger, "Failed to resolve graft block"; "error" => err.to_string());
             SubgraphRegistrarError::ManifestValidationError(vec![
                 SubgraphManifestValidationError::BlockNotFound(format!(
-                    "graft base block {} not found",
-                    base.block
+                    "graft base {} block {} not found",
+                    base.base, base.block
                 )),
             ])
         })
 }
 
-async fn create_subgraph_version<C: Blockchain, S: SubgraphStore>(
+async fn create_subgraph_version<
+    C: Blockchain,
+    S: SubgraphStore,
+    P: SubgraphAssignmentProviderTrait,
+>(
     logger: &Logger,
     store: Arc<S>,
     chains: Arc<BlockchainMap>,
@@ -618,9 +632,13 @@ 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>,
+    provider: Arc<P>,
 ) -> 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 +648,38 @@ 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, P>(
+                graft,
+                resolver,
+                logger,
+                &store,
+                &chains,
+                &name,
+                &node_id,
+                &debug_fork,
+                version_switching_mode,
+                history_blocks_override,
+                depth,
+                provider,
+            ))
+            .await?;
+        } else {
+            warn!(
+                logger,
+                "auto-graft-sync: 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 +772,115 @@ 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, P: SubgraphAssignmentProviderTrait>(
+    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,
+    provider: Arc<P>,
+) -> Result<DeploymentLocator, SubgraphRegistrarError> {
+    info!(
+        logger,
+        "auto-graft-sync: begin graft sync";
+        "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 deployment = graft.base.clone();
+
+    let name = &deployment[deployment.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-graft-sync: create 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" => deployment.to_string());
+
+    let locator = create_subgraph_version::<C, S, P>(
+        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),
+        provider.clone(),
+    )
+    .await?;
+
+    info!(
+        logger,
+        "auto-graft-sync: awaiting subgraph sync";
+        "subgraph" => name.to_string(),
+        "hash" => graft.base.to_string()
+    );
+
+    info!(&logger, "auto-graft-sync: starting graft sync"; "subgraph" => name.to_string(), "hash" => graft.base.to_string());
+    provider
+        .start(locator.clone(), Some(graft.block))
+        .await
+        .map_err(SubgraphRegistrarError::AutoGraftSubgraphAssignmentError)?;
+
+    info!(&logger, "auto-graft-sync: waiting for graft sync"; "subgraph" => name.to_string(), "hash" => graft.base.to_string());
+    graft
+        .await_sync(store.clone(), Duration::from_secs(1))
+        .await?;
+
+    info!(
+        logger,
+        "auto-graft-sync: sync complete";
+        "subgraph" => name.to_string(),
+        "graft-hash" => graft.base.to_string(),
+        "depth" => depth,
+        "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/blockchain/block_stream.rs

@@ -118,6 +118,14 @@ pub trait BlockRefetcher<C: Blockchain>: Send + Sync {
 /// BlockStreamBuilder is an abstraction that would separate the logic for building streams from the blockchain trait
 #[async_trait]
 pub trait BlockStreamBuilder<C: Blockchain>: Send + Sync {
+    fn can_directly_resolve_blocks(&self) -> bool {
+        false
+    }
+
+    fn directly_resolve_block_from_number(&self, _number: BlockNumber) -> Option<BlockPtr> {
+        None
+    }
+
     async fn build_firehose(
         &self,
         chain: &C,