upstream(core): Track and store unchanged loaded runtime objects

crates/iota-analytics-indexer/src/handlers/transaction_handler.rs

@@ -223,11 +223,13 @@ mod tests {
 
         // Create a checkpoint which should include the transaction we executed.
         let checkpoint = sim.create_checkpoint();
-        let checkpoint_data = sim.get_checkpoint_data(
-            checkpoint.clone(),
-            sim.get_checkpoint_contents_by_digest(&checkpoint.contents_digest)
-                .unwrap(),
-        );
+        let checkpoint_data: iota_types::full_checkpoint_content::CheckpointData = sim
+            .get_checkpoint_data(
+                checkpoint.clone(),
+                sim.get_checkpoint_contents_by_digest(&checkpoint.contents_digest)
+                    .unwrap(),
+            )
+            .into();
         let shared_checkpoint_data = Arc::new(checkpoint_data);
         let txn_handler = TransactionHandler::new();
         txn_handler

crates/iota-core/src/authority.rs

@@ -106,7 +106,8 @@ use iota_types::{
     move_authenticator::MoveAuthenticatorExt,
     object::{Object, ObjectRead, PastObjectRead, bounded_visitor::BoundedVisitor},
     storage::{
-        BackingPackageStore, BackingStore, ObjectKey, ObjectOrTombstone, ObjectStore, WriteKind,
+        BackingPackageStore, BackingStore, ObjectKey, ObjectOrTombstone, ObjectStore,
+        TrackingBackingStore, WriteKind,
     },
     supported_protocol_versions::{
         ProtocolConfig, SupportedProtocolVersions, SupportedProtocolVersionsWithHashes,
@@ -1649,20 +1650,21 @@ impl AuthorityState {
         // errors). However, all errors from this function occur before we have
         // written anything to the db, so we commit the tx guard and rely on the
         // client to retry the tx (if it was transient).
-        let (inner_temporary_store, effects, execution_error_opt) = match self.execute_transaction(
-            &execution_guard,
-            transaction,
-            tx_input_objects,
-            per_authenticator_inputs,
-            epoch_store,
-        ) {
-            Err(e) => {
-                info!(name = ?self.name, ?digest, "Error preparing transaction: {e}");
-                tx_guard.release();
-                return Err(e);
-            }
-            Ok(res) => res,
-        };
+        let (inner_temporary_store, effects, execution_error_opt, unchanged_loaded_runtime_objects) =
+            match self.execute_transaction(
+                &execution_guard,
+                transaction,
+                tx_input_objects,
+                per_authenticator_inputs,
+                epoch_store,
+            ) {
+                Err(e) => {
+                    info!(name = ?self.name, ?digest, "Error preparing transaction: {e}");
+                    tx_guard.release();
+                    return Err(e);
+                }
+                Ok(res) => res,
+            };
 
         if let Some(expected_effects_digest) = expected_effects_digest {
             if effects.digest() != expected_effects_digest {
@@ -1707,6 +1709,7 @@ impl AuthorityState {
             transaction,
             inner_temporary_store,
             &effects,
+            unchanged_loaded_runtime_objects,
             tx_guard,
             execution_guard,
             epoch_store,
@@ -1740,6 +1743,7 @@ impl AuthorityState {
         transaction: &VerifiedExecutableTransaction,
         inner_temporary_store: InnerTemporaryStore,
         effects: &TransactionEffects,
+        unchanged_loaded_runtime_objects: Vec<ObjectKey>,
         tx_guard: TxGuard,
         _execution_guard: ExecutionLockReadGuard<'_>,
         epoch_store: &Arc<AuthorityPerEpochStore>,
@@ -1775,6 +1779,7 @@ impl AuthorityState {
             transaction.clone().into_unsigned(),
             effects.clone(),
             inner_temporary_store,
+            unchanged_loaded_runtime_objects,
         );
         self.get_cache_writer()
             .try_write_transaction_outputs(epoch_store.epoch(), transaction_outputs.into())?;
@@ -1860,6 +1865,7 @@ impl AuthorityState {
         InnerTemporaryStore,
         TransactionEffects,
         Option<ExecutionError>,
+        Vec<ObjectKey>,
     )> {
         let _scope = monitored_scope("Execution::execute_certificate");
         let _metrics_guard = self.metrics.prepare_certificate_latency.start_timer();
@@ -1875,7 +1881,8 @@ impl AuthorityState {
             .epoch_data()
             .epoch_start_timestamp();
 
-        let backing_store = self.get_backing_store().as_ref();
+        let tracking_store = TrackingBackingStore::new(self.get_backing_store().as_ref());
+        let backing_store = &tracking_store;
 
         let tx_digest = *transaction.digest();
 
@@ -2082,7 +2089,19 @@ impl AuthorityState {
                 .observe(effects.gas_cost_summary().computation_cost as f64 / elapsed);
         }
 
-        Ok((inner_temp_store, effects, execution_error_opt.err()))
+        let unchanged_loaded_runtime_objects =
+            iota_types::storage::unchanged_loaded_runtime_objects(
+                tx,
+                &effects,
+                &tracking_store.into_read_objects(),
+            );
+
+        Ok((
+            inner_temp_store,
+            effects,
+            execution_error_opt.err(),
+            unchanged_loaded_runtime_objects,
+        ))
     }
 
     pub fn prepare_transaction_for_benchmark(
@@ -2105,6 +2124,9 @@ impl AuthorityState {
             vec![],
             epoch_store,
         )
+        .map(|(inner_temp_store, effects, execution_error, _)| {
+            (inner_temp_store, effects, execution_error)
+        })
     }
 
     /// Simulate a transaction without committing it.
@@ -2305,6 +2327,7 @@ impl AuthorityState {
 
         // Execute the simulation
         let (kind, signer, gas_data) = transaction.execution_parts();
+
         let (inner_temp_store, _, effects, execution_result) = executor.dev_inspect_transaction(
             self.get_backing_store().as_ref(),
             protocol_config,
@@ -2325,8 +2348,16 @@ impl AuthorityState {
             checks.disabled(),
         );
 
+        let mut input_objects = inner_temp_store.input_objects;
+        iota_types::storage::extend_input_objects_with_loaded_runtime_objects(
+            &mut input_objects,
+            &effects,
+            &inner_temp_store.loaded_runtime_objects,
+            self.get_backing_store().as_object_store(),
+        );
+
         Ok(SimulateTransactionResult {
-            input_objects: inner_temp_store.input_objects,
+            input_objects,
             output_objects: inner_temp_store.written,
             events: effects.events_digest().map(|_| inner_temp_store.events),
             effects,
@@ -5265,7 +5296,7 @@ impl AuthorityState {
         let (input_objects, _) =
             self.read_objects_for_execution(&tx_lock, &executable_tx, epoch_store)?;
 
-        let (temporary_store, effects, _execution_error_opt) = self.execute_transaction(
+        let (temporary_store, effects, _execution_error_opt, _) = self.execute_transaction(
             &execution_guard,
             &executable_tx,
             input_objects,

crates/iota-core/src/authority/authority_store.rs

@@ -406,6 +406,15 @@ impl AuthorityStore {
             .collect::<Result<Vec<_>, _>>()?)
     }
 
+    pub fn get_unchanged_loaded_runtime_objects(
+        &self,
+        digest: &TransactionDigest,
+    ) -> Result<Option<Vec<ObjectKey>>, TypedStoreError> {
+        self.perpetual_tables
+            .unchanged_loaded_runtime_objects
+            .get(digest)
+    }
+
     pub fn multi_get_effects<'a>(
         &self,
         effects_digests: impl Iterator<Item = &'a TransactionEffectsDigest>,
@@ -835,6 +844,7 @@ impl AuthorityStore {
             deleted,
             written,
             events,
+            unchanged_loaded_runtime_objects,
             live_object_markers_to_delete,
             new_live_object_markers_to_init,
             ..
@@ -885,6 +895,14 @@ impl AuthorityStore {
             )?;
         }
 
+        // Write unchanged_loaded_runtime_objects
+        if !unchanged_loaded_runtime_objects.is_empty() {
+            write_batch.insert_batch(
+                &self.perpetual_tables.unchanged_loaded_runtime_objects,
+                [(transaction_digest, unchanged_loaded_runtime_objects)],
+            )?;
+        }
+
         self.initialize_live_object_markers_impl(write_batch, new_live_object_markers_to_init)?;
 
         // Note: deletes live object markers for received objects as well (but not for

crates/iota-core/src/authority/authority_store_pruner.rs

@@ -347,7 +347,7 @@ impl AuthorityStorePruner {
         perpetual_batch.delete_batch(&perpetual_db.executed_effects, transactions.iter())?;
         perpetual_batch.delete_batch(
             &perpetual_db.executed_transactions_to_checkpoint,
-            transactions,
+            transactions.iter(),
         )?;
 
         let mut effect_digests = vec![];
@@ -361,6 +361,10 @@ impl AuthorityStorePruner {
                     .delete_batch(&perpetual_db.events_2, [effects.transaction_digest()])?;
             }
         }
+        perpetual_batch.delete_batch(
+            &perpetual_db.unchanged_loaded_runtime_objects,
+            transactions.iter(),
+        )?;
         perpetual_batch.delete_batch(&perpetual_db.effects, effect_digests)?;
 
         let mut checkpoints_batch = checkpoint_db.tables.certified_checkpoints.batch();

crates/iota-core/src/authority/authority_store_tables.rs

@@ -109,6 +109,9 @@ pub struct AuthorityPerpetualTables {
     // Events keyed by the digest of the transaction that produced them.
     pub(crate) events_2: DBMap<TransactionDigest, TransactionEvents>,
 
+    // Loaded (and unchanged) runtime object references.
+    pub(crate) unchanged_loaded_runtime_objects: DBMap<TransactionDigest, Vec<ObjectKey>>,
+
     /// Epoch and checkpoint of transactions finalized by checkpoint
     /// executor. Currently, mainly used to implement JSON RPC `ReadApi`.
     /// Note, there is a table with the same name in

crates/iota-core/src/checkpoints/checkpoint_executor/data_ingestion_handler.rs

@@ -2,13 +2,17 @@
 // Modifications Copyright (c) 2024 IOTA Stiftung
 // SPDX-License-Identifier: Apache-2.0
 
-use std::{collections::HashMap, path::Path};
+use std::{
+    collections::{BTreeSet, HashMap},
+    path::Path,
+};
 
 use iota_storage::blob::{Blob, BlobEncoding};
 use iota_types::{
     effects::TransactionEffectsAPI,
     error::{IotaError, IotaResult},
-    full_checkpoint_content::{CheckpointData, CheckpointTransaction},
+    full_checkpoint_content::{Checkpoint, CheckpointData, ExecutedTransaction},
+    object::ObjectSet,
     storage::ObjectStore,
 };
 
@@ -17,19 +21,19 @@ use crate::{
     execution_cache::TransactionCacheRead,
 };
 
-pub(crate) fn load_checkpoint_data(
+pub(crate) fn load_checkpoint(
     checkpoint_exec_data: &CheckpointExecutionData,
     checkpoint_tx_data: &CheckpointTransactionData,
     object_store: &dyn ObjectStore,
     transaction_cache_reader: &dyn TransactionCacheRead,
-) -> IotaResult<CheckpointData> {
+) -> IotaResult<Checkpoint> {
     let event_tx_digests = checkpoint_tx_data
         .effects
         .iter()
         .flat_map(|fx| fx.events_digest().map(|_| fx.transaction_digest()).copied())
         .collect::<Vec<_>>();
 
-    let events = transaction_cache_reader
+    let mut events = transaction_cache_reader
         .try_multi_get_events(&event_tx_digests)?
         .into_iter()
         .zip(event_tx_digests)
@@ -40,39 +44,66 @@ pub(crate) fn load_checkpoint_data(
         })
         .collect::<IotaResult<HashMap<_, _>>>()?;
 
-    let mut full_transactions = Vec::with_capacity(checkpoint_tx_data.transactions.len());
+    let mut transactions = Vec::with_capacity(checkpoint_tx_data.transactions.len());
     for (tx, fx) in checkpoint_tx_data
         .transactions
         .iter()
         .zip(checkpoint_tx_data.effects.iter())
     {
         let events = fx.events_digest().map(|_event_digest| {
             events
-                .get(fx.transaction_digest())
-                .cloned()
+                .remove(fx.transaction_digest())
                 .expect("event was already checked to be present")
         });
 
-        let input_objects = iota_types::storage::get_transaction_input_objects(object_store, fx)
-            .map_err(|e| IotaError::Unknown(e.to_string()))?;
-        let output_objects = iota_types::storage::get_transaction_output_objects(object_store, fx)
-            .map_err(|e| IotaError::Unknown(e.to_string()))?;
-
-        let full_transaction = CheckpointTransaction {
-            transaction: (*tx).clone().into_unsigned().into(),
+        let transaction = ExecutedTransaction {
+            transaction: tx.data().transaction().clone(),
+            signatures: tx.data().signatures().to_vec(),
             effects: fx.clone(),
             events,
-            input_objects,
-            output_objects,
+            unchanged_loaded_runtime_objects: transaction_cache_reader
+                .get_unchanged_loaded_runtime_objects(tx.digest())
+                // We don't write empty sets to the DB to save space, so if this load went
+                // through the writeback cache to the DB itself it wouldn't find an entry.
+                .unwrap_or_default(),
         };
-        full_transactions.push(full_transaction);
+        transactions.push(transaction);
     }
-    let checkpoint_data = CheckpointData {
-        checkpoint_summary: checkpoint_exec_data.checkpoint.clone().into(),
-        checkpoint_contents: checkpoint_exec_data.checkpoint_contents.clone(),
-        transactions: full_transactions,
+
+    let object_set = {
+        let refs = transactions
+            .iter()
+            .flat_map(|tx| {
+                iota_types::storage::get_transaction_object_set(
+                    &tx.transaction,
+                    &tx.effects,
+                    &tx.unchanged_loaded_runtime_objects,
+                )
+            })
+            .collect::<BTreeSet<_>>()
+            .into_iter()
+            .collect::<Vec<_>>();
+
+        let objects = object_store.multi_get_objects_by_key(&refs);
+
+        let mut object_set = ObjectSet::default();
+        for (idx, object) in objects.into_iter().enumerate() {
+            object_set.insert(object.ok_or_else(|| {
+                iota_types::storage::error::Error::custom(format!(
+                    "unable to load object {:?}",
+                    refs[idx]
+                ))
+            })?);
+        }
+        object_set
+    };
+    let checkpoint = Checkpoint {
+        summary: checkpoint_exec_data.checkpoint.clone().into(),
+        contents: checkpoint_exec_data.checkpoint_contents.clone(),
+        transactions,
+        object_set,
     };
-    Ok(checkpoint_data)
+    Ok(checkpoint)
 }
 
 pub(crate) fn store_checkpoint_locally(