From ae5c999d15af003280482b00cdb020eaf45cbe65 Mon Sep 17 00:00:00 2001 From: owen Date: Fri, 17 Jul 2026 19:18:04 +0100 Subject: [PATCH 1/6] ensures dehydrated blocks are included for block merging --- crates/relay/src/block_merging/tile.rs | 28 +++++++++++++------------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/crates/relay/src/block_merging/tile.rs b/crates/relay/src/block_merging/tile.rs index ad02662d..d43d5a56 100644 --- a/crates/relay/src/block_merging/tile.rs +++ b/crates/relay/src/block_merging/tile.rs @@ -600,22 +600,22 @@ impl BlockMergingTile { signed } // hydrating also feeds this submission's new txs into the cache - Submission::Dehydrated(d) => { - match self.hydration_cache.hydrate(d.clone(), self.chain_info.max_blobs_per_block()) - { - Ok(h) => { - if !is_replay { - self.stats.forwarded_hydrated += 1; - } - hydrated = h.submission; - &hydrated - } - Err(_) => { - self.stats.hydration_failed += 1; - return; + Submission::Dehydrated(d) => match self + .hydration_cache + .hydrate(d.clone(), self.chain_info.max_blobs_per_block()) + { + Ok(h) => { + if !is_replay { + self.stats.forwarded_hydrated += 1; } + hydrated = h.submission; + &hydrated } - } + Err(_) => { + self.stats.hydration_failed += 1; + return; + } + }, }; let mut merge_orders = Vec::with_capacity(merging.merge_orders.len()); From eac50ab6497b4f8dfaa73672b4054cf11a963833 Mon Sep 17 00:00:00 2001 From: owen Date: Fri, 17 Jul 2026 19:53:24 +0100 Subject: [PATCH 2/6] fix lints --- crates/relay/src/block_merging/tile.rs | 28 +++++++++++++------------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/crates/relay/src/block_merging/tile.rs b/crates/relay/src/block_merging/tile.rs index d43d5a56..ad02662d 100644 --- a/crates/relay/src/block_merging/tile.rs +++ b/crates/relay/src/block_merging/tile.rs @@ -600,22 +600,22 @@ impl BlockMergingTile { signed } // hydrating also feeds this submission's new txs into the cache - Submission::Dehydrated(d) => match self - .hydration_cache - .hydrate(d.clone(), self.chain_info.max_blobs_per_block()) - { - Ok(h) => { - if !is_replay { - self.stats.forwarded_hydrated += 1; + Submission::Dehydrated(d) => { + match self.hydration_cache.hydrate(d.clone(), self.chain_info.max_blobs_per_block()) + { + Ok(h) => { + if !is_replay { + self.stats.forwarded_hydrated += 1; + } + hydrated = h.submission; + &hydrated + } + Err(_) => { + self.stats.hydration_failed += 1; + return; } - hydrated = h.submission; - &hydrated - } - Err(_) => { - self.stats.hydration_failed += 1; - return; } - }, + } }; let mut merge_orders = Vec::with_capacity(merging.merge_orders.len()); From bc97f94eddf8099a21d1885451504bbdd21e6f59 Mon Sep 17 00:00:00 2001 From: owen Date: Fri, 17 Jul 2026 20:40:16 +0100 Subject: [PATCH 3/6] add missing merging_data for Dehydrated blocks --- crates/relay/src/bid_decoder/tile.rs | 20 +++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/crates/relay/src/bid_decoder/tile.rs b/crates/relay/src/bid_decoder/tile.rs index 3a35ba05..d3b44f55 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,8 +18,8 @@ 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 zstd::zstd_safe::WriteBuf; @@ -232,8 +232,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 +247,15 @@ impl DecoderTile { None } } - } else { - None } + // transactions aren't resolved yet, so orders referencing tx indices can't be + // expanded here; only append-only submissions reach this point (see + // SubmissionDecoder::decode_dehydrated), so merge_orders is always empty. + 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 From a66e0a597393eb5f5527682f74798dc3eabd339f Mon Sep 17 00:00:00 2001 From: owen Date: Mon, 20 Jul 2026 14:33:05 +0100 Subject: [PATCH 4/6] improve tile stats logs --- crates/relay/src/api/builder/top_bid.rs | 82 ++++++++++++++++- crates/relay/src/auctioneer/bid_sorter.rs | 12 ++- crates/relay/src/auctioneer/mod.rs | 31 +++++++ crates/relay/src/bid_decoder/tile.rs | 84 ++++++++++++++++- crates/relay/src/data_gatherer/tile.rs | 44 +++++++++ crates/relay/src/housekeeper/tile.rs | 106 ++++++++++++++++++++-- crates/relay/src/main.rs | 1 + crates/relay/src/registration/tile.rs | 40 +++++++- crates/relay/src/simulator/tile.rs | 16 +++- crates/relay/src/tcp_bid_recv/mod.rs | 67 +++++++++++++- 10 files changed, 461 insertions(+), 22 deletions(-) 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/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 d3b44f55..3277be2c 100644 --- a/crates/relay/src/bid_decoder/tile.rs +++ b/crates/relay/src/bid_decoder/tile.rs @@ -21,7 +21,8 @@ use helix_types::{ 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, } } @@ -390,6 +452,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/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..0d5389b6 100644 --- a/crates/relay/src/main.rs +++ b/crates/relay/src/main.rs @@ -258,6 +258,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(); } } From faaaec401688e29e072aa4cf9c60bf897f84514f Mon Sep 17 00:00:00 2001 From: owen Date: Mon, 20 Jul 2026 16:32:52 +0100 Subject: [PATCH 5/6] fix for connection issues --- crates/relay/src/block_merging/tile.rs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/crates/relay/src/block_merging/tile.rs b/crates/relay/src/block_merging/tile.rs index ad02662d..a1924c46 100644 --- a/crates/relay/src/block_merging/tile.rs +++ b/crates/relay/src/block_merging/tile.rs @@ -224,7 +224,12 @@ impl BlockMergingTile { // 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, From a735fc82332b11cf8b29e1f227e269c8f89745d1 Mon Sep 17 00:00:00 2001 From: owen Date: Mon, 20 Jul 2026 19:57:14 +0100 Subject: [PATCH 6/6] mv blk merging hydration to merge builder --- config.example.yml | 6 +- crates/common/src/config.rs | 3 +- crates/relay/src/block_merging/tile.rs | 231 ++++++++++-------- crates/tcp-types/src/merging/mod.rs | 2 +- crates/tcp-types/src/merging/order.rs | 13 + .../tcp-types/src/merging/relay_to_builder.rs | 11 + 6 files changed, 154 insertions(+), 112 deletions(-) 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/relay/src/block_merging/tile.rs b/crates/relay/src/block_merging/tile.rs index a1924c46..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,17 +218,12 @@ 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. @@ -235,8 +240,9 @@ impl BlockMergingTile { 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, @@ -256,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, @@ -312,7 +315,8 @@ impl BlockMergingTile { // poll, all reactions are buffered. let Self { connector, - conns, + token: my_token, + conn, slot, stats, to_disconnect, @@ -328,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 @@ -373,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) { @@ -407,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 { @@ -455,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) { @@ -481,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(); @@ -519,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); } @@ -554,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" ); } @@ -567,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 { @@ -630,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, @@ -650,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); @@ -660,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) { @@ -697,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. @@ -729,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/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,