Handle empty responses from subgraphs and failed entity calls (#500)

This PR tackles an issue where the subgraph error could end up being
omitted entirely in the client response.
It now also handles cases where the upstream response is empty.

In cases where a subgraph call fails, there's no point to clone the
error for every entity representation. It'd result in a bloated response
with the same thing repeated multiple times (imagine fetching 1000s
representations).
Instead of cloning and adding `path` property to associate the error
with a point in GraphQL document, we attach `affectedPath` that
represents the Flatten(path) where the error occurred.

Author: kamilkisiela

lib/executor/src/context.rs

@@ -45,19 +45,24 @@ impl<'a> ExecutionContext<'a> {
     pub fn handle_errors(
         &mut self,
         subgraph_name: &str,
+        affected_path: Option<String>,
         errors: Option<Vec<GraphQLError>>,
         entity_index_error_map: Option<HashMap<&usize, Vec<GraphQLErrorPath>>>,
     ) {
         if let Some(response_errors) = errors {
             for response_error in response_errors {
-                let response_error_with_subgraph_name =
-                    response_error.add_subgraph_name(subgraph_name);
+                let mut processed_error = response_error.add_subgraph_name(subgraph_name);
+
+                if let Some(path) = &affected_path {
+                    processed_error = processed_error.add_affected_path(path.clone());
+                }
+
                 if let Some(entity_index_error_map) = &entity_index_error_map {
-                    let normalized_errors = response_error_with_subgraph_name
-                        .normalize_entity_error(entity_index_error_map);
+                    let normalized_errors =
+                        processed_error.normalize_entity_error(entity_index_error_map);
                     self.errors.extend(normalized_errors);
                 } else {
-                    self.errors.push(response_error_with_subgraph_name);
+                    self.errors.push(processed_error);
                 }
             }
         }

lib/executor/src/execution/plan.rs

@@ -509,7 +509,7 @@ impl<'exec> Executor<'exec> {
                     job.response.body,
                     job.fetch_node_id,
                 ) {
-                    ctx.handle_errors(&job.subgraph_name, response.errors, None);
+                    ctx.handle_errors(&job.subgraph_name, None, response.errors, None);
                     if let Some(output_rewrites) = output_rewrites {
                         for output_rewrite in output_rewrites {
                             output_rewrite
@@ -588,9 +588,18 @@ impl<'exec> Executor<'exec> {
                         );
                         ctx.handle_errors(
                             &job.subgraph_name,
+                            Some(job.flatten_node_path.to_string()),
                             response.errors,
                             entity_index_error_map,
                         );
+                    } else if let Some(errors) = response.errors {
+                        // No entities were returned, but there are errors to handle.
+                        // We associate them with the flattened path and subgraph.
+                        let affected_path = job.flatten_node_path.to_string();
+                        ctx.errors.extend(errors.into_iter().map(|e| {
+                            e.add_subgraph_name(&job.subgraph_name)
+                                .add_affected_path(affected_path.clone())
+                        }));
                     }
                 }
             }
@@ -902,6 +911,7 @@ mod tests {
         }];
         ctx.handle_errors(
             "subgraph_a",
+            None,
             Some(response_errors),
             Some(entity_index_error_map),
         );

lib/executor/src/executors/http.rs

@@ -162,16 +162,24 @@ impl HTTPSubgraphExecutor {
         );
 
         let (parts, body) = res.into_parts();
+        let body = body
+            .collect()
+            .await
+            .map_err(|e| {
+                SubgraphExecutorError::RequestFailure(self.endpoint.to_string(), e.to_string())
+            })?
+            .to_bytes();
+
+        if body.is_empty() {
+            return Err(SubgraphExecutorError::RequestFailure(
+                self.endpoint.to_string(),
+                "Empty response body".to_string(),
+            ));
+        }
 
         Ok(SharedResponse {
             status: parts.status,
-            body: body
-                .collect()
-                .await
-                .map_err(|e| {
-                    SubgraphExecutorError::RequestFailure(self.endpoint.to_string(), e.to_string())
-                })?
-                .to_bytes(),
+            body,
             headers: parts.headers,
         })
     }

lib/executor/src/response/graphql_error.rs

@@ -113,6 +113,11 @@ impl GraphQLError {
             .get_or_insert("DOWNSTREAM_SERVICE_ERROR".to_string());
         self
     }
+
+    pub fn add_affected_path(mut self, affected_path: String) -> Self {
+        self.extensions.affected_path = Some(affected_path);
+        self
+    }
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
@@ -185,6 +190,9 @@ pub struct GraphQLErrorExtensions {
     pub code: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub service_name: Option<String>,
+    /// Corresponds to a path of a Flatten(Fetch) node that caused the error.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub affected_path: Option<String>,
     #[serde(flatten)]
     pub extensions: HashMap<String, Value>,
 }
@@ -251,16 +259,20 @@ impl GraphQLErrorExtensions {
         GraphQLErrorExtensions {
             code: Some(code.to_string()),
             service_name: None,
+            affected_path: None,
             extensions: HashMap::new(),
         }
     }
+
     pub fn new_from_code_and_service_name(code: &str, service_name: &str) -> Self {
         GraphQLErrorExtensions {
             code: Some(code.to_string()),
             service_name: Some(service_name.to_string()),
+            affected_path: None,
             extensions: HashMap::new(),
         }
     }
+
     pub fn get(&self, key: &str) -> Option<&Value> {
         self.extensions.get(key)
     }
@@ -270,6 +282,9 @@ impl GraphQLErrorExtensions {
     }
 
     pub fn is_empty(&self) -> bool {
-        self.code.is_none() && self.service_name.is_none() && self.extensions.is_empty()
+        self.code.is_none()
+            && self.service_name.is_none()
+            && self.affected_path.is_none()
+            && self.extensions.is_empty()
     }
 }