core, tests: Remove IndexingContext.filter

There's no real reason to hang on to the filter after the block stream has
been constructed

Author: lutter

core/src/subgraph/context/mod.rs

@@ -6,7 +6,7 @@ use crate::polling_monitor::{
 use anyhow::{self, Error};
 use bytes::Bytes;
 use graph::{
-    blockchain::{Blockchain, TriggerFilterWrapper},
+    blockchain::Blockchain,
     components::{store::DeploymentId, subgraph::HostMetrics},
     data::subgraph::SubgraphManifest,
     data_source::{
@@ -74,7 +74,6 @@ where
     pub(crate) instance: SubgraphInstance<C, T>,
     pub instances: SubgraphKeepAlive,
     pub offchain_monitor: OffchainMonitor,
-    pub filter: Option<TriggerFilterWrapper<C>>,
     pub(crate) trigger_processor: Box<dyn TriggerProcessor<C, T>>,
     pub(crate) decoder: Box<Decoder<C, T>>,
 }
@@ -101,7 +100,6 @@ impl<C: Blockchain, T: RuntimeHostBuilder<C>> IndexingContext<C, T> {
             instance,
             instances,
             offchain_monitor,
-            filter: None,
             trigger_processor,
             decoder,
         }

core/src/subgraph/runner/mod.rs

@@ -223,15 +223,11 @@ where
         let block_stream_canceler = CancelGuard::new();
         let block_stream_cancel_handle = block_stream_canceler.handle();
         // TriggerFilter needs to be rebuilt eveytime the blockstream is restarted
-        self.ctx.filter = Some(self.build_filter());
+        let filter = self.build_filter();
 
-        let block_stream = new_block_stream(
-            &self.inputs,
-            self.ctx.filter.clone().unwrap(), // Safe to unwrap as we just called `build_filter` in the previous line
-            &self.metrics.subgraph,
-        )
-        .await?
-        .cancelable(&block_stream_canceler);
+        let block_stream = new_block_stream(&self.inputs, filter, &self.metrics.subgraph)
+            .await?
+            .cancelable(&block_stream_canceler);
 
         self.cancel_handle = Some(block_stream_cancel_handle);
 
@@ -262,7 +258,7 @@ where
     /// Returns the next state to transition to:
     /// - `Restarting` to start the block stream (normal case)
     /// - `Stopped` if the max end block was already reached
-    async fn initialize(&mut self) -> Result<RunnerState<C>, SubgraphRunnerError> {
+    async fn initialize(&self) -> Result<RunnerState<C>, SubgraphRunnerError> {
         self.update_deployment_synced_metric();
 
         // If a subgraph failed for deterministic reasons, before start indexing, we first

tests/tests/runner_tests.rs

@@ -448,7 +448,7 @@ async fn end_block() -> anyhow::Result<()> {
     ) {
         let runner = ctx.runner(block_ptr.clone()).await;
         let runner = runner.run_for_test(false).await.unwrap();
-        let filter = runner.context().filter.as_ref().unwrap();
+        let filter = runner.build_filter_for_test();
         let addresses = filter
             .chain_filter
             .log()