Add txn monitoring in op-rbuilder pool (flashbots#500)

## 📝 Summary

This PR introduces a monitoring service that prints all the transactions
that arrive to the transaction pool in op-rbuilder.

## 💡 Motivation and Context

<!--- (Optional) Why is this change required? What problem does it
solve? Remove this section if not applicable. -->

---

## ✅ I have completed the following steps:

* [ ] Run `make lint`
* [ ] Run `make test`
* [ ] Added tests (if applicable)

Author: ferranbt

crates/op-rbuilder/src/args.rs

@@ -38,4 +38,8 @@ pub struct OpRbuilderArgs {
         env = "FLASHBLOCK_BLOCK_TIME"
     )]
     pub flashblock_block_time: u64,
+
+    /// Signals whether to log pool transactions events
+    #[arg(long = "builder.log-pool-transactions", default_value = "false")]
+    pub log_pool_transactions: bool,
 }

crates/op-rbuilder/src/main.rs

@@ -9,13 +9,15 @@ use reth_optimism_node::OpNode;
 use payload_builder::CustomOpPayloadBuilder;
 #[cfg(not(feature = "flashblocks"))]
 use payload_builder_vanilla::CustomOpPayloadBuilder;
+use reth_transaction_pool::TransactionPool;
 
 /// CLI argument parsing.
 pub mod args;
 pub mod generator;
 #[cfg(test)]
 mod integration;
 mod metrics;
+mod monitor_tx_pool;
 mod monitoring;
 #[cfg(feature = "flashblocks")]
 pub mod payload_builder;
@@ -25,6 +27,7 @@ mod primitives;
 #[cfg(test)]
 mod tester;
 mod tx_signer;
+use monitor_tx_pool::monitor_tx_pool;
 
 fn main() {
     Cli::<OpChainSpecParser, args::OpRbuilderArgs>::parse()
@@ -50,6 +53,16 @@ fn main() {
                     let new_canonical_blocks = ctx.provider().canonical_state_stream();
                     let builder_signer = builder_args.builder_signer;
 
+                    if builder_args.log_pool_transactions {
+                        tracing::info!("Logging pool transactions");
+                        ctx.task_executor.spawn_critical(
+                            "txlogging",
+                            Box::pin(async move {
+                                monitor_tx_pool(ctx.pool.all_transactions_event_listener()).await;
+                            }),
+                        );
+                    }
+
                     ctx.task_executor.spawn_critical(
                         "monitoring",
                         Box::pin(async move {

crates/op-rbuilder/src/monitor_tx_pool.rs

@@ -0,0 +1,62 @@
+use futures_util::StreamExt;
+use reth_optimism_node::txpool::OpPooledTransaction;
+use reth_transaction_pool::{AllTransactionsEvents, FullTransactionEvent};
+use tracing::debug;
+
+pub async fn monitor_tx_pool(mut new_transactions: AllTransactionsEvents<OpPooledTransaction>) {
+    while let Some(event) = new_transactions.next().await {
+        transaction_event_log(event);
+    }
+}
+
+fn transaction_event_log(event: FullTransactionEvent<OpPooledTransaction>) {
+    match event {
+        FullTransactionEvent::Pending(hash) => {
+            debug!(
+                tx_hash = hash.to_string(),
+                kind = "pending",
+                "Transaction event received"
+            )
+        }
+        FullTransactionEvent::Queued(hash) => {
+            debug!(
+                tx_hash = hash.to_string(),
+                kind = "queued",
+                "Transaction event received"
+            )
+        }
+        FullTransactionEvent::Mined {
+            tx_hash,
+            block_hash,
+        } => debug!(
+            tx_hash = tx_hash.to_string(),
+            kind = "mined",
+            block_hash = block_hash.to_string(),
+            "Transaction event received"
+        ),
+        FullTransactionEvent::Replaced {
+            transaction,
+            replaced_by,
+        } => debug!(
+            tx_hash = transaction.hash().to_string(),
+            kind = "replaced",
+            replaced_by = replaced_by.to_string(),
+            "Transaction event received"
+        ),
+        FullTransactionEvent::Discarded(hash) => {
+            debug!(
+                tx_hash = hash.to_string(),
+                kind = "discarded",
+                "Transaction event received"
+            )
+        }
+        FullTransactionEvent::Invalid(hash) => {
+            debug!(
+                tx_hash = hash.to_string(),
+                kind = "invalid",
+                "Transaction event received"
+            )
+        }
+        FullTransactionEvent::Propagated(_propagated) => {}
+    }
+}