In-process rbuilder orderpool <> reth transaction-pool connection (#339)

Adds a new method to `LiveBuilder` `connect_to_transaction_pool`
allowing in-process connection between the reth transaction pool and
rbuilder orderpool.

This fixes an issue where the orderpool does not receive any
transactions that were cached on disk by reth on start up, and finally
removes the requirement for an ipc connection when running in-process.

Author: liamaharon

op-rbuilder/node/src/node.rs

@@ -27,7 +27,7 @@ use reth_primitives::TransactionSigned;
 use reth_provider::{BlockReader, CanonStateSubscriptions, DatabaseProviderFactory};
 use reth_tracing::tracing::{debug, info};
 use reth_transaction_pool::{
-    blobstore::DiskFileBlobStore, CoinbaseTipOrdering, EthPooledTransaction,
+    blobstore::DiskFileBlobStore, CoinbaseTipOrdering, EthPooledTransaction, Pool,
     TransactionValidationTaskExecutor,
 };
 use reth_trie_db::MerklePatriciaTrie;
@@ -194,16 +194,16 @@ where
                 .require_l1_data_gas_fee(!ctx.config().dev.dev)
         });
 
-        let bundle_ops = BundlePoolOps::new(ctx.provider().clone(), self.config)
-            .await
-            .expect("Failed to instantiate RbuilderBundlePoolOps");
-        let transaction_pool = OpRbuilderTransactionPool::new(
+        let tx_pool = Pool::new(
             validator,
             CoinbaseTipOrdering::default(),
             blob_store,
-            bundle_ops,
             ctx.pool_config(),
         );
+        let bundle_ops = BundlePoolOps::new(ctx.provider().clone(), tx_pool.clone(), self.config)
+            .await
+            .expect("Failed to instantiate RbuilderBundlePoolOps");
+        let transaction_pool = OpRbuilderTransactionPool::new(tx_pool, bundle_ops);
 
         info!(target: "reth::cli", "Transaction pool initialized");
         let transactions_path = data_dir.txpool_transactions();

op-rbuilder/payload_builder/Cargo.toml

@@ -48,4 +48,5 @@ optimism = [
 	"reth-provider/optimism",
 	"revm/optimism",
 	"reth-optimism-consensus/optimism",
+	"reth-optimism-payload-builder/optimism"
 ]

transaction-pool-bundle-ext/bundle_pool_ops/rbuilder/Cargo.toml

@@ -10,7 +10,9 @@ rbuilder = { path = "../../../rbuilder" }
 alloy-primitives.workspace = true
 alloy-rpc-types-beacon.workspace = true
 reth-primitives = { workspace = true }
-reth-provider = {workspace = true}
+reth-transaction-pool = { workspace = true }
+reth-provider = { workspace = true }
+
 derive_more = { workspace = true }
 eyre = { workspace = true }
 tokio = { workspace = true }

transaction-pool-bundle-ext/bundle_pool_ops/rbuilder/src/lib.rs

@@ -30,6 +30,9 @@ use rbuilder::{
 };
 use reth_primitives::TransactionSigned;
 use reth_provider::{BlockReader, DatabaseProviderFactory, HeaderProvider};
+use reth_transaction_pool::{
+    BlobStore, EthPooledTransaction, Pool, TransactionOrdering, TransactionValidator,
+};
 use tokio::{
     sync::{
         mpsc::{self, error::SendError},
@@ -88,13 +91,20 @@ impl SlotSource for OurSlotSource {
 }
 
 impl BundlePoolOps {
-    pub async fn new<P>(provider: P, config: Config) -> Result<Self, Error>
+    pub async fn new<P, V, T, S>(
+        provider: P,
+        pool: Pool<V, T, S>,
+        config: Config,
+    ) -> Result<Self, Error>
     where
         P: DatabaseProviderFactory<Provider: BlockReader>
             + reth_provider::StateProviderFactory
             + HeaderProvider
             + Clone
             + 'static,
+        V: TransactionValidator<Transaction = EthPooledTransaction> + 'static,
+        T: TransactionOrdering<Transaction = <V as TransactionValidator>::Transaction>,
+        S: BlobStore,
     {
         // Create the payload source to trigger new block building
         let cancellation_token = CancellationToken::new();
@@ -108,7 +118,6 @@ impl BundlePoolOps {
             block_building_helper_tx,
         };
 
-        // Spawn the builder!
         let builder_strategy = BuilderConfig {
             name: "mp-ordering".to_string(),
             builder: SpecificBuilderConfig::OrderingBuilder(OrderingBuilderConfig {
@@ -162,6 +171,10 @@ impl BundlePoolOps {
             .await
             .expect("Failed to start full telemetry server");
 
+            builder
+                .connect_to_transaction_pool(pool)
+                .await
+                .expect("Failed to connect to reth pool");
             builder.run().await.unwrap();
 
             Ok::<(), ()>

transaction-pool-bundle-ext/src/bundle_supported_pool.rs

@@ -9,10 +9,10 @@ use reth_primitives::PooledTransactionsElement;
 use reth_transaction_pool::{
     AllPoolTransactions, AllTransactionsEvents, BestTransactions, BestTransactionsAttributes,
     BlobStore, BlobStoreError, BlockInfo, CanonicalStateUpdate, EthPoolTransaction,
-    GetPooledTransactionLimit, NewBlobSidecar, NewTransactionEvent, Pool, PoolConfig, PoolResult,
-    PoolSize, PropagatedTransactions, TransactionEvents, TransactionListenerKind,
-    TransactionOrdering, TransactionOrigin, TransactionPool,
-    TransactionPoolExt as TransactionPoolBlockInfoExt, TransactionValidator, ValidPoolTransaction,
+    GetPooledTransactionLimit, NewBlobSidecar, NewTransactionEvent, Pool, PoolResult, PoolSize,
+    PropagatedTransactions, TransactionEvents, TransactionListenerKind, TransactionOrdering,
+    TransactionOrigin, TransactionPool, TransactionPoolExt as TransactionPoolBlockInfoExt,
+    TransactionValidator, ValidPoolTransaction,
 };
 use std::{collections::HashSet, future::Future, sync::Arc};
 use tokio::sync::mpsc::Receiver;
@@ -67,15 +67,9 @@ where
     S: BlobStore,
     B: BundlePoolOperations,
 {
-    pub fn new(
-        validator: V,
-        ordering: T,
-        blob_store: S,
-        bundle_ops: B,
-        tx_pool_config: PoolConfig,
-    ) -> Self {
+    pub fn new(pool: Pool<V, T, S>, bundle_ops: B) -> Self {
         Self {
-            tx_pool: Pool::<V, T, S>::new(validator, ordering, blob_store, tx_pool_config),
+            tx_pool: pool,
             bundle_pool: Arc::new(BundlePool::<B>::new(bundle_ops)),
         }
     }