Merge pull request #125 from decipherhub/fix/eth-subscribe

fix: broadcast blocks to eth_subscribe subscribers

Author: qj0r9j0vc2

crates/node/src/node.rs

@@ -756,6 +756,16 @@ impl Node {
             None
         };
 
+        // Create subscription manager for WebSocket subscriptions (eth_subscribe)
+        // This needs to be shared between the RPC server and the event loop
+        // so that new blocks can be broadcast to subscribers
+        let subscription_manager = if rpc_storage.is_some() {
+            use cipherbft_rpc::SubscriptionManager;
+            Some(Arc::new(SubscriptionManager::default()))
+        } else {
+            None
+        };
+
         // Start RPC server if enabled
         if let Some(ref storage) = rpc_storage {
             use cipherbft_rpc::{
@@ -772,8 +782,16 @@ impl Node {
             let executor = Arc::new(StubExecutionApi::new());
             let network = Arc::new(StubNetworkApi::new());
 
-            let rpc_server =
-                RpcServer::new(rpc_config, storage.clone(), mempool, executor, network);
+            // Use with_subscription_manager to share the subscription manager
+            // between the RPC server and the event loop for broadcasting blocks
+            let rpc_server = RpcServer::with_subscription_manager(
+                rpc_config,
+                storage.clone(),
+                mempool,
+                executor,
+                network,
+                subscription_manager.clone().unwrap(),
+            );
 
             let http_port = self.config.rpc_http_port;
             let ws_port = self.config.rpc_ws_port;
@@ -819,6 +837,7 @@ impl Node {
             &mut decided_rx,
             execution_bridge,
             rpc_storage,
+            subscription_manager,
         )
         .await;
 
@@ -840,6 +859,7 @@ impl Node {
     }
 
     /// Internal event loop that handles messages and can be cancelled.
+    #[allow(clippy::too_many_arguments)]
     async fn run_event_loop(
         cancel_token: CancellationToken,
         primary_incoming_rx: &mut mpsc::Receiver<(ValidatorId, DclMessage)>,
@@ -848,6 +868,7 @@ impl Node {
         decided_rx: &mut mpsc::Receiver<(ConsensusHeight, Cut)>,
         execution_bridge: Option<Arc<ExecutionBridge>>,
         rpc_storage: Option<Arc<cipherbft_rpc::MdbxRpcStorage<InMemoryProvider>>>,
+        subscription_manager: Option<Arc<cipherbft_rpc::SubscriptionManager>>,
     ) -> Result<()> {
         loop {
             tokio::select! {
@@ -940,6 +961,14 @@ impl Node {
                                         error!("Failed to store block {} to MDBX: {}", height.0, e);
                                     } else {
                                         debug!("Stored block {} to MDBX with hash {}", height.0, block_result.block_hash);
+
+                                        // Broadcast to WebSocket subscribers (eth_subscribe("newHeads"))
+                                        if let Some(ref sub_mgr) = subscription_manager {
+                                            use cipherbft_rpc::storage_block_to_rpc_block;
+                                            let rpc_block = storage_block_to_rpc_block(block.clone(), false);
+                                            sub_mgr.broadcast_block(rpc_block);
+                                            debug!("Broadcast block {} to WebSocket subscribers", height.0);
+                                        }
                                     }
 
                                     // Store receipts for eth_getBlockReceipts queries

crates/rpc/src/adapters.rs

@@ -1811,6 +1811,76 @@ impl NetworkApi for StubNetworkApi {
     }
 }
 
+/// Convert a storage block to an RPC block format.
+///
+/// This is a standalone function for use by the node when broadcasting
+/// new blocks to WebSocket subscribers via `eth_subscribe("newHeads")`.
+pub fn storage_block_to_rpc_block(
+    storage_block: cipherbft_storage::blocks::Block,
+    full_txs: bool,
+) -> Block {
+    use alloy_primitives::{Bloom, B64};
+
+    // Convert transaction hashes to B256
+    let tx_hashes: Vec<B256> = storage_block
+        .transaction_hashes
+        .iter()
+        .map(|h| B256::from(*h))
+        .collect();
+
+    // Build transactions field based on full_txs flag
+    let transactions = if full_txs {
+        // TODO: In the future, this should return full Transaction objects
+        BlockTransactions::Hashes(tx_hashes)
+    } else {
+        BlockTransactions::Hashes(tx_hashes)
+    };
+
+    // Build the consensus header
+    let consensus_header = alloy_consensus::Header {
+        parent_hash: B256::from(storage_block.parent_hash),
+        ommers_hash: B256::from(storage_block.ommers_hash),
+        beneficiary: Address::from(storage_block.beneficiary),
+        state_root: B256::from(storage_block.state_root),
+        transactions_root: B256::from(storage_block.transactions_root),
+        receipts_root: B256::from(storage_block.receipts_root),
+        logs_bloom: Bloom::from_slice(&storage_block.logs_bloom),
+        difficulty: U256::from_be_bytes(storage_block.difficulty),
+        number: storage_block.number,
+        gas_limit: storage_block.gas_limit,
+        gas_used: storage_block.gas_used,
+        timestamp: storage_block.timestamp,
+        extra_data: Bytes::from(storage_block.extra_data.clone()),
+        mix_hash: B256::from(storage_block.mix_hash),
+        nonce: B64::from(storage_block.nonce),
+        base_fee_per_gas: storage_block.base_fee_per_gas,
+        withdrawals_root: None,
+        blob_gas_used: None,
+        excess_blob_gas: None,
+        parent_beacon_block_root: None,
+        requests_hash: None,
+    };
+
+    // Build the RPC header with hash and total difficulty
+    let block_hash = B256::from(storage_block.hash);
+    let total_difficulty = U256::from_be_bytes(storage_block.total_difficulty);
+
+    let rpc_header = Header {
+        hash: block_hash,
+        inner: consensus_header,
+        total_difficulty: Some(total_difficulty),
+        size: None,
+    };
+
+    // Build the final RPC block
+    Block {
+        header: rpc_header,
+        uncles: Vec::new(),
+        transactions,
+        withdrawals: None,
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/rpc/src/lib.rs

@@ -167,6 +167,9 @@ pub use adapters::{StubExecutionApi, StubMempoolApi, StubNetworkApi, StubRpcStor
 // Real implementations backed by storage
 pub use adapters::{EvmExecutionApi, MdbxRpcStorage, PoolMempoolApi, ProviderBasedRpcStorage};
 
+// Block conversion utilities for subscription broadcasting
+pub use adapters::storage_block_to_rpc_block;
+
 // RPC server traits (for method registration)
 pub use eth::EthRpcServer;
 pub use net::NetRpcServer;

crates/rpc/src/server.rs

@@ -72,14 +72,37 @@ where
         mempool: Arc<M>,
         executor: Arc<E>,
         network: Arc<N>,
+    ) -> Self {
+        Self::with_subscription_manager(
+            config,
+            storage,
+            mempool,
+            executor,
+            network,
+            Arc::new(SubscriptionManager::default()),
+        )
+    }
+
+    /// Create a new RPC server with an external subscription manager.
+    ///
+    /// This allows sharing the subscription manager with other components
+    /// (e.g., the node's event loop) to enable broadcasting events to
+    /// WebSocket subscribers.
+    pub fn with_subscription_manager(
+        config: RpcConfig,
+        storage: Arc<S>,
+        mempool: Arc<M>,
+        executor: Arc<E>,
+        network: Arc<N>,
+        subscription_manager: Arc<SubscriptionManager>,
     ) -> Self {
         Self {
             config: Arc::new(config),
             storage,
             mempool,
             executor,
             network,
-            subscription_manager: Arc::new(SubscriptionManager::default()),
+            subscription_manager,
             state: Arc::new(RwLock::new(ServerState::Stopped)),
             http_handle: Arc::new(RwLock::new(None)),
             ws_handle: Arc::new(RwLock::new(None)),
@@ -96,6 +119,14 @@ where
         &self.subscription_manager
     }
 
+    /// Get the subscription manager as an Arc.
+    ///
+    /// This is useful for sharing the subscription manager with other components
+    /// that need to broadcast events to WebSocket subscribers.
+    pub fn subscription_manager_arc(&self) -> Arc<SubscriptionManager> {
+        Arc::clone(&self.subscription_manager)
+    }
+
     /// Get the current server state.
     pub async fn state(&self) -> ServerState {
         *self.state.read().await