Performance improvements (#202)

- No preserve_order, we manually serialize the response on our own
Closes #201
- We also manually serialize the representations sent to the subgraphs
- Avoid extra allocations in the executor
- Parsing and normalization caching in the gateway #245
- Better parallelism in ParallelNode execution - Use FuturesUnordered
instead of a vector #226
- Replace serde_json deserializer with sonic_rs deserializer #257
Closes #215
- Replace reqwest with lower level hyper_util combining with above #254
- Use mimalloc as an allocator #256
- Avoid extra allocations and cloning in the gateway layers #261
- Remove HttpRequestParams layer to prevent extra copy/clone/allocations
#287
- Projection planning to handle conflicting fields on the selection set
PR #285 & Bug why it is
needed? #249
- Use io::write instead of concatenating strings
#288
Closes #284
- Other kind of improvements can be followed with the commits

---------

Co-authored-by: Kamil Kisiela <[email protected]>

Author: ardatan

Cargo.lock

@@ -5,11 +5,11 @@ import { githubComment } from "https://raw.githubusercontent.com/dotansimha/k6-g
 
 const endpoint = __ENV.GATEWAY_ENDPOINT || "http://0.0.0.0:4000/graphql";
 const vus = __ENV.BENCH_VUS ? parseInt(__ENV.BENCH_VUS) : 50;
-const time = __ENV.BENCH_OVER_TIME || "30s";
+const duration = __ENV.BENCH_OVER_TIME || "30s";
 
 export const options = {
-  vus: vus,
-  duration: time,
+  vus,
+  duration,
 };
 
 export function setup() {
@@ -57,7 +57,7 @@ export function handleSummary(data) {
       },
     });
   }
-  return handleBenchmarkSummary(data, { vus, time });
+  return handleBenchmarkSummary(data, { vus, duration });
 }
 
 let printIdentifiersMap = {};

bench/k6.js

@@ -12,4 +12,4 @@ query-planner = { path = "../../lib/query-planner" }
 graphql-parser = "0.4.1"
 tracing-subscriber = { version = "0.3.19", features = ["env-filter"] }
 tracing-tree = "0.4.0"
-serde_json = { version = "1.0.120", features = ["preserve_order"] }
+serde_json = "1.0.140"

bin/dev-cli/Cargo.toml

@@ -11,6 +11,8 @@ path = "src/main.rs"
 query-planner = { path = "../../lib/query-planner" }
 query-plan-executor = { path = "../../lib/query-plan-executor" }
 
+mimalloc = { version = "0.1.47", features = ["override"] }
+
 # Tokio runtime
 tokio = { version = "1.38.0", features = ["full"] }
 
@@ -20,7 +22,8 @@ graphql-tools = "0.4.0"  # Using version from original file
 
 # Serialization
 serde = { version = "1.0.203", features = ["derive"] }
-serde_json = { version = "1.0.120", features = ["preserve_order"] }
+serde_json = "1.0.140"
+sonic-rs = "0.3"
 
 # HTTP client and caching
 moka = { version = "0.12.8", features = ["future"] }
@@ -48,6 +51,7 @@ tower-http = { version = "0.6.6", features = [
   "cors",
   "request-id",
 ] }
+http-body-util = "0.1.3"
 
 # Utils
 thiserror = "2.0.12"

bin/gateway/Cargo.toml

@@ -18,6 +18,7 @@ use axum::{
     Router,
 };
 use http::Request;
+use mimalloc::MiMalloc;
 use tokio::signal;
 
 use axum::Extension;
@@ -28,7 +29,6 @@ use crate::pipeline::{
     coerce_variables_service::CoerceVariablesService, execution_service::ExecutionService,
     graphiql_service::GraphiQLResponderService,
     graphql_request_params::GraphQLRequestParamsExtractor,
-    http_request_params::HttpRequestParamsExtractor,
     normalize_service::GraphQLOperationNormalizationService, parser_service::GraphQLParserService,
     progressive_override_service::ProgressiveOverrideExtractor,
     query_plan_service::QueryPlanService, validation_service::GraphQLValidationService,
@@ -43,6 +43,9 @@ use tower_http::{
 };
 use tracing::info;
 
+#[global_allocator]
+static GLOBAL: MiMalloc = MiMalloc;
+
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let perfetto_file = env::var("PERFETTO_OUT").ok().is_some_and(|v| v == "1");
@@ -95,14 +98,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 )
             }),
         )
