Add special transmitter for OCR2 feeds

core/services/ocrcommon/transmitter.go

@@ -3,6 +3,7 @@ package ocrcommon
 import (
 	"context"
 	"math/big"
+	"slices"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/pkg/errors"
@@ -63,6 +64,48 @@ func NewTransmitter(
 	}, nil
 }
 
+type txManagerOCR2 interface {
+	CreateTransaction(ctx context.Context, txRequest txmgr.TxRequest) (tx txmgr.Tx, err error)
+	GetForwarderForEOAOCR2Feeds(ctx context.Context, eoa, ocr2AggregatorID common.Address) (forwarder common.Address, err error)
+}
+
+type ocr2FeedsTransmitter struct {
+	txManagerOCR2
+	transmitter
+}
+
+// NewOCR2FeedsTransmitter creates a new eth transmitter that handles OCR2 Feeds specific logic surrounding forwarders.
+// ocr2FeedsTransmitter validates forwarders before every transmission, enabling smooth onchain config changes without job restarts.
+func NewOCR2FeedsTransmitter(
+	txm txManagerOCR2,
+	fromAddresses []common.Address,
+	gasLimit uint64,
+	effectiveTransmitterAddress common.Address,
+	strategy types.TxStrategy,
+	checker txmgr.TransmitCheckerSpec,
+	chainID *big.Int,
+	keystore roundRobinKeystore,
+) (Transmitter, error) {
+	// Ensure that a keystore is provided.
+	if keystore == nil {
+		return nil, errors.New("nil keystore provided to transmitter")
+	}
+
+	return &ocr2FeedsTransmitter{
+		txManagerOCR2: txm,
+		transmitter: transmitter{
+			txm:                         txm,
+			fromAddresses:               fromAddresses,
+			gasLimit:                    gasLimit,
+			effectiveTransmitterAddress: effectiveTransmitterAddress,
+			strategy:                    strategy,
+			checker:                     checker,
+			chainID:                     chainID,
+			keystore:                    keystore,
+		},
+	}, nil
+}
+
 func (t *transmitter) CreateEthTransaction(ctx context.Context, toAddress common.Address, payload []byte, txMeta *txmgr.TxMeta) error {
 	roundRobinFromAddress, err := t.keystore.GetRoundRobinAddress(ctx, t.chainID, t.fromAddresses...)
 	if err != nil {
@@ -94,3 +137,47 @@ func (t *transmitter) forwarderAddress() common.Address {
 	}
 	return t.effectiveTransmitterAddress
 }
+
+func (t *ocr2FeedsTransmitter) CreateEthTransaction(ctx context.Context, toAddress common.Address, payload []byte, txMeta *txmgr.TxMeta) error {
+	roundRobinFromAddress, err := t.keystore.GetRoundRobinAddress(ctx, t.chainID, t.fromAddresses...)
+	if err != nil {
+		return errors.Wrap(err, "skipped OCR transmission, error getting round-robin address")
+	}
+
+	forwarderAddress, err := t.forwarderAddress(ctx, roundRobinFromAddress, toAddress)
+	if err != nil {
+		return err
+	}
+
+	_, err = t.txm.CreateTransaction(ctx, txmgr.TxRequest{
+		FromAddress:      roundRobinFromAddress,
+		ToAddress:        toAddress,
+		EncodedPayload:   payload,
+		FeeLimit:         t.gasLimit,
+		ForwarderAddress: forwarderAddress,
+		Strategy:         t.strategy,
+		Checker:          t.checker,
+		Meta:             txMeta,
+	})
+
+	return errors.Wrap(err, "skipped OCR transmission")
+}
+
+func (t *ocr2FeedsTransmitter) forwarderAddress(ctx context.Context, eoa, ocr2Aggregator common.Address) (common.Address, error) {
+	//	If effectiveTransmitterAddress is in fromAddresses, then forwarders aren't set.
+	if slices.Contains(t.fromAddresses, t.effectiveTransmitterAddress) {
+		return common.Address{}, nil
+	}
+
+	forwarderAddress, err := t.GetForwarderForEOAOCR2Feeds(ctx, eoa, ocr2Aggregator)
+	if err != nil {
+		return common.Address{}, err
+	}
+
+	// if forwarder address is in fromAddresses, then none of the forwarders are valid
+	if slices.Contains(t.fromAddresses, forwarderAddress) {
+		forwarderAddress = common.Address{}
+	}
+
+	return forwarderAddress, nil
+}

core/services/relay/evm/evm.go

@@ -566,17 +566,33 @@ func newOnChainContractTransmitter(ctx context.Context, lggr logger.Logger, rarg
 		gasLimit = uint64(*opts.pluginGasLimit)
 	}
 
-	transmitter, err := ocrcommon.NewTransmitter(
-		configWatcher.chain.TxManager(),
-		fromAddresses,
-		gasLimit,
-		effectiveTransmitterAddress,
-		strategy,
-		checker,
-		configWatcher.chain.ID(),
-		ethKeystore,
-	)
+	var transmitter Transmitter
+	var err error
 
+	switch commontypes.OCR2PluginType(rargs.ProviderType) {
+	case commontypes.Median:
+		transmitter, err = ocrcommon.NewOCR2FeedsTransmitter(
+			configWatcher.chain.TxManager(),
+			fromAddresses,
+			gasLimit,
+			effectiveTransmitterAddress,
+			strategy,
+			checker,
+			configWatcher.chain.ID(),
+			ethKeystore,
+		)
+	default:
+		transmitter, err = ocrcommon.NewTransmitter(
+			configWatcher.chain.TxManager(),
+			fromAddresses,
+			gasLimit,
+			effectiveTransmitterAddress,
+			strategy,
+			checker,
+			configWatcher.chain.ID(),
+			ethKeystore,
+		)
+	}
 	if err != nil {
 		return nil, pkgerrors.Wrap(err, "failed to create transmitter")
 	}