refactor: fan out flashblock publication#559
Merged
Merged
Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
This PR refactors flashblock publication so post-build side effects (websocket publish, p2p forwarding, engine feedback) are decoupled from the payload build path and handled by dedicated subscriber tasks consuming a fanout broadcast bus. This isolates transient publication failures (e.g., websocket disconnects) from aborting payload construction and replaces “send failed” counters with per-subscriber lag/drop counters.
Changes:
- Introduces a broadcast-based fanout bus (
fanout.rs) and subscriber tasks for websocket publish, p2p forward, and engine feedback. - Updates the builder/service wiring so built payloads are emitted once onto the bus;
PayloadHandlernow only processes peer-received payloads. - Improves slot-offset timing robustness (saturating pre-UNIX-epoch case) and adds a regression test; updates task metrics + lag counters.
Reviewed changes
Copilot reviewed 9 out of 9 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| crates/op-rbuilder/src/tokio_metrics.rs | Adds task monitors/recording for the new subscriber tasks. |
| crates/op-rbuilder/src/metrics.rs | Replaces built-payload send-failure counters with per-subscriber lag/drop counters. |
| crates/op-rbuilder/src/builder/timing.rs | Uses saturating subtraction for slot start calculation; adds test. |
| crates/op-rbuilder/src/builder/service.rs | Wires up the fanout bus and spawns websocket/p2p/engine subscriber tasks. |
| crates/op-rbuilder/src/builder/payload.rs | Emits FlashblockEvents onto the fanout bus instead of doing side effects inline. |
| crates/op-rbuilder/src/builder/payload_handler.rs | Narrows responsibility to handling peer-received payloads only. |
| crates/op-rbuilder/src/builder/mod.rs | Registers the new fanout module. |
| crates/op-rbuilder/src/builder/fanout.rs | New fanout bus + subscribers with lag handling and targeted tests. |
| crates/op-rbuilder/src/builder/continuous/publish.rs | Routes continuous candidate publication through the fanout bus. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
julio4
force-pushed
the
refactor/flashblock-publish-bus
branch
from
July 20, 2026 04:28
77b0752 to
387c79a
Compare
julio4
force-pushed
the
refactor/flashblock-publish-bus
branch
from
July 20, 2026 04:36
387c79a to
2fb7355
Compare
avalonche
reviewed
Jul 21, 2026
avalonche
approved these changes
Jul 21, 2026
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 11 out of 11 changed files in this pull request and generated 1 comment.
Comments suppressed due to low confidence (1)
crates/op-rbuilder/src/metrics.rs:222
- Same as above: these counters also apply to fallback/other built-payload events on the fanout bus, not only flashblocks. Also, the double-space in the doc comment is probably unintentional.
/// 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,
julio4
force-pushed
the
refactor/flashblock-publish-bus
branch
from
July 23, 2026 07:30
5fa461b to
3677966
Compare
julio4
force-pushed
the
refactor/flashblock-publish-bus
branch
from
July 23, 2026 07:46
3677966 to
88e8b0b
Compare
julio4
enabled auto-merge (squash)
July 23, 2026 07:46
| use tokio::sync::{broadcast, mpsc}; | ||
| use tracing::{debug, warn}; | ||
|
|
||
| /// ~6 slots of events at 200ms flashblock |
julio4
force-pushed
the
refactor/flashblock-publish-bus
branch
from
July 23, 2026 08:06
88e8b0b to
32741fa
Compare
Comment on lines
+273
to
+285
| 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)); | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Publication side effects (ws publish, p2p forward, engine feedback) now decoupled from the build path.
Every built payload (fallback, flashblock, continuous candidate) is sent once on a broadcast bus (
fanout.rs), consumed by dedicated subscriber tasks with fanout guarantees.PayloadHandlernow only handles payloads received from peers.Changes:
built_*_send_failedcounters from feat(observability): count dropped built-payload channel sends #555 are replaced by per-subscriberws_publish_lagged/p2p_forward_lagged/engine_feedback_laggedcounters.Supersede / Close #518