Fix coprocessor empty body panic

If a coprocessor responds with an empty body at the supergraph stage then the router would panic as there was a false expectation that the query had already been validated.

There is a deeper issue around coprocessors and their ability to modify the query before processing. As query analysis and pluginss may use the query to populate context modifying the query won't necessarily do what the user wants. In fact the router will simply ignore the modified query as the parsed version is already in context.

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,105 @@ 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