Better payload job log message

Author: karim-agha

src/pipelines/job.rs

@@ -1,20 +1,23 @@
 use {
 	super::traits,
 	crate::{
+		alloy,
 		pipelines::{exec::PipelineExecutor, service::ServiceContext},
 		prelude::*,
-		reth::{
-			api::PayloadBuilderAttributes,
-			payload::builder::{PayloadJob as RethPayloadJobTrait, *},
-		},
+		reth,
 	},
+	alloy::consensus::BlockHeader,
 	core::{
 		pin::Pin,
 		task::{Context, Poll},
 	},
 	futures::{FutureExt, future::Shared},
-	reth_node_builder::BuiltPayload,
-	std::sync::Arc,
+	reth::{
+		api::PayloadBuilderAttributes,
+		node::builder::{BlockBody, BuiltPayload},
+		payload::builder::{PayloadJob as RethPayloadJobTrait, *},
+	},
+	std::{fmt::Write as _, sync::Arc, time::Instant},
 	tracing::{debug, warn},
 };
 
@@ -146,6 +149,7 @@ where
 	Provider: traits::ProviderBounds<P>,
 {
 	payload_id: PayloadId,
+	started_at: Instant,
 	state: ExecutorFutureState<P, Provider>,
 }
 
@@ -173,10 +177,58 @@ where
 {
 	pub fn new(executor: PipelineExecutor<P, Provider>) -> Self {
 		Self {
+			started_at: Instant::now(),
 			payload_id: executor.payload_id(),
 			state: ExecutorFutureState::Future(executor.shared()),
 		}
 	}
+
+	fn log_payload_job_result(
+		&self,
+		result: &Result<types::BuiltPayload<P>, Arc<PayloadBuilderError>>,
+	) -> core::fmt::Result {
+		match &result {
+			Ok(built_payload) => {
+				let built_block = built_payload.block();
+				let fees = built_payload.fees();
+
+				let mut out = String::new();
+				writeln!(&mut out, "Payload job {}:", self.payload_id)?;
+				writeln!(&mut out, "  Status: Success")?;
+				writeln!(&mut out, "  Duration: {:?}", self.started_at.elapsed())?;
+				writeln!(&mut out, "  Fees: {fees}")?;
+				writeln!(&mut out, "  Built Block:")?;
+				writeln!(&mut out, "    Number: {:?}", built_block.number())?;
+				writeln!(&mut out, "    Hash: {:?}", built_block.hash())?;
+				writeln!(&mut out, "    Parent: {:?}", built_block.parent_hash())?;
+				writeln!(&mut out, "    Timestamp: {:?}", built_block.timestamp())?;
+				writeln!(
+					&mut out,
+					"    Gas: {}/{} ({}%)",
+					built_block.gas_used(),
+					built_block.gas_limit(),
+					built_block.gas_used() * 100 / built_block.gas_limit()
+				)?;
+				writeln!(
+					&mut out,
+					"    Transactions: {}",
+					built_block.body().transactions().len()
+				)?;
+
+				debug!("{out}");
+			}
+			Err(error) => {
+				let mut out = String::new();
+				writeln!(&mut out, "Payload job {}:", self.payload_id)?;
+				writeln!(&mut out, "  Status: Failed")?;
+				writeln!(&mut out, "  Duration: {:?}", self.started_at.elapsed())?;
+				writeln!(&mut out, "  Error: {error:#?}")?;
+				warn!("{out}");
+			}
+		}
+
+		Ok(())
+	}
 }
 
 impl<P, Provider> Future for ExecutorFuture<P, Provider>
