Decoupled Pipelines from Reth Transaction Pool

Author: karim-agha

Cargo.toml

@@ -7,7 +7,7 @@ members = [
 resolver = "2"
 
 [workspace.package]
-version = "0.2.2"
+version = "0.3.0"
 edition = "2024"
 rust-version = "1.87"
 license = "MIT"

src/pipelines/exec/mod.rs

@@ -11,7 +11,7 @@
 //! [`BlockBuilder::apply_pre_execution_changes`] applied to its state.
 
 use {
-	super::service::ServiceContext,
+	super::{StepInstance, service::ServiceContext},
 	crate::{prelude::*, reth},
 	core::{
 		pin::Pin,
@@ -38,10 +38,9 @@ type PipelineOutput<P: Platform> =
 /// 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, Pool>
+pub(super) struct PipelineExecutor<P, Provider>
 where
 	P: Platform,
-	Pool: traits::PoolBounds<P>,
 	Provider: traits::ProviderBounds<P>,
 {
 	/// The current state of the executor state machine.
@@ -54,24 +53,21 @@ where
 	block: BlockContext<P>,
 
 	// The reth payload builder service context that is running this payload job.
-	service: Arc<ServiceContext<P, Provider, Pool>>,
+	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>,
 }
 
-impl<
-	P: Platform,
-	Provider: traits::ProviderBounds<P>,
-	Pool: traits::PoolBounds<P>,
-> PipelineExecutor<P, Provider, Pool>
+impl<P: Platform, Provider: traits::ProviderBounds<P>>
+	PipelineExecutor<P, Provider>
 {
 	/// Begins the execution of a pipeline for a new block/payload job.
 	pub fn run(
 		pipeline: Arc<Pipeline<P>>,
 		block: BlockContext<P>,
-		service: Arc<ServiceContext<P, Provider, Pool>>,
+		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()));
@@ -89,7 +85,6 @@ impl<
 			cursor: Cursor::<P>::Initializing({
 				let block = block.clone();
 				let pipeline = Arc::clone(&pipeline);
-				let service = Arc::clone(&service);
 				let scope = Arc::clone(&root);
 
 				async move {
@@ -102,7 +97,7 @@ impl<
 							 implementation.",
 						);
 						let scope = scope.of(&step).expect("invalid step path");
-						let ctx = StepContext::new(&block, &service, &navi, scope);
+						let ctx = StepContext::new(&block, &navi, scope);
 						navi.instance().before_job(ctx).await?;
 					}
 					Ok(())
@@ -123,11 +118,8 @@ impl<
 }
 
 /// private implementation details for the `PipelineExecutor`.
-impl<
-	P: Platform,
-	Provider: traits::ProviderBounds<P>,
-	Pool: traits::PoolBounds<P>,
-> PipelineExecutor<P, Provider, Pool>
+impl<P: Platform, Provider: traits::ProviderBounds<P>>
+	PipelineExecutor<P, Provider>
 {
 	/// This method creates a future that encapsulates the execution an an async
 	/// step. The created future will be held inside `Cursor::StepInProgress` and
@@ -151,7 +143,7 @@ impl<
 			 implementation.",
 		);
 
-		let ctx = StepContext::new(&self.block, &self.service, &step_navi, scope);
+		let ctx = StepContext::new(&self.block, &step_navi, scope);
 		let step = Arc::clone(step_navi.instance());
 		async move { step.step(input, ctx).await }.boxed()
 	}
@@ -236,7 +228,6 @@ impl<
 		let pipeline = Arc::clone(&self.pipeline);
 		let block = self.block.clone();
 		let pipeline = Arc::clone(&pipeline);
-		let service = Arc::clone(&self.service);
 		let scope = Arc::clone(&self.scope);
 
 		async move {
@@ -249,7 +240,7 @@ impl<
 					 implementation.",
 				);
 				let scope = scope.of(&step).expect("invalid step path");
-				let ctx = StepContext::new(&block, &service, &navi, scope);
+				let ctx = StepContext::new(&block, &navi, scope);
 				navi.instance().after_job(ctx, output.clone()).await?;
 			}
 
@@ -264,11 +255,10 @@ impl<
 	}
 }
 
-impl<P, Provider, Pool> Future for PipelineExecutor<P, Provider, Pool>
+impl<P, Provider> Future for PipelineExecutor<P, Provider>
 where
 	P: Platform,
 	Provider: traits::ProviderBounds<P>,
