all: Remove 'wasm block' processing

This was only used for substreams

Author: lutter

core/src/subgraph/context/instance/mod.rs

@@ -255,8 +255,4 @@ where
     pub fn hosts_len(&self) -> usize {
         self.onchain_hosts.len() + self.offchain_hosts.len()
     }
-
-    pub fn first_host(&self) -> Option<&Arc<T::Host>> {
-        self.onchain_hosts.hosts().first()
-    }
 }

core/src/subgraph/context/mod.rs

@@ -6,11 +6,8 @@ use crate::polling_monitor::{
 use anyhow::{self, Error};
 use bytes::Bytes;
 use graph::{
-    blockchain::{BlockTime, Blockchain, TriggerFilterWrapper},
-    components::{
-        store::{DeploymentId, SubgraphFork},
-        subgraph::{HostMetrics, MappingError, RuntimeHost as _, SharedProofOfIndexing},
-    },
+    blockchain::{Blockchain, TriggerFilterWrapper},
+    components::{store::DeploymentId, subgraph::HostMetrics},
     data::subgraph::SubgraphManifest,
     data_source::{
         causality_region::CausalityRegionSeq,
@@ -20,14 +17,13 @@ use graph::{
     derive::CheapClone,
     ipfs::IpfsContext,
     prelude::{
-        BlockNumber, BlockPtr, BlockState, CancelGuard, CheapClone, DeploymentHash,
-        MetricsRegistry, RuntimeHostBuilder, SubgraphCountMetric, SubgraphInstanceMetrics,
-        TriggerProcessor,
+        BlockNumber, CancelGuard, CheapClone, DeploymentHash, MetricsRegistry, RuntimeHostBuilder,
+        SubgraphCountMetric, TriggerProcessor,
     },
     slog::Logger,
 };
+use std::collections::HashMap;
 use std::sync::{Arc, RwLock};
-use std::{collections::HashMap, time::Instant};
 use tokio::sync::mpsc;
 
 use self::instance::SubgraphInstance;
@@ -109,59 +105,6 @@ impl<C: Blockchain, T: RuntimeHostBuilder<C>> IndexingContext<C, T> {
         }
     }
 
-    pub async fn process_block(
-        &self,
-        logger: &Logger,
-        block_ptr: BlockPtr,
-        block_time: BlockTime,
-        block_data: Box<[u8]>,
-        handler: String,
-        mut state: BlockState,
-        proof_of_indexing: &SharedProofOfIndexing,
-        causality_region: &str,
-        debug_fork: &Option<Arc<dyn SubgraphFork>>,
-        subgraph_metrics: &Arc<SubgraphInstanceMetrics>,
-        instrument: bool,
-    ) -> Result<BlockState, MappingError> {
-        let error_count = state.deterministic_errors.len();
-
-        proof_of_indexing.start_handler(causality_region);
-
-        let start = Instant::now();
-
-        // This flow is expected to have a single data source(and a corresponding host) which
-        // gets executed every block.
-        state = self
-            .instance
-            .first_host()
-            .expect("Expected this flow to have exactly one host")
-            .process_block(
-                logger,
-                block_ptr,
-                block_time,
-                block_data,
-                handler,
-                state,
-                proof_of_indexing.cheap_clone(),
-                debug_fork,
-                instrument,
-            )
-            .await?;
-
-        let elapsed = start.elapsed().as_secs_f64();
-        subgraph_metrics.observe_trigger_processing_duration(elapsed);
-
-        if state.deterministic_errors.len() != error_count {
-            assert!(state.deterministic_errors.len() == error_count + 1);
-
-            // If a deterministic error has happened, write a new
-            // ProofOfIndexingEvent::DeterministicError to the SharedProofOfIndexing.
-            proof_of_indexing.write_deterministic_error(logger, causality_region);
-        }
-
-        Ok(state)
-    }
-
     /// Removes data sources hosts with a creation block greater or equal to `reverted_block`, so
     /// that they are no longer candidates for `process_trigger`.
     ///

core/src/subgraph/runner.rs

@@ -47,7 +47,6 @@ const MINUTE: Duration = Duration::from_secs(60);
 const SKIP_PTR_UPDATES_THRESHOLD: Duration = Duration::from_secs(60 * 5);
 const HANDLE_REVERT_SECTION_NAME: &str = "handle_revert";
 const PROCESS_BLOCK_SECTION_NAME: &str = "process_block";
-const PROCESS_WASM_BLOCK_SECTION_NAME: &str = "process_wasm_block";
 const PROCESS_TRIGGERS_SECTION_NAME: &str = "process_triggers";
 const HANDLE_CREATED_DS_SECTION_NAME: &str = "handle_new_data_sources";
 
@@ -869,37 +868,6 @@ where
         Ok(Arc::new(block))
     }
 
-    async fn process_wasm_block(
-        &mut self,
-        proof_of_indexing: &SharedProofOfIndexing,
-        block_ptr: BlockPtr,
-        block_time: BlockTime,
-        block_data: Box<[u8]>,
-        handler: String,
-        causality_region: &str,
-    ) -> Result<BlockState, MappingError> {
-        let block_state = BlockState::new(
-            self.inputs.store.clone(),
-            std::mem::take(&mut self.state.entity_lfu_cache),
-        );
-
-        self.ctx
-            .process_block(
-                &self.logger,
-                block_ptr,
-                block_time,
-                block_data,
-                handler,
-                block_state,
-                proof_of_indexing,
-                causality_region,
-                &self.inputs.debug_fork,
-                &self.metrics.subgraph,
-                self.inputs.instrument,
-            )
-            .await
-    }
-
     fn create_dynamic_data_sources(
         &mut self,
         created_data_sources: Vec<InstanceDSTemplateInfo>,
@@ -1171,20 +1139,6 @@ where
     ) -> Result<Action, Error> {
         let stopwatch = &self.metrics.stream.stopwatch;
         let action = match event {
-            Some(Ok(BlockStreamEvent::ProcessWasmBlock(
-                block_ptr,
-                block_time,
-                data,
-                handler,
-                cursor,
-            ))) => {
-                let _section = stopwatch.start_section(PROCESS_WASM_BLOCK_SECTION_NAME);
-                let res = self
-                    .handle_process_wasm_block(block_ptr.clone(), block_time, data, handler, cursor)
-                    .await;
-                let start = Instant::now();
-                self.handle_action(start, block_ptr, res).await?
-            }
             Some(Ok(BlockStreamEvent::ProcessBlock(block, cursor))) => {
                 let _section = stopwatch.start_section(PROCESS_BLOCK_SECTION_NAME);
                 self.handle_process_block(block, cursor).await?
@@ -1335,82 +1289,6 @@ where
     C: Blockchain,
     T: RuntimeHostBuilder<C>,
 {
-    async fn handle_process_wasm_block(
-        &mut self,
-        block_ptr: BlockPtr,
-        block_time: BlockTime,
-        block_data: Box<[u8]>,
-        handler: String,
-        cursor: FirehoseCursor,
-    ) -> Result<Action, ProcessingError> {
-        let logger = self.logger.new(o!(
-                "block_number" => format!("{:?}", block_ptr.number),
-                "block_hash" => format!("{}", block_ptr.hash)
-        ));
-
-        debug!(logger, "Start processing wasm block";);
-
-        self.metrics
-            .stream
-            .deployment_head
-            .set(block_ptr.number as f64);
-
-        let proof_of_indexing =
-            SharedProofOfIndexing::new(block_ptr.number, self.inputs.poi_version);
-
-        // Causality region for onchain triggers.
-        let causality_region = PoICausalityRegion::from_network(&self.inputs.network);
-
-        let block_state = {
-            match self
-                .process_wasm_block(
-                    &proof_of_indexing,
-                    block_ptr.clone(),
-                    block_time,
-                    block_data,
-                    handler,
-                    &causality_region,
-                )
-                .await
-            {
-                // Triggers processed with no errors or with only deterministic errors.
-                Ok(block_state) => block_state,
-
-                // Some form of unknown or non-deterministic error ocurred.
-                Err(MappingError::Unknown(e)) => return Err(ProcessingError::Unknown(e).into()),
-                Err(MappingError::PossibleReorg(e)) => {
-                    info!(logger,
-                        "Possible reorg detected, retrying";
-                        "error" => format!("{:#}", e),
-                    );
-
-                    // In case of a possible reorg, we want this function to do nothing and restart the
-                    // block stream so it has a chance to detect the reorg.
-                    //
-                    // The state is unchanged at this point, except for having cleared the entity cache.
-                    // Losing the cache is a bit annoying but not an issue for correctness.
-                    //
-                    // See also b21fa73b-6453-4340-99fb-1a78ec62efb1.
-                    return Ok(Action::Restart);
-                }
-            }
-        };
-
-        self.transact_block_state(
-            &logger,
-            block_ptr.clone(),
-            cursor.clone(),
-            block_time,
-            block_state,
-            proof_of_indexing,
-            vec![],
-            vec![],
-        )
-        .await?;
-
-        Ok(Action::Continue)
-    }
-
     async fn handle_process_block(
         &mut self,
         block: BlockWithTriggers<C>,

graph/src/blockchain/block_stream.rs

@@ -10,7 +10,7 @@ use std::sync::Arc;
 use thiserror::Error;
 use tokio::sync::mpsc::{self, Receiver, Sender};
 
-use super::{Block, BlockPtr, BlockTime, Blockchain, Trigger, TriggerFilterWrapper};
+use super::{Block, BlockPtr, Blockchain, Trigger, TriggerFilterWrapper};
 use crate::anyhow::Result;
 use crate::components::store::{BlockNumber, DeploymentLocator, SourceableStore};
 use crate::data::subgraph::UnifiedMappingApiVersion;
@@ -721,7 +721,6 @@ pub enum BlockStreamEvent<C: Blockchain> {
     Revert(BlockPtr, FirehoseCursor),
 
     ProcessBlock(BlockWithTriggers<C>, FirehoseCursor),
-    ProcessWasmBlock(BlockPtr, BlockTime, Box<[u8]>, String, FirehoseCursor),
 }
 
 #[derive(Clone)]

graph/src/components/subgraph/host.rs

@@ -6,7 +6,6 @@ use anyhow::Error;
 use async_trait::async_trait;
 use futures01::sync::mpsc;
 
-use crate::blockchain::BlockTime;
 use crate::components::metrics::gas::GasMetrics;
 use crate::components::store::SubgraphFork;
 use crate::data_source::{
@@ -71,19 +70,6 @@ pub trait RuntimeHost<C: Blockchain>: Send + Sync + 'static {
         logger: &Logger,
     ) -> Result<Option<TriggerWithHandler<MappingTrigger<C>>>, Error>;
 
-    async fn process_block(
-        &self,
-        logger: &Logger,
-        block_ptr: BlockPtr,
-        block_time: BlockTime,
-        block_data: Box<[u8]>,
-        handler: String,
-        state: BlockState,
-        proof_of_indexing: SharedProofOfIndexing,
-        debug_fork: &Option<Arc<dyn SubgraphFork>>,
-        instrument: bool,
-    ) -> Result<BlockState, MappingError>;
-
     async fn process_mapping_trigger(
         &self,
         logger: &Logger,