Skip to content

Commit aeb507a

Browse files
committed
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.
1 parent a60fdc9 commit aeb507a

File tree

3 files changed

+82
-22
lines changed

3 files changed

+82
-22
lines changed

docs/config.md

Lines changed: 10 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -5,12 +5,11 @@ CLI. The location of the file is passed with the `--config` command line switch.
55
configuration file, it is not possible to use the options `--postgres-url`,
66
`--postgres-secondary-hosts`, and `--postgres-host-weights`.
77

8-
The TOML file consists of four sections:
8+
The TOML file consists of three sections:
99

10-
- `[chains]` sets the endpoints to blockchain clients.
11-
- `[store]` describes the available databases.
12-
- `[ingestor]` sets the name of the node responsible for block ingestion.
13-
- `[deployment]` describes how to place newly deployed subgraphs.
10+
- [`[chains]`](#configuring-chains) lists the available chains and how to access them.
11+
- [`[store]`](#configuring-multiple-databases) describes the available databases.
12+
- [`[deployment]`](#controlling-deployment) describes how to place newly deployed subgraphs.
1413

1514
Some of these sections support environment variable expansion out of the box,
1615
most notably Postgres connection strings. The official `graph-node` Docker image
@@ -104,8 +103,12 @@ names that will use this configuration file.
104103

105104
The `[chains]` section controls the providers that `graph-node`
106105
connects to, and where blocks and other metadata for each chain are
107-
stored. The section consists of the name of the node doing block ingestion
108-
(currently not used), and a list of chains.
106+
stored. The section consists of the name of the node responsible for block
107+
ingestion and a list of chains. Block ingestion only runs on the node
108+
whose `--node-id` matches the `ingestor` value. The
109+
`--disable-block-ingestor` flag (or `DISABLE_BLOCK_INGESTOR` env var)
110+
acts as a hard override that always prevents ingestion regardless of
111+
the config.
109112

110113
The configuration for a chain `name` is specified in the section
111114
`[chains.<name>]`, with the following:

node/src/config.rs

Lines changed: 59 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,8 @@ impl Default for Opt {
7272
pub struct Config {
7373
#[serde(skip, default = "default_node_id")]
7474
pub node: NodeId,
75+
#[serde(skip)]
76+
pub disable_block_ingestor: bool,
7577
pub general: Option<GeneralSection>,
7678
#[serde(rename = "store")]
7779
pub stores: BTreeMap<String, Shard>,
@@ -104,6 +106,15 @@ fn validate_name(s: &str) -> Result<()> {
104106
}
105107

106108
impl Config {
109+
/// Returns `true` if this node should run block ingestion.
110+
///
111+
/// Block ingestion runs when the `--disable-block-ingestor` kill switch
112+
/// is **not** set and this node's id matches the `ingestor` value from
113+
/// the `[chains]` config section.
114+
pub fn is_block_ingestor(&self) -> bool {
115+
!self.disable_block_ingestor && self.chains.ingestor == self.node.to_string()
116+
}
117+
107118
pub fn chain_ids(&self) -> Vec<ChainName> {
108119
self.chains
109120
.chains
@@ -182,7 +193,7 @@ impl Config {
182193
/// a config from the command line arguments in `opt`
183194
pub fn load(logger: &Logger, opt: &Opt) -> Result<Config> {
184195
if let Some(config) = &opt.config {
185-
Self::from_file(logger, config, &opt.node_id)
196+
Self::from_file(logger, config, opt)
186197
} else {
187198
info!(
188199
logger,
@@ -192,14 +203,16 @@ impl Config {
192203
}
193204
}
194205

195-
pub fn from_file(logger: &Logger, path: &str, node: &str) -> Result<Config> {
206+
pub fn from_file(logger: &Logger, path: &str, opt: &Opt) -> Result<Config> {
196207
info!(logger, "Reading configuration file `{}`", path);
197-
Self::from_str(&read_to_string(path)?, node)
208+
Self::from_str(&read_to_string(path)?, opt)
198209
}
199210

200-
pub fn from_str(config: &str, node: &str) -> Result<Config> {
211+
pub fn from_str(config: &str, opt: &Opt) -> Result<Config> {
201212
let mut config: Config = toml::from_str(config)?;
202-
config.node = NodeId::new(node).map_err(|node| anyhow!("invalid node id {}", node))?;
213+
config.node =
214+
NodeId::new(&opt.node_id).map_err(|node| anyhow!("invalid node id {}", node))?;
215+
config.disable_block_ingestor = opt.disable_block_ingestor;
203216
config.validate()?;
204217
Ok(config)
205218
}
@@ -213,6 +226,7 @@ impl Config {
213226
stores.insert(PRIMARY_SHARD.to_string(), Shard::from_opt(true, opt)?);
214227
Ok(Config {
215228
node,
229+
disable_block_ingestor: opt.disable_block_ingestor,
216230
general: None,
217231
stores,
218232
chains,
@@ -486,13 +500,7 @@ impl ChainSection {
486500
}
487501

488502
fn from_opt(opt: &Opt) -> Result<Self> {
489-
// If we are not the block ingestor, set the node name
490-
// to something that is definitely not our node_id
491-
let ingestor = if opt.disable_block_ingestor {
492-
format!("{} is not ingesting", opt.node_id)
493-
} else {
494-
opt.node_id.clone()
495-
};
503+
let ingestor = opt.node_id.clone();
496504
let mut chains = BTreeMap::new();
497505
Self::parse_networks(&mut chains, Transport::Rpc, &opt.ethereum_rpc)?;
498506
Self::parse_networks(&mut chains, Transport::Ws, &opt.ethereum_ws)?;
@@ -2059,6 +2067,7 @@ fdw_pool_size = [
20592067

20602068
let config = Config {
20612069
node: NodeId::new("test").unwrap(),
2070+
disable_block_ingestor: false,
20622071
general: None,
20632072
stores: {
20642073
let mut s = std::collections::BTreeMap::new();
@@ -2090,4 +2099,42 @@ fdw_pool_size = [
20902099
graph::components::network_provider::ChainName::from("unknown")
20912100
);
20922101
}
2102+
2103+
#[test]
2104+
fn is_block_ingestor() {
2105+
let make_config = |node: &str, ingestor: &str, disable: bool| -> Config {
2106+
let section: ChainSection =
2107+
toml::from_str(&format!(r#"ingestor = "{}""#, ingestor)).unwrap();
2108+
Config {
2109+
node: NodeId::new(node).unwrap(),
2110+
disable_block_ingestor: disable,
2111+
general: None,
2112+
stores: {
2113+
let mut s = std::collections::BTreeMap::new();
2114+
s.insert(
2115+
"primary".to_string(),
2116+
toml::from_str::<Shard>(r#"connection = "postgresql://u:p@h/db""#).unwrap(),
2117+
);
2118+
s
2119+
},
2120+
chains: section,
2121+
deployment: toml::from_str(
2122+
"[[rule]]\nshards = [\"primary\"]\nindexers = [\"test\"]",
2123+
)
2124+
.unwrap(),
2125+
}
2126+
};
2127+
2128+
// Node matches ingestor, not disabled
2129+
assert!(make_config("index-0", "index-0", false).is_block_ingestor());
2130+
2131+
// Node does not match ingestor
2132+
assert!(!make_config("query-0", "index-0", false).is_block_ingestor());
2133+
2134+
// Node matches but kill switch is on
2135+
assert!(!make_config("index-0", "index-0", true).is_block_ingestor());
2136+
2137+
// Node does not match and kill switch is on
2138+
assert!(!make_config("query-0", "index-0", true).is_block_ingestor());
2139+
}
20932140
}

node/src/launcher.rs

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -551,9 +551,19 @@ pub async fn run(
551551
)
552552
.await;
553553

554-
// see comment on cleanup_ethereum_shallow_blocks
555-
if !opt.disable_block_ingestor {
554+
if config.is_block_ingestor() {
555+
// see comment on cleanup_ethereum_shallow_blocks
556556
cleanup_ethereum_shallow_blocks(&blockchain_map, &network_store).await;
557+
} else if opt.disable_block_ingestor {
558+
info!(
559+
logger,
560+
"Block ingestor disabled by --disable-block-ingestor"
561+
);
562+
} else {
563+
info!(
564+
logger,
565+
"Not running block ingestion, ingestor is `{}`", config.chains.ingestor
566+
);
557567
}
558568

559569
let graphql_server = build_graphql_server(
@@ -573,7 +583,7 @@ pub async fn run(
573583
amp_client.cheap_clone(),
574584
);
575585

576-
if !opt.disable_block_ingestor {
586+
if config.is_block_ingestor() {
577587
spawn_block_ingestor(
578588
&logger,
579589
&blockchain_map,

0 commit comments

Comments
 (0)