feat(beacon): serve the Beacon API eventstream, GET /eth/v1/events - #626

🤖 Codex Code Review

I found 1 correctness issue worth fixing; otherwise the eventstream work looks solid.

Findings

• crates/net/rpc/src/beacon/pool.rs:313 — POST /eth/v2/validator/aggregate_and_proofs inserts the aggregate into the local attestation_pool before confirming publish_beacon_aggregate(...) succeeds. If the P2P actor is down, the request returns failure, but the node still keeps the aggregate for local block production. That creates an inconsistent partial-accept path: the aggregate can influence local block contents / fork choice visibility even though the API reported it “was not published.” Move insert_aggregate(inner) after a successful p2p.publish_beacon_aggregate(...), or explicitly treat “local accept, gossip failed” as success with a different response. This is the main bug.

Why it matters

• For consensus code, partial acceptance is risky because operator-visible failure no longer matches local state mutation.
• It also complicates debugging and can produce surprising behavior under transient actor failures.

What looks good

• crates/blockchain/src/lib.rs:2265 and crates/net/p2p/src/gossipsub/handler.rs:608 correctly route validator-submitted aggregates back into the chain actor so the node sees its own aggregate despite gossipsub self-suppression.
• crates/blockchain/src/events.rs:883 emits chain_reorg before head, and gates stale head events while still allowing reorg/finality events; that ordering and recency behavior look consistent with the docs.
• crates/net/rpc/src/beacon/events.rs:49 cleanly handles repeated and comma-separated topics, de-duplicates, and rejects non-Beacon topics.
• crates/net/rpc/src/events.rs:78 adding X-Accel-Buffering: no is a good operational improvement for SSE.

Validation note

• I couldn’t run cargo test here because the sandbox blocks rustup from creating temp files under /home/runner/.rustup/... during toolchain resolution.

If you want, I can turn this into a ready-to-post GitHub review comment set.

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

🤖 Claude Code Review

Review: PR 626 — Beacon API /eth/v1/events and chain-event plumbing

I read the diff against origin/beacon-chain-integration. I did not build it or run the tests. I went through the RPC, p2p, storage, API-trait and events.rs (non-test) changes closely. I only skimmed the blockchain/src/lib.rs diff, so I can't vouch for the actor wiring there.

The design holds up. No blocking issues found.

Good

• BlockAnnouncement is separate from BlockSource. This lets block_gossip mean "passed validation" on the beacon wire. Lean keeps announcing every gossip block. Only Outcome::Accept maps to Announce (verdict.rs), and a test pins all three verdict cases.
• EventView is a persistent view, and advance_view always advances it, even with no subscribers. Only the diff is skipped, so a later subscriber never sees a stale baseline. The expensive work (message_hash_tree_root, the attestation clone) sits behind has_subscribers().
• prune_beacon_optimistic_roots(slot, keep) fixes a real bug. A skipped epoch boundary leaves the finalized root below the slot bound, and pruning it would report execution_optimistic: false for an unvalidated block. The test mirrors the existing el_block_hashes one.
• Handing a validator client's own aggregate to the chain actor through publish_beacon_aggregate is correct, since gossipsub never loops a node's own messages back. The attesting_indices come from the same stateful_checks call that validated the aggregate, so there is no second validation path.
• /eth/v1/events parsing handles repeated and comma-separated topics, collapses duplicates, and returns the Beacon-shaped 400. Lean-only topics are refused. Topic::LEAN and Topic::BEACON keep the two surfaces from bleeding into each other.

Nits and minor points

