From 65fc2c91073c6c8c5442d7814720d3ebf356cd74 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 2 Nov 2026 15:24:12 -0300 Subject: [PATCH 1/2] feat(rpc): emit single_attestation for gossip-accepted subnet attestations ethPandaOps' contributoor subscribes to single_attestation and never to attestation, and ethlambda accepted that topic without ever sending it, so the lab showed no attestation data for our nodes. Subnet attestations are validated in ethlambda-p2p and never reach the chain actor, so p2p reports an accepted one through the GossipEventSink that #748 added, and the event bus maps it to a ChainEvent on Topic::SingleAttestation. It is the highest-rate event the node has (every attestation on its backbone subnets), so the attestation is cloned only when a client is subscribed, and the payload is boxed to keep the broadcast ring sized for the smaller variants. The payload is the existing electra SingleAttestation container, so its JSON form is the specification's. --- crates/blockchain/src/events.rs | 45 ++++++++++++++++++- crates/net/api/src/lib.rs | 5 +++ crates/net/p2p/src/beacon/verdict.rs | 64 ++++++++++++++++++++++++++-- docs/rpc.md | 6 +-- 4 files changed, 112 insertions(+), 8 deletions(-) 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/docs/rpc.md b/docs/rpc.md index 297d08f0a..4ec582b66 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. 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 From 6a621064e924c602d0e87334d7c34f55c598bcb0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 2 Nov 2026 15:25:31 -0300 Subject: [PATCH 2/2] feat(rpc): emit single_attestation for API-submitted attestations This is the specification's "(from P2P or API)" half of single_attestation: POST /eth/v2/beacon/pool/attestations now reports each attestation that passed validation and was handed to the network actor, the way the other API handlers report what a validator client or builder submits. It is a separate commit because emitting our own validators' attestations, with near-zero propagation time, can link validator indices to this node, which is why contributoor makes the topic opt-in. Dropping this commit leaves the gossip-accepted events alone; docs/spec_deviations.md would then need a line saying API submissions are not reported. --- crates/net/rpc/src/beacon/pool.rs | 55 +++++++++++++++++++++++++++++-- docs/rpc.md | 2 +- 2 files changed, 53 insertions(+), 4 deletions(-) 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 4ec582b66..63767a5e1 100644 --- a/docs/rpc.md +++ b/docs/rpc.md @@ -885,7 +885,7 @@ 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. This is the highest-rate topic (every attestation on the node's backbone subnets), so nothing is built unless a client is subscribed | +| `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: the operation topics (`voluntary_exit`,