all: Introduce subgraph-specific settings

It is now possible to set the history_blocks for a subgraph from a config
file that matches on subgraph names.

Settings are matched in the order in which they appear in the config file.

`graphman config check` checks that config file for validity if one is set
in the environment

Author: lutter

core/src/subgraph/registrar.rs

@@ -6,6 +6,7 @@ use graph::blockchain::Blockchain;
 use graph::blockchain::BlockchainKind;
 use graph::blockchain::BlockchainMap;
 use graph::components::store::{DeploymentId, DeploymentLocator, SubscriptionManager};
+use graph::components::subgraph::Settings;
 use graph::data::subgraph::schema::DeploymentCreate;
 use graph::data::subgraph::Graft;
 use graph::prelude::{
@@ -24,6 +25,7 @@ pub struct SubgraphRegistrar<P, S, SM> {
     node_id: NodeId,
     version_switching_mode: SubgraphVersionSwitchingMode,
     assignment_event_stream_cancel_guard: CancelGuard, // cancels on drop
+    settings: Arc<Settings>,
 }
 
 impl<P, S, SM> SubgraphRegistrar<P, S, SM>
@@ -41,6 +43,7 @@ where
         chains: Arc<BlockchainMap>,
         node_id: NodeId,
         version_switching_mode: SubgraphVersionSwitchingMode,
+        settings: Arc<Settings>,
     ) -> Self {
         let logger = logger_factory.component_logger("SubgraphRegistrar", None);
         let logger_factory = logger_factory.with_parent(logger.clone());
@@ -58,6 +61,7 @@ where
             node_id,
             version_switching_mode,
             assignment_event_stream_cancel_guard: CancelGuard::new(),
+            settings,
         }
     }
 
@@ -297,6 +301,10 @@ where
             SubgraphRegistrarError::ResolveError(SubgraphManifestResolveError::ResolveError(e))
         })?;
 
