node: Wire file watching in dev mode to redeploy subgraphs

Author: incrypto32

node/src/bin/dev.rs

@@ -3,8 +3,17 @@ use std::{path::Path, sync::Arc};
 use anyhow::{Context, Result};
 use clap::Parser;
 use git_testament::{git_testament, render_testament};
-use graph::{components::link_resolver::FileLinkResolver, env::EnvVars, tokio};
-use graph_node::{launcher, opt::Opt};
+use graph::{
+    components::link_resolver::FileLinkResolver,
+    env::EnvVars,
+    log::logger,
+    tokio::{self, sync::mpsc},
+};
+use graph_node::{
+    dev::{helpers::DevModeContext, watcher::watch_subgraph_dir},
+    launcher,
+    opt::Opt,
+};
 use lazy_static::lazy_static;
 use pgtemp::PgTempDBBuilder;
 
@@ -101,15 +110,16 @@ fn get_build_dir(manifest_path_str: &str) -> Result<std::path::PathBuf> {
         .context("Failed to canonicalize build directory path")
 }
 
-async fn run_graph_node(opt: Opt, file_link_resolver: Arc<FileLinkResolver>) -> Result<()> {
+async fn run_graph_node(opt: Opt, ctx: Option<DevModeContext>) -> Result<()> {
     let env_vars = Arc::new(EnvVars::from_env().context("Failed to load environment variables")?);
 
-    launcher::run(opt, env_vars, Some(file_link_resolver)).await;
+    launcher::run(opt, env_vars, ctx).await;
     Ok(())
 }
 
 #[tokio::main]
 async fn main() -> Result<()> {
+    env_logger::init();
     let dev_opt = DevOpt::parse();
 
     let build_dir = get_build_dir(&dev_opt.manifest)?;
@@ -121,11 +131,39 @@ async fn main() -> Result<()> {
         .start_async()
         .await;
 
+    let (tx, rx) = mpsc::channel(1);
     let opt = build_args(&dev_opt, &db.connection_uri(), &dev_opt.manifest)?;
     let file_link_resolver = Arc::new(FileLinkResolver::with_base_dir(&build_dir));
 
+    let ctx = DevModeContext {
+        watch: dev_opt.watch,
+        file_link_resolver,
+        updates_rx: rx,
+    };
+
+    let subgraph = opt.subgraph.clone().unwrap();
+
+    // Set up logger
+    let logger = logger(opt.debug);
+
     // Run graph node
-    run_graph_node(opt, file_link_resolver).await?;
+    graph::spawn(async move {
+        let _ = run_graph_node(opt, Some(ctx)).await;
+    });
+
+    if dev_opt.watch {
+        graph::spawn_blocking(async move {
+            watch_subgraph_dir(
+                &logger,
+                build_dir,
+                subgraph,
+                vec!["pgtemp-*".to_string()],
+                tx,
+            )
+            .await;
+        });
+    }
 
+    graph::futures03::future::pending::<()>().await;
     Ok(())
 }

node/src/launcher.rs

@@ -6,11 +6,12 @@ use graph::futures03::compat::Future01CompatExt;
 use graph::futures03::future::TryFutureExt;
 
 use crate::config::Config;
+use crate::dev::helpers::{watch_subgraph_updates, DevModeContext};
 use crate::network_setup::Networks;
 use crate::opt::Opt;
 use crate::store_builder::StoreBuilder;
 use graph::blockchain::{Blockchain, BlockchainKind, BlockchainMap};
-use graph::components::link_resolver::{ArweaveClient, FileLinkResolver, FileSizeLimit};
+use graph::components::link_resolver::{ArweaveClient, FileSizeLimit};
 use graph::components::subgraph::Settings;
 use graph::data::graphql::load_manager::LoadManager;
 use graph::endpoint::EndpointMetrics;
@@ -347,12 +348,7 @@ fn build_graphql_server(
     graphql_server
 }
 
-pub async fn run(
-    opt: Opt,
-    env_vars: Arc<EnvVars>,
-    file_link_resolver: Option<Arc<FileLinkResolver>>,
-) {
-    env_logger::init();
+pub async fn run(opt: Opt, env_vars: Arc<EnvVars>, dev_ctx: Option<DevModeContext>) {
     // Set up logger
     let logger = logger(opt.debug);
 
@@ -440,9 +436,8 @@ pub async fn run(
 
     // Convert the clients into a link resolver. Since we want to get past
     // possible temporary DNS failures, make the resolver retry
-    let link_resolver: Arc<dyn LinkResolver> = if let Some(file_link_resolver) = file_link_resolver
-    {
-        file_link_resolver
+    let link_resolver: Arc<dyn LinkResolver> = if let Some(dev_ctx) = &dev_ctx {
+        dev_ctx.file_link_resolver.clone()
     } else {
         Arc::new(IpfsResolver::new(ipfs_client, env_vars.cheap_clone()))
     };
@@ -563,7 +558,7 @@ pub async fn run(
 
         // Add the CLI subgraph with a REST request to the admin server.
         if let Some(subgraph) = subgraph {
-            deploy_subgraph_from_flag(subgraph, &opt, subgraph_registrar.clone(), node_id);
+            deploy_subgraph_from_flag(subgraph, &opt, subgraph_registrar.clone(), node_id.clone());
         }
 
         // Serve GraphQL queries over HTTP
@@ -578,6 +573,23 @@ pub async fn run(
                 .await
                 .expect("Failed to start metrics server")
         });
+
+        // If we are in dev mode, watch for subgraph updates
+        // And drop and recreate the subgraph when it changes
+        if let Some(dev_ctx) = dev_ctx {
+            if dev_ctx.watch {
+                graph::spawn(async move {
+                    watch_subgraph_updates(
+                        &logger,
+                        network_store.subgraph_store(),
+                        subgraph_registrar.clone(),
+                        node_id.clone(),
+                        dev_ctx.updates_rx,
+                    )
+                    .await;
+                });
+            }
+        }
     };
 
     graph::spawn(launch_services(logger.clone(), env_vars.cheap_clone()));

node/src/main.rs

@@ -23,6 +23,7 @@ fn main() {
 }
 
 async fn main_inner() {
+    env_logger::init();
     let env_vars = Arc::new(EnvVars::from_env().unwrap());
     let opt = opt::Opt::parse();