Dual transmission for 2.19

.changeset/cool-owls-laugh.md

@@ -0,0 +1,5 @@
+---
+"chainlink": minor
+---
+
+Bump to start the next version

.github/CODEOWNERS

@@ -39,6 +39,10 @@
 /core/services/ccip @smartcontractkit/ccip
 /core/services/ocr2/plugins/ccip @smartcontractkit/ccip
 
+# CCIP
+/core/services/ccip @smartcontractkit/ccip
+/core/services/ocr2/plugins/ccip @smartcontractkit/ccip
+
 # VRF-related services
 /core/services/vrf @smartcontractkit/dev-services @smartcontractkit/core
 /core/services/blockhashstore @smartcontractkit/dev-services @smartcontractkit/core

.mockery.yaml

@@ -93,6 +93,12 @@ packages:
       BalanceMonitor:
         config:
           dir: "{{ .InterfaceDir }}/../mocks"
+  github.com/smartcontractkit/chainlink/v2/core/chains/evm/txm:
+    interfaces:
+      Client:
+      TxStore:
+      AttemptBuilder:
+      Keystore:
   github.com/smartcontractkit/chainlink/v2/core/chains/evm/txmgr:
     interfaces:
       ChainConfig:

common/txmgr/confirmer.go

@@ -282,6 +282,23 @@
 		return err
 	}
 	ec.lggr.Debugw("Finished RebroadcastWhereNecessary", "headNum", head.BlockNumber(), "time", time.Since(mark), "id", "confirmer")
+	mark = time.Now()
+
+	if err := ec.EnsureConfirmedTransactionsInLongestChain(ctx, head); err != nil {
+		return fmt.Errorf("EnsureConfirmedTransactionsInLongestChain failed: %w", err)
+	}
+
+	ec.lggr.Debugw("Finished EnsureConfirmedTransactionsInLongestChain", "headNum", head.BlockNumber(), "time", time.Since(mark), "id", "confirmer")
+
+	if ec.resumeCallback != nil {
+		mark = time.Now()
+		if err := ec.ResumePendingTaskRuns(ctx, head.BlockNumber(), latestFinalizedHead.BlockNumber()); err != nil {
+			return fmt.Errorf("ResumePendingTaskRuns failed: %w", err)
+		}
+
+		ec.lggr.Debugw("Finished ResumePendingTaskRuns", "headNum", head.BlockNumber(), "time", time.Since(mark), "id", "confirmer")
+	}
+
 	ec.lggr.Debugw("processHead finish", "headNum", head.BlockNumber(), "id", "confirmer")
 
 	return nil
@@ -800,6 +817,139 @@
 	}
 }
 
