diff --git a/crates/op-rbuilder/src/builder/continuous/publish.rs b/crates/op-rbuilder/src/builder/continuous/publish.rs index ad41c68e..048a10e8 100644 --- a/crates/op-rbuilder/src/builder/continuous/publish.rs +++ b/crates/op-rbuilder/src/builder/continuous/publish.rs @@ -10,22 +10,21 @@ use crate::{ builder_tx::BuilderTransactions, cancellation::FlashblockJobCancellation, context::OpPayloadJobCtx, + fanout::{self, FlashblockEvent}, payload::{ BuildProgress, BuildState, FlashblocksState, JobDeps, OpPayloadBuilder, PayloadBuildStats, }, - timing::compute_slot_offset_ms, }, - metrics::record_flashblock_publish_timing, primitives::reth::ExecutionInfo, traits::{ClientBounds, PoolBounds}, }; use alloy_primitives::B256; use reth_node_api::PayloadBuilderError; use reth_optimism_node::OpBuiltPayload; -use std::{ops::ControlFlow, time::Instant}; +use std::{ops::ControlFlow, sync::Arc, time::Instant}; use tokio::sync::watch; -use tracing::{debug, info, metadata::Level, span}; +use tracing::{info, metadata::Level, span}; // === Per-interval publishing and advancement ================= // @@ -101,17 +100,14 @@ where Client: ClientBounds + 'static, BuilderTx: BuilderTransactions + Send + Sync + 'static, { - /// Publish a candidate flashblock immediately. The context is only used for - /// publish timing metadata, so this can run before awaiting the build task. - /// Returns `(byte_size, slot_offset_ms)` so the caller can include - /// slot_offset_ms in the fb_published tx_trace log. + /// Publish a candidate flashblock immediately, before awaiting the build task. fn publish_candidate( &self, candidate: &BestCandidate, best_payload_tx: &watch::Sender>, fb_span: &tracing::Span, ctx: &OpPayloadJobCtx, - ) -> Result<(usize, f64), PayloadBuilderError> { + ) { let _publish_span = if fb_span.is_none() { tracing::Span::none() } else { @@ -120,16 +116,18 @@ where .entered(); let (_, ref new_payload, ref fb_payload_delta) = candidate.result; - let flashblock_byte_size = self - .ws_pub() - .publish(fb_payload_delta) - .map_err(PayloadBuilderError::other)?; best_payload_tx.send_replace(Some(new_payload.clone())); - self.notify_built_payload(new_payload.clone()); - let slot_offset_ms = - compute_slot_offset_ms(ctx.attributes().timestamp(), self.config().block_time); - record_flashblock_publish_timing(candidate.fb_state.flashblock_index(), slot_offset_ms); - Ok((flashblock_byte_size, slot_offset_ms)) + fanout::emit( + self.flashblock_tx(), + self.metrics(), + FlashblockEvent { + fb_payload: Arc::new(fb_payload_delta.clone()), + built: Arc::new(new_payload.clone()), + attributes_timestamp_secs: ctx.attributes().timestamp(), + ws_eligible: true, + tx_trace_total_txs: candidate.info.executed_transactions.len(), + }, + ); } pub(super) async fn publish_and_spawn_next( @@ -285,12 +283,7 @@ where ) -> Result, PayloadBuilderError> { match outcome { TriggerOutcome::PublishAndAdvance | TriggerOutcome::PublishAndStop => { - let (byte_size, slot_offset_ms) = self.publish_candidate( - &candidate, - deps.best_payload_tx, - fb_span, - &candidate_ctx, - )?; + self.publish_candidate(&candidate, deps.best_payload_tx, fb_span, &candidate_ctx); self.record_continuous_candidate_metrics( fb_span, candidates_evaluated, @@ -303,8 +296,6 @@ where fb_span, candidate_ctx, candidate, - byte_size, - slot_offset_ms, candidates_evaluated, candidates_improved, new_fb_cancel, @@ -375,8 +366,6 @@ where fb_span: &tracing::Span, ctx: OpPayloadJobCtx, candidate: BestCandidate, - byte_size: usize, - slot_offset_ms: f64, candidates_evaluated: u64, candidates_improved: u64, new_fb_cancel: FlashblockJobCancellation, @@ -413,24 +402,6 @@ where .address_limiter() .restore_pending(&limiter_snapshot); - if self.config().enable_tx_tracking_debug_logs { - debug!( - target: "tx_trace", - payload_id = %base_state.ctx.payload_id(), - block_number = base_state.ctx.block_number(), - flashblock_index = base_state.fb_state.flashblock_index(), - byte_size, - total_txs = base_state.info.executed_transactions.len(), - slot_offset_ms, - stage = "fb_published" - ); - } - - base_state - .ctx - .metrics - .flashblock_byte_size_histogram - .record(byte_size as f64); base_state .ctx .metrics diff --git a/crates/op-rbuilder/src/builder/fanout.rs b/crates/op-rbuilder/src/builder/fanout.rs new file mode 100644 index 00000000..a3252597 --- /dev/null +++ b/crates/op-rbuilder/src/builder/fanout.rs @@ -0,0 +1,370 @@ +use crate::{ + builder::{p2p::Message, timing::compute_slot_offset_ms, wspub::WebSocketPublisher}, + metrics::{OpRBuilderMetrics, record_flashblock_publish_timing}, +}; +use op_alloy_rpc_types_engine::OpFlashblockPayload; +use reth_node_builder::Events; +use reth_optimism_node::{OpBuiltPayload, OpEngineTypes}; +use std::{io, sync::Arc, time::Duration}; +use tokio::sync::{broadcast, mpsc}; +use tracing::{debug, warn}; + +/// ~6 slots of events at 200ms flashblock +pub(crate) const FLASHBLOCK_BUS_CAPACITY: usize = 32; + +#[derive(Clone, Debug)] +pub(crate) struct FlashblockEvent { + /// Wire payload published to websocket subscribers. + pub fb_payload: Arc, + /// Built payload forwarded to p2p and engine feedback subscribers. + pub built: Arc, + /// Payload attributes timestamp used to calculate publication timing. + pub attributes_timestamp_secs: u64, + /// Whether this event should be published over websocket. + pub ws_eligible: bool, + /// Total transaction count included in tx-trace publication logs. + pub tx_trace_total_txs: usize, +} + +pub(crate) fn channel() -> ( + broadcast::Sender, + broadcast::Receiver, +) { + broadcast::channel(FLASHBLOCK_BUS_CAPACITY) +} + +pub(crate) fn emit( + sender: &broadcast::Sender, + metrics: &OpRBuilderMetrics, + event: FlashblockEvent, +) -> bool { + let payload_id = event.fb_payload.payload_id; + if sender.send(event).is_err() { + warn!( + target: "payload_builder", + id = %payload_id, + "flashblock fanout bus has no active subscribers; dropping event" + ); + metrics.fanout_no_subscribers.increment(1); + false + } else { + true + } +} + +async fn recv_event( + receiver: &mut broadcast::Receiver, + mut on_lagged: impl FnMut(u64), +) -> Option { + loop { + match receiver.recv().await { + Ok(event) => return Some(event), + Err(broadcast::error::RecvError::Lagged(n)) => on_lagged(n), + Err(broadcast::error::RecvError::Closed) => return None, + } + } +} + +async fn recv_ws_event( + receiver: &mut broadcast::Receiver, + mut on_lagged: impl FnMut(u64), +) -> Option { + loop { + let event = recv_event(receiver, &mut on_lagged).await?; + if event.ws_eligible { + return Some(event); + } + } +} + +pub(crate) async fn websocket_subscriber( + receiver: broadcast::Receiver, + ws_pub: WebSocketPublisher, + metrics: Arc, + block_time: Duration, + tx_tracking_logs: bool, +) { + let lag_metrics = metrics.clone(); + websocket_subscriber_with( + receiver, + move |payload| ws_pub.publish(payload), + move |byte_size| metrics.flashblock_byte_size_histogram.record(byte_size), + move |n| lag_metrics.ws_publish_lagged.increment(n), + block_time, + tx_tracking_logs, + ) + .await; +} + +async fn websocket_subscriber_with( + mut receiver: broadcast::Receiver, + mut publish: impl FnMut(&OpFlashblockPayload) -> io::Result, + mut record_byte_size: impl FnMut(f64), + mut on_lagged: impl FnMut(u64), + block_time: Duration, + tx_tracking_logs: bool, +) { + while let Some(event) = recv_ws_event(&mut receiver, |n| { + on_lagged(n); + }) + .await + { + // WebSocket failures are isolated from payload construction so a + // rollup-boost disconnect cannot stop block building. + let byte_size = match publish(&event.fb_payload) { + Ok(byte_size) => byte_size, + Err(error) => { + warn!( + target: "payload_builder", + %error, + "failed to publish flashblock via websocket" + ); + continue; + } + }; + let slot_offset_ms = compute_slot_offset_ms(event.attributes_timestamp_secs, block_time); + record_flashblock_publish_timing(event.fb_payload.index, slot_offset_ms); + + if tx_tracking_logs { + debug!( + target: "tx_trace", + payload_id = %event.fb_payload.payload_id, + block_number = event.fb_payload.metadata.block_number, + flashblock_index = event.fb_payload.index, + byte_size, + total_txs = event.tx_trace_total_txs, + slot_offset_ms, + stage = "fb_published" + ); + } + + record_byte_size(byte_size as f64); + } +} + +pub(crate) async fn p2p_subscriber( + mut receiver: broadcast::Receiver, + p2p_tx: mpsc::Sender, + metrics: Arc, +) { + while let Some(event) = recv_event(&mut receiver, |n| { + metrics.p2p_forward_lagged.increment(n); + }) + .await + { + // A closed channel is expected when p2p is disabled. + let _ = p2p_tx.send((*event.built).clone().into()).await; + } +} + +pub(crate) async fn engine_feedback_subscriber( + mut receiver: broadcast::Receiver, + payload_events_handle: broadcast::Sender>, + metrics: Arc, +) { + while let Some(event) = recv_event(&mut receiver, |n| { + metrics.engine_feedback_lagged.increment(n); + }) + .await + { + if let Err(error) = payload_events_handle.send(Events::BuiltPayload((*event.built).clone())) + { + warn!( + target: "payload_builder", + %error, + "failed to send BuiltPayload event" + ); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use alloy_consensus::{BlockBody, Header}; + use alloy_primitives::{B256, Bloom, U256}; + use metrics::Counter; + use op_alloy_rpc_types_engine::{OpFlashblockPayloadDelta, OpFlashblockPayloadMetadata}; + use reth_optimism_primitives::OpBlock; + use reth_payload_builder::PayloadId; + use reth_primitives_traits::Block as _; + use std::{ + collections::BTreeMap, + sync::atomic::{AtomicU64, Ordering}, + }; + + fn event(index: u64, ws_eligible: bool) -> FlashblockEvent { + let payload_id = PayloadId::new([index as u8; 8]); + FlashblockEvent { + fb_payload: Arc::new(OpFlashblockPayload { + payload_id, + index, + base: None, + diff: OpFlashblockPayloadDelta { + state_root: B256::ZERO, + receipts_root: B256::ZERO, + logs_bloom: Bloom::ZERO, + gas_used: 0, + block_hash: B256::ZERO, + transactions: Vec::new(), + withdrawals: Vec::new(), + withdrawals_root: B256::ZERO, + blob_gas_used: None, + }, + metadata: OpFlashblockPayloadMetadata { + block_number: index, + new_account_balances: BTreeMap::new(), + receipts: BTreeMap::new(), + }, + }), + built: Arc::new(OpBuiltPayload::new( + payload_id, + Arc::new(OpBlock::new(Header::default(), BlockBody::default()).seal_slow()), + U256::ZERO, + None, + )), + attributes_timestamp_secs: 0, + ws_eligible, + tx_trace_total_txs: 0, + } + } + + #[test] + fn emit_without_receivers_counts_total_subscriber_loss() { + let (tx, receiver) = channel(); + drop(receiver); + let no_subscribers = Arc::new(AtomicU64::new(0)); + let metrics = OpRBuilderMetrics { + fanout_no_subscribers: Counter::from_arc(no_subscribers.clone()), + ..Default::default() + }; + + assert!(!emit(&tx, &metrics, event(0, true))); + assert_eq!(no_subscribers.load(Ordering::Relaxed), 1); + } + + #[tokio::test] + async fn fans_out_to_multiple_subscribers_in_order() { + let (tx, mut first) = channel(); + let mut second = tx.subscribe(); + for index in 0..3 { + tx.send(event(index, true)).unwrap(); + } + drop(tx); + + let mut first_indexes = Vec::new(); + let mut second_indexes = Vec::new(); + while let Some(event) = recv_event(&mut first, |_| {}).await { + first_indexes.push(event.fb_payload.index); + } + while let Some(event) = recv_event(&mut second, |_| {}).await { + second_indexes.push(event.fb_payload.index); + } + + assert_eq!(first_indexes, [0, 1, 2]); + assert_eq!(second_indexes, [0, 1, 2]); + } + + #[tokio::test] + async fn lagged_receiver_counts_drops_and_continues() { + let (tx, mut receiver) = broadcast::channel(2); + for index in 0..4 { + tx.send(event(index, true)).unwrap(); + } + + let lagged = AtomicU64::new(0); + let received = recv_event(&mut receiver, |n| { + lagged.fetch_add(n, Ordering::Relaxed); + }) + .await + .unwrap(); + + assert_eq!(lagged.load(Ordering::Relaxed), 2); + assert_eq!(received.fb_payload.index, 2); + } + + #[tokio::test] + async fn ineligible_event_skips_websocket_but_reaches_engine_feedback() { + let (tx, mut ws_receiver) = channel(); + let engine_receiver = tx.subscribe(); + let (payload_events_handle, mut payload_events_receiver) = broadcast::channel(2); + tokio::spawn(engine_feedback_subscriber( + engine_receiver, + payload_events_handle, + Arc::new(OpRBuilderMetrics::default()), + )); + tx.send(event(0, false)).unwrap(); + tx.send(event(1, true)).unwrap(); + + let engine_event = payload_events_receiver.recv().await.unwrap(); + let ws_event = recv_ws_event(&mut ws_receiver, |_| {}).await.unwrap(); + + let Events::BuiltPayload(engine_payload) = engine_event else { + panic!("expected built payload event"); + }; + assert_eq!(engine_payload.id(), PayloadId::new([0; 8])); + assert_eq!(ws_event.fb_payload.index, 1); + } + + #[tokio::test] + async fn p2p_subscriber_forwards_events_in_order() { + let (tx, receiver) = channel(); + let (p2p_tx, mut p2p_rx) = mpsc::channel(3); + let task = tokio::spawn(p2p_subscriber( + receiver, + p2p_tx, + Arc::new(OpRBuilderMetrics::default()), + )); + + for index in 0..3 { + tx.send(event(index, true)).unwrap(); + } + drop(tx); + + let mut ids = Vec::new(); + while let Some(Message::OpBuiltPayload(payload)) = p2p_rx.recv().await { + ids.push(payload.id); + } + task.await.unwrap(); + + assert_eq!( + ids, + [ + PayloadId::new([0; 8]), + PayloadId::new([1; 8]), + PayloadId::new([2; 8]), + ] + ); + } + + #[tokio::test] + async fn websocket_subscriber_filters_and_records_successful_publications() { + let (tx, receiver) = channel(); + tx.send(event(0, false)).unwrap(); + tx.send(event(1, true)).unwrap(); + tx.send(event(2, true)).unwrap(); + drop(tx); + + let mut published = Vec::new(); + let mut recorded_sizes = Vec::new(); + websocket_subscriber_with( + receiver, + |payload| { + published.push(payload.index); + if payload.index == 1 { + Err(io::Error::other("test publish failure")) + } else { + Ok(100 + payload.index as usize) + } + }, + |byte_size| recorded_sizes.push(byte_size), + |_| {}, + Duration::from_secs(1), + false, + ) + .await; + + assert_eq!(published, [1, 2]); + assert_eq!(recorded_sizes, [102.0]); + } +} diff --git a/crates/op-rbuilder/src/builder/mod.rs b/crates/op-rbuilder/src/builder/mod.rs index 535cc04c..6d847537 100644 --- a/crates/op-rbuilder/src/builder/mod.rs +++ b/crates/op-rbuilder/src/builder/mod.rs @@ -16,6 +16,7 @@ pub(crate) mod cancellation; mod config; mod context; mod continuous; +mod fanout; mod flashblocks_builder_tx; mod generator; mod p2p; diff --git a/crates/op-rbuilder/src/builder/payload.rs b/crates/op-rbuilder/src/builder/payload.rs index cd9882be..96b1486d 100644 --- a/crates/op-rbuilder/src/builder/payload.rs +++ b/crates/op-rbuilder/src/builder/payload.rs @@ -1,4 +1,7 @@ -use super::{state_root::StateRootCalculator, wspub::WebSocketPublisher}; +use super::{ + fanout::{self, FlashblockEvent}, + state_root::StateRootCalculator, +}; use crate::{ builder::{ BuilderConfig, @@ -7,12 +10,12 @@ use crate::{ cancellation::{CancellationReason, FlashblockJobCancellation, PayloadJobCancellation}, context::{OpPayloadBuilderCtx, OpPayloadJobCtx}, generator::{BuildArguments, PayloadBuilder}, - timing::{FlashblockScheduler, compute_slot_offset_ms}, + timing::FlashblockScheduler, }, evm::OpBlockEvmFactory, hardforks::ActiveHardforks, limiter::AddressLimiter, - metrics::{OpRBuilderMetrics, record_flashblock_publish_timing}, + metrics::OpRBuilderMetrics, primitives::reth::ExecutionInfo, runtime_ext::RuntimeExt, tokio_metrics::FlashblocksTaskMetrics, @@ -42,7 +45,7 @@ use std::{ sync::{Arc, atomic::AtomicU64}, time::{Duration, Instant}, }; -use tokio::sync::{mpsc, watch}; +use tokio::sync::{broadcast, mpsc, watch}; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, info_span, metadata::Level, span, warn}; @@ -360,15 +363,8 @@ pub(crate) struct OpPayloadBuilderInner { pool: Pool, /// Node client client: Client, - /// Sender for sending built flashblock payloads to [`PayloadHandler`], - /// which broadcasts outgoing flashblock payloads via p2p. - built_fb_payload_tx: mpsc::Sender, - /// Sender for sending built full block payloads to [`PayloadHandler`], - /// which updates the engine tree state. - built_payload_tx: mpsc::Sender, - /// WebSocket publisher for broadcasting flashblocks - /// to all connected subscribers. - ws_pub: WebSocketPublisher, + /// Non-blocking publication bus for locally built flashblocks. + flashblock_tx: broadcast::Sender, /// System configuration for the builder config: BuilderConfig, /// The end of builder transaction type @@ -390,12 +386,8 @@ impl OpPayloadBuilderInner { &self.client } - pub(crate) fn ws_pub(&self) -> &WebSocketPublisher { - &self.ws_pub - } - - pub(crate) fn config(&self) -> &BuilderConfig { - &self.config + pub(crate) fn flashblock_tx(&self) -> &broadcast::Sender { + &self.flashblock_tx } pub(crate) fn metrics(&self) -> &OpRBuilderMetrics { @@ -442,9 +434,7 @@ where client: Client, config: BuilderConfig, builder_tx: BuilderTx, - built_fb_payload_tx: mpsc::Sender, - built_payload_tx: mpsc::Sender, - ws_pub: WebSocketPublisher, + flashblock_tx: broadcast::Sender, metrics: Arc, task_metrics: Arc, pool_change_epoch: Arc, @@ -474,9 +464,7 @@ where builder_ctx, pool, client, - built_fb_payload_tx, - built_payload_tx, - ws_pub, + flashblock_tx, config, builder_tx, task_metrics, @@ -672,43 +660,26 @@ where )); } + let payload_id = fb_payload.payload_id; best_payload_tx.send_replace(Some(payload.clone())); - self.notify_built_payload(payload); + fanout::emit( + &self.flashblock_tx, + self.metrics(), + FlashblockEvent { + fb_payload: Arc::new(fb_payload), + built: Arc::new(payload), + attributes_timestamp_secs: config.attributes.timestamp(), + ws_eligible: !ctx.attributes().no_tx_pool, + tx_trace_total_txs: info.executed_transactions.len(), + }, + ); info!( target: "payload_builder", - id = %fb_payload.payload_id, + id = %payload_id, "Fallback block built" ); - // not emitting flashblock if no_tx_pool in FCU, it's just syncing - if !ctx.attributes().no_tx_pool { - let flashblock_byte_size = self - .ws_pub - .publish(&fb_payload) - .map_err(PayloadBuilderError::other)?; - - let slot_offset_ms = - compute_slot_offset_ms(config.attributes.timestamp(), self.config.block_time); - record_flashblock_publish_timing(fb_payload.index, slot_offset_ms); - - if self.config.enable_tx_tracking_debug_logs { - debug!( - target: "tx_trace", - payload_id = %ctx.payload_id(), - block_number = ctx.block_number(), - flashblock_index = fb_payload.index, - byte_size = flashblock_byte_size, - total_txs = info.executed_transactions.len(), - slot_offset_ms, - stage = "fb_published" - ); - } - ctx.metrics - .flashblock_byte_size_histogram - .record(flashblock_byte_size as f64); - } - if ctx.attributes().no_tx_pool { info!( target: "payload_builder", @@ -736,7 +707,7 @@ where let target_flashblocks = flashblock_scheduler.target_flashblocks(); info!( target: "payload_builder", - id = %fb_payload.payload_id, + id = %payload_id, target_flashblocks, schedule = ?flashblock_scheduler, "Computed flashblock timing schedule" @@ -803,7 +774,7 @@ where tx, payload_cancel.clone(), fb_cancel, - fb_payload.payload_id, + payload_id, )), ); @@ -955,7 +926,6 @@ where .publish_flashblock_payload( &state.ctx, deps.best_payload_tx, - &state.fb_state, deps.payload_cancel, built_flashblock, ) @@ -1061,37 +1031,10 @@ where }) } - pub(crate) fn notify_built_payload(&self, payload: OpBuiltPayload) { - if let Err(e) = self.built_fb_payload_tx.try_send(payload.clone()) { - self.builder_ctx - .metrics - .built_fb_payload_send_failed - .increment(1); - warn!( - target: "payload_builder", - error = %e, - "Failed to send built flashblock payload to handler" - ); - } - - if let Err(e) = self.built_payload_tx.try_send(payload) { - self.builder_ctx - .metrics - .built_payload_send_failed - .increment(1); - warn!( - target: "payload_builder", - error = %e, - "Failed to send updated payload" - ); - } - } - fn publish_flashblock_payload( &self, ctx: &OpPayloadJobCtx, best_payload_tx: &watch::Sender>, - fb_state: &FlashblocksState, payload_cancel: &PayloadJobCancellation, built_flashblock: BuiltFlashblockOutput, ) -> eyre::Result> { @@ -1109,38 +1052,25 @@ where return Ok(None); } - // After this point, all side effects are synchronous. If cancellation wins the race after - // this check, still publish the local payload so getPayload can include this flashblock. - let flashblock_byte_size = self - .ws_pub - .publish(&fb_payload) - .wrap_err("failed to publish flashblock via websocket")?; + // The publish decision (watch update + bus enqueue) is synchronous after this gate: no + // await can suppress a gated-in event or enqueue a gated-out one. Delivery is async and + // ordered per subscriber; cancellation after enqueue is indistinguishable from latency. let flashblock_tx_count = fb_payload.raw_transactions().len(); best_payload_tx.send_replace(Some(new_payload.clone())); - self.notify_built_payload(new_payload); - - let slot_offset_ms = - compute_slot_offset_ms(ctx.attributes().timestamp(), self.config.block_time); - record_flashblock_publish_timing(fb_state.flashblock_index(), slot_offset_ms); - - if self.config.enable_tx_tracking_debug_logs { - debug!( - target: "tx_trace", - payload_id = %ctx.payload_id(), - block_number = ctx.block_number(), - flashblock_index = fb_state.flashblock_index(), - byte_size = flashblock_byte_size, - total_txs = flashblock_tx_count, - slot_offset_ms, - stage = "fb_published" - ); - } + fanout::emit( + &self.flashblock_tx, + self.metrics(), + FlashblockEvent { + fb_payload: Arc::new(fb_payload), + built: Arc::new(new_payload), + attributes_timestamp_secs: ctx.attributes().timestamp(), + ws_eligible: true, + tx_trace_total_txs: flashblock_tx_count, + }, + ); ctx.metrics.flashblock_build_duration.record(build_duration); - ctx.metrics - .flashblock_byte_size_histogram - .record(flashblock_byte_size as f64); ctx.metrics .flashblock_num_tx_histogram .record(flashblock_tx_count as f64); diff --git a/crates/op-rbuilder/src/builder/payload_handler.rs b/crates/op-rbuilder/src/builder/payload_handler.rs index 81cb5dd1..fe8d5700 100644 --- a/crates/op-rbuilder/src/builder/payload_handler.rs +++ b/crates/op-rbuilder/src/builder/payload_handler.rs @@ -35,19 +35,12 @@ use std::sync::Arc; use tokio::sync::mpsc; use tracing::{error, info, trace, warn}; -/// Handles newly built or received flashblock payloads. +/// Handles flashblock payloads received from peers. /// -/// In the case of a payload built by this node, it is broadcast to peers and an event is sent to the payload builder. -/// In the case of a payload received from a peer, it is executed and if successful, an event is sent to the payload builder. +/// A payload received from a peer is executed and, if successful, sent to the payload builder. pub(crate) struct PayloadHandler { - // receives new flashblock payloads built by this builder. - built_fb_payload_rx: mpsc::Receiver, - // receives new full block payloads built by this builder. - built_payload_rx: mpsc::Receiver, // receives incoming p2p messages from peers. p2p_rx: mpsc::Receiver, - // outgoing p2p channel to broadcast new payloads to peers. - p2p_tx: mpsc::Sender, // sends a `Events::BuiltPayload` to the reth payload builder when a new payload is received. payload_events_handle: tokio::sync::broadcast::Sender>, // context required for execution of blocks during syncing @@ -64,12 +57,8 @@ impl PayloadHandler where Client: ClientBounds + 'static, { - #[allow(clippy::too_many_arguments)] pub(crate) fn new( - built_fb_payload_rx: mpsc::Receiver, - built_payload_rx: mpsc::Receiver, p2p_rx: mpsc::Receiver, - p2p_tx: mpsc::Sender, payload_events_handle: tokio::sync::broadcast::Sender>, syncer_config: OpPayloadSyncerConfig, client: Client, @@ -77,10 +66,7 @@ where metrics: Arc, ) -> Self { Self { - built_fb_payload_rx, - built_payload_rx, p2p_rx, - p2p_tx, payload_events_handle, syncer_config, client, @@ -91,10 +77,7 @@ where pub(crate) async fn run(self) { let Self { - mut built_fb_payload_rx, - mut built_payload_rx, mut p2p_rx, - p2p_tx, payload_events_handle, syncer_config, client, @@ -106,20 +89,6 @@ where loop { tokio::select! { - Some(payload) = built_fb_payload_rx.recv() => { - // ignore error here; if p2p was disabled, the channel will be closed. - let _ = p2p_tx.send(payload.into()).await; - } - Some(payload) = built_payload_rx.recv() => { - // Update engine tree state with locally built block payloads - if let Err(e) = payload_events_handle.send(Events::BuiltPayload(payload.clone())) { - warn!( - target: "payload_builder", - error = %e, - "failed to send BuiltPayload event" - ); - } - } Some(message) = p2p_rx.recv() => { match message { Message::OpBuiltPayload(payload) => { diff --git a/crates/op-rbuilder/src/builder/service.rs b/crates/op-rbuilder/src/builder/service.rs index 4002edd4..2bb40138 100644 --- a/crates/op-rbuilder/src/builder/service.rs +++ b/crates/op-rbuilder/src/builder/service.rs @@ -3,6 +3,7 @@ use crate::{ builder::{ BuilderConfig, builder_tx::BuilderTransactions, + fanout, flashblocks_builder_tx::{FlashblocksBuilderTx, FlashblocksNumberBuilderTx}, generator::BlockPayloadJobGenerator, p2p::{AGENT_VERSION, FLASHBLOCKS_STREAM_PROTOCOL, Message}, @@ -116,6 +117,8 @@ impl FlashblocksServiceBuilder { let metrics = Arc::new(OpRBuilderMetrics::default()); let task_metrics = Arc::new(FlashblocksTaskMetrics::new()); let pool_change_epoch = Arc::new(AtomicU64::new(0)); + let block_time = self.0.block_time; + let tx_tracking_logs = self.0.enable_tx_tracking_debug_logs; if flashblocks_config.continuous_build { let mut pending_txs = @@ -129,9 +132,9 @@ impl FlashblocksServiceBuilder { }); } - // Channels for built flashblock payloads - let (built_fb_payload_tx, built_fb_payload_rx) = tokio::sync::mpsc::channel(16); - let (built_payload_tx, built_payload_rx) = tokio::sync::mpsc::channel(16); + let (flashblock_tx, ws_receiver) = fanout::channel(); + let p2p_receiver = flashblock_tx.subscribe(); + let engine_feedback_receiver = flashblock_tx.subscribe(); let ws_pub = WebSocketPublisher::new( flashblocks_config.ws_addr, @@ -146,9 +149,7 @@ impl FlashblocksServiceBuilder { ctx.provider().clone(), self.0.clone(), builder_tx, - built_fb_payload_tx, - built_payload_tx, - ws_pub, + flashblock_tx, metrics.clone(), task_metrics.clone(), pool_change_epoch, @@ -170,16 +171,47 @@ impl FlashblocksServiceBuilder { OpPayloadSyncerConfig::new(self.0, OpEvmConfig::optimism(ctx.chain_spec())) .wrap_err("failed to create flashblocks payload builder context")?; + let payload_events_handle = payload_service.payload_events_handle(); let payload_handler = PayloadHandler::new( - built_fb_payload_rx, - built_payload_rx, incoming_message_rx, - outgoing_message_tx, - payload_service.payload_events_handle(), + payload_events_handle.clone(), syncer_config, ctx.provider().clone(), ctx.task_executor().clone(), - metrics, + metrics.clone(), + ); + + ctx.task_executor().spawn_critical_task( + "flashblocks websocket subscriber", + task_metrics + .websocket_subscriber + .instrument(fanout::websocket_subscriber( + ws_receiver, + ws_pub, + metrics.clone(), + block_time, + tx_tracking_logs, + )), + ); + ctx.task_executor().spawn_critical_task( + "flashblocks p2p subscriber", + task_metrics + .p2p_subscriber + .instrument(fanout::p2p_subscriber( + p2p_receiver, + outgoing_message_tx, + metrics.clone(), + )), + ); + ctx.task_executor().spawn_critical_task( + "flashblocks engine feedback subscriber", + task_metrics + .engine_feedback_subscriber + .instrument(fanout::engine_feedback_subscriber( + engine_feedback_receiver, + payload_events_handle, + metrics, + )), ); ctx.task_executor().spawn_critical_task( diff --git a/crates/op-rbuilder/src/builder/timing.rs b/crates/op-rbuilder/src/builder/timing.rs index 851b6830..d2ff7c44 100644 --- a/crates/op-rbuilder/src/builder/timing.rs +++ b/crates/op-rbuilder/src/builder/timing.rs @@ -255,7 +255,8 @@ impl std::fmt::Debug for FlashblockScheduler { /// Compute the elapsed time in ms since the start of the slot. /// Slot start defined as `payload_timestamp - block_time`. pub(super) fn compute_slot_offset_ms(payload_timestamp: u64, block_time: Duration) -> f64 { - let slot_start = SystemTime::UNIX_EPOCH + Duration::from_secs(payload_timestamp) - block_time; + let slot_start = + SystemTime::UNIX_EPOCH + Duration::from_secs(payload_timestamp).saturating_sub(block_time); SystemTime::now() .duration_since(slot_start) .unwrap_or_default() @@ -267,6 +268,22 @@ pub(super) fn compute_slot_offset_ms(payload_timestamp: u64, block_time: Duratio mod tests { use super::*; + #[test] + fn slot_offset_saturates_before_unix_epoch() { + let before = SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_secs_f64() + * 1000.0; + let offset_ms = compute_slot_offset_ms(1, Duration::from_secs(2)); + let after = SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_secs_f64() + * 1000.0; + assert!((before..=after).contains(&offset_ms)); + } + struct ComputeSendTimesTestCase { first_flashblock_offset_ms: u64, deadline_ms: u64, diff --git a/crates/op-rbuilder/src/metrics.rs b/crates/op-rbuilder/src/metrics.rs index 18f359ae..1ca75e70 100644 --- a/crates/op-rbuilder/src/metrics.rs +++ b/crates/op-rbuilder/src/metrics.rs @@ -211,13 +211,15 @@ pub struct OpRBuilderMetrics { pub payload_job_cancellation_complete: Counter, /// Payload job ended due to a build error pub payload_job_cancellation_error: Counter, - /// Built flashblock payload dropped because the channel to `PayloadHandler`'s - /// p2p fan-out was full or closed. Nonzero means peers missed a flashblock. - pub built_fb_payload_send_failed: Counter, - /// Built payload dropped because the channel to `PayloadHandler`'s engine-tree - /// feedback was full or closed. Nonzero means the engine tree missed a - /// pre-cached payload and had to re-validate our own block the expensive way. - pub built_payload_send_failed: Counter, + /// Flashblock events skipped by the WebSocket subscriber after it lagged. + pub ws_publish_lagged: Counter, + /// Flashblock events skipped by the p2p subscriber after it lagged. + pub p2p_forward_lagged: Counter, + /// Flashblock events skipped by the engine feedback subscriber after it lagged. + pub engine_feedback_lagged: Counter, + /// Flashblock bus sends with no active receivers. + /// Nonzero means total fanout loss. + pub fanout_no_subscribers: Counter, } impl OpRBuilderMetrics { diff --git a/crates/op-rbuilder/src/tokio_metrics.rs b/crates/op-rbuilder/src/tokio_metrics.rs index 81d4d449..e67c3218 100644 --- a/crates/op-rbuilder/src/tokio_metrics.rs +++ b/crates/op-rbuilder/src/tokio_metrics.rs @@ -199,6 +199,12 @@ pub struct FlashblocksTaskMetrics { pub payload_handler: MonitoredTask, /// Monitor for the websocket listener task pub websocket_publisher: MonitoredTask, + /// Monitor for the websocket publication subscriber + pub websocket_subscriber: MonitoredTask, + /// Monitor for the p2p forwarding subscriber + pub p2p_subscriber: MonitoredTask, + /// Monitor for the engine feedback subscriber + pub engine_feedback_subscriber: MonitoredTask, /// Global runtime metrics recorder runtime_recorder: TokioRuntimeMetricsRecorder, } @@ -216,6 +222,9 @@ impl FlashblocksTaskMetrics { payload_builder_service: MonitoredTask::new("payload_builder_service"), payload_handler: MonitoredTask::new("payload_handler"), websocket_publisher: MonitoredTask::new("websocket_publisher"), + websocket_subscriber: MonitoredTask::new("websocket_subscriber"), + p2p_subscriber: MonitoredTask::new("p2p_subscriber"), + engine_feedback_subscriber: MonitoredTask::new("engine_feedback_subscriber"), runtime_recorder: TokioRuntimeMetricsRecorder::default(), } } @@ -237,6 +246,11 @@ impl FlashblocksTaskMetrics { let mut payload_builder_intervals = metrics.payload_builder_service.monitor.intervals(); let mut payload_handler_intervals = metrics.payload_handler.monitor.intervals(); let mut websocket_publisher_intervals = metrics.websocket_publisher.monitor.intervals(); + let mut websocket_subscriber_intervals = + metrics.websocket_subscriber.monitor.intervals(); + let mut p2p_subscriber_intervals = metrics.p2p_subscriber.monitor.intervals(); + let mut engine_feedback_subscriber_intervals = + metrics.engine_feedback_subscriber.monitor.intervals(); loop { timer.tick().await; @@ -261,6 +275,17 @@ impl FlashblocksTaskMetrics { if let Some(task_metrics) = websocket_publisher_intervals.next() { metrics.websocket_publisher.record_metrics(&task_metrics); } + if let Some(task_metrics) = websocket_subscriber_intervals.next() { + metrics.websocket_subscriber.record_metrics(&task_metrics); + } + if let Some(task_metrics) = p2p_subscriber_intervals.next() { + metrics.p2p_subscriber.record_metrics(&task_metrics); + } + if let Some(task_metrics) = engine_feedback_subscriber_intervals.next() { + metrics + .engine_feedback_subscriber + .record_metrics(&task_metrics); + } } }); } @@ -276,6 +301,15 @@ impl fmt::Debug for FlashblocksTaskMetrics { ) .field("payload_handler", &self.payload_handler.task_name()) .field("websocket_publisher", &self.websocket_publisher.task_name()) + .field( + "websocket_subscriber", + &self.websocket_subscriber.task_name(), + ) + .field("p2p_subscriber", &self.p2p_subscriber.task_name()) + .field( + "engine_feedback_subscriber", + &self.engine_feedback_subscriber.task_name(), + ) .field("runtime_monitor", &"enabled") .finish() }