1. crates/net/p2p/src/beacon/verdict.rs (around the new only_an_accepted_gossip_block_is_announced test): the doc comment of an_overloaded_aggregate_is_not_forwarded ("Ignore/Reject too… must never reach the chain actor") is now stacked above the new test's doc comment. It reads as one garbled comment on the wrong test. Move the old paragraph back down to sit directly on an_overloaded_aggregate_is_not_forwarded.
2. publish_beacon_head_change in events.rs (around line 901): if the head state is missing, it warns and returns. The view has already advanced, so that head transition and any reorg event are lost permanently rather than retried. The only trigger is a store inconsistency, so this is acceptable. A comment saying "dropped, not retried" would make the intent explicit.
3. publish_beacon_aggregate (gossipsub/handler.rs): the chain hand-off uses let _ = ….inspect_err(…). That matches the surrounding code. I'm only confirming the early-return path on a failed gossip publish also skips the hand-off. Hand-off after a failed publish would be wrong, because the API reports failure but fork choice would still apply the aggregate.
4. /eth/v1/events has no cap on concurrent subscribers, and each one gets a broadcast receiver. This matches the lean endpoint and isn't a regression. It's worth a follow-up if the beacon API port is exposed beyond localhost.
5. CLAUDE.md (working-tree modification): the "HTTP Servers" section still says the beacon arm "takes no … EventBus" and that the chain-events stream is part of the lean surface. This PR reverses that, so update the text.

The new tests look sufficient: framing with quoted integers, topic filtering, the lag comment, the lean 400 for beacon-only topics, and the optimistic-root retention.

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

🤖 Kimi Code Review

I'll review this PR which adds Beacon API eventstream support (/eth/v1/events) alongside the existing lean events, with significant refactoring of chain event publishing. Let me analyze the key areas.

Overall Assessment

This is a substantial, well-structured PR that unifies lean and beacon event publishing. The architecture is sound: a persistent EventView replaces snapshot-based diffing, solving missed-window bugs. However, I've identified several correctness issues, potential panics, and API concerns.

Critical Issues

1. Potential Division by Zero / Modulo by Zero in insert_event_block test helper

File: crates/blockchain/src/lib.rs
Line: ~6058 (in test helper insert_event_block)

inner.block_roots[below as usize % preset::SLOTS_PER_HISTORICAL_ROOT] = latest;

When slot == 0, the loop for below in 0..slot doesn't execute, so this is fine. But if SLOTS_PER_HISTORICAL_ROOT were ever 0, this would panic. Given it's a preset constant, this is low risk, but the pattern is worth noting.

More importantly: The test helper doesn't handle the case where slot >= SLOTS_PER_HISTORICAL_ROOT correctly for the block_roots array — it wraps via modulo, but a real state transition would have rotated this array. For tests with slots > 8192 (mainnet), this would be incorrect. The tests only use small slots, so this is acceptable for test code.

2. common_ancestor Can Panic on Malformed Store (Unwrap in Debug Build)

File: crates/blockchain/src/events.rs
Line: ~770-790

fn common_ancestor(store: &Store, a: H256, b: H256, floor_slot: u64) -> Option<(H256, u64)> {
    let mut a = (a, store.block_entry(&a)?);  // Returns None if missing — OK
    let mut b = (b, store.block_entry(&b)?);  // Returns None if missing — OK
    while a.0 != b.0 {
        let side = if a.1.0 >= b.1.0 { &mut a } else { &mut b };
        let (slot, parent) = side.1;
        if slot < floor_slot {
            return None;
        }
        *side = (parent, store.block_entry(&parent)?);  // Returns None if missing — OK
    }
    Some((a.0, a.1.0))
}

Actually, this looks correct — all ? propagate None. Let me re-check... Yes, this is fine. The block_entry returns Option<(u64, H256)> and all uses are with ?.

3. REAL ISSUE: advance_view Uses std::mem::replace Then Conditionally Returns

File: crates/blockchain/src/events.rs
Line: ~560-565

fn advance_view(&mut self, store: &Store) -> Option<EventView> {
    let previous = std::mem::replace(&mut self.view, EventView::capture(store));
    self.bus.has_subscribers().then_some(previous)
}

This is correct by design but subtle: the view is always advanced, even when there are no subscribers. The comment explains why — so a later subscriber doesn't see stale moves. This is good.

However, there's a potential issue: EventView::capture calls store.head(), store.latest_justified(), and store.latest_finalized(), all of which .expect("..."). In a test store or corrupted state, this could panic. The previous ChainEventSnapshot::capture had the same issue, so this isn't a regression.

