all: Remove special handling of subscriptions from QueryStore

Author: lutter

graph/src/components/store/traits.rs

@@ -439,12 +439,9 @@ pub trait QueryStoreManager: Send + Sync + 'static {
     /// which deployment will be queried. It is not possible to use the id of the
     /// metadata subgraph, though the resulting store can be used to query
     /// metadata about the deployment `id` (but not metadata about other deployments).
-    ///
-    /// If `for_subscription` is true, the main replica will always be used.
     async fn query_store(
         &self,
         target: QueryTarget,
-        for_subscription: bool,
     ) -> Result<Arc<dyn QueryStore + Send + Sync>, QueryExecutionError>;
 }
 

graphql/src/runner.rs

@@ -103,7 +103,7 @@ where
         // point, and everything needs to go through the `store` we are
         // setting up here
 
-        let store = self.store.query_store(target.clone(), false).await?;
+        let store = self.store.query_store(target.clone()).await?;
         let state = store.deployment_state().await?;
         let network = Some(store.network_name().to_string());
         let schema = store.api_schema()?;

node/src/manager/commands/copy.rs

@@ -106,10 +106,10 @@ pub async fn create(
     let subgraph_store = store.subgraph_store();
     let src = src.locate_unique(&primary)?;
     let query_store = store
-        .query_store(
-            QueryTarget::Deployment(src.hash.clone(), Default::default()),
-            true,
-        )
+        .query_store(QueryTarget::Deployment(
+            src.hash.clone(),
+            Default::default(),
+        ))
         .await?;
     let network = query_store.network_name();
 

server/graphman/tests/deployment_query.rs

@@ -60,10 +60,10 @@ fn graphql_returns_deployment_info() {
         let namespace = format!("sgd{}", locator.id);
         let node = SUBGRAPH_STORE.assigned_node(&locator).unwrap().unwrap();
         let qs = STORE
-            .query_store(
-                QueryTarget::Deployment(locator.hash.clone(), Default::default()),
-                false,
-            )
+            .query_store(QueryTarget::Deployment(
+                locator.hash.clone(),
+                Default::default(),
+            ))
             .await
             .expect("could get a query store");
         let shard = qs.shard();

store/postgres/src/deployment_store.rs

@@ -1434,23 +1434,16 @@ impl DeploymentStore {
         Ok(())
     }
 
-    pub(crate) fn replica_for_query(
-        &self,
-        for_subscription: bool,
-    ) -> Result<ReplicaId, StoreError> {
+    pub(crate) fn replica_for_query(&self) -> Result<ReplicaId, StoreError> {
         use std::sync::atomic::Ordering;
 
-        let replica_id = match for_subscription {
-            // Pick a weighted ReplicaId. `replica_order` contains a list of
-            // replicas with repetitions according to their weight
-            false => {
-                let weights_count = self.replica_order.len();
-                let index =
-                    self.conn_round_robin_counter.fetch_add(1, Ordering::SeqCst) % weights_count;
-                *self.replica_order.get(index).unwrap()
-            }
-            // Subscriptions always go to the main replica.
-            true => ReplicaId::Main,
+        // Pick a weighted ReplicaId. `replica_order` contains a list of
+        // replicas with repetitions according to their weight
+        let replica_id = {
+            let weights_count = self.replica_order.len();
+            let index =
+                self.conn_round_robin_counter.fetch_add(1, Ordering::SeqCst) % weights_count;
+            *self.replica_order.get(index).unwrap()
         };
 
         Ok(replica_id)

store/postgres/src/store.rs

@@ -70,7 +70,6 @@ impl QueryStoreManager for Store {
     async fn query_store(
         &self,
         target: graph::data::query::QueryTarget,
-        for_subscription: bool,
     ) -> Result<
         Arc<dyn graph::prelude::QueryStore + Send + Sync>,
         graph::prelude::QueryExecutionError,
@@ -80,7 +79,7 @@ impl QueryStoreManager for Store {
         let target = target.clone();
         let (store, site, replica) = graph::spawn_blocking_allow_panic(move || {
             store
-                .replica_for_query(target.clone(), for_subscription)
+                .replica_for_query(target.clone())
                 .map_err(|e| e.into())
         })
         .await

store/postgres/src/subgraph_store.rs

@@ -811,15 +811,14 @@ impl SubgraphStoreInner {
     pub(crate) fn replica_for_query(
         &self,
         target: QueryTarget,
-        for_subscription: bool,
     ) -> Result<(Arc<DeploymentStore>, Arc<Site>, ReplicaId), StoreError> {
         let id = match target {
             QueryTarget::Name(name, _) => self.mirror.current_deployment_for_subgraph(&name)?,
             QueryTarget::Deployment(id, _) => id,
         };
 
         let (store, site) = self.store(&id)?;
-        let replica = store.replica_for_query(for_subscription)?;
+        let replica = store.replica_for_query()?;
 
         Ok((store.clone(), site, replica))
     }

store/test-store/src/store.rs

@@ -530,7 +530,7 @@ async fn execute_subgraph_query_internal(
     let deployment = query.schema.id().clone();
     let store = STORE
         .clone()
-        .query_store(QueryTarget::Deployment(deployment, version.clone()), false)
+        .query_store(QueryTarget::Deployment(deployment, version.clone()))
         .await
         .unwrap();
     let state = store.deployment_state().await.unwrap();
@@ -571,10 +571,10 @@ async fn execute_subgraph_query_internal(
 
 pub async fn deployment_state(store: &Store, subgraph_id: &DeploymentHash) -> DeploymentState {
     store
-        .query_store(
-            QueryTarget::Deployment(subgraph_id.clone(), Default::default()),
-            false,
-        )
+        .query_store(QueryTarget::Deployment(
+            subgraph_id.clone(),
+            Default::default(),
+        ))
         .await
         .expect("could get a query store")
         .deployment_state()

store/test-store/tests/postgres/chain_head.rs

@@ -219,10 +219,10 @@ fn test_get_block_number() {
             create_test_subgraph(&subgraph, "type Dummy @entity { id: ID! }").await;
 
             let query_store = subgraph_store
-                .query_store(
-                    QueryTarget::Deployment(subgraph.cheap_clone(), Default::default()),
-                    false,
-                )
+                .query_store(QueryTarget::Deployment(
+                    subgraph.cheap_clone(),
+                    Default::default(),
+                ))
                 .await
                 .unwrap();
 

store/test-store/tests/postgres/subgraph.rs

@@ -696,10 +696,10 @@ fn fatal_vs_non_fatal() {
     run_test_sequentially(|store| async move {
         let deployment = setup().await;
         let query_store = store
-            .query_store(
-                QueryTarget::Deployment(deployment.hash.clone(), Default::default()),
-                false,
-            )
+            .query_store(QueryTarget::Deployment(
+                deployment.hash.clone(),
+                Default::default(),
+            ))
             .await
             .unwrap();
 
@@ -757,10 +757,10 @@ fn fail_unfail_deterministic_error() {
         let deployment = setup().await;
 
         let query_store = store
-            .query_store(
-                QueryTarget::Deployment(deployment.hash.clone(), Default::default()),
-                false,
-            )
+            .query_store(QueryTarget::Deployment(
+                deployment.hash.clone(),
+                Default::default(),
+            ))
             .await
             .unwrap();