feat: initial support for graph-node versions

graph-gateway/src/config.rs

@@ -45,7 +45,10 @@ pub struct Config {
     /// L2 gateway to forward client queries to
     #[serde_as(as = "Option<DisplayFromStr>")]
     pub l2_gateway: Option<Url>,
-    /// Minimum indexer software version that will receive queries
+    /// Minimum graph-node version that will receive queries
+    #[serde_as(as = "DisplayFromStr")]
+    pub min_graph_node_version: Version,
+    /// Minimum indexer-service version that will receive queries
     #[serde_as(as = "DisplayFromStr")]
     pub min_indexer_version: Version,
     /// Network subgraph query path

graph-gateway/src/indexers/indexing.rs

@@ -33,12 +33,14 @@ pub struct Status {
 pub async fn statuses(
     deployments: Eventual<Ptr<HashMap<DeploymentId, Arc<Deployment>>>>,
     client: reqwest::Client,
-    min_version: Version,
+    min_graph_node_version: Version,
+    min_indexer_service_version: Version,
     geoip: Option<GeoIP>,
 ) -> Eventual<Ptr<HashMap<Indexing, Status>>> {
     let (indexing_statuses_tx, indexing_statuses_rx) = Eventual::new();
     let actor: &'static Mutex<Actor> = Box::leak(Box::new(Mutex::new(Actor {
-        min_version,
+        min_graph_node_version,
+        min_indexer_service_version,
         geoip,
         dns_resolver: DNSResolver::tokio_from_system_conf().unwrap(),
         geoblocking_cache: Default::default(),
@@ -65,7 +67,8 @@ pub async fn statuses(
 }
 
 struct Actor {
-    min_version: Version,
+    min_graph_node_version: Version,
+    min_indexer_service_version: Version,
     geoip: Option<GeoIP>,
     dns_resolver: DNSResolver,
     geoblocking_cache: HashMap<String, Result<(), String>>,
@@ -133,12 +136,22 @@ async fn update_indexer(
     let service_version = version::query_indexer_service_version(client, version_url)
         .await
         .map_err(|err| anyhow::anyhow!("IndexerVersionError({err})"))?;
+    let status_url = url.join("status")?;
+    let graph_node_version = version::query_graph_node_version(client, status_url).await;
 
     let mut locked_actor = actor.lock().await;
     ensure!(
-        service_version >= locked_actor.min_version,
+        service_version >= locked_actor.min_indexer_service_version,
         "IndexerServiceVersionBelowMinimum({service_version})",
     );
+    // TODO: Strongly enforce graph-node version, by removing this statement, after more indexers
+    // update their indexer-service.
+    let graph_node_version =
+        graph_node_version.unwrap_or_else(|_| locked_actor.min_graph_node_version.clone());
+    ensure!(
+        graph_node_version >= locked_actor.min_graph_node_version,
+        "GraphNodeVersionBelowMinimum({graph_node_version})",
+    );
     apply_geoblocking(&mut locked_actor, &url).await?;
     drop(locked_actor);
 

graph-gateway/src/indexers/version.rs

@@ -1,3 +1,4 @@
+use graphql_http::http_client::ReqwestExt as _;
 use semver::Version;
 use serde::Deserialize;
 
@@ -14,6 +15,20 @@ pub async fn query_indexer_service_version(
     Ok(response.version)
 }
 
+pub async fn query_graph_node_version(
+    client: &reqwest::Client,
+    status_url: reqwest::Url,
+) -> anyhow::Result<Version> {
+    let query = "{ version { version } }";
+    let response: GraphNodeVersion = client.post(status_url).send_graphql(query).await??;
+    Ok(response.version.version)
+}
+
+#[derive(Debug, Deserialize)]
+struct GraphNodeVersion {
+    version: IndexerVersion,
+}
+
 #[derive(Debug, Deserialize)]
 struct IndexerVersion {
     version: Version,

graph-gateway/src/main.rs

@@ -173,6 +173,7 @@ async fn main() {
     let indexing_statuses = indexing::statuses(
         network.deployments.clone(),
         http_client.clone(),
+        config.min_graph_node_version,
         config.min_indexer_version,
         geoip,
     )