Make state value metadata upgradable at runtime

Remove StateValueMetadataKind=Option<StateValueMetadata>, instead, add
StateValueMetadata::Dummy as a in memory representation of a None
metadata. This way, we are able to upgrade the metadata format at
runtime (when we charge and refund the storage fee). This is to prepare
for the refundable permanent bytes.

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

@@ -589,10 +589,16 @@ mod test {
         let sub_op = delta_sub(100, 200);
 
         let add_result = state_view.try_convert_aggregator_v1_delta_into_write_op(&KEY, &add_op);
-        assert_ok_eq!(add_result, WriteOp::Modification(serialize(&200).into()));
+        assert_ok_eq!(
+            add_result,
+            WriteOp::legacy_modification(serialize(&200).into())
+        );
 
         let sub_result = state_view.try_convert_aggregator_v1_delta_into_write_op(&KEY, &sub_op);
-        assert_ok_eq!(sub_result, WriteOp::Modification(serialize(&0).into()));
+        assert_ok_eq!(
+            sub_result,
+            WriteOp::legacy_modification(serialize(&0).into())
+        );
     }
 
     #[test]

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

@@ -15,7 +15,7 @@ use aptos_types::{
     aggregator::PanicError,
     state_store::{
         state_key::StateKey,
-        state_value::{StateValue, StateValueMetadataKind},
+        state_value::{StateValue, StateValueMetadata},
     },
     write_set::WriteOp,
 };
@@ -64,7 +64,7 @@ pub trait TAggregatorV1View {
     fn get_aggregator_v1_state_value_metadata(
         &self,
         id: &Self::Identifier,
-    ) -> anyhow::Result<Option<StateValueMetadataKind>> {
+    ) -> anyhow::Result<Option<StateValueMetadata>> {
         // When getting state value metadata for aggregator V1, we need to do a
         // precise read.
         let maybe_state_value = self.get_aggregator_v1_state_value(id)?;
@@ -120,7 +120,7 @@ pub trait TAggregatorV1View {
                     )))
                     .into_vm_status()
             })
-            .map(|result| WriteOp::Modification(serialize(&result).into()))
+            .map(|result| WriteOp::legacy_modification(serialize(&result).into()))
     }
 }
 

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

@@ -5,7 +5,6 @@ use aptos_types::{
     state_store::state_value::StateValueMetadata,
     write_set::{TransactionWrite, WriteOp, WriteOpSize},
 };
-use claims::assert_none;
 use move_core_types::{language_storage::StructTag, value::MoveTypeLayout};
 use std::{collections::BTreeMap, sync::Arc};
 
@@ -55,15 +54,13 @@ impl AbstractResourceWriteOp {
             }) => {
                 use WriteOp::*;
                 match write_op {
-                    Creation(_) | CreationWithMetadata { .. } => WriteOpSize::Creation {
+                    Creation { .. } => WriteOpSize::Creation {
                         write_len: materialized_size.expect("Creation must have size"),
                     },
-                    Modification(_) | ModificationWithMetadata { .. } => {
-                        WriteOpSize::Modification {
-                            write_len: materialized_size.expect("Modification must have size"),
-                        }
+                    Modification { .. } => WriteOpSize::Modification {
+                        write_len: materialized_size.expect("Modification must have size"),
                     },
-                    Deletion | DeletionWithMetadata { .. } => WriteOpSize::Deletion,
+                    Deletion { .. } => WriteOpSize::Deletion,
                 }
             },
             InPlaceDelayedFieldChange(InPlaceDelayedFieldChangeOp {
@@ -80,16 +77,20 @@ impl AbstractResourceWriteOp {
 
     /// Deposit amount is inserted into metadata at a different time than the WriteOp is created.
     /// So this method is needed to be able to update metadata generically across different variants.
-    pub fn get_metadata_mut(&mut self) -> Option<&mut StateValueMetadata> {
+    pub fn get_metadata_mut(&mut self) -> &mut StateValueMetadata {
         use AbstractResourceWriteOp::*;
         match self {
             Write(write_op)
             | WriteWithDelayedFields(WriteWithDelayedFieldsOp { write_op, .. })
             | WriteResourceGroup(GroupWrite {
                 metadata_op: write_op,
                 ..
+            })
+            | ResourceGroupInPlaceDelayedFieldChange(ResourceGroupInPlaceDelayedFieldChangeOp {
+                metadata_op: write_op,
+                ..
             }) => write_op.get_metadata_mut(),
-            InPlaceDelayedFieldChange(_) | ResourceGroupInPlaceDelayedFieldChange(_) => None,
+            InPlaceDelayedFieldChange(InPlaceDelayedFieldChangeOp { metadata, .. }) => metadata,
         }
     }
 
@@ -148,7 +149,10 @@ impl GroupWrite {
             metadata_op
         );
         for (_tag, (v, _layout)) in &inner_ops {
-            assert_none!(v.metadata(), "Group inner ops must have no metadata");
+            assert!(
+                v.metadata().is_none(),
+                "Group inner ops must have no metadata"
+            )
         }
 
         let maybe_group_op_size = (!metadata_op.is_deletion()).then_some(group_size);
@@ -193,6 +197,7 @@ pub struct WriteWithDelayedFieldsOp {
 pub struct InPlaceDelayedFieldChangeOp {
     pub layout: Arc<MoveTypeLayout>,
     pub materialized_size: u64,
+    pub metadata: StateValueMetadata,
 }
 
 /// Actual information of which individual tag has delayed fields was read,

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

@@ -154,6 +154,7 @@ impl VMChangeSet {
                                                     ),
                                                 )
                                             })?,
+                                        metadata: w.into_metadata(),
                                     },
                                 ),
                             ))
