iota_core/
storage.rs

1// Copyright (c) Mysten Labs, Inc.
2// Modifications Copyright (c) 2024 IOTA Stiftung
3// SPDX-License-Identifier: Apache-2.0
4
5use std::sync::Arc;
6
7use iota_node_storage::{GrpcIndexes, GrpcStateReader};
8use iota_sdk_types::{
9    CheckpointContentsDigest, CheckpointDigest, StructTag, TransactionDigest,
10    checkpoint::{CheckpointContents, EndOfEpochData},
11};
12use iota_types::{
13    committee::{Committee, EpochId},
14    effects::{TransactionEffects, TransactionEvents},
15    error::IotaError,
16    messages_checkpoint::{
17        CheckpointContentsExt, CheckpointSequenceNumber, FullCheckpointContents,
18        VerifiedCheckpoint, VerifiedCheckpointContents,
19    },
20    object::Object,
21    storage::{
22        ObjectKey, ObjectStore, ReadStore, WriteStore,
23        error::{Error as StorageError, Result},
24    },
25    transaction::VerifiedTransaction,
26};
27use parking_lot::Mutex;
28use tracing::instrument;
29
30use crate::{
31    authority::AuthorityState, checkpoints::CheckpointStore,
32    epoch::committee_store::CommitteeStore, execution_cache::ExecutionCacheTraitPointers,
33    grpc_indexes::GrpcIndexesStore,
34};
35
36#[derive(Clone)]
37pub struct RocksDbStore {
38    cache_traits: ExecutionCacheTraitPointers,
39
40    committee_store: Arc<CommitteeStore>,
41    checkpoint_store: Arc<CheckpointStore>,
42    // in memory checkpoint watermark sequence numbers
43    highest_verified_checkpoint: Arc<Mutex<Option<u64>>>,
44    highest_synced_checkpoint: Arc<Mutex<Option<u64>>>,
45}
46
47impl RocksDbStore {
48    pub fn new(
49        cache_traits: ExecutionCacheTraitPointers,
50        committee_store: Arc<CommitteeStore>,
51        checkpoint_store: Arc<CheckpointStore>,
52    ) -> Self {
53        Self {
54            cache_traits,
55            committee_store,
56            checkpoint_store,
57            highest_verified_checkpoint: Arc::new(Mutex::new(None)),
58            highest_synced_checkpoint: Arc::new(Mutex::new(None)),
59        }
60    }
61
62    pub fn get_objects(&self, object_keys: &[ObjectKey]) -> Result<Vec<Option<Object>>, IotaError> {
63        self.cache_traits
64            .object_cache_reader
65            .try_multi_get_objects_by_key(object_keys)
66    }
67
68    pub fn get_last_executed_checkpoint(&self) -> Result<Option<VerifiedCheckpoint>, IotaError> {
69        Ok(self.checkpoint_store.get_highest_executed_checkpoint()?)
70    }
71}
72
73impl ReadStore for RocksDbStore {
74    fn try_get_checkpoint_by_digest(
75        &self,
76        digest: &CheckpointDigest,
77    ) -> Result<Option<VerifiedCheckpoint>, StorageError> {
78        self.checkpoint_store
79            .get_checkpoint_by_digest(digest)
80            .map_err(Into::into)
81    }
82
83    fn try_get_checkpoint_by_sequence_number(
84        &self,
85        sequence_number: CheckpointSequenceNumber,
86    ) -> Result<Option<VerifiedCheckpoint>, StorageError> {
87        self.checkpoint_store
88            .get_checkpoint_by_sequence_number(sequence_number)
89            .map_err(Into::into)
90    }
91
92    fn try_get_highest_verified_checkpoint(&self) -> Result<VerifiedCheckpoint, StorageError> {
93        self.checkpoint_store
94            .get_highest_verified_checkpoint()
95            .map(|maybe_checkpoint| {
96                maybe_checkpoint
97                    .expect("storage should have been initialized with genesis checkpoint")
98            })
99            .map_err(Into::into)
100    }
101
102    fn try_get_highest_synced_checkpoint(&self) -> Result<VerifiedCheckpoint, StorageError> {
103        self.checkpoint_store
104            .get_highest_synced_checkpoint()
105            .map(|maybe_checkpoint| {
106                maybe_checkpoint
107                    .expect("storage should have been initialized with genesis checkpoint")
108            })
109            .map_err(Into::into)
110    }
111
112    fn try_get_lowest_available_checkpoint(
113        &self,
114    ) -> Result<CheckpointSequenceNumber, StorageError> {
115        if let Some(highest_pruned_cp) = self
116            .checkpoint_store
117            .get_highest_pruned_checkpoint_seq_number()
118            .map_err(Into::<StorageError>::into)?
119        {
120            Ok(highest_pruned_cp + 1)
121        } else {
122            Ok(0)
123        }
124    }
125
126    fn try_get_full_checkpoint_contents_by_sequence_number(
127        &self,
128        sequence_number: CheckpointSequenceNumber,
129    ) -> Result<Option<FullCheckpointContents>, StorageError> {
130        Ok(self
131            .checkpoint_store
132            .get_full_checkpoint_contents_by_sequence_number(sequence_number)
133            .map(|contents| contents.as_ref().clone()))
134    }
135
136    fn try_get_full_checkpoint_contents(
137        &self,
138        digest: &CheckpointContentsDigest,
139    ) -> Result<Option<FullCheckpointContents>, StorageError> {
140        // First look to see if the in-memory cache still holds the complete
141        // contents.
142        if let Some(contents) = self
143            .checkpoint_store
144            .get_full_checkpoint_contents_by_digest(digest)
145        {
146            return Ok(Some(contents.as_ref().clone()));
147        }
148
149        // Otherwise gather it from the individual components.
150        self.checkpoint_store
151            .get_checkpoint_contents(digest)
152            .map_err(iota_types::storage::error::Error::custom)?
153            .map(|contents| {
154                let mut transactions = Vec::with_capacity(contents.len());
155                for tx in contents.iter() {
156                    if let (Some(t), Some(e)) = (
157                        self.try_get_transaction(&tx.transaction)?,
158                        self.cache_traits
159                            .transaction_cache_reader
160                            .try_get_effects(&tx.effects)
161                            .map_err(iota_types::storage::error::Error::custom)?,
162                    ) {
163                        transactions.push(iota_types::base_types::ExecutionData::new(
164                            (*t).clone().into_inner(),
165                            e,
166                        ))
167                    } else {
168                        return Result::<
169                            Option<FullCheckpointContents>,
170                            iota_types::storage::error::Error,
171                        >::Ok(None);
172                    }
173                }
174                Ok(Some(
175                    FullCheckpointContents::from_contents_and_execution_data(
176                        contents,
177                        transactions.into_iter(),
178                    ),
179                ))
180            })
181            .transpose()
182            .map(|contents| contents.flatten())
183            .map_err(iota_types::storage::error::Error::custom)
184    }
185
186    fn try_get_committee(
187        &self,
188        epoch: EpochId,
189    ) -> Result<Option<Arc<Committee>>, iota_types::storage::error::Error> {
190        Ok(self.committee_store.get_committee(&epoch).unwrap())
191    }
192
193    fn try_get_transaction(
194        &self,
195        digest: &TransactionDigest,
196    ) -> Result<Option<Arc<VerifiedTransaction>>, StorageError> {
197        self.cache_traits
198            .transaction_cache_reader
199            .try_get_transaction_block(digest)
200            .map_err(StorageError::custom)
201    }
202
203    fn try_get_transaction_effects(
204        &self,
205        digest: &TransactionDigest,
206    ) -> Result<Option<TransactionEffects>, StorageError> {
207        self.cache_traits
208            .transaction_cache_reader
209            .try_get_executed_effects(digest)
210            .map_err(StorageError::custom)
211    }
212
213    fn try_get_events(
214        &self,
215        digest: &TransactionDigest,
216    ) -> Result<Option<TransactionEvents>, StorageError> {
217        self.cache_traits
218            .transaction_cache_reader
219            .try_get_events(digest)
220            .map_err(StorageError::custom)
221    }
222
223    fn try_get_latest_checkpoint(&self) -> iota_types::storage::error::Result<VerifiedCheckpoint> {
224        self.checkpoint_store
225            .get_highest_executed_checkpoint()
226            .map_err(iota_types::storage::error::Error::custom)?
227            .ok_or_else(|| {
228                iota_types::storage::error::Error::missing("unable to get latest checkpoint")
229            })
230    }
231
232    fn try_get_checkpoint_contents_by_digest(
233        &self,
234        digest: &CheckpointContentsDigest,
235    ) -> iota_types::storage::error::Result<Option<CheckpointContents>> {
236        self.checkpoint_store
237            .get_checkpoint_contents(digest)
238            .map_err(iota_types::storage::error::Error::custom)
239    }
240
241    fn try_get_checkpoint_contents_by_sequence_number(
242        &self,
243        sequence_number: CheckpointSequenceNumber,
244    ) -> iota_types::storage::error::Result<Option<CheckpointContents>> {
245        match self.try_get_checkpoint_by_sequence_number(sequence_number) {
246            Ok(Some(checkpoint)) => {
247                self.try_get_checkpoint_contents_by_digest(&checkpoint.contents_digest)
248            }
249            Ok(None) => Ok(None),
250            Err(e) => Err(e),
251        }
252    }
253}
254
255impl ObjectStore for RocksDbStore {
256    fn try_get_object(
257        &self,
258        object_id: &iota_sdk_types::ObjectId,
259    ) -> iota_types::storage::error::Result<Option<Object>> {
260        self.cache_traits.object_store.try_get_object(object_id)
261    }
262
263    fn try_get_object_by_key(
264        &self,
265        object_id: &iota_sdk_types::ObjectId,
266        version: iota_types::base_types::VersionNumber,
267    ) -> iota_types::storage::error::Result<Option<Object>> {
268        self.cache_traits
269            .object_store
270            .try_get_object_by_key(object_id, version)
271    }
272}
273
274impl WriteStore for RocksDbStore {
275    #[instrument(level = "trace", skip_all)]
276    fn try_insert_checkpoint(
277        &self,
278        checkpoint: &VerifiedCheckpoint,
279    ) -> Result<(), iota_types::storage::error::Error> {
280        if let Some(EndOfEpochData {
281            next_epoch_committee,
282            ..
283        }) = checkpoint.end_of_epoch_data.as_ref()
284        {
285            let committee = Committee::from_committee_members(
286                checkpoint.epoch().checked_add(1).unwrap(),
287                next_epoch_committee,
288            );
289            self.try_insert_committee(committee)?;
290        }
291
292        self.checkpoint_store
293            .insert_verified_checkpoint(checkpoint)
294            .map_err(Into::into)
295    }
296
297    fn try_update_highest_synced_checkpoint(
298        &self,
299        checkpoint: &VerifiedCheckpoint,
300    ) -> Result<(), iota_types::storage::error::Error> {
301        let mut locked = self.highest_synced_checkpoint.lock();
302        if locked.is_some() && locked.unwrap() >= checkpoint.sequence_number {
303            return Ok(());
304        }
305        self.checkpoint_store
306            .update_highest_synced_checkpoint(checkpoint)
307            .map_err(iota_types::storage::error::Error::custom)?;
308        *locked = Some(checkpoint.sequence_number);
309        Ok(())
310    }
311
312    fn try_update_highest_verified_checkpoint(
313        &self,
314        checkpoint: &VerifiedCheckpoint,
315    ) -> Result<(), iota_types::storage::error::Error> {
316        let mut locked = self.highest_verified_checkpoint.lock();
317        if locked.is_some() && locked.unwrap() >= checkpoint.sequence_number {
318            return Ok(());
319        }
320        self.checkpoint_store
321            .update_highest_verified_checkpoint(checkpoint)
322            .map_err(iota_types::storage::error::Error::custom)?;
323        *locked = Some(checkpoint.sequence_number);
324        Ok(())
325    }
326
327    fn try_insert_checkpoint_contents(
328        &self,
329        checkpoint: &VerifiedCheckpoint,
330        contents: VerifiedCheckpointContents,
331    ) -> Result<(), iota_types::storage::error::Error> {
332        self.cache_traits
333            .state_sync_store
334            .try_multi_insert_transaction_and_effects(contents.transactions())
335            .map_err(iota_types::storage::error::Error::custom)?;
336        self.checkpoint_store
337            .insert_verified_checkpoint_contents(checkpoint, contents)
338            .map_err(Into::into)
339    }
340
341    fn try_insert_committee(
342        &self,
343        new_committee: Committee,
344    ) -> Result<(), iota_types::storage::error::Error> {
345        self.committee_store
346            .insert_new_committee(&new_committee)
347            .unwrap();
348        Ok(())
349    }
350}
351
352pub struct GrpcReadStore {
353    state: Arc<AuthorityState>,
354    rocks: RocksDbStore,
355}
356
357impl GrpcReadStore {
358    pub fn new(state: Arc<AuthorityState>, rocks: RocksDbStore) -> Self {
359        Self { state, rocks }
360    }
361
362    fn grpc_indexes_store(&self) -> iota_types::storage::error::Result<&GrpcIndexesStore> {
363        self.state.grpc_indexes_store.as_deref().ok_or_else(|| {
364            iota_types::storage::error::Error::custom("gRPC index store is disabled")
365        })
366    }
367}
368
369impl ObjectStore for GrpcReadStore {
370    fn try_get_object(
371        &self,
372        object_id: &iota_sdk_types::ObjectId,
373    ) -> iota_types::storage::error::Result<Option<Object>> {
374        self.rocks.try_get_object(object_id)
375    }
376
377    fn try_get_object_by_key(
378        &self,
379        object_id: &iota_sdk_types::ObjectId,
380        version: iota_types::base_types::VersionNumber,
381    ) -> iota_types::storage::error::Result<Option<Object>> {
382        self.rocks.try_get_object_by_key(object_id, version)
383    }
384}
385
386impl ReadStore for GrpcReadStore {
387    fn try_get_committee(
388        &self,
389        epoch: EpochId,
390    ) -> iota_types::storage::error::Result<Option<Arc<Committee>>> {
391        self.rocks.try_get_committee(epoch)
392    }
393
394    fn try_get_latest_checkpoint(&self) -> iota_types::storage::error::Result<VerifiedCheckpoint> {
395        self.rocks.try_get_latest_checkpoint()
396    }
397
398    fn try_get_highest_verified_checkpoint(
399        &self,
400    ) -> iota_types::storage::error::Result<VerifiedCheckpoint> {
401        self.rocks.try_get_highest_verified_checkpoint()
402    }
403
404    fn try_get_highest_synced_checkpoint(
405        &self,
406    ) -> iota_types::storage::error::Result<VerifiedCheckpoint> {
407        self.rocks.try_get_highest_synced_checkpoint()
408    }
409
410    fn try_get_lowest_available_checkpoint(
411        &self,
412    ) -> iota_types::storage::error::Result<CheckpointSequenceNumber> {
413        self.rocks.try_get_lowest_available_checkpoint()
414    }
415
416    fn try_get_checkpoint_by_digest(
417        &self,
418        digest: &CheckpointDigest,
419    ) -> iota_types::storage::error::Result<Option<VerifiedCheckpoint>> {
420        self.rocks.try_get_checkpoint_by_digest(digest)
421    }
422
423    fn try_get_checkpoint_by_sequence_number(
424        &self,
425        sequence_number: CheckpointSequenceNumber,
426    ) -> iota_types::storage::error::Result<Option<VerifiedCheckpoint>> {
427        self.rocks
428            .try_get_checkpoint_by_sequence_number(sequence_number)
429    }
430
431    fn try_get_checkpoint_contents_by_digest(
432        &self,
433        digest: &CheckpointContentsDigest,
434    ) -> iota_types::storage::error::Result<Option<CheckpointContents>> {
435        self.rocks.try_get_checkpoint_contents_by_digest(digest)
436    }
437
438    fn try_get_checkpoint_contents_by_sequence_number(
439        &self,
440        sequence_number: CheckpointSequenceNumber,
441    ) -> iota_types::storage::error::Result<Option<CheckpointContents>> {
442        self.rocks
443            .try_get_checkpoint_contents_by_sequence_number(sequence_number)
444    }
445
446    fn try_get_transaction(
447        &self,
448        digest: &TransactionDigest,
449    ) -> iota_types::storage::error::Result<Option<Arc<VerifiedTransaction>>> {
450        self.rocks.try_get_transaction(digest)
451    }
452
453    fn try_get_transaction_effects(
454        &self,
455        digest: &TransactionDigest,
456    ) -> iota_types::storage::error::Result<Option<TransactionEffects>> {
457        self.rocks.try_get_transaction_effects(digest)
458    }
459
460    fn try_get_events(
461        &self,
462        digest: &TransactionDigest,
463    ) -> iota_types::storage::error::Result<Option<TransactionEvents>> {
464        self.rocks.try_get_events(digest)
465    }
466
467    fn try_get_full_checkpoint_contents_by_sequence_number(
468        &self,
469        sequence_number: CheckpointSequenceNumber,
470    ) -> iota_types::storage::error::Result<Option<FullCheckpointContents>> {
471        self.rocks
472            .try_get_full_checkpoint_contents_by_sequence_number(sequence_number)
473    }
474
475    fn try_get_full_checkpoint_contents(
476        &self,
477        digest: &CheckpointContentsDigest,
478    ) -> iota_types::storage::error::Result<Option<FullCheckpointContents>> {
479        self.rocks.try_get_full_checkpoint_contents(digest)
480    }
481}
482
483impl GrpcStateReader for GrpcReadStore {
484    fn get_lowest_available_checkpoint_objects(
485        &self,
486    ) -> iota_types::storage::error::Result<CheckpointSequenceNumber> {
487        Ok(self
488            .state
489            .get_object_cache_reader()
490            .try_get_highest_pruned_checkpoint()
491            .map_err(StorageError::custom)?
492            .map(|cp| cp + 1)
493            .unwrap_or(0))
494    }
495
496    fn get_chain_identifier(&self) -> Result<iota_types::digests::ChainIdentifier> {
497        Ok(self.state.get_chain_identifier())
498    }
499
500    fn get_epoch_last_checkpoint(
501        &self,
502        epoch_id: EpochId,
503    ) -> iota_types::storage::error::Result<Option<VerifiedCheckpoint>> {
504        self.rocks
505            .checkpoint_store
506            .get_epoch_last_checkpoint(epoch_id)
507            .map_err(iota_types::storage::error::Error::custom)
508    }
509
510    fn get_epoch_info(
511        &self,
512        epoch: EpochId,
513    ) -> iota_types::storage::error::Result<Option<iota_types::storage::EpochInfoV2>> {
514        self.rocks
515            .checkpoint_store
516            .get_epoch_info(epoch)
517            .map_err(iota_types::storage::error::Error::custom)
518    }
519
520    fn grpc_indexes(&self) -> Option<&dyn GrpcIndexes> {
521        self.grpc_indexes_store().ok().map(|index| index as _)
522    }
523
524    fn get_struct_layout(
525        &self,
526        struct_tag: &StructTag,
527    ) -> Result<Option<move_core_types::annotated_value::MoveTypeLayout>> {
528        self.state
529            .load_epoch_store_one_call_per_task()
530            .executor()
531            // TODO(cache) - must read through cache
532            .type_layout_resolver(Box::new(self.state.get_backing_package_store().as_ref()))
533            .get_annotated_layout(struct_tag)
534            .map(|layout| layout.into_layout())
535            .map(Some)
536            .map_err(StorageError::custom)
537    }
538}
iota_core/storage.rs

iota_core/
storage.rs
