gnd: Integration tests

Cargo.toml

@@ -55,6 +55,7 @@ diesel = { version = "2.2.7", features = [
     "chrono",
     "i-implement-a-third-party-backend-and-opt-into-breaking-changes",
 ] }
+pq-sys = { version = "0.6", features = ["bundled"] }
 diesel-derive-enum = { version = "2.1.0", features = ["postgres"] }
 diesel-dynamic-schema = { version = "0.2.3", features = ["postgres"] }
 diesel_derives = "2.2.3"

node/Cargo.toml

@@ -48,4 +48,4 @@ globset = "0.4.16"
 notify = "8.0.0"
 
 [target.'cfg(unix)'.dependencies]
-pgtemp = { git = "https://github.com/incrypto32/pgtemp", branch = "initdb-args" }
+pgtemp = { git = "https://github.com/graphprotocol/pgtemp", branch = "initdb-args" }

node/src/bin/dev.rs

@@ -1,4 +1,4 @@
-use std::{mem, path::Path, sync::Arc};
+use std::{path::Path, sync::Arc};
 
 use anyhow::{Context, Result};
 use clap::Parser;
@@ -13,14 +13,15 @@ use graph::{
 };
 use graph_core::polling_monitor::ipfs_service;
 use graph_node::{
+    dev::watcher,
     dev::watcher::{parse_manifest_args, watch_subgraphs},
     launcher,
     opt::Opt,
 };
 use lazy_static::lazy_static;
 
 #[cfg(unix)]
-use pgtemp::PgTempDBBuilder;
+use pgtemp::{PgTempDB, PgTempDBBuilder};
 
 git_testament!(TESTAMENT);
 lazy_static! {
@@ -90,6 +91,35 @@ pub struct DevOpt {
         default_value = "https://api.thegraph.com/ipfs"
     )]
     pub ipfs: Vec<String>,
+    #[clap(
+        long,
+        default_value = "8000",
+        value_name = "PORT",
+        help = "Port for the GraphQL HTTP server",
+        env = "GRAPH_GRAPHQL_HTTP_PORT"
+    )]
+    pub http_port: u16,
+    #[clap(
+        long,
+        default_value = "8030",
+        value_name = "PORT",
+        help = "Port for the index node server"
+    )]
+    pub index_node_port: u16,
+    #[clap(
+        long,
+        default_value = "8020",
+        value_name = "PORT",
+        help = "Port for the JSON-RPC admin server"
+    )]
+    pub admin_port: u16,
+    #[clap(
+        long,
+        default_value = "8040",
+        value_name = "PORT",
+        help = "Port for the Prometheus metrics server"
+    )]
+    pub metrics_port: u16,
 }
 
 /// Builds the Graph Node options from DevOpt
@@ -109,7 +139,12 @@ fn build_args(dev_opt: &DevOpt, db_url: &str) -> Result<Opt> {
     args.push("--postgres-url".to_string());
     args.push(db_url.to_string());
 
-    let opt = Opt::parse_from(args);
+    let mut opt = Opt::parse_from(args);
+
+    opt.http_port = dev_opt.http_port;
+    opt.admin_port = dev_opt.admin_port;
+    opt.metrics_port = dev_opt.metrics_port;
+    opt.index_node_port = dev_opt.index_node_port;
 
     Ok(opt)
 }
@@ -118,7 +153,7 @@ async fn run_graph_node(
     logger: &Logger,
     opt: Opt,
     link_resolver: Arc<dyn LinkResolver>,
-    subgraph_updates_channel: Option<mpsc::Receiver<(DeploymentHash, SubgraphName)>>,
+    subgraph_updates_channel: mpsc::Receiver<(DeploymentHash, SubgraphName)>,
 ) -> Result<()> {
     let env_vars = Arc::new(EnvVars::from_env().context("Failed to load environment variables")?);
 
@@ -139,16 +174,19 @@ async fn run_graph_node(
         env_vars,
         ipfs_service,
         link_resolver,
-        subgraph_updates_channel,
+        Some(subgraph_updates_channel),
     )
     .await;
     Ok(())
 }
 
 /// Get the database URL, either from the provided option or by creating a temporary database
