diff --git a/Cargo.lock b/Cargo.lock index 8dfa23e00a6..ce9fa057760 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -90,6 +90,16 @@ version = "0.9.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eab1c04a571841102f5345a8fc0f6bb3d31c315dec879b5c6e42e40ce7ffa34e" +[[package]] +name = "assert-json-diff" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e4f2b81832e72834d7518d8487a0396a28cc408186a2e8854c0f98011faf12" +dependencies = [ + "serde", + "serde_json", +] + [[package]] name = "async-recursion" version = "1.0.0" @@ -1944,6 +1954,7 @@ dependencies = [ "pin-utils", "postgres", "postgres-openssl", + "pretty_assertions", "rand", "serde", "stable-hash 0.3.3", @@ -1956,6 +1967,7 @@ name = "graph-tests" version = "0.29.0" dependencies = [ "anyhow", + "assert-json-diff", "async-stream", "bollard", "cid", diff --git a/chain/substreams/src/trigger.rs b/chain/substreams/src/trigger.rs index d18b5bbe6b4..237f9e09793 100644 --- a/chain/substreams/src/trigger.rs +++ b/chain/substreams/src/trigger.rs @@ -4,11 +4,11 @@ use anyhow::Error; use graph::{ blockchain::{self, block_stream::BlockWithTriggers, BlockPtr, EmptyNodeCapabilities}, components::{ - store::{DeploymentLocator, EntityKey, SubgraphFork}, + store::{DeploymentLocator, EntityKey, EntityType, SubgraphFork}, subgraph::{MappingError, ProofOfIndexingEvent, SharedProofOfIndexing}, }, data::store::scalar::Bytes, - data_source, + data_source::{self, CausalityRegion}, prelude::{ anyhow, async_trait, BigDecimal, BigInt, BlockHash, BlockNumber, BlockState, Entity, RuntimeHostBuilder, Value, @@ -183,7 +183,11 @@ where Operation::Create | Operation::Update => { let entity_type: &str = &entity_change.entity; let entity_id: String = entity_change.id.clone(); - let key = EntityKey::data(entity_type.to_string(), entity_id.clone()); + let key = EntityKey { + entity_type: EntityType::new(entity_type.to_string()), + entity_id: entity_id.clone().into(), + causality_region: CausalityRegion::ONCHAIN, // Substreams don't currently support offchain data + }; let mut data: HashMap = HashMap::from_iter(vec![]); for field in entity_change.fields.iter() { @@ -214,7 +218,11 @@ where Operation::Delete => { let entity_type: &str = &entity_change.entity; let entity_id: String = entity_change.id.clone(); - let key = EntityKey::data(entity_type.to_string(), entity_id.clone()); + let key = EntityKey { + entity_type: EntityType::new(entity_type.to_string()), + entity_id: entity_id.clone().into(), + causality_region: CausalityRegion::ONCHAIN, // Substreams don't currently support offchain data + }; state.entity_cache.remove(key); diff --git a/core/src/subgraph/provider.rs b/core/src/subgraph/provider.rs index 11b325af3f9..4d3a0cab51d 100644 --- a/core/src/subgraph/provider.rs +++ b/core/src/subgraph/provider.rs @@ -82,9 +82,7 @@ impl SubgraphAssignmentProviderTrait for SubgraphAss { // Shut down subgraph processing self.instance_manager.stop_subgraph(deployment).await; - Ok(()) - } else { - Err(SubgraphAssignmentProviderError::NotRunning(deployment)) } + Ok(()) } } diff --git a/core/src/subgraph/registrar.rs b/core/src/subgraph/registrar.rs index 8e3bddeff26..3ca52d98bef 100644 --- a/core/src/subgraph/registrar.rs +++ b/core/src/subgraph/registrar.rs @@ -88,8 +88,7 @@ where // - The event stream sees a Remove event for subgraph B, but the table query finds that // subgraph B has already been removed. // The `handle_assignment_events` function handles these cases by ignoring AlreadyRunning - // (on subgraph start) or NotRunning (on subgraph stop) error types, which makes the - // operations idempotent. + // (on subgraph start) which makes the operations idempotent. Subgraph stop is already idempotent. // Start event stream let assignment_event_stream = self.assignment_events(); @@ -455,7 +454,6 @@ async fn handle_assignment_event( node_id: _, } => match provider.stop(deployment).await { Ok(()) => Ok(()), - Err(SubgraphAssignmentProviderError::NotRunning(_)) => Ok(()), Err(e) => Err(CancelableError::Error(e)), }, } @@ -620,11 +618,30 @@ async fn create_subgraph_version( "block" => format!("{:?}", base_block.as_ref().map(|(_,ptr)| ptr.number)) ); + // Entity types that may be touched by offchain data sources need a causality region column. + let needs_causality_region = manifest + .data_sources + .iter() + .filter_map(|ds| ds.as_offchain()) + .map(|ds| ds.mapping.entities.iter()) + .chain( + manifest + .templates + .iter() + .filter_map(|ds| ds.as_offchain()) + .map(|ds| ds.mapping.entities.iter()), + ) + .flatten() + .cloned() + .collect(); + // Apply the subgraph versioning and deployment operations, // creating a new subgraph deployment if one doesn't exist. let deployment = DeploymentCreate::new(raw_string, &manifest, start_block) .graft(base_block) - .debug(debug_fork); + .debug(debug_fork) + .entities_with_causality_region(needs_causality_region); + deployment_store .create_subgraph_deployment( name, diff --git a/core/src/subgraph/runner.rs b/core/src/subgraph/runner.rs index af5a338dcb2..515fa53c066 100644 --- a/core/src/subgraph/runner.rs +++ b/core/src/subgraph/runner.rs @@ -17,7 +17,7 @@ use graph::data::subgraph::{ SubgraphFeature, }; use graph::data_source::{ - offchain, DataSource, DataSourceCreationError, DataSourceTemplate, TriggerData, + offchain, CausalityRegion, DataSource, DataSourceCreationError, DataSourceTemplate, TriggerData, }; use graph::env::EnvVars; use graph::prelude::*; @@ -653,6 +653,7 @@ where let mut block_state = BlockState::::new(EmptyStore::new(schema), LfuCache::new()); // PoI ignores offchain events. + // See also: poi-ignores-offchain let proof_of_indexing = None; let causality_region = ""; @@ -998,7 +999,14 @@ async fn update_proof_of_indexing( // Create the special POI entity key specific to this causality_region let entity_key = EntityKey { entity_type: POI_OBJECT.to_owned(), + + // There are two things called causality regions here, one is the causality region for + // the poi which is a string and the PoI entity id. The other is the data source + // causality region to which the PoI belongs as an entity. Currently offchain events do + // not affect PoI so it is assumed to be `ONCHAIN`. + // See also: poi-ignores-offchain entity_id: causality_region.into(), + causality_region: CausalityRegion::ONCHAIN, }; // Grab the current digest attribute on this entity diff --git a/graph/src/components/store/entity_cache.rs b/graph/src/components/store/entity_cache.rs index 252bad33073..50fc357a890 100644 --- a/graph/src/components/store/entity_cache.rs +++ b/graph/src/components/store/entity_cache.rs @@ -1,11 +1,9 @@ use anyhow::anyhow; -use std::collections::{BTreeMap, HashMap}; +use std::collections::HashMap; use std::fmt::{self, Debug}; use std::sync::Arc; -use crate::components::store::{ - self as s, Entity, EntityKey, EntityOp, EntityOperation, EntityType, -}; +use crate::components::store::{self as s, Entity, EntityKey, EntityOp, EntityOperation}; use crate::prelude::{Schema, ENV_VARS}; use crate::util::lfu_cache::LfuCache; @@ -102,6 +100,10 @@ impl EntityCache { // Get the current entity, apply any updates from `updates`, then // from `handler_updates`. let mut entity = self.current.get_entity(&*self.store, eref)?; + + // Always test the cache consistency in debug mode. + debug_assert!(entity == self.store.get(&eref).unwrap()); + if let Some(op) = self.updates.get(eref).cloned() { entity = op.apply_to(entity) } @@ -233,22 +235,8 @@ impl EntityCache { // violation in the database, ensuring correctness let missing = missing.filter(|key| !self.schema.is_immutable(&key.entity_type)); - let mut missing_by_type: BTreeMap<&EntityType, Vec<&str>> = BTreeMap::new(); - for key in missing { - missing_by_type - .entry(&key.entity_type) - .or_default() - .push(&key.entity_id); - } - - for (entity_type, entities) in self.store.get_many(missing_by_type)? { - for entity in entities { - let key = EntityKey { - entity_type: entity_type.clone(), - entity_id: entity.id().unwrap().into(), - }; - self.current.insert(key, Some(entity)); - } + for (entity_key, entity) in self.store.get_many(missing.cloned().collect())? { + self.current.insert(entity_key, Some(entity)); } let mut mods = Vec::new(); diff --git a/graph/src/components/store/mod.rs b/graph/src/components/store/mod.rs index c20b58cc295..d03c0c26e91 100644 --- a/graph/src/components/store/mod.rs +++ b/graph/src/components/store/mod.rs @@ -4,6 +4,7 @@ mod traits; pub use entity_cache::{EntityCache, ModificationsAndCache}; +use diesel::types::{FromSql, ToSql}; pub use err::StoreError; use itertools::Itertools; pub use traits::*; @@ -12,13 +13,14 @@ use futures::stream::poll_fn; use futures::{Async, Poll, Stream}; use graphql_parser::schema as s; use serde::{Deserialize, Serialize}; +use std::borrow::Borrow; use std::collections::btree_map::Entry; use std::collections::{BTreeMap, BTreeSet, HashSet}; -use std::fmt; use std::fmt::Display; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, RwLock}; use std::time::Duration; +use std::{fmt, io}; use crate::blockchain::Block; use crate::data::store::scalar::Bytes; @@ -71,6 +73,12 @@ impl<'a> From<&s::InterfaceType<'a, String>> for EntityType { } } +impl Borrow for EntityType { + fn borrow(&self) -> &str { + &self.0 + } +} + // This conversion should only be used in tests since it makes it too // easy to convert random strings into entity types #[cfg(debug_assertions)] @@ -82,6 +90,22 @@ impl From<&str> for EntityType { impl CheapClone for EntityType {} +impl FromSql for EntityType { + fn from_sql(bytes: Option<&[u8]>) -> diesel::deserialize::Result { + let s = >::from_sql(bytes)?; + Ok(EntityType::new(s)) + } +} + +impl ToSql for EntityType { + fn to_sql( + &self, + out: &mut diesel::serialize::Output, + ) -> diesel::serialize::Result { + >::to_sql(self.0.as_str(), out) + } +} + #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord, Hash)] pub struct EntityFilterDerivative(bool); @@ -104,13 +128,23 @@ pub struct EntityKey { /// ID of the individual entity. pub entity_id: Word, + + /// This is the causality region of the data source that created the entity. + /// + /// In the case of an entity lookup, this is the causality region of the data source that is + /// doing the lookup. So if the entity exists but was created on a different causality region, + /// the lookup will return empty. + pub causality_region: CausalityRegion, } impl EntityKey { - pub fn data(entity_type: String, entity_id: String) -> Self { + // For use in tests only + #[cfg(debug_assertions)] + pub fn data(entity_type: impl Into, entity_id: impl Into) -> Self { Self { - entity_type: EntityType::new(entity_type), - entity_id: entity_id.into(), + entity_type: EntityType::new(entity_type.into()), + entity_id: entity_id.into().into(), + causality_region: CausalityRegion::ONCHAIN, } } } @@ -1071,10 +1105,7 @@ impl ReadStore for EmptyStore { Ok(None) } - fn get_many( - &self, - _ids_for_type: BTreeMap<&EntityType, Vec<&str>>, - ) -> Result>, StoreError> { + fn get_many(&self, _: BTreeSet) -> Result, StoreError> { Ok(BTreeMap::new()) } diff --git a/graph/src/components/store/traits.rs b/graph/src/components/store/traits.rs index 24d3918907e..6a6ef2bc79d 100644 --- a/graph/src/components/store/traits.rs +++ b/graph/src/components/store/traits.rs @@ -171,12 +171,11 @@ pub trait ReadStore: Send + Sync + 'static { /// Looks up an entity using the given store key at the latest block. fn get(&self, key: &EntityKey) -> Result, StoreError>; - /// Look up multiple entities as of the latest block. Returns a map of - /// entities by type. + /// Look up multiple entities as of the latest block. fn get_many( &self, - ids_for_type: BTreeMap<&EntityType, Vec<&str>>, - ) -> Result>, StoreError>; + keys: BTreeSet, + ) -> Result, StoreError>; fn input_schema(&self) -> Arc; } @@ -189,9 +188,9 @@ impl ReadStore for Arc { fn get_many( &self, - ids_for_type: BTreeMap<&EntityType, Vec<&str>>, - ) -> Result>, StoreError> { - (**self).get_many(ids_for_type) + keys: BTreeSet, + ) -> Result, StoreError> { + (**self).get_many(keys) } fn input_schema(&self) -> Arc { diff --git a/graph/src/data/graphql/object_or_interface.rs b/graph/src/data/graphql/object_or_interface.rs index 48053555e76..7764769a1a0 100644 --- a/graph/src/data/graphql/object_or_interface.rs +++ b/graph/src/data/graphql/object_or_interface.rs @@ -117,7 +117,7 @@ impl<'a> ObjectOrInterface<'a> { ObjectOrInterface::Object(object) => Some(vec![object]), ObjectOrInterface::Interface(interface) => schema .types_for_interface() - .get(&interface.into()) + .get(interface.name.as_str()) .map(|object_types| object_types.iter().collect()), } } @@ -131,7 +131,7 @@ impl<'a> ObjectOrInterface<'a> { ) -> bool { match self { ObjectOrInterface::Object(o) => o.name == typename, - ObjectOrInterface::Interface(i) => types_for_interface[&i.into()] + ObjectOrInterface::Interface(i) => types_for_interface[i.name.as_str()] .iter() .any(|o| o.name == typename), } diff --git a/graph/src/data/subgraph/mod.rs b/graph/src/data/subgraph/mod.rs index e21423c18b6..114c2f66936 100644 --- a/graph/src/data/subgraph/mod.rs +++ b/graph/src/data/subgraph/mod.rs @@ -28,7 +28,7 @@ use crate::{ blockchain::{BlockPtr, Blockchain, DataSource as _}, components::{ link_resolver::LinkResolver, - store::{DeploymentLocator, StoreError, SubgraphStore}, + store::{StoreError, SubgraphStore}, }, data::{ graphql::TryFromValue, @@ -304,8 +304,6 @@ pub enum SubgraphAssignmentProviderError { /// Occurs when attempting to remove a subgraph that's not hosted. #[error("Subgraph with ID {0} already running")] AlreadyRunning(DeploymentHash), - #[error("Subgraph with ID {0} is not running")] - NotRunning(DeploymentLocator), #[error("Subgraph provider error: {0}")] Unknown(#[from] anyhow::Error), } diff --git a/graph/src/data/subgraph/schema.rs b/graph/src/data/subgraph/schema.rs index eda88f9f522..11f4ed6cfea 100644 --- a/graph/src/data/subgraph/schema.rs +++ b/graph/src/data/subgraph/schema.rs @@ -5,6 +5,7 @@ use hex; use lazy_static::lazy_static; use rand::rngs::OsRng; use rand::Rng; +use std::collections::BTreeSet; use std::str::FromStr; use std::{fmt, fmt::Display}; @@ -116,7 +117,7 @@ impl DeploymentCreate { start_block: Option, ) -> Self { Self { - manifest: SubgraphManifestEntity::new(raw_manifest, source_manifest), + manifest: SubgraphManifestEntity::new(raw_manifest, source_manifest, Vec::new()), start_block: start_block.cheap_clone(), graft_base: None, graft_block: None, @@ -136,6 +137,15 @@ impl DeploymentCreate { self.debug_fork = fork; self } + + pub fn entities_with_causality_region( + mut self, + entities_with_causality_region: BTreeSet, + ) -> Self { + self.manifest.entities_with_causality_region = + entities_with_causality_region.into_iter().collect(); + self + } } /// The representation of a subgraph deployment when reading an existing @@ -169,10 +179,15 @@ pub struct SubgraphManifestEntity { pub features: Vec, pub schema: String, pub raw_yaml: Option, + pub entities_with_causality_region: Vec, } impl SubgraphManifestEntity { - pub fn new(raw_yaml: String, manifest: &super::SubgraphManifest) -> Self { + pub fn new( + raw_yaml: String, + manifest: &super::SubgraphManifest, + entities_with_causality_region: Vec, + ) -> Self { Self { spec_version: manifest.spec_version.to_string(), description: manifest.description.clone(), @@ -180,6 +195,7 @@ impl SubgraphManifestEntity { features: manifest.features.iter().map(|f| f.to_string()).collect(), schema: manifest.schema.document.clone().to_string(), raw_yaml: Some(raw_yaml), + entities_with_causality_region, } } diff --git a/graph/src/data/value.rs b/graph/src/data/value.rs index f3b94a2c9fa..ddb2d0a7134 100644 --- a/graph/src/data/value.rs +++ b/graph/src/data/value.rs @@ -43,6 +43,12 @@ impl From for Word { } } +impl From for String { + fn from(w: Word) -> Self { + w.0.into() + } +} + impl Serialize for Word { fn serialize(&self, serializer: S) -> Result where diff --git a/graph/src/data_source/causality_region.rs b/graph/src/data_source/causality_region.rs index 2c297a5c930..808538b6027 100644 --- a/graph/src/data_source/causality_region.rs +++ b/graph/src/data_source/causality_region.rs @@ -8,6 +8,8 @@ use diesel::{ use std::fmt; use std::io; +use crate::components::subgraph::Entity; + /// The causality region of a data source. All onchain data sources share the same causality region, /// but each offchain data source is assigned its own. This isolates offchain data sources from /// onchain and from each other. @@ -19,7 +21,7 @@ use std::io; /// This necessary for determinism because offchain data sources don't have a deterministic order of /// execution, for example an IPFS file may become available at any point in time. The isolation /// rules make the indexing result reproducible, given a set of available files. -#[derive(Debug, Copy, Clone, PartialEq, Eq, FromSqlRow)] +#[derive(Debug, Copy, Clone, PartialEq, Eq, FromSqlRow, Hash, PartialOrd, Ord)] pub struct CausalityRegion(i32); impl fmt::Display for CausalityRegion { @@ -47,6 +49,14 @@ impl CausalityRegion { pub const fn next(self) -> Self { CausalityRegion(self.0 + 1) } + + pub fn from_entity(entity: &Entity) -> Self { + entity + .get("causality_region") + .and_then(|v| v.as_int()) + .map(CausalityRegion) + .unwrap_or(CausalityRegion::ONCHAIN) + } } /// A subgraph will assign causality regions to offchain data sources from a sequence. diff --git a/graph/src/data_source/mod.rs b/graph/src/data_source/mod.rs index a2b5d5b97c2..aba6498682b 100644 --- a/graph/src/data_source/mod.rs +++ b/graph/src/data_source/mod.rs @@ -206,6 +206,13 @@ impl DataSource { Self::Offchain(_) => vec![], } } + + pub fn causality_region(&self) -> CausalityRegion { + match self { + Self::Onchain(_) => CausalityRegion::ONCHAIN, + Self::Offchain(ds) => ds.causality_region, + } + } } #[derive(Debug)] @@ -250,6 +257,13 @@ impl DataSourceTemplate { } } + pub fn as_offchain(&self) -> Option<&offchain::DataSourceTemplate> { + match self { + Self::Onchain(_) => None, + Self::Offchain(t) => Some(&t), + } + } + pub fn into_onchain(self) -> Option { match self { Self::Onchain(ds) => Some(ds), diff --git a/graph/src/data_source/tests.rs b/graph/src/data_source/tests.rs index 0312fd3dbd6..30421fca84f 100644 --- a/graph/src/data_source/tests.rs +++ b/graph/src/data_source/tests.rs @@ -1,6 +1,11 @@ use cid::Cid; -use crate::{components::subgraph::Entity, ipfs_client::CidFile, prelude::Link}; +use crate::{ + blockchain::mock::{MockBlockchain, MockDataSource}, + components::subgraph::Entity, + ipfs_client::CidFile, + prelude::Link, +}; use super::{ offchain::{Mapping, Source}, @@ -45,6 +50,21 @@ fn offchain_mark_processed_error() { x.mark_processed_at(-1) } +#[test] +fn data_source_helpers() { + let offchain = new_datasource(); + let offchain_ds = DataSource::::Offchain(offchain.clone()); + assert!(offchain_ds.causality_region() == offchain.causality_region); + assert!(offchain_ds + .as_offchain() + .unwrap() + .is_duplicate_of(&offchain)); + + let onchain = DataSource::::Onchain(MockDataSource); + assert!(onchain.causality_region() == CausalityRegion::ONCHAIN); + assert!(onchain.as_offchain().is_none()); +} + fn new_datasource() -> offchain::DataSource { offchain::DataSource::new( "theKind".into(), diff --git a/graph/tests/entity_cache.rs b/graph/tests/entity_cache.rs index 5cff8d0a251..ffe2388d058 100644 --- a/graph/tests/entity_cache.rs +++ b/graph/tests/entity_cache.rs @@ -6,7 +6,7 @@ use graph::data_source::CausalityRegion; use graph::prelude::{Schema, StopwatchMetrics, StoreError, UnfailOutcome}; use lazy_static::lazy_static; use slog::Logger; -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; use graph::components::store::{ @@ -14,7 +14,7 @@ use graph::components::store::{ }; use graph::{ components::store::{DeploymentId, DeploymentLocator}, - prelude::{anyhow, DeploymentHash, Entity, EntityCache, EntityModification, Value}, + prelude::{DeploymentHash, Entity, EntityCache, EntityModification, Value}, }; lazy_static! { @@ -38,33 +38,24 @@ lazy_static! { } struct MockStore { - get_many_res: BTreeMap>, + get_many_res: BTreeMap, } impl MockStore { - fn new(get_many_res: BTreeMap>) -> Self { + fn new(get_many_res: BTreeMap) -> Self { Self { get_many_res } } } impl ReadStore for MockStore { fn get(&self, key: &EntityKey) -> Result, StoreError> { - match self.get_many_res.get(&key.entity_type) { - Some(entities) => Ok(entities - .iter() - .find(|entity| entity.id().ok().as_deref() == Some(key.entity_id.as_str())) - .cloned()), - None => Err(StoreError::Unknown(anyhow!( - "nothing for type {}", - key.entity_type - ))), - } + Ok(self.get_many_res.get(&key).cloned()) } fn get_many( &self, - _ids_for_type: BTreeMap<&EntityType, Vec<&str>>, - ) -> Result>, StoreError> { + _keys: BTreeSet, + ) -> Result, StoreError> { Ok(self.get_many_res.clone()) } @@ -170,6 +161,7 @@ fn make_band(id: &'static str, data: Vec<(&str, Value)>) -> (EntityKey, Entity) EntityKey { entity_type: EntityType::new("Band".to_string()), entity_id: id.into(), + causality_region: CausalityRegion::ONCHAIN, }, Entity::from(data), ) @@ -227,12 +219,16 @@ fn insert_modifications() { ); } -fn entity_version_map( - entity_type: &str, - entities: Vec, -) -> BTreeMap> { +fn entity_version_map(entity_type: &str, entities: Vec) -> BTreeMap { let mut map = BTreeMap::new(); - map.insert(EntityType::from(entity_type), entities); + for entity in entities { + let key = EntityKey { + entity_type: EntityType::new(entity_type.to_string()), + entity_id: entity.id().unwrap().into(), + causality_region: CausalityRegion::ONCHAIN, + }; + map.insert(key, entity); + } map } diff --git a/graphql/src/introspection/resolver.rs b/graphql/src/introspection/resolver.rs index ed95e0066b9..f71ee350797 100644 --- a/graphql/src/introspection/resolver.rs +++ b/graphql/src/introspection/resolver.rs @@ -135,7 +135,7 @@ fn interface_type_object( description: interface_type.description.clone(), fields: field_objects(schema, type_objects, &interface_type.fields), - possibleTypes: schema.types_for_interface()[&interface_type.into()] + possibleTypes: schema.types_for_interface()[interface_type.name.as_str()] .iter() .map(|object_type| r::Value::String(object_type.name.to_owned())) .collect::>(), diff --git a/graphql/tests/query.rs b/graphql/tests/query.rs index a6ee479a88c..e906f06ed7a 100644 --- a/graphql/tests/query.rs +++ b/graphql/tests/query.rs @@ -1,7 +1,7 @@ #[macro_use] extern crate pretty_assertions; -use graph::components::store::{EntityKey, EntityType}; +use graph::components::store::EntityKey; use graph::data::subgraph::schema::DeploymentCreate; use graph::entity; use graph::prelude::SubscriptionResult; @@ -320,12 +320,10 @@ async fn insert_test_entities( async fn insert_at(entities: Vec, deployment: &DeploymentLocator, block_ptr: BlockPtr) { let insert_ops = entities.into_iter().map(|data| EntityOperation::Set { - key: EntityKey { - entity_type: EntityType::new( - data.get("__typename").unwrap().clone().as_string().unwrap(), - ), - entity_id: data.get("id").unwrap().clone().as_string().unwrap().into(), - }, + key: EntityKey::data( + data.get("__typename").unwrap().clone().as_string().unwrap(), + data.get("id").unwrap().clone().as_string().unwrap(), + ), data, }); diff --git a/runtime/test/src/test.rs b/runtime/test/src/test.rs index cad61b5c897..939daf0e074 100644 --- a/runtime/test/src/test.rs +++ b/runtime/test/src/test.rs @@ -423,10 +423,7 @@ fn make_thing(id: &str, value: &str) -> (String, EntityModification) { data.set("id", id); data.set("value", value); data.set("extra", USER_DATA); - let key = EntityKey { - entity_type: EntityType::new("Thing".to_string()), - entity_id: id.into(), - }; + let key = EntityKey::data("Thing".to_string(), id); ( format!("{{ \"id\": \"{}\", \"value\": \"{}\"}}", id, value), EntityModification::Insert { key, data }, diff --git a/runtime/wasm/src/host_exports.rs b/runtime/wasm/src/host_exports.rs index d6eaf0d0f24..8c64233c975 100644 --- a/runtime/wasm/src/host_exports.rs +++ b/runtime/wasm/src/host_exports.rs @@ -15,7 +15,7 @@ use graph::components::subgraph::{ PoICausalityRegion, ProofOfIndexingEvent, SharedProofOfIndexing, }; use graph::data::store; -use graph::data_source::{DataSource, DataSourceTemplate, EntityTypeAccess}; +use graph::data_source::{CausalityRegion, DataSource, DataSourceTemplate, EntityTypeAccess}; use graph::ensure; use graph::prelude::ethabi::param_type::Reader; use graph::prelude::ethabi::{decode, encode, Token}; @@ -64,12 +64,13 @@ pub struct HostExports { subgraph_network: String, data_source_context: Arc>, entity_type_access: EntityTypeAccess, + data_source_causality_region: CausalityRegion, /// Some data sources have indeterminism or different notions of time. These /// need to be each be stored separately to separate causality between them, /// and merge the results later. Right now, this is just the ethereum /// networks but will be expanded for ipfs and the availability chain. - causality_region: String, + poi_causality_region: String, templates: Arc>>, pub(crate) link_resolver: Arc, ens_lookup: Arc, @@ -91,7 +92,8 @@ impl HostExports { data_source_address: data_source.address().unwrap_or_default(), data_source_context: data_source.context().cheap_clone(), entity_type_access: data_source.entities(), - causality_region: PoICausalityRegion::from_network(&subgraph_network), + data_source_causality_region: data_source.causality_region(), + poi_causality_region: PoICausalityRegion::from_network(&subgraph_network), subgraph_network, templates, link_resolver, @@ -165,7 +167,7 @@ impl HostExports { id: &entity_id, data: &data, }, - &self.causality_region, + &self.poi_causality_region, logger, ); poi_section.end(); @@ -173,6 +175,7 @@ impl HostExports { let key = EntityKey { entity_type: EntityType::new(entity_type), entity_id: entity_id.into(), + causality_region: self.data_source_causality_region, }; self.check_entity_type_access(&key.entity_type)?; @@ -199,12 +202,13 @@ impl HostExports { entity_type: &entity_type, id: &entity_id, }, - &self.causality_region, + &self.poi_causality_region, logger, ); let key = EntityKey { entity_type: EntityType::new(entity_type), entity_id: entity_id.into(), + causality_region: self.data_source_causality_region, }; self.check_entity_type_access(&key.entity_type)?; @@ -225,6 +229,7 @@ impl HostExports { let store_key = EntityKey { entity_type: EntityType::new(entity_type), entity_id: entity_id.into(), + causality_region: self.data_source_causality_region, }; self.check_entity_type_access(&store_key.entity_type)?; diff --git a/store/postgres/Cargo.toml b/store/postgres/Cargo.toml index a753d4ca942..7b990348bd7 100644 --- a/store/postgres/Cargo.toml +++ b/store/postgres/Cargo.toml @@ -33,6 +33,7 @@ git-testament = "0.2.2" itertools = "0.10.5" pin-utils = "0.1" hex = "0.4.3" +pretty_assertions = "1.3.0" [dev-dependencies] futures = "0.3" diff --git a/store/postgres/examples/layout.rs b/store/postgres/examples/layout.rs index 1f73934adda..1bf156d5f21 100644 --- a/store/postgres/examples/layout.rs +++ b/store/postgres/examples/layout.rs @@ -2,6 +2,7 @@ extern crate clap; extern crate graph_store_postgres; use clap::{arg, Command}; +use std::collections::BTreeSet; use std::process::exit; use std::{fs, sync::Arc}; @@ -145,7 +146,7 @@ pub fn main() { ); let site = Arc::new(make_dummy_site(subgraph, namespace, "anet".to_string())); let catalog = ensure( - Catalog::for_tests(site.clone()), + Catalog::for_tests(site.clone(), BTreeSet::new()), "Failed to construct catalog", ); let layout = ensure( diff --git a/store/postgres/migrations/2022-11-10-185105_add_has_causality_region_column/down.sql b/store/postgres/migrations/2022-11-10-185105_add_has_causality_region_column/down.sql new file mode 100644 index 00000000000..4775fa58a3b --- /dev/null +++ b/store/postgres/migrations/2022-11-10-185105_add_has_causality_region_column/down.sql @@ -0,0 +1 @@ +alter table subgraphs.subgraph_manifest drop column entities_with_causality_region; diff --git a/store/postgres/migrations/2022-11-10-185105_add_has_causality_region_column/up.sql b/store/postgres/migrations/2022-11-10-185105_add_has_causality_region_column/up.sql new file mode 100644 index 00000000000..d82d5ad2628 --- /dev/null +++ b/store/postgres/migrations/2022-11-10-185105_add_has_causality_region_column/up.sql @@ -0,0 +1 @@ +alter table subgraphs.subgraph_manifest add column if not exists entities_with_causality_region text[] not null default array[]::text[]; diff --git a/store/postgres/src/block_range.rs b/store/postgres/src/block_range.rs index aafaeca29ef..a8d85e4c21c 100644 --- a/store/postgres/src/block_range.rs +++ b/store/postgres/src/block_range.rs @@ -15,6 +15,9 @@ use crate::relational::{SqlName, Table}; /// entities pub(crate) const BLOCK_RANGE_COLUMN: &str = "block_range"; +/// The name of the column that stores the causality region of an entity. +pub(crate) const CAUSALITY_REGION_COLUMN: &str = "causality_region"; + /// The SQL clause we use to check that an entity version is current; /// that version has an unbounded block range, but checking for /// `upper_inf(block_range)` is slow and can't use the exclusion diff --git a/store/postgres/src/catalog.rs b/store/postgres/src/catalog.rs index 77d8bd25be2..6228ce66842 100644 --- a/store/postgres/src/catalog.rs +++ b/store/postgres/src/catalog.rs @@ -6,9 +6,10 @@ use diesel::{ sql_types::{Array, Double, Nullable, Text}, ExpressionMethods, QueryDsl, }; +use graph::components::store::EntityType; use graph::components::store::VersionStats; use itertools::Itertools; -use std::collections::{BTreeMap, HashMap, HashSet}; +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::fmt::Write; use std::iter::FromIterator; use std::sync::Arc; @@ -115,13 +116,6 @@ lazy_static! { SqlName::verbatim("__diesel_schema_migrations".to_string()); } -// In debug builds (for testing etc.) create exclusion constraints, in -// release builds for production, skip them -#[cfg(debug_assertions)] -const CREATE_EXCLUSION_CONSTRAINT: bool = true; -#[cfg(not(debug_assertions))] -const CREATE_EXCLUSION_CONSTRAINT: bool = false; - pub struct Locale { collate: String, ctype: String, @@ -177,11 +171,15 @@ impl Locale { pub struct Catalog { pub site: Arc, text_columns: HashMap>, + pub use_poi: bool, /// Whether `bytea` columns are indexed with just a prefix (`true`) or /// in their entirety. This influences both DDL generation and how /// queries are generated pub use_bytea_prefix: bool, + + /// Set of tables which have an explicit causality region column. + pub(crate) entities_with_causality_region: BTreeSet, } impl Catalog { @@ -190,6 +188,7 @@ impl Catalog { conn: &PgConnection, site: Arc, use_bytea_prefix: bool, + entities_with_causality_region: Vec, ) -> Result { let text_columns = get_text_columns(conn, &site.namespace)?; let use_poi = supports_proof_of_indexing(conn, &site.namespace)?; @@ -198,11 +197,15 @@ impl Catalog { text_columns, use_poi, use_bytea_prefix, + entities_with_causality_region: entities_with_causality_region.into_iter().collect(), }) } /// Return a new catalog suitable for creating a new subgraph - pub fn for_creation(site: Arc) -> Self { + pub fn for_creation( + site: Arc, + entities_with_causality_region: BTreeSet, + ) -> Self { Catalog { site, text_columns: HashMap::default(), @@ -211,18 +214,23 @@ impl Catalog { // DDL generation creates indexes for prefixes of bytes columns // see: attr-bytea-prefix use_bytea_prefix: true, + entities_with_causality_region, } } /// Make a catalog as if the given `schema` did not exist in the database /// yet. This function should only be used in situations where a database /// connection is definitely not available, such as in unit tests - pub fn for_tests(site: Arc) -> Result { + pub fn for_tests( + site: Arc, + entities_with_causality_region: BTreeSet, + ) -> Result { Ok(Catalog { site, text_columns: HashMap::default(), use_poi: false, use_bytea_prefix: true, + entities_with_causality_region, }) } @@ -234,12 +242,6 @@ impl Catalog { .map(|cols| cols.contains(column.as_str())) .unwrap_or(false) } - - /// Whether to create exclusion indexes; if false, create gist indexes - /// w/o an exclusion constraint - pub fn create_exclusion_constraint() -> bool { - CREATE_EXCLUSION_CONSTRAINT - } } fn get_text_columns( diff --git a/store/postgres/src/deployment.rs b/store/postgres/src/deployment.rs index fdc4e6df617..32b7b68c41f 100644 --- a/store/postgres/src/deployment.rs +++ b/store/postgres/src/deployment.rs @@ -13,11 +13,14 @@ use diesel::{ sql_query, sql_types::{Nullable, Text}, }; -use graph::prelude::{ - anyhow, bigdecimal::ToPrimitive, hex, web3::types::H256, BigDecimal, BlockNumber, BlockPtr, - DeploymentHash, DeploymentState, Schema, StoreError, -}; use graph::{blockchain::block_stream::FirehoseCursor, data::subgraph::schema::SubgraphError}; +use graph::{ + components::store::EntityType, + prelude::{ + anyhow, bigdecimal::ToPrimitive, hex, web3::types::H256, BigDecimal, BlockNumber, BlockPtr, + DeploymentHash, DeploymentState, Schema, StoreError, + }, +}; use graph::{ data::subgraph::{ schema::{DeploymentCreate, SubgraphManifestEntity}, @@ -114,6 +117,10 @@ table! { start_block_number -> Nullable, start_block_hash -> Nullable, raw_yaml -> Nullable, + + // Entity types that have a `causality_region` column. + // Names stored as present in the schema, not in snake case. + entities_with_causality_region -> Array, } } @@ -769,6 +776,19 @@ pub(crate) fn health(conn: &PgConnection, id: DeploymentId) -> Result Result, StoreError> { + use subgraph_manifest as sm; + + sm::table + .filter(sm::id.eq(id)) + .select(sm::entities_with_causality_region) + .get_result(conn) + .map_err(|e| e.into()) +} + /// Reverts the errors and updates the subgraph health if necessary. pub(crate) fn revert_subgraph_errors( conn: &PgConnection, @@ -926,6 +946,7 @@ pub fn create_deployment( features, schema, raw_yaml, + entities_with_causality_region, }, start_block, graft_base, @@ -933,6 +954,7 @@ pub fn create_deployment( debug_fork, } = deployment; let earliest_block_number = start_block.as_ref().map(|ptr| ptr.number).unwrap_or(0); + let entities_with_causality_region = Vec::from_iter(entities_with_causality_region.into_iter()); let deployment_values = ( d::id.eq(site.id), @@ -968,6 +990,7 @@ pub fn create_deployment( m::start_block_hash.eq(b(&start_block)), m::start_block_number.eq(start_block.as_ref().map(|ptr| ptr.number)), m::raw_yaml.eq(raw_yaml), + m::entities_with_causality_region.eq(entities_with_causality_region), ); if exists && replace { diff --git a/store/postgres/src/deployment_store.rs b/store/postgres/src/deployment_store.rs index a1fc1f95e1f..ec73672c7ad 100644 --- a/store/postgres/src/deployment_store.rs +++ b/store/postgres/src/deployment_store.rs @@ -175,6 +175,8 @@ impl DeploymentStore { let exists = deployment::exists(&conn, &site)?; // Create (or update) the metadata. Update only happens in tests + let entities_with_causality_region = + deployment.manifest.entities_with_causality_region.clone(); if replace || !exists { deployment::create_deployment(&conn, &site, deployment, exists, replace)?; }; @@ -184,7 +186,12 @@ impl DeploymentStore { let query = format!("create schema {}", &site.namespace); conn.batch_execute(&query)?; - let layout = Layout::create_relational_schema(&conn, site.clone(), schema)?; + let layout = Layout::create_relational_schema( + &conn, + site.clone(), + schema, + entities_with_causality_region.into_iter().collect(), + )?; // See if we are grafting and check that the graft is permissible if let Some(base) = graft_base { let errors = layout.can_copy_from(&base); @@ -274,7 +281,7 @@ impl DeploymentStore { .interfaces_for_type(&key.entity_type) .into_iter() .flatten() - .flat_map(|interface| &types_for_interface[&interface.into()]) + .flat_map(|interface| &types_for_interface[&EntityType::from(interface)]) .map(EntityType::from) .filter(|type_name| type_name != &key.entity_type), ); @@ -1043,17 +1050,17 @@ impl DeploymentStore { ) -> Result, StoreError> { let conn = self.get_conn()?; let layout = self.layout(&conn, site)?; - layout.find(&conn, &key.entity_type, &key.entity_id, block) + layout.find(&conn, &key, block) } - /// Retrieve all the entities matching `ids_for_type` from the - /// deployment `site`. Only consider entities as of the given `block` + /// Retrieve all the entities matching `ids_for_type`, both the type and causality region, from + /// the deployment `site`. Only consider entities as of the given `block` pub(crate) fn get_many( &self, site: Arc, - ids_for_type: &BTreeMap<&EntityType, Vec<&str>>, + ids_for_type: &BTreeMap<(EntityType, CausalityRegion), Vec>, block: BlockNumber, - ) -> Result>, StoreError> { + ) -> Result, StoreError> { if ids_for_type.is_empty() { return Ok(BTreeMap::new()); } diff --git a/store/postgres/src/detail.rs b/store/postgres/src/detail.rs index 8e6698d1b53..f67ffbae34a 100644 --- a/store/postgres/src/detail.rs +++ b/store/postgres/src/detail.rs @@ -10,6 +10,7 @@ use diesel::prelude::{ use diesel_derives::Associations; use git_testament::{git_testament, git_testament_macros}; use graph::blockchain::BlockHash; +use graph::components::store::EntityType; use graph::data::subgraph::schema::{SubgraphError, SubgraphManifestEntity}; use graph::prelude::{ bigdecimal::ToPrimitive, BigDecimal, BlockPtr, DeploymentHash, StoreError, @@ -338,6 +339,7 @@ struct StoredSubgraphManifest { start_block_number: Option, start_block_hash: Option, raw_yaml: Option, + entities_with_causality_region: Vec, } impl From for SubgraphManifestEntity { @@ -349,6 +351,7 @@ impl From for SubgraphManifestEntity { features: value.features, schema: value.schema, raw_yaml: value.raw_yaml, + entities_with_causality_region: value.entities_with_causality_region, } } } diff --git a/store/postgres/src/relational.rs b/store/postgres/src/relational.rs index c1a532a2fdf..4c038f79669 100644 --- a/store/postgres/src/relational.rs +++ b/store/postgres/src/relational.rs @@ -24,6 +24,7 @@ use graph::constraint_violation; use graph::data::graphql::TypeExt as _; use graph::data::query::Trace; use graph::data::value::Word; +use graph::data_source::CausalityRegion; use graph::prelude::{q, s, EntityQuery, StopwatchMetrics, ENV_VARS}; use graph::slog::warn; use inflector::Inflector; @@ -313,6 +314,9 @@ impl Layout { &enums, &id_types, i as u32, + catalog + .entities_with_causality_region + .contains(&EntityType::from(obj_type.clone())), ) }) .collect::, _>>()?; @@ -393,6 +397,7 @@ impl Layout { position: position as u32, is_account_like: false, immutable: false, + has_causality_region: false, } } @@ -404,8 +409,9 @@ impl Layout { conn: &PgConnection, site: Arc, schema: &Schema, + entities_with_causality_region: BTreeSet, ) -> Result { - let catalog = Catalog::for_creation(site.cheap_clone()); + let catalog = Catalog::for_creation(site.cheap_clone(), entities_with_causality_region); let layout = Self::new(site, schema, catalog)?; let sql = layout .as_ddl() @@ -484,31 +490,31 @@ impl Layout { pub fn find( &self, conn: &PgConnection, - entity: &EntityType, - id: &str, + key: &EntityKey, block: BlockNumber, ) -> Result, StoreError> { - let table = self.table_for_entity(entity)?; - FindQuery::new(table.as_ref(), id, block) + let table = self.table_for_entity(&key.entity_type)?; + FindQuery::new(table.as_ref(), key, block) .get_result::(conn) .optional()? .map(|entity_data| entity_data.deserialize_with_layout(self, None, true)) .transpose() } + // An optimization when looking up multiple entities, it will generate a single sql query using `UNION ALL`. pub fn find_many( &self, conn: &PgConnection, - ids_for_type: &BTreeMap<&EntityType, Vec<&str>>, + ids_for_type: &BTreeMap<(EntityType, CausalityRegion), Vec>, block: BlockNumber, - ) -> Result>, StoreError> { + ) -> Result, StoreError> { if ids_for_type.is_empty() { return Ok(BTreeMap::new()); } let mut tables = Vec::new(); - for entity_type in ids_for_type.keys() { - tables.push(self.table_for_entity(entity_type)?.as_ref()); + for (entity_type, cr) in ids_for_type.keys() { + tables.push((self.table_for_entity(entity_type)?.as_ref(), *cr)); } let query = FindManyQuery { _namespace: &self.catalog.site.namespace, @@ -516,17 +522,22 @@ impl Layout { tables, block, }; - let mut entities_for_type: BTreeMap> = BTreeMap::new(); + let mut entities: BTreeMap = BTreeMap::new(); for data in query.load::(conn)? { let entity_type = data.entity_type(); let entity_data: Entity = data.deserialize_with_layout(self, None, true)?; - entities_for_type - .entry(entity_type) - .or_default() - .push(entity_data); + let key = EntityKey { + entity_type, + entity_id: entity_data.id()?.into(), + causality_region: CausalityRegion::from_entity(&entity_data), + }; + let overwrite = entities.insert(key, entity_data).is_some(); + if overwrite { + return Err(constraint_violation!("duplicate entity in result set")); + } } - Ok(entities_for_type) + Ok(entities) } pub fn find_changes( @@ -553,18 +564,15 @@ impl Layout { for entity_data in inserts_or_updates.into_iter() { let entity_type = entity_data.entity_type(); - let mut data: Entity = entity_data.deserialize_with_layout(self, None, false)?; + let data: Entity = entity_data.deserialize_with_layout(self, None, true)?; let entity_id = Word::from(data.id().expect("Invalid ID for entity.")); processed_entities.insert((entity_type.clone(), entity_id.clone())); - // `__typename` is not a real field. - data.remove("__typename") - .expect("__typename expected; this is a bug"); - changes.push(EntityOperation::Set { key: EntityKey { entity_type, entity_id, + causality_region: CausalityRegion::from_entity(&data), }, data, }); @@ -581,6 +589,7 @@ impl Layout { key: EntityKey { entity_type, entity_id, + causality_region: del.causality_region(), }, }); } @@ -1215,6 +1224,10 @@ pub struct Table { /// Entities in this table are immutable, i.e., will never be updated or /// deleted pub(crate) immutable: bool, + + /// Whether this table has an explicit `causality_region` column. If `false`, then the column is + /// not present and the causality region for all rows is implicitly `0` (equivalent to CasualityRegion::ONCHAIN). + pub(crate) has_causality_region: bool, } impl Table { @@ -1225,6 +1238,7 @@ impl Table { enums: &EnumMap, id_types: &IdTypeMap, position: u32, + has_causality_region: bool, ) -> Result { SqlName::check_valid_identifier(&*defn.name, "object")?; @@ -1250,6 +1264,7 @@ impl Table { columns, position, immutable, + has_causality_region, }; Ok(table) } @@ -1265,6 +1280,7 @@ impl Table { is_account_like: self.is_account_like, position: self.position, immutable: self.immutable, + has_causality_region: self.has_causality_region, }; Arc::new(other) @@ -1379,7 +1395,8 @@ impl LayoutCache { fn load(conn: &PgConnection, site: Arc) -> Result, StoreError> { let (subgraph_schema, use_bytea_prefix) = deployment::schema(conn, site.as_ref())?; - let catalog = Catalog::load(conn, site.clone(), use_bytea_prefix)?; + let has_causality_region = deployment::entities_with_causality_region(conn, site.id)?; + let catalog = Catalog::load(conn, site.clone(), use_bytea_prefix, has_causality_region)?; let layout = Arc::new(Layout::new(site.clone(), &subgraph_schema, catalog)?); layout.refresh(conn, site) } diff --git a/store/postgres/src/relational/ddl.rs b/store/postgres/src/relational/ddl.rs index 4c9d6a18927..5b20be43d29 100644 --- a/store/postgres/src/relational/ddl.rs +++ b/store/postgres/src/relational/ddl.rs @@ -5,13 +5,21 @@ use std::{ use graph::prelude::BLOCK_NUMBER_MAX; +use crate::block_range::CAUSALITY_REGION_COLUMN; use crate::relational::{ - Catalog, ColumnType, BLOCK_COLUMN, BLOCK_RANGE_COLUMN, BYTE_ARRAY_PREFIX_SIZE, - STRING_PREFIX_SIZE, VID_COLUMN, + ColumnType, BLOCK_COLUMN, BLOCK_RANGE_COLUMN, BYTE_ARRAY_PREFIX_SIZE, STRING_PREFIX_SIZE, + VID_COLUMN, }; use super::{Column, Layout, SqlName, Table}; +// In debug builds (for testing etc.) unconditionally create exclusion constraints, in release +// builds for production, skip them +#[cfg(debug_assertions)] +const CREATE_EXCLUSION_CONSTRAINT: bool = true; +#[cfg(not(debug_assertions))] +const CREATE_EXCLUSION_CONSTRAINT: bool = false; + impl Layout { /// Generate the DDL for the entire layout, i.e., all `create table` /// and `create index` etc. statements needed in the database schema @@ -77,30 +85,38 @@ impl Table { fn columns_ddl(table: &Table) -> Result { let mut cols = String::new(); let mut first = true; + + if table.has_causality_region { + first = false; + write!( + cols, + "{causality_region} int not null", + causality_region = CAUSALITY_REGION_COLUMN + )?; + } + for column in &table.columns { if !first { writeln!(cols, ",")?; - } else { - writeln!(cols)?; + write!(cols, " ")?; } - write!(cols, " ")?; column.as_ddl(&mut cols)?; first = false; } + Ok(cols) } if self.immutable { writeln!( out, - r#" - create table {qname} ( - {vid} bigserial primary key, - {block} int not null, - {cols}, - unique({id}) - ); - "#, + " + create table {qname} ( + {vid} bigserial primary key, + {block} int not null,\n\ + {cols}, + unique({id}) + );", qname = self.qualified_name, cols = columns_ddl(self)?, vid = VID_COLUMN, @@ -111,19 +127,18 @@ impl Table { writeln!( out, r#" - create table {qname} ( - {vid} bigserial primary key, - {block_range} int4range not null, - {cols} - ); - "#, + create table {qname} ( + {vid} bigserial primary key, + {block_range} int4range not null, + {cols} + );"#, qname = self.qualified_name, cols = columns_ddl(self)?, vid = VID_COLUMN, block_range = BLOCK_RANGE_COLUMN )?; - self.exclusion_ddl(out, Catalog::create_exclusion_constraint()) + self.exclusion_ddl(out) } } @@ -266,14 +281,22 @@ impl Table { self.create_attribute_indexes(out) } - pub fn exclusion_ddl(&self, out: &mut String, as_constraint: bool) -> fmt::Result { + pub fn exclusion_ddl(&self, out: &mut String) -> fmt::Result { + // Tables with causality regions need to use exclusion constraints for correctness, + // to catch violations of write isolation. + let as_constraint = self.has_causality_region || CREATE_EXCLUSION_CONSTRAINT; + + self.exclusion_ddl_inner(out, as_constraint) + } + + // `pub` for tests. + pub(crate) fn exclusion_ddl_inner(&self, out: &mut String, as_constraint: bool) -> fmt::Result { if as_constraint { writeln!( out, - r#" - alter table {qname} - add constraint {bare_name}_{id}_{block_range}_excl exclude using gist ({id} with =, {block_range} with &&); - "#, + " + alter table {qname} + add constraint {bare_name}_{id}_{block_range}_excl exclude using gist ({id} with =, {block_range} with &&);", qname = self.qualified_name, bare_name = self.name, id = self.primary_key().name, @@ -282,10 +305,10 @@ impl Table { } else { writeln!( out, - r#" + " create index {bare_name}_{id}_{block_range}_excl on {qname} using gist ({id}, {block_range}); - "#, + ", qname = self.qualified_name, bare_name = self.name, id = self.primary_key().name, @@ -303,7 +326,6 @@ impl Column { /// See the unit tests at the end of this file for the actual DDL that /// gets generated fn as_ddl(&self, out: &mut String) -> fmt::Result { - write!(out, " ")?; write!(out, "{:20} {}", self.name.quoted(), self.sql_type())?; if self.is_list() { write!(out, "[]")?; diff --git a/store/postgres/src/relational/ddl_tests.rs b/store/postgres/src/relational/ddl_tests.rs index abcf8f1e595..72a3b8bed7b 100644 --- a/store/postgres/src/relational/ddl_tests.rs +++ b/store/postgres/src/relational/ddl_tests.rs @@ -1,4 +1,5 @@ use itertools::Itertools; +use pretty_assertions::assert_eq; use super::*; @@ -11,7 +12,8 @@ fn test_layout(gql: &str) -> Layout { let schema = Schema::parse(gql, subgraph.clone()).expect("Test schema invalid"); let namespace = Namespace::new("sgd0815".to_owned()).unwrap(); let site = Arc::new(make_dummy_site(subgraph, namespace, "anet".to_string())); - let catalog = Catalog::for_tests(site.clone()).expect("Can not create catalog"); + let catalog = Catalog::for_tests(site.clone(), BTreeSet::from_iter(["FileThing".into()])) + .expect("Can not create catalog"); Layout::new(site, &schema, catalog).expect("Failed to construct Layout") } @@ -57,7 +59,7 @@ fn check_eqv(left: &str, right: &str) { fn generate_ddl() { let layout = test_layout(THING_GQL); let sql = layout.as_ddl().expect("Failed to generate DDL"); - check_eqv(THING_DDL, &sql); + assert_eq!(THING_DDL, &sql); // Use `assert_eq!` to also test the formatting. let layout = test_layout(MUSIC_GQL); let sql = layout.as_ddl().expect("Failed to generate DDL"); @@ -86,7 +88,7 @@ fn exlusion_ddl() { // When `as_constraint` is false, just create an index let mut out = String::new(); table - .exclusion_ddl(&mut out, false) + .exclusion_ddl_inner(&mut out, false) .expect("can write exclusion DDL"); check_eqv( r#"create index thing_id_block_range_excl on "sgd0815"."thing" using gist (id, block_range);"#, @@ -96,7 +98,7 @@ fn exlusion_ddl() { // When `as_constraint` is true, add an exclusion constraint let mut out = String::new(); table - .exclusion_ddl(&mut out, true) + .exclusion_ddl_inner(&mut out, true) .expect("can write exclusion DDL"); check_eqv( r#"alter table "sgd0815"."thing" add constraint thing_id_block_range_excl exclude using gist (id with =, block_range with &&);"#, @@ -198,20 +200,27 @@ const THING_GQL: &str = r#" bytes: Bytes, bigInt: BigInt, color: Color, - }"#; + } + + type FileThing @entity { + id: ID! + } + "#; const THING_DDL: &str = r#"create type sgd0815."color" as enum ('BLUE', 'red', 'yellow'); create type sgd0815."size" as enum ('large', 'medium', 'small'); -create table "sgd0815"."thing" ( + + create table "sgd0815"."thing" ( vid bigserial primary key, block_range int4range not null, "id" text not null, "big_thing" text not null -); -alter table "sgd0815"."thing" - add constraint thing_id_block_range_excl exclude using gist (id with =, block_range with &&); + ); + + alter table "sgd0815"."thing" + add constraint thing_id_block_range_excl exclude using gist (id with =, block_range with &&); create index brin_thing on "sgd0815"."thing" using brin(lower(block_range), coalesce(upper(block_range), 2147483647), vid); @@ -223,7 +232,8 @@ create index attr_0_0_thing_id create index attr_0_1_thing_big_thing on "sgd0815"."thing" using gist("big_thing", block_range); -create table "sgd0815"."scalar" ( + + create table "sgd0815"."scalar" ( vid bigserial primary key, block_range int4range not null, "id" text not null, @@ -234,9 +244,10 @@ create table "sgd0815"."scalar" ( "bytes" bytea, "big_int" numeric, "color" "sgd0815"."color" -); -alter table "sgd0815"."scalar" - add constraint scalar_id_block_range_excl exclude using gist (id with =, block_range with &&); + ); + + alter table "sgd0815"."scalar" + add constraint scalar_id_block_range_excl exclude using gist (id with =, block_range with &&); create index brin_scalar on "sgd0815"."scalar" using brin(lower(block_range), coalesce(upper(block_range), 2147483647), vid); @@ -260,6 +271,25 @@ create index attr_1_6_scalar_big_int create index attr_1_7_scalar_color on "sgd0815"."scalar" using btree("color"); + + create table "sgd0815"."file_thing" ( + vid bigserial primary key, + block_range int4range not null, + causality_region int not null, + "id" text not null + ); + + alter table "sgd0815"."file_thing" + add constraint file_thing_id_block_range_excl exclude using gist (id with =, block_range with &&); +create index brin_file_thing + on "sgd0815"."file_thing" + using brin(lower(block_range), coalesce(upper(block_range), 2147483647), vid); +create index file_thing_block_range_closed + on "sgd0815"."file_thing"(coalesce(upper(block_range), 2147483647)) + where coalesce(upper(block_range), 2147483647) < 2147483647; +create index attr_2_0_file_thing_id + on "sgd0815"."file_thing" using btree("id"); + "#; const MUSIC_GQL: &str = r#"type Musician @entity { diff --git a/store/postgres/src/relational/query_tests.rs b/store/postgres/src/relational/query_tests.rs index 2c7cacaf07a..acb4610b301 100644 --- a/store/postgres/src/relational/query_tests.rs +++ b/store/postgres/src/relational/query_tests.rs @@ -1,4 +1,4 @@ -use std::sync::Arc; +use std::{collections::BTreeSet, sync::Arc}; use diesel::{debug_query, pg::Pg}; use graph::{ @@ -33,7 +33,8 @@ fn test_layout(gql: &str) -> Layout { let schema = Schema::parse(gql, subgraph.clone()).expect("Test schema invalid"); let namespace = Namespace::new("sgd0815".to_owned()).unwrap(); let site = Arc::new(make_dummy_site(subgraph, namespace, "anet".to_string())); - let catalog = Catalog::for_tests(site.clone()).expect("Can not create catalog"); + let catalog = + Catalog::for_tests(site.clone(), BTreeSet::new()).expect("Can not create catalog"); Layout::new(site, &schema, catalog).expect("Failed to construct Layout") } diff --git a/store/postgres/src/relational_queries.rs b/store/postgres/src/relational_queries.rs index cbf85bbf973..e311835437b 100644 --- a/store/postgres/src/relational_queries.rs +++ b/store/postgres/src/relational_queries.rs @@ -14,6 +14,7 @@ use diesel::Connection; use graph::components::store::EntityKey; use graph::data::value::Word; +use graph::data_source::CausalityRegion; use graph::prelude::{ anyhow, r, serde_json, Attribute, BlockNumber, ChildMultiplicity, Entity, EntityCollection, EntityFilter, EntityLink, EntityOrder, EntityRange, EntityWindow, ParentLink, @@ -39,7 +40,7 @@ use crate::sql_value::SqlValue; use crate::{ block_range::{ BlockRangeColumn, BlockRangeLowerBoundClause, BlockRangeUpperBoundClause, BLOCK_COLUMN, - BLOCK_RANGE_COLUMN, BLOCK_RANGE_CURRENT, + BLOCK_RANGE_COLUMN, BLOCK_RANGE_CURRENT, CAUSALITY_REGION_COLUMN, }, primary::{Namespace, Site}, }; @@ -448,6 +449,8 @@ pub struct EntityDeletion { entity: String, #[sql_type = "Text"] id: String, + #[sql_type = "Integer"] + causality_region: CausalityRegion, } impl EntityDeletion { @@ -458,6 +461,10 @@ impl EntityDeletion { pub fn id(&self) -> &str { &self.id } + + pub fn causality_region(&self) -> CausalityRegion { + self.causality_region + } } /// Helper struct for retrieving entities from the database. With diesel, we @@ -1443,10 +1450,12 @@ impl<'a> QueryFragment for QueryFilter<'a> { } } +/// A query that finds an entity by key. Used during indexing. +/// See also `FindManyQuery`. #[derive(Debug, Clone, Constructor)] pub struct FindQuery<'a> { table: &'a Table, - id: &'a str, + key: &'a EntityKey, block: BlockNumber, } @@ -1454,6 +1463,12 @@ impl<'a> QueryFragment for FindQuery<'a> { fn walk_ast(&self, mut out: AstPass) -> QueryResult<()> { out.unsafe_to_cache_prepared(); + let EntityKey { + entity_type: _, + entity_id, + causality_region, + } = self.key; + // Generate // select '..' as entity, to_jsonb(e.*) as data // from schema.table e where id = $1 @@ -1463,8 +1478,13 @@ impl<'a> QueryFragment for FindQuery<'a> { out.push_sql(" from "); out.push_sql(self.table.qualified_name.as_str()); out.push_sql(" e\n where "); - self.table.primary_key().eq(self.id, &mut out)?; + self.table.primary_key().eq(entity_id, &mut out)?; out.push_sql(" and "); + if self.table.has_causality_region { + out.push_sql("causality_region = "); + out.push_bind_param::(causality_region)?; + out.push_sql(" and "); + } BlockRangeColumn::new(self.table, "e.", self.block).contains(&mut out) } } @@ -1553,7 +1573,13 @@ impl<'a> QueryFragment for FindPossibleDeletionsQuery<'a> { } out.push_sql("select "); out.push_bind_param::(&table.object.as_str())?; - out.push_sql(" as entity, e.id\n"); + out.push_sql(" as entity, "); + if table.has_causality_region { + out.push_sql("causality_region, "); + } else { + out.push_sql("0 as causality_region, "); + } + out.push_sql("e.id\n"); out.push_sql(" from "); out.push_sql(table.qualified_name.as_str()); out.push_sql(" e\n where "); @@ -1581,10 +1607,10 @@ impl<'a, Conn> RunQueryDsl for FindPossibleDeletionsQuery<'a> {} #[derive(Debug, Clone, Constructor)] pub struct FindManyQuery<'a> { pub(crate) _namespace: &'a Namespace, - pub(crate) tables: Vec<&'a Table>, + pub(crate) tables: Vec<(&'a Table, CausalityRegion)>, // Maps object name to ids. - pub(crate) ids_for_type: &'a BTreeMap<&'a EntityType, Vec<&'a str>>, + pub(crate) ids_for_type: &'a BTreeMap<(EntityType, CausalityRegion), Vec>, pub(crate) block: BlockNumber, } @@ -1600,7 +1626,7 @@ impl<'a> QueryFragment for FindManyQuery<'a> { // from schema. e where {id.is_in($ids1)) // union all // ... - for (i, table) in self.tables.iter().enumerate() { + for (i, (table, cr)) in self.tables.iter().enumerate() { if i > 0 { out.push_sql("\nunion all\n"); } @@ -1612,8 +1638,13 @@ impl<'a> QueryFragment for FindManyQuery<'a> { out.push_sql(" e\n where "); table .primary_key() - .is_in(&self.ids_for_type[&table.object], &mut out)?; + .is_in(&self.ids_for_type[&(table.object.clone(), *cr)], &mut out)?; out.push_sql(" and "); + if table.has_causality_region { + out.push_sql("causality_region = "); + out.push_bind_param::(cr)?; + out.push_sql(" and "); + } BlockRangeColumn::new(table, "e.", self.block).contains(&mut out)?; } Ok(()) @@ -1745,12 +1776,16 @@ impl<'a> QueryFragment for InsertQuery<'a> { out.push_sql(", "); } self.br_column.name(&mut out); + if self.table.has_causality_region { + out.push_sql(", "); + out.push_sql(CAUSALITY_REGION_COLUMN); + }; out.push_sql(") values\n"); // Use a `Peekable` iterator to help us decide how to finalize each line. - let mut iter = self.entities.iter().map(|(_key, entity)| entity).peekable(); - while let Some(entity) = iter.next() { + let mut iter = self.entities.iter().peekable(); + while let Some((key, entity)) = iter.next() { out.push_sql("("); for column in &self.unique_columns { // If the column name is not within this entity's fields, we will issue the @@ -1763,6 +1798,10 @@ impl<'a> QueryFragment for InsertQuery<'a> { out.push_sql(", "); } self.br_column.literal_range_current(&mut out)?; + if self.table.has_causality_region { + out.push_sql(", "); + out.push_bind_param::(&key.causality_region)?; + }; out.push_sql(")"); // finalize line according to remaining entities to insert @@ -3488,6 +3527,12 @@ impl<'a> QueryFragment for CopyEntityBatchQuery<'a> { } else { out.push_sql(BLOCK_RANGE_COLUMN); } + + if self.dst.has_causality_region { + out.push_sql(", "); + out.push_sql(CAUSALITY_REGION_COLUMN); + }; + out.push_sql(")\nselect "); for column in &self.columns { out.push_identifier(column.name.as_str())?; @@ -3534,6 +3579,24 @@ impl<'a> QueryFragment for CopyEntityBatchQuery<'a> { } (false, false) => out.push_sql(BLOCK_RANGE_COLUMN), } + + match (self.src.has_causality_region, self.dst.has_causality_region) { + (false, false) => (), + (true, true) => { + out.push_sql(", "); + out.push_sql(CAUSALITY_REGION_COLUMN); + } + (false, true) => { + out.push_sql(", 0"); + } + (true, false) => { + return Err(constraint_violation!( + "can not copy entity type {} to {} because the src has a causality region but the dst does not", + self.src.object.as_str(), + self.dst.object.as_str() + )); + } + } out.push_sql(" from "); out.push_sql(self.src.qualified_name.as_str()); out.push_sql(" where vid >= "); diff --git a/store/postgres/src/writable.rs b/store/postgres/src/writable.rs index 926ad38a43a..204af9f7fe3 100644 --- a/store/postgres/src/writable.rs +++ b/store/postgres/src/writable.rs @@ -1,3 +1,4 @@ +use std::collections::BTreeSet; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Mutex; use std::time::Duration; @@ -274,12 +275,20 @@ impl SyncStore { fn get_many( &self, - ids_for_type: BTreeMap<&EntityType, Vec<&str>>, + keys: BTreeSet, block: BlockNumber, - ) -> Result>, StoreError> { + ) -> Result, StoreError> { + let mut by_type: BTreeMap<(EntityType, CausalityRegion), Vec> = BTreeMap::new(); + for key in keys { + by_type + .entry((key.entity_type, key.causality_region)) + .or_default() + .push(key.entity_id.into()); + } + self.retry("get_many", || { self.writable - .get_many(self.site.cheap_clone(), &ids_for_type, block) + .get_many(self.site.cheap_clone(), &by_type, block) }) } @@ -708,15 +717,15 @@ impl Queue { /// Get many entities at once by looking at both the queue and the store fn get_many( &self, - mut ids_for_type: BTreeMap<&EntityType, Vec<&str>>, - ) -> Result>, StoreError> { + mut keys: BTreeSet, + ) -> Result, StoreError> { // See the implementation of `get` for how we handle reverts let mut tracker = BlockTracker::new(); // Get entities from entries in the queue - let mut map = self.queue.fold( + let entities_in_queue = self.queue.fold( BTreeMap::new(), - |mut map: BTreeMap>, req| { + |mut map: BTreeMap>, req| { tracker.update(req.as_ref()); match req.as_ref() { Request::Write { @@ -725,26 +734,9 @@ impl Queue { if tracker.visible(block_ptr) { for emod in mods { let key = emod.entity_ref(); - if let Some(ids) = ids_for_type.get_mut(&key.entity_type) { - if let Some(idx) = - ids.iter().position(|id| *id == key.entity_id.as_str()) - { - // We are looking for the entity - // underlying this modification. Add - // it to the result map, but also - // remove it from `ids_for_type` so - // that we don't look for it any - // more - if let Some(entity) = emod.entity() { - map.entry(key.entity_type.clone()) - .or_default() - .push(entity.clone()); - } - ids.swap_remove(idx); - if ids.is_empty() { - ids_for_type.remove(&key.entity_type); - } - } + // The key must be removed to avoid overwriting it with a stale value. + if let Some(key) = keys.take(key) { + map.insert(key, emod.entity().cloned()); } } } @@ -755,20 +747,17 @@ impl Queue { }, ); - // Whatever remains in `ids_for_type` needs to be gotten from the - // store. Take extra care to not unnecessarily copy maps - if !ids_for_type.is_empty() { - let store_map = self.store.get_many(ids_for_type, tracker.query_block())?; - if !store_map.is_empty() { - if map.is_empty() { - map = store_map - } else { - for (entity_type, mut entities) in store_map { - map.entry(entity_type).or_default().append(&mut entities); - } - } + // Whatever remains in `keys` needs to be gotten from the store + let mut map = self.store.get_many(keys, tracker.query_block())?; + + // Extend the store results with the entities from the queue. + for (key, entity) in entities_in_queue { + if let Some(entity) = entity { + let overwrite = map.insert(key, entity).is_some(); + assert!(!overwrite); } } + Ok(map) } @@ -921,11 +910,11 @@ impl Writer { fn get_many( &self, - ids_for_type: BTreeMap<&EntityType, Vec<&str>>, - ) -> Result>, StoreError> { + keys: BTreeSet, + ) -> Result, StoreError> { match self { - Writer::Sync(store) => store.get_many(ids_for_type, BLOCK_NUMBER_MAX), - Writer::Async(queue) => queue.get_many(ids_for_type), + Writer::Sync(store) => store.get_many(keys, BLOCK_NUMBER_MAX), + Writer::Async(queue) => queue.get_many(keys), } } @@ -1006,9 +995,9 @@ impl ReadStore for WritableStore { fn get_many( &self, - ids_for_type: BTreeMap<&EntityType, Vec<&str>>, - ) -> Result>, StoreError> { - self.writer.get_many(ids_for_type) + keys: BTreeSet, + ) -> Result, StoreError> { + self.writer.get_many(keys) } fn input_schema(&self) -> Arc { diff --git a/store/postgres/tests/graft.rs b/store/postgres/tests/graft.rs index 4a7539b61af..660661778e0 100644 --- a/store/postgres/tests/graft.rs +++ b/store/postgres/tests/graft.rs @@ -259,10 +259,7 @@ fn create_test_entity( ); EntityOperation::Set { - key: EntityKey { - entity_type: EntityType::new(entity_type.to_string()), - entity_id: id.into(), - }, + key: EntityKey::data(entity_type.to_string(), id), data: test_entity, } } @@ -325,10 +322,7 @@ async fn check_graft( // Make our own entries for block 2 shaq.set("email", "shaq@gmail.com"); let op = EntityOperation::Set { - key: EntityKey { - entity_type: EntityType::new(USER.to_owned()), - entity_id: "3".into(), - }, + key: EntityKey::data(USER.to_owned(), "3"), data: shaq, }; transact_and_wait(&store, &deployment, BLOCKS[2].clone(), vec![op]) diff --git a/store/postgres/tests/relational.rs b/store/postgres/tests/relational.rs index 4cf67adeca5..986b4cde872 100644 --- a/store/postgres/tests/relational.rs +++ b/store/postgres/tests/relational.rs @@ -16,6 +16,7 @@ use graph_store_postgres::layout_for_tests::SqlName; use hex_literal::hex; use lazy_static::lazy_static; use std::borrow::Cow; +use std::collections::BTreeSet; use std::panic; use std::str::FromStr; use std::sync::Arc; @@ -426,7 +427,7 @@ fn create_schema(conn: &PgConnection) -> Layout { let query = format!("create schema {}", NAMESPACE.as_str()); conn.batch_execute(&*query).unwrap(); - Layout::create_relational_schema(&conn, Arc::new(site), &schema) + Layout::create_relational_schema(&conn, Arc::new(site), &schema, BTreeSet::new()) .expect("Failed to create relational schema") } @@ -494,19 +495,31 @@ fn find() { // Happy path: find existing entity let entity = layout - .find(conn, &*SCALAR, "one", BLOCK_NUMBER_MAX) + .find( + conn, + &EntityKey::data(SCALAR.as_str(), "one"), + BLOCK_NUMBER_MAX, + ) .expect("Failed to read Scalar[one]") .unwrap(); assert_entity_eq!(scrub(&*SCALAR_ENTITY), entity); // Find non-existing entity let entity = layout - .find(conn, &*SCALAR, "noone", BLOCK_NUMBER_MAX) + .find( + conn, + &EntityKey::data(SCALAR.as_str(), "noone"), + BLOCK_NUMBER_MAX, + ) .expect("Failed to read Scalar[noone]"); assert!(entity.is_none()); // Find for non-existing entity type - let err = layout.find(conn, &*NO_ENTITY, "one", BLOCK_NUMBER_MAX); + let err = layout.find( + conn, + &EntityKey::data(NO_ENTITY.as_str(), "one"), + BLOCK_NUMBER_MAX, + ); match err { Err(e) => assert_eq!("unknown table 'NoEntity'", e.to_string()), _ => { @@ -529,7 +542,11 @@ fn insert_null_fulltext_fields() { // Find entity with null string values let entity = layout - .find(conn, &*NULLABLE_STRINGS, "one", BLOCK_NUMBER_MAX) + .find( + conn, + &EntityKey::data(NULLABLE_STRINGS.as_str(), "one"), + BLOCK_NUMBER_MAX, + ) .expect("Failed to read NullableStrings[one]") .unwrap(); assert_entity_eq!(scrub(&*EMPTY_NULLABLESTRINGS_ENTITY), entity); @@ -555,7 +572,11 @@ fn update() { .expect("Failed to update"); let actual = layout - .find(conn, &*SCALAR, "one", BLOCK_NUMBER_MAX) + .find( + conn, + &EntityKey::data(SCALAR.as_str(), "one"), + BLOCK_NUMBER_MAX, + ) .expect("Failed to read Scalar[one]") .unwrap(); assert_entity_eq!(scrub(&entity), actual); @@ -611,9 +632,13 @@ fn update_many() { // check updates took effect let updated: Vec = ["one", "two", "three"] .iter() - .map(|id| { + .map(|&id| { layout - .find(conn, &*SCALAR, id, BLOCK_NUMBER_MAX) + .find( + conn, + &EntityKey::data(SCALAR.as_str(), id), + BLOCK_NUMBER_MAX, + ) .expect(&format!("Failed to read Scalar[{}]", id)) .unwrap() }) @@ -679,7 +704,11 @@ fn serialize_bigdecimal() { .expect("Failed to update"); let actual = layout - .find(conn, &*SCALAR, "one", BLOCK_NUMBER_MAX) + .find( + conn, + &EntityKey::data(SCALAR.as_str(), "one"), + BLOCK_NUMBER_MAX, + ) .expect("Failed to read Scalar[one]") .unwrap(); assert_entity_eq!(entity, actual); @@ -880,7 +909,7 @@ fn revert_block() { let assert_fred = |name: &str| { let fred = layout - .find(conn, &EntityType::from("Cat"), id, BLOCK_NUMBER_MAX) + .find(conn, &EntityKey::data("Cat", id), BLOCK_NUMBER_MAX) .unwrap() .expect("there's a fred"); assert_eq!(name, fred.get("name").unwrap().as_str().unwrap()) diff --git a/store/postgres/tests/relational_bytes.rs b/store/postgres/tests/relational_bytes.rs index 7198085e9e4..663084a7100 100644 --- a/store/postgres/tests/relational_bytes.rs +++ b/store/postgres/tests/relational_bytes.rs @@ -3,11 +3,13 @@ use diesel::connection::SimpleConnection as _; use diesel::pg::PgConnection; use graph::components::store::EntityKey; use graph::data::store::scalar; +use graph::data_source::CausalityRegion; use graph::prelude::EntityQuery; use graph_mock::MockMetricsRegistry; use hex_literal::hex; use lazy_static::lazy_static; use std::borrow::Cow; +use std::collections::BTreeSet; use std::str::FromStr; use std::{collections::BTreeMap, sync::Arc}; @@ -127,7 +129,7 @@ fn create_schema(conn: &PgConnection) -> Layout { NAMESPACE.clone(), NETWORK_NAME.to_string(), ); - Layout::create_relational_schema(conn, Arc::new(site), &schema) + Layout::create_relational_schema(conn, Arc::new(site), &schema, BTreeSet::new()) .expect("Failed to create relational schema") } @@ -193,7 +195,11 @@ fn bad_id() { // We test that we get errors for various strings that are not // valid 'Bytes' strings; we use `find` to force the conversion // from String -> Bytes internally - let res = layout.find(conn, &*THING, "bad", BLOCK_NUMBER_MAX); + let res = layout.find( + conn, + &EntityKey::data(THING.as_str(), "bad"), + BLOCK_NUMBER_MAX, + ); assert!(res.is_err()); assert_eq!( "store error: Odd number of digits", @@ -201,7 +207,11 @@ fn bad_id() { ); // We do not allow the `\x` prefix that Postgres uses - let res = layout.find(conn, &*THING, "\\xbadd", BLOCK_NUMBER_MAX); + let res = layout.find( + conn, + &EntityKey::data(THING.as_str(), "\\xbadd"), + BLOCK_NUMBER_MAX, + ); assert!(res.is_err()); assert_eq!( "store error: Invalid character \'\\\\\' at position 0", @@ -209,11 +219,19 @@ fn bad_id() { ); // Having the '0x' prefix is ok - let res = layout.find(conn, &*THING, "0xbadd", BLOCK_NUMBER_MAX); + let res = layout.find( + conn, + &EntityKey::data(THING.as_str(), "0xbadd"), + BLOCK_NUMBER_MAX, + ); assert!(res.is_ok()); // Using non-hex characters is also bad - let res = layout.find(conn, &*THING, "nope", BLOCK_NUMBER_MAX); + let res = layout.find( + conn, + &EntityKey::data(THING.as_str(), "nope"), + BLOCK_NUMBER_MAX, + ); assert!(res.is_err()); assert_eq!( "store error: Invalid character \'n\' at position 0", @@ -231,14 +249,19 @@ fn find() { // Happy path: find existing entity let entity = layout - .find(conn, &*THING, ID, BLOCK_NUMBER_MAX) + .find(conn, &EntityKey::data(THING.as_str(), ID), BLOCK_NUMBER_MAX) .expect("Failed to read Thing[deadbeef]") .unwrap(); assert_entity_eq!(scrub(&*BEEF_ENTITY), entity); + assert!(CausalityRegion::from_entity(&entity) == CausalityRegion::ONCHAIN); // Find non-existing entity let entity = layout - .find(conn, &*THING, "badd", BLOCK_NUMBER_MAX) + .find( + conn, + &EntityKey::data(THING.as_str(), "badd"), + BLOCK_NUMBER_MAX, + ) .expect("Failed to read Thing[badd]"); assert!(entity.is_none()); }); @@ -254,24 +277,29 @@ fn find_many() { insert_thing(conn, layout, ID, NAME); insert_thing(conn, layout, ID2, NAME2); - let mut id_map: BTreeMap<&EntityType, Vec<&str>> = BTreeMap::default(); - id_map.insert(&*THING, vec![ID, ID2, "badd"]); + let mut id_map = BTreeMap::default(); + id_map.insert( + (THING.clone(), CausalityRegion::ONCHAIN), + vec![ID.to_string(), ID2.to_string(), "badd".to_string()], + ); let entities = layout .find_many(conn, &id_map, BLOCK_NUMBER_MAX) .expect("Failed to read many things"); - assert_eq!(1, entities.len()); - - let ids = entities - .get(&*THING) - .expect("We got some things") - .iter() - .map(|thing| thing.id().unwrap()) - .collect::>(); - - assert_eq!(2, ids.len()); - assert!(ids.contains(&ID.to_owned()), "Missing ID"); - assert!(ids.contains(&ID2.to_owned()), "Missing ID2"); + assert_eq!(2, entities.len()); + + let id_key = EntityKey { + entity_id: ID.into(), + entity_type: THING.clone(), + causality_region: CausalityRegion::ONCHAIN, + }; + let id2_key = EntityKey { + entity_id: ID2.into(), + entity_type: THING.clone(), + causality_region: CausalityRegion::ONCHAIN, + }; + assert!(entities.contains_key(&id_key), "Missing ID"); + assert!(entities.contains_key(&id2_key), "Missing ID2"); }); } @@ -293,7 +321,11 @@ fn update() { .expect("Failed to update"); let actual = layout - .find(conn, &*THING, &entity_id, BLOCK_NUMBER_MAX) + .find( + conn, + &EntityKey::data(THING.as_str(), &entity_id), + BLOCK_NUMBER_MAX, + ) .expect("Failed to read Thing[deadbeef]") .unwrap(); diff --git a/store/test-store/src/store.rs b/store/test-store/src/store.rs index c3d00035988..513af19aca6 100644 --- a/store/test-store/src/store.rs +++ b/store/test-store/src/store.rs @@ -3,6 +3,7 @@ use graph::data::graphql::effort::LoadManager; use graph::data::query::QueryResults; use graph::data::query::QueryTarget; use graph::data::subgraph::schema::{DeploymentCreate, SubgraphError}; +use graph::data_source::CausalityRegion; use graph::log; use graph::prelude::{QueryStoreManager as _, SubgraphStore as _, *}; use graph::semver::Version; @@ -355,6 +356,7 @@ pub async fn insert_entities( key: EntityKey { entity_type, entity_id: data.get("id").unwrap().clone().as_string().unwrap().into(), + causality_region: CausalityRegion::ONCHAIN, }, data, }); diff --git a/tests/Cargo.toml b/tests/Cargo.toml index b0893f97ba9..cdc636c546e 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -24,6 +24,7 @@ hex = "0.4.3" serde_yaml = "0.8" hyper = "0.14" serde = "1.0" +assert-json-diff = "2.0.2" [dev-dependencies] bollard = "0.10" diff --git a/tests/integration-tests/file-data-sources/abis/Contract.abi b/tests/integration-tests/file-data-sources/abis/Contract.abi index 1e3ec7217af..9d9f56b9263 100644 --- a/tests/integration-tests/file-data-sources/abis/Contract.abi +++ b/tests/integration-tests/file-data-sources/abis/Contract.abi @@ -1 +1,15 @@ -[ ] +[ + { + "anonymous": false, + "inputs": [ + { + "indexed": false, + "internalType": "string", + "name": "testCommand", + "type": "string" + } + ], + "name": "TestEvent", + "type": "event" + } +] diff --git a/tests/integration-tests/file-data-sources/src/mapping.ts b/tests/integration-tests/file-data-sources/src/mapping.ts index f928b1ba419..dbd8013908d 100644 --- a/tests/integration-tests/file-data-sources/src/mapping.ts +++ b/tests/integration-tests/file-data-sources/src/mapping.ts @@ -1,16 +1,31 @@ import { ethereum, dataSource, BigInt, Bytes, DataSourceContext } from '@graphprotocol/graph-ts' +import { TestEvent } from '../generated/Contract/Contract' import { IpfsFile, IpfsFile1 } from '../generated/schema' +// CID of `file-data-sources/abis/Contract.abi` after being processed by graph-cli. +const KNOWN_CID = "QmQ2REmceVtzawp7yrnxLQXgNNCtFHEnig6fL9aqE1kcWq" + export function handleBlock(block: ethereum.Block): void { + let entity = new IpfsFile("onchain") + entity.content = "onchain" + entity.save() + // This will create the same data source twice, once at block 0 and another at block 2. // The creation at block 2 should be detected as a duplicate and therefore a noop. if (block.number == BigInt.fromI32(0) || block.number == BigInt.fromI32(2)) { - // CID QmVkvoPGi9jvvuxsHDVJDgzPEzagBaWSZRYoRDzU244HjZ is the file - // `file-data-sources/abis/Contract.abi` after being processed by graph-cli. - dataSource.create("File", ["QmVkvoPGi9jvvuxsHDVJDgzPEzagBaWSZRYoRDzU244HjZ"]) + dataSource.create("File", [KNOWN_CID]) } if (block.number == BigInt.fromI32(1)) { + let entity = IpfsFile.load("onchain")! + assert(entity.content == "onchain") + + // The test assumes file data sources are processed in the block in which they are created. + // So the ds created at block 0 will have been processed. + // + // Test that onchain data sources cannot read offchain data. + assert(IpfsFile.load(KNOWN_CID) == null); + // Test that using an invalid CID will be ignored dataSource.create("File", ["hi, I'm not valid"]) } @@ -19,19 +34,50 @@ export function handleBlock(block: ethereum.Block): void { // This will invoke File1 data source with same CID, which will be used // to test whether same cid is triggered across different data source. if (block.number == BigInt.fromI32(3)) { - dataSource.create("File1", ["QmVkvoPGi9jvvuxsHDVJDgzPEzagBaWSZRYoRDzU244HjZ"]) + // Test that onchain data sources cannot read offchain data (again, but this time more likely to hit the DB than the write queue). + assert(IpfsFile.load(KNOWN_CID) == null); + + dataSource.create("File1", [KNOWN_CID]) } +} + +export function handleTestEvent(event: TestEvent): void { + let command = event.params.testCommand; - // Will fail the subgraph when processed due to mismatch in the entity type and 'entities'. - if (block.number == BigInt.fromI32(5)) { - dataSource.create("File2", ["QmVkvoPGi9jvvuxsHDVJDgzPEzagBaWSZRYoRDzU244HjZ"]) + if (command == "createFile2") { + // Will fail the subgraph when processed due to mismatch in the entity type and 'entities'. + dataSource.create("File2", [KNOWN_CID]) + } else if (command == "saveConflictingEntity") { + // Will fail the subgraph because the same entity has been created in a file data source. + let entity = new IpfsFile(KNOWN_CID) + entity.content = "empty" + entity.save() + } else if (command == "createFile1") { + // Will fail the subgraph with a conflict between two entities created by offchain data sources. + let context = new DataSourceContext(); + context.setBytes("hash", event.block.hash); + dataSource.createWithContext("File1", [KNOWN_CID], context) + } else { + assert(false, "Unknown command: " + command); } } export function handleFile(data: Bytes): void { + // Test that offchain data sources cannot read onchain data. + assert(IpfsFile.load("onchain") == null); + + if (dataSource.stringParam() != "QmVkvoPGi9jvvuxsHDVJDgzPEzagBaWSZRYoRDzU244HjZ") { + // Test that an offchain data source cannot read from another offchain data source. + assert(IpfsFile.load("QmVkvoPGi9jvvuxsHDVJDgzPEzagBaWSZRYoRDzU244HjZ") == null); + } + let entity = new IpfsFile(dataSource.stringParam()) entity.content = data.toString() entity.save() + + // Test that an offchain data source can load its own entities + let loaded_entity = IpfsFile.load(dataSource.stringParam())! + assert(loaded_entity.content == entity.content) } export function handleFile1(data: Bytes): void { diff --git a/tests/integration-tests/file-data-sources/subgraph.yaml b/tests/integration-tests/file-data-sources/subgraph.yaml index ee9d3aac689..502faccdff6 100644 --- a/tests/integration-tests/file-data-sources/subgraph.yaml +++ b/tests/integration-tests/file-data-sources/subgraph.yaml @@ -6,7 +6,7 @@ dataSources: name: Contract network: test source: - address: "0xCfEB869F69431e42cdB54A4F4f105C19C080A601" + address: "0x0000000000000000000000000000000000000000" abi: Contract mapping: kind: ethereum/events @@ -19,6 +19,9 @@ dataSources: file: ./abis/Contract.abi blockHandlers: - handler: handleBlock + eventHandlers: + - event: TestEvent(string) + handler: handleTestEvent file: ./src/mapping.ts templates: - kind: file/ipfs diff --git a/tests/src/fixture.rs b/tests/src/fixture.rs index f0487a65a5b..654bea65014 100644 --- a/tests/src/fixture.rs +++ b/tests/src/fixture.rs @@ -99,14 +99,36 @@ pub async fn build_subgraph_with_yarn_cmd(dir: &str, yarn_cmd: &str) -> Deployme } pub fn test_ptr(n: BlockNumber) -> BlockPtr { + test_ptr_reorged(n, 0) +} + +// Set n as the low bits and `reorg_n` as the high bits of the hash. +pub fn test_ptr_reorged(n: BlockNumber, reorg_n: u32) -> BlockPtr { + let mut hash = H256::from_low_u64_be(n as u64); + hash[0..4].copy_from_slice(&reorg_n.to_be_bytes()); BlockPtr { - hash: H256::from_low_u64_be(n as u64).into(), + hash: hash.into(), number: n, } } type GraphQlRunner = graph_graphql::prelude::GraphQlRunner; +pub struct TestChain { + pub chain: Arc, + pub block_stream_builder: Arc>, +} + +impl TestChain { + pub fn set_block_stream(&self, blocks: Vec>) + where + C::TriggerData: Clone, + { + let static_block_stream = Arc::new(StaticStreamBuilder { chain: blocks }); + *self.block_stream_builder.0.lock().unwrap() = static_block_stream; + } +} + pub struct TestContext { pub logger: Logger, pub provider: Arc< @@ -120,6 +142,7 @@ pub struct TestContext { pub instance_manager: SubgraphInstanceManager, pub link_resolver: Arc, pub env_vars: Arc, + pub ipfs: IpfsClient, graphql_runner: Arc, indexing_status_service: Arc>, } @@ -184,6 +207,9 @@ impl TestContext { } pub async fn start_and_sync_to(&self, stop_block: BlockPtr) { + // In case the subgraph has been previously started. + self.provider.stop(self.deployment.clone()).await.unwrap(); + self.provider .start(self.deployment.clone(), Some(stop_block.number)) .await @@ -200,8 +226,11 @@ impl TestContext { } pub async fn start_and_sync_to_error(&self, stop_block: BlockPtr) -> SubgraphError { + // In case the subgraph has been previously started. + self.provider.stop(self.deployment.clone()).await.unwrap(); + self.provider - .start(self.deployment.clone(), Some(stop_block.number)) + .start(self.deployment.clone(), None) .await .expect("unable to start subgraph"); @@ -259,6 +288,12 @@ impl TestContext { serde_json::from_str(&serde_json::to_string(&value).unwrap()).unwrap(); query_res.indexing_status_for_current_version } + + pub fn rewind(&self, block_ptr_to: BlockPtr) { + self.store + .rewind(self.deployment.hash.clone(), block_ptr_to) + .unwrap() + } } impl Drop for TestContext { @@ -332,7 +367,7 @@ pub async fn setup( subgraph_name: SubgraphName, hash: &DeploymentHash, stores: &Stores, - chain: Arc, + chain: &TestChain, graft_block: Option, env_vars: Option, ) -> TestContext { @@ -351,7 +386,7 @@ pub async fn setup( cleanup(&subgraph_store, &subgraph_name, hash).unwrap(); let mut blockchain_map = BlockchainMap::new(); - blockchain_map.insert(stores.network_name.clone(), chain); + blockchain_map.insert(stores.network_name.clone(), chain.chain.clone()); let static_filters = env_vars.experimental_static_filters; @@ -361,7 +396,7 @@ pub async fn setup( Default::default(), )); let ipfs_service = ipfs_service( - ipfs, + ipfs.cheap_clone(), env_vars.mappings.max_ipfs_file_bytes as u64, env_vars.mappings.ipfs_timeout, env_vars.mappings.ipfs_request_limit, @@ -445,6 +480,7 @@ pub async fn setup( link_resolver, env_vars, indexing_status_service, + ipfs, } } @@ -481,8 +517,11 @@ pub async fn wait_for_sync( }; let status = store.status_for_id(deployment.id); + if let Some(fatal_error) = status.fatal_error { - return Err(fatal_error); + if fatal_error.block_ptr.as_ref().unwrap() == &stop_block { + return Err(fatal_error); + } } if block_ptr == stop_block { @@ -514,6 +553,48 @@ impl BlockRefetcher for StaticBlockRefetcher { } } +pub struct MutexBlockStreamBuilder(pub Mutex>>); + +#[async_trait] +impl BlockStreamBuilder for MutexBlockStreamBuilder { + async fn build_firehose( + &self, + chain: &C, + deployment: DeploymentLocator, + block_cursor: FirehoseCursor, + start_blocks: Vec, + subgraph_current_block: Option, + filter: Arc<::TriggerFilter>, + unified_api_version: graph::data::subgraph::UnifiedMappingApiVersion, + ) -> anyhow::Result>> { + let builder = self.0.lock().unwrap().clone(); + + builder + .build_firehose( + chain, + deployment, + block_cursor, + start_blocks, + subgraph_current_block, + filter, + unified_api_version, + ) + .await + } + + async fn build_polling( + &self, + _chain: Arc, + _deployment: DeploymentLocator, + _start_blocks: Vec, + _subgraph_current_block: Option, + _filter: Arc<::TriggerFilter>, + _unified_api_version: graph::data::subgraph::UnifiedMappingApiVersion, + ) -> anyhow::Result>> { + unimplemented!("only firehose mode should be used for tests") + } +} + /// `chain` is the sequence of chain heads to be processed. If the next block to be processed in the /// chain is not a descendant of the previous one, reorgs will be emitted until it is. /// diff --git a/tests/src/fixture/ethereum.rs b/tests/src/fixture/ethereum.rs index 5d941b45c0d..3e94385ece5 100644 --- a/tests/src/fixture/ethereum.rs +++ b/tests/src/fixture/ethereum.rs @@ -1,16 +1,17 @@ use std::marker::PhantomData; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use std::time::Duration; use super::{ - test_ptr, NoopAdapterSelector, NoopRuntimeAdapter, StaticBlockRefetcher, StaticStreamBuilder, - Stores, NODE_ID, + test_ptr, MutexBlockStreamBuilder, NoopAdapterSelector, NoopRuntimeAdapter, + StaticBlockRefetcher, StaticStreamBuilder, Stores, TestChain, NODE_ID, }; use graph::blockchain::{BlockPtr, TriggersAdapterSelector}; use graph::cheap_clone::CheapClone; use graph::firehose::{FirehoseEndpoint, FirehoseEndpoints}; use graph::prelude::ethabi::ethereum_types::H256; -use graph::prelude::{LightEthereumBlock, LoggerFactory, NodeId}; +use graph::prelude::web3::types::{Address, Log, Transaction, H160}; +use graph::prelude::{ethabi, tiny_keccak, LightEthereumBlock, LoggerFactory, NodeId}; use graph::{blockchain::block_stream::BlockWithTriggers, prelude::ethabi::ethereum_types::U64}; use graph_chain_ethereum::network::EthereumNetworkAdapters; use graph_chain_ethereum::{ @@ -24,7 +25,7 @@ pub async fn chain( blocks: Vec>, stores: &Stores, triggers_adapter: Option>>, -) -> Chain { +) -> TestChain { let triggers_adapter = triggers_adapter.unwrap_or(Arc::new(NoopAdapterSelector { triggers_in_block_sleep: Duration::ZERO, x: PhantomData, @@ -47,7 +48,10 @@ pub async fn chain( ))] .into(); - Chain::new( + let static_block_stream = Arc::new(StaticStreamBuilder { chain: blocks }); + let block_stream_builder = Arc::new(MutexBlockStreamBuilder(Mutex::new(static_block_stream))); + + let chain = Chain::new( logger_factory.clone(), stores.network_name.clone(), node_id, @@ -57,14 +61,19 @@ pub async fn chain( firehose_endpoints, EthereumNetworkAdapters { adapters: vec![] }, stores.chain_head_listener.cheap_clone(), - Arc::new(StaticStreamBuilder { chain: blocks }), + block_stream_builder.clone(), Arc::new(StaticBlockRefetcher { x: PhantomData }), triggers_adapter, Arc::new(NoopRuntimeAdapter { x: PhantomData }), ENV_VARS.reorg_threshold, // We assume the tested chain is always ingestible for now true, - ) + ); + + TestChain { + chain: Arc::new(chain), + block_stream_builder, + } } pub fn genesis() -> BlockWithTriggers { @@ -86,13 +95,44 @@ pub fn empty_block( assert!(ptr != parent_ptr); assert!(ptr.number > parent_ptr.number); + // A 0x000.. transaction is used so `push_test_log` can use it + let transactions = vec![Transaction { + hash: H256::zero(), + block_hash: Some(H256::from_slice(ptr.hash.as_slice().into())), + block_number: Some(ptr.number.into()), + transaction_index: Some(0.into()), + from: Some(H160::zero()), + to: Some(H160::zero()), + ..Default::default() + }]; + BlockWithTriggers:: { block: BlockFinality::Final(Arc::new(LightEthereumBlock { hash: Some(H256::from_slice(ptr.hash.as_slice())), number: Some(U64::from(ptr.number)), parent_hash: H256::from_slice(parent_ptr.hash.as_slice()), + transactions, ..Default::default() })), trigger_data: vec![EthereumTrigger::Block(ptr, EthereumBlockTriggerType::Every)], } } + +pub fn push_test_log(block: &mut BlockWithTriggers, payload: impl Into) { + block.trigger_data.push(EthereumTrigger::Log( + Arc::new(Log { + address: Address::zero(), + topics: vec![tiny_keccak::keccak256(b"TestEvent(string)").into()], + data: ethabi::encode(&[ethabi::Token::String(payload.into())]).into(), + block_hash: Some(H256::from_slice(block.ptr().hash.as_slice())), + block_number: Some(block.ptr().number.into()), + transaction_hash: Some(H256::from_low_u64_be(0).into()), + transaction_index: Some(0.into()), + log_index: Some(0.into()), + transaction_log_index: Some(0.into()), + log_type: None, + removed: None, + }), + None, + )) +} diff --git a/tests/tests/runner.rs b/tests/tests/runner.rs index a5385a81136..52b422da03b 100644 --- a/tests/tests/runner.rs +++ b/tests/tests/runner.rs @@ -3,6 +3,7 @@ use std::sync::atomic::{self, AtomicBool}; use std::sync::Arc; use std::time::Duration; +use assert_json_diff::assert_json_eq; use graph::blockchain::block_stream::BlockWithTriggers; use graph::blockchain::{Block, BlockPtr, Blockchain}; use graph::data::subgraph::schema::{SubgraphError, SubgraphHealth}; @@ -11,8 +12,10 @@ use graph::env::EnvVars; use graph::object; use graph::prelude::ethabi::ethereum_types::H256; use graph::prelude::{CheapClone, SubgraphAssignmentProvider, SubgraphName, SubgraphStore}; -use graph_tests::fixture::ethereum::{chain, empty_block, genesis}; -use graph_tests::fixture::{self, stores, test_ptr, MockAdapterSelector, NoopAdapterSelector}; +use graph_tests::fixture::ethereum::{chain, empty_block, genesis, push_test_log}; +use graph_tests::fixture::{ + self, stores, test_ptr, test_ptr_reorged, MockAdapterSelector, NoopAdapterSelector, +}; use slog::{o, Discard, Logger}; #[tokio::test] @@ -39,16 +42,8 @@ async fn data_source_revert() -> anyhow::Result<()> { vec![block0, block1, block1_reorged, block2, block3, block4] }; - let chain = Arc::new(chain(blocks.clone(), &stores, None).await); - let ctx = fixture::setup( - subgraph_name.clone(), - &hash, - &stores, - chain.clone(), - None, - None, - ) - .await; + let chain = chain(blocks.clone(), &stores, None).await; + let ctx = fixture::setup(subgraph_name.clone(), &hash, &stores, &chain, None, None).await; let stop_block = test_ptr(2); ctx.start_and_sync_to(stop_block).await; @@ -70,7 +65,7 @@ async fn data_source_revert() -> anyhow::Result<()> { subgraph_name.clone(), &hash, &stores, - chain, + &chain, graft_block, None, ) @@ -119,8 +114,8 @@ async fn typename() -> anyhow::Result<()> { let stop_block = blocks.last().unwrap().block.ptr(); let stores = stores("./integration-tests/config.simple.toml").await; - let chain = Arc::new(chain(blocks, &stores, None).await); - let ctx = fixture::setup(subgraph_name.clone(), &hash, &stores, chain, None, None).await; + let chain = chain(blocks, &stores, None).await; + let ctx = fixture::setup(subgraph_name.clone(), &hash, &stores, &chain, None, None).await; ctx.start_and_sync_to(stop_block).await; @@ -143,10 +138,10 @@ async fn file_data_sources() { let block_2 = empty_block(block_1.ptr(), test_ptr(2)); let block_3 = empty_block(block_2.ptr(), test_ptr(3)); let block_4 = empty_block(block_3.ptr(), test_ptr(4)); - let block_5 = empty_block(block_4.ptr(), test_ptr(5)); + let mut block_5 = empty_block(block_4.ptr(), test_ptr(5)); + push_test_log(&mut block_5, "createFile2"); vec![block_0, block_1, block_2, block_3, block_4, block_5] }; - let stop_block = test_ptr(1); // This test assumes the file data sources will be processed in the same block in which they are // created. But the test might fail due to a race condition if for some reason it takes longer @@ -155,31 +150,28 @@ async fn file_data_sources() { // attempt to ensure the monitor has enough time to fetch the file. let adapter_selector = NoopAdapterSelector { x: PhantomData, - triggers_in_block_sleep: Duration::from_millis(100), + triggers_in_block_sleep: Duration::from_millis(150), }; - let chain = Arc::new(chain(blocks, &stores, Some(Arc::new(adapter_selector))).await); - let ctx = fixture::setup(subgraph_name.clone(), &hash, &stores, chain, None, None).await; - ctx.start_and_sync_to(stop_block).await; - - // CID QmVkvoPGi9jvvuxsHDVJDgzPEzagBaWSZRYoRDzU244HjZ is the file - // `file-data-sources/abis/Contract.abi` after being processed by graph-cli. - let id = "QmVkvoPGi9jvvuxsHDVJDgzPEzagBaWSZRYoRDzU244HjZ"; - + let chain = chain(blocks.clone(), &stores, Some(Arc::new(adapter_selector))).await; + let ctx = fixture::setup(subgraph_name.clone(), &hash, &stores, &chain, None, None).await; + ctx.start_and_sync_to(test_ptr(1)).await; + + // CID of `file-data-sources/abis/Contract.abi` after being processed by graph-cli. + let id = "QmQ2REmceVtzawp7yrnxLQXgNNCtFHEnig6fL9aqE1kcWq"; + let content_bytes = ctx.ipfs.cat_all(id, Duration::from_secs(10)).await.unwrap(); + let content = String::from_utf8(content_bytes.into()).unwrap(); let query_res = ctx .query(&format!(r#"{{ ipfsFile(id: "{id}") {{ id, content }} }}"#,)) .await .unwrap(); - assert_eq!( + assert_json_eq!( query_res, - Some(object! { ipfsFile: object!{ id: id.clone() , content: "[]" } }) + Some(object! { ipfsFile: object!{ id: id.clone() , content: content.clone() } }) ); // assert whether duplicate data sources are created. - ctx.provider.stop(ctx.deployment.clone()).await.unwrap(); - let stop_block = test_ptr(2); - - ctx.start_and_sync_to(stop_block).await; + ctx.start_and_sync_to(test_ptr(2)).await; let store = ctx.store.cheap_clone(); let writable = store @@ -189,24 +181,19 @@ async fn file_data_sources() { let datasources = writable.load_dynamic_data_sources(vec![]).await.unwrap(); assert!(datasources.len() == 1); - ctx.provider.stop(ctx.deployment.clone()).await.unwrap(); - let stop_block = test_ptr(3); - ctx.start_and_sync_to(stop_block).await; + ctx.start_and_sync_to(test_ptr(3)).await; let query_res = ctx .query(&format!(r#"{{ ipfsFile1(id: "{id}") {{ id, content }} }}"#,)) .await .unwrap(); - assert_eq!( + assert_json_eq!( query_res, - Some(object! { ipfsFile1: object!{ id: id , content: "[]" } }) + Some(object! { ipfsFile1: object!{ id: id , content: content } }) ); - ctx.provider.stop(ctx.deployment.clone()).await.unwrap(); - let stop_block = test_ptr(4); - ctx.start_and_sync_to(stop_block).await; - ctx.provider.stop(ctx.deployment.clone()).await.unwrap(); + ctx.start_and_sync_to(test_ptr(4)).await; let writable = ctx .store .clone() @@ -224,17 +211,72 @@ async fn file_data_sources() { } let stop_block = test_ptr(5); - let err = ctx.start_and_sync_to_error(stop_block).await; + let err = ctx.start_and_sync_to_error(stop_block.clone()).await; let message = "entity type `IpfsFile1` is not on the 'entities' list for data source `File2`. \ - Hint: Add `IpfsFile1` to the 'entities' list, which currently is: `IpfsFile`.\twasm backtrace:\t 0: 0x33bf - !src/mapping/handleFile1\t in handler `handleFile1` at block #5 ()".to_string(); + Hint: Add `IpfsFile1` to the 'entities' list, which currently is: `IpfsFile`.\twasm backtrace:\t 0: 0x365d - !src/mapping/handleFile1\t in handler `handleFile1` at block #5 ()".to_string(); let expected_err = SubgraphError { subgraph_id: ctx.deployment.hash.clone(), message, - block_ptr: Some(test_ptr(5)), + block_ptr: Some(stop_block), handler: None, deterministic: false, }; assert_eq!(err, expected_err); + + // Unfail the subgraph to test a conflict between an onchain and offchain entity + { + ctx.rewind(test_ptr(4)); + + // Replace block number 5 with one that contains a different event + let mut blocks = blocks.clone(); + blocks.pop(); + let block_5_1_ptr = test_ptr_reorged(5, 1); + let mut block_5_1 = empty_block(test_ptr(4), block_5_1_ptr.clone()); + push_test_log(&mut block_5_1, "saveConflictingEntity"); + blocks.push(block_5_1); + + chain.set_block_stream(blocks); + + // Errors in the store pipeline can be observed by using the runner directly. + let runner = ctx.runner(block_5_1_ptr.clone()).await; + let err = runner + .run() + .await + .err() + .unwrap_or_else(|| panic!("subgraph ran successfully but an error was expected")); + + let message = + "store error: conflicting key value violates exclusion constraint \"ipfs_file_id_block_range_excl\"" + .to_string(); + assert_eq!(err.to_string(), message); + } + + // Unfail the subgraph to test a conflict between an onchain and offchain entity + { + // Replace block number 5 with one that contains a different event + let mut blocks = blocks.clone(); + blocks.pop(); + let block_5_2_ptr = test_ptr_reorged(5, 2); + let mut block_5_2 = empty_block(test_ptr(4), block_5_2_ptr.clone()); + push_test_log(&mut block_5_2, "createFile1"); + blocks.push(block_5_2); + + chain.set_block_stream(blocks); + + // Errors in the store pipeline can be observed by using the runner directly. + let err = ctx + .runner(block_5_2_ptr.clone()) + .await + .run() + .await + .err() + .unwrap_or_else(|| panic!("subgraph ran successfully but an error was expected")); + + let message = + "store error: conflicting key value violates exclusion constraint \"ipfs_file_1_id_block_range_excl\"" + .to_string(); + assert_eq!(err.to_string(), message); + } } #[tokio::test] @@ -254,7 +296,7 @@ async fn template_static_filters_false_positives() { vec![block_0, block_1, block_2] }; let stop_block = test_ptr(1); - let chain = Arc::new(chain(blocks, &stores, None).await); + let chain = chain(blocks, &stores, None).await; let mut env_vars = EnvVars::default(); env_vars.experimental_static_filters = true; @@ -263,7 +305,7 @@ async fn template_static_filters_false_positives() { subgraph_name.clone(), &hash, &stores, - chain, + &chain, None, Some(env_vars), ) @@ -329,7 +371,7 @@ async fn retry_create_ds() { triggers_in_block_sleep: Duration::ZERO, triggers_in_block, }); - let chain = Arc::new(chain(blocks, &stores, Some(triggers_adapter)).await); + let chain = chain(blocks, &stores, Some(triggers_adapter)).await; let mut env_vars = EnvVars::default(); env_vars.subgraph_error_retry_ceil = Duration::from_secs(1); @@ -338,7 +380,7 @@ async fn retry_create_ds() { subgraph_name.clone(), &hash, &stores, - chain, + &chain, None, Some(env_vars), ) @@ -373,8 +415,8 @@ async fn fatal_error() -> anyhow::Result<()> { let stop_block = blocks.last().unwrap().block.ptr(); let stores = stores("./integration-tests/config.simple.toml").await; - let chain = Arc::new(chain(blocks, &stores, None).await); - let ctx = fixture::setup(subgraph_name.clone(), &hash, &stores, chain, None, None).await; + let chain = chain(blocks, &stores, None).await; + let ctx = fixture::setup(subgraph_name.clone(), &hash, &stores, &chain, None, None).await; ctx.start_and_sync_to_error(stop_block).await; @@ -387,7 +429,7 @@ async fn fatal_error() -> anyhow::Result<()> { assert!(err.deterministic); // Test that rewind unfails the subgraph. - ctx.store.rewind(ctx.deployment.hash.clone(), test_ptr(1))?; + ctx.rewind(test_ptr(1)); let status = ctx.indexing_status().await; assert!(status.health == SubgraphHealth::Healthy); assert!(status.fatal_error.is_none());