diff --git a/core/services/job/job_orm_test.go b/core/services/job/job_orm_test.go index 7a310d6f791..6e7a0b2a034 100644 --- a/core/services/job/job_orm_test.go +++ b/core/services/job/job_orm_test.go @@ -2119,7 +2119,8 @@ func TestORM_CreateJob_OCR2_With_DualTransmission(t *testing.T) { require.ErrorContains(t, jobORM.CreateJob(ctx, &jb), "invalid transmitter address in dual transmission config") dtTransmitterAddress := cltest.MustGenerateRandomKey(t) - completeDualTransmissionSpec := fmt.Sprintf(` + + metaNotSliceDualTransmissionSpec := fmt.Sprintf(` enableDualTransmission=true [relayConfig.dualTransmission] contractAddress = '0x613a38AC1659769640aaE063C651F48E0250454C' @@ -2130,6 +2131,96 @@ func TestORM_CreateJob_OCR2_With_DualTransmission(t *testing.T) { `, dtTransmitterAddress.Address.String()) + jb, err = ocr2validate.ValidatedOracleSpecToml(testutils.Context(t), config.OCR2(), config.Insecure(), baseJobSpec+metaNotSliceDualTransmissionSpec, nil) + require.NoError(t, err) + require.ErrorContains(t, jobORM.CreateJob(ctx, &jb), "dual transmission meta value key1 is not a slice") + + hintNotValidDualTransmissionSpec := fmt.Sprintf(` + enableDualTransmission=true + [relayConfig.dualTransmission] + contractAddress = '0x613a38AC1659769640aaE063C651F48E0250454C' + transmitterAddress = '%s' + [relayConfig.dualTransmission.meta] + hint = ['some-invalid-hint'] + key2 = ['val2','val3'] + `, + dtTransmitterAddress.Address.String()) + + jb, err = ocr2validate.ValidatedOracleSpecToml(testutils.Context(t), config.OCR2(), config.Insecure(), baseJobSpec+hintNotValidDualTransmissionSpec, nil) + require.NoError(t, err) + require.ErrorContains(t, jobORM.CreateJob(ctx, &jb), "dual transmission meta.hint value some-invalid-hint should be one of the following [contract_address function_selector logs calldata default_logs]") + + invalidRefundFormatDualTransmissionSpec := fmt.Sprintf(` + enableDualTransmission=true + [relayConfig.dualTransmission] + contractAddress = '0x613a38AC1659769640aaE063C651F48E0250454C' + transmitterAddress = '%s' + [relayConfig.dualTransmission.meta] + hint = ['calldata','logs'] + refund = ['0x00'] + `, + dtTransmitterAddress.Address.String()) + + jb, err = ocr2validate.ValidatedOracleSpecToml(testutils.Context(t), config.OCR2(), config.Insecure(), baseJobSpec+invalidRefundFormatDualTransmissionSpec, nil) + require.NoError(t, err) + require.ErrorContains(t, jobORM.CreateJob(ctx, &jb), "invalid dual transmission refund, format should be
:") + + invalidRefundAddressFormatDualTransmissionSpec := fmt.Sprintf(` + enableDualTransmission=true + [relayConfig.dualTransmission] + contractAddress = '0x613a38AC1659769640aaE063C651F48E0250454C' + transmitterAddress = '%s' + [relayConfig.dualTransmission.meta] + hint = ['calldata','logs'] + refund = ['0x000:50'] + `, + dtTransmitterAddress.Address.String()) + + jb, err = ocr2validate.ValidatedOracleSpecToml(testutils.Context(t), config.OCR2(), config.Insecure(), baseJobSpec+invalidRefundAddressFormatDualTransmissionSpec, nil) + require.NoError(t, err) + require.ErrorContains(t, jobORM.CreateJob(ctx, &jb), "invalid dual transmission refund address, 0x000 is not a valid address") + + invalidRefundPercentFormatDualTransmissionSpec := fmt.Sprintf(` + enableDualTransmission=true + [relayConfig.dualTransmission] + contractAddress = '0x613a38AC1659769640aaE063C651F48E0250454C' + transmitterAddress = '%s' + [relayConfig.dualTransmission.meta] + hint = ['calldata','logs'] + refund = ['0x0000000000000000000000000000000000000000:A'] + `, + dtTransmitterAddress.Address.String()) + + jb, err = ocr2validate.ValidatedOracleSpecToml(testutils.Context(t), config.OCR2(), config.Insecure(), baseJobSpec+invalidRefundPercentFormatDualTransmissionSpec, nil) + require.NoError(t, err) + require.ErrorContains(t, jobORM.CreateJob(ctx, &jb), "invalid dual transmission refund percent, A is not a number") + + invalidRefundPercentTotalFormatDualTransmissionSpec := fmt.Sprintf(` + enableDualTransmission=true + [relayConfig.dualTransmission] + contractAddress = '0x613a38AC1659769640aaE063C651F48E0250454C' + transmitterAddress = '%s' + [relayConfig.dualTransmission.meta] + hint = ['calldata','logs'] + refund = ['0x0000000000000000000000000000000000000000:50','0x0000000000000000000000000000000000000001:50'] + `, + dtTransmitterAddress.Address.String()) + + jb, err = ocr2validate.ValidatedOracleSpecToml(testutils.Context(t), config.OCR2(), config.Insecure(), baseJobSpec+invalidRefundPercentTotalFormatDualTransmissionSpec, nil) + require.NoError(t, err) + require.ErrorContains(t, jobORM.CreateJob(ctx, &jb), "invalid dual transmission refund percentages, total sum of percentages must be less than 100") + + completeDualTransmissionSpec := fmt.Sprintf(` + enableDualTransmission=true + [relayConfig.dualTransmission] + contractAddress = '0x613a38AC1659769640aaE063C651F48E0250454C' + transmitterAddress = '%s' + [relayConfig.dualTransmission.meta] + key1 = ['val1'] + key2 = ['val2','val3'] + `, + dtTransmitterAddress.Address.String()) + jb, err = ocr2validate.ValidatedOracleSpecToml(testutils.Context(t), config.OCR2(), config.Insecure(), baseJobSpec+completeDualTransmissionSpec, nil) require.NoError(t, err) diff --git a/core/services/job/orm.go b/core/services/job/orm.go index fa64404ec3a..62cc2cd596a 100644 --- a/core/services/job/orm.go +++ b/core/services/job/orm.go @@ -7,6 +7,8 @@ import ( "fmt" "reflect" "slices" + "strconv" + "strings" "time" "github.com/ethereum/go-ethereum/common" @@ -326,9 +328,19 @@ func (o *orm) CreateJob(ctx context.Context, jb *Job) error { return errors.New("invalid transmitter address in dual transmission config") } + rawMeta, ok := dualTransmissionConfig["meta"].(map[string]interface{}) + if !ok { + return errors.New("invalid dual transmission meta") + } + + if err = validateDualTransmissionMeta(rawMeta); err != nil { + return err + } + if err = validateKeyStoreMatchForRelay(ctx, jb.OCR2OracleSpec.Relay, tx.keyStore, dtTransmitterAddress); err != nil { return errors.Wrap(err, "unknown dual transmission transmitterAddress") } + } specID, err := tx.insertOCR2OracleSpec(ctx, jb.OCR2OracleSpec) @@ -1669,3 +1681,67 @@ func (r legacyGasStationServerSpecRow) toLegacyGasStationServerSpec() *LegacyGas func (o *orm) loadJobSpecErrors(ctx context.Context, jb *Job) error { return errors.Wrapf(o.ds.SelectContext(ctx, &jb.JobSpecErrors, `SELECT * FROM job_spec_errors WHERE job_id = $1`, jb.ID), "failed to load job spec errors for job %d", jb.ID) } + +func validateDualTransmissionHint(vals []interface{}) error { + accepted := []string{"contract_address", "function_selector", "logs", "calldata", "default_logs"} + for _, v := range vals { + valString, ok := v.(string) + if !ok { + return errors.Errorf("dual transmission meta value %v is not a string", v) + } + if !slices.Contains(accepted, valString) { + return errors.Errorf("dual transmission meta.hint value %s should be one of the following %s", valString, accepted) + } + } + return nil +} + +func validateDualTransmissionRefund(vals []interface{}) error { + totalRefund := 0 + for _, v := range vals { + valString, ok := v.(string) + if !ok { + return errors.Errorf("dual transmission meta value %v is not a string", v) + } + + s := strings.Split(valString, ":") + if len(s) != 2 { + return errors.New("invalid dual transmission refund, format should be
:") + } + if !common.IsHexAddress(s[0]) { + return errors.Errorf("invalid dual transmission refund address, %s is not a valid address", s[0]) + } + percent, err := strconv.Atoi(s[1]) + if err != nil { + return errors.Errorf("invalid dual transmission refund percent, %s is not a number", s[1]) + } + totalRefund += percent + } + + if totalRefund >= 100 { + return errors.New("invalid dual transmission refund percentages, total sum of percentages must be less than 100") + } + return nil +} + +func validateDualTransmissionMeta(meta map[string]interface{}) error { + for k, v := range meta { + metaFieldValues, ok := v.([]interface{}) + if !ok { + return errors.Errorf("dual transmission meta value %s is not a slice", k) + } + if k == "hint" { + if err := validateDualTransmissionHint(metaFieldValues); err != nil { + return err + } + } + + if k == "refund" { + if err := validateDualTransmissionRefund(metaFieldValues); err != nil { + return err + } + } + } + + return nil +} diff --git a/core/services/ocr2/plugins/ccip/transmitter/transmitter.go b/core/services/ocr2/plugins/ccip/transmitter/transmitter.go index 24123d03337..abb023a4251 100644 --- a/core/services/ocr2/plugins/ccip/transmitter/transmitter.go +++ b/core/services/ocr2/plugins/ccip/transmitter/transmitter.go @@ -26,6 +26,8 @@ type txManager interface { type Transmitter interface { CreateEthTransaction(ctx context.Context, toAddress common.Address, payload []byte, txMeta *txmgr.TxMeta) error FromAddress(context.Context) common.Address + + CreateSecondaryEthTransaction(context.Context, []byte, *txmgr.TxMeta) error } type transmitter struct { @@ -141,3 +143,7 @@ func (t *transmitter) forwarderAddress() common.Address { } return t.effectiveTransmitterAddress } + +func (t *transmitter) CreateSecondaryEthTransaction(ctx context.Context, bytes []byte, meta *txmgr.TxMeta) error { + return errors.New("trying to send a secondary transmission on a non dual transmitter") +} diff --git a/core/services/ocrcommon/dual_transmitter.go b/core/services/ocrcommon/dual_transmitter.go new file mode 100644 index 00000000000..efc60978f19 --- /dev/null +++ b/core/services/ocrcommon/dual_transmitter.go @@ -0,0 +1,134 @@ +package ocrcommon + +import ( + "context" + "math/big" + "net/url" + "slices" + + "github.com/ethereum/go-ethereum/common" + "github.com/pkg/errors" + + "github.com/smartcontractkit/chainlink/v2/common/txmgr/types" + "github.com/smartcontractkit/chainlink/v2/core/chains/evm/forwarders" + "github.com/smartcontractkit/chainlink/v2/core/chains/evm/txmgr" +) + +type ocr2FeedsDualTransmission struct { + txm txManager + primaryFromAddresses []common.Address + gasLimit uint64 + primaryEffectiveTransmitterAddress common.Address + strategy types.TxStrategy + checker txmgr.TransmitCheckerSpec + chainID *big.Int + keystore roundRobinKeystore + + ocr2Aggregator common.Address + txManagerOCR2 + + secondaryContractAddress common.Address + secondaryFromAddress common.Address + secondaryMeta map[string][]string +} + +func (t *ocr2FeedsDualTransmission) 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.primaryFromAddresses, t.primaryEffectiveTransmitterAddress) { + 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.primaryFromAddresses, forwarderAddress) { + forwarderAddress = common.Address{} + } + + return forwarderAddress, nil +} + +func (t *ocr2FeedsDualTransmission) CreateEthTransaction(ctx context.Context, toAddress common.Address, payload []byte, txMeta *txmgr.TxMeta) error { + roundRobinFromAddress, err := t.keystore.GetRoundRobinAddress(ctx, t.chainID, t.primaryFromAddresses...) + 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: skipped primary transmission") +} + +func (t *ocr2FeedsDualTransmission) CreateSecondaryEthTransaction(ctx context.Context, payload []byte, txMeta *txmgr.TxMeta) error { + forwarderAddress, err := t.forwarderAddress(ctx, t.secondaryFromAddress, t.secondaryContractAddress) + if err != nil { + return err + } + + if txMeta == nil { + txMeta = &txmgr.TxMeta{} + } + + dualBroadcast := true + dualBroadcastParams := t.urlParams() + + txMeta.DualBroadcast = &dualBroadcast + txMeta.DualBroadcastParams = &dualBroadcastParams + + _, err = t.txm.CreateTransaction(ctx, txmgr.TxRequest{ + FromAddress: t.secondaryFromAddress, + ToAddress: t.secondaryContractAddress, + EncodedPayload: payload, + ForwarderAddress: forwarderAddress, + FeeLimit: t.gasLimit, + Strategy: t.strategy, + Checker: t.checker, + Meta: txMeta, + }) + + return errors.Wrap(err, "skipped secondary transmission") +} + +func (t *ocr2FeedsDualTransmission) FromAddress(ctx context.Context) common.Address { + roundRobinFromAddress, err := t.keystore.GetRoundRobinAddress(ctx, t.chainID, t.primaryFromAddresses...) + if err != nil { + return t.primaryEffectiveTransmitterAddress + } + + forwarderAddress, err := t.GetForwarderForEOAOCR2Feeds(ctx, roundRobinFromAddress, t.ocr2Aggregator) + if errors.Is(err, forwarders.ErrForwarderForEOANotFound) { + // if there are no valid forwarders try to fallback to eoa + return roundRobinFromAddress + } else if err != nil { + return t.primaryEffectiveTransmitterAddress + } + + return forwarderAddress +} + +func (t *ocr2FeedsDualTransmission) urlParams() string { + values := url.Values{} + for k, v := range t.secondaryMeta { + for _, p := range v { + values.Add(k, p) + } + } + return values.Encode() +} diff --git a/core/services/ocrcommon/transmitter.go b/core/services/ocrcommon/transmitter.go index 8121f3778d2..01200bbb7cb 100644 --- a/core/services/ocrcommon/transmitter.go +++ b/core/services/ocrcommon/transmitter.go @@ -2,10 +2,7 @@ package ocrcommon import ( "context" - errors2 "errors" - "fmt" "math/big" - "net/url" "slices" "github.com/ethereum/go-ethereum/common" @@ -28,6 +25,8 @@ type txManager interface { type Transmitter interface { CreateEthTransaction(ctx context.Context, toAddress common.Address, payload []byte, txMeta *txmgr.TxMeta) error FromAddress(context.Context) common.Address + + CreateSecondaryEthTransaction(context.Context, []byte, *txmgr.TxMeta) error } type transmitter struct { @@ -99,7 +98,24 @@ func NewOCR2FeedsTransmitter( return nil, errors.New("nil keystore provided to transmitter") } - baseTransmitter := &ocr2FeedsTransmitter{ + if dualTransmissionConfig != nil { + return &ocr2FeedsDualTransmission{ + ocr2Aggregator: ocr2Aggregator, + txm: txm, + txManagerOCR2: txm, + primaryFromAddresses: fromAddresses, + gasLimit: gasLimit, + primaryEffectiveTransmitterAddress: effectiveTransmitterAddress, + strategy: strategy, + checker: checker, + chainID: chainID, + keystore: keystore, + secondaryContractAddress: dualTransmissionConfig.ContractAddress, + secondaryFromAddress: dualTransmissionConfig.TransmitterAddress, + secondaryMeta: dualTransmissionConfig.Meta, + }, nil + } + return &ocr2FeedsTransmitter{ ocr2Aggregator: ocr2Aggregator, txManagerOCR2: txm, transmitter: transmitter{ @@ -112,17 +128,7 @@ func NewOCR2FeedsTransmitter( chainID: chainID, keystore: keystore, }, - } - - if dualTransmissionConfig != nil { - return &ocr2FeedsDualTransmission{ - transmitter: *baseTransmitter, - secondaryContractAddress: dualTransmissionConfig.ContractAddress, - secondaryFromAddress: dualTransmissionConfig.TransmitterAddress, - secondaryMeta: dualTransmissionConfig.Meta, - }, nil - } - return baseTransmitter, nil + }, nil } func (t *transmitter) CreateEthTransaction(ctx context.Context, toAddress common.Address, payload []byte, txMeta *txmgr.TxMeta) error { @@ -144,6 +150,10 @@ func (t *transmitter) CreateEthTransaction(ctx context.Context, toAddress common return errors.Wrap(err, "skipped OCR transmission") } +func (t *transmitter) CreateSecondaryEthTransaction(ctx context.Context, bytes []byte, meta *txmgr.TxMeta) error { + return errors.New("trying to send a secondary transmission on a non dual transmitter") +} + func (t *transmitter) FromAddress(context.Context) common.Address { return t.effectiveTransmitterAddress } @@ -219,56 +229,6 @@ func (t *ocr2FeedsTransmitter) forwarderAddress(ctx context.Context, eoa, ocr2Ag return forwarderAddress, nil } -type ocr2FeedsDualTransmission struct { - transmitter ocr2FeedsTransmitter - - secondaryContractAddress common.Address - secondaryFromAddress common.Address - secondaryMeta map[string][]string -} - -func (t *ocr2FeedsDualTransmission) CreateEthTransaction(ctx context.Context, toAddress common.Address, payload []byte, txMeta *txmgr.TxMeta) error { - // Primary transmission - errPrimary := t.transmitter.CreateEthTransaction(ctx, toAddress, payload, txMeta) - if errPrimary != nil { - errPrimary = fmt.Errorf("skipped primary transmission: %w", errPrimary) - } - - if txMeta == nil { - txMeta = &txmgr.TxMeta{} - } - - dualBroadcast := true - dualBroadcastParams := t.urlParams() - - txMeta.DualBroadcast = &dualBroadcast - txMeta.DualBroadcastParams = &dualBroadcastParams - - // Secondary transmission - _, errSecondary := t.transmitter.txm.CreateTransaction(ctx, txmgr.TxRequest{ - FromAddress: t.secondaryFromAddress, - ToAddress: t.secondaryContractAddress, - EncodedPayload: payload, - FeeLimit: t.transmitter.gasLimit, - Strategy: t.transmitter.strategy, - Checker: t.transmitter.checker, - Meta: txMeta, - }) - - errSecondary = errors.Wrap(errSecondary, "skipped secondary transmission") - return errors2.Join(errPrimary, errSecondary) -} - -func (t *ocr2FeedsDualTransmission) FromAddress(ctx context.Context) common.Address { - return t.transmitter.FromAddress(ctx) -} - -func (t *ocr2FeedsDualTransmission) urlParams() string { - values := url.Values{} - for k, v := range t.secondaryMeta { - for _, p := range v { - values.Add(k, p) - } - } - return values.Encode() +func (t *ocr2FeedsTransmitter) CreateSecondaryEthTransaction(ctx context.Context, bytes []byte, meta *txmgr.TxMeta) error { + return errors.New("trying to send a secondary transmission on a non dual transmitter") } diff --git a/core/services/ocrcommon/transmitter_test.go b/core/services/ocrcommon/transmitter_test.go index 5f434e59c62..bb91a87d517 100644 --- a/core/services/ocrcommon/transmitter_test.go +++ b/core/services/ocrcommon/transmitter_test.go @@ -241,6 +241,7 @@ func Test_DualTransmitter(t *testing.T) { })).Twice().Return(txmgr.Tx{}, nil) require.NoError(t, transmitter.CreateEthTransaction(testutils.Context(t), toAddress, payload, nil)) + require.NoError(t, transmitter.CreateSecondaryEthTransaction(testutils.Context(t), payload, nil)) require.True(t, primaryTxConfirmed) require.True(t, secondaryTxConfirmed) diff --git a/core/services/relay/evm/contract_transmitter.go b/core/services/relay/evm/contract_transmitter.go index aead9f6ca8a..248968ec053 100644 --- a/core/services/relay/evm/contract_transmitter.go +++ b/core/services/relay/evm/contract_transmitter.go @@ -34,6 +34,8 @@ var _ ContractTransmitter = &contractTransmitter{} type Transmitter interface { CreateEthTransaction(ctx context.Context, toAddress gethcommon.Address, payload []byte, txMeta *txmgr.TxMeta) error FromAddress(context.Context) gethcommon.Address + + CreateSecondaryEthTransaction(ctx context.Context, payload []byte, txMeta *txmgr.TxMeta) error } type ReportToEthMetadata func([]byte) (*txmgr.TxMeta, error) @@ -42,28 +44,35 @@ func reportToEvmTxMetaNoop([]byte) (*txmgr.TxMeta, error) { return nil, nil } -type OCRTransmitterOption func(transmitter *contractTransmitter) +type transmitterOps struct { + reportToEvmTxMeta ReportToEthMetadata + excludeSigs bool + retention time.Duration + maxLogsKept uint64 +} + +type OCRTransmitterOption func(transmitter *transmitterOps) func WithExcludeSignatures() OCRTransmitterOption { - return func(ct *contractTransmitter) { + return func(ct *transmitterOps) { ct.excludeSigs = true } } func WithRetention(retention time.Duration) OCRTransmitterOption { - return func(ct *contractTransmitter) { + return func(ct *transmitterOps) { ct.retention = retention } } func WithMaxLogsKept(maxLogsKept uint64) OCRTransmitterOption { - return func(ct *contractTransmitter) { + return func(ct *transmitterOps) { ct.maxLogsKept = maxLogsKept } } func WithReportToEthMetadata(reportToEvmTxMeta ReportToEthMetadata) OCRTransmitterOption { - return func(ct *contractTransmitter) { + return func(ct *transmitterOps) { if reportToEvmTxMeta != nil { ct.reportToEvmTxMeta = reportToEvmTxMeta } @@ -79,10 +88,7 @@ type contractTransmitter struct { lp logpoller.LogPoller lggr logger.Logger // Options - reportToEvmTxMeta ReportToEthMetadata - excludeSigs bool - retention time.Duration - maxLogsKept uint64 + transmitterOptions *transmitterOps } func transmitterFilterName(addr common.Address) string { @@ -112,17 +118,19 @@ func NewOCRContractTransmitter( lp: lp, contractReader: caller, lggr: logger.Named(lggr, "OCRContractTransmitter"), - reportToEvmTxMeta: reportToEvmTxMetaNoop, - excludeSigs: false, - retention: 0, - maxLogsKept: 0, + transmitterOptions: &transmitterOps{ + reportToEvmTxMeta: reportToEvmTxMetaNoop, + excludeSigs: false, + retention: 0, + maxLogsKept: 0, + }, } for _, opt := range opts { - opt(newContractTransmitter) + opt(newContractTransmitter.transmitterOptions) } - err := lp.RegisterFilter(ctx, logpoller.Filter{Name: transmitterFilterName(address), EventSigs: []common.Hash{transmitted.ID}, Addresses: []common.Address{address}, Retention: newContractTransmitter.retention, MaxLogsKept: newContractTransmitter.maxLogsKept}) + err := lp.RegisterFilter(ctx, logpoller.Filter{Name: transmitterFilterName(address), EventSigs: []common.Hash{transmitted.ID}, Addresses: []common.Address{address}, Retention: newContractTransmitter.transmitterOptions.retention, MaxLogsKept: newContractTransmitter.transmitterOptions.maxLogsKept}) if err != nil { return nil, err } @@ -142,7 +150,7 @@ func (oc *contractTransmitter) Transmit(ctx context.Context, reportCtx ocrtypes. if err != nil { panic("eventTransmit(ev): error in SplitSignature") } - if !oc.excludeSigs { + if !oc.transmitterOptions.excludeSigs { rs = append(rs, r) ss = append(ss, s) vs[i] = v @@ -150,7 +158,7 @@ func (oc *contractTransmitter) Transmit(ctx context.Context, reportCtx ocrtypes. } rawReportCtx := evmutil.RawReportContext(reportCtx) - txMeta, err := oc.reportToEvmTxMeta(report) + txMeta, err := oc.transmitterOptions.reportToEvmTxMeta(report) if err != nil { oc.lggr.Warnw("failed to generate tx metadata for report", "err", err) } @@ -163,6 +171,7 @@ func (oc *contractTransmitter) Transmit(ctx context.Context, reportCtx ocrtypes. } return errors.Wrap(oc.transmitter.CreateEthTransaction(ctx, oc.contractAddress, payload, txMeta), "failed to send Eth transaction") + } type contractReader interface { diff --git a/core/services/relay/evm/contract_transmitter_test.go b/core/services/relay/evm/contract_transmitter_test.go index 5b9e1ae5981..6106389f326 100644 --- a/core/services/relay/evm/contract_transmitter_test.go +++ b/core/services/relay/evm/contract_transmitter_test.go @@ -34,6 +34,10 @@ type mockTransmitter struct { lastPayload []byte } +func (m *mockTransmitter) CreateSecondaryEthTransaction(ctx context.Context, bytes []byte, meta *txmgr.TxMeta) error { + return nil +} + func (m *mockTransmitter) CreateEthTransaction(ctx context.Context, toAddress gethcommon.Address, payload []byte, _ *txmgr.TxMeta) error { m.lastPayload = payload return nil diff --git a/core/services/relay/evm/dual_contract_transmitter.go b/core/services/relay/evm/dual_contract_transmitter.go new file mode 100644 index 00000000000..86d7d38be2e --- /dev/null +++ b/core/services/relay/evm/dual_contract_transmitter.go @@ -0,0 +1,184 @@ +package evm + +import ( + "context" + "database/sql" + "encoding/hex" + errors2 "errors" + "fmt" + "strings" + "sync" + + "github.com/ethereum/go-ethereum/accounts/abi" + "github.com/ethereum/go-ethereum/common" + gethcommon "github.com/ethereum/go-ethereum/common" + "github.com/pkg/errors" + + "github.com/smartcontractkit/libocr/offchainreporting2plus/chains/evmutil" + ocrtypes "github.com/smartcontractkit/libocr/offchainreporting2plus/types" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" + + "github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller" +) + +// TODO: Remove when new dual transmitter contracts are merged +var dtABI = `[{"inputs":[{"internalType":"bytes32[3]","name":"reportContext","type":"bytes32[3]"},{"internalType":"bytes","name":"report","type":"bytes"},{"internalType":"bytes32[]","name":"rs","type":"bytes32[]"},{"internalType":"bytes32[]","name":"ss","type":"bytes32[]"},{"internalType":"bytes32","name":"rawVs","type":"bytes32"}],"name":"transmitSecondary","outputs":[],"stateMutability":"nonpayable","type":"function"}]` + +var _ ContractTransmitter = (*dualContractTransmitter)(nil) + +type dualContractTransmitter struct { + contractAddress gethcommon.Address + contractABI abi.ABI + dualTransmissionABI abi.ABI + transmitter Transmitter + transmittedEventSig common.Hash + contractReader contractReader + lp logpoller.LogPoller + lggr logger.Logger + // Options + transmitterOptions *transmitterOps +} + +var dualTransmissionABI = sync.OnceValue(func() abi.ABI { + dualTransmissionABI, err := abi.JSON(strings.NewReader(dtABI)) + if err != nil { + panic(fmt.Errorf("failed to parse dualTransmission ABI: %w", err)) + } + return dualTransmissionABI +}) + +func NewOCRDualContractTransmitter( + ctx context.Context, + address gethcommon.Address, + caller contractReader, + contractABI abi.ABI, + transmitter Transmitter, + lp logpoller.LogPoller, + lggr logger.Logger, + opts ...OCRTransmitterOption, +) (*dualContractTransmitter, error) { + transmitted, ok := contractABI.Events["Transmitted"] + if !ok { + return nil, errors.New("invalid ABI, missing transmitted") + } + + newContractTransmitter := &dualContractTransmitter{ + contractAddress: address, + contractABI: contractABI, + dualTransmissionABI: dualTransmissionABI(), + transmitter: transmitter, + transmittedEventSig: transmitted.ID, + lp: lp, + contractReader: caller, + lggr: logger.Named(lggr, "OCRDualContractTransmitter"), + transmitterOptions: &transmitterOps{ + reportToEvmTxMeta: reportToEvmTxMetaNoop, + excludeSigs: false, + retention: 0, + maxLogsKept: 0, + }, + } + + for _, opt := range opts { + opt(newContractTransmitter.transmitterOptions) + } + + err := lp.RegisterFilter(ctx, logpoller.Filter{Name: transmitterFilterName(address), EventSigs: []common.Hash{transmitted.ID}, Addresses: []common.Address{address}, Retention: newContractTransmitter.transmitterOptions.retention, MaxLogsKept: newContractTransmitter.transmitterOptions.maxLogsKept}) + if err != nil { + return nil, err + } + return newContractTransmitter, nil +} + +// Transmit sends the report to the on-chain smart contract's Transmit method. +func (oc *dualContractTransmitter) Transmit(ctx context.Context, reportCtx ocrtypes.ReportContext, report ocrtypes.Report, signatures []ocrtypes.AttributedOnchainSignature) error { + var rs [][32]byte + var ss [][32]byte + var vs [32]byte + if len(signatures) > 32 { + return errors.New("too many signatures, maximum is 32") + } + for i, as := range signatures { + r, s, v, err := evmutil.SplitSignature(as.Signature) + if err != nil { + panic("eventTransmit(ev): error in SplitSignature") + } + if !oc.transmitterOptions.excludeSigs { + rs = append(rs, r) + ss = append(ss, s) + vs[i] = v + } + } + rawReportCtx := evmutil.RawReportContext(reportCtx) + + txMeta, err := oc.transmitterOptions.reportToEvmTxMeta(report) + if err != nil { + oc.lggr.Warnw("failed to generate tx metadata for report", "err", err) + } + + oc.lggr.Debugw("Transmitting report", "report", hex.EncodeToString(report), "rawReportCtx", rawReportCtx, "contractAddress", oc.contractAddress, "txMeta", txMeta) + + // Primary transmission + payload, err := oc.contractABI.Pack("transmit", rawReportCtx, []byte(report), rs, ss, vs) + if err != nil { + return errors.Wrap(err, "abi.Pack failed") + } + + transactionErr := errors.Wrap(oc.transmitter.CreateEthTransaction(ctx, oc.contractAddress, payload, txMeta), "failed to send primary Eth transaction") + + oc.lggr.Debugw("Created primary transaction", "error", transactionErr) + + // Secondary transmission + secondaryPayload, err := oc.dualTransmissionABI.Pack("transmitSecondary", rawReportCtx, []byte(report), rs, ss, vs) + if err != nil { + return errors.Wrap(err, "transmitSecondary abi.Pack failed") + } + + err = errors.Wrap(oc.transmitter.CreateSecondaryEthTransaction(ctx, secondaryPayload, txMeta), "failed to send secondary Eth transaction") + oc.lggr.Debugw("Created secondary transaction", "error", err) + return errors2.Join(transactionErr, err) +} + +// LatestConfigDigestAndEpoch retrieves the latest config digest and epoch from the OCR2 contract. +// It is plugin independent, in particular avoids use of the plugin specific generated evm wrappers +// by using the evm client Call directly for functions/events that are part of OCR2Abstract. +func (oc *dualContractTransmitter) LatestConfigDigestAndEpoch(ctx context.Context) (ocrtypes.ConfigDigest, uint32, error) { + latestConfigDigestAndEpoch, err := callContract(ctx, oc.contractAddress, oc.contractABI, "latestConfigDigestAndEpoch", nil, oc.contractReader) + if err != nil { + return ocrtypes.ConfigDigest{}, 0, err + } + // Panic on these conversions erroring, would mean a broken contract. + scanLogs := *abi.ConvertType(latestConfigDigestAndEpoch[0], new(bool)).(*bool) + configDigest := *abi.ConvertType(latestConfigDigestAndEpoch[1], new([32]byte)).(*[32]byte) + epoch := *abi.ConvertType(latestConfigDigestAndEpoch[2], new(uint32)).(*uint32) + if !scanLogs { + return configDigest, epoch, nil + } + + // Otherwise, we have to scan for the logs. + latest, err := oc.lp.LatestLogByEventSigWithConfs(ctx, oc.transmittedEventSig, oc.contractAddress, 1) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + // No transmissions yet + return configDigest, 0, nil + } + return ocrtypes.ConfigDigest{}, 0, err + } + return parseTransmitted(latest.Data) +} + +// FromAccount returns the account from which the transmitter invokes the contract +func (oc *dualContractTransmitter) FromAccount(ctx context.Context) (ocrtypes.Account, error) { + return ocrtypes.Account(oc.transmitter.FromAddress(ctx).String()), nil +} + +func (oc *dualContractTransmitter) Start(ctx context.Context) error { return nil } +func (oc *dualContractTransmitter) Close() error { return nil } + +// Has no state/lifecycle so it's always healthy and ready +func (oc *dualContractTransmitter) Ready() error { return nil } +func (oc *dualContractTransmitter) HealthReport() map[string]error { + return map[string]error{oc.Name(): nil} +} +func (oc *dualContractTransmitter) Name() string { return oc.lggr.Name() } diff --git a/core/services/relay/evm/dual_contract_transmitter_test.go b/core/services/relay/evm/dual_contract_transmitter_test.go new file mode 100644 index 00000000000..a5110398159 --- /dev/null +++ b/core/services/relay/evm/dual_contract_transmitter_test.go @@ -0,0 +1,164 @@ +package evm + +import ( + "context" + "encoding/hex" + "strings" + "testing" + + "github.com/ethereum/go-ethereum/accounts/abi" + gethcommon "github.com/ethereum/go-ethereum/common" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + "github.com/smartcontractkit/chainlink-common/pkg/utils/tests" + evmclimocks "github.com/smartcontractkit/chainlink/v2/core/chains/evm/client/mocks" + "github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller" + lpmocks "github.com/smartcontractkit/chainlink/v2/core/chains/evm/logpoller/mocks" + "github.com/smartcontractkit/chainlink/v2/core/chains/evm/txmgr" + "github.com/smartcontractkit/chainlink/v2/core/internal/testutils" + "github.com/smartcontractkit/chainlink/v2/core/logger" + + "github.com/smartcontractkit/libocr/gethwrappers2/ocr2aggregator" + "github.com/smartcontractkit/libocr/offchainreporting2plus/chains/evmutil" + "github.com/smartcontractkit/libocr/offchainreporting2plus/types" +) + +var sampleAddressPrimary = testutils.NewAddress() + +type mockDualTransmitter struct { + lastPrimaryPayload []byte + lastSecondaryPayload []byte +} + +func (*mockDualTransmitter) FromAddress(ctx context.Context) gethcommon.Address { + return sampleAddressPrimary +} + +func (m *mockDualTransmitter) CreateEthTransaction(ctx context.Context, toAddress gethcommon.Address, payload []byte, _ *txmgr.TxMeta) error { + m.lastPrimaryPayload = payload + return nil +} + +func (m *mockDualTransmitter) CreateSecondaryEthTransaction(ctx context.Context, payload []byte, _ *txmgr.TxMeta) error { + m.lastSecondaryPayload = payload + return nil +} + +func TestDualContractTransmitter(t *testing.T) { + t.Parallel() + + lggr := logger.TestLogger(t) + c := evmclimocks.NewClient(t) + lp := lpmocks.NewLogPoller(t) + ctx := testutils.Context(t) + // scanLogs = false + digestAndEpochDontScanLogs, _ := hex.DecodeString( + "0000000000000000000000000000000000000000000000000000000000000000" + // false + "000130da6b9315bd59af6b0a3f5463c0d0a39e92eaa34cbcbdbace7b3bfcc776" + // config digest + "0000000000000000000000000000000000000000000000000000000000000002") // epoch + c.On("CallContract", mock.Anything, mock.Anything, mock.Anything).Return(digestAndEpochDontScanLogs, nil).Once() + contractABI, _ := abi.JSON(strings.NewReader(ocr2aggregator.OCR2AggregatorMetaData.ABI)) + lp.On("RegisterFilter", mock.Anything, mock.Anything).Return(nil) + reportToEvmTxMeta := func(b []byte) (*txmgr.TxMeta, error) { + return &txmgr.TxMeta{}, nil + } + ot, err := NewOCRDualContractTransmitter(ctx, gethcommon.Address{}, c, contractABI, &mockDualTransmitter{}, lp, lggr, + WithReportToEthMetadata(reportToEvmTxMeta)) + require.NoError(t, err) + digest, epoch, err := ot.LatestConfigDigestAndEpoch(testutils.Context(t)) + require.NoError(t, err) + assert.Equal(t, "000130da6b9315bd59af6b0a3f5463c0d0a39e92eaa34cbcbdbace7b3bfcc776", hex.EncodeToString(digest[:])) + assert.Equal(t, uint32(2), epoch) + + // scanLogs = true + digestAndEpochScanLogs, _ := hex.DecodeString( + "0000000000000000000000000000000000000000000000000000000000000001" + // true + "000130da6b9315bd59af6b0a3f5463c0d0a39e92eaa34cbcbdbace7b3bfcc776" + // config digest + "0000000000000000000000000000000000000000000000000000000000000002") // epoch + c.On("CallContract", mock.Anything, mock.Anything, mock.Anything).Return(digestAndEpochScanLogs, nil).Once() + transmitted2, _ := hex.DecodeString( + "000130da6b9315bd59af6b0a3f5463c0d0a39e92eaa34cbcbdbace7b3bfcc777" + // config digest + "0000000000000000000000000000000000000000000000000000000000000002") // epoch + lp.On("LatestLogByEventSigWithConfs", + mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&logpoller.Log{ + Data: transmitted2, + }, nil) + digest, epoch, err = ot.LatestConfigDigestAndEpoch(testutils.Context(t)) + require.NoError(t, err) + assert.Equal(t, "000130da6b9315bd59af6b0a3f5463c0d0a39e92eaa34cbcbdbace7b3bfcc777", hex.EncodeToString(digest[:])) + assert.Equal(t, uint32(2), epoch) + from, err := ot.FromAccount(tests.Context(t)) + require.NoError(t, err) + assert.Equal(t, sampleAddressPrimary.String(), string(from)) +} + +func Test_dualContractTransmitterNoSignatures_Transmit_SignaturesAreNotTransmitted(t *testing.T) { + t.Parallel() + + transmitter := &mockDualTransmitter{} + + ctx := context.Background() + reportCtx := types.ReportContext{} + report := types.Report{} + var signatures = oneSignature() + + oc := createDualContractTransmitter(ctx, t, transmitter, WithExcludeSignatures()) + + err := oc.Transmit(ctx, reportCtx, report, signatures) + require.NoError(t, err) + + var emptyRs [][32]byte + var emptySs [][32]byte + var emptyVs [32]byte + emptySignaturesPayloadPrimary, err := oc.contractABI.Pack("transmit", evmutil.RawReportContext(reportCtx), []byte(report), emptyRs, emptySs, emptyVs) + require.NoError(t, err) + emptySignaturesPayloadSecondary, err := oc.dualTransmissionABI.Pack("transmitSecondary", evmutil.RawReportContext(reportCtx), []byte(report), emptyRs, emptySs, emptyVs) + require.NoError(t, err) + require.Equal(t, transmitter.lastPrimaryPayload, emptySignaturesPayloadPrimary, "primary payload not equal") + require.Equal(t, transmitter.lastSecondaryPayload, emptySignaturesPayloadSecondary, "secondary payload not equal") +} + +func Test_dualContractTransmitter_Transmit_SignaturesAreTransmitted(t *testing.T) { + t.Parallel() + + transmitter := &mockDualTransmitter{} + + ctx := context.Background() + reportCtx := types.ReportContext{} + report := types.Report{} + var signatures = oneSignature() + + oc := createDualContractTransmitter(ctx, t, transmitter) + + err := oc.Transmit(ctx, reportCtx, report, signatures) + require.NoError(t, err) + + rs, ss, vs := signaturesAsPayload(t, signatures) + withSignaturesPayloadPrimary, err := oc.contractABI.Pack("transmit", evmutil.RawReportContext(reportCtx), []byte(report), rs, ss, vs) + require.NoError(t, err) + withSignaturesPayloadSecondary, err := oc.dualTransmissionABI.Pack("transmitSecondary", evmutil.RawReportContext(reportCtx), []byte(report), rs, ss, vs) + require.NoError(t, err) + require.Equal(t, transmitter.lastPrimaryPayload, withSignaturesPayloadPrimary, "primary payload not equal") + require.Equal(t, transmitter.lastSecondaryPayload, withSignaturesPayloadSecondary, "secondary payload not equal") +} + +func createDualContractTransmitter(ctx context.Context, t *testing.T, transmitter Transmitter, ops ...OCRTransmitterOption) *dualContractTransmitter { + contractABI, err := abi.JSON(strings.NewReader(ocr2aggregator.OCR2AggregatorMetaData.ABI)) + require.NoError(t, err) + lp := lpmocks.NewLogPoller(t) + lp.On("RegisterFilter", mock.Anything, mock.Anything).Return(nil) + contractTransmitter, err := NewOCRDualContractTransmitter( + ctx, + gethcommon.Address{}, + evmclimocks.NewClient(t), + contractABI, + transmitter, + lp, + logger.TestLogger(t), + ops..., + ) + require.NoError(t, err) + return contractTransmitter +} diff --git a/core/services/relay/evm/evm.go b/core/services/relay/evm/evm.go index cf79d2aea59..0a418262988 100644 --- a/core/services/relay/evm/evm.go +++ b/core/services/relay/evm/evm.go @@ -938,6 +938,33 @@ func newOnChainContractTransmitter(ctx context.Context, lggr logger.Logger, rarg ) } +// newOnChainDualContractTransmitter creates a new dual contract transmitter. +func newOnChainDualContractTransmitter(ctx context.Context, lggr logger.Logger, rargs commontypes.RelayArgs, ethKeystore keystore.Eth, configWatcher *configWatcher, opts configTransmitterOpts, transmissionContractABI abi.ABI, ocrTransmitterOpts ...OCRTransmitterOption) (*dualContractTransmitter, error) { + transmitter, err := generateTransmitterFrom(ctx, rargs, ethKeystore, configWatcher, opts) + if err != nil { + return nil, err + } + + return NewOCRDualContractTransmitter( + ctx, + configWatcher.contractAddress, + configWatcher.chain.Client(), + transmissionContractABI, + transmitter, + configWatcher.chain.LogPoller(), + lggr, + ocrTransmitterOpts..., + ) +} + +func NewContractTransmitter(ctx context.Context, lggr logger.Logger, rargs commontypes.RelayArgs, ethKeystore keystore.Eth, configWatcher *configWatcher, opts configTransmitterOpts, transmissionContractABI abi.ABI, dualTransmission bool, ocrTransmitterOpts ...OCRTransmitterOption) (ContractTransmitter, error) { + if dualTransmission { + return newOnChainDualContractTransmitter(ctx, lggr, rargs, ethKeystore, configWatcher, opts, transmissionContractABI, ocrTransmitterOpts...) + } + + return newOnChainContractTransmitter(ctx, lggr, rargs, ethKeystore, configWatcher, opts, transmissionContractABI, ocrTransmitterOpts...) +} + func generateTransmitterFrom(ctx context.Context, rargs commontypes.RelayArgs, ethKeystore keystore.Eth, configWatcher *configWatcher, opts configTransmitterOpts) (Transmitter, error) { var relayConfig types.RelayConfig if err := json.Unmarshal(rargs.RelayConfig, &relayConfig); err != nil { @@ -1075,7 +1102,7 @@ func (r *Relayer) NewMedianProvider(ctx context.Context, rargs commontypes.Relay reportCodec := evmreportcodec.ReportCodec{} - contractTransmitter, err := newOnChainContractTransmitter(ctx, lggr, rargs, r.ks.Eth(), configWatcher, configTransmitterOpts{}, OCR2AggregatorTransmissionContractABI) + ct, err := NewContractTransmitter(ctx, lggr, rargs, r.ks.Eth(), configWatcher, configTransmitterOpts{}, OCR2AggregatorTransmissionContractABI, relayConfig.EnableDualTransmission) if err != nil { return nil, err } @@ -1089,7 +1116,7 @@ func (r *Relayer) NewMedianProvider(ctx context.Context, rargs commontypes.Relay lggr: lggr.Named("MedianProvider"), configWatcher: configWatcher, reportCodec: reportCodec, - contractTransmitter: contractTransmitter, + contractTransmitter: ct, medianContract: medianContract, }