hotshot_query_service/data_source/
storage.rs

1// Copyright (c) 2022 Espresso Systems (espressosys.com)
2// This file is part of the HotShot Query Service library.
3//
4// This program is free software: you can redistribute it and/or modify it under the terms of the GNU
5// General Public License as published by the Free Software Foundation, either version 3 of the
6// License, or (at your option) any later version.
7// This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without
8// even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
9// General Public License for more details.
10// You should have received a copy of the GNU General Public License along with this program. If not,
11// see <https://www.gnu.org/licenses/>.
12
13//! Persistent storage for data sources.
14//!
15//! Naturally, an archival query service such as this is heavily dependent on a persistent storage
16//! implementation. This module defines the interfaces required of this storage. Any storage layer
17//! implementing the appropriate interfaces can be used as the storage layer when constructing a
18//! [`FetchingDataSource`](super::FetchingDataSource), which can in turn be used to instantiate the
19//! REST APIs provided by this crate.
20//!
21//! This module also comes with a few pre-built persistence implementations:
22//! * [`SqlStorage`]
23//! * [`FileSystemStorage`]
24//!
25//! # Storage Traits vs Data Source Traits
26//!
27//! Many of the traits defined in this module (e.g. [`NodeStorage`], [`ExplorerStorage`], and
28//! others) are nearly identical to the corresponding data source traits (e.g.
29//! [`NodeDataSource`](crate::node::NodeDataSource),
30//! [`ExplorerDataSource`](crate::explorer::ExplorerDataSource), etc). They typically differ in
31//! mutability: the storage traits are intended to be implemented on storage
32//! [transactions](super::Transaction), and because even reading may update the internal
33//! state of a transaction, such as a buffer or database cursor, these traits typically take `&mut
34//! self`. This is not a barrier for concurrency since there may be many transactions open
35//! simultaneously from a single data source. The data source traits, meanwhile, are implemented on
36//! the data source itself. Internally, they usually open a fresh transaction and do all their work
37//! on the transaction, not modifying the data source itself, so they take `&self`.
38//!
39//! For traits that differ _only_ in the mutability of the `self` parameter, it is almost possible
40//! to combine them into a single trait whose methods take `self` by value, and implementing said
41//! traits for the reference types `&SomeDataSource` and `&mut SomeDataSourceTransaction`. There are
42//! two problems with this approach, which lead us to prefer the slight redundance of having
43//! separate versions of the traits with mutable and immutable methods:
44//! * The trait bounds quickly get out of hand, since we now have trait bounds not only on the type
45//!   itself, but also on references to that type, and the reference also requires the introduction
46//!   of an additional lifetime parameter.
47//! * We run into a longstanding [`rustc` bug](https://github.com/rust-lang/rust/issues/85063) in
48//!   which type inference diverges when given trait bounds on reference types, even when
49//!   theoretically the types are uniquely inferable. This issue can be worked around by [explicitly
50//!   specifying type parameters at every call site](https://users.rust-lang.org/t/type-recursion-when-trait-bound-is-added-on-reference-type/74525/2),
51//!   but this further exacerbates the ergonomic issues with this approach, past the point of
52//!   viability.
53//!
54//! Occasionally, there may be further differences between the data source traits and corresponding
55//! storage traits. For example, [`AvailabilityStorage`] also differs from
56//! [`AvailabilityDataSource`](crate::availability::AvailabilityDataSource) in fallibility.
57//!
58
59use std::ops::RangeBounds;
60
61use alloy::primitives::map::HashMap;
62use async_trait::async_trait;
63use futures::future::Future;
64use hotshot_types::{
65    data::VidShare, simple_certificate::CertificatePair, traits::node_implementation::NodeType,
66};
67use jf_merkle_tree_compat::prelude::MerkleProof;
68use tagged_base64::TaggedBase64;
69
70use crate::{
71    Header, Payload, QueryResult, Transaction,
72    availability::{
73        BlockId, BlockQueryData, Certificate2, LeafId, LeafQueryData, NamespaceId, PayloadMetadata,
74        PayloadQueryData, QueryableHeader, QueryablePayload, TransactionHash, VidCommonMetadata,
75        VidCommonQueryData,
76    },
77    explorer::{
78        query_data::{
79            BlockDetail, BlockIdentifier, BlockSummary, ExplorerSummary, GetBlockDetailError,
80            GetBlockSummariesError, GetBlockSummariesRequest, GetExplorerSummaryError,
81            GetSearchResultsError, GetTransactionDetailError, GetTransactionSummariesError,
82            GetTransactionSummariesRequest, SearchResult, TransactionDetailResponse,
83            TransactionIdentifier, TransactionSummary,
84        },
85        traits::{ExplorerHeader, ExplorerTransaction},
86    },
87    merklized_state::{MerklizedState, Snapshot},
88    node::{SyncStatusQueryData, TimeWindowQueryData, WindowStart},
89    types::HeightIndexed,
90};
91
92/// Retry a fallible storage operation whenever it fails due to a transient serialization conflict.
93///
94/// Under PostgreSQL `SERIALIZABLE` isolation, concurrent transactions can abort with SQLSTATE
95/// `40001` ("could not serialize access"). These aborts are expected and safe to retry from
96/// scratch, so every read or write should run inside this harness rather than calling
97/// [`read`](VersionedDataSource::read) / [`write`](VersionedDataSource::write) directly.
98///
99/// `f` is re-invoked from scratch on each attempt, so it must (re)open its own transaction.
100/// Backends that cannot produce serialization conflicts (e.g. the file system) implement this as a
101/// single pass-through call.
102///
103/// The operation is generic over its error type `E` (e.g. [`QueryError`], `anyhow::Error`, or an
104/// explorer error); conflicts are detected from the error's [`Display`](std::fmt::Display) output.
105#[async_trait]
106pub trait SerializableRetry {
107    /// Run `f`, retrying on serialization conflicts according to the backend's retry policy.
108    ///
109    /// `op` is a short, static label for the operation (used in diagnostic logging); the
110    /// [`serializable_retry!`](crate::serializable_retry) macro fills it in automatically with the
111    /// name of the enclosing function.
112    async fn serializable_retry<T, E, F, Fut>(&self, op: &'static str, f: F) -> Result<T, E>
113    where
114        T: Send,
115        E: std::fmt::Display + Send,
116        F: Fn() -> Fut + Send + Sync,
117        Fut: Future<Output = Result<T, E>> + Send;
118}
119
120/// Expands to the unqualified name of the enclosing function/method as a `&'static str`.
121#[macro_export]
122macro_rules! function_name {
123    () => {{
124        fn __f() {}
125        fn type_name_of<T>(_: T) -> &'static str {
126            ::std::any::type_name::<T>()
127        }
128        let full: &'static str = type_name_of(__f);
129        let trimmed: &'static str = full.strip_suffix("::__f").unwrap_or(full);
130        trimmed
131            .rsplit("::")
132            .find(|segment| *segment != "{{closure}}")
133            .unwrap_or(trimmed)
134    }};
135}
136
137/// Calls [`SerializableRetry::serializable_retry`] with `op` set to the unqualified name of the
138/// enclosing function via [`function_name!`](crate::function_name).
139#[macro_export]
140macro_rules! serializable_retry {
141    ($self:expr, $f:expr) => {
142        $self.serializable_retry($crate::function_name!(), $f)
143    };
144}
145
146pub mod fail_storage;
147pub mod fs;
148mod ledger_log;
149pub mod pruning;
150pub mod sql;
151
152#[cfg(any(test, feature = "testing"))]
153pub use fail_storage::FailStorage;
154#[cfg(feature = "file-system-data-source")]
155pub use fs::FileSystemStorage;
156#[cfg(feature = "sql-data-source")]
157pub use sql::{SqlStorage, StorageConnectionType};
158
159/// Persistent storage for a HotShot blockchain.
160///
161/// This trait defines the interface which must be provided by the storage layer in order to
162/// implement an availability data source. It is very similar to
163/// [`AvailabilityDataSource`](crate::availability::AvailabilityDataSource) with every occurrence of
164/// [`Fetch`](crate::availability::Fetch) replaced by [`QueryResult`]. This is not a coincidence.
165/// The purpose of the storage layer is to provide all of the functionality of the data source
166/// layer, but independent of an external fetcher for missing data. Thus, when the storage layer
167/// encounters missing, corrupt, or inaccessible data, it simply gives up and replaces the missing
168/// data with [`Err`], rather than creating an asynchronous fetch request to retrieve the missing
169/// data.
170///
171/// Rust gives us ways to abstract and deduplicate these two similar APIs, but they do not lead to a
172/// better interface.
173#[async_trait]
174pub trait AvailabilityStorage<Types>: Send + Sync
175where
176    Types: NodeType,
177    Header<Types>: QueryableHeader<Types>,
178    Payload<Types>: QueryablePayload<Types>,
179{
180    async fn get_leaf(&mut self, id: LeafId<Types>) -> QueryResult<LeafQueryData<Types>>;
181    async fn get_block(&mut self, id: BlockId<Types>) -> QueryResult<BlockQueryData<Types>>;
182    async fn get_header(&mut self, id: BlockId<Types>) -> QueryResult<Header<Types>>;
183    async fn get_payload(&mut self, id: BlockId<Types>) -> QueryResult<PayloadQueryData<Types>>;
184    async fn get_payload_metadata(
185        &mut self,
186        id: BlockId<Types>,
187    ) -> QueryResult<PayloadMetadata<Types>>;
188    async fn get_vid_common(
189        &mut self,
190        id: BlockId<Types>,
191    ) -> QueryResult<VidCommonQueryData<Types>>;
192    async fn get_vid_common_metadata(
193        &mut self,
194        id: BlockId<Types>,
195    ) -> QueryResult<VidCommonMetadata<Types>>;
196
197    async fn get_leaf_range<R>(
198        &mut self,
199        range: R,
200    ) -> QueryResult<Vec<QueryResult<LeafQueryData<Types>>>>
201    where
202        R: RangeBounds<usize> + Send + 'static;
203    async fn get_block_range<R>(
204        &mut self,
205        range: R,
206    ) -> QueryResult<Vec<QueryResult<BlockQueryData<Types>>>>
207    where
208        R: RangeBounds<usize> + Send + 'static;
209
210    async fn get_header_range<R>(
211        &mut self,
212        range: R,
213    ) -> QueryResult<Vec<QueryResult<Header<Types>>>>
214    where
215        R: RangeBounds<usize> + Send + 'static,
216    {
217        let blocks = self.get_block_range(range).await?;
218        Ok(blocks
219            .into_iter()
220            .map(|block| block.map(|block| block.header))
221            .collect())
222    }
223    async fn get_payload_range<R>(
224        &mut self,
225        range: R,
226    ) -> QueryResult<Vec<QueryResult<PayloadQueryData<Types>>>>
227    where
228        R: RangeBounds<usize> + Send + 'static;
229    async fn get_payload_metadata_range<R>(
230        &mut self,
231        range: R,
232    ) -> QueryResult<Vec<QueryResult<PayloadMetadata<Types>>>>
233    where
234        R: RangeBounds<usize> + Send + 'static;
235    async fn get_vid_common_range<R>(
236        &mut self,
237        range: R,
238    ) -> QueryResult<Vec<QueryResult<VidCommonQueryData<Types>>>>
239    where
240        R: RangeBounds<usize> + Send + 'static;
241    async fn get_vid_common_metadata_range<R>(
242        &mut self,
243        range: R,
244    ) -> QueryResult<Vec<QueryResult<VidCommonMetadata<Types>>>>
245    where
246        R: RangeBounds<usize> + Send + 'static;
247
248    async fn get_block_with_transaction(
249        &mut self,
250        hash: TransactionHash<Types>,
251    ) -> QueryResult<BlockQueryData<Types>>;
252}
253
254pub trait UpdateAvailabilityStorage<Types>: Send
255where
256    Types: NodeType,
257{
258    fn insert_leaf(
259        &mut self,
260        leaf: &LeafQueryData<Types>,
261    ) -> impl Send + Future<Output = anyhow::Result<()>> {
262        self.insert_leaf_range([leaf])
263    }
264
265    fn insert_leaf_with_qc_chain(
266        &mut self,
267        leaf: &LeafQueryData<Types>,
268        qc_chain: Option<[CertificatePair<Types>; 2]>,
269    ) -> impl Send + Future<Output = anyhow::Result<()>> {
270        async move {
271            self.insert_leaf(leaf).await?;
272            self.insert_qc_chain(leaf.height(), qc_chain).await?;
273            Ok(())
274        }
275    }
276
277    fn insert_block(
278        &mut self,
279        block: &BlockQueryData<Types>,
280    ) -> impl Send + Future<Output = anyhow::Result<()>> {
281        self.insert_block_range([block])
282    }
283
284    fn insert_vid<'a>(
285        &mut self,
286        common: &'a VidCommonQueryData<Types>,
287        share: Option<&'a VidShare>,
288    ) -> impl Send + Future<Output = anyhow::Result<()>> {
289        self.insert_vid_range([(common, share)])
290    }
291
292    fn insert_qc_chain(
293        &mut self,
294        height: u64,
295        qc_chain: Option<[CertificatePair<Types>; 2]>,
296    ) -> impl Send + Future<Output = anyhow::Result<()>>;
297
298    fn insert_cert2(
299        &mut self,
300        height: u64,
301        cert2: Certificate2<Types>,
302    ) -> impl Send + Future<Output = anyhow::Result<()>>;
303
304    fn insert_leaf_range<'a>(
305        &mut self,
306        leaves: impl Send + IntoIterator<IntoIter: Send, Item = &'a LeafQueryData<Types>>,
307    ) -> impl Send + Future<Output = anyhow::Result<()>>;
308    fn insert_block_range<'a>(
309        &mut self,
310        blocks: impl Send + IntoIterator<IntoIter: Send, Item = &'a BlockQueryData<Types>>,
311    ) -> impl Send + Future<Output = anyhow::Result<()>>;
312    fn insert_vid_range<'a>(
313        &mut self,
314        vid: impl Send
315        + IntoIterator<
316            IntoIter: Send,
317            Item = (&'a VidCommonQueryData<Types>, Option<&'a VidShare>),
318        >,
319    ) -> impl Send + Future<Output = anyhow::Result<()>>;
320}
321
322#[async_trait]
323pub trait NodeStorage<Types>
324where
325    Types: NodeType,
326    Header<Types>: QueryableHeader<Types>,
327{
328    async fn block_height(&mut self) -> QueryResult<usize>;
329    async fn count_transactions_in_range(
330        &mut self,
331        range: impl RangeBounds<usize> + Send,
332        namespace: Option<NamespaceId<Types>>,
333    ) -> QueryResult<usize>;
334    async fn payload_size_in_range(
335        &mut self,
336        range: impl RangeBounds<usize> + Send,
337        namespace: Option<NamespaceId<Types>>,
338    ) -> QueryResult<usize>;
339    async fn vid_share<ID>(&mut self, id: ID) -> QueryResult<VidShare>
340    where
341        ID: Into<BlockId<Types>> + Send + Sync;
342    async fn get_header_window(
343        &mut self,
344        start: impl Into<WindowStart<Types>> + Send + Sync,
345        end: u64,
346        limit: usize,
347    ) -> QueryResult<TimeWindowQueryData<Header<Types>>>;
348
349    async fn latest_qc_chain(&mut self) -> QueryResult<Option<[CertificatePair<Types>; 2]>>;
350
351    async fn load_cert2(&mut self, height: u64) -> QueryResult<Option<Certificate2<Types>>>;
352
353    /// Load the earliest cert2 whose finalized block height is at or above `height`.
354    ///
355    /// "Earliest" means the cert2 with the smallest finalized block height that is still greater
356    /// than or equal to the requested `height`.
357    async fn load_earliest_cert2(
358        &mut self,
359        height: u64,
360    ) -> QueryResult<Option<Certificate2<Types>>>;
361
362    /// Search the given range of the database for missing objects.
363    async fn sync_status_for_range(
364        &mut self,
365        from: usize,
366        to: usize,
367    ) -> QueryResult<SyncStatusQueryData>;
368}
369
370#[derive(Clone, Debug, Default)]
371pub struct Aggregate<Types: NodeType>
372where
373    Header<Types>: QueryableHeader<Types>,
374{
375    pub height: i64,
376    pub num_transactions: HashMap<Option<NamespaceId<Types>>, usize>,
377    pub payload_size: HashMap<Option<NamespaceId<Types>>, usize>,
378}
379
380pub trait AggregatesStorage<Types>
381where
382    Types: NodeType,
383    Header<Types>: QueryableHeader<Types>,
384{
385    /// The block height for which aggregate statistics are currently available.
386    fn aggregates_height(&mut self) -> impl Future<Output = anyhow::Result<usize>> + Send;
387
388    /// the last aggregate
389    fn load_prev_aggregate(
390        &mut self,
391    ) -> impl Future<Output = anyhow::Result<Option<Aggregate<Types>>>> + Send;
392}
393
394pub trait UpdateAggregatesStorage<Types>
395where
396    Types: NodeType,
397    Header<Types>: QueryableHeader<Types>,
398{
399    /// Update aggregate statistics based on a new block.
400    fn update_aggregates(
401        &mut self,
402        aggregate: Aggregate<Types>,
403        blocks: &[PayloadMetadata<Types>],
404    ) -> impl Future<Output = anyhow::Result<Aggregate<Types>>> + Send;
405}
406
407/// An interface for querying Data and Statistics from the HotShot Blockchain.
408///
409/// This interface provides methods that allows the enabling of querying data
410/// concerning the blockchain from the stored data for use with a
411/// block explorer.  It does not provide the same guarantees as the
412/// Availability data source with data fetching.  It is not concerned with
413/// being up-to-date or having all of the data required, but rather it is
414/// concerned with providing the requested data as quickly as possible, and in
415/// a way that can be easily cached.
416#[async_trait]
417pub trait ExplorerStorage<Types>
418where
419    Types: NodeType,
420    Header<Types>: ExplorerHeader<Types> + QueryableHeader<Types>,
421    Transaction<Types>: ExplorerTransaction<Types>,
422    Payload<Types>: QueryablePayload<Types>,
423{
424    /// `get_block_detail` is a method that retrieves the details of a specific
425    /// block from the blockchain.  The block is identified by the given
426    /// [BlockIdentifier].
427    async fn get_block_detail(
428        &mut self,
429        request: BlockIdentifier<Types>,
430    ) -> Result<BlockDetail<Types>, GetBlockDetailError>;
431
432    /// `get_block_summaries` is a method that retrieves a list of block
433    /// summaries from the blockchain.  The list is generated from the given
434    /// [GetBlockSummariesRequest].
435    async fn get_block_summaries(
436        &mut self,
437        request: GetBlockSummariesRequest<Types>,
438    ) -> Result<Vec<BlockSummary<Types>>, GetBlockSummariesError>;
439
440    /// `get_transaction_detail` is a method that retrieves the details of a
441    /// specific transaction from the blockchain.  The transaction is identified
442    /// by the given [TransactionIdentifier].
443    async fn get_transaction_detail(
444        &mut self,
445        request: TransactionIdentifier<Types>,
446    ) -> Result<TransactionDetailResponse<Types>, GetTransactionDetailError>;
447
448    /// `get_transaction_summaries` is a method that retrieves a list of
449    /// transaction summaries from the blockchain.  The list is generated from
450    /// the given [GetTransactionSummariesRequest].
451    async fn get_transaction_summaries(
452        &mut self,
453        request: GetTransactionSummariesRequest<Types>,
454    ) -> Result<Vec<TransactionSummary<Types>>, GetTransactionSummariesError>;
455
456    /// `get_explorer_summary` is a method that retrieves a summary overview of
457    /// the blockchain.  This is useful for displaying information that
458    /// indicates the overall status of the block chain.
459    async fn get_explorer_summary(
460        &mut self,
461    ) -> Result<ExplorerSummary<Types>, GetExplorerSummaryError>;
462
463    /// `get_search_results` is a method that retrieves the results of a search
464    /// query against the blockchain.  The results are generated from the given
465    /// query string.
466    async fn get_search_results(
467        &mut self,
468        query: TaggedBase64,
469    ) -> Result<SearchResult<Types>, GetSearchResultsError>;
470}
471
472/// This trait defines methods that a data source should implement
473/// It enables retrieval of the membership path for a leaf node, which can be used to reconstruct the Merkle tree state.
474#[async_trait]
475pub trait MerklizedStateStorage<Types, State, const ARITY: usize>
476where
477    Types: NodeType,
478    State: MerklizedState<Types, ARITY>,
479{
480    async fn get_path(
481        &mut self,
482        snapshot: Snapshot<Types, State, ARITY>,
483        key: State::Key,
484    ) -> QueryResult<MerkleProof<State::Entry, State::Key, State::T, ARITY>>;
485}
486
487#[async_trait]
488pub trait MerklizedStateHeightStorage {
489    async fn get_last_state_height(&mut self) -> QueryResult<usize>;
490}
hotshot_query_service/data_source/storage.rs

hotshot_query_service/data_source/
storage.rs
