diff --git a/crates/blockchain/src/events.rs b/crates/blockchain/src/events.rs index 66386ffd..1d7f9fb3 100644 --- a/crates/blockchain/src/events.rs +++ b/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, + 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); diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index 73d3a9cd..c3468b11 100644 --- a/crates/blockchain/src/lib.rs +++ b/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 = 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 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(), diff --git a/crates/net/rpc/src/events.rs b/crates/net/rpc/src/events.rs index 69482239..6d4bf9a3 100644 --- a/crates/net/rpc/src/events.rs +++ b/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); diff --git a/docs/rpc.md b/docs/rpc.md index f560692d..2c540002 100644 --- a/docs/rpc.md +++ b/docs/rpc.md @@ -86,9 +86,9 @@ 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 | |-------|---------|--------------| @@ -96,6 +96,8 @@ Payload fields mirror the Ethereum beacon-API eventstream: `block` is the block | `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`