diff --git a/Cargo.lock b/Cargo.lock index 0eeb5a9a..5dbe4ce6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2434,7 +2434,6 @@ dependencies = [ "fake", "futures-lite", "graph-networks-registry 0.7.0", - "indoc", "jsonrpsee", "rand 0.9.2", "reqwest 0.12.28", @@ -2445,7 +2444,6 @@ dependencies = [ "sqlx", "test-with", "thegraph-core", - "thegraph-graphql-http", "thiserror 2.0.18", "time", "tokio", @@ -6555,19 +6553,6 @@ dependencies = [ "thiserror 2.0.18", ] -[[package]] -name = "thegraph-graphql-http" -version = "0.4.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "47039e9174267bf71f7ad9ff0af478f31df576d3ec6adcbd88e406cf19546e4a" -dependencies = [ - "async-trait", - "reqwest 0.12.28", - "serde", - "serde_json", - "thiserror 1.0.69", -] - [[package]] name = "thiserror" version = "1.0.69" diff --git a/bin/dipper-service/Cargo.toml b/bin/dipper-service/Cargo.toml index faff2d72..4d235892 100644 --- a/bin/dipper-service/Cargo.toml +++ b/bin/dipper-service/Cargo.toml @@ -15,7 +15,6 @@ dipper-pgregistry = { path = "../../dipper-pgregistry" } dipper-rpc = { version = "0.1.0", path = "../../dipper-rpc", features = ["admin-rpc", "indexer-rpc"] } futures-lite = { version = "2.4.0", default-features = false } graph-networks-registry.workspace = true -indoc = "2.0.5" jsonrpsee = { workspace = true, features = ["server"] } rand = "0.9.0" reqwest = { workspace = true, features = ["json"] } @@ -25,7 +24,6 @@ serde_with.workspace = true snmalloc-rs = "0.3.6" sqlx.workspace = true thegraph-core = { workspace = true, features = ["alloy-signer-local", "serde", "signed-message"] } -thegraph-graphql-http = { version = "0.4.0", features = ["reqwest"] } thiserror.workspace = true time = { workspace = true, features = ["parsing", "formatting"] } tokio = { workspace = true, features = ["rt-multi-thread"] } diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs index ff152dff..30b0ee3d 100644 --- a/bin/dipper-service/src/config.rs +++ b/bin/dipper-service/src/config.rs @@ -9,12 +9,9 @@ use std::{ use dipper_core::config::{Hidden, HiddenSecretKeyAsHexStr}; use serde_with::serde_as; -use thegraph_core::{ - DeploymentId, - alloy::{ - primitives::{Address, ChainId, U256}, - signers::k256::SecretKey, - }, +use thegraph_core::alloy::{ + primitives::{Address, ChainId, U256}, + signers::k256::SecretKey, }; use url::Url; @@ -989,21 +986,26 @@ pub struct DbConfig { #[serde_as] #[derive(custom_debug::CustomDebug, serde::Deserialize)] pub struct NetworkConfig { - /// The graph network gateway URL + /// The indexing-payments subgraph query endpoint used to look up + /// registered indexer URLs. Same form as `chain_listener.subgraph_endpoint`: + /// a full query URL, gateway-served or self-hosted. #[debug(with = std::fmt::Display::fmt)] #[serde_as(as = "serde_with::DisplayFromStr")] - pub gateway_url: Url, - - /// The graph network API key - pub api_key: Hidden, + pub subgraph_endpoint: Url, - /// The graph network subgraph deployment ID - #[serde_as(as = "serde_with::DisplayFromStr")] - pub deployment_id: DeploymentId, + /// Bearer token for the endpoint (needed for gateway-served subgraphs) + #[serde(default)] + pub api_key: Option>, - /// The update interval for the network service + /// The update interval for the indexer URL lookup service #[serde_as(as = "serde_with::DurationSeconds")] pub update_interval: Duration, + + /// Boot even if the subgraph reports 0 registered indexers, instead of + /// exiting after the startup retries. For environments that come up before + /// any indexer has registered (e.g. local networks); leave off elsewhere. + #[serde(default)] + pub allow_empty_at_startup: bool, } /// The configuration for the signer diff --git a/bin/dipper-service/src/main.rs b/bin/dipper-service/src/main.rs index cc73759d..a649be33 100644 --- a/bin/dipper-service/src/main.rs +++ b/bin/dipper-service/src/main.rs @@ -31,6 +31,11 @@ static ALLOC: snmalloc_rs::SnMalloc = snmalloc_rs::SnMalloc; /// over across every configured provider with its own backoff. const RCA_DOMAIN_FETCH_MAX_RETRIES: u32 = 5; +/// Extra attempts to fetch the initial indexer URL snapshot after the first. The fetch runs +/// before the admin RPC port (the readiness probe's target) opens, so retrying forever on a +/// stalled subgraph would leave the pod hanging unready with no restart; exit visibly instead. +const INDEXER_URLS_FETCH_MAX_RETRIES: u32 = 5; + #[tokio::main] pub async fn main() -> anyhow::Result<()> { // Set up logging @@ -169,55 +174,78 @@ pub async fn main() -> anyhow::Result<()> { ); //- The network services - let (network_topology_handle, network_topology_service) = { - let network_subgraph_url = conf - .network - .gateway_url - .join(&format!( - "/api/deployments/id/{}", - conf.network.deployment_id - )) - .expect("invalid network subgraph URL"); - - let network_subgraph_client = network::fetch::Client::new( - reqwest::Client::new(), - network_subgraph_url, - conf.network.api_key.into_inner(), - ); - - // Fetch the initial topology snapshot, retrying with exponential backoff. - // The gateway may be temporarily unavailable (e.g. during a chain halt). - let topology_init_snapshot = { + let (indexer_urls_handle, indexer_urls_service) = { + let subgraph_endpoint = conf.network.subgraph_endpoint; + let api_key = conf.network.api_key.map(|key| key.into_inner()); + + // Fetch the initial snapshot with bounded exponential-backoff retries. + // The subgraph may be temporarily unavailable (e.g. during a chain halt). + let init_snapshot = { + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(30)) + .build() + .expect("failed to build HTTP client"); let mut attempt: u32 = 0; loop { - match network::service::topology::fetch_snapshot(&network_subgraph_client).await { - Ok(s) => break s, - Err(err) => { - attempt += 1; - let delay = std::time::Duration::from_secs(2u64.pow(attempt.min(5))); - tracing::info!( - attempt, - delay_secs = delay.as_secs(), - error = %err, - "initial topology fetch failed, retrying" + match network::service::indexer_urls::fetch_snapshot( + &client, + &subgraph_endpoint, + api_key.as_deref(), + ) + .await + { + Ok(s) if !s.is_empty() => break s, + Ok(s) if conf.network.allow_empty_at_startup => { + tracing::warn!( + "subgraph reports 0 registered indexers; starting with an empty \ + snapshot (network.allow_empty_at_startup)" ); - tokio::time::sleep(delay).await; + break s; + } + // An empty snapshot is useless at startup (no offer can be + // sent) and usually means a wrong endpoint: retry like an + // error. The refresh loop tolerates empties separately. + result => { + let err = match result { + Ok(_) => anyhow::anyhow!("subgraph returned 0 registered indexers"), + Err(err) => err, + }; + if attempt < INDEXER_URLS_FETCH_MAX_RETRIES { + attempt += 1; + let delay = std::time::Duration::from_secs(2u64.pow(attempt.min(5))); + tracing::warn!( + attempt, + delay_secs = delay.as_secs(), + error = %err, + "initial indexer URLs fetch failed, retrying" + ); + tokio::time::sleep(delay).await; + } else { + anyhow::bail!( + "failed to fetch a non-empty initial indexer URL snapshot \ + after {} attempts: {err}", + INDEXER_URLS_FETCH_MAX_RETRIES + 1 + ) + } } } } }; - network::service::topology::new( - network_subgraph_client, - conf.network.update_interval, - topology_init_snapshot, + network::service::indexer_urls::new( + network::service::indexer_urls::Ctx { + endpoint: subgraph_endpoint, + api_key, + update_interval: conf.network.update_interval, + }, + init_snapshot, ) }; - tracing::info!("initialized Graph network service"); + tracing::info!("initialized indexer URLs service"); //- The network provider component let network_provider = - network::provider::NetworkProviderService::new(network_topology_handle.clone()); + network::provider::NetworkProviderService::new(indexer_urls_handle.clone()); //- The IISA HTTP client // Verify IISA is reachable before accepting traffic (deployment ordering) @@ -524,8 +552,8 @@ pub async fn main() -> anyhow::Result<()> { // Construct the task tree let mut task_tree = JoinSet::new(); - let network_topology_task_handle = task_tree.spawn(network_topology_service); - tracing::debug!(task_id=%network_topology_task_handle.id(), "Graph network topology service started"); + let indexer_urls_task_handle = task_tree.spawn(indexer_urls_service); + tracing::debug!(task_id=%indexer_urls_task_handle.id(), "Indexer URLs service started"); let worker_task_handle = task_tree.spawn(worker_service); tracing::debug!(task_id=%worker_task_handle.id(), "Worker service started"); @@ -641,9 +669,9 @@ pub async fn main() -> anyhow::Result<()> { worker_handle.stop().await; tracing::trace!("stopped Worker service"); - tracing::trace!("stopping Graph network service"); - network_topology_handle.stop().await; - tracing::trace!("stopped Graph network service"); + tracing::trace!("stopping indexer URLs service"); + indexer_urls_handle.stop().await; + tracing::trace!("stopped indexer URLs service"); tracing::trace!("shutting down DB connection pool"); db_conn.close().await; diff --git a/bin/dipper-service/src/network.rs b/bin/dipper-service/src/network.rs index 52539711..b33f95fe 100644 --- a/bin/dipper-service/src/network.rs +++ b/bin/dipper-service/src/network.rs @@ -1,10 +1,9 @@ //! A service providing information about the indexers in the network. -pub mod fetch; pub mod provider; pub mod service; #[cfg(test)] mod tests { - mod it_fetch_subgraph_topology_data; + mod it_fetch_indexer_urls; } diff --git a/bin/dipper-service/src/network/fetch.rs b/bin/dipper-service/src/network/fetch.rs deleted file mode 100644 index c3d928a5..00000000 --- a/bin/dipper-service/src/network/fetch.rs +++ /dev/null @@ -1,15 +0,0 @@ -//! The network subgraph client API of the network service. -//! -//! The network subgraph client API provides functionality for querying the network subgraph -//! for information about the indexers in the network. This includes information such as the -//! indexers' ID, URL, allocations, and more. -//! -//! The retrieved information is preprocessed and returned in a structured format. Further -//! processing should be done to verify the information and to use it in the application. -//! The client module provides a high-level client API to query subgraphs. - -mod client; -pub(super) mod indexer_operators; -pub(super) mod indexer_subgraphs; - -pub use client::Client; diff --git a/bin/dipper-service/src/network/fetch/client.rs b/bin/dipper-service/src/network/fetch/client.rs deleted file mode 100644 index 64ab2431..00000000 --- a/bin/dipper-service/src/network/fetch/client.rs +++ /dev/null @@ -1,46 +0,0 @@ -//! The Graph network subgraph indexes the Graph network smart contract which is responsible, -//! among other things, to act as an on-chain registry for subgraphs and their deployments. -//! -//! This module contains the logic necessary to query the Graph to get the latest state of the -//! network subgraph. - -use super::{indexer_operators, indexer_subgraphs}; - -mod paginated_client; -mod queries; - -/// The Graph network subgraph client. -#[derive(Clone)] -pub struct Client { - client: paginated_client::Client, -} - -impl Client { - /// Creates a new [`Client`] instance. - pub fn new(http_client: reqwest::Client, url: reqwest::Url, auth: String) -> Self { - Self { - client: paginated_client::Client::new(http_client, url, auth), - } - } - - pub async fn fetch_subgraphs(&self) -> anyhow::Result> { - self.client - .paginated_query(indexer_subgraphs::GRAPHQL_QUERY_FRAGMENT, 1000) - .await - .map_err(Into::into) - } - - pub async fn fetch_indexer_operators( - &self, - ) -> anyhow::Result> { - self.client - .paginated_query(indexer_operators::GRAPHQL_QUERY_FRAGMENT, 1000) - .await - .map_err(Into::into) - } -} - -#[cfg(test)] -mod tests { - mod it_subgraph_paginated_client; -} diff --git a/bin/dipper-service/src/network/fetch/client/paginated_client.rs b/bin/dipper-service/src/network/fetch/client/paginated_client.rs deleted file mode 100644 index 2a355cee..00000000 --- a/bin/dipper-service/src/network/fetch/client/paginated_client.rs +++ /dev/null @@ -1,265 +0,0 @@ -use std::sync::{Arc, atomic::AtomicU64}; - -use serde::de::Deserialize; -use thegraph_core::{BlockPointer, alloy::primitives::BlockNumber}; -use thegraph_graphql_http::{graphql::IntoDocument, http_client::ResponseError}; -use tracing::{Instrument, instrument}; - -use super::queries::{ - meta::send_bootstrap_meta_query, - page::{BlockHeight, SubgraphPageQueryResponseOpaqueEntry, send_subgraph_page_query}, -}; - -/// Error message returned by the indexer typically when a reorg happens. -const SUBGRAPH_REORG_ERROR: &str = "no block with that hash found"; - -/// Errors that can occur while sending a paginated query. -#[derive(Debug, Clone, thiserror::Error)] -pub enum PaginatedQueryError { - /// The bootstrap meta query failed. - #[error("bootstrap meta query failed: {0}")] - BootstrapMetaQueryFailed(String), - - /// The page response was empty. - /// - /// A page query response should always contain at least the 'meta' field response. If the - /// response is empty, it means that the subgraph is not returning any data. - #[error("empty response")] - EmptyResponse, - - /// A reorg was detected. - /// - /// The indexer responded with an error message indicating that a reorg was detected. - #[error("reorg detected")] - ReorgDetected, - - /// An error occurred while sending one of the requests. - #[error("request error: {0}")] - RequestError(String), - - /// An error occurred while processing the query. - /// - /// This error contains the error messages returned by the indexer when an error occurred while - /// processing one of the page requests. - #[error("response error: {0:?}")] - ResponseError(Vec), - - /// Response deserialization error. - /// - /// An error occurred while deserializing the response. - #[error("deserialization error: {0}")] - DeserializationError(String), -} - -async fn send_paginated_query Deserialize<'de>>( - client: &reqwest::Client, - subgraph_url: &reqwest::Url, - query: impl IntoDocument + Clone, - auth: Option<&str>, - page_size: usize, - mut block_height: BlockHeight, -) -> Result<(Vec, Option), PaginatedQueryError> { - debug_assert_ne!(page_size, 0, "page size must be greater than 0"); - - // Block at which the query is executed. - let mut block_pointer: Option = None; - - // The last id of the previous batch. - let mut last_id: Option = None; - - // Vector to store the results of the paginated query. - let mut results = Vec::new(); - - loop { - tracing::trace!( - last_id = %last_id.as_deref().unwrap_or("none"), - "sending page query request" - ); - - let response = send_subgraph_page_query( - client, - subgraph_url.clone(), - auth, - query.clone(), - block_height, - page_size, - last_id, - ) - .await - .map_err(PaginatedQueryError::RequestError)?; - - let resp = match response { - Ok(data) => data, - Err(err) => { - return match err { - ResponseError::Empty => Err(PaginatedQueryError::EmptyResponse), - ResponseError::Failure { errors } => { - // Check if the error message contains the reorg error message. - if errors - .iter() - .any(|err| err.message.contains(SUBGRAPH_REORG_ERROR)) - { - tracing::debug!(errors=?errors, "reorg detected"); - return Err(PaginatedQueryError::ReorgDetected); - } - - let errors = errors - .into_iter() - .map(|err| err.message) - .collect::>(); - Err(PaginatedQueryError::ResponseError(errors)) - } - }; - } - }; - - // Return if the page response was empty (no results). - if resp.results.is_empty() { - return Ok((results, block_pointer)); - } - - // Extract the page's last entry ID from the response. - last_id = { - let raw_data = resp.results.last().unwrap().get(); - match serde_json::from_str::(raw_data) { - Ok(item) => Some(item.id), - Err(err) => { - tracing::debug!(error = %err, "failed to extract 'id' for last page entry"); - return Err(PaginatedQueryError::DeserializationError( - "failed to extract 'id' for last page entry".to_string(), - )); - } - } - }; - - tracing::trace!( - block_number = %resp.meta.block.number, - block_hash = %resp.meta.block.hash, - page_items_count = %resp.results.len(), - page_items_last_id = %last_id.as_deref().unwrap_or_default(), - "received page query response" - ); - - block_height = BlockHeight::Hash(resp.meta.block.hash); - block_pointer = Some(resp.meta.block); - - // Deserialize the response data and push them to the results vector - for entity in resp.results { - match serde_json::from_str::(entity.get()) { - Ok(data) => results.push(data), - Err(err) => { - return Err(PaginatedQueryError::DeserializationError(err.to_string())); - } - } - } - } -} - -/// A client for interacting with a subgraph. -#[derive(Clone)] -pub struct Client { - pub http_client: reqwest::Client, - pub subgraph_url: reqwest::Url, - - /// The request authentication bearer token. - /// - /// This is token is inserted in the `Authentication` header. - pub auth_token: Option, - - /// The latest block number that the subgraph has progressed to. - /// - /// By default, this value is 0, and is updated after each paginated query. - latest_block: Arc, -} - -impl Client { - /// Create a new client. - pub fn new(http_client: reqwest::Client, subgraph_url: reqwest::Url, auth: String) -> Self { - Self { - http_client, - subgraph_url, - auth_token: Some(auth), - latest_block: Arc::new(AtomicU64::new(0)), - } - } - - /// Get the latest block number. - fn latest_block(&self) -> BlockNumber { - self.latest_block.load(std::sync::atomic::Ordering::Relaxed) - } - - /// Update the client's latest block number. - /// - /// The function ensures that the latest block number is always increasing - /// - /// Returns the latest block number. - fn update_latest_block(&self, new_value: BlockNumber) -> BlockNumber { - // Ensure that the latest block number is always increasing - self.latest_block - .fetch_max(new_value, std::sync::atomic::Ordering::Relaxed) - .max(new_value) - } - - /// Send a paginated query to the subgraph. - /// - /// The query is sent with a page size of `page_size` and the latest block number that the - /// subgraph has progressed to. - /// - /// In the case of a reorg, the function will return an error. - #[instrument(level = "debug", skip_all, fields(url = %self.subgraph_url, page_size = %page_size))] - pub async fn paginated_query Deserialize<'de>>( - &self, - query: impl IntoDocument + Clone, - page_size: usize, - ) -> Result, PaginatedQueryError> { - // Send a bootstrap meta query if the latest block number is 0. - // - // Graph-node is rejecting values of `number_gte:0` on subgraphs with a larger `startBlock`. - // This forces us to request the latest block number from the subgraph before sending the - // paginated query. - let mut latest_block = self.latest_block(); - if latest_block == 0 { - tracing::debug!("sending bootstrap meta query"); - let bootstrap_block = send_bootstrap_meta_query( - &self.http_client, - self.subgraph_url.clone(), - self.auth_token.as_deref(), - ) - .in_current_span() - .await - .map_err(PaginatedQueryError::BootstrapMetaQueryFailed)?; - - tracing::debug!( - block_number = bootstrap_block.meta.block.number, - block_hash = %bootstrap_block.meta.block.hash, - "received bootstrap meta query response" - ); - - // Update the latest block number - latest_block = self.update_latest_block(bootstrap_block.meta.block.number); - } - - // Send the paginated query request - tracing::debug!(block_number = %latest_block ,"sending request"); - - let (results, block) = send_paginated_query( - &self.http_client, - &self.subgraph_url, - query, - self.auth_token.as_deref(), - page_size, - BlockHeight::NumberGte(latest_block), - ) - .in_current_span() - .await?; - - // Update the latest block number - if let Some(block) = block { - self.update_latest_block(block.number); - } - - tracing::debug!(total_items_count = %results.len(), "received response"); - - Ok(results) - } -} diff --git a/bin/dipper-service/src/network/fetch/client/queries.rs b/bin/dipper-service/src/network/fetch/client/queries.rs deleted file mode 100644 index 59fe548b..00000000 --- a/bin/dipper-service/src/network/fetch/client/queries.rs +++ /dev/null @@ -1,176 +0,0 @@ -use serde::Deserialize; -use thegraph_graphql_http::{ - http::request::IntoRequestParameters, - http_client::{ReqwestExt, ResponseResult}, -}; - -/// Send an authenticated GraphQL query to a subgraph. -pub async fn send_query( - client: &reqwest::Client, - url: reqwest::Url, - auth: Option<&str>, - query: impl IntoRequestParameters + Send, -) -> Result, String> -where - T: for<'de> Deserialize<'de>, -{ - let mut builder = client.post(url); - - if let Some(auth) = auth { - builder = builder.bearer_auth(auth) - } - - let res = builder - .send_graphql(query) - .await - .map_err(|err| err.to_string())?; - - Ok(res) -} - -/// Subgraphs sometimes fall behind, be it due to failing or the Graph Node may be having issues. -/// The `_meta` field can now be added to any query so that it is possible to determine against -/// which block the query was effectively executed. -pub mod meta { - use serde::Deserialize; - use thegraph_core::BlockPointer; - - use super::send_query; - - const SUBGRAPH_META_QUERY_DOCUMENT: &str = r#"{ meta: _meta { block { number hash } } }"#; - - #[derive(Debug, Deserialize)] - pub struct SubgraphMetaQueryResponse { - pub meta: Meta, - } - - #[derive(Debug, Deserialize)] - pub struct Meta { - pub block: BlockPointer, - } - - pub async fn send_bootstrap_meta_query( - client: &reqwest::Client, - subgraph_url: reqwest::Url, - auth: Option<&str>, - ) -> Result { - send_query(client, subgraph_url, auth, SUBGRAPH_META_QUERY_DOCUMENT) - .await - .map_err(|err| format!("Error sending subgraph meta query: {}", err))? - .map_err(|err| err.to_string()) - } -} - -pub mod page { - use serde::{Deserialize, Serialize, Serializer, ser::SerializeMap as _}; - use serde_json::value::RawValue; - use thegraph_core::alloy::primitives::{BlockHash, BlockNumber}; - use thegraph_graphql_http::{ - graphql::{Document, IntoDocument, IntoDocumentWithVariables}, - http_client::ResponseResult, - }; - - use super::{meta::Meta, send_query}; - - /// The block at which the query should be executed. - /// - /// This is part of the input arguments of the [`SubgraphPageQuery`]. - #[derive(Clone, Debug, Default)] - pub enum BlockHeight { - #[default] - Latest, - Hash(BlockHash), - NumberGte(BlockNumber), - } - - impl Serialize for BlockHeight { - fn serialize(&self, s: S) -> Result { - let mut obj = s.serialize_map(Some(1))?; - match self { - Self::Latest => (), - Self::Hash(hash) => obj.serialize_entry("hash", hash)?, - Self::NumberGte(number) => obj.serialize_entry("number_gte", number)?, - } - obj.end() - } - } - - /// The arguments of the [`SubgraphPageQuery`] query. - #[derive(Clone, Debug, Serialize)] - pub struct SubgraphPageQueryVars { - /// The block at which the query should be executed. - block: BlockHeight, - /// The maximum number of entities to fetch. - first: usize, - /// The ID of the last entity fetched. - last: String, - } - - pub struct SubgraphPageQuery { - query: Document, - vars: SubgraphPageQueryVars, - } - - impl SubgraphPageQuery { - pub fn new( - query: impl IntoDocument, - block: BlockHeight, - first: usize, - last: String, - ) -> Self { - Self { - query: query.into_document(), - vars: SubgraphPageQueryVars { block, first, last }, - } - } - } - - impl IntoDocumentWithVariables for SubgraphPageQuery { - type Variables = SubgraphPageQueryVars; - - fn into_document_with_variables(self) -> (Document, Self::Variables) { - let query = indoc::formatdoc! { - r#"query ($block: Block_height!, $first: Int!, $last: String!) {{ - meta: _meta(block: $block) {{ block {{ number hash }} }} - results: {query} - }}"#, - query = self.query, - }; - - (query.into_document(), self.vars) - } - } - - #[derive(Debug, Deserialize)] - pub struct SubgraphPageQueryResponse { - pub meta: Meta, - pub results: Vec>, - } - - /// An opaque entry in the response of a subgraph page query. - /// - /// This is used to determine the ID of the last entity fetched. - #[derive(Debug, Deserialize)] - pub struct SubgraphPageQueryResponseOpaqueEntry { - pub id: String, - } - - pub async fn send_subgraph_page_query( - client: &reqwest::Client, - subgraph_url: reqwest::Url, - auth: Option<&str>, - query: impl IntoDocument, - block_height: BlockHeight, - batch_size: usize, - last: Option, - ) -> Result, String> { - send_query( - client, - subgraph_url, - auth, - SubgraphPageQuery::new(query, block_height, batch_size, last.unwrap_or_default()), - ) - .await - .map_err(|err| format!("Error sending subgraph graphql query: {}", err)) - } -} diff --git a/bin/dipper-service/src/network/fetch/client/tests/it_subgraph_paginated_client.rs b/bin/dipper-service/src/network/fetch/client/tests/it_subgraph_paginated_client.rs deleted file mode 100644 index a29c915c..00000000 --- a/bin/dipper-service/src/network/fetch/client/tests/it_subgraph_paginated_client.rs +++ /dev/null @@ -1,241 +0,0 @@ -//! Integration tests for the subgraph client. - -use std::time::Duration; - -use reqwest::Url; -use serde::Deserialize; -use thegraph_core::SubgraphId; -use tracing_subscriber::{EnvFilter, fmt::TestWriter}; - -use crate::network::fetch::client::{ - paginated_client::Client, - queries::{ - meta::send_bootstrap_meta_query, - page::{BlockHeight, send_subgraph_page_query}, - }, -}; - -/// Initialize the tests tracing subscriber. -fn init_test_tracing() { - let _ = tracing_subscriber::fmt() - .with_env_filter(EnvFilter::from_default_env()) - .compact() - .with_writer(TestWriter::default()) - .try_init(); -} - -/// Test helper to get the gateway base url from the environment. -fn test_gateway_base_url() -> Url { - std::env::var("IT_TEST_ARBITRUM_GATEWAY_URL") - .expect("Missing IT_TEST_ARBITRUM_GATEWAY_URL") - .parse() - .expect("Invalid IT_TEST_ARBITRUM_GATEWAY_URL") -} - -/// Test helper to get the test auth token from the environment. -fn test_auth_token() -> String { - std::env::var("IT_TEST_ARBITRUM_GATEWAY_AUTH").expect("Missing IT_TEST_ARBITRUM_GATEWAY_AUTH") -} - -/// Test helper to build the subgraph url with the given subgraph ID. -fn test_subgraph_url(subgraph: impl AsRef) -> Url { - test_gateway_base_url() - .join(&format!("api/subgraphs/id/{}", subgraph.as_ref())) - .expect("Invalid URL") -} - -/// The Graph Network Arbitrum subgraph in the network. -/// -/// https://thegraph.com/explorer/subgraphs/DZz4kDTdmzWLWsV373w2bSmoar3umKKH9y82SUKr5qmp?view=About&chain=arbitrum-one -const GRAPH_NETWORK_ARBITRUM_SUBGRAPH_ID: &str = "DZz4kDTdmzWLWsV373w2bSmoar3umKKH9y82SUKr5qmp"; - -#[test_with::env(IT_TEST_ARBITRUM_GATEWAY_URL, IT_TEST_ARBITRUM_GATEWAY_AUTH)] -#[tokio::test] -async fn send_subgraph_meta_query_request() { - init_test_tracing(); - - //* Given - let subgraph_url = test_subgraph_url(GRAPH_NETWORK_ARBITRUM_SUBGRAPH_ID); - let auth_token = test_auth_token(); - - let http_client = reqwest::Client::new(); - - //* When - let res = tokio::time::timeout( - Duration::from_secs(20), - send_bootstrap_meta_query(&http_client, subgraph_url, Some(&auth_token)), - ) - .await - .expect("Timeout on subgraph meta query"); - - //* Then - // Assert the query succeeded, and we get a non-empty block number and hash. - let response = res.expect("Failed to fetch subgraph meta"); - - assert!(response.meta.block.number > 0); - assert!(!response.meta.block.hash.is_empty()); -} - -#[test_with::env(IT_TEST_ARBITRUM_GATEWAY_URL, IT_TEST_ARBITRUM_GATEWAY_AUTH)] -#[tokio::test] -async fn send_subgraph_page_query_request() { - init_test_tracing(); - - //* Given - const PAGE_REQUEST_BATCH_SIZE: usize = 6; - - let subgraph_url = test_subgraph_url(GRAPH_NETWORK_ARBITRUM_SUBGRAPH_ID); - let auth_token = test_auth_token(); - - let http_client = reqwest::Client::new(); - - // Query all subgraph ids. - const SUBGRAPHS_QUERY_DOCUMENT: &str = r#" - subgraphs( - block: $block - orderBy: id, orderDirection: asc - first: $first - where: { - id_gt: $last - entityVersion: 2 - } - ) { - id - } - "#; - - //* When - let res = tokio::time::timeout( - Duration::from_secs(20), - send_subgraph_page_query( - &http_client, - subgraph_url, - Some(&auth_token), - SUBGRAPHS_QUERY_DOCUMENT, - BlockHeight::NumberGte(18627000), - PAGE_REQUEST_BATCH_SIZE, - None, - ), - ) - .await - .expect("Timeout on subgraph meta query"); - - //* Then - let page_res = res.expect("Failed to fetch subgraph page"); - let page_response = page_res.expect("Failed to fetch subgraph page"); - - // Assert meta data is present and valid. - assert!(page_response.meta.block.number > 0); - assert!(!page_response.meta.block.hash.is_empty()); - - // Assert the results are present and the correct size. - assert_eq!(page_response.results.len(), PAGE_REQUEST_BATCH_SIZE); -} - -#[test_with::env(IT_TEST_ARBITRUM_GATEWAY_URL, IT_TEST_ARBITRUM_GATEWAY_AUTH)] -#[tokio::test] -async fn send_subgraph_paginated() { - init_test_tracing(); - - //* Given - let subgraph_url = test_subgraph_url(GRAPH_NETWORK_ARBITRUM_SUBGRAPH_ID); - let auth_token = test_auth_token(); - - let http_client = reqwest::Client::new(); - - let client = Client::new(http_client, subgraph_url, auth_token); - - // Query all subgraph ids. - const SUBGRAPHS_QUERY_DOCUMENT: &str = r#" - subgraphs( - block: $block - orderBy: id, orderDirection: asc - first: $first - where: { - id_gt: $last - entityVersion: 2 - } - ) { - id - } - "#; - - #[derive(Debug, Deserialize)] - #[serde(rename_all = "camelCase")] - pub struct Subgraph { - #[allow(dead_code)] - pub id: SubgraphId, - } - - //* When - let res = tokio::time::timeout( - Duration::from_secs(20), - client.paginated_query::(SUBGRAPHS_QUERY_DOCUMENT, 200), - ) - .await - .expect("Timeout on subgraph paginated query"); - - //* Then - // Assert the query succeeded, and we got a non-empty list of active subscriptions. - let response = res.expect("Failed to fetch subgraphs"); - assert!(!response.is_empty()); - assert!( - response - .iter() - .all(|subgraph| subgraph.id != SubgraphId::ZERO) - ); -} - -#[test_with::env(IT_TEST_ARBITRUM_GATEWAY_URL, IT_TEST_ARBITRUM_GATEWAY_AUTH)] -#[tokio::test] -async fn send_subgraph_paginated_empty_response() { - init_test_tracing(); - - //* Given - let subgraph_url = test_subgraph_url(GRAPH_NETWORK_ARBITRUM_SUBGRAPH_ID); - let auth_token = test_auth_token(); - - let http_client = reqwest::Client::new(); - - let client = Client::new(http_client, subgraph_url, auth_token); - - // Query all subgraph ids. As 'entityVersion' is set to 9999, we expect no results. - const SUBGRAPHS_QUERY_DOCUMENT: &str = r#" - subgraphs( - block: $block - orderBy: id, orderDirection: asc - first: $first - where: { - id_gt: $last - entityVersion: 9999 - } - ) { - id - } - "#; - - #[derive(Debug, Deserialize)] - #[serde(rename_all = "camelCase")] - pub struct Subgraph { - #[allow(dead_code)] - pub id: SubgraphId, - } - - //* When - let res = tokio::time::timeout( - Duration::from_secs(20), - client.paginated_query::(SUBGRAPHS_QUERY_DOCUMENT, 200), - ) - .await - .expect("Timeout on subgraph paginated query"); - - //* Then - // Assert the query succeeded, and we got a empty list of active subscriptions - let response = res.expect("Failed to fetch subgraphs"); - assert!(response.is_empty()); - assert!( - response - .iter() - .all(|subgraph| subgraph.id != SubgraphId::ZERO) - ); -} diff --git a/bin/dipper-service/src/network/fetch/indexer_operators.rs b/bin/dipper-service/src/network/fetch/indexer_operators.rs deleted file mode 100644 index 0f3ea1e1..00000000 --- a/bin/dipper-service/src/network/fetch/indexer_operators.rs +++ /dev/null @@ -1,53 +0,0 @@ -pub(super) const GRAPHQL_QUERY_FRAGMENT: &str = indoc::indoc! {r#" - indexers( - block: $block - orderBy: id, orderDirection: asc - first: $first - where: { - id_gt: $last - url_not: "" - } - ) { - id - url - account { - operators( - first: 100 - orderBy: id, orderDirection: asc - ) { - id - } - } - }"#, -}; - -/// The Graph network indexer operator query response types. -/// -///
-/// These types are used to deserialize the response from the Graph network subgraph. -/// These types are not meant to be used directly by the project logic. -/// -/// Please, DO NOT mix or merge them. -///
-/// -/// See: https://github.com/graphprotocol/graph-network-subgraph/blob/master/schema.graphql -pub(in crate::network) mod types { - use thegraph_core::{IndexerId, alloy::primitives::Address}; - - #[derive(Debug, Clone, serde::Deserialize)] - pub struct Indexer { - pub id: IndexerId, - pub url: Option, - pub account: Account, - } - - #[derive(Debug, Clone, serde::Deserialize)] - pub struct Account { - pub operators: Vec, - } - - #[derive(Debug, Clone, serde::Deserialize)] - pub struct Operator { - pub id: Address, - } -} diff --git a/bin/dipper-service/src/network/fetch/indexer_subgraphs.rs b/bin/dipper-service/src/network/fetch/indexer_subgraphs.rs deleted file mode 100644 index ce998a2d..00000000 --- a/bin/dipper-service/src/network/fetch/indexer_subgraphs.rs +++ /dev/null @@ -1,97 +0,0 @@ -pub(super) const GRAPHQL_QUERY_FRAGMENT: &str = indoc::indoc! {r#" - subgraphs( - block: $block - orderBy: id, orderDirection: asc - first: $first - where: { - id_gt: $last - entityVersion: 2 - versionCount_gte: 1 - } - ) { - id - versions(orderBy: version, orderDirection: desc) { - version - subgraphDeployment { - ipfsHash - indexerAllocations( - first: 100 - orderBy: closedAtEpoch, orderDirection: desc - ) { - id - allocatedTokens - createdAtEpoch - closedAtEpoch - indexer { - id - url - } - } - } - } - }"#, -}; - -/// The Graph network indexer subgraph query response types. -/// -///
-/// These types are used to deserialize the response from the Graph network subgraph. -/// These types are not meant to be used directly by the project logic. -/// -/// Please, DO NOT mix or merge them. -///
-/// -/// See: https://github.com/graphprotocol/graph-network-subgraph/blob/master/schema.graphql -pub(in crate::network) mod types { - use serde_with::serde_as; - use thegraph_core::{AllocationId, DeploymentId, IndexerId, ProofOfIndexing, SubgraphId}; - - #[derive(Debug, Clone, serde::Deserialize)] - #[serde(rename_all = "camelCase")] - pub struct Subgraph { - pub id: SubgraphId, - pub versions: Vec, - } - - #[derive(Debug, Clone, serde::Deserialize)] - #[serde(rename_all = "camelCase")] - pub struct SubgraphVersion { - pub version: u32, - pub subgraph_deployment: SubgraphDeployment, - } - - #[derive(Debug, Clone, serde::Deserialize)] - #[serde(rename_all = "camelCase")] - pub struct SubgraphDeployment { - #[serde(rename = "ipfsHash")] - pub id: DeploymentId, - #[serde(rename = "indexerAllocations")] - pub allocations: Vec, - } - - #[serde_as] - #[derive(Debug, Clone, serde::Deserialize)] - #[serde(rename_all = "camelCase")] - pub struct Allocation { - #[serde(rename = "id")] - pub _id: AllocationId, - #[serde(rename = "createdAtEpoch")] - pub _created_at_epoch: u32, - #[serde(rename = "closedAtEpoch")] - pub _closed_at_epoch: Option, - #[serde_as(as = "serde_with::DisplayFromStr")] - #[serde(rename = "allocatedTokens")] - pub _allocated_tokens: u128, - pub indexer: Indexer, - #[serde(rename = "poi")] - pub _poi: Option, - } - - #[serde_as] - #[derive(Debug, Clone, serde::Deserialize)] - #[serde(rename_all = "camelCase")] - pub struct Indexer { - pub id: IndexerId, - pub url: Option, - } -} diff --git a/bin/dipper-service/src/network/provider.rs b/bin/dipper-service/src/network/provider.rs index 437489a9..49f89d63 100644 --- a/bin/dipper-service/src/network/provider.rs +++ b/bin/dipper-service/src/network/provider.rs @@ -13,24 +13,23 @@ pub struct Indexer { #[derive(Clone)] pub struct NetworkProviderService { - /// The network provider topology service handler - topology: service::topology::Handle, + /// The indexer URLs service handle + indexer_urls: service::indexer_urls::Handle, } impl NetworkProviderService { /// Creates a new network provider service instance. - pub fn new(topology: service::topology::Handle) -> Self { - Self { topology } + pub fn new(indexer_urls: service::indexer_urls::Handle) -> Self { + Self { indexer_urls } } /// Get an indexer by its ID. pub fn get_indexer_by_id(&self, indexer_id: &IndexerId) -> Option { - self.topology - .snapshot() - .get_indexer(indexer_id) - .map(|indexer| Indexer { - id: indexer.id, - url: indexer.url.clone(), + self.indexer_urls + .get_indexer_url(indexer_id) + .map(|url| Indexer { + id: *indexer_id, + url, }) } } diff --git a/bin/dipper-service/src/network/service.rs b/bin/dipper-service/src/network/service.rs index bc7bf2bb..1e566e01 100644 --- a/bin/dipper-service/src/network/service.rs +++ b/bin/dipper-service/src/network/service.rs @@ -3,6 +3,6 @@ pub mod chain_listener; pub mod entity_count_cache; pub mod escrow_reconciler; pub mod expiration; +pub mod indexer_urls; pub mod liveness_checker; pub mod reassignment; -pub mod topology; diff --git a/bin/dipper-service/src/network/service/indexer_urls.rs b/bin/dipper-service/src/network/service/indexer_urls.rs new file mode 100644 index 00000000..27be2433 --- /dev/null +++ b/bin/dipper-service/src/network/service/indexer_urls.rs @@ -0,0 +1,335 @@ +//! Periodic background service that polls the indexing-payments subgraph's +//! `Indexer` entities (URLs from SubgraphService registration) and exposes +//! an indexer ID to URL lookup for proposal sending and liveness checks. + +use std::{collections::BTreeMap, future::Future, time::Duration}; + +use thegraph_core::IndexerId; +use tokio::{ + sync::{mpsc, watch}, + time::MissedTickBehavior, +}; +use url::Url; + +/// Timeout for subgraph queries. +const QUERY_TIMEOUT: Duration = Duration::from_secs(30); + +/// A snapshot of the registered indexers' URLs, keyed by indexer ID. +pub type Snapshot = BTreeMap; + +/// Parse and validate an indexer URL: `Some(Url)` if it parses, uses an +/// HTTP(S) scheme, and has a host component; `None` otherwise. +fn parse_indexer_url(raw: &str) -> Option { + let url = raw.parse::().ok()?; + (matches!(url.scheme(), "http" | "https") && url.has_host()).then_some(url) +} + +/// Handle for interacting with the indexer URLs service. +#[derive(Clone)] +pub struct Handle { + /// The receiver for the latest snapshot + rx_snapshot: watch::Receiver, + + /// The stop signal for the service + tx_stop: mpsc::Sender<()>, +} + +impl Handle { + /// Look up the registered URL for an indexer. + pub fn get_indexer_url(&self, id: &IndexerId) -> Option { + self.rx_snapshot.borrow().get(id).cloned() + } + + /// Signal the service to stop and wait for it to shut down; returns + /// immediately if it is already stopped. + pub async fn stop(&self) { + if self.tx_stop.is_closed() { + return; + } + + let _ = self.tx_stop.send(()).await; + + // Wait for the channel to close + self.tx_stop.closed().await; + } +} + +/// Configuration for the indexer URLs service. +#[derive(Clone)] +pub struct Ctx { + /// The indexing-payments subgraph query endpoint. + pub endpoint: Url, + /// Bearer token for the endpoint (needed for gateway-served subgraphs). + pub api_key: Option, + /// Refresh interval. + pub update_interval: Duration, +} + +/// Fetch a full snapshot of registered indexer URLs from the subgraph. +/// Fails on query or decode errors; an empty result is not an error (no +/// indexers have registered yet). +pub async fn fetch_snapshot( + client: &reqwest::Client, + endpoint: &Url, + api_key: Option<&str>, +) -> anyhow::Result { + let mut snapshot = Snapshot::new(); + // `Indexer.id` is Bytes (an address); the all-zero address sorts below + // any real id, so it works as the initial keyset cursor. + let mut last_id = ZERO_ADDRESS.to_string(); + + loop { + let body = serde_json::json!({ + "query": PAGINATED_QUERY, + "variables": { + "lastId": last_id, + "first": PAGE_SIZE, + }, + }); + + let mut builder = client.post(endpoint.as_str()).json(&body); + if let Some(key) = api_key { + builder = builder.bearer_auth(key); + } + + let response: PageResponse = builder + .send() + .await + .map_err(|err| anyhow::anyhow!("failed to query indexers page: {err}"))? + .json() + .await + .map_err(|err| anyhow::anyhow!("failed to decode indexers page: {err}"))?; + + let Some(data) = response.data else { + anyhow::bail!("subgraph errors in indexers page: {:?}", response.errors); + }; + + let page_size = data.indexers.len(); + + for entry in data.indexers { + // Advance the cursor unconditionally so pagination continues + // even if this entry fails to parse. + last_id = entry.id.clone(); + if let Some((id, url)) = parse_indexer_entry(&entry) { + snapshot.insert(id, url); + } else { + tracing::warn!( + indexer = %entry.id, + url = %entry.url, + "skipping indexer with unparseable id or invalid URL" + ); + } + } + + if page_size < PAGE_SIZE { + break; + } + } + + Ok(snapshot) +} + +/// Parse a single indexer entry into an ID and validated URL. +/// Returns None if either field fails to parse (entry is skipped). +fn parse_indexer_entry(entry: &IndexerEntity) -> Option<(IndexerId, Url)> { + let id: IndexerId = entry.id.parse().ok()?; + let url = parse_indexer_url(&entry.url)?; + Some((id, url)) +} + +/// Create a new indexer URLs service: refetches the full set of registered +/// indexers at regular intervals and publishes the result. Failed or empty +/// refreshes preserve the previous snapshot. +pub fn new(ctx: Ctx, init: Snapshot) -> (Handle, impl Future>) { + let (tx_stop, mut rx_stop) = mpsc::channel(1); + let (tx_snapshot, rx_snapshot) = watch::channel(init); + + let service = async move { + let client = reqwest::Client::builder() + .timeout(QUERY_TIMEOUT) + .build() + .unwrap_or_default(); + + let mut timer = tokio::time::interval(ctx.update_interval); + timer.set_missed_tick_behavior(MissedTickBehavior::Skip); + // Skip the first tick: startup already fetched the initial snapshot. + timer.tick().await; + + let mut prev_count: usize = tx_snapshot.borrow().len(); + + loop { + tokio::select! { + _ = rx_stop.recv() => break, + _ = timer.tick() => {}, + } + + let snapshot = + match fetch_snapshot(&client, &ctx.endpoint, ctx.api_key.as_deref()).await { + Ok(snapshot) => snapshot, + Err(err) => { + tracing::warn!(error = %err, "failed to fetch indexer URLs update"); + continue; + } + }; + + // Guard against refreshes that produce zero indexers: a subgraph + // serving a partially indexed dataset would otherwise wipe every + // URL and stall proposal sending until the next good refresh. + let count = snapshot.len(); + if count == 0 { + tracing::warn!( + "indexer URLs refresh produced 0 indexers -- preserving previous snapshot" + ); + continue; + } + + if prev_count != count { + tracing::info!( + "indexer URLs updated: {}->{} ({:+})", + prev_count, + count, + count as isize - prev_count as isize, + ); + } else { + tracing::debug!(indexers = count, "indexer URLs refresh completed"); + } + prev_count = count; + + // Send the snapshot to the receiver; if no listener is available, + // finish the service + if let Err(err) = tx_snapshot.send(snapshot) { + tracing::debug!(error = %err, "failed to send indexer URLs update"); + break; + } + } + + tracing::debug!("indexer URLs service stopped"); + + Ok(()) + }; + + ( + Handle { + rx_snapshot, + tx_stop, + }, + service, + ) +} + +const PAGE_SIZE: usize = 1000; + +const ZERO_ADDRESS: &str = "0x0000000000000000000000000000000000000000"; + +const PAGINATED_QUERY: &str = r#" + query RegisteredIndexers($lastId: Bytes!, $first: Int!) { + indexers( + first: $first + where: { id_gt: $lastId } + orderBy: id + orderDirection: asc + ) { + id + url + } + } +"#; + +#[derive(Debug, serde::Deserialize)] +struct PageResponse { + data: Option, + errors: Option>, +} + +#[derive(Debug, serde::Deserialize)] +struct PageData { + indexers: Vec, +} + +#[derive(Debug, serde::Deserialize)] +struct IndexerEntity { + id: String, + url: String, +} + +#[cfg(test)] +mod tests { + use super::*; + + const INDEXER: &str = "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; + + fn entry(id: &str, url: &str) -> IndexerEntity { + IndexerEntity { + id: id.to_string(), + url: url.to_string(), + } + } + + #[test] + fn test_parse_indexer_entry_valid() { + let (id, url) = parse_indexer_entry(&entry(INDEXER, "https://indexer.example.com/")) + .expect("entry should parse"); + assert_eq!(id, INDEXER.parse::().unwrap()); + assert_eq!(url.as_str(), "https://indexer.example.com/"); + } + + #[test] + fn test_parse_indexer_entry_invalid_id() { + assert!( + parse_indexer_entry(&entry("not-an-address", "https://indexer.example.com")).is_none() + ); + } + + #[test] + fn test_parse_indexer_entry_invalid_url() { + assert!(parse_indexer_entry(&entry(INDEXER, "")).is_none()); + assert!(parse_indexer_entry(&entry(INDEXER, "not a url")).is_none()); + // Non-HTTP scheme + assert!(parse_indexer_entry(&entry(INDEXER, "ftp://indexer.example.com")).is_none()); + // Scheme merely prefixed with "http" is not http(s) + assert!(parse_indexer_entry(&entry(INDEXER, "httpsx://indexer.example.com")).is_none()); + // No host + assert!(parse_indexer_entry(&entry(INDEXER, "http://")).is_none()); + } + + #[test] + fn test_handle_lookup() { + let id: IndexerId = INDEXER.parse().unwrap(); + let url: Url = "https://indexer.example.com".parse().unwrap(); + let init = Snapshot::from([(id, url.clone())]); + + let (handle, _service) = new( + Ctx { + endpoint: "http://localhost:9999/subgraphs/name/test".parse().unwrap(), + api_key: None, + update_interval: Duration::from_secs(3600), + }, + init, + ); + + assert_eq!(handle.get_indexer_url(&id), Some(url)); + let other: IndexerId = "0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + .parse() + .unwrap(); + assert_eq!(handle.get_indexer_url(&other), None); + } + + #[tokio::test] + async fn test_stop_signal() { + let (handle, fut) = new( + Ctx { + endpoint: "http://localhost:9999/subgraphs/name/test".parse().unwrap(), + api_key: None, + update_interval: Duration::from_secs(3600), + }, + Snapshot::new(), + ); + + let task = tokio::spawn(fut); + handle.stop().await; + + // Should complete without error + let result = tokio::time::timeout(Duration::from_secs(2), task).await; + assert!(result.is_ok()); + } +} diff --git a/bin/dipper-service/src/network/service/topology.rs b/bin/dipper-service/src/network/service/topology.rs deleted file mode 100644 index 8feec260..00000000 --- a/bin/dipper-service/src/network/service/topology.rs +++ /dev/null @@ -1,458 +0,0 @@ -//! Network topology service for tracking indexers, subgraphs, deployments, and allocations. -//! -//! This module provides a service that periodically fetches network topology data from -//! a subgraph and maintains an up-to-date snapshot of the network state. The snapshot -//! includes information about: -//! -//! - Indexers: Entities that index subgraph data and provide query services -//! - Subgraphs: Collections of data sources and mappings that define how to index blockchain data -//! - Deployments: Specific versions of subgraphs deployed to the network -//! - Allocations: Staked tokens by indexers on specific deployments -//! -//! The service runs in the background and provides a handle for accessing the latest -//! network topology snapshot. - -use std::{ - collections::{BTreeMap, BTreeSet}, - future::Future, - time::Duration, -}; - -use anyhow::Context; -use thegraph_core::{DeploymentId, IndexerId, SubgraphId, alloy::primitives::Address}; -use tokio::{ - sync::{mpsc, watch, watch::Ref}, - time::MissedTickBehavior, -}; -use url::Url; - -use crate::network::fetch::{Client as SubgraphClient, indexer_operators, indexer_subgraphs}; - -/// Parse and validate an indexer URL. -/// -/// Returns `Some(Url)` if the URL is present, parses successfully, uses an HTTP(S) -/// scheme, and has a host component. Returns `None` otherwise. -fn parse_indexer_url(raw: Option) -> Option { - let url = raw?.parse::().ok()?; - (url.scheme().starts_with("http") && url.has_host()).then_some(url) -} - -/// Fetches the latest network topology snapshot from the subgraph -pub async fn fetch_snapshot(client: &SubgraphClient) -> anyhow::Result { - let subgraphs = client - .fetch_subgraphs() - .await - .context("failed to fetch subgraphs info")?; - let operators = client - .fetch_indexer_operators() - .await - .context("failed to fetch indexer operators info")?; - - let mut snapshot = Snapshot::new(); - snapshot.extend(subgraphs); - snapshot.extend(operators); - Ok(snapshot) -} - -/// Handle for interacting with the network topology service. -/// -/// This handle provides access to the current network topology snapshot -/// and allows stopping the service. -#[derive(Clone)] -pub struct Handle { - /// The receiver for the service data - rx_snapshot: watch::Receiver, - - /// The stop signal for the service - tx_stop: mpsc::Sender<()>, -} - -impl Handle { - /// Get the current network topology snapshot. - /// - /// Returns a reference to the latest snapshot of the network topology. - pub fn snapshot(&self) -> Ref<'_, Snapshot> { - self.rx_snapshot.borrow() - } - - /// Stop the network topology service. - /// - /// This method sends a stop signal to the service and waits for it to shut down. - /// If the service is already stopped, this method returns immediately. - pub async fn stop(&self) { - if self.tx_stop.is_closed() { - return; - } - - let _ = self.tx_stop.send(()).await; - - // Wait for the channel to close - self.tx_stop.closed().await; - } -} - -/// Create a new network topology service that fetches data from the subgraph. -/// -/// The service will fetch data from the subgraph at regular intervals and update the internal -/// state. It periodically queries for indexers, subgraphs, deployments, and allocations to -/// maintain an up-to-date view of the network topology. -pub fn new( - client: SubgraphClient, - update_interval: Duration, - init: Snapshot, -) -> (Handle, impl Future>) { - let (tx_stop, mut rx_stop) = mpsc::channel(1); - let (tx_snapshot, rx_snapshot) = watch::channel(init); - - let service = async move { - let mut timer = tokio::time::interval(update_interval); - timer.set_missed_tick_behavior(MissedTickBehavior::Skip); - - let mut prev_indexers: usize = 0; - let mut prev_deployments: usize = 0; - let mut prev_allocations: usize = 0; - - loop { - tokio::select! { - _ = rx_stop.recv() => break, - _ = timer.tick() => {}, - } - - // Create a new snapshot - let mut snapshot = Snapshot::new(); - - match client.fetch_subgraphs().await { - Ok(data) if !data.is_empty() => { - snapshot.extend(data); - } - Ok(_) => { - tracing::warn!("empty network subgraph update"); - continue; - } - Err(err) => { - tracing::warn!(error=%err, "failed to fetch network subgraph update"); - continue; - } - }; - - match client.fetch_indexer_operators().await { - Ok(data) if !data.is_empty() => { - snapshot.extend(data); - } - Ok(_) => { - tracing::warn!("empty network indexer operator update"); - continue; - } - Err(err) => { - tracing::warn!(error=%err, "failed to fetch network indexer operator update"); - continue; - } - } - - // Guard against topology updates that produce zero indexers. The - // per-fetch empty checks above handle empty API responses, but the - // snapshot can still end up with zero indexers if all returned entries - // have invalid URLs and are filtered out during extend. - let new_count = snapshot.indexers.len(); - if new_count == 0 { - tracing::warn!( - "topology refresh produced 0 indexers despite non-empty fetch \ - responses -- preserving previous snapshot" - ); - continue; - } - - // Log topology summary on every successful refresh. - let cur_deployments = snapshot.deployments.len(); - let cur_allocations: usize = snapshot - .deployments - .values() - .map(|d| d.indexings.len()) - .sum(); - - tracing::info!( - indexers = new_count, - deployments = cur_deployments, - allocations = cur_allocations, - "topology refresh completed" - ); - - // Log deltas when any count changed since the last successful refresh. - if prev_indexers != new_count - || prev_deployments != cur_deployments - || prev_allocations != cur_allocations - { - tracing::info!( - "topology updated: indexers {}->{} ({:+}), deployments {}->{} ({:+}), allocations {}->{} ({:+})", - prev_indexers, - new_count, - new_count as isize - prev_indexers as isize, - prev_deployments, - cur_deployments, - cur_deployments as isize - prev_deployments as isize, - prev_allocations, - cur_allocations, - cur_allocations as isize - prev_allocations as isize, - ); - } - - prev_indexers = new_count; - prev_deployments = cur_deployments; - prev_allocations = cur_allocations; - - // Send the snapshot to the receiver, if no listener is available, finish the service - if let Err(err) = tx_snapshot.send(snapshot) { - tracing::debug!(error = %err, "failed to send network subgraph update"); - break; - } - } - - tracing::debug!("network subgraph service stopped"); - - Ok(()) - }; - - ( - Handle { - rx_snapshot, - tx_stop, - }, - service, - ) -} - -/// A snapshot of the network topology state at a given point in time. -/// -/// This structure contains the complete state of the network topology, -/// including indexers, subgraphs, deployments, and allocations. -#[derive(Debug, Clone)] -pub struct Snapshot { - /// The indexers table - /// - /// See [Indexer] for more information - indexers: BTreeMap, - /// The subgraphs table - /// - /// See [Subgraph] for more information - subgraphs: BTreeMap, - /// The deployments table - /// - /// See [Deployment] for more information - deployments: BTreeMap, -} - -impl Snapshot { - /// Create a new empty network snapshot with the current timestamp. - /// - /// Returns an empty snapshot with no indexers, subgraphs, deployments, or allocations. - pub fn new() -> Self { - Self { - indexers: Default::default(), - subgraphs: Default::default(), - deployments: Default::default(), - } - } - - /// Get an [Indexer] by its [IndexerId]. - pub fn get_indexer(&self, id: &IndexerId) -> Option<&Indexer> { - self.indexers.get(id) - } -} - -impl Extend for Snapshot { - /// Extend the network snapshot with a list of subgraphs. - /// - /// This method processes subgraph data from the network and updates the snapshot - /// with new subgraphs, deployments, indexers, and allocations. - fn extend(&mut self, iter: T) - where - T: IntoIterator, - { - for sub in iter { - let subgraph_id = sub.id; - - // Add subgraph to the network snapshot - self.subgraphs - .entry(subgraph_id) - .or_insert_with(|| Subgraph { - _id: subgraph_id, - versions: Default::default(), - }); - - for sub_version in sub.versions { - let deployment_id = sub_version.subgraph_deployment.id; - let deployment_subgraph_id = subgraph_id; - let deployment_version_num = sub_version.version; - - // Add subgraph version to the subgraph - self.subgraphs.entry(subgraph_id).and_modify(|subgraph| { - subgraph.versions.push(SubgraphVersion { - _num: deployment_version_num, - _deployment: deployment_id, - }); - }); - - // Add deployment to the network snapshot - self.deployments - .entry(deployment_id) - .or_insert_with(|| Deployment { - _id: deployment_id, - _subgraph: deployment_subgraph_id, - _version: deployment_version_num, - indexings: Default::default(), - }); - - for allocation in sub_version.subgraph_deployment.allocations { - let indexer_id = allocation.indexer.id; - - let indexer_url = match parse_indexer_url(allocation.indexer.url) { - Some(url) => url, - None => continue, - }; - - // Add the indexer to the network snapshot indexers table - self.indexers - .entry(indexer_id) - .and_modify(|indexer| { - indexer.indexings.insert(deployment_id); - }) - .or_insert_with(|| Indexer { - id: indexer_id, - url: indexer_url, - indexings: BTreeSet::from([deployment_id]), - operators: Default::default(), - }); - - // Add the indexer to the deployment indexings - self.deployments - .entry(deployment_id) - .and_modify(|deployment| { - deployment.indexings.insert(indexer_id); - }); - } - } - } - } -} - -impl Extend for Snapshot { - /// Extend the network snapshot with indexer data and operator relationships. - /// - /// Creates indexer entries for any registered indexer with a valid URL, - /// regardless of whether they have active allocations. This ensures idle - /// indexers are visible to the proposal pipeline. - fn extend(&mut self, iter: T) - where - T: IntoIterator, - { - for indexer_data in iter { - let indexer_id = indexer_data.id; - - let indexer_url = match parse_indexer_url(indexer_data.url) { - Some(url) => url, - None => continue, - }; - - let operators: BTreeSet
= indexer_data - .account - .operators - .into_iter() - .map(|op| op.id) - .collect(); - - // Only create new entries for indexers that have operators, - // since proposals require an operator to sign. - if operators.is_empty() && !self.indexers.contains_key(&indexer_id) { - continue; - } - - self.indexers - .entry(indexer_id) - .and_modify(|indexer| { - indexer.operators.extend(operators.iter().copied()); - }) - .or_insert_with(|| Indexer { - id: indexer_id, - url: indexer_url, - indexings: BTreeSet::new(), - operators, - }); - } - } -} - -/// An indexer in the network. -/// -/// Indexers are entities that stake tokens on subgraph deployments and provide -/// query services for those deployments. They earn rewards for correctly indexing -/// and serving subgraph data. -#[derive(Debug, Clone)] -pub struct Indexer { - /// The indexer ID - /// - /// The indexer ID is a unique identifier for the indexer and coincides with - /// the Ethereum address of the indexer. - pub id: IndexerId, - /// The indexer URL - /// - /// The URL where the indexer's GraphQL API can be accessed. - pub url: Url, - /// The deployments that the indexer has allocations for and is indexing - /// - /// This set contains the IDs of all deployments the indexer is currently indexing. - pub indexings: BTreeSet, - /// Associated indexer operator account addresses - /// - /// These are Ethereum addresses that have permission to operate on behalf of the indexer. - pub operators: BTreeSet
, -} - -/// A subgraph in the network. -/// -/// Subgraphs define how to index and transform blockchain data into a structured GraphQL API. -/// They can have multiple versions, each corresponding to a specific deployment. -#[derive(Debug, Clone)] -pub struct Subgraph { - /// The subgraph ID (stored for data model completeness; lookups use the BTreeMap key) - _id: SubgraphId, - /// The versions of the subgraph - /// - /// Each version corresponds to a specific deployment of the subgraph. - /// See [SubgraphVersion] for more information. - pub versions: Vec, -} - -/// A version of a [Subgraph]. -/// -/// Each subgraph can have multiple versions, each with a different deployment. -/// Newer versions typically contain improvements or fixes to the subgraph. -#[derive(Debug, Clone)] -pub struct SubgraphVersion { - /// The version number - /// - /// A sequential number identifying the version, with higher numbers - /// indicating newer versions. - _num: u32, - /// The deployment ID - /// - /// The ID of the deployment corresponding to this version. - _deployment: DeploymentId, -} - -/// A deployment of a [Subgraph] to the network. -/// -/// A deployment represents a specific version of a subgraph that has been -/// published to the network and can be indexed by indexers. -#[derive(Debug, Clone)] -pub struct Deployment { - /// The deployment ID (stored for data model completeness; lookups use the BTreeMap key) - _id: DeploymentId, - /// The subgraph this deployment belongs to - _subgraph: SubgraphId, - /// The deployment version number within the subgraph - _version: u32, - /// The indexers that are indexing the deployment - /// - /// The indexers are stored in a BTreeSet to ensure that they are unique. - pub indexings: BTreeSet, -} diff --git a/bin/dipper-service/src/network/tests/it_fetch_indexer_urls.rs b/bin/dipper-service/src/network/tests/it_fetch_indexer_urls.rs new file mode 100644 index 00000000..f1eebd98 --- /dev/null +++ b/bin/dipper-service/src/network/tests/it_fetch_indexer_urls.rs @@ -0,0 +1,245 @@ +//! Integration tests for the indexer URLs service against a wiremock subgraph: +//! `fetch_snapshot` (auth, pagination, invalid-entry skipping, error surfacing) +//! and the refresh loop (bad or empty refreshes preserve, good ones replace). + +use std::time::Duration; + +use thegraph_core::IndexerId; +use url::Url; +use wiremock::{ + Mock, MockServer, ResponseTemplate, + matchers::{body_string_contains, header, method}, +}; + +use crate::network::service::indexer_urls::{Ctx, Handle, Snapshot, fetch_snapshot, new}; + +const ZERO_CURSOR: &str = "0x0000000000000000000000000000000000000000"; + +/// 20-byte hex address with `n` as its numeric value. +fn addr(n: u32) -> String { + format!("0x{n:040x}") +} + +fn page_body(indexers: &[(String, String)]) -> serde_json::Value { + let entries: Vec<_> = indexers + .iter() + .map(|(id, url)| serde_json::json!({ "id": id, "url": url })) + .collect(); + serde_json::json!({ "data": { "indexers": entries } }) +} + +#[tokio::test] +async fn fetch_paginates_past_a_full_page_and_skips_invalid_entries() { + //* Given + // A full first page of 1,000 indexers forces a second request; the + // second page mixes a valid entry with an invalid URL and a bad id. + let server = MockServer::start().await; + + let page1: Vec<_> = (1..=1000) + .map(|n| (addr(n), format!("https://indexer-{n}.example.com/"))) + .collect(); + let page2 = vec![ + (addr(1001), "https://indexer-1001.example.com/".to_string()), + (addr(1002), "not a url".to_string()), + ( + "not-an-address".to_string(), + "https://x.example.com/".to_string(), + ), + ]; + + Mock::given(method("POST")) + .and(header("authorization", "Bearer it-test-key")) + .and(body_string_contains(ZERO_CURSOR)) + .respond_with(ResponseTemplate::new(200).set_body_json(page_body(&page1))) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(header("authorization", "Bearer it-test-key")) + .and(body_string_contains(addr(1000))) + .respond_with(ResponseTemplate::new(200).set_body_json(page_body(&page2))) + .mount(&server) + .await; + + //* When + let client = reqwest::Client::new(); + let endpoint = server.uri().parse().unwrap(); + let snapshot = fetch_snapshot(&client, &endpoint, Some("it-test-key")) + .await + .expect("fetch should succeed"); + + //* Then + // 1,000 from page 1 plus the single valid entry from page 2. + assert_eq!(snapshot.len(), 1_001); + let valid: IndexerId = addr(1001).parse().unwrap(); + assert_eq!( + snapshot.get(&valid).map(|url| url.as_str()), + Some("https://indexer-1001.example.com/") + ); + let invalid: IndexerId = addr(1002).parse().unwrap(); + assert!(!snapshot.contains_key(&invalid)); +} + +#[tokio::test] +async fn fetch_without_api_key_sends_no_authorization_header() { + //* Given + let server = MockServer::start().await; + + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(page_body(&[( + addr(7), + "https://indexer-7.example.com/".to_string(), + )]))) + .mount(&server) + .await; + + //* When + let client = reqwest::Client::new(); + let endpoint = server.uri().parse().unwrap(); + let snapshot = fetch_snapshot(&client, &endpoint, None) + .await + .expect("fetch should succeed"); + + //* Then + assert_eq!(snapshot.len(), 1); + let received = server.received_requests().await.unwrap(); + assert!( + received + .iter() + .all(|req| !req.headers.contains_key("authorization")) + ); +} + +#[tokio::test] +async fn fetch_surfaces_graphql_errors() { + //* Given + let server = MockServer::start().await; + + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "errors": [{ "message": "Type `Bytes` is not a valid input type" }] + }))) + .mount(&server) + .await; + + //* When + let client = reqwest::Client::new(); + let endpoint = server.uri().parse().unwrap(); + let result = fetch_snapshot(&client, &endpoint, None).await; + + //* Then + let err = result.expect_err("fetch should fail").to_string(); + assert!(err.contains("subgraph errors"), "unexpected error: {err}"); +} + +/// Spawn the refresh service against `server` with a 50 ms interval and a +/// single-entry init snapshot; returns the handle and the init entry. +fn spawn_service(server: &MockServer) -> (Handle, IndexerId, Url) { + let id: IndexerId = addr(1).parse().unwrap(); + let url: Url = "https://indexer-1.example.com/".parse().unwrap(); + let init = Snapshot::from([(id, url.clone())]); + + let (handle, service) = new( + Ctx { + endpoint: server.uri().parse().unwrap(), + api_key: None, + update_interval: Duration::from_millis(50), + }, + init, + ); + tokio::spawn(service); + (handle, id, url) +} + +/// Wait until the mock server has seen at least `count` refresh requests. +async fn wait_for_requests(server: &MockServer, count: usize) { + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + loop { + if server.received_requests().await.unwrap().len() >= count { + break; + } + assert!( + tokio::time::Instant::now() < deadline, + "timed out waiting for {count} refresh requests" + ); + tokio::time::sleep(Duration::from_millis(25)).await; + } +} + +#[tokio::test] +async fn refresh_with_zero_indexers_preserves_the_previous_snapshot() { + //* Given + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(page_body(&[]))) + .mount(&server) + .await; + + let (handle, id, url) = spawn_service(&server); + + //* When + // Let at least 2 empty refreshes complete. + wait_for_requests(&server, 2).await; + + //* Then + assert_eq!(handle.get_indexer_url(&id), Some(url)); + handle.stop().await; +} + +#[tokio::test] +async fn failed_refresh_preserves_the_previous_snapshot() { + //* Given + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(500)) + .mount(&server) + .await; + + let (handle, id, url) = spawn_service(&server); + + //* When + // Let at least 2 failing refreshes complete. + wait_for_requests(&server, 2).await; + + //* Then + assert_eq!(handle.get_indexer_url(&id), Some(url)); + handle.stop().await; +} + +#[tokio::test] +async fn successful_refresh_replaces_the_snapshot() { + //* Given + // The subgraph now reports a different indexer than the init snapshot's. + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(page_body(&[( + addr(2), + "https://indexer-2.example.com/".to_string(), + )]))) + .mount(&server) + .await; + + let (handle, init_id, _) = spawn_service(&server); + + //* When + let new_id: IndexerId = addr(2).parse().unwrap(); + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + loop { + if handle.get_indexer_url(&new_id).is_some() { + break; + } + assert!( + tokio::time::Instant::now() < deadline, + "timed out waiting for the refreshed snapshot" + ); + tokio::time::sleep(Duration::from_millis(25)).await; + } + + //* Then + // The snapshot is replaced wholesale, not merged into. + assert_eq!( + handle.get_indexer_url(&new_id).map(|url| url.to_string()), + Some("https://indexer-2.example.com/".to_string()) + ); + assert_eq!(handle.get_indexer_url(&init_id), None); + handle.stop().await; +} diff --git a/bin/dipper-service/src/network/tests/it_fetch_subgraph_topology_data.rs b/bin/dipper-service/src/network/tests/it_fetch_subgraph_topology_data.rs deleted file mode 100644 index c341dbf3..00000000 --- a/bin/dipper-service/src/network/tests/it_fetch_subgraph_topology_data.rs +++ /dev/null @@ -1,94 +0,0 @@ -//! Integration tests for the network subgraph client. - -use std::time::Duration; - -use reqwest::Url; -use tracing_subscriber::{EnvFilter, fmt::TestWriter}; - -use crate::network::fetch::Client as NetworkSubgraphClient; - -/// Initialize the tests tracing subscriber. -fn init_test_tracing() { - let _ = tracing_subscriber::fmt() - .with_env_filter(EnvFilter::from_default_env()) - .compact() - .with_writer(TestWriter::default()) - .try_init(); -} - -/// Test helper to get the gateway base url from the environment. -fn test_gateway_base_url() -> Url { - std::env::var("IT_TEST_ARBITRUM_GATEWAY_URL") - .expect("Missing IT_TEST_ARBITRUM_GATEWAY_URL") - .parse() - .expect("Invalid IT_TEST_ARBITRUM_GATEWAY_URL") -} - -/// Test helper to get the test auth token from the environment. -fn test_auth_token() -> String { - std::env::var("IT_TEST_ARBITRUM_GATEWAY_AUTH").expect("Missing IT_TEST_ARBITRUM_GATEWAY_AUTH") -} - -/// Test helper to build the subgraph url with the given subgraph ID. -fn test_subgraph_url(subgraph: impl AsRef) -> Url { - test_gateway_base_url() - .join(&format!("api/subgraphs/id/{}", subgraph.as_ref())) - .expect("Invalid URL") -} - -/// Graph Network Arbitrum subgraph, addressed by subgraph ID so the gateway routes -/// to the current deployment rather than a pinned hash that can rot to a dead indexer set. -/// https://thegraph.com/explorer/subgraphs/DZz4kDTdmzWLWsV373w2bSmoar3umKKH9y82SUKr5qmp -const GRAPH_NETWORK_ARBITRUM_SUBGRAPH_ID: &str = "DZz4kDTdmzWLWsV373w2bSmoar3umKKH9y82SUKr5qmp"; - -#[test_with::env(IT_TEST_ARBITRUM_GATEWAY_URL, IT_TEST_ARBITRUM_GATEWAY_AUTH)] -#[tokio::test] -async fn fetch_subgraph_data() { - init_test_tracing(); - - //* Given - let subgraph_url = test_subgraph_url(GRAPH_NETWORK_ARBITRUM_SUBGRAPH_ID); - let auth_token = test_auth_token(); - - let network_subgraph_client = - NetworkSubgraphClient::new(reqwest::Client::new(), subgraph_url, auth_token); - - //* When - let res = tokio::time::timeout( - Duration::from_secs(90), - network_subgraph_client.fetch_subgraphs(), - ) - .await - .expect("Timeout on network fetch subgraph data query"); - - //* Then - let response = res.expect("Failed to fetch data"); - - assert!(!response.is_empty()); -} - -#[test_with::env(IT_TEST_ARBITRUM_GATEWAY_URL, IT_TEST_ARBITRUM_GATEWAY_AUTH)] -#[tokio::test] -async fn fetch_indexer_operators_data() { - init_test_tracing(); - - //* Given - let subgraph_url = test_subgraph_url(GRAPH_NETWORK_ARBITRUM_SUBGRAPH_ID); - let auth_token = test_auth_token(); - - let network_subgraph_client = - NetworkSubgraphClient::new(reqwest::Client::new(), subgraph_url, auth_token); - - //* When - let res = tokio::time::timeout( - Duration::from_secs(45), - network_subgraph_client.fetch_indexer_operators(), - ) - .await - .expect("Timeout on network fetch indexer operators data query"); - - //* Then - let response = res.expect("Failed to fetch data"); - - assert!(!response.is_empty()); -} diff --git a/k8s/configmap-example.yaml b/k8s/configmap-example.yaml index 9261502f..353e7264 100644 --- a/k8s/configmap-example.yaml +++ b/k8s/configmap-example.yaml @@ -41,10 +41,10 @@ data: "max_connections": 10 }, "network": { - "gateway_url": "https://gateway.thegraph.com", + "subgraph_endpoint": "https://gateway.thegraph.com/api/deployments/id/REPLACE_ME", "api_key": "REPLACE_ME", - "deployment_id": "REPLACE_ME", - "update_interval": 60 + "update_interval": 60, + "allow_empty_at_startup": false }, "signer": { "secret_key": "REPLACE_ME",