kabu_node_reth_api/
provider_wrapper.rs

1//! Provider wrapper that adds canonical state notification capabilities
2//!
3//! This wrapper allows any provider to emit canonical state notifications,
4//! which is essential for both testing and remote node connections.
5
6use eyre::Result;
7use reth_chain_state::{CanonStateNotification, CanonStateSubscriptions};
8use reth_ethereum_primitives::EthPrimitives;
9use reth_execution_types::Chain;
10use reth_primitives::Block;
11use reth_primitives_traits::block::RecoveredBlock;
12use reth_storage_api::NodePrimitivesProvider;
13use std::fmt::Debug;
14use std::sync::Arc;
15use tokio::sync::broadcast;
16use tracing::trace;
17
18/// A wrapper around a provider that adds canonical state notification capabilities.
19///
20/// This is necessary because not all providers (like RpcBlockchainProvider) implement
21/// CanonStateSubscriptions natively. This wrapper adds that capability by maintaining
22/// a broadcast channel for canonical state notifications.
23#[derive(Debug, Clone)]
24pub struct KabuRethProviderWrapper<P> {
25    /// The inner provider that implements the actual blockchain access
26    inner: P,
27    /// Sender for canonical state notifications
28    pub canon_state_notification_sender: broadcast::Sender<CanonStateNotification<EthPrimitives>>,
29}
30
31impl<P> KabuRethProviderWrapper<P> {
32    /// Create a new provider wrapper with canonical state notification support
33    pub fn new(inner: P) -> Self {
34        // Create a broadcast channel for canonical state notifications
35        // Buffer size of 256 should be sufficient for most use cases
36        let (canon_state_notification_sender, _) = broadcast::channel(256);
37        Self { inner, canon_state_notification_sender }
38    }
39
40    /// Send a canonical commit notification for a single block
41    ///
42    /// The caller needs to provide the block and execution outcome in the appropriate format
43    pub fn send_canonical_commit(
44        &self,
45        block_with_senders: RecoveredBlock<Block>,
46        execution_outcome: reth_execution_types::ExecutionOutcome,
47    ) -> Result<()> {
48        // Create a chain with a single block
49        let chain = Chain::new(vec![block_with_senders], execution_outcome, None);
50
51        // Send the commit notification
52        let notification = CanonStateNotification::Commit { new: Arc::new(chain) };
53
54        // Ignore send errors (no receivers is ok)
55        let _ = self.canon_state_notification_sender.send(notification);
56        trace!("Sent canonical commit notification");
57
58        Ok(())
59    }
60
61    /// Get access to the inner provider
62    pub fn inner(&self) -> &P {
63        &self.inner
64    }
65
66    /// Get a mutable reference to the inner provider
67    pub fn inner_mut(&mut self) -> &mut P {
68        &mut self.inner
69    }
70
71    /// Consume the wrapper and return the inner provider
72    pub fn into_inner(self) -> P {
73        self.inner
74    }
75}
76
77// Implement CanonStateSubscriptions for the wrapper
78impl<P> CanonStateSubscriptions for KabuRethProviderWrapper<P>
79where
80    P: NodePrimitivesProvider<Primitives = EthPrimitives> + Send + Sync,
81{
82    fn subscribe_to_canonical_state(&self) -> reth_chain_state::CanonStateNotifications<Self::Primitives> {
83        self.canon_state_notification_sender.subscribe()
84    }
85
86    // canonical_state_stream has a default implementation in the trait
87}
88
89// Implement std::ops::Deref to allow transparent access to inner provider methods
90impl<P> std::ops::Deref for KabuRethProviderWrapper<P> {
91    type Target = P;
92
93    fn deref(&self) -> &Self::Target {
94        &self.inner
95    }
96}
97
98// Implement std::ops::DerefMut for mutable access
99impl<P> std::ops::DerefMut for KabuRethProviderWrapper<P> {
100    fn deref_mut(&mut self) -> &mut Self::Target {
101        &mut self.inner
102    }
103}
104
105// Implement NodePrimitivesProvider by delegating to the inner provider
106impl<P> NodePrimitivesProvider for KabuRethProviderWrapper<P>
107where
108    P: NodePrimitivesProvider,
109{
110    type Primitives = P::Primitives;
111}
kabu_node_reth_api/provider_wrapper.rs

kabu_node_reth_api/
provider_wrapper.rs
