Check, validate and skip if we got any deposit events already present in DB

packages/cli/src/options/beaconNodeOptions/eth1.ts

@@ -7,6 +7,7 @@ export interface IEth1Args {
   "eth1.providerUrls": string[];
   "eth1.depositContractDeployBlock": number;
   "eth1.disableEth1DepositDataTracker": boolean;
+  "eth1.unsafeAllowDepositDataOverwrite": boolean;
 }
 
 export function parseArgs(args: IEth1Args): IBeaconNodeOptions["eth1"] {
@@ -22,6 +23,7 @@ export function parseArgs(args: IEth1Args): IBeaconNodeOptions["eth1"] {
     providerUrls: providerUrls,
     depositContractDeployBlock: args["eth1.depositContractDeployBlock"],
     disableEth1DepositDataTracker: args["eth1.disableEth1DepositDataTracker"],
+    unsafeAllowDepositDataOverwrite: args["eth1.unsafeAllowDepositDataOverwrite"],
   };
 }
 
@@ -61,4 +63,13 @@ export const options: ICliCommandOptions<IEth1Args> = {
     defaultDescription: String(defaultOptions.eth1.disableEth1DepositDataTracker),
     group: "eth1",
   },
+
+  "eth1.unsafeAllowDepositDataOverwrite": {
+    hidden: true,
+    description:
+      "Allow the deposit tracker to overwrite previously fetched and saved deposit event data. Warning!!! This is an unsafe operation, so enable this flag only if you know what you are doing.",
+    type: "boolean",
+    defaultDescription: String(defaultOptions.eth1.unsafeAllowDepositDataOverwrite),
+    group: "eth1",
+  },
 };

packages/cli/test/unit/options/beaconNodeOptions.test.ts

