Add push-based peer head subscription

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

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://<public-key>@<block-peer>: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

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::<EvolveChainSpecParser, EvolveArgs>::parse().run(|builder, _evolve_args| async move {
+        Cli::<EvolveChainSpecParser, EvolveArgs>::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
         })

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

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<Url>,
+}
+
+fn parse_websocket_url(value: &str) -> Result<Url, String> {
+    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());
+    }
+}