(Requires Audit) PLEX-2473 LP Sparce Block Store (Part 3)

CONFIG.md

@@ -28,6 +28,7 @@ FlagsContractAddress = '0xae4E781a6218A8031764928E88d457937A954fC3' # Example
 LinkContractAddress = '0x538aAaB4ea120b2bC2fe5D296852D948F07D849e' # Example
 LogBackfillBatchSize = 1000 # Default
 LogPollInterval = '15s' # Default
+LogPollerSkipEmptyBlocks = false # Default
 LogKeepBlocksDepth = 100000 # Default
 LogPrunePageSize = 0 # Default
 BackupLogPollerBlockDelay = 100 # Default
@@ -172,6 +173,13 @@ LogPollInterval = '15s' # Default
 ```
 LogPollInterval works in conjunction with Feature.LogPoller. Controls how frequently the log poller polls for logs. Defaults to the block production rate.
 
+### LogPollerSkipEmptyBlocks
+:warning: **_ADVANCED_**: _Do not change this setting unless you know what you are doing._
+```toml
+LogPollerSkipEmptyBlocks = false # Default
+```
+LogPollerSkipEmptyBlocks works in conjunction with Feature.LogPoller. Controls whether the log poller skips blocks with no logs. Setting this to true can reduce DB load, however it can affect product performance. Consult with specific Chainlink product team before changing this setting.
+
 ### LogKeepBlocksDepth
 :warning: **_ADVANCED_**: _Do not change this setting unless you know what you are doing._
 ```toml

pkg/chains/legacyevm/chain.go

@@ -246,6 +246,7 @@ func newChain(cfg *config.ChainScoped, nodes []*toml.Node, opts ChainRelayOpts,
 				LogPrunePageSize:         int64(cfg.EVM().LogPrunePageSize()),
 				BackupPollerBlockDelay:   int64(cfg.EVM().BackupLogPollerBlockDelay()),
 				ClientErrors:             cfg.EVM().NodePool().Errors(),
+				SkipEmptyBlocks:          cfg.EVM().LogPollerSkipEmptyBlocks(),
 			}
 
 			lpORM, err := logpoller.NewObservedORM(chainID, opts.DS, l)

pkg/config/chain_scoped.go

@@ -95,6 +95,10 @@ func (e *EVMConfig) LogPollInterval() time.Duration {
 	return e.C.LogPollInterval.Duration()
 }
 
+func (e *EVMConfig) LogPollerSkipEmptyBlocks() bool {
+	return *e.C.LogPollerSkipEmptyBlocks
+}
+
 func (e *EVMConfig) FinalityDepth() uint32 {
 	return *e.C.FinalityDepth
 }

pkg/config/config.go

