From 0d94c57f221fb23fab1bd1160f36b9ee44c69c26 Mon Sep 17 00:00:00 2001 From: Onyeka Obi Date: Tue, 21 Jul 2026 21:11:15 -0700 Subject: [PATCH] feat: tag orders with their source (mempool vs RPC) Closes #122. The block-building algorithm needs to know whether an order arrived from the public mempool or was submitted directly to the builder's RPC server, but that origin was only inferable indirectly (the Order::Tx vs Order::Bundle shape, plus the out-of-band MempoolTxsDetector hash set). Record the origin as a first-class OrderSource on Order metadata: - add OrderSource { Mempool, RpcServer, Unknown } (default Unknown) and a `source` field on Metadata, with Order::source() / Order::set_source(); - stamp Mempool at both mempool ingresses (the IPC/WS txpool fetcher and the reth-embedded transaction-pool path) and RpcServer at the RPC ingress (send_order funnels eth_sendRawTransaction and eth_sendBundle); - Metadata is excluded from Order equality/hashing, so tagging the source never affects order identity or deduplication. The tag is deliberately not carried over the wire: it is a local record of how the order reached this builder, so a remote submitter cannot assert it. Orders reconstructed outside the live ingestion paths (backtests, tests) keep OrderSource::Unknown. Signed-off-by: Onyeka Obi --- crates/rbuilder-primitives/src/lib.rs | 114 +++++++++++++++++- .../backtest/build_block/synthetic_orders.rs | 4 +- crates/rbuilder/src/live_builder/mod.rs | 5 +- .../live_builder/order_input/rpc_server.rs | 7 +- .../order_input/txpool_fetcher.rs | 12 +- 5 files changed, 132 insertions(+), 10 deletions(-) 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()