diff --git a/config.example.yml b/config.example.yml index 9bf61e26..8447b00a 100644 --- a/config.example.yml +++ b/config.example.yml @@ -38,9 +38,9 @@ block_merging_config: is_enabled: false max_merged_bid_age_ms: 250 # tcp: - # builders: - # - addr: "127.0.0.1:9876" - # api_key: "00000000-0000-0000-0000-000000000001" + # builder: + # addr: "127.0.0.1:9876" + # api_key: "00000000-0000-0000-0000-000000000001" # relay_fee_recipient: "0x0000000000000000000000000000000000000000" # multisend_contract: "0x0000000000000000000000000000000000000000" # relay_bps: 2500 diff --git a/crates/common/src/config.rs b/crates/common/src/config.rs index 5d25f4e5..b39dfe5c 100644 --- a/crates/common/src/config.rs +++ b/crates/common/src/config.rs @@ -263,7 +263,8 @@ pub struct BlockMergingConfig { /// on helix-tcp-types. #[derive(Serialize, Deserialize, Clone)] pub struct BlockMergingTcpConfig { - pub builders: Vec, + /// The single merging builder this relay dials. + pub builder: MergingBuilderEndpoint, pub relay_fee_recipient: Address, pub multisend_contract: Address, pub relay_bps: u64, diff --git a/crates/common/src/decoder.rs b/crates/common/src/decoder.rs index 6172b930..cd43f38d 100644 --- a/crates/common/src/decoder.rs +++ b/crates/common/src/decoder.rs @@ -8,9 +8,9 @@ use flate2::read::GzDecoder; use flux_profiler::timed; use helix_types::{ BidAdjustmentData, BlockMergingData, Compression, DehydratedBidSubmission, - DehydratedBidSubmissionFuluWithAdjustments, ForkName, ForkVersionDecode, MergeType, - SignedBidSubmission, SignedBidSubmissionWithAdjustments, SignedBidSubmissionWithMergingData, - Submission, + DehydratedBidSubmissionFuluWithAdjustments, DehydratedBidSubmissionFuluWithMergingData, + ForkName, ForkVersionDecode, MergeType, SignedBidSubmission, SignedBidSubmissionWithAdjustments, + SignedBidSubmissionWithMergingData, Submission, }; use http::{ HeaderMap, HeaderValue, StatusCode, @@ -279,6 +279,14 @@ impl SubmissionDecoder { body: &[u8], ) -> Result<(Submission, Option, Option), DecoderError> { + if self.merge_type == MergeType::Mergeable { + let sub_with_merging: DehydratedBidSubmissionFuluWithMergingData = + self.decode_by_fork(body, self.fork_name)?; + let (submission, merging_data) = sub_with_merging.split(); + + return Ok((Submission::Dehydrated(submission), Some(merging_data), None)); + } + let (submission, bid_adjustment) = if self.with_adjustments { let sub_with_adjustment: DehydratedBidSubmissionFuluWithAdjustments = self.decode_by_fork(body, self.fork_name)?; @@ -292,11 +300,7 @@ impl SubmissionDecoder { }; let merging_data = match self.merge_type { - MergeType::Mergeable => { - //Should this return an error instead? - error!("mergeable dehydrated submissions are not supported"); - None - } + MergeType::Mergeable => unreachable!("handled above"), MergeType::AppendOnly => { Some(BlockMergingData::append_only(submission.fee_recipient())) } @@ -474,7 +478,8 @@ fn gzip_size_hint(buf: &[u8]) -> Option { #[cfg(test)] mod tests { - use helix_types::MergeType; + use helix_types::{DehydratedBidSubmissionFuluWithMergingData, MergeType, TestRandom}; + use ssz::Encode; use super::*; @@ -520,4 +525,42 @@ mod tests { assert!("invalid".parse::().is_err()); assert!("AppendOnly".parse::().is_err()); // CamelCase should fail } + + #[test] + fn decode_dehydrated_mergeable_carries_real_merge_orders() { + // `random_for_test` may produce an empty `merge_orders` (it's randomly sized); retry + // until we get a non-empty one, since that's specifically what this test is proving + // survives the decode (previously it was always dropped/forced-empty for dehydrated + // submissions). + let (body, expected_merging_data) = (0..100) + .find_map(|_| { + let submission = + DehydratedBidSubmissionFuluWithMergingData::random_for_test(&mut rand::rng()); + let (_, merging_data) = submission.clone().split(); + if merging_data.merge_orders.is_empty() { + return None; + } + Some((submission.as_ssz_bytes(), merging_data)) + }) + .expect("should produce a submission with non-empty merge_orders within 100 tries"); + + let params = SubmissionDecoderParams { + compression: Compression::None, + encoding: Encoding::Ssz, + merge_type: MergeType::Mergeable, + is_dehydrated: true, + with_mergeable_data: false, + with_adjustments: false, + block_merging_dry_run: false, + fork_name: ForkName::Fulu, + }; + let mut decoder = SubmissionDecoder::new(¶ms); + let mut buf = Vec::new(); + let (decoded_submission, merging_data, bid_adjustment_data) = + decoder.decode(&body, &mut buf).expect("decode should succeed"); + + assert!(matches!(decoded_submission, Submission::Dehydrated(_))); + assert!(bid_adjustment_data.is_none()); + assert_eq!(merging_data.expect("mergeable submission should carry merging data"), expected_merging_data); + } } diff --git a/crates/relay/src/api/builder/top_bid.rs b/crates/relay/src/api/builder/top_bid.rs index 8ea84738..b0de24ff 100644 --- a/crates/relay/src/api/builder/top_bid.rs +++ b/crates/relay/src/api/builder/top_bid.rs @@ -11,7 +11,7 @@ use helix_common::{ }; use hyper::HeaderMap; use tokio_tungstenite::tungstenite::{Error as WebSocketError, Message}; -use tracing::{debug, error}; +use tracing::{debug, error, info}; use super::api::BuilderApi; use crate::{ @@ -69,15 +69,70 @@ impl BuilderApi { } } +/// Per-slot counters, logged and reset when `TopBidUpdate.slot` advances. +/// Sends/pings are per-(update, connection) fan-out, a different unit from +/// `top_bid_updates_received`, which is per-update. +#[derive(Default)] +struct SlotStats { + new_connections: u32, + top_bid_updates_received: u32, + sends_ok: u32, + sends_failed: u32, + pings_sent: u32, + pings_failed: u32, + pongs_sent: u32, + pongs_failed: u32, + closes_received: u32, + read_errors: u32, +} + pub struct TopBidTile { new_connections: Receiver<(RawWebSocket, TopBidPrecision)>, connections: Vec<(RawWebSocket, TopBidMetrics, TopBidPrecision)>, last_send: Nanos, + bid_slot: u64, + stats: SlotStats, } impl TopBidTile { pub fn new(new_connections: Receiver<(RawWebSocket, TopBidPrecision)>) -> Self { - Self { new_connections, connections: vec![], last_send: Nanos::now() } + Self { + new_connections, + connections: vec![], + last_send: Nanos::now(), + bid_slot: 0, + stats: SlotStats::default(), + } + } + + fn on_top_bid_slot(&mut self, slot: u64) { + if slot <= self.bid_slot { + return; + } + self.report_slot_stats(); + self.bid_slot = slot; + } + + fn report_slot_stats(&mut self) { + if self.bid_slot == 0 { + return; + } + let stats = std::mem::take(&mut self.stats); + info!( + bid_slot = self.bid_slot, + connections = self.connections.len(), + new_connections = stats.new_connections, + top_bid_updates_received = stats.top_bid_updates_received, + sends_ok = stats.sends_ok, + sends_failed = stats.sends_failed, + pings_sent = stats.pings_sent, + pings_failed = stats.pings_failed, + pongs_sent = stats.pongs_sent, + pongs_failed = stats.pongs_failed, + closes_received = stats.closes_received, + read_errors = stats.read_errors, + "top bid slot stats" + ); } } @@ -87,17 +142,24 @@ impl Tile for TopBidTile { while let Ok((web_socket, precision)) = self.new_connections.try_recv() { let metric = TopBidMetrics::connection(); self.connections.push((web_socket, metric, precision)); + self.stats.new_connections += 1; } // send top bids - note that `send` call is non-blocking. adapter.consume(|top_bid: TopBidUpdate, _producers| { + self.on_top_bid_slot(top_bid.slot); + self.stats.top_bid_updates_received += 1; let mut i = 0; while i < self.connections.len() { let precision = self.connections[i].2; let payload = top_bid.as_ssz_bytes_with_precision(precision); match self.connections[i].0.send(Message::Binary(payload)) { - Ok(_) => i += 1, + Ok(_) => { + self.stats.sends_ok += 1; + i += 1; + } Err(e) => { + self.stats.sends_failed += 1; error!(error=?e, peer=?self.connections[i].0.get_ref().peer_addr(), "Failed to send bid. Disconnecting."); self.connections.swap_remove(i); } @@ -111,8 +173,12 @@ impl Tile for TopBidTile { let mut i = 0; while i < self.connections.len() { match self.connections[i].0.send(Message::Ping(Bytes::new())) { - Ok(_) => i += 1, + Ok(_) => { + self.stats.pings_sent += 1; + i += 1; + } Err(e) => { + self.stats.pings_failed += 1; error!(error=?e, peer=?self.connections[i].0.get_ref().peer_addr(), "Failed to send ping. Disconnecting."); self.connections.swap_remove(i); } @@ -127,14 +193,19 @@ impl Tile for TopBidTile { match self.connections[i].0.read() { Ok(msg) => match msg { Message::Ping(data) => match self.connections[i].0.send(Message::Pong(data)) { - Ok(_) => i += 1, + Ok(_) => { + self.stats.pongs_sent += 1; + i += 1; + } Err(e) => { + self.stats.pongs_failed += 1; error!(error=?e, peer=?self.connections[i].0.get_ref().peer_addr(), "Failed to send pong. Disconnecting."); self.connections.swap_remove(i); } }, Message::Close(_) => { debug!("Received close frame."); + self.stats.closes_received += 1; self.connections.swap_remove(i); } _ => i += 1, @@ -143,6 +214,7 @@ impl Tile for TopBidTile { i += 1; } Err(e) => { + self.stats.read_errors += 1; error!(error=?e, peer=?self.connections[i].0.get_ref().peer_addr(), "Failed to read. Disconnecting."); self.connections.swap_remove(i); } diff --git a/crates/relay/src/auctioneer/bid_sorter.rs b/crates/relay/src/auctioneer/bid_sorter.rs index 9154f6dd..9ba681f9 100644 --- a/crates/relay/src/auctioneer/bid_sorter.rs +++ b/crates/relay/src/auctioneer/bid_sorter.rs @@ -68,6 +68,9 @@ struct BidSorterTelemetry { subs: u32, /// Internal bid processing time of the sorter subs_process_time: Duration, + /// Submissions rejected as stale (a newer version already recorded for + /// that builder on that fork). `subs == stale_subs + sum(fork.subs)`. + stale_subs: u32, } struct ForkState { @@ -238,7 +241,7 @@ impl BidSorter { let entry = entry.get_mut(); if entry.version >= new_bid.version { trace!("bid is stale, ignore"); - // stale + self.local_telemetry.stale_subs += 1; return false; } else { *entry = new_bid; @@ -313,6 +316,7 @@ impl BidSorter { let tel = std::mem::take(&mut self.local_telemetry); let avg_sub_process = avg_duration(tel.subs_process_time, tel.subs); + let accepted_subs: u32 = self.forks.values().map(|s| s.subs).sum(); let fork_report: Vec<_> = self .forks .iter() @@ -320,11 +324,13 @@ impl BidSorter { .collect(); info!( - slot = self.curr_bid_slot, + bid_slot = self.curr_bid_slot, valid_subs = tel.subs, + accepted_subs, + stale_subs = tel.stale_subs, ?avg_sub_process, ?fork_report, - "bid sorter telemetry" + "bid sorter slot stats" ) } } diff --git a/crates/relay/src/auctioneer/block_merger.rs b/crates/relay/src/auctioneer/block_merger.rs index 8b4b191f..80df1ad4 100644 --- a/crates/relay/src/auctioneer/block_merger.rs +++ b/crates/relay/src/auctioneer/block_merger.rs @@ -71,13 +71,18 @@ pub struct BlockMerger { config: RelayConfig, chain_info: ChainInfo, local_cache: LocalCache, - best_merged_block: Option, + best_merged_blocks: HashMap, best_mergeable_orders: BestMergeableOrders, appendable_blocks: HashMap, base_block: Option, has_new_base_block: bool, last_merge_request_time_ms: u64, base_txs_set: FxHashSet, + /// Base block hashes for which `get_header` found that the merged bid only differed + /// from the original in the payment tx (and lost out on value because of it). Checked + /// again in `prepare_merged_payload_for_storage` so we can log when the same original + /// block gets reprocessed. + flagged_payment_tx_only_blocks: FxHashSet, trimmed_orders_buf: Vec, inserted_orders_count: usize, inserted_appendable_blocks_count: usize, @@ -102,13 +107,14 @@ impl BlockMerger { config, chain_info, local_cache, - best_merged_block: None, + best_merged_blocks: HashMap::with_capacity(16), best_mergeable_orders: BestMergeableOrders::new(), appendable_blocks: HashMap::with_capacity(00), base_block: None, has_new_base_block: false, last_merge_request_time_ms: 0, base_txs_set: FxHashSet::with_capacity_and_hasher(400, FxBuildHasher), + flagged_payment_tx_only_blocks: FxHashSet::with_capacity_and_hasher(16, FxBuildHasher), trimmed_orders_buf: Vec::with_capacity(400), inserted_orders_count: 0, inserted_appendable_blocks_count: 0, @@ -127,13 +133,14 @@ impl BlockMerger { pub fn on_new_slot(&mut self, bid_slot: u64) { info!(old_slot = %self.curr_bid_slot, new_slot = %bid_slot, inserted_appendable_blocks_count = %self.inserted_appendable_blocks_count, inserted_orders_count = %self.inserted_orders_count, updated_base_block_count = %self.updated_base_block_count, fetch_merge_request_count = %self.fetch_merge_request_count, proceeding_merge_request_count = %self.proceeding_merge_request_count, no_base_block_count = %self.no_base_block_count, no_appendable_block_data_count = %self.no_appendable_block_data_count, found_orders_count = %self.found_orders_count, "resetting block merger slot"); self.curr_bid_slot = bid_slot; - self.best_merged_block = None; + self.best_merged_blocks.clear(); self.best_mergeable_orders.reset(); self.appendable_blocks.clear(); self.base_block = None; self.has_new_base_block = false; self.last_merge_request_time_ms = 0; self.base_txs_set.clear(); + self.flagged_payment_tx_only_blocks.clear(); self.trimmed_orders_buf.clear(); self.found_orders_count = 0; self.inserted_orders_count = 0; @@ -170,13 +177,15 @@ impl BlockMerger { #[timed] pub fn get_header( - &self, + &mut self, original_bid: &PayloadEntry, is_mev_boost: bool, ) -> Option { trace!("fetching merged header"); let start_time = Instant::now(); - let entry = self.best_merged_block.as_ref()?; + let coinbase = original_bid.execution_payload().fee_recipient; + let entry = self.best_merged_blocks.get(&coinbase)?; + if !merged_bid_higher( &entry.bid, original_bid, @@ -184,6 +193,16 @@ impl BlockMerger { self.config.block_merging_config.max_merged_bid_age_ms, ) { trace!("merged bid not higher"); + let original_block_hash = *original_bid.block_hash(); + if log_if_only_payment_tx_changed( + original_block_hash, + *entry.bid.block_hash(), + entry.bid.bid_data_ref().builder_pubkey, + &original_bid.execution_payload().transactions, + &entry.bid.execution_payload().transactions, + ) { + self.flagged_payment_tx_only_blocks.insert(original_block_hash); + } return None; } @@ -221,6 +240,9 @@ impl BlockMerger { } pub fn insert_merge_data(&mut self, merging_data: MergeData) { + // Dehydrated submissions never populate `orders` (see bid_decoder/tile.rs) since + // transaction indices can't be expanded to bytes before hydration runs; only + // `allow_appending`/`merge_orders` (forwarded raw to the merge-builder tile) apply to them. if !merging_data.merging_data.orders.orders.is_empty() { trace!(%merging_data.block_hash,"inserting merge orders from block"); self.inserted_orders_count += 1; @@ -340,6 +362,15 @@ impl BlockMerger { let bid_slot = self.curr_bid_slot; let max_blobs_per_block = self.chain_info.max_blobs_per_block(); + let original_block_hash = original_payload.execution_payload.block_hash; + if self.flagged_payment_tx_only_blocks.remove(&original_block_hash) { + info!( + %original_block_hash, + %builder_pubkey, + "original payload previously flagged for a payment-tx-only merge is being merged again" + ); + } + let base_block_data = match self.appendable_blocks.get(&response.base_block_hash) { Some(data) => data, None => { @@ -412,8 +443,11 @@ impl BlockMerger { let new_bid = PayloadEntry::new_gossip(payload_and_blobs, bid_data); - // Store locally to serve header requests - self.best_merged_block = Some(BestMergedBlock { + // Store locally to serve header requests, keyed by the beneficiary/coinbase + // address of the base block the merge was built from, so that a merge for one + // builder's block can never be served in place of another builder's original bid. + let coinbase = base_block_data.execution_payload.fee_recipient; + self.best_merged_blocks.insert(coinbase, BestMergedBlock { base_block_time_ms: base_block_data.time_ms, bid: new_bid.clone(), }); @@ -763,6 +797,46 @@ fn append_merged_blobs( Ok(()) } +/// Checks whether the merged block kept the original builder's tx ordering completely +/// unchanged except for the payment tx (the last tx in the original block), with any +/// additional orders appended after it. When that's the case, this is logged since it +/// indicates the merge only affected the payment tx rather than tx ordering/content. +/// Returns whether the condition was detected (and thus logged). +fn log_if_only_payment_tx_changed( + base_block_hash: B256, + merged_block_hash: B256, + builder_pubkey: &BlsPublicKeyBytes, + original_txs: &Transactions, + merged_txs: &Transactions, +) -> bool { + let Some(payment_tx_index) = original_txs.len().checked_sub(1) else { + return false; + }; + + if merged_txs.len() <= payment_tx_index { + return false; + } + + let prefix_unchanged = original_txs[..payment_tx_index] == merged_txs[..payment_tx_index]; + let payment_tx_changed = original_txs[payment_tx_index] != merged_txs[payment_tx_index]; + + if !(prefix_unchanged && payment_tx_changed) { + return false; + } + + info!( + %base_block_hash, + %merged_block_hash, + %builder_pubkey, + prefix_tx_count = payment_tx_index, + appended_tx_count = merged_txs.len() - original_txs.len(), + original_payment_tx = %original_txs[payment_tx_index].0, + merged_payment_tx = %merged_txs[payment_tx_index].0, + "merge kept original builder tx ordering unchanged except for the payment tx" + ); + true +} + fn merged_bid_higher( merged_bid: &PayloadEntry, original_bid: &PayloadEntry, diff --git a/crates/relay/src/auctioneer/get_header.rs b/crates/relay/src/auctioneer/get_header.rs index 164e3f5d..6ddbac01 100644 --- a/crates/relay/src/auctioneer/get_header.rs +++ b/crates/relay/src/auctioneer/get_header.rs @@ -43,16 +43,15 @@ impl Context { return Err(ProposerApiError::NoBidPrepared); }; - let Some(original_bid) = self.payloads.get(&best_block_hash) else { + let Some(original_bid) = self.payloads.get(&best_block_hash).cloned() else { error!("failed to get payload from bid sorter best, this should never happen!"); return Err(ProposerApiError::NoBidPrepared); }; - if let Some(merged_bid) = self.block_merger.get_header(original_bid, is_mev_boost) { + if let Some(merged_bid) = self.block_merger.get_header(&original_bid, is_mev_boost) { return Ok(merged_bid); }; - let original_bid = original_bid.clone(); if self.cache.adjustments_enabled.load(Ordering::Relaxed) { let start = Instant::now(); if let Some((adjusted_bid, sim_request, is_adjustable_slot, strategy)) = diff --git a/crates/relay/src/auctioneer/mod.rs b/crates/relay/src/auctioneer/mod.rs index c1afa644..22e09d56 100644 --- a/crates/relay/src/auctioneer/mod.rs +++ b/crates/relay/src/auctioneer/mod.rs @@ -223,10 +223,18 @@ impl State { let start = Instant::now(); let start_state = self.as_str(); let event_tag = event.as_str(); + let prev_bid_slot = self.bid_slot(); + + *tel.event_counts.entry(event_tag).or_insert(0) += 1; self._step(event, ctx, producers); let end_state = self.as_str(); + + let new_bid_slot = self.bid_slot(); + if new_bid_slot > prev_bid_slot { + tel.report_slot_stats(prev_bid_slot); + } let step_dur = start.elapsed(); tel.loop_worked += step_dur; @@ -696,6 +704,14 @@ impl State { State::Broadcasting { .. } => "Broadcasting", } } + + fn bid_slot(&self) -> u64 { + match self { + State::Slot { bid_slot, .. } => bid_slot.as_u64(), + State::Sorting(slot_data) => slot_data.bid_slot.as_u64(), + State::Broadcasting { slot_data, .. } => slot_data.bid_slot.as_u64(), + } + } } pub struct Telemetry { @@ -705,6 +721,9 @@ pub struct Telemetry { next_record: Instant, loop_start: Instant, loop_worked: Duration, + /// Count of every event handled this slot, keyed by `Event::as_str()`. + /// Sums to the total number of events the state machine processed. + event_counts: FxHashMap<&'static str, u32>, } impl Telemetry { @@ -718,7 +737,19 @@ impl Telemetry { next_record: Instant::now() + Self::REPORT_FREQ, loop_start: Instant::now(), loop_worked: Duration::ZERO, + event_counts: FxHashMap::default(), + } + } + + /// Logged on every slot transition; `bid_slot` is the slot that just + /// ended. + fn report_slot_stats(&mut self, bid_slot: u64) { + if bid_slot == 0 { + return; } + let counts = std::mem::take(&mut self.event_counts); + let total_events: u32 = counts.values().sum(); + info!(bid_slot, total_events, ?counts, "auctioneer slot stats"); } fn telemetry(&mut self, rx: &crossbeam_channel::Receiver) { diff --git a/crates/relay/src/bid_decoder/tile.rs b/crates/relay/src/bid_decoder/tile.rs index 3a35ba05..b64aadf8 100644 --- a/crates/relay/src/bid_decoder/tile.rs +++ b/crates/relay/src/bid_decoder/tile.rs @@ -1,4 +1,4 @@ -use std::{cell::RefCell, sync::Arc}; +use std::{cell::RefCell, collections::HashMap, sync::Arc}; use alloy_primitives::B256; use bytes::Bytes; @@ -18,10 +18,11 @@ use helix_common::{ record_submission_step, }; use helix_types::{ - BidAdjustmentData, BlockMergingData, BlsPublicKeyBytes, MergeableOrdersWithPref, - SignedBidSubmission, Submission, SubmissionVersion, + BidAdjustmentData, BlockMergingData, BlsPublicKeyBytes, MergeableOrders, + MergeableOrdersWithPref, SignedBidSubmission, Submission, SubmissionVersion, }; -use tracing::trace; +use rustc_hash::FxHashMap; +use tracing::{info, trace}; use zstd::zstd_safe::WriteBuf; use crate::{ @@ -32,12 +33,21 @@ use crate::{ send_submission_result, }, bid_decoder::SubmissionDataWithSpan, + housekeeper::SlotUpdate, spine::{ HelixSpineProducers, - messages::{DecodedSubmission, NewBidSubmission}, + messages::{DecodedSubmission, NewBidSubmission, SlotMsg}, }, }; +/// Per-slot decode outcomes. `decoded_ok + decode_errors.values().sum() ==` +/// total submissions the decoder saw this slot. +#[derive(Default)] +struct DecodeStats { + decoded_ok: u32, + decode_errors: FxHashMap<&'static str, u32>, +} + pub struct DecoderTile { chain_info: ChainInfo, cache: LocalCache, @@ -47,10 +57,15 @@ pub struct DecoderTile { http_submissions: Arc>, buffer: RefCell>, core: usize, + slot_events: Arc>, + bid_slot: u64, + stats: RefCell, } impl Tile for DecoderTile { fn loop_body(&mut self, adapter: &mut flux::spine::SpineAdapter) { + adapter.consume(|msg: SlotMsg, _| self.on_slot_msg(msg)); + adapter.consume_with_dcache_collaborative_internal_message( |new_bid: &InternalMessage, dcache_payload| { // dcache bypass: the dcache slot can be mutated between publish and @@ -79,6 +94,7 @@ impl Tile for DecoderTile { }, |res, producers| match res { DCacheRead::Ok((new_bid, result)) => { + self.record_decode_result(&result); let sent_at = new_bid.tracking_timestamp().publish_t(); Self::handle_result( &self.decoded, @@ -96,6 +112,7 @@ impl Tile for DecoderTile { "failed to find the payload for bid submission with id = {}", new_bid.header.id ); + self.record_decode_result(&Err(BuilderApiError::InternalError)); return send_submission_result( producers, &self.future_results, @@ -117,6 +134,7 @@ impl Tile for DecoderTile { sent_at, new_bid.expected_pubkey(), ); + self.record_decode_result(&result); Self::handle_result( &self.decoded, &self.future_results, @@ -131,6 +149,7 @@ impl Tile for DecoderTile { "dcache read failed for bid submission with id {}", new_bid.header.id ); + self.record_decode_result(&Err(BuilderApiError::InternalError)); send_submission_result( producers, &self.future_results, @@ -159,6 +178,7 @@ impl Tile for DecoderTile { } impl DecoderTile { + #[allow(clippy::too_many_arguments)] pub fn new( cache: LocalCache, chain_info: ChainInfo, @@ -166,6 +186,7 @@ impl DecoderTile { future_results: Arc>, decoded: Arc>, http_submissions: Arc>, + slot_events: Arc>, core: usize, ) -> Self { Self { @@ -177,6 +198,47 @@ impl DecoderTile { http_submissions, buffer: RefCell::new(Vec::with_capacity(MAX_PAYLOAD_LENGTH)), core, + slot_events, + bid_slot: 0, + stats: RefCell::new(DecodeStats::default()), + } + } + + fn on_slot_msg(&mut self, msg: SlotMsg) { + let Some(ev) = self.slot_events.get(msg.ix) else { return }; + let bid_slot = ev.bid_slot.as_u64(); + if bid_slot <= self.bid_slot { + return; + } + self.report_slot_stats(); + self.bid_slot = bid_slot; + } + + fn report_slot_stats(&self) { + if self.bid_slot == 0 { + return; + } + let stats = std::mem::take(&mut *self.stats.borrow_mut()); + let decode_errors: u32 = stats.decode_errors.values().sum(); + let errors_by_category = stats.decode_errors; + info!( + bid_slot = self.bid_slot, + submissions_seen = stats.decoded_ok + decode_errors, + decoded_ok = stats.decoded_ok, + decode_errors, + ?errors_by_category, + "bid decoder slot stats" + ); + } + + fn record_decode_result( + &self, + result: &Result<(SubmissionData, tracing::Span), BuilderApiError>, + ) { + let mut stats = self.stats.borrow_mut(); + match result { + Ok(_) => stats.decoded_ok += 1, + Err(e) => *stats.decode_errors.entry(error_category(e)).or_insert(0) += 1, } } @@ -232,8 +294,8 @@ impl DecoderTile { trace!("sending to auctioneer"); let merging_data = if config.block_merging_config.is_enabled { - merging_data.and_then(|data| { - if let Submission::Full(ref signed_bid_submission) = submission { + merging_data.and_then(|data| match &submission { + Submission::Full(signed_bid_submission) => { // TODO: split up mergeable order and submission processing to // avoid delaying the bid update match get_mergeable_orders(signed_bid_submission, &data) { @@ -247,9 +309,18 @@ impl DecoderTile { None } } - } else { - None } + // transactions aren't resolved yet, so orders referencing tx indices can't be + // expanded into `orders` here for either append-only or mergeable dehydrated + // submissions. `merge_orders` is carried through raw so the external + // merge-builder tile (which hydrates independently) can still resolve them; the + // local `BestMergeableOrders` pool never receives expanded orders for dehydrated + // submissions. + Submission::Dehydrated(_) => Some(MergeableOrdersWithPref { + allow_appending: data.allow_appending, + orders: MergeableOrders::new(data.builder_address, Vec::new(), HashMap::new()), + merge_orders: data.merge_orders, + }), }) } else { None @@ -384,6 +455,24 @@ impl DecoderTile { } } +/// Coarse, low-cardinality bucket for a decode failure; several +/// `BuilderApiError` variants carry per-request data unsuited to a metric key. +fn error_category(err: &BuilderApiError) -> &'static str { + match err { + BuilderApiError::JsonDecodeError(_) => "json_decode", + BuilderApiError::SszDecode(_) => "ssz_decode", + BuilderApiError::IOError(_) => "io", + BuilderApiError::PayloadDecode(_) => "payload_decode", + BuilderApiError::BidValidation(_) => "bid_validation", + BuilderApiError::SigError(_) => "sig_error", + BuilderApiError::HydrationError(_) => "hydration", + BuilderApiError::UntrustedBuilderOnDehydratedPayload => "untrusted_builder", + BuilderApiError::InvalidBuilderPubkey(..) => "invalid_pubkey", + BuilderApiError::InternalError => "internal_error", + _ => "other", + } +} + #[timed] fn verify_and_validate( submission: &mut SignedBidSubmission, diff --git a/crates/relay/src/block_merging/tile.rs b/crates/relay/src/block_merging/tile.rs index ad02662d..8fcf20fb 100644 --- a/crates/relay/src/block_merging/tile.rs +++ b/crates/relay/src/block_merging/tile.rs @@ -1,6 +1,6 @@ use std::sync::Arc; -use alloy_primitives::{B256, U256}; +use alloy_primitives::{B256, Bytes, U256, keccak256}; use flux::{ spine::SpineProducers, tile::Tile, @@ -41,12 +41,10 @@ const PING_INTERVAL_S: u64 = 5; struct Endpoint { addr: std::net::SocketAddr, api_key: [u8; 16], - token: Option, } #[derive(Default)] struct Conn { - endpoint_ix: usize, /// Handshake completed (ack received with ok status). active: bool, max_orders_per_slot: u32, @@ -55,6 +53,12 @@ struct Conn { /// Appendable block hashes forwarded on this connection this slot. forwarded: FxHashSet, activated: Option, + /// Tx hashes already sent whole on this connection this slot; a repeat + /// tx is forwarded as a hash reference instead. Cleared with the rest of + /// the per-slot state so it never outlives the builder's own per-slot + /// resolution cache, and on reconnect since a fresh connection can't + /// resolve references to bytes it was never sent. + sent_txs: FxHashSet, } impl Conn { @@ -67,6 +71,7 @@ impl Conn { self.orders_sent = 0; self.forwarded.clear(); self.activated = None; + self.sent_txs.clear(); } } @@ -79,7 +84,7 @@ struct SlotState { fee_recipient: Option, /// parent_hash -> parent_beacon_block_root. attrs: FxHashMap, - /// Appendable block hashes forwarded this slot (any connection). + /// Appendable block hashes forwarded this slot. appendable: FxHashSet, /// Decoded ixs forwarded this slot, replayed on re-handshake. mergeable_ixs: Vec, @@ -98,7 +103,7 @@ struct SlotStats { skipped_no_slot_start: usize, skipped_wrong_slot: usize, skipped_no_merging_data: usize, - /// Per-connection sends skipped over builder limits. + /// Sends skipped over the builder's advertised limits. skipped_over_limits: usize, /// Merge orders dropped for an out of range tx index. orders_dropped: usize, @@ -110,12 +115,16 @@ struct SlotStats { merged_regressed: usize, /// TopBidUpdate messages received for the current bid slot. top_bid_updates: usize, - /// Per-connection ActivateBaseBlockV1 frames sent. + /// ActivateBaseBlockV1 frames sent. activations_sent: usize, /// Gaps between consecutive top bid updates, from bid sorter send /// timestamps. top_bid_gaps_ns: Vec, last_top_bid_ns: u64, + /// Txs sent as full bytes: new to this connection this slot. + tx_bytes_sent: usize, + /// Txs sent as a hash reference: already sent whole earlier this slot. + tx_refs_sent: usize, } pub struct BlockMergingTile { @@ -123,8 +132,9 @@ pub struct BlockMergingTile { relay_id: Vec, relay_config_msg: RelayConfigV1, - endpoints: Vec, - conns: FxHashMap, + endpoint: Endpoint, + token: Option, + conn: Conn, slot: SlotState, stats: SlotStats, chain_info: ChainInfo, @@ -159,7 +169,7 @@ impl Tile for BlockMergingTile { } if self.redial.fired() { - self.dial_endpoints(); + self.dial_endpoint(); } if self.ping.fired() { self.send_pings(); @@ -171,7 +181,7 @@ impl Tile for BlockMergingTile { } fn try_init(&mut self, _adapter: &mut flux::spine::SpineAdapter) -> bool { - self.dial_endpoints(); + self.dial_endpoint(); info!("starting"); true } @@ -208,30 +218,31 @@ impl BlockMergingTile { }; relay_config_msg.validate().expect("invalid block merging relay config"); - let endpoints = config - .builders - .iter() - .map(|b| Endpoint { - addr: b.addr, - api_key: Uuid::parse_str(&b.api_key) - .expect("invalid block merging api key") - .into_bytes(), - token: None, - }) - .collect(); + let endpoint = Endpoint { + addr: config.builder.addr, + api_key: Uuid::parse_str(&config.builder.api_key) + .expect("invalid block merging api key") + .into_bytes(), + }; // TODO: enable telemetry once the per-connection shm queue leak is fixed // Disabled: per-connection shm queue leak, see tcp_bid_recv/mod.rs. let connector = TcpConnector::default() .with_telemetry(TcpTelemetry::Disabled) - .with_socket_buf_size(64 * 1024 * 1024); + .with_socket_buf_size(64 * 1024 * 1024) + // Otherwise a stale message queued for the dead socket (e.g. an + // activation or ping) gets replayed on the new one ahead of the + // fresh MergerRegistrationV1, and the builder rejects it with + // "expected registration" — killing the connection again. + .with_drop_outbound_backlog_on_disconnect(true); Self { connector, relay_id: relay_id.into_bytes(), relay_config_msg, - endpoints, - conns: FxHashMap::default(), + endpoint, + token: None, + conn: Conn::default(), slot: SlotState::default(), stats: SlotStats::default(), chain_info, @@ -251,30 +262,27 @@ impl BlockMergingTile { } } - /// Dials endpoints that have no token yet. A failed initial `connect` is - /// not retried by the connector (unlike established conns, which - /// auto-reconnect), so this runs on a repeater. - fn dial_endpoints(&mut self) { - for ix in 0..self.endpoints.len() { - if self.endpoints[ix].token.is_some() { - continue; - } - let addr = self.endpoints[ix].addr; - let Some(token) = self.connector.connect(addr) else { - warn!(%addr, "failed to dial merging builder"); - continue; - }; - info!(%addr, ?token, "dialing merging builder"); - self.endpoints[ix].token = Some(token); - self.conns.insert(token, Conn { endpoint_ix: ix, ..Default::default() }); - self.send_registration(token); + /// Dials the builder if not already connected. A failed initial `connect` + /// is not retried by the connector (unlike an established conn, which + /// auto-reconnects), so this runs on a repeater. + fn dial_endpoint(&mut self) { + if self.token.is_some() { + return; } + let addr = self.endpoint.addr; + let Some(token) = self.connector.connect(addr) else { + warn!(%addr, "failed to dial merging builder"); + return; + }; + info!(%addr, ?token, "dialing merging builder"); + self.token = Some(token); + self.conn = Conn::default(); + self.send_registration(token); } fn send_registration(&mut self, token: Token) { - let Some(conn) = self.conns.get(&token) else { return }; let msg = MergerRegistrationV1 { - api_key: self.endpoints[conn.endpoint_ix].api_key, + api_key: self.endpoint.api_key, relay_id: self.relay_id.clone(), min_version: MERGING_PROTOCOL_VERSION, max_version: MERGING_PROTOCOL_VERSION, @@ -307,7 +315,8 @@ impl BlockMergingTile { // poll, all reactions are buffered. let Self { connector, - conns, + token: my_token, + conn, slot, stats, to_disconnect, @@ -323,18 +332,21 @@ impl BlockMergingTile { PollEvent::Accept { .. } => error!("unexpected inbound connection on merging tile"), PollEvent::Reconnect { token } => { info!(?token, "reconnected to merging builder"); - if let Some(conn) = conns.get_mut(&token) { + if *my_token == Some(token) { conn.reset(); to_register.push(token); } } PollEvent::Disconnect { token } => { warn!(?token, "merging builder disconnected"); - if let Some(conn) = conns.get_mut(&token) { + if *my_token == Some(token) { conn.reset(); } } PollEvent::Message { token, payload, send_ts: _ } => { + if *my_token != Some(token) { + return; + } let header = match MergingFrameHeader::decode(payload) { Ok(header) => header, // extension ids must be ignored @@ -368,12 +380,10 @@ impl BlockMergingTile { return; } info!(?token, version = ack.version, "merging builder handshake ok"); - if let Some(conn) = conns.get_mut(&token) { - conn.active = true; - conn.max_orders_per_slot = ack.max_orders_per_slot; - conn.max_frame_bytes = ack.max_frame_bytes; - handshaken.push(token); - } + conn.active = true; + conn.max_orders_per_slot = ack.max_orders_per_slot; + conn.max_frame_bytes = ack.max_frame_bytes; + handshaken.push(token); } MergingMsgId::PongV1 => { if let Ok(pong) = PongV1::from_ssz_bytes(body) { @@ -402,9 +412,9 @@ impl BlockMergingTile { ); return; } - // Builders only guarantee per-connection monotonicity; - // filter across builders so the stored merged bid - // never regresses. + // Builders only guarantee monotonicity within a + // connection; filter so the stored merged bid never + // regresses. let best = slot.best_merged.entry(merged.base_block_hash).or_insert(U256::ZERO); if merged.proposer_value <= *best { @@ -450,8 +460,8 @@ impl BlockMergingTile { for token in std::mem::take(&mut self.to_disconnect) { // outbound: schedules an auto-reconnect, which re-handshakes self.connector.disconnect(token); - if let Some(conn) = self.conns.get_mut(&token) { - conn.reset(); + if self.token == Some(token) { + self.conn.reset(); } } for token in std::mem::take(&mut self.to_register) { @@ -476,9 +486,7 @@ impl BlockMergingTile { if bid_slot > self.slot.bid_slot { self.report_slot_stats(); self.slot = SlotState { bid_slot, ..Default::default() }; - for conn in self.conns.values_mut() { - conn.reset_slot(); - } + self.conn.reset_slot(); self.hydration_cache.clear(); // sole producer; consumed indices are stale after the transition self.merged_blocks.clear(); @@ -514,12 +522,12 @@ impl BlockMergingTile { }; debug!(slot = self.slot.bid_slot, "merging slot start"); - for (token, conn) in self.conns.iter() { - if conn.active { - self.connector.write_or_enqueue_with(SendBehavior::Single(*token), |buf| { - append_frame(buf, MergingMsgId::SlotStartV1, &msg); - }); - } + if self.conn.active && + let Some(token) = self.token + { + self.connector.write_or_enqueue_with(SendBehavior::Single(token), |buf| { + append_frame(buf, MergingMsgId::SlotStartV1, &msg); + }); } self.slot.slot_start = Some(msg); } @@ -549,6 +557,8 @@ impl BlockMergingTile { appendable_blocks = self.slot.appendable.len(), hydration_txs = self.hydration_cache.tx_count(), hydration_builders = self.hydration_cache.builder_count(), + tx_bytes_sent = stats.tx_bytes_sent, + tx_refs_sent = stats.tx_refs_sent, "block merging slot stats" ); } @@ -562,8 +572,8 @@ impl BlockMergingTile { } } - /// Forwards the decoded submission at `ix` as a `MergeableBlockV1` to all - /// active connections, or to `only` on re-handshake replay. + /// Forwards the decoded submission at `ix` as a `MergeableBlockV1`, or + /// replays it to `only` on re-handshake. fn forward_decoded(&mut self, ix: usize, only: Option) { let is_replay = only.is_some(); if is_replay { @@ -625,10 +635,9 @@ impl BlockMergingTile { None => self.stats.orders_dropped += 1, } } - let order_count = merge_orders.len() as u32; let block_hash = signed.message.block_hash; - let msg = MergeableBlockV1 { + let mut msg = MergeableBlockV1 { slot: signed.message.slot, builder_pubkey: signed.message.builder_pubkey, block_value: signed.message.value, @@ -645,6 +654,21 @@ impl BlockMergingTile { execution_payload: payload_to_v3(&signed.execution_payload), }; + // Dehydrate relative to this connection: a tx already sent whole + // earlier this slot (from this block or any other, any builder) goes + // out as a hash reference instead. See MergeableBlockV1's doc comment + // for the wire convention. + for tx in &mut msg.execution_payload.payload_inner.payload_inner.transactions { + let hash = keccak256(tx.as_ref()); + if self.conn.sent_txs.insert(hash) { + self.stats.tx_bytes_sent += 1; + } else { + *tx = Bytes::copy_from_slice(hash.as_slice()); + self.stats.tx_refs_sent += 1; + } + } + let order_count = msg.merge_orders.len() as u32; + self.encode_buf.clear(); append_frame(&mut self.encode_buf, MergingMsgId::MergeableBlockV1, &msg); @@ -655,26 +679,25 @@ impl BlockMergingTile { self.slot.mergeable_ixs.push(ix); } + let Some(token) = self.token else { return }; + if !self.conn.active || only.is_some_and(|t| t != token) { + return; + } let frame = &self.encode_buf; - for (token, conn) in self.conns.iter_mut() { - if !conn.active || only.is_some_and(|t| t != *token) { - continue; - } - if conn.orders_sent.saturating_add(order_count) > conn.max_orders_per_slot || - frame.len() > conn.max_frame_bytes as usize - { - self.stats.skipped_over_limits += 1; - debug!(?token, %block_hash, "skipping mergeable block over builder limits"); - continue; - } - conn.orders_sent += order_count; - if msg.allow_appending { - conn.forwarded.insert(block_hash); - } - self.connector.write_or_enqueue_with(SendBehavior::Single(*token), |buf| { - buf.extend_from_slice(frame); - }); + if self.conn.orders_sent.saturating_add(order_count) > self.conn.max_orders_per_slot || + frame.len() > self.conn.max_frame_bytes as usize + { + self.stats.skipped_over_limits += 1; + debug!(?token, %block_hash, "skipping mergeable block over builder limits"); + return; + } + self.conn.orders_sent += order_count; + if msg.allow_appending { + self.conn.forwarded.insert(block_hash); } + self.connector.write_or_enqueue_with(SendBehavior::Single(token), |buf| { + buf.extend_from_slice(frame); + }); } fn on_top_bid(&mut self, top_bid: TopBidUpdate) { @@ -692,20 +715,19 @@ impl BlockMergingTile { if !self.slot.appendable.contains(&top_bid.block_hash) { return; } - let msg = ActivateBaseBlockV1 { slot: top_bid.slot, block_hash: top_bid.block_hash }; - for (token, conn) in self.conns.iter_mut() { - if !conn.active || - !conn.forwarded.contains(&top_bid.block_hash) || - conn.activated == Some(top_bid.block_hash) - { - continue; - } - conn.activated = Some(top_bid.block_hash); - self.stats.activations_sent += 1; - self.connector.write_or_enqueue_with(SendBehavior::Single(*token), |buf| { - append_frame(buf, MergingMsgId::ActivateBaseBlockV1, &msg); - }); + let Some(token) = self.token else { return }; + if !self.conn.active || + !self.conn.forwarded.contains(&top_bid.block_hash) || + self.conn.activated == Some(top_bid.block_hash) + { + return; } + self.conn.activated = Some(top_bid.block_hash); + self.stats.activations_sent += 1; + let msg = ActivateBaseBlockV1 { slot: top_bid.slot, block_hash: top_bid.block_hash }; + self.connector.write_or_enqueue_with(SendBehavior::Single(token), |buf| { + append_frame(buf, MergingMsgId::ActivateBaseBlockV1, &msg); + }); } /// Median of unsorted samples; 0 if empty. @@ -724,13 +746,13 @@ impl BlockMergingTile { fn send_pings(&mut self) { self.ping_nonce += 1; - let msg = PingV1 { nonce: self.ping_nonce }; - for (token, conn) in self.conns.iter() { - if conn.active { - self.connector.write_or_enqueue_with(SendBehavior::Single(*token), |buf| { - append_frame(buf, MergingMsgId::PingV1, &msg); - }); - } + let Some(token) = self.token else { return }; + if !self.conn.active { + return; } + let msg = PingV1 { nonce: self.ping_nonce }; + self.connector.write_or_enqueue_with(SendBehavior::Single(token), |buf| { + append_frame(buf, MergingMsgId::PingV1, &msg); + }); } } diff --git a/crates/relay/src/data_gatherer/tile.rs b/crates/relay/src/data_gatherer/tile.rs index 7bcb5b5f..7803e971 100644 --- a/crates/relay/src/data_gatherer/tile.rs +++ b/crates/relay/src/data_gatherer/tile.rs @@ -7,6 +7,7 @@ use helix_common::{ S3Config, api::builder_api::TopBidUpdate, config::ClickhouseConfig, decoder::Encoding, }; use helix_types::{BlsPublicKeyBytes, Compression, MergeType}; +use tracing::info; use crate::{ HelixSpine, SubmissionDataWithSpan, @@ -17,12 +18,27 @@ use crate::{ spine::messages::{BidEvent, BidUpdate, DecodedSubmission, NewBidSubmission}, }; +/// Per-slot counters, logged and reset on slot transition. +/// `extract_ok + extract_failed + compressed_skipped ==` raw +/// `NewBidSubmission` events seen this slot. +#[derive(Default)] +struct SlotStats { + s3_uploads: u32, + extract_ok: u32, + extract_failed: u32, + compressed_skipped: u32, + decoded_seen: u32, + bid_live_events: u32, + top_bid_updates_seen: u32, +} + pub struct DataGatherer { decoded: Arc>, ch: Option, s3: Option, current_slot: u64, rt: tokio::runtime::Runtime, + stats: SlotStats, } impl DataGatherer { @@ -42,10 +58,12 @@ impl DataGatherer { s3: s3_config.map(S3Data::new), current_slot: 0, rt, + stats: SlotStats::default(), } } pub fn on_new_slot(&mut self, new_slot: u64) { + self.report_slot_stats(); self.current_slot = new_slot; if let Some(ch) = self.ch.as_mut() && let Some(future) = ch.publish_snapshot(new_slot) @@ -54,6 +72,24 @@ impl DataGatherer { } } + fn report_slot_stats(&mut self) { + if self.current_slot == 0 { + return; + } + let stats = std::mem::take(&mut self.stats); + info!( + bid_slot = self.current_slot, + s3_uploads = stats.s3_uploads, + extract_ok = stats.extract_ok, + extract_failed = stats.extract_failed, + compressed_skipped = stats.compressed_skipped, + decoded_seen = stats.decoded_seen, + bid_live_events = stats.bid_live_events, + top_bid_updates_seen = stats.top_bid_updates_seen, + "data gatherer slot stats" + ); + } + fn extract_block_hash_and_pubkey( encoding: Encoding, buf: &[u8], @@ -120,6 +156,7 @@ impl Tile for DataGatherer { |bid: &InternalMessage, payload| { let payload = &payload[bid.payload_offset..]; if let Some(s3) = self.s3.as_ref() { + self.stats.s3_uploads += 1; self.rt.spawn(s3.upload_task(bid.header, payload)); } @@ -128,6 +165,7 @@ impl Tile for DataGatherer { if let Some((slot, block_hash, builder_pubkey)) = Self::extract_block_hash_and_pubkey(bid.header.encoding, payload, is_mergeable) { + self.stats.extract_ok += 1; max_slot = max_slot.max(slot); if let Some(ch) = self.ch.as_mut() { ch.insert(block_hash, BlockInfo { @@ -140,11 +178,14 @@ impl Tile for DataGatherer { }); } } else { + self.stats.extract_failed += 1; tracing::error!( "failed to extract builder_pubkey & block hash from submission with id {}", bid.header.id ); } + } else { + self.stats.compressed_skipped += 1; } }, |_, _| {}, @@ -152,6 +193,7 @@ impl Tile for DataGatherer { adapter.consume_internal_message(|msg: &mut InternalMessage, _| { if let Some(bid) = self.decoded.get(msg.ix) { + self.stats.decoded_seen += 1; max_slot = max_slot.max(bid.submission_data.bid_slot()); if let Some(ch) = self.ch.as_mut() { let info = @@ -177,12 +219,14 @@ impl Tile for DataGatherer { // todo @nina - will we ever need other events? #[allow(irrefutable_let_patterns)] if let BidEvent::Live = msg.event { + self.stats.bid_live_events += 1; info.live_ns = Some(msg.ingestion_time().real().0 as i64); } } }); adapter.consume_internal_message(|msg: &mut InternalMessage, _| { + self.stats.top_bid_updates_seen += 1; max_slot = max_slot.max(msg.slot); if let Some(ch) = self.ch.as_mut() && let Some(info) = ch.get_mut(&msg.block_hash) diff --git a/crates/relay/src/housekeeper/tile.rs b/crates/relay/src/housekeeper/tile.rs index 6aa1f672..16e8b9ab 100644 --- a/crates/relay/src/housekeeper/tile.rs +++ b/crates/relay/src/housekeeper/tile.rs @@ -54,6 +54,33 @@ pub struct SlotUpdate { pub il: Option, } +/// Counters accumulated between two `send_slot_event` calls, logged and reset +/// alongside the slot event. Each fetch/poll's outcomes are mutually +/// exclusive, so e.g. `duties_fetch_ok + duties_fetch_empty + +/// duties_fetch_err` is the total number of duties fetches this window. +#[derive(Default)] +struct HousekeeperStats { + head_sse_events: u32, + head_sse_parse_errors: u32, + payload_attr_sse_events: u32, + payload_attr_parse_errors: u32, + duties_fetch_ok: u32, + duties_fetch_empty: u32, + duties_fetch_err: u32, + il_fetch_ok: u32, + il_fetch_timeout: u32, + il_fetch_decode_err: u32, + il_consensus_ok: u32, + il_consensus_err: u32, + primev_builders_fetched: u32, + primev_validators_fetched: u32, + known_validators_refresh_ok: u32, + known_validators_refresh_err: u32, + sync_status_ok: u32, + sync_status_err: u32, + sync_status_timeout: u32, +} + pub struct HousekeeperTile { // Internal state chain_head: ChainHead, @@ -93,6 +120,8 @@ pub struct HousekeeperTile { sync_check_pending: Vec<(usize, Instant, PendingResponse)>, sync_check_next: Instant, sync_best: Option<(usize, u64)>, + + stats: HousekeeperStats, } impl HousekeeperTile { @@ -156,6 +185,7 @@ impl HousekeeperTile { sync_check_pending: Vec::new(), sync_check_next: Instant::now(), sync_best: None, + stats: HousekeeperStats::default(), }; (tile, curr_slot_info) } @@ -231,6 +261,7 @@ impl Tile for HousekeeperTile { let mut new_slot_from_sse = false; for stream in &mut self.head_sse { if let Poll::Ready(ev) = stream.poll() { + self.stats.head_sse_events += 1; match serde_json::from_str::(&ev.data) { Ok(data) => { if self.chain_head.update(data) { @@ -238,7 +269,10 @@ impl Tile for HousekeeperTile { break; } } - Err(e) => error!(err = %e, "head SSE parse error"), + Err(e) => { + self.stats.head_sse_parse_errors += 1; + error!(err = %e, "head SSE parse error"); + } } } } @@ -249,13 +283,17 @@ impl Tile for HousekeeperTile { } for stream in &mut self.payload_attr_sse { if let Poll::Ready(ev) = stream.poll() { + self.stats.payload_attr_sse_events += 1; match serde_json::from_str::(&ev.data) { Ok(data) => process_payload_attributes( &mut self.chain_head, data, &mut self.known_payload_attributes, ), - Err(e) => error!(err = %e, "payload_attributes SSE parse error"), + Err(e) => { + self.stats.payload_attr_parse_errors += 1; + error!(err = %e, "payload_attributes SSE parse error"); + } } } } @@ -270,15 +308,20 @@ impl Tile for HousekeeperTile { self.duties_fetch = None; match result { Ok(proposer_duties) if proposer_duties.is_empty() => { + self.stats.duties_fetch_empty += 1; warn!("no proposer duties found"); } Ok(proposer_duties) => { + self.stats.duties_fetch_ok += 1; process_duties(&proposer_duties, &self.local_cache, &self.db); self.duties = proposer_duties; self.chain_head.mark_duties_done(); self.maybe_fetch_primev(); } - Err(e) => error!(%e, "failed to fetch proposer duties"), + Err(e) => { + self.stats.duties_fetch_err += 1; + error!(%e, "failed to fetch proposer duties"); + } } } @@ -287,6 +330,7 @@ impl Tile for HousekeeperTile { let Poll::Ready(builders) = state.poll() { self.primev_builders_fetch = None; + self.stats.primev_builders_fetched += 1; if !builders.is_empty() { info!(builders = builders.len(), "fetched primev builders"); let configs = build_primev_builder_configs(builders, &self.local_cache); @@ -300,6 +344,7 @@ impl Tile for HousekeeperTile { let Poll::Ready(validators) = state.poll() { self.primev_validators_fetch = None; + self.stats.primev_validators_fetched += 1; if !validators.is_empty() { info!(validators = validators.len(), "fetched primev validators"); self.local_cache.update_primev_proposers(&validators); @@ -318,6 +363,7 @@ impl Tile for HousekeeperTile { let head = self.chain_head.head(); match InclusionListWithMetadata::try_from(il.clone()) { Ok(il_meta) => { + self.stats.il_fetch_ok += 1; info!(txs = il_meta.txs.len(), "fetched inclusion list"); if let Some(parent_hash) = self.chain_head.block() && let Some(duty) = self.duties.iter().find(|d| d.slot == head) @@ -335,13 +381,17 @@ impl Tile for HousekeeperTile { } self.pending_il = Some(il_meta); } - Err(err) => warn!(%err, %head, "could not decode inclusion list RLP"), + Err(err) => { + self.stats.il_fetch_decode_err += 1; + warn!(%err, %head, "could not decode inclusion list RLP"); + } } self.chain_head.mark_il_done(); self.il_consensus_rx = self.relay_network_api.try_share_inclusion_list(head.as_u64(), il); } None => { + self.stats.il_fetch_timeout += 1; warn!("inclusion list fetch timed out"); self.chain_head.mark_il_done(); } @@ -356,10 +406,14 @@ impl Tile for HousekeeperTile { self.known_validators_fetch = None; match result { Ok(resp) => { + self.stats.known_validators_refresh_ok += 1; debug!(validators = resp.data.len(), duration = ?start.elapsed(), "refreshed known validators"); self.db.set_known_validators(resp.data); } - Err(err) => error!(%err, "failed to refresh known validators"), + Err(err) => { + self.stats.known_validators_refresh_err += 1; + error!(%err, "failed to refresh known validators"); + } } } } @@ -386,16 +440,21 @@ impl Tile for HousekeeperTile { if started.elapsed() < SYNC_STATUS_REQUEST_TIMEOUT { self.sync_check_pending.push((idx, started, req)); } else { + self.stats.sync_status_timeout += 1; warn!(idx, "sync status request timed out"); } } Poll::Ready(Ok(resp)) => { + self.stats.sync_status_ok += 1; let slot = resp.data.head_slot.as_u64(); if self.sync_best.as_ref().is_none_or(|(_, s)| slot > *s) { self.sync_best = Some((idx, slot)); } } - Poll::Ready(Err(e)) => warn!(%e, idx, "sync status fetch failed"), + Poll::Ready(Err(e)) => { + self.stats.sync_status_err += 1; + warn!(%e, idx, "sync status fetch failed"); + } } } if self.sync_check_pending.is_empty() && @@ -422,8 +481,14 @@ impl Tile for HousekeeperTile { match rx.try_recv() { Ok(Some(il)) => { match InclusionListWithMetadata::try_from(il) { - Ok(il_meta) => self.pending_il = Some(il_meta), - Err(e) => warn!(%e, "could not decode consensus IL"), + Ok(il_meta) => { + self.stats.il_consensus_ok += 1; + self.pending_il = Some(il_meta); + } + Err(e) => { + self.stats.il_consensus_err += 1; + warn!(%e, "could not decode consensus IL"); + } } self.il_consensus_rx = None; } @@ -432,6 +497,7 @@ impl Tile for HousekeeperTile { } Err(tokio::sync::oneshot::error::TryRecvError::Empty) => {} Err(tokio::sync::oneshot::error::TryRecvError::Closed) => { + self.stats.il_consensus_err += 1; warn!("IL consensus channel closed before result"); self.il_consensus_rx = None; } @@ -447,6 +513,7 @@ impl Tile for HousekeeperTile { &self.slot_events, &self.known_payload_attributes, self.pending_il.take(), + std::mem::take(&mut self.stats), ); } } @@ -456,6 +523,7 @@ impl Tile for HousekeeperTile { } } +#[allow(clippy::too_many_arguments)] fn send_slot_event( adapter: &mut flux::spine::SpineAdapter, chain_head: &mut ChainHead, @@ -464,6 +532,7 @@ fn send_slot_event( slot_events: &SharedVector, known_payload_attributes: &HashMap<(B256, Slot), PayloadAttributesUpdate>, il: Option, + stats: HousekeeperStats, ) { let slot = chain_head.head(); let bid_slot = slot + 1; @@ -493,7 +562,26 @@ fn send_slot_event( duties_available = !new_duties.is_empty(), has_next_duty = next_duty.is_some(), has_payload_attrs = !next_payload_attributes.is_empty(), - "sending slot event to auctioneer", + head_sse_events = stats.head_sse_events, + head_sse_parse_errors = stats.head_sse_parse_errors, + payload_attr_sse_events = stats.payload_attr_sse_events, + payload_attr_parse_errors = stats.payload_attr_parse_errors, + duties_fetch_ok = stats.duties_fetch_ok, + duties_fetch_empty = stats.duties_fetch_empty, + duties_fetch_err = stats.duties_fetch_err, + il_fetch_ok = stats.il_fetch_ok, + il_fetch_timeout = stats.il_fetch_timeout, + il_fetch_decode_err = stats.il_fetch_decode_err, + il_consensus_ok = stats.il_consensus_ok, + il_consensus_err = stats.il_consensus_err, + primev_builders_fetched = stats.primev_builders_fetched, + primev_validators_fetched = stats.primev_validators_fetched, + known_validators_refresh_ok = stats.known_validators_refresh_ok, + known_validators_refresh_err = stats.known_validators_refresh_err, + sync_status_ok = stats.sync_status_ok, + sync_status_err = stats.sync_status_err, + sync_status_timeout = stats.sync_status_timeout, + "housekeeper slot stats", ); let update = SlotDuties { slot, new_duties: Some(new_duties), next_duty: next_duty.clone() }; diff --git a/crates/relay/src/main.rs b/crates/relay/src/main.rs index da24bcbb..9ef88263 100644 --- a/crates/relay/src/main.rs +++ b/crates/relay/src/main.rs @@ -75,10 +75,6 @@ fn main() { init_runtime(&config); - if config.enable_flux_profiler { - flux_profiler::enable_profiler("helix"); - } - let keypair = load_keypair(); let instance_id = config @@ -168,6 +164,11 @@ async fn run( spawn_tokio_monitoring(); HelixSpine::remove_all_files(); + + if config.enable_flux_profiler { + flux_profiler::enable_profiler("helix"); + } + let spine = if let Some(spine_config) = spine_config { HelixSpine::new_with_config(None, spine_config) } else { @@ -258,6 +259,7 @@ async fn run( future_results.clone(), decoded.clone(), http_submissions.clone(), + slot_events.clone(), *core, ); attach_tile(decoder_tile, spine, TileConfig::new(*core, ThreadPriority::OSDefault)); diff --git a/crates/relay/src/registration/tile.rs b/crates/relay/src/registration/tile.rs index bd9e271a..0b779e01 100644 --- a/crates/relay/src/registration/tile.rs +++ b/crates/relay/src/registration/tile.rs @@ -7,15 +7,30 @@ use flux::{ use flux_profiler::timed; use helix_common::{chain_info::ChainInfo, utils::utcnow_ns}; use helix_types::SignedValidatorRegistration; +use tracing::info; use crate::{HelixSpine, api::proposer::ProposerApiError, registration::handle::RegWorkerJob}; +/// Registrations aren't slot-scoped (they arrive on their own schedule, not +/// tied to `bid_slot`), so this reports on the same wall-clock cadence as the +/// rest of `Telemetry` rather than on a slot boundary. +/// `registrations_seen == valid + invalid`; `batches_seen == completed + +/// aborted`. +#[derive(Default)] +struct RegStats { + valid: u32, + invalid: u32, + completed: u32, + aborted: u32, +} + struct Telemetry { work: Duration, spin: Duration, next_record: Instant, loop_start: Instant, loop_worked: Duration, + stats: RegStats, } impl Telemetry { @@ -48,6 +63,18 @@ impl Telemetry { WORKER_UTIL.with_label_values(&[id]).observe(util); WORKER_QUEUE_LEN.with_label_values(&[queue_type]).observe(rx.len() as f64); + + let stats = std::mem::take(&mut self.stats); + info!( + id, + registrations_seen = stats.valid + stats.invalid, + valid = stats.valid, + invalid = stats.invalid, + batches_seen = stats.completed + stats.aborted, + completed = stats.completed, + aborted = stats.aborted, + "registration tile stats" + ); } } } @@ -62,6 +89,7 @@ impl Default for Telemetry { Duration::from_millis(utcnow_ns() % 10 * 5), loop_start: Instant::now(), loop_worked: Default::default(), + stats: RegStats::default(), } } } @@ -90,6 +118,11 @@ impl RegistrationTile { let start_task = Instant::now(); let completed = self.process_reg_task(task); let tag = if completed { "RegistrationBatch" } else { "RegistrationBatch_Aborted" }; + if completed { + self.tel.stats.completed += 1; + } else { + self.tel.stats.aborted += 1; + } let dur = start_task.elapsed(); self.tel.loop_worked += dur; @@ -100,7 +133,7 @@ impl RegistrationTile { /// Returns whether the task completed (false = caller dropped the result channel). #[timed] - fn process_reg_task(&self, RegWorkerJob { regs, range, res_tx }: RegWorkerJob) -> bool { + fn process_reg_task(&mut self, RegWorkerJob { regs, range, res_tx }: RegWorkerJob) -> bool { use helix_common::metrics::{WORKER_TASK_COUNT, WORKER_TASK_LATENCY_US}; let mut res = Vec::with_capacity(range.len()); @@ -112,6 +145,11 @@ impl RegistrationTile { let start = Instant::now(); let valid = validate_registration(&self.chain_info, ®s[i]); + if valid.is_ok() { + self.tel.stats.valid += 1; + } else { + self.tel.stats.invalid += 1; + } res.push((i, valid.is_ok())); WORKER_TASK_COUNT.with_label_values(&["Registration", &self.id]).inc(); diff --git a/crates/relay/src/simulator/tile.rs b/crates/relay/src/simulator/tile.rs index a7fec3bf..14b7b212 100644 --- a/crates/relay/src/simulator/tile.rs +++ b/crates/relay/src/simulator/tile.rs @@ -213,6 +213,7 @@ impl SimulatorTile { self.local_telemetry.sims_sent_immediately += 1; self.spawn_sim(id, req) } else { + self.local_telemetry.queued += 1; let evicted = if fast_track { self.priority_requests.store(req, builder_pubkey, &mut self.local_telemetry) } else { @@ -282,6 +283,7 @@ impl SimulatorTile { if let Some(id) = self.next_client(|s| s.can_simulate()) && let Some(req) = self.priority_requests.next_req().or(self.requests.next_req()) { + self.local_telemetry.sims_sent_from_queue += 1; self.spawn_sim(id, req); } } @@ -487,6 +489,7 @@ impl SimulatorTile { fn report(&mut self) { let tel = std::mem::take(&mut self.local_telemetry); + let queue_left = self.requests.reqs.len() + self.priority_requests.reqs.len(); SimulatorMetrics::sim_mananger_count("sims_sent_immediately", tel.sims_sent_immediately); SimulatorMetrics::sim_mananger_count("sims_reqs_dropped", tel.sims_reqs_dropped); @@ -500,13 +503,16 @@ impl SimulatorTile { bid_slot = self.last_bid_slot, sims_reqs = tel.sims_reqs, sims_sent_immediately = tel.sims_sent_immediately, + queued = tel.queued, + sims_sent_from_queue = tel.sims_sent_from_queue, sims_reqs_dropped = tel.sims_reqs_dropped, + queue_left, stale_sim_reqs = tel.stale_sim_reqs, max_pending = tel.max_pending, max_in_flight = tel.max_in_flight, merge_reqs = tel.merge_reqs, dropped_merge_reqs = tel.dropped_merge_reqs, - "sim manager telemetry" + "simulator slot stats" ) } } @@ -557,6 +563,14 @@ struct LocalTelemetry { max_in_flight: usize, merge_reqs: usize, dropped_merge_reqs: usize, + /// Requests with no simulator free at intake, queued for later dispatch. + /// `sims_reqs == sims_sent_immediately + queued`. + queued: usize, + /// Queued requests dispatched once a simulator freed up (queued -> + /// sims_sent_from_queue, or evicted by a fresher request for the same + /// builder -> sims_reqs_dropped, or still resident at slot end -> + /// `queue_left`, read live from the queues rather than stored here). + sims_sent_from_queue: usize, } // Sim id / Simulation Result, so we can use this for merging requests diff --git a/crates/relay/src/tcp_bid_recv/mod.rs b/crates/relay/src/tcp_bid_recv/mod.rs index 051ead00..f0f3cfe5 100644 --- a/crates/relay/src/tcp_bid_recv/mod.rs +++ b/crates/relay/src/tcp_bid_recv/mod.rs @@ -1,4 +1,9 @@ -use std::{collections::HashMap, net::SocketAddr, sync::Arc}; +use std::{ + collections::HashMap, + net::SocketAddr, + sync::Arc, + time::{Duration, Instant}, +}; use bytes::Bytes; use dashmap::DashMap; @@ -12,6 +17,7 @@ use helix_common::{SubmissionTrace, metrics::SUB_CLIENT_TO_SERVER_LATENCY, utils use helix_tcp_types::BID_SUB_HEADER_SIZE; use helix_types::BlsPublicKeyBytes; use ssz::{Decode, Encode}; +use tracing::info; use uuid::Uuid; use crate::{ @@ -32,6 +38,24 @@ pub use crate::tcp_bid_recv::types::{ type SubmissionError = (Token, Option, Option, BidSubmissionError); +/// Connections aren't slot-scoped, so this reports on a wall-clock cadence +/// instead of a slot boundary. For a registered peer, every `Message` is +/// either a decoded submission or a header-parse error: +/// `submissions_received + header_parse_errors == messages_from_registered`. +/// For an unregistered peer, every `Message` is a registration attempt: +/// `registration_ok + registration_invalid == registration_attempts`. +#[derive(Default)] +struct Stats { + accepted: u32, + reconnected: u32, + disconnected: u32, + registration_ok: u32, + registration_invalid: u32, + submissions_received: u32, + header_parse_errors: u32, + results_sent: u32, +} + pub struct BidSubmissionTcpListener { listener: TcpConnector, @@ -44,9 +68,14 @@ pub struct BidSubmissionTcpListener { // dcache bypass: stage a stable copy of the payload, the dcache slot can be // mutated out from under the decoder between publish and consume. http_submissions: Arc>, + + stats: Stats, + next_report: Instant, } impl BidSubmissionTcpListener { + const REPORT_FREQ: Duration = Duration::from_millis(500); + pub fn new( listener_addr: SocketAddr, api_key_cache: Arc>>, @@ -71,8 +100,33 @@ impl BidSubmissionTcpListener { registered: HashMap::with_capacity(max_connections), submission_errors: Vec::with_capacity(max_connections), http_submissions, + stats: Stats::default(), + next_report: Instant::now() + Self::REPORT_FREQ, } } + + fn maybe_report_stats(&mut self) { + let now = Instant::now(); + if now < self.next_report { + return; + } + self.next_report = now + Self::REPORT_FREQ; + + let stats = std::mem::take(&mut self.stats); + info!( + accepted = stats.accepted, + reconnected = stats.reconnected, + disconnected = stats.disconnected, + registration_attempts = stats.registration_ok + stats.registration_invalid, + registration_ok = stats.registration_ok, + registration_invalid = stats.registration_invalid, + messages_from_registered = stats.submissions_received + stats.header_parse_errors, + submissions_received = stats.submissions_received, + header_parse_errors = stats.header_parse_errors, + results_sent = stats.results_sent, + "tcp bid recv tile stats" + ); + } } impl Tile for BidSubmissionTcpListener { @@ -80,15 +134,18 @@ impl Tile for BidSubmissionTcpListener { self.listener.poll_with_produce(&mut adapter.producers, |event| match event { PollEvent::Accept { listener: _, stream, peer_addr } => { tracing::trace!("connected to new peer {:?} with token {:?}", peer_addr, stream); + self.stats.accepted += 1; None } PollEvent::Reconnect { token } => { tracing::trace!("reconnected to peer with token {:?}", token); + self.stats.reconnected += 1; None } PollEvent::Disconnect { token } => { tracing::trace!("disconnected from peer with token {:?}", token); self.registered.remove(&token); + self.stats.disconnected += 1; None } PollEvent::Message { token, payload, send_ts } => { @@ -97,6 +154,7 @@ impl Tile for BidSubmissionTcpListener { Ok(header) => header, Err(e) => { tracing::error!("{e} failed to parse the bid submission header"); + self.stats.header_parse_errors += 1; self.submission_errors.push(( token, None, @@ -106,6 +164,7 @@ impl Tile for BidSubmissionTcpListener { return None; } }; + self.stats.submissions_received += 1; let id = Uuid::new_v4(); let _enter = tracing::info_span!( @@ -154,17 +213,20 @@ impl Tile for BidSubmissionTcpListener { .is_some_and(|p| p.value().contains(&msg.builder_pubkey)) { self.registered.insert(token, msg.builder_pubkey); + self.stats.registration_ok += 1; } else { tracing::error!( "unknown api key and pubkey pair: {} {}, disconnecting peer", api_key, msg.builder_pubkey ); + self.stats.registration_invalid += 1; self.to_disconnect.push(token); } } Err(e) => { tracing::error!(err=?e, "invalid registration message"); + self.stats.registration_invalid += 1; self.to_disconnect.push(token); } } @@ -189,9 +251,12 @@ impl Tile for BidSubmissionTcpListener { let response = response_from_submission_result(seq_num, id, r.tcp_status, r.error_msg.as_str()); tracing::debug!("submission result: {}", response); + self.stats.results_sent += 1; self.listener.write_or_enqueue_with(SendBehavior::Single(Token(token)), |buffer| { response.ssz_append(buffer); }); }); + + self.maybe_report_stats(); } } diff --git a/crates/tcp-types/src/merging/mod.rs b/crates/tcp-types/src/merging/mod.rs index b63cbb79..42c73316 100644 --- a/crates/tcp-types/src/merging/mod.rs +++ b/crates/tcp-types/src/merging/mod.rs @@ -11,7 +11,7 @@ pub mod relay_to_builder; use bitflags::bitflags; -pub const MERGING_PROTOCOL_VERSION: u16 = 1; +pub const MERGING_PROTOCOL_VERSION: u16 = 2; pub const MERGING_HEADER_SIZE: usize = 2; /// Message-type IDs. `0x00-0x0f` control, `0x10-0x3f` relay->builder, diff --git a/crates/tcp-types/src/merging/order.rs b/crates/tcp-types/src/merging/order.rs index 3cdc0e8a..cf287838 100644 --- a/crates/tcp-types/src/merging/order.rs +++ b/crates/tcp-types/src/merging/order.rs @@ -9,6 +9,19 @@ pub const MAX_BUNDLE_TXS: usize = 32; pub const MAX_ORDERS_PER_BLOCK: usize = 1024; pub const MAX_BLOCK_TXS: usize = u16::MAX as usize; +/// Length of a tx-hash reference in `MergeableBlockV1::execution_payload`'s +/// transaction list — `keccak256(rlp)` of a tx already sent once on this +/// connection. Unambiguous against a real tx: the shortest legal signed tx +/// (list prefix + nonce/gas/value/`to`/empty data + v/r/s) is well over 32 +/// bytes, so any entry of exactly this length is a reference, never a tx. +pub const TX_HASH_REF_LEN: usize = 32; + +/// True if `raw_tx` is a tx-hash reference rather than a raw transaction. See +/// [`TX_HASH_REF_LEN`]. +pub fn is_tx_hash_ref(raw_tx: &[u8]) -> bool { + raw_tx.len() == TX_HASH_REF_LEN +} + /// Identity of a mergeable order, derived by both sides from a forwarded /// `MergeableBlockV1`; traceable to the contributing builder and the block /// the order came from. Not itself a wire type — the canonical id spec. diff --git a/crates/tcp-types/src/merging/relay_to_builder.rs b/crates/tcp-types/src/merging/relay_to_builder.rs index b6f89f05..66b04fcf 100644 --- a/crates/tcp-types/src/merging/relay_to_builder.rs +++ b/crates/tcp-types/src/merging/relay_to_builder.rs @@ -21,6 +21,17 @@ pub struct SlotStartV1 { /// highest-value block on duplicates) and, when `allow_appending`, stores the /// payload as a merge-base candidate so a later `ActivateBaseBlockV1` is the /// only message on the top-bid critical path. +/// +/// `execution_payload.transactions` entries are dehydrated relative to this +/// connection: an entry of `order::TX_HASH_REF_LEN` bytes +/// (`order::is_tx_hash_ref`) is `keccak256(rlp)` of a tx already sent once on +/// this same connection (by any block, from any originating builder — the +/// cache is shared, not partitioned per source), to be resolved from that +/// side's own cache; anything else is a raw tx, which the receiver caches by +/// its hash for later references. The cache is connection-scoped: a fresh +/// connection starts empty, so the relay must resend full bytes for anything +/// replayed after a reconnect rather than reusing hashes from before the +/// drop. #[derive(Debug, Clone, PartialEq, Encode, Decode)] pub struct MergeableBlockV1 { pub slot: u64, diff --git a/crates/types/src/hydration.rs b/crates/types/src/hydration.rs index e674821d..d9b518a7 100644 --- a/crates/types/src/hydration.rs +++ b/crates/types/src/hydration.rs @@ -2,7 +2,7 @@ use std::{hash::Hasher, sync::Arc}; use alloy_primitives::{Address, B256, U256}; use flux_profiler::timed; -use lh_types::{ForkName, ForkVersionDecode}; +use lh_types::{ForkName, ForkVersionDecode, test_utils::TestRandom}; use rustc_hash::{FxHashMap, FxHasher}; use serde::{Deserialize, Serialize}; use ssz::{Decode, DecodeError}; @@ -11,8 +11,8 @@ use tracing::trace; use tree_hash::TreeHash; use crate::{ - BidTrace, Blob, BlobsBundle, BlockValidationError, BlsPublicKeyBytes, BlsSignatureBytes, - ExecutionPayload, SignedBidSubmission, + BidTrace, Blob, BlobsBundle, BlockMergingData, BlockValidationError, BlsPublicKeyBytes, + BlsSignatureBytes, ExecutionPayload, SignedBidSubmission, bid_adjustment_data::BidAdjustmentData, bid_submission, fields::{ExecutionRequests, KzgCommitment, KzgProof, Transaction}, @@ -212,6 +212,62 @@ impl ForkVersionDecode for DehydratedBidSubmissionFuluWithAdjustments { } } +#[derive(Debug, Clone, Serialize, Deserialize, Encode, Decode)] +pub struct DehydratedBidSubmissionFuluWithMergingData { + message: BidTrace, + execution_payload: ExecutionPayload, + blobs_bundle: DehydratedBlobsFulu, + execution_requests: Arc, + signature: BlsSignatureBytes, + tx_root: Option, + merging_data: BlockMergingData, +} + +impl DehydratedBidSubmissionFuluWithMergingData { + pub fn split(self) -> (DehydratedBidSubmission, BlockMergingData) { + ( + DehydratedBidSubmission::Fulu(DehydratedBidSubmissionFulu { + message: self.message, + execution_payload: self.execution_payload, + blobs_bundle: self.blobs_bundle, + execution_requests: self.execution_requests, + signature: self.signature, + tx_root: self.tx_root, + }), + self.merging_data, + ) + } +} + +impl ForkVersionDecode for DehydratedBidSubmissionFuluWithMergingData { + fn from_ssz_bytes_by_fork(bytes: &[u8], fork: ForkName) -> Result { + match fork { + ForkName::Base | + ForkName::Altair | + ForkName::Bellatrix | + ForkName::Capella | + ForkName::Deneb | + ForkName::Gloas | + ForkName::Electra => Err(DecodeError::NoMatchingVariant), + ForkName::Fulu => DehydratedBidSubmissionFuluWithMergingData::from_ssz_bytes(bytes), + } + } +} + +impl TestRandom for DehydratedBidSubmissionFuluWithMergingData { + fn random_for_test(rng: &mut impl rand::RngCore) -> Self { + Self { + message: BidTrace::random_for_test(rng), + execution_payload: ExecutionPayload::random_for_test(rng), + blobs_bundle: DehydratedBlobsFulu { commitments: vec![], new_items: vec![] }, + execution_requests: Arc::new(ExecutionRequests::random_for_test(rng)), + signature: BlsSignatureBytes::random(), + tx_root: None, + merging_data: BlockMergingData::random_for_test(rng), + } + } +} + #[derive(Debug, Clone, Serialize, Deserialize, Encode, Decode)] struct DehydratedBlobsFulu { commitments: Vec, @@ -549,3 +605,35 @@ pub enum HydrationError { #[error("too many blobs: blobs {blobs}, max {max}")] TooManyBlobs { blobs: usize, max: usize }, } + +#[cfg(test)] +mod tests { + use ssz::Encode; + + use super::*; + + #[test] + fn dehydrated_with_merging_data_ssz_round_trip() { + let submission = + DehydratedBidSubmissionFuluWithMergingData::random_for_test(&mut rand::rng()); + + let bytes = submission.as_ssz_bytes(); + let decoded = DehydratedBidSubmissionFuluWithMergingData::from_ssz_bytes(&bytes) + .expect("SSZ decode should succeed"); + + assert_eq!(submission.merging_data, decoded.merging_data); + assert_eq!(submission.message, decoded.message); + } + + #[test] + fn dehydrated_with_merging_data_split_preserves_merge_orders() { + let submission = + DehydratedBidSubmissionFuluWithMergingData::random_for_test(&mut rand::rng()); + let expected_merging_data = submission.merging_data.clone(); + + let (dehydrated, split_merging_data) = submission.split(); + + assert_eq!(split_merging_data, expected_merging_data); + assert!(matches!(dehydrated, DehydratedBidSubmission::Fulu(_))); + } +}