*: Add --readonly-db flag disabling WS and db listeners

This is useful when running a Graph Node against a Postgres read replica.

Author: Jannis

chain/ethereum/src/block_stream.rs

@@ -170,7 +170,9 @@ where
         BlockStream {
             state: Mutex::new(BlockStreamState::New),
             consecutive_err_count: 0,
-            chain_head_update_stream: chain_store.chain_head_updates(),
+            chain_head_update_stream: chain_store
+                .chain_head_updates()
+                .expect("running against a readonly database"),
             ctx: BlockStreamContext {
                 subgraph_store,
                 chain_store,

core/src/subgraph/registrar.rs

@@ -154,6 +154,7 @@ where
             .subscribe(vec![
                 SubgraphDeploymentAssignmentEntity::subgraph_entity_pair(),
             ])
+            .expect("running against a readonly database")
             .map_err(|()| format_err!("Entity change stream failed"))
             .map(|event| {
                 // We're only interested in the SubgraphDeploymentAssignment change; we

graph/src/components/store.rs

@@ -881,7 +881,7 @@ pub trait Store: Send + Sync + 'static {
     /// Subscribe to changes for specific subgraphs and entities.
     ///
     /// Returns a stream of store events that match the input arguments.
-    fn subscribe(&self, entities: Vec<SubgraphEntityPair>) -> StoreEventStreamBox;
+    fn subscribe(&self, entities: Vec<SubgraphEntityPair>) -> Option<StoreEventStreamBox>;
 
     fn resolve_subgraph_name_to_id(
         &self,
@@ -1304,7 +1304,7 @@ impl Store for MockStore {
         unimplemented!()
     }
 
-    fn subscribe(&self, _entities: Vec<SubgraphEntityPair>) -> StoreEventStreamBox {
+    fn subscribe(&self, _entities: Vec<SubgraphEntityPair>) -> Option<StoreEventStreamBox> {
         unimplemented!()
     }
 
@@ -1403,7 +1403,7 @@ pub trait ChainStore: Send + Sync + 'static {
     fn attempt_chain_head_update(&self, ancestor_count: u64) -> Result<Vec<H256>, Error>;
 
     /// Subscribe to chain head updates.
-    fn chain_head_updates(&self) -> ChainHeadUpdateStream;
+    fn chain_head_updates(&self) -> Option<ChainHeadUpdateStream>;
 
     /// Get the current head block pointer for this chain.
     /// Any changes to the head block pointer will be to a block with a larger block number, never

graph/src/data/query/error.rs

@@ -60,6 +60,7 @@ pub enum QueryExecutionError {
     Panic(String),
     EventStreamError,
     FulltextQueryRequiresFilter,
+    SubscriptionsDisabled,
 }
 
 impl Error for QueryExecutionError {
@@ -207,6 +208,7 @@ impl fmt::Display for QueryExecutionError {
             Panic(msg) => write!(f, "panic processing query: {}", msg),
             EventStreamError => write!(f, "error in the subscription event stream"),
             FulltextQueryRequiresFilter => write!(f, "fulltext search queries can only use EntityFilter::Equal"),
+            SubscriptionsDisabled => write!(f, "subscriptions are disabled"),
         }
     }
 }

graphql/src/store/resolver.rs

@@ -234,11 +234,15 @@ where
 
         // Subscribe to the store and return the entity change stream
         let deployment_id = parse_subgraph_id(object_type)?;
-        Ok(self.store.subscribe(entities).throttle_while_syncing(
-            &self.logger,
-            self.store.clone(),
-            deployment_id,
-            *SUBSCRIPTION_THROTTLE_INTERVAL,
-        ))
+
+        match self.store.subscribe(entities) {
+            Some(stream) => Ok(stream.throttle_while_syncing(
+                &self.logger,
+                self.store.clone(),
+                deployment_id,
+                *SUBSCRIPTION_THROTTLE_INTERVAL,
+            )),
+            None => Err(QueryExecutionError::SubscriptionsDisabled),
+        }
     }
 }

mock/src/store.rs

@@ -36,7 +36,7 @@ mock! {
 
         fn attempt_chain_head_update(&self, ancestor_count: u64) -> Result<Vec<H256>, Error>;
 
-        fn chain_head_updates(&self) -> ChainHeadUpdateStream;
+        fn chain_head_updates(&self) -> Option<ChainHeadUpdateStream>;
 
         fn chain_head_ptr(&self) -> Result<Option<EthereumBlockPointer>, Error>;
 
@@ -145,7 +145,7 @@ impl Store for MockStore {
         unimplemented!()
     }
 
-    fn subscribe(&self, _entities: Vec<SubgraphEntityPair>) -> StoreEventStreamBox {
+    fn subscribe(&self, _entities: Vec<SubgraphEntityPair>) -> Option<StoreEventStreamBox> {
         unimplemented!()
     }