iota_core/execution_cache/

proxy_cache.rs

// Copyright (c) Mysten Labs, Inc.
// Modifications Copyright (c) 2024 IOTA Stiftung
// SPDX-License-Identifier: Apache-2.0

use std::{sync::Arc, time::Duration};

use futures::{FutureExt, future::BoxFuture};
use iota_config::node::ExecutionCacheTypeAtomicU8;
use iota_types::{
    accumulator::Accumulator,
    base_types::{EpochId, ObjectID, ObjectRef, SequenceNumber, VerifiedExecutionData},
    digests::{TransactionDigest, TransactionEffectsDigest, TransactionEventsDigest},
    effects::{TransactionEffects, TransactionEvents},
    error::{IotaError, IotaResult},
    executable_transaction::VerifiedExecutableTransaction,
    iota_system_state::IotaSystemState,
    messages_checkpoint::CheckpointSequenceNumber,
    object::Object,
    storage::{MarkerValue, ObjectKey, ObjectOrTombstone, PackageObject},
    transaction::{VerifiedSignedTransaction, VerifiedTransaction},
};
use tracing::instrument;

use super::{
    CheckpointCache, ExecutionCacheCommit, ExecutionCacheConfig, ExecutionCacheMetrics,
    ExecutionCacheReconfigAPI, ExecutionCacheType, ExecutionCacheWrite, ObjectCacheRead,
    PassthroughCache, StateSyncAPI, TestingAPI, TransactionCacheRead, WritebackCache,
};
use crate::{
    authority::{
        AuthorityStore,
        authority_per_epoch_store::AuthorityPerEpochStore,
        authority_store::{ExecutionLockWriteGuard, IotaLockResult},
        backpressure::BackpressureManager,
        epoch_start_configuration::{EpochFlag, EpochStartConfigTrait, EpochStartConfiguration},
    },
    state_accumulator::AccumulatorStore,
    transaction_outputs::TransactionOutputs,
};

macro_rules! delegate_method {
    ($self:ident.$method:ident($($args:ident),*)) => {
        match ExecutionCacheType::from(&$self.mode.load(std::sync::atomic::Ordering::Acquire)) {
            ExecutionCacheType::PassthroughCache => $self.passthrough_cache.$method($($args),*),
            ExecutionCacheType::WritebackCache => $self.writeback_cache.$method($($args),*),
        }
    };
}

pub struct ProxyCache {
    // Note: both caches must be constructed at startup, rather than using ArcSwap
    // (or some similar strategy). This is because we need to proxy iter_live_object_set,
    // which requires that we borrow from a member of the cache. If we used ArcSwap,
    // we would be forced to borrow from a local variable after loading from the ArcSwap.
    //
    // Cache implementations are entirely passive, so the unused one will have no effect.
    passthrough_cache: PassthroughCache,
    writeback_cache: WritebackCache,
    mode: ExecutionCacheTypeAtomicU8,
}

impl ProxyCache {
    pub fn new(
        cache_config: &ExecutionCacheConfig,
        epoch_start_config: &EpochStartConfiguration,
        store: Arc<AuthorityStore>,
        metrics: Arc<ExecutionCacheMetrics>,
        backpressure_manager: Arc<BackpressureManager>,
    ) -> Self {
        let cache_type = epoch_start_config.execution_cache_type();
        tracing::info!("using cache impl {:?}", cache_type);
        let passthrough_cache = PassthroughCache::new(store.clone(), metrics.clone());

        let writeback_cache = WritebackCache::new(
            &cache_config.writeback_cache,
            store.clone(),
            metrics.clone(),
            backpressure_manager,
        );

        Self {
            passthrough_cache,
            writeback_cache,
            mode: cache_type.into(),
        }
    }

