feat(rpc): emit single_attestation on /eth/v1/events

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<gloas::PayloadAttestationMessage>),
     /// Beacon `proposer_preferences`.
     ProposerPreferences(Box<Versioned<gloas::SignedProposerPreferences>>),
+    /// 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<electra::SingleAttestation>),
 }
 
 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);
+    }
 }

crates/net/api/src/lib.rs

@@ -562,6 +562,11 @@ pub enum GossipEvent {
     PayloadAttestationMessage(PayloadAttestationMessage),
     /// A `proposer_preferences` message passed gossip validation.
     ProposerPreferences(Box<SignedProposerPreferences>),
+    /// 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<SingleAttestation>),
 }
 
 /// Where `ethlambda-p2p` reports [`GossipEvent`]s.

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<Vec<GossipEvent>>,
+    }
+
+    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 {

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<Store>,
     Extension(p2p): Extension<RpcToP2PRef>,
+    events: Option<Extension<EventBus>>,
     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<EventBus>,
     ) -> (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();

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":<SignedExecutionPayloadBid>}` | 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":<PayloadAttestationMessage>}` | 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":<AttestationData>, "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":<SignedProposerPreferences>}` | 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