4. ISSUE: BeaconHeadEvent Serialization — quoted_or_bare::serialize Used

File: crates/blockchain/src/events.rs
Line: ~295-310

pub struct BeaconHeadEvent {
    #[serde(serialize_with = "quoted_or_bare::serialize")]
    pub slot: u64,
    // ...
}

The quoted_or_bare module name suggests it can serialize either way, but for Beacon API events, integers must be quoted strings per the spec. The test at line ~1370 confirms "slot":"10" (quoted).

Question: Is quoted_or_bare::serialize actually quoting? The test passes, so yes. But the module name is confusing — it should perhaps be quoted or as_string. This is a minor naming issue, not a bug.

5. ISSUE: Topic::ALL Array Order Dependency for FromStr

File: crates/blockchain/src/events.rs
Line: ~85-115

pub const ALL: &[Topic] = &[
    Topic::Head,
    Topic::Block,
    // ...
];

And FromStr:

fn from_str(s: &str) -> Result<Self, Self::Err> {
    Topic::ALL
        .iter()
        .copied()
        .find(|topic| topic.as_str() == s)
        .ok_or_else(|| UnknownTopic(s.to_string()))
}

This is O(n) and relies on ALL being complete. The test every_topic_is_listed_once_and_named_uniquely (line ~1100) verifies uniqueness. However, if a developer adds a Topic variant but forgets to add it to ALL, from_str will fail for that name.

Suggestion: Add a compile-time check or at least a test that Topic::ALL contains all variants. The existing topic_from_str_inverts_as_str test only checks the variants in ALL, so a missing variant would be silently untested.

Actually, the test every_topic_is_listed_once_and_named_uniquely checks names.len() == Topic::ALL.len(), but if a variant is missing from both ALL and the test... The Topic enum derives Hash, so a HashSet of all variants via std::mem::variant_count (nightly) or a macro would be better. For now, this is acceptable but fragile.

6. ISSUE: BlockAnnouncement Not Used for Lean Blocks in handle_lean_block

File: crates/net/p2p/src/gossipsub/handler.rs
Line: ~166

blockchain
    .new_block(
        SignedBeaconBlock::Lean(signed_block),
        BlockSource::Gossip,
        arrival,
        BlockAnnouncement::Announce,  // Always Announce for lean
    )

Lean blocks always use BlockAnnouncement::Announce, which matches the documented behavior ("lean block from gossip, which validates nothing before import and has always announced every gossip block"). This is correct per the PR description, but note that lean has no BlockAnnouncement::Silent path for e.g. sync blocks. Looking at req_resp/handlers.rs:

// For lean blocks by range:
.new_block(block, BlockSource::Sync, BlockArrival::now(), BlockAnnouncement::Silent)

Good, sync blocks are silent. The lean behavior is preserved.

7. ISSUE: publish_beacon_aggregate Called Before Covered-Bits Check

File: crates/blockchain/src/lib.rs
Line: ~2270-2290

fn on_gossip_beacon_aggregate(...) {
    // Every aggregate that reaches here passed the gossip conditions, so
    // it is announced before the covered-bits and hold checks...
    self.events.publish_beacon_aggregate(&aggregate);

    if let Some(dropped) = self.beacon_aggregates.already_covered(&aggregate) {
        // ...returns early, but event was already emitted
        return;
    }
    // ...
}

This is intentional per the comment — the event announces aggregates that passed gossip validation, regardless of whether fork choice applies them. This matches the PR description: "An aggregate this node's own validator client submits reaches the chain actor too... That is also what puts this node's own aggregates into its fork choice."

However, this means duplicate aggregates (already covered) still emit events. The Beacon API spec's attestation event says "an attestation that has been validated" — if "validated" means "passed gossip conditions", this is correct. If it means "applied to fork choice", it's not. The PR documents this clearly, so this is a design choice, not a bug.

8. POTENTIAL BUG: DataColumnSidecarEvent Uses sidecar.index Directly

File: crates/blockchain/src/events.rs
Line: ~620-625

