refactor: extract post-seal side-effects into PostSealHook trait

Move WS publication, p2p forwarding, engine propagation, and per-flashblock
metrics out of the building loop and behind a PostSealHook trait. Hooks
are constructed once in OpPayloadBuilder::new() and dispatched after each
sealed candidate (fallback + every flashblock).

build_next_flashblock no longer performs side-effects: it returns a
FlashblockBuildResult struct and the async loop handles dispatch. The
side-effect call sites in the fallback path and the FB iteration are
collapsed to a single dispatch_post_seal call each.

p2p and engine forwarders are unified as ChannelHook(name, sender) since
they only differ in the failure log label.

Author: julio4

crates/op-rbuilder/src/builder/hooks/channel.rs

@@ -0,0 +1,34 @@
+use crate::builder::hooks::post_seal::{PostSealHook, SealedCandidate, SealedCtx};
+use reth_optimism_node::OpBuiltPayload;
+use tokio::sync::mpsc;
+use tracing::warn;
+
+/// Forwards each sealed candidate over a named mpsc channel.
+#[derive(Debug)]
+pub(in crate::builder) struct ChannelHook {
+    name: &'static str,
+    sender: mpsc::Sender<OpBuiltPayload>,
+}
+
+impl ChannelHook {
+    pub(in crate::builder) fn new(
+        name: &'static str,
+        sender: mpsc::Sender<OpBuiltPayload>,
+    ) -> Self {
+        Self { name, sender }
+    }
+}
+
+impl PostSealHook for ChannelHook {
+    fn on_sealed(&self, candidate: &SealedCandidate, ctx: &SealedCtx) {
+        if let Err(e) = self.sender.try_send(candidate.payload.clone()) {
+            warn!(
+                target: "payload_builder",
+                channel = self.name,
+                error = %e,
+                flashblock_index = ctx.flashblock_index,
+                "Failed to forward sealed payload over channel"
+            );
+        }
+    }
+}

crates/op-rbuilder/src/builder/hooks/metrics.rs

@@ -0,0 +1,29 @@
+use crate::{
+    builder::hooks::post_seal::{PostSealHook, SealedCandidate, SealedCtx},
+    metrics::OpRBuilderMetrics,
+};
+use std::sync::Arc;
+
+/// Records per-flashblock metrics that aren't tied to publication:
+/// build duration and transaction-count histogram.
+#[derive(Debug)]
+pub(in crate::builder) struct MetricsHook {
+    metrics: Arc<OpRBuilderMetrics>,
+}
+
+impl MetricsHook {
+    pub(in crate::builder) fn new(metrics: Arc<OpRBuilderMetrics>) -> Self {
+        Self { metrics }
+    }
+}
+
+impl PostSealHook for MetricsHook {
+    fn on_sealed(&self, _candidate: &SealedCandidate, ctx: &SealedCtx) {
+        if let Some(duration) = ctx.flashblock_build_duration {
+            self.metrics.flashblock_build_duration.record(duration);
+        }
+        self.metrics
+            .flashblock_num_tx_histogram
+            .record(ctx.executed_tx_count as f64);
+    }
+}

crates/op-rbuilder/src/builder/hooks/mod.rs

@@ -0,0 +1,30 @@
+//! Lifecycle hooks dispatched from the building loop.
+//!
+//! [`PostSealHook`] fires after each sealed candidate (fallback or
+//! flashblock) and isolates downstream side-effects — WS publication, p2p
+//! broadcast, engine propagation, metrics — from the building loop itself.
+//! Concrete implementations live alongside this module.
+
+mod channel;
+mod metrics;
+mod post_seal;
+mod ws;
+
+pub(super) use channel::ChannelHook;
+pub(super) use metrics::MetricsHook;
+pub(super) use post_seal::{PostSealHook, SealedCandidate, SealedCtx};
+pub(super) use ws::WsHook;
+
+/// Dispatch a sealed candidate to every hook in `hooks`.
+///
+/// Hook impls are expected to be cheap; we run them sequentially in the
+/// caller's context (sync, including from within `spawn_blocking`).
+pub(super) fn dispatch_post_seal(
+    hooks: &[Box<dyn PostSealHook>],
+    candidate: &SealedCandidate,
+    ctx: &SealedCtx,
+) {
+    for h in hooks {
+        h.on_sealed(candidate, ctx);
+    }
+}

crates/op-rbuilder/src/builder/hooks/post_seal.rs

