Fix coprocessor empty body panic (#6398)

.changesets/fix_address_dentist_buyer_frown.md

@@ -0,0 +1,13 @@
+### Fix coprocessor empty body object panic ([PR #6398](https://github.com/apollographql/router/pull/6398))
+If a coprocessor responds with an empty body object at the supergraph stage then the router would panic.
+
+```json
+{
+  ... // other fields
+  "body": {} // empty object
+}
+```
+
+This does not affect coprocessors that respond with formed responses.
+
+By [@BrynCooke](https://github.com/BrynCooke) in https://github.com/apollographql/router/pull/6398

apollo-router/src/services/supergraph/service.rs

@@ -175,11 +175,15 @@ async fn service_call(
         body.operation_name.clone(),
         context.clone(),
         schema.clone(),
+        // We cannot assume that the query is present as it may have been modified by coprocessors or plugins.
+        // There is a deeper issue here in that query analysis is doing a bunch of stuff that it should not and
+        // places the results in context. Therefore plugins that have modified the query won't actually take effect.
+        // However, this can't be resolved before looking at the pipeline again.
         req.supergraph_request
             .body()
             .query
             .clone()
-            .expect("query presence was checked before"),
+            .unwrap_or_default(),
     )
     .await
     {

apollo-router/tests/integration/coprocessor.rs

@@ -83,6 +83,127 @@ async fn test_coprocessor_limit_payload() -> Result<(), BoxError> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn test_coprocessor_response_handling() -> Result<(), BoxError> {
+    if !graph_os_enabled() {
+        return Ok(());
+    }
+    test_full_pipeline(400, "RouterRequest", empty_body_string).await;
+    test_full_pipeline(200, "RouterResponse", empty_body_string).await;
+    test_full_pipeline(500, "SupergraphRequest", empty_body_string).await;
+    test_full_pipeline(500, "SupergraphResponse", empty_body_string).await;
+    test_full_pipeline(200, "SubgraphRequest", empty_body_string).await;
+    test_full_pipeline(200, "SubgraphResponse", empty_body_string).await;
+    test_full_pipeline(500, "ExecutionRequest", empty_body_string).await;
+    test_full_pipeline(500, "ExecutionResponse", empty_body_string).await;
+
+    test_full_pipeline(500, "RouterRequest", empty_body_object).await;
+    test_full_pipeline(500, "RouterResponse", empty_body_object).await;
+    test_full_pipeline(200, "SupergraphRequest", empty_body_object).await;
+    test_full_pipeline(200, "SupergraphResponse", empty_body_object).await;
+    test_full_pipeline(200, "SubgraphRequest", empty_body_object).await;
+    test_full_pipeline(200, "SubgraphResponse", empty_body_object).await;
+    test_full_pipeline(200, "ExecutionRequest", empty_body_object).await;
+    test_full_pipeline(200, "ExecutionResponse", empty_body_object).await;
+
+    test_full_pipeline(200, "RouterRequest", remove_body).await;
+    test_full_pipeline(200, "RouterResponse", remove_body).await;
+    test_full_pipeline(200, "SupergraphRequest", remove_body).await;
+    test_full_pipeline(200, "SupergraphResponse", remove_body).await;
+    test_full_pipeline(200, "SubgraphRequest", remove_body).await;
+    test_full_pipeline(200, "SubgraphResponse", remove_body).await;
+    test_full_pipeline(200, "ExecutionRequest", remove_body).await;
+    test_full_pipeline(200, "ExecutionResponse", remove_body).await;
+
+    test_full_pipeline(500, "RouterRequest", null_out_response).await;
+    test_full_pipeline(500, "RouterResponse", null_out_response).await;
+    test_full_pipeline(500, "SupergraphRequest", null_out_response).await;
+    test_full_pipeline(500, "SupergraphResponse", null_out_response).await;
+    test_full_pipeline(200, "SubgraphRequest", null_out_response).await;
+    test_full_pipeline(200, "SubgraphResponse", null_out_response).await;
+    test_full_pipeline(500, "ExecutionRequest", null_out_response).await;
+    test_full_pipeline(500, "ExecutionResponse", null_out_response).await;
+    Ok(())
+}
+
+fn empty_body_object(mut body: serde_json::Value) -> serde_json::Value {
+    *body
+        .as_object_mut()
+        .expect("body")
+        .get_mut("body")
+        .expect("body") = serde_json::Value::Object(serde_json::Map::new());
+    body
+}
+
+fn empty_body_string(mut body: serde_json::Value) -> serde_json::Value {
+    *body
+        .as_object_mut()
+        .expect("body")
+        .get_mut("body")
+        .expect("body") = serde_json::Value::String("".to_string());
+    body
+}
+
+fn remove_body(mut body: serde_json::Value) -> serde_json::Value {
+    body.as_object_mut().expect("body").remove("body");
+    body
+}
+
+fn null_out_response(_body: serde_json::Value) -> serde_json::Value {
+    serde_json::Value::String("".to_string())
+}
+
+async fn test_full_pipeline(
+    response_status: u16,
+    stage: &'static str,
+    coprocessor: impl Fn(serde_json::Value) -> serde_json::Value + Send + Sync + 'static,
+) {
+    let mock_server = wiremock::MockServer::start().await;
+    let coprocessor_address = mock_server.uri();
+
+    // Expect a small query
+    Mock::given(method("POST"))
+        .and(path("/"))
+        .respond_with(move |req: &wiremock::Request| {
+            let mut body = req.body_json::<serde_json::Value>().expect("body");
+            if body
+                .as_object()
+                .unwrap()
+                .get("stage")
+                .unwrap()
+                .as_str()
+                .unwrap()
+                == stage
+            {
+                body = coprocessor(body);
+            }
+            ResponseTemplate::new(200).set_body_json(body)
+        })
+        .mount(&mock_server)
+        .await;
+
+    let mut router = IntegrationTest::builder()
+        .config(
+            include_str!("fixtures/coprocessor.router.yaml")
+                .replace("<replace>", &coprocessor_address),
+        )
+        .build()
+        .await;
+
+    router.start().await;
+    router.assert_started().await;
+
+    let (_trace_id, response) = router.execute_default_query().await;
+    assert_eq!(
+        response.status(),
+        response_status,
+        "Failed at stage {}",
+        stage
+    );
+
+    router.graceful_shutdown().await;
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn test_coprocessor_demand_control_access() -> Result<(), BoxError> {
     if !graph_os_enabled() {

apollo-router/tests/integration/fixtures/coprocessor.router.yaml

@@ -0,0 +1,24 @@
+# This coprocessor doesn't point to anything
+coprocessor:
+  url: "<replace>"
+  router:
+    request:
+      body: true
+    response:
+      body: true
+  supergraph:
+    request:
+      body: true
+    response:
+      body: true
+  subgraph:
+    all:
+      request:
+        body: true
+      response:
+        body: true
+  execution:
+    request:
+      body: true
+    response:
+      body: true