chain: Add debug log for duplicate triggers

Author: evaporei

chain/arweave/src/chain.rs

@@ -191,7 +191,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
 
     async fn triggers_in_block(
         &self,
-        _logger: &Logger,
+        logger: &Logger,
         block: codec::Block,
         filter: &TriggerFilter,
     ) -> Result<BlockWithTriggers<Chain>, Error> {
@@ -223,7 +223,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
             trigger_data.push(ArweaveTrigger::Block(shared_block.cheap_clone()));
         }
 
-        Ok(BlockWithTriggers::new(block, trigger_data))
+        Ok(BlockWithTriggers::new(block, trigger_data, logger))
     }
 
     async fn is_on_main_chain(&self, _ptr: BlockPtr) -> Result<bool, Error> {

chain/cosmos/src/chain.rs

@@ -197,7 +197,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
 
     async fn triggers_in_block(
         &self,
-        _logger: &Logger,
+        logger: &Logger,
         block: codec::Block,
         filter: &TriggerFilter,
     ) -> Result<BlockWithTriggers<Chain>, Error> {
@@ -277,7 +277,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
             triggers.push(CosmosTrigger::Block(shared_block.cheap_clone()));
         }
 
-        Ok(BlockWithTriggers::new(block, triggers))
+        Ok(BlockWithTriggers::new(block, triggers, logger))
     }
 
     async fn is_on_main_chain(&self, _ptr: BlockPtr) -> Result<bool, Error> {

chain/ethereum/src/chain.rs

@@ -533,7 +533,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
                 ));
                 triggers.append(&mut parse_call_triggers(&filter.call, &full_block)?);
                 triggers.append(&mut parse_block_triggers(&filter.block, &full_block));
-                Ok(BlockWithTriggers::new(block, triggers))
+                Ok(BlockWithTriggers::new(block, triggers, logger))
             }
         }
     }

chain/ethereum/src/ethereum_adapter.rs

@@ -1433,13 +1433,16 @@ pub(crate) async fn blocks_with_triggers(
     block_hashes.insert(to_hash);
     triggers_by_block.entry(to).or_insert(Vec::new());
 
+    let logger2 = logger.cheap_clone();
+
     let blocks = adapter
         .load_blocks(logger.cheap_clone(), chain_store.clone(), block_hashes)
         .and_then(
             move |block| match triggers_by_block.remove(&(block.number() as BlockNumber)) {
                 Some(triggers) => Ok(BlockWithTriggers::new(
                     BlockFinality::Final(block),
                     triggers,
+                    &logger2,
                 )),
                 None => Err(anyhow!(
                     "block {} not found in `triggers_by_block`",

chain/ethereum/src/tests.rs

@@ -6,6 +6,7 @@ use graph::{
         web3::types::{Address, Bytes, Log, H160, H256, U64},
         EthereumCall,
     },
+    slog::{self, o, Logger},
 };
 
 use crate::{
@@ -86,9 +87,14 @@ fn test_trigger_ordering() {
         log1.clone(),
     ];
 
+    let logger = Logger::root(slog::Discard, o!());
+
     // Test that `BlockWithTriggers` sorts the triggers.
-    let block_with_triggers =
-        BlockWithTriggers::<crate::Chain>::new(BlockFinality::Final(Default::default()), triggers);
+    let block_with_triggers = BlockWithTriggers::<crate::Chain>::new(
+        BlockFinality::Final(Default::default()),
+        triggers,
+        &logger,
+    );
 
     assert_eq!(
         block_with_triggers.trigger_data,

chain/near/src/chain.rs

@@ -235,7 +235,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
 
     async fn triggers_in_block(
         &self,
-        _logger: &Logger,
+        logger: &Logger,
         block: codec::Block,
         filter: &TriggerFilter,
     ) -> Result<BlockWithTriggers<Chain>, Error> {
@@ -292,7 +292,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
             trigger_data.push(NearTrigger::Block(shared_block.cheap_clone()));
         }
 
-        Ok(BlockWithTriggers::new(block, trigger_data))
+        Ok(BlockWithTriggers::new(block, trigger_data, logger))
     }
 
     async fn is_on_main_chain(&self, _ptr: BlockPtr) -> Result<bool, Error> {

chain/substreams/src/mapper.rs

@@ -16,7 +16,7 @@ pub struct Mapper {}
 impl SubstreamsMapper<Chain> for Mapper {
     async fn to_block_stream_event(
         &self,
-        _logger: &Logger,
+        logger: &Logger,
         block_scoped_data: &BlockScopedData,
     ) -> Result<Option<BlockStreamEvent<Chain>>, SubstreamsError> {
         let step = ForkStep::from_i32(block_scoped_data.step).unwrap_or_else(|| {
@@ -52,7 +52,7 @@ impl SubstreamsMapper<Chain> for Mapper {
 
                         // TODO(filipe): Fix once either trigger data can be empty
                         // or we move the changes into trigger data.
-                        BlockWithTriggers::new(changes, vec![TriggerData {}]),
+                        BlockWithTriggers::new(changes, vec![TriggerData {}], logger),
                         FirehoseCursor::from(cursor.clone()),
                     ))),
                     StepUndo => {

graph/src/blockchain/block_stream.rs

@@ -173,12 +173,29 @@ where
 }
 
 impl<C: Blockchain> BlockWithTriggers<C> {
-    pub fn new(block: C::Block, mut trigger_data: Vec<C::TriggerData>) -> Self {
+    pub fn new(block: C::Block, mut trigger_data: Vec<C::TriggerData>, logger: &Logger) -> Self {
         // This is where triggers get sorted.
         trigger_data.sort();
+
+        let old_len = trigger_data.len();
+
         // This is removing the duplicate triggers in the case of multiple
         // data sources fetching the same event/call/etc.
         trigger_data.dedup();
+
+        let new_len = trigger_data.len();
+
+        if new_len != old_len {
+            debug!(
+                logger,
+                "Trigger data had duplicate triggers";
+                "block_number" => block.number(),
+                "block_hash" => block.hash().hash_hex(),
+                "old_length" => old_len,
+                "new_length" => new_len,
+            );
+        }
+
         Self {
             block,
             trigger_data,

tests/src/fixture.rs

@@ -512,12 +512,12 @@ impl<C: Blockchain> TriggersAdapter<C> for NoopTriggersAdapter<C> {
 
     async fn triggers_in_block(
         &self,
-        _logger: &Logger,
+        logger: &Logger,
         block: <C as Blockchain>::Block,
         _filter: &<C as Blockchain>::TriggerFilter,
     ) -> Result<BlockWithTriggers<C>, Error> {
         // Return no triggers on data source reprocessing.
-        Ok(BlockWithTriggers::new(block, Vec::new()))
+        Ok(BlockWithTriggers::new(block, Vec::new(), logger))
     }
 
     async fn is_on_main_chain(&self, _ptr: BlockPtr) -> Result<bool, Error> {