subscriptions: No backpressure to notify queue

Author: leoyvens

graph/src/components/store.rs

@@ -623,6 +623,8 @@ pub struct StoreEventStream<S> {
 pub type StoreEventStreamBox =
     StoreEventStream<Box<dyn Stream<Item = Arc<StoreEvent>, Error = ()> + Send>>;
 
+pub type UnitStream = Box<dyn futures03::Stream<Item = ()> + Unpin + Send + Sync>;
+
 impl<S> Stream for StoreEventStream<S>
 where
     S: Stream<Item = Arc<StoreEvent>, Error = ()> + Send,
@@ -661,6 +663,8 @@ where
     /// on the returned stream as a single `StoreEvent`; the events are
     /// combined by using the maximum of all sources and the concatenation
     /// of the changes of the `StoreEvents` received during the interval.
+    //
+    // Currently unused, needs to be made compatible with `subscribe_no_payload`.
     pub async fn throttle_while_syncing(
         self,
         logger: &Logger,
@@ -849,6 +853,9 @@ pub trait SubscriptionManager: Send + Sync + 'static {
     ///
     /// Returns a stream of store events that match the input arguments.
     fn subscribe(&self, entities: Vec<SubscriptionFilter>) -> StoreEventStreamBox;
+
+    /// If the payload is not required, use for a more efficient subscription mechanism backed by a watcher.
+    fn subscribe_no_payload(&self, entities: Vec<SubscriptionFilter>) -> UnitStream;
 }
 
 /// An internal identifer for the specific instance of a deployment. The

graph/src/data/subscription/result.rs

@@ -1,10 +1,10 @@
 use crate::prelude::QueryResult;
-use std::marker::Unpin;
+use std::pin::Pin;
 use std::sync::Arc;
 
 /// A stream of query results for a subscription.
 pub type QueryResultStream =
-    Box<dyn futures03::stream::Stream<Item = Arc<QueryResult>> + Send + Unpin>;
+    Pin<Box<dyn futures03::stream::Stream<Item = Arc<QueryResult>> + Send>>;
 
 /// The result of running a subscription, if successful.
 pub type SubscriptionResult = QueryResultStream;

graphql/src/execution/resolver.rs

@@ -1,7 +1,8 @@
 use std::collections::HashMap;
 
 use crate::execution::ExecutionContext;
-use graph::prelude::{async_trait, q, s, tokio, Error, QueryExecutionError, StoreEventStreamBox};
+use graph::components::store::UnitStream;
+use graph::prelude::{async_trait, q, s, tokio, Error, QueryExecutionError};
 use graph::{
     data::graphql::{ext::DocumentExt, ObjectOrInterface},
     prelude::{r, QueryResult},
@@ -109,12 +110,12 @@ pub trait Resolver: Sized + Send + Sync + 'static {
     }
 
     // Resolves a change stream for a given field.
-    async fn resolve_field_stream(
+    fn resolve_field_stream(
         &self,
         _schema: &s::Document,
         _object_type: &s::ObjectType,
         _field: &q::Field,
-    ) -> Result<StoreEventStreamBox, QueryExecutionError> {
+    ) -> Result<UnitStream, QueryExecutionError> {
         Err(QueryExecutionError::NotSupported(String::from(
             "Resolving field streams is not supported by this resolver",
         )))

graphql/src/runner.rs

@@ -347,7 +347,6 @@ where
                 result_size: self.result_size.clone(),
             },
         )
-        .await
     }
 
     fn load_manager(&self) -> Arc<LoadManager> {

graphql/src/store/resolver.rs

@@ -19,6 +19,7 @@ use crate::store::query::collect_entities_from_query_field;
 /// A resolver that fetches entities from a `Store`.
 #[derive(Clone)]
 pub struct StoreResolver {
+    #[allow(dead_code)]
     logger: Logger,
     pub(crate) store: Arc<dyn QueryStore>,
     subscription_manager: Arc<dyn SubscriptionManager>,
@@ -298,25 +299,17 @@ impl Resolver for StoreResolver {
         }
     }
 
-    async fn resolve_field_stream(
+    fn resolve_field_stream(
         &self,
         schema: &s::Document,
         object_type: &s::ObjectType,
         field: &q::Field,
-    ) -> result::Result<StoreEventStreamBox, QueryExecutionError> {
+    ) -> result::Result<UnitStream, QueryExecutionError> {
         // Collect all entities involved in the query field
         let entities = collect_entities_from_query_field(schema, object_type, field);
 
         // Subscribe to the store and return the entity change stream
-        Ok(self
-            .subscription_manager
-            .subscribe(entities)
-            .throttle_while_syncing(
-                &self.logger,
-                self.store.clone(),
-                *SUBSCRIPTION_THROTTLE_INTERVAL,
-            )
-            .await)
+        Ok(self.subscription_manager.subscribe_no_payload(entities))
     }
 
     fn post_process(&self, result: &mut QueryResult) -> Result<(), anyhow::Error> {

graphql/src/subscription/mod.rs

@@ -3,6 +3,7 @@ use std::iter;
 use std::result::Result;
 use std::time::{Duration, Instant};
 
+use graph::components::store::UnitStream;
 use graph::{components::store::SubscriptionManager, prelude::*};
 
 use crate::runner::ResultSizeMetrics;
@@ -40,7 +41,7 @@ pub struct SubscriptionExecutionOptions {
     pub result_size: Arc<ResultSizeMetrics>,
 }
 
-pub async fn execute_subscription(
+pub fn execute_subscription(
     subscription: Subscription,
     schema: Arc<ApiSchema>,
     options: SubscriptionExecutionOptions,
@@ -53,10 +54,10 @@ pub async fn execute_subscription(
         options.max_complexity,
         options.max_depth,
     )?;
-    execute_prepared_subscription(query, options).await
+    execute_prepared_subscription(query, options)
 }
 
-pub(crate) async fn execute_prepared_subscription(
+pub(crate) fn execute_prepared_subscription(
     query: Arc<crate::execution::Query>,
     options: SubscriptionExecutionOptions,
 ) -> Result<SubscriptionResult, SubscriptionError> {
@@ -72,15 +73,15 @@ pub(crate) async fn execute_prepared_subscription(
         "query" => &query.query_text,
     );
 
-    let source_stream = create_source_event_stream(query.clone(), &options).await?;
+    let source_stream = create_source_event_stream(query.clone(), &options)?;
     let response_stream = map_source_to_response_stream(query, options, source_stream);
     Ok(response_stream)
 }
 
-async fn create_source_event_stream(
+fn create_source_event_stream(
     query: Arc<crate::execution::Query>,
     options: &SubscriptionExecutionOptions,
-) -> Result<StoreEventStreamBox, SubscriptionError> {
+) -> Result<UnitStream, SubscriptionError> {
     let resolver = StoreResolver::for_subscription(
         &options.logger,
         query.schema.id().clone(),
@@ -123,35 +124,31 @@ async fn create_source_event_stream(
     let field = fields.1[0];
     let argument_values = coerce_argument_values(&ctx.query, subscription_type.as_ref(), field)?;
 
-    resolve_field_stream(&ctx, &subscription_type, field, argument_values).await
+    resolve_field_stream(&ctx, &subscription_type, field, argument_values)
 }
 
-async fn resolve_field_stream(
+fn resolve_field_stream(
     ctx: &ExecutionContext<impl Resolver>,
     object_type: &s::ObjectType,
     field: &q::Field,
     _argument_values: HashMap<&str, r::Value>,
-) -> Result<StoreEventStreamBox, SubscriptionError> {
+) -> Result<UnitStream, SubscriptionError> {
     ctx.resolver
         .resolve_field_stream(&ctx.query.schema.document(), object_type, field)
-        .await
         .map_err(SubscriptionError::from)
 }
 
 fn map_source_to_response_stream(
     query: Arc<crate::execution::Query>,
     options: SubscriptionExecutionOptions,
-    source_stream: StoreEventStreamBox,
+    source_stream: UnitStream,
 ) -> QueryResultStream {
     // Create a stream with a single empty event. By chaining this in front
     // of the real events, we trick the subscription into executing its query
     // at least once. This satisfies the GraphQL over Websocket protocol
     // requirement of "respond[ing] with at least one GQL_DATA message", see
     // https://github.com/apollographql/subscriptions-transport-ws/blob/master/PROTOCOL.md#gql_data
-    let trigger_stream = futures03::stream::iter(vec![Ok(Arc::new(StoreEvent {
-        tag: 0,
-        changes: Default::default(),
-    }))]);
+    let trigger_stream = futures03::stream::once(async {});
 
     let SubscriptionExecutionOptions {
         logger,
@@ -165,43 +162,34 @@ fn map_source_to_response_stream(
         result_size,
     } = options;
 
-    Box::new(
-        trigger_stream
-            .chain(source_stream.compat())
-            .then(move |res| match res {
-                Err(()) => {
-                    futures03::future::ready(Arc::new(QueryExecutionError::EventStreamError.into()))
-                        .boxed()
-                }
-                Ok(event) => execute_subscription_event(
-                    logger.clone(),
-                    store.clone(),
-                    subscription_manager.cheap_clone(),
-                    query.clone(),
-                    event,
-                    timeout,
-                    max_first,
-                    max_skip,
-                    result_size.cheap_clone(),
-                )
-                .boxed(),
-            }),
-    )
+    trigger_stream
+        .chain(source_stream)
+        .then(move |()| {
+            execute_subscription_event(
+                logger.clone(),
+                store.clone(),
+                subscription_manager.cheap_clone(),
+                query.clone(),
+                timeout,
+                max_first,
+                max_skip,
+                result_size.cheap_clone(),
+            )
+            .boxed()
+        })
+        .boxed()
 }
 
 async fn execute_subscription_event(
     logger: Logger,
     store: Arc<dyn QueryStore>,
     subscription_manager: Arc<dyn SubscriptionManager>,
     query: Arc<crate::execution::Query>,
-    event: Arc<StoreEvent>,
     timeout: Option<Duration>,
     max_first: u32,
     max_skip: u32,
     result_size: Arc<ResultSizeMetrics>,
 ) -> Arc<QueryResult> {
-    debug!(logger, "Execute subscription event"; "event" => format!("{:?}", event));
-
     let resolver = match StoreResolver::at_block(
         &logger,
         store,

graphql/tests/query.rs

@@ -887,9 +887,8 @@ fn query_complexity_subscriptions() {
 
         // This query is exactly at the maximum complexity.
         // FIXME: Not collecting the stream because that will hang the test.
-        let _ignore_stream = execute_subscription(Subscription { query }, schema.clone(), options)
-            .await
-            .unwrap();
+        let _ignore_stream =
+            execute_subscription(Subscription { query }, schema.clone(), options).unwrap();
 
         let query = Query::new(
             graphql_parser::parse_query(
@@ -934,7 +933,7 @@ fn query_complexity_subscriptions() {
         };
 
         // The extra introspection causes the complexity to go over.
-        let result = execute_subscription(Subscription { query }, schema, options).await;
+        let result = execute_subscription(Subscription { query }, schema, options);
         match result {
             Err(SubscriptionError::GraphQLError(e)) => match e[0] {
                 QueryExecutionError::TooComplex(1_010_200, _) => (), // Expected
@@ -1314,9 +1313,7 @@ fn subscription_gets_result_even_without_events() {
         };
         // Execute the subscription and expect at least one result to be
         // available in the result stream
-        let stream = execute_subscription(Subscription { query }, schema, options)
-            .await
-            .unwrap();
+        let stream = execute_subscription(Subscription { query }, schema, options).unwrap();
         let results: Vec<_> = stream
             .take(1)
             .collect()

node/src/manager/mod.rs

@@ -1,5 +1,5 @@
 use graph::{
-    components::store::SubscriptionManager,
+    components::store::{SubscriptionManager, UnitStream},
     prelude::{StoreEventStreamBox, SubscriptionFilter},
 };
 
@@ -15,4 +15,8 @@ impl SubscriptionManager for PanicSubscriptionManager {
     fn subscribe(&self, _: Vec<SubscriptionFilter>) -> StoreEventStreamBox {
         panic!("we were never meant to call `subscribe`");
     }
+
+    fn subscribe_no_payload(&self, _: Vec<SubscriptionFilter>) -> UnitStream {
+        panic!("we were never meant to call `subscribe_no_payload`");
+    }
 }

store/postgres/src/store_events.rs

@@ -3,12 +3,13 @@ use graph::tokio_stream::wrappers::ReceiverStream;
 use std::sync::{atomic::Ordering, Arc, RwLock};
 use std::{collections::HashMap, sync::atomic::AtomicUsize};
 use tokio::sync::mpsc::{channel, Sender};
+use tokio::sync::watch;
 use uuid::Uuid;
 
 use crate::notification_listener::{NotificationListener, SafeChannelName};
-use graph::components::store::SubscriptionManager as SubscriptionManagerTrait;
+use graph::components::store::{SubscriptionManager as SubscriptionManagerTrait, UnitStream};
 use graph::prelude::serde_json;
-use graph::prelude::*;
+use graph::{prelude::*, tokio_stream};
 
 pub struct StoreEventListener {
     notification_listener: NotificationListener,
@@ -100,6 +101,17 @@ impl EventSink for Sender<Arc<StoreEvent>> {
     }
 }
 
+#[async_trait]
+impl EventSink for watch::Sender<()> {
+    async fn send(&self, _event: Arc<StoreEvent>) -> Result<(), Error> {
+        Ok(self.send(())?)
+    }
+
+    fn is_closed(&self) -> bool {
+        self.is_closed()
+    }
+}
+
 /// Manage subscriptions to the `StoreEvent` stream. Keep a list of
 /// currently active subscribers and forward new events to each of them
 pub struct SubscriptionManager {
@@ -204,4 +216,17 @@ impl SubscriptionManagerTrait for SubscriptionManager {
         StoreEventStream::new(Box::new(ReceiverStream::new(receiver).map(Ok).compat()))
             .filter_by_entities(entities)
     }
+
+    fn subscribe_no_payload(&self, entities: Vec<SubscriptionFilter>) -> UnitStream {
+        let id = Uuid::new_v4().to_string();
+
+        let (sender, receiver) = watch::channel(());
+
+        self.subscriptions
+            .write()
+            .unwrap()
+            .insert(id, (Arc::new(entities.clone()), Arc::new(sender)));
+
+        Box::new(tokio_stream::wrappers::WatchStream::new(receiver))
+    }
 }