[fix] Adding delta ops in TransactionOutput and ChangeSet (aptos-labs…

…#2222) Fixes aptos-labs#2057. Now deltas do not escape from executor, and are stored in `TransactionOutputExt` and `ChangeSetExt` wrappers.
bowenyang007 · Jul 29, 2022 · c2b4ff3 · c2b4ff3
1 parent 8cc3edc
commit c2b4ff3
Show file tree

Hide file tree

Showing 22 changed files with 284 additions and 109 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/api/src/transactions.rs b/api/src/transactions.rs
@@ -349,6 +349,10 @@ impl Transactions {
             TransactionStatus::Keep(exec_status) => exec_status,
             _ => ExecutionStatus::MiscellaneousError(None),
         };
+
+        // TODO: Here we need to materialize deltas.
+        let (_, output) = output.into();
+
         let zero_hash = HashValue::zero();
         let info = TransactionInfo::new(
             zero_hash,

diff --git a/api/types/src/convert.rs b/api/types/src/convert.rs
@@ -274,8 +274,6 @@ impl<'a, R: MoveResolverExt + ?Sized> MoveConverter<'a, R> {
                     data: self.try_into_resource(&typ, &val)?,
                 }),
             },
-            // Deltas never use access paths.
-            WriteOp::Delta(..) => unreachable!("unexpected conversion"),
         };
         Ok(ret)
     }
@@ -312,8 +310,6 @@ impl<'a, R: MoveResolverExt + ?Sized> MoveConverter<'a, R> {
                     data,
                 })
             }
-            // Deltas are materialized into WriteOP::Value(..) in executor.
-            WriteOp::Delta(..) => unreachable!("unexpected conversion"),
         };
         Ok(ret)
     }

diff --git a/aptos-move/aptos-vm/src/adapter_common.rs b/aptos-move/aptos-vm/src/adapter_common.rs
@@ -5,7 +5,9 @@ use crate::{counters::*, data_cache::StateViewCache};
 use anyhow::Result;
 use aptos_state_view::StateView;
 use aptos_types::{
-    transaction::{SignatureCheckedTransaction, SignedTransaction, VMValidatorResult},
+    transaction::{
+        SignatureCheckedTransaction, SignedTransaction, TransactionOutputExt, VMValidatorResult,
+    },
     vm_status::{StatusCode, VMStatus},
 };
 
@@ -64,7 +66,7 @@ pub trait VMAdapter {
         txn: &PreprocessedTransaction,
         data_cache: &S,
         log_context: &AdapterLogSchema,
-    ) -> Result<(VMStatus, TransactionOutput, Option<String>), VMStatus>;
+    ) -> Result<(VMStatus, TransactionOutputExt, Option<String>), VMStatus>;
 }
 
 /// Validate a signed transaction by performing the following:
@@ -204,6 +206,8 @@ pub(crate) fn execute_block_impl<A: VMAdapter, S: StateView>(
             &data_cache.as_move_resolver(),
             &log_context,
         )?;
+        // TODO: apply deltas.
+        let (_, output) = output.into();
         if !output.status().is_discarded() {
             data_cache.push_write_set(output.write_set());
         } else {
@@ -273,7 +277,7 @@ pub(crate) fn preprocess_transaction<A: VMAdapter>(txn: Transaction) -> Preproce
     }
 }
 
