remove Buffer from query deduplication

NEXT_CHANGELOG.md

@@ -117,4 +117,10 @@ This removes `tower::Buffer` usage from the Automated Persisted Queries implemen
 
 By [@Geal](https://github.com/Geal) in https://github.com/apollographql/router/pull/1641
 
+### Remove `Buffer` from query deduplication ([PR #1889](https://github.com/apollographql/router/pull/1889))
+
+This removes `tower::Buffer` usage from the query deduplication implementation to improve reliability.
+
+By [@Geal](https://github.com/Geal) in https://github.com/apollographql/router/pull/1889
+
 ## 📚 Documentation

apollo-router/src/executable.rs

@@ -415,7 +415,7 @@ fn setup_panic_handler(dispatcher: Dispatch) {
     let show_backtraces =
         backtrace_env.as_deref() == Ok("1") || backtrace_env.as_deref() == Ok("full");
     if show_backtraces {
-        tracing::warn!("RUST_BACKTRACE={} detected. This use useful for diagnostics but will have a performance impact and may leak sensitive information", backtrace_env.as_ref().unwrap());
+        tracing::warn!("RUST_BACKTRACE={} detected. This is useful for diagnostics but will have a performance impact and may leak sensitive information", backtrace_env.as_ref().unwrap());
     }
     std::panic::set_hook(Box::new(move |e| {
         with_default(&dispatcher, || {

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

@@ -49,7 +49,8 @@ impl Clone for CloneSubgraphResponse {
     }
 }
 
-pub(crate) struct QueryDeduplicationService<S> {
+#[derive(Clone)]
+pub(crate) struct QueryDeduplicationService<S: Clone> {
     service: S,
     wait_map: WaitMap,
 }

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

@@ -2,14 +2,12 @@
 //!
 //! Currently includes:
 //! * Query deduplication
-//!
-//! Future functionality:
-//! * APQ (already written, but config needs to be moved here)
-//! * Caching
+//! * Timeout
+//! * Compression
 //! * Rate limiting
 //!
 
-mod deduplication;
+pub(crate) mod deduplication;
 mod rate;
 mod timeout;
 
@@ -32,10 +30,8 @@ pub(crate) use self::rate::RateLimited;
 pub(crate) use self::timeout::Elapsed;
 use self::timeout::TimeoutLayer;
 use crate::error::ConfigurationError;
-use crate::layers::ServiceBuilderExt;
 use crate::plugin::Plugin;
 use crate::plugin::PluginInit;
-use crate::plugins::traffic_shaping::deduplication::QueryDeduplicationLayer;
 use crate::register_plugin;
 use crate::services::subgraph;
 use crate::services::subgraph_service::Compression;
@@ -209,12 +205,6 @@ impl Plugin for TrafficShaping {
                     .clone()
             });
             ServiceBuilder::new()
-                .option_layer(config.deduplicate_query.unwrap_or_default().then(|| {
-                    // Buffer is required because dedup layer requires a clone service.
-                    ServiceBuilder::new()
-                        .layer(QueryDeduplicationLayer::default())
-                        .buffered()
-                }))
                 .layer(TimeoutLayer::new(
                     config
                     .timeout

apollo-router/src/router_factory.rs

@@ -7,14 +7,17 @@ use multimap::MultiMap;
 use serde_json::Map;
 use serde_json::Value;
 use tower::service_fn;
+use tower::util::option_layer;
 use tower::BoxError;
+use tower::Layer;
 use tower::ServiceExt;
 use tower_service::Service;
 
 use crate::configuration::Configuration;
 use crate::configuration::ConfigurationError;
 use crate::plugin::DynPlugin;
 use crate::plugin::Handler;
+use crate::plugins::traffic_shaping::deduplication::QueryDeduplicationLayer;
 use crate::services::new_service::NewService;
 use crate::services::RouterCreator;
 use crate::services::SubgraphService;
@@ -115,11 +118,23 @@ impl SupergraphServiceConfigurator for YamlSupergraphServiceFactory {
         // Process the plugins.
         let plugins = create_plugins(&configuration, &schema, extra_plugins).await?;
 
+        let deduplicate_queries = configuration
+            .as_ref()
+            .plugins()
+            .iter()
+            .find(|(name, _)| name == "apollo.traffic_shaping")
+            .and_then(|(_, shaping)| shaping.get("deduplicate_query"))
+            == Some(&serde_json::Value::Bool(true));
+
         let mut builder = PluggableSupergraphServiceBuilder::new(schema.clone());
         builder = builder.with_configuration(configuration);
 
         for (name, _) in schema.subgraphs() {
-            builder = builder.with_subgraph_service(name, SubgraphService::new(name));
+            builder = builder.with_subgraph_service(
+                name,
+                option_layer(deduplicate_queries.then(QueryDeduplicationLayer::default))
+                    .layer(SubgraphService::new(name)),
+            );
         }
 
         for (plugin_name, plugin) in plugins {