pub(crate) fn publish_data_column(&self, block_root: H256, sidecar: &fulu::DataColumnSidecar) {
    self.bus.emit(ChainEvent::DataColumnSidecar(DataColumnSidecarEvent {
        block_root,
        index: sidecar.index,  // u64 directly
        slot: sidecar.signed_block_header.message.slot,
    }));
}

And the struct:

pub struct DataColumnSidecarEvent {
    pub block_root: H256,
    #[serde(serialize_with = "quoted_or_bare::serialize")]
    pub index: u64,
    #[serde(serialize_with = "quoted_or_bare::serialize")]
    pub slot: u64,
}

The test expects "index": "1" (quoted). This is correct for Beacon API. But is sidecar.index the right type? DataColumnSidecar likely uses ColumnIndex which may be u64. This seems fine.

9. ISSUE: events Field Added to BeaconApiHandles but Not Documented in main.rs

File: bin/ethlambda/src/main.rs
Line: ~758

events: rpc_events,

This is passed through correctly. Looking at the BeaconApiHandles struct:

pub struct BeaconApiHandles {
    pub p2p: RpcToP2PRef,
    pub attestation_pool: Arc<Mutex<AttestationPool>>,
    pub engine: Option<ethlambda_engine::EngineClient>,
    pub events: EventBus,  // NEW
}

And in start_beacon_rpc_server:

.layer(Extension(handles.events));

This is correct. The events field is properly wired.

10. ISSUE: serde_json Added to blockchain Crate but Only Used in Tests

File: crates/blockchain/Cargo.toml
Line: ~42

serde_json.workspace = true

And in events.rs tests, serde_json is used extensively. This is a test-only dependency that isn't marked as such. Should be:

serde_json = { workspace = true, optional = true }  # or dev-dependency

Actually, looking more carefully — serde_json is used in #[cfg(test)] blocks. It should be a [dev-dependencies] entry, not a regular dependency. This bloats the release binary unnecessarily.

Wait — let me check if serde_json is used outside tests in events.rs... Scanning through... No, all serde_json uses are in #[cfg(test)] modules. This should be moved to [dev-dependencies].

11. ISSUE: EventBus::has_subscribers Race Condition

File: crates/blockchain/src/events.rs
Line: ~195-200

pub fn has_subscribers(&self) -> bool {
    self.tx.receiver_count() > 0
}

This is called to gate expensive payload construction, but there's a TOCTOU race: between has_subscribers() returning true and emit() being called, the last subscriber could drop. emit() handles this gracefully (returns Err which is ignored), so the worst case is wasted work. This is acceptable for a best-effort bus.

However, there's a reverse race: has_subscribers() returns false, we skip work, but a subscriber joins right after. They miss the event. This is by design ("a slow subscriber loses events") and documented.

12. ISSUE: publish_block_gossip for Beacon Uses block.message_hash_tree_root()

File: crates/blockchain/src/events.rs
Line: ~580-595

pub(crate) fn publish_block_gossip(...) {
    // ...
    let root = block.message_hash_tree_root();
    let event = match store.chain() {
        Chain::Lean => ChainEvent::BlockGossip { slot, block: root },
        Chain::Beacon => ChainEvent::BeaconBlockGossip(BeaconBlockGossipEvent { slot, block: root }),
    };
}

For beacon blocks, message_hash_tree_root() computes the hash tree root of the BeaconBlock message (without signature). This is correct — the block_gossip event should use the same root that gossip validation uses, which is the signed block's message root.

But wait — is message_hash_tree_root() the same as SignedBeaconBlock::message_hash_tree_root()? Let me check... The method is called on &SignedBeaconBlock, and it likely returns the hash tree root of the inner message. This is correct.

13. ISSUE: common_ancestor_slot for Beacon Doesn't Use NewChain::contains for Finalized Check

File: crates/blockchain/src/events.rs
Line: ~920-945

