Added config options to control HeadTracker's support of finality tags

.changeset/sour-owls-grab.md

@@ -0,0 +1,7 @@
+---
+"chainlink": patch
+---
+
+Added config option `HeadTracker.FinalityTagBypass` to force `HeadTracker` to track blocks up to `FinalityDepth` even if `FinalityTagsEnabled = true`. This option is a temporary measure to address high CPU usage on chains with extremely large actual finality depth (gap between the current head and the latest finalized block). #added
+
+Added config option `HeadTracker.MaxAllowedFinalityDepth` maximum gap between current head to the latest finalized block that `HeadTracker` considers healthy. #added

common/headtracker/head_tracker.go

@@ -119,7 +119,7 @@ func (ht *headTracker[HTH, S, ID, BLOCK_HASH]) Start(ctx context.Context) error
 			if ctx.Err() != nil {
 				return ctx.Err()
 			}
-			ht.log.Errorw("Error handling initial head", "err", err)
+			ht.log.Errorw("Error handling initial head", "err", err.Error())
 		}
 
 		ht.wgDone.Add(3)
@@ -338,9 +338,23 @@ func (ht *headTracker[HTH, S, ID, BLOCK_HASH]) backfillLoop() {
 // calculateLatestFinalized - returns latest finalized block. It's expected that currentHeadNumber - is the head of
 // canonical chain. There is no guaranties that returned block belongs to the canonical chain. Additional verification
 // must be performed before usage.
-func (ht *headTracker[HTH, S, ID, BLOCK_HASH]) calculateLatestFinalized(ctx context.Context, currentHead HTH) (h HTH, err error) {
-	if ht.config.FinalityTagEnabled() {
-		return ht.client.LatestFinalizedBlock(ctx)
+func (ht *headTracker[HTH, S, ID, BLOCK_HASH]) calculateLatestFinalized(ctx context.Context, currentHead HTH) (latestFinalized HTH, err error) {
+	if ht.config.FinalityTagEnabled() && !ht.htConfig.FinalityTagBypass() {
+		latestFinalized, err = ht.client.LatestFinalizedBlock(ctx)
+		if err != nil {
+			return latestFinalized, fmt.Errorf("failed to get latest finalized block: %w", err)
+		}
+
+		if !latestFinalized.IsValid() {
+			return latestFinalized, fmt.Errorf("failed to get valid latest finalized block")
+		}
+
+		if currentHead.BlockNumber()-latestFinalized.BlockNumber() > int64(ht.htConfig.MaxAllowedFinalityDepth()) {
+			return latestFinalized, fmt.Errorf("gap between latest finalized block (%d) and current head (%d) is too large (> %d)",
+				latestFinalized.BlockNumber(), currentHead.BlockNumber(), ht.htConfig.MaxAllowedFinalityDepth())
+		}
+
+		return latestFinalized, nil
 	}
 	// no need to make an additional RPC call on chains with instant finality
 	if ht.config.FinalityDepth() == 0 {

common/headtracker/types/config.go

@@ -12,4 +12,6 @@ type HeadTrackerConfig interface {
 	HistoryDepth() uint32
 	MaxBufferSize() uint32
 	SamplingInterval() time.Duration
+	FinalityTagBypass() bool
+	MaxAllowedFinalityDepth() uint32
 }

core/chains/evm/config/chain_scoped_head_tracker.go

@@ -21,3 +21,11 @@ func (h *headTrackerConfig) MaxBufferSize() uint32 {
 func (h *headTrackerConfig) SamplingInterval() time.Duration {
 	return h.c.SamplingInterval.Duration()
 }
+
+func (h *headTrackerConfig) FinalityTagBypass() bool {
+	return *h.c.FinalityTagBypass
+}
+
+func (h *headTrackerConfig) MaxAllowedFinalityDepth() uint32 {
+	return *h.c.MaxAllowedFinalityDepth
+}

core/chains/evm/config/config.go

@@ -71,6 +71,8 @@ type HeadTracker interface {
 	HistoryDepth() uint32
 	MaxBufferSize() uint32
 	SamplingInterval() time.Duration
+	FinalityTagBypass() bool
+	MaxAllowedFinalityDepth() uint32
 }
 
 type BalanceMonitor interface {

core/chains/evm/config/config_test.go

@@ -376,6 +376,8 @@ func TestChainScopedConfig_HeadTracker(t *testing.T) {
 	assert.Equal(t, uint32(100), ht.HistoryDepth())
 	assert.Equal(t, uint32(3), ht.MaxBufferSize())
 	assert.Equal(t, time.Second, ht.SamplingInterval())
+	assert.Equal(t, true, ht.FinalityTagBypass())
+	assert.Equal(t, uint32(10000), ht.MaxAllowedFinalityDepth())
 }
 
 func Test_chainScopedConfig_Validate(t *testing.T) {

core/chains/evm/config/toml/config.go

@@ -740,9 +740,11 @@ func (e *KeySpecificGasEstimator) setFrom(f *KeySpecificGasEstimator) {
 }
 
 type HeadTracker struct {
-	HistoryDepth     *uint32
-	MaxBufferSize    *uint32
-	SamplingInterval *commonconfig.Duration
+	HistoryDepth            *uint32
+	MaxBufferSize           *uint32
+	SamplingInterval        *commonconfig.Duration
+	MaxAllowedFinalityDepth *uint32
+	FinalityTagBypass       *bool
 }
 
 func (t *HeadTracker) setFrom(f *HeadTracker) {
@@ -755,6 +757,21 @@ func (t *HeadTracker) setFrom(f *HeadTracker) {
 	if v := f.SamplingInterval; v != nil {
 		t.SamplingInterval = v
 	}
+	if v := f.MaxAllowedFinalityDepth; v != nil {
+		t.MaxAllowedFinalityDepth = v
+	}
+	if v := f.FinalityTagBypass; v != nil {
+		t.FinalityTagBypass = v
+	}
+}
+
+func (t *HeadTracker) ValidateConfig() (err error) {
+	if *t.MaxAllowedFinalityDepth < 1 {
+		err = multierr.Append(err, commonconfig.ErrInvalid{Name: "MaxAllowedFinalityDepth", Value: *t.MaxAllowedFinalityDepth,
+			Msg: "must be greater than or equal to 1"})
+	}
+
+	return
 }
 
 type ClientErrors struct {

core/chains/evm/config/toml/defaults/Avalanche_Fuji.toml

@@ -14,3 +14,6 @@ PriceMin = '25 gwei'
 
 [GasEstimator.BlockHistory]
 BlockHistorySize = 24
+
+[HeadTracker]
+FinalityTagBypass = false

core/chains/evm/config/toml/defaults/BSC_Testnet.toml

@@ -15,6 +15,11 @@ BumpThreshold = 5
 [GasEstimator.BlockHistory]
 BlockHistorySize = 24
 
+[HeadTracker]
+HistoryDepth = 100
+SamplingInterval = '1s'
+FinalityTagBypass = false
+
 [OCR]
 DatabaseTimeout = '2s'
 ContractTransmitterTransmitTimeout = '2s'

core/chains/evm/config/toml/defaults/Ethereum_Sepolia.toml

@@ -11,3 +11,6 @@ TransactionPercentile = 50
 
 [OCR2.Automation]
 GasLimit = 10500000
+
+[HeadTracker]
+FinalityTagBypass = false

core/chains/evm/config/toml/defaults/Linea_Sepolia.toml

@@ -10,4 +10,4 @@ PriceMin = '1 wei'
 ResendAfterThreshold = '3m'
 
 [HeadTracker]
-HistoryDepth = 1000
+HistoryDepth = 1000

core/chains/evm/config/toml/defaults/WeMix_Testnet.toml

@@ -12,3 +12,6 @@ ContractConfirmations = 1
 [GasEstimator]
 EIP1559DynamicFees = true
 TipCapDefault = '100 gwei'
+
+[HeadTracker]
+FinalityTagBypass = false

core/chains/evm/config/toml/defaults/fallback.toml

@@ -57,6 +57,8 @@ TransactionPercentile = 60
 HistoryDepth = 100
 MaxBufferSize = 3
 SamplingInterval = '1s'
+FinalityTagBypass = true
+MaxAllowedFinalityDepth = 10000
 
 [NodePool]
 PollFailureThreshold = 5

core/chains/evm/headtracker/head_saver_test.go

@@ -36,6 +36,13 @@ func (h *headTrackerConfig) MaxBufferSize() uint32 {
 	return uint32(0)
 }
 
+func (h *headTrackerConfig) FinalityTagBypass() bool {
+	return false
+}
+func (h *headTrackerConfig) MaxAllowedFinalityDepth() uint32 {
+	return 10000
+}
+
 type config struct {
 	finalityDepth                     uint32
 	blockEmissionIdleWarningThreshold time.Duration

core/chains/evm/headtracker/head_tracker_test.go

@@ -205,11 +205,27 @@ func TestHeadTracker_Start(t *testing.T) {
 	t.Parallel()
 
 	const historyDepth = 100
-	newHeadTracker := func(t *testing.T) *headTrackerUniverse {
+	const finalityDepth = 50
+	type opts struct {
+		FinalityTagEnable       *bool
+		MaxAllowedFinalityDepth *uint32
+		FinalityTagBypass       *bool
+	}
+	newHeadTracker := func(t *testing.T, opts opts) *headTrackerUniverse {
 		db := pgtest.NewSqlxDB(t)
 		gCfg := configtest.NewGeneralConfig(t, func(c *chainlink.Config, _ *chainlink.Secrets) {
-			c.EVM[0].FinalityTagEnabled = ptr[bool](true)
+			if opts.FinalityTagEnable != nil {
+				c.EVM[0].FinalityTagEnabled = opts.FinalityTagEnable
+			}
 			c.EVM[0].HeadTracker.HistoryDepth = ptr[uint32](historyDepth)
+			c.EVM[0].FinalityDepth = ptr[uint32](finalityDepth)
+			if opts.MaxAllowedFinalityDepth != nil {
+				c.EVM[0].HeadTracker.MaxAllowedFinalityDepth = opts.MaxAllowedFinalityDepth
+			}
+
+			if opts.FinalityTagBypass != nil {
+				c.EVM[0].HeadTracker.FinalityTagBypass = opts.FinalityTagBypass
+			}
 		})
 		config := evmtest.NewChainScopedConfig(t, gCfg)
 		orm := headtracker.NewORM(cltest.FixtureChainID, db)
@@ -219,44 +235,59 @@ func TestHeadTracker_Start(t *testing.T) {
 
 	t.Run("Fail start if context was canceled", func(t *testing.T) {
 		ctx, cancel := context.WithCancel(testutils.Context(t))
-		ht := newHeadTracker(t)
+		ht := newHeadTracker(t, opts{})
 		ht.ethClient.On("HeadByNumber", mock.Anything, (*big.Int)(nil)).Run(func(args mock.Arguments) {
 			cancel()
 		}).Return(cltest.Head(0), context.Canceled)
 		err := ht.headTracker.Start(ctx)
 		require.ErrorIs(t, err, context.Canceled)
 	})
 	t.Run("Starts even if failed to get initialHead", func(t *testing.T) {
-		ht := newHeadTracker(t)
+		ht := newHeadTracker(t, opts{})
 		ht.ethClient.On("HeadByNumber", mock.Anything, (*big.Int)(nil)).Return(cltest.Head(0), errors.New("failed to get init head"))
 		ht.Start(t)
 		tests.AssertLogEventually(t, ht.observer, "Error handling initial head")
 	})
 	t.Run("Starts even if received invalid head", func(t *testing.T) {
-		ht := newHeadTracker(t)
+		ht := newHeadTracker(t, opts{})
 		ht.ethClient.On("HeadByNumber", mock.Anything, (*big.Int)(nil)).Return(nil, nil)
 		ht.Start(t)
 		tests.AssertLogEventually(t, ht.observer, "Got nil initial head")
 	})
 	t.Run("Starts even if fails to get finalizedHead", func(t *testing.T) {
-		ht := newHeadTracker(t)
+		ht := newHeadTracker(t, opts{FinalityTagEnable: ptr(true), FinalityTagBypass: ptr(false)})
 		head := cltest.Head(1000)
 		ht.ethClient.On("HeadByNumber", mock.Anything, (*big.Int)(nil)).Return(head, nil).Once()
 		ht.ethClient.On("LatestFinalizedBlock", mock.Anything).Return(nil, errors.New("failed to load latest finalized")).Once()
 		ht.Start(t)
 		tests.AssertLogEventually(t, ht.observer, "Error handling initial head")
 	})
 	t.Run("Starts even if latest finalizedHead is nil", func(t *testing.T) {
-		ht := newHeadTracker(t)
+		ht := newHeadTracker(t, opts{FinalityTagEnable: ptr(true), FinalityTagBypass: ptr(false)})
 		head := cltest.Head(1000)
 		ht.ethClient.On("HeadByNumber", mock.Anything, (*big.Int)(nil)).Return(head, nil).Once()
 		ht.ethClient.On("LatestFinalizedBlock", mock.Anything).Return(nil, nil).Once()
+		ht.ethClient.On("SubscribeNewHead", mock.Anything, mock.Anything).Return(nil, errors.New("failed to connect")).Maybe()
 		ht.Start(t)
 		tests.AssertLogEventually(t, ht.observer, "Error handling initial head")
 	})
-	t.Run("Happy path", func(t *testing.T) {
+	t.Run("Logs error if finality gap is too big", func(t *testing.T) {
+		ht := newHeadTracker(t, opts{FinalityTagEnable: ptr(true), FinalityTagBypass: ptr(false), MaxAllowedFinalityDepth: ptr(uint32(10))})
+		head := cltest.Head(1000)
+		ht.ethClient.On("HeadByNumber", mock.Anything, (*big.Int)(nil)).Return(head, nil).Once()
+		ht.ethClient.On("LatestFinalizedBlock", mock.Anything).Return(cltest.Head(989), nil).Once()
+		ht.ethClient.On("SubscribeNewHead", mock.Anything, mock.Anything).Return(nil, errors.New("failed to connect")).Maybe()
+		ht.Start(t)
+		tests.AssertEventually(t, func() bool {
+			// must exactly match the error passed to logger
+			field := zap.String("err", "failed to calculate latest finalized head: gap between latest finalized block (989) and current head (1000) is too large (> 10)")
+			filtered := ht.observer.FilterMessage("Error handling initial head").FilterField(field)
+			return filtered.Len() > 0
+		})
+	})
+	t.Run("Happy path (finality tag)", func(t *testing.T) {
 		head := cltest.Head(1000)
-		ht := newHeadTracker(t)
+		ht := newHeadTracker(t, opts{FinalityTagEnable: ptr(true), FinalityTagBypass: ptr(false)})
 		ctx := testutils.Context(t)
 		require.NoError(t, ht.orm.IdempotentInsertHead(ctx, cltest.Head(799)))
 		ht.ethClient.On("HeadByNumber", mock.Anything, (*big.Int)(nil)).Return(head, nil).Once()
@@ -265,9 +296,46 @@ func TestHeadTracker_Start(t *testing.T) {
 		ht.ethClient.On("LatestFinalizedBlock", mock.Anything).Return(finalizedHead, nil).Once()
 		// on backfill
 		ht.ethClient.On("LatestFinalizedBlock", mock.Anything).Return(nil, errors.New("backfill call to finalized failed")).Maybe()
+		ht.ethClient.On("SubscribeNewHead", mock.Anything, mock.Anything).Return(nil, errors.New("failed to connect")).Maybe()
 		ht.Start(t)
 		tests.AssertLogEventually(t, ht.observer, "Loaded chain from DB")
 	})
+	happyPathFD := func(t *testing.T, opts opts) {
+		head := cltest.Head(1000)
+		ht := newHeadTracker(t, opts)
+		ht.ethClient.On("HeadByNumber", mock.Anything, (*big.Int)(nil)).Return(head, nil).Once()
+		finalizedHead := cltest.Head(head.Number - finalityDepth)
+		ht.ethClient.On("HeadByNumber", mock.Anything, big.NewInt(finalizedHead.Number)).Return(finalizedHead, nil).Once()
+		ctx := testutils.Context(t)
+		require.NoError(t, ht.orm.IdempotentInsertHead(ctx, cltest.Head(finalizedHead.Number-1)))
+		// on backfill
+		ht.ethClient.On("HeadByNumber", mock.Anything, mock.Anything).Return(nil, errors.New("backfill call to finalized failed")).Maybe()
+		ht.ethClient.On("SubscribeNewHead", mock.Anything, mock.Anything).Return(nil, errors.New("failed to connect")).Maybe()
+		ht.Start(t)
+		tests.AssertLogEventually(t, ht.observer, "Loaded chain from DB")
+	}
+	testCases := []struct {
+		Name string
+		Opts opts
+	}{
+		{
+			Name: "Happy path (Chain FT is disabled & HeadTracker's FT is disabled)",
+			Opts: opts{FinalityTagEnable: ptr(false), FinalityTagBypass: ptr(true)},
+		},
+		{
+			Name: "Happy path (Chain FT is disabled & HeadTracker's FT is enabled, but ignored)",
+			Opts: opts{FinalityTagEnable: ptr(false), FinalityTagBypass: ptr(false)},
+		},
+		{
+			Name: "Happy path (Chain FT is enabled & HeadTracker's FT is disabled)",
+			Opts: opts{FinalityTagEnable: ptr(true), FinalityTagBypass: ptr(true)},
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.Name, func(t *testing.T) {
+			happyPathFD(t, tc.Opts)
+		})
+	}
 }
 
 func TestHeadTracker_CallsHeadTrackableCallbacks(t *testing.T) {

core/config/docs/chains-evm.toml

@@ -311,6 +311,14 @@ MaxBufferSize = 3 # Default
 # **ADVANCED**
 # SamplingInterval means that head tracker callbacks will at maximum be made once in every window of this duration. This is a performance optimisation for fast chains. Set to 0 to disable sampling entirely.
 SamplingInterval = '1s' # Default
+# FinalityTagBypass disables FinalityTag support in HeadTracker and makes it track blocks up to FinalityDepth from the most recent head.
+# It should only be used on chains with an extremely large actual finality depth (the number of blocks between the most recent head and the latest finalized block).
+# Has no effect if `FinalityTagsEnabled` = false
+FinalityTagBypass = true # Default
+# MaxAllowedFinalityDepth - defines maximum number of blocks between the most recent head and the latest finalized block.
+# If actual finality depth exceeds this number, HeadTracker aborts backfill and returns an error.
+# Has no effect if `FinalityTagsEnabled` = false
+MaxAllowedFinalityDepth = 10000 # Default
 
 [[EVM.KeySpecific]]
 # Key is the account to apply these settings to

core/services/chainlink/config_test.go

@@ -24,6 +24,7 @@ import (
 	coscfg "github.com/smartcontractkit/chainlink-cosmos/pkg/cosmos/config"
 	solcfg "github.com/smartcontractkit/chainlink-solana/pkg/solana/config"
 	stkcfg "github.com/smartcontractkit/chainlink-starknet/relayer/pkg/chainlink/config"
+
 	commonconfig "github.com/smartcontractkit/chainlink/v2/common/config"
 	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/assets"
 
@@ -572,9 +573,11 @@ func TestConfig_Marshal(t *testing.T) {
 				},
 
 				HeadTracker: evmcfg.HeadTracker{
-					HistoryDepth:     ptr[uint32](15),
-					MaxBufferSize:    ptr[uint32](17),
-					SamplingInterval: &hour,
+					HistoryDepth:            ptr[uint32](15),
+					MaxBufferSize:           ptr[uint32](17),
+					SamplingInterval:        &hour,
+					FinalityTagBypass:       ptr[bool](false),
+					MaxAllowedFinalityDepth: ptr[uint32](1500),
 				},
 
 				NodePool: evmcfg.NodePool{
@@ -1034,6 +1037,8 @@ TransactionPercentile = 15
 HistoryDepth = 15
 MaxBufferSize = 17
 SamplingInterval = '1h0m0s'
+MaxAllowedFinalityDepth = 1500
+FinalityTagBypass = false
 
 [[EVM.KeySpecific]]
 Key = '0x2a3e23c6f242F5345320814aC8a1b4E58707D292'
@@ -1275,7 +1280,7 @@ func TestConfig_Validate(t *testing.T) {
 					- WSURL: missing: required for primary nodes
 					- HTTPURL: missing: required for all nodes
 				- 1.HTTPURL: missing: required for all nodes
-		- 1: 9 errors:
+		- 1: 10 errors:
 			- ChainType: invalid value (Foo): must not be set with this chain id
 			- Nodes: missing: must have at least one node
 			- ChainType: invalid value (Foo): must be one of arbitrum, celo, gnosis, kroma, metis, optimismBedrock, scroll, wemix, xlayer, zkevm, zksync or omitted
@@ -1286,6 +1291,7 @@ func TestConfig_Validate(t *testing.T) {
 			- GasEstimator: 2 errors:
 				- FeeCapDefault: invalid value (101 wei): must be equal to PriceMax (99 wei) since you are using FixedPrice estimation with gas bumping disabled in EIP1559 mode - PriceMax will be used as the FeeCap for transactions instead of FeeCapDefault
 				- PriceMax: invalid value (1 gwei): must be greater than or equal to PriceDefault
+			- HeadTracker.MaxAllowedFinalityDepth: invalid value (0): must be greater than or equal to 1
 			- KeySpecific.Key: invalid value (0xde709f2102306220921060314715629080e2fb77): duplicate - must be unique
 		- 2: 5 errors:
 			- ChainType: invalid value (Arbitrum): only "optimismBedrock" can be used with this chain id

core/services/chainlink/testdata/config-full.toml

@@ -335,6 +335,8 @@ TransactionPercentile = 15
 HistoryDepth = 15
 MaxBufferSize = 17
 SamplingInterval = '1h0m0s'
+MaxAllowedFinalityDepth = 1500
+FinalityTagBypass = false
 
 [[EVM.KeySpecific]]
 Key = '0x2a3e23c6f242F5345320814aC8a1b4E58707D292'