feat(flashblocks): shift atomic gossip op, wspub on seq after p2p gossip

[sieniven]

Summary

Fixes the websocket publishing logic on the flashblocks builder into the p2p node service instead. This ensures zero re-org on subscribed flashblocks RPC nodes by ensuring p2p gossip to follower sequencers is a blocking successful operation first before gossiping the flashblocks to RPC nodes.

This allows for atomicity on flashblocks replay during sequencer switches / failures of the leader sequencer, to replay consistent gossiped flashblocks across RPC nodes. This ensures flashblocks re-org is guaranteed protected on RPC nodes.

[sieniven]

@claude review this PR

[github-actions]

Claude finished @sieniven's task in 3m 31s —— View job

---

PR Review: feat(flashblocks): shift atomic gossip op, wspub on seq after p2p gossip

• Gather context and understand the PR
• Read changed files
• Analyze logic correctness and atomicity guarantees
• Check for bugs, edge cases, and error handling
• Provide review feedback

---

Summary

Good architectural direction — removing ws_pub from the builder and consolidating publish responsibility into the p2p Node simplifies the builder and keeps WS publishing coordinated with p2p gossip. The Message trait → concrete Message enum refactor also reduces unnecessary generics. However, there are several issues worth addressing:

---

🔴 Critical: WS publishes before p2p gossip in Node::run() — opposite of the stated guarantee

crates/builder/src/p2p/mod.rs, lines ~157–162:

if let Message::OpFlashblockPayload(ref fb_payload) = message {
    let flashblock_byte_size = ws_pub.publish(fb_payload)?;   // ← WS first
    metrics.flashblock_byte_size_histogram.record(flashblock_byte_size as f64);
}
if let Err(e) = outgoing_streams_handler.broadcast_message(message).await {  // ← p2p second
    warn!(target: "flashblocks-p2p", "...");
}

The PR description states: "p2p gossip to follower sequencers is a blocking successful operation first before gossiping the flashblocks to RPC nodes."

But the code publishes to WebSocket subscribers before broadcast_message() to peers. This means the leader's own WS subscribers (RPC nodes) receive the flashblock before follower sequencers do, which is the exact re-org risk the PR claims to fix. The broadcast_message() call should come first, and ws_pub.publish() only after it returns (or at least returns without error).

Fix this →

---

🟡 Medium: ws_pub.publish()? kills the p2p node if broadcast channel has no receivers

crates/builder/src/p2p/mod.rs:

let flashblock_byte_size = ws_pub.publish(fb_payload)?;

WebSocketPublisher::publish() calls self.pipe.send(utf8_bytes) (a broadcast::Sender), which returns Err when there are no active receivers. The initial receiver created in WebSocketPublisher::new() is held by the listener_loop task, so in normal operation this is unlikely to fire. However, if the listener_loop task ever panics or exits unexpectedly, subsequent calls to ws_pub.publish() will kill the p2p node entirely via ?.

The previous code in builder.rs handled this with map_err(PayloadBuilderError::other)?, which only terminated the current build job (restartable), not the p2p node service. The handler's follower path (in handler.rs) correctly uses if let Err(e) = ws_pub.publish(...) { warn!(...) }. The Node path should follow the same pattern.

Fix this →

---

🟡 Medium: flashblock_byte_size_histogram metric silently drops when p2p is disabled

crates/builder/src/flashblocks/builder.rs (removed) vs crates/builder/src/p2p/mod.rs (added):

Previously, ctx.metrics.flashblock_byte_size_histogram.record(...) was called directly in the builder — always, regardless of p2p state. Now it's only recorded inside Node::run(), which only runs when p2p_enabled = true. Deployments with p2p_enabled = false will silently lose this metric after this change.

---

🟢 Good: behavior fix for no_tx_pool path

crates/builder/src/flashblocks/builder.rs:

Previously, built_fb_payload_tx.try_send(fb_payload) for the initial flashblock was guarded only by !rebuild_external_payload, meaning syncing FCUs (where no_tx_pool=true) would still trigger p2p gossip. The new code correctly gates it with !no_tx_pool && !rebuild_external_payload, consistent with the comment "not emitting flashblock if no_tx_pool in FCU, it's just syncing". This is a correct fix.

---

🟢 Good: Message trait → enum refactor

Removing the Message generic trait in favor of a concrete Message enum in types.rs is the right call for an internal protocol — reduces boilerplate, removes the generic parameter from Node<M>, and makes deserialization straightforward (serde_json::from_str::<Message>). The types.rs module is well-organized.

---

Minor: Default removed for NodeBuilder

Since NodeBuilder::new() now requires ws_pub and metrics, removing Default is correct. Worth verifying no external call sites use NodeBuilder::default() (doesn't appear to be any in the diff, but worth a quick grep if not already done).

---

Test coverage note

The test in p2p/mod.rs now uses OpFlashblockPayload::default() as the round-trip message and binds WebSocketPublisher to 127.0.0.1:0. Consider adding a test case that verifies the WS publish is called (or not) at the right point relative to p2p gossip, once the ordering fix is applied.

---

TL;DR: The main concern is the WS-before-p2p ordering in Node::run(), which inverts the core atomicity guarantee the PR is trying to establish. Suggest swapping ws_pub.publish() to run after broadcast_message(), and converting the ? to a warn! to avoid killing the p2p node on publish failure.