fn common_ancestor_slot(store: &Store, new_chain: &NewChain<'_>, previous: &EventView) -> u64 {
    let floor = previous.finalized.slot;
    let mut root = previous.head;
    while let Some((slot, parent)) = store.block_entry(&root) {
        if new_chain.contains(root, slot) {
            return slot;
        }
        if slot < floor {
            break;
        }
        root = parent;
    }
    // ...warn and return floor
}

This walks parent links from the old head. The new_chain.contains check uses get_block_root_at_slot on the new head's state. If the old head's chain goes below the finalized checkpoint without finding a common ancestor, it returns floor (the finalized slot).

Potential issue: What if store.block_entry(&root) returns None mid-walk? This happens if a block is missing. The loop breaks, warns, and returns floor. This is safe — the depth will be old_slot.saturating_sub(floor), which may overestimate but is bounded.

14. ISSUE: NewChain::dependent_root with epoch == 0

File: crates/blockchain/src/events.rs
Line: ~955-965

fn dependent_root(&self, epoch: Option<u64>) -> H256 {
    let slot = epoch.and_then(|epoch| compute_start_slot_at_epoch(epoch).checked_sub(1));
    match slot {
        Some(slot) => get_block_root_at_slot(self.state, slot).unwrap_or(self.root),
        None => self.genesis_root(),
    }
}

For epoch = Some(0): compute_start_slot_at_epoch(0) = 0, checked_sub(1) = None, so we call genesis_root(). Correct.

For epoch = None (underflow from epoch - 1): genesis_root(). Correct.

For epoch = Some(1): compute_start_slot_at_epoch(1) = 32 (mainnet), 32 - 1 = 31. get_block_root_at_slot(state, 31). If the head is at slot 32+, this should be in block_roots. Correct.

15. ISSUE: BeaconAttestationEvent::from Clones the Entire Attestation

File: crates/blockchain/src/events.rs
Line: ~345-355

impl From<&SignedAggregateAndProof> for BeaconAttestationEvent {
    fn from(aggregate: &SignedAggregateAndProof) -> Self {
        match aggregate {
            SignedAggregateAndProof::Phase0(signed) => {
                Self::Phase0(signed.message.aggregate.clone())
            }
            // ...
        }
    }
}

This clones the entire Attestation (including aggregation bits, signature, etc.). For electra, this includes committee_bits which is a 64-bit bitfield. The comment says "Boxed, since it is the one payload much larger than the rest". The Box is in ChainEvent::BeaconAttestation(Box<BeaconAttestationEvent>), but the clone happens before boxing. This is necessary for the event to own its data. Acceptable.

16. ISSUE: get_block_root_at_slot Can Fail for Very Old Slots

File: crates/blockchain/src/events.rs (implied in NewChain::contains)
Line: ~940

fn contains(&self, root: H256, slot: u64) -> bool {
    root == self.root
        || (slot < self.slot
            && get_block_root_at_slot(self.state, slot).is_ok_and(|on_chain| on_chain == root))
}

get_block_root_at_slot returns Err if the slot is too old (not in block_roots or historical_roots). For a reorg deeper than SLOTS_PER_HISTORICAL_ROOT (8192 slots), this would return false even if root is actually an ancestor.

However, the fallback is the parent-link walk in common_ancestor_slot, which would eventually reach the same answer (or the finalized checkpoint). So this is a performance optimization, not a correctness issue.

17. ISSUE: publish_beacon_changes Doesn't Check has_subscribers Before State Load

File: crates/blockchain/src/events.rs
Line: ~570-590

pub(crate) fn publish_beacon_changes(&mut self, store: &Store, wall_clock_slot: u64) {
    let Some(previous) = self.advance_view(store) else {
        return;  // No subscribers, early return
    };
    // ...expensive state load happens after this...

advance_view returns None when !has_subscribers(), so this is actually correct! The state load only happens when there are subscribers.

But wait — advance_view does EventView::capture(store) which calls store.head(), store.latest_justified(), store.latest_finalized(). These are cheap metadata reads. The expensive store.get_state(&head) happens in publish_beacon_head_change, which is only called when self.view.head != previous.head. And publish_beacon_head_change

Automated review by Kimi (Moonshot AI) · kimi-k2.6 · custom prompt