    async fn reconfigure_cache_impl(&self, epoch_start_config: &EpochStartConfiguration) {
        let cache_type = epoch_start_config.execution_cache_type();
        tracing::info!("switching to cache impl {:?}", cache_type);
        if matches!(cache_type, ExecutionCacheType::PassthroughCache) {
            // we may switch back to the writeback cache next epoch, at which point its
            // caches will be stale if not cleared now

            // When we call invalidate_all on Moka caches, it sets the valid after time
            // stamp to the current time. Upon retrieval, it ignores entries
            // whose insertion time is strictly less than the valid-after
            // time. In the simulator, time remains constant for the duration of a single
            // task poll, so it is possible that entries have been inserted in
            // the same tick, and will therefore not be invalidated
            // properly. So, this sleep is necessary for passing tests, and it also is some
            // insurance against hitting the same issue in production. (It
            // should be more or less impossible for two consecutive
            // calls to Instant::now() to return the same value in production, but there is
            // no harm in having a short sleep here just to be sure).
            tokio::time::sleep(Duration::from_nanos(100)).await;
            self.writeback_cache.clear_caches_and_assert_empty();
        }
        self.mode
            .store(cache_type.into(), std::sync::atomic::Ordering::Release);
    }
}

impl ObjectCacheRead for ProxyCache {
    fn try_get_package_object(&self, package_id: &ObjectID) -> IotaResult<Option<PackageObject>> {
        delegate_method!(self.try_get_package_object(package_id))
    }

    fn force_reload_system_packages(&self, system_package_ids: &[ObjectID]) {
        delegate_method!(self.force_reload_system_packages(system_package_ids))
    }

    fn try_get_object(&self, id: &ObjectID) -> IotaResult<Option<Object>> {
        delegate_method!(self.try_get_object(id))
    }

    fn try_get_object_by_key(
        &self,
        object_id: &ObjectID,
        version: SequenceNumber,
    ) -> IotaResult<Option<Object>> {
        delegate_method!(self.try_get_object_by_key(object_id, version))
    }

    fn try_multi_get_objects_by_key(
        &self,
        object_keys: &[ObjectKey],
    ) -> Result<Vec<Option<Object>>, IotaError> {
        delegate_method!(self.try_multi_get_objects_by_key(object_keys))
    }

    fn try_object_exists_by_key(
        &self,
        object_id: &ObjectID,
        version: SequenceNumber,
    ) -> IotaResult<bool> {
        delegate_method!(self.try_object_exists_by_key(object_id, version))
    }

    fn try_multi_object_exists_by_key(&self, object_keys: &[ObjectKey]) -> IotaResult<Vec<bool>> {
        delegate_method!(self.try_multi_object_exists_by_key(object_keys))
    }

    fn try_get_latest_object_ref_or_tombstone(
        &self,
        object_id: ObjectID,
    ) -> IotaResult<Option<ObjectRef>> {
        delegate_method!(self.try_get_latest_object_ref_or_tombstone(object_id))
    }

    fn try_get_latest_object_or_tombstone(
        &self,
        object_id: ObjectID,
    ) -> Result<Option<(ObjectKey, ObjectOrTombstone)>, IotaError> {
        delegate_method!(self.try_get_latest_object_or_tombstone(object_id))
    }

    fn try_find_object_lt_or_eq_version(
        &self,
        object_id: ObjectID,
        version: SequenceNumber,
    ) -> IotaResult<Option<Object>> {
        delegate_method!(self.try_find_object_lt_or_eq_version(object_id, version))
    }

    fn try_get_lock(
        &self,
        obj_ref: ObjectRef,
        epoch_store: &AuthorityPerEpochStore,
    ) -> IotaLockResult {
        delegate_method!(self.try_get_lock(obj_ref, epoch_store))
    }

    fn _try_get_live_objref(&self, object_id: ObjectID) -> IotaResult<ObjectRef> {
        delegate_method!(self._try_get_live_objref(object_id))
    }

    fn try_check_owned_objects_are_live(&self, owned_object_refs: &[ObjectRef]) -> IotaResult {
        delegate_method!(self.try_check_owned_objects_are_live(owned_object_refs))
    }

    fn try_get_iota_system_state_object_unsafe(&self) -> IotaResult<IotaSystemState> {
        delegate_method!(self.try_get_iota_system_state_object_unsafe())
    }

    fn try_get_marker_value(
        &self,
        object_id: &ObjectID,
        version: SequenceNumber,
        epoch_id: EpochId,
    ) -> IotaResult<Option<MarkerValue>> {
        delegate_method!(self.try_get_marker_value(object_id, version, epoch_id))
    }

