[vm] Introducing a delta write (aptos-labs#2057)

Introduced a new `WriteOp::Delta`. Delta is parametrised by a partial update
function, such as f(x) = + x or f(x) = - x, operating on 128-bit unsigned int,
and a limit (a postcondition that ensures the result of delta application does
not overflow).

Note: in the future, we would like to keep deltas private to executor, for
example by modifying `ChangeSet` or `TransactionOutput` instead.

api/types/src/convert.rs

@@ -257,6 +257,8 @@ 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)
     }
@@ -282,6 +284,8 @@ impl<'a, R: MoveResolverExt + ?Sized> MoveConverter<'a, R> {
                 key,
                 value: value.into(),
             },
+            // Deltas are materialized into WriteOP::Value(..) in executor.
+            WriteOp::Delta(..) => unreachable!("unexpected conversion"),
         };
         Ok(ret)
     }

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

@@ -72,6 +72,9 @@ 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")
+                }
             }
         }
     }

aptos-move/aptos-vm/src/parallel_executor/storage_wrapper.rs

@@ -35,6 +35,9 @@ 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),
         }

aptos-move/e2e-tests/src/data_store.rs

@@ -50,6 +50,7 @@ impl FakeDataStore {
                 WriteOp::Deletion => {
                     self.remove(state_key);
                 }
+                WriteOp::Delta(..) => unreachable!("deltas are only used in executor"),
             }
         }
     }

aptos-move/genesis-viewer/src/main.rs