@@ -42,6 +42,7 @@ type EVM interface {
 	LogKeepBlocksDepth() uint32
 	BackupLogPollerBlockDelay() uint64
 	LogPollInterval() time.Duration
+	LogPollerSkipEmptyBlocks() bool
 	LogPrunePageSize() uint32
 	MinContractPayment() *commonassets.Link
 	MinIncomingConfirmations() uint32

pkg/config/toml/config.go

@@ -403,6 +403,7 @@ type Chain struct {
 	LinkContractAddress          *types.EIP55Address
 	LogBackfillBatchSize         *uint32
 	LogPollInterval              *commonconfig.Duration
+	LogPollerSkipEmptyBlocks     *bool
 	LogKeepBlocksDepth           *uint32
 	LogPrunePageSize             *uint32
 	BackupLogPollerBlockDelay    *uint64

pkg/config/toml/config_test.go

@@ -276,6 +276,7 @@ var fullConfig = EVMConfig{
 		LinkContractAddress:          ptr(types.MustEIP55Address("0x538aAaB4ea120b2bC2fe5D296852D948F07D849e")),
 		LogBackfillBatchSize:         ptr[uint32](17),
 		LogPollInterval:              config.MustNewDuration(time.Minute),
+		LogPollerSkipEmptyBlocks:     ptr(false),
 		LogKeepBlocksDepth:           ptr[uint32](100000),
 		LogPrunePageSize:             ptr[uint32](0),
 		BackupLogPollerBlockDelay:    ptr[uint64](532),

pkg/config/toml/defaults.go

@@ -229,6 +229,9 @@ func (c *Chain) SetFrom(f *Chain) {
 	if v := f.LogPollInterval; v != nil {
 		c.LogPollInterval = v
 	}
+	if v := f.LogPollerSkipEmptyBlocks; v != nil {
+		c.LogPollerSkipEmptyBlocks = v
+	}
 	if v := f.LogKeepBlocksDepth; v != nil {
 		c.LogKeepBlocksDepth = v
 	}

pkg/config/toml/defaults/fallback.toml

@@ -7,6 +7,7 @@ FinalityTagEnabled = false
 SafeTagSupported = true
 LogBackfillBatchSize = 1000
 LogPollInterval = '15s'
+LogPollerSkipEmptyBlocks = false
 LogKeepBlocksDepth = 100000
 LogPrunePageSize = 0
 BackupLogPollerBlockDelay = 100

pkg/config/toml/docs.toml

@@ -71,6 +71,9 @@ LogBackfillBatchSize = 1000 # Default
 # LogPollInterval works in conjunction with Feature.LogPoller. Controls how frequently the log poller polls for logs. Defaults to the block production rate.
 LogPollInterval = '15s' # Default
 # **ADVANCED**
+# LogPollerSkipEmptyBlocks works in conjunction with Feature.LogPoller. Controls whether the log poller skips blocks with no logs. Setting this to true can reduce DB load, however it can affect product performance. Consult with specific Chainlink product team before changing this setting.
+LogPollerSkipEmptyBlocks = false # Default
+# **ADVANCED**
 # LogKeepBlocksDepth works in conjunction with Feature.LogPoller. Controls how many blocks the poller will keep, must be greater than FinalityDepth+1.
 LogKeepBlocksDepth = 100000 # Default
 # **ADVANCED**

pkg/config/toml/testdata/config-full.toml

@@ -12,6 +12,7 @@ FlagsContractAddress = '0xae4E781a6218A8031764928E88d457937A954fC3'
 LinkContractAddress = '0x538aAaB4ea120b2bC2fe5D296852D948F07D849e'
 LogBackfillBatchSize = 17
 LogPollInterval = '1m0s'
+LogPollerSkipEmptyBlocks = false
 LogKeepBlocksDepth = 100000
 LogPrunePageSize = 0
 BackupLogPollerBlockDelay = 532

pkg/logpoller/log_poller.go

@@ -119,11 +119,12 @@ type logPoller struct {
 	latencyMonitor           LatencyMonitor
 	lggr                     logger.SugaredLogger
 	pollPeriod               time.Duration // poll period set by block production rate
-	useFinalityTag           bool          // indicates whether logPoller should use chain's finality or pick a fixed depth for finality
-	finalityDepth            int64         // finality depth is taken to mean that block (head - finality) is finalized. If `useFinalityTag` is set to true, this value is ignored, because finalityDepth is fetched from chain
-	keepFinalizedBlocksDepth int64         // the number of blocks behind the last finalized block we keep in database
-	backfillBatchSize        int64         // batch size to use when backfilling finalized logs
-	rpcBatchSize             int64         // batch size to use for fallback RPC calls made in GetBlocks
+	skipEmptyBlocks          bool
+	useFinalityTag           bool  // indicates whether logPoller should use chain's finality or pick a fixed depth for finality
+	finalityDepth            int64 // finality depth is taken to mean that block (head - finality) is finalized. If `useFinalityTag` is set to true, this value is ignored, because finalityDepth is fetched from chain
+	keepFinalizedBlocksDepth int64 // the number of blocks behind the last finalized block we keep in database
+	backfillBatchSize        int64 // batch size to use when backfilling finalized logs
+	rpcBatchSize             int64 // batch size to use for fallback RPC calls made in GetBlocks
 	logPrunePageSize         int64
 	clientErrors             config.ClientErrors
 	backupPollerNextBlock    int64 // next block to be processed by Backup LogPoller
@@ -159,6 +160,7 @@ type Opts struct {
 	BackupPollerBlockDelay   int64
 	LogPrunePageSize         int64
 	ClientErrors             config.ClientErrors
+	SkipEmptyBlocks          bool
 }
 
 // NewLogPoller creates a log poller. Note there is an assumption
@@ -182,6 +184,7 @@ func NewLogPoller(orm ORM, ec Client, lggr logger.Logger, headTracker HeadTracke
 		replayStart:              make(chan int64),
 		replayComplete:           make(chan error),
 		pollPeriod:               opts.PollPeriod,
+		skipEmptyBlocks:          opts.SkipEmptyBlocks,
 		backupPollerBlockDelay:   opts.BackupPollerBlockDelay,
 		finalityDepth:            opts.FinalityDepth,
 		useFinalityTag:           opts.UseFinalityTag,
@@ -1090,7 +1093,7 @@ func (lp *logPoller) handleReorg(ctx context.Context, currentBlock *evmtypes.Hea
 	// There can be another reorg while we're finding the LCA.
 	// That is ok, since we'll detect it on the next iteration.
 	// Since we go currentBlock by currentBlock for unfinalized logs, the mismatch starts at currentBlockNumber - 1.
-	blockAfterLCA, err2 := lp.findBlockAfterLCA(ctx, currentBlock, latestBlock.FinalizedBlockNumber)
+	blockAfterLCA, err2 := lp.findBlockAfterLCA(ctx, currentBlock.Number, latestBlock.FinalizedBlockNumber)
 	if err2 != nil {
 		return nil, fmt.Errorf("unable to find LCA after reorg: %w", err2)
 	}
@@ -1243,10 +1246,15 @@ func (e *reorgError) Error() string {
 	return fmt.Sprintf("reorg detected at block %d", e.ReorgedAt.Number)
 }
 
-func (lp *logPoller) getUnfinalizedLogs(ctx context.Context, currentBlock *evmtypes.Head, latest, safe, finalized int64, isReplay bool) ([]Block, []Log, error) {
+func (lp *logPoller) getUnfinalizedLogs(ctx context.Context, currentBlock *evmtypes.Head, latest, safe, finalized int64, isReplay bool) (blocks []Block, logs []Log, err error) {
 	const maxUnfinalizedBlocks = 2000
-	var logs []Log
-	var blocks []Block
+	var block *Block
+	defer func() {
+		// ensure that we always include the last block even if it's empty to use it as check point for next poll.
+		if block != nil && (len(blocks) == 0 || blocks[len(blocks)-1].BlockNumber != block.BlockNumber) {
+			blocks = append(blocks, *block)
+		}
+	}()
 	for {
 		h := currentBlock.Hash
 		rpcLogs, err := lp.latencyMonitor.FilterLogs(ctx, lp.Filter(nil, nil, &h))
@@ -1255,15 +1263,18 @@ func (lp *logPoller) getUnfinalizedLogs(ctx context.Context, currentBlock *evmty
 			return blocks, logs, fmt.Errorf("unable to query for logs: %w", err)
 		}
 		lp.lggr.Debugw("Unfinalized log query", "logs", len(logs), "currentBlockNumber", currentBlock.Number, "blockHash", currentBlock.Hash, "timestamp", currentBlock.Timestamp)
-		block := Block{
+		block = &Block{
 			BlockHash:            h,
 			BlockNumber:          currentBlock.Number,
 			BlockTimestamp:       currentBlock.Timestamp,
 			FinalizedBlockNumber: finalized,
 			SafeBlockNumber:      safe,
 		}
-		logs = append(logs, convertLogs(rpcLogs, []Block{block}, lp.lggr, lp.ec.ConfiguredChainID())...)
-		blocks = append(blocks, block)
+		logs = append(logs, convertLogs(rpcLogs, []Block{*block}, lp.lggr, lp.ec.ConfiguredChainID())...)
+		// Skip empty blocks if configured to do so.
+		if len(rpcLogs) > 0 || !lp.skipEmptyBlocks {
+			blocks = append(blocks, *block)
+		}
 
 		if currentBlock.Number >= latest {
 			return blocks, logs, nil
@@ -1340,39 +1351,50 @@ func (lp *logPoller) latestSafeBlock(ctx context.Context, latestFinalizedBlockNu
 
 // Find the first place where our chain and their chain have the same block,
 // that block number is the LCA. Return the block after that, where we want to resume polling.
-func (lp *logPoller) findBlockAfterLCA(ctx context.Context, current *evmtypes.Head, latestFinalizedBlockNumber int64) (*evmtypes.Head, error) {
-	// Current is where the mismatch starts.
-	// Check its parent to see if its the same as ours saved.
-	parent, err := lp.latencyMonitor.HeadByHash(ctx, current.ParentHash)
-	if err != nil {
-		return nil, err
+func (lp *logPoller) findBlockAfterLCA(ctx context.Context, currentHeadNumber int64, dbLatestFinalizedBlockNumber int64) (*evmtypes.Head, error) {
+	if currentHeadNumber < dbLatestFinalizedBlockNumber {
+		lp.lggr.Criticalw("Unexpected state. Current head number is lower than latest finalized block number", "currentHeadNumber", currentHeadNumber, "dbLatestFinalizedBlockNumber", dbLatestFinalizedBlockNumber)
+		return nil, fmt.Errorf("current head number %d is lower than latest finalized block number %d: %w", currentHeadNumber, dbLatestFinalizedBlockNumber, commontypes.ErrFinalityViolated)
 	}
-	blockAfterLCA := current
+
 	// We expect reorgs up to the block after latestFinalizedBlock
-	// We loop via parent instead of current so current always holds the LCA+1.
 	// If the parent block number becomes < the first finalized block our reorg is too deep.
-	// This can happen only if finalityTag is not enabled and fixed finalityDepth is provided via config.
-	var ourParentBlockHash common.Hash
-	for parent.Number >= latestFinalizedBlockNumber {
-		outParentBlock, err := lp.orm.SelectBlockByNumber(ctx, parent.Number)
+	// This can happen only if finalityTag is not enabled and fixed finalityDepth is provided via config or chain violates finality guarantees.
+	for {
+		// Since we do not store all blocks in the db, it's possible that we do not have the parent block in our db.
+		// Find the nearest ancestor that we have in our db and check if it still belongs to canonical chain.
+		ourParent, err := lp.orm.SelectNewestBlock(ctx, currentHeadNumber-1)
 		if err != nil {
-			return nil, err
-		}
-		ourParentBlockHash = outParentBlock.BlockHash
-		if parent.Hash == ourParentBlockHash {
-			// If we do have the blockhash, return blockAfterLCA
-			return blockAfterLCA, nil
+			if errors.Is(err, sql.ErrNoRows) {
+				lp.lggr.Warnw("No ancestor block found in db, this means that the reorg is deeper than the number of blocks we have in the db.", "err", err, "currentHeadNumber", currentHeadNumber, "dbLatestFinalizedBlockNumber", dbLatestFinalizedBlockNumber)
+				// we should return currentHeadNumber as the block after LCA, to avoid drifting too far back.
+				return lp.headerByNumber(ctx, currentHeadNumber)
+			}
+			return nil, fmt.Errorf("failed to select ancestor for current block %d: %w", currentHeadNumber-1, err)
 		}
-		// Otherwise get a new parent and update blockAfterLCA.
-		blockAfterLCA = parent
-		parent, err = lp.latencyMonitor.HeadByHash(ctx, parent.ParentHash)
+
+		// Since we are looking for block after LCA, fetch child of ourParent.
+		// If new current points to ourParent, we found the LCA and can return block after it. Otherwise, keep looking for ancestors.
+		rpcChild, err := lp.headerByNumber(ctx, ourParent.BlockNumber+1)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("failed to fetch block after ancestor %d: %w", ourParent.BlockNumber+1, err)
+		}
+		if ourParent.BlockHash == rpcChild.ParentHash {
+			return rpcChild, nil
+		}
+
+		if ourParent.BlockNumber <= dbLatestFinalizedBlockNumber {
+			lp.lggr.Criticalw("Reorg greater than finality depth detected", "finalityTag", lp.useFinalityTag,
+				"current", rpcChild.Number,
+				"latestFinalized", dbLatestFinalizedBlockNumber,
+				"ourParentHash", ourParent.BlockHash,
+				"expectedParentHash", rpcChild.ParentHash,
+				"childHash", rpcChild.Hash)
+			return nil, fmt.Errorf("%w: finalized block with hash %s is not parent of canonical block at height %d, with parent hash %s", commontypes.ErrFinalityViolated, ourParent.BlockHash, rpcChild.Number, rpcChild.ParentHash)
 		}
-	}
 
-	lp.lggr.Criticalw("Reorg greater than finality depth detected", "finalityTag", lp.useFinalityTag, "current", current.Number, "latestFinalized", latestFinalizedBlockNumber)
-	return nil, fmt.Errorf("%w: finalized block hash %s does not match RPC's %s at height %d", commontypes.ErrFinalityViolated, ourParentBlockHash, blockAfterLCA.Hash, blockAfterLCA.Number)
+		currentHeadNumber = ourParent.BlockNumber
+	}
 }
 
 // PruneOldBlocks removes blocks that are > lp.keepFinalizedBlocksDepth behind the latest finalized block.