Client/Blockchain: receipt reorg logic

packages/blockchain/src/blockchain.ts

@@ -9,6 +9,7 @@
 } from '@ethereumjs/common'
 import { genesisStateRoot as genGenesisStateRoot } from '@ethereumjs/trie'
 import {
+  AsyncEventEmitter,
   BIGINT_0,
   BIGINT_1,
   BIGINT_8,
@@ -33,6 +34,7 @@
 import { DBTarget } from './db/operation.js'
 
 import type {
+  BlockchainEvents,
   BlockchainInterface,
   BlockchainOptions,
   Consensus,
@@ -50,6 +52,7 @@
   consensus: Consensus
   db: DB<Uint8Array | string, Uint8Array | string | DBObject>
   dbManager: DBManager
+  events: AsyncEventEmitter<BlockchainEvents>
 
   private _genesisBlock?: Block /** The genesis block of this blockchain */
   private _customGenesisState?: GenesisState /** Custom genesis state */
@@ -80,6 +83,13 @@
   private readonly _validateConsensus: boolean
   private readonly _validateBlocks: boolean
 
+  /**
+   * This is used to track which canonical blocks are deleted. After a method calls
+   * `_deleteCanonicalChainReferences`, if this array has any items, the
+   * `deletedCanonicalBlocks` event is emitted with the array as argument.
+   */
+  private _deletedBlocks: Block[] = []
+
   /**
    * Safe creation of a new Blockchain object awaiting the initialization function,
    * encouraged method to use when creating a blockchain object.
@@ -144,6 +154,8 @@
 
     this.dbManager = new DBManager(this.db, this.common)
 
+    this.events = new AsyncEventEmitter()
+
     if (opts.consensus) {
       this.consensus = opts.consensus
     } else {
@@ -436,6 +448,10 @@
       await this.dbManager.batch(ops)
       await this.checkAndTransitionHardForkByNumber(canonicalHead, td, header.timestamp)
     })
+    if (this._deletedBlocks.length > 0) {
+      this.events.emit('deletedCanonicalBlocks', this._deletedBlocks)
+      this._deletedBlocks = []
+    }
   }
 
   /**
@@ -566,6 +582,10 @@
         throw e
       }
     })
+    if (this._deletedBlocks.length > 0) {
+      this.events.emit('deletedCanonicalBlocks', this._deletedBlocks)
+      this._deletedBlocks = []
+    }
   }
 
   /**
@@ -916,6 +936,11 @@
     }
 
     await this.dbManager.batch(dbOps)
+
+    if (this._deletedBlocks.length > 0) {
+      this.events.emit('deletedCanonicalBlocks', this._deletedBlocks)
+      this._deletedBlocks = []
+    }
   }
 
   /**
@@ -1127,34 +1152,48 @@
     headHash: Uint8Array,
     ops: DBOp[]
   ) {
-    let hash: Uint8Array | false
+    try {
+      let hash: Uint8Array | false
 
-    hash = await this.safeNumberToHash(blockNumber)
-    while (hash !== false) {
-      ops.push(DBOp.del(DBTarget.NumberToHash, { blockNumber }))
+      hash = await this.safeNumberToHash(blockNumber)
+      while (hash !== false) {
+        ops.push(DBOp.del(DBTarget.NumberToHash, { blockNumber }))
 
-      // reset stale iterator heads to current canonical head this can, for
-      // instance, make the VM run "older" (i.e. lower number blocks than last
-      // executed block) blocks to verify the chain up to the current, actual,
-      // head.
-      for (const name of Object.keys(this._heads)) {
-        if (equalsBytes(this._heads[name], hash)) {
-          this._heads[name] = headHash
+        if (this.events.listenerCount('deletedCanonicalBlocks') > 0) {
+          const block = await this.getBlock(blockNumber)
+          this._deletedBlocks.push(block)
         }
-      }
 
-      // reset stale headHeader to current canonical
-      if (this._headHeaderHash !== undefined && equalsBytes(this._headHeaderHash, hash) === true) {
-        this._headHeaderHash = headHash
-      }
-      // reset stale headBlock to current canonical
-      if (this._headBlockHash !== undefined && equalsBytes(this._headBlockHash, hash) === true) {
-        this._headBlockHash = headHash
-      }
+        // reset stale iterator heads to current canonical head this can, for
+        // instance, make the VM run "older" (i.e. lower number blocks than last
+        // executed block) blocks to verify the chain up to the current, actual,
+        // head.
+        for (const name of Object.keys(this._heads)) {
+          if (equalsBytes(this._heads[name], hash)) {
+            this._heads[name] = headHash
+          }
+        }
 
-      blockNumber++
+        // reset stale headHeader to current canonical
+        if (
+          this._headHeaderHash !== undefined &&
+          equalsBytes(this._headHeaderHash, hash) === true
+        ) {
+          this._headHeaderHash = headHash
+        }
+        // reset stale headBlock to current canonical
+        if (this._headBlockHash !== undefined && equalsBytes(this._headBlockHash, hash) === true) {
+          this._headBlockHash = headHash
+        }
 
-      hash = await this.safeNumberToHash(blockNumber)
+        blockNumber++
+
+        hash = await this.safeNumberToHash(blockNumber)
+      }
+    } catch (e) {
+      // Ensure that if this method throws, `_deletedBlocks` is reset to the empty array
+      this._deletedBlocks = []
+      throw e
     }
   }
 

packages/blockchain/src/types.ts

@@ -1,10 +1,14 @@
 import type { Blockchain } from '.'
 import type { Block, BlockHeader } from '@ethereumjs/block'
 import type { Common, ConsensusAlgorithm } from '@ethereumjs/common'
-import type { DB, DBObject, GenesisState } from '@ethereumjs/util'
+import type { AsyncEventEmitter, DB, DBObject, GenesisState } from '@ethereumjs/util'
 
 export type OnBlock = (block: Block, reorg: boolean) => Promise<void> | void
 
+export type BlockchainEvents = {
+  deletedCanonicalBlocks: (data: Block[], resolve?: (result?: any) => void) => void
+}
+
 export interface BlockchainInterface {
   consensus: Consensus
   /**
@@ -79,6 +83,11 @@ export interface BlockchainInterface {
    * Returns the latest full block in the canonical chain.
    */
   getCanonicalHeadBlock(): Promise<Block>
+
+  /**
+   * Optional events emitter
+   */
+  events?: AsyncEventEmitter<BlockchainEvents>
 }
 
 export interface GenesisOptions {

packages/client/src/execution/receipt.ts

@@ -107,6 +107,11 @@ export class ReceiptsManager extends MetaDBManager {
     void this.updateIndex(IndexOperation.Save, IndexType.TxHash, block)
   }
 
+  async deleteReceipts(block: Block) {
+    await this.delete(DBKey.Receipts, block.hash())
+    void this.updateIndex(IndexOperation.Delete, IndexType.TxHash, block)
+  }
+
   /**
    * Returns receipts for given blockHash
    * @param blockHash the block hash

packages/client/src/execution/vmexecution.ts

@@ -23,11 +23,25 @@
 import type { Block } from '@ethereumjs/block'
 import type { RunBlockOpts, TxReceipt } from '@ethereumjs/vm'
 
+export enum ExecStatus {
+  VALID = 'VALID',
+  INVALID = 'INVALID',
+}
+
+type ChainStatus = {
+  height: bigint
+  status: ExecStatus
+  hash: Uint8Array
+  root: Uint8Array
+}
+
 export class VMExecution extends Execution {
   private _lock = new Lock()
 
   public vm: VM
   public hardfork: string = ''
+  /* Whether canonical chain execution has stayed valid or ran into an invalid block */
+  public chainStatus: ChainStatus | null = null
 
   public receiptsManager?: ReceiptsManager
   private pendingReceipts?: Map<string, TxReceipt[]>
@@ -104,6 +118,18 @@
         metaDB: this.metaDB,
       })
       this.pendingReceipts = new Map()
