[BCI-3573] - Remove dependence on FinalityDepth in EVM TXM code

.changeset/itchy-bugs-clean.md

@@ -0,0 +1,5 @@
+---
+"chainlink": minor
+---
+
+Added a finalizer component to the TXM to mark transactions as finalized #internal

common/txmgr/confirmer.go

@@ -297,7 +297,7 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) pro
 		return fmt.Errorf("CheckConfirmedMissingReceipt failed: %w", err)
 	}
 
-	if err := ec.CheckForReceipts(ctx, head.BlockNumber()); err != nil {
+	if err := ec.CheckForReceipts(ctx, head.BlockNumber(), head.LatestFinalizedHead().BlockNumber()); err != nil {
 		return fmt.Errorf("CheckForReceipts failed: %w", err)
 	}
 
@@ -395,8 +395,8 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Che
 	return
 }
 
-// CheckForReceipts finds attempts that are still pending and checks to see if a receipt is present for the given block number
-func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) CheckForReceipts(ctx context.Context, blockNum int64) error {
+// CheckForReceipts finds attempts that are still pending and checks to see if a receipt is present for the given block number.
+func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) CheckForReceipts(ctx context.Context, blockNum int64, latestFinalizedBlockNum int64) error {
 	attempts, err := ec.txStore.FindTxAttemptsRequiringReceiptFetch(ctx, ec.chainID)
 	if err != nil {
 		return fmt.Errorf("FindTxAttemptsRequiringReceiptFetch failed: %w", err)
@@ -443,7 +443,7 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Che
 		return fmt.Errorf("unable to mark txes as 'confirmed_missing_receipt': %w", err)
 	}
 
-	if err := ec.txStore.MarkOldTxesMissingReceiptAsErrored(ctx, blockNum, ec.chainConfig.FinalityDepth(), ec.chainID); err != nil {
+	if err := ec.txStore.MarkOldTxesMissingReceiptAsErrored(ctx, blockNum, latestFinalizedBlockNum, ec.chainID); err != nil {
 		return fmt.Errorf("unable to confirm buried unconfirmed txes': %w", err)
 	}
 	return nil
@@ -1000,22 +1000,22 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) han
 	}
 }
 