+// EnsureConfirmedTransactionsInLongestChain finds all confirmed txes up to the earliest head
+// 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 {
+	logArgs := []interface{}{
+		"chainLength", head.ChainLength(),
+	}
+
+	//Here, we rely on the finalized block provided in the chain instead of the one
+	//provided via a dedicated method to avoid the false warning of the chain being
+	//too short. When `FinalityTagBypass = true,` HeadTracker tracks `finality depth
+	//+ history depth` to prevent excessive CPU usage. Thus, the provided chain may
+	//be shorter than the chain from the latest to the latest finalized, marked with
+	//a tag. A proper fix of this issue and complete switch to finality tag support
+	//will be introduced in BCFR-620
+	latestFinalized := head.LatestFinalizedHead()
+	if latestFinalized == nil || !latestFinalized.IsValid() {
+		if ec.nConsecutiveBlocksChainTooShort > logAfterNConsecutiveBlocksChainTooShort {
+			warnMsg := "Chain length supplied for re-org detection was shorter than the depth from the latest head to the finalized head. 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 the depth from the latest head to the finalized head"
+			ec.lggr.Debugw(logMsg, append(logArgs, "nConsecutiveBlocksChainTooShort", ec.nConsecutiveBlocksChainTooShort)...)
+		}
+		ec.nConsecutiveBlocksChainTooShort++
+	} else {
+		ec.nConsecutiveBlocksChainTooShort = 0
+	}
+	etxs, err := ec.txStore.FindTransactionsConfirmedInBlockRange(ctx, head.BlockNumber(), head.EarliestHeadInChain().BlockNumber(), ec.chainID)
+	if err != nil {
+		return fmt.Errorf("findTransactionsConfirmedInBlockRange failed: %w", err)
+	}
+
+	for _, etx := range etxs {
+		if !hasReceiptInLongestChain(*etx, head) {
+			if err := ec.markForRebroadcast(ctx, *etx, head); err != nil {
+				return fmt.Errorf("markForRebroadcast failed for etx %v: %w", etx.ID, err)
+			}
+		}
+	}
+
+	// It is safe to process separate keys concurrently
+	// NOTE: This design will block one key if another takes a really long time to execute
+	var wg sync.WaitGroup
+	errors := []error{}
+	var errMu sync.Mutex
+	wg.Add(len(ec.enabledAddresses))
+	for _, address := range ec.enabledAddresses {
+		go func(fromAddress ADDR) {
+			if err := ec.handleAnyInProgressAttempts(ctx, fromAddress, head.BlockNumber()); err != nil {
+				errMu.Lock()
+				errors = append(errors, err)
+				errMu.Unlock()
+				ec.lggr.Errorw("Error in handleAnyInProgressAttempts", "err", err, "fromAddress", fromAddress)
+			}
+
+			wg.Done()
+		}(address)
+	}
+
+	wg.Wait()
+
+	return multierr.Combine(errors...)
+}
+
+func hasReceiptInLongestChain[
+	CHAIN_ID types.ID,
+	ADDR types.Hashable,
+	TX_HASH, BLOCK_HASH types.Hashable,
+	SEQ types.Sequence,
+	FEE feetypes.Fee,
+](etx txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], head types.Head[BLOCK_HASH]) bool {
+	for {
+		for _, attempt := range etx.TxAttempts {
+			for _, receipt := range attempt.Receipts {
+				if receipt.GetBlockHash().String() == head.BlockHash().String() && receipt.GetBlockNumber().Int64() == head.BlockNumber() {
+					return true
+				}
+			}
+		}
+
+		head = head.GetParent()
+		if head == nil {
+			return false
+		}
+	}
+}
+
+func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) markForRebroadcast(ctx context.Context, etx txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], head types.Head[BLOCK_HASH]) error {
+	if len(etx.TxAttempts) == 0 {
+		return fmt.Errorf("invariant violation: expected tx %v to have at least one attempt", etx.ID)
+	}
+
+	// Rebroadcast the one with the highest gas price
+	attempt := etx.TxAttempts[0]
+	var receipt txmgrtypes.ChainReceipt[TX_HASH, BLOCK_HASH]
+	if len(attempt.Receipts) > 0 {
+		receipt = attempt.Receipts[0]
+	}
+
+	logValues := []interface{}{
+		"txhash", attempt.Hash.String(),
+		"currentBlockNum", head.BlockNumber(),
+		"currentBlockHash", head.BlockHash().String(),
+		"txID", etx.ID,
+		"attemptID", attempt.ID,
+		"nReceipts", len(attempt.Receipts),
+		"id", "confirmer",
+	}
+
+	// nil check on receipt interface
+	if receipt != nil {
+		logValues = append(logValues,
+			"replacementBlockHashAtConfirmedHeight", head.HashAtHeight(receipt.GetBlockNumber().Int64()),
+			"confirmedInBlockNum", receipt.GetBlockNumber(),
+			"confirmedInBlockHash", receipt.GetBlockHash(),
+			"confirmedInTxIndex", receipt.GetTransactionIndex(),
+		)
+	}
+
+	ec.lggr.Infow(fmt.Sprintf("Re-org detected. Rebroadcasting transaction %s which may have been re-org'd out of the main chain", attempt.Hash.String()), logValues...)
+
+	// Put it back in progress and delete all receipts (they do not apply to the new chain)
+	if err := ec.txStore.UpdateTxForRebroadcast(ctx, etx, attempt); err != nil {
+		return fmt.Errorf("markForRebroadcast failed: %w", err)
+	}
+
+	return nil
+}
+
 // ForceRebroadcast sends a transaction for every sequence in the given sequence range at the given gas price.
 // If an tx exists for this sequence, we re-send the existing tx with the supplied parameters.
 // If an tx doesn't exist for this sequence, we send a zero transaction.

common/txmgr/txmgr.go

