graph: order subgraph triggers by source_idx first

Author: incrypto32

core/src/subgraph/runner.rs

@@ -146,6 +146,7 @@ where
                     .iter()
                     .map(|handler| handler.entity.clone())
                     .collect(),
+                manifest_idx: ds.manifest_idx,
             })
             .collect::<Vec<_>>();
 

graph/src/blockchain/block_stream.rs

@@ -357,11 +357,10 @@ impl<C: Blockchain> TriggersAdapterWrapper<C> {
             SubgraphTriggerScanRange::Range(from, to) => {
                 let hash_to_entities = self.fetch_entities_for_filters(filters, from, to).await?;
 
-                let block_numbers: BTreeSet<BlockNumber> = hash_to_entities
+                let block_numbers = hash_to_entities
                     .iter()
-                    .flat_map(|(_, entities)| entities.keys().copied())
-                    .chain(std::iter::once(to))
-                    .collect();
+                    .flat_map(|(_, entities, _)| entities.keys().copied())
+                    .collect::<BTreeSet<_>>();
 
                 let blocks = self
                     .adapter
@@ -384,6 +383,7 @@ impl<C: Blockchain> TriggersAdapterWrapper<C> {
         Vec<(
             DeploymentHash,
             BTreeMap<BlockNumber, Vec<EntitySourceOperation>>,
+            u32,
         )>,
         Error,
     > {
@@ -399,7 +399,7 @@ impl<C: Blockchain> TriggersAdapterWrapper<C> {
                         async move {
                             let entities =
                                 get_entities_for_range(&store, filter, &schema, from, to).await?;
-                            Ok::<_, Error>((filter.subgraph.clone(), entities))
+                            Ok::<_, Error>((filter.subgraph.clone(), entities, filter.manifest_idx))
                         }
                     })
             })
@@ -416,12 +416,14 @@ impl<C: Blockchain> TriggersAdapterWrapper<C> {
 fn create_subgraph_trigger_from_entities(
     subgraph: &DeploymentHash,
     entities: Vec<EntitySourceOperation>,
+    manifest_idx: u32,
 ) -> Vec<subgraph::TriggerData> {
     entities
         .into_iter()
         .map(|entity| subgraph::TriggerData {
             source: subgraph.clone(),
             entity,
+            source_idx: manifest_idx,
         })
         .collect()
 }
@@ -432,6 +434,7 @@ async fn create_subgraph_triggers<C: Blockchain>(
     subgraph_data: Vec<(
         DeploymentHash,
         BTreeMap<BlockNumber, Vec<EntitySourceOperation>>,
+        u32,
     )>,
 ) -> Result<Vec<BlockWithTriggers<C>>, Error> {
     let logger_clone = logger.cheap_clone();
@@ -441,10 +444,13 @@ async fn create_subgraph_triggers<C: Blockchain>(
             let block_number = block.number();
             let mut all_trigger_data = Vec::new();
 
-            for (hash, entities) in subgraph_data.iter() {
+            for (hash, entities, manifest_idx) in subgraph_data.iter() {
                 if let Some(block_entities) = entities.get(&block_number) {
-                    let trigger_data =
-                        create_subgraph_trigger_from_entities(hash, block_entities.clone());
+                    let trigger_data = create_subgraph_trigger_from_entities(
+                        hash,
+                        block_entities.clone(),
+                        *manifest_idx,
+                    );
                     all_trigger_data.extend(trigger_data);
                 }
             }

graph/src/blockchain/mod.rs

@@ -266,6 +266,7 @@ pub struct SubgraphFilter {
     pub subgraph: DeploymentHash,
     pub start_block: BlockNumber,
     pub entities: Vec<String>,
+    pub manifest_idx: u32,
 }
 
 impl<C: Blockchain> TriggerFilterWrapper<C> {
@@ -467,7 +468,7 @@ where
             (Trigger::Chain(data1), Trigger::Chain(data2)) => data1.cmp(data2),
             (Trigger::Subgraph(_), Trigger::Chain(_)) => std::cmp::Ordering::Greater,
             (Trigger::Chain(_), Trigger::Subgraph(_)) => std::cmp::Ordering::Less,
-            (Trigger::Subgraph(t1), Trigger::Subgraph(t2)) => t1.entity.vid.cmp(&t2.entity.vid),
+            (Trigger::Subgraph(t1), Trigger::Subgraph(t2)) => t1.cmp(t2),
         }
     }
 }

graph/src/data_source/subgraph.rs

@@ -354,18 +354,34 @@ pub struct MappingEntityTrigger {
 pub struct TriggerData {
     pub source: DeploymentHash,
     pub entity: EntitySourceOperation,
+    pub source_idx: u32,
 }
 
 impl TriggerData {
-    pub fn new(source: DeploymentHash, entity: EntitySourceOperation) -> Self {
-        Self { source, entity }
+    pub fn new(source: DeploymentHash, entity: EntitySourceOperation, source_idx: u32) -> Self {
+        Self { source, entity, source_idx }
     }
 
     pub fn entity_type(&self) -> &str {
         self.entity.entity_type.as_str()
     }
 }
 
+impl Ord for TriggerData {
+    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
+        match self.source_idx.cmp(&other.source_idx) {
+            std::cmp::Ordering::Equal => self.entity.vid.cmp(&other.entity.vid),
+            ord => ord,
+        }
+    }
+}
+
+impl PartialOrd for TriggerData {
+    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
+        Some(self.cmp(other))
+    }
+}
+
 impl fmt::Debug for TriggerData {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         write!(

graph/tests/subgraph_datasource_tests.rs

@@ -190,6 +190,7 @@ async fn test_triggers_adapter_with_entities() {
         subgraph: id,
         start_block: 0,
         entities: vec!["User".to_string()], // Only monitoring User entities
+        manifest_idx: 0,
     };
 
     let logger = Logger::root(slog::Discard, slog::o!());

tests/src/fixture/ethereum.rs

@@ -169,6 +169,7 @@ pub fn push_test_subgraph_trigger(
     entity_type: EntityType,
     entity_op: EntityOperationKind,
     vid: i64,
+    source_idx: u32,
 ) {
     let entity = EntitySourceOperation {
         entity: entity,
@@ -179,7 +180,7 @@ pub fn push_test_subgraph_trigger(
 
     block
         .trigger_data
-        .push(Trigger::Subgraph(subgraph::TriggerData { source, entity }));
+        .push(Trigger::Subgraph(subgraph::TriggerData { source, entity, source_idx }));
 }
 
 pub fn push_test_command(

tests/tests/runner_tests.rs

@@ -1122,6 +1122,7 @@ async fn subgraph_data_sources() {
             entity_type,
             EntityOperationKind::Create,
             1,
+            0,
         );
 
         let block_2 = empty_block(block_1.ptr(), test_ptr(2));