diff --git a/crates/rbuilder-primitives/src/lib.rs b/crates/rbuilder-primitives/src/lib.rs index d80b6b826..b6c5e2b18 100644 --- a/crates/rbuilder-primitives/src/lib.rs +++ b/crates/rbuilder-primitives/src/lib.rs @@ -41,6 +41,24 @@ use uuid::Uuid; use crate::serialize::TxEncoding; +/// Where an [`Order`] entered the builder from. +/// +/// The building algorithm can treat orders differently depending on how they +/// arrived: transactions observed in the public mempool versus orders submitted +/// directly to the builder's RPC server. See +/// . +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)] +pub enum OrderSource { + /// Derived from a transaction observed in the public mempool. + Mempool, + /// Submitted directly to the builder's order-input RPC server + /// (`eth_sendRawTransaction` or `eth_sendBundle`). + RpcServer, + /// Origin not recorded, e.g. orders reconstructed during backtests or tests. + #[default] + Unknown, +} + /// Extra metadata for an order. #[derive(Debug, Clone, PartialEq, Eq, Hash)] pub struct Metadata { @@ -52,6 +70,9 @@ pub struct Metadata { pub refund_identity: Option
, /// `RawBundle` field, round-tripped through `Bundle`. Not consumed by rbuilder. pub disable_cross_region_sharing: bool, + /// Where the order entered the builder from. Defaults to + /// [`OrderSource::Unknown`]. + pub source: OrderSource, } impl Default for Metadata { @@ -73,6 +94,7 @@ impl Metadata { is_system: false, refund_identity: None, disable_cross_region_sharing: false, + source: OrderSource::Unknown, } } @@ -102,6 +124,17 @@ impl Metadata { self.disable_cross_region_sharing = disable_cross_region_sharing; self } + + /// Set the order source and return the metadata. + pub fn with_source(mut self, source: OrderSource) -> Self { + self.set_source(source); + self + } + + /// Set the order source. + pub fn set_source(&mut self, source: OrderSource) { + self.source = source; + } } impl InMemorySize for Metadata { @@ -109,7 +142,8 @@ impl InMemorySize for Metadata { mem::size_of::() + // received_at_timestamp mem::size_of::>() + // refund_identity mem::size_of::() + // is_system - mem::size_of::() // disable_cross_region_sharing + mem::size_of::() + // disable_cross_region_sharing + mem::size_of::() // source } } @@ -902,6 +936,22 @@ impl Order { Order::Tx(tx) => &tx.tx_with_blobs.metadata, } } + + /// Where this order entered the builder from (mempool vs RPC server). + pub fn source(&self) -> OrderSource { + self.metadata().source + } + + /// Record where this order entered the builder from. + /// + /// [`Metadata`] is excluded from [`Order`] equality and hashing, so tagging + /// the source does not affect order identity or deduplication. + pub fn set_source(&mut self, source: OrderSource) { + match self { + Order::Bundle(bundle) => bundle.metadata.set_source(source), + Order::Tx(tx) => tx.tx_with_blobs.metadata.set_source(source), + } + } } #[derive(Debug, Default, Clone, PartialEq, Eq, Hash)] @@ -1225,6 +1275,68 @@ mod tests { use reth_ethereum_primitives::{Transaction, TransactionSigned}; use uuid::uuid; + #[test] + fn order_source_defaults_to_unknown() { + assert_eq!(OrderSource::default(), OrderSource::Unknown); + assert_eq!(Metadata::new_received_now().source, OrderSource::Unknown); + } + + #[test] + fn metadata_set_source_round_trips() { + let with = Metadata::new_received_now().with_source(OrderSource::RpcServer); + assert_eq!(with.source, OrderSource::RpcServer); + + let mut meta = Metadata::new_received_now(); + meta.set_source(OrderSource::Mempool); + assert_eq!(meta.source, OrderSource::Mempool); + } + + #[test] + /// Tagging an order's source must update `source()` for both variants while + /// leaving order identity untouched (`Metadata` is ignored by `Order` Eq/Hash). + fn order_set_source_tags_both_variants_without_changing_identity() { + let mut data = TestDataGenerator::default(); + + let tx_sender = AccountNonce { + nonce: 0, + account: data.create_address(), + }; + let mut tx_order = data.create_tx_order(tx_sender); + assert_eq!(tx_order.source(), OrderSource::Unknown); + let original_tx = tx_order.clone(); + tx_order.set_source(OrderSource::Mempool); + assert_eq!(tx_order.source(), OrderSource::Mempool); + assert_eq!(tx_order.id(), original_tx.id()); + assert_eq!(tx_order, original_tx); + + let bundle_tx = BundledTxInfo { + nonce: AccountNonce { + nonce: 0, + account: data.create_address(), + }, + optional: false, + }; + let mut bundle_order = data.create_bundle_multi_tx_order(1, &[bundle_tx], None); + assert_eq!(bundle_order.source(), OrderSource::Unknown); + let original_bundle = bundle_order.clone(); + bundle_order.set_source(OrderSource::RpcServer); + assert_eq!(bundle_order.source(), OrderSource::RpcServer); + assert_eq!(bundle_order.id(), original_bundle.id()); + assert_eq!(bundle_order, original_bundle); + + // Metadata is excluded from Order's Hash as well: each tagged order hashes + // equal to its untagged original, so the two distinct orders collapse to + // exactly two entries in a HashSet. Order carries interior-mutable blob + // state that its Hash impl deliberately ignores, so mutable_key_type is a + // false positive here. + #[allow(clippy::mutable_key_type)] + let deduped: std::collections::HashSet = + [tx_order, original_tx, bundle_order, original_bundle] + .into_iter() + .collect(); + assert_eq!(deduped.len(), 2); + } + #[test] /// A bundle with a single optional tx paying enough gas should be considered executable fn can_execute_single_optional_tx() { diff --git a/crates/rbuilder/src/backtest/build_block/synthetic_orders.rs b/crates/rbuilder/src/backtest/build_block/synthetic_orders.rs index 3f337eef7..8d358816b 100644 --- a/crates/rbuilder/src/backtest/build_block/synthetic_orders.rs +++ b/crates/rbuilder/src/backtest/build_block/synthetic_orders.rs @@ -2,7 +2,8 @@ use alloy_primitives::B256; use clap::Parser; use rbuilder_config::load_toml_config; use rbuilder_primitives::{ - Bundle, MempoolTx, Metadata, Order, TransactionSignedEcRecoveredWithBlobs, LAST_BUNDLE_VERSION, + Bundle, MempoolTx, Metadata, Order, OrderSource, TransactionSignedEcRecoveredWithBlobs, + LAST_BUNDLE_VERSION, }; use reth_provider::test_utils::MockNodeTypesWithDB; use std::sync::Arc; @@ -99,6 +100,7 @@ impl SyntheticOrdersSource { is_system: false, refund_identity: None, disable_cross_region_sharing: false, + source: OrderSource::Unknown, }, dropping_tx_hashes: Default::default(), refund: None, diff --git a/crates/rbuilder/src/live_builder/mod.rs b/crates/rbuilder/src/live_builder/mod.rs index 4ca09fb29..82a1f8baf 100644 --- a/crates/rbuilder/src/live_builder/mod.rs +++ b/crates/rbuilder/src/live_builder/mod.rs @@ -40,7 +40,7 @@ use futures::{stream::FuturesUnordered, StreamExt}; use jsonrpsee::RpcModule; use order_input::{mempool_txs_detector::MempoolTxsDetector, ReplaceableOrderPoolCommand}; use payload_events::{InternalPayloadId, MevBoostSlotDataGenerator}; -use rbuilder_primitives::{MempoolTx, Order, TransactionSignedEcRecoveredWithBlobs}; +use rbuilder_primitives::{MempoolTx, Order, OrderSource, TransactionSignedEcRecoveredWithBlobs}; use reth::transaction_pool::{ BlobStore, EthPooledTransaction, Pool, TransactionListenerKind, TransactionOrdering, TransactionPool, TransactionValidator, @@ -554,7 +554,8 @@ async fn try_send_to_orderpool( Ok(tx) => { let tx_hash = tx.hash(); mempool_detector.add_tx(tx_hash); - let order = Order::Tx(MempoolTx::new(tx)); + let mut order = Order::Tx(MempoolTx::new(tx)); + order.set_source(OrderSource::Mempool); let command = ReplaceableOrderPoolCommand::Order(Arc::new(order)); if let Err(e) = orderpool_sender.send(command).await { mempool_detector.remove_tx(tx_hash); diff --git a/crates/rbuilder/src/live_builder/order_input/rpc_server.rs b/crates/rbuilder/src/live_builder/order_input/rpc_server.rs index e04e2fd53..a3ff990c4 100644 --- a/crates/rbuilder/src/live_builder/order_input/rpc_server.rs +++ b/crates/rbuilder/src/live_builder/order_input/rpc_server.rs @@ -11,7 +11,7 @@ use jsonrpsee::{ }; use rbuilder_primitives::{ serialize::{RawBundle, RawBundleDecodeResult, RawTx, TxEncoding}, - BundleReplacementData, BundleReplacementKey, MempoolTx, Order, OrderId, + BundleReplacementData, BundleReplacementKey, MempoolTx, Order, OrderId, OrderSource, }; use serde::Deserialize; use std::{ @@ -217,11 +217,14 @@ async fn handle_eth_send_bundle( } async fn send_order( - order: Order, + mut order: Order, channel: &mpsc::Sender, timeout: Duration, received_at: OffsetDateTime, ) { + // Every order reaching this helper was submitted directly to the builder's + // RPC server: both eth_sendRawTransaction and eth_sendBundle funnel here. + order.set_source(OrderSource::RpcServer); send_command( ReplaceableOrderPoolCommand::Order(Arc::new(order)), channel, diff --git a/crates/rbuilder/src/live_builder/order_input/txpool_fetcher.rs b/crates/rbuilder/src/live_builder/order_input/txpool_fetcher.rs index 6b0b6ec59..8dff32261 100644 --- a/crates/rbuilder/src/live_builder/order_input/txpool_fetcher.rs +++ b/crates/rbuilder/src/live_builder/order_input/txpool_fetcher.rs @@ -4,7 +4,7 @@ use alloy_primitives::FixedBytes; use alloy_provider::{IpcConnect, Provider, ProviderBuilder}; use futures::StreamExt; use rbuilder_primitives::{ - serialize::TxEncoding, MempoolTx, Order, RawTransactionDecodable, + serialize::TxEncoding, MempoolTx, Order, OrderSource, RawTransactionDecodable, TransactionSignedEcRecoveredWithBlobs, }; use std::{pin::pin, sync::Arc, time::Instant}; @@ -71,7 +71,8 @@ pub async fn subscribe_to_txpool_with_blobs( } }; let tx = MempoolTx::new(tx_with_blobs); - let order = Order::Tx(tx); + let mut order = Order::Tx(tx); + order.set_source(OrderSource::Mempool); let parse_duration = start.elapsed(); trace!(order = ?order.id(), parse_duration_mus = parse_duration.as_micros(), "Mempool transaction received with blobs"); add_txfetcher_time_to_query(parse_duration); @@ -182,9 +183,11 @@ mod test { let pending_tx = provider.send_transaction(tx).await.unwrap(); let recv_tx = receiver.recv().await.unwrap(); - let tx_with_blobs = match recv_tx { + let (tx_with_blobs, source) = match recv_tx { ReplaceableOrderPoolCommand::Order(order) => match order.as_ref() { - Order::Tx(MempoolTx { tx_with_blobs }) => Some(tx_with_blobs.clone()), + Order::Tx(MempoolTx { tx_with_blobs }) => { + Some((tx_with_blobs.clone(), order.source())) + } _ => None, }, _ => None, @@ -193,6 +196,7 @@ mod test { assert_eq!(tx_with_blobs.hash(), *pending_tx.tx_hash()); assert_eq!(tx_with_blobs.blobs_len(), 1); + assert_eq!(source, OrderSource::Mempool); // send another tx without blobs let tx = TransactionRequest::default()