-        .layer(HttpRequestParamsExtractor::new_layer())
         .layer(GraphiQLResponderService::new_layer())
         .layer(GraphQLRequestParamsExtractor::new_layer())
         .layer(GraphQLParserService::new_layer())
+        .layer(GraphQLValidationService::new_layer())
         .layer(ProgressiveOverrideExtractor::new_layer())
         .layer(GraphQLOperationNormalizationService::new_layer())
         .layer(CoerceVariablesService::new_layer())
-        .layer(GraphQLValidationService::new_layer())
         .layer(QueryPlanService::new_layer())
         .layer(PropagateRequestIdLayer::new(REQUEST_ID_HEADER_NAME.clone()))
         .service(ExecutionService::new(expose_query_plan));

bin/gateway/src/main.rs

@@ -2,17 +2,17 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use axum::body::Body;
-use http::Request;
+use http::{Method, Request};
 use query_plan_executor::variables::collect_variables;
-use query_plan_executor::ExecutionRequest;
+use query_planner::state::supergraph_state::OperationKind;
 use serde_json::Value;
-use tracing::{trace, warn};
+use tracing::{error, trace, warn};
 
-use crate::pipeline::error::{PipelineError, PipelineErrorVariant};
+use crate::pipeline::error::{PipelineError, PipelineErrorFromAcceptHeader, PipelineErrorVariant};
 use crate::pipeline::gateway_layer::{
     GatewayPipelineLayer, GatewayPipelineStepDecision, ProcessorLayer,
 };
-use crate::pipeline::http_request_params::HttpRequestParams;
+use crate::pipeline::graphql_request_params::ExecutionRequest;
 use crate::pipeline::normalize_service::GraphQLNormalizationPayload;
 use crate::shared_state::GatewaySharedState;
 
@@ -35,30 +35,44 @@ impl GatewayPipelineLayer for CoerceVariablesService {
     #[tracing::instrument(level = "trace", name = "CoerceVariablesService", skip_all)]
     async fn process(
         &self,
-        mut req: Request<Body>,
-    ) -> Result<(Request<Body>, GatewayPipelineStepDecision), PipelineError> {
+        req: &mut Request<Body>,
+    ) -> Result<GatewayPipelineStepDecision, PipelineError> {
         let normalized_operation = req
             .extensions()
-            .get::<GraphQLNormalizationPayload>()
+            .get::<Arc<GraphQLNormalizationPayload>>()
             .ok_or_else(|| {
-                PipelineErrorVariant::InternalServiceError("GraphQLNormalizationPayload is missing")
+                req.new_pipeline_error(PipelineErrorVariant::InternalServiceError(
+                    "GraphQLNormalizationPayload is missing",
+                ))
             })?;
 
-        let http_payload = req.extensions().get::<HttpRequestParams>().ok_or_else(|| {
-            PipelineErrorVariant::InternalServiceError("HttpRequestParams is missing")
-        })?;
-
         let execution_params = req.extensions().get::<ExecutionRequest>().ok_or_else(|| {
-            PipelineErrorVariant::InternalServiceError("ExecutionRequest is missing")
+            req.new_pipeline_error(PipelineErrorVariant::InternalServiceError(
+                "ExecutionRequest is missing",
+            ))
         })?;
 
         let app_state = req
             .extensions()
             .get::<Arc<GatewaySharedState>>()
             .ok_or_else(|| {
-                PipelineErrorVariant::InternalServiceError("GatewaySharedState is missing")
+                req.new_pipeline_error(PipelineErrorVariant::InternalServiceError(
+                    "GatewaySharedState is missing",
+                ))
             })?;
 
+        if req.method() == Method::GET {
+            if let Some(OperationKind::Mutation) =
+                normalized_operation.operation_for_plan.operation_kind
+            {
+                error!("Mutation is not allowed over GET, stopping");
+
+                return Err(
+                    req.new_pipeline_error(PipelineErrorVariant::MutationNotAllowedOverHttpGet)
+                );
+            }
+        }
+
         match collect_variables(
             &normalized_operation.operation_for_plan,
             &execution_params.variables,
@@ -74,18 +88,16 @@ impl GatewayPipelineLayer for CoerceVariablesService {
                     variables_map: values,
                 });
 
-                Ok((req, GatewayPipelineStepDecision::Continue))
+                Ok(GatewayPipelineStepDecision::Continue)
             }
             Err(err_msg) => {
                 warn!(
                     "failed to collect variables from incoming request: {}",
                     err_msg
                 );
-
-                return Err(PipelineError::new_with_accept_header(
-                    PipelineErrorVariant::VariablesCoercionError(err_msg),
-                    http_payload.accept_header.clone(),
-                ));
+                return Err(
+                    req.new_pipeline_error(PipelineErrorVariant::VariablesCoercionError(err_msg))
+                );
             }
         }
     }

