graph: use ReadStore instead of WriteStore for source subgraph stores

incrypto32 · incrypto32 · commit 9069486cedbf · 2024-11-12T14:28:30.000+04:00
diff --git a/chain/arweave/src/chain.rs b/chain/arweave/src/chain.rs
@@ -7,7 +7,7 @@ use graph::blockchain::{
 };
 use graph::cheap_clone::CheapClone;
 use graph::components::adapter::ChainId;
-use graph::components::store::{DeploymentCursorTracker, WritableStore};
+use graph::components::store::{DeploymentCursorTracker, ReadStore};
 use graph::data::subgraph::UnifiedMappingApiVersion;
 use graph::env::EnvVars;
 use graph::firehose::FirehoseEndpoint;
@@ -121,7 +121,7 @@ impl Blockchain for Chain {
         deployment: DeploymentLocator,
         store: impl DeploymentCursorTracker,
         start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
         filter: Arc<TriggerFilterWrapper<Self>>,
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error> {
diff --git a/chain/cosmos/src/chain.rs b/chain/cosmos/src/chain.rs
@@ -12,7 +12,7 @@ use graph::blockchain::block_stream::{BlockStreamError, BlockStreamMapper, Fireh
 use graph::blockchain::client::ChainClient;
 use graph::blockchain::{BasicBlockchainBuilder, BlockchainBuilder, NoopRuntimeAdapter};
 use graph::cheap_clone::CheapClone;
-use graph::components::store::{DeploymentCursorTracker, WritableStore};
+use graph::components::store::{DeploymentCursorTracker, ReadStore};
 use graph::data::subgraph::UnifiedMappingApiVersion;
 use graph::{
     blockchain::{
@@ -114,7 +114,7 @@ impl Blockchain for Chain {
         deployment: DeploymentLocator,
         store: impl DeploymentCursorTracker,
         start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
         filter: Arc<TriggerFilterWrapper<Self>>,
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error> {
diff --git a/chain/ethereum/src/chain.rs b/chain/ethereum/src/chain.rs
@@ -7,7 +7,7 @@ use graph::blockchain::{
     TriggersAdapterSelector,
 };
 use graph::components::adapter::ChainId;
-use graph::components::store::{DeploymentCursorTracker, WritableStore};
+use graph::components::store::{DeploymentCursorTracker, ReadStore};
 use graph::data::subgraph::UnifiedMappingApiVersion;
 use graph::firehose::{FirehoseEndpoint, ForkStep};
 use graph::futures03::compat::Future01CompatExt;
@@ -128,7 +128,7 @@ impl BlockStreamBuilder<Chain> for EthereumStreamBuilder {
         chain: &Chain,
         deployment: DeploymentLocator,
         start_blocks: Vec<BlockNumber>,
-        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
+        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
         subgraph_current_block: Option<BlockPtr>,
         filter: Arc<TriggerFilterWrapper<Chain>>,
         unified_api_version: UnifiedMappingApiVersion,
@@ -150,7 +150,7 @@ impl BlockStreamBuilder<Chain> for EthereumStreamBuilder {
         chain: &Chain,
         deployment: DeploymentLocator,
         start_blocks: Vec<BlockNumber>,
-        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
+        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
         subgraph_current_block: Option<BlockPtr>,
         filter: Arc<TriggerFilterWrapper<Chain>>,
         unified_api_version: UnifiedMappingApiVersion,
@@ -437,7 +437,7 @@ impl Blockchain for Chain {
         deployment: DeploymentLocator,
         store: impl DeploymentCursorTracker,
         start_blocks: Vec<BlockNumber>,
-        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
+        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
         filter: Arc<TriggerFilterWrapper<Self>>,
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error> {
diff --git a/chain/near/src/chain.rs b/chain/near/src/chain.rs
@@ -8,7 +8,7 @@ use graph::blockchain::{
 };
 use graph::cheap_clone::CheapClone;
 use graph::components::adapter::ChainId;
-use graph::components::store::{DeploymentCursorTracker, WritableStore};
+use graph::components::store::{DeploymentCursorTracker, ReadStore};
 use graph::data::subgraph::UnifiedMappingApiVersion;
 use graph::env::EnvVars;
 use graph::firehose::FirehoseEndpoint;
@@ -152,7 +152,7 @@ impl BlockStreamBuilder<Chain> for NearStreamBuilder {
         _chain: &Chain,
         _deployment: DeploymentLocator,
         _start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
         _subgraph_current_block: Option<BlockPtr>,
         _filter: Arc<TriggerFilterWrapper<Chain>>,
         _unified_api_version: UnifiedMappingApiVersion,
@@ -232,7 +232,7 @@ impl Blockchain for Chain {
         deployment: DeploymentLocator,
         store: impl DeploymentCursorTracker,
         start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
         filter: Arc<TriggerFilterWrapper<Self>>,
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error> {
diff --git a/chain/starknet/src/chain.rs b/chain/starknet/src/chain.rs
@@ -16,7 +16,7 @@ use graph::{
     cheap_clone::CheapClone,
     components::{
         adapter::ChainId,
-        store::{DeploymentCursorTracker, DeploymentLocator, WritableStore},
+        store::{DeploymentCursorTracker, DeploymentLocator, ReadStore},
     },
     data::subgraph::UnifiedMappingApiVersion,
     env::EnvVars,
@@ -116,7 +116,7 @@ impl Blockchain for Chain {
         deployment: DeploymentLocator,
         store: impl DeploymentCursorTracker,
         start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
         filter: Arc<TriggerFilterWrapper<Self>>,
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error> {
@@ -240,7 +240,7 @@ impl BlockStreamBuilder<Chain> for StarknetStreamBuilder {
         _chain: &Chain,
         _deployment: DeploymentLocator,
         _start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
         _subgraph_current_block: Option<BlockPtr>,
         _filter: Arc<TriggerFilterWrapper<Chain>>,
         _unified_api_version: UnifiedMappingApiVersion,
diff --git a/chain/substreams/src/block_stream.rs b/chain/substreams/src/block_stream.rs
@@ -9,7 +9,7 @@ use graph::{
         substreams_block_stream::SubstreamsBlockStream,
         Blockchain, TriggerFilterWrapper,
     },
-    components::store::{DeploymentLocator, WritableStore},
+    components::store::{DeploymentLocator, ReadStore},
     data::subgraph::UnifiedMappingApiVersion,
     prelude::{async_trait, BlockNumber, BlockPtr, DeploymentHash},
     schema::InputSchema,
@@ -104,7 +104,7 @@ impl BlockStreamBuilderTrait<Chain> for BlockStreamBuilder {
         _chain: &Chain,
         _deployment: DeploymentLocator,
         _start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
         _subgraph_current_block: Option<BlockPtr>,
         _filter: Arc<TriggerFilterWrapper<Chain>>,
         _unified_api_version: UnifiedMappingApiVersion,
diff --git a/chain/substreams/src/chain.rs b/chain/substreams/src/chain.rs
@@ -7,7 +7,7 @@ use graph::blockchain::{
     NoopRuntimeAdapter, TriggerFilterWrapper,
 };
 use graph::components::adapter::ChainId;
-use graph::components::store::{DeploymentCursorTracker, WritableStore};
+use graph::components::store::{DeploymentCursorTracker, ReadStore};
 use graph::env::EnvVars;
 use graph::prelude::{
     BlockHash, CheapClone, DeploymentHash, Entity, LoggerFactory, MetricsRegistry,
@@ -142,7 +142,7 @@ impl Blockchain for Chain {
         deployment: DeploymentLocator,
         store: impl DeploymentCursorTracker,
         _start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
         filter: Arc<TriggerFilterWrapper<Self>>,
         _unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error> {
diff --git a/core/src/subgraph/inputs.rs b/core/src/subgraph/inputs.rs
@@ -1,7 +1,7 @@
 use graph::{
     blockchain::{block_stream::TriggersAdapterWrapper, Blockchain},
     components::{
-        store::{DeploymentLocator, SubgraphFork, WritableStore},
+        store::{DeploymentLocator, ReadStore, SubgraphFork, WritableStore},
         subgraph::ProofOfIndexingVersion,
     },
     data::subgraph::{SubgraphFeature, UnifiedMappingApiVersion},
@@ -16,7 +16,7 @@ pub struct IndexingInputs<C: Blockchain> {
     pub features: BTreeSet<SubgraphFeature>,
     pub start_blocks: Vec<BlockNumber>,
     pub end_blocks: BTreeSet<BlockNumber>,
-    pub source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
+    pub source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
     pub stop_block: Option<BlockNumber>,
     pub store: Arc<dyn WritableStore>,
     pub debug_fork: Option<Arc<dyn SubgraphFork>>,
diff --git a/core/src/subgraph/instance_manager.rs b/core/src/subgraph/instance_manager.rs
@@ -9,7 +9,7 @@ use crate::subgraph::runner::SubgraphRunner;
 use graph::blockchain::block_stream::{BlockStreamMetrics, TriggersAdapterWrapper};
 use graph::blockchain::{Blockchain, BlockchainKind, DataSource, NodeCapabilities};
 use graph::components::metrics::gas::GasMetrics;
-use graph::components::store::WritableStore;
+use graph::components::store::ReadStore;
 use graph::components::subgraph::ProofOfIndexingVersion;
 use graph::data::subgraph::{UnresolvedSubgraphManifest, SPEC_VERSION_0_0_6};
 use graph::data::value::Word;
@@ -204,14 +204,14 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
         }
     }
 
-    pub async fn hashes_to_writable_store<C: Blockchain>(
+    pub async fn hashes_to_read_store<C: Blockchain>(
         &self,
         logger: &Logger,
         link_resolver: &Arc<dyn LinkResolver>,
         hashes: Vec<DeploymentHash>,
         max_spec_version: Version,
         is_runner_test: bool,
-    ) -> anyhow::Result<Vec<(DeploymentHash, Arc<dyn WritableStore>)>> {
+    ) -> anyhow::Result<Vec<(DeploymentHash, Arc<dyn ReadStore>)>> {
         let mut writable_stores = Vec::new();
         let subgraph_store = self.subgraph_store.clone();
 
@@ -235,16 +235,16 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
                 .active_locator(&hash)?
                 .ok_or_else(|| anyhow!("no active deployment for hash {}", hash))?;
 
-            let writable_store = subgraph_store
-                .clone() // Clone the Arc again for each iteration
-                .writable(
+            let readable_store = subgraph_store
+                .clone()
+                .readable(
                     logger.clone(),
                     loc.id.clone(),
                     Arc::new(manifest.template_idx_and_name().collect()),
                 )
                 .await?;
 
-            writable_stores.push((loc.hash, writable_store));
+            writable_stores.push((loc.hash, readable_store));
         }
 
         Ok(writable_stores)
@@ -491,8 +491,8 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
 
         let decoder = Box::new(Decoder::new(decoder_hook));
 
-        let subgraph_data_source_writables = self
-            .hashes_to_writable_store::<C>(
+        let subgraph_data_source_read_stores = self
+            .hashes_to_read_store::<C>(
                 &logger,
                 &link_resolver,
                 subgraph_ds_source_deployments,
@@ -503,15 +503,15 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
 
         let triggers_adapter = Arc::new(TriggersAdapterWrapper::new(
             triggers_adapter,
-            subgraph_data_source_writables.clone(),
+            subgraph_data_source_read_stores.clone(),
         ));
 
         let inputs = IndexingInputs {
             deployment: deployment.clone(),
             features,
             start_blocks,
             end_blocks,
-            source_subgraph_stores: subgraph_data_source_writables,
+            source_subgraph_stores: subgraph_data_source_read_stores,
             stop_block,
             store,
             debug_fork,
diff --git a/graph/src/blockchain/block_stream.rs b/graph/src/blockchain/block_stream.rs
@@ -19,7 +19,7 @@ use super::{
     Block, BlockPtr, BlockTime, Blockchain, SubgraphFilter, Trigger, TriggerFilterWrapper,
 };
 use crate::anyhow::Result;
-use crate::components::store::{BlockNumber, DeploymentLocator, WritableStore};
+use crate::components::store::{BlockNumber, DeploymentLocator, ReadStore};
 use crate::data::subgraph::UnifiedMappingApiVersion;
 use crate::firehose::{self, FirehoseEndpoint};
 use crate::futures03::stream::StreamExt as _;
@@ -149,7 +149,7 @@ pub trait BlockStreamBuilder<C: Blockchain>: Send + Sync {
         chain: &C,
         deployment: DeploymentLocator,
         start_blocks: Vec<BlockNumber>,
-        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
+        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
         subgraph_current_block: Option<BlockPtr>,
         filter: Arc<TriggerFilterWrapper<C>>,
         unified_api_version: UnifiedMappingApiVersion,
@@ -160,7 +160,7 @@ pub trait BlockStreamBuilder<C: Blockchain>: Send + Sync {
         chain: &C,
         deployment: DeploymentLocator,
         start_blocks: Vec<BlockNumber>,
-        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
+        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
         subgraph_current_block: Option<BlockPtr>,
         filter: Arc<TriggerFilterWrapper<C>>,
         unified_api_version: UnifiedMappingApiVersion,
@@ -317,13 +317,13 @@ impl<C: Blockchain> BlockWithTriggers<C> {
 
 pub struct TriggersAdapterWrapper<C: Blockchain> {
     pub adapter: Arc<dyn TriggersAdapter<C>>,
-    pub source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
+    pub source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
 }
 
 impl<C: Blockchain> TriggersAdapterWrapper<C> {
     pub fn new(
         adapter: Arc<dyn TriggersAdapter<C>>,
-        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
+        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
     ) -> Self {
         Self {
             adapter,
diff --git a/graph/src/blockchain/mock.rs b/graph/src/blockchain/mock.rs
@@ -2,7 +2,7 @@ use crate::{
     bail,
     components::{
         link_resolver::LinkResolver,
-        store::{BlockNumber, DeploymentCursorTracker, DeploymentLocator, WritableStore},
+        store::{BlockNumber, DeploymentCursorTracker, DeploymentLocator, ReadStore},
         subgraph::InstanceDSTemplateInfo,
     },
     data::subgraph::UnifiedMappingApiVersion,
@@ -386,7 +386,7 @@ impl Blockchain for MockBlockchain {
         _deployment: DeploymentLocator,
         _store: impl DeploymentCursorTracker,
         _start_blocks: Vec<BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
         _filter: Arc<TriggerFilterWrapper<Self>>,
         _unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error> {
diff --git a/graph/src/blockchain/mod.rs b/graph/src/blockchain/mod.rs
@@ -20,9 +20,7 @@ use crate::{
     components::{
         adapter::ChainId,
         metrics::subgraph::SubgraphInstanceMetrics,
-        store::{
-            DeploymentCursorTracker, DeploymentLocator, StoredDynamicDataSource, WritableStore,
-        },
+        store::{DeploymentCursorTracker, DeploymentLocator, ReadStore, StoredDynamicDataSource},
         subgraph::{HostMetrics, InstanceDSTemplateInfo, MappingError},
         trigger_processor::RunnableTriggers,
     },
@@ -191,7 +189,7 @@ pub trait Blockchain: Debug + Sized + Send + Sync + Unpin + 'static {
         deployment: DeploymentLocator,
         store: impl DeploymentCursorTracker,
         start_blocks: Vec<BlockNumber>,
-        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn WritableStore>)>,
+        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
         filter: Arc<TriggerFilterWrapper<Self>>,
         unified_api_version: UnifiedMappingApiVersion,
     ) -> Result<Box<dyn BlockStream<Self>>, Error>;
diff --git a/graph/src/components/store/traits.rs b/graph/src/components/store/traits.rs
@@ -185,6 +185,13 @@ pub trait SubgraphStore: Send + Sync + 'static {
         manifest_idx_and_name: Arc<Vec<(u32, String)>>,
     ) -> Result<Arc<dyn WritableStore>, StoreError>;
 
+    async fn readable(
+        self: Arc<Self>,
+        logger: Logger,
+        deployment: DeploymentId,
+        manifest_idx_and_name: Arc<Vec<(u32, String)>>,
+    ) -> Result<Arc<dyn ReadStore>, StoreError>;
+
     /// Initiate a graceful shutdown of the writable that a previous call to
     /// `writable` might have started
     async fn stop_subgraph(&self, deployment: &DeploymentLocator) -> Result<(), StoreError>;
diff --git a/store/postgres/src/subgraph_store.rs b/store/postgres/src/subgraph_store.rs
diff --git a/tests/src/fixture/mod.rs b/tests/src/fixture/mod.rs
