feat: tag orders with their source (mempool vs RPC)

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
+/// <https://github.com/flashbots/rbuilder/issues/122>.
+#[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<Address>,
     /// `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,14 +124,26 @@ 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 {
     fn size(&self) -> usize {
         mem::size_of::<time::OffsetDateTime>() + // received_at_timestamp
             mem::size_of::<Option<Address>>() + // refund_identity
             mem::size_of::<bool>() + // is_system
-            mem::size_of::<bool>() // disable_cross_region_sharing
+            mem::size_of::<bool>() + // disable_cross_region_sharing
+            mem::size_of::<OrderSource>() // 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<Order> =
+            [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() {

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<ConfigType: LiveBuilderConfig> SyntheticOrdersSource<ConfigType> {
                     is_system: false,
                     refund_identity: None,
                     disable_cross_region_sharing: false,
+                    source: OrderSource::Unknown,
                 },
                 dropping_tx_hashes: Default::default(),
                 refund: None,

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<V, T, S>(
         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);

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<ReplaceableOrderPoolCommand>,
     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,

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()