@@ -188,6 +188,7 @@ fn print_modules(ws: &WriteSet) {
             AccessPath::try_from(k.clone()).expect("State key can't be converted to access path");
         match v {
             WriteOp::Deletion => panic!("found WriteOp::Deletion in WriteSet"),
+            WriteOp::Delta(..) => panic!("found WriteOp::Delta in WriteSet"),
             WriteOp::Value(blob) => {
                 let tag = ap.path.get(0).expect("empty blob in WriteSet");
                 if *tag == 0 {
@@ -213,6 +214,7 @@ fn print_resources(storage: &impl MoveResolverExt, ws: &WriteSet) {
             AccessPath::try_from(k.clone()).expect("State key can't be converted to access path");
         match v {
             WriteOp::Deletion => panic!("found WriteOp::Deletion in WriteSet"),
+            WriteOp::Delta(..) => panic!("found WriteOp::Delta in WriteSet"),
             WriteOp::Value(blob) => {
                 let tag = ap.path.get(0).expect("empty blob in WriteSet");
                 if *tag == 1 {
@@ -243,6 +245,7 @@ fn print_account_states(storage: &impl MoveResolverExt, ws: &WriteSet) {
             AccessPath::try_from(k.clone()).expect("State key can't be converted to access path");
         match v {
             WriteOp::Deletion => panic!("found WriteOp::Deletion in WriteSet"),
+            WriteOp::Delta(..) => panic!("found WriteOp::Delta in WriteSet"),
             WriteOp::Value(blob) => {
                 let tag = ap.path.get(0).expect("empty blob in WriteSet");
                 if *tag == 1 {

aptos-move/transaction-replay/src/lib.rs

@@ -177,10 +177,12 @@ impl AptosDebugger {
                 access_path::Path::Resource(tag) => match op {
                     WriteOp::Deletion => state_view.delete_resource(addr, tag)?,
                     WriteOp::Value(bytes) => state_view.save_resource(addr, tag, bytes)?,
+                    WriteOp::Delta(..) => unreachable!("deltas are only used in executor"),
                 },
                 access_path::Path::Code(module_id) => match op {
                     WriteOp::Deletion => state_view.delete_module(&module_id)?,
                     WriteOp::Value(bytes) => state_view.save_module(&module_id, bytes)?,
+                    WriteOp::Delta(..) => unreachable!("deltas are only used in executor"),
                 },
             }
         }

documentation/specifications/move_adapter/README.md

@@ -531,13 +531,28 @@ pub struct TransactionOutput {
 /// `WriteSet` contains all access paths that one transaction modifies.
 /// Each of them is a `WriteOp` where `Value(val)` means that serialized
 /// representation should be updated to `val`, and `Deletion` means that
-/// we are going to delete this access path.
+/// we are going to delete this access path. A special `WriteOp` used by
+/// aggregator - `Delta(op, limit)`, means that `op` should be applied to
+/// deserialized representation and a postcondition `result <= limit` must be
+/// ensured.
 pub struct WriteSet {
     write_set: Vec<(AccessPath, WriteOp)>,
 }
 
+/// `DeltaOperation` specifies the partial function which is used to apply
+/// delta.
+pub enum DeltaOperation {
+    Addition(u128),
+    Subtraction(u128),
+}
+
+/// Value when delta application overflows, i.e. the postcondition of delta
+/// application.
+pub struct DeltaLimit(pub u128);
+
 pub enum WriteOp {
     Deletion,
+    Delta(DeltaOperation, DeltaLimit),
     Value(Vec<u8>),
 }
 

execution/executor-types/src/in_memory_state_calculator.rs

@@ -244,6 +244,7 @@ fn process_state_key_write_op(
     let state_value = match write_op {
         WriteOp::Value(new_value) => StateValue::from(new_value),
         WriteOp::Deletion => StateValue::empty(),
+        WriteOp::Delta(..) => unreachable!("deltas are only used in executor"),
     };
     match state_cache.entry(state_key.clone()) {
         hash_map::Entry::Occupied(mut entry) => {

types/src/write_set.rs

@@ -2,24 +2,54 @@
 // SPDX-License-Identifier: Apache-2.0
 
 //! For each transaction the VM executes, the VM will output a `WriteSet` that contains each access
-//! path it updates. For each access path, the VM can either give its new value or delete it.
+//! path it updates. For each access path, the VM can either give its new value or delete it. For
+//! aggregator, delta updates are used (note: this is a temporary solution and ideally we should
+//! modify `ChangeSet` and `TransactionOutput` to keep deltas internal to executor).
 
 use crate::state_store::state_key::StateKey;
 use anyhow::Result;
 use aptos_crypto_derive::{BCSCryptoHash, CryptoHasher};
 use serde::{Deserialize, Serialize};
 
+/// Specifies partial function such as +X or -X to use with `WriteOp::Delta`.
+#[derive(Copy, Clone, Hash, PartialOrd, Ord, PartialEq, Eq)]
+pub enum DeltaOperation {
+    Addition(u128),
+    Subtraction(u128),
+}
+
+impl std::fmt::Debug for DeltaOperation {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            DeltaOperation::Addition(value) => write!(f, "+{}", value),
+            DeltaOperation::Subtraction(value) => write!(f, "-{}", value),
+        }
+    }
+}
+
+#[derive(Copy, Clone, Hash, PartialOrd, Ord, PartialEq, Eq)]
+pub struct DeltaLimit(pub u128);
+
+impl std::fmt::Debug for DeltaLimit {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "result <= {}", self.0)
+    }
+}
+
 #[derive(Clone, Eq, Hash, PartialEq, Serialize, Deserialize)]
 pub enum WriteOp {
     Deletion,
     Value(#[serde(with = "serde_bytes")] Vec<u8>),
+    #[serde(skip)]
+    Delta(DeltaOperation, DeltaLimit),
 }
 
 impl WriteOp {
     #[inline]
     pub fn is_deletion(&self) -> bool {
         match self {
             WriteOp::Deletion => true,
+            WriteOp::Delta(..) => false,
             WriteOp::Value(_) => false,
         }
     }
@@ -36,6 +66,9 @@ impl std::fmt::Debug for WriteOp {
                     .map(|byte| format!("{:02x}", byte))
                     .collect::<String>()
             ),
+            WriteOp::Delta(op, limit) => {
+                write!(f, "Delta({:?} ensures {:?})", op, limit)
+            }
             WriteOp::Deletion => write!(f, "Deletion"),
         }
     }