replace ReadSTore with SourceableStore

Zoran Cvetkov · Zoran Cvetkov · commit 38c428656d48 · 2024-12-03T11:37:59.000+02: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, ReadStore};
+use graph::components::store::{DeploymentCursorTracker, SourceableStore};
 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 ReadStore>)>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
         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, ReadStore};
+use graph::components::store::{DeploymentCursorTracker, SourceableStore};
 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 ReadStore>)>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
         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, ReadStore};
+use graph::components::store::{DeploymentCursorTracker, SourceableStore};
 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 ReadStore>)>,
+        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
         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 ReadStore>)>,
+        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
         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 ReadStore>)>,
+        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
         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, ReadStore};
+use graph::components::store::{DeploymentCursorTracker, SourceableStore};
 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 ReadStore>)>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
         _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 ReadStore>)>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
         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, ReadStore},
+        store::{DeploymentCursorTracker, DeploymentLocator, SourceableStore},
     },
     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 ReadStore>)>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
         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 ReadStore>)>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
         _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, ReadStore},
+    components::store::{DeploymentLocator, SourceableStore},
     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 ReadStore>)>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
         _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, ReadStore};
+use graph::components::store::{DeploymentCursorTracker, SourceableStore};
 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 ReadStore>)>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
         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, ReadStore, SubgraphFork, WritableStore},
+        store::{DeploymentLocator, SourceableStore, 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 ReadStore>)>,
+    pub source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
     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::ReadStore;
+use graph::components::store::SourceableStore;
 use graph::components::subgraph::ProofOfIndexingVersion;
 use graph::data::subgraph::{UnresolvedSubgraphManifest, SPEC_VERSION_0_0_6};
 use graph::data::value::Word;
@@ -211,7 +211,7 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
         hashes: Vec<DeploymentHash>,
         max_spec_version: Version,
         is_runner_test: bool,
-    ) -> anyhow::Result<Vec<(DeploymentHash, Arc<dyn ReadStore>)>> {
+    ) -> anyhow::Result<Vec<(DeploymentHash, Arc<dyn SourceableStore>)>> {
         let mut writable_stores = Vec::new();
         let subgraph_store = self.subgraph_store.clone();
 
@@ -237,7 +237,7 @@ impl<S: SubgraphStore> SubgraphInstanceManager<S> {
 
             let readable_store = subgraph_store
                 .clone()
-                .readable(
+                .sourceable(
                     logger.clone(),
                     loc.id.clone(),
                     Arc::new(manifest.template_idx_and_name().collect()),
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, ReadStore};
+use crate::components::store::{BlockNumber, DeploymentLocator, SourceableStore};
 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 ReadStore>)>,
+        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
         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 ReadStore>)>,
+        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
         subgraph_current_block: Option<BlockPtr>,
         filter: Arc<TriggerFilterWrapper<C>>,
         unified_api_version: UnifiedMappingApiVersion,
@@ -320,13 +320,13 @@ impl<C: Blockchain> BlockWithTriggers<C> {
 /// logic for each chain, increasing code repetition.
 pub struct TriggersAdapterWrapper<C: Blockchain> {
     pub adapter: Arc<dyn TriggersAdapter<C>>,
-    pub source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
+    pub source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
 }
 
 impl<C: Blockchain> TriggersAdapterWrapper<C> {
     pub fn new(
         adapter: Arc<dyn TriggersAdapter<C>>,
-        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
+        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
     ) -> 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, ReadStore},
+        store::{BlockNumber, DeploymentCursorTracker, DeploymentLocator, SourceableStore},
         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 ReadStore>)>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
         _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,7 +20,9 @@ use crate::{
     components::{
         adapter::ChainId,
         metrics::subgraph::SubgraphInstanceMetrics,
-        store::{DeploymentCursorTracker, DeploymentLocator, ReadStore, StoredDynamicDataSource},
+        store::{
+            DeploymentCursorTracker, DeploymentLocator, SourceableStore, StoredDynamicDataSource,
+        },
         subgraph::{HostMetrics, InstanceDSTemplateInfo, MappingError},
         trigger_processor::RunnableTriggers,
     },
@@ -189,7 +191,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 ReadStore>)>,
+        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
         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
@@ -186,16 +186,11 @@ pub trait SubgraphStore: Send + Sync + 'static {
         manifest_idx_and_name: Arc<Vec<(u32, String)>>,
     ) -> Result<Arc<dyn WritableStore>, StoreError>;
 
