Test coverage for payload limits scaling and inheritance

Author: karim-agha

bin/flashblocks/grafana/standard.json

@@ -110,7 +110,7 @@
       },
       "gridPos": {
         "h": 4,
-        "w": 2,
+        "w": 3,
         "x": 0,
         "y": 1
       },
@@ -182,8 +182,8 @@
       },
       "gridPos": {
         "h": 4,
-        "w": 3,
-        "x": 2,
+        "w": 4,
+        "x": 3,
         "y": 1
       },
       "id": 37,
@@ -218,7 +218,7 @@
           "refId": "A"
         }
       ],
-      "title": "Payload Job Deadline",
+      "title": "Payload Deadline (block time)",
       "type": "stat"
     },
     {
@@ -248,7 +248,7 @@
       "gridPos": {
         "h": 4,
         "w": 3,
-        "x": 5,
+        "x": 7,
         "y": 1
       },
       "id": 18,
@@ -316,8 +316,8 @@
       },
       "gridPos": {
         "h": 4,
-        "w": 4,
-        "x": 8,
+        "w": 3,
+        "x": 10,
         "y": 1
       },
       "id": 5,
@@ -352,7 +352,7 @@
           "refId": "A"
         }
       ],
-      "title": "Total Transactions Count",
+      "title": "Transactions Built",
       "type": "stat"
     },
     {
@@ -382,8 +382,8 @@
       },
       "gridPos": {
         "h": 4,
-        "w": 4,
-        "x": 12,
+        "w": 3,
+        "x": 13,
         "y": 1
       },
       "id": 6,
@@ -621,7 +621,7 @@
           "refId": "A"
         }
       ],
-      "title": "Failed Payload Jobs",
+      "title": "Failed Jobs",
       "type": "stat"
     },
     {
@@ -3216,7 +3216,6 @@
   "timepicker": {},
   "timezone": "browser",
   "title": "Flashblock - Standard Builder",
-  "uid": "5211c89f-e5bc-4e58-8210-237af32b0d89",
-  "version": 4,
+  "version": 5,
   "weekStart": ""
 }

src/pipelines/mod.rs

@@ -119,10 +119,15 @@ impl<P: Platform> Pipeline<P> {
 	}
 
 	/// Sets payload building limits for the pipeline.
+	///
+	/// Here we can either use an instance of `LimitsFactory` that generates
+	/// limits dynamically according to a user-defined logic, or we can use a
+	/// fixed `Limits` instance.
 	#[must_use]
-	pub fn with_limits<L: LimitsFactory<P>>(self, limits: L) -> Self {
+	pub fn with_limits<T, L: IntoLimitsFactory<P, T>>(self, limits: L) -> Self {
 		let mut this = self;
-		this.limits = Some(Box::new(limits) as Box<dyn LimitsFactory<P>>);
+		let factory = limits.into_limits_factory();
+		this.limits = Some(Box::new(factory) as Box<dyn LimitsFactory<P>>);
 		this
 	}
 }

src/platform/limits.rs

@@ -97,3 +97,274 @@ impl Limits {
 		}
 	}
 }
