feat: pending logs subscription

bin/reth/Cargo.toml

@@ -92,8 +92,6 @@ comfy-table = "7.0"
 crossterm = "0.27.0"
 ratatui = "0.25.0"
 human_bytes = "0.4.1"
-
-# async
 tokio = { workspace = true, features = ["sync", "macros", "time", "rt-multi-thread"] }
 futures.workspace = true
 pin-project.workspace = true

crates/rpc/rpc-builder/Cargo.toml

@@ -42,6 +42,7 @@ strum = { workspace = true, features = ["derive"] }
 serde = { workspace = true, features = ["derive"] }
 thiserror.workspace = true
 tracing.workspace = true
+tokio.workspace = true
 
 [dev-dependencies]
 reth-tracing.workspace = true

crates/rpc/rpc-builder/src/auth.rs

@@ -4,6 +4,9 @@ use crate::{
     error::{RpcError, ServerKind},
     EthConfig,
 };
+
+use tokio::sync::watch;
+
 use hyper::header::AUTHORIZATION;
 pub use jsonrpsee::server::ServerBuilder;
 use jsonrpsee::{
@@ -13,10 +16,12 @@ use jsonrpsee::{
 };
 use reth_network_api::{NetworkInfo, Peers};
 use reth_node_api::EngineTypes;
+use reth_primitives::{Receipt, SealedBlock};
 use reth_provider::{
     BlockReaderIdExt, ChainSpecProvider, EvmEnvProvider, HeaderProvider, ReceiptProviderIdExt,
     StateProviderFactory,
 };
+
 use reth_rpc::{
     eth::{
         cache::EthStateCache, gas_oracle::GasPriceOracle, EthFilterConfig, FeeHistoryCache,
@@ -68,6 +73,9 @@ where
 
     let fee_history_cache =
         FeeHistoryCache::new(eth_cache.clone(), FeeHistoryCacheConfig::default());
+
+    let initial_value: Option<(SealedBlock, Vec<Receipt>)> = None;
+    let (local_pending_block_watcher_tx, _) = watch::channel(initial_value);
     let eth_api = EthApi::with_spawner(
         provider.clone(),
         pool.clone(),
@@ -78,6 +86,7 @@ where
         Box::new(executor.clone()),
         BlockingTaskPool::build().expect("failed to build tracing pool"),
         fee_history_cache,
+        local_pending_block_watcher_tx,
     );
     let config = EthFilterConfig::default()
         .max_logs_per_response(DEFAULT_MAX_LOGS_PER_RESPONSE)

crates/rpc/rpc-builder/src/lib.rs

@@ -144,24 +144,19 @@ use std::{
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
 
+use constants::*;
+use error::{RpcError, ServerKind};
 use hyper::{header::AUTHORIZATION, HeaderMap};
 pub use jsonrpsee::server::ServerBuilder;
 use jsonrpsee::{
     server::{IdProvider, Server, ServerHandle},
     Methods, RpcModule,
 };
-use reth_node_api::EngineTypes;
-use serde::{Deserialize, Serialize, Serializer};
-use strum::{AsRefStr, EnumIter, EnumVariantNames, IntoStaticStr, ParseError, VariantNames};
-use tower::layer::util::{Identity, Stack};
-use tower_http::cors::CorsLayer;
-use tracing::{instrument, trace};
-
-use constants::*;
-use error::{RpcError, ServerKind};
 use reth_ipc::server::IpcServer;
 pub use reth_ipc::server::{Builder as IpcServerBuilder, Endpoint};
 use reth_network_api::{noop::NoopNetwork, NetworkInfo, Peers};
+use reth_node_api::EngineTypes;
+use reth_primitives::{Receipt, SealedBlock};
 use reth_provider::{
     AccountReader, BlockReader, BlockReaderIdExt, CanonStateSubscriptions, ChainSpecProvider,
     ChangeSetReader, EvmEnvProvider, StateProviderFactory,
@@ -180,6 +175,12 @@ use reth_rpc::{
 use reth_rpc_api::{servers::*, EngineApiServer};
 use reth_tasks::{TaskSpawner, TokioTaskExecutor};
 use reth_transaction_pool::{noop::NoopTransactionPool, TransactionPool};
+use serde::{Deserialize, Serialize, Serializer};
+use strum::{AsRefStr, EnumIter, EnumVariantNames, IntoStaticStr, ParseError, VariantNames};
+use tokio::sync::watch;
+use tower::layer::util::{Identity, Stack};
+use tower_http::cors::CorsLayer;
+use tracing::{instrument, trace};
 
 use crate::{
     auth::AuthRpcModule, error::WsHttpSamePortError, metrics::RpcServerMetrics,
@@ -1224,6 +1225,11 @@ where
             let executor = Box::new(self.executor.clone());
             let blocking_task_pool =
                 BlockingTaskPool::build().expect("failed to build tracing pool");
+
+            let initial_value: Option<(SealedBlock, Vec<Receipt>)> = None; // or some initial value
+            let (local_pending_block_watcher_tx, local_pending_block_watcher_rx) =
+                watch::channel(initial_value);
+
             let api = EthApi::with_spawner(
                 self.provider.clone(),
                 self.pool.clone(),
@@ -1234,6 +1240,7 @@ where
                 executor.clone(),
                 blocking_task_pool.clone(),
                 fee_history_cache,
+                local_pending_block_watcher_tx,
             );
             let filter = EthFilter::new(
                 self.provider.clone(),
@@ -1249,6 +1256,7 @@ where
                 self.events.clone(),
                 self.network.clone(),
                 executor,
+                local_pending_block_watcher_rx,
             );
 
             let eth = EthHandlers { api, cache, filter, pubsub, blocking_task_pool };

crates/rpc/rpc/src/eth/api/block.rs

@@ -157,7 +157,7 @@ where
     }
 
     /// Returns the block object for the given block id.
-    pub(crate) async fn block_with_senders(
+    pub async fn block_with_senders(
         &self,
         block_id: impl Into<BlockId>,
     ) -> EthResult<Option<reth_primitives::SealedBlockWithSenders>> {

crates/rpc/rpc/src/eth/api/mod.rs

@@ -17,7 +17,8 @@ use reth_interfaces::RethResult;
 use reth_network_api::NetworkInfo;
 use reth_primitives::{
     revm_primitives::{BlockEnv, CfgEnv},
-    Address, BlockId, BlockNumberOrTag, ChainInfo, SealedBlockWithSenders, B256, U256, U64,
+    Address, BlockId, BlockNumberOrTag, ChainInfo, Receipt, SealedBlock, SealedBlockWithSenders,
+    B256, U256, U64,
 };
 
 use reth_provider::{
@@ -33,7 +34,7 @@ use std::{
     time::{Duration, Instant},
 };
 
-use tokio::sync::{oneshot, Mutex};
+use tokio::sync::{oneshot, watch, Mutex};
 
 mod block;
 mod call;
@@ -50,6 +51,12 @@ mod transactions;
 use crate::BlockingTaskPool;
 pub use transactions::{EthTransactions, TransactionSource};
 
+/// Type alias for a watch receiver that receives the recent pending block with receipts
+pub(crate) type LocalPendingBlockWatcherReceiver =
+    watch::Receiver<Option<(SealedBlock, Vec<Receipt>)>>;
+/// Type alias for a watch sender that sends the recent local pending block with receipts
+type LocalPendingBlockWatcherSender = watch::Sender<Option<(SealedBlock, Vec<Receipt>)>>;
+
 /// `Eth` API trait.
 ///
 /// Defines core functionality of the `eth` API implementation.
@@ -102,6 +109,7 @@ where
         gas_cap: impl Into<GasCap>,
         blocking_task_pool: BlockingTaskPool,
         fee_history_cache: FeeHistoryCache,
+        local_pending_block_sender: LocalPendingBlockWatcherSender,
     ) -> Self {
         Self::with_spawner(
             provider,
@@ -113,6 +121,7 @@ where
             Box::<TokioTaskExecutor>::default(),
             blocking_task_pool,
             fee_history_cache,
+            local_pending_block_sender,
         )
     }
 
@@ -128,6 +137,7 @@ where
         task_spawner: Box<dyn TaskSpawner>,
         blocking_task_pool: BlockingTaskPool,
         fee_history_cache: FeeHistoryCache,
+        local_pending_block_sender: LocalPendingBlockWatcherSender,
     ) -> Self {
         // get the block number of the latest block
         let latest_block = provider
@@ -152,6 +162,7 @@ where
             fee_history_cache,
             #[cfg(feature = "optimism")]
             http_client: reqwest::Client::builder().use_rustls_tls().build().unwrap(),
+            local_pending_block_sender,
         };
 
         Self { inner: Arc::new(inner) }
@@ -211,6 +222,11 @@ where
     pub fn fee_history_cache(&self) -> &FeeHistoryCache {
         &self.inner.fee_history_cache
     }
+
+    /// Returns local pending block sender
+    pub fn local_pending_block_sender(&self) -> &LocalPendingBlockWatcherSender {
+        &self.inner.local_pending_block_sender
+    }
 }
 
 // === State access helpers ===
@@ -341,6 +357,11 @@ where
                 expires_at: now + Duration::from_secs(3),
             });
 
+            let receipts = this.cache().get_receipts(pending_block.hash).await?.unwrap_or_default();
+            let pending_block_clone = pending_block.block.clone();
+            this.local_pending_block_sender()
+                .send_modify(|v| *v = Some((pending_block_clone, receipts.to_vec())));
+
             Ok(Some(pending_block))
         })
         .await
@@ -468,6 +489,8 @@ struct EthApiInner<Provider, Pool, Network> {
     blocking_task_pool: BlockingTaskPool,
     /// Cache for block fees history
     fee_history_cache: FeeHistoryCache,
+    /// A tokio watch sender to notify about the most recent locally built pending block
+    local_pending_block_sender: LocalPendingBlockWatcherSender,
     /// An http client for communicating with sequencers.
     #[cfg(feature = "optimism")]
     http_client: reqwest::Client,

crates/rpc/rpc/src/eth/api/server.rs

@@ -404,7 +404,7 @@ mod tests {
     use reth_network_api::noop::NoopNetwork;
     use reth_primitives::{
         basefee::calculate_next_block_base_fee, constants::ETHEREUM_BLOCK_GAS_LIMIT, BaseFeeParams,
-        Block, BlockNumberOrTag, Header, TransactionSigned, B256, U256,
+        Block, BlockNumberOrTag, Header, Receipt, SealedBlock, TransactionSigned, B256, U256,
     };
     use reth_provider::{
         test_utils::{MockEthProvider, NoopProvider},
@@ -413,6 +413,7 @@ mod tests {
     use reth_rpc_api::EthApiServer;
     use reth_rpc_types::FeeHistory;
     use reth_transaction_pool::test_utils::{testing_pool, TestPool};
+    use tokio::sync::watch;
 
     fn build_test_eth_api<
         P: BlockReaderIdExt
@@ -430,6 +431,8 @@ mod tests {
 
         let fee_history_cache =
             FeeHistoryCache::new(cache.clone(), FeeHistoryCacheConfig::default());
+        let initial_value: Option<(SealedBlock, Vec<Receipt>)> = None; // or some initial value
+        let (local_pending_block_watcher_tx, _) = watch::channel(initial_value);
 
         EthApi::new(
             provider.clone(),
@@ -440,6 +443,7 @@ mod tests {
             ETHEREUM_BLOCK_GAS_LIMIT,
             BlockingTaskPool::build().expect("failed to build tracing pool"),
             fee_history_cache,
+            local_pending_block_watcher_tx,
         )
     }
 

crates/rpc/rpc/src/eth/api/state.rs

@@ -131,17 +131,22 @@ mod tests {
         },
         BlockingTaskPool,
     };
-    use reth_primitives::{constants::ETHEREUM_BLOCK_GAS_LIMIT, StorageKey, StorageValue};
+    use reth_primitives::{
+        constants::ETHEREUM_BLOCK_GAS_LIMIT, Receipt, SealedBlock, StorageKey, StorageValue,
+    };
     use reth_provider::test_utils::{ExtendedAccount, MockEthProvider, NoopProvider};
     use reth_transaction_pool::test_utils::testing_pool;
     use std::collections::HashMap;
+    use tokio::sync::watch;
 
     #[tokio::test]
     async fn test_storage() {
         // === Noop ===
         let pool = testing_pool();
 
         let cache = EthStateCache::spawn(NoopProvider::default(), Default::default());
+        let initial_value: Option<(SealedBlock, Vec<Receipt>)> = None;
+        let (local_pending_block_watcher_tx, _) = watch::channel(initial_value.clone());
         let eth_api = EthApi::new(
             NoopProvider::default(),
             pool.clone(),
@@ -151,6 +156,7 @@ mod tests {
             ETHEREUM_BLOCK_GAS_LIMIT,
             BlockingTaskPool::build().expect("failed to build tracing pool"),
             FeeHistoryCache::new(cache.clone(), FeeHistoryCacheConfig::default()),
+            local_pending_block_watcher_tx,
         );
         let address = Address::random();
         let storage = eth_api.storage_at(address, U256::ZERO.into(), None).unwrap();
@@ -164,6 +170,7 @@ mod tests {
         let account = ExtendedAccount::new(0, U256::ZERO).extend_storage(storage);
         mock_provider.add_account(address, account);
 
+        let (local_pending_block_watcher_tx, _) = watch::channel(initial_value);
         let cache = EthStateCache::spawn(mock_provider.clone(), Default::default());
         let eth_api = EthApi::new(
             mock_provider.clone(),
@@ -174,6 +181,7 @@ mod tests {
             ETHEREUM_BLOCK_GAS_LIMIT,
             BlockingTaskPool::build().expect("failed to build tracing pool"),
             FeeHistoryCache::new(cache.clone(), FeeHistoryCacheConfig::default()),
+            local_pending_block_watcher_tx,
         );
 
         let storage_key: U256 = storage_key.into();

crates/rpc/rpc/src/eth/api/transactions.rs

@@ -1274,6 +1274,7 @@ mod tests {
     use reth_primitives::{constants::ETHEREUM_BLOCK_GAS_LIMIT, hex_literal::hex, Bytes};
     use reth_provider::test_utils::NoopProvider;
     use reth_transaction_pool::{test_utils::testing_pool, TransactionPool};
+    use tokio::sync::watch;
 
     #[tokio::test]
     async fn send_raw_transaction() {
@@ -1285,6 +1286,9 @@ mod tests {
         let cache = EthStateCache::spawn(noop_provider, Default::default());
         let fee_history_cache =
             FeeHistoryCache::new(cache.clone(), FeeHistoryCacheConfig::default());
+        let initial_value: Option<(SealedBlock, Vec<Receipt>)> = None; // or some initial value
+        let (local_pending_block_watcher_tx, _) = watch::channel(initial_value);
+
         let eth_api = EthApi::new(
             noop_provider,
             pool.clone(),
@@ -1294,6 +1298,7 @@ mod tests {
             ETHEREUM_BLOCK_GAS_LIMIT,
             BlockingTaskPool::build().expect("failed to build tracing pool"),
             fee_history_cache,
+            local_pending_block_watcher_tx,
         );
 
         // https://etherscan.io/tx/0xa694b71e6c128a2ed8e2e0f6770bddbe52e3bb8f10e8472f9a79ab81497a8b5d

crates/rpc/rpc/src/eth/mod.rs

@@ -8,7 +8,7 @@ mod filter;
 pub mod gas_oracle;
 mod id_provider;
 mod logs_utils;
-mod pubsub;
+pub mod pubsub;
 pub mod revm_utils;
 mod signer;
 pub(crate) mod utils;