@@ -24,6 +24,7 @@ describe("options / beaconNodeOptions", () => {
       "eth1.providerUrls": ["http://my.node:8545"],
       "eth1.depositContractDeployBlock": 1625314,
       "eth1.disableEth1DepositDataTracker": true,
+      "eth1.unsafeAllowDepositDataOverwrite": false,
 
       "execution.urls": ["http://localhost:8550"],
       "execution.timeout": 12000,
@@ -78,6 +79,7 @@ describe("options / beaconNodeOptions", () => {
         providerUrls: ["http://my.node:8545"],
         depositContractDeployBlock: 1625314,
         disableEth1DepositDataTracker: true,
+        unsafeAllowDepositDataOverwrite: false,
       },
       executionEngine: {
         urls: ["http://localhost:8550"],

packages/lodestar/src/eth1/errors.ts

@@ -21,6 +21,8 @@ export enum Eth1ErrorCode {
   DUPLICATE_DISTINCT_LOG = "ETH1_ERROR_DUPLICATE_DISTINCT_LOG",
   /** Attempted to insert a log with index != prev + 1 into the Eth1DepositsCache */
   NON_CONSECUTIVE_LOGS = "ETH1_ERROR_NON_CONSECUTIVE_LOGS",
+  /** Expected a deposit log in the db for the index, missing log implies a corrupted db */
+  MISSING_DEPOSIT_LOG = "ETH1_ERROR_MISSING_DEPOSIT_LOG",
 }
 
 export type Eth1ErrorType =
@@ -31,7 +33,8 @@ export type Eth1ErrorType =
   | {code: Eth1ErrorCode.NO_DEPOSITS_FOR_BLOCK_RANGE; fromBlock: number; toBlock: number}
   | {code: Eth1ErrorCode.NO_DEPOSIT_ROOT; depositCount: number}
   | {code: Eth1ErrorCode.NOT_ENOUGH_DEPOSIT_ROOTS; index: number; treeLength: number}
-  | {code: Eth1ErrorCode.DUPLICATE_DISTINCT_LOG; newIndex: number; prevIndex: number}
-  | {code: Eth1ErrorCode.NON_CONSECUTIVE_LOGS; newIndex: number; prevIndex: number};
+  | {code: Eth1ErrorCode.DUPLICATE_DISTINCT_LOG; newIndex: number; lastLogIndex: number}
+  | {code: Eth1ErrorCode.NON_CONSECUTIVE_LOGS; newIndex: number; lastLogIndex: number}
+  | {code: Eth1ErrorCode.MISSING_DEPOSIT_LOG; newIndex: number; lastLogIndex: number};
 
 export class Eth1Error extends LodestarError<Eth1ErrorType> {}

packages/lodestar/src/eth1/eth1DepositDataTracker.ts

@@ -54,7 +54,7 @@ export class Eth1DepositDataTracker {
     this.signal = signal;
     this.logger = logger;
     this.eth1Provider = eth1Provider;
-    this.depositsCache = new Eth1DepositsCache(config, db);
+    this.depositsCache = new Eth1DepositsCache(opts, config, db);
     this.eth1DataCache = new Eth1DataCache(config, db);
     this.lastProcessedDepositBlockNumber = null;
 

packages/lodestar/src/eth1/eth1DepositsCache.ts

@@ -1,19 +1,23 @@
 import {phase0, ssz} from "@chainsafe/lodestar-types";
+import {byteArrayEquals} from "@chainsafe/ssz";
 import {IFilterOptions} from "@chainsafe/lodestar-db";
 import {IChainForkConfig} from "@chainsafe/lodestar-config";
+
 import {IBeaconDb} from "../db";
 import {getEth1DataForBlocks} from "./utils/eth1Data";
 import {assertConsecutiveDeposits} from "./utils/eth1DepositEvent";
 import {getDepositsWithProofs} from "./utils/deposits";
 import {Eth1Error, Eth1ErrorCode} from "./errors";
 
 export class Eth1DepositsCache {
+  unsafeAllowDepositDataOverwrite: boolean;
   db: IBeaconDb;
   config: IChainForkConfig;
 
-  constructor(config: IChainForkConfig, db: IBeaconDb) {
+  constructor(opts: {unsafeAllowDepositDataOverwrite: boolean}, config: IChainForkConfig, db: IBeaconDb) {
     this.config = config;
     this.db = db;
+    this.unsafeAllowDepositDataOverwrite = opts.unsafeAllowDepositDataOverwrite;
   }
 
   /**
@@ -38,14 +42,54 @@ export class Eth1DepositsCache {
 
     const lastLog = await this.db.depositEvent.lastValue();
     const firstEvent = depositEvents[0];
+
+    // Check, validate and skip if we got any deposit events already present in DB
+    // This can happen if the remote eth1/EL resets its head in these four scenarios:
+    //   1. Remote eth1/EL resynced/restarted from head behind its previous head pre-merge
+    //   2. In a post merge scenario, Lodestar restarted from finalized state from DB which
+    //      generally is a few epochs behind the last synced head. This causes eth1 tracker to reset
+    //      and refetch the deposits as the lodestar syncs further along (Post merge there is 1-1
+    //      correspondence between EL and CL blocks)
+    //   3. The EL reorged beyond the eth1 follow distance.
+    //
+    // While 1. & 2. are benign and we handle them below by checking if the duplicate log fetched
+    // is same as one written in DB. Refer to this issue for some data dump of how this happens
+    // https://github.com/ChainSafe/lodestar/issues/3674
+    //
+    // If the duplicate log fetched is not same as written in DB then its probablu scenario 3.
+    // which would be a catastrophic event for the network (or we messed up real bad!!!).
+    //
+    // So we provide for a way to overwrite this log without deleting full db via
+    // --unsafeAllowDepositDataOverwrite cli flag which will just overwrite the previous tracker data
+    // if any. This option as indicated by its name is unsafe and to be only used if you know what
+    // you are doing.
     if (lastLog !== null && firstEvent !== undefined) {
       const newIndex = firstEvent.index;
-      const prevIndex = lastLog.index;
-      if (newIndex <= prevIndex) {
-        throw new Eth1Error({code: Eth1ErrorCode.DUPLICATE_DISTINCT_LOG, newIndex, prevIndex});
-      }
-      if (newIndex > prevIndex + 1) {
-        throw new Eth1Error({code: Eth1ErrorCode.NON_CONSECUTIVE_LOGS, newIndex, prevIndex});
+      const lastLogIndex = lastLog.index;
+
+      if (!this.unsafeAllowDepositDataOverwrite && firstEvent.index <= lastLog.index) {
+        // lastLogIndex - newIndex + 1 events are duplicate since this is a consecutive log
+        // as asserted by assertConsecutiveDeposits. Splice those events out from depositEvents.
+        const skipEvents = depositEvents.splice(0, lastLogIndex - newIndex + 1);
+        // After splicing skipEvents will contain duplicate events to be checked and validated
+        // and rest of the remaining events in depositEvents could be safely written to DB and
+        // move the tracker along.
+        for (const depositEvent of skipEvents) {
+          const prevDBSerializedEvent = await this.db.depositEvent.getBinary(depositEvent.index);
+          if (!prevDBSerializedEvent) {
+            throw new Eth1Error({code: Eth1ErrorCode.MISSING_DEPOSIT_LOG, newIndex, lastLogIndex});
+          }
+          const serializedEvent = ssz.phase0.DepositEvent.serialize(depositEvent);
+          if (!byteArrayEquals(prevDBSerializedEvent, serializedEvent)) {
+            throw new Eth1Error({code: Eth1ErrorCode.DUPLICATE_DISTINCT_LOG, newIndex, lastLogIndex});
+          }
+        }
+      } else if (newIndex > lastLogIndex + 1) {
+        // deposit events need to be consective, the way we fetch our tracker. If the deposit event
+        // is not consecutive it means either our tracker, or the corresponding eth1/EL
+        // node or the database has messed up. All these failures are critical and the tracker
+        // shouldn't proceed without the resolution of this error.
+        throw new Eth1Error({code: Eth1ErrorCode.NON_CONSECUTIVE_LOGS, newIndex, lastLogIndex});
       }
     }
 

packages/lodestar/src/eth1/options.ts

@@ -3,10 +3,12 @@ export type Eth1Options = {
   disableEth1DepositDataTracker?: boolean;
   providerUrls: string[];
   depositContractDeployBlock?: number;
+  unsafeAllowDepositDataOverwrite: boolean;
 };
 
 export const defaultEth1Options: Eth1Options = {
   enabled: true,
   providerUrls: ["http://localhost:8545"],
   depositContractDeployBlock: 0,
+  unsafeAllowDepositDataOverwrite: false,
 };

packages/lodestar/test/e2e/eth1/eth1ForBlockProduction.test.ts

@@ -66,6 +66,7 @@ describe("eth1 / Eth1Provider", function () {
       enabled: true,
       providerUrls: [getGoerliRpcUrl()],
       depositContractDeployBlock: medallaTestnetConfig.depositBlock,
+      unsafeAllowDepositDataOverwrite: false,
     };
     const eth1Provider = new Eth1Provider(config, eth1Options, controller.signal);
 

packages/lodestar/test/e2e/eth1/eth1MergeBlockTracker.test.ts

@@ -37,6 +37,7 @@ describe("eth1 / Eth1MergeBlockTracker", function () {
       enabled: true,
       providerUrls: [getGoerliRpcUrl()],
       depositContractDeployBlock: 0,
+      unsafeAllowDepositDataOverwrite: false,
     };
   });
 

packages/lodestar/test/e2e/eth1/eth1Provider.test.ts

@@ -24,6 +24,7 @@ describe("eth1 / Eth1Provider", function () {
       enabled: true,
       providerUrls: [getGoerliRpcUrl()],
       depositContractDeployBlock: 0,
+      unsafeAllowDepositDataOverwrite: false,
     };
     return new Eth1Provider(config, eth1Options, controller.signal);
   }

packages/lodestar/test/e2e/eth1/stream.test.ts

@@ -22,6 +22,7 @@ describe("Eth1 streams", function () {
       enabled: true,
       providerUrls: [getGoerliRpcUrl()],
       depositContractDeployBlock: 0,
+      unsafeAllowDepositDataOverwrite: false,
     };
     return new Eth1Provider(config, eth1Options, controller.signal);
   }