    fn try_get_latest_marker(
        &self,
        object_id: &ObjectID,
        epoch_id: EpochId,
    ) -> IotaResult<Option<(SequenceNumber, MarkerValue)>> {
        delegate_method!(self.try_get_latest_marker(object_id, epoch_id))
    }

    fn try_get_highest_pruned_checkpoint(&self) -> IotaResult<CheckpointSequenceNumber> {
        delegate_method!(self.try_get_highest_pruned_checkpoint())
    }
}

impl TransactionCacheRead for ProxyCache {
    #[instrument(level = "trace", skip_all)]
    fn try_multi_get_transaction_blocks(
        &self,
        digests: &[TransactionDigest],
    ) -> IotaResult<Vec<Option<Arc<VerifiedTransaction>>>> {
        delegate_method!(self.try_multi_get_transaction_blocks(digests))
    }

    #[instrument(level = "trace", skip_all)]
    fn try_multi_get_executed_effects_digests(
        &self,
        digests: &[TransactionDigest],
    ) -> IotaResult<Vec<Option<TransactionEffectsDigest>>> {
        delegate_method!(self.try_multi_get_executed_effects_digests(digests))
    }

    #[instrument(level = "trace", skip_all)]
    fn try_multi_get_effects(
        &self,
        digests: &[TransactionEffectsDigest],
    ) -> IotaResult<Vec<Option<TransactionEffects>>> {
        delegate_method!(self.try_multi_get_effects(digests))
    }

    #[instrument(level = "trace", skip_all)]
    fn try_notify_read_executed_effects_digests<'a>(
        &'a self,
        digests: &'a [TransactionDigest],
    ) -> BoxFuture<'a, IotaResult<Vec<TransactionEffectsDigest>>> {
        delegate_method!(self.try_notify_read_executed_effects_digests(digests))
    }

    #[instrument(level = "trace", skip_all)]
    fn try_multi_get_events(
        &self,
        event_digests: &[TransactionEventsDigest],
    ) -> IotaResult<Vec<Option<TransactionEvents>>> {
        delegate_method!(self.try_multi_get_events(event_digests))
    }
}

impl ExecutionCacheWrite for ProxyCache {
    fn try_write_transaction_outputs(
        &self,
        epoch_id: EpochId,
        tx_outputs: Arc<TransactionOutputs>,
    ) -> IotaResult {
        delegate_method!(self.try_write_transaction_outputs(epoch_id, tx_outputs))
    }

    #[instrument(level = "trace", skip_all)]
    fn try_acquire_transaction_locks<'a>(
        &'a self,
        epoch_store: &'a AuthorityPerEpochStore,
        owned_input_objects: &'a [ObjectRef],
        transaction: VerifiedSignedTransaction,
    ) -> IotaResult {
        delegate_method!(self.try_acquire_transaction_locks(
            epoch_store,
            owned_input_objects,
            transaction
        ))
    }
}

impl AccumulatorStore for ProxyCache {
    fn get_root_state_accumulator_for_epoch(
        &self,
        epoch: EpochId,
    ) -> IotaResult<Option<(CheckpointSequenceNumber, Accumulator)>> {
        delegate_method!(self.get_root_state_accumulator_for_epoch(epoch))
    }

    fn get_root_state_accumulator_for_highest_epoch(
        &self,
    ) -> IotaResult<Option<(EpochId, (CheckpointSequenceNumber, Accumulator))>> {
        delegate_method!(self.get_root_state_accumulator_for_highest_epoch())
    }

    fn insert_state_accumulator_for_epoch(
        &self,
        epoch: EpochId,
        checkpoint_seq_num: &CheckpointSequenceNumber,
        acc: &Accumulator,
    ) -> IotaResult {
        delegate_method!(self.insert_state_accumulator_for_epoch(epoch, checkpoint_seq_num, acc))
    }

    fn iter_live_object_set(
        &self,
    ) -> Box<dyn Iterator<Item = crate::authority::authority_store_tables::LiveObject> + '_> {
        delegate_method!(self.iter_live_object_set())
    }

    fn iter_cached_live_object_set_for_testing(
        &self,
    ) -> Box<dyn Iterator<Item = crate::authority::authority_store_tables::LiveObject> + '_> {
        delegate_method!(self.iter_cached_live_object_set_for_testing())
    }
}

