Add finalizer component to TXM

.changeset/itchy-bugs-clean.md

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

common/txmgr/models.go

@@ -11,4 +11,5 @@ const (
 	TxUnconfirmed             = txmgrtypes.TxState("unconfirmed")
 	TxConfirmed               = txmgrtypes.TxState("confirmed")
 	TxConfirmedMissingReceipt = txmgrtypes.TxState("confirmed_missing_receipt")
+	TxFinalized               = txmgrtypes.TxState("finalized")
 )

common/txmgr/reaper.go

@@ -14,7 +14,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
@@ -25,10 +24,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"),
@@ -103,13 +101,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

@@ -108,6 +108,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
@@ -143,6 +144,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]{
@@ -165,13 +167,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")
 	}
@@ -199,6 +202,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()
@@ -293,6 +300,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() {
@@ -415,6 +423,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
@@ -446,6 +455,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
@@ -644,9 +657,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
-		// Return unconfirmed if tx receipt's block is newer than the latest finalized block
+		// Return unconfirmed for confirmed transactions because they are not yet finalized
 		return commontypes.Unconfirmed, nil
+	case TxFinalized:
+		return commontypes.Finalized, nil
 	case TxFatalError:
 		// Use an ErrorClassifier to determine if the transaction is considered Fatal
 		txErr := b.newErrorClassifier(tx.GetError())

common/txmgr/types/config.go

@@ -5,7 +5,6 @@ import "time"
 type TransactionManagerChainConfig interface {
 	BroadcasterChainConfig
 	ConfirmerChainConfig
-	ReaperChainConfig
 }
 
 type TransactionManagerFeeConfig interface {
@@ -74,11 +73,6 @@ type ResenderTransactionsConfig interface {
 	MaxInFlight() uint32
 }
 
-// ReaperConfig is the config subset used by the reaper
-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
+}

common/txmgr/types/tx_store.go

@@ -105,11 +105,10 @@ type TransactionStore[
 	UpdateTxUnstartedToInProgress(ctx context.Context, etx *Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], attempt *TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) error
 	UpdateTxFatalError(ctx context.Context, etx *Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) error
 	UpdateTxForRebroadcast(ctx context.Context, etx Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], etxAttempt TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) error
-	IsTxFinalized(ctx context.Context, blockHeight int64, txID int64, chainID CHAIN_ID) (finalized bool, err error)
 }
 
 type TxHistoryReaper[CHAIN_ID types.ID] interface {
-	ReapTxHistory(ctx context.Context, minBlockNumberToKeep int64, timeThreshold time.Time, chainID CHAIN_ID) error
+	ReapTxHistory(ctx context.Context, timeThreshold time.Time, chainID CHAIN_ID) error
 }
 
 type UnstartedTxQueuePruner interface {