Dispatch errors from the primary response to deferred responses

NEXT_CHANGELOG.md

@@ -117,6 +117,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 affected to
+deferred responses.
+
+By [@Geal](https://github.com/geal) in https://github.com/apollographql/router/pull/2192
 
 ## 🛠 Maintenance
 

apollo-router/Cargo.toml

@@ -150,7 +150,7 @@ reqwest = { version = "0.11.13", default-features = false, features = [
     "json",
     "stream",
 ] }
-router-bridge = "0.1.11"
+router-bridge = "0.1.12"
 rust-embed="6.4.2"
 schemars = { version = "0.8.11", features = ["url"] }
 shellexpand = "2.1.2"
@@ -190,6 +190,7 @@ urlencoding = "2.1.2"
 uuid = { version = "1.2.2", features = ["serde", "v4"] }
 yaml-rust = "0.4.5"
 askama = "0.11.1"
+apollo-encoder = "0.4.0"
 
 [target.'cfg(macos)'.dependencies]
 uname = "0.1.1"

apollo-router/src/axum_factory/snapshots/apollo_router__axum_factory__tests__defer_is_not_buffered.snap

@@ -0,0 +1,38 @@
+---
+source: apollo-router/src/axum_factory/tests.rs
+expression: "json!([{\n        \"data\" :\n        {\n            \"topProducts\" :\n            [{ \"upc\" : \"1\", \"name\" : \"Table\", \"reviews\" : null },\n            { \"upc\" : \"2\", \"name\" : \"Couch\", \"reviews\" : null }]\n        }, \"errors\" :\n        [{\n            \"message\" :\n            \"couldn't find mock for query {\\\"query\\\":\\\"query TopProducts__reviews__1($representations:[_Any!]!){_entities(representations:$representations){...on Product{reviews{__typename id product{__typename upc}}}}}\\\",\\\"operationName\\\":\\\"TopProducts__reviews__1\\\",\\\"variables\\\":{\\\"representations\\\":[{\\\"__typename\\\":\\\"Product\\\",\\\"upc\\\":\\\"1\\\"},{\\\"__typename\\\":\\\"Product\\\",\\\"upc\\\":\\\"2\\\"}]}}\"\n        },\n        {\n            \"message\" :\n            \"Subgraph response from 'reviews' was missing key `_entities`\",\n            \"path\" : [\"topProducts\", \"@\"]\n        }], \"hasNext\" : true,\n    }, { \"hasNext\" : false }])"
+---
+[
+  {
+    "data": {
+      "topProducts": [
+        {
+          "upc": "1",
+          "name": "Table",
+          "reviews": null
+        },
+        {
+          "upc": "2",
+          "name": "Couch",
+          "reviews": null
+        }
+      ]
+    },
+    "errors": [
+      {
+        "message": "couldn't find mock for query {\"query\":\"query TopProducts__reviews__1($representations:[_Any!]!){_entities(representations:$representations){...on Product{reviews{__typename id product{__typename upc}}}}}\",\"operationName\":\"TopProducts__reviews__1\",\"variables\":{\"representations\":[{\"__typename\":\"Product\",\"upc\":\"1\"},{\"__typename\":\"Product\",\"upc\":\"2\"}]}}"
+      },
+      {
+        "message": "Subgraph response from 'reviews' was missing key `_entities`",
+        "path": [
+          "topProducts",
+          "@"
+        ]
+      }
+    ],
+    "hasNext": true
+  },
+  {
+    "hasNext": false
+  }
+]

apollo-router/src/axum_factory/tests.rs

@@ -1923,31 +1923,7 @@ async fn test_defer_is_not_buffered() {
 
     let (parts, counts): (Vec<_>, Vec<_>) = parts.map(|part| (part, counter.get())).unzip().await;
     let parts = serde_json::Value::Array(parts);
-    assert_eq!(
-        parts,
-        json!([
-            {
-                "data": {
-                    "topProducts": [
-                        {"upc": "1", "name": "Table", "reviews": null},
-                        {"upc": "2", "name": "Couch", "reviews": null}
-                    ]
-                },
-                "errors": [
-                    {
-                        "message": "couldn't find mock for query {\"query\":\"query TopProducts__reviews__1($representations:[_Any!]!){_entities(representations:$representations){...on Product{reviews{__typename id product{__typename upc}}}}}\",\"operationName\":\"TopProducts__reviews__1\",\"variables\":{\"representations\":[{\"__typename\":\"Product\",\"upc\":\"1\"},{\"__typename\":\"Product\",\"upc\":\"2\"}]}}"
-                    },
-                    {
-                        "message": "Subgraph response from 'reviews' was missing key `_entities`",
-                        "path": [ "topProducts", "@" ]
-                    }],
-                "hasNext": true,
-            },
-            {"hasNext": false}
-        ]),
-        "{}",
-        serde_json::to_string(&parts).unwrap()
-    );
+    insta::assert_json_snapshot!(parts);
 
     // Non-regression test for https://github.com/apollographql/router/issues/1572
     //

apollo-router/src/json_ext.rs

@@ -242,6 +242,7 @@ impl ValueExt for Value {
                         .get_mut(k.as_str())
                         .expect("the value at that key was just inserted");
                 }
+                PathElement::Fragment(_) => {}
             }
         }
 