+      this.chain.blockchain.events.addListener(
+        'deletedCanonicalBlocks',
+        async (blocks, resolve) => {
+          // Once a block gets deleted from the chain, delete the receipts also
+          for (const block of blocks) {
+            await this.receiptsManager?.deleteReceipts(block)
+          }
+          if (resolve !== undefined) {
+            resolve()
+          }
+        }
+      )
     }
   }
 
@@ -139,7 +165,14 @@
         throw new Error('cannot get iterator head: blockchain has no getIteratorHead function')
       }
       const headBlock = await this.vm.blockchain.getIteratorHead()
-      const { number, timestamp } = headBlock.header
+      const { number, timestamp, stateRoot } = headBlock.header
+      this.chainStatus = {
+        height: number,
+        status: ExecStatus.VALID,
+        root: stateRoot,
+        hash: headBlock.hash(),
+      }
+
       if (typeof this.vm.blockchain.getTotalDifficulty !== 'function') {
         throw new Error('cannot get iterator head: blockchain has no getTotalDifficulty function')
       }
@@ -258,7 +291,7 @@
       for (const block of blocks) {
         const receipts = this.pendingReceipts?.get(bytesToHex(block.hash()))
         if (receipts) {
-          void this.receiptsManager?.saveReceipts(block, receipts)
+          await this.receiptsManager?.saveReceipts(block, receipts)
           this.pendingReceipts?.delete(bytesToHex(block.hash()))
         }
       }
