diff --git a/crates/blockchain/src/events.rs b/crates/blockchain/src/events.rs index d42fbe5bc..5dc46fc60 100644 --- a/crates/blockchain/src/events.rs +++ b/crates/blockchain/src/events.rs @@ -79,8 +79,10 @@ pub enum Topic { /// The new head does not descend from the previous one. On both surfaces. ChainReorg, DataColumnSidecar, - // Accepted on the beacon surface; `docs/rpc.md` lists which are emitted. + /// Beacon: a subnet attestation that passed the + /// `beacon_attestation_{subnet_id}` rules. SingleAttestation, + // The rest are accepted on the beacon surface; `docs/rpc.md` lists which are emitted. VoluntaryExit, BlsToExecutionChange, ProposerSlashing, @@ -340,6 +342,11 @@ pub enum ChainEvent { PayloadAttestationMessage(Versioned), /// Beacon `proposer_preferences`. ProposerPreferences(Box>), + /// Beacon `single_attestation`: a subnet attestation passed the + /// `beacon_attestation_{subnet_id}` rules. Boxed for the reason + /// [`ChainEvent::BeaconAttestation`] is; it is also the highest-rate + /// topic, so the clone is only made when someone listens. + SingleAttestation(Box), } impl ChainEvent { @@ -363,6 +370,7 @@ impl ChainEvent { ChainEvent::ExecutionPayloadBid(_) => Topic::ExecutionPayloadBid, ChainEvent::PayloadAttestationMessage(_) => Topic::PayloadAttestationMessage, ChainEvent::ProposerPreferences(_) => Topic::ProposerPreferences, + ChainEvent::SingleAttestation(_) => Topic::SingleAttestation, } } } @@ -663,6 +671,9 @@ impl GossipEventSink for EventBus { GossipEvent::ProposerPreferences(preferences) => { ChainEvent::ProposerPreferences(Box::new(Versioned::gloas(*preferences))) } + GossipEvent::SingleAttestation(attestation) => { + ChainEvent::SingleAttestation(attestation) + } }; EventBus::emit(self, event); } @@ -2099,4 +2110,36 @@ mod tests { let event = ChainEvent::BeaconAttestation(Box::new(BeaconAttestationEvent::from(&signed))); assert_matches_example(event, Topic::Attestation, example); } + + /// The specification's `single_attestation` example, carried through the + /// event unchanged: quoted integers, `0x` hex, electra's field names. + #[test] + fn single_attestation_matches_the_specification_example() { + let example = r#"{"committee_index":"1", "attester_index":"1", "data":{"slot":"1", "index":"1", "beacon_block_root":"0xcf8e0d4e9587369b2301d0790347320302cc0943d5a1884560367e8208d920f2", "source":{"epoch":"1", "root":"0xcf8e0d4e9587369b2301d0790347320302cc0943d5a1884560367e8208d920f2"}, "target":{"epoch":"1", "root":"0xcf8e0d4e9587369b2301d0790347320302cc0943d5a1884560367e8208d920f2"}}, "signature":"0x1b66ac1fb663c9bc59509846d6ec05345bd908eda73e670af888da41af171505cc411d61252fb6cb3fa0017b679f8bb2305b26a285fa2737f175668d0dff91cc1b66ac1fb663c9bc59509846d6ec05345bd908eda73e670af888da41af171505"}"#; + let attestation: electra::SingleAttestation = serde_json::from_str(example).unwrap(); + let event = ChainEvent::SingleAttestation(Box::new(attestation)); + assert_matches_example(event, Topic::SingleAttestation, example); + } + + #[tokio::test] + async fn the_bus_turns_a_gossip_single_attestation_into_its_topic() { + let bus = EventBus::new(8); + let mut rx = bus.subscribe(); + let sink: &dyn GossipEventSink = &bus; + + let attestation = electra::SingleAttestation { + committee_index: 1, + attester_index: 2, + data: Default::default(), + signature: Default::default(), + }; + sink.emit(GossipEvent::SingleAttestation(Box::new( + attestation.clone(), + ))); + + let ChainEvent::SingleAttestation(received) = rx.try_recv().unwrap() else { + panic!("expected a single_attestation event"); + }; + assert_eq!(*received, attestation); + } } diff --git a/crates/net/api/src/lib.rs b/crates/net/api/src/lib.rs index e1d692459..47769766f 100644 --- a/crates/net/api/src/lib.rs +++ b/crates/net/api/src/lib.rs @@ -562,6 +562,11 @@ pub enum GossipEvent { PayloadAttestationMessage(PayloadAttestationMessage), /// A `proposer_preferences` message passed gossip validation. ProposerPreferences(Box), + /// A `beacon_attestation_{subnet_id}` single attestation passed gossip + /// validation (the `single_attestation` event). Boxed: it is the largest + /// payload here after the bids, and the event ring is sized for the + /// largest variant. + SingleAttestation(Box), } /// Where `ethlambda-p2p` reports [`GossipEvent`]s. diff --git a/crates/net/p2p/src/beacon/verdict.rs b/crates/net/p2p/src/beacon/verdict.rs index 8dae1f8c9..53ef8d82e 100644 --- a/crates/net/p2p/src/beacon/verdict.rs +++ b/crates/net/p2p/src/beacon/verdict.rs @@ -405,10 +405,10 @@ 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). +/// Report an accepted object on the Beacon API's event stream, for the topics +/// whose objects never reach the chain actor that publishes the rest (bids +/// and preferences stop here, a payload attestation message only ever feeds +/// fork choice, and a subnet attestation is relayed and nothing more). /// /// 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 @@ -434,6 +434,11 @@ fn publish_gossip_event(server: &P2PServer, object: &Validated) { Validated::PayloadAttestation(message) if sink.has_subscribers() => { GossipEvent::PayloadAttestationMessage(message.clone()) } + // The highest-rate event the node has (every attestation on its + // backbone subnets), so nothing is cloned for a node nobody listens to. + Validated::Attestation { attestation, .. } if sink.has_subscribers() => { + GossipEvent::SingleAttestation(attestation.clone()) + } _ => return, }; sink.emit(event); @@ -1150,6 +1155,57 @@ mod tests { assert!(chain.aggregates.lock().unwrap().is_empty()); } + /// A sink that keeps what p2p reported, with a switchable audience. + #[derive(Default)] + struct RecordingSink { + subscribed: bool, + events: std::sync::Mutex>, + } + + impl ethlambda_network_api::GossipEventSink for RecordingSink { + fn has_subscribers(&self) -> bool { + self.subscribed + } + + fn emit(&self, event: GossipEvent) { + self.events.lock().unwrap().push(event); + } + } + + /// A subnet attestation reaches the event sink on `Accept` alone, and only + /// when someone listens. + #[tokio::test] + async fn an_accepted_subnet_attestation_reaches_the_event_sink_only_when_listened_to() { + let mut server = unconnected_beacon_server(Config::mainnet(), 0).await; + let forward = |server: &P2PServer, outcome| { + Validated::Attestation { + attestation: Box::new(electra_attestation(5, 1)), + subnet_id: 0, + } + .forward(server, Instant::now(), outcome) + }; + + let silent = Arc::new(RecordingSink::default()); + server.gossip_events = Some(silent.clone()); + forward(&server, Outcome::Accept); + assert!(silent.events.lock().unwrap().is_empty()); + + let sink = Arc::new(RecordingSink { + subscribed: true, + ..Default::default() + }); + server.gossip_events = Some(sink.clone()); + forward(&server, Outcome::Ignore(IgnoreReason::Overloaded)); + forward(&server, Outcome::Reject(RejectReason::BadSignature)); + assert!(sink.events.lock().unwrap().is_empty()); + forward(&server, Outcome::Accept); + let events = sink.events.lock().unwrap(); + let [GossipEvent::SingleAttestation(attestation)] = events.as_slice() else { + panic!("expected exactly one single attestation event, got {events:?}"); + }; + assert_eq!(**attestation, electra_attestation(5, 1)); + } + /// What `forward` handed the chain actor, by method. #[derive(Debug, PartialEq)] enum Sent { diff --git a/crates/net/rpc/src/beacon/pool.rs b/crates/net/rpc/src/beacon/pool.rs index 7131b67f4..a9fd51522 100644 --- a/crates/net/rpc/src/beacon/pool.rs +++ b/crates/net/rpc/src/beacon/pool.rs @@ -25,8 +25,9 @@ use axum::{ response::{IntoResponse, Response}, routing::{get, post}, }; +use ethlambda_blockchain::EventBus; use ethlambda_engine::EngineClient; -use ethlambda_network_api::RpcToP2PRef; +use ethlambda_network_api::{GossipEvent, RpcToP2PRef}; use ethlambda_state_transition::beacon::{ bls, gossip::attestation::compute_subnet_for_attestation, @@ -52,7 +53,7 @@ use serde::{Deserialize, Serialize}; use tracing::{debug, warn}; use crate::beacon::{ - ApiError, decode_list, + ApiError, decode_list, emit_accepted, validator::{head, require_execution_client, require_validated}, }; @@ -91,6 +92,7 @@ pub(crate) struct Failure { async fn post_pool_attestations( State(store): State, Extension(p2p): Extension, + events: Option>, headers: HeaderMap, body: Bytes, ) -> Response { @@ -134,7 +136,14 @@ async fn post_pool_attestations( checked.committee_position, checked.committee_len, ); + let for_event = attestation.clone(); p2p.publish_beacon_attestation(checked.subnet_id, attestation) + .map(|()| { + // The `(from P2P or API)` half of `single_attestation`. + emit_accepted(events.as_deref(), || { + GossipEvent::SingleAttestation(Box::new(for_event)) + }) + }) .map_err(|_| "the network actor is not running") }); match published { @@ -659,11 +668,23 @@ mod tests { fixture: &Fixture, attestations: &[SingleAttestation], version: &str, + ) -> (StatusCode, serde_json::Value) { + submit_with_events(fixture, attestations, version, None).await + } + + async fn submit_with_events( + fixture: &Fixture, + attestations: &[SingleAttestation], + version: &str, + events: Option, ) -> (StatusCode, serde_json::Value) { let network: RpcToP2PRef = fixture.network.clone(); - let app = routes() + let mut app = routes() .with_state(fixture.store.clone()) .layer(Extension(network)); + if let Some(events) = events { + app = app.layer(Extension(events)); + } let request = Request::post("/eth/v2/beacon/pool/attestations") .header("content-type", "application/json") .header("eth-consensus-version", version) @@ -680,6 +701,34 @@ mod tests { (status, json) } + /// An accepted attestation goes out on the event stream next to its + /// publication; one that fails validation does not. + #[tokio::test] + async fn an_accepted_attestation_is_reported_on_the_event_stream() { + let fixture = fixture(); + let good = attestation(&fixture, 0, 0); + let mut bad = attestation(&fixture, 0, 1); + bad.signature = Default::default(); + let events = EventBus::new(8); + let mut stream = events.subscribe(); + + let (status, _) = submit_with_events( + &fixture, + &[bad, good.clone()], + fixture.version, + Some(events), + ) + .await; + assert_eq!(status, StatusCode::BAD_REQUEST); + + let event = stream.try_recv().expect("the accepted attestation"); + let ethlambda_blockchain::ChainEvent::SingleAttestation(reported) = event else { + panic!("expected a single_attestation event, got {event:?}"); + }; + assert_eq!(*reported, good); + assert!(stream.try_recv().is_err(), "the refused one stays off"); + } + #[tokio::test] async fn a_valid_attestation_is_published_on_its_subnet() { let fixture = fixture(); diff --git a/docs/rpc.md b/docs/rpc.md index 297d08f0a..63767a5e1 100644 --- a/docs/rpc.md +++ b/docs/rpc.md @@ -885,10 +885,10 @@ these are ever sent: | `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` | +| `single_attestation` | `{"committee_index":"1", "attester_index":"1", "data":, "signature":"0x…"}` | A subnet attestation passed the `beacon_attestation_{subnet_id}` gossip rules, or was accepted by `POST /eth/v2/beacon/pool/attestations`. This is the highest-rate topic (every attestation on the node's backbone subnets), so nothing is built unless a client is subscribed | | `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`, +Never emitted: 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 `fast_confirmation` (no producer). @@ -897,7 +897,7 @@ 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 +`payload_attestation_message`, `proposer_preferences` and `single_attestation`; and the API handlers publish what a validator client or builder submits. **`execution_payload_available` is not a separate moment.** The specification