espresso_node/
proposal_fetcher.rs

1use std::{sync::Arc, time::Duration};
2
3use anyhow::Context;
4use async_channel::{Receiver, Sender};
5use clap::Parser;
6use committable::Commitment;
7use derivative::Derivative;
8use espresso_types::{ValidatedState, parse_duration, v0::traits::SequencerPersistence};
9use futures::stream::StreamExt;
10use hotshot_types::{
11    data::{Leaf2, QuorumProposalWrapper, ViewNumber},
12    message::Proposal,
13    new_protocol::CoordinatorEvent,
14    traits::{
15        ValidatedState as _,
16        metrics::{Counter, Gauge, Metrics},
17    },
18};
19use serde::Serialize;
20use tokio::time::{sleep, timeout};
21use tracing::Instrument;
22
23use crate::{SeqTypes, consensus_handle::ConsensusHandle, context::TaskList};
24
25#[derive(Clone, Copy, Debug, Parser, Serialize)]
26pub struct ProposalFetcherConfig {
27    #[clap(
28        long = "proposal-fetcher-num-workers",
29        env = "ESPRESSO_NODE_PROPOSAL_FETCHER_NUM_WORKERS",
30        default_value = "2"
31    )]
32    pub num_workers: usize,
33
34    #[clap(
35        long = "proposal-fetcher-fetch-timeout",
36        env = "ESPRESSO_NODE_PROPOSAL_FETCHER_FETCH_TIMEOUT",
37        default_value = "2s",
38        value_parser = parse_duration,
39    )]
40    pub fetch_timeout: Duration,
41}
42
43impl Default for ProposalFetcherConfig {
44    fn default() -> Self {
45        Self::parse_from(std::iter::empty::<String>())
46    }
47}
48
49impl ProposalFetcherConfig {
50    pub(crate) fn spawn<P>(
51        self,
52        tasks: &mut TaskList,
53        consensus_handle: Arc<ConsensusHandle<SeqTypes, Arc<P>>>,
54        persistence: Arc<P>,
55        metrics: &(impl Metrics + ?Sized),
56    ) where
57        P: SequencerPersistence,
58    {
59        let (sender, receiver) = async_channel::unbounded();
60        let fetcher = ProposalFetcher {
61            sender,
62            consensus_handle,
63            persistence,
64            cfg: self,
65            metrics: ProposalFetcherMetrics::new(metrics),
66        };
67
68        tasks.spawn("proposal scanner", fetcher.clone().scan());
69        for i in 0..self.num_workers {
70            tasks.spawn(
71                format!("proposal fetcher {i}"),
72                fetcher.clone().fetch(receiver.clone()),
73            );
74        }
75    }
76}
77
78#[derive(Clone, Debug)]
79struct ProposalFetcherMetrics {
80    fetched: Arc<dyn Counter>,
81    failed: Arc<dyn Counter>,
82    queue_len: Arc<dyn Gauge>,
83    last_seen: Arc<dyn Gauge>,
84    last_fetched: Arc<dyn Gauge>,
85}
86
87impl ProposalFetcherMetrics {
88    fn new(metrics: &(impl Metrics + ?Sized)) -> Self {
89        let metrics = metrics.subgroup("proposal_fetcher".into());
90        Self {
91            fetched: metrics.create_counter("fetched".into(), None).into(),
92            failed: metrics.create_counter("failed".into(), None).into(),
93            queue_len: metrics.create_gauge("queue_len".into(), None).into(),
94            last_seen: metrics
95                .create_gauge("last_seen".into(), Some("view".into()))
96                .into(),
97            last_fetched: metrics
98                .create_gauge("last_fetched".into(), Some("view".into()))
99                .into(),
100        }
101    }
102}
103
104type Request = (ViewNumber, Commitment<Leaf2<SeqTypes>>);
105
106#[derive(Derivative)]
107#[derivative(Clone(bound = ""), Debug(bound = ""))]
108struct ProposalFetcher<P>
109where
110    P: SequencerPersistence,
111{
112    sender: Sender<Request>,
113    #[derivative(Debug = "ignore")]
114    consensus_handle: Arc<ConsensusHandle<SeqTypes, Arc<P>>>,
115    #[derivative(Debug = "ignore")]
116    persistence: Arc<P>,
117    cfg: ProposalFetcherConfig,
118    metrics: ProposalFetcherMetrics,
119}
120
121impl<P> ProposalFetcher<P>
122where
123    P: SequencerPersistence,
124{
125    #[tracing::instrument(skip_all)]
126    async fn scan(self) {
127        let mut events = self.consensus_handle.event_stream();
128        while let Some(event) = events.next().await {
129            let (parent_view, parent_leaf) = match event {
130                CoordinatorEvent::QuorumProposal { proposal, .. } => {
131                    let parent_view = proposal.data.justify_qc.view_number;
132                    let parent_leaf = proposal.data.justify_qc.data.leaf_commit;
133                    (parent_view, parent_leaf)
134                },
135                _ => continue,
136            };
137            // Whenever we see a quorum proposal, ensure we have the chain of proposals stretching back
138            // to the anchor. This allows state replay from the decided state.
139            self.request((parent_view, parent_leaf)).await;
140        }
141    }
142
143    #[tracing::instrument(skip_all)]
144    async fn fetch(self, receiver: Receiver<(ViewNumber, Commitment<Leaf2<SeqTypes>>)>) {
145        let mut receiver = std::pin::pin!(receiver);
146        while let Some(req) = receiver.next().await {
147            self.fetch_request(req).await;
148        }
149    }
150
151    async fn request(&self, req: Request) {
152        self.sender.send(req).await.ok();
153        self.metrics.queue_len.set(self.sender.len());
154        self.metrics.last_seen.set(req.0.u64() as usize);
155    }
156
157    async fn fetch_request(&self, (view, leaf): Request) {
158        let span = tracing::warn_span!("fetch proposal", ?view, %leaf);
159        let res: anyhow::Result<()> = async {
160            let mut decided_view = self.consensus_handle.decided_leaf().await.view_number();
161            if decided_view == ViewNumber::genesis() {
162                // A freshly restarted node may not have decided anything
163                // yet sofall back to the persisted anchor view
164                decided_view = self
165                    .persistence
166                    .load_anchor_view()
167                    .await
168                    .context("loading anchor view")?;
169            }
170            if view <= decided_view {
171                tracing::debug!(?decided_view, "skipping already-decided proposal");
172                return Ok(());
173            }
174
175            match self.persistence.load_quorum_proposal(view).await {
176                Ok(proposal) => {
177                    // If we already have the proposal in storage, keep traversing the chain to its
178                    // parent.
179                    let view = proposal.data.justify_qc().view_number;
180                    let leaf = proposal.data.justify_qc().data.leaf_commit;
181                    self.request((view, leaf)).await;
182                    return Ok(());
183                },
184                Err(err) => {
185                    tracing::info!("proposal missing from storage; fetching from network: {err:#}");
186                },
187            }
188
189            let future = self.consensus_handle.request_proposal(view, leaf).await?;
190            let proposal: Proposal<SeqTypes, QuorumProposalWrapper<SeqTypes>> =
191                timeout(self.cfg.fetch_timeout, future)
192                    .await
193                    .context("timed out fetching proposal")?
194                    .context("error fetching proposal")?;
195            self.persistence
196                .append_quorum_proposal2(&proposal)
197                .await
198                .context("error saving fetched proposal")?;
199
200            // Add the fetched leaf to consensus state, so consensus can make use of it.
201            // Only update if the view is missing or DA-only (state() returns None for
202            // both cases) — don't overwrite an existing Leaf view.
203            let leaf = Leaf2::from_quorum_proposal(&proposal.data);
204            if self.consensus_handle.state(view).await.is_none() {
205                let state = Arc::new(ValidatedState::from_header(leaf.block_header()));
206                if let Err(err) = self.consensus_handle.update_leaf(leaf, state, None).await {
207                    tracing::info!("unable to update leaf: {err:#}");
208                }
209            }
210
211            self.metrics.last_fetched.set(view.u64() as usize);
212            self.metrics.fetched.add(1);
213
214            Ok(())
215        }
216        .instrument(span)
217        .await;
218        if let Err(err) = res {
219            tracing::warn!(?view, err = %format!("{err:#}"), "failed to fetch proposal");
220            self.metrics.failed.add(1);
221
222            // Avoid busy loop when operations are failing.
223            sleep(Duration::from_secs(1)).await;
224
225            // If we fail fetching the proposal, don't let it clog up the fetching task. Just push
226            // it back onto the queue and move onto the next proposal.
227            self.request((view, leaf)).await;
228        }
229    }
230}
espresso_node/proposal_fetcher.rs

espresso_node/
proposal_fetcher.rs
