adds unbundling check for merged blocks

crates/relay/Cargo.toml

@@ -6,6 +6,10 @@ repository.workspace = true
 rust-version.workspace = true
 version.workspace = true
 
+[features]
+# Exposes block-merging internals for benches/unbundling.rs.
+bench-internals = []
+
 [dependencies]
 alloy-consensus.workspace = true
 alloy-eips.workspace = true
@@ -80,6 +84,12 @@ zstd.workspace = true
 
 [dev-dependencies]
 tracing-subscriber.workspace = true
+criterion.workspace = true
 
 [build-dependencies]
 tonic-build.workspace = true
+
+[[bench]]
+harness = false
+name = "unbundling"
+required-features = ["bench-internals"]

crates/relay/benches/unbundling.rs

@@ -0,0 +1,164 @@
+//! Run with: `cargo bench -p helix-relay --features bench-internals --bench unbundling`
+
+use std::hint::black_box;
+
+use alloy_primitives::{B256, Bytes, keccak256};
+use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
+use helix_relay::{OrderTxs, find_unbundled_txs};
+use helix_tcp_types::merging::order::{BundleOrderRef, MergeOrderRef};
+use rustc_hash::FxHashMap;
+
+const BUNDLE_SIZE: usize = 3;
+
+fn idx_hash(i: u64) -> B256 {
+    let mut bytes = [0u8; 32];
+    bytes[24..].copy_from_slice(&i.to_be_bytes());
+    B256::from(bytes)
+}
+
+fn make_order(hashes: Vec<B256>) -> OrderTxs {
+    let order_ref = MergeOrderRef::Bundle(BundleOrderRef {
+        txs: (0..hashes.len() as u16).collect(),
+        reverting_txs: vec![],
+        dropping_txs: vec![],
+    });
+    OrderTxs::from_ref(&order_ref, &hashes)
+}
+
+/// `included_orders` are fully satisfied (drawn from `final_txs`); the rest
+/// reference hashes absent from `final_txs`, the common case per slot.
+fn build_scenario(
+    total_orders: usize,
+    included_orders: usize,
+    final_tx_count: usize,
+) -> (Vec<B256>, Vec<OrderTxs>) {
+    assert!(included_orders * BUNDLE_SIZE <= final_tx_count);
+    assert!(included_orders <= total_orders);
+
+    let final_txs: Vec<B256> = (0..final_tx_count as u64).map(idx_hash).collect();
+
+    let mut orders = Vec::with_capacity(total_orders);
+    for i in 0..included_orders {
+        let start = i * BUNDLE_SIZE;
+        orders.push(make_order(final_txs[start..start + BUNDLE_SIZE].to_vec()));
+    }
+
+    let mut next = final_tx_count as u64 + 1;
+    for _ in included_orders..total_orders {
+        let hashes: Vec<B256> = (0..BUNDLE_SIZE as u64).map(|j| idx_hash(next + j)).collect();
+        next += BUNDLE_SIZE as u64;
+        orders.push(make_order(hashes));
+    }
+
+    (final_txs, orders)
+}
+
+/// Scaling with orders sent to a builder connection this slot.
+fn bench_by_order_count(c: &mut Criterion) {
+    let mut group = c.benchmark_group("unbundling_by_order_count");
+    const FINAL_TX_COUNT: usize = 300;
+    const INCLUDED_ORDERS: usize = 30;
+
+    for total_orders in [100usize, 1_000, 5_000, 20_000] {
+        let (final_txs, orders) = build_scenario(total_orders, INCLUDED_ORDERS, FINAL_TX_COUNT);
+        group.throughput(Throughput::Elements(total_orders as u64));
+        group.bench_with_input(
+            BenchmarkId::from_parameter(total_orders),
+            &(final_txs, orders),
+            |b, (final_txs, orders)| {
+                let mut bundled = Vec::new();
+                let mut covered = Vec::new();
+                b.iter(|| {
+                    black_box(find_unbundled_txs(final_txs, orders, &mut bundled, &mut covered))
+                });
+            },
+        );
+    }
+    group.finish();
+}
+
+/// Scaling with the merged block's own tx count.
+fn bench_by_final_tx_count(c: &mut Criterion) {
+    let mut group = c.benchmark_group("unbundling_by_final_tx_count");
+    const TOTAL_ORDERS: usize = 2_000;
+    const INCLUDED_ORDERS: usize = 30;
+
+    for final_tx_count in [100usize, 300, 600] {
+        let (final_txs, orders) = build_scenario(TOTAL_ORDERS, INCLUDED_ORDERS, final_tx_count);
+        group.throughput(Throughput::Elements(final_tx_count as u64));
+        group.bench_with_input(
+            BenchmarkId::from_parameter(final_tx_count),
+            &(final_txs, orders),
+            |b, (final_txs, orders)| {
+                let mut bundled = Vec::new();
+                let mut covered = Vec::new();
+                b.iter(|| {
+                    black_box(find_unbundled_txs(final_txs, orders, &mut bundled, &mut covered))
+                });
+            },
+        );
+    }
+    group.finish();
+}
+
+/// Cost of resolving a single wire order ref into `OrderTxs`.
+fn bench_order_construction(c: &mut Criterion) {
+    let tx_hashes: Vec<B256> = (0..BUNDLE_SIZE as u64).map(idx_hash).collect();
+    let order_ref = MergeOrderRef::Bundle(BundleOrderRef {
+        txs: (0..BUNDLE_SIZE as u16).collect(),
+        reverting_txs: vec![],
+        dropping_txs: vec![],
+    });
+    c.bench_function("order_txs_from_ref", |b| {
+        b.iter(|| black_box(OrderTxs::from_ref(&order_ref, &tx_hashes)));
+    });
+}
+
+/// Cost of hashing a merged block's own tx list, the step that produces
+/// `find_unbundled_txs`'s `final_txs` input in `tile.rs`.
+fn bench_final_tx_hashing(c: &mut Criterion) {
+    let mut group = c.benchmark_group("final_tx_hashing");
+    for (label, tx_size, count) in
+        [("small", 110usize, 300usize), ("typical", 250, 300), ("large", 250, 600)]
+    {
+        let txs: Vec<Vec<u8>> = (0..count).map(|i| vec![(i % 256) as u8; tx_size]).collect();
+        group.throughput(Throughput::Elements(count as u64));
+        group.bench_function(label, |b| {
+            b.iter(|| {
+                let hashes: Vec<B256> = txs.iter().map(|tx| keccak256(tx.as_slice())).collect();
+                black_box(hashes)
+            });
+        });
+    }
+    group.finish();
+}
+
+/// Steady-state cost once a slot's tx bytes are cached: same tx set as
+/// `final_tx_hashing/typical`, but the cache is warmed before `b.iter`, as it
+/// would be after the outgoing forwarding pass or an earlier merged block.
+fn bench_final_tx_hashing_cached(c: &mut Criterion) {
+    let txs: Vec<Bytes> = (0..300).map(|i| Bytes::from(vec![(i % 256) as u8; 250])).collect();
+    let mut cache: FxHashMap<Bytes, B256> = FxHashMap::default();
+    for tx in &txs {
+        cache.insert(tx.clone(), keccak256(tx.as_ref()));
+    }
+    c.bench_function("final_tx_hashing_cached", |b| {
+        b.iter(|| {
+            let hashes: Vec<B256> = txs
+                .iter()
+                .map(|tx| *cache.entry(tx.clone()).or_insert_with(|| keccak256(tx.as_ref())))
+                .collect();
+            black_box(hashes)
+        });
+    });
+}
+
+criterion_group!(
+    benches,
+    bench_by_order_count,
+    bench_by_final_tx_count,
+    bench_order_construction,
+    bench_final_tx_hashing,
+    bench_final_tx_hashing_cached
+);
+criterion_main!(benches);