-fn get_database_url(postgres_url: Option<&String>, database_dir: &Path) -> Result<String> {
+fn get_database_url(
+    postgres_url: Option<&String>,
+    database_dir: &Path,
+) -> Result<(String, Option<PgTempDB>)> {
     if let Some(url) = postgres_url {
-        Ok(url.clone())
+        Ok((url.clone(), None))
     } else {
         #[cfg(unix)]
         {
@@ -162,13 +200,14 @@ fn get_database_url(postgres_url: Option<&String>, database_dir: &Path) -> Resul
 
             let db = PgTempDBBuilder::new()
                 .with_data_dir_prefix(database_dir)
-                .with_initdb_param("-E", "UTF8")
-                .with_initdb_param("--locale", "C")
+                .persist_data(false)
+                .with_initdb_arg("-E", "UTF8")
+                .with_initdb_arg("--locale", "C")
                 .start();
             let url = db.connection_uri().to_string();
-            // Prevent the database from being dropped by forgetting it
-            mem::forget(db);
-            Ok(url)
+            // Return the handle so it lives for the lifetime of the program; dropping it will
+            // shut down Postgres and remove the temporary directory automatically.
+            Ok((url, Some(db)))
         }
 
         #[cfg(not(unix))]
@@ -182,49 +221,75 @@ fn get_database_url(postgres_url: Option<&String>, database_dir: &Path) -> Resul
 
 #[tokio::main]
 async fn main() -> Result<()> {
+    std::env::set_var("ETHEREUM_REORG_THRESHOLD", "10");
+    std::env::set_var("GRAPH_NODE_DISABLE_DEPLOYMENT_HASH_VALIDATION", "true");
     env_logger::init();
     let dev_opt = DevOpt::parse();
 
     let database_dir = Path::new(&dev_opt.database_dir);
 
     let logger = logger(true);
 
-    info!(logger, "Starting Graph Node Dev");
+    info!(logger, "Starting Graph Node Dev 1");
     info!(logger, "Database directory: {}", database_dir.display());
 
-    // Get the database URL
-    let db_url = get_database_url(dev_opt.postgres_url.as_ref(), database_dir)?;
+    // Get the database URL and keep the temporary database handle alive for the life of the
+    // program so that it is dropped (and cleaned up) on graceful shutdown.
+    let (db_url, mut temp_db_opt) = get_database_url(dev_opt.postgres_url.as_ref(), database_dir)?;
 
     let opt = build_args(&dev_opt, &db_url)?;
 
     let (manifests_paths, source_subgraph_aliases) =
         parse_manifest_args(dev_opt.manifests, dev_opt.sources, &logger)?;
     let file_link_resolver = Arc::new(FileLinkResolver::new(None, source_subgraph_aliases.clone()));
 
-    let (tx, rx) = dev_opt.watch.then(|| mpsc::channel(1)).unzip();
+    let (tx, rx) = mpsc::channel(1);
 
     let logger_clone = logger.clone();
     graph::spawn(async move {
         let _ = run_graph_node(&logger_clone, opt, file_link_resolver, rx).await;
     });
 
-    if let Some(tx) = tx {
+    if let Err(e) =
+        watcher::deploy_all_subgraphs(&logger, &manifests_paths, &source_subgraph_aliases, &tx)
+            .await
+    {
+        error!(logger, "Error deploying subgraphs"; "error" => e.to_string());
+        std::process::exit(1);
+    }
+
+    if dev_opt.watch {
+        let logger_clone_watch = logger.clone();
         graph::spawn_blocking(async move {
             if let Err(e) = watch_subgraphs(
-                &logger,
+                &logger_clone_watch,
                 manifests_paths,
                 source_subgraph_aliases,
                 vec!["pgtemp-*".to_string()],
                 tx,
             )
             .await
             {
-                error!(logger, "Error watching subgraphs"; "error" => e.to_string());
+                error!(logger_clone_watch, "Error watching subgraphs"; "error" => e.to_string());
                 std::process::exit(1);
             }
         });
     }
 
-    graph::futures03::future::pending::<()>().await;
+    // Wait for Ctrl+C so we can shut down cleanly and drop the temporary database, which removes
+    // the data directory.
+    tokio::signal::ctrl_c()
+        .await
+        .expect("Failed to listen for Ctrl+C signal");
+    info!(logger, "Received Ctrl+C, shutting down.");
+
+    // Explicitly shut down and clean up the temporary database directory if we started one.
+    if let Some(db) = temp_db_opt.take() {
+        db.shutdown();
+    }
+
+    std::process::exit(0);
+
+    #[allow(unreachable_code)]
     Ok(())
 }

node/src/dev/watcher.rs

@@ -255,7 +255,7 @@ fn is_relevant_event(event: &Event, watched_dirs: Vec<PathBuf>, exclusion_set: &
 }
 
 /// Redeploys all subgraphs in the order it appears in the manifests_paths
-async fn deploy_all_subgraphs(
+pub async fn deploy_all_subgraphs(
     logger: &Logger,
     manifests_paths: &Vec<PathBuf>,
     source_subgraph_aliases: &HashMap<String, PathBuf>,

store/postgres/Cargo.toml

@@ -9,6 +9,7 @@ blake3 = "1.8"
 chrono = { workspace = true }
 derive_more = { version = "2.0.1", features = ["full"] }
 diesel = { workspace = true }
+pq-sys = { workspace = true}
 diesel-dynamic-schema = { workspace = true }
 diesel-derive-enum = { workspace = true }
 diesel_derives = { workspace = true }

tests/src/config.rs

@@ -1,3 +1,4 @@
+use std::sync::OnceLock;
 use std::time::{Duration, Instant};
 use std::{fs, path::PathBuf};
 
@@ -13,6 +14,15 @@ use crate::status;
 
 lazy_static! {
     pub static ref CONFIG: Config = Config::default();
+    static ref DEV_MODE: OnceLock<bool> = OnceLock::new();
+}
+
+pub fn set_dev_mode(val: bool) {
+    DEV_MODE.set(val).expect("DEV_MODE already set");
+}
+
+pub fn dev_mode() -> bool {
+    *DEV_MODE.get().unwrap_or(&false)
 }
 
 #[derive(Clone, Debug)]
@@ -117,6 +127,26 @@ impl GraphNodeConfig {
             }
         }
     }
+
+    pub fn from_env() -> Self {
+        if dev_mode() {
+            Self::gnd()
+        } else {
+            Self::default()
+        }
+    }
+
+    fn gnd() -> Self {
+        let bin = fs::canonicalize("../target/debug/gnd")
+            .expect("failed to infer `graph-node` program location. (Was it built already?)");
+
+        Self {
+            bin,
+            ports: GraphNodePorts::default(),
+            ipfs_uri: "http://localhost:3001".to_string(),
+            log_file: TestFile::new("integration-tests/graph-node.log"),
+        }
+    }
 }
 
 impl Default for GraphNodeConfig {
@@ -145,6 +175,13 @@ pub struct Config {
 
 impl Config {
     pub async fn spawn_graph_node(&self) -> anyhow::Result<Child> {
+        self.spawn_graph_node_with_args(&[]).await
+    }
+
+    pub async fn spawn_graph_node_with_args(
+        &self,
+        additional_args: &[&str],
+    ) -> anyhow::Result<Child> {
         let ports = &self.graph_node.ports;
 
         let args = [
@@ -163,6 +200,12 @@ impl Config {
             "--metrics-port",
             &ports.metrics.to_string(),
         ];
+
+        let args = args
+            .iter()
+            .chain(additional_args.iter())
+            .cloned()
+            .collect::<Vec<_>>();
         let stdout = self.graph_node.log_file.create();
         let stderr = stdout.try_clone()?;
         status!(
@@ -174,7 +217,7 @@ impl Config {
         command
             .stdout(stdout)
             .stderr(stderr)
-            .args(args)
+            .args(args.clone())
             .env("GRAPH_STORE_WRITE_BATCH_DURATION", "5")
             .env("ETHEREUM_REORG_THRESHOLD", "0");
 
@@ -254,7 +297,7 @@ impl Default for Config {
                 port: 3021,
                 host: "localhost".to_string(),
             },
-            graph_node: GraphNodeConfig::default(),
+            graph_node: GraphNodeConfig::from_env(),
             graph_cli,
             num_parallel_tests,
             timeout: Duration::from_secs(600),