coprocessor: fix parsing of graphql responses with null as data (#7141)

Author: IvanGoncharov

.changesets/fix_fix_coprocessor_data_null.md

@@ -0,0 +1,27 @@
+### Fix Parsing of Coprocessor GraphQL Responses ([PR #7141](https://github.com/apollographql/router/pull/7141))
+
+Previously Router ignored `data: null` property inside GraphQL response returned by coprocessor.
+According to [GraphQL Spectification](https://spec.graphql.org/draft/#sel-FAPHLJCAACEBxlY):
+
+> If an error was raised during the execution that prevented a valid response, the "data" entry in the response should be null.
+
+That means if coprocessor returned valid execution error, for example:
+
+```json
+{
+  "data": null,
+  "errors": [{ "message": "Some execution error" }]
+}
+```
+
+Router violated above restriction from GraphQL Specification by returning following response to client:
+
+```json
+{
+  "errors": [{ "message": "Some execution error" }]
+}
+```
+
+This fix ensures full compliance with the GraphQL specification by preserving the complete structure of error responses from coprocessors.
+
+Contributed by [@IvanGoncharov](https://github.com/IvanGoncharov) in [#7141](https://github.com/apollographql/router/pull/7141)

apollo-router/src/graphql/mod.rs

@@ -13,6 +13,7 @@ use futures::Stream;
 use heck::ToShoutySnakeCase;
 pub use request::Request;
 pub use response::IncrementalResponse;
+use response::MalformedResponseError;
 pub use response::Response;
 use serde::Deserialize;
 use serde::Serialize;
@@ -21,7 +22,6 @@ use serde_json_bytes::Map as JsonMap;
 use serde_json_bytes::Value;
 pub(crate) use visitor::ResponseVisitor;
 
-use crate::error::FetchError;
 use crate::json_ext::Object;
 use crate::json_ext::Path;
 pub use crate::json_ext::Path as JsonPath;
@@ -125,41 +125,39 @@ impl Error {
         }
     }
 
-    pub(crate) fn from_value(service_name: &str, value: Value) -> Result<Error, FetchError> {
-        let mut object =
-            ensure_object!(value).map_err(|error| FetchError::SubrequestMalformedResponse {
-                service: service_name.to_string(),
-                reason: format!("invalid error within `errors`: {}", error),
-            })?;
+    pub(crate) fn from_value(value: Value) -> Result<Error, MalformedResponseError> {
+        let mut object = ensure_object!(value).map_err(|error| MalformedResponseError {
+            reason: format!("invalid error within `errors`: {}", error),
+        })?;
 
         let extensions =
             extract_key_value_from_object!(object, "extensions", Value::Object(o) => o)
-                .map_err(|err| FetchError::SubrequestMalformedResponse {
-                    service: service_name.to_string(),
+                .map_err(|err| MalformedResponseError {
                     reason: format!("invalid `extensions` within error: {}", err),
                 })?
                 .unwrap_or_default();
-        let message = extract_key_value_from_object!(object, "message", Value::String(s) => s)
-            .map_err(|err| FetchError::SubrequestMalformedResponse {
-                service: service_name.to_string(),
+        let message = match extract_key_value_from_object!(object, "message", Value::String(s) => s)
+        {
+            Ok(Some(s)) => Ok(s.as_str().to_string()),
+            Ok(None) => Err(MalformedResponseError {
+                reason: "missing required `message` property within error".to_owned(),
+            }),
+            Err(err) => Err(MalformedResponseError {
                 reason: format!("invalid `message` within error: {}", err),
-            })?
-            .map(|s| s.as_str().to_string())
-            .unwrap_or_default();
+            }),
+        }?;
         let locations = extract_key_value_from_object!(object, "locations")
             .map(skip_invalid_locations)
             .map(serde_json_bytes::from_value)
             .transpose()
-            .map_err(|err| FetchError::SubrequestMalformedResponse {
-                service: service_name.to_string(),
+            .map_err(|err| MalformedResponseError {
                 reason: format!("invalid `locations` within error: {}", err),
             })?
             .unwrap_or_default();
         let path = extract_key_value_from_object!(object, "path")
             .map(serde_json_bytes::from_value)
             .transpose()
-            .map_err(|err| FetchError::SubrequestMalformedResponse {
-                service: service_name.to_string(),
+            .map_err(|err| MalformedResponseError {
                 reason: format!("invalid `path` within error: {}", err),
             })?;
 

apollo-router/src/graphql/request.rs

@@ -6,10 +6,10 @@ use serde::de::DeserializeSeed;
 use serde::de::Error;
 use serde_json_bytes::ByteString;
 use serde_json_bytes::Map as JsonMap;
-use serde_json_bytes::Value;
 
 use crate::configuration::BatchingMode;
 use crate::json_ext::Object;
+use crate::json_ext::Value;
 
 /// A GraphQL `Request` used to represent both supergraph and subgraph requests.
 #[derive(Clone, Derivative, Serialize, Deserialize, Default)]
@@ -169,34 +169,33 @@ impl Request {
     pub(crate) fn batch_from_urlencoded_query(
         url_encoded_query: String,
     ) -> Result<Vec<Request>, serde_json::Error> {
-        let value: serde_json::Value = serde_urlencoded::from_bytes(url_encoded_query.as_bytes())
+        let value: Value = serde_urlencoded::from_bytes(url_encoded_query.as_bytes())
             .map_err(serde_json::Error::custom)?;
 
-        Request::process_query_values(&value)
+        Request::process_query_values(value)
     }
 
     /// Convert Bytes into a GraphQL [`Request`].
     ///
     /// An error will be produced in the event that the bytes array cannot be
     /// turned into a valid GraphQL `Request`.
     pub(crate) fn batch_from_bytes(bytes: &[u8]) -> Result<Vec<Request>, serde_json::Error> {
-        let value: serde_json::Value =
-            serde_json::from_slice(bytes).map_err(serde_json::Error::custom)?;
+        let value: Value = serde_json::from_slice(bytes).map_err(serde_json::Error::custom)?;
 
-        Request::process_batch_values(&value)
+        Request::process_batch_values(value)
     }
 
-    fn allocate_result_array(value: &serde_json::Value) -> Vec<Request> {
+    fn allocate_result_array(value: &Value) -> Vec<Request> {
         match value.as_array() {
             Some(array) => Vec::with_capacity(array.len()),
             None => Vec::with_capacity(1),
         }
     }
 
-    fn process_batch_values(value: &serde_json::Value) -> Result<Vec<Request>, serde_json::Error> {
-        let mut result = Request::allocate_result_array(value);
+    fn process_batch_values(value: Value) -> Result<Vec<Request>, serde_json::Error> {
+        let mut result = Request::allocate_result_array(&value);
 
-        if let serde_json::Value::Array(entries) = value {
+        if let Value::Array(entries) = value {
             u64_histogram!(
                 "apollo.router.operations.batching.size",
                 "Number of queries contained within each query batch",
@@ -211,20 +210,20 @@ impl Request {
                 mode = BatchingMode::BatchHttpLink.to_string() // Only supported mode right now
             );
             for entry in entries {
-                let bytes = serde_json::to_vec(entry)?;
+                let bytes = serde_json::to_vec(&entry)?;
                 result.push(Request::deserialize_from_bytes(&bytes.into())?);
             }
         } else {
-            let bytes = serde_json::to_vec(value)?;
+            let bytes = serde_json::to_vec(&value)?;
             result.push(Request::deserialize_from_bytes(&bytes.into())?);
         }
         Ok(result)
     }
 
-    fn process_query_values(value: &serde_json::Value) -> Result<Vec<Request>, serde_json::Error> {
-        let mut result = Request::allocate_result_array(value);
+    fn process_query_values(value: Value) -> Result<Vec<Request>, serde_json::Error> {
+        let mut result = Request::allocate_result_array(&value);
 
-        if let serde_json::Value::Array(entries) = value {
+        if let Value::Array(entries) = value {
             u64_histogram!(
                 "apollo.router.operations.batching.size",
                 "Number of queries contained within each query batch",
@@ -239,42 +238,36 @@ impl Request {
                 mode = BatchingMode::BatchHttpLink.to_string() // Only supported mode right now
             );
             for entry in entries {
-                result.push(Request::process_value(entry)?);
+                result.push(Request::process_value(&entry)?);
             }
         } else {
-            result.push(Request::process_value(value)?)
+            result.push(Request::process_value(&value)?)
         }
         Ok(result)
     }
 
-    fn process_value(value: &serde_json::Value) -> Result<Request, serde_json::Error> {
-        let operation_name =
-            if let Some(serde_json::Value::String(operation_name)) = value.get("operationName") {
-                Some(operation_name.clone())
-            } else {
-                None
-            };
-
-        let query = if let Some(serde_json::Value::String(query)) = value.get("query") {
-            Some(query.as_str())
-        } else {
-            None
-        };
-        let variables: Object = get_from_urlencoded_value(value, "variables")?.unwrap_or_default();
-        let extensions: Object =
-            get_from_urlencoded_value(value, "extensions")?.unwrap_or_default();
-
-        let request_builder = Self::builder()
+    fn process_value(value: &Value) -> Result<Request, serde_json::Error> {
+        let operation_name = value.get("operationName").and_then(Value::as_str);
+        let query = value.get("query").and_then(Value::as_str).map(String::from);
+        let variables: Object = value
+            .get("variables")
+            .and_then(Value::as_str)
+            .map(serde_json::from_str)
+            .transpose()?
+            .unwrap_or_default();
+        let extensions: Object = value
+            .get("extensions")
+            .and_then(Value::as_str)
+            .map(serde_json::from_str)
+            .transpose()?
+            .unwrap_or_default();
+
+        let request = Self::builder()
+            .and_query(query)
             .variables(variables)
             .and_operation_name(operation_name)
-            .extensions(extensions);
-
-        let request = if let Some(query_str) = query {
-            request_builder.query(query_str).build()
-        } else {
-            request_builder.build()
-        };
-
+            .extensions(extensions)
+            .build();
         Ok(request)
     }
 
@@ -284,25 +277,13 @@ impl Request {
     /// An error will be produced in the event that the query string parameters
     /// cannot be turned into a valid GraphQL `Request`.
     pub fn from_urlencoded_query(url_encoded_query: String) -> Result<Request, serde_json::Error> {
-        let urldecoded: serde_json::Value =
-            serde_urlencoded::from_bytes(url_encoded_query.as_bytes())
-                .map_err(serde_json::Error::custom)?;
+        let urldecoded: Value = serde_urlencoded::from_bytes(url_encoded_query.as_bytes())
+            .map_err(serde_json::Error::custom)?;
 
         Request::process_value(&urldecoded)
     }
 }
 
-fn get_from_urlencoded_value<'a, T: Deserialize<'a>>(
-    object: &'a serde_json::Value,
-    key: &str,
-) -> Result<Option<T>, serde_json::Error> {
-    if let Some(serde_json::Value::String(byte_string)) = object.get(key) {
-        Some(serde_json::from_str(byte_string.as_str())).transpose()
-    } else {
-        Ok(None)
-    }
-}
-
 struct RequestFromBytesSeed<'data>(&'data Bytes);
 
 impl<'de> DeserializeSeed<'de> for RequestFromBytesSeed<'_> {