cleaning up aggregators part 2 (aptos-labs#11939)

aptos-move/aptos-aggregator/src/types.rs

@@ -39,7 +39,7 @@ impl<T: std::fmt::Debug> PanicOr<T> {
 
 pub fn code_invariant_error<M: std::fmt::Debug>(message: M) -> PanicError {
  let msg = format!(
- "Delayed field / resource group code invariant broken (a bug in the code), {:?}",
+ "Delayed materialization code invariant broken (there is a bug in the code), {:?}",
  message
  );
  error!("{}", msg);
@@ -75,7 +75,9 @@ impl From<DelayedFieldsSpeculativeError> for PartialVMError {
 impl<T: std::fmt::Debug> From<&PanicOr<T>> for StatusCode {
  fn from(err: &PanicOr<T>) -> Self {
  match err {
- PanicOr::CodeInvariantError(_) => StatusCode::DELAYED_FIELDS_CODE_INVARIANT_ERROR,
+ PanicOr::CodeInvariantError(_) => {
+ StatusCode::DELAYED_MATERIALIZATION_CODE_INVARIANT_ERROR
+ },
  PanicOr::Or(_) => StatusCode::SPECULATIVE_EXECUTION_ABORT_ERROR,
  }
  }
@@ -85,7 +87,7 @@ impl<T: std::fmt::Debug> From<PanicOr<T>> for PartialVMError {
  fn from(err: PanicOr<T>) -> Self {
  match err {
  PanicOr::CodeInvariantError(msg) => {
- PartialVMError::new(StatusCode::DELAYED_FIELDS_CODE_INVARIANT_ERROR)
+ PartialVMError::new(StatusCode::DELAYED_MATERIALIZATION_CODE_INVARIANT_ERROR)
  .with_message(msg)
  },
  PanicOr::Or(err) => PartialVMError::new(StatusCode::SPECULATIVE_EXECUTION_ABORT_ERROR)

aptos-move/aptos-vm-types/src/change_set.rs

@@ -179,7 +179,7 @@ impl VMChangeSet {
  .write_len()
  .ok_or_else(|| {
  PartialVMError::new(
- StatusCode::DELAYED_FIELDS_CODE_INVARIANT_ERROR,
+ StatusCode::DELAYED_MATERIALIZATION_CODE_INVARIANT_ERROR,
  )
  .with_message(
  "Read with exchange cannot be a delete."
@@ -211,7 +211,7 @@ impl VMChangeSet {
  let (key, value) = element?;
  if acc.insert(key, value).is_some() {
  Err(PartialVMError::new(
- StatusCode::DELAYED_FIELDS_CODE_INVARIANT_ERROR,
+ StatusCode::DELAYED_MATERIALIZATION_CODE_INVARIANT_ERROR,
  )
  .with_message(
  "Found duplicate key across resource change sets.".to_string(),

aptos-move/aptos-vm-types/src/output.rs

@@ -108,7 +108,7 @@ impl VMOutput {
  self.try_materialize(resolver)?;
  Self::convert_to_transaction_output(self).map_err(|e| {
  VMStatus::error(
- StatusCode::DELAYED_FIELDS_CODE_INVARIANT_ERROR,
+ StatusCode::DELAYED_MATERIALIZATION_CODE_INVARIANT_ERROR,
  Some(e.to_string()),
  )
  })
@@ -120,7 +120,7 @@ impl VMOutput {
  let output = VMOutput::new(change_set, fee_statement, status);
  Self::convert_to_transaction_output(output).map_err(|e| {
  VMStatus::error(
- StatusCode::DELAYED_FIELDS_CODE_INVARIANT_ERROR,
+ StatusCode::DELAYED_MATERIALIZATION_CODE_INVARIANT_ERROR,
  Some(e.to_string()),
  )
  })

aptos-move/aptos-vm-types/src/resource_group_adapter.rs

@@ -124,10 +124,11 @@ impl<'r> ResourceGroupAdapter<'r> {
  gas_feature_version: u64,
  resource_group_charge_as_size_sum_enabled: bool,
  ) -> Self {
- // TODO[agg_v2](fix) - when is_resource_group_split_in_change_set_capable is false,
+ // when is_resource_group_split_in_change_set_capable is false,
  // but resource_group_charge_as_size_sum_enabled is true, we still don't set
  // group_size_kind to GroupSizeKind::AsSum, meaning that
- // is_resource_group_split_in_change_set_capable affects gas charging. make sure that is correct
+ // is_resource_group_split_in_change_set_capable affects gas charging.
+ // Onchain execution always needs to go through capable resolvers.
 
  let group_size_kind = GroupSizeKind::from_gas_feature_version(
  gas_feature_version,

aptos-move/aptos-vm-types/src/tests/test_change_set.rs

@@ -253,7 +253,8 @@ macro_rules! assert_invariant_violation {
  // TODO[agg_v2](test): Uniformize errors for write op squashing.
  assert!(
  err.major_status() == StatusCode::UNKNOWN_INVARIANT_VIOLATION_ERROR
- || err.major_status() == StatusCode::DELAYED_FIELDS_CODE_INVARIANT_ERROR
+ || err.major_status()
+ == StatusCode::DELAYED_MATERIALIZATION_CODE_INVARIANT_ERROR
  );
  };
 

aptos-move/aptos-vm/src/block_executor/mod.rs

@@ -21,6 +21,7 @@ use aptos_infallible::Mutex;
 use aptos_types::{
  block_executor::config::BlockExecutorConfig,
  contract_event::ContractEvent,
+ delayed_fields::PanicError,
  executable::ExecutableTestType,
  fee_statement::FeeStatement,
  state_store::{state_key::StateKey, StateView, StateViewId},
@@ -272,7 +273,7 @@ impl BlockExecutorTransactionOutput for AptosTransactionOutput {
  aggregator_v1_writes: Vec<(StateKey, WriteOp)>,
  patched_resource_write_set: Vec<(StateKey, WriteOp)>,
  patched_events: Vec<ContractEvent>,
- ) {
+ ) -> Result<(), PanicError> {
  assert!(
  self.committed_output
  .set(
@@ -284,13 +285,12 @@ impl BlockExecutorTransactionOutput for AptosTransactionOutput {
  aggregator_v1_writes,
  patched_resource_write_set,
  patched_events,
- )
- // TODO[agg_v2](fix) - propagate issued to block executor.
- .expect("Materialization must not fail"),
+ )?,
  )
  .is_ok(),
  "Could not combine VMOutput with the patched resource and event data"
  );
+ Ok(())
  }
 
  fn set_txn_output_for_non_dynamic_change_set(&self) {

aptos-move/aptos-vm/src/block_executor/vm_wrapper.rs

@@ -78,7 +78,8 @@ impl<'a, S: 'a + StateView + Sync> ExecutorTask for AptosExecutorTask<'a, S> {
  ExecutionStatus::SpeculativeExecutionAbortError(
  vm_status.message().cloned().unwrap_or_default(),
  )
- } else if vm_status.status_code() == StatusCode::DELAYED_FIELDS_CODE_INVARIANT_ERROR
+ } else if vm_status.status_code()
+ == StatusCode::DELAYED_MATERIALIZATION_CODE_INVARIANT_ERROR
  {
  ExecutionStatus::DelayedFieldsCodeInvariantError(
  vm_status.message().cloned().unwrap_or_default(),
@@ -104,7 +105,9 @@ impl<'a, S: 'a + StateView + Sync> ExecutorTask for AptosExecutorTask<'a, S> {
  ExecutionStatus::SpeculativeExecutionAbortError(
  err.message().cloned().unwrap_or_default(),
  )
- } else if err.status_code() == StatusCode::DELAYED_FIELDS_CODE_INVARIANT_ERROR {
+ } else if err.status_code()
+ == StatusCode::DELAYED_MATERIALIZATION_CODE_INVARIANT_ERROR
+ {
  ExecutionStatus::DelayedFieldsCodeInvariantError(
  err.message().cloned().unwrap_or_default(),
  )

aptos-move/aptos-vm/src/data_cache.rs

@@ -381,7 +381,7 @@ pub(crate) mod tests {
  };
 
  let group_adapter = ResourceGroupAdapter::new(
- // TODO[agg_v2](fix) add a converter for StateView for tests that implements ResourceGroupView
+ // TODO[agg_v2](test) add a converter for StateView for tests that implements ResourceGroupView
  None,
  state_view,
  gas_feature_version,

aptos-move/aptos-vm/src/errors.rs

@@ -212,7 +212,7 @@ pub fn expect_only_successful_execution(
  e @ VMStatus::Error {
  status_code:
  StatusCode::SPECULATIVE_EXECUTION_ABORT_ERROR
- | StatusCode::DELAYED_FIELDS_CODE_INVARIANT_ERROR,
+ | StatusCode::DELAYED_MATERIALIZATION_CODE_INVARIANT_ERROR,
  ..
  } => e,
  status => {

aptos-move/aptos-vm/src/move_vm_ext/respawned_session.rs

@@ -380,15 +380,8 @@ impl<'r> TResourceGroupView for ExecutorViewWithChangeSet<'r> {
  WriteResourceGroup(group_write) => Some(Ok(group_write)),
  ResourceGroupInPlaceDelayedFieldChange(_) => None,
  Write(_) | WriteWithDelayedFields(_) | InPlaceDelayedFieldChange(_) => {
- // TODO[agg_v2](fix): Revisit this error whether it should be invariant violation
- // or a panic/fallback.
- Some(Err(PartialVMError::new(
- StatusCode::UNKNOWN_INVARIANT_VIOLATION_ERROR,
- )
- .with_message(
- "Non-ResourceGroup write found for key in get_resource_from_group call"
- .to_string(),
- )))
+ // There should be no colisions, we cannot have group key refer to a resource.
+ Some(Err(code_invariant_error(format!("Non-ResourceGroup write found for key in get_resource_from_group call for key {group_key:?}"))))
  },
  })
  .transpose()?

aptos-move/aptos-vm/src/move_vm_ext/write_op_converter.rs

@@ -416,7 +416,7 @@ mod tests {
  }
  }
 
- // TODO[agg_v2](fix) make as_resolver_with_group_size_kind support AsSum
+ // TODO[agg_v2](test) make as_resolver_with_group_size_kind support AsSum
  // #[test]
  #[allow(unused)]
  fn size_computation_delete_modify_ops() {
@@ -474,7 +474,7 @@ mod tests {
  );
  }
 
- // TODO[agg_v2](fix) make as_resolver_with_group_size_kind support AsSum
+ // TODO[agg_v2](test) make as_resolver_with_group_size_kind support AsSum
  // #[test]
  #[allow(unused)]
  fn size_computation_new_op() {
@@ -515,7 +515,7 @@ mod tests {
  );
  }
 
- // TODO[agg_v2](fix) make as_resolver_with_group_size_kind support AsSum
+ // TODO[agg_v2](test) make as_resolver_with_group_size_kind support AsSum
  // #[test]
  #[allow(unused)]
  fn size_computation_new_group() {
@@ -541,7 +541,7 @@ mod tests {
  );
  }
 
- // TODO[agg_v2](fix) make as_resolver_with_group_size_kind support AsSum
+ // TODO[agg_v2](test) make as_resolver_with_group_size_kind support AsSum
  // #[test]
  #[allow(unused)]
  fn size_computation_delete_group() {

aptos-move/block-executor/src/captured_reads.rs

@@ -61,7 +61,7 @@ pub(crate) enum DataRead<V> {
  // (version implies value equality, but not vice versa). TODO: when
  // comparing the instances of V is cheaper, compare those instead.
  #[derivative(PartialEq = "ignore", Debug = "ignore")] Arc<V>,
- Option<Arc<MoveTypeLayout>>,
+ #[derivative(PartialEq = "ignore", Debug = "ignore")] Option<Arc<MoveTypeLayout>>,
  ),
  Metadata(Option<StateValueMetadata>),
  Exists(bool),

aptos-move/block-executor/src/executor.rs

@@ -887,7 +887,7 @@ where
  .chain(serialized_groups)
  .collect(),
  patched_events,
- );
+ )?;
  if let Some(txn_commit_listener) = &self.transaction_commit_hook {
  let txn_output = last_input_output.txn_output(txn_idx).unwrap();
  let execution_status = txn_output.output_status();
@@ -1370,18 +1370,20 @@ where
  Self::serialize_groups(patched_finalized_groups, idx as TxnIndex)
  .map_err(BlockExecutionError::FallbackToSequential)?;
 
- // TODO[agg_v2] patch resources in groups and provide explicitly
- output.incorporate_materialized_txn_output(
- // No aggregator v1 delta writes are needed for sequential execution.
- // They are already handled because we passed materialize_deltas=true
- // to execute_transaction.
- vec![],
- patched_resource_write_set
- .into_iter()
- .chain(serialized_groups.into_iter())
- .collect(),
- patched_events,
- );
+ output
+ .incorporate_materialized_txn_output(
+ // No aggregator v1 delta writes are needed for sequential execution.
+ // They are already handled because we passed materialize_deltas=true
+ // to execute_transaction.
+ vec![],
+ patched_resource_write_set
+ .into_iter()
+ .chain(serialized_groups.into_iter())
+ .collect(),
+ patched_events,
+ )
+ .map_err(PanicOr::from)
+ .map_err(BlockExecutionError::FallbackToSequential)?;
  }
  // If dynamic change set is disabled, this can be used to assert nothing needs patching instead:
  // output.set_txn_output_for_non_dynamic_change_set();

aptos-move/block-executor/src/proptest_types/types.rs

@@ -17,6 +17,7 @@ use aptos_types::{
  access_path::AccessPath,
  account_address::AccountAddress,
  contract_event::TransactionEvent,
+ delayed_fields::PanicError,
  executable::ModulePath,
  fee_statement::FeeStatement,
  on_chain_config::CurrentTimeMicroseconds,
@@ -1114,10 +1115,11 @@ where
  <Self::Txn as Transaction>::Value,
  )>,
  _patched_events: Vec<<Self::Txn as Transaction>::Event>,
- ) {
+ ) -> Result<(), PanicError> {
  assert_ok!(self.materialized_delta_writes.set(aggregator_v1_writes));
  // TODO[agg_v2](tests): Set the patched resource write set and events. But that requires the function
  // to take &mut self as input
+ Ok(())
  }
 
  fn set_txn_output_for_non_dynamic_change_set(&self) {

aptos-move/block-executor/src/task.rs

@@ -8,8 +8,8 @@ use aptos_aggregator::{
 };
 use aptos_mvhashmap::types::TxnIndex;
 use aptos_types::{
- fee_statement::FeeStatement, transaction::BlockExecutableTransaction as Transaction,
- write_set::WriteOp,
+ delayed_fields::PanicError, fee_statement::FeeStatement,
+ transaction::BlockExecutableTransaction as Transaction, write_set::WriteOp,
 };
 use aptos_vm_types::resolver::{TExecutorView, TResourceGroupView};
 use move_core_types::value::MoveTypeLayout;
@@ -179,7 +179,7 @@ pub trait TransactionOutput: Send + Sync + Debug {
  <Self::Txn as Transaction>::Value,
  )>,
  patched_events: Vec<<Self::Txn as Transaction>::Event>,
- );
+ ) -> Result<(), PanicError>;
 
  fn set_txn_output_for_non_dynamic_change_set(&self);
 

aptos-move/block-executor/src/txn_last_input_output.rs

@@ -12,8 +12,8 @@ use aptos_aggregator::types::{code_invariant_error, PanicOr};
 use aptos_logger::warn;
 use aptos_mvhashmap::types::{TxnIndex, ValueWithLayout};
 use aptos_types::{
- fee_statement::FeeStatement, transaction::BlockExecutableTransaction as Transaction,
- write_set::WriteOp,
+ delayed_fields::PanicError, fee_statement::FeeStatement,
+ transaction::BlockExecutableTransaction as Transaction, write_set::WriteOp,
 };
 use arc_swap::ArcSwapOption;
 use crossbeam::utils::CachePadded;
@@ -411,7 +411,7 @@ impl<T: Transaction, O: TransactionOutput<Txn = T>, E: Debug + Send + Clone>
  delta_writes: Vec<(T::Key, WriteOp)>,
  patched_resource_write_set: Vec<(T::Key, T::Value)>,
  patched_events: Vec<T::Event>,
- ) {
+ ) -> Result<(), PanicError> {
  match &self.outputs[txn_idx as usize]
  .load_full()
  .expect("Output must exist")
@@ -422,12 +422,13 @@ impl<T: Transaction, O: TransactionOutput<Txn = T>, E: Debug + Send + Clone>
  delta_writes,
  patched_resource_write_set,
  patched_events,
- );
+ )?;
  },
  ExecutionStatus::Abort(_)
  | ExecutionStatus::SpeculativeExecutionAbortError(_)
  | ExecutionStatus::DelayedFieldsCodeInvariantError(_) => {},
  };
+ Ok(())
  }
 
  pub(crate) fn get_txn_read_write_summary(&self, txn_idx: TxnIndex) -> ReadWriteSummary<T> {

aptos-move/block-executor/src/view.rs

@@ -1050,7 +1050,7 @@ impl<'a, T: Transaction, S: TStateView<Key = T::Key>, X: Executable> LatestView<
  );
  self.mark_incorrect_use();
  return Err(PartialVMError::new(
- StatusCode::DELAYED_FIELDS_CODE_INVARIANT_ERROR,
+ StatusCode::DELAYED_MATERIALIZATION_CODE_INVARIANT_ERROR,
  )
  .with_message(format!("{}", err)));
  },

aptos-move/mvhashmap/src/versioned_delayed_fields.rs

@@ -167,7 +167,7 @@ impl<K: Copy + Clone + Debug + Eq> VersionedValue<K> {
  )))
  },
  } {
- // TODO[agg_v2](fix): handle invalidation when we change read_estimate_deltas
+ // TODO[agg_v2](optimize): See if we want to invalidate, when we change read_estimate_deltas
  self.read_estimate_deltas = false;
  }
  o.insert(CachePadded::new(entry));

third_party/move/move-core/types/src/vm_status.rs

@@ -729,11 +729,12 @@ pub enum StatusCode {
  // Failed to resolve type due to linking being broken after verification
  TYPE_RESOLUTION_FAILURE = 2021,
  DUPLICATE_NATIVE_FUNCTION = 2022,
- // code invariant error while handling delayed fields, should never happen,
+ // code invariant error while handling delayed materialization, should never happen,
  // always indicates a code bug.
+ // Delayed materialization includes handling of Resource Groups and Delayed Fields.
  // Unlike regular CODE_INVARIANT_ERROR, this is a signal to BlockSTM,
  // which it might do something about (i.e. fallback to sequential execution)
- DELAYED_FIELDS_CODE_INVARIANT_ERROR = 2023,
+ DELAYED_MATERIALIZATION_CODE_INVARIANT_ERROR = 2023,
  // Speculative error means that there was an issue because of speculative
  // reads provided to the transaction, and the transaction needs to
  // be re-executed.