Merge pull request #1383 from apollographql/api-1.0

Api 1.0

NEXT_CHANGELOG.md

@@ -25,6 +25,16 @@ By [@USERNAME](https://github.com/USERNAME) in https://github.com/apollographql/
 
 # [0.10.1] (unreleased) - 2022-mm-dd
 ## ❗ BREAKING ❗
+
+### Relax plugin api mutability ([PR #1340](https://github.com/apollographql/router/pull/1340) ([PR #1289](https://github.com/apollographql/router/pull/1289)
+
+the `Plugin::*_service()` methods were taking a `&mut self` as argument, but since
+they work like a tower Layer, they can use `&self` instead. This change
+then allows us to move from Buffer to service factories for the query
+planner, execution and subgraph services
+
+By [@Geal](https://github.com/Geal) in https://github.com/apollographql/router/pull/1340 https://github.com/apollographql/router/pull/1289
+
 ## 🚀 Features
 
 ### Add support to add custom resources on metrics. [PR #1354](https://github.com/apollographql/router/pull/1354)
@@ -66,17 +76,11 @@ By [@bnjjj](https://github.com/bnjjj) in https://github.com/apollographql/router
 
 ## 🛠 Maintenance
 
-### execute the query plan's first response directly  ([PR #1357](https://github.com/apollographql/router/issues/1357))
-
-The query plan was entirely executed in a spawned task to prepare for the `@defer` implementation, but we can actually
-generate the first response right inside the same future.
-
-By [@Geal](https://github.com/Geal) in https://github.com/apollographql/router/pull/1357
-
-### Remove deprecated `failure` crate from the dependency tree [PR #1373](https://github.com/apollographql/router/pull/1373)
+### Replace Buffers of tower services with service factories([PR #1289](https://github.com/apollographql/router/pull/1289) [PR #1355](https://github.com/apollographql/router/pull/1355))
 
-This should fix automated reports about [GHSA-jq66-xh47-j9f3](https://github.com/advisories/GHSA-jq66-xh47-j9f3).
+tower services should be used by creating a new service instance for each new session
+instead of going through a Buffer.
 
-By [@yanns](https://github.com/yanns) in https://github.com/apollographql/router/pull/1373
+By [@Geal](https://github.com/Geal) in https://github.com/apollographql/router/pull/1289  https://github.com/apollographql/router/pull/1355
 
 ## 📚 Documentation

apollo-router-benchmarks/benches/basic_composition.rs

@@ -10,9 +10,9 @@ fn from_elem(c: &mut Criterion) {
 
         let builder = setup();
 
-        let (router, _) = runtime.block_on(builder.build()).unwrap();
+        let router = runtime.block_on(builder.build()).unwrap();
         b.to_async(runtime)
-            .iter(|| basic_composition_benchmark(router.clone()));
+            .iter(|| basic_composition_benchmark(router.test_service()));
     });
 }
 

apollo-router-benchmarks/src/lib.rs

@@ -7,7 +7,7 @@ pub mod tests {
         let runtime = tokio::runtime::Runtime::new().unwrap();
 
         let builder = setup();
-        let (router, _) = runtime.block_on(builder.build()).unwrap();
-        runtime.block_on(basic_composition_benchmark(router.clone()));
+        let router = runtime.block_on(builder.build()).unwrap();
+        runtime.block_on(basic_composition_benchmark(router.test_service()));
     }
 }

apollo-router-scaffold/templates/plugin/src/plugins/{{snake_name}}.rs

@@ -52,7 +52,7 @@ impl Plugin for {{pascal_name}} {
 
     // Delete this function if you are not customizing it.
     fn router_service(
-        &mut self,
+        &self,
         service: BoxService<RouterRequest, RouterResponse<BoxStream<'static, Response>>, BoxError>,
     ) -> BoxService<RouterRequest, RouterResponse<BoxStream<'static, Response>>, BoxError> {
         // Always use service builder to compose your plugins.
@@ -68,23 +68,23 @@ impl Plugin for {{pascal_name}} {
 
     // Delete this function if you are not customizing it.
     fn query_planning_service(
-        &mut self,
+        &self,
         service: BoxService<QueryPlannerRequest, QueryPlannerResponse, BoxError>,
     ) -> BoxService<QueryPlannerRequest, QueryPlannerResponse, BoxError> {
         service
     }
 
     // Delete this function if you are not customizing it.
     fn execution_service(
-        &mut self,
+        &self,
         service: BoxService<ExecutionRequest, ExecutionResponse<BoxStream<'static, Response>>, BoxError>,
     ) -> BoxService<ExecutionRequest, ExecutionResponse<BoxStream<'static, Response>>, BoxError> {
         service
     }
 
     // Delete this function if you are not customizing it.
     fn subgraph_service(
-        &mut self,
+        &self,
         _name: &str,
         service: BoxService<SubgraphRequest, SubgraphResponse, BoxError>,
     ) -> BoxService<SubgraphRequest, SubgraphResponse, BoxError> {
@@ -104,7 +104,7 @@ impl Plugin for {{pascal_name}} {
     }
 
     fn router_service(
-        &mut self,
+        &self,
         service: BoxService<RouterRequest, RouterResponse<BoxStream<'static, Response>>, BoxError>,
     ) -> BoxService<RouterRequest, RouterResponse<BoxStream<'static, Response>>, BoxError> {
 
@@ -131,7 +131,7 @@ impl Plugin for {{pascal_name}} {
     }
 
     fn router_service(
-        &mut self,
+        &self,
         service: BoxService<RouterRequest, RouterResponse<BoxStream<'static, Response>>, BoxError>,
     ) -> BoxService<RouterRequest, RouterResponse<BoxStream<'static, Response>>, BoxError> {
 

apollo-router/src/axum_http_server_factory.rs

@@ -38,7 +38,6 @@ use tokio::net::TcpListener;
 #[cfg(unix)]
 use tokio::net::UnixListener;
 use tokio::sync::Notify;
-use tower::buffer::Buffer;
 use tower::util::BoxService;
 use tower::BoxError;
 use tower::MakeService;
@@ -58,9 +57,9 @@ use crate::http_server_factory::HttpServerFactory;
 use crate::http_server_factory::HttpServerHandle;
 use crate::http_server_factory::Listener;
 use crate::http_server_factory::NetworkStream;
-use crate::layers::DEFAULT_BUFFER_SIZE;
 use crate::plugin::Handler;
 use crate::router::ApolloRouterError;
+use crate::router_factory::RouterServiceFactory;
 
 /// A basic http server using Axum.
 /// Uses streaming as primary method of response.
@@ -74,38 +73,19 @@ impl AxumHttpServerFactory {
     }
 }
 
-type BufferedService = Buffer<
-    BoxService<
-        http_ext::Request<graphql::Request>,
-        http_ext::Response<BoxStream<'static, graphql::Response>>,
-        BoxError,
-    >,
-    http_ext::Request<graphql::Request>,
->;
-
 impl HttpServerFactory for AxumHttpServerFactory {
     type Future = Pin<Box<dyn Future<Output = Result<HttpServerHandle, ApolloRouterError>> + Send>>;
 
-    fn create<RS>(
+    fn create<RF>(
         &self,
-        service: RS,
+        service_factory: RF,
         configuration: Arc<Configuration>,
         listener: Option<Listener>,
         plugin_handlers: HashMap<String, Handler>,
     ) -> Self::Future
     where
-        RS: Service<
-                http_ext::Request<graphql::Request>,
-                Response = http_ext::Response<BoxStream<'static, graphql::Response>>,
-                Error = BoxError,
-            > + Send
-            + Sync
-            + Clone
-            + 'static,
-
-        <RS as Service<http_ext::Request<graphql::Request>>>::Future: std::marker::Send,
+        RF: RouterServiceFactory,
     {
-        let boxed_service = Buffer::new(service.boxed(), DEFAULT_BUFFER_SIZE);
         Box::pin(async move {
             let (shutdown_sender, shutdown_receiver) = oneshot::channel::<()>();
             let listen_address = configuration.server.listen.clone();
@@ -136,12 +116,31 @@ impl HttpServerFactory for AxumHttpServerFactory {
                     get({
                         let display_landing_page = configuration.server.landing_page;
                         move |host: Host,
-                              service: Extension<BufferedService>,
+                              Extension(service): Extension<RF>,
                               http_request: Request<Body>| {
-                            handle_get(host, service, http_request, display_landing_page)
+                            handle_get(
+                                host,
+                                service.new_service().boxed(),
+                                http_request,
+                                display_landing_page,
+                            )
                         }
                     })
-                    .post(handle_post),
+                    .post({
+                        move |host: Host,
+                              uri: OriginalUri,
+                              request: Json<graphql::Request>,
+                              Extension(service): Extension<RF>,
+                              header_map: HeaderMap| {
+                            handle_post(
+                                host,
+                                uri,
+                                request,
+                                service.new_service().boxed(),
+                                header_map,
+                            )
+                        }
+                    }),
                 )
                 .layer(middleware::from_fn(decompress_request_body))
                 .layer(
@@ -162,7 +161,7 @@ impl HttpServerFactory for AxumHttpServerFactory {
                         }),
                 )
                 .route(&configuration.server.health_check_path, get(health_check))
-                .layer(Extension(boxed_service))
+                .layer(Extension(service_factory))
                 .layer(cors)
                 .layer(CompressionLayer::new()); // To compress response body
 
@@ -418,7 +417,11 @@ async fn custom_plugin_handler(
 
 async fn handle_get(
     Host(host): Host,
-    Extension(service): Extension<BufferedService>,
+    service: BoxService<
+        http_ext::Request<graphql::Request>,
+        http_ext::Response<BoxStream<'static, graphql::Response>>,
+        BoxError,
+    >,
     http_request: Request<Body>,
     display_landing_page: bool,
 ) -> impl IntoResponse {
@@ -452,7 +455,11 @@ async fn handle_post(
     Host(host): Host,
     OriginalUri(uri): OriginalUri,
     Json(request): Json<graphql::Request>,
-    Extension(service): Extension<BufferedService>,
+    service: BoxService<
+        http_ext::Request<graphql::Request>,
+        http_ext::Response<BoxStream<'static, graphql::Response>>,
+        BoxError,
+    >,
     header_map: HeaderMap,
 ) -> impl IntoResponse {
     let mut http_request = Request::post(
@@ -477,10 +484,17 @@ async fn health_check() -> impl IntoResponse {
     Json(json!({ "status": "pass" }))
 }
 
-async fn run_graphql_request(
-    service: BufferedService,
+async fn run_graphql_request<RS>(
+    service: RS,
     http_request: Request<graphql::Request>,
-) -> impl IntoResponse {
+) -> impl IntoResponse
+where
+    RS: Service<
+            http_ext::Request<graphql::Request>,
+            Response = http_ext::Response<BoxStream<'static, graphql::Response>>,
+            Error = BoxError,
+        > + Send,
+{
     match service.ready_oneshot().await {
         Ok(mut service) => {
             let (head, body) = http_request.into_parts();
@@ -679,6 +693,7 @@ mod tests {
     use super::*;
     use crate::configuration::Cors;
     use crate::http_ext::Request;
+    use crate::services::new_service::NewService;
 
     macro_rules! assert_header {
         ($response:expr, $header:expr, $expected:expr $(, $msg:expr)?) => {
@@ -726,6 +741,32 @@ mod tests {
         }
     }
 
+    type MockRouterServiceType = tower_test::mock::Mock<
+        http_ext::Request<graphql::Request>,
+        http_ext::Response<Pin<Box<dyn Stream<Item = graphql::Response> + Send>>>,
+    >;
+
+    #[derive(Clone)]
+    struct TestRouterServiceFactory {
+        inner: MockRouterServiceType,
+    }
+
+    impl NewService<Request<graphql::Request>> for TestRouterServiceFactory {
+        type Service = MockRouterServiceType;
+
+        fn new_service(&self) -> Self::Service {
+            self.inner.clone()
+        }
+    }
+
+    impl RouterServiceFactory for TestRouterServiceFactory {
+        type RouterService = MockRouterServiceType;
+
+        type Future = <<TestRouterServiceFactory as NewService<
+            http_ext::Request<graphql::Request>,
+        >>::Service as Service<http_ext::Request<graphql::Request>>>::Future;
+    }
+
     async fn init(mut mock: MockRouterService) -> (HttpServerHandle, Client) {
         let server_factory = AxumHttpServerFactory::new();
         let (service, mut handle) = tower_test::mock::spawn();
@@ -742,7 +783,9 @@ mod tests {
         });
         let server = server_factory
             .create(
-                service.into_inner(),
+                TestRouterServiceFactory {
+                    inner: service.into_inner(),
+                },
                 Arc::new(
                     Configuration::builder()
                         .server(
@@ -792,7 +835,14 @@ mod tests {
             }
         });
         let server = server_factory
-            .create(service.into_inner(), Arc::new(conf), None, plugin_handlers)
+            .create(
+                TestRouterServiceFactory {
+                    inner: service.into_inner(),
+                },
+                Arc::new(conf),
+                None,
+                plugin_handlers,
+            )
             .await
             .expect("Failed to create server factory");
         let mut default_headers = HeaderMap::new();
@@ -826,7 +876,9 @@ mod tests {
         });
         let server = server_factory
             .create(
-                service.into_inner(),
+                TestRouterServiceFactory {
+                    inner: service.into_inner(),
+                },
                 Arc::new(
                     Configuration::builder()
                         .server(