Fix wrong epoch hash in db for receipts. (#1421)

In the old implementation, if

1. Block A is on pivot chain and executed.
2. Pivot chain switches, and Block A is referred and executed in the epoch of block B.
3. Pivot chain switches back where A is on pivot chain.

Our receipts for Block A in db will still be the one that it's executed in
the epoch of block B.

This bugfix is just a patch, we should refactor/optimize this receipt
storage to make it more clear and efficient. The current implemenration may
need to insert TransactionIndex and BlockExecutionResult duplicatedly
due to pivot chain switch, especially when optimistic execution is
enabled.

client/src/rpc/impls/pubsub.rs

@@ -322,7 +322,8 @@ impl ChainNotificationHandler {
 
         for iter in 0..NUM_POLLS {
             match self.data_man.block_execution_result_by_hash_with_epoch(
-                &block, &pivot, true, /* update_cache */
+                &block, &pivot, false, /* update_pivot_assumption */
+                true,  /* update_cache */
             ) {
                 Some(res) => return Some(res.block_receipts.clone()),
                 None => {

core/src/block_data_manager/block_data_types.rs

@@ -73,40 +73,49 @@ pub struct BlockExecutionResultWithEpoch(
 #[derive(Default, Debug)]
 pub struct BlockReceiptsInfo {
     info_with_epoch: Vec<BlockExecutionResultWithEpoch>,
+    // The current pivot epoch that this block is executed.
+    // This should be consistent with the epoch hash in database.
+    pivot_epoch: EpochIndex,
 }
 
 impl BlockReceiptsInfo {
-    /// `epoch` is the index of the epoch id in consensus arena
+    /// Return None if we do not have a corresponding ExecutionResult in the
+    /// given `epoch`. Return `(ExecutionResult, is_on_pivot)` otherwise.
     pub fn get_receipts_at_epoch(
         &self, epoch: &EpochIndex,
-    ) -> Option<BlockExecutionResult> {
+    ) -> Option<(BlockExecutionResult, bool)> {
         for BlockExecutionResultWithEpoch(e_id, receipts) in
             &self.info_with_epoch
         {
             if *e_id == *epoch {
-                return Some(receipts.clone());
+                return Some((receipts.clone(), epoch == &self.pivot_epoch));
             }
         }
         None
     }
 
+    pub fn set_pivot_hash(&mut self, epoch: EpochIndex) {
+        self.pivot_epoch = epoch;
+    }
+
     /// Insert the tx fee when the block is included in epoch `epoch`
     pub fn insert_receipts_at_epoch(
         &mut self, epoch: &EpochIndex, receipts: BlockExecutionResult,
     ) {
-        // If it's inserted before, the fee must be the same, so we do not add
-        // duplicate entry
+        // If it's inserted before, we do not need to push a duplicated entry.
         if self.get_receipts_at_epoch(epoch).is_none() {
             self.info_with_epoch
                 .push(BlockExecutionResultWithEpoch(*epoch, receipts));
         }
+        self.pivot_epoch = *epoch;
     }
 
     /// Only keep the tx fee in the given `epoch`
     /// Called after we process rewards, and other fees will not be used w.h.p.
     pub fn retain_epoch(&mut self, epoch: &EpochIndex) {
         self.info_with_epoch
             .retain(|BlockExecutionResultWithEpoch(e_id, _)| *e_id == *epoch);
+        self.pivot_epoch = *epoch;
     }
 }
 

core/src/block_data_manager/mod.rs

@@ -576,26 +576,44 @@ impl BlockDataManager {
     }
 
     /// Return None if receipts for corresponding epoch is not computed before
-    /// or has been overwritten by another new pivot chain in db
+    /// or has been overwritten by another new pivot chain in db.
+    /// If `update_pivot_assumption` is true and we have execution results of
+    /// `assumed_epoch` in memory, we will also ensure `assumed_epoch`
+    /// is persisted as the pivot hash in db.
     ///
-    /// This function will require lock of block_receipts
+    /// This function will require lock of block_receipts.
     pub fn block_execution_result_by_hash_with_epoch(
-        &self, hash: &H256, assumed_epoch: &H256, update_cache: bool,
-    ) -> Option<BlockExecutionResult> {
-        let maybe_receipts =
-            self.block_receipts
-                .read()
-                .get(hash)
-                .and_then(|receipt_info| {
-                    receipt_info.get_receipts_at_epoch(assumed_epoch)
-                });
-        if maybe_receipts.is_some() {
+        &self, hash: &H256, assumed_epoch: &H256,
+        update_pivot_assumption: bool, update_cache: bool,
+    ) -> Option<BlockExecutionResult>
+    {
+        if let Some((receipts, is_on_pivot)) = self
+            .block_receipts
+            .write()
+            .get_mut(hash)
+            .and_then(|receipt_info| {
+                let r = receipt_info.get_receipts_at_epoch(assumed_epoch);
+                if update_pivot_assumption {
+                    receipt_info.set_pivot_hash(*assumed_epoch);
+                }
+                r
+            })
+        {
             if update_cache {
                 self.cache_man
                     .lock()
                     .note_used(CacheId::BlockReceipts(*hash));
             }
-            return maybe_receipts;
+            if update_pivot_assumption && !is_on_pivot {
+                self.db_manager.insert_block_execution_result_to_db(
+                    hash,
+                    &BlockExecutionResultWithEpoch(
+                        *assumed_epoch,
+                        receipts.clone(),
+                    ),
+                )
+            }
+            return Some(receipts);
         }
         let BlockExecutionResultWithEpoch(epoch, receipts) =
             self.db_manager.block_execution_result_from_db(hash)?;
@@ -970,7 +988,8 @@ impl BlockDataManager {
             let mut epoch_receipts = Vec::new();
             for h in epoch_block_hashes {
                 if let Some(r) = self.block_execution_result_by_hash_with_epoch(
-                    h, epoch_hash, true, /* update_cache */
+                    h, epoch_hash, true, /* update_pivot_assumption */
+                    true, /* update_cache */
                 ) {
                     epoch_receipts.push(r.block_receipts);
                 } else {

core/src/consensus/consensus_inner/consensus_executor.rs

@@ -1314,7 +1314,8 @@ impl ConsensusExecutionHandler {
                 .block_execution_result_by_hash_with_epoch(
                     &block_hash,
                     &reward_epoch_hash,
-                    true, /* update_cache */
+                    false, /* update_pivot_assumption */
+                    true,  /* update_cache */
                 ) {
                 Some(block_exec_result) => block_exec_result.block_receipts,
                 None => {

core/src/consensus/consensus_inner/mod.rs

@@ -2149,6 +2149,7 @@ impl ConsensusGraphInner {
                     self.data_man.block_execution_result_by_hash_with_epoch(
                         hash,
                         &epoch,
+                        false, /* update_pivot_assumption */
                         update_cache,
                     )?;
                 Some(BlockExecutionResultWithEpoch(epoch, execution_result))

core/src/consensus/mod.rs

@@ -800,7 +800,9 @@ impl ConsensusGraph {
         if let Some(block_log_bloom) = self
             .data_man
             .block_execution_result_by_hash_with_epoch(
-                block_hash, epoch_hash, false, /* update_cache */
+                block_hash, epoch_hash,
+                false, /* update_pivot_assumption */
+                false, /* update_cache */
             )
             .map(|r| r.bloom)
         {

core/src/light_protocol/common/ledger_info.rs

@@ -185,7 +185,8 @@ impl LedgerInfo {
                 self.consensus
                     .get_data_manager()
                     .block_execution_result_by_hash_with_epoch(
-                        &h, &pivot, false, /* update_cache */
+                        &h, &pivot, false, /* update_pivot_assumption */
+                        false, /* update_cache */
                     )
                     .map(|res| (*res.block_receipts).clone())
                     .ok_or(ErrorKind::InternalError.into())
@@ -209,7 +210,8 @@ impl LedgerInfo {
                 self.consensus
                     .get_data_manager()
                     .block_execution_result_by_hash_with_epoch(
-                        &h, &pivot, false, /* update_cache */
+                        &h, &pivot, false, /* update_pivot_assumption */
+                        false, /* update_cache */
                     )
                     .map(|res| res.bloom)
                     .ok_or(ErrorKind::InternalError.into())

core/src/sync/state/snapshot_manifest_request.rs

@@ -119,6 +119,7 @@ impl SnapshotManifestRequest {
                                 .block_execution_result_by_hash_with_epoch(
                                     hash,
                                     &epoch_hash,
+                                    false, /* update_pivot_assumption */
                                     false, /* update_cache */
                                 ) {
                                 Some(block_execution_result) => {