graph, store: Load dynamic data sources directly from the store

Load the parts of dynamic data sources that can't be reconstructed from the
manifest we have in memory

Author: lutter

graph/src/components/store.rs

@@ -15,9 +15,9 @@ use std::sync::{Arc, RwLock};
 use std::time::{Duration, Instant};
 use web3::types::{Address, H256};
 
-use crate::data::store::*;
 use crate::data::subgraph::schema::*;
 use crate::data::subgraph::status;
+use crate::data::{store::*, subgraph::Source};
 use crate::prelude::*;
 use crate::util::lfu_cache::LfuCache;
 
@@ -796,6 +796,12 @@ pub enum TransactionAbortError {
     Other(String),
 }
 
+pub struct StoredDynamicDataSource {
+    pub name: String,
+    pub source: Source,
+    pub context: Option<String>,
+}
+
 /// Common trait for store implementations.
 #[async_trait]
 pub trait Store: Send + Sync + 'static {
@@ -1005,6 +1011,12 @@ pub trait Store: Send + Sync + 'static {
     fn query_store(self: Arc<Self>, for_subscription: bool) -> Arc<dyn QueryStore + Send + Sync>;
 
     fn status(&self, filter: status::Filter) -> Result<Vec<status::Info>, StoreError>;
+
+    /// Load the dynamic data sources for the given deployment
+    fn load_dynamic_data_sources(
+        &self,
+        subgraph_id: &SubgraphDeploymentId,
+    ) -> Result<Vec<StoredDynamicDataSource>, StoreError>;
 }
 
 mock! {
@@ -1181,6 +1193,13 @@ impl Store for MockStore {
     fn status(&self, _: status::Filter) -> Result<Vec<status::Info>, StoreError> {
         unimplemented!()
     }
+
+    fn load_dynamic_data_sources(
+        &self,
+        _subgraph_id: &SubgraphDeploymentId,
+    ) -> Result<Vec<StoredDynamicDataSource>, StoreError> {
+        unimplemented!()
+    }
 }
 
 #[automock]

mock/src/store.rs

@@ -2,9 +2,9 @@ use mockall::predicate::*;
 use mockall::*;
 use std::collections::BTreeMap;
 
-use graph::data::subgraph::schema::*;
 use graph::data::subgraph::status;
 use graph::prelude::*;
+use graph::{components::store::StoredDynamicDataSource, data::subgraph::schema::SubgraphError};
 use graph_graphql::prelude::api_schema;
 use web3::types::{Address, H256};
 
@@ -218,6 +218,13 @@ impl Store for MockStore {
     fn status(&self, _: status::Filter) -> Result<Vec<status::Info>, StoreError> {
         unimplemented!()
     }
+
+    fn load_dynamic_data_sources(
+        &self,
+        _: &SubgraphDeploymentId,
+    ) -> Result<Vec<StoredDynamicDataSource>, StoreError> {
+        unimplemented!()
+    }
 }
 
 pub fn mock_store_with_users_subgraph() -> (Arc<MockStore>, SubgraphDeploymentId) {

store/postgres/src/dynds.rs

@@ -0,0 +1,131 @@
+//! SQL queries to load dynamic data sources
+
+use diesel::pg::PgConnection;
+use diesel::prelude::{ExpressionMethods, JoinOnDsl, QueryDsl, RunQueryDsl};
+
+use graph::{
+    components::store::StoredDynamicDataSource,
+    data::subgraph::Source,
+    prelude::{bigdecimal::ToPrimitive, web3::types::H160, BigDecimal, StoreError},
+};
+
+// Diesel tables for some of the metadata
+// See also: ed42d219c6704a4aab57ce1ea66698e7
+// Changes to the GraphQL schema might require changes to these tables.
+// The definitions of the tables can be generated with
+//    cargo run -p graph-store-postgres --example layout -- \
+//      -g diesel store/postgres/src/subgraphs.graphql subgraphs
+// BEGIN GENERATED CODE
+table! {
+    subgraphs.dynamic_ethereum_contract_data_source (vid) {
+        vid -> BigInt,
+        id -> Text,
+        kind -> Text,
+        name -> Text,
+        network -> Nullable<Text>,
+        source -> Text,
+        mapping -> Text,
+        templates -> Nullable<Array<Text>>,
+        ethereum_block_hash -> Binary,
+        ethereum_block_number -> Numeric,
+        deployment -> Text,
+        context -> Nullable<Text>,
+        block_range -> Range<Integer>,
+    }
+}
+
+table! {
+    subgraphs.ethereum_contract_source (vid) {
+        vid -> BigInt,
+        id -> Text,
+        address -> Nullable<Binary>,
+        abi -> Text,
+        start_block -> Nullable<Numeric>,
+        block_range -> Range<Integer>,
+    }
+}
+
+// END GENERATED CODE
+
+allow_tables_to_appear_in_same_query!(
+    dynamic_ethereum_contract_data_source,
+    ethereum_contract_source
+);
+
+fn to_source(
+    deployment: &str,
+    ds_id: &str,
+    (address, abi, start_block): (Option<Vec<u8>>, String, Option<BigDecimal>),
+) -> Result<Source, StoreError> {
+    // Treat a missing address as an error. TODO: Is that correct?
+    let address = match address {
+        Some(address) => address,
+        None => {
+            return Err(StoreError::ConstraintViolation(format!(
+                "Dynamic data source {} for deployment {} is missing an address",
+                ds_id, deployment
+            )));
+        }
+    };
+    if address.len() != 20 {
+        return Err(StoreError::ConstraintViolation(format!(
+            "Data source address 0x`{:?}` for dynamic data source {} in deployment {} should have be 20 bytes long but is {} bytes long",
+            address, ds_id, deployment,
+            address.len()
+        )));
+    }
+    let address = Some(H160::from_slice(address.as_slice()));
+
+    // Assume a missing start block is the same as 0
+    let start_block = start_block
+        .map(|s| {
+            s.to_u64().ok_or_else(|| {
+                StoreError::ConstraintViolation(format!(
+                    "Start block {:?} for dynamic data source {} in deployment {} is not a u64",
+                    s, ds_id, deployment
+                ))
+            })
+        })
+        .transpose()?
+        .unwrap_or(0);
+
+    Ok(Source {
+        address,
+        abi,
+        start_block,
+    })
+}
+
+#[allow(dead_code)]
+pub fn load(conn: &PgConnection, id: &str) -> Result<Vec<StoredDynamicDataSource>, StoreError> {
+    use dynamic_ethereum_contract_data_source as decds;
+    use ethereum_contract_source as ecs;
+
+    let dds: Vec<_> = decds::table
+        .inner_join(ecs::table.on(decds::source.eq(ecs::id)))
+        .filter(decds::deployment.eq(id))
+        .select((
+            decds::id,
+            decds::name,
+            decds::context,
+            (ecs::address, ecs::abi, ecs::start_block),
+        ))
+        .load::<(
+            String,
+            String,
+            Option<String>,
+            (Option<Vec<u8>>, String, Option<BigDecimal>),
+        )>(conn)?;
+
+    let mut data_sources = Vec::new();
+    for (ds_id, name, context, source) in dds.into_iter() {
+        let source = to_source(id, &ds_id, source)?;
+        let data_source = StoredDynamicDataSource {
+            name,
+            source,
+            context,
+        };
+        data_sources.push(data_source);
+    }
+    Ok(data_sources)
+}

store/postgres/src/entities.rs

@@ -26,7 +26,6 @@ use diesel::sql_types::{Integer, Text};
 use diesel::Connection as _;
 use diesel::ExpressionMethods;
 use diesel::{OptionalExtension, QueryDsl, RunQueryDsl};
-use lazy_static::lazy_static;
 use maybe_owned::MaybeOwned;
 use std::collections::hash_map::DefaultHasher;
 use std::collections::{BTreeMap, HashMap};
@@ -50,7 +49,7 @@ use crate::notification_listener::JsonNotification;
 use crate::relational::{Catalog, Layout};
 
 #[cfg(debug_assertions)]
-lazy_static! {
+lazy_static::lazy_static! {
 /// Tests set this to true so that `send_store_event` will store a copy
 /// of each event sent in `EVENT_TAP`
 pub static ref EVENT_TAP_ENABLED: Mutex<bool> = Mutex::new(false);

store/postgres/src/lib.rs

@@ -27,6 +27,7 @@ mod chain_store;
 pub mod connection_pool;
 mod db_schema;
 mod detail;
+mod dynds;
 mod entities;
 mod functions;
 mod jsonb;

store/postgres/src/network_store.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use graph::{
+    components::store::StoredDynamicDataSource,
     data::subgraph::schema::SubgraphError,
     data::subgraph::status,
     prelude::{
@@ -250,6 +251,13 @@ impl StoreTrait for NetworkStore {
     fn status(&self, filter: status::Filter) -> Result<Vec<status::Info>, StoreError> {
         self.store.status(filter)
     }
+
+    fn load_dynamic_data_sources(
+        &self,
+        subgraph_id: &SubgraphDeploymentId,
+    ) -> Result<Vec<StoredDynamicDataSource>, StoreError> {
+        self.store.load_dynamic_data_sources(subgraph_id)
+    }
 }
 
 impl SubgraphDeploymentStore for NetworkStore {

store/postgres/src/store.rs

@@ -4,9 +4,12 @@ use diesel::prelude::*;
 use diesel::r2d2::{ConnectionManager, PooledConnection};
 use diesel::{insert_into, update};
 use futures03::FutureExt as _;
-use graph::prelude::{
-    CancelGuard, CancelHandle, CancelToken, CancelableError, NodeId, PoolWaitStats,
-    SubgraphVersionSwitchingMode,
+use graph::{
+    components::store::StoredDynamicDataSource,
+    prelude::{
+        CancelGuard, CancelHandle, CancelToken, CancelableError, NodeId, PoolWaitStats,
+        SubgraphVersionSwitchingMode,
+    },
 };
 use graph::{data::subgraph::status, prelude::TryFutureExt};
 use lazy_static::lazy_static;
@@ -1370,6 +1373,14 @@ impl StoreTrait for Store {
     fn status(&self, filter: status::Filter) -> Result<Vec<status::Info>, StoreError> {
         self.status_internal(filter)
     }
+
+    fn load_dynamic_data_sources(
+        &self,
+        id: &SubgraphDeploymentId,
+    ) -> Result<Vec<StoredDynamicDataSource>, StoreError> {
+        let econn = self.get_entity_conn(&*SUBGRAPHS_ID, ReplicaId::Main)?;
+        econn.transaction(|| crate::dynds::load(&econn.conn, id.as_str()))
+    }
 }
 
 impl SubgraphDeploymentStore for Store {

store/test-store/src/lib.rs

@@ -17,7 +17,6 @@ use graph_store_postgres::{
 };
 use hex_literal::hex;
 use lazy_static::lazy_static;
-use std::collections::BTreeSet;
 use std::env;
 use std::sync::Mutex;
 use std::time::Instant;
@@ -154,7 +153,7 @@ fn create_subgraph(
         id: subgraph_id.clone(),
         location: String::new(),
         spec_version: "1".to_owned(),
-        features: BTreeSet::new(),
+        features: std::collections::BTreeSet::new(),
         description: None,
         repository: None,
         schema: schema.clone(),