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    CheckpointContents, CheckpointContentsDigest, CheckpointDigest, EndOfEpochData, StructTag,
10    TransactionDigest, TransactionEffects, TransactionEvents,
11};
12use iota_types::{
13    committee::{Committee, EpochId},
14    error::IotaError,
15    messages_checkpoint::{
16        CheckpointContentsExt, CheckpointSequenceNumber, FullCheckpointContents,
17        VerifiedCheckpoint, VerifiedCheckpointContents,
18    },
19    object::Object,
20    storage::{
21        ObjectKey, ObjectStore, ReadStore, WriteStore,
22        error::{Error as StorageError, Result},
23    },
24    transaction::VerifiedTransaction,
25};
26use parking_lot::Mutex;
27use tracing::instrument;
28
29use crate::{
30    authority::AuthorityState,
31    checkpoints::CheckpointStore,
32    epoch::committee_store::CommitteeStore,
33    execution_cache::ExecutionCacheTraitPointers,
34    rpc_indexes::{RpcIndexesStore, schema::IndexGroup},
35};
36
37#[derive(Clone)]
38pub struct RocksDbStore {
39    cache_traits: ExecutionCacheTraitPointers,
40
41    committee_store: Arc<CommitteeStore>,
42    checkpoint_store: Arc<CheckpointStore>,
43    // Lower bounds on the watermark rows, not mirrors of them: a row already
44    // ahead leaves its cache behind, which costs a repeated call and nothing
45    // else. Held so that the read and the write of a row happen under one
46    // lock.
47    highest_verified_checkpoint: Arc<Mutex<Option<u64>>>,
48    highest_synced_checkpoint: Arc<Mutex<Option<u64>>>,
49}
50
51impl RocksDbStore {
52    pub fn new(
53        cache_traits: ExecutionCacheTraitPointers,
54        committee_store: Arc<CommitteeStore>,
55        checkpoint_store: Arc<CheckpointStore>,
56    ) -> Self {
57        Self {
58            cache_traits,
59            committee_store,
60            checkpoint_store,
61            highest_verified_checkpoint: Arc::new(Mutex::new(None)),
62            highest_synced_checkpoint: Arc::new(Mutex::new(None)),
63        }
64    }
65
66    pub fn get_objects(&self, object_keys: &[ObjectKey]) -> Result<Vec<Option<Object>>, IotaError> {
67        self.cache_traits
68            .object_cache_reader
69            .try_multi_get_objects_by_key(object_keys)
70    }
71
72    pub fn get_last_executed_checkpoint(&self) -> Result<Option<VerifiedCheckpoint>, IotaError> {
73        Ok(self.checkpoint_store.get_highest_executed_checkpoint()?)
74    }
75
76    /// Marks a consecutive run of checkpoints as synced, writing the watermark
77    /// once for the last one but notifying waiters of every checkpoint.
78    /// Returns whether the watermark moved.
79    fn update_highest_synced_checkpoints(
80        &self,
81        checkpoints: &[VerifiedCheckpoint],
82    ) -> Result<bool, iota_types::storage::error::Error> {
83        let Some(last) = checkpoints.last() else {
84            return Ok(false);
85        };
86        let mut locked = self.highest_synced_checkpoint.lock();
87        if locked.is_some_and(|seq| seq >= last.sequence_number) {
88            return Ok(false);
89        }
90        let moved = self
91            .checkpoint_store
92            .multi_update_highest_synced_checkpoint(checkpoints)
93            .map_err(iota_types::storage::error::Error::custom)?;
94        if moved {
95            *locked = locked.max(Some(last.sequence_number));
96        }
97        Ok(moved)
98    }
99}
100
101impl ReadStore for RocksDbStore {
102    fn try_get_checkpoint_by_digest(
103        &self,
104        digest: &CheckpointDigest,
105    ) -> Result<Option<VerifiedCheckpoint>, StorageError> {
106        self.checkpoint_store
107            .get_checkpoint_by_digest(digest)
108            .map_err(Into::into)
109    }
110
111    fn try_get_checkpoint_by_sequence_number(
112        &self,
113        sequence_number: CheckpointSequenceNumber,
114    ) -> Result<Option<VerifiedCheckpoint>, StorageError> {
115        self.checkpoint_store
116            .get_checkpoint_by_sequence_number(sequence_number)
117            .map_err(Into::into)
118    }
119
120    fn try_get_highest_verified_checkpoint(&self) -> Result<VerifiedCheckpoint, StorageError> {
121        self.checkpoint_store
122            .get_highest_verified_checkpoint()
123            .map(|maybe_checkpoint| {
124                maybe_checkpoint
125                    .expect("storage should have been initialized with genesis checkpoint")
126            })
127            .map_err(Into::into)
128    }
129
130    fn try_get_highest_verified_checkpoint_seq_number(
131        &self,
132    ) -> Result<CheckpointSequenceNumber, StorageError> {
133        Ok(self
134            .checkpoint_store
135            .get_highest_verified_checkpoint_seq_number()?
136            .expect("storage should have been initialized with genesis checkpoint"))
137    }
138
139    fn try_get_highest_synced_checkpoint(&self) -> Result<VerifiedCheckpoint, StorageError> {
140        self.checkpoint_store
141            .get_highest_synced_checkpoint()
142            .map(|maybe_checkpoint| {
143                maybe_checkpoint
144                    .expect("storage should have been initialized with genesis checkpoint")
145            })
146            .map_err(Into::into)
147    }
148
149    fn try_get_highest_synced_checkpoint_seq_number(
150        &self,
151    ) -> Result<CheckpointSequenceNumber, StorageError> {
152        Ok(self
153            .checkpoint_store
154            .get_highest_synced_checkpoint_seq_number()?
155            .expect("storage should have been initialized with genesis checkpoint"))
156    }
157
158    fn try_get_lowest_available_checkpoint(
159        &self,
160    ) -> Result<CheckpointSequenceNumber, StorageError> {
161        if let Some(highest_pruned_cp) = self
162            .checkpoint_store
163            .get_highest_pruned_checkpoint_seq_number()
164            .map_err(Into::<StorageError>::into)?
165        {
166            Ok(highest_pruned_cp + 1)
167        } else {
168            Ok(0)
169        }
170    }
171
172    fn try_get_full_checkpoint_contents_by_sequence_number(
173        &self,
174        sequence_number: CheckpointSequenceNumber,
175    ) -> Result<Option<FullCheckpointContents>, StorageError> {
176        Ok(self
177            .checkpoint_store
178            .get_full_checkpoint_contents_by_sequence_number(sequence_number)
179            .map(|contents| contents.as_ref().clone()))
180    }
181
182    fn try_get_full_checkpoint_contents(
183        &self,
184        digest: &CheckpointContentsDigest,
185    ) -> Result<Option<FullCheckpointContents>, StorageError> {
186        // First look to see if the in-memory cache still holds the complete
187        // contents.
188        if let Some(contents) = self
189            .checkpoint_store
190            .get_full_checkpoint_contents_by_digest(digest)
191        {
192            return Ok(Some(contents.as_ref().clone()));
193        }
194
195        // Otherwise gather it from the individual components: in two batched
196        // reads of the checkpoint's epoch, where a checkpoint's transactions
197        // and effects are stored, and one by one through every epoch for what
198        // those miss, such as a transaction still only in memory.
199        let Some((epoch, contents)) = self
200            .checkpoint_store
201            .get_checkpoint_contents_with_epoch(digest)
202            .map_err(iota_types::storage::error::Error::custom)?
203        else {
204            return Ok(None);
205        };
206        let reader = &self.cache_traits.transaction_cache_reader;
207        let tx_digests: Vec<_> = contents.iter().map(|tx| tx.transaction).collect();
208        let fx_digests: Vec<_> = contents.iter().map(|tx| tx.effects).collect();
209        let transactions = reader
210            .try_multi_get_transaction_blocks_in_epoch(epoch, &tx_digests)
211            .map_err(iota_types::storage::error::Error::custom)?;
212        let effects = reader
213            .try_multi_get_effects_in_epoch(epoch, &fx_digests)
214            .map_err(iota_types::storage::error::Error::custom)?;
215        let mut execution_data = Vec::with_capacity(contents.len());
216        for (((tx_digest, fx_digest), transaction), effects) in tx_digests
217            .iter()
218            .zip(&fx_digests)
219            .zip(transactions)
220            .zip(effects)
221        {
222            let transaction = match transaction {
223                Some(transaction) => transaction,
224                None => match self.try_get_transaction(tx_digest)? {
225                    Some(transaction) => transaction,
226                    None => return Ok(None),
227                },
228            };
229            let effects = match effects {
230                Some(effects) => effects,
231                None => match reader
232                    .try_get_effects(fx_digest)
233                    .map_err(iota_types::storage::error::Error::custom)?
234                {
235                    Some(effects) => effects,
236                    None => return Ok(None),
237                },
238            };
239            execution_data.push(iota_types::base_types::ExecutionData::new(
240                (*transaction).clone().into_inner(),
241                effects,
242            ));
243        }
244        FullCheckpointContents::from_contents_and_execution_data(
245            contents,
246            execution_data.into_iter(),
247        )
248        .map(Some)
249        .map_err(iota_types::storage::error::Error::custom)
250    }
251
252    fn try_get_committee(
253        &self,
254        epoch: EpochId,
255    ) -> Result<Option<Arc<Committee>>, iota_types::storage::error::Error> {
256        Ok(self.committee_store.get_committee(&epoch).unwrap())
257    }
258
259    fn try_get_transaction(
260        &self,
261        digest: &TransactionDigest,
262    ) -> Result<Option<Arc<VerifiedTransaction>>, StorageError> {
263        self.cache_traits
264            .transaction_cache_reader
265            .try_get_transaction_block(digest)
266            .map_err(StorageError::custom)
267    }
268
269    fn try_get_transaction_effects(
270        &self,
271        digest: &TransactionDigest,
272    ) -> Result<Option<TransactionEffects>, StorageError> {
273        self.cache_traits
274            .transaction_cache_reader
275            .try_get_executed_effects(digest)
276            .map_err(StorageError::custom)
277    }
278
279    fn try_get_events(
280        &self,
281        digest: &TransactionDigest,
282    ) -> Result<Option<TransactionEvents>, StorageError> {
283        self.cache_traits
284            .transaction_cache_reader
285            .try_get_events(digest)
286            .map_err(StorageError::custom)
287    }
288
289    fn try_get_latest_checkpoint(&self) -> iota_types::storage::error::Result<VerifiedCheckpoint> {
290        self.checkpoint_store
291            .get_highest_executed_checkpoint()
292            .map_err(iota_types::storage::error::Error::custom)?
293            .ok_or_else(|| {
294                iota_types::storage::error::Error::missing("unable to get latest checkpoint")
295            })
296    }
297
298    fn try_get_checkpoint_contents_by_digest(
299        &self,
300        digest: &CheckpointContentsDigest,
301    ) -> iota_types::storage::error::Result<Option<CheckpointContents>> {
302        self.checkpoint_store
303            .get_checkpoint_contents(digest)
304            .map_err(iota_types::storage::error::Error::custom)
305    }
306
307    fn try_get_checkpoint_contents_by_sequence_number(
308        &self,
309        sequence_number: CheckpointSequenceNumber,
310    ) -> iota_types::storage::error::Result<Option<CheckpointContents>> {
311        match self.try_get_checkpoint_by_sequence_number(sequence_number) {
312            Ok(Some(checkpoint)) => {
313                self.try_get_checkpoint_contents_by_digest(&checkpoint.contents_digest)
314            }
315            Ok(None) => Ok(None),
316            Err(e) => Err(e),
317        }
318    }
319}
320
321impl ObjectStore for RocksDbStore {
322    fn try_get_object(
323        &self,
324        object_id: &iota_sdk_types::ObjectId,
325    ) -> iota_types::storage::error::Result<Option<Object>> {
326        self.cache_traits.object_store.try_get_object(object_id)
327    }
328
329    fn try_get_object_by_key(
330        &self,
331        object_id: &iota_sdk_types::ObjectId,
332        version: iota_types::base_types::VersionNumber,
333    ) -> iota_types::storage::error::Result<Option<Object>> {
334        self.cache_traits
335            .object_store
336            .try_get_object_by_key(object_id, version)
337    }
338}
339
340impl WriteStore for RocksDbStore {
341    #[instrument(level = "trace", skip_all)]
342    fn try_insert_checkpoint(
343        &self,
344        checkpoint: &VerifiedCheckpoint,
345    ) -> Result<(), iota_types::storage::error::Error> {
346        if let Some(EndOfEpochData {
347            next_epoch_committee,
348            ..
349        }) = checkpoint.end_of_epoch_data.as_ref()
350        {
351            let committee = Committee::from_committee_members(
352                checkpoint.epoch().checked_add(1).unwrap(),
353                next_epoch_committee,
354            );
355            self.try_insert_committee(committee)?;
356        }
357
358        self.checkpoint_store
359            .insert_certified_checkpoint(checkpoint)
360            .map_err(iota_types::storage::error::Error::custom)?;
361        // Not `insert_verified_checkpoint`, which would write the watermark
362        // straight to the store: taking the same lock the archive path takes
363        // is what stops two writers leaving the row behind where one of them
364        // had already seen it.
365        self.try_update_highest_verified_checkpoint(checkpoint)
366    }
367
368    fn try_update_highest_verified_checkpoint(
369        &self,
370        checkpoint: &VerifiedCheckpoint,
371    ) -> Result<(), iota_types::storage::error::Error> {
372        let mut locked = self.highest_verified_checkpoint.lock();
373        if locked.is_some() && locked.unwrap() >= checkpoint.sequence_number {
374            return Ok(());
375        }
376        self.checkpoint_store
377            .update_highest_verified_checkpoint(checkpoint)
378            .map_err(iota_types::storage::error::Error::custom)?;
379        *locked = locked.max(Some(checkpoint.sequence_number));
380        Ok(())
381    }
382
383    fn try_update_highest_synced_checkpoint(
384        &self,
385        checkpoint: &VerifiedCheckpoint,
386    ) -> Result<bool, iota_types::storage::error::Error> {
387        self.update_highest_synced_checkpoints(std::slice::from_ref(checkpoint))
388    }
389
390    fn try_insert_checkpoint_contents(
391        &self,
392        checkpoint: &VerifiedCheckpoint,
393        contents: VerifiedCheckpointContents,
394    ) -> Result<(), iota_types::storage::error::Error> {
395        // Transactions go into the bucket of the epoch that executed them, the
396        // contents into that of the epoch that closed the checkpoint. State
397        // sync runs ahead of execution, so either may be an epoch this node
398        // has not begun executing.
399        self.cache_traits
400            .state_sync_store
401            .try_multi_insert_transaction_and_effects(contents.transactions())
402            .map_err(iota_types::storage::error::Error::custom)?;
403        self.checkpoint_store
404            .insert_verified_checkpoint_contents(checkpoint, contents)
405            .map_err(Into::into)
406    }
407
408    fn try_insert_committee(
409        &self,
410        new_committee: Committee,
411    ) -> Result<(), iota_types::storage::error::Error> {
412        self.committee_store
413            .insert_new_committee(&new_committee)
414            .unwrap();
415        Ok(())
416    }
417
418    fn try_get_highest_executed_checkpoint_seq_number(
419        &self,
420    ) -> Result<Option<CheckpointSequenceNumber>, iota_types::storage::error::Error> {
421        self.checkpoint_store
422            .get_highest_executed_checkpoint_seq_number()
423            .map_err(Into::into)
424    }
425
426    async fn wait_for_executed_checkpoint(&self, sequence_number: CheckpointSequenceNumber) {
427        self.checkpoint_store
428            .notify_read_executed_checkpoint(sequence_number)
429            .await;
430    }
431
432    fn try_insert_synced_checkpoints(
433        &self,
434        checkpoints: Vec<(VerifiedCheckpoint, VerifiedCheckpointContents)>,
435    ) -> Result<(), iota_types::storage::error::Error> {
436        let summaries: Vec<VerifiedCheckpoint> = checkpoints
437            .iter()
438            .map(|(checkpoint, _)| checkpoint.clone())
439            .collect();
440        let Some(last) = summaries.last() else {
441            return Ok(());
442        };
443
444        for checkpoint in &summaries {
445            if let Some(EndOfEpochData {
446                next_epoch_committee,
447                ..
448            }) = checkpoint.end_of_epoch_data.as_ref()
449            {
450                let committee = Committee::from_committee_members(
451                    checkpoint.epoch().checked_add(1).unwrap(),
452                    next_epoch_committee,
453                );
454                self.try_insert_committee(committee)?;
455            }
456        }
457
458        self.checkpoint_store
459            .multi_insert_certified_checkpoints(&summaries)?;
460        self.try_update_highest_verified_checkpoint(last)?;
461
462        // Transactions and effects must be durable before their contents
463        // rows (see `CheckpointStore::cache_full_checkpoint_contents`).
464        for (_, contents) in &checkpoints {
465            self.cache_traits
466                .state_sync_store
467                .try_multi_insert_transaction_and_effects(contents.transactions())
468                .map_err(iota_types::storage::error::Error::custom)?;
469        }
470        self.checkpoint_store
471            .multi_insert_verified_checkpoint_contents(checkpoints)?;
472
473        self.update_highest_synced_checkpoints(&summaries)?;
474        Ok(())
475    }
476}
477
478pub struct GrpcReadStore {
479    state: Arc<AuthorityState>,
480    rocks: RocksDbStore,
481}
482
483impl GrpcReadStore {
484    pub fn new(state: Arc<AuthorityState>, rocks: RocksDbStore) -> Self {
485        Self { state, rocks }
486    }
487
488    /// The index store when this node maintains the gRPC group's tables.
489    fn grpc_indexes_store(&self) -> iota_types::storage::error::Result<&RpcIndexesStore> {
490        self.state
491            .rpc_indexes_store
492            .as_deref()
493            .filter(|store| store.serves(IndexGroup::Grpc))
494            .ok_or_else(|| {
495                iota_types::storage::error::Error::custom("gRPC index store is disabled")
496            })
497    }
498}
499
500impl ObjectStore for GrpcReadStore {
501    fn try_get_object(
502        &self,
503        object_id: &iota_sdk_types::ObjectId,
504    ) -> iota_types::storage::error::Result<Option<Object>> {
505        self.rocks.try_get_object(object_id)
506    }
507
508    /// Unlike [`RocksDbStore`], falls back to the historic buckets on a live
509    /// miss, since gRPC responses read past versions by exact key and those
510    /// leave the live table when their checkpoint commits. `RocksDbStore` has
511    /// no fallback because state sync reads through it, where a miss is a bug.
512    fn try_get_object_by_key(
513        &self,
514        object_id: &iota_sdk_types::ObjectId,
515        version: iota_types::base_types::VersionNumber,
516    ) -> iota_types::storage::error::Result<Option<Object>> {
517        self.state
518            .get_object_with_historic_fallback(&ObjectKey(*object_id, version))
519            .map_err(StorageError::custom)
520    }
521}
522
523impl ReadStore for GrpcReadStore {
524    fn try_get_committee(
525        &self,
526        epoch: EpochId,
527    ) -> iota_types::storage::error::Result<Option<Arc<Committee>>> {
528        self.rocks.try_get_committee(epoch)
529    }
530
531    fn try_get_latest_checkpoint(&self) -> iota_types::storage::error::Result<VerifiedCheckpoint> {
532        self.rocks.try_get_latest_checkpoint()
533    }
534
535    fn try_get_highest_verified_checkpoint(
536        &self,
537    ) -> iota_types::storage::error::Result<VerifiedCheckpoint> {
538        self.rocks.try_get_highest_verified_checkpoint()
539    }
540
541    fn try_get_highest_verified_checkpoint_seq_number(
542        &self,
543    ) -> iota_types::storage::error::Result<CheckpointSequenceNumber> {
544        self.rocks.try_get_highest_verified_checkpoint_seq_number()
545    }
546
547    fn try_get_highest_synced_checkpoint(
548        &self,
549    ) -> iota_types::storage::error::Result<VerifiedCheckpoint> {
550        self.rocks.try_get_highest_synced_checkpoint()
551    }
552
553    fn try_get_highest_synced_checkpoint_seq_number(
554        &self,
555    ) -> iota_types::storage::error::Result<CheckpointSequenceNumber> {
556        self.rocks.try_get_highest_synced_checkpoint_seq_number()
557    }
558
559    fn try_get_lowest_available_checkpoint(
560        &self,
561    ) -> iota_types::storage::error::Result<CheckpointSequenceNumber> {
562        self.rocks.try_get_lowest_available_checkpoint()
563    }
564
565    fn try_get_checkpoint_by_digest(
566        &self,
567        digest: &CheckpointDigest,
568    ) -> iota_types::storage::error::Result<Option<VerifiedCheckpoint>> {
569        self.rocks.try_get_checkpoint_by_digest(digest)
570    }
571
572    fn try_get_checkpoint_by_sequence_number(
573        &self,
574        sequence_number: CheckpointSequenceNumber,
575    ) -> iota_types::storage::error::Result<Option<VerifiedCheckpoint>> {
576        self.rocks
577            .try_get_checkpoint_by_sequence_number(sequence_number)
578    }
579
580    fn try_get_checkpoint_contents_by_digest(
581        &self,
582        digest: &CheckpointContentsDigest,
583    ) -> iota_types::storage::error::Result<Option<CheckpointContents>> {
584        self.rocks.try_get_checkpoint_contents_by_digest(digest)
585    }
586
587    fn try_get_checkpoint_contents_by_sequence_number(
588        &self,
589        sequence_number: CheckpointSequenceNumber,
590    ) -> iota_types::storage::error::Result<Option<CheckpointContents>> {
591        self.rocks
592            .try_get_checkpoint_contents_by_sequence_number(sequence_number)
593    }
594
595    fn try_get_transaction(
596        &self,
597        digest: &TransactionDigest,
598    ) -> iota_types::storage::error::Result<Option<Arc<VerifiedTransaction>>> {
599        self.rocks.try_get_transaction(digest)
600    }
601
602    fn try_get_transaction_effects(
603        &self,
604        digest: &TransactionDigest,
605    ) -> iota_types::storage::error::Result<Option<TransactionEffects>> {
606        self.rocks.try_get_transaction_effects(digest)
607    }
608
609    fn try_get_events(
610        &self,
611        digest: &TransactionDigest,
612    ) -> iota_types::storage::error::Result<Option<TransactionEvents>> {
613        self.rocks.try_get_events(digest)
614    }
615
616    fn try_get_full_checkpoint_contents_by_sequence_number(
617        &self,
618        sequence_number: CheckpointSequenceNumber,
619    ) -> iota_types::storage::error::Result<Option<FullCheckpointContents>> {
620        self.rocks
621            .try_get_full_checkpoint_contents_by_sequence_number(sequence_number)
622    }
623
624    fn try_get_full_checkpoint_contents(
625        &self,
626        digest: &CheckpointContentsDigest,
627    ) -> iota_types::storage::error::Result<Option<FullCheckpointContents>> {
628        self.rocks.try_get_full_checkpoint_contents(digest)
629    }
630}
631
632impl GrpcStateReader for GrpcReadStore {
633    fn get_lowest_available_checkpoint_objects(
634        &self,
635    ) -> iota_types::storage::error::Result<CheckpointSequenceNumber> {
636        self.rocks
637            .checkpoint_store
638            .lowest_checkpoint_with_retained_objects(
639                self.state.get_historic_objects().earliest_readable_epoch(),
640                self.state.get_historic_objects().objects_pruned_through,
641                self.state.get_historic_objects().backlog_sweep_epoch(),
642            )
643            .map_err(StorageError::custom)
644    }
645
646    fn get_chain_identifier(&self) -> Result<iota_types::digests::ChainIdentifier> {
647        Ok(self.state.get_chain_identifier())
648    }
649
650    fn get_epoch_last_checkpoint(
651        &self,
652        epoch_id: EpochId,
653    ) -> iota_types::storage::error::Result<Option<VerifiedCheckpoint>> {
654        self.rocks
655            .checkpoint_store
656            .get_epoch_last_checkpoint(epoch_id)
657            .map_err(iota_types::storage::error::Error::custom)
658    }
659
660    fn get_epoch_info(
661        &self,
662        epoch: EpochId,
663    ) -> iota_types::storage::error::Result<Option<iota_types::storage::EpochInfoV2>> {
664        self.rocks
665            .checkpoint_store
666            .get_epoch_info(epoch)
667            .map_err(iota_types::storage::error::Error::custom)
668    }
669
670    fn grpc_indexes(&self) -> Option<&dyn GrpcIndexes> {
671        self.grpc_indexes_store().ok().map(|index| index as _)
672    }
673
674    /// Answered from the ledger rather than the RPC index, so a node
675    /// retaining fewer index epochs than ledger epochs still reports
676    /// confirmation for every transaction it has. Reported only once the RPC
677    /// indexes have caught up with the checkpoint, so that an index-backed
678    /// read sent right after sees the transaction.
679    fn get_transaction_checkpoint(
680        &self,
681        digest: &TransactionDigest,
682    ) -> iota_types::storage::error::Result<Option<CheckpointSequenceNumber>> {
683        let indexed_through = self.state.get_latest_checkpoint_sequence_number().ok();
684        Ok(self
685            .state
686            .get_checkpoint_cache()
687            .try_get_transaction_perpetual_checkpoint(digest)
688            .map_err(iota_types::storage::error::Error::custom)?
689            .map(|(_epoch, checkpoint)| checkpoint)
690            .filter(|checkpoint| indexed_through.is_some_and(|i| *checkpoint <= i)))
691    }
692
693    fn get_struct_layout(
694        &self,
695        struct_tag: &StructTag,
696    ) -> Result<Option<move_core_types::annotated_value::MoveTypeLayout>> {
697        self.state
698            .load_epoch_store_one_call_per_task()
699            .executor()
700            // TODO(cache) - must read through cache
701            .type_layout_resolver(Box::new(self.state.get_backing_package_store().as_ref()))
702            .get_annotated_layout(struct_tag)
703            .map(|layout| layout.into_layout())
704            .map(Some)
705            .map_err(StorageError::custom)
706    }
707}
iota_core/storage.rs

iota_core/
storage.rs