-// EnsureConfirmedTransactionsInLongestChain finds all confirmed txes up to the depth
+// EnsureConfirmedTransactionsInLongestChain finds all confirmed txes up to the LatestFinalizedHead
 // of the given chain and ensures that every one has a receipt with a block hash that is
 // in the given chain.
 //
 // If any of the confirmed transactions does not have a receipt in the chain, it has been
 // re-org'd out and will be rebroadcast.
 func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) EnsureConfirmedTransactionsInLongestChain(ctx context.Context, head types.Head[BLOCK_HASH]) error {
-	if head.ChainLength() < ec.chainConfig.FinalityDepth() {
+	if head.ChainLength() < uint32(head.LatestFinalizedHead().BlockNumber()) {
 		logArgs := []interface{}{
-			"chainLength", head.ChainLength(), "finalityDepth", ec.chainConfig.FinalityDepth(),
+			"chainLength", head.ChainLength(), "latestFinalizedHead", head.LatestFinalizedHead().BlockNumber(),
 		}
 		if ec.nConsecutiveBlocksChainTooShort > logAfterNConsecutiveBlocksChainTooShort {
-			warnMsg := "Chain length supplied for re-org detection was shorter than FinalityDepth. Re-org protection is not working properly. This could indicate a problem with the remote RPC endpoint, a compatibility issue with a particular blockchain, a bug with this particular blockchain, heads table being truncated too early, remote node out of sync, or something else. If this happens a lot please raise a bug with the Chainlink team including a log output sample and details of the chain and RPC endpoint you are using."
+			warnMsg := "Chain length supplied for re-org detection was shorter than LatestFinalizedHead. Re-org protection is not working properly. This could indicate a problem with the remote RPC endpoint, a compatibility issue with a particular blockchain, a bug with this particular blockchain, heads table being truncated too early, remote node out of sync, or something else. If this happens a lot please raise a bug with the Chainlink team including a log output sample and details of the chain and RPC endpoint you are using."
 			ec.lggr.Warnw(warnMsg, append(logArgs, "nConsecutiveBlocksChainTooShort", ec.nConsecutiveBlocksChainTooShort)...)
 		} else {
-			logMsg := "Chain length supplied for re-org detection was shorter than FinalityDepth"
+			logMsg := "Chain length supplied for re-org detection was shorter than LatestFinalizedHead"
 			ec.lggr.Debugw(logMsg, append(logArgs, "nConsecutiveBlocksChainTooShort", ec.nConsecutiveBlocksChainTooShort)...)
 		}
 		ec.nConsecutiveBlocksChainTooShort++
@@ -1100,6 +1100,7 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) mar
 		"txID", etx.ID,
 		"attemptID", attempt.ID,
 		"nReceipts", len(attempt.Receipts),
+		"finalized", etx.Finalized,
 		"id", "confirmer",
 	}
 

common/txmgr/reaper.go

@@ -16,7 +16,6 @@ import (
 // Reaper handles periodic database cleanup for Txm
 type Reaper[CHAIN_ID types.ID] struct {
 	store          txmgrtypes.TxHistoryReaper[CHAIN_ID]
-	config         txmgrtypes.ReaperChainConfig
 	txConfig       txmgrtypes.ReaperTransactionsConfig
 	chainID        CHAIN_ID
 	log            logger.Logger
@@ -27,10 +26,9 @@ type Reaper[CHAIN_ID types.ID] struct {
 }
 
 // NewReaper instantiates a new reaper object
-func NewReaper[CHAIN_ID types.ID](lggr logger.Logger, store txmgrtypes.TxHistoryReaper[CHAIN_ID], config txmgrtypes.ReaperChainConfig, txConfig txmgrtypes.ReaperTransactionsConfig, chainID CHAIN_ID) *Reaper[CHAIN_ID] {
+func NewReaper[CHAIN_ID types.ID](lggr logger.Logger, store txmgrtypes.TxHistoryReaper[CHAIN_ID], txConfig txmgrtypes.ReaperTransactionsConfig, chainID CHAIN_ID) *Reaper[CHAIN_ID] {
 	r := &Reaper[CHAIN_ID]{
 		store,
-		config,
 		txConfig,
 		chainID,
 		logger.Named(lggr, "Reaper"),
@@ -106,13 +104,12 @@ func (r *Reaper[CHAIN_ID]) ReapTxes(headNum int64) error {
 		r.log.Debug("Transactions.ReaperThreshold  set to 0; skipping ReapTxes")
 		return nil
 	}
-	minBlockNumberToKeep := headNum - int64(r.config.FinalityDepth())
 	mark := time.Now()
 	timeThreshold := mark.Add(-threshold)
 
-	r.log.Debugw(fmt.Sprintf("reaping old txes created before %s", timeThreshold.Format(time.RFC3339)), "ageThreshold", threshold, "timeThreshold", timeThreshold, "minBlockNumberToKeep", minBlockNumberToKeep)
+	r.log.Debugw(fmt.Sprintf("reaping old txes created before %s", timeThreshold.Format(time.RFC3339)), "ageThreshold", threshold, "timeThreshold", timeThreshold)
 
-	if err := r.store.ReapTxHistory(ctx, minBlockNumberToKeep, timeThreshold, r.chainID); err != nil {
+	if err := r.store.ReapTxHistory(ctx, timeThreshold, r.chainID); err != nil {
 		return err
 	}
 

common/txmgr/txmgr.go

@@ -110,6 +110,7 @@ type Txm[
 	broadcaster        *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]
 	confirmer          *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]
 	tracker            *Tracker[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]
+	finalizer          txmgrtypes.Finalizer[BLOCK_HASH, HEAD]
 	fwdMgr             txmgrtypes.ForwarderManager[ADDR]
 	txAttemptBuilder   txmgrtypes.TxAttemptBuilder[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]
 	newErrorClassifier NewErrorClassifier
@@ -145,6 +146,7 @@ func NewTxm[
 	confirmer *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE],
 	resender *Resender[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE],
 	tracker *Tracker[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE],
+	finalizer txmgrtypes.Finalizer[BLOCK_HASH, HEAD],
 	newErrorClassifierFunc NewErrorClassifier,
 ) *Txm[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE] {
 	b := Txm[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]{
@@ -167,13 +169,14 @@ func NewTxm[
 		resender:           resender,
 		tracker:            tracker,
 		newErrorClassifier: newErrorClassifierFunc,
+		finalizer:          finalizer,
 	}
 
 	if txCfg.ResendAfterThreshold() <= 0 {
 		b.logger.Info("Resender: Disabled")
 	}
 	if txCfg.ReaperThreshold() > 0 && txCfg.ReaperInterval() > 0 {
-		b.reaper = NewReaper[CHAIN_ID](lggr, b.txStore, cfg, txCfg, chainId)
+		b.reaper = NewReaper[CHAIN_ID](lggr, b.txStore, txCfg, chainId)
 	} else {
 		b.logger.Info("TxReaper: Disabled")
 	}
@@ -201,6 +204,10 @@ func (b *Txm[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Start(ctx
 			return fmt.Errorf("Txm: Tracker failed to start: %w", err)
 		}
 
+		if err := ms.Start(ctx, b.finalizer); err != nil {
+			return fmt.Errorf("Txm: Finalizer failed to start: %w", err)
+		}
+
 		b.logger.Info("Txm starting runLoop")
 		b.wg.Add(1)
 		go b.runLoop()
@@ -295,6 +302,7 @@ func (b *Txm[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) HealthRepo
 		services.CopyHealth(report, b.broadcaster.HealthReport())
 		services.CopyHealth(report, b.confirmer.HealthReport())
 		services.CopyHealth(report, b.txAttemptBuilder.HealthReport())
+		services.CopyHealth(report, b.finalizer.HealthReport())
 	})
 
 	if b.txConfig.ForwardersEnabled() {
@@ -417,6 +425,7 @@ func (b *Txm[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) runLoop()
 		case head := <-b.chHeads:
 			b.confirmer.mb.Deliver(head)
 			b.tracker.mb.Deliver(head.BlockNumber())
+			b.finalizer.DeliverLatestHead(head)
 		case reset := <-b.reset:
 			// This check prevents the weird edge-case where you can select
 			// into this block after chStop has already been closed and the
@@ -448,6 +457,10 @@ func (b *Txm[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) runLoop()
 			if err != nil && (!errors.Is(err, services.ErrAlreadyStopped) || !errors.Is(err, services.ErrCannotStopUnstarted)) {
 				b.logger.Errorw(fmt.Sprintf("Failed to Close Tracker: %v", err), "err", err)
 			}
+			err = b.finalizer.Close()
+			if err != nil && (!errors.Is(err, services.ErrAlreadyStopped) || !errors.Is(err, services.ErrCannotStopUnstarted)) {
+				b.logger.Errorw(fmt.Sprintf("Failed to Close Finalizer: %v", err), "err", err)
+			}
 			return
 		case <-keysChanged:
 			// This check prevents the weird edge-case where you can select
@@ -646,7 +659,10 @@ func (b *Txm[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) GetTransac
 		// Return unconfirmed for ConfirmedMissingReceipt since a receipt is required to determine if it is finalized
 		return commontypes.Unconfirmed, nil
 	case TxConfirmed:
-		// TODO: Check for finality and return finalized status
+		if tx.Finalized {
+			// Return finalized if tx receipt's block is equal or older than the latest finalized block
+			return commontypes.Finalized, nil
+		}
 		// Return unconfirmed if tx receipt's block is newer than the latest finalized block
 		return commontypes.Unconfirmed, nil
 	case TxFatalError:

common/txmgr/types/config.go

@@ -5,7 +5,6 @@ import "time"
 type TransactionManagerChainConfig interface {
 	BroadcasterChainConfig
 	ConfirmerChainConfig
-	ReaperChainConfig
 }
 
 type TransactionManagerFeeConfig interface {
@@ -74,13 +73,6 @@ type ResenderTransactionsConfig interface {
 	MaxInFlight() uint32
 }
 
-// ReaperConfig is the config subset used by the reaper
-//
-//go:generate mockery --quiet --name ReaperChainConfig --structname ReaperConfig --output ./mocks/ --case=underscore
-type ReaperChainConfig interface {
-	FinalityDepth() uint32
-}
-
 type ReaperTransactionsConfig interface {
 	ReaperInterval() time.Duration
 	ReaperThreshold() time.Duration

common/txmgr/types/finalizer.go

@@ -0,0 +1,12 @@
+package types
+
+import (
+	"github.com/smartcontractkit/chainlink-common/pkg/services"
+	"github.com/smartcontractkit/chainlink/v2/common/types"
+)
+
+type Finalizer[BLOCK_HASH types.Hashable, HEAD types.Head[BLOCK_HASH]] interface {
+	// interfaces for running the underlying estimator
+	services.Service
+	DeliverLatestHead(head HEAD) bool
+}