Dispatch errors from the primary response to deferred responses

NEXT_CHANGELOG.md

@@ -125,6 +125,12 @@ By [@Geal](https://github.com/geal) in https://github.com/apollographql/router/p
 When we drop Telemetry we spawn a thread to perform the global opentelemetry trace provider shutdown. The documentation of this function indicates that "This will invoke the shutdown method on all span processors. span processors should export remaining spans before return". We should give that process some time to complete (5 seconds currently) before returning from the `drop`. This will provide more opportunity for spans to be exported.
 
 By [@garypen](https://github.com/garypen) in https://github.com/apollographql/router/pull/2191
+### Dispatch errors from the primary response to deferred responses ([Issue #1818](https://github.com/apollographql/router/issues/1818), [Issue #2185](https://github.com/apollographql/router/issues/2185))
+
+When errors are generated during the primary execution, some of them can be assigned to
+deferred responses.
+
+By [@Geal](https://github.com/geal) in https://github.com/apollographql/router/pull/2192
 
 ### Reconstruct deferred queries with knowledge about fragments ([Issue #2105](https://github.com/apollographql/router/issues/2105))
 

apollo-router/src/query_planner/execution.rs

@@ -221,7 +221,8 @@ impl PlanNode {
                             HashMap::new();
                         let mut futures = Vec::new();
 
-                        let (primary_sender, _) = tokio::sync::broadcast::channel::<Value>(1);
+                        let (primary_sender, _) =
+                            tokio::sync::broadcast::channel::<(Value, Vec<Error>)>(1);
 
                         for deferred_node in deferred {
                             let fut = deferred_node
@@ -266,10 +267,10 @@ impl PlanNode {
                             errors.extend(err.into_iter());
                             subselection = primary_subselection.clone();
 
-                            let _ = primary_sender.send(value.clone());
+                            let _ = primary_sender.send((value.clone(), errors.clone()));
                         } else {
                             subselection = primary_subselection.clone();
-                            let _ = primary_sender.send(value.clone());
+                            let _ = primary_sender.send((value.clone(), errors.clone()));
                         }
                     }
                     .instrument(tracing::info_span!(
@@ -357,7 +358,7 @@ impl DeferredNode {
         parameters: &'a ExecutionParameters<'a, SF>,
         parent_value: &Value,
         sender: futures::channel::mpsc::Sender<Response>,
-        primary_sender: &Sender<Value>,
+        primary_sender: &Sender<(Value, Vec<Error>)>,
         deferred_fetches: &mut HashMap<String, Sender<(Value, Vec<Error>)>>,
     ) -> impl Future<Output = ()>
     where
@@ -406,8 +407,10 @@ impl DeferredNode {
             let mut errors = Vec::new();
 
             if is_depends_empty {
-                let primary_value = primary_receiver.recv().await.unwrap_or_default();
+                let (primary_value, primary_errors) =
+                    primary_receiver.recv().await.unwrap_or_default();
                 value.deep_merge(primary_value);
+                errors.extend(primary_errors.into_iter())
             } else {
                 while let Some((v, _remaining)) = stream.next().await {
                     // a Err(RecvError) means either that the fetch was not performed and the
@@ -449,8 +452,10 @@ impl DeferredNode {
                     .await;
 
                 if !is_depends_empty {
-                    let primary_value = primary_receiver.recv().await.unwrap_or_default();
+                    let (primary_value, primary_errors) =
+                        primary_receiver.recv().await.unwrap_or_default();
                     v.deep_merge(primary_value);
+                    errors.extend(primary_errors.into_iter())
                 }
 
                 if let Err(e) = tx
@@ -473,8 +478,10 @@ impl DeferredNode {
                 };
                 tx.disconnect();
             } else {
-                let primary_value = primary_receiver.recv().await.unwrap_or_default();
+                let (primary_value, primary_errors) =
+                    primary_receiver.recv().await.unwrap_or_default();
                 value.deep_merge(primary_value);
+                errors.extend(primary_errors.into_iter());
 
                 if let Err(e) = tx
                     .send(

apollo-router/src/services/execution_service.rs

@@ -24,6 +24,7 @@ use super::subgraph_service::SubgraphServiceFactory;
 use super::Plugins;
 use crate::graphql::IncrementalResponse;
 use crate::graphql::Response;
+use crate::json_ext::Object;
 use crate::json_ext::Path;
 use crate::json_ext::PathElement;
 use crate::json_ext::ValueExt;
@@ -128,6 +129,10 @@ where
                                 response.has_next = Some(has_next);
                             }
 
+                            response.errors.retain(|error| match &error.path {
+                                    None => true,
+                                    Some(error_path) => query.contains_error_path(operation_name.as_deref(), response.subselection.as_deref(), response.path.as_ref(), error_path),
+                                });
                             ready(Some(response))
                         }
                         // if the deferred response specified a path, we must extract the
@@ -157,6 +162,9 @@ where
                                 }
                             });
 
+                            let query = query.clone();
+                            let operation_name = operation_name.clone();
+
                             let incremental = sub_responses
                                 .into_iter()
                                 .filter_map(move |(path, data)| {
@@ -166,25 +174,67 @@ where
                                         .iter()
                                         .filter(|error| match &error.path {
                                             None => false,
-                                            Some(err_path) => err_path.starts_with(&path),
+                                            Some(error_path) =>query.contains_error_path(operation_name.as_deref(), response.subselection.as_deref(), response.path.as_ref(), error_path) &&  error_path.starts_with(&path),
+
                                         })
                                         .cloned()
                                         .collect::<Vec<_>>();
 
+                                        let extensions: Object = response
+                                        .extensions
+                                        .iter()
+                                        .map(|(key, value)| {
+                                            if key.as_str() == "valueCompletion" {
+                                                let value = match value.as_array() {
+                                                    None => Value::Null,
+                                                    Some(v) => Value::Array(
+                                                        v.iter()
+                                                            .filter(|ext| {
+                                                                match ext
+                                                                    .as_object()
+                                                                    .as_ref()
+                                                                    .and_then(|ext| {
+                                                                        ext.get("path")
+                                                                    })
+                                                                    .and_then(|v| {
+                                                                        let p:Option<Path> = serde_json_bytes::from_value(v.clone()).ok();
+                                                                        p
+                                                                    }) {
+                                                                    None => false,
+                                                                    Some(ext_path) => {
+                                                                        ext_path
+                                                                            .starts_with(
+                                                                                &path,
+                                                                            )
+                                                                    }
+                                                                }
+                                                            })
+                                                            .cloned()
+                                                            .collect(),
+                                                    ),
+                                                };
+
+                                                (key.clone(), value)
+                                            } else {
+                                                (key.clone(), value.clone())
+                                            }
+                                        })
+                                        .collect();
+
                                     // an empty response should not be sent
                                     // still, if there's an error or extension to show, we should
                                     // send it
                                     if !data.is_null()
                                         || !errors.is_empty()
-                                        || !response.extensions.is_empty()
+                                        || !extensions.is_empty()
                                     {
                                         Some(
                                             IncrementalResponse::builder()
                                                 .and_label(response.label.clone())
                                                 .data(data)
                                                 .path(path)
                                                 .errors(errors)
-                                                .extensions(response.extensions.clone())
+                                                .extensions(extensions)
                                                 .build(),
                                         )
                                     } else {
@@ -199,6 +249,7 @@ where
                                     .incremental(incremental)
                                     .build(),
                             ))
+
                         }
                     }
                 })

apollo-router/src/services/snapshots/apollo_router__services__supergraph_service__tests__deferred_fragment_bounds_nullability-2.snap

@@ -23,24 +23,6 @@ expression: stream.next_response().await.unwrap()
               "suborga",
               0
             ]
-          },
-          {
-            "message": "Cannot return null for non-nullable field Organization.nonNullId",
-            "path": [
-              "currentUser",
-              "activeOrganization",
-              "suborga",
-              1
-            ]
-          },
-          {
-            "message": "Cannot return null for non-nullable field Organization.nonNullId",
-            "path": [
-              "currentUser",
-              "activeOrganization",
-              "suborga",
-              2
-            ]
           }
         ]
       }