impl ExecutionCacheCommit for ProxyCache {
    fn try_commit_transaction_outputs(
        &self,
        epoch: EpochId,
        digests: &[TransactionDigest],
    ) -> IotaResult {
        delegate_method!(self.try_commit_transaction_outputs(epoch, digests))
    }

    fn try_persist_transaction(&self, tx: &VerifiedExecutableTransaction) -> IotaResult {
        delegate_method!(self.try_persist_transaction(tx))
    }

    fn approximate_pending_transaction_count(&self) -> u64 {
        delegate_method!(self.approximate_pending_transaction_count())
    }
}

impl CheckpointCache for ProxyCache {
    fn try_get_transaction_perpetual_checkpoint(
        &self,
        digest: &TransactionDigest,
    ) -> IotaResult<Option<(EpochId, CheckpointSequenceNumber)>> {
        delegate_method!(self.try_get_transaction_perpetual_checkpoint(digest))
    }

    fn try_multi_get_transactions_perpetual_checkpoints(
        &self,
        digests: &[TransactionDigest],
    ) -> IotaResult<Vec<Option<(EpochId, CheckpointSequenceNumber)>>> {
        delegate_method!(self.try_multi_get_transactions_perpetual_checkpoints(digests))
    }

    fn try_insert_finalized_transactions_perpetual_checkpoints(
        &self,
        digests: &[TransactionDigest],
        epoch: EpochId,
        sequence: CheckpointSequenceNumber,
    ) -> IotaResult {
        delegate_method!(
            self.try_insert_finalized_transactions_perpetual_checkpoints(digests, epoch, sequence)
        )
    }
}

impl ExecutionCacheReconfigAPI for ProxyCache {
    fn try_insert_genesis_object(&self, object: Object) -> IotaResult {
        delegate_method!(self.try_insert_genesis_object(object))
    }

    fn try_bulk_insert_genesis_objects(&self, objects: &[Object]) -> IotaResult {
        delegate_method!(self.try_bulk_insert_genesis_objects(objects))
    }

    fn try_revert_state_update(&self, digest: &TransactionDigest) -> IotaResult {
        delegate_method!(self.try_revert_state_update(digest))
    }

    fn try_set_epoch_start_configuration(
        &self,
        epoch_start_config: &EpochStartConfiguration,
    ) -> IotaResult {
        delegate_method!(self.try_set_epoch_start_configuration(epoch_start_config))
    }

    fn update_epoch_flags_metrics(&self, old: &[EpochFlag], new: &[EpochFlag]) {
        delegate_method!(self.update_epoch_flags_metrics(old, new))
    }

    fn clear_state_end_of_epoch(&self, execution_guard: &ExecutionLockWriteGuard<'_>) {
        delegate_method!(self.clear_state_end_of_epoch(execution_guard))
    }

    fn try_expensive_check_iota_conservation(
        &self,
        old_epoch_store: &AuthorityPerEpochStore,
        epoch_supply_change: Option<i64>,
    ) -> IotaResult {
        delegate_method!(
            self.try_expensive_check_iota_conservation(old_epoch_store, epoch_supply_change)
        )
    }

    fn try_checkpoint_db(&self, path: &std::path::Path) -> IotaResult {
        delegate_method!(self.try_checkpoint_db(path))
    }

    fn reconfigure_cache<'a>(
        &'a self,
        epoch_start_config: &'a EpochStartConfiguration,
    ) -> BoxFuture<'a, ()> {
        self.reconfigure_cache_impl(epoch_start_config).boxed()
    }
}

impl StateSyncAPI for ProxyCache {
    fn try_insert_transaction_and_effects(
        &self,
        transaction: &VerifiedTransaction,
        transaction_effects: &TransactionEffects,
    ) -> IotaResult {
        delegate_method!(self.try_insert_transaction_and_effects(transaction, transaction_effects))
    }

    fn try_multi_insert_transaction_and_effects(
        &self,
        transactions_and_effects: &[VerifiedExecutionData],
    ) -> IotaResult {
        delegate_method!(self.try_multi_insert_transaction_and_effects(transactions_and_effects))
    }
}

impl TestingAPI for ProxyCache {
    fn database_for_testing(&self) -> Arc<AuthorityStore> {
        delegate_method!(self.database_for_testing())
    }
}