core: Extract TriggerRunner component for unified trigger processing

Eliminate the duplicated trigger processing loops in runner.rs by
extracting a TriggerRunner component that handles the execution of
triggers against runtime hosts. This component is now used for both
initial trigger processing and dynamic data source trigger processing.

This is Phase 1 of the runner refactor as described in
docs/plans/runner-refactor.md.

Author: lutter

core/src/subgraph/runner.rs core/src/subgraph/runner/mod.rscore/src/subgraph/runner.rs renamed to core/src/subgraph/runner/mod.rs

@@ -1,3 +1,5 @@
+mod trigger_runner;
+
 use crate::subgraph::context::IndexingContext;
 use crate::subgraph::error::{
     ClassifyErrorHelper as _, DetailHelper as _, NonDeterministicErrorHelper as _, ProcessingError,
@@ -42,6 +44,8 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 use std::vec;
 
+use self::trigger_runner::TriggerRunner;
+
 const MINUTE: Duration = Duration::from_secs(60);
 
 const SKIP_PTR_UPDATES_THRESHOLD: Duration = Duration::from_secs(60 * 5);
@@ -615,44 +619,32 @@ where
 
         // Match and decode all triggers in the block
         let hosts_filter = |trigger: &TriggerData<C>| self.ctx.instance.hosts_for_trigger(trigger);
-        let match_res = self
+        let runnables = self
             .match_and_decode_many(&logger, &block, triggers, hosts_filter)
             .await;
 
         // Process events one after the other, passing in entity operations
         // collected previously to every new event being processed
-        let mut res = Ok(block_state);
-        match match_res {
+        let trigger_runner = TriggerRunner::new(
+            self.ctx.trigger_processor.as_ref(),
+            &self.logger,
+            &self.metrics.subgraph,
+            &self.inputs.debug_fork,
+            self.inputs.instrument,
+        );
+        let res = match runnables {
             Ok(runnables) => {
-                for runnable in runnables {
-                    let process_res = self
-                        .ctx
-                        .trigger_processor
-                        .process_trigger(
-                            &self.logger,
-                            runnable.hosted_triggers,
-                            &block,
-                            res.unwrap(),
-                            &proof_of_indexing,
-                            &causality_region,
-                            &self.inputs.debug_fork,
-                            &self.metrics.subgraph,
-                            self.inputs.instrument,
-                        )
-                        .await
-                        .map_err(|e| e.add_trigger_context(&runnable.trigger));
-                    match process_res {
-                        Ok(state) => res = Ok(state),
-                        Err(e) => {
-                            res = Err(e);
-                            break;
-                        }
-                    }
-                }
-            }
-            Err(e) => {
-                res = Err(e);
+                trigger_runner
+                    .execute(
+                        &block,
+                        runnables,
+                        block_state,
+                        &proof_of_indexing,
+                        &causality_region,
+                    )
+                    .await
             }
+            Err(e) => Err(e),
         };
 
         match res {
@@ -751,43 +743,31 @@ where
                 let hosts_filter = |_: &'_ TriggerData<C>| -> Box<dyn Iterator<Item = _> + Send> {
                     Box::new(runtime_hosts.iter().map(Arc::as_ref))
                 };
-                let match_res: Result<Vec<_>, _> = self
+                let runnables = self
                     .match_and_decode_many(&logger, &block, triggers, hosts_filter)
                     .await;
 
-                let mut res = Ok(block_state);
-                match match_res {
+                let trigger_runner = TriggerRunner::new(
+                    self.ctx.trigger_processor.as_ref(),
+                    &self.logger,
+                    &self.metrics.subgraph,
+                    &self.inputs.debug_fork,
+                    self.inputs.instrument,
+                );
+                let res = match runnables {
                     Ok(runnables) => {
-                        for runnable in runnables {
-                            let process_res = self
-                                .ctx
-                                .trigger_processor
-                                .process_trigger(
-                                    &self.logger,
-                                    runnable.hosted_triggers,
-                                    &block,
-                                    res.unwrap(),
-                                    &proof_of_indexing,
-                                    &causality_region,
-                                    &self.inputs.debug_fork,
-                                    &self.metrics.subgraph,
-                                    self.inputs.instrument,
-                                )
-                                .await
-                                .map_err(|e| e.add_trigger_context(&runnable.trigger));
-                            match process_res {
-                                Ok(state) => res = Ok(state),
-                                Err(e) => {
-                                    res = Err(e);
-                                    break;
-                                }
-                            }
-                        }
-                    }
-                    Err(e) => {
-                        res = Err(e);
+                        trigger_runner
+                            .execute(
+                                &block,
+                                runnables,
+                                block_state,
+                                &proof_of_indexing,
+                                &causality_region,
+                            )
+                            .await
                     }
-                }
+                    Err(e) => Err(e),
+                };
 
                 block_state = res.map_err(|e| {
                     // This treats a `PossibleReorg` as an ordinary error which will fail the subgraph.

core/src/subgraph/runner/trigger_runner.rs

@@ -0,0 +1,79 @@
+use std::sync::Arc;
+
+use graph::blockchain::Blockchain;
+use graph::components::store::SubgraphFork;
+use graph::components::subgraph::{MappingError, SharedProofOfIndexing};
+use graph::components::trigger_processor::RunnableTriggers;
+use graph::prelude::{BlockState, RuntimeHostBuilder, SubgraphInstanceMetrics, TriggerProcessor};
+use graph::slog::Logger;
+
+/// Handles the execution of triggers against runtime hosts, accumulating state.
+///
+/// This component unifies the trigger processing loop that was previously duplicated
+/// for initial triggers and dynamically created data source triggers.
+pub struct TriggerRunner<'a, C: Blockchain, T: RuntimeHostBuilder<C>> {
+    processor: &'a dyn TriggerProcessor<C, T>,
+    logger: &'a Logger,
+    metrics: &'a Arc<SubgraphInstanceMetrics>,
+    debug_fork: &'a Option<Arc<dyn SubgraphFork>>,
+    instrument: bool,
+}
+
+impl<'a, C, T> TriggerRunner<'a, C, T>
+where
+    C: Blockchain,
+    T: RuntimeHostBuilder<C>,
+{
+    /// Create a new TriggerRunner with the given dependencies.
+    pub fn new(
+        processor: &'a dyn TriggerProcessor<C, T>,
+        logger: &'a Logger,
+        metrics: &'a Arc<SubgraphInstanceMetrics>,
+        debug_fork: &'a Option<Arc<dyn SubgraphFork>>,
+        instrument: bool,
+    ) -> Self {
+        Self {
+            processor,
+            logger,
+            metrics,
+            debug_fork,
+            instrument,
+        }
+    }
+
+    /// Execute a sequence of runnable triggers, accumulating state changes.
+    ///
+    /// Processes each trigger in order. If any trigger fails with a non-deterministic
+    /// error, processing stops and the error is returned. Deterministic errors are
+    /// accumulated in the block state.
+    pub async fn execute(
+        &self,
+        block: &Arc<C::Block>,
+        runnables: Vec<RunnableTriggers<'a, C>>,
+        block_state: BlockState,
+        proof_of_indexing: &SharedProofOfIndexing,
+        causality_region: &str,
+    ) -> Result<BlockState, MappingError> {
+        let mut state = block_state;
+
+        for runnable in runnables {
+            state = self
+                .processor
+                .process_trigger(
+                    self.logger,
+                    runnable.hosted_triggers,
+                    block,
+                    state,
+                    proof_of_indexing,
+                    causality_region,
+                    self.debug_fork,
+                    self.metrics,
+                    self.instrument,
+                )
+                .await
+                .map_err(|e| e.add_trigger_context(&runnable.trigger))?;
+        }
+
+        Ok(state)
+    }
+}

docs/plans/runner-refactor.md

@@ -211,10 +211,10 @@ Each phase is complete when:
 
 ### Phase 1: Extract TriggerRunner Component
 
-- [ ] Create `TriggerRunner` struct with execute method
-- [ ] Replace first trigger loop (lines 616-656)
-- [ ] Replace second trigger loop (lines 754-790)
-- [ ] Verify tests pass
+- [x] Create `TriggerRunner` struct with execute method
+- [x] Replace first trigger loop (lines 616-656)
+- [x] Replace second trigger loop (lines 754-790)
+- [x] Verify tests pass
 
 ### Phase 2: Define RunnerState Enum