set the response path for deferred responses

NEXT_CHANGELOG.md

@@ -418,6 +418,20 @@ This will make containers stop faster as they will not have to wait until a SIGK
 
 By [@Geal](https://github.com/Geal) in https://github.com/apollographql/router/pull/1497
 
+### Set the response path for deferred responses ([PR #1529](https://github.com/apollographql/router/pull/1529))
+
+Some GraphQL clients rely on the response path to find out which
+fragment created a deferred response, and generate code that checks the
+type of the value at that path.
+Previously the router was generating a value that starts at the root
+for every deferred response. Now it checks the path returned by the query
+plan execution and creates a response for each value that matches that
+path.
+In particular, for deferred fragments on an ojbect inside an array, it
+will create a separate response for each element of the array.
+
+By [@Geal](https://github.com/Geal) in https://github.com/apollographql/router/pull/1529
+
 ## 🛠 Maintenance
 
 ### Display licenses.html diff in CI if the check failed ([#1524](https://github.com/apollographql/router/issues/1524))

apollo-router/src/query_planner/mod.rs

@@ -869,7 +869,7 @@ pub(crate) mod fetch {
                 Some(Variables { variables, paths })
             } else {
                 // with nested operations (Query or Mutation has an operation returning a Query or Mutation),
-                // when the first fetch fails, the query plan wwill still execute up until the second fetch,
+                // when the first fetch fails, the query plan will still execute up until the second fetch,
                 // where `requires` is empty (not a federated fetch), the current dir is not emmpty (child of
                 // the previous operation field) and the data is null. In that case, we recognize that we
                 // should not perform the next fetch

apollo-router/src/services/supergraph_service.rs

@@ -5,6 +5,8 @@ use std::task::Poll;
 
 use futures::future::ready;
 use futures::future::BoxFuture;
+use futures::future::Either;
+use futures::stream;
 use futures::stream::once;
 use futures::stream::BoxStream;
 use futures::stream::StreamExt;
@@ -33,6 +35,7 @@ use crate::graphql;
 use crate::graphql::Response;
 use crate::http_ext::Request;
 use crate::introspection::Introspection;
+use crate::json_ext::ValueExt;
 use crate::layers::DEFAULT_BUFFER_SIZE;
 use crate::plugin::DynPlugin;
 use crate::plugin::Handler;
@@ -176,7 +179,7 @@ where
                             response: http::Response::from_parts(
                                 parts,
                                 response_stream
-                                    .map(move |mut response: Response| {
+                                    .flat_map(move |mut response: Response| {
                                         tracing::debug_span!("format_response").in_scope(|| {
                                             query.format_response(
                                                 &mut response,
@@ -186,15 +189,51 @@ where
                                             )
                                         });
 
-                                        // we use the path to look up the subselections, but the generated response
-                                        // is an object starting at the root so the path should be empty
-                                        response.path = None;
-
-                                        if is_deferred {
-                                            response.has_next = Some(true);
+                                        match (response.path.as_ref(), response.data.as_ref()) {
+                                            (None, _) | (_, None) => {
+                                                if is_deferred {
+                                                    response.has_next = Some(true);
+                                                }
+
+                                                Either::Left(once(ready(response)))
+                                            }
+                                            // if the deferred response specified a path, we must extract the
+                                            //values matched by that path and create a separate response for
+                                            //each of them.
+                                            // While { "data": { "a": { "b": 1 } } } and { "data": { "b": 1 }, "path: ["a"] }
+                                            // would merge in the same ways, some clients will generate code
+                                            // that checks the specific type of the deferred response at that
+                                            // path, instead of starting from the root object, so to support
+                                            // this, we extract the value at that path.
+                                            // In particular, that means that a deferred fragment in an object
+                                            // under an array would generate one response par array element
+                                            (Some(response_path), Some(response_data)) => {
+                                                let mut sub_responses = Vec::new();
+                                                response_data.select_values_and_paths(
+                                                    response_path,
+                                                    |path, value| {
+                                                        sub_responses
+                                                            .push((path.clone(), value.clone()));
+                                                    },
+                                                );
+
+                                                Either::Right(stream::iter(
+                                                    sub_responses.into_iter().map(
+                                                        move |(path, data)| Response {
+                                                            label: response.label.clone(),
+                                                            data: Some(data),
+                                                            path: Some(path),
+                                                            errors: response.errors.clone(),
+                                                            extensions: response.extensions.clone(),
+                                                            has_next: Some(true),
+                                                            subselection: response
+                                                                .subselection
+                                                                .clone(),
+                                                        },
+                                                    ),
+                                                ))
+                                            }
                                         }
-
-                                        response
                                     })
                                     .in_current_span()
                                     .boxed(),

apollo-router/tests/integration_tests.rs

@@ -637,6 +637,152 @@ async fn query_just_at_recursion_limit() {
     assert_eq!(registry.totals(), expected_service_hits);
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn defer_path() {
+    let config = serde_json::json!({
+        "server": {
+            "experimental_defer_support": true
+        },
+        "plugins": {
+            "experimental.include_subgraph_errors": {
+                "all": true
+            }
+        }
+    });
+    let request = Request::builder()
+        .query(
+            r#"{
+            me {
+                id
+                ...@defer {
+                    name
+                }
+            }
+        }"#,
+        )
+        .build();
+
+    let request = http_ext::Request::fake_builder()
+        .body(request)
+        .header("content-type", "application/json")
+        .build()
+        .expect("expecting valid request");
+
+    let (router, _) = setup_router_and_registry(config).await;
+
+    let mut stream = router.oneshot(request.into()).await.unwrap();
+
+    let first = stream.next_response().await.unwrap();
+    println!("first: {:?}", first);
+    assert_eq!(
+        first.data.unwrap(),
+        serde_json_bytes::json! {{
+            "me":{
+                "id":"1"
+            }
+        }}
+    );
+    assert_eq!(first.path, None);
+
+    let second = stream.next_response().await.unwrap();
+    println!("second: {:?}", second);
+    assert_eq!(
+        second.data.unwrap(),
+        serde_json_bytes::json! {{
+            "name": "Ada Lovelace"
+        }}
+    );
+    assert_eq!(second.path.unwrap().to_string(), "/me");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn defer_path_in_array() {
+    let config = serde_json::json!({
+        "server": {
+            "experimental_defer_support": true
+        },
+        "plugins": {
+            "experimental.include_subgraph_errors": {
+                "all": true
+            }
+        }
+    });
+    let request = Request::builder()
+        .query(
+            r#"{
+                me {
+                    reviews {
+                        id
+                        author {
+                            id
+                            ... @defer {
+                            name
+                            }
+                        }
+                    }
+                }
+            }"#,
+        )
+        .build();
+
+    let request = http_ext::Request::fake_builder()
+        .body(request)
+        .header("content-type", "application/json")
+        .build()
+        .expect("expecting valid request");
+
+    let (router, _) = setup_router_and_registry(config).await;
+
+    let mut stream = router.oneshot(request.into()).await.unwrap();
+
+    let first = stream.next_response().await.unwrap();
+    println!("first: {:?}", first);
+    assert_eq!(
+        first.data.unwrap(),
+        serde_json_bytes::json! {{
+            "me":{
+                "reviews":[
+                    {
+                        "id": "1",
+                        "author":
+                        {
+                            "id": "1"
+                        }
+                    },
+                    {
+                        "id": "2",
+                        "author":
+                        {
+                            "id": "1"
+                        }
+                    }
+                ]
+            }
+        }}
+    );
+    assert_eq!(first.path, None);
+
+    let second = stream.next_response().await.unwrap();
+    println!("second: {:?}", second);
+    assert_eq!(
+        second.data.unwrap(),
+        serde_json_bytes::json! {{
+            "name": "Ada Lovelace"
+        }}
+    );
+    assert_eq!(second.path.unwrap().to_string(), "/me/reviews/0/author");
+
+    let third = stream.next_response().await.unwrap();
+    println!("third: {:?}", third);
+    assert_eq!(
+        third.data.unwrap(),
+        serde_json_bytes::json! {{
+            "name": "Ada Lovelace"
+        }}
+    );
+    assert_eq!(third.path.unwrap().to_string(), "/me/reviews/1/author");
+}
+
 async fn query_node(
     request: &graphql::Request,
 ) -> Result<graphql::Response, apollo_router::error::FetchError> {