node: Wire up ChainSection.ingestor to control block ingestion

The `[chains] ingestor` config field was validated and logged but never
actually consulted when deciding whether to run block ingestion. Add
`Config::is_block_ingestor()` as the single source of truth: ingestion
runs only when `--disable-block-ingestor` is not set and the node's id
matches the configured ingestor value.

Author: lutter

docs/config.md

@@ -5,12 +5,11 @@ CLI. The location of the file is passed with the `--config` command line switch.
 configuration file, it is not possible to use the options `--postgres-url`,
 `--postgres-secondary-hosts`, and `--postgres-host-weights`.
 
-The TOML file consists of four sections:
+The TOML file consists of three sections:
 
-- `[chains]` sets the endpoints to blockchain clients.
-- `[store]` describes the available databases.
-- `[ingestor]` sets the name of the node responsible for block ingestion.
-- `[deployment]` describes how to place newly deployed subgraphs.
+- [`[chains]`](#configuring-chains) lists the available chains and how to access them.
+- [`[store]`](#configuring-multiple-databases) describes the available databases.
+- [`[deployment]`](#controlling-deployment) describes how to place newly deployed subgraphs.
 
 Some of these sections support environment variable expansion out of the box,
 most notably Postgres connection strings. The official `graph-node` Docker image
@@ -104,8 +103,12 @@ names that will use this configuration file.
 
 The `[chains]` section controls the providers that `graph-node`
 connects to, and where blocks and other metadata for each chain are
-stored. The section consists of the name of the node doing block ingestion
-(currently not used), and a list of chains.
+stored. The section consists of the name of the node responsible for block
+ingestion and a list of chains. Block ingestion only runs on the node
+whose `--node-id` matches the `ingestor` value. The
+`--disable-block-ingestor` flag (or `DISABLE_BLOCK_INGESTOR` env var)
+acts as a hard override that always prevents ingestion regardless of
+the config.
 
 The configuration for a chain `name` is specified in the section
 `[chains.<name>]`, with the following:

gnd/src/commands/test/runner.rs

@@ -48,7 +48,7 @@ use graph_chain_ethereum::{
 };
 use graph_core::polling_monitor::{arweave_service, ipfs_service};
 use graph_graphql::prelude::GraphQlRunner;
-use graph_node::config::Config;
+use graph_node::config::{Config, Opt};
 use graph_node::manager::PanicSubscriptionManager;
 use graph_node::store_builder::StoreBuilder;
 use graph_store_postgres::{ChainHeadUpdateListener, ChainStore, Store, SubgraphStore};
@@ -434,7 +434,7 @@ ingestor = "default"
         db.url()
     );
 
-    let config = Config::from_str(&config_str, "default")
+    let config = Config::from_str(&config_str, &Opt::default())
         .map_err(|e| anyhow!("Failed to parse config: {}", e))?;
 
     let mock_registry = Arc::new(MetricsRegistry::mock());

node/src/config.rs

@@ -72,6 +72,8 @@ impl Default for Opt {
 pub struct Config {
     #[serde(skip, default = "default_node_id")]
     pub node: NodeId,
+    #[serde(skip)]
+    pub disable_block_ingestor: bool,
     pub general: Option<GeneralSection>,
     #[serde(rename = "store")]
     pub stores: BTreeMap<String, Shard>,
@@ -104,6 +106,15 @@ fn validate_name(s: &str) -> Result<()> {
 }
 
 impl Config {
+    /// Returns `true` if this node should run block ingestion.
+    ///
+    /// Block ingestion runs when the `--disable-block-ingestor` kill switch
+    /// is **not** set and this node's id matches the `ingestor` value from
+    /// the `[chains]` config section.
+    pub fn is_block_ingestor(&self) -> bool {
+        !self.disable_block_ingestor && self.chains.ingestor == self.node.to_string()
+    }
+
     pub fn chain_ids(&self) -> Vec<ChainName> {
         self.chains
             .chains
@@ -182,7 +193,7 @@ impl Config {
     /// a config from the command line arguments in `opt`
     pub fn load(logger: &Logger, opt: &Opt) -> Result<Config> {
         if let Some(config) = &opt.config {
-            Self::from_file(logger, config, &opt.node_id)
+            Self::from_file(logger, config, opt)
         } else {
             info!(
                 logger,
@@ -192,14 +203,16 @@ impl Config {
         }
     }
 
-    pub fn from_file(logger: &Logger, path: &str, node: &str) -> Result<Config> {
+    pub fn from_file(logger: &Logger, path: &str, opt: &Opt) -> Result<Config> {
         info!(logger, "Reading configuration file `{}`", path);
-        Self::from_str(&read_to_string(path)?, node)
+        Self::from_str(&read_to_string(path)?, opt)
     }
 
-    pub fn from_str(config: &str, node: &str) -> Result<Config> {
+    pub fn from_str(config: &str, opt: &Opt) -> Result<Config> {
         let mut config: Config = toml::from_str(config)?;
-        config.node = NodeId::new(node).map_err(|node| anyhow!("invalid node id {}", node))?;
+        config.node =
+            NodeId::new(&opt.node_id).map_err(|node| anyhow!("invalid node id {}", node))?;
+        config.disable_block_ingestor = opt.disable_block_ingestor;
         config.validate()?;
         Ok(config)
     }
@@ -213,6 +226,7 @@ impl Config {
         stores.insert(PRIMARY_SHARD.to_string(), Shard::from_opt(true, opt)?);
         Ok(Config {
             node,
+            disable_block_ingestor: opt.disable_block_ingestor,
             general: None,
             stores,
             chains,
@@ -486,13 +500,7 @@ impl ChainSection {
     }
 
     fn from_opt(opt: &Opt) -> Result<Self> {
-        // If we are not the block ingestor, set the node name
-        // to something that is definitely not our node_id
-        let ingestor = if opt.disable_block_ingestor {
-            format!("{} is not ingesting", opt.node_id)
-        } else {
-            opt.node_id.clone()
-        };
+        let ingestor = opt.node_id.clone();
         let mut chains = BTreeMap::new();
         Self::parse_networks(&mut chains, Transport::Rpc, &opt.ethereum_rpc)?;
         Self::parse_networks(&mut chains, Transport::Ws, &opt.ethereum_ws)?;
@@ -2059,6 +2067,7 @@ fdw_pool_size = [
 
         let config = Config {
             node: NodeId::new("test").unwrap(),
+            disable_block_ingestor: false,
             general: None,
             stores: {
                 let mut s = std::collections::BTreeMap::new();
@@ -2090,4 +2099,42 @@ fdw_pool_size = [
             graph::components::network_provider::ChainName::from("unknown")
         );
     }
+
+    #[test]
+    fn is_block_ingestor() {
+        let make_config = |node: &str, ingestor: &str, disable: bool| -> Config {
+            let section: ChainSection =
+                toml::from_str(&format!(r#"ingestor = "{}""#, ingestor)).unwrap();
+            Config {
+                node: NodeId::new(node).unwrap(),
+                disable_block_ingestor: disable,
+                general: None,
+                stores: {
+                    let mut s = std::collections::BTreeMap::new();
+                    s.insert(
+                        "primary".to_string(),
+                        toml::from_str::<Shard>(r#"connection = "postgresql://u:p@h/db""#).unwrap(),
+                    );
+                    s
+                },
+                chains: section,
+                deployment: toml::from_str(
+                    "[[rule]]\nshards = [\"primary\"]\nindexers = [\"test\"]",
+                )
+                .unwrap(),
+            }
+        };
+
+        // Node matches ingestor, not disabled
+        assert!(make_config("index-0", "index-0", false).is_block_ingestor());
+
+        // Node does not match ingestor
+        assert!(!make_config("query-0", "index-0", false).is_block_ingestor());
+
+        // Node matches but kill switch is on
+        assert!(!make_config("index-0", "index-0", true).is_block_ingestor());
+
+        // Node does not match and kill switch is on
+        assert!(!make_config("query-0", "index-0", true).is_block_ingestor());
+    }
 }

node/src/launcher.rs

@@ -551,9 +551,19 @@ pub async fn run(
         )
         .await;
 
-        // see comment on cleanup_ethereum_shallow_blocks
-        if !opt.disable_block_ingestor {
+        if config.is_block_ingestor() {
+            // see comment on cleanup_ethereum_shallow_blocks
             cleanup_ethereum_shallow_blocks(&blockchain_map, &network_store).await;
+        } else if opt.disable_block_ingestor {
+            info!(
+                logger,
+                "Block ingestor disabled by --disable-block-ingestor"
+            );
+        } else {
+            info!(
+                logger,
+                "Not running block ingestion, ingestor is `{}`", config.chains.ingestor
+            );
         }
 
         let graphql_server = build_graphql_server(
@@ -573,7 +583,7 @@ pub async fn run(
             amp_client.cheap_clone(),
         );
 
-        if !opt.disable_block_ingestor {
+        if config.is_block_ingestor() {
             spawn_block_ingestor(
                 &logger,
                 &blockchain_map,

tests/src/fixture/mod.rs

@@ -52,6 +52,7 @@ use graph_chain_ethereum::chain::RuntimeAdapterBuilder;
 use graph_chain_ethereum::network::EthereumNetworkAdapters;
 use graph_chain_ethereum::Chain;
 use graph_core::polling_monitor::{arweave_service, ipfs_service};
+use graph_node::config::Opt;
 use graph_node::manager::PanicSubscriptionManager;
 use graph_node::{config::Config, store_builder::StoreBuilder};
 use graph_runtime_wasm::RuntimeHostBuilder;
@@ -377,7 +378,7 @@ pub async fn stores(test_name: &str, store_config_path: &str) -> Stores {
             Err(e) => panic!("{}", e.to_string()),
         };
         let config = config.replace("$THEGRAPH_STORE_POSTGRES_DIESEL_URL", &db_url);
-        Config::from_str(&config, "default").expect("failed to create configuration")
+        Config::from_str(&config, &Opt::default()).expect("failed to create configuration")
     };
 
     let logger = test_logger(test_name);