@@ -65,6 +65,19 @@ type TxManager[
 	GetTransactionStatus(ctx context.Context, transactionID string) (state commontypes.TransactionStatus, err error)
 }
 
+type TxmV2Wrapper[
+	CHAIN_ID types.ID,
+	HEAD types.Head[BLOCK_HASH],
+	ADDR types.Hashable,
+	TX_HASH types.Hashable,
+	BLOCK_HASH types.Hashable,
+	SEQ types.Sequence,
+	FEE feetypes.Fee,
+] interface {
+	services.Service
+	CreateTransaction(ctx context.Context, txRequest txmgrtypes.TxRequest[ADDR, TX_HASH]) (etx txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], err error)
+}
+
 type reset struct {
 	// f is the function to execute between stopping/starting the
 	// Broadcaster and Confirmer
@@ -112,6 +125,7 @@ type Txm[
 	fwdMgr             txmgrtypes.ForwarderManager[ADDR]
 	txAttemptBuilder   txmgrtypes.TxAttemptBuilder[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]
 	newErrorClassifier NewErrorClassifier
+	txmv2wrapper       TxmV2Wrapper[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]
 }
 
 func (b *Txm[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) RegisterResumeCallback(fn ResumeCallback) {
@@ -147,6 +161,7 @@ func NewTxm[
 	tracker *Tracker[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE],
 	finalizer txmgrtypes.Finalizer[BLOCK_HASH, HEAD],
 	newErrorClassifierFunc NewErrorClassifier,
+	txmv2wrapper TxmV2Wrapper[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE],
 ) *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]{
 		logger:             logger.Sugared(lggr),
@@ -169,6 +184,7 @@ func NewTxm[
 		tracker:            tracker,
 		newErrorClassifier: newErrorClassifierFunc,
 		finalizer:          finalizer,
+		txmv2wrapper:       txmv2wrapper,
 	}
 
 	if txCfg.ResendAfterThreshold() <= 0 {
@@ -207,6 +223,12 @@ func (b *Txm[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Start(ctx
 			return fmt.Errorf("Txm: Finalizer failed to start: %w", err)
 		}
 
+		if b.txmv2wrapper != nil {
+			if err := ms.Start(ctx, b.txmv2wrapper); err != nil {
+				return fmt.Errorf("Txm: Txmv2 failed to start: %w", err)
+			}
+		}
+
 		b.logger.Info("Txm starting runLoop")
 		b.wg.Add(1)
 		go b.runLoop()
@@ -460,6 +482,12 @@ 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 Finalizer: %v", err), "err", err)
 			}
+			if b.txmv2wrapper != nil {
+				err = b.txmv2wrapper.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
@@ -513,11 +541,14 @@ func (b *Txm[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Trigger(ad
 func (b *Txm[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) CreateTransaction(ctx context.Context, txRequest txmgrtypes.TxRequest[ADDR, TX_HASH]) (tx txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], err error) {
 	// Check for existing Tx with IdempotencyKey. If found, return the Tx and do nothing
 	// Skipping CreateTransaction to avoid double send
+	if b.txmv2wrapper != nil && txRequest.Meta != nil && txRequest.Meta.DualBroadcast != nil && *txRequest.Meta.DualBroadcast {
+		return b.txmv2wrapper.CreateTransaction(ctx, txRequest)
+	}
 	if txRequest.IdempotencyKey != nil {
 		var existingTx *txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]
 		existingTx, err = b.txStore.FindTxWithIdempotencyKey(ctx, *txRequest.IdempotencyKey, b.chainID)
 		if err != nil {
-			return tx, fmt.Errorf("Failed to search for transaction with IdempotencyKey: %w", err)
+			return tx, fmt.Errorf("failed to search for transaction with IdempotencyKey: %w", err)
 		}
 		if existingTx != nil {
 			b.logger.Infow("Found a Tx with IdempotencyKey. Returning existing Tx without creating a new one.", "IdempotencyKey", *txRequest.IdempotencyKey)

core/capabilities/ccip/ocrimpls/contract_transmitter_test.go

@@ -476,7 +476,8 @@ func makeTestEvmTxm(
 		lp,
 		keyStore,
 		estimator,
-		ht)
+		ht,
+		nil)
 	require.NoError(t, err, "can't create tx manager")
 
 	_, unsub := broadcaster.Subscribe(txm)
@@ -603,8 +604,8 @@ func (d *TestDAOracleConfig) OracleType() *toml.DAOracleType {
 	return &oracleType
 }
 
-func (d *TestDAOracleConfig) OracleAddress() *evmtypes.EIP55Address {
-	a, err := evmtypes.NewEIP55Address("0x420000000000000000000000000000000000000F")
+func (d *TestDAOracleConfig) OracleAddress() *types.EIP55Address {
+	a, err := types.NewEIP55Address("0x420000000000000000000000000000000000000F")
 	if err != nil {
 		panic(err)
 	}

core/chains/evm/config/chain_scoped.go

@@ -52,6 +52,10 @@ func (e *EVMConfig) BalanceMonitor() BalanceMonitor {
 	return &balanceMonitorConfig{c: e.C.BalanceMonitor}
 }
 
+func (e *EVMConfig) TxmV2() TxmV2 {
+	return &txmv2Config{c: e.C.TxmV2}
+}
+
 func (e *EVMConfig) Transactions() Transactions {
 	return &transactionsConfig{c: e.C.Transactions}
 }

core/chains/evm/config/chain_scoped_txmv2.go

@@ -0,0 +1,25 @@
+package config
+
+import (
+	"net/url"
+	"time"
+
+	"github.com/smartcontractkit/chainlink/v2/core/chains/evm/config/toml"
+)
+
+type txmv2Config struct {
+	c toml.TxmV2
+}
+
+func (t *txmv2Config) Enabled() bool {
+	return *t.c.Enabled
+}
+
+func (t *txmv2Config) BlockTime() *time.Duration {
+	d := t.c.BlockTime.Duration()
+	return &d
+}
+
+func (t *txmv2Config) CustomURL() *url.URL {
+	return t.c.CustomURL.URL()
+}

core/chains/evm/config/chaintype/chaintype.go

@@ -23,6 +23,7 @@ const (
 	ChainZkEvm           ChainType = "zkevm"
 	ChainZkSync          ChainType = "zksync"
 	ChainZircuit         ChainType = "zircuit"
+	ChainDualBroadcast   ChainType = "dualBroadcast"
 )
 
 // IsL2 returns true if this chain is a Layer 2 chain. Notably:
@@ -39,7 +40,7 @@ func (c ChainType) IsL2() bool {
 
 func (c ChainType) IsValid() bool {
 	switch c {
-	case "", ChainArbitrum, ChainAstar, ChainCelo, ChainGnosis, ChainHedera, ChainKroma, ChainMantle, ChainMetis, ChainOptimismBedrock, ChainScroll, ChainWeMix, ChainXLayer, ChainZkEvm, ChainZkSync, ChainZircuit:
+	case "", ChainArbitrum, ChainAstar, ChainCelo, ChainGnosis, ChainHedera, ChainKroma, ChainMantle, ChainMetis, ChainOptimismBedrock, ChainScroll, ChainWeMix, ChainXLayer, ChainZkEvm, ChainZkSync, ChainZircuit, ChainDualBroadcast:
 		return true
 	}
 	return false
@@ -77,6 +78,8 @@ func FromSlug(slug string) ChainType {
 		return ChainZkSync
 	case "zircuit":
 		return ChainZircuit
+	case "dualBroadcast":
+		return ChainDualBroadcast
 	default:
 		return ChainType(slug)
 	}
@@ -144,4 +147,5 @@ var ErrInvalid = fmt.Errorf("must be one of %s or omitted", strings.Join([]strin
 	string(ChainZkEvm),
 	string(ChainZkSync),
 	string(ChainZircuit),
+	string(ChainDualBroadcast),
 }, ", "))

core/chains/evm/config/config.go

@@ -18,6 +18,7 @@ import (
 type EVM interface {
 	HeadTracker() HeadTracker
 	BalanceMonitor() BalanceMonitor
+	TxmV2() TxmV2
 	Transactions() Transactions
 	GasEstimator() GasEstimator
 	OCR() OCR
@@ -102,6 +103,12 @@ type ClientErrors interface {
 	TooManyResults() string
 }
 
+type TxmV2 interface {
+	Enabled() bool
+	BlockTime() *time.Duration
+	CustomURL() *url.URL
+}
+
 type Transactions interface {
 	ForwardersEnabled() bool
 	ReaperInterval() time.Duration