@@ -274,6 +307,13 @@
         }
       }
       await this.chain.blockchain.setIteratorHead('vm', vmHeadBlock.hash())
+      this.chainStatus = {
+        height: vmHeadBlock.header.number,
+        root: vmHeadBlock.header.stateRoot,
+        hash: vmHeadBlock.hash(),
+        status: ExecStatus.VALID,
+      }
+
       if (safeBlock !== undefined) {
         await this.chain.blockchain.setIteratorHead('safe', safeBlock.hash())
       }
@@ -425,7 +465,7 @@
                     this.config.logger.warn(msg)
                   }
 
-                  void this.receiptsManager?.saveReceipts(block, result.receipts)
+                  await this.receiptsManager?.saveReceipts(block, result.receipts)
 
                   txCounter += block.transactions.length
                   // set as new head block
@@ -445,39 +485,14 @@
             // Ensure to catch and not throw as this would lead to unCaughtException with process exit
             .catch(async (error) => {
               if (errorBlock !== undefined) {
-                // TODO: determine if there is a way to differentiate between the cases
-                // a) a bad block is served by a bad peer -> delete the block and restart sync
-                //    sync from parent block
-                // b) there is a consensus error in the VM -> stop execution until the
-                //    consensus error is fixed
-                //
-                // For now only option b) is implemented, atm this is a very likely case
-                // and the implemented behavior helps on debugging.
-                // Option a) would likely need some block comparison of the same blocks
-                // received by different peer to decide on bad blocks
-                // (minimal solution: receive block from 3 peers and take block if there is
-                // is equally served from at least 2 peers)
-                /*try {
-            // remove invalid block
-              await blockchain!.delBlock(block.header.hash())
-            } catch (error: any) {
-              this.config.logger.error(
-                `Error deleting block number=${blockNumber} hash=${hash} on failed execution`
-              )
-            }
-            this.config.logger.warn(
-              `Deleted block number=${blockNumber} hash=${hash} on failed execution`
-            )
+                // set the chainStatus to invalid
+                this.chainStatus = {
+                  height: errorBlock.header.number,
+                  root: errorBlock.header.stateRoot,
+                  hash: errorBlock.hash(),
+                  status: ExecStatus.INVALID,
+                }
 
-            const hardfork = this.config.execCommon.getHardforkBy({ blockNumber })
-            if (hardfork !== this.hardfork) {
-              this.config.logger.warn(
-                `Set back hardfork along block deletion number=${blockNumber} hash=${hash} old=${this.hardfork} new=${hardfork}`
-              )
-              this.config.execCommon.setHardforkBy({ blockNumber, td })
-            }*/
-                // Option a): set iterator head to the parent block so that an
-                // error can repeatedly processed for debugging
                 const { number } = errorBlock.header
                 const hash = short(errorBlock.hash())
                 const errorMsg = `Execution of block number=${number} hash=${hash} hardfork=${this.hardfork} failed`