From 4ff59df11c63efce58bc5ca0ea27c0294d46f9dd Mon Sep 17 00:00:00 2001 From: Theo Butler Date: Mon, 25 Nov 2024 12:19:37 -0500 Subject: [PATCH] add graph-indexed header (#5710) This adds a `graph-indexed` header to query responses. The header value contains the block hash, number, and timestamp for the most recently processed block in the subgraph. This avoids the need to rewrite all queries to include `_meta { block { hash number timestamp } }` in either the indexer-service or gateway. Related: https://github.com/edgeandnode/gateway/issues/900, https://github.com/graphprotocol/indexer-rs/issues/494 --- graph/src/data/query/mod.rs | 2 +- graph/src/data/query/result.rs | 33 ++++++++++++++++++++++++++++----- graphql/src/runner.rs | 19 +++++++++++++++++-- store/test-store/src/store.rs | 2 +- 4 files changed, 47 insertions(+), 9 deletions(-) diff --git a/graph/src/data/query/mod.rs b/graph/src/data/query/mod.rs index 7b5a901908f..73a6f1fe220 100644 --- a/graph/src/data/query/mod.rs +++ b/graph/src/data/query/mod.rs @@ -7,5 +7,5 @@ mod trace; pub use self::cache_status::CacheStatus; pub use self::error::{QueryError, QueryExecutionError}; pub use self::query::{Query, QueryTarget, QueryVariables}; -pub use self::result::{QueryResult, QueryResults}; +pub use self::result::{LatestBlockInfo, QueryResult, QueryResults}; pub use self::trace::Trace; diff --git a/graph/src/data/query/result.rs b/graph/src/data/query/result.rs index 60b58fc4759..787c1b2524c 100644 --- a/graph/src/data/query/result.rs +++ b/graph/src/data/query/result.rs @@ -4,7 +4,7 @@ use crate::cheap_clone::CheapClone; use crate::components::server::query::ServerResponse; use crate::data::value::Object; use crate::derive::CacheWeight; -use crate::prelude::{r, CacheWeight, DeploymentHash}; +use crate::prelude::{r, BlockHash, BlockNumber, CacheWeight, DeploymentHash}; use http_body_util::Full; use hyper::header::{ ACCESS_CONTROL_ALLOW_HEADERS, ACCESS_CONTROL_ALLOW_METHODS, ACCESS_CONTROL_ALLOW_ORIGIN, @@ -48,6 +48,13 @@ where ser.end() } +fn serialize_block_hash(data: &BlockHash, serializer: S) -> Result +where + S: Serializer, +{ + serializer.serialize_str(&data.to_string()) +} + pub type Data = Object; #[derive(Debug)] @@ -55,13 +62,23 @@ pub type Data = Object; pub struct QueryResults { results: Vec>, pub trace: Trace, + pub indexed_block: Option, +} + +#[derive(Debug, Serialize)] +pub struct LatestBlockInfo { + #[serde(serialize_with = "serialize_block_hash")] + pub hash: BlockHash, + pub number: BlockNumber, + pub timestamp: Option, } impl QueryResults { - pub fn empty(trace: Trace) -> Self { + pub fn empty(trace: Trace, indexed_block: Option) -> Self { QueryResults { results: Vec::new(), trace, + indexed_block, } } @@ -155,6 +172,7 @@ impl From for QueryResults { QueryResults { results: vec![Arc::new(x.into())], trace: Trace::None, + indexed_block: None, } } } @@ -164,6 +182,7 @@ impl From for QueryResults { QueryResults { results: vec![Arc::new(x)], trace: Trace::None, + indexed_block: None, } } } @@ -173,6 +192,7 @@ impl From> for QueryResults { QueryResults { results: vec![x], trace: Trace::None, + indexed_block: None, } } } @@ -182,6 +202,7 @@ impl From for QueryResults { QueryResults { results: vec![Arc::new(x.into())], trace: Trace::None, + indexed_block: None, } } } @@ -191,6 +212,7 @@ impl From> for QueryResults { QueryResults { results: vec![Arc::new(x.into())], trace: Trace::None, + indexed_block: None, } } } @@ -205,6 +227,7 @@ impl QueryResults { pub fn as_http_response(&self) -> ServerResponse { let json = serde_json::to_string(&self).unwrap(); let attestable = self.results.iter().all(|r| r.is_attestable()); + let indexed_block = serde_json::to_string(&self.indexed_block).unwrap(); Response::builder() .status(200) .header(ACCESS_CONTROL_ALLOW_ORIGIN, "*") @@ -212,7 +235,8 @@ impl QueryResults { .header(ACCESS_CONTROL_ALLOW_HEADERS, "Content-Type, User-Agent") .header(ACCESS_CONTROL_ALLOW_METHODS, "GET, OPTIONS, POST") .header(CONTENT_TYPE, "application/json") - .header("Graph-Attestable", attestable.to_string()) + .header("graph-attestable", attestable.to_string()) + .header("graph-indexed", indexed_block) .body(Full::from(json)) .unwrap() } @@ -386,8 +410,7 @@ fn multiple_data_items() { let obj1 = make_obj("key1", "value1"); let obj2 = make_obj("key2", "value2"); - let trace = Trace::None; - let mut res = QueryResults::empty(trace); + let mut res = QueryResults::empty(Trace::None, None); res.append(obj1, CacheStatus::default()); res.append(obj2, CacheStatus::default()); diff --git a/graphql/src/runner.rs b/graphql/src/runner.rs index 1c55384a142..79a13b0e04e 100644 --- a/graphql/src/runner.rs +++ b/graphql/src/runner.rs @@ -17,7 +17,7 @@ use graph::{ }; use graph::{data::graphql::load_manager::LoadManager, prelude::QueryStoreManager}; use graph::{ - data::query::{QueryResults, QueryTarget}, + data::query::{LatestBlockInfo, QueryResults, QueryTarget}, prelude::QueryStore, }; @@ -117,6 +117,20 @@ where let network = Some(store.network_name().to_string()); let schema = store.api_schema()?; + let latest_block = match store.block_ptr().await.ok().flatten() { + Some(block) => Some(LatestBlockInfo { + timestamp: store + .block_number_with_timestamp_and_parent_hash(&block.hash) + .await + .ok() + .flatten() + .and_then(|(_, t, _)| t), + hash: block.hash, + number: block.number, + }), + None => None, + }; + // Test only, see c435c25decbc4ad7bbbadf8e0ced0ff2 #[cfg(debug_assertions)] let state = INITIAL_DEPLOYMENT_STATE_FOR_TESTS @@ -148,7 +162,8 @@ where let by_block_constraint = StoreResolver::locate_blocks(store.as_ref(), &state, &query).await?; let mut max_block = 0; - let mut result: QueryResults = QueryResults::empty(query.root_trace(do_trace)); + let mut result: QueryResults = + QueryResults::empty(query.root_trace(do_trace), latest_block); let mut query_res_futures: Vec<_> = vec![]; let setup_elapsed = execute_start.elapsed(); diff --git a/store/test-store/src/store.rs b/store/test-store/src/store.rs index 59a65535cf3..afb088f6bf6 100644 --- a/store/test-store/src/store.rs +++ b/store/test-store/src/store.rs @@ -526,7 +526,7 @@ async fn execute_subgraph_query_internal( 100, graphql_metrics(), )); - let mut result = QueryResults::empty(query.root_trace(trace)); + let mut result = QueryResults::empty(query.root_trace(trace), None); let deployment = query.schema.id().clone(); let store = STORE .clone()