From a2ed701a52078bb471466c0af939e5b0ce877600 Mon Sep 17 00:00:00 2001
From: Ash Kunda <18058966+akundaz@users.noreply.github.com>
Date: Thu, 8 Jan 2026 23:17:07 -0500
Subject: [PATCH 1/6] fix: remove unnecessary wakes

---
 src/pipelines/exec/mod.rs | 132 ++++++++++++++++++--------------------
 1 file changed, 64 insertions(+), 68 deletions(-)

diff --git a/src/pipelines/exec/mod.rs b/src/pipelines/exec/mod.rs
index 7f1f5d7..6297686 100644
--- a/src/pipelines/exec/mod.rs
+++ b/src/pipelines/exec/mod.rs
@@ -280,69 +280,63 @@ where
 
 		// The executor has not run any steps yet, it is invoking the `before_job`
 		// method of each step in the pipeline.
-		if let Cursor::Initializing(ref mut future) = executor.cursor {
-			if let Poll::Ready(output) = future.as_mut().poll_unpin(cx) {
-				match output {
-					Ok(checkpoint) => {
-						trace!("{} initialized successfully", executor.pipeline);
-						executor.cursor = executor.first_step(checkpoint);
-					}
-					Err(error) => {
-						trace!(
-							"{} initialization failed with error: {error:?}",
-							executor.pipeline
-						);
-						// If the initialization failed, we immediately finalize the
-						// pipeline with the error that occurred during initialization
-						// and not attempt to run any steps.
-						executor.cursor =
-							Cursor::Finalizing(executor.finalize(Err(error.into())));
-					}
+		if let Cursor::Initializing(ref mut future) = executor.cursor
+			&& let Poll::Ready(output) = future.as_mut().poll_unpin(cx)
+		{
+			match output {
+				Ok(checkpoint) => {
+					trace!("{} initialized successfully", executor.pipeline);
+					executor.cursor = executor.first_step(checkpoint);
+				}
+				Err(error) => {
+					trace!(
+						"{} initialization failed with error: {error:?}",
+						executor.pipeline
+					);
+					// If the initialization failed, we immediately finalize the
+					// pipeline with the error that occurred during initialization
+					// and not attempt to run any steps.
+					executor.cursor =
+						Cursor::Finalizing(executor.finalize(Err(error.into())));
 				}
-				trace!("{} initializing completed", executor.pipeline);
 			}
-
-			// tell the async runtime to poll again because we are still initializing
-			cx.waker().wake_by_ref();
+			trace!("{} initializing completed", executor.pipeline);
 		}
 
 		// the pipeline has completed executing all steps or encountered and error.
 		// Now we are running the `after_job` of each step in the pipeline.
-		if let Cursor::Finalizing(ref mut future) = executor.cursor {
-			if let Poll::Ready(output) = future.as_mut().poll_unpin(cx) {
-				trace!("{} completed with output: {output:#?}", executor.pipeline);
-
-				// Execution of this pipeline has completed, This resolves the
-				// executor future with the final output of the pipeline. Also
-				// emit an appropriate system event and record metrics.
-
-				let payload_id = executor.block.payload_id();
-				let events_bus = &executor.pipeline.events;
-				let metrics = executor.service.metrics();
-
-				match &output {
-					Ok(built_payload) => {
-						events_bus.publish(PayloadJobCompleted::<P> {
-							payload_id,
-							built_payload: built_payload.clone(),
-						});
-						metrics.jobs_completed.increment(1);
-						metrics.record_payload::<P>(built_payload, &executor.block);
-					}
-					Err(error) => {
-						events_bus.publish(PayloadJobFailed {
-							payload_id,
-							error: error.clone(),
-						});
-						metrics.jobs_failed.increment(1);
-					}
+		if let Cursor::Finalizing(ref mut future) = executor.cursor
+			&& let Poll::Ready(output) = future.as_mut().poll_unpin(cx)
+		{
+			trace!("{} completed with output: {output:#?}", executor.pipeline);
+
+			// Execution of this pipeline has completed, This resolves the
+			// executor future with the final output of the pipeline. Also
+			// emit an appropriate system event and record metrics.
+
+			let payload_id = executor.block.payload_id();
+			let events_bus = &executor.pipeline.events;
+			let metrics = executor.service.metrics();
+
+			match &output {
+				Ok(built_payload) => {
+					events_bus.publish(PayloadJobCompleted::<P> {
+						payload_id,
+						built_payload: built_payload.clone(),
+					});
+					metrics.jobs_completed.increment(1);
+					metrics.record_payload::<P>(built_payload, &executor.block);
+				}
+				Err(error) => {
+					events_bus.publish(PayloadJobFailed {
+						payload_id,
+						error: error.clone(),
+					});
+					metrics.jobs_failed.increment(1);
 				}
-
-				return Poll::Ready(output);
 			}
 
-			// tell the async runtime to poll again because we are still finalizing
-			cx.waker().wake_by_ref();
+			return Poll::Ready(output);
 		}
 
 		if matches!(executor.cursor, Cursor::BeforeStep(_, _)) {
@@ -359,23 +353,25 @@ where
 			trace!("{} will execute step {path}", executor.pipeline);
 			let future = executor.execute_step(&path, input);
 			executor.cursor = Cursor::StepInProgress(path, future);
-			cx.waker().wake_by_ref(); // tell the async runtime to poll again
-		}
 
-		if let Cursor::StepInProgress(ref path, ref mut future) = executor.cursor {
-			// If the cursor is in the StepInProgress state, we to poll the
-			// future instance of that step to see if it has completed.
-			if let Poll::Ready(output) = future.as_mut().poll_unpin(cx) {
-				trace!(
-					"{} step {path:?} completed with output: {output:#?}",
-					executor.pipeline
-				);
+			// This wake is necessary because we have a new future/at a state
+			// transition boundary and directly return Poll::Pending without polling
+			// it first, so there is no registered waker to drive progress without it
+			cx.waker().wake_by_ref();
+		}
 
-				// step has completed, we can advance the cursor
-				executor.cursor = executor.advance_cursor(path, output);
-			}
+		// If the cursor is in the StepInProgress state, we to poll the
+		// future instance of that step to see if it has completed.
+		if let Cursor::StepInProgress(ref path, ref mut future) = executor.cursor
+			&& let Poll::Ready(output) = future.as_mut().poll_unpin(cx)
+		{
+			trace!(
+				"{} step {path:?} completed with output: {output:#?}",
+				executor.pipeline
+			);
 
-			cx.waker().wake_by_ref(); // tell the async runtime to poll again
+			// step has completed, we can advance the cursor
+			executor.cursor = executor.advance_cursor(path, output);
 		}
 
 		Poll::Pending

From 52279ae428eb1e97e91c25dad48ac3e66715f889 Mon Sep 17 00:00:00 2001
From: Ash Kunda <18058966+akundaz@users.noreply.github.com>
Date: Wed, 14 Jan 2026 12:28:31 -0500
Subject: [PATCH 2/6] run a loop for the PipelineExecutor future

---
 src/pipelines/exec/mod.rs | 218 +++++++++++++++++++-------------------
 1 file changed, 111 insertions(+), 107 deletions(-)

diff --git a/src/pipelines/exec/mod.rs b/src/pipelines/exec/mod.rs
index 6297686..e16cf83 100644
--- a/src/pipelines/exec/mod.rs
+++ b/src/pipelines/exec/mod.rs
@@ -278,108 +278,128 @@ where
 	fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
 		let executor = self.get_mut();
 
-		// The executor has not run any steps yet, it is invoking the `before_job`
-		// method of each step in the pipeline.
-		if let Cursor::Initializing(ref mut future) = executor.cursor
-			&& let Poll::Ready(output) = future.as_mut().poll_unpin(cx)
-		{
-			match output {
-				Ok(checkpoint) => {
-					trace!("{} initialized successfully", executor.pipeline);
-					executor.cursor = executor.first_step(checkpoint);
+		loop {
+			match executor.cursor {
+				// The executor has not run any steps yet, it is invoking the
+				// `before_job` method of each step in the pipeline.
+				Cursor::Initializing(ref mut future) => {
+					match future.as_mut().poll_unpin(cx) {
+						Poll::Ready(Ok(checkpoint)) => {
+							trace!("{} initialized successfully", executor.pipeline);
+							executor.cursor = executor.first_step(checkpoint);
+						}
+						Poll::Ready(Err(error)) => {
+							trace!(
+								"{} initialization failed with error: {error:?}",
+								executor.pipeline
+							);
+							// If the initialization failed, we immediately finalize the
+							// pipeline with the error that occurred during initialization
+							// and not attempt to run any steps.
+							executor.cursor =
+								Cursor::Finalizing(executor.finalize(Err(error.into())));
+						}
+						Poll::Pending => return Poll::Pending,
+					}
 				}
-				Err(error) => {
-					trace!(
-						"{} initialization failed with error: {error:?}",
-						executor.pipeline
-					);
-					// If the initialization failed, we immediately finalize the
-					// pipeline with the error that occurred during initialization
-					// and not attempt to run any steps.
-					executor.cursor =
-						Cursor::Finalizing(executor.finalize(Err(error.into())));
+				// The pipeline has completed executing all steps or encountered an
+				// error. Now we are running the `after_job` of each step.
+				Cursor::Finalizing(ref mut future) => match future
+					.as_mut()
+					.poll_unpin(cx)
+				{
+					Poll::Ready(output) => {
+						trace!("{} completed with output: {output:#?}", executor.pipeline);
+
+						// Execution of this pipeline has completed, This resolves the
+						// executor future with the final output of the pipeline. Also
+						// emit an appropriate system event and record metrics.
+
+						let payload_id = executor.block.payload_id();
+						let events_bus = &executor.pipeline.events;
+						let metrics = executor.service.metrics();
+
+						// Record metrics for the payload job
+						match &output {
+							Ok(built_payload) => {
+								events_bus.publish(PayloadJobCompleted::<P> {
+									payload_id,
+									built_payload: built_payload.clone(),
+								});
+								metrics.jobs_completed.increment(1);
+								metrics.record_payload::<P>(built_payload, &executor.block);
+							}
+							Err(error) => {
+								events_bus.publish(PayloadJobFailed {
+									payload_id,
+									error: error.clone(),
+								});
+								metrics.jobs_failed.increment(1);
+							}
+						}
+
+						return Poll::Ready(output);
+					}
+					Poll::Pending => return Poll::Pending,
+				},
+				// If the cursor is in the BeforeStep state, we need to run the next
+				// step of the pipeline. Steps are async futures, so we need to store
+				// their instance while they are running and being polled until
+				// resolved.
+				Cursor::BeforeStep(_, _) => {
+					let Cursor::BeforeStep(path, input) =
+						std::mem::replace(&mut executor.cursor, Cursor::PreparingStep)
+					else {
+						unreachable!("bug in PipelineExecutor state machine");
+					};
+
+					trace!("{} will execute step {path}", executor.pipeline);
+					let future = executor.execute_step(&path, input);
+					executor.cursor = Cursor::StepInProgress(path, future);
 				}
-			}
-			trace!("{} initializing completed", executor.pipeline);
-		}
-
-		// the pipeline has completed executing all steps or encountered and error.
-		// Now we are running the `after_job` of each step in the pipeline.
-		if let Cursor::Finalizing(ref mut future) = executor.cursor
-			&& let Poll::Ready(output) = future.as_mut().poll_unpin(cx)
-		{
-			trace!("{} completed with output: {output:#?}", executor.pipeline);
-
-			// Execution of this pipeline has completed, This resolves the
-			// executor future with the final output of the pipeline. Also
-			// emit an appropriate system event and record metrics.
-
-			let payload_id = executor.block.payload_id();
-			let events_bus = &executor.pipeline.events;
-			let metrics = executor.service.metrics();
-
-			match &output {
-				Ok(built_payload) => {
-					events_bus.publish(PayloadJobCompleted::<P> {
-						payload_id,
-						built_payload: built_payload.clone(),
-					});
-					metrics.jobs_completed.increment(1);
-					metrics.record_payload::<P>(built_payload, &executor.block);
+				// If the cursor is in the StepInProgress state, poll the future
+				// instance of that step to see if it has completed.
+				Cursor::StepInProgress(ref path, ref mut future) => {
+					match future.as_mut().poll_unpin(cx) {
+						Poll::Ready(output) => {
+							trace!(
+								"{} step {path:?} completed with output: {output:#?}",
+								executor.pipeline
+							);
+
+							// step has completed, we can advance the cursor
+							executor.cursor = executor.advance_cursor(path, output);
+						}
+						Poll::Pending => return Poll::Pending,
+					}
 				}
-				Err(error) => {
-					events_bus.publish(PayloadJobFailed {
-						payload_id,
-						error: error.clone(),
-					});
-					metrics.jobs_failed.increment(1);
+				// Transient state that should never be observed here
+				Cursor::PreparingStep => {
+					unreachable!("bug in PipelineExecutor state machine")
 				}
 			}
-
-			return Poll::Ready(output);
-		}
-
-		if matches!(executor.cursor, Cursor::BeforeStep(_, _)) {
-			// If the cursor is in the BeforeStep state, we need to run the next
-			// step of the pipeline. Steps are async futures, so we need to store
-			// their instance while they are running and being polled until resolved.
-
-			let Cursor::BeforeStep(path, input) =
-				std::mem::replace(&mut executor.cursor, Cursor::PreparingStep)
-			else {
-				unreachable!("bug in PipelineExecutor state machine");
-			};
-
-			trace!("{} will execute step {path}", executor.pipeline);
-			let future = executor.execute_step(&path, input);
-			executor.cursor = Cursor::StepInProgress(path, future);
-
-			// This wake is necessary because we have a new future/at a state
-			// transition boundary and directly return Poll::Pending without polling
-			// it first, so there is no registered waker to drive progress without it
-			cx.waker().wake_by_ref();
 		}
-
-		// If the cursor is in the StepInProgress state, we to poll the
-		// future instance of that step to see if it has completed.
-		if let Cursor::StepInProgress(ref path, ref mut future) = executor.cursor
-			&& let Poll::Ready(output) = future.as_mut().poll_unpin(cx)
-		{
-			trace!(
-				"{} step {path:?} completed with output: {output:#?}",
-				executor.pipeline
-			);
-
-			// step has completed, we can advance the cursor
-			executor.cursor = executor.advance_cursor(path, output);
-		}
-
-		Poll::Pending
 	}
 }
 
 /// Keeps track of the current pipeline execution progress.
 enum Cursor<P: Platform> {
+	/// The pipeline is currently initializing all steps for a new payload job.
+	///
+	/// This happens once before any step is executed, and it calls the
+	/// `before_job` method of each step in the pipeline.
+	Initializing(
+		Pin<
+			Box<
+				dyn Future<Output = Result<Checkpoint<P>, PayloadBuilderError>> + Send,
+			>,
+		>,
+	),
+
+	/// This state occurs after the `Completed` state is reached. It calls
+	/// the `after_job` method of each step in the pipeline with the output.
+	Finalizing(Pin<Box<dyn Future<Output = PipelineOutput<P>> + Send>>),
+
 	/// The pipeline will execute this step on the next iteration.
 	BeforeStep(StepPath, Checkpoint<P>),
 
@@ -397,22 +417,6 @@ enum Cursor<P: Platform> {
 		Pin<Box<dyn Future<Output = ControlFlow<P>> + Send>>,
 	),
 
-	/// The pipeline is currently initializing all steps for a new payload job.
-	///
-	/// This happens once before any step is executed, and it calls the
-	/// `before_job` method of each step in the pipeline.
-	Initializing(
-		Pin<
-			Box<
-				dyn Future<Output = Result<Checkpoint<P>, PayloadBuilderError>> + Send,
-			>,
-		>,
-	),
-
-	/// This state occurs after the `Completed` state is reached. It calls
-	/// the `after_job` method of each step in the pipeline with the output.
-	Finalizing(Pin<Box<dyn Future<Output = PipelineOutput<P>> + Send>>),
-
 	/// The pipeline is currently preparing to execute the next step.
 	/// We are in this state only for a brief moment inside the `poll` method, and
 	/// it will never be seen by the `run_step` method.

From 6299d4a348fac75ebe497e28a48e7a886d363e6b Mon Sep 17 00:00:00 2001
From: Ash Kunda <18058966+akundaz@users.noreply.github.com>
Date: Wed, 14 Jan 2026 15:13:36 -0500
Subject: [PATCH 3/6] ai magic

---
 src/pipelines/exec/mod.rs   | 472 ++++++++++--------------------------
 src/pipelines/exec/scope.rs |   2 +-
 src/pipelines/job.rs        |  39 ++-
 3 files changed, 151 insertions(+), 362 deletions(-)

diff --git a/src/pipelines/exec/mod.rs b/src/pipelines/exec/mod.rs
index e16cf83..15b58ea 100644
--- a/src/pipelines/exec/mod.rs
+++ b/src/pipelines/exec/mod.rs
@@ -15,412 +15,202 @@
 use {
 	super::{StepInstance, service::ServiceContext},
 	crate::{prelude::*, reth},
-	core::{
-		pin::Pin,
-		task::{Context, Poll},
-	},
+	core::pin::Pin,
 	futures::FutureExt,
 	navi::{StepNavigator, StepPath},
 	reth::payload::builder::{PayloadBuilderError, PayloadId},
 	scope::RootScope,
-	std::{sync::Arc, time::Instant},
+	std::sync::Arc,
 	tracing::{debug, trace},
 };
 
 pub(super) mod navi;
 pub(super) mod scope;
 
-type PipelineOutput<P: Platform> =
+pub(super) type PipelineOutput<P> =
 	Result<types::BuiltPayload<P>, Arc<PayloadBuilderError>>;
 
+pub(super) type PipelineFuture<P> =
+	Pin<Box<dyn Future<Output = PipelineOutput<P>> + Send>>;
+
 /// This type is responsible for executing a single run of a pipeline.
 ///
-/// The execution is driven by the future poll that it implements. Each call to
-/// `poll` will run one step of the pipeline at a time, or parts of a step if
-/// the step is async and needs many polls before it completes. The executor
-/// future will resolve when the whole pipeline has been executed, or when an
-/// error occurs.
-pub(super) struct PipelineExecutor<P, Provider>
+/// The execution is driven by an async function. Call `into_future()` to get
+/// the future that executes the pipeline.
+pub(super) struct PipelineExecutor<P>
 where
 	P: Platform,
-	Provider: traits::ProviderBounds<P>,
 {
-	/// The current state of the executor state machine.
-	cursor: Cursor<P>,
-
-	// The pipeline that is being executed.
-	pipeline: Arc<Pipeline<P>>,
-
-	// The block context that is being built.
-	block: BlockContext<P>,
-
-	// The reth payload builder service context that is running this payload job.
-	service: Arc<ServiceContext<P, Provider>>,
-
-	/// Execution scopes. This root scope represents the top-level pipeline that
-	/// may contain nested scopes for each nested pipeline.
-	scope: Arc<RootScope<P>>,
+	payload_id: PayloadId,
+	future: PipelineFuture<P>,
 }
 
-impl<P: Platform, Provider: traits::ProviderBounds<P>>
-	PipelineExecutor<P, Provider>
-{
-	/// Begins the execution of a pipeline for a new block/payload job.
-	pub(super) fn run(
+impl<P: Platform> PipelineExecutor<P> {
+	pub(super) fn run<Provider: traits::ProviderBounds<P>>(
 		pipeline: Arc<Pipeline<P>>,
 		block: BlockContext<P>,
 		service: Arc<ServiceContext<P, Provider>>,
 	) -> Self {
-		// Emit a system event for this new payload job and record initial metrics.
 		pipeline.events.publish(PayloadJobStarted(block.clone()));
 		service.metrics().jobs_started.increment(1);
 		service
 			.metrics()
 			.record_payload_job_attributes::<P>(block.attributes());
 
-		// Create the initial payload checkpoint, this will implicitly capture the
-		// time we started executing the pipeline for this payload job.
-		let checkpoint = block.start();
+		let payload_id = block.payload_id();
+		let future = Self::execute(pipeline, block, service).boxed();
 
-		// initialize pipeline scopes
-		let root = Arc::new(RootScope::new(&pipeline, &checkpoint));
-
-		// Initially set the execution cursor to initializing state, that will call
-		// all `before_job` methods of the steps in the pipeline.
-		Self {
-			cursor: Cursor::<P>::Initializing({
-				let block = block.clone();
-				let pipeline = Arc::clone(&pipeline);
-				let scope = Arc::clone(&root);
-
-				async move {
-					for step in pipeline.iter_steps() {
-						let navi = step.navigator(&pipeline).expect(
-							"Invalid step path. This is a bug in the pipeline executor \
-							 implementation.",
-						);
-						let limits = scope.limits_of(&step).expect("invalid step path");
-						let ctx = StepContext::new(&block, &navi, limits, None);
-						navi.instance().before_job(ctx).await?;
-					}
-
-					Ok(checkpoint)
-				}
-				.boxed()
-			}),
-			pipeline,
-			block,
-			service,
-			scope: root,
-		}
+		Self { payload_id, future }
 	}
 
-	/// Returns the payload id for which we are building a payload.
 	pub(super) fn payload_id(&self) -> PayloadId {
-		self.block.payload_id()
+		self.payload_id
 	}
-}
 
-/// private implementation details for the `PipelineExecutor`.
-impl<P: Platform, Provider: traits::ProviderBounds<P>>
-	PipelineExecutor<P, Provider>
-{
-	/// This method creates a future that encapsulates the execution as an async
-	/// step. The created future will be held inside `Cursor::StepInProgress` and
-	/// polled until it resolves.
-	///
-	/// It will prepare the step context and all the information a pipeline step
-	/// needs to execute, then create a future object that will be stored in the
-	/// cursor and polled whenever the executor is polled.
-	fn execute_step(
-		&self,
-		path: &StepPath,
-		input: Checkpoint<P>,
-	) -> Pin<Box<dyn Future<Output = ControlFlow<P>> + Send>> {
-		let limits = self.scope.limits_of(path).expect(
-			"Invalid step path. This is a bug in the pipeline executor \
-			 implementation.",
-		);
-
-		let navi = path.navigator(&self.pipeline).expect(
-			"Invalid step path. This is a bug in the pipeline executor \
-			 implementation.",
-		);
-
-		let entered_at = self.scope.entered_at(path);
-		let ctx = StepContext::new(&self.block, &navi, limits, entered_at);
-		let step = Arc::clone(navi.instance());
-		async move { step.step(input, ctx).await }.boxed()
+	pub(super) fn into_future(self) -> PipelineFuture<P> {
+		self.future
 	}
 
-	/// This method handles the control flow of the pipeline execution.
-	///
-	/// Once a step is executed it determines the next step to execute based on
-	/// the output of the step, the current cursor state and the pipeline
-	/// structure.
-	fn advance_cursor(
-		&self,
-		current_path: &StepPath,
-		output: ControlFlow<P>,
-	) -> Cursor<P> {
-		// we need this type to determine the next step to execute
-		// based on the current step output.
-		let navigator = current_path.navigator(&self.pipeline).expect(
-			"Step path is unreachable. This is a bug in the pipeline executor \
-			 implementation.",
-		);
-
-		// identify the next step to execute based on the output of the previously
-		// executed step.
-		let (step, input) = match output {
-			// If the step output is a failure, we terminate the execution of the
-			// whole pipeline and return the error as the final output on next future
-			// poll.
-			ControlFlow::Fail(payload_builder_error) => {
-				return Cursor::Finalizing(
-					self.finalize(Err(Arc::new(payload_builder_error))),
-				);
-			}
+	async fn execute<Provider: traits::ProviderBounds<P>>(
+		pipeline: Arc<Pipeline<P>>,
+		block: BlockContext<P>,
+		service: Arc<ServiceContext<P, Provider>>,
+	) -> PipelineOutput<P> {
+		let checkpoint = block.start();
+		let scope = Arc::new(RootScope::new(&pipeline, &checkpoint));
 
-			// not a failure, chose the next step based on the control flow output
-			ControlFlow::Break(checkpoint) => (navigator.next_break(), checkpoint),
-			ControlFlow::Ok(checkpoint) => (navigator.next_ok(), checkpoint),
-		};
+		let init_result =
+			Self::initialize(&pipeline, &block, &scope, checkpoint).await;
 
-		let Some(step) = step else {
-			// If there is no next step, we are done with the pipeline execution.
-			// We can finalize the pipeline and return the output as the final
-			// result of the pipeline run.
-			return Cursor::Finalizing(
-				self.finalize(input.build_payload().map_err(Arc::new)),
-			);
+		let output = match init_result {
+			Ok(checkpoint) => {
+				Self::run_steps(&pipeline, &block, &scope, checkpoint).await
+			}
+			Err(e) => Err(Arc::new(e)),
 		};
 
-		// there is a next step to be executed, create a cursor that will
-		// start running the next step with the output of the current step
-		// as input on next executor future poll
+		Self::finalize(&pipeline, &block, &service, &scope, output).await
+	}
 
-		// enter the scope of the next step
-		self.scope.switch_context(step.path(), &input);
+	async fn initialize(
+		pipeline: &Arc<Pipeline<P>>,
+		block: &BlockContext<P>,
+		scope: &Arc<RootScope<P>>,
+		checkpoint: Checkpoint<P>,
+	) -> Result<Checkpoint<P>, PayloadBuilderError> {
+		for step in pipeline.iter_steps() {
+			let navi = step
+				.navigator(pipeline)
+				.expect("Invalid step path in pipeline executor");
+			let limits = scope.limits_of(&step).expect("invalid step path");
+			let ctx = StepContext::new(block, &navi, limits, None);
+			navi.instance().before_job(ctx).await?;
+		}
 
-		// schedule execution on next future poll
-		Cursor::BeforeStep(step.into(), input)
+		trace!("{pipeline} initialized successfully");
+		Ok(checkpoint)
 	}
 
-	/// After pipeline steps are initialized, this method will identify the first
-	/// step to execute in the pipeline and prepare the cursor to run it.
-	fn first_step(&self, checkpoint: Checkpoint<P>) -> Cursor<P> {
-		let Some(navigator) = StepNavigator::entrypoint(&self.pipeline) else {
+	async fn run_steps(
+		pipeline: &Arc<Pipeline<P>>,
+		block: &BlockContext<P>,
+		scope: &Arc<RootScope<P>>,
+		mut checkpoint: Checkpoint<P>,
+	) -> PipelineOutput<P> {
+		let Some(mut navigator) = StepNavigator::entrypoint(pipeline) else {
 			debug!(
 				"empty pipeline, building empty payload for attributes: {:?}",
-				self.block.attributes()
-			);
-
-			return Cursor::<P>::Finalizing(
-				self.finalize(self.block.start().build_payload().map_err(Arc::new)),
+				block.attributes()
 			);
+			return block.start().build_payload().map_err(Arc::new);
 		};
 
-		// enter the scope of the root pipeline
-		self.scope.enter(&checkpoint);
+		scope.enter(&checkpoint);
 
-		// Begin executing the first step of the pipeline in the next future poll
-		Cursor::BeforeStep(navigator.into(), checkpoint)
+		loop {
+			let path: StepPath = navigator.path().clone();
+			scope.switch_context(&path, &checkpoint);
+
+			let limits = scope.limits_of(&path).expect("invalid step path");
+			let entered_at = scope.entered_at(&path);
+			let ctx = StepContext::new(block, &navigator, limits, entered_at);
+			let step = Arc::clone(navigator.instance());
+
+			trace!("{pipeline} will execute step {path}");
+			let output = step.step(checkpoint, ctx).await;
+			trace!("{pipeline} step {path:?} completed with output: {output:#?}");
+
+			let (next_navigator, next_checkpoint) = match output {
+				ControlFlow::Ok(cp) => (navigator.next_ok(), cp),
+				ControlFlow::Break(cp) => (navigator.next_break(), cp),
+				ControlFlow::Fail(e) => return Err(Arc::new(e)),
+			};
+
+			checkpoint = next_checkpoint;
+
+			match next_navigator {
+				Some(next) => navigator = next,
+				None => return checkpoint.build_payload().map_err(Arc::new),
+			}
+		}
 	}
 
-	/// This method will walk through the pipeline steps and invoke the
-	/// `after_job` method of each step in the pipeline with the final output.
-	fn finalize(
-		&self,
+	async fn finalize<Provider: traits::ProviderBounds<P>>(
+		pipeline: &Arc<Pipeline<P>>,
+		block: &BlockContext<P>,
+		service: &Arc<ServiceContext<P, Provider>>,
+		scope: &Arc<RootScope<P>>,
 		output: PipelineOutput<P>,
-	) -> Pin<Box<dyn Future<Output = PipelineOutput<P>> + Send>> {
+	) -> PipelineOutput<P> {
 		let output = Arc::new(output.map_err(|e| clone_payload_error_lossy(&e)));
-		let pipeline = Arc::clone(&self.pipeline);
-		let block = self.block.clone();
-		let pipeline = Arc::clone(&pipeline);
-		let scope = Arc::clone(&self.scope);
-
-		async move {
-			// invoke the `after_job` method of each step in the pipeline
-			// if any of them fails, we fail the pipeline execution, otherwise
-			// we return the output of the pipeline.
-			for step in pipeline.iter_steps() {
-				let navi = step.navigator(&pipeline).expect(
-					"Invalid step path. This is a bug in the pipeline executor \
-					 implementation.",
-				);
-				let limits = scope.limits_of(&step).expect("invalid step path");
-				let ctx = StepContext::new(&block, &navi, limits, None);
-				navi.instance().after_job(ctx, output.clone()).await?;
-			}
 
-			// leave the scope of the root pipeline (if entered, we never enter the
-			// root scope only in empty pipelines).
-			if scope.is_active() {
-				scope.leave();
+		for step in pipeline.iter_steps() {
+			let navi = step
+				.navigator(pipeline)
+				.expect("Invalid step path in pipeline executor");
+			let limits = scope.limits_of(&step).expect("invalid step path");
+			let ctx = StepContext::new(block, &navi, limits, None);
+
+			if let Err(e) = navi.instance().after_job(ctx, output.clone()).await {
+				trace!("{pipeline} finalization failed with error: {e:?}");
+				return Err(Arc::new(e));
 			}
+		}
 
-			Arc::into_inner(output)
-				.expect("unexpected > 1 strong reference count")
-				.map_err(Arc::new)
+		if scope.is_active() {
+			scope.leave();
 		}
-		.boxed()
-	}
-}
 
-impl<P, Provider> Future for PipelineExecutor<P, Provider>
-where
-	P: Platform,
-	Provider: traits::ProviderBounds<P>,
-{
-	type Output = PipelineOutput<P>;
+		let result = Arc::into_inner(output)
+			.expect("unexpected > 1 strong reference count")
+			.map_err(Arc::new);
 
-	fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-		let executor = self.get_mut();
+		trace!("{pipeline} completed with output: {result:#?}");
 
-		loop {
-			match executor.cursor {
-				// The executor has not run any steps yet, it is invoking the
-				// `before_job` method of each step in the pipeline.
-				Cursor::Initializing(ref mut future) => {
-					match future.as_mut().poll_unpin(cx) {
-						Poll::Ready(Ok(checkpoint)) => {
-							trace!("{} initialized successfully", executor.pipeline);
-							executor.cursor = executor.first_step(checkpoint);
-						}
-						Poll::Ready(Err(error)) => {
-							trace!(
-								"{} initialization failed with error: {error:?}",
-								executor.pipeline
-							);
-							// If the initialization failed, we immediately finalize the
-							// pipeline with the error that occurred during initialization
-							// and not attempt to run any steps.
-							executor.cursor =
-								Cursor::Finalizing(executor.finalize(Err(error.into())));
-						}
-						Poll::Pending => return Poll::Pending,
-					}
-				}
-				// The pipeline has completed executing all steps or encountered an
-				// error. Now we are running the `after_job` of each step.
-				Cursor::Finalizing(ref mut future) => match future
-					.as_mut()
-					.poll_unpin(cx)
-				{
-					Poll::Ready(output) => {
-						trace!("{} completed with output: {output:#?}", executor.pipeline);
-
-						// Execution of this pipeline has completed, This resolves the
-						// executor future with the final output of the pipeline. Also
-						// emit an appropriate system event and record metrics.
-
-						let payload_id = executor.block.payload_id();
-						let events_bus = &executor.pipeline.events;
-						let metrics = executor.service.metrics();
-
-						// Record metrics for the payload job
-						match &output {
-							Ok(built_payload) => {
-								events_bus.publish(PayloadJobCompleted::<P> {
-									payload_id,
-									built_payload: built_payload.clone(),
-								});
-								metrics.jobs_completed.increment(1);
-								metrics.record_payload::<P>(built_payload, &executor.block);
-							}
-							Err(error) => {
-								events_bus.publish(PayloadJobFailed {
-									payload_id,
-									error: error.clone(),
-								});
-								metrics.jobs_failed.increment(1);
-							}
-						}
-
-						return Poll::Ready(output);
-					}
-					Poll::Pending => return Poll::Pending,
-				},
-				// If the cursor is in the BeforeStep state, we need to run the next
-				// step of the pipeline. Steps are async futures, so we need to store
-				// their instance while they are running and being polled until
-				// resolved.
-				Cursor::BeforeStep(_, _) => {
-					let Cursor::BeforeStep(path, input) =
-						std::mem::replace(&mut executor.cursor, Cursor::PreparingStep)
-					else {
-						unreachable!("bug in PipelineExecutor state machine");
-					};
-
-					trace!("{} will execute step {path}", executor.pipeline);
-					let future = executor.execute_step(&path, input);
-					executor.cursor = Cursor::StepInProgress(path, future);
-				}
-				// If the cursor is in the StepInProgress state, poll the future
-				// instance of that step to see if it has completed.
-				Cursor::StepInProgress(ref path, ref mut future) => {
-					match future.as_mut().poll_unpin(cx) {
-						Poll::Ready(output) => {
-							trace!(
-								"{} step {path:?} completed with output: {output:#?}",
-								executor.pipeline
-							);
-
-							// step has completed, we can advance the cursor
-							executor.cursor = executor.advance_cursor(path, output);
-						}
-						Poll::Pending => return Poll::Pending,
-					}
-				}
-				// Transient state that should never be observed here
-				Cursor::PreparingStep => {
-					unreachable!("bug in PipelineExecutor state machine")
-				}
+		let payload_id = block.payload_id();
+		let events_bus = &pipeline.events;
+		let metrics = service.metrics();
+
+		match &result {
+			Ok(built_payload) => {
+				events_bus.publish(PayloadJobCompleted::<P> {
+					payload_id,
+					built_payload: built_payload.clone(),
+				});
+				metrics.jobs_completed.increment(1);
+				metrics.record_payload::<P>(built_payload, block);
+			}
+			Err(error) => {
+				events_bus.publish(PayloadJobFailed {
+					payload_id,
+					error: error.clone(),
+				});
+				metrics.jobs_failed.increment(1);
 			}
 		}
-	}
-}
 
-/// Keeps track of the current pipeline execution progress.
-enum Cursor<P: Platform> {
-	/// The pipeline is currently initializing all steps for a new payload job.
-	///
-	/// This happens once before any step is executed, and it calls the
-	/// `before_job` method of each step in the pipeline.
-	Initializing(
-		Pin<
-			Box<
-				dyn Future<Output = Result<Checkpoint<P>, PayloadBuilderError>> + Send,
-			>,
-		>,
-	),
-
-	/// This state occurs after the `Completed` state is reached. It calls
-	/// the `after_job` method of each step in the pipeline with the output.
-	Finalizing(Pin<Box<dyn Future<Output = PipelineOutput<P>> + Send>>),
-
-	/// The pipeline will execute this step on the next iteration.
-	BeforeStep(StepPath, Checkpoint<P>),
-
-	/// a pipeline step execution is in progress.
-	///
-	/// This state is set when the pipeline executor has began executing a step
-	/// but has not yet completed it between two polls. Steps are asynchronous
-	/// and may span multiple polls. Until the step future is resolved, we need
-	/// to store its instance and poll it.
-	///
-	/// Here we store the step path that is currently being executed
-	/// and the pinned future that is executing the step.
-	StepInProgress(
-		StepPath,
-		Pin<Box<dyn Future<Output = ControlFlow<P>> + Send>>,
-	),
-
-	/// The pipeline is currently preparing to execute the next step.
-	/// We are in this state only for a brief moment inside the `poll` method, and
-	/// it will never be seen by the `run_step` method.
-	PreparingStep,
+		result
+	}
 }
 
 pub(crate) fn clone_payload_error_lossy(
diff --git a/src/pipelines/exec/scope.rs b/src/pipelines/exec/scope.rs
index af8c0bb..045b414 100644
--- a/src/pipelines/exec/scope.rs
+++ b/src/pipelines/exec/scope.rs
@@ -51,7 +51,7 @@ use {
 	core::{cell::RefCell, time::Duration},
 	metrics::{Counter, Histogram},
 	parking_lot::RwLock,
-	std::collections::HashMap,
+	std::{collections::HashMap, time::Instant},
 };
 
 /// Keeps track of the currently active pipeline execution scope relative to the
diff --git a/src/pipelines/job.rs b/src/pipelines/job.rs
index 133f4b0..cc41b4f 100644
--- a/src/pipelines/job.rs
+++ b/src/pipelines/job.rs
@@ -2,7 +2,10 @@ use {
 	super::traits,
 	crate::{
 		alloy,
-		pipelines::{exec::PipelineExecutor, service::ServiceContext},
+		pipelines::{
+			exec::{PipelineExecutor, PipelineFuture},
+			service::ServiceContext,
+		},
 		prelude::*,
 		reth,
 	},
@@ -41,9 +44,9 @@ where
 	Provider: traits::ProviderBounds<P>,
 {
 	block: BlockContext<P>,
-	fut: ExecutorFuture<P, Provider>,
-	/// The deadline when this job should resolve.
+	fut: ExecutorFuture<P>,
 	deadline: Pin<Box<Sleep>>,
+	_provider: std::marker::PhantomData<fn() -> Provider>,
 }
 
 impl<P, Provider> PayloadJob<P, Provider>
@@ -78,6 +81,7 @@ where
 			block,
 			fut,
 			deadline,
+			_provider: std::marker::PhantomData::<fn() -> Provider>,
 		}
 	}
 }
@@ -89,7 +93,7 @@ where
 {
 	type BuiltPayload = types::BuiltPayload<P>;
 	type PayloadAttributes = types::PayloadBuilderAttributes<P>;
-	type ResolvePayloadFuture = ExecutorFuture<P, Provider>;
+	type ResolvePayloadFuture = ExecutorFuture<P>;
 
 	fn best_payload(&self) -> Result<Self::BuiltPayload, PayloadBuilderError> {
 		unimplemented!("PayloadJob::best_payload is not implemented");
@@ -173,14 +177,13 @@ where
 /// This future wraps the `PipelineExecutor` and is used to poll the
 /// internal executor of the pipeline. Once this future is resolved, it
 /// can be polled again and will return copy of the resolved payload.
-pub(super) struct ExecutorFuture<P, Provider>
+pub(super) struct ExecutorFuture<P>
 where
 	P: Platform,
-	Provider: traits::ProviderBounds<P>,
 {
 	payload_id: PayloadId,
 	started_at: Instant,
-	state: ExecutorFutureState<P, Provider>,
+	state: ExecutorFutureState<P>,
 }
 
 /// This enum allows us to wrap the `PipelineExecutor` future
@@ -191,25 +194,23 @@ where
 /// Whenever any of the copies of the future is polled, it will poll the
 /// executor, if any copy resolved, all copies will also resolve with the same
 /// result.
-enum ExecutorFutureState<P, Provider>
+enum ExecutorFutureState<P>
 where
 	P: Platform,
-	Provider: traits::ProviderBounds<P>,
 {
 	Ready(Result<types::BuiltPayload<P>, Arc<PayloadBuilderError>>),
-	Future(Shared<PipelineExecutor<P, Provider>>),
+	Future(Shared<PipelineFuture<P>>),
 }
 
-impl<P, Provider> ExecutorFuture<P, Provider>
+impl<P> ExecutorFuture<P>
 where
 	P: Platform,
-	Provider: traits::ProviderBounds<P>,
 {
-	pub(super) fn new(executor: PipelineExecutor<P, Provider>) -> Self {
+	pub(super) fn new(executor: PipelineExecutor<P>) -> Self {
 		Self {
 			started_at: Instant::now(),
 			payload_id: executor.payload_id(),
-			state: ExecutorFutureState::Future(executor.shared()),
+			state: ExecutorFutureState::Future(executor.into_future().shared()),
 		}
 	}
 
@@ -261,10 +262,9 @@ where
 	}
 }
 
-impl<P, Provider> Future for ExecutorFuture<P, Provider>
+impl<P> Future for ExecutorFuture<P>
 where
 	P: Platform,
-	Provider: traits::ProviderBounds<P>,
 {
 	type Output = Result<types::BuiltPayload<P>, PayloadBuilderError>;
 
@@ -279,8 +279,8 @@ where
 			),
 
 			// we are still in progress. keep polling the inner executor future.
-			ExecutorFutureState::Future(ref mut executor) => {
-				match executor.poll_unpin(cx) {
+			ExecutorFutureState::Future(ref mut future) => {
+				match future.poll_unpin(cx) {
 					Poll::Ready(result) => {
 						// got a result. All future polls will return the result directly
 						// without polling the executor again.
@@ -304,10 +304,9 @@ where
 /// We want this to be cloneable because the `resolve_kind` method could
 /// potentially return multiple copies of the future, and we want all of them to
 /// resolve with the same result at the same time.
-impl<P, Provider> Clone for ExecutorFuture<P, Provider>
+impl<P> Clone for ExecutorFuture<P>
 where
 	P: Platform,
-	Provider: traits::ProviderBounds<P>,
 {
 	fn clone(&self) -> Self {
 		Self {

From 8d0e57b0697a99252b1079a30b97dd63574fae2f Mon Sep 17 00:00:00 2001
From: Ash Kunda <18058966+akundaz@users.noreply.github.com>
Date: Wed, 14 Jan 2026 15:28:37 -0500
Subject: [PATCH 4/6] add a yield point to address tokio executor starvation

---
 src/pipelines/exec/mod.rs | 32 +++++++++++++++++++++++++++++++-
 1 file changed, 31 insertions(+), 1 deletion(-)

diff --git a/src/pipelines/exec/mod.rs b/src/pipelines/exec/mod.rs
index 15b58ea..169d674 100644
--- a/src/pipelines/exec/mod.rs
+++ b/src/pipelines/exec/mod.rs
@@ -149,7 +149,37 @@ impl<P: Platform> PipelineExecutor<P> {
 			checkpoint = next_checkpoint;
 
 			match next_navigator {
-				Some(next) => navigator = next,
+				Some(next) => {
+					navigator = next;
+					// Cooperative yielding: The `run_steps()` loop can execute multiple
+					// pipeline steps in a tight sequence without yielding back to
+					// the tokio scheduler, especially when steps complete quickly
+					// (synchronously). This can cause executor starvation where other
+					// tasks are delayed.
+					//
+					// The `yield_now().await` here provides a yield point to prevent
+					// starvation. This yield point is not at a fixed location - it
+					// could be placed at several alternative points in the execution
+					// loop. For example:
+					//   - After step execution completes
+					//   - Before executing the next step
+					//   - At the end of each loop iteration
+					//
+					// The critical requirement is that SOME yield point exists in the
+					// loop to allow the tokio scheduler to run other tasks. The
+					// exact placement is a trade-off that hasn't been fully explored:
+					//   - Here: Minimal overhead (one yield per step), but may yield
+					//     unnecessarily
+					//   - Before step: Never misses a yield, but always adds overhead
+					//   - Conditional (e.g., "if step completed synchronously"): Optimal,
+					//     but requires additional bookkeeping/changes to the step API
+					//
+					// If executor starvation persists, consider:
+					//   1. Adding yield points to `initialize()` and `finalize()` loops
+					//      (which lack await points between iterations)
+					//   2. Moving yield point to different location in this loop
+					tokio::task::yield_now().await;
+				}
 				None => return checkpoint.build_payload().map_err(Arc::new),
 			}
 		}

From d1d46fff4294d457885b848d5ec8fb3549db470c Mon Sep 17 00:00:00 2001
From: Ash Kunda <18058966+akundaz@users.noreply.github.com>
Date: Wed, 14 Jan 2026 15:51:49 -0500
Subject: [PATCH 5/6] make PayloadJob not generic over `Provider`

---
 src/pipelines/exec/mod.rs   | 30 +++++++++++------------
 src/pipelines/exec/scope.rs |  2 +-
 src/pipelines/job.rs        | 48 +++++++++----------------------------
 src/pipelines/service.rs    | 44 +++++++++++++++++++---------------
 4 files changed, 51 insertions(+), 73 deletions(-)

diff --git a/src/pipelines/exec/mod.rs b/src/pipelines/exec/mod.rs
index 169d674..318d94e 100644
--- a/src/pipelines/exec/mod.rs
+++ b/src/pipelines/exec/mod.rs
@@ -13,7 +13,7 @@
 //! [`BlockBuilder::apply_pre_execution_changes`]: crate::reth::evm::execute::BlockBuilder::apply_pre_execution_changes
 
 use {
-	super::{StepInstance, service::ServiceContext},
+	super::{StepInstance, metrics},
 	crate::{prelude::*, reth},
 	core::pin::Pin,
 	futures::FutureExt,
@@ -46,19 +46,17 @@ where
 }
 
 impl<P: Platform> PipelineExecutor<P> {
-	pub(super) fn run<Provider: traits::ProviderBounds<P>>(
+	pub(super) fn run(
 		pipeline: Arc<Pipeline<P>>,
 		block: BlockContext<P>,
-		service: Arc<ServiceContext<P, Provider>>,
+		metrics: Arc<metrics::Payload>,
 	) -> Self {
 		pipeline.events.publish(PayloadJobStarted(block.clone()));
-		service.metrics().jobs_started.increment(1);
-		service
-			.metrics()
-			.record_payload_job_attributes::<P>(block.attributes());
+		metrics.jobs_started.increment(1);
+		metrics.record_payload_job_attributes::<P>(block.attributes());
 
 		let payload_id = block.payload_id();
-		let future = Self::execute(pipeline, block, service).boxed();
+		let future = Self::execute(pipeline, block, metrics).boxed();
 
 		Self { payload_id, future }
 	}
@@ -71,10 +69,10 @@ impl<P: Platform> PipelineExecutor<P> {
 		self.future
 	}
 
-	async fn execute<Provider: traits::ProviderBounds<P>>(
+	async fn execute(
 		pipeline: Arc<Pipeline<P>>,
 		block: BlockContext<P>,
-		service: Arc<ServiceContext<P, Provider>>,
+		metrics: Arc<metrics::Payload>,
 	) -> PipelineOutput<P> {
 		let checkpoint = block.start();
 		let scope = Arc::new(RootScope::new(&pipeline, &checkpoint));
@@ -89,7 +87,7 @@ impl<P: Platform> PipelineExecutor<P> {
 			Err(e) => Err(Arc::new(e)),
 		};
 
-		Self::finalize(&pipeline, &block, &service, &scope, output).await
+		Self::finalize(&pipeline, &block, &metrics, &scope, output).await
 	}
 
 	async fn initialize(
@@ -185,10 +183,10 @@ impl<P: Platform> PipelineExecutor<P> {
 		}
 	}
 
-	async fn finalize<Provider: traits::ProviderBounds<P>>(
+	async fn finalize(
 		pipeline: &Arc<Pipeline<P>>,
 		block: &BlockContext<P>,
-		service: &Arc<ServiceContext<P, Provider>>,
+		metrics: &Arc<metrics::Payload>,
 		scope: &Arc<RootScope<P>>,
 		output: PipelineOutput<P>,
 	) -> PipelineOutput<P> {
@@ -199,9 +197,10 @@ impl<P: Platform> PipelineExecutor<P> {
 				.navigator(pipeline)
 				.expect("Invalid step path in pipeline executor");
 			let limits = scope.limits_of(&step).expect("invalid step path");
-			let ctx = StepContext::new(block, &navi, limits, None);
+			let step_ctx = StepContext::new(block, &navi, limits, None);
 
-			if let Err(e) = navi.instance().after_job(ctx, output.clone()).await {
+			if let Err(e) = navi.instance().after_job(step_ctx, output.clone()).await
+			{
 				trace!("{pipeline} finalization failed with error: {e:?}");
 				return Err(Arc::new(e));
 			}
@@ -219,7 +218,6 @@ impl<P: Platform> PipelineExecutor<P> {
 
 		let payload_id = block.payload_id();
 		let events_bus = &pipeline.events;
-		let metrics = service.metrics();
 
 		match &result {
 			Ok(built_payload) => {
diff --git a/src/pipelines/exec/scope.rs b/src/pipelines/exec/scope.rs
index 045b414..ed9a7af 100644
--- a/src/pipelines/exec/scope.rs
+++ b/src/pipelines/exec/scope.rs
@@ -48,8 +48,8 @@
 
 use {
 	super::*,
+	::metrics::{Counter, Histogram},
 	core::{cell::RefCell, time::Duration},
-	metrics::{Counter, Histogram},
 	parking_lot::RwLock,
 	std::{collections::HashMap, time::Instant},
 };
diff --git a/src/pipelines/job.rs b/src/pipelines/job.rs
index cc41b4f..a07517b 100644
--- a/src/pipelines/job.rs
+++ b/src/pipelines/job.rs
@@ -1,11 +1,8 @@
 use {
-	super::traits,
+	super::metrics,
 	crate::{
 		alloy,
-		pipelines::{
-			exec::{PipelineExecutor, PipelineFuture},
-			service::ServiceContext,
-		},
+		pipelines::exec::{PipelineExecutor, PipelineFuture},
 		prelude::*,
 		reth,
 	},
@@ -13,6 +10,7 @@ use {
 	core::{
 		pin::Pin,
 		task::{Context, Poll},
+		time::Duration,
 	},
 	futures::{FutureExt, future::Shared},
 	reth::{
@@ -38,26 +36,18 @@ use {
 ///
 /// This job will automatically terminate after a deadline even `resolve_kind`
 /// is not called.
-pub(super) struct PayloadJob<P, Provider>
-where
-	P: Platform,
-	Provider: traits::ProviderBounds<P>,
-{
+pub(super) struct PayloadJob<P: Platform> {
 	block: BlockContext<P>,
 	fut: ExecutorFuture<P>,
 	deadline: Pin<Box<Sleep>>,
-	_provider: std::marker::PhantomData<fn() -> Provider>,
 }
 
-impl<P, Provider> PayloadJob<P, Provider>
-where
-	P: Platform,
-	Provider: traits::ProviderBounds<P>,
-{
+impl<P: Platform> PayloadJob<P> {
 	pub(super) fn new(
 		pipeline: &Arc<Pipeline<P>>,
 		block: BlockContext<P>,
-		service: &Arc<ServiceContext<P, Provider>>,
+		metrics: Arc<metrics::Payload>,
+		deadline: Duration,
 	) -> Self {
 		debug!(
 			"New Payload Job {} with block context: {block:#?}",
@@ -67,30 +57,18 @@ where
 		let fut = ExecutorFuture::new(PipelineExecutor::run(
 			Arc::clone(pipeline),
 			block.clone(),
-			Arc::clone(service),
+			metrics,
 		));
 
-		// Job should complete within deadline (12s) even if GetPayload is never
-		// called. This prevents job accumulation.
-		// TODO: when FCU update avalanche is fixed we could replace
-		// builder.deadline with payload.attributes.timeout
-		let deadline =
-			Box::pin(tokio::time::sleep(service.node_config().builder.deadline));
-
 		Self {
 			block,
 			fut,
-			deadline,
-			_provider: std::marker::PhantomData::<fn() -> Provider>,
+			deadline: Box::pin(tokio::time::sleep(deadline)),
 		}
 	}
 }
 
-impl<P, Provider> RethPayloadJobTrait for PayloadJob<P, Provider>
-where
-	P: Platform,
-	Provider: traits::ProviderBounds<P>,
-{
+impl<P: Platform> RethPayloadJobTrait for PayloadJob<P> {
 	type BuiltPayload = types::BuiltPayload<P>;
 	type PayloadAttributes = types::PayloadBuilderAttributes<P>;
 	type ResolvePayloadFuture = ExecutorFuture<P>;
@@ -132,11 +110,7 @@ where
 /// This future is polled for the first time by the Reth runtime when the
 /// `PayloadJob` is created. Here we want to immediately start executing
 /// the pipeline instead of waiting for the `resolve_kind` to be called.
-impl<P, Provider> Future for PayloadJob<P, Provider>
-where
-	P: Platform,
-	Provider: traits::ProviderBounds<P>,
-{
+impl<P: Platform> Future for PayloadJob<P> {
 	type Output = Result<(), PayloadBuilderError>;
 
 	fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
diff --git a/src/pipelines/service.rs b/src/pipelines/service.rs
index 4cb2337..384ea9e 100644
--- a/src/pipelines/service.rs
+++ b/src/pipelines/service.rs
@@ -56,13 +56,12 @@ where
 
 		let provider = Arc::new(ctx.provider().clone());
 
+		let metrics =
+			metrics::Payload::with_scope(&format!("{}_payloads", pipeline.name()));
+
 		let service = ServiceContext {
 			provider: Arc::clone(&provider),
 			node_config: ctx.config().clone(),
-			metrics: metrics::Payload::with_scope(&format!(
-				"{}_payloads",
-				pipeline.name()
-			)),
 		};
 
 		// assign metric names to each step in the pipeline.
@@ -86,7 +85,7 @@ where
 		}
 
 		let (service, builder) = PayloadBuilderService::new(
-			JobGenerator::new(pipeline, service),
+			JobGenerator::new(pipeline, service, metrics),
 			ctx.provider().canonical_state_stream(),
 		);
 
@@ -101,14 +100,13 @@ where
 /// There is one service context instance per reth node. This type gives
 /// individual jobs access to the node state, transaction pool and other
 /// runtime facilities that are managed by reth.
-pub(super) struct ServiceContext<Plat, Provider>
+struct ServiceContext<Plat, Provider>
 where
 	Plat: Platform,
 	Provider: traits::ProviderBounds<Plat>,
 {
 	provider: Arc<Provider>,
 	node_config: NodeConfig<types::ChainSpec<Plat>>,
-	metrics: metrics::Payload,
 }
 
 impl<Plat, Provider> ServiceContext<Plat, Provider>
@@ -116,23 +114,17 @@ where
 	Plat: Platform,
 	Provider: traits::ProviderBounds<Plat>,
 {
-	pub(super) fn provider(&self) -> &Provider {
+	fn provider(&self) -> &Provider {
 		&self.provider
 	}
 
-	pub(super) const fn node_config(
-		&self,
-	) -> &NodeConfig<types::ChainSpec<Plat>> {
+	const fn node_config(&self) -> &NodeConfig<types::ChainSpec<Plat>> {
 		&self.node_config
 	}
 
-	pub(super) const fn chain_spec(&self) -> &Arc<types::ChainSpec<Plat>> {
+	const fn chain_spec(&self) -> &Arc<types::ChainSpec<Plat>> {
 		&self.node_config().chain
 	}
-
-	pub(super) const fn metrics(&self) -> &metrics::Payload {
-		&self.metrics
-	}
 }
 
 /// This type is stored inside the [`PayloadBuilderService`] type in Reth.
@@ -149,6 +141,7 @@ where
 {
 	pipeline: Arc<Pipeline<Plat>>,
 	service: Arc<ServiceContext<Plat, Provider>>,
+	metrics: Arc<metrics::Payload>,
 	pre_cached: Option<PrecachedState>,
 }
 
@@ -157,9 +150,10 @@ where
 	Plat: Platform,
 	Provider: traits::ProviderBounds<Plat>,
 {
-	pub(super) fn new(
+	fn new(
 		pipeline: Pipeline<Plat>,
 		service: ServiceContext<Plat, Provider>,
+		metrics: metrics::Payload,
 	) -> Self {
 		let pipeline = Arc::new(pipeline);
 		let service = Arc::new(service);
@@ -167,6 +161,7 @@ where
 		Self {
 			pipeline,
 			service,
+			metrics: Arc::new(metrics),
 			pre_cached: None,
 		}
 	}
@@ -190,7 +185,7 @@ where
 	Plat: Platform,
 	Provider: traits::ProviderBounds<Plat>,
 {
-	type Job = PayloadJob<Plat, Provider>;
+	type Job = PayloadJob<Plat>;
 
 	fn new_payload_job(
 		&self,
@@ -218,7 +213,18 @@ where
 		)
 		.map_err(PayloadBuilderError::other)?;
 
-		Ok(PayloadJob::new(&self.pipeline, block_ctx, &self.service))
+		// Job should complete within deadline (12s) even if GetPayload is never
+		// called. This prevents job accumulation.
+		// TODO: when FCU update avalanche is fixed we could replace
+		// builder.deadline with payload.attributes.timeout
+		let deadline = self.service.node_config().builder.deadline;
+
+		Ok(PayloadJob::new(
+			&self.pipeline,
+			block_ctx,
+			self.metrics.clone(),
+			deadline,
+		))
 	}
 
 	fn on_new_state<N: NodePrimitives>(

From c714ef9be7f29c760838daf0e5174ee97ec648ee Mon Sep 17 00:00:00 2001
From: Ash Kunda <18058966+akundaz@users.noreply.github.com>
Date: Wed, 14 Jan 2026 17:14:43 -0500
Subject: [PATCH 6/6] try consume_budget fn

---
 src/pipelines/exec/mod.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/pipelines/exec/mod.rs b/src/pipelines/exec/mod.rs
index 318d94e..3c48c42 100644
--- a/src/pipelines/exec/mod.rs
+++ b/src/pipelines/exec/mod.rs
@@ -176,7 +176,7 @@ impl<P: Platform> PipelineExecutor<P> {
 					//   1. Adding yield points to `initialize()` and `finalize()` loops
 					//      (which lack await points between iterations)
 					//   2. Moving yield point to different location in this loop
-					tokio::task::yield_now().await;
+					tokio::task::consume_budget().await;
 				}
 				None => return checkpoint.build_payload().map_err(Arc::new),
 			}