Provide a backpressure enabled pipeline

.changesets/feat_garypen_next_backpressure.md

@@ -0,0 +1,9 @@
+### Enabling back-pressure in the request processing pipeline ([PR #6486](https://github.com/apollographql/router/pull/6486))
+
+In Router 1.x, back-pressure was not maintained. Requests would be accepted by the router. This could cause issue for routers which were accepting high levels of traffic.
+
+We are now improving the handling of back-pressure so that traffic shaping measures are more effective and integration with telemetry is improved. In particular, this means that telemetry events will not be lost due to traffic shaping and that traffic shaping now works more precisely. This will make the behaviour of the router more predictable.
+
+For more details about how these improvements effect the router please refer to the [migrating from 1.x guide](reference/migration/from-router-v1.mdx).
+
+By [@garypen](https://github.com/garypen) in https://github.com/apollographql/router/pull/6486

apollo-router/src/axum_factory/tests.rs

@@ -1,4 +1,3 @@
-use std::collections::HashMap;
 use std::io;
 use std::net::SocketAddr;
 use std::pin::Pin;
@@ -67,28 +66,18 @@ use crate::graphql;
 use crate::http_server_factory::HttpServerFactory;
 use crate::http_server_factory::HttpServerHandle;
 use crate::json_ext::Path;
-use crate::plugin::test::MockSubgraph;
-use crate::query_planner::QueryPlannerService;
-use crate::router_factory::create_plugins;
 use crate::router_factory::Endpoint;
 use crate::router_factory::RouterFactory;
-use crate::services::execution;
-use crate::services::layers::persisted_queries::PersistedQueryLayer;
-use crate::services::layers::query_analysis::QueryAnalysisLayer;
 use crate::services::layers::static_page::home_page_content;
 use crate::services::layers::static_page::sandbox_page_content;
 use crate::services::new_service::ServiceFactory;
 use crate::services::router;
-use crate::services::router::service::RouterCreator;
 use crate::services::supergraph;
-use crate::services::HasSchema;
-use crate::services::PluggableSupergraphServiceBuilder;
 use crate::services::RouterRequest;
 use crate::services::RouterResponse;
 use crate::services::SupergraphResponse;
 use crate::services::MULTIPART_DEFER_ACCEPT;
 use crate::services::MULTIPART_DEFER_CONTENT_TYPE;
-use crate::spec::Schema;
 use crate::test_harness::http_client;
 use crate::test_harness::http_client::MaybeMultipart;
 use crate::uplink::license_enforcement::LicenseState;
@@ -2406,114 +2395,3 @@ async fn test_supergraph_and_health_check_same_port_different_listener() {
         error.to_string()
     );
 }
