diff --git a/Cargo.lock b/Cargo.lock index 6c6acbdc..eb1a0e93 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3009,6 +3009,7 @@ dependencies = [ "eyre", "futures", "hex", + "jsonrpsee", "reth-basic-payload-builder", "reth-chainspec", "reth-cli", @@ -3055,6 +3056,7 @@ dependencies = [ "tokio", "tracing", "tracing-subscriber 0.3.23", + "url", ] [[package]] @@ -3173,6 +3175,7 @@ dependencies = [ "eyre", "futures", "hex", + "jsonrpsee", "rand 0.10.2", "reqwest 0.12.28", "reth-basic-payload-builder", @@ -3205,6 +3208,7 @@ dependencies = [ "tempfile", "tokio", "tracing", + "url", ] [[package]] diff --git a/Dockerfile.cross b/Dockerfile.cross index 8da01728..6cbbcf42 100644 --- a/Dockerfile.cross +++ b/Dockerfile.cross @@ -9,7 +9,7 @@ ARG BUILDPLATFORM # Copy the pre-built binary based on the target platform COPY dist/bin/${TARGETPLATFORM}/ev-reth /usr/local/bin/ev-reth -RUN apt-get update && apt-get install -y --no-install-recommends curl jq tini && rm -rf /var/lib/apt/lists/* +RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates curl jq tini && rm -rf /var/lib/apt/lists/* # Expose default ports EXPOSE 8545 8546 30303 6060 9001 diff --git a/README.md b/README.md index cc03200c..c40be498 100644 --- a/README.md +++ b/README.md @@ -164,6 +164,31 @@ With custom configuration: --ws.api all ``` +### Peer Head Subscription + +The sequencer continues to run its EV node and ev-reth. Other full nodes can run ev-reth without an +EV node by subscribing to any authoritative ev-reth peer. The WebSocket pushes only chain identity +and forkchoice references; native Reth P2P fetches and validates the referenced blocks. + +Enable WebSocket RPC on the publishing peer with `--ws`, then start a subscribing full node with +`--subscribe-peer` and at least one P2P peer that has the chain: + +```bash +./target/release/ev-reth node \ + --chain /path/to/genesis.json \ + --datadir /var/lib/ev-reth-subscriber \ + --subscribe-peer wss://peer.example/rpc \ + --trusted-peers enode://@:30303 \ + --http --http.api eth,net,web3 +``` + +The WebSocket publisher and P2P block-data peer may be different nodes. A synchronized subscriber +can also enable `--ws` and relay validated forkchoice updates downstream. + +See the [Peer Head Subscription Guide](docs/guide/peer-head-subscription.md) for architecture, +publishing-peer setup, private P2P configuration, relay topology, verification, security, and +troubleshooting. + ### Lightweight Chainspec Startup Large custom genesis files can be expensive to parse on every restart because the `alloc` map is @@ -564,6 +589,7 @@ ev-reth/ │ │ ├── builder.rs # Payload builder implementation │ │ ├── executor.rs # Block executor for EvTxEnvelope │ │ ├── evm_executor.rs # EVM executor and receipt builder +│ │ ├── head.rs # Push-based peer forkchoice subscription │ │ ├── payload_types.rs # EvBuiltPayload and conversions │ │ ├── rpc.rs # RPC types with feePayer support │ │ ├── txpool.rs # EvNode txpool validator diff --git a/bin/ev-reth/src/main.rs b/bin/ev-reth/src/main.rs index 79c35162..f9050225 100644 --- a/bin/ev-reth/src/main.rs +++ b/bin/ev-reth/src/main.rs @@ -16,7 +16,10 @@ use tracing::info; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter, Layer}; use url::Url; -use ev_node::{log_startup, EvolveArgs, EvolveChainSpecParser, EvolveNode}; +use ev_node::{ + head::{HeadApiServer, HeadPublisher}, + log_startup, EvolveArgs, EvolveChainSpecParser, EvolveNode, +}; #[global_allocator] static ALLOC: reth_cli_util::allocator::Allocator = reth_cli_util::allocator::new_allocator(); @@ -96,8 +99,14 @@ fn main() { init_tracing(); if let Err(err) = - Cli::::parse().run(|builder, _evolve_args| async move { + Cli::::parse().run(|builder, evolve_args| async move { log_startup(); + let head_publisher = HeadPublisher::new( + builder.config().chain.chain().id(), + builder.config().chain.genesis_hash(), + ); + let rpc_head_publisher = head_publisher.clone(); + let handle = builder .node(EvolveNode::new()) .extend_rpc_modules(move |ctx| { @@ -108,11 +117,34 @@ fn main() { // Merge into all enabled transports (HTTP / WS) ctx.modules.merge_configured(evolve_txpool.into_rpc())?; + ctx.modules + .merge_configured(rpc_head_publisher.into_rpc())?; Ok(()) }) .launch() .await?; + ev_node::head::spawn_publisher( + handle.node.task_executor.clone(), + handle.node.provider.clone(), + handle + .node + .add_ons_handle + .consensus_engine_events() + .new_listener(), + head_publisher, + ); + + if let Some(peer_url) = evolve_args.subscribe_peer { + ev_node::head::spawn_subscriber( + handle.node.task_executor.clone(), + handle.node.add_ons_handle.beacon_engine_handle.clone(), + peer_url, + handle.node.config.chain.chain().id(), + handle.node.config.chain.genesis_hash(), + ); + } + info!("=== EV-RETH: Node launched successfully with ev-reth payload builder ==="); handle.node_exit_future.await }) diff --git a/crates/node/Cargo.toml b/crates/node/Cargo.toml index ea7bbb66..3eab7bd9 100644 --- a/crates/node/Cargo.toml +++ b/crates/node/Cargo.toml @@ -52,6 +52,7 @@ reth-rpc-eth-api.workspace = true reth-rpc-eth-types.workspace = true reth-engine-primitives.workspace = true reth-ethereum-primitives.workspace = true +reth-tasks.workspace = true # Alloy dependencies alloy-rpc-types.workspace = true @@ -71,7 +72,9 @@ c-kzg = "2.1.6" eyre.workspace = true tracing.workspace = true tokio = { workspace = true, features = ["full"] } +jsonrpsee = { workspace = true, features = ["macros", "server", "ws-client", "client-ws-transport-tls"] } serde = { workspace = true, features = ["derive"] } +url.workspace = true serde_json.workspace = true thiserror.workspace = true async-trait.workspace = true diff --git a/crates/node/src/args.rs b/crates/node/src/args.rs index 65f45a2d..f51fa320 100644 --- a/crates/node/src/args.rs +++ b/crates/node/src/args.rs @@ -1,5 +1,61 @@ use clap::Args; +use url::Url; -/// Evolve CLI arguments (currently empty; reserved for future toggles). +/// Evolve CLI arguments. #[derive(Debug, Clone, Default, Args)] -pub struct EvolveArgs {} +pub struct EvolveArgs { + /// Subscribe to valid forkchoice updates from an authoritative ev-reth peer. + /// + /// The subscriber receives only chain identity and forkchoice references from this endpoint. + /// Block headers and bodies are fetched through the configured native Reth P2P peers. + #[arg( + long, + value_name = "WS_URL", + env = "EV_SUBSCRIBE_PEER", + value_parser = parse_websocket_url + )] + pub subscribe_peer: Option, +} + +fn parse_websocket_url(value: &str) -> Result { + let url = Url::parse(value).map_err(|error| error.to_string())?; + if matches!(url.scheme(), "ws" | "wss") { + Ok(url) + } else { + Err("peer endpoint must use ws:// or wss://".into()) + } +} + +#[cfg(test)] +mod tests { + use super::EvolveArgs; + use clap::Parser; + + #[derive(Parser)] + struct TestCli { + #[command(flatten)] + evolve: EvolveArgs, + } + + #[test] + fn parses_peer_websocket_url() { + let cli = + TestCli::try_parse_from(["ev-reth", "--subscribe-peer", "wss://peer.example/rpc"]) + .expect("valid peer subscription argument"); + + assert_eq!( + cli.evolve.subscribe_peer.expect("configured URL").as_str(), + "wss://peer.example/rpc" + ); + } + + #[test] + fn rejects_non_websocket_peer_url() { + assert!(TestCli::try_parse_from([ + "ev-reth", + "--subscribe-peer", + "https://peer.example/rpc", + ]) + .is_err()); + } +} diff --git a/crates/node/src/head.rs b/crates/node/src/head.rs new file mode 100644 index 00000000..137f72dc --- /dev/null +++ b/crates/node/src/head.rs @@ -0,0 +1,794 @@ +//! Push-based forkchoice sharing between ev-reth peers. +//! +//! Peers exchange only chain identity and forkchoice references over WebSocket. Missing block +//! headers and bodies are fetched by Reth's native P2P downloader after the local engine receives +//! the pushed forkchoice state. + +use crate::EvolveEngineTypes; +use alloy_eips::BlockNumHash; +use alloy_primitives::B256; +use alloy_rpc_types_engine::{ForkchoiceState, PayloadStatusEnum}; +use async_trait::async_trait; +use ev_primitives::EvPrimitives; +use futures::{Stream, StreamExt}; +use jsonrpsee::{ + core::SubscriptionResult, proc_macros::rpc, ws_client::WsClientBuilder, + PendingSubscriptionSink, SubscriptionMessage, +}; +use reth_engine_primitives::{ConsensusEngineEvent, ConsensusEngineHandle}; +use reth_storage_api::{BlockIdReader, BlockNumReader}; +use reth_tasks::TaskExecutor; +use serde::{Deserialize, Serialize}; +use std::time::{Duration, Instant}; +use tokio::{ + sync::watch, + time::{interval, sleep, timeout, MissedTickBehavior}, +}; +use tracing::{debug, error, info, warn}; +use url::Url; + +const RPC_TIMEOUT: Duration = Duration::from_secs(15); +const RETRY_INTERVAL: Duration = Duration::from_secs(3); +const INITIAL_RECONNECT_DELAY: Duration = Duration::from_secs(1); +const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(30); +const SUBSCRIPTION_BUFFER: usize = 16; + +/// A complete, self-identifying forkchoice update pushed by an ev-reth peer. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct HeadUpdate { + /// Chain ID of the publishing peer. + pub chain_id: u64, + /// Genesis hash of the publishing peer. + pub genesis_hash: B256, + /// Forkchoice state accepted by the publishing peer. + pub forkchoice_state: ForkchoiceState, + /// Block number corresponding to `head_block_hash`. + pub head_block_number: u64, + /// Block number corresponding to `safe_block_hash`. + pub safe_block_number: u64, + /// Block number corresponding to `finalized_block_hash`. + pub finalized_block_number: u64, +} + +/// Push-based ev-reth head sharing API. +#[rpc(server, client, namespace = "ev")] +pub trait HeadApi { + /// Subscribes to the latest valid forkchoice state accepted by this node. + #[subscription( + name = "subscribeForkchoice", + unsubscribe = "unsubscribeForkchoice", + item = HeadUpdate + )] + fn subscribe_forkchoice(&self) -> SubscriptionResult; +} + +/// Latest-value publisher backing `ev_subscribeForkchoice`. +/// +/// A watch channel intentionally retains only one update. Slow subscribers can reconnect and +/// receive the newest valid forkchoice without accumulating an unbounded history. +#[derive(Clone, Debug)] +pub struct HeadPublisher { + chain_id: u64, + genesis_hash: B256, + latest: watch::Sender>, +} + +impl HeadPublisher { + /// Creates a publisher for one chain. + pub fn new(chain_id: u64, genesis_hash: B256) -> Self { + let (latest, receiver) = watch::channel(None); + drop(receiver); + Self { + chain_id, + genesis_hash, + latest, + } + } + + /// Publishes a valid forkchoice state with its corresponding block numbers. + /// + /// Returns `true` when the latest value changed. + pub fn publish( + &self, + forkchoice_state: ForkchoiceState, + head_block_number: u64, + safe_block_number: u64, + finalized_block_number: u64, + ) -> bool { + let update = HeadUpdate { + chain_id: self.chain_id, + genesis_hash: self.genesis_hash, + forkchoice_state, + head_block_number, + safe_block_number, + finalized_block_number, + }; + self.latest.send_if_modified(|current| { + if *current == Some(update) { + false + } else { + *current = Some(update); + true + } + }) + } + + /// Number of active subscription receivers. + pub fn subscriber_count(&self) -> usize { + self.latest.receiver_count() + } + + fn publish_state

(&self, provider: &P, state: ForkchoiceState) -> Result + where + P: BlockNumReader, + { + let head = block_number(provider, state.head_block_hash, "head")?; + let safe = block_number(provider, state.safe_block_hash, "safe")?; + let finalized = block_number(provider, state.finalized_block_hash, "finalized")?; + Ok(self.publish(state, head, safe, finalized)) + } + + fn publish_provider_snapshot

(&self, provider: &P) -> Result + where + P: BlockNumReader + BlockIdReader, + { + let chain_info = provider.chain_info().map_err(|error| error.to_string())?; + let safe = provider + .safe_block_num_hash() + .map_err(|error| error.to_string())? + .ok_or_else(|| "local safe block is unavailable".to_string())?; + let finalized = provider + .finalized_block_num_hash() + .map_err(|error| error.to_string())? + .ok_or_else(|| "local finalized block is unavailable".to_string())?; + let head = BlockNumHash::new(chain_info.best_number, chain_info.best_hash); + Ok(self.publish( + ForkchoiceState { + head_block_hash: head.hash, + safe_block_hash: safe.hash, + finalized_block_hash: finalized.hash, + }, + head.number, + safe.number, + finalized.number, + )) + } +} + +impl HeadApiServer for HeadPublisher { + fn subscribe_forkchoice( + &self, + pending_subscription: PendingSubscriptionSink, + ) -> SubscriptionResult { + let mut updates = self.latest.subscribe(); + tokio::spawn(async move { + let sink = match pending_subscription.accept().await { + Ok(sink) => sink, + Err(error) => { + warn!(%error, "failed to accept forkchoice subscription"); + return; + } + }; + + loop { + let update = *updates.borrow_and_update(); + if let Some(update) = update { + let message = match serde_json::value::to_raw_value(&update) { + Ok(value) => SubscriptionMessage::from(value), + Err(error) => { + error!(%error, "failed to serialize forkchoice update"); + return; + } + }; + if sink.send(message).await.is_err() { + return; + } + } + + tokio::select! { + _ = sink.closed() => return, + changed = updates.changed() => { + if changed.is_err() { + return; + } + } + } + } + }); + Ok(()) + } +} + +/// Publishes valid forkchoice events from the local Reth engine. +pub fn spawn_publisher( + executor: TaskExecutor, + provider: P, + mut engine_events: S, + publisher: HeadPublisher, +) where + P: BlockNumReader + BlockIdReader + Send + Sync + 'static, + S: Stream> + Send + Unpin + 'static, +{ + executor.spawn_with_graceful_shutdown_signal(move |shutdown| async move { + if let Err(error) = publisher.publish_provider_snapshot(&provider) { + debug!(%error, "valid forkchoice snapshot is not available yet"); + } + + tokio::pin!(shutdown); + loop { + tokio::select! { + _ = &mut shutdown => { + info!("forkchoice publisher stopped during node shutdown"); + return; + } + event = engine_events.next() => { + let Some(event) = event else { + warn!("local engine event stream ended; forkchoice publisher stopped"); + return; + }; + if let ConsensusEngineEvent::ForkchoiceUpdated(state, status) = event { + if status.is_valid() { + match publisher.publish_state(&provider, state) { + Ok(true) => { + debug!(head = ?state.head_block_hash, "published valid forkchoice"); + } + Ok(false) => {} + Err(error) => { + warn!(%error, ?state, "could not resolve valid forkchoice locally"); + } + } + } + } + } + } + } + }); +} + +/// Starts the non-critical, shutdown-aware peer subscriber. +/// +/// An unavailable peer is deliberately not node-fatal: the local node continues to serve its +/// existing chain and P2P connections while this task reconnects with bounded backoff. +pub fn spawn_subscriber( + executor: TaskExecutor, + engine: ConsensusEngineHandle, + peer_url: Url, + expected_chain_id: u64, + expected_genesis_hash: B256, +) { + spawn_with_sink( + executor, + InProcessForkchoiceSink(engine), + peer_url, + expected_chain_id, + expected_genesis_hash, + ); +} + +/// Starts a peer subscriber with a caller-provided local forkchoice sink. +pub fn spawn_with_sink( + executor: TaskExecutor, + sink: S, + peer_url: Url, + expected_chain_id: u64, + expected_genesis_hash: B256, +) where + S: ForkchoiceSink + 'static, +{ + executor.spawn_with_graceful_shutdown_signal(move |shutdown| async move { + tokio::select! { + _ = shutdown => info!("peer head subscriber stopped during node shutdown"), + () = subscribe(sink, peer_url, expected_chain_id, expected_genesis_hash) => {} + } + }); +} + +/// Runs the peer subscription until its endpoint violates the trust policy or the future is +/// cancelled by its owner. +pub async fn subscribe( + sink: S, + peer_url: Url, + expected_chain_id: u64, + expected_genesis_hash: B256, +) { + let mut reconnect_delay = INITIAL_RECONNECT_DELAY; + let mut tracker = ForkchoiceTracker::default(); + + loop { + let session_started = Instant::now(); + match run_session( + &sink, + &peer_url, + expected_chain_id, + expected_genesis_hash, + &mut tracker, + ) + .await + { + Ok(()) => { + warn!(endpoint = %peer_url, "peer forkchoice subscription ended; reconnecting"); + } + Err( + SubscriptionError::IncompatibleChain(message) + | SubscriptionError::MalformedFinality(message), + ) => { + error!(endpoint = %peer_url, %message, "peer violated head subscription trust policy; stopping subscriber"); + return; + } + Err(error) => { + warn!(endpoint = %peer_url, %error, "peer head subscription failed; reconnecting"); + } + } + + if session_started.elapsed() >= MAX_RECONNECT_DELAY { + reconnect_delay = INITIAL_RECONNECT_DELAY; + } + sleep(reconnect_delay).await; + reconnect_delay = reconnect_delay.saturating_mul(2).min(MAX_RECONNECT_DELAY); + } +} + +async fn run_session( + sink: &S, + peer_url: &Url, + expected_chain_id: u64, + expected_genesis_hash: B256, + tracker: &mut ForkchoiceTracker, +) -> Result<(), SubscriptionError> { + let client = timeout( + RPC_TIMEOUT, + WsClientBuilder::default() + .request_timeout(RPC_TIMEOUT) + .max_concurrent_requests(4) + .max_buffer_capacity_per_subscription(SUBSCRIPTION_BUFFER) + .build(peer_url.as_str()), + ) + .await + .map_err(|_| SubscriptionError::Transient("WebSocket connection timed out".into()))? + .map_err(|error| SubscriptionError::Transient(error.to_string()))?; + + let mut updates = timeout(RPC_TIMEOUT, HeadApiClient::subscribe_forkchoice(&client)) + .await + .map_err(|_| SubscriptionError::Transient("forkchoice subscription timed out".into()))? + .map_err(|error| SubscriptionError::Transient(error.to_string()))?; + info!(endpoint = %peer_url, "subscribed to peer forkchoice updates"); + + let mut retry = interval(RETRY_INTERVAL); + retry.set_missed_tick_behavior(MissedTickBehavior::Skip); + retry.tick().await; + + loop { + tokio::select! { + update = updates.next() => { + let update = update + .ok_or_else(|| SubscriptionError::Transient("forkchoice subscription ended".into()))? + .map_err(|error| SubscriptionError::Transient(error.to_string()))?; + process_update( + sink, + tracker, + update, + expected_chain_id, + expected_genesis_hash, + ) + .await?; + } + _ = retry.tick() => { + apply_pending(sink, tracker).await?; + } + } + } +} + +async fn process_update( + sink: &S, + tracker: &mut ForkchoiceTracker, + update: HeadUpdate, + expected_chain_id: u64, + expected_genesis_hash: B256, +) -> Result<(), SubscriptionError> { + if update.chain_id != expected_chain_id { + return Err(SubscriptionError::IncompatibleChain(format!( + "chain ID is {}, expected {expected_chain_id}", + update.chain_id + ))); + } + if update.genesis_hash != expected_genesis_hash { + return Err(SubscriptionError::IncompatibleChain(format!( + "genesis hash is {}, expected {expected_genesis_hash}", + update.genesis_hash + ))); + } + + let state = update.forkchoice_state; + let head = RemoteBlock { + hash: state.head_block_hash, + number: update.head_block_number, + }; + let safe = RemoteBlock { + hash: state.safe_block_hash, + number: update.safe_block_number, + }; + let finalized = RemoteBlock { + hash: state.finalized_block_hash, + number: update.finalized_block_number, + }; + validate_forkchoice(head, safe, finalized, tracker.last_finality)?; + + let resolved = ResolvedForkchoice { + state, + finality: Finality { safe, finalized }, + }; + if tracker.should_skip(resolved) { + return Ok(()); + } + tracker.queue(resolved); + apply_pending(sink, tracker).await +} + +async fn apply_pending( + sink: &S, + tracker: &mut ForkchoiceTracker, +) -> Result<(), SubscriptionError> { + let Some(pending) = tracker.pending else { + return Ok(()); + }; + + let status = sink + .apply_forkchoice(pending.state) + .await + .map_err(SubscriptionError::Transient)?; + match &status { + PayloadStatusEnum::Valid => { + tracker.accept(pending); + debug!(head = ?pending.state.head_block_hash, "applied peer forkchoice to local engine"); + } + PayloadStatusEnum::Syncing | PayloadStatusEnum::Accepted => { + debug!(head = ?pending.state.head_block_hash, ?status, "local engine is fetching peer target through P2P"); + } + PayloadStatusEnum::Invalid { validation_error } => { + error!(head = ?pending.state.head_block_hash, %validation_error, "local engine rejected peer forkchoice; waiting for a changed state"); + tracker.reject(pending.state); + } + } + Ok(()) +} + +fn block_number(provider: &P, hash: B256, label: &str) -> Result { + if hash.is_zero() { + return Err(format!("{label} block hash is zero")); + } + provider + .block_number(hash) + .map_err(|error| error.to_string())? + .ok_or_else(|| format!("{label} block {hash} is unavailable locally")) +} + +fn validate_forkchoice( + head: RemoteBlock, + safe: RemoteBlock, + finalized: RemoteBlock, + previous: Option, +) -> Result<(), SubscriptionError> { + if finalized.number > safe.number || safe.number > head.number { + return Err(SubscriptionError::MalformedFinality(format!( + "finalized={}, safe={}, head={}", + finalized.number, safe.number, head.number + ))); + } + if (finalized.number == safe.number && finalized.hash != safe.hash) + || (safe.number == head.number && safe.hash != head.hash) + { + return Err(SubscriptionError::MalformedFinality( + "equal-height canonical references have different hashes".into(), + )); + } + if let Some(previous) = previous { + reject_regression("safe", safe, previous.safe)?; + reject_regression("finalized", finalized, previous.finalized)?; + } + Ok(()) +} + +fn reject_regression( + name: &str, + current: RemoteBlock, + previous: RemoteBlock, +) -> Result<(), SubscriptionError> { + if current.number < previous.number + || (current.number == previous.number && current.hash != previous.hash) + { + return Err(SubscriptionError::MalformedFinality(format!( + "{name} regressed from {} ({}) to {} ({})", + previous.number, previous.hash, current.number, current.hash + ))); + } + Ok(()) +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +struct RemoteBlock { + hash: B256, + number: u64, +} + +#[derive(Clone, Copy, Debug)] +struct Finality { + safe: RemoteBlock, + finalized: RemoteBlock, +} + +#[derive(Clone, Copy, Debug)] +struct ResolvedForkchoice { + state: ForkchoiceState, + finality: Finality, +} + +#[derive(Default)] +struct ForkchoiceTracker { + accepted: Option, + pending: Option, + rejected: Option, + last_finality: Option, +} + +impl ForkchoiceTracker { + fn should_skip(&self, resolved: ResolvedForkchoice) -> bool { + self.accepted == Some(resolved.state) || self.rejected == Some(resolved.state) + } + + const fn queue(&mut self, resolved: ResolvedForkchoice) { + self.rejected = None; + self.pending = Some(resolved); + } + + const fn accept(&mut self, resolved: ResolvedForkchoice) { + self.accepted = Some(resolved.state); + self.pending = None; + self.last_finality = Some(resolved.finality); + } + + const fn reject(&mut self, state: ForkchoiceState) { + self.pending = None; + self.rejected = Some(state); + } +} + +#[derive(Debug)] +enum SubscriptionError { + IncompatibleChain(String), + MalformedFinality(String), + Transient(String), +} + +impl std::fmt::Display for SubscriptionError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::IncompatibleChain(message) + | Self::MalformedFinality(message) + | Self::Transient(message) => formatter.write_str(message), + } + } +} + +impl std::error::Error for SubscriptionError {} + +/// Applies a payload-free forkchoice update to the local subscriber. +#[async_trait] +pub trait ForkchoiceSink: Send + Sync { + /// Returns the local engine's payload status. Implementations must not submit a payload. + async fn apply_forkchoice(&self, state: ForkchoiceState) -> Result; +} + +/// Production sink which communicates directly with the local Reth consensus engine. +#[derive(Debug)] +pub struct InProcessForkchoiceSink(ConsensusEngineHandle); + +#[async_trait] +impl ForkchoiceSink for InProcessForkchoiceSink { + async fn apply_forkchoice(&self, state: ForkchoiceState) -> Result { + self.0 + .fork_choice_updated(state, None) + .await + .map(|response| response.payload_status.status) + .map_err(|error| error.to_string()) + } +} + +#[cfg(test)] +mod tests { + use super::{ + process_update, Finality, ForkchoiceSink, ForkchoiceTracker, HeadPublisher, HeadUpdate, + RemoteBlock, ResolvedForkchoice, SubscriptionError, + }; + use alloy_primitives::B256; + use alloy_rpc_types_engine::{ForkchoiceState, PayloadStatusEnum}; + use async_trait::async_trait; + use std::sync::{Arc, Mutex}; + + fn block(number: u64, byte: u8) -> RemoteBlock { + RemoteBlock { + hash: B256::repeat_byte(byte), + number, + } + } + + fn update( + chain_id: u64, + genesis_hash: B256, + head: RemoteBlock, + safe: RemoteBlock, + finalized: RemoteBlock, + ) -> HeadUpdate { + HeadUpdate { + chain_id, + genesis_hash, + forkchoice_state: ForkchoiceState { + head_block_hash: head.hash, + safe_block_hash: safe.hash, + finalized_block_hash: finalized.hash, + }, + head_block_number: head.number, + safe_block_number: safe.number, + finalized_block_number: finalized.number, + } + } + + #[derive(Clone, Default)] + struct RecordingSink(Arc>>); + + #[async_trait] + impl ForkchoiceSink for RecordingSink { + async fn apply_forkchoice( + &self, + state: ForkchoiceState, + ) -> Result { + self.0.lock().expect("recording sink lock").push(state); + Ok(PayloadStatusEnum::Valid) + } + } + + #[test] + fn publisher_retains_only_latest_update() { + let publisher = HeadPublisher::new(10, B256::repeat_byte(1)); + let state = ForkchoiceState { + head_block_hash: B256::repeat_byte(4), + safe_block_hash: B256::repeat_byte(3), + finalized_block_hash: B256::repeat_byte(2), + }; + assert!(publisher.publish(state, 4, 3, 2)); + assert!(!publisher.publish(state, 4, 3, 2)); + + let receiver = publisher.latest.subscribe(); + assert_eq!( + *receiver.borrow(), + Some(HeadUpdate { + chain_id: 10, + genesis_hash: B256::repeat_byte(1), + forkchoice_state: state, + head_block_number: 4, + safe_block_number: 3, + finalized_block_number: 2, + }) + ); + } + + #[tokio::test] + async fn applies_valid_push_without_remote_block_queries() { + let genesis = B256::repeat_byte(1); + let head = block(10, 10); + let safe = block(9, 9); + let finalized = block(8, 8); + let sink = RecordingSink::default(); + let mut tracker = ForkchoiceTracker::default(); + + process_update( + &sink, + &mut tracker, + update(7, genesis, head, safe, finalized), + 7, + genesis, + ) + .await + .expect("valid pushed update"); + + assert_eq!( + sink.0.lock().expect("recording sink lock").as_slice(), + &[ForkchoiceState { + head_block_hash: head.hash, + safe_block_hash: safe.hash, + finalized_block_hash: finalized.hash, + }] + ); + } + + #[tokio::test] + async fn rejects_incompatible_peer_identity() { + let genesis = B256::repeat_byte(1); + let error = process_update( + &RecordingSink::default(), + &mut ForkchoiceTracker::default(), + update(8, genesis, block(10, 10), block(9, 9), block(8, 8)), + 7, + genesis, + ) + .await + .expect_err("chain identity must match"); + assert!(matches!(error, SubscriptionError::IncompatibleChain(_))); + } + + #[tokio::test] + async fn applies_lower_head_reorg_without_finality_regression() { + let genesis = B256::repeat_byte(1); + let safe = block(8, 8); + let sink = RecordingSink::default(); + let mut tracker = ForkchoiceTracker::default(); + + process_update( + &sink, + &mut tracker, + update(7, genesis, block(10, 10), safe, safe), + 7, + genesis, + ) + .await + .expect("initial update"); + process_update( + &sink, + &mut tracker, + update(7, genesis, block(9, 19), safe, safe), + 7, + genesis, + ) + .await + .expect("lower canonical head is a valid reorg"); + + let states = sink.0.lock().expect("recording sink lock"); + assert_eq!(states.len(), 2); + assert_eq!(states[1].head_block_hash, B256::repeat_byte(19)); + } + + #[test] + fn retains_finality_across_reconnects() { + let accepted = ResolvedForkchoice { + state: ForkchoiceState { + head_block_hash: block(10, 10).hash, + safe_block_hash: block(9, 9).hash, + finalized_block_hash: block(8, 8).hash, + }, + finality: Finality { + safe: block(9, 9), + finalized: block(8, 8), + }, + }; + let mut tracker = ForkchoiceTracker::default(); + tracker.accept(accepted); + + let error = super::validate_forkchoice( + block(11, 11), + block(8, 8), + block(8, 8), + tracker.last_finality, + ) + .expect_err("a reconnect must not forget accepted finality"); + assert!(matches!(error, SubscriptionError::MalformedFinality(_))); + } + + #[test] + fn invalid_forkchoice_is_not_retried_until_state_changes() { + let resolved = ResolvedForkchoice { + state: ForkchoiceState { + head_block_hash: block(10, 10).hash, + safe_block_hash: block(9, 9).hash, + finalized_block_hash: block(8, 8).hash, + }, + finality: Finality { + safe: block(9, 9), + finalized: block(8, 8), + }, + }; + let mut tracker = ForkchoiceTracker::default(); + tracker.reject(resolved.state); + assert!(tracker.should_skip(resolved)); + } +} diff --git a/crates/node/src/lib.rs b/crates/node/src/lib.rs index 543d5196..785dad04 100644 --- a/crates/node/src/lib.rs +++ b/crates/node/src/lib.rs @@ -21,6 +21,8 @@ pub mod error; pub mod evm_executor; /// Executor wiring for EV aware execution. pub mod executor; +/// Push-based forkchoice sharing between ev-reth peers. +pub mod head; /// Node composition and payload types. pub mod node; /// Payload service integration. diff --git a/crates/tests/Cargo.toml b/crates/tests/Cargo.toml index eab44b44..56a737aa 100644 --- a/crates/tests/Cargo.toml +++ b/crates/tests/Cargo.toml @@ -70,6 +70,8 @@ async-trait.workspace = true futures.workspace = true eyre.workspace = true tracing.workspace = true +jsonrpsee = { workspace = true, features = ["server"] } +url.workspace = true [lints] workspace = true diff --git a/crates/tests/src/e2e_tests.rs b/crates/tests/src/e2e_tests.rs index 29f32ad5..3e484a38 100644 --- a/crates/tests/src/e2e_tests.rs +++ b/crates/tests/src/e2e_tests.rs @@ -1,4 +1,6 @@ -use alloy_consensus::{transaction::TxHashRef, SignableTransaction, TxEnvelope, TxReceipt}; +use alloy_consensus::{ + transaction::TxHashRef, SignableTransaction, TransactionEnvelope, TxEnvelope, TxReceipt, +}; use alloy_eips::{eip2718::Encodable2718, eip2930::AccessList, BlockNumberOrTag}; use alloy_network::{eip2718::Decodable2718, ReceiptResponse}; use alloy_primitives::{address, Address, Bytes, Signature, TxKind, B256, U256}; @@ -12,11 +14,13 @@ use alloy_rpc_types::{ use alloy_rpc_types_engine::{ForkchoiceState, PayloadAttributes, PayloadStatusEnum}; use alloy_signer::SignerSync; use alloy_sol_types::{sol, SolCall}; +use async_trait::async_trait; use eyre::Result; use futures::future; +use jsonrpsee::server::ServerBuilder; use reth_e2e_test_utils::{ testsuite::{ - actions::MakeCanonical, + actions::{Action, MakeCanonical, WaitForSync}, setup::{NetworkSetup, Setup}, BlockInfo, Environment, TestBuilder, }, @@ -24,13 +28,17 @@ use reth_e2e_test_utils::{ wallet::Wallet, }; use reth_rpc_api::clients::{EngineApiClient, EthApiClient}; +use std::time::Duration; use crate::common::{ create_test_chain_spec, create_test_chain_spec_with_base_fee_sink, create_test_chain_spec_with_deploy_allowlist, create_test_chain_spec_with_mint_admin, e2e_test_tree_config, TEST_CHAIN_ID, }; -use ev_node::rpc::{EvRpcReceipt, EvRpcTransaction, EvTransactionRequest}; +use ev_node::{ + head::{ForkchoiceSink, HeadApiServer, HeadPublisher}, + rpc::{EvRpcReceipt, EvRpcTransaction, EvTransactionRequest}, +}; use ev_precompiles::mint::MINT_PRECOMPILE_ADDR; use ev_primitives::{Call, EvNodeTransaction, EvTxEnvelope}; @@ -70,6 +78,30 @@ const REVERT_INITCODE: [u8; 17] = [ 0xfd, ]; +async fn start_head_peer( + publisher: HeadPublisher, +) -> Result<(url::Url, jsonrpsee::server::ServerHandle)> { + let server = ServerBuilder::default().build("127.0.0.1:0").await?; + let endpoint = format!("ws://{}", server.local_addr()?).parse()?; + Ok((endpoint, server.start(publisher.into_rpc()))) +} + +#[derive(Clone)] +struct EngineHttpForkchoiceSink(C); + +#[async_trait] +impl ForkchoiceSink for EngineHttpForkchoiceSink +where + C: jsonrpsee::core::client::ClientT + Send + Sync, +{ + async fn apply_forkchoice(&self, state: ForkchoiceState) -> Result { + EngineApiClient::::fork_choice_updated_v3(&self.0, state, None) + .await + .map(|response| response.payload_status.status) + .map_err(|error| error.to_string()) + } +} + /// Computes the contract address that will be created by a deployer at a given nonce. /// /// Uses the CREATE opcode address derivation formula: keccak256(rlp([sender, nonce])). @@ -243,6 +275,200 @@ async fn test_e2e_single_node_produces_blocks() -> Result<()> { .await } +/// Verifies that an ev-reth peer can fetch and execute Evolve blocks through native Reth P2P. +/// +/// Node 0 builds a three-block chain locally, including a type `0x76` `EvNode` transaction. Node +/// 1 receives only a pushed forkchoice update for Node 0's final head; it never receives a payload +/// via `engine_newPayload` or a block through WebSocket. It must therefore retrieve headers and +/// bodies through Reth P2P before the final forkchoice update can be accepted as valid. +#[tokio::test(flavor = "multi_thread")] +async fn test_e2e_subscriber_syncs_evnode_block_over_native_reth_p2p() -> Result<()> { + reth_tracing::init_test_tracing(); + + let chain_spec = create_test_chain_spec(); + let chain_id = chain_spec.chain().id(); + let mut setup = Setup::::default() + .with_chain_spec(chain_spec) + .with_network(NetworkSetup::multi_node(2)) + .with_dev_mode(false) + .with_tree_config(e2e_test_tree_config()); + let mut env = Environment::::default(); + setup.apply::(&mut env).await?; + + let parent_block = env.node_clients[0] + .get_block_by_number(BlockNumberOrTag::Latest) + .await? + .expect("sequencer genesis block should exist"); + let mut parent_hash = parent_block.header.hash; + let mut parent_timestamp = parent_block.header.inner.timestamp; + let mut parent_number = parent_block.header.inner.number; + let gas_limit = parent_block.header.inner.gas_limit; + + let mut wallets = Wallet::new(3).with_chain_id(chain_id).wallet_gen(); + let executor = wallets.remove(0); + let sponsor = wallets.remove(0); + let recipient = Address::random(); + let executor_address = executor.address(); + let executor_nonce = EthApiClient::< + TransactionRequest, + Transaction, + Block, + Receipt, + Header, + Bytes, + >::transaction_count( + &env.node_clients[0].rpc, + executor_address, + Some(BlockId::latest()), + ) + .await?; + let executor_nonce = u64::try_from(executor_nonce).expect("nonce fits into u64"); + + let transfer_value = U256::from(1_000_000_000_000_000u64); + let ev_tx = EvNodeTransaction { + chain_id, + nonce: executor_nonce, + max_priority_fee_per_gas: 1_000_000_000, + max_fee_per_gas: 2_000_000_000, + gas_limit: 100_000, + calls: vec![Call { + to: TxKind::Call(recipient), + value: transfer_value, + input: Bytes::default(), + }], + access_list: AccessList::default(), + fee_payer_signature: None, + }; + let executor_sig = executor + .sign_hash_sync(&ev_tx.signature_hash()) + .expect("executor signature"); + let mut signed = ev_tx.into_signed(executor_sig); + let sponsor_hash = signed.tx().sponsor_signing_hash(executor_address); + signed.tx_mut().fee_payer_signature = Some( + sponsor + .sign_hash_sync(&sponsor_hash) + .expect("sponsor signature"), + ); + let ev_tx = EvTxEnvelope::EvNode(signed); + let ev_tx_hash = *ev_tx.tx_hash(); + + build_block_with_transactions( + &mut env, + &mut parent_hash, + &mut parent_number, + &mut parent_timestamp, + Some(gas_limit), + vec![ev_tx.encoded_2718().into()], + Address::ZERO, + ) + .await?; + build_block_with_transactions( + &mut env, + &mut parent_hash, + &mut parent_number, + &mut parent_timestamp, + Some(gas_limit), + vec![], + Address::ZERO, + ) + .await?; + build_block_with_transactions( + &mut env, + &mut parent_hash, + &mut parent_number, + &mut parent_timestamp, + Some(gas_limit), + vec![], + Address::ZERO, + ) + .await?; + + let source_head = BlockInfo { + hash: parent_hash, + number: parent_number, + timestamp: parent_timestamp, + }; + let follower_head_before = env.node_clients[1] + .get_block_by_number(BlockNumberOrTag::Latest) + .await? + .expect("follower genesis block should exist"); + assert_ne!( + follower_head_before.header.hash, source_head.hash, + "subscriber must not receive the source blocks before forkchoice is sent" + ); + + let genesis_hash = parent_block.header.hash; + let publisher = HeadPublisher::new(chain_id, genesis_hash); + let (peer_ws, ws_server) = start_head_peer(publisher.clone()).await?; + let peer_subscriber = tokio::spawn(ev_node::head::subscribe( + EngineHttpForkchoiceSink(env.node_clients[1].engine.http_client()), + peer_ws, + chain_id, + genesis_hash, + )); + // Wait for the actual custom subscription path before publishing the source's current state. + tokio::time::timeout(Duration::from_secs(5), async { + while publisher.subscriber_count() == 0 { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("peer must subscribe to forkchoice updates"); + assert!(publisher.publish( + ForkchoiceState { + head_block_hash: source_head.hash, + safe_block_hash: source_head.hash, + finalized_block_hash: source_head.hash, + }, + source_head.number, + source_head.number, + source_head.number, + )); + + let mut wait_for_sync = WaitForSync::new(0, 1).with_timeout(60); + wait_for_sync.execute(&mut env).await?; + + let follower_head = env.node_clients[1] + .get_block_by_number(BlockNumberOrTag::Latest) + .await? + .expect("follower head should exist after synchronization"); + assert_eq!(follower_head.header.hash, source_head.hash); + assert_eq!(follower_head.header.inner.number, source_head.number); + + let type_0x76_tx = EthApiClient::< + EvTransactionRequest, + EvRpcTransaction, + Block, + EvRpcReceipt, + Header, + Bytes, + >::transaction_by_hash(&env.node_clients[1].rpc, ev_tx_hash) + .await? + .expect("follower should serve the type 0x76 transaction fetched through P2P"); + assert_eq!( + type_0x76_tx.inner().inner.tx_type(), + ev_primitives::EVNODE_TX_TYPE_ID + ); + + let follower_recipient_balance = + EthApiClient::::balance( + &env.node_clients[1].rpc, + recipient, + Some(BlockId::latest()), + ) + .await?; + assert_eq!( + follower_recipient_balance, transfer_value, + "follower must execute the P2P-fetched type 0x76 transaction" + ); + + peer_subscriber.abort(); + ws_server.stop()?; + + drop(setup); + Ok(()) +} + /// Tests that the base fee sink address correctly receives base fees and priority tips. /// /// # Test Flow diff --git a/docs/guide/peer-head-subscription.md b/docs/guide/peer-head-subscription.md new file mode 100644 index 00000000..84179ed3 --- /dev/null +++ b/docs/guide/peer-head-subscription.md @@ -0,0 +1,326 @@ +# Peer Head Subscription Guide + +This guide explains how to run ev-reth full nodes without an EV node. The sequencer still runs its +EV node and ev-reth. Other ev-reth nodes subscribe to canonical forkchoice updates over WebSocket +and fetch the referenced blocks through native Reth P2P. + +## When to Use This Mode + +Use `--subscribe-peer` for an ev-reth full node that: + +- does not produce blocks; +- does not run an EV node; +- needs to follow the canonical head selected by another ev-reth node; and +- has native P2P connectivity to at least one node that stores the referenced blocks. + +Do not use this mode to replace the sequencer or its EV node. The subscription distributes +forkchoice decisions; it does not create blocks or run consensus. + +## Architecture + +The control plane and block-data plane are separate: + +```text + control plane: WebSocket +EV node ──Engine API──> ev-reth A ──ev_subscribeForkchoice──> ev-reth B + sequencer or subscribing + synchronized peer full node + │ + │ native Reth P2P + ▼ + ev-reth C + block-data peer +``` + +`ev-reth A` and `ev-reth C` may be the same node. They do not have to be. A synchronized full node +can also act as `ev-reth A`, which allows subscriber-to-subscriber relay without adding an EV node. + +The two connections serve different purposes: + +| Connection | Carries | Establishes | +| --- | --- | --- | +| WebSocket | Chain identity and head, safe, and finalized references | Which valid forkchoice the subscriber should follow | +| Native Reth P2P | Headers, block bodies, and transactions | Where the subscriber obtains block data | + +The WebSocket never carries execution payloads or blocks. The subscriber does not poll +`eth_blockNumber` or call `eth_getBlockByNumber` on the publishing peer. + +## How an Update Moves Through the System + +1. The sequencer's EV node builds and imports a block through the Engine API as usual. +2. The publishing ev-reth validates the block and accepts a forkchoice update as `VALID`. +3. The publisher resolves the head, safe, and finalized block numbers from its local database. +4. `ev_subscribeForkchoice` pushes the update to connected subscribers. +5. A subscriber checks the chain ID, genesis hash, and finality ordering. +6. The subscriber passes the payload-free forkchoice state directly to its local Reth engine. +7. If the target is unknown, Reth returns `SYNCING` and downloads the missing chain through P2P. +8. The subscriber retries the pending forkchoice locally every three seconds until Reth returns + `VALID` or `INVALID`. +9. After accepting the update as `VALID`, the subscriber can publish the same state to downstream + subscribers. + +Only locally valid forkchoices are published. A node that is still downloading a target does not +relay it as valid. + +## Requirements + +Before starting a subscriber, confirm: + +- The publishing and subscribing nodes run an ev-reth version that supports + `ev_subscribeForkchoice`. +- Both nodes use the same chain ID and genesis hash. Use the same chainspec for the safest setup. +- The publishing peer has WebSocket RPC enabled. +- The subscriber can reach the publishing peer's WebSocket endpoint. +- The subscriber has at least one native P2P peer that has the announced blocks. +- Firewalls permit the selected WebSocket and P2P ports. + +Default ports used in the examples: + +| Service | Default | Required direction | +| --- | --- | --- | +| WebSocket RPC | TCP 8546 | Subscriber to publishing peer | +| Reth P2P | TCP 30303 | Between the subscriber and block-data peers | +| Discv4 discovery | UDP 30303 | Optional when static trusted peers are configured | +| HTTP RPC | TCP 8545 | Optional, for monitoring and user RPC | +| Authenticated Engine API | TCP 8551 | EV node to sequencer ev-reth only | + +The subscription does not use the authenticated Engine API on the full node. + +## Configure a Publishing Peer + +Every ev-reth node starts the forkchoice publisher internally. Enable WebSocket RPC to make the +subscription available: + +```bash +./target/release/ev-reth node \ + --chain /path/to/genesis.json \ + --datadir /var/lib/ev-reth-source \ + --ws \ + --ws.addr 0.0.0.0 \ + --ws.port 8546 \ + --addr 0.0.0.0 \ + --port 30303 \ + --http \ + --http.addr 127.0.0.1 \ + --http.api eth,net +``` + +The sequencer's EV node and Engine API configuration remain unchanged. Enabling this WebSocket +endpoint does not move block production into ev-reth. + +The `ev_subscribeForkchoice` method is a custom module merged into every enabled WebSocket server. +Do not add `ev` to `--ws.api`; `ev` is not a standard Reth module name and no `--ws.api` change is +required. + +Binding `--ws.addr 0.0.0.0` exposes the endpoint on every interface. Use it only on a private +network or with firewall rules that restrict subscriber addresses. + +## Configure Native P2P + +The subscriber needs a native Reth peer with the announced blocks. For a static private topology, +configure one or more comma-separated enode URLs: + +```text +--trusted-peers enode://@:30303 +``` + +The publishing peer is usually the simplest block-data peer, but a different synchronized ev-reth +node works as well. The WebSocket URL and `--trusted-peers` enode do not need to identify the same +machine. + +For a closed topology, add: + +```text +--trusted-only --disable-discovery +``` + +This restricts the node to configured trusted peers and disables discovery. Provide more than one +trusted peer in production if the node must continue syncing through a peer outage. + +`--trusted-peers` controls connection policy. It does not make block data authoritative. Reth still +validates downloaded headers, bodies, transactions, and state transitions locally. + +## Start a Subscribing Full Node + +Start the full node with the same chainspec, a peer WebSocket URL, and native P2P peers: + +```bash +./target/release/ev-reth node \ + --chain /path/to/genesis.json \ + --datadir /var/lib/ev-reth-subscriber \ + --subscribe-peer wss://peer.example/rpc \ + --trusted-peers enode://@:30303 \ + --http \ + --http.addr 127.0.0.1 \ + --http.api eth,net,web3 +``` + +For a direct private-network connection without TLS, use `ws://host:8546`. Use `wss://` when a TLS +proxy terminates the connection in front of ev-reth. + +The equivalent environment variable is: + +```bash +EV_SUBSCRIBE_PEER=wss://peer.example/rpc \ +./target/release/ev-reth node \ + --chain /path/to/genesis.json \ + --datadir /var/lib/ev-reth-subscriber \ + --trusted-peers enode://@:30303 +``` + +Only one subscription endpoint can be configured. If that endpoint is unavailable, the full node +continues running and reconnects to the same endpoint with bounded backoff. + +## Relay Through a Synchronized Full Node + +A subscriber can serve downstream subscribers after its local engine validates the received +forkchoice. Enable WebSocket RPC on that node: + +```bash +./target/release/ev-reth node \ + --chain /path/to/genesis.json \ + --datadir /var/lib/ev-reth-relay \ + --subscribe-peer wss://upstream.example/rpc \ + --trusted-peers enode://@:30303 \ + --ws \ + --ws.addr 0.0.0.0 \ + --ws.port 8546 +``` + +Downstream nodes can then use `wss://relay.example/rpc` as their `--subscribe-peer` endpoint. The +relay still needs P2P connectivity because it validates and stores blocks before publishing their +forkchoice. + +## Verify the Setup + +### 1. Verify the WebSocket method + +With `websocat` installed, connect to the publishing peer: + +```bash +websocat ws://127.0.0.1:8546 +``` + +Send: + +```json +{"jsonrpc":"2.0","id":1,"method":"ev_subscribeForkchoice","params":[]} +``` + +The server returns a subscription ID. If a valid state is already available, it also sends the +latest update. The update result has this shape: + +```json +{ + "chainId": 1234, + "genesisHash": "0x...", + "forkchoiceState": { + "headBlockHash": "0x...", + "safeBlockHash": "0x...", + "finalizedBlockHash": "0x..." + }, + "headBlockNumber": 42, + "safeBlockNumber": 42, + "finalizedBlockNumber": 42 +} +``` + +### 2. Check subscriber logs + +Use module-specific debug logging: + +```bash +RUST_LOG=info,ev_node::head=debug \ +./target/release/ev-reth node \ + --chain /path/to/genesis.json \ + --datadir /var/lib/ev-reth-subscriber \ + --subscribe-peer ws://peer.internal:8546 \ + --trusted-peers enode://@:30303 +``` + +Expected messages include: + +```text +subscribed to peer forkchoice updates +local engine is fetching peer target through P2P +applied peer forkchoice to local engine +``` + +The second message is logged only while the target is missing, and the last two require debug +logging. + +### 3. Check P2P connectivity + +If HTTP RPC exposes the `net` module: + +```bash +curl -s \ + -H 'content-type: application/json' \ + --data '{"jsonrpc":"2.0","id":1,"method":"net_peerCount","params":[]}' \ + http://127.0.0.1:8545 +``` + +A zero peer count explains why an unknown head cannot be downloaded. + +### 4. Compare block height + +Query `eth_blockNumber` on the publishing and subscribing nodes: + +```bash +curl -s \ + -H 'content-type: application/json' \ + --data '{"jsonrpc":"2.0","id":1,"method":"eth_blockNumber","params":[]}' \ + http://127.0.0.1:8545 +``` + +The subscriber may lag while it downloads and executes blocks. It should converge to the +publishing peer's height when both the WebSocket control path and P2P data path are healthy. + +## Failure Handling + +The subscription task is non-critical. A transient subscription failure does not stop the node or +its other RPC and P2P services. + +| Symptom or log | Meaning | Action | +| --- | --- | --- | +| `peer head subscription failed; reconnecting` | The WebSocket connection, RPC method, or subscription failed | Check the URL, proxy, firewall, TLS certificate, and ev-reth version on the peer | +| `peer forkchoice subscription ended; reconnecting` | An established WebSocket subscription closed | Check proxy idle timeouts and peer restarts | +| `local engine is fetching peer target through P2P` repeats | The control path works, but the target is not locally available yet | Check `net_peerCount`, enode addresses, P2P ports, network ID, and whether peers have the block | +| `peer violated head subscription trust policy; stopping subscriber` | Chain identity differs or safe/finalized regressed | Compare chainspecs and upstream state; correct the endpoint, then restart the subscriber | +| `local engine rejected peer forkchoice` | The downloaded chain or announced forkchoice failed local validation | Inspect the validation error and publishing peer; do not bypass validation | +| `could not resolve valid forkchoice locally` on a publisher | The publisher accepted an event but cannot resolve every referenced hash in its local database | Check local database/provider health and confirm safe and finalized references are available | +| WebSocket connects but no update arrives | The publisher has not accepted a complete valid forkchoice yet | Check block production and Engine API activity on the publishing peer | + +Connection and subscription requests time out after 15 seconds. Reconnect delay starts at one +second and grows to at most 30 seconds. A stable session resets the delay. + +The publisher retains one latest state instead of an unbounded history. A slow or reconnecting +subscriber receives current state and uses P2P to fill any missing history. + +## Trust and Security + +The configured WebSocket endpoint is authoritative for forkchoice. It can choose which valid fork +the subscriber follows. Local execution validation prevents invalid blocks from being imported, +but it does not choose between two otherwise valid forks. + +Apply these controls: + +- Restrict the WebSocket endpoint to known subscriber networks. +- Terminate `wss://` at a TLS proxy because ev-reth's WS listener is plain WebSocket. +- Do not expose `admin`, `debug`, or other unnecessary RPC modules on the subscription endpoint. +- Monitor unexpected finality regressions and repeated validation failures. +- Use multiple P2P peers for data availability, even when one WebSocket peer controls forkchoice. + +`--subscribe-peer` currently accepts only a URL. It has no option for custom authorization headers +or client certificates. Protect the endpoint with network controls or a compatible TLS proxy, and +do not assume the Engine API JWT protects the regular WebSocket server. The Engine API and its JWT +remain a separate connection between the EV node and the sequencer's ev-reth. + +## Current Limitations + +- One authoritative subscription endpoint per node. +- No quorum or comparison across multiple forkchoice publishers. +- No block or payload transfer over WebSocket. +- No historical event replay; the publisher retains only the latest valid state. +- No custom WebSocket authentication headers from the subscriber. +- Vanilla Reth does not expose `ev_subscribeForkchoice`; the publishing peer must run ev-reth.