feat(rpc): add ?topics= filtering to the events stream

[MegaRedHand]

Adds Beacon-API-style server-side filtering to GET /lean/v0/events: ?topics= takes comma-separated topic names, an unknown name returns 400, and a missing or empty parameter subscribes to all topics. The Beacon API makes topics required; defaulting to all is a documented divergence in docs/rpc.md.

Filtering lives in the event mechanism rather than the handler: EventBus::subscribe(TopicSet) returns an EventSubscription whose recv skips unmatched events (a TopicSet is a Copy bitmask, one AND per event), so non-SSE consumers get filtering for free and lag is still surfaced, never swallowed. The handler bridges recv to a stream via futures_util::stream::unfold, replacing the tokio-stream/futures-core pair with futures-util (already in the lock transitively).

Stacked on #517.

Has unit and handler tests (including the 400 path) and passes clippy with -D warnings.

[github-actions]

🤖 Kimi Code Review

The PR introduces topic-filtered SSE subscriptions with a bitmask-based filter (TopicSet). The implementation is clean and well-tested, but there are a few correctness and robustness issues to address.

Critical: Potential overflow in TopicSet

crates/blockchain/src/events.rs, line 87:

self.0 & (1 << topic as u16) != 0

If the Topic enum grows beyond 16 variants, 1 << topic as u16 will panic in debug builds or truncate in release. The same applies to from_iter (line 96).

Suggestion: Add a compile-time assertion or document the invariant:

impl TopicSet {
    // ...
    pub fn contains(self, topic: Topic) -> bool {
        let discriminant = topic as u16;
        debug_assert!(discriminant < 16, "Topic discriminant overflow");
        self.0 & (1 << discriminant) != 0
    }
}

Alternatively, use 1u16.wrapping_shl(topic as u32) and document that topics beyond bit 15 are ignored.

Minor: Whitespace handling in topic parsing

crates/net/rpc/src/events.rs, line 45:

list.split(',').map(Topic::from_str).collect()

This does not handle whitespace around commas. A request to ?topics=head, block will fail with "unknown topic: ' block'".

Suggestion: Trim whitespace:

list.split(',').map(str::trim).map(Topic::from_str).collect()

Minor: Unbounded string in error variant

crates/blockchain/src/events.rs, lines 42-50:
The UnknownTopic(String) error stores the raw input. While likely mitigated by HTTP query string limits, consider truncating excessively long strings in the Display impl to prevent accidental memory pressure from malicious "topic" parameters.

Code quality observations

1. Good: The switch from tokio-stream to futures_util::stream::unfold reduces dependencies while maintaining correctness.
2. Good: The SSE stream correctly handles RecvError::Lagged by logging and continuing, and RecvError::Closed by terminating the stream.
3. Good: The TopicSet bitset approach is efficient (Copy, no allocation) and appropriate for the expected small number of topics.

Documentation

The divergence from the Beacon API (optional vs mandatory topics parameter) is clearly documented in docs/rpc.md.

Summary
The PR is well-structured and the tests cover the new functionality. Address the potential 16-topic overflow limitation and consider trimming whitespace in topic parsing before merging.

---

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

[github-actions]

🤖 Codex Code Review

No findings.

The PR is narrowly scoped to SSE event delivery and topic parsing in crates/blockchain/src/events.rs and crates/net/rpc/src/events.rs. I did not see correctness, security, or performance regressions in the touched paths. The filtering remains outside consensus execution, so fork choice, attestation validation, justification/finalization, state transition, XMSS, and SSZ logic are unaffected by this change.

Verification gap: I could not run the Rust tests in this environment because cargo/rustup attempted to write under read-only /home/runner/.rustup and /home/runner/.cargo while resolving the pinned toolchain and git dependencies.

---

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

[greptile-apps]

Greptile Summary

This PR adds topic filtering to the chain-event SSE endpoint. The main changes are:

• Parses comma-separated topic names from the topics query parameter.
• Adds filtered subscriptions to the blockchain event bus.
• Returns a bad request response for unknown topics.
• Replaces the previous broadcast stream adapter with a futures-util stream.
• Documents the endpoint's filtering behavior and Beacon API divergence.

Confidence Score: 5/5

The latest changes look safe to merge.

• No blocking issue tied to an eligible earlier review thread was found.
• Filtering, invalid-topic handling, lag behavior, and the unfiltered default remain consistent across the changed layers.

Important Files Changed

Filename	Overview
crates/blockchain/src/events.rs	Adds topic parsing, topic bitsets, and filtered event subscriptions while preserving lag errors.
crates/net/rpc/src/events.rs	Adds query parsing, validation, filtered SSE streaming, and handler coverage.
crates/blockchain/src/lib.rs	Exports the new subscription types and updates existing tests to request all topics.
crates/net/rpc/Cargo.toml	Replaces the previous stream dependencies with futures-util.
docs/rpc.md	Documents accepted topics, response statuses, and unfiltered defaults.

_{Reviews (3): Last reviewed commit: "feat(rpc): add ?topics= filtering to the..." | Re-trigger Greptile}

[github-actions]

🤖 Claude Code Review

Not related — that's gossipsub topics, a separate concept. I have enough to write the review now.