-
-#[tokio::test]
-async fn test_supergraph_timeout() {
-    let config = serde_json::json!({
-        "supergraph": {
-            "listen": "127.0.0.1:0",
-            "defer_support": false,
-        },
-        "traffic_shaping": {
-            "router": {
-                "timeout": "1ns"
-            }
-        },
-    });
-
-    let conf: Arc<Configuration> = Arc::new(serde_json::from_value(config).unwrap());
-
-    let schema = include_str!("..//testdata/minimal_supergraph.graphql");
-    let schema = Arc::new(Schema::parse(schema, &conf).unwrap());
-    let planner = QueryPlannerService::new(schema.clone(), conf.clone())
-        .await
-        .unwrap();
-
-    // we do the entire supergraph rebuilding instead of using `from_supergraph_mock_callback_and_configuration`
-    // because we need the plugins to apply on the supergraph
-    let subgraph_schemas = Arc::new(
-        planner
-            .subgraph_schemas()
-            .iter()
-            .map(|(k, v)| (k.clone(), v.schema.clone()))
-            .collect(),
-    );
-    let mut plugins = create_plugins(&conf, &schema, subgraph_schemas, None, None)
-        .await
-        .unwrap();
-
-    plugins.insert("delay".into(), Box::new(Delay));
-
-    struct Delay;
-
-    #[async_trait::async_trait]
-    impl crate::plugin::Plugin for Delay {
-        type Config = ();
-
-        async fn new(_: crate::plugin::PluginInit<()>) -> Result<Self, BoxError> {
-            Ok(Self)
-        }
-
-        fn execution_service(&self, service: execution::BoxService) -> execution::BoxService {
-            service
-                .map_future(|fut| async {
-                    tokio::time::sleep(Duration::from_millis(10)).await;
-                    fut.await
-                })
-                .boxed()
-        }
-    }
-
-    let builder = PluggableSupergraphServiceBuilder::new(planner)
-        .with_configuration(conf.clone())
-        .with_subgraph_service("accounts", MockSubgraph::new(HashMap::new()));
-
-    let supergraph_creator = builder
-        .with_plugins(Arc::new(plugins))
-        .build()
-        .await
-        .unwrap();
-
-    let service = RouterCreator::new(
-        QueryAnalysisLayer::new(supergraph_creator.schema(), Arc::clone(&conf)).await,
-        Arc::new(PersistedQueryLayer::new(&conf).await.unwrap()),
-        Arc::new(supergraph_creator),
-        conf.clone(),
-    )
-    .await
-    .unwrap()
-    .make();
-
-    // keep the server handle around otherwise it will immediately shutdown
-    let (server, client) = init_with_config(service, conf.clone(), MultiMap::new())
-        .await
-        .unwrap();
-    let url = server
-        .graphql_listen_address()
-        .as_ref()
-        .unwrap()
-        .to_string();
-
-    let response = client
-        .post(url)
-        .body(r#"{ "query": "{ me }" }"#)
-        .send()
-        .await
-        .unwrap();
-
-    assert_eq!(response.status(), StatusCode::GATEWAY_TIMEOUT);
-
-    let body = response.bytes().await.unwrap();
-    let body: serde_json::Value = serde_json::from_slice(&body).unwrap();
-    assert_eq!(
-        body,
-        json!({
-             "errors": [{
-                 "message": "Request timed out",
-                 "extensions": {
-                     "code": "REQUEST_TIMEOUT"
-                 }
-             }]
-        })
-    );
-}

apollo-router/src/batching.rs

@@ -281,7 +281,7 @@ impl Batch {
                                 request, sender, ..
                             } in cancelled_requests
                             {
-                                let subgraph_name = request.subgraph_name.ok_or(SubgraphBatchingError::MissingSubgraphName)?;
+                                let subgraph_name = request.subgraph_name;
                                 if let Err(log_error) = sender.send(Err(Box::new(FetchError::SubrequestBatchingError {
                                         service: subgraph_name.clone(),
                                         reason: format!("request cancelled: {reason}"),
@@ -365,7 +365,7 @@ impl Batch {
                 sender: tx,
             } in all_in_one
             {
-                let subgraph_name = sg_request.subgraph_name.clone().ok_or(SubgraphBatchingError::MissingSubgraphName)?;
+                let subgraph_name = sg_request.subgraph_name.clone();
                 let value = svc_map
                     .entry(
                         subgraph_name,
@@ -583,7 +583,7 @@ mod tests {
                     .body(graphql::Response::builder().data(data.clone()).build())
                     .unwrap(),
                 context: Context::new(),
-                subgraph_name: None,
+                subgraph_name: String::default(),
                 id: SubgraphRequestId(String::new()),
             };
 

apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__schema_generation.snap

@@ -5380,6 +5380,13 @@ expression: "&schema"
     "RouterShaping": {
       "additionalProperties": false,
       "properties": {
+        "concurrency_limit": {
+          "description": "The global concurrency limit",
+          "format": "uint",
+          "minimum": 0.0,
+          "nullable": true,
+          "type": "integer"
+        },
         "global_rate_limit": {
           "$ref": "#/definitions/RateLimitConf",
           "description": "#/definitions/RateLimitConf",

apollo-router/src/error.rs

@@ -630,8 +630,6 @@ impl std::fmt::Display for ValidationErrors {
 pub(crate) enum SubgraphBatchingError {
     /// Sender unavailable
     SenderUnavailable,
-    /// Request does not have a subgraph name
-    MissingSubgraphName,
     /// Requests is empty
     RequestsIsEmpty,
     /// Batch processing failed: {0}