diff --git a/bin/ethlambda/src/main.rs b/bin/ethlambda/src/main.rs index db3f62368..09ac8767b 100644 --- a/bin/ethlambda/src/main.rs +++ b/bin/ethlambda/src/main.rs @@ -791,6 +791,7 @@ async fn run_node(options: Options) -> eyre::Result<()> { payload_attestation_pool.clone(), builder_market.clone(), sync_committee_pool.clone(), + Some(Arc::new(events.clone())), ) .await .wrap_err("failed to start discv5 discovery")?; diff --git a/crates/blockchain/src/beacon_envelope.rs b/crates/blockchain/src/beacon_envelope.rs index a893a651c..b0a993895 100644 --- a/crates/blockchain/src/beacon_envelope.rs +++ b/crates/blockchain/src/beacon_envelope.rs @@ -574,6 +574,14 @@ impl BlockChainServer { held.envelope.message.payload.block_hash, ); self.envelopes.verified.push_back(root); + let message = &held.envelope.message; + self.events.publish_envelope_imported( + &self.store, + slot, + message.builder_index, + message.payload.block_hash, + root, + ); true } @@ -2329,6 +2337,60 @@ mod tests { envelope.message.payload.block_hash } + /// The actor's import of an envelope is what publishes `execution_payload`, + /// and `execution_payload_available` right after it. + #[tokio::test] + #[cfg_attr( + not(feature = "beacon-spec-tests"), + ignore = "needs the fork choice fixtures; run `make consensus-spec-tests`" + )] + async fn an_imported_envelope_emits_execution_payload_then_available() { + let case = Case::new("on_execution_payload_envelope", FULL_CHILD_CASE); + let mut server = case.server(12); + server + .on_block(case.block(FULL_PARENT), ImportTimings::default()) + .await; + let mut rx = server.events.subscribe(); + let envelope = valid_envelope_for_full_parent(); + let builder_index = envelope.message.builder_index; + let block_hash = payload_hash(&envelope); + + server.receive_envelope(envelope, 0).await; + + let payload_events: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()) + .filter(|event| { + matches!( + event, + crate::ChainEvent::ExecutionPayload(_) + | crate::ChainEvent::ExecutionPayloadAvailable(_) + ) + }) + .collect(); + let slot = server + .store + .block_slot_and_state_root(&root(FULL_PARENT)) + .expect("the parent is imported") + .0; + match payload_events.as_slice() { + [ + crate::ChainEvent::ExecutionPayload(payload), + crate::ChainEvent::ExecutionPayloadAvailable(available), + ] => { + assert_eq!( + (payload.slot, payload.builder_index, payload.block_hash), + (slot, builder_index, block_hash) + ); + assert_eq!(payload.block_root, root(FULL_PARENT)); + assert!(!payload.execution_optimistic); + assert_eq!( + (available.slot, available.block_root), + (slot, root(FULL_PARENT)) + ); + } + other => panic!("expected execution_payload then available, got {other:?}"), + } + } + #[tokio::test] #[cfg_attr( not(feature = "beacon-spec-tests"), diff --git a/crates/blockchain/src/events.rs b/crates/blockchain/src/events.rs index 1e82f89e8..d42fbe5bc 100644 --- a/crates/blockchain/src/events.rs +++ b/crates/blockchain/src/events.rs @@ -1,9 +1,19 @@ //! Chain-event pub-sub bus. //! -//! The [`crate::BlockChainServer`] actor is the **sole publisher**: it emits a +//! The [`crate::BlockChainServer`] actor is the **main publisher**: it emits a //! [`ChainEvent`] whenever consensus state changes (block import, head move, -//! justification, finalization). Consumers subscribe read-only receivers and -//! never write back into the actor, keeping the write flow one-directional. +//! justification, finalization, an envelope imported). Consumers subscribe +//! read-only receivers and never write back into the actor, keeping the write +//! flow one-directional. +//! +//! Two other places publish the topics that name an object *accepted*, not +//! one that changed consensus state, since those objects never reach the +//! actor. The Beacon API handlers publish what a validator client or builder +//! submits (they hold the [`EventBus`] directly), and `ethlambda-p2p` +//! publishes gloas bids, proposer preferences, payload attestation messages +//! and accepted envelopes off gossip, through the [`GossipEventSink`] that +//! [`EventBus`] implements: p2p does not depend on this crate, so the sink +//! trait lives in `ethlambda-network-api`. //! //! The bus is intentionally best-effort: emission never blocks the actor, and //! a slow subscriber loses events (the bounded broadcast channel overwrites @@ -16,7 +26,7 @@ //! surface accepts its own set of [`Topic`] names ([`Topic::LEAN`], //! [`Topic::BEACON`]). -use ethlambda_network_api::BlockAnnouncement; +use ethlambda_network_api::{BlockAnnouncement, GossipEvent, GossipEventSink}; use ethlambda_state_transition::beacon::helpers::accessors::get_block_root_at_slot; use ethlambda_storage::{Chain, Store}; use ethlambda_types::ShortRoot; @@ -25,6 +35,8 @@ use ethlambda_types::beacon::containers::{ BeaconState, DataColumnSidecar, SignedAggregateAndProof, SignedBeaconBlock, electra, gloas, phase0, }; +use ethlambda_types::beacon::fork::ForkName; +use ethlambda_types::beacon::fork_choice::PayloadStatus; use ethlambda_types::beacon::serde_helpers::quoted_or_bare; use ethlambda_types::beacon::signing::{compute_epoch_at_slot, compute_start_slot_at_epoch}; use ethlambda_types::checkpoint::Checkpoint; @@ -43,10 +55,11 @@ use tracing::warn; /// One enum for both surfaces, so a name means the same thing on each: every /// Beacon API eventstream topic, plus the two ethlambda extensions lean serves /// (`justified_checkpoint`, `aggregate`). [`FromStr`] parses all of them; which -/// ones a surface *accepts* is [`Topic::LEAN`] or [`Topic::BEACON`]. Most beacon -/// names are accepted and never emitted: the node has no producer for them (no -/// operation pools, no light-client server, no gloas), and the specification -/// lists them all as valid subscriptions. +/// ones a surface *accepts* is [`Topic::LEAN`] or [`Topic::BEACON`]. Some beacon +/// names are accepted and never emitted: the node has no producer for them +/// (no light-client server, no payload-attributes stream, no fast +/// confirmation), and the specification lists them all as valid +/// subscriptions. `docs/rpc.md` has the table. #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] pub enum Topic { Head, @@ -66,7 +79,7 @@ pub enum Topic { /// The new head does not descend from the previous one. On both surfaces. ChainReorg, DataColumnSidecar, - // Accepted on the beacon surface, never emitted. See `docs/rpc.md`. + // Accepted on the beacon surface; `docs/rpc.md` lists which are emitted. SingleAttestation, VoluntaryExit, BlsToExecutionChange, @@ -309,6 +322,24 @@ pub enum ChainEvent { BeaconAttestation(Box), /// Beacon `data_column_sidecar`: a column this node custodies was stored. DataColumnSidecar(DataColumnSidecarEvent), + /// Beacon `head_v2`: `head` with the head's payload status, version-wrapped. + BeaconHeadV2(Versioned), + /// Beacon `execution_payload`: an envelope was imported. + ExecutionPayload(ExecutionPayloadEvent), + /// Beacon `execution_payload_gossip`: an envelope passed gossip validation + /// or was accepted through the API. + ExecutionPayloadGossip(ExecutionPayloadGossipEvent), + /// Beacon `execution_payload_available`: the payload and its blobs are + /// verified available. + ExecutionPayloadAvailable(ExecutionPayloadAvailableEvent), + /// Beacon `execution_payload_bid`. Boxed with the other payloads that + /// carry KZG commitments, for the reason [`ChainEvent::BeaconAttestation`] + /// is. + ExecutionPayloadBid(Box>), + /// Beacon `payload_attestation_message`. + PayloadAttestationMessage(Versioned), + /// Beacon `proposer_preferences`. + ProposerPreferences(Box>), } impl ChainEvent { @@ -325,11 +356,43 @@ impl ChainEvent { ChainEvent::Aggregate { .. } => Topic::Aggregate, ChainEvent::ChainReorg { .. } | ChainEvent::BeaconChainReorg(_) => Topic::ChainReorg, ChainEvent::DataColumnSidecar(_) => Topic::DataColumnSidecar, + ChainEvent::BeaconHeadV2(_) => Topic::HeadV2, + ChainEvent::ExecutionPayload(_) => Topic::ExecutionPayload, + ChainEvent::ExecutionPayloadGossip(_) => Topic::ExecutionPayloadGossip, + ChainEvent::ExecutionPayloadAvailable(_) => Topic::ExecutionPayloadAvailable, + ChainEvent::ExecutionPayloadBid(_) => Topic::ExecutionPayloadBid, + ChainEvent::PayloadAttestationMessage(_) => Topic::PayloadAttestationMessage, + ChainEvent::ProposerPreferences(_) => Topic::ProposerPreferences, } } } -/// The Beacon API `head` payload. +/// A payload in the `{"version", "data"}` wrapper the Beacon API gives the +/// event topics whose shape depends on the fork. +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +pub struct Versioned { + /// The lowercase fork name, [`ForkName::as_str`]. + pub version: &'static str, + pub data: T, +} + +impl Versioned { + /// `data` tagged with `fork`. + pub fn new(fork: ForkName, data: T) -> Self { + Self { + version: fork.as_str(), + data, + } + } + + /// `data` tagged gloas, for the containers that exist from that fork on. + pub fn gloas(data: T) -> Self { + Self::new(ForkName::Gloas, data) + } +} + +/// The Beacon API `head` payload (deprecated by the specification in favour of +/// [`BeaconHeadV2Event`], and still emitted). /// /// The two dependent roots are `get_block_root_at_slot(state, /// compute_start_slot_at_epoch(epoch) - 1)` (current) and the same one epoch @@ -350,13 +413,94 @@ pub struct BeaconHeadEvent { pub execution_optimistic: bool, } +/// The Beacon API `head_v2` payload, inside [`Versioned`] by the head +/// state's fork. +/// +/// `head`'s two dependent roots under the specification's new names: +/// `current_epoch_dependent_root` is the root at +/// `compute_start_slot_at_epoch(epoch - 1) - 1` and `next_epoch_dependent_root` +/// the root at `compute_start_slot_at_epoch(epoch) - 1`, `epoch` being the +/// head's own, so they are `head`'s previous and current duty dependent roots +/// respectively. Both are the genesis block root on underflow. +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +pub struct BeaconHeadV2Event { + #[serde(serialize_with = "quoted_or_bare::serialize")] + pub slot: u64, + pub block: H256, + pub state: H256, + /// `"full"` or `"empty"`; always `"full"` before gloas. + pub payload_status: &'static str, + /// Whether the head's epoch is later than the previous head's. False on + /// the second event a block gets when its payload arrives. + pub epoch_transition: bool, + pub current_epoch_dependent_root: H256, + pub next_epoch_dependent_root: H256, + pub execution_optimistic: bool, +} + /// The Beacon API `block` payload. +/// +/// From gloas on it also names the block's bid's builder and payload hash; +/// both are left out of an earlier block's JSON. #[derive(Clone, Debug, PartialEq, Eq, Serialize)] pub struct BeaconBlockEvent { #[serde(serialize_with = "quoted_or_bare::serialize")] pub slot: u64, pub block: H256, pub execution_optimistic: bool, + #[serde( + skip_serializing_if = "Option::is_none", + serialize_with = "serialize_quoted_option" + )] + pub builder_index: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub block_hash: Option, +} + +/// `value` as a quoted decimal, for the field that [`skip_serializing_if`] +/// has already kept `None` out of. +/// +/// [`skip_serializing_if`]: https://serde.rs/field-attrs.html#skip_serializing_if +fn serialize_quoted_option( + value: &Option, + serializer: S, +) -> Result { + match value { + Some(value) => quoted_or_bare::serialize(value, serializer), + None => serializer.serialize_none(), + } +} + +/// The Beacon API `execution_payload` payload: an envelope was imported. +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +pub struct ExecutionPayloadEvent { + #[serde(serialize_with = "quoted_or_bare::serialize")] + pub slot: u64, + #[serde(serialize_with = "quoted_or_bare::serialize")] + pub builder_index: u64, + pub block_hash: H256, + pub block_root: H256, + pub execution_optimistic: bool, +} + +/// The Beacon API `execution_payload_gossip` payload: `execution_payload`'s +/// fields less `execution_optimistic`, since the envelope is not imported. +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +pub struct ExecutionPayloadGossipEvent { + #[serde(serialize_with = "quoted_or_bare::serialize")] + pub slot: u64, + #[serde(serialize_with = "quoted_or_bare::serialize")] + pub builder_index: u64, + pub block_hash: H256, + pub block_root: H256, +} + +/// The Beacon API `execution_payload_available` payload. +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +pub struct ExecutionPayloadAvailableEvent { + #[serde(serialize_with = "quoted_or_bare::serialize")] + pub slot: u64, + pub block_root: H256, } /// The Beacon API `block_gossip` payload. @@ -492,6 +636,38 @@ impl EventBus { } } +impl GossipEventSink for EventBus { + fn has_subscribers(&self) -> bool { + EventBus::has_subscribers(self) + } + + fn emit(&self, event: GossipEvent) { + let event = match event { + GossipEvent::ExecutionPayloadGossip { + slot, + builder_index, + block_hash, + block_root, + } => ChainEvent::ExecutionPayloadGossip(ExecutionPayloadGossipEvent { + slot, + builder_index, + block_hash, + block_root, + }), + GossipEvent::ExecutionPayloadBid(bid) => { + ChainEvent::ExecutionPayloadBid(Box::new(Versioned::gloas(*bid))) + } + GossipEvent::PayloadAttestationMessage(message) => { + ChainEvent::PayloadAttestationMessage(Versioned::gloas(message)) + } + GossipEvent::ProposerPreferences(preferences) => { + ChainEvent::ProposerPreferences(Box::new(Versioned::gloas(*preferences))) + } + }; + EventBus::emit(self, event); + } +} + impl Default for EventBus { /// A live bus with the default channel capacity /// ([`CHAIN_EVENT_CHANNEL_CAPACITY`]). @@ -568,8 +744,9 @@ impl ChainEvents { } } - /// Beacon: `chain_reorg`, `head` and `finalized_checkpoint`, in that order, - /// for whatever moved since the last diff. + /// Beacon: `chain_reorg`, `head`, `head_v2` and `finalized_checkpoint`, in + /// that order, for whatever moved since the last diff, and `head_v2` alone + /// again when the head's payload status went from empty to full. /// /// Called at the end of `recompute_beacon_head`, where every beacon head /// and finality move ends. Only `head` is gated, against @@ -587,6 +764,11 @@ impl ChainEvents { self.view.head, wall_clock_slot, ); + } else if previous.payload_status == Some(PayloadStatus::Empty) + && self.view.payload_status == Some(PayloadStatus::Full) + { + // The same head, now with its payload: `head_v2` says so again. + publish_beacon_payload_arrived(store, &self.bus, self.view.head, wall_clock_slot); } if self.view.finalized != previous.finalized { publish_beacon_finalized_checkpoint(store, &self.bus, self.view.finalized); @@ -602,18 +784,60 @@ impl ChainEvents { } /// `block`: `block` was imported at `slot`, in the store's chain's shape. - pub(crate) fn publish_block(&self, store: &Store, slot: u64, block: H256) { + /// + /// `bid` is a gloas block's bid's `(builder_index, block_hash)`, `None` + /// for any other block. + pub(crate) fn publish_block( + &self, + store: &Store, + slot: u64, + block: H256, + bid: Option<(u64, H256)>, + ) { let event = match store.chain() { Chain::Lean => ChainEvent::Block { slot, block }, Chain::Beacon => ChainEvent::BeaconBlock(BeaconBlockEvent { slot, block, execution_optimistic: store.is_beacon_optimistic(block), + builder_index: bid.map(|(builder_index, _)| builder_index), + block_hash: bid.map(|(_, block_hash)| block_hash), }), }; self.bus.emit(event); } + /// Beacon `execution_payload` and `execution_payload_available` for an + /// envelope the actor just imported. + /// + /// Both here, in that order: the actor imports an envelope only once its + /// payload verified and every sampled column of its blobs is stored (see + /// `apply_or_hold_envelope`), which is what `execution_payload_available` + /// reports, so the two moments are one. + pub(crate) fn publish_envelope_imported( + &self, + store: &Store, + slot: u64, + builder_index: u64, + block_hash: H256, + block_root: H256, + ) { + if !self.bus.has_subscribers() { + return; + } + self.bus + .emit(ChainEvent::ExecutionPayload(ExecutionPayloadEvent { + slot, + builder_index, + block_hash, + block_root, + execution_optimistic: store.is_beacon_optimistic(block_root), + })); + self.bus.emit(ChainEvent::ExecutionPayloadAvailable( + ExecutionPayloadAvailableEvent { slot, block_root }, + )); + } + /// `block_gossip` for an arriving block, if its sender asked for it. /// /// The sender decides: see [`BlockAnnouncement`]. Behind the subscriber @@ -715,6 +939,9 @@ pub(crate) const HEAD_EVENT_RECENCY_SLOTS: u64 = 32; /// its `LiveChain` rows. struct EventView { head: H256, + /// The head node's payload status as recorded for `head` (beacon only; + /// `None` on lean, and on beacon until fork choice has recorded one). + payload_status: Option, /// Lean only: beacon has no justified-checkpoint topic. justified: Checkpoint, finalized: Checkpoint, @@ -724,6 +951,7 @@ impl EventView { fn capture(store: &Store) -> Self { Self { head: store.head().expect("head block exists"), + payload_status: store.head_payload_status(), justified: store .latest_justified() .expect("latest justified checkpoint exists"), @@ -937,15 +1165,114 @@ fn publish_beacon_head_change( return; } let epoch = compute_epoch_at_slot(new_slot); + let epoch_transition = epoch > compute_epoch_at_slot(old_slot); + let current_dependent = new_chain.dependent_root(Some(epoch)); + let previous_dependent = new_chain.dependent_root(epoch.checked_sub(1)); events.emit(ChainEvent::BeaconHead(BeaconHeadEvent { slot: new_slot, block: head, state: new_state, - epoch_transition: epoch > compute_epoch_at_slot(old_slot), - previous_duty_dependent_root: new_chain.dependent_root(epoch.checked_sub(1)), - current_duty_dependent_root: new_chain.dependent_root(Some(epoch)), + epoch_transition, + previous_duty_dependent_root: previous_dependent, + current_duty_dependent_root: current_dependent, execution_optimistic, })); + // `head` is deprecated in the specification's favour of `head_v2`, and + // clients on either keep working: both go out for every head move. + events.emit(head_v2_event( + store, + &head_state, + &new_chain, + new_state, + epoch_transition, + execution_optimistic, + )); +} + +/// `head_v2` for the head it already announced, now that its payload is in: +/// the second event the specification asks for when the payload status goes +/// from empty to full. Gated on recency like the first. +fn publish_beacon_payload_arrived( + store: &Store, + events: &EventBus, + head: H256, + wall_clock_slot: u64, +) { + let Some((slot, state_root)) = store.block_slot_and_state_root(&head) else { + warn!( + head_root = %ShortRoot(&head.0), + "Head block missing while emitting head_v2; skipping" + ); + return; + }; + if slot + HEAD_EVENT_RECENCY_SLOTS < wall_clock_slot { + return; + } + let head_state = match store.get_state(&head) { + Ok(Some(state)) => state, + Ok(None) | Err(_) => { + warn!( + head_root = %ShortRoot(&head.0), + "Head state missing while emitting head_v2; skipping" + ); + return; + } + }; + let new_chain = NewChain { + state: &head_state, + root: head, + slot, + }; + // Same block, same epoch: never a transition. + events.emit(head_v2_event( + store, + &head_state, + &new_chain, + state_root, + false, + store.is_beacon_optimistic(head), + )); +} + +/// The `head_v2` event for `new_chain`'s head, versioned by its state's fork. +fn head_v2_event( + store: &Store, + head_state: &BeaconState, + new_chain: &NewChain<'_>, + state_root: H256, + epoch_transition: bool, + execution_optimistic: bool, +) -> ChainEvent { + let fork = head_state.fork_name(); + let epoch = compute_epoch_at_slot(new_chain.slot); + ChainEvent::BeaconHeadV2(Versioned::new( + fork, + BeaconHeadV2Event { + slot: new_chain.slot, + block: new_chain.root, + state: state_root, + payload_status: payload_status_label(fork, store.head_payload_status()), + epoch_transition, + current_epoch_dependent_root: new_chain.dependent_root(epoch.checked_sub(1)), + next_epoch_dependent_root: new_chain.dependent_root(Some(epoch)), + execution_optimistic, + }, + )) +} + +/// The `payload_status` a `head_v2` event names for a head of `fork` whose +/// recorded status is `recorded`. +/// +/// Every head before gloas is `full`: its block carries its payload, so there +/// is nothing to wait for. From gloas on, only a head node fork choice picked +/// as FULL is; one not recorded yet (a head move whose status is still +/// pending) is `empty`, the state the block is in until its envelope arrives. +fn payload_status_label(fork: ForkName, recorded: Option) -> &'static str { + if fork < ForkName::Gloas || recorded == Some(PayloadStatus::Full) { + "full" + } else { + "empty" + } } fn publish_beacon_finalized_checkpoint(store: &Store, events: &EventBus, finalized: Checkpoint) { @@ -1503,6 +1830,8 @@ mod tests { slot: 10, block: root(BLOCK), execution_optimistic: false, + builder_index: None, + block_hash: None, }); assert_matches_example( event, @@ -1511,6 +1840,184 @@ mod tests { ); } + /// The same example with the two fields a gloas block adds, quoted and + /// hex as the specification writes them. Before gloas they are absent + /// rather than null (the test above). + #[test] + fn a_gloas_block_event_carries_its_bids_builder_and_payload_hash() { + let event = ChainEvent::BeaconBlock(BeaconBlockEvent { + slot: 10, + block: root(BLOCK), + execution_optimistic: false, + builder_index: Some(42), + block_hash: Some(root(PAYLOAD_HASH)), + }); + assert_matches_example( + event, + Topic::Block, + r#"{"slot":"10", "block":"0x9a2fefd2fdb57f74993c7780ea5b9030d2897b615b89f808011ca5aebed54eaf", "execution_optimistic": false, "builder_index":"42", "block_hash":"0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef"}"#, + ); + } + + const PAYLOAD_HASH: &str = "0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef"; + + #[test] + fn beacon_head_v2_matches_the_specification_example() { + let dependent = root("0x5e0043f107cb57913498fbf2f99ff55e730bf1e151f02f221e977c91a90a0e91"); + let event = ChainEvent::BeaconHeadV2(Versioned::new( + ForkName::Gloas, + BeaconHeadV2Event { + slot: 10, + block: root(BLOCK), + state: root(STATE), + payload_status: "empty", + epoch_transition: false, + current_epoch_dependent_root: dependent, + next_epoch_dependent_root: dependent, + execution_optimistic: false, + }, + )); + assert_matches_example( + event, + Topic::HeadV2, + r#"{"version": "gloas", "data":{"slot":"10", "block":"0x9a2fefd2fdb57f74993c7780ea5b9030d2897b615b89f808011ca5aebed54eaf", "state":"0x600e852a08c1200654ddf11025f1ceacb3c2e74bdd5c630cde0838b2591b69f9", "payload_status": "empty", "epoch_transition":false, "current_epoch_dependent_root":"0x5e0043f107cb57913498fbf2f99ff55e730bf1e151f02f221e977c91a90a0e91", "next_epoch_dependent_root":"0x5e0043f107cb57913498fbf2f99ff55e730bf1e151f02f221e977c91a90a0e91", "execution_optimistic": false}}"#, + ); + } + + #[test] + fn execution_payload_matches_the_specification_example() { + let event = ChainEvent::ExecutionPayload(ExecutionPayloadEvent { + slot: 10, + builder_index: 42, + block_hash: root(PAYLOAD_HASH), + block_root: root(BLOCK), + execution_optimistic: false, + }); + assert_matches_example( + event, + Topic::ExecutionPayload, + r#"{"slot":"10", "builder_index":"42", "block_hash":"0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef", "block_root":"0x9a2fefd2fdb57f74993c7780ea5b9030d2897b615b89f808011ca5aebed54eaf", "execution_optimistic": false}"#, + ); + } + + #[test] + fn execution_payload_gossip_matches_the_specification_example() { + let event = ChainEvent::ExecutionPayloadGossip(ExecutionPayloadGossipEvent { + slot: 10, + builder_index: 42, + block_hash: root(PAYLOAD_HASH), + block_root: root(BLOCK), + }); + assert_matches_example( + event, + Topic::ExecutionPayloadGossip, + r#"{"slot":"10", "builder_index":"42", "block_hash":"0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef", "block_root":"0x9a2fefd2fdb57f74993c7780ea5b9030d2897b615b89f808011ca5aebed54eaf"}"#, + ); + } + + #[test] + fn execution_payload_available_matches_the_specification_example() { + let event = ChainEvent::ExecutionPayloadAvailable(ExecutionPayloadAvailableEvent { + slot: 10, + block_root: root(BLOCK), + }); + assert_matches_example( + event, + Topic::ExecutionPayloadAvailable, + r#"{"slot": "10", "block_root": "0x9a2fefd2fdb57f74993c7780ea5b9030d2897b615b89f808011ca5aebed54eaf"}"#, + ); + } + + const SIGNATURE: &str = "0x1b66ac1fb663c9bc59509846d6ec05345bd908eda73e670af888da41af171505cc411d61252fb6cb3fa0017b679f8bb2305b26a285fa2737f175668d0dff91cc1b66ac1fb663c9bc59509846d6ec05345bd908eda73e670af888da41af171505"; + + /// The containers carry their own serde, so an event for each is the + /// specification's example decoded and wrapped, and it must come back out + /// as written: the quoting is theirs, the wrapper is this module's. + #[test] + fn the_wrapped_gloas_containers_match_the_specification_examples() { + let zero_address = "0x0000000000000000000000000000000000000000"; + let commitment = format!("0x{}", "ab".repeat(48)); + let bid = format!( + r#"{{"message": {{"parent_block_hash": "{BLOCK}", "parent_block_root": "{BLOCK}", "block_hash": "{PAYLOAD_HASH}", "prev_randao": "{BLOCK}", "fee_recipient": "{zero_address}", "gas_limit": "30000000", "builder_index": "42", "slot": "10", "value": "1000000000", "execution_payment": "0", "blob_kzg_commitments": ["{commitment}"], "execution_requests_root": "{BLOCK}"}}, "signature": "{SIGNATURE}"}}"# + ); + let message = format!( + r#"{{"validator_index": "123", "data": {{"beacon_block_root": "{BLOCK}", "slot": "10", "payload_present": true, "blob_data_available": true}}, "signature": "{SIGNATURE}"}}"# + ); + let preferences = format!( + r#"{{"message": {{"dependent_root": "{BLOCK}", "proposal_slot": "32", "validator_index": "123", "fee_recipient": "{zero_address}", "target_gas_limit": "60000000"}}, "signature": "{SIGNATURE}"}}"# + ); + let wrap = |data: &str| format!(r#"{{"version": "gloas", "data": {data}}}"#); + + let event = ChainEvent::ExecutionPayloadBid(Box::new(Versioned::gloas( + serde_json::from_str(&bid).unwrap(), + ))); + assert_matches_example(event, Topic::ExecutionPayloadBid, &wrap(&bid)); + let event = ChainEvent::PayloadAttestationMessage(Versioned::gloas( + serde_json::from_str(&message).unwrap(), + )); + assert_matches_example(event, Topic::PayloadAttestationMessage, &wrap(&message)); + let event = ChainEvent::ProposerPreferences(Box::new(Versioned::gloas( + serde_json::from_str(&preferences).unwrap(), + ))); + assert_matches_example(event, Topic::ProposerPreferences, &wrap(&preferences)); + } + + /// What p2p reports through the sink becomes the matching topic's event. + #[tokio::test] + async fn the_bus_turns_gossip_events_into_their_topics() { + let bus = EventBus::new(8); + let mut rx = bus.subscribe(); + let sink: &dyn GossipEventSink = &bus; + assert!(sink.has_subscribers()); + + sink.emit(GossipEvent::ExecutionPayloadGossip { + slot: 10, + builder_index: 42, + block_hash: root(PAYLOAD_HASH), + block_root: root(BLOCK), + }); + sink.emit(GossipEvent::ExecutionPayloadBid(Box::default())); + sink.emit(GossipEvent::ProposerPreferences(Box::default())); + sink.emit(GossipEvent::PayloadAttestationMessage( + gloas::PayloadAttestationMessage { + validator_index: 1, + data: Default::default(), + signature: Default::default(), + }, + )); + + let topics: Vec = std::iter::from_fn(|| rx.try_recv().ok()) + .map(|event| event.topic()) + .collect(); + assert_eq!( + topics, + [ + Topic::ExecutionPayloadGossip, + Topic::ExecutionPayloadBid, + Topic::ProposerPreferences, + Topic::PayloadAttestationMessage, + ] + ); + } + + #[test] + fn every_head_before_gloas_is_full_and_a_gloas_head_follows_its_node() { + for fork in [ForkName::Phase0, ForkName::Fulu] { + for recorded in [None, Some(PayloadStatus::Empty), Some(PayloadStatus::Full)] { + assert_eq!(payload_status_label(fork, recorded), "full"); + } + } + assert_eq!( + payload_status_label(ForkName::Gloas, Some(PayloadStatus::Full)), + "full" + ); + assert_eq!( + payload_status_label(ForkName::Gloas, Some(PayloadStatus::Empty)), + "empty" + ); + assert_eq!(payload_status_label(ForkName::Gloas, None), "empty"); + } + #[test] fn beacon_block_gossip_matches_the_specification_example() { let event = ChainEvent::BeaconBlockGossip(BeaconBlockGossipEvent { diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index 251048b7a..f8afd9a5c 100644 --- a/crates/blockchain/src/lib.rs +++ b/crates/blockchain/src/lib.rs @@ -1722,6 +1722,10 @@ impl BlockChainServer { // post-state the store already holds before ever reaching here. let slot = signed_block.slot(); let block_root = signed_block.message_hash_tree_root(); + // Two words, taken before the match moves the block: the `block` event + // of a gloas block names its bid's builder and payload hash. + let bid = crate::beacon_payloads::bid_of(&signed_block) + .map(|bid| (bid.builder_index, bid.block_hash)); let is_new = !self .store .has_state(&block_root) @@ -1943,7 +1947,8 @@ impl BlockChainServer { // `block` goes out first so subscribers see it ahead of the // justified/head/finalized moves its import triggers. if is_new { - self.events.publish_block(&self.store, slot, block_root); + self.events + .publish_block(&self.store, slot, block_root, bid); } // Lean only: a beacon import moves no head of its own, and the // cascade's head recompute publishes beacon's events once it runs. @@ -5828,9 +5833,11 @@ mod tests { use crate::events::{ BeaconAttestationEvent, BeaconChainReorgEvent, BeaconFinalizedCheckpointEvent, - BeaconHeadEvent, DataColumnSidecarEvent, HEAD_EVENT_RECENCY_SLOTS, + BeaconHeadEvent, BeaconHeadV2Event, DataColumnSidecarEvent, ExecutionPayloadAvailableEvent, + ExecutionPayloadEvent, HEAD_EVENT_RECENCY_SLOTS, Versioned, }; use ethlambda_storage::ForkCheckpoints; + use ethlambda_types::beacon::fork_choice::PayloadStatus; use tokio::sync::broadcast; const G: H256 = H256::repeat_byte(0x10); @@ -5901,8 +5908,12 @@ mod tests { .expect("update head"); } + /// What is on the channel but `head_v2`, which rides every `head` and is + /// tested apart (see [`drain_all`]). fn drain(rx: &mut broadcast::Receiver) -> Vec { - std::iter::from_fn(|| rx.try_recv().ok()).collect() + std::iter::from_fn(|| rx.try_recv().ok()) + .filter(|event| !matches!(event, ChainEvent::BeaconHeadV2(_))) + .collect() } fn as_head(event: &ChainEvent) -> &BeaconHeadEvent { @@ -6235,4 +6246,185 @@ mod tests { }; assert_eq!((gossip.slot, gossip.block), (4, root)); } + + // ----------------------------------------------------------------- + // `head_v2` + // ----------------------------------------------------------------- + + /// Everything on the channel, `head_v2` included. [`drain`] leaves it + /// out, so the tests of `head` and `chain_reorg` keep reading the + /// events they name. + fn drain_all(rx: &mut broadcast::Receiver) -> Vec { + std::iter::from_fn(|| rx.try_recv().ok()).collect() + } + + fn as_head_v2(event: &ChainEvent) -> &Versioned { + match event { + ChainEvent::BeaconHeadV2(head) => head, + other => panic!("expected a head_v2 event, got {other:?}"), + } + } + + /// Every head move announces `head`, then `head_v2` for the same block, + /// and a pre-gloas head is `full`. + #[test] + fn a_head_move_emits_head_v2_right_after_head_and_it_is_full_before_gloas() { + let (mut store, mut events, mut rx) = fork_fixture(A); + set_head(&mut store, B); + events.publish_beacon_changes(&store, 2); + + let emitted = drain_all(&mut rx); + assert_eq!(emitted.len(), 2, "{emitted:?}"); + assert_eq!(as_head(&emitted[0]).block, B); + let v2 = as_head_v2(&emitted[1]); + assert_eq!(v2.version, "phase0"); + assert_eq!( + v2.data, + BeaconHeadV2Event { + slot: 2, + block: B, + state: post_state_root(B), + payload_status: "full", + epoch_transition: false, + // Epoch 0: both dependent roots underflow to genesis. + current_epoch_dependent_root: G, + next_epoch_dependent_root: G, + execution_optimistic: false, + } + ); + } + + /// `head_v2`'s roots are `head`'s shifted one epoch: current is the root + /// at `start_slot(epoch - 1) - 1`, next the root at `start_slot(epoch) - 1`, + /// both the genesis block's on underflow. + #[test] + fn head_v2_dependent_roots_follow_the_epoch_and_underflow_to_genesis() { + let x = H256::repeat_byte(0x21); + let y = H256::repeat_byte(0x22); + let z = H256::repeat_byte(0x23); + let mut store = event_store(GENESIS_TIME); + insert_event_block(&mut store, x, 31, G, &[(0, G)]); + insert_event_block(&mut store, y, 32, x, &[(0, G), (31, x)]); + insert_event_block(&mut store, z, 65, y, &[(0, G), (31, x), (32, y)]); + set_head(&mut store, x); + let mut events = ChainEvents::new(EventBus::new(16), &store); + let mut rx = events.subscribe(); + + set_head(&mut store, y); + events.publish_beacon_changes(&store, 32); + let emitted = drain_all(&mut rx); + let head = &as_head_v2(&emitted[1]).data; + assert!(head.epoch_transition); + // Epoch 1 (slot 32): `start_slot(0) - 1` underflows, `start_slot(1) - 1` + // is slot 31. + assert_eq!(head.current_epoch_dependent_root, G); + assert_eq!(head.next_epoch_dependent_root, x); + + set_head(&mut store, z); + events.publish_beacon_changes(&store, 65); + let emitted = drain_all(&mut rx); + let head = &as_head_v2(&emitted[1]).data; + assert!(head.epoch_transition); + // Epoch 2 (slot 65): slots 31 and 63. + assert_eq!(head.current_epoch_dependent_root, x); + assert_eq!(head.next_epoch_dependent_root, y); + } + + /// A head whose payload arrives gets `head_v2` a second time, for the same + /// block and with no epoch transition; nothing else moves. + #[test] + fn a_head_whose_payload_arrives_gets_head_v2_again() { + let (store, mut events, mut rx) = fork_fixture(B); + store.set_head_payload_status(B, PayloadStatus::Empty); + // The view starts from the empty head, as after the head's own move. + events.publish_beacon_changes(&store, 2); + assert!(drain_all(&mut rx).is_empty()); + + store.set_head_payload_status(B, PayloadStatus::Full); + events.publish_beacon_changes(&store, 2); + + let emitted = drain_all(&mut rx); + assert_eq!(emitted.len(), 1, "{emitted:?}"); + let v2 = &as_head_v2(&emitted[0]).data; + assert_eq!((v2.slot, v2.block), (2, B)); + assert!(!v2.epoch_transition); + + // Reported once: the view now holds FULL. + events.publish_beacon_changes(&store, 2); + assert!(drain_all(&mut rx).is_empty()); + } + + /// Other transitions are optional in the specification and left out here: + /// full to empty, and an empty head that stays empty. + #[test] + fn only_empty_to_full_repeats_head_v2() { + let (store, mut events, mut rx) = fork_fixture(B); + store.set_head_payload_status(B, PayloadStatus::Full); + events.publish_beacon_changes(&store, 2); + store.set_head_payload_status(B, PayloadStatus::Empty); + events.publish_beacon_changes(&store, 2); + assert!(drain_all(&mut rx).is_empty()); + } + + // ----------------------------------------------------------------- + // Envelope import events + // ----------------------------------------------------------------- + + /// An envelope the chain actor imports publishes `execution_payload` and + /// `execution_payload_available`, in that order. + #[test] + fn an_imported_envelope_publishes_execution_payload_then_available() { + let (store, events, mut rx) = fork_fixture(B); + + events.publish_envelope_imported(&store, 2, 42, H256::repeat_byte(0x77), B); + + let emitted = drain_all(&mut rx); + assert_eq!(emitted.len(), 2, "{emitted:?}"); + match &emitted[0] { + ChainEvent::ExecutionPayload(payload) => assert_eq!( + payload, + &ExecutionPayloadEvent { + slot: 2, + builder_index: 42, + block_hash: H256::repeat_byte(0x77), + block_root: B, + execution_optimistic: false, + } + ), + other => panic!("expected execution_payload first, got {other:?}"), + } + match &emitted[1] { + ChainEvent::ExecutionPayloadAvailable(available) => assert_eq!( + available, + &ExecutionPayloadAvailableEvent { + slot: 2, + block_root: B + } + ), + other => panic!("expected execution_payload_available second, got {other:?}"), + } + } + + /// A gloas block's `block` event names its bid's builder and payload + /// hash; an earlier block's carries neither. + #[test] + fn the_block_event_carries_the_bid_of_a_gloas_block_only() { + let (store, events, mut rx) = fork_fixture(B); + + events.publish_block(&store, 2, B, Some((42, H256::repeat_byte(0x77)))); + events.publish_block(&store, 2, B, None); + + let emitted = drain_all(&mut rx); + let json: Vec<_> = emitted + .iter() + .map(|event| serde_json::to_value(event).unwrap()) + .collect(); + assert_eq!(json[0]["builder_index"], "42"); + assert_eq!( + json[0]["block_hash"], + format!("0x{}", "77".repeat(32)).as_str() + ); + assert!(json[1].get("builder_index").is_none()); + assert!(json[1].get("block_hash").is_none()); + } } diff --git a/crates/net/api/src/lib.rs b/crates/net/api/src/lib.rs index a0aa347d2..e1d692459 100644 --- a/crates/net/api/src/lib.rs +++ b/crates/net/api/src/lib.rs @@ -533,3 +533,47 @@ pub struct InitBlockChain { impl Message for InitBlockChain { type Result = (); } + +// --- Sink: P2P -> event stream --- + +/// A gloas gossip object that passed its topic's validation inside +/// `ethlambda-p2p`, for the Beacon API's event stream. +/// +/// These are the topics whose objects never reach the chain actor, which is +/// the event bus's usual publisher: bids and proposer preferences are +/// validated and kept in the p2p layer's `BuilderMarket`, and a payload +/// attestation message is pooled there. Carrying only what each event shows +/// keeps the sink free of a dependency on the blockchain crate, which owns +/// the bus. +#[derive(Clone, Debug)] +pub enum GossipEvent { + /// An `execution_payload` envelope passed gossip validation. Only the + /// fields the event shows are carried, since the envelope itself holds a + /// whole execution payload. + ExecutionPayloadGossip { + slot: u64, + builder_index: u64, + block_hash: H256, + block_root: H256, + }, + /// An `execution_payload_bid` passed gossip validation. + ExecutionPayloadBid(Box), + /// A `payload_attestation_message` passed gossip validation. + PayloadAttestationMessage(PayloadAttestationMessage), + /// A `proposer_preferences` message passed gossip validation. + ProposerPreferences(Box), +} + +/// Where `ethlambda-p2p` reports [`GossipEvent`]s. +/// +/// Implemented by the event bus in `ethlambda-blockchain`, so the p2p layer +/// can feed `/eth/v1/events` without depending on the crate that owns it. +/// Best-effort like the bus itself: `emit` never blocks and never fails. +pub trait GossipEventSink: Send + Sync { + /// Whether anything listens. Asked before building a payload that costs + /// a clone, so a node nobody subscribes to pays nothing. + fn has_subscribers(&self) -> bool; + + /// Publish `event` to every current subscriber. + fn emit(&self, event: GossipEvent); +} diff --git a/crates/net/p2p/src/beacon/verdict.rs b/crates/net/p2p/src/beacon/verdict.rs index ab5651c26..8dae1f8c9 100644 --- a/crates/net/p2p/src/beacon/verdict.rs +++ b/crates/net/p2p/src/beacon/verdict.rs @@ -16,7 +16,9 @@ use std::panic::{AssertUnwindSafe, catch_unwind}; use std::time::Instant; -use ethlambda_network_api::{AggregateArrival, BlockAnnouncement, BlockArrival, BlockSource}; +use ethlambda_network_api::{ + AggregateArrival, BlockAnnouncement, BlockArrival, BlockSource, GossipEvent, +}; use ethlambda_state_transition::beacon::builder_market::SharedBuilderMarket; use ethlambda_state_transition::beacon::gossip::{self, IgnoreReason, Outcome}; use ethlambda_state_transition::beacon::helpers::accessors::CommitteeCacheExt as _; @@ -269,6 +271,7 @@ impl Validated { fn forward(self, server: &P2PServer, received_at: Instant, outcome: Outcome) { if outcome == Outcome::Accept { record_liveness(server, &self); + publish_gossip_event(server, &self); } if let Self::Aggregate { aggregate, .. } = &self && outcome == Outcome::Accept @@ -391,8 +394,7 @@ impl Validated { Self::Aggregate { .. } | Self::Attestation { .. } | Self::Envelope(_) - // The SSE `execution_payload_bid` and `proposer_preferences` events - // hook in here once the events endpoint lands. + // Their events went out above, in `publish_gossip_event`. | Self::ExecutionPayloadBid { .. } | Self::ProposerPreferences(_) | Self::PayloadAttestation(_) @@ -403,6 +405,40 @@ impl Validated { } } +/// Report an accepted gloas object on the Beacon API's event stream, for the +/// four topics whose objects never reach the chain actor that publishes the +/// rest (bids and preferences stop here, and a payload attestation message +/// only ever feeds fork choice). +/// +/// Only on `Accept`, since each event names objects that passed the topic's +/// rules. A queued envelope has not, so `execution_payload_gossip` waits for +/// the copy that does. Behind the subscriber check, because a bid is cloned +/// into the event. +fn publish_gossip_event(server: &P2PServer, object: &Validated) { + let Some(sink) = &server.gossip_events else { + return; + }; + let event = match object { + Validated::Envelope(envelope) => GossipEvent::ExecutionPayloadGossip { + slot: envelope.message.payload.slot_number, + builder_index: envelope.message.builder_index, + block_hash: envelope.message.payload.block_hash, + block_root: envelope.message.beacon_block_root, + }, + Validated::ExecutionPayloadBid { bid, .. } if sink.has_subscribers() => { + GossipEvent::ExecutionPayloadBid(bid.clone()) + } + Validated::ProposerPreferences(preferences) if sink.has_subscribers() => { + GossipEvent::ProposerPreferences(preferences.clone()) + } + Validated::PayloadAttestation(message) if sink.has_subscribers() => { + GossipEvent::PayloadAttestationMessage(message.clone()) + } + _ => return, + }; + sink.emit(event); +} + /// Mark the validators an accepted aggregate or subnet attestation names as /// live for its target epoch, for `/eth/v1/validator/liveness`. /// diff --git a/crates/net/p2p/src/lib.rs b/crates/net/p2p/src/lib.rs index cc9ca2bd1..ff24a31da 100644 --- a/crates/net/p2p/src/lib.rs +++ b/crates/net/p2p/src/lib.rs @@ -38,7 +38,7 @@ use std::{ use either::Either; use ethlambda_network_api::{ - FetchRequest, InitBlockChain, MetaDataController, P2PToBlockChainRef, + FetchRequest, GossipEventSink, InitBlockChain, MetaDataController, P2PToBlockChainRef, block_chain_to_p2p::{ CheckDataColumnSidecars, FetchBlock, PublishAggregatedAttestation, PublishAttestation, PublishBlock, @@ -1180,6 +1180,7 @@ impl P2P { payload_attestation_pool: SharedPayloadAttestationPool, builder_market: SharedBuilderMarket, sync_committee_pool: SharedSyncCommitteePool, + gossip_events: Option>, ) -> Result { let (discovery, local_enr) = match discovery { Some(config) => { @@ -1263,6 +1264,7 @@ impl P2P { BUILDER_VALIDATION_PERMITS, )), sync_committee_pool, + gossip_events, aggregator_subnets: HashMap::new(), peer_scoring, }; @@ -1457,6 +1459,10 @@ pub struct P2PServer { /// it). Filled by `verdict::forward`; lean never touches it. pub(crate) sync_committee_pool: SharedSyncCommitteePool, + /// Where gloas gossip objects that passed validation are reported for the + /// Beacon API's event stream. `None` on a node with no event bus. + pub(crate) gossip_events: Option>, + /// The attestation subnets joined for a validator client's aggregators, /// each with the last slot it is needed for. Short-lived by design: never /// advertised in `attnets`, and left once the slot has passed. The @@ -3101,6 +3107,7 @@ pub(crate) mod test_support { crate::BUILDER_VALIDATION_PERMITS, )), sync_committee_pool: Default::default(), + gossip_events: None, aggregator_subnets: HashMap::new(), peer_scoring: None, } diff --git a/crates/net/p2p/src/req_resp/handlers.rs b/crates/net/p2p/src/req_resp/handlers.rs index 5801a84f2..c1a4e591f 100644 --- a/crates/net/p2p/src/req_resp/handlers.rs +++ b/crates/net/p2p/src/req_resp/handlers.rs @@ -2839,6 +2839,7 @@ pub(crate) mod tests { crate::BUILDER_VALIDATION_PERMITS, )), sync_committee_pool: Default::default(), + gossip_events: None, aggregator_subnets: HashMap::new(), peer_scoring: None, } diff --git a/crates/net/rpc/src/beacon/bids.rs b/crates/net/rpc/src/beacon/bids.rs index e7da89e78..e4b96d52c 100644 --- a/crates/net/rpc/src/beacon/bids.rs +++ b/crates/net/rpc/src/beacon/bids.rs @@ -16,7 +16,7 @@ use axum::{ response::{IntoResponse, Response}, routing::post, }; -use ethlambda_network_api::RpcToP2PRef; +use ethlambda_network_api::{GossipEvent, RpcToP2PRef}; use ethlambda_state_transition::beacon::{ builder_market::SharedBuilderMarket, gossip::{ @@ -30,7 +30,7 @@ use ethlambda_storage::Store; use ethlambda_types::beacon::{containers::gloas::SignedExecutionPayloadBid, fork::ForkName}; use tracing::{debug, warn}; -use crate::beacon::{ApiError, BodyEncoding}; +use crate::beacon::{ApiError, BodyEncoding, emit_accepted}; pub(crate) fn routes() -> Router { Router::new().route("/eth/v1/beacon/execution_payload_bids", post(post_bid)) @@ -82,6 +82,7 @@ async fn post_bid( State(store): State, Extension(p2p): Extension, Extension(market): Extension, + events: Option>, headers: HeaderMap, body: Bytes, ) -> Response { @@ -141,8 +142,12 @@ async fn post_bid( if !market.record_bid(bid.clone()) { return bad_request(describe(&Outcome::Ignore(IgnoreReason::AlreadySeen))); } + let for_event = bid.clone(); match p2p.publish_execution_payload_bid(bid) { Ok(()) => { + emit_accepted(events.as_deref(), || { + GossipEvent::ExecutionPayloadBid(Box::new(for_event)) + }); debug!( slot, builder_index, "Accepted execution payload bid for gossip" diff --git a/crates/net/rpc/src/beacon/builder_market_tests.rs b/crates/net/rpc/src/beacon/builder_market_tests.rs index 4f89fd600..384a406e9 100644 --- a/crates/net/rpc/src/beacon/builder_market_tests.rs +++ b/crates/net/rpc/src/beacon/builder_market_tests.rs @@ -1293,3 +1293,87 @@ async fn the_infinity_reveal_is_produced_only_when_the_flag_skips_its_verificati assert_eq!(reply.status, StatusCode::BAD_REQUEST); assert_eq!(reply.json()["message"], "invalid randao_reveal"); } + +// --------------------------------------------------------------------------- +// Event stream: what the POST endpoints accept is announced +// --------------------------------------------------------------------------- + +/// The events a router layered with the bus published, in order, as +/// `{topic, body}`. +fn drained( + rx: &mut tokio::sync::broadcast::Receiver, +) -> Vec { + std::iter::from_fn(|| rx.try_recv().ok()) + .map(|event| { + serde_json::json!({ + "topic": event.topic().as_str(), + "body": serde_json::to_value(&event).unwrap(), + }) + }) + .collect() +} + +#[tokio::test] +async fn an_accepted_bid_is_announced_once_in_its_version_wrapper() { + let events = ethlambda_blockchain::EventBus::new(8); + let mut rx = events.subscribe(); + let world = world_ready_for_bids(); + let app = world.app(None).layer(Extension(events)); + let bid = world.bid(5, address(0xaa)); + + let reply = send(&app, bid_request(&bid)).await; + assert_eq!(reply.status, StatusCode::OK); + // The identical bid again is answered without a publish, and so without + // an event. + let reply = send(&app, bid_request(&bid)).await; + assert_eq!(reply.status, StatusCode::OK); + + let emitted = drained(&mut rx); + assert_eq!(emitted.len(), 1, "{emitted:?}"); + assert_eq!(emitted[0]["topic"], "execution_payload_bid"); + assert_eq!(emitted[0]["body"]["version"], "gloas"); + assert_eq!( + emitted[0]["body"]["data"], + serde_json::to_value(&bid).unwrap() + ); +} + +#[tokio::test] +async fn a_refused_bid_is_not_announced() { + let events = ethlambda_blockchain::EventBus::new(8); + let mut rx = events.subscribe(); + let world = world_ready_for_bids(); + let app = world.app(None).layer(Extension(events)); + let mut bid = world.bid(5, address(0xaa)); + bid.signature = BlsSignature::default(); + + let reply = send(&app, bid_request(&bid)).await; + + assert_eq!(reply.status, StatusCode::BAD_REQUEST); + assert!(drained(&mut rx).is_empty()); +} + +#[tokio::test] +async fn accepted_preferences_are_announced() { + let events = ethlambda_blockchain::EventBus::new(8); + let mut rx = events.subscribe(); + let world = World::new(); + let app = world.app(None).layer(Extension(events)); + let signed = world.preferences(address(0xaa), 30_000_000); + + let reply = send( + &app, + preferences_request(std::slice::from_ref(&signed), Some("gloas")), + ) + .await; + assert_eq!(reply.status, StatusCode::OK); + + let emitted = drained(&mut rx); + assert_eq!(emitted.len(), 1, "{emitted:?}"); + assert_eq!(emitted[0]["topic"], "proposer_preferences"); + assert_eq!(emitted[0]["body"]["version"], "gloas"); + assert_eq!( + emitted[0]["body"]["data"], + serde_json::to_value(&signed).unwrap() + ); +} diff --git a/crates/net/rpc/src/beacon/events.rs b/crates/net/rpc/src/beacon/events.rs index 5b7902079..20f6c8060 100644 --- a/crates/net/rpc/src/beacon/events.rs +++ b/crates/net/rpc/src/beacon/events.rs @@ -14,8 +14,8 @@ //! - The payloads are the chain actor's beacon variants, which quote every //! integer, so nothing here reshapes an event. //! -//! Every event comes from the chain actor, the bus's only publisher; this -//! handler only subscribes. +//! The chain actor, p2p and the submission endpoints publish to the bus (see +//! `ethlambda_blockchain::events`); this handler only subscribes. //! //! [`ChainEvent`]: ethlambda_blockchain::ChainEvent @@ -238,6 +238,8 @@ mod tests { slot: 1, block: H256::ZERO, execution_optimistic: false, + builder_index: None, + block_hash: None, })); events.emit(head_event(1)); events.emit(ChainEvent::BeaconFinalizedCheckpoint( diff --git a/crates/net/rpc/src/beacon/gloas_proposal.rs b/crates/net/rpc/src/beacon/gloas_proposal.rs index 51b57d76e..d639fe9ff 100644 --- a/crates/net/rpc/src/beacon/gloas_proposal.rs +++ b/crates/net/rpc/src/beacon/gloas_proposal.rs @@ -44,13 +44,13 @@ use axum::{ response::{IntoResponse, Response}, routing::{get, post}, }; -use ethlambda_blockchain::checkpoint_hash; +use ethlambda_blockchain::{EventBus, checkpoint_hash}; use ethlambda_engine::{ CustodyColumns, EngineClient, ForkchoiceStateV1, building::{BuiltGloasPayload, PayloadAttributesV3, PayloadAttributesV4}, types::ClientVersionV1, }; -use ethlambda_network_api::RpcToP2PRef; +use ethlambda_network_api::{GossipEvent, RpcToP2PRef}; use ethlambda_state_transition::beacon::{ block_production::{advance_to_slot, empty_sync_aggregate, verified_sync_aggregate}, bls, @@ -108,6 +108,7 @@ use crate::beacon::{ ApiError, BodyEncoding, bid_selection::{LocalCandidate, PayloadChoice, bid_total_gwei, choose_payload, wei_u128}, builder_config::{BuilderConfig, decode_builder_config}, + emit_accepted, graffiti::{self, OwnVersion}, proposal::{Blobs, CellKzgProofs, decimal, randao_verification, require_gloas_slot}, validator::{FeeRecipients, head}, @@ -1022,12 +1023,24 @@ async fn post_envelope( State(store): State, Extension(p2p): Extension, Extension(cache): Extension, + events: Option>, headers: HeaderMap, body: Bytes, ) -> Response { - post_envelope_waiting(&store, &p2p, &cache, &headers, &body, BLOCK_WAIT).await + post_envelope_emitting( + &store, + &p2p, + &cache, + events.as_deref(), + &headers, + &body, + BLOCK_WAIT, + ) + .await } +/// [`post_envelope_emitting`] without an event bus, for the handler's tests. +#[cfg(test)] async fn post_envelope_waiting( store: &Store, p2p: &RpcToP2PRef, @@ -1035,6 +1048,18 @@ async fn post_envelope_waiting( headers: &HeaderMap, body: &[u8], block_wait: Duration, +) -> Response { + post_envelope_emitting(store, p2p, cache, None, headers, body, block_wait).await +} + +async fn post_envelope_emitting( + store: &Store, + p2p: &RpcToP2PRef, + cache: &PayloadCache, + events: Option<&EventBus>, + headers: &HeaderMap, + body: &[u8], + block_wait: Duration, ) -> Response { if consensus_version(headers) != Some(ForkName::Gloas) { return ApiError::BadRequest("Eth-Consensus-Version must be gloas").into_response(); @@ -1160,12 +1185,19 @@ async fn post_envelope_waiting( let sidecars: Vec = sidecars.into_iter().map(DataColumnSidecar::Gloas).collect(); + let accepted = GossipEvent::ExecutionPayloadGossip { + slot: block_slot, + builder_index: signed.message.builder_index, + block_hash: signed.message.payload.block_hash, + block_root, + }; if p2p .publish_execution_payload_envelope(Box::new(signed), sidecars) .is_err() { return ApiError::Internal("the network actor is not running").into_response(); } + emit_accepted(events, || accepted); StatusCode::OK.into_response() } diff --git a/crates/net/rpc/src/beacon/mod.rs b/crates/net/rpc/src/beacon/mod.rs index f9ed67066..b463cdcaa 100644 --- a/crates/net/rpc/src/beacon/mod.rs +++ b/crates/net/rpc/src/beacon/mod.rs @@ -155,6 +155,23 @@ impl BodyEncoding { } } +/// Report an object the API accepted on the event stream, when the node has an +/// [`EventBus`](ethlambda_blockchain::EventBus) and anyone listens. +/// +/// `event` builds the payload only behind that check, since a bid or an +/// envelope's fields are cloned into it. The bus is an `Option>` +/// on the handlers that call this, so a router built without one (the +/// handlers' own tests) still serves. +pub(crate) fn emit_accepted( + events: Option<ðlambda_blockchain::EventBus>, + event: impl FnOnce() -> ethlambda_network_api::GossipEvent, +) { + use ethlambda_network_api::GossipEventSink; + if let Some(events) = events.filter(|events| events.has_subscribers()) { + GossipEventSink::emit(events, event()); + } +} + /// Decode the array a batch-submission endpoint takes, as JSON or as the SSZ /// `List[T, ...]` of the same elements, by the request's `Content-Type`. pub(crate) fn decode_list( diff --git a/crates/net/rpc/src/beacon/proposer_preferences.rs b/crates/net/rpc/src/beacon/proposer_preferences.rs index bd217f0b5..96b99f0ac 100644 --- a/crates/net/rpc/src/beacon/proposer_preferences.rs +++ b/crates/net/rpc/src/beacon/proposer_preferences.rs @@ -14,7 +14,8 @@ use axum::{ response::{IntoResponse, Response}, routing::post, }; -use ethlambda_network_api::RpcToP2PRef; +use ethlambda_blockchain::EventBus; +use ethlambda_network_api::{GossipEvent, RpcToP2PRef}; use ethlambda_state_transition::beacon::{ builder_market::SharedBuilderMarket, gossip::{IgnoreReason, Outcome, proposer_preferences::validate}, @@ -28,7 +29,7 @@ use tracing::{debug, warn}; use crate::beacon::{ ApiError, bids::{describe, require_version, unix_ms}, - decode_list, + decode_list, emit_accepted, }; /// The most entries one request may carry: the SSZ list's bound, @@ -62,6 +63,7 @@ async fn post_proposer_preferences( State(store): State, Extension(p2p): Extension, Extension(market): Extension, + events: Option>, headers: HeaderMap, body: Bytes, ) -> Response { @@ -78,8 +80,11 @@ async fn post_proposer_preferences( return ApiError::BadRequest("too many signed proposer preferences").into_response(); } - let validated = - tokio::task::spawn_blocking(move || submit(&store, &p2p, &market, preferences)).await; + let events = events.map(|Extension(events)| events); + let validated = tokio::task::spawn_blocking(move || { + submit(&store, &p2p, &market, events.as_ref(), preferences) + }) + .await; let failures = match validated { Ok(failures) => failures, Err(_) => return ApiError::Internal("validating the preferences failed").into_response(), @@ -102,6 +107,7 @@ fn submit( store: &Store, p2p: &RpcToP2PRef, market: &SharedBuilderMarket, + events: Option<&EventBus>, preferences: Vec, ) -> Vec { let now_ms = unix_ms(); @@ -132,8 +138,14 @@ fn submit( }); continue; } + let for_event = signed.clone(); match p2p.publish_proposer_preferences(signed) { - Ok(()) => debug!(%slot, validator, "Accepted proposer preferences for gossip"), + Ok(()) => { + emit_accepted(events, || { + GossipEvent::ProposerPreferences(Box::new(for_event)) + }); + debug!(%slot, validator, "Accepted proposer preferences for gossip") + } Err(_) => failures.push(Failure { index, message: "the network actor is not running".to_string(), diff --git a/crates/net/rpc/src/beacon/ptc.rs b/crates/net/rpc/src/beacon/ptc.rs index be80bf330..e293e3807 100644 --- a/crates/net/rpc/src/beacon/ptc.rs +++ b/crates/net/rpc/src/beacon/ptc.rs @@ -12,6 +12,7 @@ use std::collections::{BTreeMap, HashSet}; use std::sync::Arc; +use crate::beacon::emit_accepted; use axum::{ Extension, Router, body::Bytes, @@ -20,10 +21,10 @@ use axum::{ response::{IntoResponse, Response}, routing::{get, post}, }; -use ethlambda_blockchain::SyncStatusController; use ethlambda_blockchain::metrics::SyncStatus; +use ethlambda_blockchain::{EventBus, SyncStatusController}; use ethlambda_engine::EngineClient; -use ethlambda_network_api::RpcToP2PRef; +use ethlambda_network_api::{GossipEvent, RpcToP2PRef}; use ethlambda_state_transition::beacon::{ fork_choice::get_payload_due_ms, gloas_block_production::aggregate_payload_attestations, @@ -389,6 +390,7 @@ async fn post_pool_payload_attestations( State(store): State, Extension(p2p): Extension, Extension(pool): Extension, + events: Option>, headers: HeaderMap, body: Bytes, ) -> Response { @@ -433,8 +435,14 @@ async fn post_pool_payload_attestations( debug!(%slot, validator, "Payload attestation already pooled; not republishing"); continue; } + let for_event = message.clone(); match p2p.publish_payload_attestation_message(message) { - Ok(()) => debug!(%slot, validator, "Accepted payload attestation for gossip"), + Ok(()) => { + emit_accepted(events.as_deref(), || { + GossipEvent::PayloadAttestationMessage(for_event) + }); + debug!(%slot, validator, "Accepted payload attestation for gossip") + } Err(_) => failures.push(Failure { index, message: "the network actor is not running".to_string(), diff --git a/crates/net/rpc/src/lib.rs b/crates/net/rpc/src/lib.rs index 4870c2781..4d90ece03 100644 --- a/crates/net/rpc/src/lib.rs +++ b/crates/net/rpc/src/lib.rs @@ -231,8 +231,9 @@ pub struct BeaconApiHandles { /// makes block production, attestation data and aggregation answer 503: /// with nothing validating payloads, none has a block it may vouch for. pub engine: Option, - /// The chain-event bus `/eth/v1/events` streams from. Read only there: - /// the chain actor is its only publisher, so no endpoint here writes to it. + /// The chain-event bus `/eth/v1/events` streams from. Besides the chain + /// actor and p2p, the submission endpoints publish to it: what they accept + /// is an event of its own (see `beacon::emit_accepted`). pub events: EventBus, /// This build as the Engine API identifies it. Block production appends /// its code and commit, with the execution client's, to every block's diff --git a/docs/rpc.md b/docs/rpc.md index bdcc260ec..297d08f0a 100644 --- a/docs/rpc.md +++ b/docs/rpc.md @@ -872,20 +872,40 @@ these are ever sent: | Event | Payload | Emitted when | |-------|---------|--------------| -| `head` | `{"slot":"10", "block":"0x…", "state":"0x…", "epoch_transition":false, "previous_duty_dependent_root":"0x…", "current_duty_dependent_root":"0x…", "execution_optimistic":false}` | The head changed, and is within 32 slots of the wall clock | -| `block` | `{"slot":"10", "block":"0x…", "execution_optimistic":false}` | A block is imported | +| `head` | `{"slot":"10", "block":"0x…", "state":"0x…", "epoch_transition":false, "previous_duty_dependent_root":"0x…", "current_duty_dependent_root":"0x…", "execution_optimistic":false}` | The head changed, and is within 32 slots of the wall clock. Deprecated by the specification in favour of `head_v2`, and still emitted | +| `head_v2` | `{"version":"gloas", "data":{"slot":"10", "block":"0x…", "state":"0x…", "payload_status":"full", "epoch_transition":false, "current_epoch_dependent_root":"0x…", "next_epoch_dependent_root":"0x…", "execution_optimistic":false}}` | Right after every `head` (before gloas too, with `payload_status` `full`), and a second time for the same block when its payload status goes from empty to full. `version` is the head state's fork | +| `block` | `{"slot":"10", "block":"0x…", "execution_optimistic":false}`; from gloas on also `"builder_index":"42", "block_hash":"0x…"` (the block's bid's), absent before | A block is imported | | `block_gossip` | `{"slot":"10", "block":"0x…"}` | A block passed gossip validation, or was published through `POST /eth/v2/beacon/blocks` | | `finalized_checkpoint` | `{"block":"0x…", "state":"0x…", "epoch":"2", "execution_optimistic":false}` | The finalized checkpoint advanced | | `chain_reorg` | `{"slot":"200", "depth":"50", "old_head_block":"0x…", "new_head_block":"0x…", "old_head_state":"0x…", "new_head_state":"0x…", "epoch":"2", "execution_optimistic":false}` | The new head does not descend from the previous one | | `attestation` | the aggregate's `Attestation` | An aggregate passed the `beacon_aggregate_and_proof` rules: accepted from gossip, or submitted through `POST /eth/v2/validator/aggregate_and_proofs` | | `data_column_sidecar` | `{"block_root":"0x…", "index":"1", "slot":"1"}` | A column this node custodies is stored, from any source | +| `execution_payload` | `{"slot":"10", "builder_index":"42", "block_hash":"0x…", "block_root":"0x…", "execution_optimistic":false}` | The chain actor imported a gloas envelope (from gossip, req/resp or `POST /eth/v1/beacon/execution_payload_envelopes`) | +| `execution_payload_gossip` | `{"slot":"10", "builder_index":"42", "block_hash":"0x…", "block_root":"0x…"}` | An envelope passed the `execution_payload` gossip rules, or was accepted by `POST /eth/v1/beacon/execution_payload_envelopes` | +| `execution_payload_available` | `{"slot":"10", "block_root":"0x…"}` | Right after `execution_payload` (see below) | +| `execution_payload_bid` | `{"version":"gloas", "data":}` | A bid passed the `execution_payload_bid` gossip rules, or was accepted by `POST /eth/v1/beacon/execution_payload_bids` | +| `payload_attestation_message` | `{"version":"gloas", "data":}` | A message passed the `payload_attestation_message` gossip rules, or was accepted by `POST /eth/v1/beacon/pool/payload_attestations` | +| `proposer_preferences` | `{"version":"gloas", "data":}` | Preferences passed the `proposer_preferences` gossip rules, or were accepted by `POST /eth/v1/validator/proposer_preferences` | Never emitted: `single_attestation` (subnet votes are verified and relayed but never reach the chain actor), the operation topics (`voluntary_exit`, `proposer_slashing`, `attester_slashing`, `bls_to_execution_change`, `contribution_and_proof`: no operation pools), `payload_attributes`, the light -client topics, `blob_sidecar`, and `head_v2` with every other gloas topic (this -build stops at fulu). +client topics, `blob_sidecar`, and `fast_confirmation` (no producer). + +The gloas topics are emitted by whichever part of the node accepted the object, +since several never reach the chain actor: the actor publishes `execution_payload` +and `execution_payload_available`; p2p publishes the gossip-accepted +`execution_payload_gossip`, `execution_payload_bid`, +`payload_attestation_message` and `proposer_preferences`; and the API handlers +publish what a validator client or builder submits. + +**`execution_payload_available` is not a separate moment.** The specification +asks for it when the payload and its blobs are verified available. The chain +actor imports an envelope only once both hold (the payload verified, and every +sampled column of a bid with commitments stored), so `execution_payload` and +`execution_payload_available` are emitted back to back from the same import, +`execution_payload` first. How head and finality events behave: @@ -900,7 +920,7 @@ How head and finality events behave: - **Stale heads are not reported.** A head more than 32 slots behind the wall clock (catch-up) sends no `head`; `block` events still show progress. `chain_reorg` and `finalized_checkpoint` are not gated. -- **Order:** each `block` as it imports, then `chain_reorg`, `head`, +- **Order:** each `block` as it imports, then `chain_reorg`, `head`, `head_v2`, `finalized_checkpoint`. - `epoch_transition` is whether the head's epoch is later than the previous head's. The dependent roots follow the specification, with the genesis block