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

This reverts commit 9e49c1c.

Author: leoyvens

chain/ethereum/src/block_stream.rs

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

core/src/subgraph/registrar.rs

@@ -154,7 +154,6 @@ 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>) -> Option<StoreEventStreamBox>;
+    fn subscribe(&self, entities: Vec<SubgraphEntityPair>) -> StoreEventStreamBox;
 
     fn resolve_subgraph_name_to_id(
         &self,
@@ -1304,7 +1304,7 @@ impl Store for MockStore {
         unimplemented!()
     }
 
-    fn subscribe(&self, _entities: Vec<SubgraphEntityPair>) -> Option<StoreEventStreamBox> {
+    fn subscribe(&self, _entities: Vec<SubgraphEntityPair>) -> 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) -> Option<ChainHeadUpdateStream>;
+    fn chain_head_updates(&self) -> 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

@@ -61,7 +61,6 @@ pub enum QueryExecutionError {
     Panic(String),
     EventStreamError,
     FulltextQueryRequiresFilter,
-    SubscriptionsDisabled,
 }
 
 impl Error for QueryExecutionError {
@@ -209,7 +208,6 @@ 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"),
             TooExpensive => write!(f, "query is too expensive")
         }
     }

graphql/src/store/resolver.rs

@@ -234,15 +234,11 @@ where
 
         // Subscribe to the store and return the entity change stream
         let deployment_id = parse_subgraph_id(object_type)?;
-
-        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),
-        }
+        Ok(self.store.subscribe(entities).throttle_while_syncing(
+            &self.logger,
+            self.store.clone(),
+            deployment_id,
+            *SUBSCRIPTION_THROTTLE_INTERVAL,
+        ))
     }
 }

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) -> Option<ChainHeadUpdateStream>;
+        fn chain_head_updates(&self) -> 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>) -> Option<StoreEventStreamBox> {
+    fn subscribe(&self, _entities: Vec<SubgraphEntityPair>) -> StoreEventStreamBox {
         unimplemented!()
     }