feat(events): add attestation and aggregate chain events

crates/blockchain/src/events.rs

@@ -11,6 +11,7 @@
 
 use ethlambda_storage::Store;
 use ethlambda_types::ShortRoot;
+use ethlambda_types::attestation::AttestationData;
 use ethlambda_types::checkpoint::Checkpoint;
 use ethlambda_types::primitives::H256;
 use serde::Serialize;
@@ -22,16 +23,20 @@ use tracing::warn;
 ///
 /// These are the names consumers address events by (the SSE `event:` line and,
 /// later, `?topics=` filtering), kept separate from [`ChainEvent`] so the
-/// payload stays flat with the topic travelling out-of-band. The names match
-/// the Ethereum beacon-API eventstream topics (`head`, `block`,
-/// `finalized_checkpoint`); `justified_checkpoint` is an ethlambda extension
-/// with no beacon analog.
+/// payload stays flat with the topic travelling out-of-band. Names match the
+/// Ethereum beacon-API eventstream topics where an analog exists (`head`,
+/// `block`, `finalized_checkpoint`, `attestation`); `justified_checkpoint` and
+/// `aggregate` are ethlambda extensions with no direct beacon topic.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum Topic {
     Head,
     Block,
     JustifiedCheckpoint,
     FinalizedCheckpoint,
+    /// A single validator vote seen on gossip.
+    Attestation,
+    /// A committee-signature aggregate (produced locally or seen on gossip).
+    Aggregate,
 }
 
 impl Topic {
@@ -41,6 +46,8 @@ impl Topic {
             Topic::Block => "block",
             Topic::JustifiedCheckpoint => "justified_checkpoint",
             Topic::FinalizedCheckpoint => "finalized_checkpoint",
+            Topic::Attestation => "attestation",
+            Topic::Aggregate => "aggregate",
         }
     }
 }
@@ -60,6 +67,8 @@ impl FromStr for Topic {
             "block" => Ok(Topic::Block),
             "justified_checkpoint" => Ok(Topic::JustifiedCheckpoint),
             "finalized_checkpoint" => Ok(Topic::FinalizedCheckpoint),
+            "attestation" => Ok(Topic::Attestation),
+            "aggregate" => Ok(Topic::Aggregate),
             other => Err(UnknownTopic(other.to_string())),
         }
     }
@@ -90,6 +99,19 @@ pub enum ChainEvent {
     JustifiedCheckpoint { slot: u64, block: H256, state: H256 },
     /// The finalized checkpoint advanced.
     FinalizedCheckpoint { slot: u64, block: H256, state: H256 },
+    /// A single validator vote seen on gossip. Carries the vote's
+    /// [`AttestationData`] and the attester's validator id; the ~3 KB XMSS
+    /// signature is deliberately omitted (too heavy for a high-rate stream).
+    Attestation {
+        validator_id: u64,
+        data: AttestationData,
+    },
+    /// A committee-signature aggregate: its [`AttestationData`] and the
+    /// participating validator ids. The SNARK proof bytes are omitted.
+    Aggregate {
+        participants: Vec<u64>,
+        data: AttestationData,
+    },
 }
 
 impl ChainEvent {
@@ -99,6 +121,8 @@ impl ChainEvent {
             ChainEvent::Block { .. } => Topic::Block,
             ChainEvent::JustifiedCheckpoint { .. } => Topic::JustifiedCheckpoint,
             ChainEvent::FinalizedCheckpoint { .. } => Topic::FinalizedCheckpoint,
+            ChainEvent::Attestation { .. } => Topic::Attestation,
+            ChainEvent::Aggregate { .. } => Topic::Aggregate,
         }
     }
 }
@@ -108,7 +132,14 @@ impl ChainEvent {
 /// Chosen so a briefly-stalled subscriber is skipped past (lagged) rather than
 /// back-pressuring the actor. Lagged subscribers re-sync via the blocks
 /// endpoints.
-const CHAIN_EVENT_CHANNEL_CAPACITY: usize = 256;
+///
+/// All topics share this one ring buffer, so the high-rate `attestation` and
+/// `aggregate` events (roughly one per validator per slot) dominate its
+/// occupancy: the window a slow subscriber can tolerate is
+/// `capacity / total_event_rate`, not per-topic. Sized for a few minutes of
+/// history at devnet validator counts; a per-topic split behind the
+/// [`EventBus`] facade is the escape hatch if the shared window bites.
+const CHAIN_EVENT_CHANNEL_CAPACITY: usize = 8192;
 
 /// Cloneable handle to the chain-event broadcast channel.
 ///
@@ -297,13 +328,24 @@ mod tests {
         }
     }
 
