BoxCloneService router pipeline improvements

apollo-router/src/axum_factory/tests.rs

@@ -84,7 +84,6 @@ use crate::services::RouterResponse;
 use crate::services::SupergraphResponse;
 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::pipeline_handle::PipelineRef;
 use crate::test_harness::http_client;
@@ -144,20 +143,10 @@ struct TestRouterFactory {
     inner: MockRouterServiceType,
 }
 
-impl ServiceFactory<router::Request> for TestRouterFactory {
-    type Service = MockRouterServiceType;
-
-    fn create(&self) -> Self::Service {
-        self.inner.clone()
-    }
-}
-
 impl RouterFactory for TestRouterFactory {
-    type RouterService = MockRouterServiceType;
-
-    type Future = <<TestRouterFactory as ServiceFactory<router::Request>>::Service as Service<
-        router::Request,
-    >>::Future;
+    fn create(&self) -> router::BoxCloneService {
+        self.inner.clone().boxed_clone()
+    }
 
     fn web_endpoints(&self) -> MultiMap<ListenAddr, Endpoint> {
         MultiMap::new()

apollo-router/src/layers/mod.rs

@@ -101,7 +101,8 @@ pub trait ServiceBuilderExt<L>: Sized {
 
     /// Decide if processing should continue or not, and if not allow returning of a response.
     /// Unlike checkpoint it is possible to perform async operations in the callback. However
-    /// this requires that the service is `Clone`. This can be achieved using `.buffered()`.
+    /// the resulting service requires `S: Clone`. Since `BoxCloneService` is already `Clone`,
+    /// a `.buffered()` call is no longer needed when wrapping a `BoxCloneService`.
     ///
     /// This is useful for things like authentication where you need to make an external call to
     /// check if a request should proceed or not.
@@ -158,7 +159,18 @@ pub trait ServiceBuilderExt<L>: Sized {
 
     /// Adds a buffer to the service stack with a default size.
     ///
-    /// This is useful for making services `Clone` and `Send`
+    /// The buffer spawns a dedicated worker task and queues requests in an in-memory channel.
+    /// The primary reasons to include a buffer are:
+    ///
+    /// - **Backpressure**: callers block (rather than failing immediately) when the inner
+    ///   service is busy processing previous requests.
+    /// - **`LoadShed` / `ConcurrencyLimit` / `RateLimit` interaction**: these layers
+    ///   signal overload by returning `Poll::Pending` from `poll_ready`. A buffer placed
+    ///   *before* them absorbs that pending state and prevents Tokio's cooperative-scheduling
+    ///   budget from causing spurious `Overloaded` responses.
+    ///
+    /// Now that pipeline services are `BoxCloneService`, a buffer is **no longer needed
+    /// merely to make a service `Clone` or `Send`**.
     ///
     /// # Examples
     ///

apollo-router/src/plugin/mod.rs

@@ -935,6 +935,8 @@ macro_rules! register_private_plugin {
 /// Handler represents a [`Plugin`] endpoint.
 #[derive(Clone)]
 pub(crate) struct Handler {
+    // BoxCloneService is Send but not Sync. The buffer's handle (Arc + Sender) is Sync,
+    // which is required for axum's route_service handler to be usable across threads.
     service: UnconstrainedBuffer<
         router::Request,
         <router::BoxCloneService as Service<router::Request>>::Future,

apollo-router/src/plugins/cache/entity.rs

@@ -51,7 +51,6 @@ use crate::graphql::Error;
 use crate::json_ext::Object;
 use crate::json_ext::Path;
 use crate::json_ext::PathElement;
-use crate::layers::ServiceBuilderExt;
 use crate::plugin::PluginInit;
 use crate::plugin::PluginPrivate;
 use crate::plugins::authorization::CacheKeyMetadata;
@@ -407,10 +406,7 @@ impl PluginPrivate for EntityCache {
                     response
                 })
                 .service(CacheService {
-                    service: ServiceBuilder::new()
-                        .buffered()
-                        .service(service)
-                        .boxed_clone(),
+                    service,
                     entity_type: self.entity_type.clone(),
                     name: name.to_string(),
                     storage,

apollo-router/src/plugins/cache/metrics.rs

@@ -9,15 +9,13 @@ use http::header;
 use parking_lot::Mutex;
 use serde_json_bytes::Value;
 use tower::BoxError;
-use tower::ServiceBuilder;
 use tower::ServiceExt;
 use tower_service::Service;
 
 use super::entity::REPRESENTATIONS;
 use super::entity::Ttl;
 use super::entity::hash_query;
 use super::entity::hash_vary_headers;
-use crate::layers::ServiceBuilderExt;
 use crate::services::subgraph;
 use crate::spec::TYPENAME;
 
@@ -32,10 +30,7 @@ impl CacheMetricsService {
         separate_per_type: bool,
     ) -> subgraph::BoxCloneService {
         tower::util::BoxCloneService::new(CacheMetricsService {
-            service: ServiceBuilder::new()
-                .buffered()
-                .service(service)
-                .boxed_clone(),
+            service,
             name: Arc::new(name),
             counter: Some(Arc::new(Mutex::new(CacheCounter::new(
                 ttl.map(|t| t.0).unwrap_or_else(|| Duration::from_secs(60)),

apollo-router/src/plugins/cache/tests.rs

@@ -21,7 +21,6 @@ use crate::MockedSubgraphs;
 use crate::TestHarness;
 use crate::cache::redis::RedisCacheStorage;
 use crate::plugin::test::MockSubgraph;
-use crate::plugin::test::MockSubgraphService;
 use crate::plugins::cache::entity::CONTEXT_CACHE_KEYS;
 use crate::plugins::cache::entity::CacheKeyContext;
 use crate::plugins::cache::entity::CacheKeysContext;
@@ -964,12 +963,10 @@ async fn no_data() {
         .extra_private_plugin(entity_cache)
         .subgraph_hook(|name, service| {
             if name == "orga" {
-                let mut subgraph = MockSubgraphService::new();
-                subgraph
-                    .expect_call()
-                    .times(1)
-                    .returning(move |_req: subgraph::Request| Err("orga not found".into()));
-                subgraph.boxed_clone()
+                tower::service_fn(|_req: subgraph::Request| async {
+                    Err::<subgraph::Response, _>("orga not found".into())
+                })
+                .boxed_clone()
             } else {
                 service
             }

apollo-router/src/plugins/connectors/tests/mod.rs

@@ -32,9 +32,9 @@ use crate::metrics::FutureMetricsExt;
 use crate::plugins::connectors::tests::req_asserts::Plan;
 use crate::plugins::telemetry::consts::CONNECT_SPAN_NAME;
 use crate::plugins::telemetry::consts::OTEL_STATUS_CODE;
+use crate::router_factory::RouterFactory;
 use crate::router_factory::RouterSuperServiceFactory;
 use crate::router_factory::YamlRouterFactory;
-use crate::services::new_service::ServiceFactory;
 use crate::services::router::Request;
 use crate::services::supergraph;
 use crate::uplink::license_enforcement::LicenseState;

apollo-router/src/plugins/coprocessor/connector.rs

@@ -215,7 +215,6 @@ impl ConnectorStage {
             .instrument(external_service_span())
             .option_layer(request_layer)
             .option_layer(response_layer)
-            .buffered()
             .service(service)
             .boxed_clone()
     }

apollo-router/src/plugins/coprocessor/execution.rs

@@ -172,7 +172,6 @@ impl ExecutionStage {
             .instrument(external_service_span())
             .option_layer(request_layer)
             .option_layer(response_layer)
-            .buffered() // XXX: Added during backpressure fixing
             .service(service)
             .boxed_clone()
     }

apollo-router/src/plugins/coprocessor/mod.rs

@@ -789,7 +789,6 @@ impl RouterStage {
             .instrument(external_service_span())
             .option_layer(request_layer)
             .option_layer(response_layer)
-            .buffered() // XXX: Added during backpressure fixing
             .service(service)
             .boxed_clone()
     }
@@ -921,7 +920,6 @@ impl SubgraphStage {
             .instrument(external_service_span())
             .option_layer(request_layer)
             .option_layer(response_layer)
-            .buffered() // XXX: Added during backpressure fixing
             .service(service)
             .boxed_clone()
     }

apollo-router/src/plugins/coprocessor/supergraph.rs

@@ -175,7 +175,6 @@ impl SupergraphStage {
             .instrument(external_service_span())
             .option_layer(request_layer)
             .option_layer(response_layer)
-            .buffered() // XXX: Added during backpressure fixing
             .service(service)
             .boxed_clone()
     }

apollo-router/src/plugins/file_uploads/mod.rs

@@ -80,7 +80,6 @@ impl PluginPrivate for FileUploadsPlugin {
                 }
                 .boxed()
             })
-            .buffered()
             .service(service)
             .boxed_clone()
     }
@@ -108,7 +107,6 @@ impl PluginPrivate for FileUploadsPlugin {
                 }
                 .boxed()
             })
-            .buffered()
             .service(service)
             .boxed_clone()
     }
@@ -149,7 +147,6 @@ impl PluginPrivate for FileUploadsPlugin {
                     .map(|req| Ok(ControlFlow::Continue(req)))
                     .boxed()
             })
-            .buffered()
             .service(service)
             .boxed_clone()
     }

apollo-router/src/plugins/license_enforcement/mod.rs

@@ -65,6 +65,8 @@ impl PluginPrivate for LicenseEnforcement {
     }
 
     fn router_service(&self, service: router::BoxCloneService) -> router::BoxCloneService {
+        // Buffer required before load_shed() for correct cooperative-scheduling behaviour
+        // (see ServiceBuilderExt::buffered).
         ServiceBuilder::new()
             .buffered()
             .map_future_with_request_data(

apollo-router/src/plugins/limits/layer.rs

@@ -105,13 +105,22 @@ where
     }
 }
 
-#[derive(Clone)]
 pub(crate) struct RequestBodyLimit<Body, S> {
     _phantom: std::marker::PhantomData<Body>,
     inner: S,
     initial_limit: usize,
 }
 
+impl<Body, S: Clone> Clone for RequestBodyLimit<Body, S> {
+    fn clone(&self) -> Self {
+        Self {
+            _phantom: std::marker::PhantomData,
+            inner: self.inner.clone(),
+            initial_limit: self.initial_limit,
+        }
+    }
+}
+
 impl<Body, S> RequestBodyLimit<Body, S>
 where
     S: Service<http::request::Request<super::limited::Limited<Body>>>,

apollo-router/src/plugins/limits/mod.rs

@@ -173,7 +173,6 @@ impl Plugin for LimitsPlugin {
 
     fn router_service(&self, service: router::BoxCloneService) -> router::BoxCloneService {
         ServiceBuilder::new()
-            .buffered()
             .map_future_with_request_data(
                 |r: &router::Request| r.context.clone(),
                 |ctx, f| async { Self::map_error_to_graphql(f.await, ctx) },

apollo-router/src/plugins/response_cache/plugin.rs

@@ -58,7 +58,6 @@ use crate::graphql::Error;
 use crate::json_ext::Object;
 use crate::json_ext::Path;
 use crate::json_ext::PathElement;
-use crate::layers::ServiceBuilderExt;
 use crate::plugin::PluginInit;
 use crate::plugin::PluginPrivate;
 use crate::plugins::authorization::CacheKeyMetadata;
@@ -454,10 +453,7 @@ impl PluginPrivate for ResponseCache {
                     response
                 })
                 .service(CacheService {
-                    service: ServiceBuilder::new()
-                        .buffered()
-                        .service(service)
-                        .boxed_clone(),
+                    service,
                     entity_type: self.entity_type.clone(),
                     name: name.to_string(),
                     storage: self.storage.clone(),

apollo-router/src/plugins/response_cache/tests.rs

@@ -21,7 +21,6 @@ use crate::configuration::subgraph::SubgraphConfiguration;
 use crate::graphql;
 use crate::metrics::FutureMetricsExt;
 use crate::plugin::test::MockSubgraph;
-use crate::plugin::test::MockSubgraphService;
 use crate::plugins::response_cache::debugger::CacheKeysContext;
 use crate::plugins::response_cache::invalidation::InvalidationRequest;
 use crate::plugins::response_cache::invalidation_endpoint::SubgraphInvalidationConfig;
@@ -2622,12 +2621,10 @@ async fn no_data() {
         .extra_private_plugin(response_cache)
         .subgraph_hook(|name, service| {
             if name == "orga" {
-                let mut subgraph = MockSubgraphService::new();
-                subgraph
-                    .expect_call()
-                    .times(1)
-                    .returning(move |_req: subgraph::Request| Err("orga not found".into()));
-                subgraph.boxed_clone()
+                tower::service_fn(|_req: subgraph::Request| async {
+                    Err::<subgraph::Response, _>("orga not found".into())
+                })
+                .boxed_clone()
             } else {
                 service
             }

apollo-router/src/plugins/traffic_shaping/mod.rs

@@ -328,7 +328,10 @@ impl PluginPrivate for TrafficShaping {
 
     fn router_service(&self, service: router::BoxCloneService) -> router::BoxCloneService {
         // NB: consider each triplet (map_future_with_request_data, load_shed, layer) as a unit of
-        //  behavior
+        //  behavior.
+        // The outer buffer before the first load_shed() is required for correct cooperative-
+        // scheduling behaviour: without it, Tokio's budget can cause poll_ready to return
+        // Pending spuriously and the load_shed would emit false Overloaded errors.
         ServiceBuilder::new()
             .buffered()
             .map_future_with_request_data(
@@ -433,6 +436,10 @@ impl PluginPrivate for TrafficShaping {
                         .clone()
                 });
 
+            // Outer buffer: required before load_shed() for correct cooperative-scheduling
+            // behaviour (see router_service above for the full explanation).
+            // Inner buffer (below): provides a backpressure surface so that RateLimitLayer
+            // can return poll_ready Pending and LoadShed will actually shed that load.
             ServiceBuilder::new()
                 .buffered()
                 .map_future_with_request_data(
@@ -514,6 +521,9 @@ impl PluginPrivate for TrafficShaping {
                     .clone()
             });
 
+            // Outer buffer: required before load_shed() for correct cooperative-scheduling
+            // behaviour (see router_service above for the full explanation).
+            // Inner buffer (below): provides the backpressure surface for RateLimitLayer.
             ServiceBuilder::new()
                 .buffered()
                 .map_future_with_request_data(
@@ -803,9 +813,9 @@ mod test {
         builder = builder.with_plugins(plugins);
 
         let builder = builder
-            .with_subgraph_service("accounts", account_service.clone())
-            .with_subgraph_service("reviews", review_service.clone())
-            .with_subgraph_service("products", product_service.clone());
+            .with_subgraph_service("accounts", account_service.boxed_clone())
+            .with_subgraph_service("reviews", review_service.boxed_clone())
+            .with_subgraph_service("products", product_service.boxed_clone());
 
         let supergraph_creator = builder.build().await.expect("should build");
 

apollo-router/src/query_planner/execution.rs

@@ -44,8 +44,7 @@ use crate::services::FetchRequest;
 use crate::services::fetch;
 use crate::services::fetch::ErrorMapping;
 use crate::services::fetch::SubscriptionRequest;
-use crate::services::fetch_service::FetchServiceFactory;
-use crate::services::new_service::ServiceFactory;
+use crate::services::fetch_service::FetchService;
 use crate::spec::Query;
 use crate::spec::Schema;
 
@@ -55,7 +54,7 @@ impl QueryPlan {
     pub(crate) async fn execute<'a>(
         &self,
         context: &'a Context,
-        service_factory: &'a Arc<FetchServiceFactory>,
+        service_factory: &'a FetchService,
         // The original supergraph request is used to populate variable values and for plugin
         // features like propagating headers or subgraph telemetry based on supergraph request
         // values.
@@ -117,7 +116,7 @@ impl QueryPlan {
 // holds the query plan executon arguments that do not change between calls
 pub(crate) struct ExecutionParameters<'a> {
     pub(crate) context: &'a Context,
-    pub(crate) service_factory: &'a Arc<FetchServiceFactory>,
+    pub(crate) service_factory: &'a FetchService,
     pub(crate) schema: &'a Arc<Schema>,
     pub(crate) subgraph_schemas: &'a Arc<SubgraphSchemas>,
     pub(crate) supergraph_request: &'a Arc<http::Request<Request>>,
@@ -239,7 +238,7 @@ impl PlanNode {
                             &None,
                         ) {
                             Some(variables) => {
-                                let service = parameters.service_factory.create();
+                                let service = parameters.service_factory.clone();
                                 let request = fetch::Request::Subscription(
                                     SubscriptionRequest::builder()
                                         .context(parameters.context.clone())
@@ -295,7 +294,7 @@ impl PlanNode {
                         ) {
                             Some(variables) => {
                                 let paths = variables.inverted_paths.clone();
-                                let service = parameters.service_factory.create();
+                                let service = parameters.service_factory.clone();
                                 let request = fetch::Request::Fetch(
                                     FetchRequest::builder()
                                         .context(parameters.context.clone())