RPC Consistency Proposal

CHANGELOG.md

@@ -41,6 +41,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 - [2429](https://github.com/FuelLabs/fuel-core/pull/2429): Introduce custom enum for representing result of running service tasks
 - [2377](https://github.com/FuelLabs/fuel-core/pull/2377): Add more errors that can be returned as responses when using protocol `/fuel/req_res/0.0.2`. The errors supported are `ProtocolV1EmptyResponse` (status code `0`) for converting empty responses sent via protocol `/fuel/req_res/0.0.1`, `RequestedRangeTooLarge`(status code `1`) if the client requests a range of objects such as sealed block headers or transactions too large, `Timeout` (status code `2`) if the remote peer takes too long to fulfill a request, or `SyncProcessorOutOfCapacity` if the remote peer is fulfilling too many requests concurrently.
 - [2233](https://github.com/FuelLabs/fuel-core/pull/2233): Introduce a new column `modification_history_v2` for storing the modification history in the historical rocksDB. Keys in this column are stored in big endian order. Changed the behaviour of the historical rocksDB to write changes for new block heights to the new column, and to perform lookup of values from the `modification_history_v2` table first, and then from the `modification_history` table, performing a migration upon access if necessary.
+- [2473](https://github.com/FuelLabs/fuel-core/pull/2473): Graphql requests look for a `REQUIRED_FUEL_BLOCK_HEIGHT` header. If the header is specified, the request will not be served unless the node's current fuel block height is at least the value specified in the header. All graphql responses now contain a `CURRENT_FUEL_BLOCK_HEIGHT` header which contains the block height of the last block processed by the node.
 
 #### Breaking
 - [2389](https://github.com/FuelLabs/fuel-core/pull/2258): Updated the `messageProof` GraphQL schema to return a non-nullable `MessageProof`.

crates/client/src/client.rs

@@ -240,6 +240,28 @@ impl FuelClient {
         Self::decode_response(response)
     }
 
+    pub async fn query_with_headers<ResponseData, Vars>(
+        &self,
+        q: Operation<ResponseData, Vars>,
+        headers: impl IntoIterator<Item = (String, String)>,
+    ) -> io::Result<ResponseData>
+    where
+        Vars: serde::Serialize,
+        ResponseData: serde::de::DeserializeOwned + 'static,
+    {
+        let mut request_builder = self.client.post(self.url.clone());
+        for (header_name, header_value) in headers {
+            request_builder = request_builder.header(header_name, header_value);
+        }
+
+        let response = request_builder
+            .run_graphql(q)
+            .await
+            .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
+
+        Self::decode_response(response)
+    }
+
     fn decode_response<R>(response: GraphQlResponse<R>) -> io::Result<R>
     where
         R: serde::de::DeserializeOwned + 'static,
@@ -1080,6 +1102,32 @@ impl FuelClient {
         Ok(balance.amount)
     }
 
+    pub async fn balance_with_required_block_header(
+        &self,
+        owner: &Address,
+        asset_id: Option<&AssetId>,
+        required_block_height: u32,
+    ) -> io::Result<u128> {
+        let owner: schema::Address = (*owner).into();
+        let asset_id: schema::AssetId = match asset_id {
+            Some(asset_id) => (*asset_id).into(),
+            None => schema::AssetId::default(),
+        };
+        let query = schema::balance::BalanceQuery::build(BalanceArgs { owner, asset_id });
+        let balance: types::Balance = self
+            .query_with_headers(
+                query,
+                vec![(
+                    "REQUIRED_FUEL_BLOCK_HEIGHT".to_string(),
+                    required_block_height.to_string(),
+                )],
+            )
+            .await?
+            .balance
+            .into();
+        Ok(balance.amount)
+    }
+
     // Retrieve a page of balances by their owner
     pub async fn balances(
         &self,

crates/fuel-core/src/graphql_api/api_service.rs

@@ -15,7 +15,9 @@ use crate::{
         view_extension::ViewExtension,
         Config,
     },
-    graphql_api,
+    graphql_api::{
+        self,
+    },
     schema::{
         CoreSchema,
         CoreSchemaBuilder,
@@ -42,6 +44,13 @@ use axum::{
             ACCESS_CONTROL_ALLOW_ORIGIN,
         },
         HeaderValue,
+        Request as AxumRequest,
+        Response as AxumResponse,
+        StatusCode,
+    },
+    middleware::{
+        self,
+        Next,
     },
     response::{
         sse::Event,
@@ -88,7 +97,10 @@ use tower_http::{
 pub type Service = fuel_core_services::ServiceRunner<GraphqlService>;
 
 pub use super::database::ReadDatabase;
-use super::ports::worker;
+use super::{
+    ports::worker,
+    worker_service::LAST_KNOWN_BLOCK_HEIGHT,
+};
 
 pub type BlockProducer = Box<dyn BlockProducerPort>;
 // In the future GraphQL should not be aware of `TxPool`. It should
@@ -214,6 +226,66 @@ impl RunnableTask for Task {
     }
 }
 
+const REQUIRED_FUEL_BLOCK_HEIGHT_HEADER: &str = "REQUIRED_FUEL_BLOCK_HEIGHT";
+const CURRENT_FUEL_BLOCK_HEIGHT_HEADER: &str = "CURRENT_FUEL_BLOCK_HEIGHT";
+
+async fn required_fuel_block_height<B>(
+    req: AxumRequest<B>,
+    next: Next<B>,
+) -> impl IntoResponse {
+    let last_known_block_height: BlockHeight = LAST_KNOWN_BLOCK_HEIGHT
+    .get()
+    //Maybe too strict?
+    .ok_or(StatusCode::INTERNAL_SERVER_ERROR)?
+    .load(std::sync::atomic::Ordering::Acquire)
+    .into();
+
+    let Some(required_fuel_block_height_header) = req
+        .headers()
+        .get(REQUIRED_FUEL_BLOCK_HEIGHT_HEADER)
+        .map(|value| value.to_str())
+    else {
+        // Header is not present, so we don't have any requirements.
+        let mut response = next.run(req).await;
+        add_current_fuel_block_height_header_to_response(
+            &mut response,
+            &last_known_block_height,
+        );
+
+        return Ok(response);
+    };
+
+    let raw_required_fuel_block_height =
+        required_fuel_block_height_header.map_err(|_err| StatusCode::BAD_REQUEST)?;
+
+    let required_fuel_block_height: BlockHeight = raw_required_fuel_block_height
+        .parse::<u32>()
+        .map_err(|_| StatusCode::BAD_REQUEST)?
+        .into();
+
+    if required_fuel_block_height > last_known_block_height {
+        Err(StatusCode::PRECONDITION_FAILED)
+    } else {
+        let mut response = next.run(req).await;
+        add_current_fuel_block_height_header_to_response(
+            &mut response,
+            &last_known_block_height,
+        );
+
+        Ok(response)
+    }
+}
+
+fn add_current_fuel_block_height_header_to_response<Body>(
+    response: &mut AxumResponse<Body>,
+    last_known_block_height: &BlockHeight,
+) {
+    response.headers_mut().insert(
+        CURRENT_FUEL_BLOCK_HEIGHT_HEADER,
+        HeaderValue::from_str(&last_known_block_height.to_string()).unwrap(),
+    );
+}
+
 // Need a separate Data Object for each Query endpoint, cannot be avoided
 #[allow(clippy::too_many_arguments)]
 pub fn new_service<OnChain, OffChain>(
@@ -287,6 +359,7 @@ where
         .route(
             graphql_endpoint,
             post(graphql_handler)
+                .layer(middleware::from_fn(required_fuel_block_height))
                 .layer(ConcurrencyLimitLayer::new(concurrency_limit))
                 .options(ok),
         )

crates/fuel-core/src/graphql_api/worker_service.rs

@@ -100,11 +100,18 @@ use futures::{
 use std::{
     borrow::Cow,
     ops::Deref,
+    sync::{
+        atomic::AtomicU32,
+        OnceLock,
+    },
 };
 
 #[cfg(test)]
 mod tests;
 
+// The last known block height that was processed by the GraphQL service.
+pub static LAST_KNOWN_BLOCK_HEIGHT: OnceLock<AtomicU32> = OnceLock::new();
+
 #[derive(Debug, Clone)]
 pub enum DaCompressionConfig {
     Disabled,
@@ -186,6 +193,11 @@ where
         // update the importer metrics after the block is successfully committed
         graphql_metrics().total_txs_count.set(total_tx_count as i64);
 
+        // update the block height recorded in memory
+        LAST_KNOWN_BLOCK_HEIGHT
+            .get_or_init(|| AtomicU32::new(**height))
+            .store(**height, std::sync::atomic::Ordering::Release);
+
         Ok(())
     }
 }

tests/tests/lib.rs

@@ -48,6 +48,8 @@ mod regenesis;
 #[cfg(not(feature = "only-p2p"))]
 mod relayer;
 #[cfg(not(feature = "only-p2p"))]
+mod required_fuel_block_height_header;
+#[cfg(not(feature = "only-p2p"))]
 mod snapshot;
 #[cfg(not(feature = "only-p2p"))]
 mod state_rewind;

tests/tests/required_fuel_block_height_header.rs

@@ -0,0 +1,50 @@
+use fuel_core::{
+    chain_config::StateConfig,
+    service::{
+        Config,
+        FuelService,
+    },
+};
+use fuel_core_client::client::{
+    types::primitives::{
+        Address,
+        AssetId,
+    },
+    FuelClient,
+};
+
+#[tokio::test]
+async fn balance_with_block_height_header() {
+    let owner = Address::default();
+    let asset_id = AssetId::BASE;
+
+    // setup config
+    let state_config = StateConfig::default();
+    let config = Config::local_node_with_state_config(state_config);
+
+    // setup server & client
+    let srv = FuelService::new_node(config).await.unwrap();
+    let client = FuelClient::from(srv.bound_address);
+
+    // Issue a request with wrong precondition
+    let error = client
+        .balance_with_required_block_header(&owner, Some(&asset_id), 100)
+        .await
+        .unwrap_err();
+
+    let error_str = format!("{:?}", error);
+    assert_eq!(
+        error_str,
+        "Custom { kind: Other, error: ErrorResponse(412, \"\") }"
+    );
+
+    // Meet precondition on server side
+    client.produce_blocks(100, None).await.unwrap();
+
+    // Issue request again
+    let result = client
+        .balance_with_required_block_header(&owner, Some(&asset_id), 100)
+        .await;
+
+    assert!(result.is_ok());
+}