hotshot_types/traits/
storage.rs

1// Copyright (c) 2021-2024 Espresso Systems (espressosys.com)
2// This file is part of the HotShot repository.
3
4// You should have received a copy of the MIT License
5// along with the HotShot repository. If not, see <https://mit-license.org/>.
6
7//! Abstract storage type for storing DA proposals and VID shares
8//!
9//! This modules provides the [`Storage`] trait.
10//!
11
12use std::{sync::Arc, time::Duration};
13
14use anyhow::{Result, anyhow, ensure};
15use async_trait::async_trait;
16use futures::future::BoxFuture;
17use tokio::time::sleep;
18
19use super::node_implementation::NodeType;
20use crate::{
21    data::{
22        DaProposal, DaProposal2, EpochNumber, QuorumProposal, QuorumProposal2,
23        QuorumProposalWrapper, VidCommitment, VidDisperseShare, ViewNumber,
24    },
25    drb::{DrbInput, DrbResult},
26    event::HotShotAction,
27    message::{Proposal, convert_proposal},
28    simple_certificate::{
29        LightClientStateUpdateCertificateV2, NextEpochQuorumCertificate2, QuorumCertificate,
30        QuorumCertificate2, UpgradeCertificate,
31    },
32};
33
34/// Abstraction for storing a variety of consensus payload datum.
35#[async_trait]
36pub trait Storage<TYPES: NodeType>: Send + Sync + Clone + 'static {
37    /// Add a proposal to the stored VID proposals.
38    async fn append_vid(&self, proposal: &Proposal<TYPES, VidDisperseShare<TYPES>>) -> Result<()>;
39
40    /// Add a proposal to the stored DA proposals.
41    async fn append_da(
42        &self,
43        proposal: &Proposal<TYPES, DaProposal<TYPES>>,
44        vid_commit: VidCommitment,
45    ) -> Result<()>;
46    /// Add a proposal to the stored DA proposals.
47    async fn append_da2(
48        &self,
49        proposal: &Proposal<TYPES, DaProposal2<TYPES>>,
50        vid_commit: VidCommitment,
51    ) -> Result<()> {
52        self.append_da(&convert_proposal(proposal.clone()), vid_commit)
53            .await
54    }
55    /// Add a proposal we sent to the store
56    async fn append_proposal(
57        &self,
58        proposal: &Proposal<TYPES, QuorumProposal<TYPES>>,
59    ) -> Result<()>;
60    /// Add a proposal we sent to the store
61    async fn append_proposal2(
62        &self,
63        proposal: &Proposal<TYPES, QuorumProposal2<TYPES>>,
64    ) -> Result<()>;
65    /// Add a proposal we sent to the store
66    async fn append_proposal_wrapper(
67        &self,
68        proposal: &Proposal<TYPES, QuorumProposalWrapper<TYPES>>,
69    ) -> Result<()> {
70        self.append_proposal2(&convert_proposal(proposal.clone()))
71            .await
72    }
73    /// Record a HotShotAction taken.
74    async fn record_action(
75        &self,
76        view: ViewNumber,
77        epoch: Option<EpochNumber>,
78        action: HotShotAction,
79    ) -> Result<()>;
80    /// Update the current high QC in storage.
81    async fn update_high_qc(&self, high_qc: QuorumCertificate<TYPES>) -> Result<()>;
82    /// Update the current high QC in storage.
83    async fn update_high_qc2(&self, high_qc: QuorumCertificate2<TYPES>) -> Result<()> {
84        self.update_high_qc(high_qc.to_qc()).await
85    }
86    /// Update the light client state update certificate in storage.
87    async fn update_state_cert(
88        &self,
89        state_cert: LightClientStateUpdateCertificateV2<TYPES>,
90    ) -> Result<()>;
91
92    async fn update_high_qc2_and_state_cert(
93        &self,
94        high_qc: QuorumCertificate2<TYPES>,
95        state_cert: LightClientStateUpdateCertificateV2<TYPES>,
96    ) -> Result<()> {
97        self.update_high_qc2(high_qc).await?;
98        self.update_state_cert(state_cert).await
99    }
100    /// Update the current high QC in storage.
101    async fn update_next_epoch_high_qc2(
102        &self,
103        _next_epoch_high_qc: NextEpochQuorumCertificate2<TYPES>,
104    ) -> Result<()>;
105
106    /// Update the current eQC in storage.
107    async fn update_eqc(
108        &self,
109        _high_qc: QuorumCertificate2<TYPES>,
110        _next_epoch_high_qc: NextEpochQuorumCertificate2<TYPES>,
111    ) -> Result<()>;
112
113    /// Upgrade the current decided upgrade certificate in storage.
114    async fn update_decided_upgrade_certificate(
115        &self,
116        decided_upgrade_certificate: Option<UpgradeCertificate<TYPES>>,
117    ) -> Result<()>;
118    /// Add a drb result
119    async fn store_drb_result(&self, epoch: EpochNumber, drb_result: DrbResult) -> Result<()>;
120    /// Add an epoch block header
121    async fn store_epoch_root(
122        &self,
123        epoch: EpochNumber,
124        block_header: TYPES::BlockHeader,
125    ) -> Result<()>;
126    async fn load_drb_result(&self, epoch: EpochNumber) -> Result<DrbResult> {
127        match self.load_drb_input(*epoch).await {
128            Ok(drb_input) => {
129                ensure!(drb_input.iteration == drb_input.difficulty_level);
130
131                Ok(drb_input.value)
132            },
133            Err(e) => Err(e),
134        }
135    }
136    async fn store_drb_input(&self, drb_input: DrbInput) -> Result<()>;
137    async fn load_drb_input(&self, _epoch: u64) -> Result<DrbInput>;
138}
139
140pub async fn load_drb_input_impl<TYPES: NodeType>(
141    storage: impl Storage<TYPES>,
142    epoch: u64,
143) -> Result<DrbInput> {
144    storage.load_drb_input(epoch).await
145}
146
147pub type LoadDrbProgressFn =
148    std::sync::Arc<dyn Fn(u64) -> BoxFuture<'static, Result<DrbInput>> + Send + Sync>;
149
150pub fn load_drb_progress_fn<TYPES: NodeType>(
151    storage: impl Storage<TYPES> + 'static,
152) -> LoadDrbProgressFn {
153    Arc::new(move |epoch| {
154        let storage = storage.clone();
155        Box::pin(load_drb_input_impl(storage, epoch))
156    })
157}
158
159pub fn null_load_drb_progress_fn() -> LoadDrbProgressFn {
160    Arc::new(move |_drb_input| {
161        Box::pin(async { Err(anyhow!("Using null implementation of load_drb_input")) })
162    })
163}
164
165pub async fn store_drb_input_impl<TYPES: NodeType>(
166    storage: impl Storage<TYPES>,
167    drb_input: DrbInput,
168) -> Result<()> {
169    for attempt in 1..=3 {
170        match storage.store_drb_input(drb_input.clone()).await {
171            Ok(()) => return Ok(()),
172            Err(e) if attempt < 3 => {
173                tracing::warn!("Failed to store DRB input (attempt {attempt}/3): {e}");
174                sleep(Duration::from_millis(300)).await;
175            },
176            Err(e) => {
177                tracing::warn!("Failed to store DRB input (attempt {attempt}/3): {e}");
178                return Err(e);
179            },
180        }
181    }
182
183    Ok(())
184}
185
186pub type StoreDrbProgressFn =
187    std::sync::Arc<dyn Fn(DrbInput) -> BoxFuture<'static, Result<()>> + Send + Sync>;
188
189pub fn store_drb_progress_fn<TYPES: NodeType>(
190    storage: impl Storage<TYPES> + 'static,
191) -> StoreDrbProgressFn {
192    Arc::new(move |drb_input| {
193        let storage = storage.clone();
194        Box::pin(store_drb_input_impl(storage, drb_input))
195    })
196}
197
198pub fn null_store_drb_progress_fn() -> StoreDrbProgressFn {
199    Arc::new(move |_drb_input| Box::pin(async { Ok(()) }))
200}
201
202pub type StoreDrbResultFn = Arc<
203    Box<dyn Fn(EpochNumber, DrbResult) -> BoxFuture<'static, Result<()>> + Send + Sync + 'static>,
204>;
205
206async fn store_drb_result_impl<TYPES: NodeType>(
207    storage: impl Storage<TYPES>,
208    epoch: EpochNumber,
209    drb_result: DrbResult,
210) -> Result<()> {
211    for attempt in 1..=3 {
212        match storage.store_drb_result(epoch, drb_result).await {
213            Ok(()) => return Ok(()),
214            Err(e) if attempt < 3 => {
215                tracing::warn!(
216                    "Failed to store DRB result for epoch {epoch} (attempt {attempt}/3): {e}"
217                );
218                sleep(Duration::from_millis(300)).await;
219            },
220            Err(e) => {
221                tracing::warn!(
222                    "Failed to store DRB result for epoch {epoch} (attempt {attempt}/3): {e}"
223                );
224                return Err(e);
225            },
226        }
227    }
228    Ok(())
229}
230
231/// Helper function to create a callback to add a drb result to storage
232pub fn store_drb_result_fn<TYPES: NodeType>(
233    storage: impl Storage<TYPES> + 'static,
234) -> StoreDrbResultFn {
235    Arc::new(Box::new(move |epoch, drb_result| {
236        let st = storage.clone();
237        Box::pin(store_drb_result_impl(st, epoch, drb_result))
238    }))
239}
hotshot_types/traits/storage.rs

hotshot_types/traits/
storage.rs