@@ -0,0 +1,46 @@
+use core::time::Duration;
+use op_alloy_rpc_types_engine::OpFlashblockPayload;
+use reth_optimism_node::OpBuiltPayload;
+use reth_payload_builder::PayloadId;
+
+/// A flashblock candidate that has been sealed and is ready for publication
+/// and downstream propagation.
+///
+/// `payload` is the full built payload for engine/p2p delivery; `fb_payload`
+/// is the slim, serialisable view streamed to flashblocks subscribers.
+#[derive(Debug, Clone)]
+pub(in crate::builder) struct SealedCandidate {
+    pub payload: OpBuiltPayload,
+    pub fb_payload: OpFlashblockPayload,
+}
+
+/// Context describing the slot a sealed candidate belongs to.
+///
+/// The fields are intentionally limited to data downstream hooks need.
+#[derive(Debug, Clone)]
+pub(in crate::builder) struct SealedCtx {
+    pub payload_id: PayloadId,
+    pub block_number: u64,
+    pub flashblock_index: u64,
+    /// True when the FCU specified `no_tx_pool`.
+    pub no_tx_pool: bool,
+    pub executed_tx_count: usize,
+    /// Slot start timestamp from the payload attributes.
+    pub slot_timestamp_secs: u64,
+    pub block_time: Duration,
+    /// Wall-clock time spent building this flashblock.
+    /// `None` for the fallback candidate.
+    pub flashblock_build_duration: Option<Duration>,
+    pub enable_tx_tracking_debug_logs: bool,
+}
+
+/// Hook invoked after a flashblock or fallback candidate has been sealed.
+///
+/// Implementations should be cheap and non-blocking: dispatch happens on the
+/// builder's hot path. Errors are intentionally swallowed at the dispatch site;
+/// hooks that want to surface failures should do so via metrics or logs.
+pub(in crate::builder) trait PostSealHook:
+    Send + Sync + std::fmt::Debug
+{
+    fn on_sealed(&self, candidate: &SealedCandidate, ctx: &SealedCtx);
+}

crates/op-rbuilder/src/builder/hooks/ws.rs

@@ -0,0 +1,73 @@
+use crate::{
+    builder::{
+        hooks::post_seal::{PostSealHook, SealedCandidate, SealedCtx},
+        timing::compute_slot_offset_ms,
+        wspub::WebSocketPublisher,
+    },
+    metrics::{OpRBuilderMetrics, record_flashblock_publish_timing},
+};
+use std::sync::Arc;
+use tracing::{debug, warn};
+
+/// Publishes the flashblock payload to WebSocket subscribers, record metrics.
+///
+/// Suppressed when `SealedCtx::no_tx_pool` is true
+pub(in crate::builder) struct WsHook {
+    ws_pub: Arc<WebSocketPublisher>,
+    metrics: Arc<OpRBuilderMetrics>,
+}
+
+impl std::fmt::Debug for WsHook {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("WsHook").finish_non_exhaustive()
+    }
+}
+
+impl WsHook {
+    pub(in crate::builder) fn new(
+        ws_pub: Arc<WebSocketPublisher>,
+        metrics: Arc<OpRBuilderMetrics>,
+    ) -> Self {
+        Self { ws_pub, metrics }
+    }
+}
+
+impl PostSealHook for WsHook {
+    fn on_sealed(&self, candidate: &SealedCandidate, ctx: &SealedCtx) {
+        if ctx.no_tx_pool {
+            return;
+        }
+
+        let byte_size = match self.ws_pub.publish(&candidate.fb_payload) {
+            Ok(size) => size,
+            Err(e) => {
+                warn!(
+                    target: "payload_builder",
+                    error = %e,
+                    flashblock_index = ctx.flashblock_index,
+                    "Failed to publish flashblock via websocket"
+                );
+                return;
+            }
+        };
+
+        let slot_offset_ms = compute_slot_offset_ms(ctx.slot_timestamp_secs, ctx.block_time);
+        record_flashblock_publish_timing(candidate.fb_payload.index, slot_offset_ms);
+        self.metrics
+            .flashblock_byte_size_histogram
+            .record(byte_size as f64);
+
+        if ctx.enable_tx_tracking_debug_logs {
+            debug!(
+                target: "tx_trace",
+                payload_id = %ctx.payload_id,
+                block_number = ctx.block_number,
+                flashblock_index = candidate.fb_payload.index,
+                byte_size,
+                total_txs = ctx.executed_tx_count,
+                slot_offset_ms,
+                stage = "fb_published"
+            );
+        }
+    }
+}

crates/op-rbuilder/src/builder/mod.rs

@@ -17,6 +17,7 @@ mod config;
 mod context;
 mod flashblocks_builder_tx;
 mod generator;
+mod hooks;
 mod p2p;
 mod payload;
 mod payload_handler;