+
+/// Convinience trait that allows API users to either used a factory type or a
+/// concrete limits value in [`Pipeline::with_limits`].
+pub trait IntoLimitsFactory<P: Platform, Marker = ()> {
+	/// Convert the type into a limits factory.
+	fn into_limits_factory(self) -> impl LimitsFactory<P>;
+}
+
+impl<T: LimitsFactory<P>, P: Platform> IntoLimitsFactory<P, Variant<0>> for T {
+	fn into_limits_factory(self) -> impl LimitsFactory<P> {
+		self
+	}
+}
+
+impl<P: Platform> IntoLimitsFactory<P, Variant<1>> for Limits {
+	fn into_limits_factory(self) -> impl LimitsFactory<P> {
+		struct FixedLimits(Limits);
+		impl<P: Platform> LimitsFactory<P> for FixedLimits {
+			fn create(&self, _: &BlockContext<P>, _: Option<&Limits>) -> Limits {
+				self.0.clone()
+			}
+		}
+
+		FixedLimits(self)
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use {
+		super::*,
+		crate::test_utils::*,
+		core::{
+			num::NonZero,
+			sync::atomic::{AtomicU32, Ordering},
+		},
+		std::sync::Arc,
+	};
+
+	assert_is_dyn_safe!(LimitsFactory<P>, P: Platform);
+
+	#[derive(Debug)]
+	struct TestStep {
+		must_run: bool,
+		iter: AtomicU32,
+		break_after: AtomicU32,
+
+		minimum_iterations: Option<u32>,
+		maximum_iterations: Option<u32>,
+		expected_gas_limit: Option<u64>,
+		expected_deadline: Option<Duration>,
+		sleep_on_step: Option<Duration>,
+	}
+
+	impl Default for TestStep {
+		fn default() -> Self {
+			Self {
+				must_run: false,
+				iter: AtomicU32::new(0),
+				break_after: AtomicU32::new(1),
+				expected_gas_limit: None,
+				expected_deadline: None,
+				minimum_iterations: None,
+				maximum_iterations: None,
+				sleep_on_step: None,
+			}
+		}
+	}
+
+	impl TestStep {
+		#[allow(dead_code)]
+		pub fn break_after(self, iterations: u32) -> Self {
+			self.break_after.store(iterations, Ordering::Relaxed);
+			self
+		}
+
+		pub fn expect_gas_limit(mut self, gas_limit: u64) -> Self {
+			self.expected_gas_limit = Some(gas_limit);
+			self
+		}
+
+		pub fn expect_deadline(mut self, deadline: Duration) -> Self {
+			self.expected_deadline = Some(deadline);
+			self
+		}
+
+		pub fn expect_minimum_iterations(mut self, iterations: u32) -> Self {
+			self.minimum_iterations = Some(iterations);
+			self
+		}
+
+		pub fn expect_maximum_iterations(mut self, iterations: u32) -> Self {
+			self.maximum_iterations = Some(iterations);
+			self
+		}
+
+		pub fn sleep_on_step(mut self, duration: Duration) -> Self {
+			self.sleep_on_step = Some(duration);
+			self
+		}
+
+		#[track_caller]
+		pub fn must_run(mut self) -> Self {
+			self.must_run = true;
+			self
+		}
+	}
+
+	impl Drop for TestStep {
+		fn drop(&mut self) {
+			assert!(
+				!self.must_run || self.iter.load(Ordering::Relaxed) > 0,
+				"Step was expected to run, but did not."
+			);
+
+			if let Some(minimum_iterations) = self.minimum_iterations {
+				assert!(
+					self.iter.load(Ordering::Relaxed) >= minimum_iterations,
+					"Expected step to run at least {minimum_iterations} times, but ran \
+					 {} times.",
+					self.iter.load(Ordering::Relaxed)
+				);
+			}
+
+			if let Some(maximum_iterations) = self.maximum_iterations {
+				assert!(
+					self.iter.load(Ordering::Relaxed) <= maximum_iterations,
+					"Expected step to run at most {maximum_iterations} times, but ran \
+					 {} times.",
+					self.iter.load(Ordering::Relaxed)
+				);
+			}
+		}
+	}
+
+	impl<P: Platform> Step<P> for TestStep {
+		async fn step(
+			self: Arc<Self>,
+			payload: Checkpoint<P>,
+			ctx: StepContext<P>,
+		) -> ControlFlow<P> {
+			if let Some(sleep_duration) = self.sleep_on_step {
+				tokio::time::sleep(sleep_duration).await;
+			}
+
+			let break_after = self.break_after.load(Ordering::Relaxed);
+			if self.iter.fetch_add(1, Ordering::Relaxed) >= break_after {
+				return ControlFlow::Break(payload);
+			}
+
+			if ctx.deadline_reached() {
+				return ControlFlow::Break(payload);
+			}
+
+			if let Some(expected_gas_limit) = self.expected_gas_limit {
+				assert_eq!(
+					ctx.limits().gas_limit,
+					expected_gas_limit,
+					"Expected gas limit to be {expected_gas_limit}, but got {}",
+					ctx.limits().gas_limit
+				);
+			}
+
+			if let Some(expected_deadline) = self.expected_deadline {
+				assert_eq!(
+					ctx.limits().deadline,
+					Some(expected_deadline),
+					"Expected deadline to be {expected_deadline:?}, but got {:?}",
+					ctx.limits().deadline
+				);
+			}
+
+			ControlFlow::Ok(payload)
+		}
+	}
+
+	#[rblib_test(Ethereum, Optimism)]
+	async fn one_level_flat<P: TestablePlatform>() -> eyre::Result<()> {
+		let pipeline = Pipeline::<P>::default()
+			.with_step(
+				TestStep::default()
+					.expect_gas_limit(1000)
+					.expect_deadline(Duration::from_millis(500))
+					.must_run(),
+			)
+			.with_limits(
+				Limits::with_gas_limit(1000).with_deadline(Duration::from_millis(500)),
+			);
+		P::create_test_node(pipeline).await?.next_block().await?;
+		Ok(())
+	}
+
+	#[rblib_test(Ethereum, Optimism)]
+	async fn scale_fraction<P: TestablePlatform>() -> eyre::Result<()> {
+		let pipeline = Pipeline::<P>::default()
+			.with_pipeline(
+				Once,
+				(TestStep::default()
+					.expect_gas_limit(500)
+					.expect_deadline(Duration::from_millis(250))
+					.must_run(),)
+					.with_limits(
+						Scaled::default()
+							.gas(Fraction(1, NonZero::new(2).unwrap()))
+							.deadline(Fraction(1, NonZero::new(2).unwrap())),
+					),
+			)
+			.with_limits(
+				Limits::with_gas_limit(1000).with_deadline(Duration::from_millis(500)),
+			);
+		P::create_test_node(pipeline).await?.next_block().await?;
+		Ok(())
+	}
+
+	#[rblib_test(Ethereum, Optimism)]
+	async fn scale_minus<P: TestablePlatform>() -> eyre::Result<()> {
+		let pipeline = Pipeline::<P>::default()
+			.with_pipeline(
+				Once,
+				(TestStep::default()
+					.expect_gas_limit(900)
+					.expect_deadline(Duration::from_millis(400))
+					.must_run(),)
+					.with_limits(
+						Scaled::default()
+							.gas(Minus(100))
+							.deadline(Minus(Duration::from_millis(100))),
+					),
+			)
+			.with_limits(
+				Limits::with_gas_limit(1000).with_deadline(Duration::from_millis(500)),
+			);
+		P::create_test_node(pipeline).await?.next_block().await?;
+		Ok(())
+	}
+
+	/// two levels of scaling
+	#[rblib_test(Ethereum, Optimism)]
+	async fn deadline_approaches<P: TestablePlatform>() -> eyre::Result<()> {
+		let pipeline = Pipeline::<P>::default()
+			.with_pipeline(
+				Once,
+				Pipeline::default()
+					.with_pipeline(
+						Once,
+						(TestStep::default()
+							.expect_gas_limit(400)
+							.expect_deadline(Duration::from_millis(300))
+							.expect_minimum_iterations(2)
+							.expect_maximum_iterations(3)
+							.sleep_on_step(Duration::from_millis(100))
+							.must_run(),)
+							.with_limits(
+								Scaled::default()
+									.gas(Minus(100))
+									.deadline(Minus(Duration::from_millis(100))),
+							),
+					)
+					.with_limits(
+						Scaled::default()
+							.gas(Fraction(1, NonZero::new(2).unwrap()))
+							.deadline(Fraction(1, NonZero::new(2).unwrap())),
+					),
+			)
+			.with_limits(
+				Limits::with_gas_limit(1000).with_deadline(Duration::from_millis(800)),
+			);
+		P::create_test_node(pipeline).await?.next_block().await?;
+		Ok(())
+	}
+}