-    async fn readable(
+    async fn sourceable(
         self: Arc<Self>,
         logger: Logger,
         deployment: DeploymentId,
         manifest_idx_and_name: Arc<Vec<(u32, String)>>,
-    ) -> Result<Arc<dyn ReadStore>, StoreError>;
-
-    async fn sourceable(
-        self: Arc<Self>,
-        deployment: DeploymentId,
     ) -> Result<Arc<dyn SourceableStore>, StoreError>;
 
     /// Initiate a graceful shutdown of the writable that a previous call to
diff --git a/store/postgres/src/subgraph_store.rs b/store/postgres/src/subgraph_store.rs
@@ -1537,25 +1537,20 @@ impl SubgraphStoreTrait for SubgraphStore {
             .map(|store| store as Arc<dyn store::WritableStore>)
     }
 
-    async fn readable(
+    async fn sourceable(
         self: Arc<Self>,
         logger: Logger,
         deployment: graph::components::store::DeploymentId,
         manifest_idx_and_name: Arc<Vec<(u32, String)>>,
-    ) -> Result<Arc<dyn store::ReadStore>, StoreError> {
-        self.get_or_create_writable_store(logger, deployment, manifest_idx_and_name)
-            .await
-            .map(|store| store as Arc<dyn store::ReadStore>)
-    }
-
-    async fn sourceable(
-        self: Arc<Self>,
-        deployment: graph::components::store::DeploymentId,
     ) -> Result<Arc<dyn store::SourceableStore>, StoreError> {
+        let writable = self
+            .clone()
+            .writable(logger, deployment, manifest_idx_and_name)
+            .await?;
         let deployment = deployment.into();
         let site = self.find_site(deployment)?;
         let store = self.for_site(&site)?;
-        let s = Arc::new(SourceableStore::new(site, store.clone()));
+        let s = Arc::new(SourceableStore::new(site, store.clone(), writable));
         Ok(s as Arc<dyn store::SourceableStore>)
     }
 
diff --git a/store/postgres/src/writable.rs b/store/postgres/src/writable.rs
@@ -1572,11 +1572,20 @@ impl ReadStore for WritableStore {
 pub struct SourceableStore {
     site: Arc<Site>,
     store: Arc<DeploymentStore>,
+    _writable: Arc<dyn store::WritableStore>,
 }
 
 impl SourceableStore {
-    pub fn new(site: Arc<Site>, store: Arc<DeploymentStore>) -> Self {
-        Self { site, store }
+    pub fn new(
+        site: Arc<Site>,
+        store: Arc<DeploymentStore>,
+        _writable: Arc<dyn store::WritableStore>,
+    ) -> Self {
+        Self {
+            site,
+            store,
+            _writable,
+        }
     }
 }
 
diff --git a/store/test-store/tests/postgres/writable.rs b/store/test-store/tests/postgres/writable.rs
@@ -113,7 +113,7 @@ where
             .expect("we can get a writable store");
         let sourceable = store
             .subgraph_store()
-            .sourceable(deployment.id)
+            .sourceable(LOGGER.clone(), deployment.id, Arc::new(Vec::new()))
             .await
             .expect("we can get a writable store");
 
diff --git a/tests/src/fixture/mod.rs b/tests/src/fixture/mod.rs
@@ -20,7 +20,7 @@ use graph::cheap_clone::CheapClone;
 use graph::components::adapter::ChainId;
 use graph::components::link_resolver::{ArweaveClient, ArweaveResolver, FileSizeLimit};
 use graph::components::metrics::MetricsRegistry;
-use graph::components::store::{BlockStore, DeploymentLocator, EthereumCallCache, ReadStore};
+use graph::components::store::{BlockStore, DeploymentLocator, EthereumCallCache, SourceableStore};
 use graph::components::subgraph::Settings;
 use graph::data::graphql::load_manager::LoadManager;
 use graph::data::query::{Query, QueryTarget};
@@ -723,7 +723,7 @@ impl<C: Blockchain> BlockStreamBuilder<C> for MutexBlockStreamBuilder<C> {
         chain: &C,
         deployment: DeploymentLocator,
         start_blocks: Vec<BlockNumber>,
-        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
+        source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
         subgraph_current_block: Option<BlockPtr>,
         filter: Arc<TriggerFilterWrapper<C>>,
         unified_api_version: graph::data::subgraph::UnifiedMappingApiVersion,
@@ -798,7 +798,7 @@ where
         _chain: &C,
         _deployment: DeploymentLocator,
         _start_blocks: Vec<graph::prelude::BlockNumber>,
-        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn ReadStore>)>,
+        _source_subgraph_stores: Vec<(DeploymentHash, Arc<dyn SourceableStore>)>,
         subgraph_current_block: Option<graph::blockchain::BlockPtr>,
         _filter: Arc<TriggerFilterWrapper<C>>,
         _unified_api_version: graph::data::subgraph::UnifiedMappingApiVersion,
