feat(events): add attestation and aggregate chain events

[MegaRedHand]

Continues the chain-event pub-sub series (#516 / #517 / #518, all merged). Adds the two high-rate topics.

Events added

Topic	Payload	Emitted when
attestation	{ validator_id, data: AttestationData }	A single validator vote passes gossip validation (signature omitted)
aggregate	{ participants: [u64], data: AttestationData }	A committee-signature aggregate is produced locally or accepted from gossip (proof omitted)

Both fire only for messages the store accepted (data + signature validation), so subscribers see the same votes fork choice does. A node never receives its own aggregate back over gossip, so aggregate fires at two sites: the locally produced AggregateProduced path and the gossip-received path.

Design points

• The ~3 KB XMSS signature and the SNARK proof bytes are deliberately omitted — too heavy for a high-rate stream.
• The shared broadcast channel capacity is bumped 256 → 8192. All topics share one ring buffer, so a subscriber's tolerable stall is capacity / total_event_rate (dominated by the attestation rate), not per-topic. A per-topic split behind the EventBus facade remains the escape hatch if real devnet rates show the shared window biting — see the note in docs/rpc.md.
• Emission uses the plain, no-subscriber-guarded emit. Consequence: the per-vote AttestationData is cloned even when nobody is subscribed (the guard only skips the send). If that actor-hot-path cost proves material, a payload-deferral helper can be added in a follow-up.

Testing

cargo fmt, cargo clippy -p ethlambda-blockchain -p ethlambda-rpc (clean), blockchain + rpc lib tests pass, including events_streams_attestation_with_nested_data (verifies the untagged data: nesting over the wire).

---

Independent PR, based off main. One of three sibling PRs continuing the series — alongside #533 (chain_reorg/safe_target) and #535 (block_gossip), each independently based off main. They touch overlapping regions of events.rs / docs/rpc.md, so whichever two merge later will each need a small conflict rebase.

[github-actions]

🤖 Codex Code Review

1. Replayed valid gossip now emits fresh SSE events even when nothing in local state changed. In crates/blockchain/src/lib.rs and crates/blockchain/src/lib.rs, emission is gated only on is_ok(). But the store paths are idempotent for duplicates: raw attestation inserts overwrite the same (validator_id, data_root) entry (crates/storage/src/store.rs, crates/storage/src/store.rs), and aggregate inserts silently skip equal/subset proofs (crates/storage/src/store.rs, crates/storage/src/store.rs). Result: a peer can replay the same valid attestation/aggregate and continuously fill the SSE bus without changing fork choice, which is both a correctness issue for consumers (“new event” no longer means new information) and an easy DoS vector for the shared channel.

2. The new high-rate topics materially reduce reliability for existing low-rate subscribers because filtering happens after one shared broadcast queue. The code explicitly keeps one global ring buffer (crates/blockchain/src/events.rs) and per-client filtering is applied only after recv() (crates/net/rpc/src/events.rs, crates/net/rpc/src/events.rs). With attestation emitted for every accepted gossip vote (crates/blockchain/src/lib.rs), a client subscribed only to head or finalized_checkpoint still lags out at the full attestation rate. The docs acknowledge this (docs/rpc.md), but it is still a real behavioral regression for the API: under load, the low-rate consensus events become much less observable.

No consensus-state-transition, fork-choice, XMSS verification, or SSZ logic changed in this PR; my concerns are confined to the new event-stream behavior.

I couldn’t run cargo test here because cargo attempted to update rustup under a read-only home directory.

---

Automated review by OpenAI Codex · gpt-5.4 · custom prompt

[greptile-apps]

Greptile Summary

This PR adds high-rate attestation and aggregate events to the chain event stream. The main changes are:

• New attestation and aggregate topics and payloads.
• Event emission for validated gossip and locally produced aggregates.
• A larger shared broadcast channel for the increased event rate.
• SSE payload coverage and updated RPC documentation.

Confidence Score: 5/5

This looks safe to merge, with a scaling concern for large aggregates and many SSE subscribers.

• Topic parsing, filtering, and payload serialization are consistent.
• Event emission follows successful validation or local application.
• Per-subscriber serialization can increase lag under heavy aggregate traffic.

crates/blockchain/src/events.rs

Important Files Changed

Filename	Overview
crates/blockchain/src/events.rs	Adds both event topics, payload variants, topic mappings, tests, and a larger shared channel.
crates/blockchain/src/lib.rs	Emits events after successful gossip validation and after applying locally produced aggregates.
crates/net/rpc/src/events.rs	Adds SSE coverage for the nested attestation payload.
docs/rpc.md	Documents the new topics, payloads, and shared-channel lag behavior.

Prompt To Fix All With AI

Fix the following 1 code review issue. Work through them one at a time, proposing concise fixes.

---

### Issue 1 of 1
crates/blockchain/src/events.rs:111-114
**Large Aggregates Multiply Serialization Work**

An aggregate can contain up to 4096 participant IDs, producing a large JSON frame that each SSE subscriber serializes independently. With many subscribers or a burst of aggregates, those tasks can fall behind the shared channel and lose unrelated low-rate events such as `head`; consider serializing each event once or using per-topic channels.

_{Reviews (1): Last reviewed commit: "feat(events): add attestation and aggreg..." | Re-trigger Greptile}

[github-actions]

🤖 Claude Code Review

I have what I need to write the review.

Review: PR 534 — add attestation and aggregate chain events

Overall this is a clean, well-scoped extension of the existing event-bus pattern. The gating logic (accepted only fires the event after the store has validated data + verified signature/proof) is correct, the #[serde(untagged)] shape stays unambiguous since the two new variants have distinct field sets, and the ?topics= filtering path in crates/net/rpc/src/events.rs required no changes to support the new topics — that's a good sign the original abstraction was designed well.

One notable finding and a few minor nits below.

1. PR description references an emit_with API that isn't in the diff

The PR body states:

EventBus::emit_with(impl FnOnce() -> ChainEvent) defers payload construction (the AttestationData clone) behind the no-subscriber guard, so the per-vote attestation emitter costs nothing on an idle node.

No emit_with exists anywhere in crates/blockchain/src/events.rs — only the pre-existing emit(&self, event: ChainEvent) (events.rs:166), which takes an already-constructed ChainEvent. The call sites in crates/blockchain/src/lib.rs:1215-1218 and :1235-1236 build the full event (including the AttestationData/Vec<u64> clone) before calling emit, so the "no-subscriber guard" only skips the broadcast send, not the construction/clone.

In practice the impact is small: AttestationData is a plain fixed-size struct (three Checkpoints + a u64, no heap allocation), so the clone is a cheap stack copy, not the "3 KB signature"-scale cost the surrounding comments seem to guard against. But the aggregate path does allocate a Vec<u64> (attestation.proof.participant_indices().collect(), lib.rs:1227) unconditionally per gossiped aggregate, regardless of subscriber count — again low-rate so probably fine, but it doesn't match the "deferred behind the guard" description either.

This looks like the PR description/code comments describe a design that was planned but not implemented, or is stale from an earlier iteration. Worth reconciling: either implement emit_with as described, or update the description and code comments (lib.rs:1210-1213, 1223-1226) to stop claiming the construction is guarded — right now the comment "emit's own guard drops the event on a node with no subscribers" is misleading since the expensive-looking part (the clone) already happened.

2. Minor: duplicate AttestationData clone in AggregateProduced handler

crates/blockchain/src/lib.rs:1400-1409 clones msg.output.hashed.data() twice — once for the event payload, once for the SignedAggregatedAttestation sent to p2p. Trivial cost given the type is cheap to clone, but could be a single let data = msg.output.hashed.data().clone(); reused in both places for readability.

3. Minor: redundant participant_indices() computation

crates/blockchain/src/lib.rs:1227 computes attestation.proof.participant_indices().collect() in the actor before calling the store, and on_gossip_aggregated_attestation_core (store.rs:488) independently recomputes the same thing from the same proof. Necessary given the store takes ownership of attestation, and is already explained in the comment — just flagging it's duplicated work on every gossiped aggregate (low-rate, so not a real concern).

Everything else checks out

• Topic/FromStr/as_str round-trip and the exhaustive ALL_TOPICS test correctly cover both new variants.
• New tests (emit_with_skips_closure_without_subscribers... — actually named without emit_with; double check this name still matches an existing helper, not a new one) — worth confirming the test names referenced in the PR body match what's actually in the diff, since emit_with_skips_closure_without_subscribers and emit_with_delivers_when_subscribed aren't in the diff shown (only events_streams_attestation_with_nested_data is). Same stale-description concern as Item 1.
• Channel capacity bump 256 → 8192 is applied correctly via EventBus::default() (confirmed main.rs:242 uses default(), not a hardcoded capacity), and the memory cost is negligible (~8192 × ~160 bytes ≈ 1.3 MB).
• Docs (docs/rpc.md) are updated consistently with the new topics and honestly caveat the shared-ring-buffer lag-window tradeoff.

---

Automated review by Claude (Anthropic) · sonnet · custom prompt