@@ -55,15 +37,6 @@ expression: stream.next_response().await.unwrap()
       ],
       "extensions": {
         "valueCompletion": [
-          {
-            "message": "Cannot return null for non-nullable field Organization.nonNullId",
-            "path": [
-              "currentUser",
-              "activeOrganization",
-              "suborga",
-              0
-            ]
-          },
           {
             "message": "Cannot return null for non-nullable field Organization.nonNullId",
             "path": [
@@ -72,15 +45,6 @@ expression: stream.next_response().await.unwrap()
               "suborga",
               1
             ]
-          },
-          {
-            "message": "Cannot return null for non-nullable field Organization.nonNullId",
-            "path": [
-              "currentUser",
-              "activeOrganization",
-              "suborga",
-              2
-            ]
           }
         ]
       }
@@ -95,24 +59,6 @@ expression: stream.next_response().await.unwrap()
       ],
       "extensions": {
         "valueCompletion": [
-          {
-            "message": "Cannot return null for non-nullable field Organization.nonNullId",
-            "path": [
-              "currentUser",
-              "activeOrganization",
-              "suborga",
-              0
-            ]
-          },
-          {
-            "message": "Cannot return null for non-nullable field Organization.nonNullId",
-            "path": [
-              "currentUser",
-              "activeOrganization",
-              "suborga",
-              1
-            ]
-          },
           {
             "message": "Cannot return null for non-nullable field Organization.nonNullId",
             "path": [

apollo-router/src/services/snapshots/apollo_router__services__supergraph_service__tests__errors_from_primary_on_deferred_responses-2.snap

@@ -0,0 +1,32 @@
+---
+source: apollo-router/src/services/supergraph_service.rs
+expression: stream.next_response().await.unwrap()
+---
+{
+  "hasNext": false,
+  "incremental": [
+    {
+      "data": {
+        "errorField": null
+      },
+      "path": [
+        "computer"
+      ],
+      "errors": [
+        {
+          "message": "Error field",
+          "locations": [
+            {
+              "line": 1,
+              "column": 93
+            }
+          ],
+          "path": [
+            "computer",
+            "errorField"
+          ]
+        }
+      ]
+    }
+  ]
+}

apollo-router/src/services/snapshots/apollo_router__services__supergraph_service__tests__errors_from_primary_on_deferred_responses.snap

@@ -0,0 +1,12 @@
+---
+source: apollo-router/src/services/supergraph_service.rs
+expression: stream.next_response().await.unwrap()
+---
+{
+  "data": {
+    "computer": {
+      "id": "Computer1"
+    }
+  },
+  "hasNext": true
+}