-	Pool: traits::PoolBounds<P>,
 {
 	type Output = PipelineOutput<P>;
 

src/pipelines/exec/navi.rs

@@ -5,7 +5,7 @@
 //! etc.
 
 use {
-	super::super::step::instance::StepInstance,
+	super::StepInstance,
 	crate::prelude::*,
 	core::fmt::Formatter,
 	derive_more::{From, Into},

src/pipelines/exec/scope.rs

@@ -70,6 +70,8 @@ impl RootScope {
 unsafe impl Send for RootScope {}
 unsafe impl Sync for RootScope {}
 
+/// Given a path to a step in the pipeline, returns a path to the immediate
+/// pipeline that contains it.
 #[inline]
 fn scope_of(step: &StepPath) -> StepPath {
 	step.clone().remove_leaf().unwrap_or(StepPath::empty())
@@ -79,14 +81,14 @@ fn scope_of(step: &StepPath) -> StepPath {
 ///
 /// Each pipeline has its scope that may include nested scopes for each nested
 /// pipeline. Scopes are used to manage limits and metrics for each pipeline
-/// execution. All steps in a pipeline run within the scope of the pipeline that
-/// contains it. When a scope is active, then all its parent scopes are active
-/// as well.
+/// execution. All steps in a pipeline run within the scopes of the pipelines
+/// that contain it. When a scope is active, then all its parent scopes are
+/// active as well.
 #[derive(Debug)]
 pub struct Scope {
 	limits: Limits,
 	metrics: Metrics,
-	entered: Cell<Option<Instant>>,
+	entered_at: Cell<Option<Instant>>,
 	enter_counter: AtomicU32,
 	nested: HashMap<usize, Scope>,
 }
@@ -96,18 +98,18 @@ impl Scope {
 	/// When a scope is active it means that one of its steps (or in its nested
 	/// scopes) is currently being executed,
 	pub const fn is_active(&self) -> bool {
-		self.entered.get().is_some()
+		self.entered_at.get().is_some()
 	}
 
 	/// Returns the elapsed time since the scope was entered.
 	/// This will only return a value if the scope is currently active.
 	pub fn elapsed(&self) -> Option<Duration> {
-		self.entered.get().map(|start| start.elapsed())
+		self.entered_at.get().map(|start| start.elapsed())
 	}
 
 	/// Returns when the scope was entered most recently.
 	pub fn started_at(&self) -> Option<Instant> {
-		self.entered.get()
+		self.entered_at.get()
 	}
 
 	/// Returns the payload limits for steps running within the current scope.
@@ -143,14 +145,14 @@ impl Scope {
 			limits,
 			metrics,
 			nested,
-			entered: Cell::new(None),
+			entered_at: Cell::new(None),
 			enter_counter: AtomicU32::new(0),
 		}
 	}
 
 	fn enter(&self) {
 		assert!(!self.is_active(), "Scope is already active");
-		self.entered.set(Some(Instant::now()));
+		self.entered_at.set(Some(Instant::now()));
 		self.enter_counter.fetch_add(1, Ordering::Relaxed);
 		self.metrics.iter_count_total.increment(1);
 	}
@@ -175,7 +177,7 @@ impl Scope {
 			.increment(duration.as_millis() as u64);
 
 		self.metrics.exec_duration_histogram.record(duration);
-		self.entered.replace(None);
+		self.entered_at.replace(None);
 	}
 
 	/// Returns the scope with a given path.
@@ -221,7 +223,7 @@ impl Scope {
 			limits,
 			metrics,
 			nested,
-			entered: Cell::new(None),
+			entered_at: Cell::new(None),
 			enter_counter: AtomicU32::new(0),
 		}
 	}

src/pipelines/job.rs

@@ -28,26 +28,24 @@ use {
 /// This is a long-running job that will be polled by the CL node until it is
 /// resolved. The job future must resolve within 1 second from the moment
 /// [`PayloadJob::resolve_kind`] is called with [`PaylodKind::Earliest`].
-pub struct PayloadJob<P, Provider, Pool>
+pub struct PayloadJob<P, Provider>
 where
 	P: Platform,
-	Pool: traits::PoolBounds<P>,
 	Provider: traits::ProviderBounds<P>,
 {
 	block: BlockContext<P>,
-	fut: ExecutorFuture<P, Provider, Pool>,
+	fut: ExecutorFuture<P, Provider>,
 }
 
-impl<P, Provider, Pool> PayloadJob<P, Provider, Pool>
+impl<P, Provider> PayloadJob<P, Provider>
 where
 	P: Platform,
-	Pool: traits::PoolBounds<P>,
 	Provider: traits::ProviderBounds<P>,
 {
 	pub fn new(
 		pipeline: &Arc<Pipeline<P>>,
 		block: BlockContext<P>,
-		service: &Arc<ServiceContext<P, Provider, Pool>>,
+		service: &Arc<ServiceContext<P, Provider>>,
 	) -> Self {
 		debug!(
 			"New Payload Job {} with block context: {block:#?}",
@@ -64,15 +62,14 @@ where
 	}
 }
 
-impl<P, Provider, Pool> RethPayloadJobTrait for PayloadJob<P, Provider, Pool>
+impl<P, Provider> RethPayloadJobTrait for PayloadJob<P, Provider>
 where
 	P: Platform,
-	Pool: traits::PoolBounds<P>,
 	Provider: traits::ProviderBounds<P>,
 {
 	type BuiltPayload = types::BuiltPayload<P>;
 	type PayloadAttributes = types::PayloadBuilderAttributes<P>;
-	type ResolvePayloadFuture = ExecutorFuture<P, Provider, Pool>;
+	type ResolvePayloadFuture = ExecutorFuture<P, Provider>;
 
 	fn best_payload(&self) -> Result<Self::BuiltPayload, PayloadBuilderError> {
 		unimplemented!("PayloadJob::best_payload is not implemented");
@@ -111,10 +108,9 @@ 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, Pool> Future for PayloadJob<P, Provider, Pool>
+impl<P, Provider> Future for PayloadJob<P, Provider>
 where
 	P: Platform,
-	Pool: traits::PoolBounds<P>,
 	Provider: traits::ProviderBounds<P>,
 {
 	type Output = Result<(), PayloadBuilderError>;
@@ -144,14 +140,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 copie of the resolved payload.
-pub struct ExecutorFuture<P, Provider, Pool>
+pub struct ExecutorFuture<P, Provider>
 where
 	P: Platform,
-	Pool: traits::PoolBounds<P>,
 	Provider: traits::ProviderBounds<P>,
 {
 	payload_id: PayloadId,
-	state: ExecutorFutureState<P, Provider, Pool>,
+	state: ExecutorFutureState<P, Provider>,
 }
 
 /// This enum allows us to wrap the `PipelineExecutor` future
@@ -162,35 +157,32 @@ 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, Pool>
+enum ExecutorFutureState<P, Provider>
 where
 	P: Platform,
-	Pool: traits::PoolBounds<P>,
 	Provider: traits::ProviderBounds<P>,
 {
 	Ready(Result<types::BuiltPayload<P>, Arc<PayloadBuilderError>>),
-	Future(Shared<PipelineExecutor<P, Provider, Pool>>),
+	Future(Shared<PipelineExecutor<P, Provider>>),
 }
 
-impl<P, Provider, Pool> ExecutorFuture<P, Provider, Pool>
+impl<P, Provider> ExecutorFuture<P, Provider>
 where
 	P: Platform,
-	Pool: traits::PoolBounds<P>,
 	Provider: traits::ProviderBounds<P>,
 {
-	pub fn new(executor: PipelineExecutor<P, Provider, Pool>) -> Self {
+	pub fn new(executor: PipelineExecutor<P, Provider>) -> Self {
 		Self {
 			payload_id: executor.payload_id(),
 			state: ExecutorFutureState::Future(executor.shared()),
 		}
 	}
 }
 
-impl<P, Provider, Pool> Future for ExecutorFuture<P, Provider, Pool>
+impl<P, Provider> Future for ExecutorFuture<P, Provider>
 where
 	P: Platform,
 	Provider: traits::ProviderBounds<P>,
-	Pool: traits::PoolBounds<P>,
 {
 	type Output = Result<types::BuiltPayload<P>, PayloadBuilderError>;
 
@@ -230,11 +222,10 @@ where
 /// We want this to be clonable 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, Pool> Clone for ExecutorFuture<P, Provider, Pool>
+impl<P, Provider> Clone for ExecutorFuture<P, Provider>
 where
 	P: Platform,
 	Provider: traits::ProviderBounds<P>,
-	Pool: traits::PoolBounds<P>,
 {
 	fn clone(&self) -> Self {
 		Self {

src/pipelines/mod.rs

@@ -10,10 +10,9 @@ use {
 	futures::Stream,
 	pipelines_macros::impl_into_pipeline_steps,
 	std::sync::Arc,
-	step::instance::StepInstance,
+	step::StepInstance,
 };
 
-mod context;
 mod events;
 mod exec;
 mod iter;
@@ -28,9 +27,8 @@ mod tests;
 // public API exports
 pub use {
 	Behavior::{Loop, Once},
-	context::StepContext,
 	events::system_events::*,
-	step::{ControlFlow, InitContext, PayloadBuilderError, Step},
+	step::{ControlFlow, InitContext, PayloadBuilderError, Step, StepContext},
 };
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]