crates/relay/src/block_merging/mod.rs

@@ -4,6 +4,7 @@
 //! into the auction.
 
 mod tile;
+mod unbundling;
 
 use std::collections::HashMap;
 
@@ -21,6 +22,9 @@ use helix_types::{
 use rustc_hash::FxHashMap;
 use ssz::Encode;
 pub use tile::BlockMergingTile;
+// Bench-only visibility, see benches/unbundling.rs.
+#[cfg(feature = "bench-internals")]
+pub use unbundling::{OrderTxs, find_unbundled_txs};
 
 use crate::simulator::BlockMergeResponse;
 

crates/relay/src/block_merging/tile.rs

@@ -32,6 +32,7 @@ use crate::{
     block_merging::{
         append_frame, merged_block_to_response, order_ref_hash, order_to_ref,
         submission_blob_sidecars,
+        unbundling::{OrderTxs, find_unbundled_txs},
     },
     housekeeper::SlotUpdate,
     simulator::BlockMergeResponse,
@@ -100,6 +101,9 @@ struct SlotState {
     mergeable_ixs: Vec<usize>,
     /// base_block_hash -> best proposer_value across all builders.
     best_merged: FxHashMap<B256, U256>,
+    /// Orders actually sent to the merge builder this slot, for the
+    /// unbundling check on incoming merged blocks.
+    order_txs: Vec<OrderTxs>,
 }
 
 /// Per-slot counters, logged and reset on slot transition.
@@ -125,6 +129,8 @@ struct SlotStats {
     merged_regressed: usize,
     /// Merged blocks dropped because an appended blob's sidecar wasn't in our cache.
     merged_blob_missing: usize,
+    /// Merged blocks dropped because the builder broke an order's atomicity.
+    merged_unbundled: usize,
     /// TopBidUpdate messages received for the current bid slot.
     top_bid_updates: usize,
     /// ActivateBaseBlockV1 frames sent.
@@ -157,6 +163,9 @@ pub struct BlockMergingTile {
     /// versioned hash, so an appended blob tx can be re-attached regardless of which
     /// builder's submission it originated from. Cleared on slot transition.
     blob_sidecars: FxHashMap<B256, BlobWithMetadata>,
+    /// Memoizes keccak256(tx bytes), shared by the outgoing per-submission
+    /// hashing and the incoming merged-block check. Cleared on slot transition.
+    tx_hash_cache: FxHashMap<Bytes, B256>,
 
     redial: Repeater,
     ping: Repeater,
@@ -174,6 +183,9 @@ pub struct BlockMergingTile {
     pongs: Vec<(Token, u64)>,
     merged_ixs: Vec<usize>,
     encode_buf: Vec<u8>,
+    // Scratch space for `find_unbundled_txs`, reused across calls.
+    unbundled_scratch_bundled: Vec<bool>,
+    unbundled_scratch_covered: Vec<bool>,
 }
 
 impl Tile<HelixSpine> for BlockMergingTile {
@@ -264,6 +276,7 @@ impl BlockMergingTile {
             chain_info,
             hydration_cache: HydrationCache::new(),
             blob_sidecars: FxHashMap::default(),
+            tx_hash_cache: FxHashMap::default(),
             redial: Repeater::every(Duration::from_secs(REDIAL_INTERVAL_S)),
             ping: Repeater::every(Duration::from_secs(PING_INTERVAL_S)),
             ping_nonce: 0,
@@ -276,6 +289,8 @@ impl BlockMergingTile {
             pongs: Vec::new(),
             merged_ixs: Vec::new(),
             encode_buf: Vec::new(),
+            unbundled_scratch_bundled: Vec::new(),
+            unbundled_scratch_covered: Vec::new(),
         }
     }
 
@@ -343,6 +358,9 @@ impl BlockMergingTile {
             merged_ixs,
             merged_blocks,
             blob_sidecars,
+            tx_hash_cache,
+            unbundled_scratch_bundled,
+            unbundled_scratch_covered,
             ..
         } = self;
 
@@ -448,6 +466,31 @@ impl BlockMergingTile {
                             );
                             return;
                         };
+                        let final_txs: Vec<B256> = response
+                            .execution_payload
+                            .transactions
+                            .iter()
+                            .map(|tx| {
+                                *tx_hash_cache
+                                    .entry(tx.0.clone())
+                                    .or_insert_with(|| keccak256(tx.as_ref()))
+                            })
+                            .collect();
+                        let unbundled = find_unbundled_txs(
+                            &final_txs,
+                            &slot.order_txs,
+                            unbundled_scratch_bundled,
+                            unbundled_scratch_covered,
+                        );
+                        if !unbundled.is_empty() {
+                            stats.merged_unbundled += 1;
+                            warn!(
+                                ?token,
+                                count = unbundled.len(),
+                                "merge builder unbundled an order, dropping merged block"
+                            );
+                            return;
+                        }
                         stats.merged_blocks += 1;
                         let ix = merged_blocks.push(response);
                         merged_ixs.push(ix);
@@ -515,6 +558,7 @@ impl BlockMergingTile {
             self.conn.reset_slot();
             self.hydration_cache.clear();
             self.blob_sidecars.clear();
+            self.tx_hash_cache.clear();
             // sole producer; consumed indices are stale after the transition
             self.merged_blocks.clear();
         }
@@ -582,6 +626,7 @@ impl BlockMergingTile {
             merged_stale = stats.merged_stale,
             merged_regressed = stats.merged_regressed,
             merged_blob_missing = stats.merged_blob_missing,
+            merged_unbundled = stats.merged_unbundled,
             appendable_blocks = self.slot.appendable.len(),
             hydration_txs = self.hydration_cache.tx_count(),
             hydration_builders = self.hydration_cache.builder_count(),
@@ -696,7 +741,9 @@ impl BlockMergingTile {
             .payload_inner
             .transactions
             .iter()
-            .map(|tx| keccak256(tx.as_ref()))
+            .map(|tx| {
+                *self.tx_hash_cache.entry(tx.clone()).or_insert_with(|| keccak256(tx.as_ref()))
+            })
             .collect();
 
         // Dehydrate relative to this connection: a tx already sent whole
@@ -757,6 +804,9 @@ impl BlockMergingTile {
             return;
         }
         self.conn.orders_sent.extend(order_hashes);
+        self.slot
+            .order_txs
+            .extend(msg.merge_orders.iter().map(|o| OrderTxs::from_ref(o, &tx_hashes)));
         if msg.allow_appending {
             self.conn.forwarded.insert(block_hash);
         }