reth_consensus_debug_client/
client.rs

1use alloy_primitives::B256;
2use futures::FutureExt;
3use reth_node_api::{ConsensusEngineHandle, ExecutionPayload, PayloadTypes};
4use ringbuffer::{AllocRingBuffer, RingBuffer};
5use std::future::Future;
6use tokio::sync::mpsc;
7use tracing::warn;
8
9/// Supplies consensus client with new execution payloads sent in `tx` and a callback to find
10/// specific payloads by number to fetch past finalized and safe block hashes.
11#[auto_impl::auto_impl(&, Arc, Box)]
12pub trait PayloadProvider: Send + Sync + 'static {
13    /// The execution payload data type.
14    type ExecutionData: ExecutionPayload;
15
16    /// Runs a provider to send new execution payloads to the given sender.
17    ///
18    /// Note: This is expected to be spawned in a separate task, and as such it should ignore
19    /// errors.
20    fn subscribe_payloads(
21        &self,
22        tx: mpsc::Sender<Self::ExecutionData>,
23    ) -> impl Future<Output = ()> + Send;
24
25    /// Get a past execution payload by block number.
26    fn get_payload(
27        &self,
28        block_number: u64,
29    ) -> impl Future<Output = eyre::Result<Self::ExecutionData>> + Send;
30
31    /// Get previous block hash using previous block hash buffer. If it isn't available (buffer
32    /// started more recently than `offset`), fetch it using `get_payload`.
33    fn get_or_fetch_previous_block(
34        &self,
35        previous_block_hashes: &AllocRingBuffer<B256>,
36        current_block_number: u64,
37        offset: usize,
38    ) -> impl Future<Output = eyre::Result<B256>> + Send {
39        async move {
40            if let Some(hash) = get_hash_at_offset(previous_block_hashes, offset) {
41                return Ok(hash);
42            }
43
44            // Return zero hash if the chain isn't long enough to have the block at the offset.
45            let previous_block_number = match current_block_number.checked_sub(offset as u64) {
46                Some(number) => number,
47                None => return Ok(B256::default()),
48            };
49            let payload = self.get_payload(previous_block_number).await?;
50            Ok(payload.block_hash())
51        }
52    }
53}
54
55/// Debug consensus client that sends FCUs and new payloads using recent payloads from an external
56/// provider like Etherscan or an RPC endpoint.
57#[derive(Debug)]
58pub struct DebugConsensusClient<P: PayloadProvider, T: PayloadTypes> {
59    /// Handle to execution client.
60    engine_handle: ConsensusEngineHandle<T>,
61    /// Provider to get consensus payloads from.
62    block_provider: P,
63}
64
65impl<P: PayloadProvider, T: PayloadTypes> DebugConsensusClient<P, T> {
66    /// Create a new debug consensus client with the given handle to execution
67    /// client and block provider.
68    pub const fn new(engine_handle: ConsensusEngineHandle<T>, block_provider: P) -> Self {
69        Self { engine_handle, block_provider }
70    }
71}
72
73impl<P, T> DebugConsensusClient<P, T>
74where
75    P: PayloadProvider<ExecutionData = T::ExecutionData> + Clone,
76    T: PayloadTypes,
77{
78    /// Spawn the client to start sending FCUs and new payloads by periodically fetching recent
79    /// payloads.
80    pub async fn run(self) {
81        let mut previous_block_hashes = AllocRingBuffer::new(65);
82        let mut payload_stream = {
83            let (tx, rx) = mpsc::channel::<P::ExecutionData>(64);
84            let block_provider = self.block_provider.clone();
85            tokio::spawn(
86                async move {
87                    block_provider.subscribe_payloads(tx).await;
88                }
89                .boxed(),
90            );
91            rx
92        };
93
94        while let Some(payload) = payload_stream.recv().await {
95            let block_hash = payload.block_hash();
96            let block_number = payload.block_number();
97
98            previous_block_hashes.enqueue(block_hash);
99
100            // Send new events to execution client
101            let _ = self.engine_handle.new_payload(payload).await;
102
103            // Load previous block hashes. We're using (head - 32) and (head - 64) as the safe and
104            // finalized block hashes.
105            let safe_block_hash = self.block_provider.get_or_fetch_previous_block(
106                &previous_block_hashes,
107                block_number,
108                32,
109            );
110            let finalized_block_hash = self.block_provider.get_or_fetch_previous_block(
111                &previous_block_hashes,
112                block_number,
113                64,
114            );
115            let (safe_block_hash, finalized_block_hash) =
116                tokio::join!(safe_block_hash, finalized_block_hash);
117            let (safe_block_hash, finalized_block_hash) = match (
118                safe_block_hash,
119                finalized_block_hash,
120            ) {
121                (Ok(safe_block_hash), Ok(finalized_block_hash)) => {
122                    (safe_block_hash, finalized_block_hash)
123                }
124                (safe_block_hash, finalized_block_hash) => {
125                    warn!(target: "consensus::debug-client", ?safe_block_hash, ?finalized_block_hash, "failed to fetch safe or finalized hash from etherscan");
126                    continue;
127                }
128            };
129            let state = alloy_rpc_types_engine::ForkchoiceState {
130                head_block_hash: block_hash,
131                safe_block_hash,
132                finalized_block_hash,
133            };
134            let _ = self.engine_handle.fork_choice_updated(state, None).await;
135        }
136    }
137}
138
139/// Looks up a block hash from the ring buffer at the given offset from the most recent entry.
140///
141/// Returns `None` if the buffer doesn't have enough entries to satisfy the offset.
142fn get_hash_at_offset(buffer: &AllocRingBuffer<B256>, offset: usize) -> Option<B256> {
143    buffer.len().checked_sub(offset + 1).and_then(|index| buffer.get(index).copied())
144}
145
146#[cfg(test)]
147mod tests {
148    use super::*;
149
150    #[test]
151    fn test_get_hash_at_offset() {
152        let mut buffer: AllocRingBuffer<B256> = AllocRingBuffer::new(65);
153
154        // Empty buffer returns None for any offset
155        assert_eq!(get_hash_at_offset(&buffer, 0), None);
156        assert_eq!(get_hash_at_offset(&buffer, 1), None);
157
158        // Push hashes 0..65
159        for i in 0..65u8 {
160            buffer.enqueue(B256::with_last_byte(i));
161        }
162
163        // offset=0 should return the most recent (64)
164        assert_eq!(get_hash_at_offset(&buffer, 0), Some(B256::with_last_byte(64)));
165
166        // offset=32 (safe block) should return hash 32
167        assert_eq!(get_hash_at_offset(&buffer, 32), Some(B256::with_last_byte(32)));
168
169        // offset=64 (finalized block) should return hash 0 (the oldest)
170        assert_eq!(get_hash_at_offset(&buffer, 64), Some(B256::with_last_byte(0)));
171
172        // offset=65 exceeds buffer, should return None
173        assert_eq!(get_hash_at_offset(&buffer, 65), None);
174    }
175
176    #[test]
177    fn test_get_hash_at_offset_insufficient_entries() {
178        let mut buffer: AllocRingBuffer<B256> = AllocRingBuffer::new(65);
179
180        // With only 1 entry, only offset=0 works
181        buffer.enqueue(B256::with_last_byte(1));
182        assert_eq!(get_hash_at_offset(&buffer, 0), Some(B256::with_last_byte(1)));
183        assert_eq!(get_hash_at_offset(&buffer, 1), None);
184        assert_eq!(get_hash_at_offset(&buffer, 32), None);
185        assert_eq!(get_hash_at_offset(&buffer, 64), None);
186
187        // With 33 entries, offset=32 works but offset=64 doesn't
188        for i in 2..=33u8 {
189            buffer.enqueue(B256::with_last_byte(i));
190        }
191        assert_eq!(get_hash_at_offset(&buffer, 32), Some(B256::with_last_byte(1)));
192        assert_eq!(get_hash_at_offset(&buffer, 64), None);
193    }
194}
reth_consensus_debug_client/client.rs

reth_consensus_debug_client/
client.rs
