espresso_node/api/
update.rs

1//! Update loop for query API state.
2
3use std::{fmt::Debug, sync::Arc};
4
5use anyhow::bail;
6use async_trait::async_trait;
7use derivative::Derivative;
8use derive_more::From;
9use espresso_types::v0::traits::NullEventConsumer;
10use hotshot_query_service::{
11    availability::{BlockInfo, UpdateAvailabilityData},
12    data_source::UpdateDataSource,
13    node::NodeDataSource,
14    types::HeightIndexed,
15};
16use hotshot_types::new_protocol::CoordinatorEvent;
17
18use super::{StorageState, data_source::SequencerDataSource};
19use crate::{EventConsumer, SeqTypes};
20
21#[async_trait]
22pub trait DecideSink: Send + Sync + Debug {
23    async fn append(&self, info: BlockInfo<SeqTypes>) -> anyhow::Result<()>;
24
25    /// The first height a resuming node has to append.
26    async fn block_height(&self) -> anyhow::Result<u64>;
27}
28
29#[async_trait]
30impl<T> DecideSink for Box<T>
31where
32    T: DecideSink + ?Sized,
33{
34    async fn append(&self, info: BlockInfo<SeqTypes>) -> anyhow::Result<()> {
35        (**self).append(info).await
36    }
37
38    async fn block_height(&self) -> anyhow::Result<u64> {
39        (**self).block_height().await
40    }
41}
42
43pub trait ApiSink: EventConsumer + DecideSink {}
44
45impl<T: EventConsumer + DecideSink> ApiSink for T {}
46
47/// Without a query module there is nowhere to store decided blocks, so a node that hands them
48/// here fails on the first one instead of reporting progress it never persisted.
49#[async_trait]
50impl DecideSink for NullEventConsumer {
51    async fn append(&self, info: BlockInfo<SeqTypes>) -> anyhow::Result<()> {
52        bail!(
53            "cannot store block {}: the API has no query storage",
54            info.height()
55        )
56    }
57
58    async fn block_height(&self) -> anyhow::Result<u64> {
59        Ok(0)
60    }
61}
62
63#[derive(Derivative, From)]
64#[derivative(Clone(bound = ""), Debug(bound = "D: Debug"))]
65pub(crate) struct ApiEventConsumer<D> {
66    inner: Arc<StorageState<D>>,
67}
68
69#[async_trait]
70impl<D> EventConsumer for ApiEventConsumer<D>
71where
72    D: SequencerDataSource + Debug + Send + Sync + 'static,
73{
74    async fn handle_event(&self, event: &CoordinatorEvent<SeqTypes>) -> anyhow::Result<()> {
75        if let Err(height) = self.inner.update(event).await {
76            bail!("failed to update API state after {height}: {event:?}",);
77        }
78        Ok(())
79    }
80}
81
82#[async_trait]
83impl<D> DecideSink for ApiEventConsumer<D>
84where
85    D: SequencerDataSource + UpdateAvailabilityData<SeqTypes> + Debug + Send + Sync + 'static,
86{
87    async fn append(&self, info: BlockInfo<SeqTypes>) -> anyhow::Result<()> {
88        self.inner.append(info).await
89    }
90
91    async fn block_height(&self) -> anyhow::Result<u64> {
92        Ok(NodeDataSource::block_height(&*self.inner).await? as u64)
93    }
94}
espresso_node/api/update.rs

espresso_node/api/
update.rs
