Add test checkpoint data builder (#20749)

## Description 

This PR adds a builder that could generate arbitrary checkpoint data.
It will make it easy to test the indexer.

## Test plan 

Added unit tests.

---

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [ ] Nodes (Validators and Full nodes): 
- [ ] gRPC:
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:

crates/sui-benchmark/tests/simtest.rs

@@ -593,7 +593,7 @@ mod test {
     // Tests cluster defense against failing transaction floods Traffic Control
     #[sim_test(config = "test_config()")]
     async fn test_simulated_load_expected_failure_traffic_control() {
-        // TODO: can we get away with significatly increasing this?
+        // TODO: can we get away with significantly increasing this?
         let target_qps = get_var("SIM_STRESS_TEST_QPS", 10);
         let num_workers = get_var("SIM_STRESS_TEST_WORKERS", 10);
 
@@ -662,7 +662,7 @@ mod test {
     async fn test_data_ingestion_pipeline() {
         let path = nondeterministic!(TempDir::new().unwrap()).into_path();
         let test_cluster = Arc::new(
-            init_test_cluster_builder(4, 1000)
+            init_test_cluster_builder(4, 5000)
                 .with_data_ingestion_dir(path.clone())
                 .build()
                 .await,

crates/sui-core/src/authority/authority_test_utils.rs

@@ -218,7 +218,11 @@ pub async fn init_state_with_ids_and_versions<
 ) -> Arc<AuthorityState> {
     let state = TestAuthorityBuilder::new().build().await;
     for (address, object_id, version) in objects {
-        let obj = Object::with_id_owner_version_for_testing(object_id, version, address);
+        let obj = Object::with_id_owner_version_for_testing(
+            object_id,
+            version,
+            Owner::AddressOwner(address),
+        );
         state.insert_genesis_object(obj).await;
     }
     state

crates/sui-core/src/execution_cache/unit_tests/writeback_cache_tests.rs

@@ -1244,7 +1244,11 @@ async fn latest_object_cache_race_test() {
         std::thread::spawn(move || {
             let mut version = OBJECT_START_VERSION;
             while start.elapsed() < Duration::from_secs(2) {
-                let object = Object::with_id_owner_version_for_testing(object_id, version, owner);
+                let object = Object::with_id_owner_version_for_testing(
+                    object_id,
+                    version,
+                    Owner::AddressOwner(owner),
+                );
 
                 cache
                     .write_object_entry(&object_id, version, object.into())
@@ -1284,8 +1288,11 @@ async fn latest_object_cache_race_test() {
                     std::thread::sleep(Duration::from_micros(1));
                 }
 
-                let object =
-                    Object::with_id_owner_version_for_testing(object_id, latest_version, owner);
+                let object = Object::with_id_owner_version_for_testing(
+                    object_id,
+                    latest_version,
+                    Owner::AddressOwner(owner),
+                );
 
                 // because we obtained the ticket before reading the object, we will not write a stale
                 // version to the cache.

crates/sui-core/src/unit_tests/authority_tests.rs

@@ -1094,7 +1094,7 @@ async fn test_dry_run_dev_inspect_max_gas_version() {
     let gas_object = Object::with_id_owner_version_for_testing(
         gas_object_id,
         SequenceNumber::from_u64(SequenceNumber::MAX.value() - 1),
-        sender,
+        Owner::AddressOwner(sender),
     );
     let gas_object_ref = gas_object.compute_object_reference();
     validator.insert_genesis_object(gas_object.clone()).await;
@@ -5950,8 +5950,16 @@ async fn test_consensus_handler_congestion_control_transaction_cancellation() {
     let gas_objects = create_gas_objects(3, sender);
     let gas_objects_cancelled_txn = create_gas_objects(1, sender);
     let owned_objects_cancelled_txn = vec![
-        Object::with_id_owner_version_for_testing(ObjectID::random(), 1.into(), sender),
-        Object::with_id_owner_version_for_testing(ObjectID::random(), 2.into(), sender),
+        Object::with_id_owner_version_for_testing(
+            ObjectID::random(),
+            1.into(),
+            Owner::AddressOwner(sender),
+        ),
+        Object::with_id_owner_version_for_testing(
+            ObjectID::random(),
+            2.into(),
+            Owner::AddressOwner(sender),
+        ),
     ];
 
     // Create the cluster with controlled per object congestion control and cancellation.

crates/sui-core/src/unit_tests/transaction_manager_tests.rs

@@ -5,6 +5,7 @@ use std::{time::Duration, vec};
 
 use sui_test_transaction_builder::TestTransactionBuilder;
 use sui_types::executable_transaction::VerifiedExecutableTransaction;
+use sui_types::object::Owner;
 use sui_types::transaction::VerifiedTransaction;
 use sui_types::{
     base_types::{ObjectID, SequenceNumber},
@@ -120,8 +121,11 @@ async fn transaction_manager_basics() {
     transaction_manager.check_empty_for_testing();
 
     // Enqueue a transaction with a new gas object, empty input.
-    let gas_object_new =
-        Object::with_id_owner_version_for_testing(ObjectID::random(), 0.into(), owner);
+    let gas_object_new = Object::with_id_owner_version_for_testing(
+        ObjectID::random(),
+        0.into(),
+        Owner::AddressOwner(owner),
+    );
     let transaction = make_transaction(gas_object_new.clone(), vec![]);
     let tx_start_time = Instant::now();
     transaction_manager.enqueue(vec![transaction.clone()], &state.epoch_store_for_testing());
@@ -386,7 +390,11 @@ async fn transaction_manager_receiving_notify_commit() {
     let obj_id = ObjectID::random();
     let object_arguments: Vec<_> = (0..10)
         .map(|i| {
-            let object = Object::with_id_owner_version_for_testing(obj_id, i.into(), owner);
+            let object = Object::with_id_owner_version_for_testing(
+                obj_id,
+                i.into(),
+                Owner::AddressOwner(owner),
+            );
             // Every other transaction receives the object, and we create a run of multiple receives in
             // a row at the beginning to test that the TM doesn't get stuck in either configuration of:
             // ImmOrOwnedObject => Receiving,
@@ -475,8 +483,10 @@ async fn transaction_manager_receiving_object_ready_notifications() {
     transaction_manager.check_empty_for_testing();
 
     let obj_id = ObjectID::random();
-    let receiving_object_new0 = Object::with_id_owner_version_for_testing(obj_id, 0.into(), owner);
-    let receiving_object_new1 = Object::with_id_owner_version_for_testing(obj_id, 1.into(), owner);
+    let receiving_object_new0 =
+        Object::with_id_owner_version_for_testing(obj_id, 0.into(), Owner::AddressOwner(owner));
+    let receiving_object_new1 =
+        Object::with_id_owner_version_for_testing(obj_id, 1.into(), Owner::AddressOwner(owner));
     let receiving_object_arg0 =
         ObjectArg::Receiving(receiving_object_new0.compute_object_reference());
     let receive_object_transaction0 = make_transaction(
@@ -561,8 +571,10 @@ async fn transaction_manager_receiving_object_ready_notifications_multiple_of_sa
     transaction_manager.check_empty_for_testing();
 
     let obj_id = ObjectID::random();
-    let receiving_object_new0 = Object::with_id_owner_version_for_testing(obj_id, 0.into(), owner);
-    let receiving_object_new1 = Object::with_id_owner_version_for_testing(obj_id, 1.into(), owner);
+    let receiving_object_new0 =
+        Object::with_id_owner_version_for_testing(obj_id, 0.into(), Owner::AddressOwner(owner));
+    let receiving_object_new1 =
+        Object::with_id_owner_version_for_testing(obj_id, 1.into(), Owner::AddressOwner(owner));
     let receiving_object_arg0 =
         ObjectArg::Receiving(receiving_object_new0.compute_object_reference());
     let receive_object_transaction0 = make_transaction(
@@ -661,8 +673,11 @@ async fn transaction_manager_receiving_object_ready_if_current_version_greater()
             Object::with_id_owner_for_testing(gas_object_id, owner)
         })
         .collect();
-    let receiving_object =
-        Object::with_id_owner_version_for_testing(ObjectID::random(), 10.into(), owner);
+    let receiving_object = Object::with_id_owner_version_for_testing(
+        ObjectID::random(),
+        10.into(),
+        Owner::AddressOwner(owner),
+    );
     gas_objects.push(receiving_object.clone());
     let state = init_state_with_objects(gas_objects.clone()).await;
 
@@ -674,10 +689,16 @@ async fn transaction_manager_receiving_object_ready_if_current_version_greater()
     // TM should be empty at the beginning.
     transaction_manager.check_empty_for_testing();
 
-    let receiving_object_new0 =
-        Object::with_id_owner_version_for_testing(receiving_object.id(), 0.into(), owner);
-    let receiving_object_new1 =
-        Object::with_id_owner_version_for_testing(receiving_object.id(), 1.into(), owner);
+    let receiving_object_new0 = Object::with_id_owner_version_for_testing(
+        receiving_object.id(),
+        0.into(),
+        Owner::AddressOwner(owner),
+    );
+    let receiving_object_new1 = Object::with_id_owner_version_for_testing(
+        receiving_object.id(),
+        1.into(),
+        Owner::AddressOwner(owner),
+    );
     let receiving_object_arg0 =
         ObjectArg::Receiving(receiving_object_new0.compute_object_reference());
     let receive_object_transaction0 = make_transaction(

crates/sui-types/src/effects/test_effects_builder.rs

@@ -19,6 +19,15 @@ pub struct TestEffectsBuilder {
     /// Provide the assigned versions for all shared objects.
     shared_input_versions: BTreeMap<ObjectID, SequenceNumber>,
     events_digest: Option<TransactionEventsDigest>,
+    created_objects: Vec<(ObjectID, Owner)>,
+    /// Objects that are mutated: (ID, old version, new owner).
+    mutated_objects: Vec<(ObjectID, SequenceNumber, Owner)>,
+    /// Objects that are deleted: (ID, old version).
+    deleted_objects: Vec<(ObjectID, SequenceNumber)>,
+    /// Objects that are wrapped: (ID, old version).
+    wrapped_objects: Vec<(ObjectID, SequenceNumber)>,
+    /// Objects that are unwrapped: (ID, new owner).
+    unwrapped_objects: Vec<(ObjectID, Owner)>,
 }
 
 impl TestEffectsBuilder {
@@ -28,6 +37,11 @@ impl TestEffectsBuilder {
             status: None,
             shared_input_versions: BTreeMap::new(),
             events_digest: None,
+            created_objects: vec![],
+            mutated_objects: vec![],
+            deleted_objects: vec![],
+            wrapped_objects: vec![],
+            unwrapped_objects: vec![],
         }
     }
 
@@ -50,7 +64,49 @@ impl TestEffectsBuilder {
         self
     }
 
+    pub fn with_created_objects(
+        mut self,
+        objects: impl IntoIterator<Item = (ObjectID, Owner)>,
+    ) -> Self {
+        self.created_objects.extend(objects);
+        self
+    }
+
+    pub fn with_mutated_objects(
+        mut self,
+        // Object ID, old version, and new owner.
+        objects: impl IntoIterator<Item = (ObjectID, SequenceNumber, Owner)>,
+    ) -> Self {
+        self.mutated_objects.extend(objects);
+        self
+    }
+
+    pub fn with_wrapped_objects(
+        mut self,
+        objects: impl IntoIterator<Item = (ObjectID, SequenceNumber)>,
+    ) -> Self {
+        self.wrapped_objects.extend(objects);
+        self
+    }
+
+    pub fn with_unwrapped_objects(
+        mut self,
+        objects: impl IntoIterator<Item = (ObjectID, Owner)>,
+    ) -> Self {
+        self.unwrapped_objects.extend(objects);
+        self
+    }
+
+    pub fn with_deleted_objects(
+        mut self,
+        objects: impl IntoIterator<Item = (ObjectID, SequenceNumber)>,
+    ) -> Self {
+        self.deleted_objects.extend(objects);
+        self
+    }
+
     pub fn build(self) -> TransactionEffects {
+        let lamport_version = self.get_lamport_version();
         let status = self.status.unwrap_or_else(|| ExecutionStatus::Success);
         // TODO: This does not yet support deleted shared objects.
         let shared_objects = self
@@ -59,22 +115,6 @@ impl TestEffectsBuilder {
             .map(|(id, version)| SharedInput::Existing((*id, *version, ObjectDigest::MIN)))
             .collect();
         let executed_epoch = 0;
-        let lamport_version = SequenceNumber::lamport_increment(
-            self.transaction
-                .transaction_data()
-                .input_objects()
-                .unwrap()
-                .iter()
-                .filter_map(|kind| kind.version())
-                .chain(
-                    self.transaction
-                        .transaction_data()
-                        .receiving_objects()
-                        .iter()
-                        .map(|oref| oref.1),
-                )
-                .chain(self.shared_input_versions.values().copied()),
-        );
         let sender = self.transaction.transaction_data().sender();
         // TODO: Include receiving objects in the object changes as well.
         let changed_objects = self
@@ -130,6 +170,72 @@ impl TestEffectsBuilder {
                     },
                 )),
             })
+            .chain(self.created_objects.into_iter().map(|(id, owner)| {
+                (
+                    id,
+                    EffectsObjectChange {
+                        input_state: ObjectIn::NotExist,
+                        output_state: ObjectOut::ObjectWrite((ObjectDigest::random(), owner)),
+                        id_operation: IDOperation::Created,
+                    },
+                )
+            }))
+            .chain(
+                self.mutated_objects
+                    .into_iter()
+                    .map(|(id, version, owner)| {
+                        (
+                            id,
+                            EffectsObjectChange {
+                                input_state: ObjectIn::Exist((
+                                    (version, ObjectDigest::random()),
+                                    Owner::AddressOwner(sender),
+                                )),
+                                output_state: ObjectOut::ObjectWrite((
+                                    ObjectDigest::random(),
+                                    owner,
+                                )),
+                                id_operation: IDOperation::None,
+                            },
+                        )
+                    }),
+            )
+            .chain(self.deleted_objects.into_iter().map(|(id, version)| {
+                (
+                    id,
+                    EffectsObjectChange {
+                        input_state: ObjectIn::Exist((
+                            (version, ObjectDigest::random()),
+                            Owner::AddressOwner(sender),
+                        )),
+                        output_state: ObjectOut::NotExist,
+                        id_operation: IDOperation::Deleted,
+                    },
+                )
+            }))
+            .chain(self.wrapped_objects.into_iter().map(|(id, version)| {
+                (
+                    id,
+                    EffectsObjectChange {
+                        input_state: ObjectIn::Exist((
+                            (version, ObjectDigest::random()),
+                            Owner::AddressOwner(sender),
+                        )),
+                        output_state: ObjectOut::NotExist,
+                        id_operation: IDOperation::None,
+                    },
+                )
+            }))
+            .chain(self.unwrapped_objects.into_iter().map(|(id, owner)| {
+                (
+                    id,
+                    EffectsObjectChange {
+                        input_state: ObjectIn::NotExist,
+                        output_state: ObjectOut::ObjectWrite((ObjectDigest::random(), owner)),
+                        id_operation: IDOperation::None,
+                    },
+                )
+            }))
             .collect();
         let gas_object_id = self.transaction.transaction_data().gas()[0].0;
         let event_digest = self.events_digest;
@@ -148,4 +254,26 @@ impl TestEffectsBuilder {
             dependencies,
         )
     }
+
+    fn get_lamport_version(&self) -> SequenceNumber {
+        SequenceNumber::lamport_increment(
+            self.transaction
+                .transaction_data()
+                .input_objects()
+                .unwrap()
+                .iter()
+                .filter_map(|kind| kind.version())
+                .chain(
+                    self.transaction
+                        .transaction_data()
+                        .receiving_objects()
+                        .iter()
+                        .map(|oref| oref.1),
+                )
+                .chain(self.shared_input_versions.values().copied())
+                .chain(self.mutated_objects.iter().map(|(_, v, _)| *v))
+                .chain(self.deleted_objects.iter().map(|(_, v)| *v))
+                .chain(self.wrapped_objects.iter().map(|(_, v)| *v)),
+        )
+    }
 }

crates/sui-types/src/lib.rs

@@ -80,6 +80,7 @@ pub mod sui_sdk_types_conversions;
 pub mod sui_serde;
 pub mod sui_system_state;
 pub mod supported_protocol_versions;
+pub mod test_checkpoint_data_builder;
 pub mod traffic_control;
 pub mod transaction;
 pub mod transaction_executor;