@@ -319,6 +320,7 @@ impl ValueExt for Value {
                         })
                     }
                 },
+                PathElement::Fragment(_) => {}
             }
         }
 
@@ -402,8 +404,17 @@ where
                     iterate_path(parent, &path[1..], value, f);
                     parent.pop();
                 }
+            } else if let Value::Array(array) = data {
+                for (i, value) in array.iter().enumerate() {
+                    parent.push(PathElement::Index(i));
+                    iterate_path(parent, path, value, f);
+                    parent.pop();
+                }
             }
         }
+        Some(PathElement::Fragment(_)) => {
+            iterate_path(parent, &path[1..], data, f);
+        }
     }
 }
 
@@ -422,6 +433,10 @@ pub enum PathElement {
     /// An index path element.
     Index(usize),
 
+    /// A fragment application
+    #[serde(deserialize_with = "deserialize_fragment")]
+    Fragment(String),
+
     /// A key path element.
     Key(String),
 }
@@ -464,6 +479,37 @@ where
     serializer.serialize_str("@")
 }
 
+fn deserialize_fragment<'de, D>(deserializer: D) -> Result<String, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    deserializer.deserialize_str(FragmentVisitor)
+}
+
+struct FragmentVisitor;
+
+impl<'de> serde::de::Visitor<'de> for FragmentVisitor {
+    type Value = String;
+
+    fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
+        write!(formatter, "a string that begins with '... '")
+    }
+
+    fn visit_str<E>(self, s: &str) -> Result<Self::Value, E>
+    where
+        E: serde::de::Error,
+    {
+        if s.starts_with("... ") {
+            Ok(s.to_string())
+        } else {
+            Err(serde::de::Error::invalid_value(
+                serde::de::Unexpected::Str(s),
+                &self,
+            ))
+        }
+    }
+}
+
 /// A path into the result document.
 ///
 /// This can be composed of strings and numbers
@@ -587,6 +633,7 @@ impl fmt::Display for Path {
                 PathElement::Index(index) => write!(f, "{}", index)?,
                 PathElement::Key(key) => write!(f, "{}", key)?,
                 PathElement::Flatten => write!(f, "@")?,
+                PathElement::Fragment(fragment) => write!(f, "{fragment}")?,
             }
         }
         Ok(())

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,11 +267,12 @@ 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!(
                         DEFER_SPAN_NAME,
@@ -357,7 +359,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
@@ -389,7 +391,7 @@ impl DeferredNode {
         let mut stream: stream::FuturesUnordered<_> = deferred_receivers.into_iter().collect();
         //FIXME/ is there a solution without cloning the entire node? Maybe it could be moved instead?
         let deferred_inner = self.node.clone();
-        let deferred_path = self.path.clone();
+        let deferred_path = self.query_path.clone();
         let subselection = self.subselection();
         let label = self.label.clone();
         let mut tx = sender;
@@ -406,8 +408,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 +453,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 +479,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(