@@ -341,7 +342,7 @@ impl VMChangeSet {
     /// So this method is needed to be able to update metadata generically across different variants.
     pub fn write_set_iter_mut(
         &mut self,
-    ) -> impl Iterator<Item = (&StateKey, WriteOpSize, Option<&mut StateValueMetadata>)> {
+    ) -> impl Iterator<Item = (&StateKey, WriteOpSize, &mut StateValueMetadata)> {
         self.resource_write_set
             .iter_mut()
             .map(|(k, v)| (k, v.materialized_size(), v.get_metadata_mut()))
@@ -476,10 +477,7 @@ impl VMChangeSet {
             if let Some(write_op) = aggregator_v1_write_set.get_mut(&state_key) {
                 // In this case, delta follows a write op.
                 match write_op {
-                    Creation(data)
-                    | Modification(data)
-                    | CreationWithMetadata { data, .. }
-                    | ModificationWithMetadata { data, .. } => {
+                    Creation { data, .. } | Modification { data, .. } => {
                         // Apply delta on top of creation or modification.
                         // TODO[agg_v1](cleanup): This will not be needed anymore once aggregator
                         // change sets carry non-serialized information.
@@ -491,7 +489,7 @@ impl VMChangeSet {
                             .map_err(|e| e.finish(Location::Undefined).into_vm_status())?;
                         *data = serialize(&value).into();
                     },
-                    Deletion | DeletionWithMetadata { .. } => {
+                    Deletion { .. } => {
                         // This case (applying a delta to deleted item) should
                         // never happen. Let's still return an error instead of
                         // panicking.

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

@@ -10,7 +10,7 @@ use aptos_types::{
     state_store::{
         state_key::StateKey,
         state_storage_usage::StateStorageUsage,
-        state_value::{StateValue, StateValueMetadataKind},
+        state_value::{StateValue, StateValueMetadata},
     },
     write_set::WriteOp,
 };
@@ -45,7 +45,7 @@ pub trait TResourceView {
     fn get_resource_state_value_metadata(
         &self,
         state_key: &Self::Key,
-    ) -> anyhow::Result<Option<StateValueMetadataKind>> {
+    ) -> anyhow::Result<Option<StateValueMetadata>> {
         // For metadata, layouts are not important.
         self.get_resource_state_value(state_key, None)
             .map(|maybe_state_value| maybe_state_value.map(StateValue::into_metadata))
@@ -151,7 +151,7 @@ pub trait TModuleView {
     fn get_module_state_value_metadata(
         &self,
         state_key: &Self::Key,
-    ) -> anyhow::Result<Option<StateValueMetadataKind>> {
+    ) -> anyhow::Result<Option<StateValueMetadata>> {
         let maybe_state_value = self.get_module_state_value(state_key)?;
         Ok(maybe_state_value.map(StateValue::into_metadata))
     }
@@ -275,13 +275,13 @@ pub trait StateValueMetadataResolver {
     fn get_module_state_value_metadata(
         &self,
         state_key: &StateKey,
-    ) -> anyhow::Result<Option<StateValueMetadataKind>>;
+    ) -> anyhow::Result<Option<StateValueMetadata>>;
 
     /// Can also be used to get the metadata of a resource group at a provided group key.
     fn get_resource_state_value_metadata(
         &self,
         state_key: &StateKey,
-    ) -> anyhow::Result<Option<StateValueMetadataKind>>;
+    ) -> anyhow::Result<Option<StateValueMetadata>>;
 }
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]

aptos-move/aptos-vm-types/src/storage/space_pricing.rs

@@ -33,7 +33,7 @@ impl DiskSpacePricing {
         params: &TransactionGasParameters,
         key: &StateKey,
         op_size: &WriteOpSize,
-        metadata: Option<&mut StateValueMetadata>,
+        metadata: &mut StateValueMetadata,
     ) -> ChargeAndRefund {
         match self {
             Self::V1 => Self::charge_refund_write_op_v1(params, key, op_size, metadata),
@@ -97,7 +97,7 @@ impl DiskSpacePricing {
         params: &TransactionGasParameters,
         key: &StateKey,
         op_size: &WriteOpSize,
-        metadata: Option<&mut StateValueMetadata>,
+        metadata: &mut StateValueMetadata,
     ) -> ChargeAndRefund {
         use WriteOpSize::*;
 
@@ -107,8 +107,8 @@ impl DiskSpacePricing {
                 let bytes_fee = Self::discounted_write_op_size_for_v1(params, key, *write_len)
                     * params.storage_fee_per_excess_state_byte;
 
-                if let Some(m) = metadata {
-                    m.set_deposit(slot_fee.into())
+                if !metadata.is_none() {
+                    metadata.set_slot_deposit(slot_fee.into())
                 }
 
                 ChargeAndRefund {
@@ -125,17 +125,9 @@ impl DiskSpacePricing {
                     refund: 0.into(),
                 }
             },
-            Deletion => {
-                let refund = match metadata {
-                    None => 0,
-                    Some(m) => m.deposit(),
-                }
-                .into();
-
-                ChargeAndRefund {
-                    charge: 0.into(),
-                    refund,
-                }
+            Deletion => ChargeAndRefund {
+                charge: 0.into(),
+                refund: metadata.total_deposit().into(),
             },
         }
     }