-pub(crate) fn discard_error_vm_status(err: VMStatus) -> (VMStatus, TransactionOutput) {
+pub(crate) fn discard_error_vm_status(err: VMStatus) -> (VMStatus, TransactionOutputExt) {
     let vm_status = err.clone();
     let error_code = match err.keep_or_discard() {
         Ok(_) => {
@@ -285,12 +289,12 @@ pub(crate) fn discard_error_vm_status(err: VMStatus) -> (VMStatus, TransactionOu
     (vm_status, discard_error_output(error_code))
 }
 
-pub(crate) fn discard_error_output(err: StatusCode) -> TransactionOutput {
+pub(crate) fn discard_error_output(err: StatusCode) -> TransactionOutputExt {
     // Since this transaction will be discarded, no writeset will be included.
-    TransactionOutput::new(
+    TransactionOutputExt::from(TransactionOutput::new(
         WriteSet::default(),
         vec![],
         0,
         TransactionStatus::Discard(err),
-    )
+    ))
 }
diff --git a/aptos-move/aptos-vm/src/aptos_vm.rs b/aptos-move/aptos-vm/src/aptos_vm.rs
@@ -32,8 +32,8 @@ use aptos_types::{
     on_chain_config::{new_epoch_event_key, VMConfig, VMPublishingOption, Version},
     transaction::{
         ChangeSet, ExecutionStatus, ModuleBundle, SignatureCheckedTransaction, SignedTransaction,
-        Transaction, TransactionOutput, TransactionPayload, TransactionStatus, VMValidatorResult,
-        WriteSetPayload,
+        Transaction, TransactionOutput, TransactionOutputExt, TransactionPayload,
+        TransactionStatus, VMValidatorResult, WriteSetPayload,
     },
     vm_status::{StatusCode, VMStatus},
     write_set::{WriteSet, WriteSetMut},
@@ -154,7 +154,7 @@ impl AptosVM {
         txn_data: &TransactionMetadata,
         storage: &S,
         log_context: &AdapterLogSchema,
-    ) -> TransactionOutput {
+    ) -> TransactionOutputExt {
         self.failed_transaction_cleanup_and_keep_vm_status(
             error_code,
             gas_status,
@@ -172,7 +172,7 @@ impl AptosVM {
         txn_data: &TransactionMetadata,
         storage: &S,
         log_context: &AdapterLogSchema,
-    ) -> (VMStatus, TransactionOutput) {
+    ) -> (VMStatus, TransactionOutputExt) {
         gas_status.set_metering(false);
         let mut session = self.0.new_session(storage, SessionId::txn_meta(txn_data));
         match TransactionStatus::from(error_code.clone()) {
@@ -212,7 +212,7 @@ impl AptosVM {
         gas_status: &mut GasStatus,
         txn_data: &TransactionMetadata,
         log_context: &AdapterLogSchema,
-    ) -> Result<(VMStatus, TransactionOutput), VMStatus> {
+    ) -> Result<(VMStatus, TransactionOutputExt), VMStatus> {
         gas_status.set_metering(false);
         self.0
             .run_success_epilogue(&mut session, gas_status, txn_data, log_context)?;
@@ -236,7 +236,7 @@ impl AptosVM {
         txn_data: &TransactionMetadata,
         payload: &TransactionPayload,
         log_context: &AdapterLogSchema,
-    ) -> Result<(VMStatus, TransactionOutput), VMStatus> {
+    ) -> Result<(VMStatus, TransactionOutputExt), VMStatus> {
         fail_point!("move_adapter::execute_script_or_script_function", |_| {
             Err(VMStatus::Error(
                 StatusCode::UNKNOWN_INVARIANT_VIOLATION_ERROR,
@@ -395,7 +395,7 @@ impl AptosVM {
         txn_data: &TransactionMetadata,
         modules: &ModuleBundle,
         log_context: &AdapterLogSchema,
-    ) -> Result<(VMStatus, TransactionOutput), VMStatus> {
+    ) -> Result<(VMStatus, TransactionOutputExt), VMStatus> {
         fail_point!("move_adapter::execute_module", |_| {
             Err(VMStatus::Error(
                 StatusCode::UNKNOWN_INVARIANT_VIOLATION_ERROR,
@@ -495,7 +495,7 @@ impl AptosVM {
         storage: &S,
         txn: &SignatureCheckedTransaction,
         log_context: &AdapterLogSchema,
-    ) -> (VMStatus, TransactionOutput) {
+    ) -> (VMStatus, TransactionOutputExt) {
         macro_rules! unwrap_or_discard {
             ($res: expr) => {
                 match $res {
@@ -570,7 +570,7 @@ impl AptosVM {
         writeset_payload: &WriteSetPayload,
         txn_sender: Option<AccountAddress>,
         session_id: SessionId,
-    ) -> Result<ChangeSet, Result<(VMStatus, TransactionOutput), VMStatus>> {
+    ) -> Result<ChangeSet, Result<(VMStatus, TransactionOutputExt), VMStatus>> {
         let mut gas_status = GasStatus::new_unmetered();
 
         Ok(match writeset_payload {
@@ -656,7 +656,7 @@ impl AptosVM {
         storage: &S,
         writeset_payload: WriteSetPayload,
         log_context: &AdapterLogSchema,
-    ) -> Result<(VMStatus, TransactionOutput), VMStatus> {
+    ) -> Result<(VMStatus, TransactionOutputExt), VMStatus> {
         // TODO: user specified genesis id to distinguish different genesis write sets
         let genesis_id = HashValue::zero();
         let change_set = match self.execute_writeset(
@@ -674,7 +674,12 @@ impl AptosVM {
         SYSTEM_TRANSACTIONS_EXECUTED.inc();
         Ok((
             VMStatus::Executed,
-            TransactionOutput::new(write_set, events, 0, VMStatus::Executed.into()),
+            TransactionOutputExt::from(TransactionOutput::new(
+                write_set,
+                events,
+                0,
+                VMStatus::Executed.into(),
+            )),
         ))
     }
 
@@ -683,7 +688,7 @@ impl AptosVM {
         storage: &S,
         block_metadata: BlockMetadata,
         log_context: &AdapterLogSchema,
-    ) -> Result<(VMStatus, TransactionOutput), VMStatus> {
+    ) -> Result<(VMStatus, TransactionOutputExt), VMStatus> {
         fail_point!("move_adapter::process_block_prologue", |_| {
             Err(VMStatus::Error(
                 StatusCode::UNKNOWN_INVARIANT_VIOLATION_ERROR,
@@ -761,7 +766,7 @@ impl AptosVM {
         storage: &S,
         txn: &SignatureCheckedTransaction,
         log_context: &AdapterLogSchema,
-    ) -> Result<(VMStatus, TransactionOutput), VMStatus> {
+    ) -> Result<(VMStatus, TransactionOutputExt), VMStatus> {
         fail_point!("move_adapter::process_writeset_transaction", |_| {
             Err(VMStatus::Error(
                 StatusCode::UNKNOWN_INVARIANT_VIOLATION_ERROR,
@@ -804,7 +809,7 @@ impl AptosVM {
         writeset_payload: &WriteSetPayload,
         txn_data: TransactionMetadata,
         log_context: &AdapterLogSchema,
-    ) -> Result<(VMStatus, TransactionOutput), VMStatus> {
+    ) -> Result<(VMStatus, TransactionOutputExt), VMStatus> {
         let change_set = match self.execute_writeset(
             storage,
             writeset_payload,
@@ -884,12 +889,12 @@ impl AptosVM {
 
         Ok((
             VMStatus::Executed,
-            TransactionOutput::new(
+            TransactionOutputExt::from(TransactionOutput::new(
                 write_set,
                 events,
                 0,
                 TransactionStatus::Keep(ExecutionStatus::Success),
-            ),
+            )),
         ))
     }
 
@@ -911,7 +916,7 @@ impl AptosVM {
     pub fn simulate_signed_transaction(
         txn: &SignedTransaction,
         state_view: &impl StateView,
-    ) -> (VMStatus, TransactionOutput) {
+    ) -> (VMStatus, TransactionOutputExt) {
         let vm = AptosVM::new(state_view);
         let simulation_vm = AptosSimulationVM(vm);
         let log_context = AdapterLogSchema::new(state_view.id(), 0);
@@ -1048,7 +1053,7 @@ impl VMAdapter for AptosVM {
         txn: &PreprocessedTransaction,
         data_cache: &S,
         log_context: &AdapterLogSchema,
-    ) -> Result<(VMStatus, TransactionOutput, Option<String>), VMStatus> {
+    ) -> Result<(VMStatus, TransactionOutputExt, Option<String>), VMStatus> {
         Ok(match txn {
             PreprocessedTransaction::BlockMetadata(block_metadata) => {
                 let (vm_status, output) =
@@ -1097,7 +1102,11 @@ impl VMAdapter for AptosVM {
                     0,
                     TransactionStatus::Keep(ExecutionStatus::Success),
                 );
-                (VMStatus::Executed, output, Some("state_checkpoint".into()))
+                (
+                    VMStatus::Executed,
+                    TransactionOutputExt::from(output),
+                    Some("state_checkpoint".into()),
+                )
             }
         })
     }
@@ -1136,7 +1145,7 @@ impl AptosSimulationVM {
         storage: &S,
         txn: &SignedTransaction,
         log_context: &AdapterLogSchema,
-    ) -> (VMStatus, TransactionOutput) {
+    ) -> (VMStatus, TransactionOutputExt) {
         // simulation transactions should not carry valid signatures, otherwise malicious fullnodes
         // may execute them without user's explicit permission.
         if txn.clone().check_signature().is_ok() {
@@ -1185,13 +1194,14 @@ impl AptosSimulationVM {
                 if txn_status.is_discarded() {
                     discard_error_vm_status(err)
                 } else {
-                    self.0.failed_transaction_cleanup_and_keep_vm_status(
+                    let (vm_status, output) = self.0.failed_transaction_cleanup_and_keep_vm_status(
                         err,
                         &mut gas_status,
                         &txn_data,
                         storage,
                         log_context,
-                    )
+                    );
+                    (vm_status, output)
                 }
             }
         }

diff --git a/aptos-move/aptos-vm/src/aptos_vm_impl.rs b/aptos-move/aptos-vm/src/aptos_vm_impl.rs
@@ -18,7 +18,7 @@ use aptos_types::{
     on_chain_config::{
         ConfigStorage, OnChainConfig, VMConfig, VMPublishingOption, Version, APTOS_VERSION_3,
     },
-    transaction::{ExecutionStatus, TransactionOutput, TransactionStatus},
+    transaction::{ExecutionStatus, TransactionOutput, TransactionOutputExt, TransactionStatus},
     vm_status::{StatusCode, VMStatus},
 };
 use fail::fail_point;
@@ -569,18 +569,17 @@ pub(crate) fn get_transaction_output<A: AccessPathCache, S: MoveResolverExt>(
     gas_left: GasUnits<GasCarrier>,
     txn_data: &TransactionMetadata,
     status: ExecutionStatus,
-) -> Result<TransactionOutput, VMStatus> {
+) -> Result<TransactionOutputExt, VMStatus> {
     let gas_used: u64 = txn_data.max_gas_amount().sub(gas_left).get();
 
     let session_out = session.finish().map_err(|e| e.into_vm_status())?;
-    let (write_set, events) = session_out.into_change_set(ap_cache)?.into_inner();
-
-    Ok(TransactionOutput::new(
-        write_set,
-        events,
-        gas_used,
-        TransactionStatus::Keep(status),
-    ))
+    let (delta_change_set, change_set) = session_out.into_change_set_ext(ap_cache)?.into_inner();
+    let (write_set, events) = change_set.into_inner();
+
+    let txn_output =
+        TransactionOutput::new(write_set, events, gas_used, TransactionStatus::Keep(status));
+
+    Ok(TransactionOutputExt::new(delta_change_set, txn_output))
 }
 
 #[test]

diff --git a/aptos-move/aptos-vm/src/data_cache.rs b/aptos-move/aptos-vm/src/data_cache.rs
@@ -72,9 +72,6 @@ impl<'a, S: StateView> StateViewCache<'a, S> {
                     self.data_map.remove(ap);
                     self.data_map.insert(ap.clone(), None);
                 }
-                WriteOp::Delta(..) => {
-                    unimplemented!("sequential execution is not supported for deltas")
-                }
             }
         }
     }

diff --git a/aptos-move/aptos-vm/src/move_vm_ext/session.rs b/aptos-move/aptos-vm/src/move_vm_ext/session.rs
@@ -10,8 +10,9 @@ use aptos_crypto_derive::{BCSCryptoHash, CryptoHasher};
 use aptos_types::{
     block_metadata::BlockMetadata,
     contract_event::ContractEvent,
+    delta_change_set::DeltaChangeSet,
     state_store::state_key::StateKey,
-    transaction::{ChangeSet, SignatureCheckedTransaction},
+    transaction::{ChangeSet, ChangeSetExt, SignatureCheckedTransaction},
     write_set::{WriteOp, WriteSetMut},
 };
 use move_deps::{
@@ -194,6 +195,16 @@ impl SessionOutput {
         Ok(ChangeSet::new(write_set, events))
     }
 
+    pub fn into_change_set_ext<C: AccessPathCache>(
+        self,
+        ap_cache: &mut C,
+    ) -> Result<ChangeSetExt, VMStatus> {
+        // TODO: extract `DeltaChangeSet` from Aggregator extension (when it lands)
+        // and initialize `ChangeSetExt` properly.
+        self.into_change_set(ap_cache)
+            .map(|change_set| ChangeSetExt::new(DeltaChangeSet::empty(), change_set))
+    }
+
     pub fn squash(&mut self, other: Self) -> Result<(), VMStatus> {
         self.change_set
             .squash(other.change_set)

diff --git a/aptos-move/aptos-vm/src/parallel_executor/storage_wrapper.rs b/aptos-move/aptos-vm/src/parallel_executor/storage_wrapper.rs
@@ -35,9 +35,6 @@ impl<'a, S: StateView> StateView for VersionedView<'a, S> {
             Some(v) => Ok(match v.as_ref() {
                 WriteOp::Value(w) => Some(w.clone()),
                 WriteOp::Deletion => None,
-                WriteOp::Delta(..) => {
-                    unimplemented!("parallel execution is not supported for deltas")
-                }
             }),
             None => self.base_view.get_state_value(state_key),
         }

diff --git a/aptos-move/aptos-vm/src/parallel_executor/vm_wrapper.rs b/aptos-move/aptos-vm/src/parallel_executor/vm_wrapper.rs
@@ -67,6 +67,9 @@ impl<'a, S: 'a + StateView> ExecutorTask for AptosVMWrapper<'a, S> {
             .execute_single_transaction(txn, &versioned_view, &log_context)
         {
             Ok((vm_status, output, sender)) => {
+                // TODO: pass deltas to `AptosTransactionOutput` once we support parallel execution.
+                let (_, output) = output.into();
+
                 if output.status().is_discarded() {
                     match sender {
                         Some(s) => trace!(

diff --git a/aptos-move/e2e-tests/src/data_store.rs b/aptos-move/e2e-tests/src/data_store.rs
@@ -50,7 +50,6 @@ impl FakeDataStore {
                 WriteOp::Deletion => {
                     self.remove(state_key);
                 }
-                WriteOp::Delta(..) => unreachable!("deltas are only used in executor"),
             }
         }
     }
-Original file line number
+Diff line change
@@ Expand Up / @@ -50,7 +50,6 @@ impl FakeDataStore { @@
                     WriteOp::Deletion => {
                         self.remove(state_key);
                     }
-                    WriteOp::Delta(..) => unreachable!("deltas are only used in executor"),
                 }
             }
         }
@@ Expand Down @@