Review: feat(rpc): add ?topics= filtering to the events stream (PR #518)

Overall this is a clean, well-scoped PR. Moving the filter into EventBus/EventSubscription rather than the SSE handler is the right call — it keeps the handler transport-only and gives every future consumer the same skip semantics. The TopicSet bitmask is a good fit (Copy, no allocation), the lag-vs-filter distinction is preserved correctly (Lagged is surfaced, non-matching topics are silently skipped), and the dependency swap (tokio-stream+futures-core → futures-util) is justified since futures-util was already pulled in transitively. Test coverage is good, including the 400 path and confirming a filtered-out event doesn't leak onto the wire.

A few points below, none of which are correctness bugs in the current 4-topic system, but worth a look before merge:

1. ?topics=head, / ?topics=head,,block produce a 400 for an empty segment (crates/net/rpc/src/events.rs:46) — list.split(',') yields "" for a trailing/doubled comma, and Topic::from_str("") fails with unknown topic: ''. This is safe (fails closed) but may surprise clients who build the query string by joining a list with trailing separators. Worth deciding if that's the desired behavior or if segments should be trimmed/filtered of empties before parsing. Not a blocker, just a UX nit worth confirming is intentional.

2. No whitespace trimming — ?topics=head, block (space after comma, a very typical way to write this) fails the second segment as unknown topic: ' block' rather than trimming to "block". Consider .map(str::trim) before Topic::from_str if you want to be lenient here, or leave as-is if strict parsing is the intended contract (worth a one-line doc note either way).

3. TopicSet(u16) bitmask caps at 16 topics (crates/blockchain/src/events.rs:75-83) — 1 << topic as u16 will panic in debug builds (shift-overflow) if Topic ever grows past 16 variants. Only 4 variants exist today so this is entirely theoretical, but since the doc comment already claims forward-compat ("including ones added later"), a debug_assert! or a comment noting the 16-topic ceiling would make the constraint explicit rather than silently latent.

4. Filtered subscribers still occupy the shared ring buffer for topics they don't want (crates/blockchain/src/events.rs:180-182) — correctly documented as a deliberate tradeoff (filter at receive-time, lag counted globally), so a narrow subscriber (e.g. ?topics=finalized_checkpoint) can be lagged out by high-frequency topics (e.g. block) it never asked for. This is a known and stated design choice, not a bug, but worth being aware of if a client wants a large event-boundary correctness guarantee, since it's stricter than it may appear from the endpoint name alone.

Nothing above blocks merging — items 3 and 4 are latent/documented design notes rather than defects, and items 1–2 are minor UX/strictness questions for the author to confirm are intentional.

---

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

[github-actions]

🤖 Kimi Code Review

The PR introduces topic-filtered SSE subscriptions with a bitmask-based filter (TopicSet). Code is generally correct, well-tested, and idiomatic.

Issues & Suggestions:

1. Silent bitmask overflow in TopicSet (crates/blockchain/src/events.rs:88-95)
   TopicSet uses u16 bit-shifting (1 << topic as u16). If Topic ever grows beyond 16 variants, shifts ≥16 wrap to 0 (due to as u16 cast), breaking contains() and from_iter().
   Fix: Add a static assertion or explicit check:

   const _: () = assert!(std::mem::variant_count::<Topic>() <= 16);

   Also update the docstring on TopicSet::ALL ("including ones added later") to clarify the 16-topic limit.

2. MSRV consideration (crates/net/rpc/src/events.rs:70)
   Uses Result::inspect_err, stabilized in Rust 1.76. If the project MSRV is lower, replace with:

   match frame {
       Ok(f) => return Some((Ok(f), subscription)),
       Err(err) => {
           tracing::warn!(%err, "Failed to serialize SSE chain event");
           continue;
       }
   }

3. Clone allocation in error path (crates/blockchain/src/events.rs:58)
   UnknownTopic(String) allocates on parse failure. Acceptable for the error path, but consider UnknownTopic(Box<str>) or &'static str if this becomes hot. Not critical.

4. Documentation nit (docs/rpc.md)
   The divergence note ("lean defaults a missing or empty parameter to all topics") is helpful. Consider explicitly stating that the filter is case-sensitive (e.g., Head ≠ head).

Security: No concerns. Input validation correctly rejects unknown topics with 400.

Performance: The design trades buffer space for simplicity (filtered events still occupy broadcast slots). This is acceptable per the documented "best-effort" contract, but ensure CHAIN_EVENT_CHANNEL_CAPACITY accounts for high-volume topics like block if clients frequently filter for rare topics like finalized_checkpoint.

Consensus Safety: No consensus logic touched; this is purely RPC/telemetry.

Verdict: Approvable once the 16-topic limit is documented or guarded (Item 1). The rest are minor nits.

---

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

[github-actions]

🤖 Codex Code Review

1. Low: TopicSet is documented as “every topic, including ones added later”, but the implementation is only safe up to 16 variants because it stores a u16 mask and shifts by topic as u16 in both contains and FromIterator. If Topic grows past bit 15, TopicSet::ALL stops matching the documented contract and the shift becomes invalid. Either widen the mask and add a guard/static assertion on topic count, or derive ALL from an explicit topic list instead of u16::MAX. Refs: crates/blockchain/src/events.rs, crates/blockchain/src/events.rs, crates/blockchain/src/events.rs

Otherwise this looks reasonable. The PR stays out of fork choice / STF / attestation / SSZ paths, and I did not see any immediate correctness or security regressions in the SSE filtering flow.

I couldn’t run the Rust tests in this environment because cargo/rustup needs write access outside the workspace (/home/runner/.rustup and /home/runner/.cargo).

---

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

[github-actions]

🤖 Claude Code Review

Review: feat(rpc): add ?topics= filtering to the events stream

This is a clean, well-scoped change. The design — pushing the filter into EventBus::subscribe/EventSubscription::recv rather than the SSE handler — is the right call: it gives every consumer (not just SSE) the same skip semantics for free, and it's honestly documented that skipped events still occupy channel slots and count against the lag window. Tests cover the happy path, the filter, and the 400 rejection. The futures-util-only dependency swap (dropping tokio-stream + futures-core) is a nice simplification.

A couple of things worth a look, nothing blocking:

1. TopicSet's u16 bitmask has a silent 16-topic ceiling (crates/blockchain/src/events.rs:75-91). contains/from_iter compute 1 << topic as u16. With overflow checks off (release builds), shifting by an amount ≥ the type's bit width wraps the shift amount modulo 16 rather than panicking — so a hypothetical 17th Topic variant (topic as u16 == 16) would compute 1 << 16 == 1 << 0, silently aliasing with Topic::Head's bit. Today there are only 4 topics, so this isn't live, but it's a landmine for whoever adds topic number 17 and won't get a compiler or debug-build signal (only debug builds with overflow-checks panic; a real "topics silently miscategorized" bug could ship in release). Worth a debug_assert!/const check tying the bitmask width to the variant count, or a comment flagging the ceiling explicitly (the current comment on TopicSet::ALL, "every topic, including ones added later," reads as if it's future-proof, which it isn't past 16).

2. No whitespace trimming in ?topics= values (crates/net/rpc/src/events.rs:46, Topic::from_str at crates/blockchain/src/events.rs:58). ?topics=head, block (space after comma, a natural way to write the param) 400s on " block" since from_str requires an exact match. Intentional per the "exact inverse of as_str" comment, but likely to surprise API users who copy-paste a comma-space-separated list. Consider trimming each segment before parsing if that's not a deliberate strictness choice.

3. Minor test-coverage gap: the Some("") branch (?topics= with an empty value) that maps to TopicSet::ALL isn't directly exercised — existing tests cover "no topics param" and "non-empty valid/invalid," but not the explicit-empty-string case. Cheap to add given the existing events_response helper.

Everything else — the FromStr/Display/Error impls for Topic, the EventSubscription::recv loop distinguishing Lagged from Closed, the unfold-based stream bridge, and the Response/StatusCode::BAD_REQUEST handling in the handler — looks correct and idiomatic for this codebase.

---

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

[github-actions]

🤖 Codex Code Review

Findings:

1. Medium: filtered subscriptions can report Lagged even when only unrelated topics were dropped. In crates/blockchain/src/events.rs, EventSubscription::recv() surfaces broadcast::RecvError::Lagged from the shared bus cursor, not from the filtered topic set. A subscriber interested only in head can therefore see Lagged because block or checkpoint traffic filled the ring, even if no head event was lost. The SSE path hides this in crates/net/rpc/src/events.rs, but the new public EventSubscription API and its docs suggest topic-local skip semantics. That can lead future consumers to resync unnecessarily or treat a clean filtered stream as lossy.

2. Low: TopicSet::ALL is not actually future-proof as documented. In crates/blockchain/src/events.rs, the mask is hard-coded to u16, while the comment at line 78 says it covers “every topic, including ones added later.” Adding a 17th topic will overflow 1 << topic as u16 in contains / from_iter and make ALL incorrect. Either widen/assert the bit width or drop the claim that it automatically handles future topics.

No consensus, fork-choice, attestation, STF, XMSS, or SSZ logic is touched in this PR. Aside from the two points above, the SSE filtering change looks reasonable and the error-path/tests are in decent shape.

I could not run cargo test in this environment because the sandbox blocks rustup/dependency fetches.

---

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

[github-actions]

🤖 Kimi Code Review

Overall Assessment: Solid PR. Clean implementation of server-side topic filtering for SSE events with proper error handling and good test coverage. No consensus-critical bugs (this is RPC-layer only).

Issues & Suggestions:

1. Potential overflow in TopicSet bitmask (crates/blockchain/src/events.rs:86, 90)
If the Topic enum ever grows beyond 16 variants, 1 << topic as u16 will panic in debug mode or silently wrap in release. Since TopicSet uses u16, this is a hard limit.

Recommendation: Add a compile-time assertion or document the constraint:

// In TopicSet impl:
const _: () = assert!(std::mem::variant_count::<Topic>() <= 16, "TopicSet cannot hold more than 16 topics");

2. Silent event dropping on serialization failure (crates/net/rpc/src/events.rs:65-71)
If json_data fails, the event is skipped with only a log warning. For a consensus client, consider if this is acceptable or if it should close the stream to force client re-sync.

3. TopicSet::ALL includes reserved bits (crates/blockchain/src/events.rs:83)
TopicSet(u16::MAX) sets bits for potentially undefined topics. While safe (since contains only checks valid enum discriminants), it could mask logic errors if bitwise ops are used elsewhere.

4. Allocation on unknown topic error (crates/blockchain/src/events.rs:58)
UnknownTopic(other.to_string()) allocates on every parse failure. For high-volume invalid requests, consider Box<str> or storing the original String only when necessary.

5. Dependency version consistency (crates/net/rpc/Cargo.toml:29)
futures-util = "0.3" should ideally use workspace inheritance (futures-util.workspace = true) like other dependencies, matching the pattern used for serde and tokio.

Code Quality Positives:

• Proper FromStr implementation with dedicated error type (UnknownTopic).
• Correct handling of RecvError::Lagged vs RecvError::Closed in the SSE stream.
• Good test coverage including 400 response for invalid topics and filtered subscription behavior.
• Documentation clearly notes divergence from Beacon API specification.

Consensus Safety Note:
This PR only touches the event publication layer (post-consensus) and does not affect fork choice, attestation validation, justification/finalization, or state transitions. The filtering is purely for RPC efficiency and does not alter internal state machine behavior.

---

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

[github-actions]

🤖 Claude Code Review

I have enough context to complete the review without building. Here's my analysis:

Review: feat(rpc): add ?topics= filtering to the events stream (PR 518)

Overall this is a clean, well-scoped change. The design decision to push filtering into EventBus/EventSubscription rather than the SSE handler is the right call — it keeps the skip semantics uniform for any future non-SSE consumer, and the lag-vs-filter distinction (RecvError::Lagged still propagates, only matching events are consumed silently) is preserved correctly from the original implementation. Test coverage is solid: default (unfiltered), filtered-skip, and the 400 path are all exercised, plus the Topic::from_str inverse-of-as_str round trip.

Correctness

• TopicSet::contains (crates/blockchain/src/events.rs:81-83): 1 << topic as u16 relies on as binding tighter than <<, which is correct Rust precedence, and with only 4 Topic variants this is safe. Worth flagging as a latent footgun: if Topic ever grows past 16 variants, 1u16 << topic as u16 will overflow the shift amount (panics in debug, is masked in release). Not a bug today, but since the doc comment on TopicSet::ALL explicitly says "including ones added later," a const _: () = assert!(...) or a comment noting the 16-variant ceiling would make the invariant self-documenting rather than implicit.
• event_stream (crates/net/rpc/src/events.rs:60-88): correctly upgrades the old BroadcastStreamRecvError (which only exposed Lagged) to the raw tokio::sync::broadcast::error::RecvError, so Closed is now handled explicitly (return None) instead of relying on the wrapper stream silently ending. This is a real (minor) correctness improvement over the previous code, not just a refactor.
• Query parsing (crates/net/rpc/src/events.rs:43-52): Result<TopicSet, _> collect short-circuits on the first invalid name via the standard FromIterator<Result<_,_>> impl, so a bad list doesn't do wasted work. None | Some("") correctly maps both "no param" and "empty param" to TopicSet::ALL.
• All previous call sites of EventBus::subscribe() (blockchain lib.rs tests, events.rs tests) were updated to pass TopicSet::ALL — no stale call sites left, confirmed by grep across the workspace.

Minor / nit-level

• Topic list parsing doesn't trim whitespace, so ?topics=head, block yields a 400 for " block" rather than trimming and matching Topic::Block. Probably fine (Beacon API conventions don't expect spaces either), but worth a one-line doc note in docs/rpc.md if this is intentional rather than an oversight.
• EventSubscription::try_recv is a reasonable symmetric API, but only exercised by same-crate tests today (blockchain/src/lib.rs) — not a defect, just noting it has no production caller yet.

Dependency swap

tokio-stream/futures-core → futures-util (crates/net/rpc/Cargo.toml) is clean; grepped the crate and confirmed no leftover tokio_stream/futures_core references remain.

No correctness, security, or consensus-layer concerns — this endpoint is read-only and doesn't touch fork choice, attestation processing, or state transition. Nothing here blocks merging; the shift-overflow note is the only thing I'd want acknowledged (even just as a code comment) before considering this fully closed out.

---

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