Update block handler input data

core/src/subgraph/instance.rs

@@ -135,7 +135,7 @@ where
     async fn process_trigger(
         &self,
         logger: &Logger,
-        block: &Arc<LightEthereumBlock>,
+        block: &Arc<EthereumBlockType>,
         trigger: EthereumTrigger,
         state: BlockState,
         proof_of_indexing: SharedProofOfIndexing,
@@ -154,7 +154,7 @@ where
     async fn process_trigger_in_runtime_hosts(
         logger: &Logger,
         hosts: &[Arc<T::Host>],
-        block: &Arc<LightEthereumBlock>,
+        block: &Arc<EthereumBlockType>,
         trigger: EthereumTrigger,
         mut state: BlockState,
         proof_of_indexing: SharedProofOfIndexing,
@@ -165,12 +165,11 @@ where
 
                 let transaction = block
                     .transaction_for_log(&log)
-                    .map(Arc::new)
+                    .map(|tx| Arc::new(tx.clone()))
                     .context("Found no transaction for event")?;
                 let matching_hosts = hosts.iter().filter(|host| host.matches_log(&log));
                 // Process the log in each host in the same order the corresponding data
                 // sources appear in the subgraph manifest
-                let transaction = Arc::new(transaction);
                 for host in matching_hosts {
                     state = host
                         .process_log(
@@ -189,8 +188,8 @@ where
 
                 let transaction = block
                     .transaction_for_call(&call)
+                    .map(|tx| Arc::new(tx.clone()))
                     .context("Found no transaction for call")?;
-                let transaction = Arc::new(transaction);
                 let matching_hosts = hosts.iter().filter(|host| host.matches_call(&call));
 
                 for host in matching_hosts {
@@ -206,16 +205,16 @@ where
                         .await?;
                 }
             }
-            EthereumTrigger::Block(ptr, trigger_type) => {
+            EthereumTrigger::Block(ptr, block_trigger_) => {
                 let matching_hosts = hosts
                     .iter()
-                    .filter(|host| host.matches_block(&trigger_type, ptr.number));
+                    .filter(|host| host.matches_block(&block_trigger_.trigger_type, ptr.number));
                 for host in matching_hosts {
                     state = host
                         .process_block(
                             logger,
                             block,
-                            &trigger_type,
+                            &block_trigger_,
                             state,
                             proof_of_indexing.cheap_clone(),
                         )

core/src/subgraph/instance_manager.rs

@@ -638,7 +638,7 @@ where
     }
 
     // Obtain current and new block pointer (after this block is processed)
-    let light_block = Arc::new(block.light_block());
+    let light_block = Arc::new(EthereumBlockType::Light(block.light_block()));
     let block_ptr_after = EthereumBlockPointer::from(&block);
     let block_ptr_for_new_data_sources = block_ptr_after.clone();
 
@@ -879,7 +879,7 @@ async fn process_triggers<B: BlockStreamBuilder, T: RuntimeHostBuilder, S: Send
     mut block_state: BlockState,
     proof_of_indexing: SharedProofOfIndexing,
     ctx: IndexingContext<B, T, S>,
-    block: &Arc<LightEthereumBlock>,
+    block: &Arc<EthereumBlockType>,
     triggers: Vec<EthereumTrigger>,
 ) -> Result<(IndexingContext<B, T, S>, BlockState), CancelableError<Error>> {
     for trigger in triggers.into_iter() {

graph/src/components/ethereum/adapter.rs

@@ -411,6 +411,7 @@ impl From<EthereumBlockFilter> for EthereumCallFilter {
 pub struct EthereumBlockFilter {
     pub contract_addresses: HashSet<(u64, Address)>,
     pub trigger_every_block: bool,
+    pub block_type: BlockType,
 }
 
 impl EthereumBlockFilter {
@@ -421,8 +422,7 @@ impl EthereumBlockFilter {
                 let has_block_handler_with_call_filter = data_source
                     .mapping
                     .block_handlers
-                    .clone()
-                    .into_iter()
+                    .iter()
                     .any(|block_handler| match block_handler.filter {
                         Some(ref filter) if *filter == BlockHandlerFilter::Call => return true,
                         _ => return false,
@@ -431,12 +431,21 @@ impl EthereumBlockFilter {
                 let has_block_handler_without_filter = data_source
                     .mapping
                     .block_handlers
-                    .clone()
-                    .into_iter()
+                    .iter()
                     .any(|block_handler| block_handler.filter.is_none());
 
+                let block_type = data_source
+                    .mapping
+                    .block_handlers
+                    .iter()
+                    .map(|handler| handler.block_format)
+                    .max()
+                    .unwrap_or(EthereumBlockHandlerData::Block)
+                    .into();
+
                 filter_opt.extend(Self {
                     trigger_every_block: has_block_handler_without_filter,
+                    block_type,
                     contract_addresses: if has_block_handler_with_call_filter {
                         vec![(
                             data_source.source.start_block,
@@ -454,6 +463,8 @@ impl EthereumBlockFilter {
 
     pub fn extend(&mut self, other: EthereumBlockFilter) {
         self.trigger_every_block = self.trigger_every_block || other.trigger_every_block;
+        self.block_type = self.block_type.max(other.block_type);
+
         self.contract_addresses = self.contract_addresses.iter().cloned().fold(
             HashSet::new(),
             |mut addresses, (start_block, address)| {
@@ -803,20 +814,30 @@ fn parse_block_triggers(
 ) -> Vec<EthereumTrigger> {
     let block_ptr = EthereumBlockPointer::from(&block.ethereum_block);
     let trigger_every_block = block_filter.trigger_every_block;
+    let block_type = block_filter.block_type;
     let call_filter = EthereumCallFilter::from(block_filter);
     let mut triggers = block.calls.as_ref().map_or(vec![], |calls| {
         calls
             .iter()
             .filter(move |call| call_filter.matches(call))
             .map(move |call| {
-                EthereumTrigger::Block(block_ptr, EthereumBlockTriggerType::WithCallTo(call.to))
+                EthereumTrigger::Block(
+                    block_ptr,
+                    EthereumBlockTrigger {
+                        block_type,
+                        trigger_type: EthereumBlockTriggerType::WithCallTo(call.to),
+                    },
+                )
             })
             .collect::<Vec<EthereumTrigger>>()
     });
     if trigger_every_block {
         triggers.push(EthereumTrigger::Block(
             block_ptr,
-            EthereumBlockTriggerType::Every,
+            EthereumBlockTrigger {
+                block_type,
+                trigger_type: EthereumBlockTriggerType::Every,
+            },
         ));
     }
     triggers
@@ -921,21 +942,33 @@ pub fn blocks_with_triggers(
                 .block_range_to_ptrs(logger.clone(), from, to)
                 .map(move |ptrs| {
                     ptrs.into_iter()
-                        .map(|ptr| EthereumTrigger::Block(ptr, EthereumBlockTriggerType::Every))
+                        .map(|ptr| {
+                            EthereumTrigger::Block(
+                                ptr,
+                                EthereumBlockTrigger {
+                                    block_type: block_filter.block_type,
+                                    trigger_type: EthereumBlockTriggerType::Every,
+                                },
+                            )
+                        })
                         .collect()
                 }),
         ))
     } else if !block_filter.contract_addresses.is_empty() {
         // To determine which blocks include a call to addresses
         // in the block filter, transform the `block_filter` into
         // a `call_filter` and run `blocks_with_calls`
+        let block_type = block_filter.block_type;
         let call_filter = EthereumCallFilter::from(block_filter);
         trigger_futs.push(Box::new(
             eth.calls_in_block_range(&logger, subgraph_metrics.clone(), from, to, call_filter)
-                .map(|call| {
+                .map(move |call| {
                     EthereumTrigger::Block(
                         EthereumBlockPointer::from(&call),
-                        EthereumBlockTriggerType::WithCallTo(call.to),
+                        EthereumBlockTrigger {
+                            block_type,
+                            trigger_type: EthereumBlockTriggerType::WithCallTo(call.to),
+                        },
                     )
                 })
                 .collect(),

graph/src/components/ethereum/mod.rs

@@ -13,8 +13,10 @@ pub use self::adapter::{
 pub use self::listener::{ChainHeadUpdate, ChainHeadUpdateListener, ChainHeadUpdateStream};
 pub use self::stream::{BlockStream, BlockStreamBuilder, BlockStreamEvent};
 pub use self::types::{
-    BlockFinality, EthereumBlock, EthereumBlockData, EthereumBlockPointer,
-    EthereumBlockTriggerType, EthereumBlockWithCalls, EthereumBlockWithTriggers, EthereumCall,
-    EthereumCallData, EthereumEventData, EthereumTransactionData, EthereumTrigger,
-    LightEthereumBlock, LightEthereumBlockExt,
+    BlockFinality, BlockType, EthereumBlock, EthereumBlockData, EthereumBlockPointer,
+    EthereumBlockTrigger, EthereumBlockTriggerType, EthereumBlockType, EthereumBlockWithCalls,
+    EthereumBlockWithTriggers, EthereumCall, EthereumCallData, EthereumEventData,
+    EthereumTransactionData, EthereumTransactionReceiptData, EthereumTrigger,
+    FullEthereumBlockData, FullEthereumBlockDataWithReceipts, LightEthereumBlock,
+    LightEthereumBlockExt,
 };