feat(l1): add eth_subscribe/eth_unsubscribe WebSocket subscription support#6496
feat(l1): add eth_subscribe/eth_unsubscribe WebSocket subscription support#6496avilagaston9 wants to merge 25 commits intomainfrom
Conversation
Implement eth_subscribe("newHeads") in both the L1 and L2 RPC WebSocket
servers. When a new block is produced, all subscribers receive the block
header as a JSON-RPC notification envelope.
L1 RPC (ethrex-rpc):
- Upgrade handle_websocket from a simple request-response loop to a
select!-based loop that multiplexes incoming messages, subscription
notification draining, and outbound message sending.
- Add handle_eth_subscribe, handle_eth_unsubscribe, generate_subscription_id,
drain_subscriptions, build_subscription_notification as public functions
so L2 can delegate to L1's implementation.
- Add new_heads_sender: Option<broadcast::Sender<Value>> field to
RpcApiContext; None when WS is disabled or subscriptions not needed.
- Add NEW_HEADS_CHANNEL_CAPACITY constant (128) for the broadcast channel.
- Export all new symbols from lib.rs including tokio::sync::broadcast.
L2 RPC (ethrex-l2-rpc):
- Add optional WS server startup in start_api when ws_addr is Some.
- handle_websocket delegates eth_subscribe/eth_unsubscribe to L1's
implementations via context.l1_ctx, eliminating duplicate code.
- Add ws_addr: Option<SocketAddr> and new_heads_sender parameters to
start_api; pass new_heads_sender through to the embedded l1_ctx.
- Re-export NEW_HEADS_CHANNEL_CAPACITY and broadcast from ethrex_rpc.
- Add axum ws feature to Cargo.toml.
L2 CLI:
- Add --l2.ws-enabled, --l2.ws-addr, --l2.ws-port flags to Options.
- Create broadcast channel in init_l2 when WS is enabled; pass sender
clone to both init_rpc_api and start_l2 (for the block producer).
L2 block producer:
- Add new_heads_sender: Option<broadcast::Sender<Value>> field.
- After sealing and storing each block, serialize the header, inject
the computed block hash, and broadcast to all WS subscribers.
- Add new_heads_sender parameter to BlockProducer::new and spawn.
Tests:
- 11 unit tests in ethrex-l2-rpc covering subscription ID generation,
broadcast channel delivery, notification format, and unsubscribe
behavior (happy path, missing subscription, missing params).
Lines of code reportTotal lines added: Detailed view |
it as a parameter. Follows the same pattern as get_http_socket_addr(opts).
creating two separate variables (new_heads_sender + new_heads_sender_for_block_producer).
… doc improvements - Remove duplicate L2 WS CLI flags (l2.ws-enabled/addr/port) — L2 now uses L1's --ws.enabled / --ws.addr / --ws.port flags - Broadcast newHeads from engine fork_choice after successful canonical head advancement - Create broadcast channel in L1 initializer when WS is enabled - Update channel capacity from 128 to 10 (matching Geth's chainEvChanSize)
sender together. Remove separate ws_addr and new_heads_sender parameters from start_api — now a single Option<WebSocketConfig>. Remove duplicate L2 WS CLI flags (l2.ws-enabled/addr/port); L2 inherits L1 flags.
paths, remove L2-specific comments and log prefixes, drop unnecessary doc comment about L1 crate delegation.
…t subscriptions
Introduce a SubscriptionManager actor (spawned-concurrency GenServer) that owns
all eth_subscribe("newHeads") subscription state. Each WebSocket connection gets
an mpsc channel and a subscription ID when it connects; the actor fans out new
block headers to all registered channels on new_head messages.
This removes the tokio broadcast channel approach, which had a "lagged subscriber"
problem and required each WS handler to maintain a HashMap of broadcast receivers.
The actor approach is simpler: two branches in the select! loop instead of three,
and dead subscribers are cleaned up automatically when their mpsc channel closes.
Changes:
- Add crates/networking/rpc/subscription_manager.rs with the SubscriptionManager
actor, SubscriptionManagerProtocol trait (#[protocol] macro), and
build_subscription_notification helper.
- WebSocketConfig now holds ActorRef<SubscriptionManager> instead of
broadcast::Sender<Value>. NEW_HEADS_CHANNEL_CAPACITY is removed.
- handle_websocket simplified to 2 select! branches; subscribe on connect,
unsubscribe on disconnect.
- handle_eth_subscribe/unsubscribe updated to use the actor; broadcast-based
drain_subscriptions and generate_subscription_id are removed.
- fork_choice.rs sends new heads via actor.new_head() instead of sender.send().
- block_producer.rs replaces broadcast::Sender with ActorRef<SubscriptionManager>.
- Both initializers.rs files spawn the actor instead of creating a broadcast channel.
- L2 sequencer mod.rs and block_producer.rs updated to pass ActorRef through.
- L2 networking rpc updated similarly; tests rewritten against the actor API.
new_head() now accepts BlockHeader directly instead of serde_json::Value. The actor handles JSON serialization and hash injection, so callers just pass the raw header.
Previously, every new WebSocket connection was auto-subscribed to
newHeads. Now subscription only happens when the client explicitly
sends eth_subscribe("newHeads"). handle_eth_subscribe calls
subscription_manager.subscribe() and stores the ID. The ID is
cleared on eth_unsubscribe or connection close.
Make handle_websocket public so L2 calls it with context.l1_ctx instead of duplicating the entire select loop, subscribe/unsubscribe routing, and cleanup logic. L2-specific RPC methods are not available over WS for now (only standard Ethereum methods). Remove all WS unit tests added by this PR.
of unbounded. When a slow client's channel is full, the notification is dropped via try_send rather than accumulating unbounded memory. The actor is never blocked.
…ptionManager tests
Change subscription tracking from Option<String> to Vec<String> so each
eth_subscribe call creates a new subscription with its own ID, matching
the Ethereum JSON-RPC spec. Previously a duplicate eth_subscribe("newHeads")
on the same connection returned the existing ID instead of creating a new one.
Also add 8 tests for the SubscriptionManager actor covering subscribe/unsubscribe,
fan-out, dead subscriber cleanup, and notification format.
-32700 Parse error with null id for unparseable requests (per JSON-RPC 2.0 spec), and use BadParams instead of Internal for unsupported subscription types since it is a client error.
and optimize fan-out serialization in SubscriptionManager. Sequential IDs leaked server activity information; replaced with 128-bit random hex IDs. Added MAX_SUBSCRIPTIONS_PER_CONNECTION (128) check in handle_eth_subscribe to prevent subscription spam. Optimized new_head fan-out to serialize the header result JSON once and build each subscriber's notification envelope via string formatting instead of repeated serde_json serialization.
can route L2-specific RPC methods (ethrexL2_*, sponsored txs, etc.) through its own map_http_requests when serving WebSocket connections. Previously the L2 WS handler delegated to the L1 context, meaning only L1 methods were reachable over WebSocket. Added Clone to RpcRequest and RpcRequestId to support passing requests by value into the routing closure.
avilagaston9
force-pushed
the
feat/l1/ws-subscriptions
branch
from
9d040d9 to
abc812f
Compare
of the repository configuration.
…thin the subscription_manager module.
avilagaston9
requested review from
a team,
ManuelBilbao and
ilitteri
as code owners
🤖 Claude Code ReviewCode Review: PR #6496 —
|
🤖 Codex Code ReviewFindings:
No other correctness/security problems stood out in the diff itself. I could not run Automated review by OpenAI Codex · gpt-5.4 · custom prompt |
There was a problem hiding this comment.
Pull request overview
Adds Ethereum JSON-RPC WebSocket subscription support (eth_subscribe / eth_unsubscribe) for "newHeads" by introducing an actor-owned subscription registry and wiring canonical-head updates to WS fan-out for both L1 and L2.
Changes:
- Introduce an actor-based
SubscriptionManagerwith bounded per-subscriber channels and ID-based subscribe/unsubscribe. - Extend the WS RPC handler to support
eth_subscribe("newHeads")/eth_unsubscribealongside normal request/response routing, and plumb WS config viaWebSocketConfig. - Emit
newHeadsnotifications from L1 fork-choice canonical head updates and from the L2 block producer after sealing.
Reviewed changes
Copilot reviewed 14 out of 14 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| test/tests/rpc/subscription_manager_tests.rs | Adds unit tests for SubscriptionManager subscribe/unsubscribe and fan-out/cleanup behavior. |
| test/tests/rpc/mod.rs | Registers the new subscription manager test module. |
| crates/networking/rpc/utils.rs | Derives Clone for RpcRequestId and RpcRequest to support WS routing flow. |
| crates/networking/rpc/test_utils.rs | Updates test API startup/context helpers for the new ws: Option<WebSocketConfig> plumbing. |
| crates/networking/rpc/subscription_manager.rs | New actor implementing subscription state and newHeads notification fan-out. |
| crates/networking/rpc/rpc.rs | Introduces WebSocketConfig, makes handle_websocket public/generic, and implements WS subscribe/unsubscribe handling. |
| crates/networking/rpc/lib.rs | Exposes subscription manager module and re-exports WS-related APIs/types. |
| crates/networking/rpc/engine/fork_choice.rs | Sends canonical head updates to the subscription manager on successful fork choice updates. |
| crates/l2/sequencer/mod.rs | Threads optional subscription manager handle into L2 startup for block producer notifications. |
| crates/l2/sequencer/block_producer.rs | Broadcasts newHeads after sealing/applying fork choice, using the optional subscription manager. |
| crates/l2/networking/rpc/rpc.rs | Starts an optional WS server in L2 and reuses L1’s handle_websocket with L2 request routing. |
| crates/l2/networking/rpc/Cargo.toml | Enables Axum’s ws feature for L2 RPC crate. |
| cmd/ethrex/l2/initializers.rs | Builds WebSocketConfig/SubscriptionManager when WS is enabled and passes it to both RPC and sequencer. |
| cmd/ethrex/initializers.rs | Builds WebSocketConfig/SubscriptionManager for L1 when WS is enabled and passes it into start_api. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| match req.method.as_str() { | ||
| "eth_subscribe" => { | ||
| let result = handle_eth_subscribe(&req, context, out_tx, subscription_ids).await; | ||
| let resp = rpc_response(req.id, result).ok()?; | ||
| Some(resp.to_string()) |
There was a problem hiding this comment.
New WebSocket behavior (handle_ws_request routing + eth_subscribe/eth_unsubscribe) isn’t covered by tests in this crate. Adding targeted unit/integration tests would help prevent regressions (e.g., batch requests over WS, invalid-request vs parse-error responses, subscription limit enforcement, and unsubscribe scoping to a single connection).
| let req: RpcRequest = match serde_json::from_str(body) { | ||
| Ok(r) => r, | ||
| Err(_) => { | ||
| // JSON-RPC 2.0 spec: parse error responses must have "id": null. | ||
| let resp = serde_json::json!({ | ||
| "jsonrpc": "2.0", | ||
| "id": null, | ||
| "error": { | ||
| "code": -32700, | ||
| "message": "Parse error" | ||
| } | ||
| }); | ||
| return Some(resp.to_string()); | ||
| } |
There was a problem hiding this comment.
handle_ws_request deserializes directly into RpcRequest and treats any deserialization error as a JSON parse error (-32700). This will incorrectly classify valid JSON that isn’t a valid request object (e.g., missing required fields), and it also regresses prior WS support for batch requests (arrays), which used to work via handle_http_request/RpcRequestWrapper. Consider parsing into RpcRequestWrapper (or serde_json::Value first) so you can: (1) return -32700 only for invalid JSON, (2) return -32600 (Invalid Request) for structurally-invalid JSON-RPC objects, and (3) preserve batch handling over WebSocket.
| #[allow(clippy::too_many_arguments)] | ||
| pub async fn start_api( | ||
| http_addr: SocketAddr, | ||
| ws_addr: Option<SocketAddr>, | ||
| ws: Option<WebSocketConfig>, | ||
| authrpc_addr: SocketAddr, | ||
| storage: Store, |
There was a problem hiding this comment.
The start_api doc comment (and argument docs) still refer to ws_addr even though the function now accepts ws: Option<WebSocketConfig>. Updating the docs here will prevent confusion for callers and keep generated docs accurate (e.g., describe WebSocketConfig.addr + subscription_manager instead of a bare socket addr).
| msg: subscription_manager_protocol::Subscribe, | ||
| _ctx: &Context<Self>, | ||
| ) -> String { | ||
| let id = generate_subscription_id(); |
There was a problem hiding this comment.
handle_subscribe inserts a freshly-generated random ID into self.subscribers without checking whether that ID already exists. While collisions are unlikely, a collision would overwrite an existing subscription sender and effectively drop a subscriber. Consider looping until an unused ID is generated (or switch to a monotonic/UUID-based ID source) before inserting.
| let id = generate_subscription_id(); | |
| let id = loop { | |
| let id = generate_subscription_id(); | |
| if !self.subscribers.contains_key(&id) { | |
| break id; | |
| } | |
| }; |
Greptile SummaryAdds Confidence Score: 4/5Safe to merge once CI passes; all findings are P2 style/robustness suggestions with no functional blockers. All identified issues are P2: JSON string interpolation, deprecated thread_rng, async{}/async move{} inconsistency, and the silent non-response edge case. The core logic — actor lifecycle, bounded channels, dead-subscriber cleanup, cleanup on disconnect, L2 reuse — is sound. Score is 4 rather than 5 only because CI is still pending. crates/networking/rpc/subscription_manager.rs (JSON building and RNG), crates/networking/rpc/rpc.rs (async capture inconsistency)
|
| Filename | Overview |
|---|---|
| crates/networking/rpc/subscription_manager.rs | New actor managing all eth_subscribe state; clean design with bounded channels and dead-subscriber cleanup, but JSON building uses string interpolation and thread_rng is deprecated in newer rand |
| crates/networking/rpc/rpc.rs | Core WS loop refactored to support eth_subscribe/eth_unsubscribe; L1 handler uses async {} while L2 counterpart uses async move {} — inconsistent but functional; silent non-response on rpc_response failure |
| crates/networking/rpc/engine/fork_choice.rs | Broadcasts newHead after canonical update via fire-and-forget actor send; correctly placed after the chain is updated |
| crates/l2/sequencer/block_producer.rs | Clones header before block is consumed, notifies subscription manager after apply_fork_choice succeeds; correct ordering |
| cmd/ethrex/l2/initializers.rs | Single SubscriptionManager actor cloned and shared between RPC server and block producer correctly; ws_config.clone() is safe since ActorRef is a cheap reference |
| test/tests/rpc/subscription_manager_tests.rs | New unit tests cover subscribe uniqueness, unsubscribe round-trips, dead-subscriber cleanup, fan-out, and hash injection; good coverage of the actor's core paths |
| crates/l2/networking/rpc/rpc.rs | L2 correctly reuses L1's handle_websocket via ctx.l1_ctx; ws config is set on the embedded L1 context so subscriptions are wired up properly |
| crates/networking/rpc/utils.rs | RpcRequest and RpcRequestId gain Clone derives; needed so req.id can be kept after req is moved into route_request |
Sequence Diagram
sequenceDiagram
participant CL as Consensus / BlockProducer
participant FC as fork_choice / block_producer
participant SM as SubscriptionManager (actor)
participant WS as handle_websocket loop
participant Client as WS Client
Client->>WS: eth_subscribe("newHeads")
WS->>SM: subscribe(out_tx)
SM-->>WS: subscription_id
WS-->>Client: {"result":"0x..."}
CL->>FC: engine_forkchoiceUpdated / seal_block
FC->>SM: new_head(BlockHeader) [fire-and-forget]
SM->>SM: serialize header + inject hash
SM->>WS: try_send(notification JSON)
WS-->>Client: eth_subscription notification
Client->>WS: eth_unsubscribe("0x...")
WS->>SM: unsubscribe(id)
SM-->>WS: true/false
WS-->>Client: {"result":true}
Note over WS,SM: On disconnect: cleanup loop unsubscribes all owned IDs
Comments Outside Diff (1)
-
crates/networking/rpc/rpc.rs, line 579-582 (link)Silent non-response on
rpc_responseserialization failureIn
handle_ws_request, all three match arms userpc_response(...).ok()?. Ifrpc_responsefails to serialize (extremely unlikely but possible for pathological payloads), the function returnsNone, and the calling loop sends no reply for the request. The WS client will hang waiting for a response. A fallback error envelope would be more robust:let resp = rpc_response(req.id, result).unwrap_or_else(|_| { serde_json::json!({"jsonrpc":"2.0","id":null,"error":{"code":-32603,"message":"Internal error"}}) }); Some(resp.to_string())
Prompt To Fix With AI
This is a comment left during a code review. Path: crates/networking/rpc/rpc.rs Line: 579-582 Comment: **Silent non-response on `rpc_response` serialization failure** In `handle_ws_request`, all three match arms use `rpc_response(...).ok()?`. If `rpc_response` fails to serialize (extremely unlikely but possible for pathological payloads), the function returns `None`, and the calling loop sends no reply for the request. The WS client will hang waiting for a response. A fallback error envelope would be more robust: ```rust let resp = rpc_response(req.id, result).unwrap_or_else(|_| { serde_json::json!({"jsonrpc":"2.0","id":null,"error":{"code":-32603,"message":"Internal error"}}) }); Some(resp.to_string()) ``` How can I resolve this? If you propose a fix, please make it concise.
Prompt To Fix All With AI
This is a comment left during a code review.
Path: crates/networking/rpc/subscription_manager.rs
Line: 153-157
Comment:
**JSON built via string interpolation**
`build_subscription_notification` embeds `result_json` verbatim into a hand-rolled format string. This works today because `sub_id` is hex-only and `result_json` is serde_json output, but any future change that makes either value contain a `"` or `}` in an unexpected place would silently produce malformed JSON. Prefer proper serialization:
```rust
fn build_subscription_notification(sub_id: &str, result_json: &str) -> String {
serde_json::json!({
"jsonrpc": "2.0",
"method": "eth_subscription",
"params": {
"subscription": sub_id,
"result": serde_json::from_str::<serde_json::Value>(result_json).unwrap_or(serde_json::Value::Null)
}
})
.to_string()
}
```
Alternatively, keep the pre-serialisation optimisation by holding `header_value` (a `serde_json::Value`) and passing it directly to `serde_json::json!` instead of converting to a string and back.
How can I resolve this? If you propose a fix, please make it concise.
---
This is a comment left during a code review.
Path: crates/networking/rpc/subscription_manager.rs
Line: 160-163
Comment:
**`rand::thread_rng()` in an actor's async handler**
`rand::thread_rng()` is thread-local. The actor runs under `spawned_concurrency`'s `Backend::Blocking`, which pools threads, so each invocation could draw from a different thread's RNG state. This is fine for randomness quality but `thread_rng` is deprecated in rand 0.9+. Prefer `rand::rng()` (rand 0.9 API) or `OsRng` to stay forward-compatible and thread-agnostic:
```suggestion
fn generate_subscription_id() -> String {
let mut bytes = [0u8; 16];
rand::rng().fill_bytes(&mut bytes);
format!("0x{}", hex::encode(bytes))
}
```
How can I resolve this? If you propose a fix, please make it concise.
---
This is a comment left during a code review.
Path: crates/networking/rpc/rpc.rs
Line: 596-604
Comment:
**`async {}` vs `async move {}` inconsistency between L1 and L2**
The L1 handler uses `async {}` (capture by reference) while the equivalent L2 handler at `crates/l2/networking/rpc/rpc.rs:164` uses `async move {}`. Both appear to compile because `ctx` ends up being moved into the inner `on_upgrade` callback regardless. However, the inconsistency is confusing and the non-`move` form relies on implicit capture semantics. For clarity and alignment with the L2 version:
```suggestion
let ws_handler = |ws: WebSocketUpgrade, State(ctx): State<RpcApiContext>| async move {
ws.on_upgrade(|mut socket| async move {
handle_websocket(&mut socket, &ctx, |req| {
let c = ctx.clone();
async move { map_http_requests(&req, c).await }
})
.await;
})
};
```
How can I resolve this? If you propose a fix, please make it concise.
---
This is a comment left during a code review.
Path: crates/networking/rpc/rpc.rs
Line: 579-582
Comment:
**Silent non-response on `rpc_response` serialization failure**
In `handle_ws_request`, all three match arms use `rpc_response(...).ok()?`. If `rpc_response` fails to serialize (extremely unlikely but possible for pathological payloads), the function returns `None`, and the calling loop sends no reply for the request. The WS client will hang waiting for a response. A fallback error envelope would be more robust:
```rust
let resp = rpc_response(req.id, result).unwrap_or_else(|_| {
serde_json::json!({"jsonrpc":"2.0","id":null,"error":{"code":-32603,"message":"Internal error"}})
});
Some(resp.to_string())
```
How can I resolve this? If you propose a fix, please make it concise.Reviews (1): Last reviewed commit: "Make build_subscription_notification pri..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| fn build_subscription_notification(sub_id: &str, result_json: &str) -> String { | ||
| format!( | ||
| r#"{{"jsonrpc":"2.0","method":"eth_subscription","params":{{"subscription":"{sub_id}","result":{result_json}}}}}"# | ||
| ) | ||
| } |
There was a problem hiding this comment.
build_subscription_notification embeds result_json verbatim into a hand-rolled format string. This works today because sub_id is hex-only and result_json is serde_json output, but any future change that makes either value contain a " or } in an unexpected place would silently produce malformed JSON. Prefer proper serialization:
fn build_subscription_notification(sub_id: &str, result_json: &str) -> String {
serde_json::json!({
"jsonrpc": "2.0",
"method": "eth_subscription",
"params": {
"subscription": sub_id,
"result": serde_json::from_str::<serde_json::Value>(result_json).unwrap_or(serde_json::Value::Null)
}
})
.to_string()
}Alternatively, keep the pre-serialisation optimisation by holding header_value (a serde_json::Value) and passing it directly to serde_json::json! instead of converting to a string and back.
Prompt To Fix With AI
This is a comment left during a code review.
Path: crates/networking/rpc/subscription_manager.rs
Line: 153-157
Comment:
**JSON built via string interpolation**
`build_subscription_notification` embeds `result_json` verbatim into a hand-rolled format string. This works today because `sub_id` is hex-only and `result_json` is serde_json output, but any future change that makes either value contain a `"` or `}` in an unexpected place would silently produce malformed JSON. Prefer proper serialization:
```rust
fn build_subscription_notification(sub_id: &str, result_json: &str) -> String {
serde_json::json!({
"jsonrpc": "2.0",
"method": "eth_subscription",
"params": {
"subscription": sub_id,
"result": serde_json::from_str::<serde_json::Value>(result_json).unwrap_or(serde_json::Value::Null)
}
})
.to_string()
}
```
Alternatively, keep the pre-serialisation optimisation by holding `header_value` (a `serde_json::Value`) and passing it directly to `serde_json::json!` instead of converting to a string and back.
How can I resolve this? If you propose a fix, please make it concise.| fn generate_subscription_id() -> String { | ||
| let mut bytes = [0u8; 16]; | ||
| rand::thread_rng().fill_bytes(&mut bytes); | ||
| format!("0x{}", hex::encode(bytes)) |
There was a problem hiding this comment.
rand::thread_rng() in an actor's async handler
rand::thread_rng() is thread-local. The actor runs under spawned_concurrency's Backend::Blocking, which pools threads, so each invocation could draw from a different thread's RNG state. This is fine for randomness quality but thread_rng is deprecated in rand 0.9+. Prefer rand::rng() (rand 0.9 API) or OsRng to stay forward-compatible and thread-agnostic:
| fn generate_subscription_id() -> String { | |
| let mut bytes = [0u8; 16]; | |
| rand::thread_rng().fill_bytes(&mut bytes); | |
| format!("0x{}", hex::encode(bytes)) | |
| fn generate_subscription_id() -> String { | |
| let mut bytes = [0u8; 16]; | |
| rand::rng().fill_bytes(&mut bytes); | |
| format!("0x{}", hex::encode(bytes)) | |
| } |
Prompt To Fix With AI
This is a comment left during a code review.
Path: crates/networking/rpc/subscription_manager.rs
Line: 160-163
Comment:
**`rand::thread_rng()` in an actor's async handler**
`rand::thread_rng()` is thread-local. The actor runs under `spawned_concurrency`'s `Backend::Blocking`, which pools threads, so each invocation could draw from a different thread's RNG state. This is fine for randomness quality but `thread_rng` is deprecated in rand 0.9+. Prefer `rand::rng()` (rand 0.9 API) or `OsRng` to stay forward-compatible and thread-agnostic:
```suggestion
fn generate_subscription_id() -> String {
let mut bytes = [0u8; 16];
rand::rng().fill_bytes(&mut bytes);
format!("0x{}", hex::encode(bytes))
}
```
How can I resolve this? If you propose a fix, please make it concise.| let ws_handler = |ws: WebSocketUpgrade, State(ctx): State<RpcApiContext>| async { | ||
| ws.on_upgrade(|mut socket| async move { | ||
| handle_websocket(&mut socket, &ctx, |req| { | ||
| let c = ctx.clone(); | ||
| async move { map_http_requests(&req, c).await } | ||
| }) | ||
| .await; | ||
| }) | ||
| }; |
There was a problem hiding this comment.
async {} vs async move {} inconsistency between L1 and L2
The L1 handler uses async {} (capture by reference) while the equivalent L2 handler at crates/l2/networking/rpc/rpc.rs:164 uses async move {}. Both appear to compile because ctx ends up being moved into the inner on_upgrade callback regardless. However, the inconsistency is confusing and the non-move form relies on implicit capture semantics. For clarity and alignment with the L2 version:
| let ws_handler = |ws: WebSocketUpgrade, State(ctx): State<RpcApiContext>| async { | |
| ws.on_upgrade(|mut socket| async move { | |
| handle_websocket(&mut socket, &ctx, |req| { | |
| let c = ctx.clone(); | |
| async move { map_http_requests(&req, c).await } | |
| }) | |
| .await; | |
| }) | |
| }; | |
| let ws_handler = |ws: WebSocketUpgrade, State(ctx): State<RpcApiContext>| async move { | |
| ws.on_upgrade(|mut socket| async move { | |
| handle_websocket(&mut socket, &ctx, |req| { | |
| let c = ctx.clone(); | |
| async move { map_http_requests(&req, c).await } | |
| }) | |
| .await; | |
| }) | |
| }; |
Prompt To Fix With AI
This is a comment left during a code review.
Path: crates/networking/rpc/rpc.rs
Line: 596-604
Comment:
**`async {}` vs `async move {}` inconsistency between L1 and L2**
The L1 handler uses `async {}` (capture by reference) while the equivalent L2 handler at `crates/l2/networking/rpc/rpc.rs:164` uses `async move {}`. Both appear to compile because `ctx` ends up being moved into the inner `on_upgrade` callback regardless. However, the inconsistency is confusing and the non-`move` form relies on implicit capture semantics. For clarity and alignment with the L2 version:
```suggestion
let ws_handler = |ws: WebSocketUpgrade, State(ctx): State<RpcApiContext>| async move {
ws.on_upgrade(|mut socket| async move {
handle_websocket(&mut socket, &ctx, |req| {
let c = ctx.clone();
async move { map_http_requests(&req, c).await }
})
.await;
})
};
```
How can I resolve this? If you propose a fix, please make it concise.Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
Motivation
Implement
eth_subscribe("newHeads")WebSocket subscription support. This is standard Ethereum JSON-RPC functionality needed by external tools (e.g., the Credible Layer sidecar) that track new blocks in real time.Description
Add a
SubscriptionManagerGenServer actor that owns all subscription state. When a new block becomes canonical (via fork choice update on L1, or block production on L2), the headBlockHeaderis sent to the actor, which serializes it and fans out notifications to all subscribed WebSocket clients through bounded per-connection channels.SubscriptionManageractor (crates/networking/rpc/subscription_manager.rs): handlesSubscribe,Unsubscribe, andNewHeadmessages. Dead subscribers are cleaned up automatically on the nextNewHeadfan-out.WebSocketConfigstruct bundles the WS socket address and the actor handle.handle_websocketis public so L2 reuses it directly — zero duplicate WS code in L2.eth_subscribe("newHeads"), not on connection open.try_sendinstead of accumulating unbounded memory. The actor is never blocked.newHeadsfromengine/fork_choice.rsafter successful canonical head advancement.--ws.enabled/--ws.addr/--ws.portCLI flags — no L2-specific WS flags.How to Test
Checklist
cargo check -p ethrex --features l2)