[types] Verify accumulator consistency proofs in trusted state

This diff modifies the `TrustedState` to support verifying accumulator
consistency proofs. Now when verifying state proofs, we will correctly
verify the consistency proof up to the last verifiable ledger info.

In order to verify a consistency proof, the trusted state needs to keep
track of a verified transaction accumulator summary; ratcheting a state
proof will also move the accumulator summary forward to the newly
trusted state.

When verifying from a waypoint, we haven't yet built an accumulator
summary. `verify_and_ratchet` now also takes an optional, untrusted
`initial_accumulator`, which the trusted state will verify and use when
it does not yet have an accumulator summary built up.

One consequence of verifying the consistency proofs is that we can no
longer verify a state proof with a newer latest li but a partially
trusted/verified prefix of epoch changes, since consistency proofs can
only be verified from the exact accumulator summary at their starting
version. This means users like the verifying client must
verify_and_ratchet from the trusted state they had at the start of their
request rather than reading the current trusted state when they verify
the response.

This diff also adds support for verifying the initializing the
accumulator and verifying consistency proofs to the verifying client.

execution/executor-test-helpers/src/integration_test_impl.rs

@@ -240,13 +240,18 @@ pub fn test_execution_with_storage_impl() -> Arc<DiemDB> {
         .commit_blocks(vec![block1_id], ledger_info_with_sigs)
         .unwrap();
 
-    let (li, epoch_change_proof, _accumulator_consistency_proof) =
-        db.reader.get_state_proof(0).unwrap();
-    let mut trusted_state = TrustedState::from(waypoint);
-    match trusted_state.verify_and_ratchet(&li, &epoch_change_proof) {
-        Ok(TrustedStateChange::Epoch { new_state, .. }) => trusted_state = new_state,
+    let initial_accumulator = db.reader.get_accumulator_summary(0).unwrap();
+    let (li, epoch_change_proof, consistency_proof) = db.reader.get_state_proof(0).unwrap();
+    let trusted_state = TrustedState::from_epoch_waypoint(waypoint);
+    let trusted_state = match trusted_state.verify_and_ratchet(
+        &li,
+        &epoch_change_proof,
+        &consistency_proof,
+        Some(&initial_accumulator),
+    ) {
+        Ok(TrustedStateChange::Epoch { new_state, .. }) => new_state,
         _ => panic!("unexpected state change"),
-    }
+    };
     let current_version = li.ledger_info().version();
     assert_eq!(trusted_state.version(), 9);
 
@@ -416,11 +421,15 @@ pub fn test_execution_with_storage_impl() -> Arc<DiemDB> {
         .commit_blocks(vec![block2_id], ledger_info_with_sigs)
         .unwrap();
 
-    let (li, epoch_change_proof, _accumulator_consistency_proof) =
+    let (li, epoch_change_proof, consistency_proof) =
         db.reader.get_state_proof(trusted_state.version()).unwrap();
-    trusted_state
-        .verify_and_ratchet(&li, &epoch_change_proof)
+    let trusted_state_change = trusted_state
+        .verify_and_ratchet(&li, &epoch_change_proof, &consistency_proof, None)
         .unwrap();
+    assert!(matches!(
+        trusted_state_change,
+        TrustedStateChange::Version { .. }
+    ));
     let current_version = li.ledger_info().version();
     assert_eq!(current_version, 23);
 

execution/executor/tests/db_bootstrapper_test.rs

@@ -68,11 +68,25 @@ fn test_empty_db() {
         Waypoint::new_epoch_boundary(startup_info.latest_ledger_info.ledger_info()).unwrap(),
         waypoint
     );
-    let (li, epoch_change_proof, _) = db_rw.reader.get_state_proof(waypoint.version()).unwrap();
-    let trusted_state = TrustedState::from(waypoint);
-    trusted_state
-        .verify_and_ratchet(&li, &epoch_change_proof)
+
+    let initial_accumulator = db_rw
+        .reader
+        .get_accumulator_summary(waypoint.version())
+        .unwrap();
+    let trusted_state = TrustedState::from_epoch_waypoint(waypoint);
+    let (li, epoch_change_proof, consistency_proof) = db_rw
+        .reader
+        .get_state_proof(trusted_state.version())
         .unwrap();
+    let trusted_state_change = trusted_state
+        .verify_and_ratchet(
+            &li,
+            &epoch_change_proof,
+            &consistency_proof,
+            Some(&initial_accumulator),
+        )
+        .unwrap();
+    assert!(trusted_state_change.is_epoch_change());
 
     // `maybe_bootstrap()` does nothing on non-empty DB.
     assert!(!maybe_bootstrap::<DiemVM>(&db_rw, &genesis_txn, waypoint).unwrap());
@@ -304,11 +318,25 @@ fn test_pre_genesis() {
     // Bootstrap DB on top of pre-genesis state.
     let waypoint = generate_waypoint::<DiemVM>(&db_rw, &genesis_txn).unwrap();
     assert!(maybe_bootstrap::<DiemVM>(&db_rw, &genesis_txn, waypoint).unwrap());
-    let (li, epoch_change_proof, _) = db_rw.reader.get_state_proof(waypoint.version()).unwrap();
-    let trusted_state = TrustedState::from(waypoint);
-    trusted_state
-        .verify_and_ratchet(&li, &epoch_change_proof)
+
+    let trusted_state = TrustedState::from_epoch_waypoint(waypoint);
+    let initial_accumulator = db_rw
+        .reader
+        .get_accumulator_summary(trusted_state.version())
+        .unwrap();
+    let (li, epoch_change_proof, consistency_proof) = db_rw
+        .reader
+        .get_state_proof(trusted_state.version())
+        .unwrap();
+    let trusted_state_change = trusted_state
+        .verify_and_ratchet(
+            &li,
+            &epoch_change_proof,
+            &consistency_proof,
+            Some(&initial_accumulator),
+        )
         .unwrap();
+    assert!(trusted_state_change.is_epoch_change());
 
     // Effect of bootstrapping reflected.
     assert_eq!(get_balance(&account1, &db_rw), 1000);
@@ -336,11 +364,23 @@ fn test_new_genesis() {
     execute_and_commit(vec![txn1, txn2, txn3, txn4], &db, &signer);
     assert_eq!(get_balance(&account1, &db), 2_000_000);
     assert_eq!(get_balance(&account2, &db), 2_000_000);
-    let (li, epoch_change_proof, _) = db.reader.get_state_proof(waypoint.version()).unwrap();
-    let trusted_state = TrustedState::from(waypoint);
-    trusted_state
-        .verify_and_ratchet(&li, &epoch_change_proof)
+
+    let trusted_state = TrustedState::from_epoch_waypoint(waypoint);
+    let initial_accumulator = db
+        .reader
+        .get_accumulator_summary(trusted_state.version())
         .unwrap();
+    let (li, epoch_change_proof, consistency_proof) =
+        db.reader.get_state_proof(trusted_state.version()).unwrap();
+    let trusted_state_change = trusted_state
+        .verify_and_ratchet(
+            &li,
+            &epoch_change_proof,
+            &consistency_proof,
+            Some(&initial_accumulator),
+        )
+        .unwrap();
+    assert!(trusted_state_change.is_epoch_change());
 
     // New genesis transaction: set validator set, bump epoch and overwrite account1 balance.
     let configuration = get_configuration(&db);
@@ -375,14 +415,25 @@ fn test_new_genesis() {
     assert_eq!(waypoint.version(), 5);
 
     // Client bootable from waypoint.
-    let trusted_state = TrustedState::from(waypoint);
-    let (li, epoch_change_proof, accumulator_consistency_proof) =
+    let trusted_state = TrustedState::from_epoch_waypoint(waypoint);
+    let initial_accumulator = db
+        .reader
+        .get_accumulator_summary(trusted_state.version())
+        .unwrap();
+    let (li, epoch_change_proof, consistency_proof) =
         db.reader.get_state_proof(trusted_state.version()).unwrap();
-    assert_eq!(li.ledger_info().version(), 5);
-    assert!(accumulator_consistency_proof.subtrees().is_empty());
-    trusted_state
-        .verify_and_ratchet(&li, &epoch_change_proof)
+    let trusted_state_change = trusted_state
+        .verify_and_ratchet(
+            &li,
+            &epoch_change_proof,
+            &consistency_proof,
+            Some(&initial_accumulator),
+        )
         .unwrap();
+    assert!(trusted_state_change.is_epoch_change());
+    let trusted_state = trusted_state_change.new_state().unwrap();
+    assert_eq!(trusted_state.version(), 5);
+    assert!(consistency_proof.is_empty());
 
     // Effect of bootstrapping reflected.
     assert_eq!(get_balance(&account1, &db), 1_000_000);

execution/executor/tests/storage_integration_test.rs

@@ -11,7 +11,7 @@ use diem_types::{
     account_state::AccountState,
     block_metadata::BlockMetadata,
     transaction::{Script, Transaction, WriteSetPayload},
-    trusted_state::{TrustedState, TrustedStateChange},
+    trusted_state::TrustedState,
     validator_signer::ValidatorSigner,
 };
 use executor_test_helpers::{
@@ -30,12 +30,21 @@ fn test_genesis() {
     let genesis = vm_genesis::test_genesis_transaction();
     let (_, db, _executor, waypoint) = create_db_and_executor(path.path(), &genesis);
 
-    let (li, epoch_change_proof, _accumulator_consistency_proof) =
-        db.reader.get_state_proof(0).unwrap();
+    let trusted_state = TrustedState::from_epoch_waypoint(waypoint);
+    let initial_accumulator = db
+        .reader
+        .get_accumulator_summary(trusted_state.version())
+        .unwrap();
+    let (li, epoch_change_proof, consistency_proof) =
+        db.reader.get_state_proof(trusted_state.version()).unwrap();
 
-    let trusted_state = TrustedState::from(waypoint);
     trusted_state
-        .verify_and_ratchet(&li, &epoch_change_proof)
+        .verify_and_ratchet(
+            &li,
+            &epoch_change_proof,
+            &consistency_proof,
+            Some(&initial_accumulator),
+        )
         .unwrap();
     let li = li.ledger_info();
     assert_eq!(li.version(), 0);
@@ -326,14 +335,25 @@ fn test_change_publishing_option_to_custom() {
         .commit_blocks(vec![block1_id], ledger_info_with_sigs)
         .unwrap();
 
-    let (li, epoch_change_proof, _accumulator_consistency_proof) =
-        db.reader.get_state_proof(0).unwrap();
-    let mut trusted_state = TrustedState::from(waypoint);
-    match trusted_state.verify_and_ratchet(&li, &epoch_change_proof) {
-        Ok(TrustedStateChange::Epoch { new_state, .. }) => trusted_state = new_state,
-        _ => panic!("unexpected state change"),
-    }
-    let current_version = li.ledger_info().version();
+    let trusted_state = TrustedState::from_epoch_waypoint(waypoint);
+    let initial_accumulator = db
+        .reader
+        .get_accumulator_summary(trusted_state.version())
+        .unwrap();
+    let (li, epoch_change_proof, consistency_proof) =
+        db.reader.get_state_proof(trusted_state.version()).unwrap();
+    let trusted_state_change = trusted_state
+        .verify_and_ratchet(
+            &li,
+            &epoch_change_proof,
+            &consistency_proof,
+            Some(&initial_accumulator),
+        )
+        .unwrap();
+    assert!(trusted_state_change.is_epoch_change());
+    let trusted_state = trusted_state_change.new_state().unwrap();
+
+    let current_version = trusted_state.version();
     assert_eq!(current_version, 3);
     // Transaction 1 is committed as it's in the allowlist
     let txn1 = db
@@ -376,11 +396,13 @@ fn test_change_publishing_option_to_custom() {
         .commit_blocks(vec![block2_id], ledger_info_with_sigs)
         .unwrap();
 
-    let (li, epoch_change_proof, _accumulator_consistency_proof) =
+    let (li, epoch_change_proof, consistency_proof) =
         db.reader.get_state_proof(current_version).unwrap();
-    trusted_state
-        .verify_and_ratchet(&li, &epoch_change_proof)
+    let trusted_state_change = trusted_state
+        .verify_and_ratchet(&li, &epoch_change_proof, &consistency_proof, None)
         .unwrap();
+    assert!(!trusted_state_change.is_epoch_change());
+
     let current_version = li.ledger_info().version();
     assert_eq!(current_version, 5);
     // Transaction 2 is committed.

sdk/client/src/error.rs

@@ -121,11 +121,8 @@ impl Error {
         )
     }
 
-    pub(crate) fn stale(expected: &super::State, recieved: &super::State) -> Self {
-        Self::new(
-            Kind::StaleResponse,
-            Some(format!("expected: {:?} recieved: {:?}", expected, recieved)),
-        )
+    pub(crate) fn stale<E: Into<BoxError>>(e: E) -> Self {
+        Self::new(Kind::StaleResponse, Some(e))
     }
 
     cfg_async! {

sdk/client/src/response.rs

@@ -155,6 +155,18 @@ impl MethodResponse {
         }
     }
 
+    pub fn try_into_get_accumulator_consistency_proof(
+        self,
+    ) -> Result<AccumulatorConsistencyProofView, Error> {
+        match self {
+            MethodResponse::GetAccumulatorConsistencyProof(proof) => Ok(proof),
+            _ => Err(Error::rpc_response(format!(
+                "expected MethodResponse::GetAccumulatorConsistencyProof found MethodResponse::{:?}",
+                self.method()
+            ))),
+        }
+    }
+
     pub fn try_into_get_account(self) -> Result<Option<AccountView>, Error> {
         match self {
             MethodResponse::GetAccount(account_view) => Ok(account_view),

sdk/client/src/state.rs

@@ -58,7 +58,11 @@ cfg_async_or_blocking! {
                     return Err(Error::chain_id(state.chain_id, resp_state.chain_id));
                 }
                 if resp_state < state {
-                    return Err(Error::stale(state, resp_state));
+                    return Err(Error::stale(format!(
+                        "received response with stale metadata: {:?}, expected a response more recent than: {:?}",
+                        resp_state,
+                        state,
+                    )));
                 }
             }
             *state_writer = Some(resp_state.clone());