espresso_types/v0/v0_1/
l1.rs

1#[cfg(feature = "node")]
2use std::time::Instant;
3use std::{
4    num::{NonZeroU64, NonZeroUsize},
5    sync::Arc,
6    time::Duration,
7};
8
9use alloy::primitives::{B256, U256};
10#[cfg(feature = "node")]
11use alloy::{
12    network::Ethereum,
13    providers::{
14        Identity, Provider, RootProvider,
15        fillers::{FillProvider, JoinFill, RecommendedFillers},
16    },
17    transports::http::{Client, Http},
18};
19use alloy_compat::ethers_serde;
20#[cfg(feature = "node")]
21use async_broadcast::{InactiveReceiver, Sender};
22use clap::Parser;
23#[cfg(feature = "node")]
24use derive_more::Deref;
25use espresso_utils::redact::redact_urls;
26#[cfg(feature = "node")]
27use hotshot_types::traits::metrics::{Counter, Gauge};
28use hotshot_types::traits::metrics::{Metrics, NoMetrics};
29#[cfg(feature = "node")]
30use lru::LruCache;
31#[cfg(feature = "node")]
32use parking_lot::RwLock;
33use serde::{Deserialize, Serialize};
34use thiserror::Error;
35#[cfg(feature = "node")]
36use tokio::{
37    sync::{Mutex, Notify},
38    task::JoinHandle,
39};
40use url::Url;
41
42use crate::v0::utils::parse_duration;
43
44#[derive(Clone, Copy, Debug, Default, Deserialize, Serialize, Hash, PartialEq, Eq)]
45pub struct L1BlockInfo {
46    pub number: u64,
47    #[serde(with = "ethers_serde::u256")]
48    pub timestamp: U256,
49    #[serde(with = "ethers_serde::b256")]
50    pub hash: B256,
51}
52
53#[cfg_attr(not(feature = "node"), allow(dead_code))]
54#[derive(Clone, Copy, Debug, PartialOrd, Ord, Hash, PartialEq, Eq)]
55pub(crate) struct L1BlockInfoWithParent {
56    pub(crate) info: L1BlockInfo,
57    pub(crate) parent_hash: B256,
58}
59
60#[derive(Clone, Copy, Debug, Default, Deserialize, Serialize, Hash, PartialEq, Eq)]
61pub struct L1Snapshot {
62    /// The relevant snapshot of the L1 includes a reference to the current head of the L1 chain.
63    ///
64    /// Note that the L1 head is subject to changing due to a reorg. However, no reorg will change
65    /// the _number_ of this block in the chain: L1 block numbers will always be sequentially
66    /// increasing. Therefore, the sequencer does not have to worry about reorgs invalidating this
67    /// snapshot.
68    pub head: u64,
69
70    /// The snapshot also includes information about the latest finalized L1 block.
71    ///
72    /// Since this block is finalized (ie cannot be reorged) we can include specific information
73    /// about the particular block, such as its hash and timestamp.
74    ///
75    /// This block may be `None` in the rare case where Espresso has started shortly after the
76    /// genesis of the L1, and the L1 has yet to finalize a block. In all other cases it will be
77    /// `Some`.
78    pub finalized: Option<L1BlockInfo>,
79}
80
81/// How far below the finalized head a block must be before its finality is trusted without
82/// verification. `None` means unlimited: every block is hash-chain verified.
83#[derive(Clone, Copy, Debug, PartialEq, Eq)]
84pub struct L1SafetyMargin(pub(crate) Option<NonZeroU64>);
85
86#[derive(Debug, Error)]
87#[error("invalid safety margin {input:?}: expected a nonzero block count or `unlimited`")]
88pub struct ParseL1SafetyMarginError {
89    pub(crate) input: String,
90}
91
92/// Configuration for an L1 client.
93#[derive(Clone, derive_more::Debug, Parser)]
94pub struct L1ClientOptions {
95    /// Delay when retrying failed L1 queries.
96    #[clap(
97        long,
98        env = "ESPRESSO_L1_RETRY_DELAY",
99        default_value = "1s",
100        value_parser = parse_duration,
101    )]
102    pub l1_retry_delay: Duration,
103
104    /// Request rate when polling L1.
105    #[clap(
106        long,
107        env = "ESPRESSO_L1_POLLING_INTERVAL",
108        default_value = "7s",
109        value_parser = parse_duration,
110    )]
111    pub l1_polling_interval: Duration,
112
113    /// How often `wait_for_block`, `wait_for_finalized_block` and
114    /// `wait_for_latest_finalized_at_least` refresh their snapshot from the RPC while waiting, on
115    /// top of listening for poller events.
116    #[clap(
117        long,
118        env = "ESPRESSO_L1_WAIT_REFRESH_INTERVAL",
119        default_value = "500ms",
120        value_parser = parse_duration,
121    )]
122    pub l1_wait_refresh_interval: Duration,
123
124    /// How long a single refresh RPC call may run before a `wait_for_block`,
125    /// `wait_for_finalized_block` or `wait_for_latest_finalized_at_least` refresh gives up on it.
126    /// Longer than `l1_wait_refresh_interval` so a call isn't cut off right as it's issued; bounded
127    /// so a hung request can't park a waiter indefinitely.
128    #[clap(
129        long,
130        env = "ESPRESSO_L1_WAIT_REFRESH_TIMEOUT",
131        default_value = "1500ms",
132        value_parser = parse_duration,
133    )]
134    pub l1_wait_refresh_timeout: Duration,
135
136    /// Timeout for L1 head and finalized block requests; log queries are not bounded.
137    #[clap(
138        long,
139        env = "ESPRESSO_L1_REQUEST_TIMEOUT",
140        default_value = "10s",
141        value_parser = parse_duration,
142    )]
143    pub l1_request_timeout: Duration,
144
145    /// Maximum number of L1 blocks to keep in cache at once.
146    #[clap(long, env = "ESPRESSO_L1_BLOCKS_CACHE_SIZE", default_value = "100")]
147    pub l1_blocks_cache_size: NonZeroUsize,
148
149    /// Number of L1 events to buffer before discarding.
150    #[clap(
151        long,
152        env = "ESPRESSO_L1_EVENTS_CHANNEL_CAPACITY",
153        default_value = "100"
154    )]
155    pub l1_events_channel_capacity: usize,
156
157    /// Maximum number of L1 blocks that can be scanned for events in a single query.
158    #[clap(
159        long,
160        env = "ESPRESSO_L1_EVENTS_MAX_BLOCK_RANGE",
161        default_value = "10000"
162    )]
163    pub l1_events_max_block_range: u64,
164
165    /// Maximum time to wait for new heads before considering a stream invalid and reconnecting.
166    #[clap(
167        long,
168        env = "ESPRESSO_L1_SUBSCRIPTION_TIMEOUT",
169        default_value = "1m",
170        value_parser = parse_duration,
171    )]
172    pub subscription_timeout: Duration,
173
174    /// Fail over to another provider if the current provider fails twice within this window.
175    #[clap(
176        long,
177        env = "ESPRESSO_L1_FREQUENT_FAILURE_TOLERANCE",
178        default_value = "1m",
179        value_parser = parse_duration,
180    )]
181    pub l1_frequent_failure_tolerance: Duration,
182
183    /// Fail over to another provider if the current provider fails many times in a row, within any
184    /// time window.
185    #[clap(
186        long,
187        env = "ESPRESSO_L1_CONSECUTIVE_FAILURE_TOLERANCE",
188        default_value = "10"
189    )]
190    pub l1_consecutive_failure_tolerance: usize,
191
192    /// Revert back to the first provider this duration after failing over.
193    #[clap(
194        long,
195        env = "ESPRESSO_L1_FAILOVER_REVERT",
196        default_value = "30m",
197        value_parser = parse_duration,
198    )]
199    pub l1_failover_revert: Duration,
200
201    /// Amount of time to wait after receiving a 429 response before making more L1 RPC requests.
202    ///
203    /// If not set, the general l1-retry-delay will be used.
204    #[clap(
205        long,
206        env = "ESPRESSO_L1_RATE_LIMIT_DELAY",
207        value_parser = parse_duration,
208    )]
209    pub l1_rate_limit_delay: Option<Duration>,
210
211    /// Separate provider to use for subscription feeds.
212    ///
213    /// Typically this would be a WebSockets endpoint while the main provider uses HTTP.
214    #[clap(long, env = "ESPRESSO_L1_WS_PROVIDER", value_delimiter = ',')]
215    #[debug("{:?}", l1_ws_provider.as_ref().map(redact_urls))]
216    pub l1_ws_provider: Option<Vec<Url>>,
217
218    /// Interval at which the background update loop polls the L1 stake table contract for new events
219    /// and updates local persistence.
220    ///
221    #[clap(
222        long,
223        env = "ESPRESSO_NODE_L1_STAKE_TABLE_UPDATE_INTERVAL",
224        default_value = "60m",
225        value_parser = parse_duration,
226    )]
227    pub stake_table_update_interval: Duration,
228
229    /// Maximum duration to retry fetching L1 events before panicking.
230    ///
231    /// This prevents infinite retries by panicking if the total number of retries exceed the maximum duration.
232    /// This is helpful in cases where the RPC block range limit or the event return limit is hit,
233    /// or if there is an outage. In such cases, panicking ensures that the node operator can take
234    /// action instead of the node getting stuck indefinitely. This is necessary because the stake table is constructed
235    /// from the fetched events, and is required for node to participate in consensus.
236    #[clap(
237        long,
238        env = "ESPRESSO_L1_EVENTS_MAX_RETRY_DURATION",
239        default_value = "20m",
240        value_parser = parse_duration,
241    )]
242    pub l1_events_max_retry_duration: Duration,
243
244    /// A block range which is expected to contain the finalized heads of all L1 provider chains.
245    ///
246    /// It is assumed that if a block `n` is known to be finalized according to a certain
247    /// provider, then any block less than `n - L1_FINALIZED_SAFETY_MARGIN` is finalized
248    /// _according to any provider_. In other words, if we fail over from one provider to another,
249    /// the second provider will never be lagging the first by more than this margin.
250    ///
251    /// This allows us to quickly query for very old finalized blocks by number. Without this
252    /// assumption, we always need to verify that a block is finalized by fetching all blocks in a
253    /// hash chain between the known finalized block and the desired block, recomputing and checking
254    /// the hashes. This is fine and good for blocks very near the finalized head, but for
255    /// extremely old blocks it is prohibitively expensive, and these old blocks are extremely
256    /// unlikely to be unfinalized anyways.
257    ///
258    /// Set to `unlimited` to hash-chain verify every block instead.
259    #[clap(
260        long,
261        env = "ESPRESSO_L1_FINALIZED_SAFETY_MARGIN",
262        default_value = "100"
263    )]
264    pub l1_finalized_safety_margin: L1SafetyMargin,
265
266    #[clap(skip = Arc::<Box<dyn Metrics>>::new(Box::new(NoMetrics)))]
267    pub metrics: Arc<Box<dyn Metrics>>,
268}
269
270/// Type alias for alloy provider
271#[cfg(feature = "node")]
272pub type L1Provider = FillProvider<
273    JoinFill<Identity, <Ethereum as RecommendedFillers>::RecommendedFillers>,
274    RootProvider,
275>;
276
277#[cfg(feature = "node")]
278#[derive(Clone, Debug, Deref)]
279/// An Ethereum provider and configuration to interact with the L1.
280///
281/// This client runs asynchronously, updating an in-memory snapshot of the relevant L1 information
282/// each time a new L1 block is published. The main advantage of this is that we can update the L1
283/// state at the pace of the L1, instead of the much faster pace of HotShot consensus.This makes it
284/// easy to use a subscription instead of polling for new blocks, vastly reducing the number of L1
285/// RPC calls we make.
286pub struct L1Client {
287    /// The alloy provider used for L1 communication with wallet and default fillers
288    #[deref]
289    pub provider: L1Provider,
290    /// Actual transport used in `self.provider`
291    /// i.e. the `t` variable in `ProviderBuilder::new().on_client(RpcClient::new(t, is_local))`
292    pub transport: SwitchingTransport,
293    /// Shared state updated by an asynchronous task which polls the L1.
294    pub(crate) state: Arc<Mutex<L1State>>,
295    /// Channel used by the async update task to send events to clients.
296    pub(crate) sender: Sender<L1Event>,
297    /// Receiver for events from the async update task.
298    pub(crate) receiver: InactiveReceiver<L1Event>,
299    /// Async task which updates the shared state.
300    pub(crate) update_task: Arc<L1UpdateTask>,
301}
302
303#[cfg(feature = "node")]
304impl Provider for L1Client {
305    fn root(&self) -> &RootProvider {
306        self.provider.root()
307    }
308}
309
310/// In-memory view of the L1 state, updated asynchronously.
311#[cfg(feature = "node")]
312#[derive(Debug)]
313pub(crate) struct L1State {
314    pub(crate) snapshot: L1Snapshot,
315    pub(crate) finalized: LruCache<u64, L1BlockInfoWithParent>,
316    pub(crate) last_finalized: Option<u64>,
317    /// When a waiter last refreshed the head from the RPC, throttling `refresh_head` to one call
318    /// per `WAIT_REFRESH_INTERVAL` across every concurrent waiter.
319    pub(crate) last_head_refresh: Option<Instant>,
320    /// When a waiter last refreshed the finalized block from the RPC, throttling
321    /// `refresh_finalized` to one call per `WAIT_REFRESH_INTERVAL` across every concurrent
322    /// waiter. Independent of `last_head_refresh`, so a head wait can't starve a finalized wait
323    /// or vice versa.
324    pub(crate) last_finalized_refresh: Option<Instant>,
325}
326
327#[cfg(feature = "node")]
328#[derive(Clone, Debug)]
329pub(crate) enum L1Event {
330    NewHead { head: u64 },
331    NewFinalized { finalized: L1BlockInfoWithParent },
332}
333
334#[cfg(feature = "node")]
335#[derive(Debug, Default)]
336pub(crate) struct L1UpdateTask(pub(crate) Mutex<Option<JoinHandle<()>>>);
337
338#[cfg(feature = "node")]
339#[derive(Clone, Debug)]
340pub(crate) struct L1ClientMetrics {
341    pub(crate) head: Arc<dyn Gauge>,
342    pub(crate) finalized: Arc<dyn Gauge>,
343    pub(crate) reconnects: Arc<dyn Counter>,
344    pub(crate) failovers: Arc<dyn Counter>,
345    pub(crate) failures: Arc<Vec<Box<dyn Counter>>>,
346}
347
348/// An RPC client with multiple remote (HTTP) providers.
349///
350/// This client utilizes one RPC provider at a time, but if it detects that the provider is in a
351/// failing state, it will automatically switch to the next provider in its list.
352#[cfg(feature = "node")]
353#[derive(Clone, derive_more::Debug)]
354pub struct SwitchingTransport {
355    /// The transport currently being used by the client
356    pub(crate) current_transport: Arc<RwLock<SingleTransport>>,
357    /// The list of configured HTTP URLs to use for RPC requests
358    #[debug("{:?}", redact_urls(urls.iter()))]
359    pub(crate) urls: Arc<Vec<Url>>,
360    pub(crate) opt: Arc<L1ClientOptions>,
361    pub(crate) metrics: L1ClientMetrics,
362    pub(crate) switch_notify: Arc<Notify>,
363}
364
365/// The state of the current provider being used by a [`SwitchingTransport`].
366/// This is cloneable and returns a reference to the same underlying data.
367#[cfg(feature = "node")]
368#[derive(derive_more::Debug, Clone)]
369pub(crate) struct SingleTransport {
370    pub(crate) generation: usize,
371    pub(crate) redacted_url: String,
372    #[debug(skip)]
373    pub(crate) client: Http<Client>,
374    pub(crate) status: Arc<RwLock<SingleTransportStatus>>,
375    /// Time at which to revert back to the primary provider after a failover.
376    pub(crate) revert_at: Option<Instant>,
377}
378
379/// The status of a single transport
380#[cfg(feature = "node")]
381#[derive(Debug, Default)]
382pub(crate) struct SingleTransportStatus {
383    pub(crate) last_failure: Option<Instant>,
384    pub(crate) consecutive_failures: usize,
385    pub(crate) consecutive_rate_limits: usize,
386    pub(crate) rate_limited_until: Option<Instant>,
387    /// Whether or not this current transport is being shut down (switching to the next transport)
388    pub(crate) shutting_down: bool,
389}
espresso_types/v0/v0_1/l1.rs

espresso_types/v0/v0_1/
l1.rs