-    const ALL_TOPICS: [Topic; 4] = [
+    const ALL_TOPICS: [Topic; 6] = [
         Topic::Head,
         Topic::Block,
         Topic::JustifiedCheckpoint,
         Topic::FinalizedCheckpoint,
+        Topic::Attestation,
+        Topic::Aggregate,
     ];
 
+    fn test_attestation_data(slot: u64) -> AttestationData {
+        AttestationData {
+            slot,
+            head: Checkpoint::default(),
+            target: Checkpoint::default(),
+            source: Checkpoint::default(),
+        }
+    }
+
     #[tokio::test]
     async fn subscriber_receives_emitted_event() {
         let bus = EventBus::default();
@@ -356,6 +398,20 @@ mod tests {
                 },
                 Topic::FinalizedCheckpoint,
             ),
+            (
+                ChainEvent::Attestation {
+                    validator_id: 0,
+                    data: test_attestation_data(1),
+                },
+                Topic::Attestation,
+            ),
+            (
+                ChainEvent::Aggregate {
+                    participants: vec![0, 1],
+                    data: test_attestation_data(1),
+                },
+                Topic::Aggregate,
+            ),
         ];
         for (event, topic) in cases {
             assert_eq!(event.topic(), topic);

crates/blockchain/src/lib.rs

@@ -1203,13 +1203,38 @@ impl BlockChainServer {
         // if the admin API just toggled, the first gossip after the toggle
         // should already use the new value.
         let is_aggregator = self.aggregator.is_enabled();
-        let _ = store::on_gossip_attestation(&mut self.store, attestation, is_aggregator)
-            .inspect_err(|err| warn!(%err, "Failed to process gossiped attestation"));
+        let accepted = store::on_gossip_attestation(&mut self.store, attestation, is_aggregator)
+            .inspect_err(|err| warn!(%err, "Failed to process gossiped attestation"))
+            .is_ok();
+
+        // Surface only votes that passed data validation and signature
+        // verification, so subscribers see the same attestations fork choice
+        // does. The ~3 KB XMSS signature is not carried. `emit`'s own guard
+        // drops the event on a node with no subscribers.
+        if accepted {
+            self.events.emit(ChainEvent::Attestation {
+                validator_id: attestation.validator_id,
+                data: attestation.data.clone(),
+            });
+        }
     }
 
     fn on_gossip_aggregated_attestation(&mut self, attestation: SignedAggregatedAttestation) {
-        let _ = store::on_gossip_aggregated_attestation(&mut self.store, attestation)
-            .inspect_err(|err| warn!(%err, "Failed to process gossiped aggregated attestation"));
+        // The store consumes the aggregate, so snapshot the event inputs first.
+        // Aggregates are low-rate (~one per subnet per slot), so building these
+        // unconditionally is cheap; `emit`'s own guard drops them on an
+        // unsubscribed node. The SNARK proof bytes are not carried.
+        let participants: Vec<u64> = attestation.proof.participant_indices().collect();
+        let data = attestation.data.clone();
+        let accepted = store::on_gossip_aggregated_attestation(&mut self.store, attestation)
+            .inspect_err(|err| warn!(%err, "Failed to process gossiped aggregated attestation"))
+            .is_ok();
+
+        // Emit only for aggregates the store accepted, mirroring `attestation`.
+        if accepted {
+            self.events
+                .emit(ChainEvent::Aggregate { participants, data });
+        }
     }
 
     fn update_sync_status(&mut self, current_slot: u64) {
@@ -1369,6 +1394,14 @@ impl Handler<AggregateProduced> for BlockChainServer {
         // the aggregate is safe to apply and gossip immediately.
         aggregation::apply_aggregated_group(&mut self.store, &msg.output);
 
+        // Surface our own freshly produced aggregate, the counterpart of the
+        // gossip-received path in `on_gossip_aggregated_attestation` (we never
+        // receive our own aggregate back over gossip). Low-rate; proof omitted.
+        self.events.emit(ChainEvent::Aggregate {
+            participants: msg.output.participants.clone(),
+            data: msg.output.hashed.data().clone(),
+        });
+
         if let Some(ref p2p) = self.p2p {
             let aggregate = SignedAggregatedAttestation {
                 data: msg.output.hashed.data().clone(),

crates/net/rpc/src/events.rs

@@ -255,6 +255,48 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn events_streams_attestation_with_nested_data() {
+        use ethlambda_types::attestation::AttestationData;
+        use ethlambda_types::checkpoint::Checkpoint;
+
+        let events = EventBus::new(16);
+
+        let resp = events_response(&events, "/lean/v0/events?topics=attestation").await;
+        assert_eq!(resp.status(), StatusCode::OK);
+
+        events.emit(ChainEvent::Attestation {
+            validator_id: 7,
+            data: AttestationData {
+                slot: 12,
+                head: Checkpoint::default(),
+                target: Checkpoint::default(),
+                source: Checkpoint::default(),
+            },
+        });
+
+        let text = first_frame(resp).await;
+
+        assert!(
+            text.contains("event:attestation") || text.contains("event: attestation"),
+            "missing attestation event name in frame: {text}"
+        );
+        // `validator_id` at top level and the vote nested under `data` (with the
+        // beacon-style checkpoint fields), confirming the untagged enum keeps
+        // this variant's shape flat-but-nested rather than tag-wrapped.
+        assert!(
+            text.contains("\"validator_id\":7"),
+            "missing top-level validator_id: {text}"
+        );
+        assert!(
+            text.contains("\"data\":")
+                && text.contains("\"head\":")
+                && text.contains("\"target\":")
+                && text.contains("\"source\":"),
+            "missing nested attestation data fields: {text}"
+        );
+    }
+
     #[tokio::test]
     async fn events_unknown_topic_returns_400() {
         let events = EventBus::new(16);

docs/rpc.md

@@ -86,16 +86,18 @@ SSZ-encoded `SignedBlock` at the latest finalized checkpoint. The genesis/anchor
 
 ### `GET /lean/v0/events`
 
-Server-Sent Events stream (`Content-Type: text/event-stream`) of live chain events published by the blockchain actor. Four event types:
+Server-Sent Events stream (`Content-Type: text/event-stream`) of live chain events published by the blockchain actor. Six event types:
 
-Payload fields mirror the Ethereum beacon-API eventstream: `block` is the block root, `state` the state root, and `slot` stands in for the beacon `epoch`.
+Payload fields mirror the Ethereum beacon-API eventstream where an analog exists: `block` is the block root, `state` the state root, and `slot` stands in for the beacon `epoch`. `justified_checkpoint` and `aggregate` are ethlambda extensions with no beacon topic.
 
 | Event | Payload | Emitted when |
 |-------|---------|--------------|
 | `head` | `{ "slot": 128, "block": "0x…", "state": "0x…" }` | Fork choice selects a new head within `HEAD_EVENT_RECENCY_SLOTS` (32 slots) of the wall clock; no head events fire during catch-up |
 | `block` | `{ "slot": 128, "block": "0x…" }` | A block is imported into the store |
 | `justified_checkpoint` | `{ "slot": 120, "block": "0x…", "state": "0x…" }` | The justified checkpoint advances |
 | `finalized_checkpoint` | `{ "slot": 96, "block": "0x…", "state": "0x…" }` | The finalized checkpoint advances |
+| `attestation` | `{ "validator_id": 4, "data": { "slot": 128, "head": {…}, "target": {…}, "source": {…} } }` | A single validator vote passes gossip validation (signature omitted) |
+| `aggregate` | `{ "participants": [0, 3, 4], "data": { "slot": 128, "head": {…}, "target": {…}, "source": {…} } }` | A committee-signature aggregate is produced locally or accepted from gossip (proof omitted) |
 
 The topic name travels only on the SSE `event:` line; the `data:` line carries the flat JSON payload. Example frame:
 
@@ -112,14 +114,16 @@ A **required** comma-separated list of event names selects which events to strea
 curl -N 'http://127.0.0.1:5052/lean/v0/events?topics=head,finalized_checkpoint'
 ```
 
-Valid values are exactly the event names above: `head`, `block`, `justified_checkpoint`, `finalized_checkpoint`. As in the Beacon API `eventstream` endpoint, `topics` is mandatory: there is no "subscribe to everything" default; list the topics you want.
+Valid values are exactly the event names above: `head`, `block`, `justified_checkpoint`, `finalized_checkpoint`, `attestation`, `aggregate`. As in the Beacon API `eventstream` endpoint, `topics` is mandatory: there is no "subscribe to everything" default; list the topics you want.
 
 | Status | Condition |
 |--------|-----------|
 | `200` | Stream opened for the listed topics |
 | `400` | `topics` is missing or empty, or any listed name is not a known topic (body names the offending value) |
 
-Events are fanned out over a bounded broadcast channel. A client that reads too slowly skips past the events it missed: they are dropped for that subscriber rather than back-pressured onto the actor, so treat the stream as best-effort and re-sync via the blocks endpoints after a gap. A client that falls behind receives an SSE comment line `: error - dropped N messages` marking the gap (wire-compatible with Lighthouse) before the stream continues; re-sync via the blocks endpoints rather than trusting the skipped range. Keep-alive comments are sent periodically to hold idle connections open.
+Events are fanned out over a single bounded broadcast channel shared by all topics. A client that reads too slowly skips past the events it missed: they are dropped for that subscriber rather than back-pressured onto the actor, so treat the stream as best-effort and re-sync via the blocks endpoints after a gap. A client that falls behind receives an SSE comment line `: error - dropped N messages` marking the gap (wire-compatible with Lighthouse) before the stream continues; re-sync via the blocks endpoints rather than trusting the skipped range. Keep-alive comments are sent periodically to hold idle connections open.
+
+Because the ring buffer is shared, the high-rate `attestation` events (roughly one per validator per slot) dominate its occupancy: a subscriber's tolerable stall is `capacity / total_event_rate`, not per-topic, so filtering with `?topics=` narrows what you receive but does **not** widen the lag window against an attestation flood. Subscribers that only need low-rate topics (`head`, `finalized_checkpoint`, …) are still evicted at the aggregate rate. If real usage shows this biting, the fix is a per-topic channel split behind the event bus (the `subscribe(TopicSet)` API is unaffected).
 
 ### `GET /lean/v0/blocks/{block_id}` and `/header`