bin/gateway/src/pipeline/coerce_variables_service.rs

@@ -2,28 +2,27 @@ use std::{collections::HashMap, sync::Arc};
 
 use axum::{body::Body, extract::rejection::QueryRejection, response::IntoResponse};
 use graphql_tools::validation::utils::ValidationError;
-use http::{Response, StatusCode};
+use http::{HeaderName, Method, Request, Response, StatusCode};
 use query_plan_executor::{ExecutionResult, GraphQLError};
 use query_planner::{ast::normalization::error::NormalizationError, planner::PlannerError};
 use serde_json::Value;
 
-use crate::pipeline::http_request_params::APPLICATION_JSON;
+use crate::pipeline::header::{RequestAccepts, APPLICATION_GRAPHQL_RESPONSE_JSON_STR};
 
 #[derive(Debug)]
 pub struct PipelineError {
-    pub accept_header: Option<String>,
+    pub accept_ok: bool,
     pub error: PipelineErrorVariant,
 }
 
-impl PipelineError {
-    pub fn new_with_accept_header(
-        error: PipelineErrorVariant,
-        accept_header_value: String,
-    ) -> Self {
-        Self {
-            accept_header: Some(accept_header_value),
-            error,
-        }
+pub trait PipelineErrorFromAcceptHeader {
+    fn new_pipeline_error(&self, error: PipelineErrorVariant) -> PipelineError;
+}
+
+impl PipelineErrorFromAcceptHeader for Request<Body> {
+    fn new_pipeline_error(&self, error: PipelineErrorVariant) -> PipelineError {
+        let accept_ok = !self.accepts_content_type(&APPLICATION_GRAPHQL_RESPONSE_JSON_STR);
+        PipelineError { accept_ok, error }
     }
 }
 
@@ -35,9 +34,9 @@ pub enum PipelineErrorVariant {
 
     // HTTP-related errors
     #[error("Unsupported HTTP method: {0}")]
-    UnsupportedHttpMethod(String),
+    UnsupportedHttpMethod(Method),
     #[error("Header '{0}' has invalid value")]
-    InvalidHeaderValue(String),
+    InvalidHeaderValue(HeaderName),
     #[error("Failed to read body: {0}")]
     FailedToReadBodyBytes(axum::Error),
     #[error("Content-Type header is missing")]
@@ -55,11 +54,11 @@ pub enum PipelineErrorVariant {
 
     // GraphQL-specific errors
     #[error("Failed to parse GraphQL request payload")]
-    FailedToParseBody(serde_json::Error),
+    FailedToParseBody(sonic_rs::Error),
     #[error("Failed to parse GraphQL variables JSON")]
-    FailedToParseVariables(serde_json::Error),
+    FailedToParseVariables(sonic_rs::Error),
     #[error("Failed to parse GraphQL extensions JSON")]
-    FailedToParseExtensions(serde_json::Error),
+    FailedToParseExtensions(sonic_rs::Error),
     #[error("Failed to parse GraphQL operation")]
     FailedToParseOperation(graphql_parser::query::ParseError),
     #[error("Failed to normalize GraphQL operation")]
@@ -78,19 +77,27 @@ impl PipelineErrorVariant {
             Self::UnsupportedHttpMethod(_) => "METHOD_NOT_ALLOWED",
             Self::PlannerError(_) => "QUERY_PLAN_BUILD_FAILED",
             Self::InternalServiceError(_) => "INTERNAL_SERVER_ERROR",
+            Self::FailedToParseOperation(_) => "GRAPHQL_PARSE_FAILED",
+            Self::ValidationErrors(_) => "GRAPHQL_VALIDATION_FAILED",
+            Self::VariablesCoercionError(_) => "BAD_USER_INPUT",
+            Self::NormalizationError(NormalizationError::OperationNotFound) => {
+                "OPERATION_RESOLUTION_FAILURE"
+            }
+            Self::NormalizationError(NormalizationError::SpecifiedOperationNotFound {
+                operation_name: _,
+            }) => "OPERATION_RESOLUTION_FAILURE",
+            Self::NormalizationError(NormalizationError::MultipleMatchingOperationsFound) => {
+                "OPERATION_RESOLUTION_FAILURE"
+            }
             _ => "BAD_REQUEST",
         }
     }
 
     pub fn graphql_error_message(&self) -> String {
         match self {
-            Self::PlannerError(_) | Self::InternalServiceError(_) => {
-                return "Unexpected error".to_string()
-            }
-            _ => {}
+            Self::PlannerError(_) | Self::InternalServiceError(_) => "Unexpected error".to_string(),
+            _ => self.to_string(),
         }
-
-        self.to_string()
     }
 
     pub fn default_status_code(&self, prefer_ok: bool) -> StatusCode {
@@ -111,34 +118,17 @@ impl PipelineErrorVariant {
             (Self::VariablesCoercionError(_), false) => StatusCode::BAD_REQUEST,
             (Self::VariablesCoercionError(_), true) => StatusCode::OK,
             (Self::MutationNotAllowedOverHttpGet, _) => StatusCode::METHOD_NOT_ALLOWED,
-            (Self::ValidationErrors(_), _) => StatusCode::BAD_REQUEST,
-            (Self::MissingContentTypeHeader, _) => StatusCode::BAD_REQUEST,
-            (Self::UnsupportedContentType, _) => StatusCode::BAD_REQUEST,
-        }
-    }
-}
-
-impl From<PipelineError> for PipelineErrorVariant {
-    fn from(error: PipelineError) -> Self {
-        error.error
-    }
-}
-
-impl From<PipelineErrorVariant> for PipelineError {
-    fn from(error: PipelineErrorVariant) -> Self {
-        Self {
-            error,
-            accept_header: None,
+            (Self::ValidationErrors(_), true) => StatusCode::OK,
+            (Self::ValidationErrors(_), false) => StatusCode::BAD_REQUEST,
+            (Self::MissingContentTypeHeader, _) => StatusCode::NOT_ACCEPTABLE,
+            (Self::UnsupportedContentType, _) => StatusCode::UNSUPPORTED_MEDIA_TYPE,
         }
     }
 }
 
 impl IntoResponse for PipelineError {
     fn into_response(self) -> Response<Body> {
-        let accept_ok = self
-            .accept_header
-            .is_some_and(|v| v.contains(APPLICATION_JSON.to_str().unwrap()));
-        let status = self.error.default_status_code(accept_ok);
+        let status = self.error.default_status_code(self.accept_ok);
 
         if let PipelineErrorVariant::ValidationErrors(validation_errors) = self.error {
             let validation_error_result = ExecutionResult {
@@ -148,7 +138,7 @@ impl IntoResponse for PipelineError {
             };
 
             return (
-                StatusCode::OK,
+                status,
                 serde_json::to_string(&validation_error_result).unwrap(),
             )
                 .into_response();