Fix Replay Reorg Detection Bypassed When Parent Block Is Missing (#422)

* Test: reorg during replay

* Fix reorg handling during replay

* Fix nits

* ORM methods to batch insert

* switch to batch logs/blocks inserts

* nit fixes

* switch ubig to sqlutil big

* fix orm_test

* Refactor tests in preparation for sparse blocks

* Store only blocks with logs and checkpoints

* Add feature flag to disable/enable sparse blocks storage

* Fix Replay Reorg Detection Bypassed When Parent Block Is Missing

* Fix flaky test

* Improve backfill tests

* Fix linter issues

Author: dhaidashenko

pkg/.mockery.yaml

@@ -92,6 +92,13 @@ packages:
   github.com/smartcontractkit/chainlink-evm/pkg/logpoller:
     interfaces:
       LogPoller:
+      ORM:
+        config:
+          dir: "{{ .InterfaceDir }}"
+          outpkg: logpoller
+          filename: "mock_orm_test.go"
+          mockname: "MockORM"
+          inpackage: true
   github.com/smartcontractkit/chainlink-evm/pkg/round:
     interfaces:
       RequestRoundDB:

pkg/logpoller/helper_test.go

@@ -176,16 +176,6 @@ func (th *TestHarness) assertDontHave(t *testing.T, start, end int) {
 	}
 }
 