@@ -202,7 +254,7 @@ where
 					Poll::Ready(result) => {
 						// got a result. All future polls will return the result directly
 						// without polling the executor again.
-						log_payload_job_result::<P>(this.payload_id, &result);
+						let _ = this.log_payload_job_result(&result);
 
 						this.state = ExecutorFutureState::Ready(result.clone());
 						Poll::Ready(
@@ -230,6 +282,7 @@ where
 	fn clone(&self) -> Self {
 		Self {
 			payload_id: self.payload_id,
+			started_at: self.started_at,
 			state: match &self.state {
 				ExecutorFutureState::Ready(result) => {
 					ExecutorFutureState::Ready(result.clone())
@@ -241,22 +294,3 @@ where
 		}
 	}
 }
-
-fn log_payload_job_result<P: Platform>(
-	payload_id: PayloadId,
-	result: &Result<types::BuiltPayload<P>, Arc<PayloadBuilderError>>,
-) {
-	match &result {
-		Ok(built_payload) => {
-			let built_block = built_payload.block();
-			let fees = built_payload.fees();
-
-			debug!(
-				"Payload job {payload_id} completed: {built_block:#?}, fees: {fees}"
-			);
-		}
-		Err(error) => {
-			warn!("Payload job {payload_id} failed: {error:#?}");
-		}
-	}
-}

src/pool/host.rs

@@ -11,6 +11,7 @@ use {
 			StateProviderFactory,
 		},
 		tasks::shutdown::Shutdown,
+		transaction_pool::TransactionPool,
 	},
 	std::sync::OnceLock,
 	tracing::debug,
@@ -104,6 +105,14 @@ impl<P: Platform> HostNode<P> {
 	pub fn system_pool(&self) -> Option<&impl traits::PoolBounds<P>> {
 		self.instances.get().map(|i| &i.system_pool)
 	}
+
+	/// If attached to a host node, this will remove a transaction from the reth
+	/// native transaction pool.
+	pub fn remove_transaction(&self, txhash: TxHash) {
+		if let Some(pool) = self.instances.get().map(|i| &i.system_pool) {
+			pool.remove_transactions(vec![txhash]);
+		}
+	}
 }
 
 impl<P: Platform> HostNode<P> {

src/pool/maintain.rs

@@ -49,6 +49,7 @@ impl<P: Platform> OrderPool<P> {
 				tokio::select! {
 					Some(OrderInclusionAttempt(order, payload_id)) = inclusion.next() => {
 						tracing::trace!(">--> order inclusion attempt: {order} in payload job {payload_id}");
+						order_pool.report_inclusion_attempt(order, payload_id);
 					}
 					Some(OrderInclusionSuccess(order, payload_id)) = success.next() => {
 						tracing::trace!(">--> order inclusion success: {order} in payload job {payload_id}");

src/pool/mod.rs

@@ -108,12 +108,14 @@ impl<P: Platform> OrderPool<P> {
 
 	/// Removes an order from the pool and makes it no longer available through
 	/// `best_orders()`.
-	pub fn remove(&self, order_hash: &B256) -> Option<Order<P>> {
-		self.inner.orders.remove(order_hash).map(|(_, order)| order)
+	pub fn remove(&self, order_hash: &B256) {
+		self.inner.orders.remove(order_hash);
+		self.inner.host.remove_transaction(*order_hash);
 	}
 
 	/// Removes all orders that contain the a specific transaction hash.
 	pub fn remove_any_with(&self, txhash: TxHash) {
+		self.inner.host.remove_transaction(txhash);
 		if let Some((_, orders)) = self.inner.txmap.remove(&txhash) {
 			for order_hash in orders {
 				self.remove(&order_hash);

src/pool/report.rs

@@ -7,6 +7,7 @@
 use {
 	super::*,
 	reth::{node::builder::BlockBody, primitives::SealedBlock},
+	reth_payload_builder::PayloadId,
 	tracing::trace,
 };
 
@@ -48,12 +49,8 @@ impl<P: Platform> OrderPool<P> {
 	/// and there was an attempt to include it in a payload. Once an order was
 	/// proposed once by the pool, it will be removed from the orders list and
 	/// will not be proposed again.
-	pub fn report_inclusion_attempt(
-		&self,
-		_order_hash: B256,
-		_block: &BlockContext<P>,
-	) {
-		// todo
+	pub fn report_inclusion_attempt(&self, _order_hash: B256, _: PayloadId) {
+		// self.remove(&order_hash);
 	}
 
 	/// Signals to the order pool that a block has been committed to the chain.

src/pool/step.rs

@@ -6,6 +6,7 @@ use {
 	dashmap::DashSet,
 	metrics::{Counter, Histogram},
 	reth::{
+		chainspec::MIN_TRANSACTION_GAS,
 		ethereum::primitives::SignedTransaction,
 		payload::builder::PayloadId,
 	},
@@ -279,7 +280,13 @@ impl<'a, P: Platform> Run<'a, P> {
 			return true;
 		}
 
-		if self.payload.cumulative_gas_used() >= self.limits().gas_limit {
+		let remaining_gas = self
+			.limits()
+			.gas_limit
+			.saturating_sub(self.payload.cumulative_gas_used());
+
+		if remaining_gas < MIN_TRANSACTION_GAS {
+			// won't be able to fit any more transactions
 			return true;
 		}
 
@@ -388,7 +395,6 @@ impl<'a, P: Platform> Run<'a, P> {
 			if candidate.is_bundle() {
 				self.step.metrics.bundles_skipped.increment(1);
 			}
-
 			return;
 		}