kabu_node_reth_api/
lib.rs

1mod provider_wrapper;
2mod traits;
3
4use alloy_provider::Provider;
5use alloy_transport::TransportResult;
6use async_trait::async_trait;
7use futures_util::StreamExt;
8use reth_chain_state::CanonStateNotification;
9use reth_ethereum_primitives::EthPrimitives;
10use reth_node_types::NodeTypes;
11use reth_storage_rpc_provider::RpcBlockchainProvider;
12use reth_tasks::TaskExecutor;
13use tracing::{error, info, warn};
14
15pub use provider_wrapper::KabuRethProviderWrapper;
16pub use traits::KabuRethFullProvider;
17
18#[async_trait]
19pub trait RethApi: Send + Sync {
20    async fn subscribe_chain_notifications(&self) -> TransportResult<alloy_pubsub::Subscription<CanonStateNotification<EthPrimitives>>>;
21}
22
23#[async_trait]
24impl<P> RethApi for P
25where
26    P: Provider,
27{
28    async fn subscribe_chain_notifications(&self) -> TransportResult<alloy_pubsub::Subscription<CanonStateNotification<EthPrimitives>>> {
29        self.root().client().pubsub_frontend().ok_or_else(alloy_transport::TransportErrorKind::pubsub_unavailable)?;
30
31        let mut call = self.client().request("reth_subscribeChainNotifications", ());
32        call.set_is_subscription();
33        let id = call.await?;
34        self.root().get_subscription(id).await
35    }
36}
37
38pub async fn chain_notifications_forwarder<P, Node, N>(
39    executor: &TaskExecutor,
40    provider: P,
41    reth_provider: RpcBlockchainProvider<P, Node, N>,
42) -> eyre::Result<()>
43where
44    P: RethApi + Send + Sync + 'static,
45    Node: NodeTypes<Primitives = EthPrimitives>,
46    N: alloy_network::Network,
47{
48    // Try to subscribe to Reth-specific chain notifications
49    match provider.subscribe_chain_notifications().await {
50        Ok(chain_notifications) => {
51            info!("Successfully subscribed to Reth chain notifications");
52
53            // Spawn task to forward canon state notifications from node to the reth provider
54            executor.spawn_critical("chain-notifications-forwarder", async move {
55                let mut stream = chain_notifications.into_stream();
56                while let Some(canon_state_notification) = stream.next().await {
57                    if let Err(err) = reth_provider.canon_state_notification().send(canon_state_notification) {
58                        error!(?err, "Failed to send canon state notification");
59                    }
60                }
61                warn!("Chain notifications stream ended");
62            });
63
64            Ok(())
65        }
66        Err(err) => {
67            // Log warning instead of error, as this is expected for non-Reth nodes
68            warn!(
69                ?err,
70                "Failed to subscribe to Reth chain notifications. This is expected for non-Reth nodes. The node will continue without Reth-specific features."
71            );
72
73            // Don't fail completely - allow the system to continue
74            // The BlockHistoryComponent and other components should handle missing notifications gracefully
75            Ok(())
76        }
77    }
78}
kabu_node_reth_api/lib.rs

kabu_node_reth_api/
lib.rs