-func (th *TestHarness) assertHaveCanonical(t *testing.T, start, end int) {
-	for i := start; i < end; i++ {
-		blk, err := th.ORM.SelectBlockByNumber(testutils.Context(t), int64(i))
-		require.NoError(t, err, "block %v", i)
-		chainBlk, err := th.Client.BlockByNumber(testutils.Context(t), big.NewInt(int64(i)))
-		require.NoError(t, err)
-		assert.Equal(t, chainBlk.Hash().String(), blk.BlockHash.String(), "block %v", i)
-	}
-}
-
 // Simulates an RPC failover event to an alternate rpc server. This can also be used to
 // simulate switching back to the primary rpc after it recovers.
 func (th *TestHarness) SetActiveClient(backend evmtypes.Backend, chainType chaintype.ChainType) {

pkg/logpoller/log_poller.go

@@ -1005,37 +1005,55 @@ func (lp *logPoller) headerByNumber(ctx context.Context, blockNumber int64) (*ev
 // 2. Delete all logs and blocks after the LCA
 // 3. Return the LCA+1, i.e. our new current (unprocessed) block.
 func (lp *logPoller) getCurrentBlockMaybeHandleReorg(ctx context.Context, currentBlockNumber int64, currentBlock *evmtypes.Head, isReplay bool) (head *evmtypes.Head, err error) {
-	var err1 error
 	if currentBlock == nil {
-		currentBlock, err1 = lp.headerByNumber(ctx, currentBlockNumber)
-		if err1 != nil {
-			return nil, fmt.Errorf("unable to get current block header for block number %d: %w", currentBlockNumber, err1)
+		currentBlock, err = lp.headerByNumber(ctx, currentBlockNumber)
+		if err != nil {
+			return nil, fmt.Errorf("unable to get current block header for block number %d: %w", currentBlockNumber, err)
 		}
 	}
 	// Does this currentBlock point to the same parent that we have saved?
 	// If not, there was a reorg, so we need to rewind.
-	expectedParent, err1 := lp.orm.SelectBlockByNumber(ctx, currentBlockNumber-1)
-	if err1 != nil && !pkgerrors.Is(err1, sql.ErrNoRows) {
+	expectedParent, err := lp.orm.SelectBlockByNumber(ctx, currentBlockNumber-1)
+	if err != nil && !pkgerrors.Is(err, sql.ErrNoRows) {
 		// If err is not a 'no rows' error, assume transient db issue and retry
-		lp.lggr.Warnw("Unable to read latestBlockNumber currentBlock saved", "err", err1, "currentBlockNumber", currentBlockNumber)
+		lp.lggr.Warnw("Unable to read latestBlockNumber currentBlock saved", "err", err, "currentBlockNumber", currentBlockNumber)
 		return nil, pkgerrors.New("Unable to read latestBlockNumber currentBlock saved")
 	}
 	// We will not have the previous currentBlock on initial poll.
-	havePreviousBlock := err1 == nil
-	if !havePreviousBlock {
-		lp.lggr.Infow("Do not have previous block, first poll ever on new chain", "currentBlockNumber", currentBlockNumber)
-		return currentBlock, nil
-	}
-	// Check for reorg.
-	if currentBlock.ParentHash != expectedParent.BlockHash {
-		return lp.handleReorg(ctx, currentBlock)
-	}
+	havePreviousBlock := err == nil
+	if havePreviousBlock {
+		// Check for reorg.
+		if currentBlock.ParentHash != expectedParent.BlockHash {
+			return lp.handleReorg(ctx, currentBlock)
+		}
 
-	if !isReplay {
-		// During normal polling DB does not have any blocks after currentBlockNumber, so no reorg is possible. We can skip extra checks and just return currentBlock.
-		return currentBlock, nil
+		if !isReplay {
+			// During normal polling DB does not have any blocks after currentBlockNumber, so no reorg is possible. We can skip extra checks and just return currentBlock.
+			return currentBlock, nil
+		}
+	} else {
+		// previous block is not present, it could only in case of initial poll or replay.
+		// If that's not a replay, let's double-check that it's an initial poll by verifying that the DB is empty.
+		if !isReplay {
+			_, err = lp.orm.SelectLatestBlock(ctx)
+			if err == nil {
+				err = fmt.Errorf("unexpected state: no previous block found for block number %d, but db is not empty", currentBlockNumber)
+				lp.lggr.Criticalw("Invariant violation: expected to always have previous block except replay and first poll for a new chain", "currentBlockNumber", currentBlockNumber, "err", err)
+				return nil, err
+			}
+
+			if !errors.Is(err, sql.ErrNoRows) {
+				// Real DB error
+				return nil, fmt.Errorf("failed to check for first poll ever on new chain: %w", err)
+			}
+
+			// This is expected, we just don't have any blocks in the DB yet.
+			lp.lggr.Infow("Do not have previous block, first poll ever on new chain", "currentBlockNumber", currentBlockNumber)
+			return currentBlock, nil
+		}
 	}
 
+	// In case of replay currentBlock may be in the middle of DB range, lets do additional checks to handle possible reorgs.
 	// Ensure that if DB contains current block it matches the current block from RPC.
 	currentBlockDB, err := lp.orm.SelectBlockByNumber(ctx, currentBlockNumber)
 	if err != nil && !pkgerrors.Is(err, sql.ErrNoRows) {
@@ -1047,12 +1065,12 @@ func (lp *logPoller) getCurrentBlockMaybeHandleReorg(ctx context.Context, curren
 	}
 
 	// No reorg for current block, but during replay it's possible that current block is older than the latest block, let's check it too to avoid false positives on finality violation.
-	latestBlockDB, err1 := lp.orm.SelectLatestBlock(ctx)
-	if err1 != nil {
-		if pkgerrors.Is(err1, sql.ErrNoRows) {
-			lp.lggr.Criticalw("Unexpected state. Expected at least one block to be present in the db when checking for reorg during replay, but got no rows", "currentBlockNumber", currentBlockNumber, "err", err1)
+	latestBlockDB, err := lp.orm.SelectLatestBlock(ctx)
+	if err != nil {
+		if pkgerrors.Is(err, sql.ErrNoRows) {
+			lp.lggr.Criticalw("Unexpected state. Expected at least one block to be present in the db when checking for reorg during replay, but got no rows", "currentBlockNumber", currentBlockNumber, "err", err)
 		}
-		return nil, pkgerrors.Wrap(err1, "unable to get latest block")
+		return nil, pkgerrors.Wrap(err, "unable to get latest block")
 	}
 
 	if currentBlock.BlockNumber() >= latestBlockDB.BlockNumber {