Add support for minimum block constraint (#2868)

This PR adds support for a minimum block constraint on queries via the `number_gte` block argument. The minimum block constraint will be used by the gateway to reduce the likelihood that a client receives query results from blocks prior to those from past queries from the same client.

Author: Theodus

graphql/src/query/ext.rs

@@ -58,6 +58,9 @@ impl ValueExt for q::Value {
 pub enum BlockConstraint {
     Hash(H256),
     Number(BlockNumber),
+    /// Execute the query on the latest block only if the the subgraph has progressed to or past the
+    /// given block number.
+    Min(BlockNumber),
     Latest,
 }
 
@@ -82,6 +85,10 @@ impl TryFromValue for BlockConstraint {
             Ok(BlockConstraint::Number(BlockNumber::try_from_value(
                 number_value,
             )?))
+        } else if let Some(number_value) = map.get("number_gte") {
+            Ok(BlockConstraint::Min(BlockNumber::try_from_value(
+                number_value,
+            )?))
         } else {
             Err(anyhow!("invalid `BlockConstraint`"))
         }

graphql/src/schema/api.rs

@@ -227,6 +227,14 @@ fn add_block_height_type(schema: &mut Document) {
                 default_value: None,
                 directives: vec![],
             },
+            InputValue {
+                position: Pos::default(),
+                description: None,
+                name: "number_gte".to_owned(),
+                value_type: Type::NamedType("Int".to_owned()),
+                default_value: None,
+                directives: vec![],
+            },
         ],
     });
     let def = Definition::TypeDefinition(typedef);
@@ -641,9 +649,12 @@ fn block_argument() -> InputValue {
         position: Pos::default(),
         description: Some(
             "The block at which the query should be executed. \
-             Can either be an `{ number: Int }` containing the block number \
-             or a `{ hash: Bytes }` value containing a block hash. Defaults \
-             to the latest block when omitted."
+             Can either be a `{ hash: Bytes }` value containing a block hash, \
+             a `{ number: Int }` containing the block number, \
+             or a `{ number_gte: Int }` containing the minimum block number. \
+             In the case of `number_gte`, the query will be executed on the latest block only if \
+             the subgraph has progressed to or past the minimum block number. \
+             Defaults to the latest block when omitted."
                 .to_owned(),
         ),
         name: "block".to_string(),

graphql/src/schema/meta.graphql

@@ -20,6 +20,8 @@ type _Block_ {
   hash: Bytes
   "The block number"
   number: Int!
+  "The minimum block number"
+  number_gte: Int!
 }
 
 enum _SubgraphErrorPolicy_ {

graphql/src/store/resolver.rs

@@ -109,31 +109,25 @@ impl StoreResolver {
         bc: BlockConstraint,
         subgraph: DeploymentHash,
     ) -> Result<BlockPtr, QueryExecutionError> {
+        fn check_ptr(
+            subgraph: DeploymentHash,
+            ptr: Option<BlockPtr>,
+            min: BlockNumber,
+        ) -> Result<BlockPtr, QueryExecutionError> {
+            let ptr = ptr.expect("we should have already checked that the subgraph exists");
+            if ptr.number < min {
+                return Err(QueryExecutionError::ValueParseError(
+                    "block.number".to_owned(),
+                    format!(
+                        "subgraph {} has only indexed up to block number {} \
+                            and data for block number {} is therefore not yet available",
+                        subgraph, ptr.number, min
+                    ),
+                ));
+            }
+            Ok(ptr)
+        }
         match bc {
-            BlockConstraint::Number(number) => store
-                .block_ptr()
-                .map_err(|e| StoreError::from(e).into())
-                .and_then(|ptr| {
-                    let ptr = ptr.expect("we should have already checked that the subgraph exists");
-                    if ptr.number < number {
-                        Err(QueryExecutionError::ValueParseError(
-                            "block.number".to_owned(),
-                            format!(
-                                "subgraph {} has only indexed up to block number {} \
-                                 and data for block number {} is therefore not yet available",
-                                subgraph, ptr.number, number
-                            ),
-                        ))
-                    } else {
-                        // We don't have a way here to look the block hash up from
-                        // the database, and even if we did, there is no guarantee
-                        // that we have the block in our cache. We therefore
-                        // always return an all zeroes hash when users specify
-                        // a block number
-                        // See 7a7b9708-adb7-4fc2-acec-88680cb07ec1
-                        Ok(BlockPtr::from((web3::types::H256::zero(), number as u64)))
-                    }
-                }),
             BlockConstraint::Hash(hash) => {
                 store
                     .block_number(hash)
@@ -149,6 +143,23 @@ impl StoreResolver {
                             .map(|number| BlockPtr::from((hash, number as u64)))
                     })
             }
+            BlockConstraint::Number(number) => store
+                .block_ptr()
+                .map_err(|e| StoreError::from(e).into())
+                .and_then(|ptr| {
+                    check_ptr(subgraph, ptr, number)?;
+                    // We don't have a way here to look the block hash up from
+                    // the database, and even if we did, there is no guarantee
+                    // that we have the block in our cache. We therefore
+                    // always return an all zeroes hash when users specify
+                    // a block number
+                    // See 7a7b9708-adb7-4fc2-acec-88680cb07ec1
+                    Ok(BlockPtr::from((web3::types::H256::zero(), number as u64)))
+                }),
+            BlockConstraint::Min(number) => store
+                .block_ptr()
+                .map_err(|e| StoreError::from(e).into())
+                .and_then(|ptr| check_ptr(subgraph, ptr, number)),
             BlockConstraint::Latest => store
                 .block_ptr()
                 .map_err(|e| StoreError::from(e).into())

graphql/tests/query.rs

@@ -1548,6 +1548,19 @@ fn query_at_block_with_vars() {
             check_musicians_at(&deployment.hash, query, var, expected, qid).await;
         }
 
+        async fn musicians_at_nr_gte(
+            deployment: &DeploymentLocator,
+            block: i32,
+            expected: Result<Vec<&str>, &str>,
+            qid: &str,
+        ) {
+            let query =
+                "query by_nr($block: Int!) { musicians(block: { number_gte: $block }) { id } }";
+            let var = Some(("block", r::Value::Int(block.into())));
+
+            check_musicians_at(&deployment.hash, query, var, expected, qid).await;
+        }
+
         async fn musicians_at_hash(
             deployment: &DeploymentLocator,
             block: &FakeBlock,
@@ -1570,6 +1583,10 @@ fn query_at_block_with_vars() {
         musicians_at_nr(&deployment, 0, Ok(vec!["m1", "m2"]), "n0").await;
         musicians_at_nr(&deployment, 1, Ok(vec!["m1", "m2", "m3", "m4"]), "n1").await;
 
+        musicians_at_nr_gte(&deployment, 7000, Err(BLOCK_NOT_INDEXED), "ngte7000").await;
+        musicians_at_nr_gte(&deployment, 0, Ok(vec!["m1", "m2", "m3", "m4"]), "ngte0").await;
+        musicians_at_nr_gte(&deployment, 1, Ok(vec!["m1", "m2", "m3", "m4"]), "ngte1").await;
+
         musicians_at_hash(&deployment, &GENESIS_BLOCK, Ok(vec!["m1", "m2"]), "h0").await;
         musicians_at_hash(
             &deployment,