Add expensive op extension

.changes/added/3181.md

@@ -0,0 +1 @@
+Add expensive op extension for full block queries on GraphQL

bin/fuel-core/src/cli/run.rs

@@ -718,6 +718,8 @@ impl Command {
                 required_fuel_block_height_timeout: graphql
                     .required_fuel_block_height_timeout
                     .into(),
+                concurrent_full_block_requests: graphql.concurrent_full_block_requests,
+                full_block_request_timeout: graphql.full_block_request_timeout.into(),
             },
             combined_db_config,
             snapshot_reader,

bin/fuel-core/src/cli/run/graphql.rs

@@ -105,6 +105,12 @@ pub struct GraphQLArgs {
 
     #[clap(flatten)]
     pub costs: QueryCosts,
+
+    #[clap(long = "concurrent-full-block-requests", default_value = "10", env)]
+    pub concurrent_full_block_requests: usize,
+
+    #[clap(long = "full-block-request-timeout", default_value = "3s", env)]
+    pub full_block_request_timeout: humantime::Duration,
 }
 
 /// Costs for individual graphql queries.

crates/fuel-core/src/graphql_api.rs

@@ -59,6 +59,9 @@ pub struct ServiceConfig {
     pub assemble_tx_estimate_predicates_limit: usize,
     /// Configurable cost parameters to limit graphql queries complexity
     pub costs: Costs,
+    /// Configurable parameters to limit number of concurrent requests to the full block API
+    pub concurrent_full_block_requests: usize,
+    pub full_block_request_timeout: Duration,
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]

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

@@ -1,7 +1,10 @@
 use crate::{
     fuel_core_graphql_api::{
         Config,
-        extensions::unify_response,
+        extensions::{
+            expensive_op_guard::ExpensiveOpGuardFactory,
+            unify_response,
+        },
         ports::{
             BlockProducerPort,
             ChainStateProvider as ChainStateProviderTrait,
@@ -297,6 +300,11 @@ where
         .extension(MetricsExtension::new(
             config.config.query_log_threshold_time,
         ))
+        .extension(ExpensiveOpGuardFactory::new(
+            Arc::new(["FullBlockByHeightQuery".to_string(), "FullBlocksQuery".to_string()]),
+            config.config.concurrent_full_block_requests,
+            config.config.full_block_request_timeout,
+        ))
         .data(config)
         .data(combined_read_database)
         .data(txpool)

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

@@ -13,6 +13,8 @@ pub(crate) mod metrics;
 pub(crate) mod required_fuel_block_height;
 pub(crate) mod validation;
 
+pub(crate) mod expensive_op_guard;
+
 // In the case of a successful query, we return the information below on
 // the `response.extensions` level.
 // But in the case of the error, `async_graphql` returns information from extensions

crates/fuel-core/src/graphql_api/extensions/expensive_op_guard.rs

@@ -0,0 +1,112 @@
+use async_graphql::{
+    Response,
+    ServerError,
+    extensions::{
+        Extension,
+        ExtensionContext,
+        ExtensionFactory,
+        NextExecute,
+    },
+};
+use std::{
+    sync::Arc,
+    time::Duration,
+};
+use tokio::sync::Semaphore;
+
+pub struct ExpensiveOpGuardFactory {
+    expensive_op_names: Arc<[String]>,
+    semaphore: Arc<Semaphore>,
+    timeout: Duration,
+}
+
+impl ExpensiveOpGuardFactory {
+    pub fn new(
+        expensive_op_names: Arc<[String]>,
+        max_in_flight: usize,
+        timeout: Duration,
+    ) -> Self {
+        Self {
+            expensive_op_names,
+            semaphore: Arc::new(Semaphore::new(max_in_flight)),
+            timeout,
+        }
+    }
+}
+
+impl ExtensionFactory for ExpensiveOpGuardFactory {
+    fn create(&self) -> Arc<dyn Extension> {
+        Arc::new(ExpensiveOpGuard {
+            expensive_op_names: self.expensive_op_names.clone(),
+            semaphore: self.semaphore.clone(),
+            timeout: self.timeout,
+        })
+    }
+}
+
+pub struct ExpensiveOpGuard {
+    expensive_op_names: Arc<[String]>,
+    semaphore: Arc<Semaphore>,
+    timeout: Duration,
+}
+
+#[async_trait::async_trait]
+impl Extension for ExpensiveOpGuard {
+    async fn execute(
+        &self,
+        ctx: &ExtensionContext<'_>,
+        operation_name: Option<&str>,
+        next: NextExecute<'_>,
+    ) -> Response {
+        let op = operation_name.clone().unwrap_or_default();
+        let is_expensive = self.expensive_op_names.iter().any(|name| op == name);
+
+        tracing::debug!(
+            "Executing operation: {:?}, and expected one of {:?}, expensive: {:?}, timeout: {:?}, semaphore_size: {:?}",
+            operation_name,
+            self.expensive_op_names,
+            is_expensive,
+            self.timeout,
+            self.semaphore.available_permits(),
+        );
+
+        if !is_expensive {
+            return next.run(ctx, operation_name).await;
+        }
+
+        // Concurrency gate (bulkhead)
+        let permit = match self.semaphore.clone().try_acquire_owned() {
+            Ok(p) => p,
+            Err(_) => {
+                let mut resp = Response::new(async_graphql::Value::Null);
+                resp.errors.push(ServerError::new(
+                    "Rate limit exceeded for this operation",
+                    None,
+                ));
+                return resp;
+            }
+        };
+
+        // Time bound (avoid request pile-ups)
+        let fut = next.run(ctx, operation_name);
+        let starting_time = tokio::time::Instant::now();
+        let out = tokio::time::timeout(self.timeout, fut).await;
+        tracing::warn!(
+            "finished executing in {:?}ns, success: {:?}",
+            starting_time.elapsed().as_nanos(),
+            out.is_ok(),
+        );
+
+        drop(permit);
+
+        match out {
+            Ok(resp) => resp,
+            Err(_) => {
+                let mut resp = Response::new(async_graphql::Value::Null);
+                resp.errors
+                    .push(ServerError::new("Operation timed out", None));
+                resp
+            }
+        }
+    }
+}

crates/fuel-core/src/service/config.rs

@@ -179,6 +179,8 @@ impl Config {
                 costs: Default::default(),
                 required_fuel_block_height_tolerance: 10,
                 required_fuel_block_height_timeout: Duration::from_secs(30),
+                concurrent_full_block_requests: 10,
+                full_block_request_timeout: Duration::from_secs(3),
             },
             combined_db_config,
             continue_on_error: false,