+        // Give priority to deployment specific history_blocks value.
+        let history_blocks =
+            history_blocks.or(self.settings.for_name(&name).map(|c| c.history_blocks));
+
         let deployment_locator = match kind {
             BlockchainKind::Arweave => {
                 create_subgraph_version::<graph_chain_arweave::Chain, _>(

graph/src/components/subgraph/mod.rs

@@ -4,6 +4,7 @@ mod instance_manager;
 mod proof_of_indexing;
 mod provider;
 mod registrar;
+mod settings;
 
 pub use crate::prelude::Entity;
 
@@ -16,3 +17,4 @@ pub use self::proof_of_indexing::{
 };
 pub use self::provider::SubgraphAssignmentProvider;
 pub use self::registrar::{SubgraphRegistrar, SubgraphVersionSwitchingMode};
+pub use self::settings::{Setting, Settings};

graph/src/components/subgraph/settings.rs

@@ -0,0 +1,94 @@
+//! Facilities for dealing with subgraph-specific settings
+use std::fs::read_to_string;
+
+use crate::{
+    anyhow,
+    prelude::{regex::Regex, SubgraphName},
+};
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub enum Predicate {
+    #[serde(alias = "name", with = "serde_regex")]
+    Name(Regex),
+}
+
+impl Predicate {
+    fn matches(&self, name: &SubgraphName) -> bool {
+        match self {
+            Predicate::Name(rx) => rx.is_match(name.as_str()),
+        }
+    }
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct Setting {
+    #[serde(alias = "match")]
+    pred: Predicate,
+    pub history_blocks: i32,
+}
+
+impl Setting {
+    fn matches(&self, name: &SubgraphName) -> bool {
+        self.pred.matches(name)
+    }
+}
+
+#[derive(Clone, Debug, Default, Deserialize, Serialize)]
+pub struct Settings {
+    #[serde(alias = "setting")]
+    settings: Vec<Setting>,
+}
+
+impl Settings {
+    pub fn from_file(path: &str) -> Result<Self, anyhow::Error> {
+        Self::from_str(&read_to_string(path)?)
+    }
+
+    pub fn from_str(toml: &str) -> Result<Self, anyhow::Error> {
+        toml::from_str::<Self>(toml).map_err(anyhow::Error::from)
+    }
+
+    pub fn for_name(&self, name: &SubgraphName) -> Option<&Setting> {
+        self.settings.iter().find(|setting| setting.matches(name))
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::{Predicate, Settings};
+
+    #[test]
+    fn parses_correctly() {
+        let content = r#"
+        [[setting]]
+        match = { name = ".*" }
+        history_blocks = 10000
+
+        [[setting]]
+        match = { name = "xxxxx" }
+        history_blocks = 10000
+
+        [[setting]]
+        match = { name = ".*!$" }
+        history_blocks = 10000
+        "#;
+
+        let section = Settings::from_str(content).unwrap();
+        assert_eq!(section.settings.len(), 3);
+
+        let rule1 = match &section.settings[0].pred {
+            Predicate::Name(name) => name,
+        };
+        assert_eq!(rule1.as_str(), ".*");
+
+        let rule2 = match &section.settings[1].pred {
+            Predicate::Name(name) => name,
+        };
+        assert_eq!(rule2.as_str(), "xxxxx");
+        let rule1 = match &section.settings[2].pred {
+            Predicate::Name(name) => name,
+        };
+        assert_eq!(rule1.as_str(), ".*!$");
+    }
+}

graph/src/env/mod.rs

@@ -172,6 +172,9 @@ pub struct EnvVars {
     /// Set by the environment variable `ETHEREUM_REORG_THRESHOLD`. The default
     /// value is 250 blocks.
     pub reorg_threshold: BlockNumber,
+    /// Set by the env var `GRAPH_EXPERIMENTAL_SUBGRAPH_SETTINGS` which should point
+    /// to a file with subgraph-specific settings
+    pub subgraph_settings: Option<String>,
 }
 
 impl EnvVars {
@@ -229,6 +232,7 @@ impl EnvVars {
             external_ws_base_url: inner.external_ws_base_url,
             static_filters_threshold: inner.static_filters_threshold,
             reorg_threshold: inner.reorg_threshold,
+            subgraph_settings: inner.subgraph_settings,
         })
     }
 
@@ -347,6 +351,8 @@ struct Inner {
     // JSON-RPC specific.
     #[envconfig(from = "ETHEREUM_REORG_THRESHOLD", default = "250")]
     reorg_threshold: BlockNumber,
+    #[envconfig(from = "GRAPH_EXPERIMENTAL_SUBGRAPH_SETTINGS")]
+    subgraph_settings: Option<String>,
 }
 
 #[derive(Clone, Debug)]

node/src/main.rs

@@ -9,6 +9,7 @@ use graph::blockchain::{
     BasicBlockchainBuilder, Blockchain, BlockchainBuilder, BlockchainKind, BlockchainMap,
 };
 use graph::components::store::BlockStore;
+use graph::components::subgraph::Settings;
 use graph::data::graphql::effort::LoadManager;
 use graph::endpoint::EndpointMetrics;
 use graph::env::EnvVars;
@@ -137,6 +138,21 @@ async fn main() {
         }
         Ok(config) => config,
     };
+
+    let subgraph_settings = match env_vars.subgraph_settings {
+        Some(ref path) => {
+            info!(logger, "Reading subgraph configuration file `{}`", path);
+            match Settings::from_file(path) {
+                Ok(rules) => rules,
+                Err(e) => {
+                    eprintln!("configuration error in subgraph settings {}: {}", path, e);
+                    std::process::exit(1);
+                }
+            }
+        }
+        None => Settings::default(),
+    };
+
     if opt.check_config {
         match config.to_json() {
             Ok(txt) => println!("{}", txt),
@@ -481,6 +497,7 @@ async fn main() {
             blockchain_map,
             node_id.clone(),
             version_switching_mode,
+            Arc::new(subgraph_settings),
         ));
         graph::spawn(
             subgraph_registrar

node/src/manager/commands/config.rs

@@ -2,7 +2,9 @@ use std::{collections::BTreeMap, sync::Arc};
 
 use graph::{
     anyhow::bail,
+    components::subgraph::Settings,
     endpoint::EndpointMetrics,
+    env::EnvVars,
     itertools::Itertools,
     prelude::{
         anyhow::{anyhow, Error},
@@ -39,13 +41,27 @@ pub fn check(config: &Config, print: bool) -> Result<(), Error> {
         Ok(txt) => {
             if print {
                 println!("{}", txt);
-            } else {
-                println!("Successfully validated configuration");
+                return Ok(());
             }
-            Ok(())
         }
-        Err(e) => Err(anyhow!("error serializing config: {}", e)),
+        Err(e) => bail!("error serializing config: {}", e),
     }
+
+    let env_vars = EnvVars::from_env().unwrap();
+    if let Some(path) = &env_vars.subgraph_settings {
+        match Settings::from_file(path) {
+            Ok(_) => {
+                println!("Successfully validated subgraph settings from {path}");
+            }
+            Err(e) => {
+                eprintln!("configuration error in subgraph settings {}: {}", path, e);
+                std::process::exit(1);
+            }
+        }
+    };
+
+    println!("Successfully validated configuration");
+    Ok(())
 }
 
 pub fn pools(config: &Config, nodes: Vec<String>, shard: bool) -> Result<(), Error> {

node/src/manager/commands/run.rs

@@ -17,6 +17,7 @@ use graph::blockchain::client::ChainClient;
 use graph::blockchain::{BlockchainKind, BlockchainMap};
 use graph::cheap_clone::CheapClone;
 use graph::components::store::{BlockStore as _, DeploymentLocator};
+use graph::components::subgraph::Settings;
 use graph::endpoint::EndpointMetrics;
 use graph::env::EnvVars;
 use graph::firehose::FirehoseEndpoints;
@@ -198,6 +199,7 @@ pub async fn run(
         blockchain_map,
         node_id.clone(),
         SubgraphVersionSwitchingMode::Instant,
+        Arc::new(Settings::default()),
     ));
 
     let (name, hash) = if subgraph.contains(':') {

tests/src/fixture/mod.rs

@@ -18,6 +18,7 @@ use graph::blockchain::{
 use graph::cheap_clone::CheapClone;
 use graph::components::metrics::MetricsRegistry;
 use graph::components::store::{BlockStore, DeploymentLocator};
+use graph::components::subgraph::Settings;
 use graph::data::graphql::effort::LoadManager;
 use graph::data::query::{Query, QueryTarget};
 use graph::data::subgraph::schema::{SubgraphError, SubgraphHealth};
@@ -408,6 +409,7 @@ pub async fn setup<C: Blockchain>(
         blockchain_map.clone(),
         node_id.clone(),
         SubgraphVersionSwitchingMode::Instant,
+        Arc::new(Settings::default()),
     ));
 
     SubgraphRegistrar::create_subgraph(subgraph_registrar.as_ref(), subgraph_name.clone())